feat(stream): chunked bulk payloads larger than one frame

ROADMAP.md

@@ -81,14 +81,21 @@ meant to be reachable by some peers and not others.
 ## M5 — bulk payloads
 
 Bodies are JSON, and a transcript or a rendered PDF should not be base64 in a
-JSON string. Two options, and the choice is not obvious yet:
-
-- [ ] File-descriptor passing over `SCM_RIGHTS`, which is zero-copy and Unix-only
-- [ ] A side-channel content store the bus hands out handles to, which works
-      everywhere and costs a write
-
-Until this lands, the convention is that large payloads travel as paths and the
-sender is responsible for the file's lifetime.
+JSON string — but nor should a 20 MB payload be undeliverable.
+
+- [x] Chunked peer-to-peer streams (`src/stream/`), flow-controlled by the
+      receiver's window and authorised by the broker-stamped `sender`. Works on
+      every transport and needs nothing from the broker, at the cost of base64
+      and a round trip per chunk.
+- [ ] File-descriptor passing over `SCM_RIGHTS`, which is zero-copy and
+      Unix-only. A fast path *under* the stream API rather than a replacement
+      for it: callers hold a `StreamRef`, so the transport underneath can change
+      without the interface changing.
+- [ ] A side-channel content store the bus hands out handles to, for payloads
+      big enough that a copy through the bus is the wrong shape entirely
+
+Passing a path remains the cheapest option when both peers can see the same
+filesystem and the sender can own the file's lifetime.
 
 ## M6 — other platforms and other languages
 

crates/tinybus/src/connection.rs

@@ -42,6 +42,10 @@ use crate::ports::Transport;
 use crate::proxy::Proxy;
 use crate::router::MatchRule;
 use crate::service::{Interface, ObjectTree};
+use crate::stream::{
+    STREAM_INTERFACE, STREAM_PATH, StreamDescriptor, StreamLimits, StreamReader, StreamRef,
+    StreamRegistry, StreamWriter,
+};
 use crate::version::{Compatibility, PeerManifest, PeerRecord};
 
 /// How long a call waits before giving up.
@@ -81,6 +85,10 @@ struct Inner {
     unique_name: std::sync::RwLock<Option<BusName>>,
     signals: broadcast::Sender<Message>,
     panic_handler: std::sync::RwLock<Option<Arc<dyn Fn() -> Error + Send + Sync>>>,
+    /// Bulk streams being received. On the connection rather than in the object
+    /// tree because a chunk has to be checked against the header's stamped
+    /// `sender`, and [`Interface`] deliberately never sees a header.
+    streams: StreamRegistry,
 }
 
 /// Closes the transport when the last [`Connection`] handle goes away.
@@ -173,6 +181,7 @@ impl Connection {
             unique_name: std::sync::RwLock::new(None),
             signals,
             panic_handler: std::sync::RwLock::new(None),
+            streams: StreamRegistry::new(),
         });
         tokio::spawn(writer_loop(inner.transport.clone(), outbound));
         tokio::spawn(dispatch_loop(inner.clone()));
@@ -629,6 +638,179 @@ impl Connection {
             .expect("panic handler lock") = Some(handler);
     }
 
+    /// What this connection will accept from peers sending it bulk streams.
+    pub fn stream_limits(&self) -> StreamLimits {
+        self.inner.streams.limits()
+    }
+
+    /// Change what this connection accepts from peers sending it bulk streams.
+    ///
+    /// Takes effect on the next `Open`; streams already running keep the window
+    /// they were opened with, because shrinking a window under a sender that is
+    /// mid-transfer would abort a transfer that was within the rules when it
+    /// started.
+    pub fn set_stream_limits(&self, limits: StreamLimits) {
+        self.inner.streams.set_limits(limits);
+    }
+
+    /// Open a bulk stream to `destination` and get the writer for it.
+    ///
+    /// The usual shape is: open, put [`StreamWriter::stream_ref`] in a method
+    /// call, issue the call, and write the payload *while the call is
+    /// outstanding*. The receiver's window is a few megabytes, so writing a
+    /// large payload before the receiving method has been dispatched stalls
+    /// against a reader that does not exist yet.
+    /// [`Connection::call_with_stream`] does the interleaving for the common
+    /// case.
+    pub async fn open_stream(
+        &self,
+        destination: &BusName,
+        descriptor: StreamDescriptor,
+    ) -> Result<StreamWriter> {
+        self.open_stream_with_timeout(destination, descriptor, DEFAULT_TIMEOUT)
+            .await
+    }
+
+    /// [`Connection::open_stream`] with an explicit deadline for every chunk.
+    ///
+    /// The deadline applies per chunk, not to the transfer: it is how long this
+    /// peer will wait for the receiver to take *one* chunk. A slow consumer of
+    /// a large payload is normal; a consumer that has stopped consuming is not.
+    pub async fn open_stream_with_timeout(
+        &self,
+        destination: &BusName,
+        descriptor: StreamDescriptor,
+        timeout: Duration,
+    ) -> Result<StreamWriter> {
+        let id: String = serde_json::from_value(
+            self.call_stream_member(
+                destination,
+                "Open",
+                serde_json::json!([descriptor]),
+                timeout,
+            )
+            .await?,
+        )?;
+        Ok(StreamWriter::new(
+            self.clone(),
+            destination.clone(),
+            id,
+            descriptor,
+            timeout,
+        ))
+    }
+
+    /// Call a method whose payload is too big for a frame, streaming `bytes`
+    /// alongside it.
+    ///
+    /// `args` is built from the [`StreamRef`] the receiver should read, so the
+    /// caller decides where in its own argument list the handle goes. The call
+    /// and the payload are in flight together, which is what keeps a sender
+    /// from stalling against its own receiver.
+    pub async fn call_with_stream<R: DeserializeOwned>(
+        &self,
+        destination: BusName,
+        path: ObjectPath,
+        interface: InterfaceName,
+        member: MemberName,
+        args: impl FnOnce(&StreamRef) -> Value,
+        bytes: &[u8],
+    ) -> Result<R> {
+        self.call_with_stream_timeout(
+            destination,
+            path,
+            interface,
+            member,
+            args,
+            bytes,
+            DEFAULT_TIMEOUT,
+        )
+        .await
+    }
+
+    /// [`Connection::call_with_stream`] with an explicit deadline.
+    ///
+    /// `timeout` bounds two different waits: how long the callee has to answer,
+    /// and how long the receiver has to take any one chunk. Both are "the peer
+    /// has stopped making progress" deadlines rather than a budget for the
+    /// whole transfer, which is why one value fits both — but the call half is
+    /// the one worth thinking about, because the callee cannot reply until it
+    /// has read the payload. A large upload to a slow-but-healthy consumer
+    /// needs more than [`DEFAULT_TIMEOUT`] here, or it fails a call that was
+    /// still making progress.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn call_with_stream_timeout<R: DeserializeOwned>(
+        &self,
+        destination: BusName,
+        path: ObjectPath,
+        interface: InterfaceName,
+        member: MemberName,
+        args: impl FnOnce(&StreamRef) -> Value,
+        bytes: &[u8],
+        timeout: Duration,
+    ) -> Result<R> {
+        let mut writer = self
+            .open_stream_with_timeout(
+                &destination,
+                StreamDescriptor::with_len(bytes.len() as u64),
+                timeout,
+            )
+            .await?;
+        let message = Message::method_call(
+            destination,
+            path,
+            interface,
+            member,
+            to_body(&args(&writer.stream_ref()))?,
+        );
+
+        // Both halves at once, and the first failure wins: the callee is
+        // reading the stream while it answers, so waiting for either one before
+        // starting the other is a deadlock, not a slow path.
+        let (reply, ()) = tokio::try_join!(self.call_raw(message, timeout), async {
+            writer.write(bytes).await?;
+            writer.finish().await.map(|_| ())
+        })?;
+        Ok(serde_json::from_value(reply)?)
+    }
+
+    /// Take the reader for a stream a peer opened on this connection.
+    ///
+    /// Once only: a stream has one consumer, because two consumers would each
+    /// get an arbitrary half of the payload.
+    pub fn accept_stream(&self, stream: &StreamRef) -> Result<StreamReader> {
+        self.inner.streams.take_reader(&stream.id)
+    }
+
+    /// Read a whole stream into memory, refusing to exceed
+    /// [`StreamLimits::max_stream_len`].
+    ///
+    /// For a payload that is too big for a frame but not too big for memory.
+    /// Anything else wants [`Connection::accept_stream`] and a loop over
+    /// [`StreamReader::next_chunk`], which never holds more than one chunk.
+    pub async fn read_stream(&self, stream: &StreamRef) -> Result<Vec<u8>> {
+        let limit = self.stream_limits().max_stream_len;
+        self.accept_stream(stream)?.read_to_end_capped(limit).await
+    }
+
+    /// Call one member of a peer's built-in stream interface.
+    pub(crate) async fn call_stream_member(
+        &self,
+        destination: &BusName,
+        member: &str,
+        args: Value,
+        timeout: Duration,
+    ) -> Result<Value> {
+        let message = Message::method_call(
+            destination.clone(),
+            ObjectPath::new(STREAM_PATH)?,
+            InterfaceName::new(STREAM_INTERFACE)?,
+            MemberName::new(member)?,
+            args,
+        );
+        self.call_raw(message, timeout).await
+    }
+
     /// Call a method on the broker's own interface.
     async fn call_bus(&self, member: &str, args: Value) -> Result<Value> {
         let message = Message::method_call(
@@ -797,6 +979,15 @@ impl<F: Future> Future for CatchUnwind<F> {
 }
 
 async fn dispatch(inner: &Inner, header: &Header, body: Value) -> Result<Value> {
+    // Streams are answered before the object tree is consulted, and without the
+    // service having exported anything: bulk transfer is bus plumbing, and a
+    // service that forgot to export it would be a service you cannot send a
+    // file to. It also means a peer cannot shadow the stream interface by
+    // exporting its own at that address.
+    if StreamRegistry::handles(header) {
+        return inner.streams.dispatch(header, body).await;
+    }
+
     let (Some(path), Some(interface), Some(member)) =
         (&header.path, &header.interface, &header.member)
     else {

crates/tinybus/src/error.rs

@@ -201,6 +201,42 @@ pub enum Error {
         detail: String,
     },
 
+    /// No such bulk stream, or not one this peer opened.
+    ///
+    /// The two cases are deliberately one error: distinguishing them would let
+    /// a peer probe for streams running between two others.
+    #[error("no stream `{id}`")]
+    UnknownStream {
+        /// The handle that was presented. Minted by this peer, so quoting it
+        /// leaks nothing.
+        id: String,
+    },
+
+    /// A bulk stream ended before it was complete.
+    #[error("stream aborted: {reason}")]
+    StreamAborted {
+        /// Why it ended. Always crate-generated — never a peer's string, which
+        /// would be a peer writing into this process's logs.
+        reason: String,
+    },
+
+    /// A bulk stream would exceed what the receiver accepts.
+    #[error("stream exceeds the {limit}-byte limit")]
+    StreamTooLarge {
+        /// The receiver's cap, in bytes.
+        limit: u64,
+    },
+
+    /// This peer already has as many streams open as the receiver allows.
+    ///
+    /// Per peer, so a peer that opens streams and never finishes them runs out
+    /// of its own slots rather than everyone's.
+    #[error("already at the limit of {limit} open streams")]
+    TooManyStreams {
+        /// The receiver's per-peer cap.
+        limit: usize,
+    },
+
     /// Filesystem or socket I/O failed.
     #[error("io: {0}")]
     Io(#[from] std::io::Error),
@@ -322,6 +358,10 @@ impl Error {
             Self::ModuleRefused { .. } => "ai.tinyhumans.tinybus.Error.ModuleRefused",
             Self::ModuleUnavailable { .. } => "ai.tinyhumans.tinybus.Error.ModuleUnavailable",
             Self::Path { .. } => "ai.tinyhumans.tinybus.Error.Path",
+            Self::UnknownStream { .. } => "ai.tinyhumans.tinybus.Error.UnknownStream",
+            Self::StreamAborted { .. } => "ai.tinyhumans.tinybus.Error.StreamAborted",
+            Self::StreamTooLarge { .. } => "ai.tinyhumans.tinybus.Error.StreamTooLarge",
+            Self::TooManyStreams { .. } => "ai.tinyhumans.tinybus.Error.TooManyStreams",
             Self::FeatureDisabled(_, _) => "ai.tinyhumans.tinybus.Error.FeatureDisabled",
             Self::Json(_) => "ai.tinyhumans.tinybus.Error.Json",
             Self::MethodFailed { name, .. } => name,
@@ -410,16 +450,30 @@ mod tests {
         let text = err.to_string();
         assert!(text.contains("expected u64"), "{text}");
         assert!(!text.contains("0xdeadbeef"), "{text}");
+        // The double-quoted half is the one serde uses for a rejected *string*,
+        // which is the shape a token or a recovery phrase arrives in.
+        assert!(!text.contains("seed phrase here"), "{text}");
     }
 
     #[test]
     fn redaction_survives_an_unclosed_quote() {
         // A truncated message must not leak the tail just because its closing
         // backtick never arrived.
         assert_eq!(redact_values("bad token `abc"), "bad token `…");
+        assert_eq!(redact_values("bad token \"abc"), "bad token \"…");
         assert_eq!(redact_values("no quotes here"), "no quotes here");
     }
 
+    #[test]
+    fn a_backtick_inside_a_quoted_value_does_not_end_the_redaction_early() {
+        // Otherwise a value chosen to contain a backtick would close the span
+        // and put its own tail back into the message.
+        assert_eq!(
+            redact_values("invalid: \"a`b`c\", expected u64"),
+            "invalid: \"…\", expected u64"
+        );
+    }
+
     #[test]
     fn a_generic_failure_falls_back_to_the_failed_name() {
         assert_eq!(Error::failed("boom").wire_name(), Error::FAILED);
@@ -505,6 +559,12 @@ mod tests {
             },
             Error::path("path", "bad"),
             Error::FeatureDisabled("thing", "uds"),
+            Error::UnknownStream { id: "s1".into() },
+            Error::StreamAborted {
+                reason: "aborted".into(),
+            },
+            Error::StreamTooLarge { limit: 1 },
+            Error::TooManyStreams { limit: 1 },
             Error::Json(serde_json::from_str::<serde_json::Value>("{").unwrap_err()),
         ];
         for error in errors {

crates/tinybus/src/lib.rs

@@ -79,6 +79,7 @@ pub mod ports;
 pub mod proxy;
 pub mod router;
 pub mod service;
+pub mod stream;
 pub mod transport;
 pub mod version;
 
@@ -99,6 +100,9 @@ pub use crate::ports::{Listener, Transport};
 pub use crate::proxy::Proxy;
 pub use crate::router::MatchRule;
 pub use crate::service::Interface;
+pub use crate::stream::{
+    MAX_CHUNK_LEN, StreamDescriptor, StreamLimits, StreamReader, StreamRef, StreamWriter,
+};
 pub use crate::version::{
     Compatibility, InterfaceVersion, PeerManifest, PeerRecord, Version, VersionRange,
 };

crates/tinybus/src/message/codec.rs

@@ -19,7 +19,8 @@ use crate::error::{Error, Result};
 /// A hard cap, not a tunable. The frame length arrives from the wire *before*
 /// the bytes do, so without this the first four bytes of a hostile or corrupt
 /// stream are a 4 GiB allocation. 16 MiB is far above any legitimate control
-/// message; bulk payloads are supposed to travel as paths, not as bodies.
+/// message; a payload that does not fit goes through [`crate::stream`], which
+/// splits it into chunks that do, rather than through a larger cap here.
 pub const MAX_FRAME_LEN: usize = 16 * 1024 * 1024;
 
 /// The length prefix's width, in bytes.

crates/tinybus/src/message/mod.rs

@@ -8,9 +8,10 @@
 //! What we get in exchange is that `tinybus monitor` is readable, a service can
 //! be written in any language in an afternoon, and `serde` derives on the
 //! kernel side are the entire client binding. The one place this would be the
-//! wrong trade is bulk binary payloads (audio, PDFs), and that is exactly what
-//! `ROADMAP.md`'s file-descriptor passing milestone is for; until then large
-//! payloads travel as paths, not as base64.
+//! wrong trade is bulk binary payloads (audio, PDFs), which is why those do not
+//! travel in a body at all: [`crate::stream`] carries them beside the call as
+//! chunks, and the body carries only a handle. A path is still cheaper when
+//! both peers can see the same filesystem.
 //!
 //! # Why the header is flat
 //!