Streaming

A stream moves a sequence of typed items from a producer to a consumer. The consumer owns a StreamAnchor<T>. A producer attaches to the anchor and gets a StreamSender<T>. The anchor has a compact u128 handle, so the consumer can send the handle to a producer on a different worker.

The handle is all that a peer needs to attach to the anchor, cancel it, or open its slot on the mux, and velo does not authenticate peers. So the local ID in the handle is random: 63 bits, and never 0. With a counter, any host that could reach a node could attach to the next handle and take over the stream of another request. Send the handle only to the producer that needs it.

Single producer (SPSC)

A default anchor accepts one producer at a time.

use futures::StreamExt;
use velo::StreamFrame;

// Consumer: create an anchor and send its handle to the producer.
let mut anchor = node_b.create_anchor::<String>();
let handle = anchor.handle();

// Producer, on the same node or another node.
let sender = node_a.attach_anchor::<String>(handle).await?;
sender.send("hello".into()).await?;
sender.finalize()?;

// Consumer.
while let Some(frame) = anchor.next().await {
    match frame? {
        StreamFrame::Item(s) => println!("{s}"),
        StreamFrame::Finalized => break,
        _ => {}
    }
}

The consumer can request graceful stop with anchor.controller().request_stop(). The producer observes sender.stop_token() and can send its final output, then call finalize(). Stop is idempotent and does not discard buffered output.

The consumer can cancel the stream with anchor.cancel() or a cloned StreamController. Cancel triggers both producer tokens and ends delivery. Dropping the consumer also cancels. Stop and cancel made before attach or ticket open are retained; the producer observes them when it opens, even if it sends no data. Transport failure remains a stream error.

Know when the output has left

finalize returns before the stream's output leaves the producer's node. Under the messenger mux, records wait on the producer for the consumer's credit, and a slow consumer can hold them there for as long as it does not read. A producer that shuts its node down in that time cuts the stream short.

sender.retired() returns a future. The future resolves once velo on the producer's node has handed on all of the stream's output; the table below says what that means on each path. It resolves after finalize, detach, or drop. Get the future before any of the three, because each one consumes the sender.

let retired = sender.retired();
sender.finalize()?;
match retired.await {
    Retirement::Sent => {}      // Handed on; see the table and the staged-item exception below.
    Retirement::Discarded => {} // Output was thrown away before it left this node.
    _ => {}                     // `Retirement` is not exhaustive.
}

Retirement::Sent means that velo handed the terminal (Finalized, Detached, or Dropped) on behind every record before it. It is a statement about the producer's node, not about the consumer: it does not say that the consumer received anything, and what "handed on" means depends on the path of the stream:

Retirement::Discarded means that the output was thrown away before velo handed it on. Over the messenger mux: the consumer cancelled or dropped its anchor, the consumer's process went away, the transport refused the batch, or the producer's node shut down while it held the output. On the same worker: the consumer dropped its anchor, so the terminal had no channel to enter; a cancel through a StreamController while the anchor lives leaves the terminal in the anchor's channel, which is Sent. On a per-stream transport: the pump's write failed, the connection ended first, or the gRPC server did not take the stream; a per-stream TCP pump cannot see a consumer that left before it writes (known limitation). A task or batch that dies with the terminal also gives Discarded, so the future does not wait forever.

The future shares the state that the sender already keeps for the stream. A sender can hand out any number of them, and each one wakes. A stream that nobody asks about pays one compare-exchange when it retires. Batched streaming shows how the mux reports it, and what it costs.

Byte items

Some streams carry items that the application has already serialized, such as response chunks. For these, use Bytes as the item type, send with send_bytes, and read with next_bytes. Inside a hand-written Stream, use poll_next_bytes.

use bytes::Bytes;
use velo::StreamFrame;

// Producer.
let sender = node_a.open_anchor_stream::<Bytes>(handle, ticket).await?;
sender.send_bytes(&chunk).await?;
sender.finalize()?;

// Consumer of `node_b.create_anchor::<Bytes>()`.
while let Some(frame) = anchor.next_bytes().await {
    match frame? {
        StreamFrame::Item(chunk) => write(chunk),
        StreamFrame::Finalized => break,
        _ => {}
    }
}

send_bytes writes the frame by hand, and the frame is the same as the one send writes for a Bytes item. So each end can use either form. next_bytes handles every frame other than an item in the same code as next, so heartbeats, errors, Detached and the end of the stream behave the same.

The difference is the copies of each payload. Over the mux:

These counts are from the source, not measured. On the same worker, there is no batch and no ingress, so send_bytes and next_bytes make one copy in total.

A kept item keeps its frame buffer in memory. The frame is the item plus at most 11 bytes.

Write an item in place

send_bytes copies a payload that the application encoded into its own buffer. To skip that copy, encode straight into the frame with ByteItemBuilder, and send the result with send_frame. The builder needs no sender, so the encode can run before the producer holds one.

use velo::streaming::ByteItemBuilder;

let mut item = ByteItemBuilder::with_capacity(size_hint);
rmp_serde::encode::write(item.buffer(), &token)?;
sender.send_frame(item.finish()?).await?;

buffer is a Vec<u8> that already holds the 11-byte item header. Append the payload to it. finish writes the payload's length into the header, and refuses a frame whose header was changed. send_frame has the rules of send_bytes for a cancel, a full channel, and the item limit.

A payload that fits the capacity costs one allocation for the whole frame, and no copy before the batch. A larger payload grows the buffer as a Vec grows.

The frame decodes as the send_bytes frame does, through next, next_bytes, and a typed consumer. The bytes differ in one way: the length is always a 5-byte bin32, because it is written after the payload. So an item under 64 KiB is up to 3 bytes longer on the wire. On the same worker, a frame with spare capacity costs the consumer one small allocation when it takes the buffer as Bytes. Over the mux, the batch copy removes that cost.

Why separate methods: next cannot skip serde for Bytes alone. Rust has no specialization, and a type check at run time needs T: 'static on the Stream impl of StreamAnchor, which is a breaking change. A marker item type would still go through serde on the producer. Why the mux still copies each record out of its batch is in Batched streaming design.

Many producers (MPSC)

create_mpsc_anchor makes an anchor that accepts many producers. Each producer gets a SenderId when it attaches, and each frame carries the SenderId of its producer. Detached and Dropped from one producer do not end the stream. The consumer ends the stream: it cancels, drops the anchor, or lets its unattached timeout (MpscAnchorConfig::unattached_timeout, none by default) fire after all producers leave. A producer cannot finalize an MPSC anchor.

How frames travel

Frames travel on one of two paths:

• A frame transport. The builder sets it with stream_config. StreamConfig::Tcp is the default. StreamConfig::Grpc needs the grpc feature. Each stream sends its own frames. This path carries a stream when the peer does not offer the mux, or when the mux is off.
• Batched streaming (the mux). The builder installs it by default. messenger_mux(MuxConfig) changes its settings or turns it off. Records from many streams to the same peer share one messenger frame, with credit-based flow control for each stream. See Batched streaming.

let node = Velo::builder()
    .add_transport(tcp)
    .stream_config(StreamConfig::Tcp(None))? // TCP streams on a port that the OS assigns
    .build()
    .await?;

Zero-RTT setup

attach_anchor makes one _anchor_attach round trip before the first item can move. Sometimes the consumer knows which worker will produce. For example, the consumer is about to send that worker a request. Then the consumer can make a ticket and put it in its own request, and the worker opens the stream with no round trip.

// Consumer: make a ticket.
let mut anchor = node_b.create_anchor::<String>();
let handle = anchor.handle();
let Some(ticket) = node_b.prebind_anchor(handle) else {
    // No ticket. Attach the usual way.
    let sender = node_a.attach_anchor::<String>(handle).await?;
    /* ... */
    return Ok(());
};

// The ticket travels inside the request of the application.

// Worker: open the stream on the ticket.
let sender = node_a.open_anchor_stream::<String>(handle, ticket).await?;
sender.send("hello".into()).await?;
sender.finalize()?;

prebind_anchor returns None when it has nothing to make. The caller then uses attach_anchor as usual:

• If no mux is installed, None is the normal result, and Velo records nothing.
• For all other causes, Velo logs at debug level and counts outcome="error" under velo_streaming_anchor_operations_total{operation="prebind"}. These causes are an MPSC anchor, the handle of another node, an anchor that is already attached or pre-bound, and an anchor that was removed. Call prebind_anchor only on SPSC anchors that this node created.

Call prebind_anchor from a runtime context, because it starts tasks. A ticket that the worker never opens does not stay forever. After the configured accept window (180 seconds by default), the bind is reclaimed and the anchor sees SenderDropped. velo_streaming_unclaimed_bind_reaped_total counts these tickets.

Ticket senders support stop_token() and cancellation_token() through messenger mux version 2. The existing slot-open exchange carries the stream session identity; no extra stream registration is needed. A peer epoch and slot generation reject stale lifecycle signals. Applications that require this lifecycle must require messenger-mux-v2; legacy attach fallback does not establish that capability.

Reservation renewal

VeloBuilder::reservation_config sets the default unattached lifetime (120 seconds) and the accept window of the messenger mux (180 seconds). The accept window applies to every mux bind: a prepared slot and an ordinary attach. The per-stream TCP transport keeps its own fixed 60-second accept window. Per-anchor AnchorConfig::unattached_timeout overrides the ordinary lifetime. The unattached default applies to SPSC anchors only. An MPSC anchor has an unattached timeout only when MpscAnchorConfig::unattached_timeout sets one, and renewal does not extend it. To make anchors wait for a sender with no limit, set unattached_timeout to Duration::MAX. An AnchorManager that drops cancels the SPSC and MPSC anchors still in its registries. Unlike stop, a drop does not ask attached senders to cancel.

ReservationConfig is #[non_exhaustive], so start from the default and change its fields:

let mut reservations = ReservationConfig::default();
reservations.prebind_timeout = Duration::from_secs(600);
let velo = Velo::builder().reservation_config(reservations).build().await?;

Capture AnchorReservation::new(handle, ticket.as_ref()) when a request takes responsibility for a reply. Keep the reservation while that request waits or executes before attachment. Periodically pass held reservations to AnchorManager::renew_reservations. The call groups them by owner and sends at most 256 entries per remote batch. It does not attach a producer. The application owns the collection, renewal cadence, and call deadline. Velo does not schedule renewal. Choose a cadence shorter than the reservation lifetime.

Renewal extends only a live, unattached reservation. Attached, expired, cancelled, and replaced reservations are successful no-ops, including attachment races. Renewal cannot revive an expired stream. A transport failure returns an error. A peer that runs an older Velo has no _anchor_renew handler, so renewal to it returns an error too. Successful delivery does not prove that a reservation still exists.

Release the held reservation after downstream acceptance, attachment, cancellation, or failure. The accept window is the only limit on an unclaimed bind. A renewal loop that outlives its request holds the bind, its buffer, and its count on a mux lane until the loop stops. Renewal does not extend an application request deadline or replace producer heartbeats. A prepared ticket still identifies one slot generation; renewal does not make that ticket reusable after detach. After a detach, renewal with that ticket extends the unattached lifetime of the anchor.