swactor/DATA_PLANE_SPSC_STREAM_MVP.md

Data-plane SPSC stream MVP

Status

Pending implementation.

This document is the authoritative plan for the remaining data-plane stream slice. Blob namespace and resolution work is defined separately in DATA_PLANE_VIRTUAL_BLOB_NAMESPACE_MVP.md.

Goal

Complete the existing public stream API for one producer and one consumer:

reader = await ctx.data.read_stream(path)

async with ctx.data.write_stream(path) as writer:
    await writer.write(data)

Swactor must resolve the logical path, match exactly one source with exactly one sink, provision the local arena rings and remote transport, preserve byte order, apply lossless backpressure, propagate terminal state, and clean up every binding.

The final Myelin acceptance path replaces the temporary Unix-socket result bridge with this implementation.

Scope

The MVP supports only:

• one producer;
• one consumer;
• ordered opaque bytes;
• lossless delivery while both endpoints remain live;
• explicit EOF and fault termination;
• bounded buffering and backpressure;
• actor-owned protocol state;
• arena-backed SPSC byte rings;
• local or Iroh-backed transfer chosen internally.

The MVP does not support:

• fanout or SPMC streams;
• multiple writers;
• topics or best-effort delivery;
• replay or resume;
• reconnecting an existing stream incarnation;
• application records, tensor schemas, JSON, dtype, or shape semantics;
• application-visible actors, nodes, edges, rings, sockets, or arena offsets.

Public semantics

A stream path names one SPSC rendezvous.

DataPath
-> waiting source or sink
-> matched source/sink incarnation
-> active transfer
-> EOF or fault

The stream is opaque ordered bytes. Application framing remains application code.

Each match creates one incarnation. A failed or closed incarnation is never silently resumed or joined to a replacement endpoint.

Writer

async with ctx.data.write_stream(path) as writer:
    await writer.write(data)

• Entering the context resolves and authorizes the path, matches the sink, provisions the transfer, and waits for readiness.
• write() completes only after Swactor has accepted the complete input into its bounded ordered pipeline.
• Backpressure suspends write() rather than dropping bytes.
• A clean context exit publishes EOF exactly once.
• Exceptional exit publishes a fault or abort terminal state.
• Writes after terminal state fail typed.

Reader

reader = await ctx.data.read_stream(path)
chunk = await reader.read()

• Opening resolves and authorizes the path, matches the source, provisions the transfer, and waits for readiness.
• Reads observe source byte order.
• EOF is distinguishable from a zero-length temporary read.
• Source, transport, ring, and session faults become typed stream failures.
• Cancellation releases any held ring view and permits cleanup.

The precise Python read method shape may follow existing binding conventions, but it must not return arena, ring, edge, actor, or transport identities.

Namespace rendezvous

The centralized namespace service from DATA_PLANE_VIRTUAL_BLOB_NAMESPACE_MVP.md gains stream entries without changing blob semantics.

Conceptually:

enum NamespaceEntry {
    Blob(BlobBinding),
    Stream(StreamBinding),
}

enum StreamBinding {
    WaitingSink {
        sink: ActorAddress,
    },
    WaitingSource {
        source: ActorAddress,
    },
    Matched {
        incarnation: u64,
        source: ActorAddress,
        sink: ActorAddress,
        coordinator: ActorAddress,
    },
}

The directory actor serializes registration and matching. A path cannot simultaneously name a blob and a stream.

For the first Myelin flow, orchestration may register the result sink before the job opens its writer. The actor protocol should not require application-managed edge IDs or socket paths.

Existing foundation to retain

crates/data-plane/src/byte_ring.rs already provides the correct stream storage foundation:

• arena-resident SPSC header and data region;
• release/acquire commit and consume cursors;
• generation and producer/consumer role enforcement;
• Data, Eof, and Fault record framing;
• lossless bounded backpressure;
• ordering and wraparound handling;
• corruption containment.

Retain its valid guarantee tests.

Also retain:

• one inherited arena descriptor;
• automatically driven child runtime;
• persistent host and child data-plane session actors;
• one operation actor per public open;
• one host binding/coordinator actor per match;
• existing actor routing and Iroh transport;
• the rule that protocol state lives in actor FSM fields rather than mutex-protected pending maps.

Required byte-ring correction

The current consuming path returns owned bytes through recv_record() -> Vec<u8>. That is not zero-copy.

Add a pinned borrowed record view:

struct PinnedRecordView {
    endpoint_guard: Arc<...>,
    generation: u64,
    kind: RecordKind,
    first_span: ...,
    second_span: Option<...>,
}

A record may wrap around the ring and therefore expose two ordered spans. The view pins the committed range. The consumer cursor advances only after the view is released or explicitly consumed.

Required behavior:

producer commits record
-> consumer receives pinned view
-> transport/Python consumes one or two spans
-> view release publishes consume cursor
-> producer capacity becomes available

Holding a view intentionally applies backpressure.

No hot-path stream operation should assemble a record into Vec<u8> merely to cross the host/child boundary or feed the transport.

Actor notification model

Actor messages replace eventfd wakeups and polling.

Messages indicate that protocol state may have advanced; ring cursors remain authoritative.

Conceptually:

producer commits data
-> ProgressAvailable notification
-> consumer checks committed cursor

consumer advances consume cursor
-> CapacityAvailable notification
-> producer retries pending reservation

Notifications may be coalesced or redundant. Correctness must not depend on wake counts.

Each operation actor owns its pending action in FSM fields. Do not add a shared request-ID map or protocol mutex.

Standard happy path

Sink registration
-> namespace records waiting sink

Python write_stream(path)
-> PyDataPlane delegates to general DataPlane
-> ChildDataPlaneSessionActor spawns WriteStreamOperationActor
-> host session authorizes canonical path
-> namespace matches source operation with waiting sink
-> StreamCoordinatorActor is created

Coordinator
-> resolves persistent endpoint actors to nodes
-> allocates and installs arena SPSC rings
-> establishes local or Iroh transfer
-> waits for source and sink readiness
-> sends Opened to both operation paths

Python writer.write(bytes)
-> reserves ring capacity
-> writes bytes into producer spans
-> release-commits Data record
-> actor notification drives downstream progress
-> transport consumes pinned source record
-> destination producer commits to consumer ring
-> reader consumes bytes in order
-> consume progress releases backpressure

Clean writer exit
-> commits EOF once
-> reader observes all preceding bytes, then EOF
-> coordinator quiesces pumps
-> rings are uninstalled and leases released
-> namespace incarnation terminates

Backpressure

Backpressure is end-to-end and lossless:

slow reader
-> destination ring fills
-> network receive stops advancing
-> source transport stops draining
-> source ring fills
-> writer.write awaits capacity

No layer may drop bytes, overwrite unread bytes, busy-spin, or create an unbounded overflow queue.

A held reader or transport view is part of the backpressure mechanism and must keep the relevant range pinned.

Ordering

For one stream incarnation:

• producer writes have one total order;
• the consumer observes exactly that order;
• EOF follows every successfully accepted data byte;
• a fault terminates the incarnation and no later data is delivered;
• transport chunk boundaries are not application-visible stream boundaries.

The existing Iroh QUIC byte stream already provides remote ordered delivery. Swactor must not implement another TCP-like acknowledgement, retransmission, congestion-control, or reorder protocol.

Terminal behavior

Each incarnation has exactly one terminal result:

EOF
Fault
Cancelled
SessionClosed

Terminal state is sticky. Duplicate or late terminal messages are ignored after the first accepted terminal transition.

A clean writer close produces EOF. Writer failure, transport failure, ring corruption, endpoint death, or exceptional writer exit produces a fault. Cancellation and session shutdown reclaim resources without presenting successful EOF.

Python binding boundary

General functionality belongs in crates/data-plane:

• stream reader and writer types;
• operation actors;
• pinned ring record views;
• ring lifetime guards;
• backpressure and terminal FSMs;
• namespace and transfer protocols.

Python owns only:

• PyStreamReader and PyStreamWriter wrappers;
• Rust-future to awaitable conversion;
• buffer exposure where used;
• async context-manager behavior;
• typed exception conversion.

Python must not manually tick the runtime or poll shared cursors.

Myelin migration

The current result path uses a temporary Unix-domain socket exposed through legacy_output and SWACTOR_DATA_PLANE_OUTPUT.

After actor-driven streams satisfy the acceptance path:

1. Register the Myelin result sink at /runs/<run-id>/results/inference.
2. Let /runs/self/results/inference resolve through JobContext.
3. Route ctx.data.write_stream(...) through the SPSC stream implementation.
4. Remove the Unix listener, socket path, legacy environment variable, connector, and compatibility code.
5. Preserve application-owned JSON/result framing.
6. Rerun the complete Tinygrad CUDA scenario.

Expected result:

CUDA -> [2.75, -8.75]

Failure behavior

Failure	Required behavior
Path absent or unmatched under chosen open policy	Typed path/rendezvous failure or pending operation, never fallback socket
Unauthorized path	Fail before ring or edge provisioning
Duplicate source or sink	Typed namespace conflict
Arena exhaustion	Fail both sides and release partial provisioning
Ring generation mismatch or corruption	Fault incarnation and stop using the ring
Iroh stream fault	Fault both endpoint operations
Writer exception	Publish fault/abort, never clean EOF
Reader or writer cancellation	Release held views and terminate binding
Child or host session death	Stop bindings, quiesce pumps, reclaim leases
EOF	Deliver all prior bytes, then complete reader exactly once

Behavioral invariants

1. One stream incarnation has exactly one producer and one consumer.
2. A live path cannot simultaneously name a blob and a stream.
3. Neither endpoint receives Opened before required rings and transport are ready.
4. Successfully accepted bytes are delivered exactly once and in order while both endpoints remain live.
5. The producer never overwrites unread bytes.
6. The consumer never observes uncommitted bytes.
7. A pinned record prevents its range from being consumed or reused.
8. Backpressure is bounded and lossless.
9. EOF follows all accepted data and appears exactly once.
10. A fault prevents subsequent data or EOF success.
11. Every incarnation reaches exactly one terminal outcome.
12. Every arena lease is released only after pumps and views are quiescent.
13. Wake notifications are hints; ring state is authoritative.
14. Python never observes actor, node, edge, ring, socket, or arena identities.
15. Swactor never interprets application byte schemas.

Implementation sequence

1. Extend the centralized namespace with SPSC stream source/sink registration and atomic matching.
2. Define shared stream session, operation, coordinator, progress, and terminal messages.
3. Implement pinned borrowed byte-ring record views with wraparound spans.
4. Add actor-driven capacity and data-progress notifications.
5. Implement general StreamReader and StreamWriter operation actors in crates/data-plane.
6. Provision local arena rings and Iroh edges through one coordinator FSM.
7. Add thin Python reader/writer wrappers and async context behavior.
8. Add focused tests for ordering, backpressure, wraparound, held views, EOF, faults, cancellation, and lease cleanup.
9. Substitute the SPSC path for Myelin's temporary Unix result bridge.
10. Remove all legacy stream socket and environment plumbing.
11. Run affected Rust and Python suites.
12. Exercise the full Tinygrad CUDA scenario and verify the deterministic result.

Verification

Focused behavioral proof must cover:

• source-first or sink-first rendezvous behavior selected for the MVP;
• concurrent independent SPSC paths;
• exact byte ordering through wraparound;
• writer suspension and resumption under backpressure;
• no consume advancement while a view is pinned;
• EOF after all prior data;
• fault dominance over later progress;
• cancellation and session-failure reclamation;
• no stream payload assembly into Vec<u8> on zero-copy host/child paths;
• absence of the legacy Unix stream bridge after migration;
• the complete Myelin CUDA result path.

Affected suites include:

cargo test -p data-plane
cargo test -p myelin --lib
PYO3_USE_ABI3_FORWARD_COMPATIBILITY=1 cargo check -p python

Python binding tests run after rebuilding the extension from crates/bindings/python.

Final boundary

The remaining stream MVP is intentionally narrow:

One logical path rendezvous-matches one writer with one reader. Swactor provisions ordered bounded SPSC byte movement, uses actor messages for progress, applies lossless backpressure, propagates EOF or faults, and removes the temporary Unix bridge without exposing transport mechanics to applications.