velo_ext/streaming.rs
1// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Frame-level transport abstraction for ordered delivery streaming.
5//!
6//! This module defines the [`FrameTransport`] trait boundary consumed by the
7//! Velo streaming runtime and implemented by the in-tree `TcpFrameTransport`,
8//! `GrpcFrameTransport` and messenger mux.
9//! Out-of-tree implementors (custom RDMA, hardware transports, alternative
10//! streaming substrates) should `impl FrameTransport for MyTransport` against
11//! this contract.
12//!
13//! # Endpoint resolution
14//!
15//! Streaming transports advertise their listener endpoint(s) into the local
16//! [`WorkerAddress`] via [`FrameTransport::address`], keyed by the same
17//! [`TransportKey`] returned by [`FrameTransport::key`]. The Velo builder
18//! merges streaming entries into the local PeerInfo's WorkerAddress alongside
19//! messenger transport entries.
20//!
21//! When a peer is registered (via `Velo::register_peer` or discovery), the
22//! runtime calls [`FrameTransport::register`] on every installed streaming
23//! transport. The transport extracts its own entry from the peer's
24//! WorkerAddress, decodes the endpoint(s), and caches the resolved socket
25//! address keyed by the peer's [`WorkerId`].
26//!
27//! [`FrameTransport::connect`] then looks up the cached address by
28//! [`WorkerId`] — no endpoint string is exchanged on the streaming attach
29//! handshake.
30
31use anyhow::Result;
32use futures::future::BoxFuture;
33
34use crate::id::{PeerInfo, TransportKey, WorkerAddress, WorkerId};
35
36/// Transport abstraction for frame-level ordered delivery.
37///
38/// # Ordered-Delivery Contract
39///
40/// All frames -- including data frames (`Item`, `Heartbeat`) **and** sentinel
41/// frames (`Dropped`, `Detached`, `Finalized`, `TransportError`) -- MUST travel
42/// the **same physical channel** established by [`FrameTransport::bind`] /
43/// [`FrameTransport::connect`]. Sentinels MUST NOT be injected via a side
44/// channel; the FIFO ordering guarantee of the underlying channel is
45/// load-bearing for the correctness of the streaming protocol.
46///
47/// Implementations MUST preserve send order: a frame sent before another MUST
48/// be received before that other frame on the corresponding
49/// [`flume::Receiver`].
50///
51/// # Usage
52///
53/// The transport operates at the raw byte level. Callers are responsible for
54/// serializing frame values to `Vec<u8>` before sending via
55/// [`flume::Sender::send_async`], and for deserializing bytes received from
56/// [`flume::Receiver::recv_async`].
57///
58/// # Async Design
59///
60/// Both `bind` and `connect` return [`BoxFuture`] to support async
61/// implementations (e.g., network setup). The heap allocation is acceptable
62/// because these are setup-path calls, not per-frame hot-path operations.
63pub trait FrameTransport: Send + Sync {
64 /// Identifies this transport's entry in [`WorkerAddress`].
65 ///
66 /// Mirrors [`crate::Transport::key`]. Used by the streaming attach
67 /// handshake to tell the client which `FrameTransport` to call
68 /// [`connect`](Self::connect) on, and by [`register`](Self::register) to
69 /// look up the peer's matching endpoint entry.
70 fn key(&self) -> TransportKey;
71
72 /// This transport's local listener endpoints, encoded for inclusion in the
73 /// local [`WorkerAddress`].
74 ///
75 /// Mirrors [`crate::Transport::address`]. Returned at builder time so the
76 /// Velo builder can merge it into the local PeerInfo's WorkerAddress.
77 /// Implementations that do not open their own listener (e.g., a transport
78 /// that piggybacks on the messenger) should return
79 /// [`WorkerAddress::default`].
80 fn address(&self) -> WorkerAddress;
81
82 /// Notify this transport that a peer's [`PeerInfo`] is now known.
83 ///
84 /// Mirrors [`crate::Transport::register`]. The transport extracts its own
85 /// entry from `peer_info.worker_address()` (using [`Self::key`]), decodes
86 /// the endpoint(s), and caches a resolved socket address keyed by the
87 /// peer's [`WorkerId`] for later use by [`Self::connect`].
88 ///
89 /// Default implementation is a no-op for transports that do not require
90 /// per-peer state (e.g., a transport that piggybacks on the messenger
91 /// which already tracks peers).
92 fn register(&self, _peer_info: &PeerInfo) -> Result<()> {
93 Ok(())
94 }
95
96 /// Bind a receive endpoint for the given anchor.
97 ///
98 /// - `anchor_id`: identifies which anchor this binding is for.
99 /// - `session_id`: unique session identifier for this attachment; used by
100 /// the transport to discriminate between successive sessions on the same
101 /// anchor so that stale frames from a prior session are not delivered.
102 ///
103 /// Returns the receiver half of the per-session frame channel. Endpoint
104 /// resolution is no longer string-based — the connecting peer resolves the
105 /// listener address from the bound worker's [`WorkerAddress`] entry for
106 /// this transport's [`Self::key`].
107 ///
108 /// The channel established by `bind` / `connect` MUST provide ordered,
109 /// loss-free delivery of all frames including sentinels.
110 fn bind(
111 &self,
112 anchor_id: u64,
113 session_id: u64,
114 ) -> BoxFuture<'_, Result<flume::Receiver<Vec<u8>>>>;
115
116 /// Connect a write endpoint to the given peer's bound anchor.
117 ///
118 /// - `peer`: the [`WorkerId`] of the worker that called [`Self::bind`].
119 /// The transport looks up the peer's cached endpoint (populated via
120 /// [`Self::register`]) to determine the actual socket address.
121 /// - `anchor_id`: identifies which anchor this writer is attached to.
122 /// - `session_id`: unique session identifier for this attachment; used by
123 /// the transport to route frames to the correct reader.
124 ///
125 /// Returns a [`flume::Sender<Vec<u8>>`] for sending frames to the bound
126 /// receiver.
127 fn connect(
128 &self,
129 peer: WorkerId,
130 anchor_id: u64,
131 session_id: u64,
132 ) -> BoxFuture<'_, Result<flume::Sender<Vec<u8>>>>;
133}
134
135#[cfg(test)]
136mod tests {
137 use super::*;
138 use crate::id::InstanceId;
139 use futures::FutureExt;
140
141 /// Compile-time proof that [`FrameTransport`] is object-safe.
142 fn _assert_object_safe(_transport: &dyn FrameTransport) {}
143
144 /// A minimal `FrameTransport` impl that exercises the trait's default
145 /// `register()` no-op. The other methods are stubs; the point is purely
146 /// to give the default body coverage.
147 struct DefaultRegisterTransport;
148
149 impl FrameTransport for DefaultRegisterTransport {
150 fn key(&self) -> TransportKey {
151 TransportKey::new("default-register-test")
152 }
153 fn address(&self) -> WorkerAddress {
154 WorkerAddress::empty()
155 }
156 fn bind(
157 &self,
158 _anchor_id: u64,
159 _session_id: u64,
160 ) -> BoxFuture<'_, Result<flume::Receiver<Vec<u8>>>> {
161 async { Err(anyhow::anyhow!("bind stub")) }.boxed()
162 }
163 fn connect(
164 &self,
165 _peer: WorkerId,
166 _anchor_id: u64,
167 _session_id: u64,
168 ) -> BoxFuture<'_, Result<flume::Sender<Vec<u8>>>> {
169 async { Err(anyhow::anyhow!("connect stub")) }.boxed()
170 }
171 }
172
173 #[test]
174 fn default_register_returns_ok() {
175 // The trait-default `register` must accept any PeerInfo and return
176 // Ok. Used by transports that piggyback on a higher layer (e.g.,
177 // the messenger mux) and need no per-peer cache.
178 let transport = DefaultRegisterTransport;
179 let peer = PeerInfo::new(InstanceId::new_v4(), WorkerAddress::empty());
180 assert!(transport.register(&peer).is_ok());
181 }
182}