feat(streaming): add native stop and ticket lifecycle

Cargo.toml

@@ -16,7 +16,7 @@ exclude = ["examples"]
 resolver = "3"
 
 [workspace.package]
-version = "0.13.0"
+version = "0.14.0"
 edition = "2024"
 description = "Velo Active Messenging Library"
 authors = ["NVIDIA Inc. <sw-dl-dynamo@nvidia.com>"]

README.md

@@ -11,6 +11,8 @@ NOTE: Velo is experimental. Its design, development, and tests are still in prog
 
 **Documentation: [the Velo book](https://ai-dynamo.github.io/velo/)**. The source is in [`docs/src/`](docs/src/SUMMARY.md).
 
+Typed streams support graceful stop, cancellation, and ordered finalization, including prebound tickets. See [Stream lifecycle](docs/src/concepts/streaming.md).
+
 ## Crates
 
 | Crate | For | Contents |

docs/src/concepts/batched-streaming.md

@@ -1,6 +1,6 @@
 # Batched streaming
 
-The messenger mux carries every stream to one peer over the Messenger connection that already exists to that peer. It packs the records for that peer into `_stream_batch` active messages. Its transport key is `messenger-mux-v1`. The mux is opt-in and is negotiated per attach. Senders do not change: `StreamSender::send` stages a record, and the layer below it decides when to write.
+The messenger mux carries every stream to one peer over the Messenger connection that already exists to that peer. It packs the records for that peer into `_stream_batch` active messages. Its transport key is `messenger-mux-v2`. The mux is opt-in and is negotiated per attach. Senders do not change: `StreamSender::send` stages a record, and the layer below it decides when to write.
 
 This chapter describes how the mux works. The [Tune batched streaming](../guides/tune-batched-streaming.md) guide tells you how to configure it. The [Batched streaming design](../development/batched-streaming-design.md) chapter records why it works this way and which alternatives were rejected.
 
@@ -85,12 +85,12 @@ _stream_batch payload:
   [16 B batch header][record_count x record]
 
 batch header:
-  [u8 mux_version = 1][u8 flags][u16 record_count][u64 peer_epoch][u32 batch_seq]
+  [u8 mux_version = 2][u8 flags][u16 record_count][u64 peer_epoch][u32 batch_seq]
 
 record:
   [u8 record_type][u32 slot][u32 frame_seq][u32 len][len bytes body]
 
-record_type: 0 = Data, 1 = OpenSlot, 2 = CloseSlot, 3 = CreditUpdate, 4 = SlotHeartbeat
+record_type: 0 = Data, 1 = OpenSlot, 2 = CloseSlot, 3 = CreditUpdate, 4 = SlotHeartbeat, 5 = LifecycleSlot
 ```
 
 Every multi-byte field is big-endian, in the header and in each record. Senders write `flags` as zero, and receivers ignore unknown bits.
@@ -105,6 +105,7 @@ Record bodies:
 - **`OpenSlot`** carries `[u64 anchor_id][u64 session_id]`. This is the 16-byte attach handshake, moved into a record.
 - **`CloseSlot`** carries `[u8 reason]`: `0` terminal sent, `1` peer gone, `2` unknown slot, `3` protocol error.
 - **`CreditUpdate`** carries `[u32 delta]` from receiver to sender.
+- **`LifecycleSlot`** carries the 64-bit session identity from `OpenSlot` and a one-byte action (0 = stop, 1 = cancel). Stop leaves the slot open for remaining output. Cancel takes precedence for the same session. Both the queued signal and the live producer check session identity, so a delayed signal cannot target a reused slot even if its compact generation has wrapped. Early stop is retained by the pre-bind until `OpenSlot` claims it.
 - **`SlotHeartbeat`** has no body. The decoder accepts it, but no sender emits it. See [Heartbeats](#heartbeats).
 
 `CloseSlot` travels in both directions and has no direction bit. The reason carries the direction. `TerminalSent` and `PeerGone` travel from slot owner to receiver. `UnknownSlot` and `ProtocolError` travel from receiver to slot owner. Both sides can hold a slot at the same dense index, and the reason tells them apart.
@@ -356,19 +357,19 @@ At serving depth, `Manual` is both higher and repeatable. A batcher that writes
 
 ## Negotiation and compatibility
 
-The attach selects the mux. No wire magic exists. `AnchorAttachRequest` and `MpscAnchorAttachRequest` carry `supported_transport_keys` (with `#[serde(default)]`, so an older sender deserializes as advertising nothing). The attach handler intersects the sender's keys with its installed transports. It picks `messenger-mux-v1` only when both sides name it. Otherwise it answers with its default key.
+The attach selects the mux. No wire magic exists. `AnchorAttachRequest` and `MpscAnchorAttachRequest` carry `supported_transport_keys` (with `#[serde(default)]`, so an older sender deserializes as advertising nothing). The attach handler intersects the sender's keys with its installed transports. It picks `messenger-mux-v2` only when both sides name it. Otherwise it answers with its default key.
 
 The sender reads the answer:
 
-- A key other than `messenger-mux-v1` is the per-stream path, and the credit fields do not apply. Every older receiver answers this way.
-- `messenger-mux-v1` with a window opens a slot that already holds that window.
-- `messenger-mux-v1` with no window (`initial_credit` zero) is refused. No shipped receiver answers this, and a node with a mux cannot advertise a zero window. A fallback to another transport reaches nothing that listens, and the stream hangs until the watchdog fires.
+- A key other than `messenger-mux-v2` is the per-stream path, and the credit fields do not apply. Every older receiver answers this way.
+- `messenger-mux-v2` with a window opens a slot that already holds that window.
+- `messenger-mux-v2` with no window (`initial_credit` zero) is refused. No shipped receiver answers this, and a node with a mux cannot advertise a zero window. A fallback to another transport reaches nothing that listens, and the stream hangs until the watchdog fires.
 
-A node with the mux enabled registers both `messenger-mux-v1` and its configured per-stream transport, so it still serves older peers. `resolve_transport` fails on an unknown key in a non-empty registry, so a receiver that answers the mux on its own breaks every older sender. SPSC and MPSC anchors both negotiate the mux.
+A node with the mux enabled registers both `messenger-mux-v2` and its configured per-stream transport, so it still serves older peers. `resolve_transport` fails on an unknown key in a non-empty registry, so a receiver that answers the mux on its own breaks every older sender. SPSC and MPSC anchors both negotiate the mux.
 
 `StreamSender::negotiated_transport()` returns the key that the attach settled on. It returns `None` for a same-worker attach, which uses no transport. Compare it with the public constant `MESSENGER_MUX_KEY`.
 
-`MuxConfig::enabled = false` is the rollback. The node stops advertising `messenger-mux-v1`, and the next attach negotiates the per-stream path with no code or wire change. See [Zero-RTT stream setup](#zero-rtt-stream-setup) for the order when tickets are in use.
+`MuxConfig::enabled = false` is the rollback. The node stops advertising `messenger-mux-v2`, and the next attach negotiates the per-stream path with no code or wire change. See [Zero-RTT stream setup](#zero-rtt-stream-setup) for the order when tickets are in use.
 
 ## Observability
 
@@ -381,3 +382,7 @@ The mux series all start with `velo_streaming_mux_`. The most important ones:
 - `velo_messenger_ordered_lane_wait_seconds{handler="_stream_batch"}` shows the ingress lane wait, one sample per batch.
 
 The [Metrics reference](../appendix/metrics.md) lists every series. [Stream saturation](../operations/saturation.md) explains how to read them under load.
+
+### Peer loss
+
+A batcher with live producer slots checks its selected message transport every five seconds. A failed check closes the producer slots and triggers their cancellation and stop tokens, including idle producers. The check is per peer, not per token or stream. It does not change the selected transport or permit a fallback.

docs/src/concepts/streaming.md

@@ -38,7 +38,9 @@ while let Some(frame) = anchor.next().await {
 | `SenderError(String)` | The producer could not serialize an item |
 | `TransportError(String)` | The network failed during delivery |
 
-The consumer can cancel the stream with `anchor.cancel()` or with a cloned `StreamController`. The producer sees the cancel through `sender.cancellation_token()`.
+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.
 
 ## Many producers (MPSC)
 
@@ -89,4 +91,4 @@ sender.finalize()?;
 
 Call `prebind_anchor` from a runtime context, because it starts tasks. A ticket that the worker never opens does not stay forever. After a 60 s accept window, the bind is reclaimed and the anchor sees `SenderDropped`. `velo_streaming_unclaimed_bind_reaped_total` counts these tickets.
 
-A sender opened from a ticket never sent `_anchor_attach`, so it has no cancel handle, and `sender.cancellation_token()` never fires. Use the result of `send` and `finalize` instead. A consumer that has gone shows as a send error.
+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.

docs/src/development/batched-streaming-design.md

@@ -126,7 +126,7 @@ The price of `Manual` is also known. A per-pass flush caps a batch at that pass'
 
 ### Negotiate per attach and register both transports
 
-The receiver answers `messenger-mux-v1` only when the sender named it. `resolve_transport` fails on an unknown key, so a receiver that chooses the mux on its own breaks every older sender. A node with the mux registers both keys. The wire version lives in the key, so incompatible versions never pair. A zero window with the mux key is refused, because a fallback connects to a transport where nothing listens.
+The receiver answers `messenger-mux-v2` only when the sender named it. `resolve_transport` fails on an unknown key, so a receiver that chooses the mux on its own breaks every older sender. A node with the mux registers both keys. The wire version lives in the key, so incompatible versions never pair. A zero window with the mux key is refused, because a fallback connects to a transport where nothing listens.
 
 ### Credit return: the pump posts, the reconcile decides
 
@@ -194,7 +194,7 @@ The mux is a user of the Messenger send path, not something a transport implemen
 
 The additions this work made to `velo-ext` were shaped to last. `Transport::max_message_size(target) -> Option<usize>` has a default, and `None` means unknown, which costs the caller a conservative budget. `SendOutcome::Pending(SendAdmission)` changed a published enum, so it shipped once, as a coordinated `velo-ext` and `velo` release. The egress recorders on `TransportObservability` have no-op defaults.
 
-The cost of deferral is bounded. Out-of-tree `FrameTransport`s get no multiplexing, and nothing else breaks. A receiver offers `messenger-mux-v1` only when the mux is installed and enabled, so a deployment without the mux degrades correctly.
+The cost of deferral is bounded. Out-of-tree `FrameTransport`s get no multiplexing, and nothing else breaks. A receiver offers `messenger-mux-v2` only when the mux is installed and enabled, so a deployment without the mux degrades correctly.
 
 Methods that the mux needs outside the trait (`advertised_limits`, `connect_negotiated`, `take_drain_signal`, `prebind`, `release_bind`, `close_claimed_slot`) are inherent on the concrete `MessengerMuxTransport`. `AnchorManager` already holds the concrete type, so none of them needs a trait change.
 

docs/src/guides/tune-batched-streaming.md

@@ -24,7 +24,7 @@ let velo = Velo::builder()
     .await?;
 ```
 
-Enable the mux on both nodes of a pair. An attach uses the mux only when both sides advertise `messenger-mux-v1`. Every other pair uses the per-stream path.
+Enable the mux on both nodes of a pair. An attach uses the mux only when both sides advertise `messenger-mux-v2`. Every other pair uses the per-stream path.
 
 Call `messenger_mux` once per `Velo` instance. A second call returns an error.
 
@@ -86,7 +86,7 @@ All settings are fields of `MuxConfig`. Always build it with `..Default::default
 
 | Field | Default | What it controls |
 |---|---|---|
-| `enabled` | `false` | Installs the mux and advertises `messenger-mux-v1`. Setting it back to `false` is the rollback. |
+| `enabled` | `false` | Installs the mux and advertises `messenger-mux-v2`. Setting it back to `false` is the rollback. |
 | `max_batch_bytes` | 60 KiB | The configured cap on one batch. The eager budget and the 64 KiB coalescing threshold also clamp it. |
 | `initial_credit` | 256 | Data credit C per slot. Each slot buffer holds C+1 records. Zero is refused at build time. |
 | `slot_byte_budget` | 1 MiB | Bytes one slot can hold in flight, and the cap on its withheld queue. Zero means the default. |
@@ -165,7 +165,7 @@ let sender = match envelope.ticket {
 };
 ```
 
-A zero-RTT sender has no cancel handle, so its `cancellation_token` never fires. When the consumer drops the anchor, the producer's next `send` returns an error. An idle producer also receives a close from the consumer.
+A ticket sender observes `stop_token()` and `cancellation_token()` through the mux's slot lifecycle. Stop leaves the stream open for final output. Cancel or consumer drop wakes an idle producer and ends delivery; it does not need a later send.
 
 The ticket stays valid for the 60-second accept window. After the window, the consumer reaps the bind and sees `SenderDropped`.
 

docs/src/operations/benchmarking.md

@@ -115,7 +115,7 @@ Each rule below exists because a measurement without it was wrong.
 5. **Compare reps at a matched backlog draw.** On the external rig, one or more mocker processes fall behind during the opening burst and keep that backlog. The number of processes that hold the backlog (the "holders") sets throughput, ITL and end-to-end latency for every arm. Compare two arms only at the same holder count.
 6. **Do not fix the arm order.** A matrix that always ran the same arm second put that arm in a degraded band in every rep. Its request-level numbers measured run position, not the setting.
 7. **Change one setting per arm.** A pair that differed in two settings attributed a result to the wrong one.
-8. **Assert the transport per request.** Fail any request whose negotiated key is not `messenger-mux-v1`. A silent fallback measures the per-stream path under a mux label.
+8. **Assert the transport per request.** Fail any request whose negotiated key is not `messenger-mux-v2`. A silent fallback measures the per-stream path under a mux label.
 9. **Record the build.** Write the velo commit and its dirty state into each rep's metadata. If the installed build and the checkout differ, fail the rep.
 10. **Report latency beside throughput.** A change that adds latency can raise throughput by letting the consumer catch up. Watch TTFT in particular, because any windowed flush policy can make it worse.
 11. **Prove a regression test before you trust it.** Revert the fix and make sure that the test fails. A test that passes with the fix reverted proves nothing.

docs/src/operations/response-plane-performance.md

@@ -164,13 +164,15 @@ UCX did not beat TCP, and it did not hold its tail. The failure chain:
 5. The reader-pump watchdog (3 × 5 s) killed 1,722 streams. Every log line showed `anchor_frame_tx_len=0` and `transport_rx_len=0`. `closed_slot` drops were 1,722 × 255 exactly, so no record had arrived before each kill.
 6. The worker was healthy the whole time. It completed all 27,702 of its requests.
 
-The defect is that the UCX send path has no backpressure edge beyond its shared ring:
+The measured revision had no UCX backpressure edge beyond its shared ring:
 
 - `ucp_am_send_nbx` never refuses. An exhausted endpoint returns a request pointer, not an error.
 - `inflight_ops` counts posted operations but does not gate admission.
 - The per-peer admission gate queues without a bound.
 - All peers share one 1,024-entry ring into the progress thread. TCP gives each connection a 256-entry channel and a blocking write.
 
+Version 0.14 bounds queued and outstanding sends per peer. A send permit is released on UCX completion or failure, and peer retirement cancels pending admission. The progress thread can still process control and endpoint cleanup when a data peer is full. This change requires a new saturation measurement; it does not change the historical results below.
+
 Because admission never blocks, the on-admission flush policy never parks, and batches collapse. UCX carried 6.97 to 8.65 records per batch against 18 to 29 over TCP with the same settings. That is 2.6 to 4.1 times the message rate for the same records.
 
 These alternatives were checked and refuted:
@@ -181,7 +183,7 @@ These alternatives were checked and refuted:
 - UCX scheduling is not unfair. The failed peer carried eight times the load of the others.
 - Spinning progress threads do not starve the CPU. There were 9 progress threads on 288 cores.
 
-No tuning setting reaches this mechanism. The fix has two parts. The UCX transport needs a backpressure edge (gate admission on in-flight operations or on a per-peer ring share). Liveness needs a heartbeat path that data cannot block, or a watchdog that can tell a starved stream from a dead one. The UCX inbound path now records frames like every other transport. The other two fixes are not built. [RDMA performance](rdma-performance.md) covers the UCX transport outside the response plane.
+No tuning setting reaches this mechanism. The fix has two parts. The UCX transport needs a backpressure edge (gate admission on in-flight operations or on a per-peer ring share). Liveness needs a heartbeat path that data cannot block, or a watchdog that can tell a starved stream from a dead one. The UCX inbound path records frames like every other transport. Version 0.14 adds the bounded admission edge. Saturation measurements must still check whether data backlog delays stream heartbeats beyond the watchdog. [RDMA performance](rdma-performance.md) covers the UCX transport outside the response plane.
 
 ## Instrumentation cost
 

docs/src/operations/saturation.md

@@ -2,7 +2,7 @@
 
 If streaming traffic shows sender errors, missed deadlines, or a `Dropped` frame with no producer crash, use this runbook. The usual cause is saturation. The producer generates frames faster than the consumer drains them. Velo reports saturation through Prometheus counters and one log line. This page tells you which signals to read and what to do.
 
-The signals differ between the per-stream path and the messenger mux. Read [The per-stream cascade](#the-per-stream-cascade) for streams with one TCP or gRPC connection each. Read [Saturation under the mux](#saturation-under-the-mux) for streams that negotiated `messenger-mux-v1`.
+The signals differ between the per-stream path and the messenger mux. Read [The per-stream cascade](#the-per-stream-cascade) for streams with one TCP or gRPC connection each. Read [Saturation under the mux](#saturation-under-the-mux) for streams that negotiated `messenger-mux-v2`.
 
 ## The per-stream cascade