diff --git a/design/routed-link.md b/design/routed-link.md new file mode 100644 index 00000000..e2a15406 --- /dev/null +++ b/design/routed-link.md @@ -0,0 +1,336 @@ +# The routed link: accept/connect transports behind a per-process router + +Status: determination 2026-07-29. This is the layer-A design of record +for transports with no innate substreams (TCP and everything shaped +like it), instantiating the router sketched in +`streaming-wire-deadlock.md` §8.4–8.5 as an in-crate generic adapter, +`rumors::link::routed`. The link contract itself (`src/link.rs`) is +unchanged; this document decides how an accept/connect byte-stream +transport satisfies it. + +## 1. Problem and shape + +A `Link` needs one persistent bidirectional control stream plus up to +`STREAM_COUNT` (17) unidirectional data streams per direction, opened +lazily mid-session, each independently flow-controlled and +half-closeable. QUIC provides all of that natively, one connection per +link. An accept/connect transport provides exactly one primitive: +dial a name, get a byte stream; listen, get byte streams. + +The mapping is one connection per stream — per-stream flow control and +half-close come from the transport itself, which is precisely what the +independence clause asks for — behind a per-process **router**: + +- Every dialed connection opens with a small **connect header** naming + the link it belongs to, by token. +- One router task per process owns the listener, reads headers, and + hands each connection *whole* to its link's bounded queue. After the + header, the connection never touches the router again: a router that + proxied bytes would become a mux and inherit the head-of-line + problem the link contract exists to preclude. +- The first connection of a link (kind `LINK`) *is* the control + stream; later connections (kind `STREAM`) are data streams. + +The adapter is generic over the caller's transport (§2) and lives in +the crate because it is pure logic over the caller's connections: no +sockets, no spawning, no timers, no new dependencies — the crate's +no-runtime promise holds (the router is a future the caller drives). + +## 2. The seam: `Addr`, `Conn`, `Dial`, `Listen` + +```rust +pub trait Addr: Clone + Send + Sync + 'static { + fn encode(&self) -> Vec; // must fit MAX_ADDR_LEN (255) + fn decode(bytes: &[u8]) -> Option; +} +pub trait Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static {} // blanket +pub trait Dial: Clone + Send + Sync + 'static { + type Addr: Addr; + type Conn: Conn; + async fn dial(&self, addr: &Self::Addr) -> io::Result; +} +pub trait Listen: Send + 'static { + type Conn: Conn; + async fn accept(&mut self) -> io::Result; // cancel-safe +} +``` + +`Addr` is a name in the *caller's* namespace — a socket address, a +Unix path, an overlay peer id. The router ferries it (opaquely, in the +`LINK` header) and never interprets it; only the caller's `Dial` does. +This is what unbinds the adapter from IP. An `Addr` impl for +`SocketAddr` ships with the module (18 bytes: v4-mapped 16-byte IP + +big-endian port) so socket transports don't each reinvent it. + +Two `Conn` requirements the bounds cannot express, documented as +contract on the trait: + +- `poll_write` makes bytes peer-visible without an explicit flush (the + session and the conformance probes never flush before awaiting a + peer's read); +- dropping the connection delivers already-written bytes and then EOF + (drop is the transport half-close the session relies on). + +`TcpStream` satisfies both; a buffering wrapper (a `BufWriter`, some +TLS rigs) does not, and fails the independence probe or truncates +final bytes. Security is the caller's layer, per the link contract's +security division: `Dial`/`Listen` are assumed pre-authenticated, and +the token routes — it does not authenticate. + +## 3. The wire + +All multi-byte integers big-endian. One header per dialed connection, +written by the dialer before any protocol byte; one ACK byte in reply +on `LINK` connections only. + +``` +magic 10 b"ROUTEDLINK" +version 1 0x01 +kind 1 0x01 LINK | 0x02 STREAM +token 16 the link's identity, minted at link establishment + +LINK only: +addr_len 1 length of the dialer's advertised name (1..=255) +addr … Addr::encode() of the name peers dial back + +ACK 1 0x01, accept-side router → dialer, LINK connections only +``` + +Per-field rationale: + +- **magic** denotes the thing itself — a routed-link connect header — + so a connection misrouted across protocols fails at the first read + with a precise error. It is deliberately not an abbreviation of the + crate name: the session preamble already owns `b"RUMORS"`, and two + near-identical magics would blunt exactly the misroute diagnosis a + magic exists for. +- **version** is the compatibility door. A future header kind that + needs more context (for instance a reusable-lease connection with + private framing) arrives as a new version or kind, never a mutation + of this one. +- **token** is 16 random bytes: wide enough that collision handling + and coordination are non-problems; minted per *link*, not per + session — sessions on a link are serialized and already disambiguate + themselves by the epoch byte in the session's own 2-byte stream + label. The router routes on the token alone. +- **addr** rides only on `LINK` because only link establishment needs + it: the accept side must dial back for its own outgoing data + streams, and the accepted connection's source is not the dialer's + listener (ephemeral ports, NAT) — only the dialer knows its own + reachable name. Length-prefixed and capped at 255 so the router's + header read is bounded by construction. +- **ACK** closes the registration race, §5. + +The fixed 28-byte prefix (magic through token) is one `read_exact`; +`LINK` reads `addr_len` and then exactly that many more bytes. The +maximum header is 284 bytes. + +The header deliberately carries **no epoch and no stream index**, +deviating from the `(link token, epoch, stream index)` shape sketched +in `streaming-wire-deadlock.md` §8.4. That sketch predates the +contract's anonymous-streams clause: `Connector::connect()` takes no +arguments, so the adapter *cannot* learn the epoch or index — the +session writes them itself as the stream's first payload bytes and +validates them on the accepting side. The router has no routing use +for them, and duplicating the session's label would mint a consistency +obligation between two copies of one fact with no checker. + +## 4. The router + +One drive future per endpoint, returned by the constructor and driven +by the caller (spawn it, select over it — the adapter never spawns). +Its loop selects over `Listen::accept` and a `FuturesUnordered` of +per-connection futures; each accepted connection becomes one such +future, so the loop itself never awaits any per-connection I/O. + +Each per-connection future owns its connection and does, bounded: + +- read the header (`read_exact` of the fixed prefix, then the + kind-determined remainder); +- `LINK`: register the token in the shared table → `try_reserve` an + incoming-links permit (no permit: kill the connection, unregister) + → write the ACK byte → split the connection into the control halves + → assemble the accept-side `Link` (its connector dials the header's + decoded `Addr`) → deliver `(LinkInfo, Link)` on the permit; +- `STREAM`: `try_send` the whole connection onto the token's queue. + Unknown token, or a queue that is gone: drop the connection. + +The §8.5 discipline, instantiated: + +- **Never await a per-destination queue.** Per-link queues are bounded + mpsc channels of capacity `STREAM_COUNT` (the control connection + never routes through the queue — it *is* the `LINK` connection, so + 17 suffices; sessions are serialized, so an honest peer can never + have more in flight). Delivery is `try_send`, never `send`. +- **Overflow kills the link, not the connection.** A full queue proves + peer misbehavior or a local bug. Dropping just the overflowing + connection would silently lose a delivery on a link that still looks + healthy — the accepting session would wait forever for a labeled + stream that never arrives, a hang where a poison was owed. Killing + the link (unregister the token, drop the queue's sender so the + acceptor's next `accept` errors) turns it into ordinary transport + failure: session errs, link poisons, application re-links. +- **Sockets, not bytes.** After the header the router never touches + the connection; flow control stays end-to-end per stream. +- **Header reads are bounded in size by construction and in count by + eviction.** §8.4 sketched header reads as small spawned tasks with a + time bound; the crate promises no spawning and no timers, so the + determination is: per-connection futures live in the drive future's + own `FuturesUnordered`, and the set is *count-bounded* — beyond the + configured cap, the oldest pending header read is aborted (its + connection drops). A connection that stalls mid-header therefore + occupies one slot until fresh arrivals displace it, and can never + park the loop. Callers who want wall-clock bounds put them where the + clock lives: a deadline-wrapping `Listen`/`Conn` in their transport + layer. The cap is hygiene, not a security boundary — the transport + is pre-authenticated, and hostile-peer economics are off-model. + +Token-table entries are removed eagerly: the link's acceptor half +holds a guard whose drop unregisters the token, so an application that +drops a link (poisoned or merely finished) revokes its routing at that +moment, whether or not the router ever hears another byte. Unknown +tokens are dropped on sight; after the ACK handshake below, the only +honest sources of one are a link the local application already +discarded, or dials chasing a stale address — both resolve as +transport failure on the dialer's side, which is the self-healing +path (poison, re-link). + +The drive future resolves `Err` when `Listen::accept` fails, and the +failure is the caller's to interpret (their `Listen` owns retry policy +for transient conditions; the adapter treats what it is given as +fatal). It never resolves otherwise: dropping the future is the +shutdown, and an undriven or dropped router routes nothing — sessions +on surviving links stall into the caller's timeout. + +## 5. Link establishment + +`Endpoint::link(peer)`: + +1. mint a token, build the link's queue, and **register the token + locally, before anything touches the wire** — the peer's reverse + dials can only follow its read of our header, so + register-before-write makes them happens-after registration; +2. dial `peer`, write the `LINK` header (token + our advertised name); +3. await the ACK byte; EOF or a non-ACK byte is a crisp rejection + (unregister and fail); +4. split the connection into the control halves and assemble the + `Link`. + +The ACK exists because of a real race on the accept side. Header reads +complete in any order (they are concurrent futures), and accept/connect +transports guarantee nothing about cross-connection ordering — so +without the ACK, a `STREAM` dial racing behind a `LINK` dial could +reach the peer's router before the `LINK` header finishes processing, +and die as an unknown token even though both ends are honest. The +protocol's own traffic happens to serialize past the window (the first +data stream follows speaker election, which follows control-stream +bytes), but the contract does not: `connect()` may legally be called +before any control byte, and the conformance suite does exactly that. +One ACK round trip, paid once per link (not per session, not per +stream), makes registration on both routers happens-before either +end's first possible `connect()` — the dialer's by rule 1, the +acceptor's because the ACK is written after registration. + +Both ends may `link()` toward each other concurrently; the result is +two independent links, each with its own token and control stream. +The adapter does not deduplicate — which link (or links) to gossip on +is application policy, as it is for every other transport. + +## 6. Contract mapping + +| Clause (src/link.rs) | How the routed link satisfies it | +| --- | --- | +| Control duplex | The control stream is one transport connection, split; both directions are the transport's own full duplex. | +| Independence | One connection per stream; flow control and buffering are per-connection in the transport. Nothing is shared: no window pool, no mux. | +| Flow control | Receiver-paced by the transport per connection, at whatever capacity the caller's transport grants (the contract needs only "positive"). | +| Concurrency | `connect()` dials a fresh connection; opens share no limiter, so no open ever serializes behind another stream's progress. The 17-per-direction complement is 35 connections worst case, well inside any listener backlog the caller configures. | +| Half-close | `Tx` *is* the connection; dropping it closes it, delivering written bytes then EOF (the documented `Conn` requirement). The control connection is held by the link itself and outlives the session's data connections structurally. | +| Cancellation | `Acceptor::accept` is a bounded-mpsc `recv()`, cancel-safe by construction: an undelivered connection stays queued for the next call. | +| Anonymous streams | The acceptor yields connections in arrival order with no routing logic; the session's own label does the pairing. | +| Security division | Inherited whole: authentication, integrity, confidentiality, freshness are the caller's transport (`Dial`/`Listen`), below this layer. The token is routing state, not a credential. | + +Accepted failure modes, stated rather than hidden: + +- A dial that reaches the wrong process (stale address, restarted + peer) dies at the header or as an unknown token; the dialing session + poisons; the application re-links. Self-healing, at session + granularity. +- A silently dead path (no reset in flight) hangs a `connect()` or a + write until the caller's session timeout cancels the session — the + same backstop every transport relies on; keepalive belongs in the + caller's `Dial`. +- Every lazy stream open pays one dial (connection setup + 28-byte + header). That is the compatibility trade named in + `streaming-wire-deadlock.md` §8.4: this instantiation exists for + deployments that cannot take QUIC, not to win latency contests. + +## 7. Decision records + +- **DECIDED (2026-07-29): in-crate module, not a sibling crate.** The + adapter is generic over `Dial`/`Listen` and needs zero new + dependencies; only its *instantiations* (TCP, TLS) touch a network + stack, and those stay caller-side (the crate's tests instantiate + over `tokio::net` from dev-dependencies). The sibling-crate ruling + in `streaming-wire-deadlock.md` §8.9 governs network bindings; an + adapter with no network dependency is core-eligible logic. +- **DECIDED (2026-07-29): caller-driven router future; count-bounded + header reads.** The crate promises no runtime, no spawning, no + timers; the router is therefore a future the caller drives, and + pending header reads are bounded by count with oldest-first + eviction instead of by wall clock. Time bounds live in the caller's + `Listen`/`Conn` wrappers, next to the runtime that can measure them. +- **DECIDED (2026-07-29): per-link token, token-only routing.** + Granularity: sessions are serialized per link and self-labeled + (epoch byte), so routing state is per-link. Content: no epoch or + stream index in the header — `connect()` cannot supply them, the + router cannot use them, and duplicating the session label creates an + unchecked consistency obligation. Supersedes the header sketched in + §8.4, which predates the contract's anonymous-streams clause. +- **DECIDED (2026-07-29): one-RTT LINK ACK.** Registration on both + routers must happen-before either end's first `connect()`; the + contract permits a `connect()` before any control byte, so control + traffic cannot be the fence. Cost: one round trip per link + establishment, amortized over every session the link carries. +- **DECIDED (2026-07-29): single-use connections; no reuse at stream + boundaries.** Reuse (the pooled-lease idea in §8.3) requires layer-A + private framing to delimit stream end without a transport close, + forfeiting the one thing connection-per-stream gets free: EOF *is* + the half-close, kernel-enforced, no trailing-byte ambiguity. It also + needs clean/dirty lease tracking and abort plumbing. The genuine + future case is per-connection security-handshake cost; the header's + version/kind fields are the door, and the change would be a new + kind, never a mutation. +- **DECIDED (2026-07-29): no connection pooling in v1; the `Dial` + trait is the pooling seam.** A warm-dial supply (pre-established + connections to hide dial latency on lazy opens) composes behind + `Dial` without touching header, router, or link code, and should be + motivated by latency measurements, not anticipation. qorb 0.4.1 was + evaluated as the supply: its pools are homogeneous (`Pool::claim()` + cannot target a backend, so per-peer means one pool per peer), its + `claim::Handle` cannot detach a connection (Deref/DerefMut/Drop + only), and single-use claims fight the recycle machinery (an + `on_recycle` error is the only way to retire a spent connection, + which reads as perpetual backend failure in its stats and + rebalancer). A qorb-backed `Dial` would hold each claim only long + enough to extract the connection (`Option::take` inside the pooled + type), never across a stream's lifetime — holding claims for live + streams re-couples stream opens to pool capacity, which is exactly + the cross-stream wait the concurrency clause forbids (and qorb's + default `max_slots = 16` is below one session's worst-case 18). + Recorded here so the future evaluation starts from these facts. +- **DECIDED (2026-07-29): no peer discovery.** `link()` takes an + explicit `Addr`. Discovery composes above the adapter (any resolver + feeding addresses in the caller's namespace) and below it (a `Dial` + that resolves names at dial time); building either in would bind the + adapter to a discovery model the seam deliberately leaves open. +- **REJECTED: deriving the dial-back address from the accepted + connection.** The accepted connection's source is an ephemeral port + (and possibly a NAT); only the dialer knows its reachable name, so + the `LINK` header carries it, opaquely, in the caller's namespace. +- **REJECTED: per-session listeners (the `tests/common/tcp.rs` + shape) as the production story.** One listener per session avoids + the router entirely and is the right trade for a test harness, but + a process gossiping with P peers over S serialized sessions would + churn P listeners per round and advertise a moving target; the + per-process router with a stable advertised name is the deployable + shape, as `tests/common/tcp.rs`'s own docs state. diff --git a/src/link.rs b/src/link.rs index 6252073b..12e591dc 100644 --- a/src/link.rs +++ b/src/link.rs @@ -132,12 +132,15 @@ //! //! [`memory`] builds the in-memory instantiation, both ends at once, for //! tests and in-process pairings; it is also the reference implementation -//! of the contract. No network instantiation ships with the crate (the -//! core stays free of network dependencies), so a real-network deployment -//! implements the two traits against its own transport (QUIC connections -//! mapping streams 1:1, or TCP with one connection per stream behind a -//! routing listener, are the natural shapes) and validates the result -//! with the `conformance` feature's link suite. +//! of the contract. A transport with native substreams implements the two +//! traits directly (QUIC connections map streams 1:1); an accept/connect +//! transport with no innate substreams (TCP and everything shaped like +//! it) gets the [`routed`] adapter, which builds the +//! one-connection-per-stream shape behind a per-process router from a +//! caller-supplied dial/listen pair. The core still ships no network +//! code — the adapter is generic, and its instantiations live with the +//! caller — so either way a deployment validates its transport with the +//! `conformance` feature's link suite. use std::future::Future; use std::io; @@ -574,6 +577,7 @@ impl Acceptor for MemoryAcceptor { } pub(crate) mod erased; +pub mod routed; #[cfg(test)] mod tests; diff --git a/src/link/routed.rs b/src/link/routed.rs new file mode 100644 index 00000000..cf0d9a7d --- /dev/null +++ b/src/link/routed.rs @@ -0,0 +1,242 @@ +//! [`Link`]s over accept/connect transports: one connection per stream, +//! routed by a per-process listener. +//! +//! A [`Link`] wants one persistent control stream plus lazily opened, +//! independently flow-controlled data streams. Transports with native +//! substreams (QUIC) map onto that directly; an accept/connect transport +//! (TCP and everything shaped like it) offers exactly one primitive: +//! dial a name, get a byte stream. This module is the adapter between +//! the two shapes, generic over the transport: the caller supplies a +//! [`Dial`]/[`Listen`] pair, and the adapter maps **every link stream to +//! its own connection**, so per-stream flow control and half-close are +//! the transport's own, which is precisely what the link contract's +//! independence clause asks for. +//! +//! What makes that routable is a small connect header. Every dialed +//! connection opens by naming the link it belongs to (a 16-byte random +//! *token* minted at link establishment), and one **router** per +//! [`Endpoint`] owns the listener: it reads each arriving connection's +//! header and hands the connection — whole, never its bytes — to that +//! link's bounded queue, where the link's [`Acceptor`] collects it. A +//! link's first connection carries the control stream; each later one +//! is a single unidirectional data stream, closed by dropping its write +//! end. +//! +//! # Driving the router +//! +//! The crate requires no runtime, so the router is a future the caller +//! drives: [`Endpoint::new`] returns it alongside the endpoint, and the +//! caller spawns it (or selects over it) for the endpoint's lifetime. +//! An undriven router accepts no links and routes no streams — sessions +//! on existing links stall until the caller's timeout cancels them. The +//! future resolves only on listener failure ([`Listen::accept`] +//! erroring is fatal to the endpoint); dropping it is the shutdown. +//! +//! # Establishing links +//! +//! [`Endpoint::link`] dials the peer's router, sends the link's token +//! together with this endpoint's own advertised name, and waits for the +//! peer router's one-byte acknowledgement; the peer's application +//! receives the other end from [`Incoming::accept`]. The advertised +//! name rides along because the accept side must dial back for its own +//! outgoing data streams, and only the dialer knows the name it is +//! reachable at (the accepted connection's source is an ephemeral +//! port). The acknowledgement makes token registration on both routers +//! happen strictly before either end can open a data stream, so a +//! stream dial can never race ahead of its own link. +//! +//! Both ends may [`link`](Endpoint::link) toward each other +//! concurrently; the result is two independent links. The adapter does +//! not deduplicate: which links to hold and gossip on is application +//! policy, as on every other transport. +//! +//! # Failure modes +//! +//! - A connection bearing an unknown token (a link the application +//! already discarded, a dial chasing a stale name) is dropped; the +//! dialing side's session fails as transport failure, poisoning its +//! link, and the application re-links. Failures heal at session +//! granularity. +//! - A link whose stream queue overflows is evicted wholesale: an +//! honest peer never has more than a session's worth of streams in +//! flight, so overflow proves misbehavior (or a local bug), and +//! evicting the link turns it into an ordinary transport failure +//! instead of a silent lost delivery. The router never blocks on a +//! full queue. +//! - A silently dead path hangs a stream open or write until the +//! caller's session timeout cancels the session, the same backstop +//! every transport relies on. Liveness probing (keepalive) belongs in +//! the caller's [`Dial`]. +//! - Every lazy stream open pays one dial. This adapter is the +//! compatibility shape, not the latency shape; a transport with +//! native substreams avoids the per-stream dial entirely. +//! +//! # What the transport must provide +//! +//! Two obligations on [`Dial::Conn`] reach beyond its trait bounds, and +//! the conformance suite observes both (see [`Conn`]): +//! +//! - bytes accepted by `poll_write` become visible to the peer without +//! an explicit flush; +//! - dropping a connection delivers already-written bytes and then +//! end-of-stream to the peer. +//! +//! Security is the transport's, per the [link contract's security +//! division](crate::link#what-securing-the-transport-means): the +//! adapter assumes `Dial`/`Listen` connections are authenticated, +//! integrity-protected, and fresh. The token routes connections; it is +//! not a credential, and the router's violation handling (dropping +//! malformed or unknown arrivals) is a conformance bug detector, not a +//! security boundary. +//! +//! # Example: a TCP instantiation +//! +//! ``` +//! # tokio::runtime::Builder::new_current_thread().enable_io().build().unwrap().block_on(async { +//! use std::io; +//! use std::net::SocketAddr; +//! +//! use rumors::link::routed::{Config, Dial, Endpoint, Listen}; +//! use tokio::net::{TcpListener, TcpStream}; +//! +//! /// Dials one TCP connection per link stream. +//! #[derive(Clone)] +//! struct TcpDial; +//! +//! impl Dial for TcpDial { +//! type Addr = SocketAddr; +//! type Conn = TcpStream; +//! +//! async fn dial(&self, addr: &SocketAddr) -> io::Result { +//! TcpStream::connect(*addr).await +//! } +//! } +//! +//! /// Yields this process's inbound connections to the router. +//! struct TcpListen(TcpListener); +//! +//! impl Listen for TcpListen { +//! type Conn = TcpStream; +//! +//! async fn accept(&mut self) -> io::Result { +//! Ok(self.0.accept().await?.0) +//! } +//! } +//! +//! // Two endpoints in one process, for the example's sake; in a +//! // deployment each process runs one. +//! let a_listener = TcpListener::bind("127.0.0.1:0").await?; +//! let a_addr = a_listener.local_addr()?; +//! let (_a, mut a_incoming, a_router) = +//! Endpoint::new(TcpListen(a_listener), a_addr, TcpDial, Config::default()); +//! tokio::spawn(a_router); +//! +//! let b_listener = TcpListener::bind("127.0.0.1:0").await?; +//! let b_addr = b_listener.local_addr()?; +//! let (b, _b_incoming, b_router) = +//! Endpoint::new(TcpListen(b_listener), b_addr, TcpDial, Config::default()); +//! tokio::spawn(b_router); +//! +//! // b initiates; a's application receives the other end. +//! let (linked, arrival) = tokio::join!(b.link(a_addr), a_incoming.accept()); +//! let mut link_at_b = linked.expect("a's router accepts the link"); +//! let (info, mut link_at_a) = arrival.expect("router is live"); +//! assert_eq!(info.peer, b_addr); +//! // Each side now runs sessions on its link: +//! // `peer.rumors().gossip(&mut link_at_b)`, etc. +//! # let _ = (&mut link_at_a, &mut link_at_b); +//! # io::Result::Ok(()) }).unwrap(); +//! ``` +//! +//! Validate any instantiation with the `conformance` feature's link +//! suite, exactly as for a hand-built [`Link`]. + +use std::future::Future; +use std::io; + +use tokio::io::{AsyncRead, AsyncWrite}; + +use super::{Acceptor, Connector, Link}; + +mod endpoint; +mod header; +mod router; +mod stream; + +#[cfg(test)] +mod tests; + +pub use endpoint::{Config, Endpoint, Incoming, LinkError, LinkInfo, RoutedLink}; +pub use header::{Addr, MAX_ADDR_LEN, Token}; +pub use stream::{StreamAcceptor, StreamConnector}; + +/// A byte-stream connection the adapter can route: one per link stream. +/// +/// Blanket-implemented for every type with the bounds; the real +/// contract is two obligations the bounds cannot express, both load- +/// bearing for the link contract: +/// +/// - **No hidden write buffering.** Bytes accepted by `poll_write` +/// must become visible to the peer without an explicit flush: the +/// session (and the conformance suite) awaits peer reactions to +/// unflushed writes. A connection wrapped in a write buffer that +/// holds bytes until it fills stalls the first such exchange. +/// - **Drop is half-close.** Dropping the connection must deliver all +/// already-written bytes and then end-of-stream to the peer: the +/// session ends every data stream by dropping its write end, and the +/// peer reads to end-of-stream. A drop that discards queued bytes, +/// or never signals the peer, breaks stream teardown. +/// +/// `tokio::net::TcpStream` satisfies both, as does anything else whose +/// writes land in the transport as they are accepted. +pub trait Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static {} + +impl Conn for T {} + +/// Opens outgoing connections to peers' routers, by name. +/// +/// This is the adapter's outgoing seam, and the place transport policy +/// lives: socket options, liveness probing (keepalive), security +/// wrapping, and dial timeouts are all the implementation's own. Dials +/// arrive concurrently through clones (one per in-flight stream open), +/// so implementations hold shared state behind cheap handles. +/// +/// # Errors +/// +/// A dial failure fails the stream open (and with it the session) +/// as transport failure; the adapter never retries. +pub trait Dial: Clone + Send + Sync + 'static { + /// The name this transport dials by; see [`Addr`]. + type Addr: Addr; + + /// The connection a dial yields. + type Conn: Conn; + + /// Open one connection to the router reachable at `addr`. + fn dial(&self, addr: &Self::Addr) -> impl Future> + Send; +} + +/// Yields inbound connections to an endpoint's router. +/// +/// The router is this listener's only consumer, selecting over +/// [`accept`](Self::accept) in its loop. +/// +/// # Errors +/// +/// An accept failure is fatal to the endpoint: the router resolves +/// with the error and stops routing. A transport whose accept can fail +/// transiently (resource exhaustion, say) handles the retry inside its +/// own `accept`. +/// +/// # Cancel safety +/// +/// The router holds `accept` in a `select!` loop, so the future is +/// dropped and re-created continuously; a connection mid-accept must +/// not be lost to the drop. +pub trait Listen: Send + 'static { + /// The connection an accept yields. + type Conn: Conn; + + /// Accept the next inbound connection. + fn accept(&mut self) -> impl Future> + Send; +} diff --git a/src/link/routed/endpoint.rs b/src/link/routed/endpoint.rs new file mode 100644 index 00000000..a79b6d82 --- /dev/null +++ b/src/link/routed/endpoint.rs @@ -0,0 +1,243 @@ +//! The endpoint: one process's routed-link identity. + +use std::collections::HashMap; +use std::io; +use std::sync::{Arc, Mutex}; + +use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf, split}; +use tokio::sync::mpsc; + +use super::header::{self, Addr, Token}; +use super::router::{self, Table}; +use super::stream::{StreamAcceptor, StreamConnector}; +use super::{Dial, Link, Listen}; + +/// The [`Link`] type the adapter builds over dialer `D`. +/// +/// Both ends of a routed link have this type: the control stream is +/// the split establishment connection, and the stream supply dials +/// [`D::Conn`](Dial::Conn) connections one per stream. +pub type RoutedLink = Link< + ReadHalf<::Conn>, + WriteHalf<::Conn>, + StreamConnector, + StreamAcceptor<::Conn>, +>; + +/// What [`Incoming`] yields per peer-established link. +pub(super) type Arrival = (LinkInfo<::Addr>, RoutedLink); + +/// Capacity knobs of an endpoint's router; [`Config::default`] suits +/// most deployments. +#[derive(Clone, Copy, Debug)] +pub struct Config { + /// Peer-established links the router holds while the application + /// catches up on [`Incoming::accept`]. + /// + /// When the backlog is full, further establishment attempts are + /// rejected (the dialer's [`Endpoint::link`] fails) rather than + /// queued without bound; an application that accepts promptly + /// never fills it. + pub incoming_backlog: usize, + /// Inbound connections the router will hold mid-header before it + /// starts evicting the oldest. + /// + /// The bound is hygiene against connections that stall inside + /// their connect header (the router has no clock, so it evicts by + /// count, oldest first); wall-clock deadlines belong in the + /// caller's [`Listen`] wrapper. Anything past the burst of + /// simultaneous dials the deployment expects is enough. + pub pending_headers: usize, +} + +/// Default [`Config::incoming_backlog`]: a burst of simultaneous +/// peers, not a queueing tier. +const DEFAULT_INCOMING_BACKLOG: usize = 16; + +/// Default [`Config::pending_headers`]: comfortably past a full +/// session complement of simultaneous dials from several peers. +const DEFAULT_PENDING_HEADERS: usize = 64; + +impl Default for Config { + fn default() -> Self { + Config { + incoming_backlog: DEFAULT_INCOMING_BACKLOG, + pending_headers: DEFAULT_PENDING_HEADERS, + } + } +} + +/// How establishing a link can fail; see [`Endpoint::link`]. +#[derive(Debug, thiserror::Error)] +pub enum LinkError { + /// The transport failed under the establishment: the dial itself, + /// or reading and writing the establishment connection. + #[error("link establishment transport failure")] + Io(#[from] io::Error), + /// The peer's router answered but did not accept the link: its + /// application is not accepting links, or the listener is not a + /// routed-link router at all. + #[error("the peer's router rejected the link")] + Rejected, +} + +/// The identity of a peer-established link, from [`Incoming::accept`]. +#[derive(Clone, Debug)] +pub struct LinkInfo { + /// The establishing peer's advertised name: where this link's + /// outgoing data streams dial, and the name to re-link with if + /// the link poisons. + pub peer: A, + /// The link's routing identity, unique per link on this endpoint. + pub token: Token, +} + +/// One process's routed-link identity: establishes outbound links and, +/// through the router it is constructed with, terminates inbound ones. +/// +/// Cloning is cheap and clones are interchangeable handles onto the +/// same endpoint. See the [module docs](super) for the architecture +/// and an instantiation example. +pub struct Endpoint { + inner: Arc>, +} + +impl Clone for Endpoint { + fn clone(&self) -> Self { + Endpoint { + inner: Arc::clone(&self.inner), + } + } +} + +/// State shared by an endpoint's clones and its router. +struct Inner { + table: Table, + dial: D, + /// The endpoint's advertised name, pre-encoded (validated at + /// construction) for the `LINK` headers this endpoint writes. + advertised: Vec, +} + +impl Endpoint { + /// Build an endpoint around its transport. + /// + /// `advertised` is the name peers dial this endpoint's `listen` at + /// — it is caller-supplied because it cannot be derived (a bound + /// address may be unroutable from outside; only the deployment + /// knows the reachable name). Returns the endpoint, the stream of + /// peer-established links, and the router future, which the caller + /// must drive for the endpoint's lifetime (see the [module + /// docs](super#driving-the-router)). + /// + /// # Panics + /// + /// If `advertised` encodes to nothing or to more than + /// [`MAX_ADDR_LEN`](super::MAX_ADDR_LEN) bytes, or if either + /// [`Config`] bound is zero: each is a configuration bug caught at + /// construction rather than at the first link. + pub fn new( + listen: impl Listen, + advertised: D::Addr, + dial: D, + config: Config, + ) -> ( + Self, + Incoming, + impl Future> + Send + 'static, + ) { + let encoded = advertised.encode(); + assert!( + (1..=header::MAX_ADDR_LEN).contains(&encoded.len()), + "advertised name must encode to 1..=255 bytes" + ); + assert!( + config.incoming_backlog > 0, + "incoming backlog must admit a link" + ); + assert!( + config.pending_headers > 0, + "pending headers must admit a connection" + ); + let table: Table = Arc::new(Mutex::new(HashMap::new())); + let (arrivals, incoming) = mpsc::channel(config.incoming_backlog); + let endpoint = Endpoint { + inner: Arc::new(Inner { + table: table.clone(), + dial: dial.clone(), + advertised: encoded, + }), + }; + let router = router::drive(listen, dial, table, arrivals, config.pending_headers); + (endpoint, Incoming { links: incoming }, router) + } + + /// Establish one link to the peer reachable at `peer`. + /// + /// One round trip: dial the peer's router, announce the link (its + /// fresh token, this endpoint's advertised name), and wait for the + /// acknowledgement that the peer's end is registered and on its + /// way to the peer's application. The connection then carries the + /// link's control stream. + /// + /// Concurrent calls (including both ends linking toward each + /// other) establish that many independent links; deduplication is + /// application policy. + /// + /// # Errors + /// + /// [`LinkError::Io`] for transport failure, [`LinkError::Rejected`] + /// when the peer answered without acknowledging (its application's + /// backlog is full, or the listener does not speak this wire). + /// Either way no link exists and nothing needs cleaning up; retry + /// policy is the caller's. + pub async fn link(&self, peer: D::Addr) -> Result, LinkError> { + // Register before the header goes out: the peer's reverse + // dials can only follow its read of the header, so they always + // find the token routable. + let (token, registration, streams) = router::register(&self.inner.table); + let mut conn = self.inner.dial.dial(&peer).await?; + conn.write_all(&header::link_header(&token, &self.inner.advertised)) + .await?; + let mut ack = [0; 1]; + match conn.read_exact(&mut ack).await { + Ok(_) if ack[0] == header::ACK => {} + // A clean close or a non-acknowledgement byte is the + // peer's router declining; transport trouble stays Io. + Ok(_) => return Err(LinkError::Rejected), + Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => { + return Err(LinkError::Rejected); + } + Err(error) => return Err(LinkError::Io(error)), + } + let (control_read, control_write) = split(conn); + Ok(Link::new( + control_read, + control_write, + StreamConnector::new(self.inner.dial.clone(), peer, token), + StreamAcceptor::new(streams, registration), + )) + } +} + +/// The links peers establish toward an endpoint, in arrival order. +/// +/// Returned by [`Endpoint::new`]; there is exactly one per endpoint, +/// and dropping it makes the router reject all further establishment +/// attempts (existing links keep routing). +pub struct Incoming { + links: mpsc::Receiver>, +} + +impl Incoming { + /// Receive the next peer-established link, or `None` once the + /// router has stopped (its future resolved or was dropped). + /// + /// # Cancel safety + /// + /// Dropping the future loses nothing: an undelivered link stays + /// queued for the next call. + pub async fn accept(&mut self) -> Option> { + self.links.recv().await + } +} diff --git a/src/link/routed/header.rs b/src/link/routed/header.rs new file mode 100644 index 00000000..64cc74ce --- /dev/null +++ b/src/link/routed/header.rs @@ -0,0 +1,255 @@ +//! The connect header: the bytes that route a connection to its link. +//! +//! Every connection dialed by this module opens with one header, written +//! by the dialer before any protocol byte, and read by the accepting +//! router as its only I/O on the connection. Two kinds exist: `LINK` +//! establishes a link (the connection becomes the control stream, and +//! the header carries the dialer's advertised name for reverse dials), +//! `STREAM` attaches one data stream to an existing link. `LINK` +//! connections receive a one-byte acknowledgement; see the [module +//! docs](super) for the race it closes. +//! +//! Layout, integers big-endian, lengths fixed per kind: +//! +//! ```text +//! magic 10 b"ROUTEDLINK" +//! version 1 0x01 +//! kind 1 0x01 LINK | 0x02 STREAM +//! token 16 the link's identity +//! LINK only: +//! addr_len 1 advertised-name length, 1..=MAX_ADDR_LEN +//! addr … Addr::encode() of the dialer's advertised name +//! ``` +//! +//! The header carries no epoch and no stream index: the session labels +//! its streams itself (the label is the stream's first payload bytes, +//! after this header), the router routes on the token alone, and a +//! second copy of the session's label would be a consistency obligation +//! with no checker. The version byte is the compatibility door: any +//! future shape (a reusable-lease kind, say) arrives as a new version +//! or kind, never a mutation of these. + +use std::fmt; +use std::io; +use std::net::{IpAddr, Ipv6Addr, SocketAddr}; + +use tokio::io::{AsyncRead, AsyncReadExt}; + +/// The header's opening magic. +/// +/// Named for what the bytes introduce (a routed-link connect header), +/// deliberately unrelated to the session preamble's own magic: the two +/// travel adjacent on the wire (a control connection carries this +/// header, then the session preamble), and distinct magics turn a +/// misrouted or misaligned connection into a precise first-read error. +const MAGIC: &[u8; 10] = b"ROUTEDLINK"; + +/// The one wire version this module speaks. +const VERSION: u8 = 1; + +/// Kind byte: this connection establishes a link and becomes its +/// control stream. +const KIND_LINK: u8 = 1; + +/// Kind byte: this connection is one data stream of an existing link. +const KIND_STREAM: u8 = 2; + +/// The fixed-width header prefix every kind shares: magic, version, +/// kind, token. +pub(super) const PREFIX_LEN: usize = MAGIC.len() + 2 + TOKEN_LEN; + +/// The acknowledgement byte a router writes back on a `LINK` +/// connection once the link's token is registered and its end of the +/// link is on its way to the application. +pub(super) const ACK: u8 = 1; + +/// Bytes in a [`Token`]. +const TOKEN_LEN: usize = 16; + +/// Longest advertised name a `LINK` header carries. +/// +/// [`Addr::encode`] must fit within it; the one-byte length prefix +/// makes the bound structural, and keeps the router's header read +/// bounded by construction. +pub const MAX_ADDR_LEN: usize = u8::MAX as usize; + +/// One link's routing identity: 16 random bytes minted at +/// establishment. +/// +/// Both routers key the link's connection queue by its token, and every +/// data-stream dial quotes it. The width makes collision handling a +/// non-problem; the token is routing state, never a credential (the +/// transport below is authenticated, per the [module docs](super)). +/// Tokens are minted by [`Endpoint::link`](super::Endpoint::link) and +/// observed through [`LinkInfo`](super::LinkInfo); they cannot be +/// constructed. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct Token([u8; TOKEN_LEN]); + +impl Token { + /// Mint a fresh random token. + pub(super) fn new() -> Self { + Token(rand::random()) + } +} + +impl fmt::Debug for Token { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "Token({})", hex::encode(self.0)) + } +} + +/// Names a dialable peer in the caller's transport namespace. +/// +/// An address is whatever the caller's [`Dial`](super::Dial) dials by: +/// a socket address, a filesystem path, an overlay peer id. The adapter +/// ferries one address per link — the dialer's advertised name, inside +/// the `LINK` header — and never interprets it; encoding exists so the +/// name survives the trip. +/// +/// `encode` must produce between 1 and [`MAX_ADDR_LEN`] bytes (an +/// endpoint's own advertised name is checked at construction), and +/// `decode` must invert it: `decode(&encode(a))` yields an address that +/// dials the same peer as `a`. `decode` returns `None` for bytes that +/// name nothing; the router drops the connection that carried them. +pub trait Addr: Clone + Send + Sync + 'static { + /// Encode this name for the wire. + fn encode(&self) -> Vec; + + /// Decode a wire name, or `None` if the bytes name nothing. + fn decode(bytes: &[u8]) -> Option; +} + +/// Socket-address bytes: a 16-byte IP (IPv4 mapped into IPv6) and a +/// big-endian port. +const SOCKET_ADDR_LEN: usize = 18; + +/// The stock instantiation for IP transports: 18 bytes, IPv4 carried +/// v4-mapped. +/// +/// Decoding canonicalizes: a v4-mapped IPv6 address comes back as the +/// IPv4 address it maps, which dials the same peer. +impl Addr for SocketAddr { + fn encode(&self) -> Vec { + let ip = match self.ip() { + IpAddr::V4(v4) => v4.to_ipv6_mapped(), + IpAddr::V6(v6) => v6, + }; + let mut bytes = Vec::with_capacity(SOCKET_ADDR_LEN); + bytes.extend_from_slice(&ip.octets()); + bytes.extend_from_slice(&self.port().to_be_bytes()); + bytes + } + + fn decode(bytes: &[u8]) -> Option { + if bytes.len() != SOCKET_ADDR_LEN { + return None; + } + let ip: [u8; 16] = bytes[..16].try_into().expect("length checked above"); + let port = u16::from_be_bytes(bytes[16..].try_into().expect("length checked above")); + let ip = Ipv6Addr::from(ip); + let ip = match ip.to_ipv4_mapped() { + Some(v4) => IpAddr::V4(v4), + None => IpAddr::V6(ip), + }; + Some(SocketAddr::new(ip, port)) + } +} + +/// One parsed connect header. +#[derive(Debug)] +pub(super) enum Header { + /// Establish a link: the connection is its control stream, and + /// `peer` is the dialer's advertised name for reverse dials. + Link { + /// The new link's identity. + token: Token, + /// Where this link's outgoing data streams dial. + peer: A, + }, + /// Attach one data stream to the link `token` names. + Stream { + /// The owning link's identity. + token: Token, + }, +} + +/// Encode a `LINK` header. +/// +/// `addr` is the dialer's advertised name, already encoded and already +/// length-checked (the endpoint validates it at construction). +pub(super) fn link_header(token: &Token, addr: &[u8]) -> Vec { + debug_assert!((1..=MAX_ADDR_LEN).contains(&addr.len())); + let mut bytes = Vec::with_capacity(PREFIX_LEN + 1 + addr.len()); + bytes.extend_from_slice(MAGIC); + bytes.push(VERSION); + bytes.push(KIND_LINK); + bytes.extend_from_slice(&token.0); + bytes.push(addr.len() as u8); + bytes.extend_from_slice(addr); + bytes +} + +/// Encode a `STREAM` header. +pub(super) fn stream_header(token: &Token) -> [u8; PREFIX_LEN] { + let mut bytes = [0; PREFIX_LEN]; + bytes[..MAGIC.len()].copy_from_slice(MAGIC); + bytes[MAGIC.len()] = VERSION; + bytes[MAGIC.len() + 1] = KIND_STREAM; + bytes[MAGIC.len() + 2..].copy_from_slice(&token.0); + bytes +} + +/// Shorthand for this module's malformed-header errors. +fn invalid(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +/// Read and parse one connect header from a fresh inbound connection. +/// +/// This is the router's only read on any connection: a fixed-width +/// prefix, then (for `LINK`) a length byte and exactly that many name +/// bytes, so the read is bounded by construction. +/// +/// # Errors +/// +/// Malformed bytes (wrong magic, unknown version or kind, a name that +/// does not decode) and truncation both fail with the underlying +/// classification; the router responds to either by dropping the +/// connection. +pub(super) async fn read(conn: &mut R) -> io::Result> { + let mut prefix = [0; PREFIX_LEN]; + conn.read_exact(&mut prefix).await?; + if &prefix[..MAGIC.len()] != MAGIC { + return Err(invalid("connect header: not a routed-link connection")); + } + if prefix[MAGIC.len()] != VERSION { + return Err(invalid("connect header: unknown version")); + } + let kind = prefix[MAGIC.len() + 1]; + let token = Token( + prefix[MAGIC.len() + 2..] + .try_into() + .expect("prefix layout fixes the token width"), + ); + match kind { + KIND_STREAM => Ok(Header::Stream { token }), + KIND_LINK => { + let mut len = [0; 1]; + conn.read_exact(&mut len).await?; + let len = usize::from(len[0]); + if len == 0 { + return Err(invalid("connect header: empty advertised name")); + } + let mut addr = vec![0; len]; + conn.read_exact(&mut addr).await?; + let peer = A::decode(&addr) + .ok_or_else(|| invalid("connect header: undecodable advertised name"))?; + Ok(Header::Link { token, peer }) + } + _ => Err(invalid("connect header: unknown kind")), + } +} + +#[cfg(test)] +mod tests; diff --git a/src/link/routed/header/tests.rs b/src/link/routed/header/tests.rs new file mode 100644 index 00000000..508136b7 --- /dev/null +++ b/src/link/routed/header/tests.rs @@ -0,0 +1,139 @@ +use std::io; +use std::net::{IpAddr, SocketAddr}; + +use proptest::prelude::*; + +use super::*; + +/// Parse a header from raw bytes, as the router would from a fresh +/// connection. +fn parse(mut bytes: &[u8]) -> io::Result> { + pollster::block_on(read::(&mut bytes)) +} + +/// A `STREAM` header carries its token through encode and parse +/// unchanged: the router routes on exactly the identity the dialer +/// quoted. +#[test] +fn stream_header_roundtrips() { + let token = Token::new(); + match parse(&stream_header(&token)).expect("a well-formed header parses") { + Header::Stream { token: parsed } => assert_eq!(parsed, token), + Header::Link { .. } => panic!("a STREAM header must parse as a stream"), + } +} + +/// A `LINK` header carries both its token and the dialer's advertised +/// name through encode and parse: the accept side dials back exactly +/// the name the dialer advertised. +#[test] +fn link_header_roundtrips() { + let token = Token::new(); + let advertised: SocketAddr = "127.0.0.1:7000".parse().expect("literal address"); + let bytes = link_header(&token, &advertised.encode()); + match parse(&bytes).expect("a well-formed header parses") { + Header::Link { + token: parsed, + peer, + } => { + assert_eq!(parsed, token); + assert_eq!(peer, advertised); + } + Header::Stream { .. } => panic!("a LINK header must parse as a link"), + } +} + +/// Bytes that do not open with the routed-link magic are rejected as +/// invalid data: a connection misrouted from another protocol fails at +/// the first read, precisely. +#[test] +fn wrong_magic_is_rejected() { + let mut bytes = stream_header(&Token::new()).to_vec(); + bytes[0] = b'X'; + let error = parse(&bytes).expect_err("a foreign magic must not parse"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +/// An unknown version byte is rejected: the version is the +/// compatibility door, and this parser speaks exactly one. +#[test] +fn unknown_version_is_rejected() { + let mut bytes = stream_header(&Token::new()).to_vec(); + bytes[MAGIC.len()] = VERSION + 1; + let error = parse(&bytes).expect_err("an unknown version must not parse"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +/// An unknown kind byte is rejected: kinds are closed vocabulary, and +/// anything else is a peer speaking a wire this router does not. +#[test] +fn unknown_kind_is_rejected() { + let mut bytes = stream_header(&Token::new()).to_vec(); + bytes[MAGIC.len() + 1] = 9; + let error = parse(&bytes).expect_err("an unknown kind must not parse"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +/// A `LINK` header advertising a zero-length name is rejected: a +/// nameless peer cannot be dialed back, so the establishment is +/// malformed by construction. +#[test] +fn empty_advertised_name_is_rejected() { + let token = Token::new(); + let mut bytes = link_header(&token, &[0]); + // Rewrite the length byte to zero and drop the placeholder name. + bytes[PREFIX_LEN] = 0; + bytes.truncate(PREFIX_LEN + 1); + let error = parse(&bytes).expect_err("an empty name must not parse"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +/// An advertised name the address type cannot decode is rejected: the +/// router refuses to build a link whose reverse dials would name +/// nothing. +#[test] +fn undecodable_advertised_name_is_rejected() { + let token = Token::new(); + let bytes = link_header(&token, b"not-a-socket-address"); + let error = parse(&bytes).expect_err("an undecodable name must not parse"); + assert_eq!(error.kind(), io::ErrorKind::InvalidData); +} + +/// A connection that closes mid-header surfaces as truncation, never a +/// partial parse: the router's read is exact-length by construction. +#[test] +fn truncation_is_rejected() { + let token = Token::new(); + let advertised: SocketAddr = "[::1]:9".parse().expect("literal address"); + let full = link_header(&token, &advertised.encode()); + for len in 0..full.len() { + let error = parse(&full[..len]).expect_err("a truncated header must not parse"); + assert_eq!(error.kind(), io::ErrorKind::UnexpectedEof); + } +} + +proptest! { + /// Every socket address survives the wire: decoding an encoded + /// address yields one that encodes identically and dials the same + /// peer. + /// + /// The wire form is canonical: IPv4 addresses come back exactly, + /// and a v4-mapped IPv6 address canonicalizes to the IPv4 it maps. + #[test] + fn socket_addr_roundtrips(ip in any::<[u8; 16]>(), v4 in any::(), port in any::()) { + let addr = if v4 { + let octets: [u8; 4] = ip[..4].try_into().expect("four bytes"); + SocketAddr::new(IpAddr::from(octets), port) + } else { + SocketAddr::new(IpAddr::from(ip), port) + }; + let encoded = addr.encode(); + prop_assert_eq!(encoded.len(), SOCKET_ADDR_LEN); + let decoded = SocketAddr::decode(&encoded).expect("an encoded address decodes"); + prop_assert_eq!(decoded.port(), addr.port()); + prop_assert_eq!(&decoded.encode(), &encoded); + if v4 { + prop_assert_eq!(decoded, addr); + } + } +} diff --git a/src/link/routed/router.rs b/src/link/routed/router.rs new file mode 100644 index 00000000..bbc6e779 --- /dev/null +++ b/src/link/routed/router.rs @@ -0,0 +1,246 @@ +//! The router: one loop that owns the listener and never blocks on +//! anyone. +//! +//! Structurally the router is a single loop funneling connections to +//! many links — the same shape whose head-of-line failure the link +//! contract exists to preclude, one layer up. The discipline here is +//! the module's core invariant: +//! +//! - **Never await a per-link queue.** Delivery is `try_send` into a +//! bounded channel; a full queue evicts that link (see +//! [`StreamAcceptor`]'s docs for what its owner observes), and the +//! loop moves on. +//! - **Hand off connections, not bytes.** After the header read a +//! connection never touches the router again, so flow control stays +//! end-to-end between the peers of each stream. A router that +//! proxied bytes would be a mux, and would inherit every coupling +//! problem the one-connection-per-stream shape exists to avoid. +//! - **Bound the header reads.** The header read is the router's only +//! I/O on a connection, bounded in size by the wire format and in +//! count by eviction: reads run as concurrent futures inside the +//! drive future (the crate spawns nothing), and past +//! [`Config::pending_headers`](super::Config::pending_headers) the +//! oldest pending read is aborted, its connection dropped. A peer +//! that dials and stalls mid-header therefore occupies one slot +//! until displaced, and can never park the loop. Wall-clock bounds +//! belong to the caller's [`Listen`]/[`Conn`](super::Conn) wrappers, +//! where a runtime's clock lives. + +use std::collections::hash_map::Entry; +use std::collections::{HashMap, VecDeque}; +use std::io; +use std::sync::{Arc, Mutex, MutexGuard}; + +use futures::future::{AbortHandle, Abortable}; +use futures::stream::{FuturesUnordered, StreamExt}; +use tokio::io::{AsyncWriteExt, split}; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TrySendError; + +use super::endpoint::{Arrival, LinkInfo}; +use super::header::{self, Header, Token}; +use super::stream::{StreamAcceptor, StreamConnector}; +use super::{Dial, Link, Listen}; +use crate::link::STREAM_COUNT; + +/// The routing table: each live link's token, mapped to the bounded +/// queue its acceptor drains. +/// +/// Shared between the router (inserts on inbound links, removes on +/// eviction), the endpoint (inserts on outbound links), and every +/// link's [`Registration`] (removes on drop). The mutex is never held +/// across an await. +pub(super) type Table = Arc>>>; + +/// Lock the table, riding through a poisoning panic. +/// +/// Every critical section is a single map operation, so a panic +/// elsewhere cannot leave the map torn; continuing lets the surviving +/// side of a panicked test observe eviction rather than a poison +/// cascade. +fn entries( + table: &Mutex>>, +) -> MutexGuard<'_, HashMap>> { + table + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +/// A link's claim on its token: dropping it revokes the route. +/// +/// Held by the link's [`StreamAcceptor`], so a +/// link that is dropped — poisoned, finished, or abandoned — stops +/// routing at that moment, whether or not the router ever hears +/// another byte. Connections that arrive for a revoked token are +/// dropped on sight. +pub(super) struct Registration { + table: Table, + token: Token, +} + +impl Registration { + /// Claim `token` in `table`; the claim ends when the value drops. + pub(super) fn new(table: Table, token: Token) -> Self { + Registration { table, token } + } +} + +impl Drop for Registration { + fn drop(&mut self) { + entries(&self.table).remove(&self.token); + } +} + +/// Register a fresh outbound link: mint an unclaimed token and claim +/// it. +/// +/// Collisions are astronomically unlikely at the token's width; the +/// loop makes the vacancy structural rather than probabilistic. +pub(super) fn register(table: &Table) -> (Token, Registration, mpsc::Receiver) { + let (sender, receiver) = mpsc::channel(STREAM_COUNT); + let mut sender = Some(sender); + let token = loop { + let token = Token::new(); + if let Entry::Vacant(vacancy) = entries(table).entry(token) { + vacancy.insert(sender.take().expect("the loop ends at the first vacancy")); + break token; + } + }; + (token, Registration::new(table.clone(), token), receiver) +} + +/// Drive one endpoint's router until the listener fails. +/// +/// The loop selects between accepting the next connection and +/// retiring finished header reads; all per-connection work (the +/// header read, the `LINK` acknowledgement, delivery) happens inside +/// the per-connection futures, so the loop itself never awaits any +/// one connection's progress. +pub(super) async fn drive( + mut listen: L, + dial: D, + table: Table, + incoming: mpsc::Sender>, + pending_headers: usize, +) -> io::Result<()> +where + D: Dial, + L: Listen, +{ + let mut pending = FuturesUnordered::new(); + // Insertion-ordered abort handles for the pending reads, so the + // count bound evicts oldest-first; ids reconcile the two + // collections because reads finish in arbitrary order. + let mut order: VecDeque<(u64, AbortHandle)> = VecDeque::new(); + let mut next_id: u64 = 0; + loop { + tokio::select! { + accepted = listen.accept() => { + let conn = accepted?; + if order.len() >= pending_headers + && let Some((_, oldest)) = order.pop_front() + { + oldest.abort(); + } + let (abort, registration) = AbortHandle::new_pair(); + let id = next_id; + next_id += 1; + order.push_back((id, abort)); + pending.push(Abortable::new( + route(conn, dial.clone(), &table, &incoming, id), + registration, + )); + } + Some(finished) = pending.next() => { + if let Ok(id) = finished { + order.retain(|(pending_id, _)| *pending_id != id); + } + } + } + } +} + +/// Read one connection's header and route it; errors drop the +/// connection. +/// +/// Returns the read's id so the drive loop can retire its abort +/// handle. +async fn route( + conn: D::Conn, + dial: D, + table: &Table, + incoming: &mpsc::Sender>, + id: u64, +) -> u64 { + // A failure here is a connection that never became anyone's + // stream: the dialer observes the drop as transport failure, and + // there is no one else to tell. + let _ = deliver(conn, dial, table, incoming).await; + id +} + +/// The routing step behind [`route`]: parse, then attach or establish. +async fn deliver( + mut conn: D::Conn, + dial: D, + table: &Table, + incoming: &mpsc::Sender>, +) -> io::Result<()> { + match header::read::(&mut conn).await? { + Header::Stream { token } => { + let Some(queue) = entries(table).get(&token).cloned() else { + // Unknown token: a revoked or never-registered link. + // Dropping the connection surfaces as transport failure + // on the dialing side, which owns the retry. + return Ok(()); + }; + match queue.try_send(conn) { + Ok(()) => {} + // A full queue proves peer misbehavior (an honest peer + // never exceeds a session's complement, and the queue + // holds one): evict the link so its owner sees a + // transport error instead of silently losing this + // stream and hanging its session. + Err(TrySendError::Full(overflow)) => { + drop(overflow); + entries(table).remove(&token); + } + // The acceptor is already gone; its registration is + // being (or has been) revoked with it. + Err(TrySendError::Closed(orphan)) => drop(orphan), + } + } + Header::Link { token, peer } => { + let (sender, receiver) = mpsc::channel(STREAM_COUNT); + { + let mut entries = entries(table); + if entries.contains_key(&token) { + // A duplicate establishment is a peer bug (tokens + // are minted fresh per link); dropping it leaves + // the live link undisturbed. + return Ok(()); + } + entries.insert(token, sender); + } + let registration = Registration::new(table.clone(), token); + // Reserve the application's slot before acknowledging: a + // full backlog rejects the link while the dialer is still + // waiting on the acknowledgement, the crisp failure. The + // registration guard revokes the token on every early + // return. + let Ok(slot) = incoming.try_reserve() else { + return Ok(()); + }; + conn.write_all(&[header::ACK]).await?; + let (control_read, control_write) = split(conn); + let link = Link::new( + control_read, + control_write, + StreamConnector::new(dial, peer.clone(), token), + StreamAcceptor::new(receiver, registration), + ); + slot.send((LinkInfo { peer, token }, link)); + } + } + Ok(()) +} diff --git a/src/link/routed/stream.rs b/src/link/routed/stream.rs new file mode 100644 index 00000000..677e17d7 --- /dev/null +++ b/src/link/routed/stream.rs @@ -0,0 +1,92 @@ +//! One link's stream supply: dial-per-open out, routed queue in. + +use std::io; + +use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; + +use super::header::{self, Token}; +use super::router::Registration; +use super::{Acceptor, Conn, Connector, Dial}; + +/// A routed link's [`Connector`]: every open dials one fresh +/// connection to the peer's router and labels it with the link's +/// token. +/// +/// The returned stream *is* the connection, so dropping it is the +/// transport half-close (the peer reads the final bytes, then +/// end-of-stream), and no open ever waits on another stream's +/// progress: the opens share nothing but the dialer. +pub struct StreamConnector { + dial: D, + peer: D::Addr, + token: Token, +} + +impl StreamConnector { + /// Bundle a link's outgoing supply: dial `peer`, quoting `token`. + pub(super) fn new(dial: D, peer: D::Addr, token: Token) -> Self { + StreamConnector { dial, peer, token } + } +} + +impl Clone for StreamConnector { + fn clone(&self) -> Self { + StreamConnector { + dial: self.dial.clone(), + peer: self.peer.clone(), + token: self.token, + } + } +} + +impl Connector for StreamConnector { + type Tx = D::Conn; + + async fn connect(&self) -> io::Result { + let mut conn = self.dial.dial(&self.peer).await?; + conn.write_all(&header::stream_header(&self.token)).await?; + Ok(conn) + } +} + +/// A routed link's [`Acceptor`]: drains the bounded queue the router +/// fills with this link's inbound stream connections. +/// +/// Receiving from the queue is cancel-safe (an undelivered connection +/// stays queued for the next accept), and the acceptor carries the +/// link's claim on its routing token, tying the routing to the link's +/// own lifetime: dropping the link revokes its token at that moment. +pub struct StreamAcceptor { + streams: mpsc::Receiver, + /// Revokes this link's token when the acceptor drops. + _registration: Registration, +} + +impl StreamAcceptor { + /// Bundle a link's incoming supply around its routed queue and + /// token claim. + pub(super) fn new(streams: mpsc::Receiver, registration: Registration) -> Self { + StreamAcceptor { + streams, + _registration: registration, + } + } +} + +impl Acceptor for StreamAcceptor { + type Rx = C; + + async fn accept(&mut self) -> io::Result { + // The router holds this queue's only sender, and removes it + // exactly when it evicts the link (a queue overflow, which + // proves peer misbehavior). Queued deliveries drain first, so + // eviction surfaces on the first accept past them. + self.streams.recv().await.ok_or_else(|| { + io::Error::new( + io::ErrorKind::BrokenPipe, + "link evicted by the router: stream queue overflowed", + ) + }) + } +} diff --git a/src/link/routed/tests.rs b/src/link/routed/tests.rs new file mode 100644 index 00000000..5e777201 --- /dev/null +++ b/src/link/routed/tests.rs @@ -0,0 +1,445 @@ +use std::future::Future; +use std::io; +use std::pin::pin; + +use futures::FutureExt; +use futures::future::{Either, select, try_join}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use super::header::{self, Token}; +use super::{Config, Dial, Endpoint, Incoming, LinkError, LinkInfo, RoutedLink}; +use crate::link::{Acceptor, Connector, Link, STREAM_COUNT}; +use crate::testing::{MemoryDial, MemoryName, MemoryNet}; + +/// One endpoint on `net`, listening at (and advertising) `name`. +fn endpoint( + net: &MemoryNet, + name: &str, + config: Config, +) -> ( + Endpoint, + Incoming, + impl Future>, +) { + let name = MemoryName::new(name); + let listen = net.listen(&name); + Endpoint::new(listen, name, net.dial(), config) +} + +/// Run `scenario` to completion while `routers` drives; a router that +/// resolves first is a failure (drive futures resolve only on +/// listener failure). +async fn drive( + routers: impl Future>, + scenario: impl Future, +) -> T { + match select(pin!(scenario), pin!(routers)).await { + Either::Left((value, _)) => value, + Either::Right((outcome, _)) => { + panic!("a router resolved mid-scenario: {outcome:?}") + } + } +} + +/// Both endpoints' routers as one drive future. +async fn routers( + a: impl Future>, + b: impl Future>, +) -> io::Result<()> { + try_join(a, b).await.map(|_| ()) +} + +/// Establish one link from `from` toward the peer named `name`, +/// collecting the peer's end from `incoming`. +async fn establish( + from: &Endpoint, + name: &str, + incoming: &mut Incoming, +) -> ( + RoutedLink, + LinkInfo, + RoutedLink, +) { + let (linked, arrival) = futures::join!(from.link(MemoryName::new(name)), incoming.accept()); + let dialer_end = linked.expect("establishment succeeds"); + let (info, acceptor_end) = arrival.expect("the router delivers the link"); + (dialer_end, info, acceptor_end) +} + +/// Open one stream from `opener` to `acceptor`, move `payload` across +/// it, close by drop, and require the receiver to observe exactly the +/// payload then end-of-stream. +async fn transfer( + opener: &Link, + acceptor: &mut Link, + payload: &[u8], +) where + Ca: Connector, + Ab: Acceptor, +{ + let open = async { + let mut tx = opener.connector.connect().await.expect("stream opens"); + tx.write_all(payload).await.expect("payload writes"); + }; + let read = async { + let mut rx = acceptor.acceptor.accept().await.expect("stream arrives"); + let mut received = Vec::new(); + rx.read_to_end(&mut received) + .await + .expect("stream reads to end-of-stream"); + received + }; + let ((), received) = futures::join!(open, read); + assert_eq!(received, payload); +} + +/// Establishment wires a full link: the control stream carries bytes +/// both ways, forward stream opens route by token, and reverse opens +/// dial the advertised name from the establishment header. +/// +/// Stream half-close is drop-driven: the receiver reads the payload, +/// then end-of-stream, with nothing left pending. +#[test] +fn establishment_connects_control_and_streams() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (a, mut a_incoming, a_router) = endpoint(&net, "a", Config::default()); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let (mut at_b, info, mut at_a) = establish(&b, "a", &mut a_incoming).await; + assert_eq!(info.peer, MemoryName::new("b")); + + // The control stream is the establishment connection: + // bytes cross in both directions independently. + at_b.control_write + .write_all(b"from b") + .await + .expect("control writes"); + let mut probe = [0; 6]; + at_a.control_read + .read_exact(&mut probe) + .await + .expect("control reads"); + assert_eq!(&probe, b"from b"); + at_a.control_write + .write_all(b"from a") + .await + .expect("control writes"); + at_b.control_read + .read_exact(&mut probe) + .await + .expect("control reads"); + assert_eq!(&probe, b"from a"); + + // Forward: the establishing side opens toward its peer. + transfer(&at_b, &mut at_a, b"forward payload").await; + // Reverse: the accepting side dials the advertised name. + transfer(&at_a, &mut at_b, b"reverse payload").await; + drop((a, b)); + }) + .await; + }); +} + +/// A stream connection quoting a token no live link owns is dropped: +/// the dialer observes end-of-stream, and no link's queue sees the +/// connection. +#[test] +fn unknown_token_is_dropped() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (_a, _a_incoming, a_router) = endpoint(&net, "a", Config::default()); + drive(a_router, async { + let mut conn = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + conn.write_all(&header::stream_header(&Token::new())) + .await + .expect("header writes"); + let mut drained = Vec::new(); + conn.read_to_end(&mut drained) + .await + .expect("the router drops the connection"); + assert!(drained.is_empty()); + }) + .await; + }); +} + +/// A stream queue driven past a full session complement proves peer +/// misbehavior and evicts the link: the queued complement still +/// drains, the next accept errors instead of hanging, and the token +/// routes nothing afterward. +#[test] +fn queue_overflow_evicts_the_link() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (a, mut a_incoming, a_router) = endpoint(&net, "a", Config::default()); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let (_at_b, info, mut at_a) = establish(&b, "a", &mut a_incoming).await; + + // A misbehaving peer: one connection past the complement, + // dialed raw so the link's own connector is not implicated. + let mut flood = Vec::new(); + for _ in 0..=STREAM_COUNT { + let mut conn = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + conn.write_all(&header::stream_header(&info.token)) + .await + .expect("header writes"); + flood.push(conn); + } + + // The queued complement drains; the accept after it + // surfaces the eviction as a transport error. + for _ in 0..STREAM_COUNT { + at_a.acceptor.accept().await.expect("queued streams drain"); + } + at_a.acceptor + .accept() + .await + .expect_err("eviction surfaces as a transport error"); + + // The token is revoked: a further stream connection is + // dropped on sight. + let mut conn = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + conn.write_all(&header::stream_header(&info.token)) + .await + .expect("header writes"); + let mut drained = Vec::new(); + conn.read_to_end(&mut drained) + .await + .expect("the router drops the connection"); + assert!(drained.is_empty()); + drop((a, b)); + }) + .await; + }); +} + +/// Dropping a link revokes its token at that moment, router +/// uninvolved: connections quoting the token afterward are dropped on +/// sight. +#[test] +fn dropping_a_link_revokes_its_token() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (a, mut a_incoming, a_router) = endpoint(&net, "a", Config::default()); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let (_at_b, info, at_a) = establish(&b, "a", &mut a_incoming).await; + drop(at_a); + let mut conn = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + conn.write_all(&header::stream_header(&info.token)) + .await + .expect("header writes"); + let mut drained = Vec::new(); + conn.read_to_end(&mut drained) + .await + .expect("the router drops the connection"); + assert!(drained.is_empty()); + drop((a, b)); + }) + .await; + }); +} + +/// A connection that stalls inside its connect header occupies one +/// pending-read slot, never the router: establishment and streams +/// proceed beside it untouched. +#[test] +fn stalled_header_does_not_park_the_router() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (a, mut a_incoming, a_router) = endpoint(&net, "a", Config::default()); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let mut stalled = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + stalled + .write_all(b"ROU") + .await + .expect("a partial magic writes"); + + let (at_b, _info, mut at_a) = establish(&b, "a", &mut a_incoming).await; + transfer(&at_b, &mut at_a, b"alive beside the stall").await; + drop((stalled, a, b)); + }) + .await; + }); +} + +/// Past the pending-header bound the oldest stalled connection is +/// evicted (its dialer observes end-of-stream), and the router keeps +/// serving: the count bound is the no-clock substitute for a header +/// deadline. +#[test] +fn pending_header_bound_evicts_oldest() { + pollster::block_on(async { + let net = MemoryNet::new(); + let config = Config { + pending_headers: 1, + ..Config::default() + }; + let (a, mut a_incoming, a_router) = endpoint(&net, "a", config); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let mut stalled = net.dial().dial(&MemoryName::new("a")).await.expect("dial"); + stalled + .write_all(b"ROU") + .await + .expect("a partial magic writes"); + + // The establishment connection displaces the stalled one. + let (_at_b, _info, _at_a) = establish(&b, "a", &mut a_incoming).await; + let mut drained = Vec::new(); + stalled + .read_to_end(&mut drained) + .await + .expect("eviction drops the stalled connection"); + assert!(drained.is_empty()); + drop((a, b)); + }) + .await; + }); +} + +/// A full incoming backlog rejects establishment while the dialer is +/// still waiting on the acknowledgement: the dialer gets a crisp +/// `Rejected`, not a queued link the application never sees. +#[test] +fn full_incoming_backlog_rejects_establishment() { + pollster::block_on(async { + let net = MemoryNet::new(); + let config = Config { + incoming_backlog: 1, + ..Config::default() + }; + let (a, a_incoming, a_router) = endpoint(&net, "a", config); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + // The first link fills the undrained backlog. + b.link(MemoryName::new("a")) + .await + .expect("the backlog admits one link"); + let Err(error) = b.link(MemoryName::new("a")).await else { + panic!("a full backlog must reject establishment"); + }; + assert!(matches!(error, LinkError::Rejected)); + drop((a, a_incoming, b)); + }) + .await; + }); +} + +/// A listener that answers establishment with anything but the +/// acknowledgement byte, or hangs up without answering, rejects the +/// link; the dialer's error says which contract failed. +#[test] +fn non_acknowledgement_rejects_establishment() { + pollster::block_on(async { + let net = MemoryNet::new(); + let fake = MemoryName::new("fake"); + let mut fake_listen = net.listen(&fake); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(b_router, async { + use super::Listen; + + // A wrong byte instead of the acknowledgement. + let (linked, served) = futures::join!(b.link(fake.clone()), async { + let mut conn = fake_listen + .accept() + .await + .expect("the fake listener accepts"); + conn.write_all(&[0x7f]) + .await + .expect("the wrong byte writes"); + conn + }); + let Err(error) = linked else { + panic!("a wrong byte must reject establishment"); + }; + assert!(matches!(error, LinkError::Rejected)); + drop(served); + + // A hang-up before any answer. + let (linked, ()) = futures::join!(b.link(fake.clone()), async { + let conn = fake_listen + .accept() + .await + .expect("the fake listener accepts"); + drop(conn); + }); + let Err(error) = linked else { + panic!("a hang-up must reject establishment"); + }; + assert!(matches!(error, LinkError::Rejected)); + }) + .await; + }); +} + +/// Establishment toward a name nothing listens at fails as transport +/// failure, not a hang: the dial's refusal passes through as `Io`. +#[test] +fn linking_to_nowhere_fails_as_transport() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(b_router, async { + let Err(error) = b.link(MemoryName::new("nowhere")).await else { + panic!("an unlistened name must refuse the dial"); + }; + assert!(matches!(error, LinkError::Io(_))); + }) + .await; + }); +} + +/// Dropped stream-open futures leave the link fully usable: whatever +/// stage the cancellation lands in, later opens route and deliver +/// exactly as before. +#[test] +fn cancelled_opens_leave_the_link_usable() { + pollster::block_on(async { + let net = MemoryNet::new(); + let (a, mut a_incoming, a_router) = endpoint(&net, "a", Config::default()); + let (b, _b_incoming, b_router) = endpoint(&net, "b", Config::default()); + drive(routers(a_router, b_router), async { + let (at_b, _info, mut at_a) = establish(&b, "a", &mut a_incoming).await; + + // Cancel opens unpolled, and cancel them after a single + // poll (the memory network resolves eagerly, so one poll + // may already have opened and immediately closed a + // connection; both fates must leave the link healthy). + drop(at_b.connector.connect()); + for _ in 0..3 { + let opened = at_b.connector.connect().now_or_never(); + drop(opened); + } + + // An open that completed before its cancellation delivers + // a connection that closes unwritten; drain any such + // orphans until the genuine payload arrives. + let open = async { + let mut tx = at_b.connector.connect().await.expect("stream opens"); + tx.write_all(b"after cancellations") + .await + .expect("payload writes"); + }; + let read = async { + loop { + let mut rx = at_a.acceptor.accept().await.expect("stream arrives"); + let mut received = Vec::new(); + rx.read_to_end(&mut received) + .await + .expect("stream reads out"); + if !received.is_empty() { + break received; + } + } + }; + let ((), received) = futures::join!(open, read); + assert_eq!(received, b"after cancellations"); + drop((a, b)); + }) + .await; + }); +} diff --git a/src/testing.rs b/src/testing.rs index 50c292c4..e0e3f37c 100644 --- a/src/testing.rs +++ b/src/testing.rs @@ -1,7 +1,9 @@ //! Executor-agnostic test support shared across protocol and API suites. +mod memnet; mod transport; +pub use memnet::{MemoryDial, MemoryListen, MemoryName, MemoryNet}; pub use transport::{ AdversarialAcceptor, AdversarialConnector, AdversarialRead, AdversarialWrite, FaultUnit as IoFaultUnit, InjectedIo, IoFault, IoPlan, IoReport, IoReportHandle, diff --git a/src/testing/memnet.rs b/src/testing/memnet.rs new file mode 100644 index 00000000..58e9c41c --- /dev/null +++ b/src/testing/memnet.rs @@ -0,0 +1,137 @@ +//! A process-local accept/connect network for exercising the routed +//! link without sockets. +//! +//! [`MemoryNet`] is a registry of named listeners; dialing a name +//! creates one [`tokio::io::duplex`] connection and delivers one end +//! to the listener, exactly the accept/connect primitive the +//! [`routed`](crate::link::routed) adapter builds on. Everything is +//! channels and buffers, so suites run deterministically under a +//! single-poll executor, and names are plain strings, which keeps the +//! address seam honest: nothing here resembles an IP address. + +use std::collections::HashMap; +use std::io; +use std::sync::{Arc, Mutex, MutexGuard}; + +use tokio::io::{DuplexStream, duplex}; +use tokio::sync::mpsc; + +use crate::link::routed::{Addr, Dial, Listen}; + +/// Bytes each connection buffers per direction before its writer +/// blocks on its reader. +const CONNECTION_CAPACITY: usize = 8 * 1024; + +/// Dialed-but-unaccepted connections a listener holds; past it, dials +/// are refused, as a full accept backlog refuses connections on a real +/// transport. +const LISTEN_BACKLOG: usize = 64; + +/// A name on a [`MemoryNet`]: an arbitrary string, encoded as its +/// UTF-8 bytes. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct MemoryName(pub String); + +impl MemoryName { + /// Name a peer on the memory network. + pub fn new(name: impl Into) -> Self { + MemoryName(name.into()) + } +} + +impl Addr for MemoryName { + fn encode(&self) -> Vec { + self.0.clone().into_bytes() + } + + fn decode(bytes: &[u8]) -> Option { + String::from_utf8(bytes.to_vec()).ok().map(MemoryName) + } +} + +/// A closed-world accept/connect network: named listeners, in-memory +/// connections. +/// +/// Clones share the network. Bind listeners with +/// [`listen`](Self::listen), dial them through the [`Dial`] handle +/// from [`dial`](Self::dial). +#[derive(Clone, Default)] +pub struct MemoryNet { + listeners: Arc>>>, +} + +impl MemoryNet { + /// An empty network. + pub fn new() -> Self { + Self::default() + } + + /// Bind a listener at `name`, displacing any earlier binding. + pub fn listen(&self, name: &MemoryName) -> MemoryListen { + let (sender, receiver) = mpsc::channel(LISTEN_BACKLOG); + self.registry().insert(name.0.clone(), sender); + MemoryListen { conns: receiver } + } + + /// A dialer onto this network. + pub fn dial(&self) -> MemoryDial { + MemoryDial { net: self.clone() } + } + + /// Lock the listener registry, riding through a poisoning panic: + /// each critical section is a single map operation, so the map is + /// never torn. + fn registry(&self) -> MutexGuard<'_, HashMap>> { + self.listeners + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// The [`Dial`] half of a [`MemoryNet`]. +#[derive(Clone)] +pub struct MemoryDial { + net: MemoryNet, +} + +impl Dial for MemoryDial { + type Addr = MemoryName; + type Conn = DuplexStream; + + async fn dial(&self, addr: &MemoryName) -> io::Result { + let listener = self.net.registry().get(&addr.0).cloned().ok_or_else(|| { + io::Error::new( + io::ErrorKind::ConnectionRefused, + "nothing listens at this name", + ) + })?; + let (dialed, accepted) = duplex(CONNECTION_CAPACITY); + // A full or abandoned backlog refuses the dial outright; a + // dial never waits on the listener's accept pace, mirroring + // the routed adapter's requirement that opens not serialize + // behind anyone else's progress. + listener.try_send(accepted).map_err(|_| { + io::Error::new( + io::ErrorKind::ConnectionRefused, + "the listener's backlog is full", + ) + })?; + Ok(dialed) + } +} + +/// The [`Listen`] half of one [`MemoryNet`] name. +pub struct MemoryListen { + conns: mpsc::Receiver, +} + +impl Listen for MemoryListen { + type Conn = DuplexStream; + + async fn accept(&mut self) -> io::Result { + self.conns + .recv() + .await + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "the memory network is gone")) + } +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs index c9bfeb50..b44022ab 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -21,7 +21,8 @@ //! - [`fault`] injects transport adversity; [`flaky`] injects //! bookmark-storage adversity. //! - [`wire`] and [`tcp`] carry the same sessions over in-memory links -//! and real sockets. +//! and real sockets; [`routed_tcp`] is the socket instantiation of +//! the routed adapter's dial/listen seam. //! - [`gossip_snapshot`] captures a session's exact bytes for the `insta` //! pins. //! @@ -35,6 +36,7 @@ pub mod flaky; pub mod gossip_snapshot; pub mod oracle; pub mod peer; +pub mod routed_tcp; pub mod schedule; pub mod sim; pub mod tcp; diff --git a/tests/common/routed_tcp.rs b/tests/common/routed_tcp.rs new file mode 100644 index 00000000..c448cb7e --- /dev/null +++ b/tests/common/routed_tcp.rs @@ -0,0 +1,75 @@ +//! The TCP instantiation of the routed link's transport seam. +//! +//! [`rumors::link::routed`] is generic over a [`Dial`]/[`Listen`] pair; +//! this is that pair over real sockets, with optional kernel-buffer +//! clamps so the conformance suite can shrink per-stream capacity to +//! the OS floor. Socket policy stops here: the adapter above sees only +//! byte streams. + +use std::io; +use std::net::SocketAddr; + +use rumors::link::routed::{Dial, Listen}; +use tokio::net::{TcpListener, TcpSocket, TcpStream}; + +/// Listener backlog: a full session complement of simultaneous stream +/// dials from a handful of peers, with room to spare. +const BACKLOG: u32 = 256; + +/// Dials one TCP connection per routed-link connection. +#[derive(Clone)] +pub struct TcpDial { + /// Send-buffer clamp for each dialed socket; `None` keeps the + /// platform default. + pub send_buffer: Option, +} + +impl Dial for TcpDial { + type Addr = SocketAddr; + type Conn = TcpStream; + + async fn dial(&self, addr: &SocketAddr) -> io::Result { + match self.send_buffer { + None => TcpStream::connect(*addr).await, + Some(send) => { + let socket = TcpSocket::new_v4()?; + socket.set_send_buffer_size(send)?; + socket.connect(*addr).await + } + } + } +} + +/// Accepts one process's inbound routed-link connections. +pub struct TcpListen(TcpListener); + +impl TcpListen { + /// Bind a fresh loopback listener, reporting the address peers + /// dial (and the endpoint advertises). + /// + /// `recv_buffer` clamps each accepted socket's receive buffer + /// toward the request (accepted sockets inherit the listener's, so + /// the clamp must land before `listen`); `None` keeps the platform + /// default. + pub async fn bind(recv_buffer: Option) -> io::Result<(Self, SocketAddr)> { + let listener = match recv_buffer { + None => TcpListener::bind("127.0.0.1:0").await?, + Some(recv) => { + let socket = TcpSocket::new_v4()?; + socket.set_recv_buffer_size(recv)?; + socket.bind(SocketAddr::from(([127, 0, 0, 1], 0)))?; + socket.listen(BACKLOG)? + } + }; + let addr = listener.local_addr()?; + Ok((TcpListen(listener), addr)) + } +} + +impl Listen for TcpListen { + type Conn = TcpStream; + + async fn accept(&mut self) -> io::Result { + Ok(self.0.accept().await?.0) + } +} diff --git a/tests/common/tcp.rs b/tests/common/tcp.rs index fd7f4c9c..7ef3cfc4 100644 --- a/tests/common/tcp.rs +++ b/tests/common/tcp.rs @@ -16,9 +16,9 @@ //! Per-stream flow control and half-close come from TCP itself, one socket //! per stream, which is exactly what the contract's independence clause //! asks for. A production deployment would keep one listener per *process* -//! and route by connect header instead (see the `link` module docs); the -//! per-session listener here trades that machinery for obviousness, which -//! is the right trade in a test. +//! and route by connect header instead ([`rumors::link::routed`] is that +//! shape); the per-session listener here trades that machinery for +//! obviousness, which is the right trade in a test. use std::io; use std::net::SocketAddr; diff --git a/tests/routed_link.rs b/tests/routed_link.rs new file mode 100644 index 00000000..e7b5369c --- /dev/null +++ b/tests/routed_link.rs @@ -0,0 +1,256 @@ +//! The routed adapter satisfies the link contract, over real sockets +//! and the in-memory network. +//! +//! [`rumors::link::routed`] is the crate's shipped shape for +//! accept/connect transports, so it answers to the same public +//! conformance suite as any caller-built transport — here over TCP at +//! default and OS-floor socket buffers, in both construction +//! orientations (the dialing and accepting ends of a routed link are +//! built by different code paths), and over the in-memory network, +//! whose string names keep the address seam honest. Real sockets need +//! real time, and every run sits under an explicit timeout because the +//! contract's liveness clauses fail as hangs. +//! +//! The mesh test then exercises the process scope the suite cannot: a +//! full mesh of endpoints converging by concurrent gossip over routed +//! links, beside a connection stalled mid-header — the router's +//! never-block law observed end to end over sockets. + +mod common; + +use std::net::SocketAddr; +use std::time::Duration; + +use rumors::link::routed::{Config, Endpoint, Incoming, RoutedLink}; +use rumors::testing::{MemoryDial, MemoryName, MemoryNet}; +use rumors::{Peer, Rumors}; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpStream; +use tokio::time::timeout; + +use crate::common::routed_tcp::{TcpDial, TcpListen}; +use crate::common::wire::bootstrap_fork_async; + +/// Bound on one whole suite run; loopback and in-memory checks finish +/// in seconds, so a run past this is a liveness violation, not a slow +/// machine. +const SUITE_TIMEOUT: Duration = Duration::from_secs(120); + +/// Bound on the mesh test's establishment and gossip rounds. +const MESH_TIMEOUT: Duration = Duration::from_secs(60); + +/// Buffer request for the minimal-capacity variants: the OS rounds a +/// socket-buffer request up to its floor, so this asks for the +/// smallest per-stream capacity the platform lawfully provides. +const MINIMAL_BUFFER_REQUEST: u32 = 1; + +/// Messages each mesh replica contributes before the gossip rounds. +const MESH_PAYLOADS: u64 = 16; + +/// One TCP endpoint with its router spawned onto the ambient runtime; +/// the router task runs (and holds the listener) for the rest of the +/// process, which is the deployment shape. +async fn tcp_endpoint(buffers: Option) -> (Endpoint, Incoming, SocketAddr) { + let (listen, addr) = TcpListen::bind(buffers) + .await + .expect("bind a loopback listener"); + let (endpoint, incoming, router) = Endpoint::new( + listen, + addr, + TcpDial { + send_buffer: buffers, + }, + Config::default(), + ); + tokio::spawn(router); + (endpoint, incoming, addr) +} + +/// Mint a fresh routed-link pair over TCP: two endpoints, one +/// establishment. +/// +/// `dialer_first` picks which construction (the dialing or the +/// accepting end) lands in the suite's first seat, so both +/// orientations get every per-side probe. +async fn tcp_pair( + buffers: Option, + dialer_first: bool, +) -> (RoutedLink, RoutedLink) { + let (_a, mut a_incoming, a_addr) = tcp_endpoint(buffers).await; + let (b, _b_incoming, _b_addr) = tcp_endpoint(buffers).await; + let (linked, arrival) = tokio::join!(b.link(a_addr), a_incoming.accept()); + let dialed = linked.expect("establishment succeeds"); + let (_info, accepted) = arrival.expect("the router delivers the link"); + if dialer_first { + (dialed, accepted) + } else { + (accepted, dialed) + } +} + +/// Run the whole conformance suite against fresh TCP pairs at the +/// given buffer sizing and construction orientation. +async fn tcp_conformance(buffers: Option, dialer_first: bool) { + timeout( + SUITE_TIMEOUT, + rumors::conformance::link::check(async || tcp_pair(buffers, dialer_first).await), + ) + .await + .expect("conformance suite ran past its liveness bound"); +} + +/// Mint a fresh routed-link pair over an in-memory network of its own. +async fn memory_pair(dialer_first: bool) -> (RoutedLink, RoutedLink) { + let net = MemoryNet::new(); + let name_a = MemoryName::new("a"); + let (_a, mut a_incoming, a_router) = Endpoint::new( + net.listen(&name_a), + name_a.clone(), + net.dial(), + Config::default(), + ); + tokio::spawn(a_router); + let name_b = MemoryName::new("b"); + let (b, _b_incoming, b_router) = + Endpoint::new(net.listen(&name_b), name_b, net.dial(), Config::default()); + tokio::spawn(b_router); + let (linked, arrival) = tokio::join!(b.link(name_a), a_incoming.accept()); + let dialed = linked.expect("establishment succeeds"); + let (_info, accepted) = arrival.expect("the router delivers the link"); + if dialer_first { + (dialed, accepted) + } else { + (accepted, dialed) + } +} + +/// At the platform's default socket buffers, the routed TCP link +/// satisfies every contract clause the suite observes. +/// +/// The clauses: independent control pipes, receiver-paced independent +/// streams, clean half-close, tolerated accept cancellation, and full +/// reconciliation sessions. +#[tokio::test] +async fn conforms_over_tcp_at_default_buffers() { + tcp_conformance(None, true).await; +} + +/// The default-buffer suite with the pair's seats swapped: the +/// accepting end (router-built) takes the first seat, so its +/// construction path gets the other half of the suite's asymmetric +/// probes. +#[tokio::test] +async fn conforms_over_tcp_at_default_buffers_swapped() { + tcp_conformance(None, false).await; +} + +/// The contract holds with per-stream kernel buffers clamped to the +/// OS floor: shrinking capacity changes when backpressure engages, +/// never whether streams stay independent, receiver-paced, and +/// half-close clean. +#[tokio::test] +async fn conforms_over_tcp_at_minimal_buffers() { + tcp_conformance(Some(MINIMAL_BUFFER_REQUEST), true).await; +} + +/// The minimal-buffer suite with the pair's seats swapped, as for the +/// default-buffer variant. +#[tokio::test] +async fn conforms_over_tcp_at_minimal_buffers_swapped() { + tcp_conformance(Some(MINIMAL_BUFFER_REQUEST), false).await; +} + +/// The adapter conforms over the in-memory network too: nothing in +/// the contract mapping leans on socket semantics, and the string +/// names prove the address seam carries non-IP namespaces. +#[tokio::test] +async fn conforms_over_the_memory_network() { + timeout( + SUITE_TIMEOUT, + rumors::conformance::link::check(async || memory_pair(true).await), + ) + .await + .expect("conformance suite ran past its liveness bound"); +} + +/// The in-memory suite with the pair's seats swapped, as for TCP. +#[tokio::test] +async fn conforms_over_the_memory_network_swapped() { + timeout( + SUITE_TIMEOUT, + rumors::conformance::link::check(async || memory_pair(false).await), + ) + .await + .expect("conformance suite ran past its liveness bound"); +} + +/// A full mesh gossips to convergence over routed TCP links while a +/// connection stalled mid-header sits in one router the whole time. +/// +/// Three endpoints, three links, concurrent sessions from shared +/// replicas: per-link routing and the pending-header bound keep every +/// live link flowing beside the stall. +#[tokio::test(flavor = "multi_thread")] +async fn mesh_converges_beside_a_stalled_header() { + timeout(MESH_TIMEOUT, async { + let (_a_ep, mut a_incoming, a_addr) = tcp_endpoint(None).await; + let (b_ep, mut b_incoming, b_addr) = tcp_endpoint(None).await; + let (c_ep, _c_incoming, _c_addr) = tcp_endpoint(None).await; + + // The stall: a connection into a's router that never finishes + // its header, held open across the whole mesh's traffic. + let mut stalled = TcpStream::connect(a_addr).await.expect("dial the stall"); + stalled + .write_all(b"ROU") + .await + .expect("a partial magic writes"); + + // The mesh: b→a, c→a, c→b. + let (linked, arrival) = tokio::join!(b_ep.link(a_addr), a_incoming.accept()); + let mut ab_at_b = linked.expect("b links a"); + let (_, mut ab_at_a) = arrival.expect("a receives b's link"); + let (linked, arrival) = tokio::join!(c_ep.link(a_addr), a_incoming.accept()); + let mut ac_at_c = linked.expect("c links a"); + let (_, mut ac_at_a) = arrival.expect("a receives c's link"); + let (linked, arrival) = tokio::join!(c_ep.link(b_addr), b_incoming.accept()); + let mut bc_at_c = linked.expect("c links b"); + let (_, mut bc_at_b) = arrival.expect("b receives c's link"); + + // Three replicas with disjoint content. + let a: Rumors = Peer::seed().sync_window_floor().into_rumors(); + let b = bootstrap_fork_async(&a).await; + let c = bootstrap_fork_async(&a).await; + for payload in 0..MESH_PAYLOADS { + a.send(payload); + b.send(100 + payload); + c.send(200 + payload); + } + + // Two rounds, each running all three pairwise sessions + // concurrently (a replica may gossip on several links at + // once); the second round settles anything a concurrent first + // round exchanged before its peers' inserts landed. + for round in 0..2 { + let ((ab_a, ab_b), (ac_a, ac_c), (bc_b, bc_c)) = tokio::join!( + async { tokio::join!(a.gossip(&mut ab_at_a), b.gossip(&mut ab_at_b)) }, + async { tokio::join!(a.gossip(&mut ac_at_a), c.gossip(&mut ac_at_c)) }, + async { tokio::join!(b.gossip(&mut bc_at_b), c.gossip(&mut bc_at_c)) }, + ); + for outcome in [ab_a, ac_a] { + outcome.unwrap_or_else(|error| panic!("a's round {round} session: {error}")); + } + for outcome in [ab_b, bc_b] { + outcome.unwrap_or_else(|error| panic!("b's round {round} session: {error}")); + } + for outcome in [ac_c, bc_c] { + outcome.unwrap_or_else(|error| panic!("c's round {round} session: {error}")); + } + } + + assert_eq!(a.snapshot().hash(), b.snapshot().hash(), "a and b diverge"); + assert_eq!(b.snapshot().hash(), c.snapshot().hash(), "b and c diverge"); + drop(stalled); + }) + .await + .expect("the mesh ran past its liveness bound"); +}