swactor/docs/development_history/STREAMS.md

Swactor Stream Primitive -- Architectural Design

Context

Swactor has a distributed actor runtime with SWIM membership, Kademlia routing, and a content-addressed datastore. The current datastore transfers blobs one chunk at a time via actor message round-trips -- extremely slow for large objects. Beyond the datastore, the system needs a general-purpose bulk data transfer primitive for ML workloads (training data, weight checkpoints, gradient exchange), real-time media (video/voice), and future game state replication.

The stream primitive is a high-performance data channel between nodes that actors negotiate and manage but do not sit on the data path of. It should achieve top-class throughput by leveraging QUIC's multiplexed streams directly, bypassing the actor mailbox system for data transfer.

Decisions made:

• Data path: StreamHandle with try_read/try_write; actors receive lightweight notification messages but data bypasses mailboxes
• Reliability: Reliable-only MVP; abstraction designed so unreliable (QUIC datagrams) can be added later
• Locality: Cross-node only; same-node actors use regular messages
• Crate: New crates/streams/ crate

---

1. Core Concept: Control Plane vs Data Plane

The fundamental architecture separates stream management (control plane) from data transfer (data plane).

Control plane -- actor messages through normal mailboxes:

• Stream negotiation (open, accept, reject)
• Parameter configuration (buffer sizes, chunk sizes, parallelism)
• Lifecycle events (established, closed, error)
• Progress/health notifications

Data plane -- bypasses actors entirely:

• Raw bytes flow through QUIC streams on the iroh transport
• Managed by async tasks on the IrohDriver's tokio runtime
• Actors interact via StreamHandle objects (try_read/try_write), not mailbox messages
• QUIC's built-in flow control handles backpressure

CONTROL PLANE (actor messages, mailboxes, worker ticks) +--------+ StreamOpen +-----------+ StreamAccept +--------+ | Actor | -----------> | Stream | <------------- | Actor | | (nodeA)| | Manager | |(nodeB) | +--------+ +-----------+ +--------+ | | | | StreamReady(handle) | | StreamReady(handle) v v v DATA PLANE (tokio tasks, QUIC streams, pre-allocated buffers) +----------+ bytes +----------+ bytes +----------+ | SendHalf | =========> | QUIC | =========> | RecvHalf | | (writer) | N parallel| streams | N parallel | (reader) | +----------+ stripes +----------+ stripes +----------+

---

2. Stream Identity and Addressing

StreamId: A 16-byte random identifier, generated by the initiator during negotiation. Deliberately not an ActorAddress -- streams are not actors, are not placed on workers, and are not discoverable via Kademlia. Keeping them out of the AddressMap avoids polluting the actor routing hot path.

Full stream address: The tuple (NodeId, StreamId) is globally unique. A node can host many concurrent streams to many peers.

ALPN separation: Streams use a new protocol identifier swactor/stream/1, separate from the existing swactor/swim/1 used for membership. This means:

• The iroh accept loop can distinguish stream connections from protocol messages immediately
• Stream data never blocks or interferes with cluster heartbeats
• Stream connections can have different tuning in the future

---

3. QUIC Stream Utilization

Parallel Stripes for Blob Transfers

For a single large transfer, multiple QUIC streams are opened in parallel on the same QUIC connection. Each stream carries a disjoint range of the data. This is the stripe count, negotiated during handshake (default: 4).

Why multiple streams? A single QUIC stream can be limited by per-stream receive-window backpressure. Multiple streams allow the sender to push data into QUIC's send buffer more aggressively, keeping the congestion window filled. Measurements from quinn/s2n-quic show 2-8 parallel streams can improve throughput 2-4x on high-bandwidth-delay-product links.

Stream layout per transfer:

• Stream 0 (control stream): Bidirectional QUIC stream. Carries the handshake header and out-of-band signals (completion, cancel, errors, health). Stays open for the transfer's lifetime.
• Streams 1..N (data stripes): Unidirectional QUIC streams, each carrying sequential chunks. Stripe assignment is round-robin by chunk index.

Connection Reuse

Multiple concurrent streams between the same two nodes share one QUIC connection (on the stream ALPN). QUIC multiplexing handles this natively. The streams crate maintains a connection cache separate from the SWIM connection cache.

---

4. Wire Format

Two layers of wire format: the stream-level protocol (negotiation + data framing) and the blob transfer application protocol that rides on top of it.

Control Stream Header (stream-level)

[2B magic: 0x53 0x57] -- "SW" [1B version: 0x01] [16B StreamId] [1B mode] -- 0x01=BlobTransfer, 0x02=ContinuousStream (future) [1B stripe_count] -- parallel data stripes (1-255) [4B frame_size (BE u32)] -- maximum frame payload size in bytes [4B metadata_len (BE u32)] [N bytes metadata] -- negotiation payload (e.g., ContentHash for blob transfer)

Data Stripe Frame Format (stream-level)

[4B frame_len (BE u32)] -- 0 = end-of-stripe [N bytes payload] -- raw data bytes

Deliberately minimal. No per-frame type tags (QUIC provides ordered reliable delivery), no per-frame checksums on the wire (QUIC provides TLS integrity for transport), no per-frame metadata. Every byte of overhead on the hot path costs throughput.

BlobTransfer Application Protocol

For blob transfers, the StreamConfig.metadata carries the 32-byte ContentHash of the requested blob (so the serve side knows what to send). The actual blob data flows over the StreamHandle with this application-level framing:

[4B manifest_json_length (u32 BE)] [N bytes manifest JSON] -- serialized ObjectManifest [chunk_0 raw bytes] -- size from manifest.chunks[0].size [chunk_1 raw bytes] -- size from manifest.chunks[1].size ...

The receiver knows each chunk's expected size and blake3 hash from the manifest. Each chunk is verified individually on arrival: blake3(chunk_data) == chunk_ref.hash. Corrupted chunks cause immediate transfer failure. This is implemented by the send_blob and recv_blob async functions in crates/datastore/src/blob_transfer.rs.

Note: the blob transfer protocol sends chunks sequentially through the StreamHandle, which distributes data frames across stripes via the data-plane layer's round-robin. Individual chunks are not split across stripes -- the stripe layer is transparent to the application protocol.

---

5. Buffering Strategy

Pre-allocated Sliding Window (Zero Allocation on Hot Path)

The buffer pool is a sliding window, not a store. It never holds the entire blob in memory -- data flows through it like water through a pipe. A 1TB transfer uses the same ~4MB of buffer memory as a 1MB transfer; only the duration changes.

All buffers are allocated during stream setup, not per-frame.

Sender pipeline (per stripe, double-buffered): Source (disk/memory/computation) → [Buffer A: being filled from source] → [Buffer B: being written to QUIC] → Buffer B recycled → becomes the next Buffer A → repeat until source exhausted One buffer is being filled while the other is being sent. When QUIC accepts Buffer B's bytes, it's recycled and refilled from the source. The source can be disk I/O, a computation producing data, or anything that yields bytes.

Receiver pipeline (per stripe, double-buffered): QUIC recv stream → [Buffer A: being filled from QUIC] → [Buffer B: being written to disk/consumed] → Buffer B recycled → becomes the next Buffer A → repeat until stream ends The receiver reads from QUIC into one buffer while the previous buffer is being written to disk (for blob transfer) or consumed by the application. Buffers are recycled, never allocated mid-transfer.

Backpressure chain (end-to-end): Source read speed → fills sender buffer pool (2 per stripe) → QUIC congestion window → network bandwidth → QUIC receive window → fills receiver buffer pool (2 per stripe) → sink write speed (disk I/O, consumer processing)

If ANY link is slow, pressure propagates backward automatically. No custom flow control needed -- QUIC handles it.

Sizing:

• Pool: stripe_count * 2 buffers per side = 8 buffers (at 4 stripes)
• Frame size: 256KB per frame (separate from the datastore's 1MB storage chunk size)
• Total memory per stream direction: 8 x 256KB = 2MB
• Total for a bidirectional transfer: ~4MB, regardless of blob size
• At ~1200 bytes per QUIC packet, 256KB = ~213 packets. Smaller blast radius on packet loss than 1MB, better interleaving across stripes, aligns with OS page sizes.

TB-Scale Considerations

For very large transfers (100GB+ ML weights, TB-scale training data), additional design considerations apply:

Manifest streaming: At 1MB datastore chunks, a 1TB blob has ~1M chunks. Each ChunkRef is ~40 bytes, so the manifest is ~40MB. This is too large for a single negotiation payload. The current implementation sends the manifest as a JSON preamble on the data stream itself (not in the negotiation metadata). For TB-scale, the manifest could be streamed progressively instead of loaded all at once.

Per-chunk verification on arrival: The receiver verifies each chunk individually as it arrives: blake3(chunk_data) == chunk_ref.hash. This is implemented in recv_blob. A corrupted chunk causes immediate transfer failure. This catches problems early rather than waiting for full reassembly.

Progressive resume tokens: Resume tokens are emitted periodically (e.g., every 1000 chunks or every 256MB, whichever comes first), not just on failure. The sender acknowledges receipt of resume tokens. On connection loss, the receiver persists the latest resume token, and a new stream can resume from that point. For a 1TB transfer, a resume token with a 1M-bit BitVec is ~125KB -- cheap to exchange. (Not yet implemented -- the ResumeToken type exists but nothing emits or consumes it.)

Disk I/O as the bottleneck: For TB-scale over fast networks (10Gbps+), disk I/O often becomes the bottleneck rather than the network. The buffering strategy handles this naturally: when disk writes slow down, the receiver's buffer pool fills, QUIC backpressure kicks in, and the sender slows to match. No special handling needed -- the pipeline self-regulates. For maximum disk throughput, the receiver can use O_DIRECT or memory-mapped writes, but this is an implementation optimization, not an architectural decision.

Stripe count scaling: For very high bandwidth links, the default 4 stripes may not be enough to saturate the connection. The stripe count should be configurable up to 16, negotiated during handshake based on the expected transfer size and link characteristics. A heuristic: min(16, max(4, total_chunks / 1000)) -- more stripes for larger transfers.

---

6. StreamHandle -- The Actor-Facing API

The StreamHandle is a lightweight, Send (but not Clone) object that actors store in their state. It communicates with the data-plane tokio tasks via channels internally.

Writer interface:

• try_write(data: &[u8]) -> Result<usize, StreamError> -- Non-blocking. Returns bytes accepted.
• flush() -- Signal that buffered data should be sent.
• close() -- Graceful close.

Reader interface:

• try_read(buf: &mut [u8]) -> Result<usize, StreamError> -- Non-blocking. Returns bytes read, 0 if none available.
• has_data() -> bool -- Check if data is available without consuming it.

BlobTransfer Async Functions

Rather than a wrapper object, blob transfer uses standalone async functions that run inside tokio tasks (spawned after StreamReady). These functions loop over try_write/try_read with tokio::task::yield_now() for cooperative scheduling:

• send_blob(send, manifest, read_chunk) -- Writes the manifest preamble, then calls read_chunk(hash) for each chunk on-demand and writes it. At most one chunk is in memory at a time on the sender side. The read_chunk callback allows any data source (BlobStore via Inbox, in-memory, etc).
• recv_blob(recv) -- Reads the manifest, then reads and blake3-verifies each chunk. Returns ReceivedBlob { manifest, chunks }.
• poll_inbox(inbox, timeout) -- Async version of the bridge.rs poll_response pattern. Yields instead of thread::sleep.

These live in crates/datastore/src/blob_transfer.rs. The key insight: since actors can't await futures, the pattern is for the actor to receive StreamReady, extract the StreamHandle via OneShot::take(), spawn a tokio task for the I/O loop, then stop itself. The tokio task sends results back to other actors via runtime.send_to().

Why non-blocking? Actor handlers are synchronous (fn handle(&mut self, ctx: &Ctx, msg)). They cannot await futures. The try_read/try_write pattern fits naturally. The tokio task bridge is the mechanism for async I/O.

---

7. Actor Integration: Negotiation Protocol

Opening a Stream (Initiator)

1. Actor sends a StreamOpen control message (through normal actor mailbox routing) to a StreamManager system actor. Contains: target_node: NodeId, mode, metadata (e.g., ContentHash + manifest for blob transfer), reply_to: ActorAddress.
2. StreamManager validates the request, allocates a StreamId, and posts an async task to the tokio runtime that:

• Opens a QUIC connection to the target (stream ALPN)
• Opens the control bidirectional stream
• Sends the stream header
• Waits for accept/reject

3. On accept: StreamManager sends StreamReady { stream_id, handle: StreamHandle } back to the requesting actor.

Accepting a Stream (Receiver)

1. IrohDriver's accept loop receives connection on stream ALPN.
2. Reads control stream header, extracts StreamId + mode + metadata.
3. Sends StreamIncoming actor message to local StreamManager.
4. StreamManager routes to registered stream acceptors (actors that called StreamListen).
5. Matching actor receives StreamOffer { stream_id, mode, metadata } in its mailbox.
6. Actor sends StreamAccept or StreamReject back to StreamManager.
7. On accept: StreamManager allocates buffers, spawns data-plane tasks, sends StreamReady { handle } to the accepting actor.

Notification Model (Hybrid)

Stream data bypasses mailboxes, but actors need to know when data is available:

• The data-plane tasks inject lightweight StreamEvent sentinel messages into the owning actor's mailbox when state changes: DataReady, WriteReady, Closed, Error.
• Coalescing: An atomic flag prevents duplicate notifications. Set when notification posted, cleared when actor handles it. A high-throughput stream generates at most one DataReady per actor tick, not one per frame.
• The actor's handle_any dispatches StreamEvent via downcast (same mechanism as Down and ActorExited today -- no core trait changes needed).
• Actors can also proactively call handle.try_read() from any handler, not just in response to DataReady.

---

8. The StreamManager Actor

A system actor spawned alongside the IrohDriver, registered under a well-known name. It is the bridge between the actor world and the stream data plane.

Responsibilities:

• Registry of active streams: StreamId -> StreamState
• Handle StreamOpen / StreamAccept / StreamReject / StreamListen / StreamClose messages
• Spawn and supervise data-plane tokio tasks
• Monitor stream-holding actors; clean up streams when actors die
• Expose stream metrics (active streams, throughput, errors) for the dashboard

Communication with tokio runtime: Uses tokio::sync::mpsc and tokio::sync::oneshot channels. Posts commands to async tasks, receives results as actor messages (via the Inbox pattern already used by DatastoreBridge).

---

9. Flow Control and Backpressure

Three layers, all leveraging what QUIC already provides:

1. QUIC-level: Per-stream and per-connection flow control (receive window, congestion window). This is the primary mechanism. Not duplicated.
2. Buffer pool saturation: When receiver's pre-allocated buffer pool is full, the recv-side tokio task stops reading from QUIC. QUIC's receive window closes, sender stops transmitting. Natural backpressure without custom protocol.
3. StreamHandle backpressure: try_write() returns 0 bytes accepted when the send buffer is full. The actor knows to back off or buffer internally.

No custom flow control protocol. QUIC's congestion control (Cubic/BBR) is battle-tested. Adding application-level flow control would add complexity and latency without benefit.

Cancellation

• Cooperative: StreamCancel signal on the control stream
• Abrupt: reset() on the QUIC streams
• Nuclear: close the QUIC connection (node shutdown only)

---

10. Error Handling and Recovery

Failure Modes

┌──────────────────┬──────────────────────┬────────────────────────────────────────────────┐ │ Failure │ Detection │ Behavior │ ├──────────────────┼──────────────────────┼────────────────────────────────────────────────┤ │ Frame corruption │ QUIC TLS + checksums │ Automatic retransmit │ ├──────────────────┼──────────────────────┼────────────────────────────────────────────────┤ │ Stream reset │ QUIC RST_STREAM │ StreamEvent::Error to owning actor │ ├──────────────────┼──────────────────────┼────────────────────────────────────────────────┤ │ Connection loss │ QUIC timeout │ StreamEvent::Error on all streams to that node │ ├──────────────────┼──────────────────────┼────────────────────────────────────────────────┤ │ Node death │ SWIM declares Dead │ StreamEvent::Error on all streams to that node │ ├──────────────────┼──────────────────────┼────────────────────────────────────────────────┤ │ Owner actor dies │ Worker cleanup phase │ Stream closed, remote side notified │ └──────────────────┴──────────────────────┴────────────────────────────────────────────────┘

Resume Tokens for Blob Transfers (Not Yet Implemented)

For large transfers, the receiver periodically emits a ResumeToken on the control channel:

ResumeToken { stream_id: StreamId, manifest_hash: ContentHash, chunks_received: BitVec, -- which chunks confirmed stored }

On failure, the initiator can open a new stream with the ResumeToken. The sender skips confirmed chunks. This avoids retransmitting terabytes when a checkpoint transfer fails near completion. Leverages the existing ObjectManifest/ChunkRef model.

The ResumeToken type is defined in crates/streams/src/types.rs but emission/consumption logic is deferred to a future stage.

---

11. Integration with Existing Datastore

The stream primitive adds a parallel transfer path to the datastore. The existing chunk-at-a-time TransferActor is preserved for compatibility; the new stream path is used when stream support is configured.

Architecture:

DOWNLOAD SIDE: SERVE SIDE:

DatastoreNode StreamListener │ DownloadViaStream (Incoming = StreamNotification) │ ctx.spawn(StreamDownloader) │ on StreamOffer → HandleStreamOffer ▼ ▼ StreamDownloader DatastoreNode (Incoming = StreamNotification) │ HandleStreamOffer │ on_start: Open → StreamManager │ ctx.spawn(StreamServer) │ StreamReady → tokio task: ▼ │ recv_blob → verify → write chunks StreamServer │ send completion to DatastoreNode (Incoming = StreamNotification) ▼ │ on_start: Accept → StreamManager DatastoreNode │ StreamReady → tokio task: │ StreamDownloadComplete │ read manifest from BlobStore (Inbox) │ persist metadata, reply to caller │ for each chunk: read from BlobStore, │ write to stream (one at a time) │ close stream

Design principles:

• No bridge/shim actors. Stream-facing actors use Incoming = StreamNotification directly.
• No preloading all chunks into memory. Chunks flow on-demand: storage → network.
• DatastoreNode stays simple (fire-and-forget coordination). The stream actors own the full I/O lifecycle.
• After receiving StreamReady, actors spawn tokio tasks for I/O. Tokio tasks communicate results back via runtime.send_to().
• StreamServer reads chunks on-demand — at most one chunk in memory at a time.

Flow:

1. DatastoreNode receives DownloadViaStream { content_hash, source_node, reply_to }.
2. Spawns a StreamDownloader actor, which sends Open to StreamManager with metadata = content_hash.0 (32 bytes).
3. Remote StreamListener receives StreamOffer, extracts ContentHash from metadata, sends HandleStreamOffer to DatastoreNode.
4. Remote DatastoreNode spawns a StreamServer actor, which sends Accept to StreamManager.
5. StreamServer receives StreamReady, spawns tokio task: reads manifest from BlobStore, then streams each chunk on-demand via send_blob.
6. StreamDownloader receives StreamReady, spawns tokio task: calls recv_blob, writes chunks to BlobStore (fire-and-forget), notifies DatastoreNode of completion.
7. DatastoreNode creates ObjectEntry and persists via MetadataActor, which sends PutOk to the original caller.

This eliminates the round-trip-per-chunk bottleneck. A 1GB object with 1MB chunks currently requires 1,024 sequential round-trips. With streams and 4 parallel stripes, the entire blob flows in a single burst limited only by network bandwidth.

---

12. Growth Path

Phase 1 (MVP): Reliable Ordered Blob Transfer — IMPLEMENTED

• StreamConfig with BlobTransfer mode only
• New ALPN swactor/stream/1 handler
• StreamOpen/StreamAccept handshake
• Parallel striped data transfer
• StreamHandle with try_read/try_write
• StreamManager actor
• Datastore integration (StreamListener, StreamDownloader, StreamServer actors)
• BlobTransfer application protocol (send_blob/recv_blob with per-chunk blake3 verification)

Remaining MVP work:

• Two-node integration test (real QUIC, full download flow)
• Resume tokens (ResumeToken type exists, emission/consumption not yet wired)

Phase 2: Continuous Streams

• ContinuousStream mode (no total size known)
• Single bidirectional QUIC stream (no striping)
• Variable-sized message frames
• Bounded ring buffer backpressure
• Enables: federated learning gradient streams, data pipelines

Phase 3: Unreliable Datagrams

• UnreliableSequenced reliability mode using QUIC datagrams
• Sequence-based frame dropping (latest-wins)
• Receiver-side jitter buffer
• Advisory StreamThrottle on control channel
• Enables: voice/video, game entity state replication

Phase 4: Priority and QoS

• priority: u8 in StreamConfig
• Priority-aware write scheduler across concurrent streams
• QUIC stream priority hints
• Per-stream health reporting and dashboard integration
• Enables: simultaneous video + checkpoint without starvation

Phase 5: Parallel Unordered Transfer

• ReliableUnordered mode: parallel QUIC streams per chunk, independent delivery
• Configurable parallelism
• Enables: gradient exchange for distributed ML (any chunk consumable independently)

---

13. Key Design Decisions Summary

┌───────────────────┬───────────────────────────────────────────────┬────────────────────────────────────────────────────────────┐ │ Decision │ Choice │ Rationale │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Data path │ StreamHandle bypass, tokio task bridge │ Max throughput; actors manage, don't bottleneck │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Stream identity │ 16-byte StreamId, not ActorAddress │ Streams are not actors; avoid polluting address space │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ ALPN │ Separate swactor/stream/1 │ Isolate from SWIM; no interference with heartbeats │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Parallel stripes │ 4 QUIC streams per blob transfer │ Saturate congestion window on high-BDP links │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Flow control │ QUIC built-in only │ Don't duplicate what the transport does well │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Wire format │ 4-byte length prefix, no type tags │ Minimal per-frame overhead │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Stream chunk size │ 256KB │ Better packet-loss resilience, page-aligned │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Buffering │ Pre-allocated slab per stream │ Zero allocation on hot path │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Blob protocol │ Async functions, not wrapper object │ Simpler; tokio tasks own the I/O loop after StreamReady │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Actor pattern │ Spawn actor → StreamReady → tokio task → stop │ Clean separation; actor negotiates, task does I/O │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Chunk I/O │ On-demand via Inbox polling (poll_inbox) │ At most 1 chunk in memory; no preloading │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Crate │ New crates/streams/ │ Optional, clean dependency graph │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ MVP scope │ Reliable ordered only │ Covers ML + datastore; unreliable added later │ ├───────────────────┼───────────────────────────────────────────────┼────────────────────────────────────────────────────────────┤ │ Locality │ Cross-node only │ Focused scope; same-node uses regular messages │ └───────────────────┴───────────────────────────────────────────────┴────────────────────────────────────────────────────────────┘