diff --git a/ROADMAP.md b/ROADMAP.md index bcc534f..2af0e1d 100644 --- a/ROADMAP.md +++ b/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 diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 74c363b..a36b8ea 100644 --- a/crates/tinybus/src/connection.rs +++ b/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>, signals: broadcast::Sender, panic_handler: std::sync::RwLock 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 { + 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 { + 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( + &self, + destination: BusName, + path: ObjectPath, + interface: InterfaceName, + member: MemberName, + args: impl FnOnce(&StreamRef) -> Value, + bytes: &[u8], + ) -> Result { + 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( + &self, + destination: BusName, + path: ObjectPath, + interface: InterfaceName, + member: MemberName, + args: impl FnOnce(&StreamRef) -> Value, + bytes: &[u8], + timeout: Duration, + ) -> Result { + 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 { + 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> { + 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 { + 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 { let message = Message::method_call( @@ -797,6 +979,15 @@ impl Future for CatchUnwind { } async fn dispatch(inner: &Inner, header: &Header, body: Value) -> Result { + // 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 { diff --git a/crates/tinybus/src/error.rs b/crates/tinybus/src/error.rs index a657480..b2c3a0a 100644 --- a/crates/tinybus/src/error.rs +++ b/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,6 +450,9 @@ 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] @@ -417,9 +460,20 @@ mod tests { // 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::("{").unwrap_err()), ]; for error in errors { diff --git a/crates/tinybus/src/lib.rs b/crates/tinybus/src/lib.rs index 38f7956..7d92da3 100644 --- a/crates/tinybus/src/lib.rs +++ b/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, }; diff --git a/crates/tinybus/src/message/codec.rs b/crates/tinybus/src/message/codec.rs index b21f310..24dc1aa 100644 --- a/crates/tinybus/src/message/codec.rs +++ b/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. diff --git a/crates/tinybus/src/message/mod.rs b/crates/tinybus/src/message/mod.rs index dc901a3..0d98bfa 100644 --- a/crates/tinybus/src/message/mod.rs +++ b/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 //! diff --git a/crates/tinybus/src/stream/base64.rs b/crates/tinybus/src/stream/base64.rs new file mode 100644 index 0000000..633ed6c --- /dev/null +++ b/crates/tinybus/src/stream/base64.rs @@ -0,0 +1,159 @@ +//! Standard base64 (RFC 4648, padded), hand-rolled. +//! +//! Hand-rolled rather than depended upon because this crate's whole argument is +//! that the kernel's dependency graph is a liability, and a chunk codec is +//! forty lines. The decoder is strict — it rejects any byte outside the +//! alphabet, wrong padding, and a trailing group that carries bits which +//! encode nothing — because a lenient decoder makes two peers disagree about +//! what a chunk contained, and the length accounting on either side of a +//! stream has to match exactly. + +use crate::error::{Error, Result}; + +const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +/// Encode `bytes` as padded standard base64. +pub fn encode(bytes: &[u8]) -> String { + let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4); + for group in bytes.chunks(3) { + let b0 = group[0] as u32; + let b1 = *group.get(1).unwrap_or(&0) as u32; + let b2 = *group.get(2).unwrap_or(&0) as u32; + let packed = (b0 << 16) | (b1 << 8) | b2; + out.push(ALPHABET[(packed >> 18) as usize & 63] as char); + out.push(ALPHABET[(packed >> 12) as usize & 63] as char); + out.push(if group.len() > 1 { + ALPHABET[(packed >> 6) as usize & 63] as char + } else { + '=' + }); + out.push(if group.len() > 2 { + ALPHABET[packed as usize & 63] as char + } else { + '=' + }); + } + out +} + +/// Decode padded standard base64. +/// +/// The error never quotes the offending input: a chunk is user data by +/// definition, and this message travels back to the peer as an error reply. +pub fn decode(text: &str) -> Result> { + let bytes = text.as_bytes(); + if !bytes.len().is_multiple_of(4) { + return Err(Error::protocol( + "base64 chunk length is not a multiple of four", + )); + } + let mut out = Vec::with_capacity(bytes.len() / 4 * 3); + for (index, group) in bytes.chunks(4).enumerate() { + let last = index == bytes.len() / 4 - 1; + let mut packed = 0u32; + let mut kept = 3; + for (position, &byte) in group.iter().enumerate() { + let value = match byte { + b'=' => { + // Padding is only ever the last one or two symbols of the + // final group; anywhere else it is a corrupt chunk, not a + // shorter one. + if !last || position < 2 { + return Err(Error::protocol("base64 chunk has misplaced padding")); + } + // Only the *first* pad fixes the length; the second is more + // of the same padding, not a shorter group again. + kept = kept.min(position - 1); + 0 + } + _ => decode_symbol(byte)?, + }; + packed = (packed << 6) | value as u32; + } + // A padded group must not carry bits below the bytes it encodes, or two + // distinct texts would decode to one chunk. + let slack = (3 - kept) * 8; + if slack > 0 && packed & ((1 << slack) - 1) != 0 { + return Err(Error::protocol("base64 chunk has non-canonical padding")); + } + for shift in (0..kept).map(|i| 16 - i * 8) { + out.push((packed >> shift) as u8); + } + } + Ok(out) +} + +fn decode_symbol(byte: u8) -> Result { + match byte { + b'A'..=b'Z' => Ok(byte - b'A'), + b'a'..=b'z' => Ok(byte - b'a' + 26), + b'0'..=b'9' => Ok(byte - b'0' + 52), + b'+' => Ok(62), + b'/' => Ok(63), + _ => Err(Error::protocol( + "base64 chunk has a symbol outside the alphabet", + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_length_modulo_three_round_trips() { + for len in 0..=64usize { + let bytes: Vec = (0..len).map(|i| (i * 7 + 3) as u8).collect(); + let text = encode(&bytes); + assert_eq!(decode(&text).unwrap(), bytes, "len {len}"); + } + } + + #[test] + fn the_encoding_matches_the_rfc_test_vectors() { + assert_eq!(encode(b""), ""); + assert_eq!(encode(b"f"), "Zg=="); + assert_eq!(encode(b"fo"), "Zm8="); + assert_eq!(encode(b"foo"), "Zm9v"); + assert_eq!(encode(b"foob"), "Zm9vYg=="); + assert_eq!(encode(b"fooba"), "Zm9vYmE="); + assert_eq!(encode(b"foobar"), "Zm9vYmFy"); + } + + #[test] + fn the_full_byte_range_survives_a_round_trip() { + let bytes: Vec = (0..=255u8).collect(); + assert_eq!(decode(&encode(&bytes)).unwrap(), bytes); + } + + #[test] + fn a_truncated_group_is_rejected_rather_than_padded_silently() { + assert!(decode("Zm9").is_err()); + } + + #[test] + fn a_symbol_outside_the_alphabet_is_rejected() { + assert!(decode("Zm9*").is_err()); + assert!(decode("Zm9 ").is_err()); + } + + #[test] + fn padding_in_the_middle_of_a_chunk_is_rejected() { + assert!(decode("Zg==Zg==").is_err()); + assert!(decode("=g==").is_err()); + } + + #[test] + fn a_padded_group_carrying_bits_it_does_not_encode_is_rejected() { + // "Zh==" decodes the same byte as "Zg==" under a lenient decoder. + assert!(decode("Zh==").is_err()); + assert!(decode("Zm9=").is_err()); + } + + #[test] + fn a_decode_failure_never_quotes_the_chunk_it_rejected() { + let secret = "recovery-phrase!"; + let error = decode(secret).unwrap_err().to_string(); + assert!(!error.contains(secret), "{error}"); + } +} diff --git a/crates/tinybus/src/stream/mod.rs b/crates/tinybus/src/stream/mod.rs new file mode 100644 index 0000000..670b212 --- /dev/null +++ b/crates/tinybus/src/stream/mod.rs @@ -0,0 +1,793 @@ +//! Bulk payloads: chunked, flow-controlled byte streams between two peers. +//! +//! A frame is capped at [`MAX_FRAME_LEN`](crate::message::codec::MAX_FRAME_LEN) +//! and that cap is not negotiable — it is what stops a peer announcing a +//! gigabyte and making the reader allocate it. A stream is how a payload larger +//! than one frame crosses the bus anyway: the sender opens a stream on the +//! receiver, writes it as a sequence of bounded chunks, and closes it. What +//! travels in the method call is a [`StreamRef`] — a handle a few dozen bytes +//! long — and the bytes travel beside it. +//! +//! # Why this is a peer-to-peer interface and not a broker feature +//! +//! Every chunk is an ordinary method call addressed to the receiving peer. The +//! broker reads the header, routes it, and forwards it, exactly as it does for +//! everything else; it never sees a stream as anything other than traffic. A +//! broker that assembled streams would be a broker that buffers every payload +//! on the bus — which is both the memory problem and the "the broker has seen +//! every credential" problem, at once. +//! +//! # Flow control +//! +//! `Write` is a call, so it has a reply and a deadline. The receiver does not +//! reply until the chunk has room in the reader's window +//! ([`StreamLimits::window_chunks`]), so a sender runs exactly as fast as the +//! receiver drains and no faster. There is no unbounded buffer anywhere: a +//! receiver that never reads stalls the sender, the stall shows up as no +//! activity on the stream, and the idle reaper aborts it. That is the +//! misbehaving-peer invariant applied to bulk transfer — one peer's refusal to +//! read costs it its own stream and nobody else's memory. +//! +//! # Ordering +//! +//! Chunks carry a sequence number and the receiver requires the next one +//! exactly. The transport is already ordered, so this catches a pipelining +//! sender rather than a reordering network: two chunks in flight at once would +//! be dispatched into two tasks on the receiver and could land either way +//! round, and silently transposing two megabytes of a PDF is worse than an +//! error. +//! +//! # Base64 +//! +//! Bodies are JSON, so a chunk is base64 and costs a third of its size in +//! overhead. That is the price of the payload travelling on the same wire as +//! everything else, and it is why `ROADMAP.md` still wants `SCM_RIGHTS`: an +//! fd-passing fast path can slot in under this same API later, because callers +//! hold a [`StreamRef`], not a byte array. + +pub mod base64; + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tokio::sync::{Mutex, mpsc}; + +use crate::error::{Error, Result}; +use crate::message::Header; +use crate::name::{BusName, MemberName}; + +/// The interface a peer serves so others can push bulk payloads at it. +/// +/// Served by every [`Connection`](crate::Connection) automatically, before the +/// object tree is consulted: a stream is bus plumbing, not something each +/// service should have to remember to export. +pub const STREAM_INTERFACE: &str = "ai.tinyhumans.tinybus.Stream"; + +/// The object path [`STREAM_INTERFACE`] lives at. +pub const STREAM_PATH: &str = "/ai/tinyhumans/tinybus/Stream"; + +/// The largest chunk a sender may put in one `Write`, before base64. +/// +/// Half a megabyte encodes to about 700 KB, which leaves the 16 MB frame cap +/// two orders of magnitude of headroom for the header and for any future field. +/// Small enough that a chunk is a cheap unit of retry and of flow control; +/// large enough that a 100 MB payload is two hundred round trips, not two +/// hundred thousand. +pub const MAX_CHUNK_LEN: usize = 512 * 1024; + +/// What a receiver will accept, and how much of itself it will spend doing it. +/// +/// A receiver's limits are its own: nothing here is negotiated with the sender, +/// because a limit a peer can talk you out of is not a limit. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StreamLimits { + /// The largest single stream, in bytes. A sender that exceeds it has its + /// stream aborted rather than being allowed to keep going. + pub max_stream_len: u64, + /// How many streams one peer may have open at once. Per peer, not global, + /// so a busy peer cannot starve every other peer of slots. + pub max_streams_per_peer: usize, + /// How many chunks may sit between the wire and the reader. This is the + /// flow-control window: the sender is never more than this far ahead. + pub window_chunks: usize, + /// How long a stream may see no writes before it is reaped. Bounds what an + /// abandoned stream — a sender that exited mid-transfer, or one stalled + /// against a reader that never reads — can hold open. + pub idle_timeout: Duration, +} + +impl Default for StreamLimits { + fn default() -> Self { + Self { + // 256 MB is a video file or a disk image, not a transcript. Past + // that a caller wants a path or a content store, not the bus. + max_stream_len: 256 * 1024 * 1024, + max_streams_per_peer: 4, + // Eight chunks is 4 MB in flight: enough that a round trip per + // chunk does not dominate throughput, bounded enough that + // `max_streams_per_peer` × this is a number you can hold in mind. + window_chunks: 8, + idle_timeout: Duration::from_secs(60), + } + } +} + +/// A handle to a stream open on the receiving peer. +/// +/// This is what travels in a method body in place of the payload. It is only +/// meaningful to the peer that minted it, and only usable by the peer that +/// opened it — the receiver checks the broker-stamped `sender` on every chunk. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct StreamRef { + /// Opaque, minted by the receiver. Never parse it. + pub id: String, + /// What the payload is, if the sender said. Advisory: a receiver that cares + /// must still validate the bytes. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_type: Option, + /// The total length, once the sender has declared or finished it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub len: Option, +} + +/// What a sender says about a payload before sending it. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StreamDescriptor { + /// A media type, if the sender knows one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_type: Option, + /// The total length, when it is known up front. + /// + /// Declaring it lets the receiver reject an oversized transfer at `Open` + /// instead of after it has already accepted 256 MB of it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_len: Option, +} + +impl StreamDescriptor { + /// A descriptor for a payload of known length and unknown type. + pub fn with_len(total_len: u64) -> Self { + Self { + content_type: None, + total_len: Some(total_len), + } + } + + /// Set the content type. + pub fn content_type(mut self, content_type: impl Into) -> Self { + self.content_type = Some(content_type.into()); + self + } +} + +/// How a stream ended, from the receiver's side. +#[derive(Debug, Clone)] +enum Outcome { + /// The sender called `Close` and the length it declared checked out. + Complete, + /// The stream will produce no more bytes, and did not finish. + /// + /// The reason is always crate-generated. A peer-supplied string would be a + /// peer writing into the receiver's logs. + Aborted(&'static str), +} + +/// One stream being received. +struct Inbound { + /// The peer that opened it, as stamped by the broker. `None` only on a + /// direct connection with no broker in the middle, where there is no sender + /// to distinguish peers in the first place. + owner: Option, + content_type: Option, + declared_len: Option, + /// Serialises the sequence check and the handoff to the reader. Without it + /// two pipelined chunks could pass the check in order and reach the reader + /// out of order, since each call is dispatched on its own task. + /// + /// It guards nothing but ordering: the counters beside it are atomics + /// precisely so that `Close` and `Abort` can read them *without* taking + /// this lock. A chunk write parks here while the window is full, and a peer + /// that could make `Close` wait on that would have found a way to wedge the + /// receiver from outside. + gate: Mutex<()>, + next_seq: AtomicU64, + received: AtomicU64, + /// Dropped to signal end-of-stream; the reader then consults `outcome` to + /// learn whether that end was a `Close` or an abort. + chunks: std::sync::Mutex>>>, + /// Taken once, by whoever reads the stream. + reader: std::sync::Mutex>>>, + /// Shared with the reader by `Arc` rather than reached through this + /// struct, because a reader must not keep the `Inbound` — and therefore the + /// channel's sending half — alive. If it did, a receiving connection that + /// died mid-stream would leave a reader parked on a channel that can never + /// close, which is precisely the hang this project exists to not have. + outcome: Arc>>, + last_activity: std::sync::Mutex, +} + +impl Inbound { + /// Record how the stream ended, keeping the first verdict. + /// + /// First rather than last because the first is the cause and anything after + /// it is a consequence — a reaped stream whose sender then aborts should + /// still read as reaped. + fn finish(&self, outcome: Outcome) { + let mut slot = self.outcome.lock().expect("stream outcome lock"); + if slot.is_none() { + *slot = Some(outcome); + } + } + + /// Close the writing half, and with it the reader's channel. + fn seal(&self) { + *self.chunks.lock().expect("stream chunks lock") = None; + } + + fn writer(&self) -> Option>> { + self.chunks.lock().expect("stream chunks lock").clone() + } + + fn touch(&self) { + *self.last_activity.lock().expect("stream activity lock") = Instant::now(); + } + + fn idle_for(&self) -> Duration { + self.last_activity + .lock() + .expect("stream activity lock") + .elapsed() + } +} + +/// Every stream one connection is receiving. +/// +/// Lives on the connection rather than on the object tree because handling a +/// chunk needs the message header — specifically the stamped `sender` — and +/// [`Interface`](crate::Interface) deliberately does not get one. +pub(crate) struct StreamRegistry { + limits: std::sync::RwLock, + inbound: std::sync::Mutex>>, + next_id: AtomicU64, +} + +impl StreamRegistry { + pub(crate) fn new() -> Self { + Self { + limits: std::sync::RwLock::new(StreamLimits::default()), + inbound: std::sync::Mutex::new(HashMap::new()), + next_id: AtomicU64::new(1), + } + } + + pub(crate) fn limits(&self) -> StreamLimits { + *self.limits.read().expect("stream limits lock") + } + + pub(crate) fn set_limits(&self, limits: StreamLimits) { + *self.limits.write().expect("stream limits lock") = limits; + } + + /// Whether a call addresses the built-in stream interface. + pub(crate) fn handles(header: &Header) -> bool { + header + .interface + .as_ref() + .is_some_and(|interface| interface.as_str() == STREAM_INTERFACE) + && header + .path + .as_ref() + .is_some_and(|path| path.as_str() == STREAM_PATH) + } + + /// Run one call against the stream interface. + pub(crate) async fn dispatch(&self, header: &Header, body: Value) -> Result { + let member = header + .member + .as_ref() + .ok_or_else(|| Error::protocol("stream call is missing a member"))?; + match member.as_str() { + "Open" => self.open(header, member, body), + "Write" => self.write(header, member, body).await, + "Close" => self.close(header, member, body), + "Abort" => self.abort(header, member, body), + _ => Err(Error::UnknownMethod { + interface: header + .interface + .clone() + .expect("dispatch only runs once the interface matched"), + member: member.clone(), + }), + } + } + + fn open(&self, header: &Header, member: &MemberName, body: Value) -> Result { + let (descriptor,): (StreamDescriptor,) = + serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?; + let limits = self.limits(); + + if descriptor + .total_len + .is_some_and(|total| total > limits.max_stream_len) + { + // Rejecting a declared oversize here rather than at the byte that + // crosses the line saves both peers the whole transfer. + return Err(Error::StreamTooLarge { + limit: limits.max_stream_len, + }); + } + + let (chunks, reader) = mpsc::channel(limits.window_chunks.max(1)); + let id = format!("s{}", self.next_id.fetch_add(1, Ordering::Relaxed)); + let inbound = Arc::new(Inbound { + owner: header.sender.clone(), + content_type: descriptor.content_type, + declared_len: descriptor.total_len, + gate: Mutex::new(()), + next_seq: AtomicU64::new(0), + received: AtomicU64::new(0), + chunks: std::sync::Mutex::new(Some(chunks)), + reader: std::sync::Mutex::new(Some(reader)), + outcome: Arc::new(std::sync::Mutex::new(None)), + last_activity: std::sync::Mutex::new(Instant::now()), + }); + + let mut streams = self.inbound.lock().expect("stream registry lock"); + // Reaping here, rather than on a timer, is enough: a stream only + // lingers once it stops being written to, and the only thing a lingering + // stream costs anyone is a slot in this check. + streams.retain(|_, stream| { + let live = stream.idle_for() < limits.idle_timeout; + if !live { + stream.finish(Outcome::Aborted("the stream went idle and was reaped")); + stream.seal(); + // Same reason `kill` does it: dropping the reading half is what + // wakes a chunk write parked against a full window. Without + // this, reaping an abandoned stream leaves its sender parked + // until its own deadline expires — the stream is gone but the + // peer is still waiting on it. + drop(stream.reader.lock().expect("stream reader lock").take()); + } + live + }); + + // A closed-but-unread stream still holds its window, so it is capped + // too — separately from live ones, because the two are different + // failures. Too many live streams is a sender running ahead of itself; + // too many closed ones is a receiver that is not collecting what it + // was sent. Evicting the oldest keeps the newest transfer — the one a + // call is most likely still waiting on — alive. + let mut live = 0usize; + let mut sealed: Vec<(String, Instant)> = Vec::new(); + for (key, stream) in streams.iter() { + if stream.owner != header.sender { + continue; + } + if stream.writer().is_some() { + live += 1; + } else { + sealed.push(( + key.clone(), + *stream.last_activity.lock().expect("stream activity lock"), + )); + } + } + if live >= limits.max_streams_per_peer { + return Err(Error::TooManyStreams { + limit: limits.max_streams_per_peer, + }); + } + if sealed.len() >= limits.max_streams_per_peer { + sealed.sort_by_key(|(_, at)| *at); + for (key, _) in sealed + .iter() + .take(sealed.len() + 1 - limits.max_streams_per_peer) + { + if let Some(stream) = streams.remove(key) { + stream.finish(Outcome::Aborted("the receiver never collected the stream")); + } + } + } + + streams.insert(id.clone(), inbound); + Ok(Value::String(id)) + } + + async fn write(&self, header: &Header, member: &MemberName, body: Value) -> Result { + let (id, seq, data): (String, u64, String) = + serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?; + let stream = self.lookup(&id, header)?; + let limits = self.limits(); + let chunk = base64::decode(&data)?; + if chunk.len() > MAX_CHUNK_LEN { + self.kill(&id, &stream, "the sender exceeded the chunk cap"); + return Err(Error::protocol(format!( + "chunk of {} bytes exceeds the {MAX_CHUNK_LEN}-byte cap", + chunk.len() + ))); + } + + let gate = stream.gate.lock().await; + let Some(chunks) = stream.writer() else { + return Err(Error::StreamAborted { + reason: "the stream is already closed".to_string(), + }); + }; + if seq != stream.next_seq.load(Ordering::Relaxed) { + drop(gate); + self.kill(&id, &stream, "the sender wrote chunks out of order"); + return Err(Error::protocol("stream chunk arrived out of order")); + } + let received = stream.received.load(Ordering::Relaxed) + chunk.len() as u64; + if received > limits.max_stream_len + || stream.declared_len.is_some_and(|total| received > total) + { + drop(gate); + self.kill(&id, &stream, "the sender exceeded the length it may write"); + return Err(Error::StreamTooLarge { + limit: limits.max_stream_len, + }); + } + stream.next_seq.fetch_add(1, Ordering::Relaxed); + stream.received.store(received, Ordering::Relaxed); + stream.touch(); + // The window is the whole flow-control story: this await is where a + // sender that has run ahead of the reader waits, and the reply it is + // waiting on carries the sender's own deadline. + let delivered = chunks.send(chunk).await; + drop(gate); + if delivered.is_err() { + // The reader was dropped. Tell the sender now rather than letting + // it push the rest of a payload nobody will ever look at. + self.kill(&id, &stream, "the receiver stopped reading"); + return Err(Error::StreamAborted { + reason: "the receiver stopped reading".to_string(), + }); + } + stream.touch(); + Ok(Value::Null) + } + + fn close(&self, header: &Header, member: &MemberName, body: Value) -> Result { + let (id, total_len): (String, u64) = + serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?; + let stream = self.lookup(&id, header)?; + + // The entry stays in the registry, sealed. A payload that fits inside + // the window can be written and closed before the receiving method has + // even been dispatched, and dropping the entry here would turn that — + // the *fast* case — into "no such stream". + let received = stream.received.load(Ordering::Relaxed); + stream.touch(); + stream.seal(); + + if received != total_len { + stream.finish(Outcome::Aborted("the sender closed a truncated stream")); + return Err(Error::protocol(format!( + "stream closed after {received} bytes, {total_len} declared" + ))); + } + stream.finish(Outcome::Complete); + Ok(Value::Null) + } + + fn abort(&self, header: &Header, member: &MemberName, body: Value) -> Result { + let (id,): (String,) = + serde_json::from_value(body).map_err(|e| Error::bad_arguments(member.clone(), e))?; + let stream = self.lookup(&id, header)?; + self.kill(&id, &stream, "the sender aborted the stream"); + Ok(Value::Null) + } + + /// Find a stream and check that the peer asking owns it. + /// + /// The ownership check is the whole authorisation story for streams, and it + /// rests on `sender` being stamped by the broker: without it, any peer that + /// guessed an id could interleave its own bytes into someone else's + /// transfer. + fn lookup(&self, id: &str, header: &Header) -> Result> { + let streams = self.inbound.lock().expect("stream registry lock"); + let stream = streams + .get(id) + .ok_or_else(|| Error::UnknownStream { id: id.to_string() })?; + if stream.owner != header.sender { + // Deliberately the same error as "no such stream": telling a peer + // that an id it does not own exists is telling it about traffic + // between two other peers. + return Err(Error::UnknownStream { id: id.to_string() }); + } + Ok(stream.clone()) + } + + /// End a stream from the receiver's side and drop it from the registry. + fn kill(&self, id: &str, stream: &Arc, reason: &'static str) { + self.inbound + .lock() + .expect("stream registry lock") + .remove(id); + stream.finish(Outcome::Aborted(reason)); + // Dropping the sending half is what wakes a reader parked on `recv`. + stream.seal(); + // And dropping the *reading* half, if nobody ever claimed it, is what + // wakes a chunk write parked against a full window: without this a + // stream killed while a sender is mid-`Write` leaves that write parked + // until the sender's own deadline expires. + drop(stream.reader.lock().expect("stream reader lock").take()); + } + + /// Hand the reading half of a stream to the caller. Once only. + /// + /// The entry leaves the registry: from here the reader owns the stream, and + /// a sender writing to it is talking to the reader's window rather than to + /// a table this connection has to keep swept. + pub(crate) fn take_reader(&self, id: &str) -> Result { + let stream = { + let mut streams = self.inbound.lock().expect("stream registry lock"); + let stream = streams + .get(id) + .cloned() + .ok_or_else(|| Error::UnknownStream { id: id.to_string() })?; + // A sealed stream has nothing left to route to it; a live one still + // needs its entry so `Write` can find it. + if stream.writer().is_none() { + streams.remove(id); + } + stream + }; + let chunks = stream + .reader + .lock() + .expect("stream reader lock") + .take() + .ok_or_else(|| Error::StreamAborted { + reason: "the stream is already being read".to_string(), + })?; + Ok(StreamReader { + content_type: stream.content_type.clone(), + declared_len: stream.declared_len, + outcome: stream.outcome.clone(), + chunks, + }) + } +} + +/// The sending half of a stream: chunks out, one at a time, at the receiver's +/// pace. +/// +/// Obtained from [`Connection::open_stream`](crate::Connection::open_stream). +/// Every write is a call with a deadline, so a receiver that stops draining +/// surfaces as an error on the write rather than as a hang — the same rule the +/// rest of the bus lives by. +pub struct StreamWriter { + connection: crate::Connection, + destination: BusName, + id: String, + content_type: Option, + declared_len: Option, + timeout: Duration, + seq: u64, + sent: u64, + finished: bool, +} + +impl StreamWriter { + pub(crate) fn new( + connection: crate::Connection, + destination: BusName, + id: String, + descriptor: StreamDescriptor, + timeout: Duration, + ) -> Self { + Self { + connection, + destination, + id, + content_type: descriptor.content_type, + declared_len: descriptor.total_len, + timeout, + seq: 0, + sent: 0, + finished: false, + } + } + + /// The handle to put in the method body that tells the receiver what these + /// bytes are for. + /// + /// Available before the payload has been written, and that is the intended + /// order: send the call first, then feed the stream. The window is only a + /// few megabytes, so a sender that writes everything before making the call + /// stalls against a reader that does not exist yet. + pub fn stream_ref(&self) -> StreamRef { + StreamRef { + id: self.id.clone(), + content_type: self.content_type.clone(), + len: self.declared_len, + } + } + + /// How many bytes have been accepted by the receiver so far. + pub fn sent(&self) -> u64 { + self.sent + } + + /// Write `bytes`, splitting them across as many chunks as it takes. + pub async fn write(&mut self, bytes: &[u8]) -> Result<()> { + for chunk in bytes.chunks(MAX_CHUNK_LEN) { + self.write_chunk(chunk).await?; + } + Ok(()) + } + + /// Write exactly one chunk, which must be no larger than [`MAX_CHUNK_LEN`]. + /// + /// There is no "already finished" case to guard against: [`Self::finish`] + /// and [`Self::abort`] both consume the writer, so the type system has + /// already ruled out a write after either of them. + pub async fn write_chunk(&mut self, chunk: &[u8]) -> Result<()> { + if chunk.len() > MAX_CHUNK_LEN { + return Err(Error::protocol(format!( + "chunk of {} bytes exceeds the {MAX_CHUNK_LEN}-byte cap", + chunk.len() + ))); + } + self.call( + "Write", + serde_json::json!([self.id, self.seq, base64::encode(chunk)]), + ) + .await?; + self.seq += 1; + self.sent += chunk.len() as u64; + Ok(()) + } + + /// Close the stream and return the handle, now carrying its final length. + pub async fn finish(mut self) -> Result { + self.call("Close", serde_json::json!([self.id, self.sent])) + .await?; + self.finished = true; + Ok(StreamRef { + id: self.id.clone(), + content_type: self.content_type.clone(), + len: Some(self.sent), + }) + } + + /// Abandon the stream, telling the receiver not to wait for the rest. + pub async fn abort(mut self) -> Result<()> { + self.finished = true; + self.call("Abort", serde_json::json!([self.id])).await?; + Ok(()) + } + + async fn call(&self, member: &str, args: Value) -> Result { + self.connection + .call_stream_member(&self.destination, member, args, self.timeout) + .await + } +} + +impl std::fmt::Debug for StreamWriter { + /// Deliberately never the payload: a writer is printed in error paths, and + /// the bytes going through it are the caller's data. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamWriter") + .field("destination", &self.destination) + .field("id", &self.id) + .field("sent", &self.sent) + .finish_non_exhaustive() + } +} + +impl Drop for StreamWriter { + /// A dropped writer aborts, so a sender that fails halfway does not leave + /// the receiver holding a window open until the idle reaper notices. + /// Best-effort by necessity: `Drop` cannot await, and the process may be on + /// its way out. + fn drop(&mut self) { + if self.finished { + return; + } + let (connection, destination, id, timeout) = ( + self.connection.clone(), + self.destination.clone(), + self.id.clone(), + self.timeout, + ); + if tokio::runtime::Handle::try_current().is_ok() { + tokio::spawn(async move { + let _ = connection + .call_stream_member(&destination, "Abort", serde_json::json!([id]), timeout) + .await; + }); + } + } +} + +/// The receiving half of a stream: chunks, in order, as they land. +/// +/// Reading incrementally is the point — a receiver writing a payload to disk +/// should never hold more than one chunk of it — but +/// [`StreamReader::read_to_end_capped`] is there for the common case where the payload +/// is merely too big for a frame, not too big for memory. +pub struct StreamReader { + chunks: mpsc::Receiver>, + /// Only the verdict is shared with the receiving connection — deliberately + /// not the whole stream record, whose drop is what closes this channel. + outcome: Arc>>, + content_type: Option, + declared_len: Option, +} + +impl std::fmt::Debug for StreamReader { + /// The metadata, never the buffered chunks. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StreamReader") + .field("content_type", &self.content_type) + .field("declared_len", &self.declared_len) + .finish_non_exhaustive() + } +} + +impl StreamReader { + /// What the sender said the payload is, if anything. + pub fn content_type(&self) -> Option<&str> { + self.content_type.as_deref() + } + + /// What the sender declared the total length to be, if it declared one. + pub fn declared_len(&self) -> Option { + self.declared_len + } + + /// The next chunk, or `None` at a clean end of stream. + /// + /// Returns an error if the sender aborted, went idle, or closed the stream + /// short of the length it declared — a truncated payload must never be + /// mistaken for a complete one. + pub async fn next_chunk(&mut self) -> Result>> { + if let Some(chunk) = self.chunks.recv().await { + return Ok(Some(chunk)); + } + match self.outcome.lock().expect("stream outcome lock").clone() { + Some(Outcome::Complete) => Ok(None), + Some(Outcome::Aborted(reason)) => Err(Error::StreamAborted { + reason: reason.to_string(), + }), + // The channel closed with no verdict recorded: the connection that + // was receiving the stream went away underneath it. + None => Err(Error::StreamAborted { + reason: "the connection closed mid-stream".to_string(), + }), + } + } + + /// Drain the whole stream into memory, refusing to exceed `limit` bytes. + /// + /// The initial reservation is **not** taken from the sender's declared + /// length. `declared_len` arrives from the peer before any payload does, so + /// sizing a buffer from it is a remote-triggered allocation — the same + /// mistake the frame-length cap in + /// [`codec`](crate::message::codec) exists to prevent, one layer up. A peer + /// could declare the maximum on each of its permitted streams and make a + /// receiver reserve gigabytes for bytes it never intends to send. Reserving + /// one chunk and letting the vector grow costs an amortised handful of + /// reallocations on a real transfer and nothing at all on a lie. + pub async fn read_to_end_capped(&mut self, limit: u64) -> Result> { + let mut out = Vec::with_capacity(limit.min(MAX_CHUNK_LEN as u64) as usize); + while let Some(chunk) = self.next_chunk().await? { + if out.len() as u64 + chunk.len() as u64 > limit { + return Err(Error::StreamTooLarge { limit }); + } + out.extend_from_slice(&chunk); + } + Ok(out) + } +} + +#[cfg(test)] +mod stream_test; diff --git a/crates/tinybus/src/stream/stream_test.rs b/crates/tinybus/src/stream/stream_test.rs new file mode 100644 index 0000000..941ca83 --- /dev/null +++ b/crates/tinybus/src/stream/stream_test.rs @@ -0,0 +1,983 @@ +//! Streams end to end, through a real broker. +//! +//! Through the broker rather than over a bare transport pair because the +//! ownership check on every chunk reads the `sender` the *broker* stamps — +//! testing it on a direct pair would test a code path where every peer looks +//! identical, which is exactly the case the check exists to rule out. + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use serde_json::Value; + +use crate::broker::Broker; +use crate::connection::Connection; +use crate::error::{Error, Result}; +use crate::message::Message; +use crate::name::{BusName, InterfaceName, MemberName, ObjectPath}; +use crate::ports::Transport; +use crate::stream::{ + MAX_CHUNK_LEN, STREAM_INTERFACE, STREAM_PATH, StreamDescriptor, StreamLimits, StreamRef, +}; +use crate::transport::memory::{MemoryBus, MemoryTransport}; + +const SINK: &str = "ai.tinyhumans.Sink"; +const SINK_PATH: &str = "/ai/tinyhumans/Sink"; + +/// A service that accepts a payload as a stream and reports what it received. +struct Sink { + connection: std::sync::Mutex>, +} + +impl Sink { + fn new() -> Arc { + Arc::new(Self { + connection: std::sync::Mutex::new(None), + }) + } + + fn connection(&self) -> Connection { + self.connection + .lock() + .unwrap() + .clone() + .expect("the sink is wired to its connection before it is called") + } +} + +#[async_trait] +impl crate::service::Interface for Arc { + fn name(&self) -> InterfaceName { + InterfaceName::new(SINK).unwrap() + } + + fn members(&self) -> Vec { + vec![ + MemberName::new("Absorb").unwrap(), + MemberName::new("Digest").unwrap(), + MemberName::new("Ignore").unwrap(), + ] + } + + async fn call(&self, member: &MemberName, args: Value) -> Result { + let (stream,): (StreamRef,) = serde_json::from_value(args)?; + let connection = self.connection(); + match member.as_str() { + // Buffer it and report the length and a checksum, so a transposed + // or dropped chunk cannot pass as a correct transfer. + "Absorb" => { + let bytes = connection.read_stream(&stream).await?; + Ok(serde_json::json!([bytes.len(), checksum(&bytes)])) + } + // Read chunk by chunk, holding one at a time. + "Digest" => { + let mut reader = connection.accept_stream(&stream)?; + let (mut len, mut sum) = (0usize, 0u64); + while let Some(chunk) = reader.next_chunk().await? { + len += chunk.len(); + sum = sum.wrapping_add(checksum(&chunk)); + } + Ok(serde_json::json!([len, sum])) + } + // Never touch the stream at all. + _ => Ok(serde_json::json!([0, 0])), + } + } +} + +fn checksum(bytes: &[u8]) -> u64 { + bytes.iter().fold(1469598103934665603u64, |hash, byte| { + (hash ^ *byte as u64).wrapping_mul(1099511628211) + }) +} + +fn payload(len: usize) -> Vec { + (0..len).map(|i| (i % 251) as u8).collect() +} + +/// A broker, a sink service that owns [`SINK`], and a client. +async fn bus() -> (Connection, Connection) { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + + let service = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let sink = Sink::new(); + *sink.connection.lock().unwrap() = Some(service.clone()); + service + .serve_at(ObjectPath::new(SINK_PATH).unwrap(), sink) + .await + .unwrap(); + service.request_name(SINK).await.unwrap(); + + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + (client, service) +} + +async fn absorb(client: &Connection, member: &str, bytes: &[u8]) -> Result<(usize, u64)> { + client + .call_with_stream( + BusName::new(SINK).unwrap(), + ObjectPath::new(SINK_PATH).unwrap(), + InterfaceName::new(SINK).unwrap(), + MemberName::new(member).unwrap(), + |stream| serde_json::json!([stream]), + bytes, + ) + .await +} + +#[tokio::test] +async fn a_payload_larger_than_one_frame_arrives_whole_and_in_order() { + let (client, _service) = bus().await; + // Comfortably past `MAX_FRAME_LEN`, so this payload could not have crossed + // as a single body no matter how it was encoded. + let bytes = payload(20 * 1024 * 1024); + let (len, sum) = absorb(&client, "Absorb", &bytes).await.unwrap(); + assert_eq!(len, bytes.len()); + assert_eq!(sum, checksum(&bytes)); +} + +#[tokio::test] +async fn a_payload_read_chunk_by_chunk_never_needs_the_whole_thing_in_memory() { + let (client, _service) = bus().await; + let bytes = payload(MAX_CHUNK_LEN * 3 + 17); + let expected = bytes + .chunks(MAX_CHUNK_LEN) + .fold(0u64, |sum, chunk| sum.wrapping_add(checksum(chunk))); + let (len, sum) = absorb(&client, "Digest", &bytes).await.unwrap(); + assert_eq!(len, bytes.len()); + assert_eq!(sum, expected); +} + +#[tokio::test] +async fn an_empty_payload_is_a_complete_stream_rather_than_an_error() { + let (client, _service) = bus().await; + let (len, sum) = absorb(&client, "Absorb", b"").await.unwrap(); + assert_eq!(len, 0); + assert_eq!(sum, checksum(b"")); +} + +#[tokio::test] +async fn a_payload_that_fits_the_window_can_finish_before_the_reader_attaches() { + let (client, _service) = bus().await; + // Written and closed with no reader in sight: the receiving method is only + // dispatched afterwards, and must still find the stream. + let mut writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::with_len(64)) + .await + .unwrap(); + let stream = writer.stream_ref(); + writer.write(&payload(64)).await.unwrap(); + writer.finish().await.unwrap(); + + let (len, _): (usize, u64) = client + .call( + BusName::new(SINK).unwrap(), + ObjectPath::new(SINK_PATH).unwrap(), + InterfaceName::new(SINK).unwrap(), + MemberName::new("Absorb").unwrap(), + serde_json::json!([stream]), + ) + .await + .unwrap(); + assert_eq!(len, 64); +} + +#[tokio::test] +async fn a_peer_cannot_write_into_a_stream_another_peer_opened() { + let (client, service) = bus().await; + let bus_again = client.clone(); + let mut writer = bus_again + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::with_len(1024), + ) + .await + .unwrap(); + let stream = writer.stream_ref(); + writer.write(&payload(16)).await.unwrap(); + + // The service is a third peer on the same bus with its own unique name. It + // knows the id — it is in this test's scope — and must still be refused, + // because the broker stamped a different sender on its call. + let forged = service + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Write", + serde_json::json!([stream.id, 1, super::base64::encode(b"intruder")]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!( + forged.wire_name(), + "ai.tinyhumans.tinybus.Error.UnknownStream", + "{forged}" + ); + + // And the legitimate owner's stream is untouched by the attempt. + writer.write(&payload(16)).await.unwrap(); + assert_eq!(writer.sent(), 32); +} + +#[tokio::test] +async fn a_chunk_out_of_order_aborts_the_stream_rather_than_transposing_it() { + let (client, _service) = bus().await; + let writer = client + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::with_len(1024), + ) + .await + .unwrap(); + let id = writer.stream_ref().id; + + let skipped = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Write", + serde_json::json!([id, 4, super::base64::encode(b"late")]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!(skipped.wire_name(), "ai.tinyhumans.tinybus.Error.Protocol"); + + // Aborted, not merely rejected: the stream is gone. + let after = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Write", + serde_json::json!([id, 0, super::base64::encode(b"first")]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!( + after.wire_name(), + "ai.tinyhumans.tinybus.Error.UnknownStream" + ); +} + +#[tokio::test] +async fn a_stream_longer_than_it_declared_is_cut_off_at_the_declaration() { + let (client, _service) = bus().await; + let mut writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::with_len(8)) + .await + .unwrap(); + let error = writer.write(&payload(9)).await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamTooLarge", + "{error}" + ); +} + +#[tokio::test] +async fn a_declared_length_over_the_receivers_cap_is_refused_before_a_byte_moves() { + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_stream_len: 1024, + ..StreamLimits::default() + }); + let error = client + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::with_len(4096), + ) + .await + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamTooLarge" + ); +} + +#[tokio::test] +async fn an_undeclared_stream_is_still_stopped_at_the_receivers_cap() { + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_stream_len: 1024, + ..StreamLimits::default() + }); + let mut writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::default()) + .await + .unwrap(); + writer.write_chunk(&payload(1024)).await.unwrap(); + let error = writer.write_chunk(b"one too many").await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamTooLarge" + ); +} + +#[tokio::test] +async fn a_peer_holding_open_more_streams_than_its_share_is_refused_a_new_one() { + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_streams_per_peer: 2, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + let _first = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let _second = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let error = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.TooManyStreams" + ); +} + +#[tokio::test] +async fn closing_short_of_the_declared_length_is_an_error_not_a_short_read() { + let (client, _service) = bus().await; + let mut writer = client + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::with_len(100), + ) + .await + .unwrap(); + writer.write(&payload(10)).await.unwrap(); + let id = writer.stream_ref().id; + // Claim a length the receiver did not get. A truncated payload accepted as + // a whole one is the failure this check exists to prevent. + let error = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Close", + serde_json::json!([id, 100]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!(error.wire_name(), "ai.tinyhumans.tinybus.Error.Protocol"); +} + +#[tokio::test] +async fn a_reader_sees_an_error_rather_than_an_eof_when_the_sender_aborts() { + let (client, service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let writer = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let stream = writer.stream_ref(); + let mut reader = service.accept_stream(&stream).unwrap(); + writer.abort().await.unwrap(); + + let error = reader.next_chunk().await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted", + "{error}" + ); +} + +#[tokio::test] +async fn a_dropped_writer_aborts_the_stream_instead_of_leaving_it_open() { + let (client, service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let stream = { + let writer = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + writer.stream_ref() + }; + let mut reader = service.accept_stream(&stream).unwrap(); + // The abort is spawned by `Drop`, so awaiting the reader is what waits for + // it — a deadline, not a sleep. + let error = tokio::time::timeout(Duration::from_secs(5), reader.next_chunk()) + .await + .expect("the dropped writer should abort promptly") + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted" + ); +} + +#[tokio::test] +async fn writing_to_a_stream_nobody_will_read_fails_instead_of_hanging_forever() { + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + window_chunks: 1, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + let mut writer = client + .open_stream_with_timeout( + &destination, + StreamDescriptor::default(), + Duration::from_millis(250), + ) + .await + .unwrap(); + + // The window takes one chunk; the second has nowhere to go, and the write's + // own deadline is what turns that into an error rather than a hang. + writer.write_chunk(b"first").await.unwrap(); + let error = writer.write_chunk(b"second").await.unwrap_err(); + assert_eq!(error.wire_name(), "ai.tinyhumans.tinybus.Error.Timeout"); +} + +#[tokio::test] +async fn a_stream_the_receiver_never_reads_does_not_stall_the_rest_of_the_bus() { + let (client, _service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let mut writer = client + .open_stream_with_timeout( + &destination, + StreamDescriptor::default(), + Duration::from_millis(250), + ) + .await + .unwrap(); + // `Ignore` never touches the stream, so the window fills and stays full. + let _ = client + .call::<(usize, u64)>( + destination.clone(), + ObjectPath::new(SINK_PATH).unwrap(), + InterfaceName::new(SINK).unwrap(), + MemberName::new("Ignore").unwrap(), + serde_json::json!([writer.stream_ref()]), + ) + .await + .unwrap(); + for _ in 0..StreamLimits::default().window_chunks { + let _ = writer.write_chunk(b"filler").await; + } + + // An ordinary call on the same connection still goes through: the stalled + // stream is the sender's problem and nobody else's. + let names = tokio::time::timeout(Duration::from_secs(5), client.list_names()) + .await + .expect("the wedged stream must not stall an unrelated call") + .unwrap(); + assert!(names.iter().any(|name| name.as_str() == SINK)); +} + +#[tokio::test] +async fn a_stream_can_only_be_read_once() { + let (client, service) = bus().await; + let writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::default()) + .await + .unwrap(); + let stream = writer.stream_ref(); + let _reader = service.accept_stream(&stream).unwrap(); + let error = service.accept_stream(&stream).unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted" + ); +} + +#[tokio::test] +async fn a_handle_that_names_no_stream_is_an_error_rather_than_an_empty_payload() { + let (_client, service) = bus().await; + let error = service + .accept_stream(&StreamRef { + id: "s404".to_string(), + content_type: None, + len: None, + }) + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.UnknownStream" + ); +} + +#[tokio::test] +async fn an_unknown_member_on_the_stream_interface_is_an_unknown_method() { + let (client, _service) = bus().await; + let error = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Rewind", + serde_json::json!([]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!(error.wire_name(), Error::UNKNOWN_METHOD); +} + +#[tokio::test] +async fn a_service_gets_the_stream_interface_without_exporting_anything() { + // A connection that has exported no objects at all still answers `Open`: + // bulk transfer is plumbing, not something each service opts into. + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let bare = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + bare.request_name("ai.tinyhumans.Bare").await.unwrap(); + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + + let mut writer = client + .open_stream( + &BusName::new("ai.tinyhumans.Bare").unwrap(), + StreamDescriptor::default().content_type("application/pdf"), + ) + .await + .unwrap(); + let stream = writer.stream_ref(); + assert_eq!(stream.content_type.as_deref(), Some("application/pdf")); + + let mut reader = bare.accept_stream(&stream).unwrap(); + writer.write(b"%PDF-1.7").await.unwrap(); + writer.finish().await.unwrap(); + assert_eq!(reader.next_chunk().await.unwrap().unwrap(), b"%PDF-1.7"); + assert_eq!(reader.next_chunk().await.unwrap(), None); +} + +#[tokio::test] +async fn a_chunk_over_the_cap_is_refused_by_the_sender_before_it_reaches_the_wire() { + let (client, _service) = bus().await; + let mut writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::default()) + .await + .unwrap(); + let error = writer + .write_chunk(&payload(MAX_CHUNK_LEN + 1)) + .await + .unwrap_err(); + assert_eq!(error.wire_name(), "ai.tinyhumans.tinybus.Error.Protocol"); +} + +#[tokio::test] +async fn a_chunk_over_the_cap_is_refused_by_the_receiver_too() { + // The sender-side check is a courtesy; the receiver's is the one that + // counts, because a peer is free not to run our sender. + let (client, _service) = bus().await; + let writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::default()) + .await + .unwrap(); + let id = writer.stream_ref().id; + let oversize = super::base64::encode(&payload(MAX_CHUNK_LEN + 1)); + let error = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Write", + serde_json::json!([id, 0, oversize]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!(error.wire_name(), "ai.tinyhumans.tinybus.Error.Protocol"); +} + +#[tokio::test] +async fn the_stream_interface_cannot_be_shadowed_by_a_service_exporting_it() { + // A service that exports its own interface at the stream address must not + // be able to intercept chunks: the connection answers streams before the + // object tree is consulted. + struct Impostor; + + #[async_trait] + impl crate::service::Interface for Impostor { + fn name(&self) -> InterfaceName { + InterfaceName::new(STREAM_INTERFACE).unwrap() + } + + fn members(&self) -> Vec { + vec![MemberName::new("Open").unwrap()] + } + + async fn call(&self, _member: &MemberName, _args: Value) -> Result { + Ok(Value::String("hijacked".to_string())) + } + } + + let (client, service) = bus().await; + service + .serve_at(ObjectPath::new(STREAM_PATH).unwrap(), Impostor) + .await + .unwrap(); + let writer = client + .open_stream(&BusName::new(SINK).unwrap(), StreamDescriptor::default()) + .await + .unwrap(); + assert_ne!(writer.stream_ref().id, "hijacked"); +} + +#[tokio::test] +async fn a_malformed_open_is_a_bad_arguments_error_that_does_not_quote_the_body() { + let (client, _service) = bus().await; + let error = client + .call_stream_member( + &BusName::new(SINK).unwrap(), + "Open", + serde_json::json!(["ceremonial-secret"]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.BadArguments" + ); + assert!(!error.to_string().contains("ceremonial-secret"), "{error}"); +} + +#[tokio::test] +async fn a_stream_call_with_no_member_is_rejected_by_the_receiver() { + // Sent down a bare transport rather than through `call_raw`, which runs + // `Message::validate` before enqueueing: going through the client would + // prove only that the *sender* refuses to build this message, and the + // property under test is that a receiver refuses to dispatch one. A peer + // running someone else's implementation is exactly who sends it. + let (mine, theirs) = MemoryTransport::pair(); + let receiver = Connection::attach(Arc::new(theirs)); + let mut malformed = Message::method_call( + BusName::new(SINK).unwrap(), + ObjectPath::new(STREAM_PATH).unwrap(), + InterfaceName::new(STREAM_INTERFACE).unwrap(), + MemberName::new("Open").unwrap(), + serde_json::json!([StreamDescriptor::default()]), + ); + malformed.header.member = None; + malformed.header.serial = 1; + mine.send(malformed).await.unwrap(); + + let reply = tokio::time::timeout(Duration::from_secs(5), mine.recv()) + .await + .expect("the receiver must answer rather than drop the call") + .unwrap() + .expect("the transport is still open"); + assert_eq!(reply.header.kind, crate::message::MessageKind::Error); + assert_eq!( + reply.header.error_name.as_deref(), + Some("ai.tinyhumans.tinybus.Error.Protocol"), + "{reply:?}" + ); + drop(receiver); +} + +#[tokio::test] +async fn a_stream_that_has_gone_idle_is_reaped_and_stops_holding_its_window() { + let (client, service) = bus().await; + // A zero idle timeout makes every existing stream idle by the time the next + // `Open` sweeps — the reaper's condition, expressed without waiting for a + // clock. + service.set_stream_limits(StreamLimits { + idle_timeout: Duration::ZERO, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + let abandoned = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let stream = abandoned.stream_ref(); + let mut reader = service.accept_stream(&stream).unwrap(); + + // The sweep runs on the next `Open`, so that is what collects the first. + let _next = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + + let error = reader.next_chunk().await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted", + "{error}" + ); + // And the sender is told, rather than writing into a stream that is gone. + let mut abandoned = abandoned; + assert!(abandoned.write_chunk(b"too late").await.is_err()); +} + +#[tokio::test] +async fn a_closed_stream_nobody_collects_is_evicted_oldest_first() { + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_streams_per_peer: 2, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + + // Three payloads written and closed, none ever read. A closed stream still + // holds its window, so the receiver must not accumulate them without bound. + let mut handles = Vec::new(); + for _ in 0..3 { + let mut writer = client + .open_stream(&destination, StreamDescriptor::with_len(4)) + .await + .unwrap(); + handles.push(writer.stream_ref()); + writer.write(b"data").await.unwrap(); + writer.finish().await.unwrap(); + } + + // The oldest is gone; the newest — the one a call is most likely still + // waiting on — survives. + let error = service.accept_stream(&handles[0]).unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.UnknownStream", + "{error}" + ); + let mut kept = service.accept_stream(&handles[2]).unwrap(); + assert_eq!(kept.next_chunk().await.unwrap().unwrap(), b"data"); +} + +#[tokio::test] +async fn a_closed_stream_still_counts_for_nothing_against_the_live_limit() { + // Closing frees the slot: a peer that finishes its transfers can keep + // opening new ones, which is the whole difference between the live cap and + // the uncollected cap. + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_streams_per_peer: 1, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + for _ in 0..3 { + let mut writer = client + .open_stream(&destination, StreamDescriptor::with_len(2)) + .await + .expect("a finished transfer must not hold its slot"); + let stream = writer.stream_ref(); + writer.write(b"hi").await.unwrap(); + writer.finish().await.unwrap(); + service.accept_stream(&stream).unwrap(); + } +} + +#[tokio::test] +async fn writing_after_close_is_refused_rather_than_appended() { + let (client, _service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let mut writer = client + .open_stream(&destination, StreamDescriptor::with_len(4)) + .await + .unwrap(); + let id = writer.stream_ref().id; + writer.write(b"data").await.unwrap(); + writer.finish().await.unwrap(); + + // A payload the receiver has already been told is complete must not grow. + let error = client + .call_stream_member( + &destination, + "Write", + serde_json::json!([id, 1, super::base64::encode(b"more")]), + Duration::from_secs(5), + ) + .await + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted", + "{error}" + ); +} + +#[tokio::test] +async fn a_sender_is_told_when_the_receiver_drops_the_reader_mid_transfer() { + let (client, service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let mut writer = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let stream = writer.stream_ref(); + let reader = service.accept_stream(&stream).unwrap(); + writer.write_chunk(b"first").await.unwrap(); + + // Nobody is going to look at the rest, so pushing it is wasted work on both + // sides — the sender learns immediately instead of at its deadline. + drop(reader); + let error = writer.write_chunk(b"second").await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted", + "{error}" + ); +} + +#[tokio::test] +async fn a_reader_reports_what_the_sender_declared_about_the_payload() { + let (client, service) = bus().await; + let writer = client + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::with_len(9).content_type("audio/wav"), + ) + .await + .unwrap(); + let reader = service.accept_stream(&writer.stream_ref()).unwrap(); + assert_eq!(reader.content_type(), Some("audio/wav")); + assert_eq!(reader.declared_len(), Some(9)); +} + +#[tokio::test] +async fn reading_past_the_callers_own_limit_is_an_error_not_a_silent_truncation() { + let (client, service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let mut writer = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + let mut reader = service.accept_stream(&writer.stream_ref()).unwrap(); + writer.write_chunk(&payload(64)).await.unwrap(); + + let error = reader.read_to_end_capped(16).await.unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamTooLarge", + "{error}" + ); +} + +#[tokio::test] +async fn a_reader_whose_connection_went_away_reports_it_rather_than_a_clean_eof() { + // No outcome is ever recorded when the receiving side simply disappears, and + // an unfinished payload must not be mistaken for a finished one. A bare + // receiver, because the shared fixture's service holds its own connection — + // a legitimate thing for a service to do, and a cycle that keeps the + // receiving side alive past the point this test needs it gone. + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let receiver = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + receiver + .request_name("ai.tinyhumans.Vanishing") + .await + .unwrap(); + let client = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + + let writer = client + .open_stream( + &BusName::new("ai.tinyhumans.Vanishing").unwrap(), + StreamDescriptor::default(), + ) + .await + .unwrap(); + let mut reader = receiver.accept_stream(&writer.stream_ref()).unwrap(); + // Leaked rather than dropped: a dropped writer aborts, which would record an + // outcome and test the wrong path. + std::mem::forget(writer); + drop(receiver); + + let error = tokio::time::timeout(Duration::from_secs(5), reader.next_chunk()) + .await + .expect("a vanished connection must not leave the reader parked") + .unwrap_err(); + assert_eq!( + error.wire_name(), + "ai.tinyhumans.tinybus.Error.StreamAborted", + "{error}" + ); +} + +#[tokio::test] +async fn neither_end_of_a_stream_prints_the_payload_when_debugged() { + // Both halves get printed in error paths, and what flows through them is + // the caller's data. + let (client, service) = bus().await; + let mut writer = client + .open_stream( + &BusName::new(SINK).unwrap(), + StreamDescriptor::default().content_type("text/plain"), + ) + .await + .unwrap(); + let reader = service.accept_stream(&writer.stream_ref()).unwrap(); + writer.write_chunk(b"recovery-phrase").await.unwrap(); + + let printed = format!("{writer:?} {reader:?}"); + assert!(!printed.contains("recovery-phrase"), "{printed}"); + assert!(printed.contains("StreamWriter"), "{printed}"); + assert!(printed.contains("StreamReader"), "{printed}"); +} + +#[tokio::test] +async fn one_peers_open_streams_do_not_consume_another_peers_slots() { + // The cap is per peer for the same reason every other queue on the bus is: + // a busy peer must not be able to starve a quiet one. + let (client, service) = bus().await; + service.set_stream_limits(StreamLimits { + max_streams_per_peer: 1, + ..StreamLimits::default() + }); + let destination = BusName::new(SINK).unwrap(); + + let _hog = client + .open_stream(&destination, StreamDescriptor::default()) + .await + .unwrap(); + // A second stream from the same peer is refused… + assert!( + client + .open_stream(&destination, StreamDescriptor::default()) + .await + .is_err() + ); + + // …while a different peer is unaffected by the first one's spending. + let other = service + .open_stream(&destination, StreamDescriptor::default()) + .await + .expect("another peer's slots are its own"); + assert!(!other.stream_ref().id.is_empty()); +} + +#[tokio::test] +async fn a_receiver_does_not_reserve_memory_for_a_length_the_sender_merely_claimed() { + // `total_len` arrives from the peer before any payload does. Sizing a + // buffer from it would let a peer declare the maximum on each stream it is + // allowed and make the receiver reserve gigabytes for bytes it never sends + // — the frame-length allocation problem, one layer up. The capacity of the + // returned buffer is what tells the two behaviours apart. + let (client, service) = bus().await; + let destination = BusName::new(SINK).unwrap(); + let claimed = 200 * 1024 * 1024; + let mut writer = client + .open_stream(&destination, StreamDescriptor::with_len(claimed)) + .await + .unwrap(); + let stream = writer.stream_ref(); + let mut reader = service.accept_stream(&stream).unwrap(); + writer.write_chunk(b"four").await.unwrap(); + // Closed cleanly at the byte count actually sent, so the read below + // succeeds and its buffer is the one the reservation produced. Aborting + // instead would hand back an empty vector and assert nothing. + writer.finish().await.unwrap(); + + let bytes = reader.read_to_end_capped(claimed).await.unwrap(); + assert_eq!(bytes, b"four"); + assert!( + bytes.capacity() as u64 <= MAX_CHUNK_LEN as u64, + "reserved {} bytes for a {claimed}-byte claim carrying {} bytes", + bytes.capacity(), + bytes.len() + ); +} diff --git a/docs/modules/README.md b/docs/modules/README.md index 99333d4..c5c718c 100644 --- a/docs/modules/README.md +++ b/docs/modules/README.md @@ -16,6 +16,7 @@ behaviours other modules rely on. | `connection` | [connection/README.md](connection/README.md) | | `proxy` | [proxy/README.md](proxy/README.md) | | `service` | [service/README.md](service/README.md) | +| `stream` | [stream/README.md](stream/README.md) | | `events` | source rustdoc (bounded domain-event fan-out) | | `global` | source rustdoc (one-time process-wide bus) | | `native` | source rustdoc (typed in-process request registry) | diff --git a/docs/modules/stream/README.md b/docs/modules/stream/README.md new file mode 100644 index 0000000..dcb5972 --- /dev/null +++ b/docs/modules/stream/README.md @@ -0,0 +1,117 @@ +# `stream` — bulk payloads larger than one frame + +A frame is capped at 16 MiB and that cap does not move: the length arrives from +the wire before the bytes, so a reader that trusted it would allocate whatever a +peer announced. `stream` is how a payload bigger than a frame crosses the bus +anyway — the sender opens a stream on the receiver, writes it as a sequence of +bounded chunks, and closes it. The method call carries a `StreamRef`, a handle a +few dozen bytes long; the bytes travel beside it. + +## Why it is a peer interface, not a broker feature + +Every chunk is an ordinary method call addressed to the receiving peer. The +broker reads the header, routes it, forwards it. It never assembles a stream and +never sees a chunk as anything but traffic — a broker that buffered payloads +would be a process holding every mail body and every recovery phrase on the bus, +which is the one thing the security boundary says it must never be. + +The other consequence: the receiver decides its own limits. Nothing about a +stream is negotiated, because a limit a sender can talk you out of is not a +limit. + +## The shape of a transfer + +```rust +let mut writer = conn.open_stream(&dest, StreamDescriptor::with_len(len)).await?; +let handle = writer.stream_ref(); // goes in the method body +// …issue the call, then feed the stream while the call is outstanding… +writer.write(&bytes).await?; +writer.finish().await?; +``` + +On the receiving side, inside the method: + +```rust +let mut reader = conn.accept_stream(&handle)?; +while let Some(chunk) = reader.next_chunk().await? { + file.write_all(&chunk).await?; // one chunk in memory, never more +} +``` + +`Connection::call_with_stream` does the interleaving for the common case, and +`Connection::read_stream` buffers a whole payload for when it is too big for a +frame but not too big for memory. + +**Order matters.** Send the call and *then* feed the stream. The receiver's +window is a few megabytes, so a sender that writes an entire payload before +making the call stalls against a reader that does not exist yet. This is not a +wart to be fixed with a bigger buffer — the bounded window is the flow control. + +## Flow control, and what a misbehaving peer costs + +`Write` is a call, so it has a reply and a deadline. The receiver does not reply +until the chunk has room in the reader's window, so a sender runs exactly as +fast as the receiver drains. There is no unbounded buffer anywhere in the path. + +That is the misbehaving-peer invariant applied to bulk transfer. Walk the cases: + +| A peer that… | Costs it | Costs anyone else | +| --- | --- | --- | +| never reads a stream sent to it | its sender's write deadline | nothing | +| opens streams and abandons them | its own per-peer slots | one window each, reaped after `idle_timeout` | +| writes past the length it declared | the stream, aborted | nothing | +| writes chunks out of order | the stream, aborted | nothing | +| exits mid-transfer | the transfer | one window, until the reaper | + +A closed-but-uncollected stream is capped separately from a live one, because +the two are different failures: too many live streams is a sender running ahead +of itself, too many closed ones is a receiver not collecting what it was sent. + +## Ownership + +Chunks are authorised by the `sender` the broker stamps, and only by that. The +peer that called `Open` is the only peer whose `Write`, `Close` or `Abort` that +stream will answer; every other peer gets `UnknownStream` — the same error as a +handle that names nothing, because distinguishing the two would let a peer probe +for transfers running between two others. + +This is the whole authorisation story for streams, and it rests entirely on the +broker overwriting `sender` on ingress. + +## Ordering + +Chunks carry a sequence number and the receiver requires the next one exactly. +The transport is already ordered, so this is not about a reordering network: it +catches a sender that pipelines. Two chunks in flight at once are dispatched +into two tasks on the receiver and could land either way round, and silently +transposing two megabytes of a PDF is worse than an error. + +## What this costs, and what replaces it later + +A chunk is base64 inside a JSON body, so a transfer pays about a third of its +size in overhead plus a round trip per 512 KiB. That is the price of bulk data +travelling the same wire as everything else, and it is why `ROADMAP.md` still +wants `SCM_RIGHTS`. When fd passing lands it slots in under this same API, +because callers hold a `StreamRef` rather than a byte array — the fast path can +change without the interface changing. + +For a payload that is genuinely huge, the pre-existing convention still applies +and is still cheaper: pass a path, and own the file's lifetime. + +## Limits + +`StreamLimits`, per receiving connection, changed with +`Connection::set_stream_limits`: + +| Field | Default | Bounds | +| --- | --- | --- | +| `max_stream_len` | 256 MiB | one transfer | +| `max_streams_per_peer` | 4 | concurrent transfers from one peer | +| `window_chunks` | 8 (≈4 MiB) | bytes in flight per transfer | +| `idle_timeout` | 60 s | how long an abandoned transfer holds its window | + +New limits apply from the next `Open`. A stream already running keeps the window +it was opened with: shrinking a window under a sender mid-transfer would abort a +transfer that was within the rules when it started. + +See [protocol.md](../../protocol.md#bulk-streams) for the wire members. diff --git a/docs/protocol.md b/docs/protocol.md index 67ceb69..ba7c84a 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -129,6 +129,51 @@ announced as `ModuleStateChanged` with body state has a safe refusal or fault reason. Name ownership changes still announce when a module attaches or stops. +## Bulk streams + +A payload larger than one frame does not travel in a body. The sender opens a +stream on the *receiving peer* and writes it as chunks; the method call carries +only a handle. The broker is not involved beyond routing — every member below is +an ordinary method call addressed to the receiving peer. + +Path `/ai/tinyhumans/tinybus/Stream`, interface `ai.tinyhumans.tinybus.Stream`. +Every peer answers it, whether or not it exported anything. + +| Member | Body | Returns | +| --- | --- | --- | +| `Open` | `[{"content_type"?, "total_len"?}]` | an opaque stream id | +| `Write` | `[id, seq, base64]` | `null` once the chunk is accepted | +| `Close` | `[id, total_len]` | `null`; `total_len` must equal what was written | +| `Abort` | `[id]` | `null` | + +The handle that travels in a method body is +`{"id": …, "content_type"?: …, "len"?: …}`. + +Rules a receiver enforces, and a sender must expect: + +- **Chunks are capped at 524 288 bytes** before base64 — a chunk plus its + encoding overhead must fit a frame with room to spare. +- **`seq` starts at 0 and increments by exactly one.** A gap aborts the stream + rather than transposing it. Do not pipeline writes: two chunks in flight can + be dispatched into two tasks and land either way round. +- **`Write` does not reply until the chunk has room** in the receiver's window. + That reply is the flow control; a sender is never more than a window ahead. + Like every call it has a deadline, so a receiver that stops reading surfaces + as an error rather than a hang. +- **Only the peer that called `Open` may write to the stream.** Authorisation is + the broker-stamped `sender` and nothing else. Any other peer gets + `UnknownStream`, which is also what an id naming nothing returns — the two are + deliberately indistinguishable. +- **`Close` declares the total.** A mismatch is an error and the payload is not + delivered as a short read. +- **Limits belong to the receiver** and are not negotiated: a maximum stream + length, a maximum number of concurrent streams per peer, a window, and an idle + timeout after which an abandoned stream is reaped. + +Send the call carrying the handle *before* writing the payload. The window is a +few megabytes, so a sender that writes everything up front stalls against a +reader that has not been dispatched yet. + ## Match rules Comma-separated `key=value`. Unset keys match anything; every set key must @@ -163,6 +208,10 @@ Bus-generated names: | `…Error.UnknownMethod` | the interface has no such member | | `…Error.BadArguments` | the body did not match the member's signature | | `…Error.Failed` | a method failed with no more specific mapping | +| `…Error.UnknownStream` | no such stream, or not one this peer opened | +| `…Error.StreamAborted` | the stream ended before it was complete | +| `…Error.StreamTooLarge` | the stream exceeds what the receiver accepts | +| `…Error.TooManyStreams` | this peer already holds its share of open streams | (`…` is `ai.tinyhumans.tinybus`.) A service should define its own dotted names under its own interface — `ai.tinyhumans.openhuman.Voice.Error.NoDevice` — for