refactor(velo)!: unify dispatch, cancellation, and ownership

CLAUDE.md

@@ -60,7 +60,7 @@ The bug class that motivated the workspace collapse: a 0.1.1 "patch" of an inter
 1. **Only `velo` and `velo-ext` are publishable.** Any new crate added to the workspace must have `publish = false` in its `Cargo.toml` unless there is a deliberate, documented reason to publish it. Adding a third publishable crate reopens the bug class.
 2. **`velo-ext` is `=`-pinned in `[workspace.dependencies]`.** The line is:
    ```toml
-   velo-ext = { path = "lib/velo-ext", version = "=0.5.4" }
+   velo-ext = { path = "lib/velo-ext", version = "=0.5.5" }
    ```
    The `=` is load-bearing. Caret (the cargo default) lets a future "compatible" patch silently re-resolve downstream lockfiles. Do not relax this to a caret requirement.
 3. **When `velo-ext` moves past its latest published version, `velo` must carry a bump over its own latest published version too.** The `=` pin in `[workspace.dependencies]` must track `velo-ext`'s version; cargo fails to resolve the workspace otherwise.

Cargo.toml

@@ -16,7 +16,7 @@ exclude = ["examples"]
 resolver = "3"
 
 [workspace.package]
-version = "0.18.1"
+version = "0.19.0"
 edition = "2024"
 description = "Velo Active Messenging Library"
 authors = ["NVIDIA Inc. <sw-dl-dynamo@nvidia.com>"]
@@ -30,7 +30,7 @@ keywords = ["nvidia", "distributed"]
 # external implementors compile against. A non-`=` requirement would let a
 # future "compatible" patch silently re-resolve downstream lockfiles, which
 # is exactly the bug that motivated the velo / velo-ext split.
-velo-ext = { path = "lib/velo-ext", version = "=0.5.4" }
+velo-ext = { path = "lib/velo-ext", version = "=0.5.5" }
 
 # External dependencies.
 #

README.md

@@ -13,6 +13,8 @@ NOTE: Velo is experimental. Its design, development, and tests are still in prog
 
 Typed streams support graceful stop, cancellation, and ordered finalization, including prebound tickets. See [Stream lifecycle](docs/src/concepts/streaming.md).
 
+Velo 0.19 gives Velo and Messenger explicit drop lifetimes, removes value-level Messenger cloning, and removes duplicate dispatch and cancellation APIs. See [Migrate ownership, dispatch, and cancellation](docs/src/guides/migrate-dispatch-and-cancellation.md).
+
 Use `Velo::shutdown(policy)` to close resources and join owned tasks while Tokio keeps running. The instance and its handles stay valid after shutdown, and shutdown does not release them from memory. Accepted requests remain counted until their replies are submitted, including error replies. See [Shutdown and drain](docs/src/concepts/shutdown.md).
 
 Batched streams give each stream a window of 32 records in flight by default. A larger window lets a saturated consumer node queue a new stream's first record behind every other stream's backlog. See [Change the credit window](docs/src/guides/tune-batched-streaming.md#change-the-credit-window).

docs/src/SUMMARY.md

@@ -17,6 +17,7 @@
 
 # Guides
 
+- [Migrate ownership, dispatch, and cancellation to 0.19](guides/migrate-dispatch-and-cancellation.md)
 - [Write an out-of-tree transport](guides/out-of-tree-transport.md)
 - [Tune batched streaming](guides/tune-batched-streaming.md)
 - [Run rendezvous over RDMA](guides/ucx-rdma.md)

docs/src/concepts/messaging.md

@@ -62,7 +62,6 @@ Handler names that start with `_` are reserved for system handlers.
 | Mode | Behavior |
 |---|---|
 | `.spawn()` (default) | One task for each message. Two messages from one peer can run in either order. |
-| `.inline()` | One task for each message, not registered with the task tracker of the messenger |
 | `.ordered()` | One lane for each sending instance. A lane handles its messages in arrival order. Different senders run in parallel. |
 | `.ordered_global()` | One lane for all senders. This gives total order and no parallelism. |
 | `.ordered_with(OrderedConfig)` | Ordered dispatch with an explicit configuration |

docs/src/concepts/shutdown.md

@@ -18,16 +18,16 @@ sequenceDiagram
 ```
 
 1. **Gate.** The drain flag goes up. New inbound requests are refused. Responses, acks, events, and the messages of open streams continue to flow, so in-flight work can finish.
-2. **Drain.** Velo waits until no admitted request is in flight. `ShutdownPolicy::WaitForever` waits with no limit. `ShutdownPolicy::Timeout(d)` bounds the drain at `d`. The close step adds the close bound of each transport to that, so the whole call takes at most `d` plus that bound (QUIC: 2.5 seconds).
-3. **Teardown.** Velo cancels the tokens and stops the transports.
+2. **Drain.** Velo waits until no admitted request is in flight. `ShutdownPolicy::WaitForever` waits with no limit. `ShutdownPolicy::Timeout(d)` bounds the drain at `d`. Joining the mux's batchers adds up to 0.5 seconds, the time a batcher stalled on a slow peer may take to send what it holds, and the close step adds the close bound of each transport (QUIC: 2.5 seconds). Teardown adds the time that the shutdown hooks of the transports take. A hook may block, so `d` does not bound it. The ZMQ hook takes at most about 6 seconds: up to 5 seconds for a send already under way to a peer that has gone, and up to 1 second for the frames that were queued before the stop. ZMQ reports each frame that it cannot send in that second as failed. The whole call takes at most `d` plus these bounds.
+3. **Teardown.** Velo cancels the tokens and stops the transports. Each transport's shutdown hook runs once, on a dedicated thread, and the call waits for all of them. (A hook runs on the caller instead if the thread cannot be created, and on the building task if the build fails or is cancelled while transports start.) If a hook panics, the other hooks still run, and the call then panics: it cannot report the instance, or its RDMA memory, as released. A later shutdown call panics at once, without draining again.
 4. **Close.** Velo waits for `Transport::closed()` on each transport. TCP and UDS return at once, because the kernel delivers what they wrote after the process exits. QUIC keeps written data in user space until the peer acknowledges it, so it waits up to 2 seconds for its writers to finish their streams and its connections to close. Then it closes the rest by force and waits up to 0.5 seconds more, so that each frame a writer still held is reported as failed. A frame that quinn accepted but the peer did not acknowledge is lost, as a frame in the kernel send buffer is lost when a TCP peer stops reading. The writer logs a warning when that can happen. The exception is an application close from the peer: a peer that closes has gone, as a TCP peer that closes has, so that end is not a warning. A process can exit when `graceful_shutdown` returns.
 
 `Velo` is `Clone`. If two clones call `graceful_shutdown` at the same time, the first runs the sequence and the second waits for it.
 
 Work that was accepted before the drain keeps flowing through it. The messenger mux sends its records and its credit as active messages, so the gate lets through the handlers that serve accepted work:
 
 - The mux batch handlers (`_stream_batch`, and `_stream_batch.1` to `_stream_batch.15`, one for each lane), which carry the records and credit of open mux streams.
-- `_stream_cancel`, the detach, finalize and cancel handlers of SPSC anchors, and the detach and cancel handlers of MPSC anchors.
+- `_stream_cancel` and `_stream_stop`, the detach, finalize and cancel handlers of SPSC anchors, and the detach and cancel handlers of MPSC anchors.
 - The rendezvous handlers that pull a staged payload and end its lease (`_rv_acquire`, `_rv_pull`, `_rv_detach`, `_rv_release`, `_rv_lease_renew`). A record or response too large for one message is staged, and the receiver pulls it with these handlers. A draining owner answers the pull chunked, never by RDMA. `_rv_metadata` and `_rv_ref` start a new consumer, so the gate refuses them. A handle that an application staged itself can still be pulled with `get` during the drain.
 - The event handlers `_event_trigger`, `_event_trigger_request` and `_event_subscribe`. `_event_trigger` completes an awaiter of work that this node already accepted. `_event_trigger_request` completes an event that this node already created, and it acknowledges the requester. The requester sends it fire-and-forget, so a refusal would never reach the requester's wait. `_event_subscribe` answers at once for an event that is already complete, or records one subscriber for a pending event. None of the three starts new work.
 
@@ -36,7 +36,7 @@ Each handler declares its exemption in the code where it is registered, so the g
 The drain counts an exempt message while its handler runs. It does not count the stream that the message serves. This has two results:
 
 - The drain finishes at the first moment that no message is in flight. A stream message is in flight only for microseconds, and consecutive batches are at least a credit round trip apart. An open stream, busy or quiet, therefore does not hold `graceful_shutdown` open.
-- Velo then stops and joins mux sends before transport teardown. Ingress slots stay in place until full shutdown detaches their readers, so orderly shutdown does not report a false sender-drop error. The shutdown timeout bounds the drain; joining mux tasks comes after it and needs the owning Tokio runtime to keep running.
+- Velo then stops and joins mux sends before transport teardown. Ingress slots stay in place until full shutdown detaches their readers, so orderly shutdown does not report a false sender-drop error. The shutdown timeout bounds the drain; joining mux tasks comes after it, takes up to 0.5 seconds more for a batcher stalled on a slow peer, and needs the owning Tokio runtime to keep running.
 
 To let open streams finish, call `begin_drain`, wait until your streams end, and then call `shutdown`. The application must wait for its streams: the messenger drain does not count their full lifetime.
 
@@ -46,13 +46,11 @@ Use `Velo::shutdown(policy)` when an application removes a Velo instance but kee
 
 If the sender of a stream is on the same instance, shutdown cancels that sender. The `cancellation_token` of the sender fires, and later sends fail. The reader of the stream ends when the application drops or finalizes the sender. Shutdown does not end the reader itself. That needs a check on every read.
 
-Call shutdown explicitly. Dropping a `Velo` value or its last `Arc` does not perform a graceful drain.
-
-This release keeps the existing strong references between the runtime and its services. A retained Messenger keeps streaming services available after Velo is dropped. Explicit shutdown closes resources and joins owned tasks, but does not release this ownership graph from memory.
+Call shutdown explicitly to drain and join. Final Velo drop cancels its streams and stops builder-owned streaming services. It also tells each remote producer that its stream ended, while the transports are still up: an attached producer through `_stream_cancel`, and a zero-RTT producer through its mux slot close. This is best effort. It needs the Tokio runtime to keep running, a stalled peer gets the slot close only if it admits it within 0.5 seconds, and without a retained Messenger the transport can close before the notice is written. A producer that is told nothing waits on its next send. When remote producers must be told, call `Velo::shutdown` and end their streams first. A retained `Arc<Messenger>` keeps active messaging available, but does not keep those streams alive. Final Messenger drop starts transport teardown on an owned thread. Explicit shutdown waits for this cleanup, including native transport joins, before it waits for transport close. Neither Drop path waits for work to finish or reports RDMA memory as released. See [Ownership changes in 0.19](../guides/migrate-dispatch-and-cancellation.md#runtime-ownership).
 
 `graceful_shutdown` keeps its existing behavior: it drains and closes the messenger and RDMA services, but does not close the per-stream TCP or gRPC transport. `shutdown` is the complete instance shutdown operation. Application handlers that exceed a timeout can still be running after it returns.
 
-The public task tracker includes receive loops, tracked handlers, and tasks added by the application. `close()` allows `wait()` to finish when all tracked tasks exit; it does not cancel them. Ensure application tasks and idle ordering lanes can exit before waiting. An ordered handler with `with_idle_lane_ttl(None)` can leave a lane waiting forever while its router remains owned. Internal shutdown joins its own receive loops separately, so an application task cannot extend its timeout.
+The public task tracker includes receive loops, ordering lanes, and tasks added by the application. `close()` allows `wait()` to finish when all tracked tasks exit; it does not cancel them. Ensure application tasks and idle ordering lanes can exit before waiting. An ordered handler with `with_idle_lane_ttl(None)` can leave a lane waiting forever while its router remains owned. Internal shutdown joins its own receive loops separately, so an application task cannot extend its timeout.
 
 A handler panic fails its waiting caller in every dispatch mode, provided the program unwinds panics. The error reply remains counted work until the transport accepts it. An ordered lane continues with the next message.
 

docs/src/guides/migrate-dispatch-and-cancellation.md

@@ -0,0 +1,41 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# Migrate ownership, dispatch, and cancellation to 0.19
+
+Velo 0.19 makes runtime ownership explicit and removes duplicate dispatch and cancellation APIs. Stream wire formats and `velo-ext` traits do not change.
+
+## Runtime ownership
+
+Keep a `Velo` owner for as long as its streams are needed. `Velo` values and `Arc<Velo>` can still be cloned. Dropping one clone does not stop streams. Dropping the final Velo owner cancels its streams and stops the builder-owned streaming listener and tasks, even if an `Arc<Messenger>` remains alive.
+
+`Messenger` values no longer implement `Clone`. Clone `Arc<Messenger>` to share the instance:
+
+```rust,ignore
+let messages = Arc::clone(velo.messenger());
+let streams = Arc::clone(&velo); // Keep this owner while streams are active.
+```
+
+The handlers that `AnchorManager::register_handlers` installs no longer keep their manager alive, so keep your own `Arc<AnchorManager>` for as long as you use it. The manager still keeps its Messenger alive: drop the manager too before you expect final Messenger drop to start teardown. `RendezvousManager::register_handlers` no longer keeps the Messenger alive. The public handler factories `streaming::control::create_anchor_{attach,detach,finalize,cancel}_handler` and `streaming::mpsc::create_mpsc_anchor_{attach,detach,cancel}_handler` are removed; call `AnchorManager::register_handlers` instead. A manager whose handlers owned it formed a cycle with its Messenger, so neither was ever dropped.
+
+A retained Messenger supports active messages after the final Velo owner is dropped; it does not keep Velo streaming services alive. Stream handles and handler contexts can also hold Messenger references. Final Messenger drop starts immediate transport teardown and wakes remote event waits. It does not complete a response wait made through that Messenger: such a wait ends at the caller's own deadline. Velo's internal tasks that wait on a response end at teardown, except the handshake task for a peer that is not yet known, which ends at its 30-second handshake timeout. Local event completion remains available through retained event handles.
+
+Drop requests cancellation and starts transport cleanup on an owned thread. It does not drain accepted work or wait for tasks or native threads to exit. A failed `VeloBuilder::build` also cleans up, in one of two ways. If anything fails while the transports start, the build stops the transports that already started on the building task, and waits for them to close before it returns the error, as step 3 of [Shutdown and drain](../concepts/shutdown.md) says. It skips that wait if a hook panicked, because a hook that panicked may never set up what its close waits on. If the build fails after its transports started, dropping what it built cleans up as Drop does, so a listener port can stay bound for a short time after the error returns. Explicit shutdown runs the same transport cleanup, and waits for it before it waits for transport close. Continue to call `Velo::shutdown(policy).await` when those guarantees are needed. RDMA users must use explicit shutdown before relying on deregistration; Drop does not report registered memory as released. Discovery registration guards and custom frame transports remain the caller's responsibility.
+
+## Handler dispatch
+
+Replace `.inline()` with `.spawn()`, or omit the selector to use the default. The `DispatchMode` enum is removed. Select behavior with the builder methods: `.spawn()`, `.ordered()`, `.ordered_global()`, or `.ordered_with(config)`.
+
+Both old modes ran one task for each message. Neither guaranteed message order. Use `.ordered()` or `.ordered_with(...)` when handlers need to run in arrival order.
+
+Accepted handler calls still count toward graceful shutdown until the handler and its response send finish. `Messenger::tracker()` remains available.
+
+## Stream cancellation
+
+`SenderEntry::rx_closer` is removed, and `SenderEntry` has a new `closed` field. Code that constructs `SenderEntry` must omit `rx_closer` and set `closed: Default::default()`. Code that dropped its receiver to cancel a sender must call `entry.cancel()` instead. If you remove an entry from `SenderRegistry` to cancel its sender, call `cancel()` on it before you drop it.
+
+The normal `StreamController::cancel()` and `MpscStreamController::cancel()` APIs need no changes. SPSC and MPSC senders check a cancelled flag before they send an item. `SenderEntry::cancel` sets the flag first, so the next send fails. Checking the token itself takes a lock on every item: 8.7 ns, against 0.3 ns for the flag. A token cancelled in another way sets the flag only once the cancelling task yields and the sender's heartbeat task runs. Until then, each send that finds space in the channel is accepted, up to the free capacity of the channel. That task runs on the Tokio runtime that built the sender: if that runtime is gone, a direct token cancel never reaches `send`. Cancel through the stream controller or `SenderEntry::cancel` instead. A cancelled token also wakes a send that waits for channel space. Item and error sends return `SendError::ChannelClosed` after cancellation.
+
+Graceful stop remains separate. For SPSC streams, `request_stop()` signals `stop_token()` and lets the producer send buffered output and call `finalize()`. Hard cancellation also signals the stop token, but rejects further item and error sends. Do not replace a graceful stop with `cancel_token.cancel()`.