From 07203f65179cf5773711c65cd75804caaa12f807 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Fri, 7 Aug 2026 23:21:29 +0000 Subject: [PATCH 1/3] Add Endpoint::local_addr, the advertised name given at construction Co-Authored-By: Claude Fable 5 --- src/link/routed/endpoint.rs | 19 ++++++++++++++----- src/link/routed/tests.rs | 10 ++++++++++ 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/src/link/routed/endpoint.rs b/src/link/routed/endpoint.rs index 35249840..39f92b8e 100644 --- a/src/link/routed/endpoint.rs +++ b/src/link/routed/endpoint.rs @@ -148,9 +148,11 @@ impl Clone for Endpoint { 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, + /// The endpoint's advertised name, as given at construction. + local_addr: D::Addr, + /// The advertised name, pre-encoded (validated at construction) for + /// the `LINK` headers this endpoint writes. + encoded: Vec, } impl Endpoint { @@ -209,13 +211,20 @@ impl Endpoint { inner: Arc::new(Inner { table: table.clone(), dial: dial.clone(), - advertised: encoded, + local_addr: advertised, + encoded, }), }; let router = router::drive(listen, dial, table, arrivals, config.pending_headers); Ok((endpoint, Incoming { links: incoming }, router)) } + /// The name peers dial this endpoint at: `advertised`, as given at + /// construction. + pub fn local_addr(&self) -> &D::Addr { + &self.inner.local_addr + } + /// Establish one link to the peer reachable at `peer`. /// /// One round trip: dial the peer's router, announce the link (its @@ -241,7 +250,7 @@ impl Endpoint { // 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)) + conn.write_all(&header::link_header(&token, &self.inner.encoded)) .await?; let mut ack = [0; 1]; match conn.read_exact(&mut ack).await { diff --git a/src/link/routed/tests.rs b/src/link/routed/tests.rs index dd7e6c8d..6ddd0740 100644 --- a/src/link/routed/tests.rs +++ b/src/link/routed/tests.rs @@ -141,6 +141,16 @@ fn establishment_connects_control_and_streams() { }); } +/// `local_addr` is the advertised name given at construction, the name +/// peers dial this endpoint at; callers need it back for policies like +/// dial tiebreaks. +#[test] +fn local_addr_is_the_constructed_name() { + let net = MemoryNet::new(); + let (a, _incoming, _router) = endpoint(&net, "a", Config::default()); + assert_eq!(*a.local_addr(), MemoryName::new("a")); +} + /// 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. From 4be4b8308b38d6961e7ccaf613394e7d5bbf3f6d Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Thu, 13 Aug 2026 19:30:17 +0000 Subject: [PATCH 2/3] Snapshot: stream completion recovers connections for reuse Work in progress. Streams already end with an in-band end control; the receiver no longer demands transport EOF behind it. Every connect and accept now pairs the stream half with a Done handle, invoked at the protocol's end of the data with the half handed back; a half dropped anywhere else is an abort, observed as EOF. The routed adapter returns completed write halves through Dial::recycle (default: drop) and read halves to the router's header wait, so a pooling Dial carries stream after stream over one connection. Native stream types ride bare: no trait to implement, plain transports pair with Done::discard(). Full suite, conformance completion probes, and the suite over a pooling TCP dial all pass, along with every gate leg this box can run (it lacks cargo-audit, and the wasm leg has not had a full run). --- benches/support/latency.rs | 9 +- design/routed-link.md | 18 +++ examples/swarm.rs | 19 ++- src/conformance/link.rs | 97 ++++++++---- src/conformance/link/tests.rs | 93 +++++++----- src/link.rs | 69 +++++++-- src/link/erased.rs | 97 ++++++++++-- src/link/routed.rs | 55 +++++-- src/link/routed/endpoint.rs | 10 +- src/link/routed/router.rs | 72 +++++---- src/link/routed/stream.rs | 35 +++-- src/link/routed/tests.rs | 140 +++++++++++++++++- src/link/tests.rs | 14 +- src/testing/transport.rs | 38 +++-- src/tests.rs | 15 +- .../streaming/remote/codec/decode/async_io.rs | 7 + .../streaming/remote/codec/encode/async_io.rs | 5 +- .../streaming/remote/proxy/tests/harness.rs | 8 +- .../streaming/remote/proxy/tests/malformed.rs | 25 ++-- .../streaming/remote/proxy/work/tests.rs | 6 +- src/tree/mirror/streaming/remote/streams.rs | 84 +++++------ .../mirror/streaming/remote/streams/tests.rs | 42 +----- src/tree/mirror/streaming/stats.rs | 10 ++ tests/common/fault.rs | 22 ++- tests/common/gossip_snapshot.rs | 8 +- tests/common/routed_tcp.rs | 36 +++++ tests/common/tcp.rs | 10 +- tests/dispute_wire.rs | 16 +- tests/hop_trace.rs | 20 ++- tests/routed_link.rs | 54 ++++++- tests/session_stats.rs | 16 +- 31 files changed, 830 insertions(+), 320 deletions(-) diff --git a/benches/support/latency.rs b/benches/support/latency.rs index 622cdbb6..a873173f 100644 --- a/benches/support/latency.rs +++ b/benches/support/latency.rs @@ -81,7 +81,7 @@ use std::time::Duration; use borsh::{BorshDeserialize, BorshSerialize}; use rumors::Rumors; -use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT}; +use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::sync::mpsc; use tokio::time::{Instant, Sleep}; @@ -303,13 +303,13 @@ pub struct DelayedConnector { impl Connector for DelayedConnector { type Tx = DelayedWriter; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let (tx, rx) = delayed_pipe(self.capacity, self.delay); self.announce .send(rx) .await .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?; - Ok(tx) + Ok((tx, Done::discard())) } } @@ -322,10 +322,11 @@ pub struct DelayedAcceptor { impl Acceptor for DelayedAcceptor { type Rx = DelayedReader; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { self.streams .recv() .await + .map(|rx| (rx, Done::discard())) .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone")) } } diff --git a/design/routed-link.md b/design/routed-link.md index e2a15406..8518b9bd 100644 --- a/design/routed-link.md +++ b/design/routed-link.md @@ -318,6 +318,24 @@ Accepted failure modes, stated rather than hidden: 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-08-13): completed streams recover their + connections.** Supersedes the single-use decision above; the future + case it named arrived (an attested-handshake transport whose RoT + serializes handshakes at roughly one per second). The door was not + the header after all: no new kind, and no layer-A framing either. + The protocol already ends every stream with an in-band end control, + so the receiver stopped demanding transport EOF behind it, and the + link contract now pairs every stream half with a completion handle + (`Done`) invoked exactly at that boundary. The write half goes back + to the `Dial` (`recycle`, defaulting to today's drop), the read half + back to the router to await its next connect header; a dropped half + remains the abort, observed as EOF. Trailing-byte ambiguity does not + arise: the codec's reads are exact, so a completed stream leaves the + next header untouched, and bytes past an end control belong to the + transport, never the session. The half-close rows in the mapping + table above describe the abort path only. The qorb facts in the + previous entry stand for the pooling `Dial` a deployment builds on + `recycle`. - **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` diff --git a/examples/swarm.rs b/examples/swarm.rs index 26df9c02..4cc20207 100644 --- a/examples/swarm.rs +++ b/examples/swarm.rs @@ -148,7 +148,7 @@ use ratatui::layout::{Constraint, Layout, Rect}; use ratatui::style::{Color, Modifier, Style}; use ratatui::text::{Line, Span}; use ratatui::widgets::{Block, Borders, Paragraph, Sparkline}; -use rumors::link::{Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink}; +use rumors::link::{Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink}; use rumors::{Key, Peer, Retire, Rumors, UnorderedMessages}; use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf}; @@ -1159,13 +1159,16 @@ struct CountConnector { impl Connector for CountConnector { type Tx = CountWrite; - async fn connect(&self) -> io::Result { - let tx = self.inner.connect().await?; - Ok(CountWrite { - inner: tx, - wire_bytes: Arc::clone(&self.wire_bytes), - rounds: None, - }) + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let (tx, _) = self.inner.connect().await?; + Ok(( + CountWrite { + inner: tx, + wire_bytes: Arc::clone(&self.wire_bytes), + rounds: None, + }, + Done::discard(), + )) } } diff --git a/src/conformance/link.rs b/src/conformance/link.rs index bb32a73d..e351fde8 100644 --- a/src/conformance/link.rs +++ b/src/conformance/link.rs @@ -63,7 +63,7 @@ use std::task::{Context, Poll}; use futures::future::{Either, join, join_all, select}; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use crate::link::{Acceptor, Connector, Link, LinkParts, STREAM_COUNT}; +use crate::link::{Acceptor, Connector, Done, Link, LinkParts, STREAM_COUNT}; use crate::{Peer, Rumors}; /// Bytes used to probe stream delivery without assuming any capacity. @@ -322,11 +322,13 @@ fn duplex_fill(tag: u8) -> Vec { (0..CONTROL_DUPLEX_FILL).map(|i| (i as u8) ^ tag).collect() } -/// An opened stream delivers its exact bytes to the peer's acceptor, and the -/// writer's drop surfaces as end-of-stream after the final byte. +/// An opened stream delivers its exact bytes to the peer's acceptor, +/// ended either way the contract allows. /// -/// Probed in both directions: each side's connector against the other -/// side's acceptor. +/// A dropped stream's abort surfaces as end-of-stream after the final +/// byte. A completed stream leaves the acceptor's next streams still +/// arriving. Probed in both directions: each side's connector against +/// the other side's acceptor. pub async fn check_streams( a: Link, b: Link, @@ -344,36 +346,79 @@ pub async fn check_streams( let mut b = b.into_parts(); probe_stream(&a.connector, &mut b.acceptor).await; probe_stream(&b.connector, &mut a.acceptor).await; + probe_completed_streams(&a.connector, &mut b.acceptor).await; + probe_completed_streams(&b.connector, &mut a.acceptor).await; } -/// One direction of [`check_streams`]: a single stream, delivered exactly. +/// One direction of [`check_streams`]: a single stream, delivered exactly +/// and ended by the abort (the halves dropped, their handles unused). async fn probe_stream(connector: &C, acceptor: &mut A) { let send = async { - let mut tx = connector + let (mut tx, done) = connector .connect() .await .expect("contract: connect succeeds while the peer link lives"); tx.write_all(PROBE).await.expect("contract: stream write"); tx.flush().await.expect("contract: stream flush"); - drop(tx); + drop((tx, done)); }; let receive = async { - let mut rx = acceptor + let (mut rx, done) = acceptor .accept() .await .expect("contract: an opened stream is accepted"); let mut bytes = Vec::new(); rx.read_to_end(&mut bytes) .await - .expect("contract: half-close surfaces as end-of-stream"); + .expect("contract: an abort surfaces as end-of-stream"); assert_eq!( bytes, PROBE, "contract: a stream delivers its exact bytes in order", ); + drop((rx, done)); }; join(send, receive).await; } +/// The completion leg of [`check_streams`]: two consecutive streams, +/// each ended by completing the write half at its final byte. +/// +/// The receiver reads exactly the probe and completes its half there, +/// never probing for end-of-stream: the contract's completion clause +/// leaves what follows the data transport-defined (nothing at all, on +/// an instantiation that recovers the connection). The second stream +/// proves the supply survives the first one's completion, wherever its +/// connection went. +async fn probe_completed_streams(connector: &C, acceptor: &mut A) { + for _ in 0..2 { + let send = async { + let (mut tx, done) = connector + .connect() + .await + .expect("contract: connect succeeds while the peer link lives"); + tx.write_all(PROBE).await.expect("contract: stream write"); + tx.flush().await.expect("contract: stream flush"); + done.complete(tx); + }; + let receive = async { + let (mut rx, done) = acceptor + .accept() + .await + .expect("contract: an opened stream is accepted"); + let mut bytes = vec![0u8; PROBE.len()]; + rx.read_exact(&mut bytes) + .await + .expect("contract: a completed stream delivers its bytes"); + assert_eq!( + bytes, PROBE, + "contract: a stream delivers its exact bytes in order", + ); + done.complete(rx); + }; + join(send, receive).await; + } +} + /// Streams are independent: a stream whose receiver never drains blocks /// nothing but itself. /// @@ -421,7 +466,7 @@ async fn probe_independence(connector: &C, acceptor: let send = async { // The stalled stream: tagged so the receiver can hold it unread // wherever it lands in arrival order. - let mut stalled = connector + let (mut stalled, _) = connector .connect() .await .expect("contract: connect succeeds"); @@ -460,7 +505,7 @@ async fn probe_independence(connector: &C, acceptor: // writing to one stream may block only on that stream's receiver. let live = async { for _ in 1..STREAM_COUNT { - let mut tx = connector.connect().await.expect("contract: connect"); + let (mut tx, _) = connector.connect().await.expect("contract: connect"); tx.write_all(&[LIVE_TAG]) .await .expect("contract: stream write"); @@ -482,7 +527,7 @@ async fn probe_independence(connector: &C, acceptor: let mut stalled = None; let mut live_seen = 0usize; for _ in 0..STREAM_COUNT { - let mut rx = acceptor + let (mut rx, _) = acceptor .accept() .await .expect("contract: later streams are accepted beside a stalled one"); @@ -544,7 +589,7 @@ async fn probe_independence_pooled(connector: &C, acc // unread wherever it lands in arrival order. let mut stalled = Vec::with_capacity(STALLED_COMPLEMENT); for _ in 0..STALLED_COMPLEMENT { - let mut tx = connector + let (mut tx, _) = connector .connect() .await .expect("contract: connect succeeds"); @@ -571,7 +616,7 @@ async fn probe_independence_pooled(connector: &C, acc })); // The one live stream must flow beside the pressured complement. let live = async { - let mut tx = connector.connect().await.expect("contract: connect"); + let (mut tx, _) = connector.connect().await.expect("contract: connect"); tx.write_all(&[LIVE_TAG]) .await .expect("contract: stream write"); @@ -594,7 +639,7 @@ async fn probe_independence_pooled(connector: &C, acc let mut held = Vec::with_capacity(STALLED_COMPLEMENT); let mut live_seen = 0usize; for _ in 0..STREAM_COUNT { - let mut rx = acceptor + let (mut rx, _) = acceptor .accept() .await .expect("contract: later streams are accepted beside stalled ones"); @@ -693,7 +738,7 @@ async fn probe_concurrency(connector: &C, acceptor: & // a capped or open-serializing supply hangs right here. let mut held = Vec::with_capacity(STREAM_COUNT); for index in 0..STREAM_COUNT { - let mut tx = connector + let (mut tx, _) = connector .connect() .await .expect("contract: a full complement of opens succeeds"); @@ -721,7 +766,7 @@ async fn probe_concurrency(connector: &C, acceptor: & let mut held: Vec> = std::iter::repeat_with(|| None).take(STREAM_COUNT).collect(); for _ in 0..STREAM_COUNT { - let mut rx = acceptor + let (mut rx, _) = acceptor .accept() .await .expect("contract: accept succeeds while the peer link lives"); @@ -821,7 +866,8 @@ async fn probe_cancellation(connector: &C, acceptor: // would deadlock the probe itself. let mut streams = Vec::with_capacity(CANCELLED_DELIVERIES); for _ in 0..CANCELLED_DELIVERIES { - streams.push(connector.connect().await.expect("contract: connect")); + let (tx, _) = connector.connect().await.expect("contract: connect"); + streams.push(tx); } let _ = connected.send(()); // The writes run concurrently: the receiver drains streams in @@ -853,7 +899,7 @@ async fn probe_cancellation(connector: &C, acceptor: // sender does not imply local acceptability (an RTT may separate // them), so the poll-drop cycles below start only once a delivery // has genuinely surfaced on this side. - let first = acceptor + let (first, _) = acceptor .accept() .await .expect("contract: accept succeeds while the peer link lives"); @@ -877,7 +923,8 @@ async fn probe_cancellation(connector: &C, acceptor: .await; match polled_once { Some(rx) => { - drain(rx.expect("contract: accept succeeds while the peer link lives")).await; + let (rx, _) = rx.expect("contract: accept succeeds while the peer link lives"); + drain(rx).await; delivered += 1; } None => { @@ -892,7 +939,7 @@ async fn probe_cancellation(connector: &C, acceptor: // Every delivery must now surface from real accepts, however many // waits were dropped above. while delivered < CANCELLED_DELIVERIES { - let rx = acceptor + let (rx, _) = acceptor .accept() .await .expect("contract: a delivery in flight across a dropped accept still arrives"); @@ -926,10 +973,10 @@ impl Clone for CountingConnector { impl Connector for CountingConnector { type Tx = C::Tx; - async fn connect(&self) -> io::Result { - let tx = self.inner.connect().await?; + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let pair = self.inner.connect().await?; self.opened.fetch_add(1, Ordering::Relaxed); - Ok(tx) + Ok(pair) } } diff --git a/src/conformance/link/tests.rs b/src/conformance/link/tests.rs index 471d54f8..a960fa26 100644 --- a/src/conformance/link/tests.rs +++ b/src/conformance/link/tests.rs @@ -18,7 +18,7 @@ use tokio::sync::mpsc; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use crate::link::{ - Acceptor, Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink, + Acceptor, Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink, STREAM_COUNT, memory, memory_with_capacity, }; use crate::testing::{Quiescence, run_to_quiescence}; @@ -49,7 +49,7 @@ fn one_byte_windows_conform() { /// loudly instead of silently. struct ReversingAcceptor { inner: A, - held: VecDeque, + held: VecDeque<(A::Rx, Done)>, /// Arrivals buffered before each reversed release. batch: usize, /// Batches of two or more released: genuine inversions. @@ -59,7 +59,7 @@ struct ReversingAcceptor { impl Acceptor for ReversingAcceptor { type Rx = A::Rx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { if let Some(held) = self.held.pop_front() { return Ok(held); } @@ -165,7 +165,7 @@ struct LossyAcceptor { impl Acceptor for LossyAcceptor { type Rx = A::Rx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { let rx = self.inner.accept().await?; // The loss window: one self-waking yield with the dequeued stream // held only in this future's state. @@ -269,7 +269,7 @@ struct MuxConnector { impl Connector for MuxConnector { type Tx = MuxTx; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(MuxTx, Done)> { let id = self.next_id.fetch_add(1, Ordering::Relaxed); // Reserve the close frame's FIFO slot up front, so the drop-time // close can be sent synchronously and is never lost to a full FIFO. @@ -283,13 +283,16 @@ impl Connector for MuxConnector { .send(MuxFrame::Open(id)) .await .map_err(|_| mux_gone())?; - Ok(MuxTx { - id, - wire: self.wire.clone(), - close: Some(close), - in_flight: None, - claimed: 0, - }) + Ok(( + MuxTx { + id, + wire: self.wire.clone(), + close: Some(close), + in_flight: None, + claimed: 0, + }, + Done::discard(), + )) } } @@ -431,19 +434,22 @@ struct MuxAcceptor { impl Acceptor for MuxAcceptor { type Rx = MuxRx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(MuxRx, Done)> { loop { { let mut state = self.demux.lock().await; if let Some(queue) = state.announced.pop_front() { - return Ok(MuxRx { - queue, - demux: self.demux.clone(), - buffer: Vec::new(), - cursor: 0, - pump: None, - ended: false, - }); + return Ok(( + MuxRx { + queue, + demux: self.demux.clone(), + buffer: Vec::new(), + cursor: 0, + pump: None, + ended: false, + }, + Done::discard(), + )); } } mux_pump(&self.demux).await?; @@ -601,18 +607,21 @@ impl AsyncWrite for CappedTx { impl Connector for CappedConnector { type Tx = CappedTx; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let permit = self .permits .clone() .acquire_owned() .await .expect("the fixture semaphore is never closed"); - let inner = self.inner.connect().await?; - Ok(CappedTx { - inner, - _permit: permit, - }) + let (inner, _) = self.inner.connect().await?; + Ok(( + CappedTx { + inner, + _permit: permit, + }, + Done::discard(), + )) } } @@ -743,12 +752,15 @@ struct WindowedConnector { impl Connector for WindowedConnector { type Tx = WindowedTx; - async fn connect(&self) -> io::Result { - let inner = self.inner.connect().await?; - Ok(WindowedTx { - inner, - window: self.window.clone(), - }) + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let (inner, _) = self.inner.connect().await?; + Ok(( + WindowedTx { + inner, + window: self.window.clone(), + }, + Done::discard(), + )) } } @@ -762,12 +774,15 @@ struct WindowedAcceptor { impl Acceptor for WindowedAcceptor { type Rx = WindowedRx; - async fn accept(&mut self) -> io::Result { - let inner = self.inner.accept().await?; - Ok(WindowedRx { - inner, - window: self.window.clone(), - }) + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { + let (inner, _) = self.inner.accept().await?; + Ok(( + WindowedRx { + inner, + window: self.window.clone(), + }, + Done::discard(), + )) } } diff --git a/src/link.rs b/src/link.rs index 12e591dc..ef2e742c 100644 --- a/src/link.rs +++ b/src/link.rs @@ -17,7 +17,7 @@ //! order the transport delivers them. //! //! Data streams are session-scoped and cheap: a session opens them lazily -//! and sparsely (up to [`STREAM_COUNT`], typically far fewer) and closes +//! and sparsely (up to [`STREAM_COUNT`], typically far fewer) and ends //! every one before it completes. Only the control stream survives into //! the next session. //! @@ -52,9 +52,15 @@ //! open at once, while [`Connector::connect`] calls arrive sparsely and //! mid-session. An instantiation must not require a full complement of //! streams, nor serialize an open behind unrelated stream progress. -//! - **Half-close.** Dropping a [`Connector::Tx`] ends that stream; the -//! peer's [`Acceptor::Rx`] then observes end-of-stream after the final -//! bytes. The control stream outlives all data streams of its session. +//! - **Completion.** A producer ends its stream by handing the +//! [`Connector::Tx`] to its [`Done`] after the final bytes, or +//! aborts by dropping it. Every written byte reaches the peer either +//! way. When the peer aborts, it surfaces at its counterparty's +//! [`Acceptor::Rx`] as end-of-stream. A completion may surface as +//! nothing at all (a reusing instantiation recovers the connection), +//! so a consumer finds the end of the data in the bytes themselves, +//! hands its `Rx` back there, and never reads past it. The control +//! stream outlives all data streams of its session. //! - **Cancellation.** A pending [`Acceptor::accept`] future may be dropped //! at any moment (session teardown is the common source, and the //! conformance suite drops them mid-session); instantiations must tolerate @@ -158,6 +164,42 @@ use tokio::sync::mpsc; /// test, so it cannot drift silently. pub const STREAM_COUNT: usize = 17; +/// Where a data-stream half goes at its stream's clean end. +/// +/// Every [`Connector::connect`] and [`Acceptor::accept`] pairs the +/// half with one of these. The session invokes it exactly at the +/// protocol's end of the data, handing the half back; a half dropped +/// anywhere else (its handle unused) is an abort. A transport that +/// reuses connections recovers them here; the rest pair every stream +/// with [`discard`](Self::discard). +pub struct Done(Box); + +impl Done { + /// For single-use streams, whose clean end is the drop. + pub fn discard() -> Self + where + Half: 'static, + { + Done(Box::new(drop)) + } + + /// Hand the completed half to `f`. + pub fn new(f: impl FnOnce(Half) + Send + 'static) -> Self { + Done(Box::new(f)) + } + + /// Release `half` at its stream's clean end. + pub fn complete(self, half: Half) { + (self.0)(half) + } +} + +impl std::fmt::Debug for Done { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Done(..)") + } +} + /// Opens outgoing unidirectional data streams for one link. /// /// Opens may arrive concurrently, through clones and through shared @@ -171,13 +213,14 @@ pub trait Connector: Clone + Send + Sync + 'static { /// The write half of one outgoing data stream. type Tx: AsyncWrite + Unpin + Send + 'static; - /// Open one outgoing unidirectional stream. + /// Open one outgoing unidirectional stream, paired with where the + /// half goes at its clean end. /// /// # Errors /// /// Fails only for transport reasons (the link is gone); the session /// treats any error as fatal to the session, never retries. - fn connect(&self) -> impl Future> + Send; + fn connect(&self) -> impl Future)>> + Send; } /// Accepts incoming unidirectional data streams for one link. @@ -190,7 +233,8 @@ pub trait Acceptor: Send { /// The read half of one incoming data stream. type Rx: AsyncRead + Unpin + Send + 'static; - /// Accept one incoming unidirectional stream. + /// Accept one incoming unidirectional stream, paired with where + /// the half goes at its clean end. /// /// Order across streams is the transport's own: the session pairs /// streams by the label written as each stream's first bytes, never by @@ -207,13 +251,13 @@ pub trait Acceptor: Send { /// teardown is the common source). A stream that was mid-delivery must /// surface from a later `accept` call; it must not be lost while the /// link stays healthy. - fn accept(&mut self) -> impl Future> + Send; + fn accept(&mut self) -> impl Future)>> + Send; } impl Acceptor for &mut A { type Rx = A::Rx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { A::accept(self).await } } @@ -549,13 +593,13 @@ pub struct MemoryConnector { impl Connector for MemoryConnector { type Tx = DuplexStream; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let (tx, rx) = tokio::io::duplex(self.capacity); self.announce .send(rx) .await .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?; - Ok(tx) + Ok((tx, Done::discard())) } } @@ -568,10 +612,11 @@ pub struct MemoryAcceptor { impl Acceptor for MemoryAcceptor { type Rx = DuplexStream; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { self.streams .recv() .await + .map(|rx| (rx, Done::discard())) .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone")) } } diff --git a/src/link/erased.rs b/src/link/erased.rs index 965bb5a2..cc88c13a 100644 --- a/src/link/erased.rs +++ b/src/link/erased.rs @@ -16,27 +16,94 @@ //! yields. use std::io; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; use futures::future::BoxFuture; -use std::sync::Arc; -use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +use super::{Acceptor, Connector, Done}; + +/// A half boxed together with its [`Done`], so completion has the +/// concrete types. +struct Bundle { + half: H, + done: Done, +} + +impl AsyncWrite for Bundle { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.half).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.half).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.half).poll_shutdown(cx) + } +} -use super::{Acceptor, Connector}; +impl AsyncRead for Bundle { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.half).poll_read(cx, buf) + } +} + +/// An erased outgoing half, completable through the box. +pub(crate) trait TxDyn: AsyncWrite + Unpin + Send { + /// Release the bundled half at its stream's clean end. + fn complete(self: Box); +} + +impl TxDyn for Bundle { + fn complete(self: Box) { + let Bundle { half, done } = *self; + done.complete(half); + } +} + +/// An erased incoming half, completable through the box. +pub(crate) trait RxDyn: AsyncRead + Unpin + Send { + /// Release the bundled half at its stream's clean end. + fn complete(self: Box); +} + +impl RxDyn for Bundle { + fn complete(self: Box) { + let Bundle { half, done } = *self; + done.complete(half); + } +} /// An owned outgoing stream half with its concrete type erased. -pub(crate) type DynTx = Box; +pub(crate) type DynTx = Box; /// An owned incoming stream half with its concrete type erased. -pub(crate) type DynRx = Box; +pub(crate) type DynRx = Box; /// Object-safe [`Connector`], for erasure behind an [`Arc`]. trait ConnectDyn: Send + Sync { - fn connect_dyn(&self) -> BoxFuture<'_, io::Result>; + fn connect_dyn(&self) -> BoxFuture<'_, io::Result<(DynTx, Done)>>; } impl ConnectDyn for C { - fn connect_dyn(&self) -> BoxFuture<'_, io::Result> { - Box::pin(async { self.connect().await.map(|tx| Box::new(tx) as DynTx) }) + fn connect_dyn(&self) -> BoxFuture<'_, io::Result<(DynTx, Done)>> { + Box::pin(async { + let (half, done) = self.connect().await?; + let erased: DynTx = Box::new(Bundle { half, done }); + Ok((erased, Done::new(TxDyn::complete))) + }) } } @@ -57,19 +124,23 @@ impl DynConnector { impl Connector for DynConnector { type Tx = DynTx; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(DynTx, Done)> { self.0.connect_dyn().await } } /// Object-safe [`Acceptor`], for erasure behind a `&mut` borrow. pub(crate) trait AcceptDyn: Send { - fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result>; + fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<(DynRx, Done)>>; } impl AcceptDyn for A { - fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result> { - Box::pin(async { self.accept().await.map(|rx| Box::new(rx) as DynRx) }) + fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<(DynRx, Done)>> { + Box::pin(async { + let (half, done) = self.accept().await?; + let erased: DynRx = Box::new(Bundle { half, done }); + Ok((erased, Done::new(RxDyn::complete))) + }) } } @@ -83,7 +154,7 @@ pub(crate) type DynAcceptor<'a> = &'a mut (dyn AcceptDyn + 'a); impl<'a, 'd> Acceptor for &'a mut (dyn AcceptDyn + 'd) { type Rx = DynRx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { // Dispatch through the object explicitly: plain method syntax would // resolve to the blanket `AcceptDyn` impl for `&mut dyn AcceptDyn` // (this very impl's `Acceptor`), recursing instead of erasing. diff --git a/src/link/routed.rs b/src/link/routed.rs index e065a357..c220182f 100644 --- a/src/link/routed.rs +++ b/src/link/routed.rs @@ -19,8 +19,16 @@ //! 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. +//! carries a single unidirectional data stream at a time. +//! +//! A data stream ends one of two ways, the link contract's completion +//! clause. Completed at its clean end, the connection outlives the +//! stream: the write half goes back to its [`Dial`] (see +//! [`Dial::recycle`]), the read half back to the router, which reads +//! the connection's next header there — so a dial that pools recycled +//! connections carries stream after stream over one connection, paying +//! connection setup once. Dropped instead, the connection closes with +//! the stream, which is how the peer observes the abort. //! //! # Driving the router //! @@ -66,10 +74,14 @@ //! - 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. +//! the caller's [`Dial`] — as does discovering that a pooled +//! connection died while idle, which surfaces as the recycled +//! stream's transport failure and heals at session granularity. +//! - Every lazy stream open pays one dial. Where the dial itself is +//! the expensive step (an authenticating transport, say), a [`Dial`] +//! that pools recycled connections pays it once per connection +//! rather than once per stream; a transport with native substreams +//! avoids the per-stream open entirely. //! //! # What the transport must provide //! @@ -158,7 +170,7 @@ use std::io; use tokio::io::{AsyncRead, AsyncWrite}; -use super::{Acceptor, Connector, Link}; +use super::{Acceptor, Connector, Done, Link}; mod endpoint; mod header; @@ -184,10 +196,12 @@ pub use stream::{StreamAcceptor, StreamConnector}; /// 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 +/// already-written bytes and then end-of-stream to the peer: an +/// aborted data stream ends by dropping its connection, and the /// peer reads to end-of-stream. A drop that discards queued bytes, -/// or never signals the peer, breaks stream teardown. +/// or never signals the peer, breaks stream teardown. (A *completed* +/// stream never drops its connection — the adapter recovers it; see +/// [`Dial::recycle`].) /// /// `tokio::net::TcpStream` satisfies both, as does anything else whose /// writes land in the transport as they are accepted. @@ -216,6 +230,27 @@ pub trait Dial: Clone + Send + Sync + 'static { /// Open one connection to the router reachable at `addr`. fn dial(&self, addr: &Self::Addr) -> impl Future> + Send; + + /// Take back a connection to `peer` whose stream completed cleanly. + /// + /// The connection rests exactly where a fresh dial's would: the + /// peer's router is reading for its next connect header, so a dial + /// that hands it out again reuses the connection for another stream + /// in place of a new connection's setup. The default drops it, + /// which suits transports whose connections are cheap; implement it + /// (a pool keyed by peer, typically) where connection setup is the + /// latency that matters. A connection whose stream failed or was + /// abandoned never comes back through here: the adapter drops it, + /// so a recycled connection is never mid-stream. + /// + /// Two cautions. Recycle runs on the session's task at the + /// stream's completion, so it must not block. And recycling + /// certifies nothing about liveness: the peer's router evicts idle + /// connections by count, so a pooled connection may be dead and + /// discovered only by the stream that draws it. + fn recycle(&self, _peer: &Self::Addr, conn: Self::Conn) { + drop(conn); + } } /// Yields inbound connections to an endpoint's router. diff --git a/src/link/routed/endpoint.rs b/src/link/routed/endpoint.rs index 39f92b8e..0ab5c777 100644 --- a/src/link/routed/endpoint.rs +++ b/src/link/routed/endpoint.rs @@ -45,8 +45,14 @@ pub struct Config { /// 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. + /// caller's [`Listen`] wrapper. Connections recovered from + /// completed streams wait here for their next header too, so a + /// deployment whose [`Dial`] pools connections sizes this past its + /// pooled idle count plus the burst of simultaneous dials it + /// expects. Eviction of an idle recovered connection is silent: no + /// invalidation reaches the dialer's pool, and the next stream + /// drawn on the dead entry fails, or hangs to the caller's session + /// timeout. Size generously. pub pending_headers: usize, } diff --git a/src/link/routed/router.rs b/src/link/routed/router.rs index bbc6e779..62d73de4 100644 --- a/src/link/routed/router.rs +++ b/src/link/routed/router.rs @@ -40,17 +40,18 @@ 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 super::{Dial, Done, Link, Listen}; use crate::link::STREAM_COUNT; /// The routing table: each live link's token, mapped to the bounded -/// queue its acceptor drains. +/// queue its acceptor drains. Each delivery carries the [`Done`] that +/// returns a completed stream's connection to the router. /// /// 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>>>; +pub(super) type Table = Arc)>>>>; /// Lock the table, riding through a poisoning panic. /// @@ -58,9 +59,10 @@ pub(super) type Table = Arc>>>; /// elsewhere cannot leave the map torn; continuing lets the surviving /// side of a panicked test observe eviction rather than a poison /// cascade. +#[allow(clippy::type_complexity)] fn entries( - table: &Mutex>>, -) -> MutexGuard<'_, HashMap>> { + table: &Mutex)>>>, +) -> MutexGuard<'_, HashMap)>>> { table .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) @@ -96,7 +98,10 @@ impl Drop for Registration { /// /// 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) { +#[allow(clippy::type_complexity)] +pub(super) fn register( + table: &Table, +) -> (Token, Registration, mpsc::Receiver<(C, Done)>) { let (sender, receiver) = mpsc::channel(STREAM_COUNT); let mut sender = Some(sender); let token = loop { @@ -127,6 +132,13 @@ where D: Dial, L: Listen, { + // Connections coming back from completed streams, to await their + // next header beside fresh arrivals. The queue is bounded by the + // same budget as the pending reads it feeds, and its send never + // blocks: a return finding it full is dropped, the eviction the + // pending bound would deal it anyway. (Declared before `pending`, + // whose futures borrow the sender.) + let (returns, mut returned) = mpsc::channel::(pending_headers); let mut pending = FuturesUnordered::new(); // Insertion-ordered abort handles for the pending reads, so the // count bound evicts oldest-first; ids reconcile the two @@ -134,29 +146,29 @@ where 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, - )); - } + let conn = tokio::select! { + accepted = listen.accept() => accepted?, + Some(conn) = returned.recv() => conn, Some(finished) = pending.next() => { if let Ok(id) = finished { order.retain(|(pending_id, _)| *pending_id != id); } + continue; } + }; + 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, &returns, id), + registration, + )); } } @@ -170,12 +182,13 @@ async fn route( dial: D, table: &Table, incoming: &mpsc::Sender>, + returns: &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; + let _ = deliver(conn, dial, table, incoming, returns).await; id } @@ -185,6 +198,7 @@ async fn deliver( dial: D, table: &Table, incoming: &mpsc::Sender>, + returns: &mpsc::Sender, ) -> io::Result<()> { match header::read::(&mut conn).await? { Header::Stream { token } => { @@ -194,7 +208,15 @@ async fn deliver( // on the dialing side, which owns the retry. return Ok(()); }; - match queue.try_send(conn) { + let returns = returns.clone(); + // A completed stream hands its connection back through this + // sender, which never blocks (the module's no-await + // discipline). A return dropped on a full queue or stopped + // router degrades to an abort, which the dialer heals. + let done = Done::new(move |conn| { + let _ = returns.try_send(conn); + }); + match queue.try_send((conn, done)) { Ok(()) => {} // A full queue proves peer misbehavior (an honest peer // never exceeds a session's complement, and the queue diff --git a/src/link/routed/stream.rs b/src/link/routed/stream.rs index 677e17d7..a38d38e4 100644 --- a/src/link/routed/stream.rs +++ b/src/link/routed/stream.rs @@ -1,4 +1,10 @@ //! One link's stream supply: dial-per-open out, routed queue in. +//! +//! Completion recovers the connection on both sides. The write half +//! goes back to its [`Dial`] through [`Dial::recycle`], and the read +//! half returns to the router, which reads its next connect header +//! there. A dropped half drops its connection instead, whose close is +//! the transport half-close the peer observes as an abort. use std::io; @@ -7,16 +13,14 @@ use tokio::sync::mpsc; use super::header::{self, Token}; use super::router::Registration; -use super::{Acceptor, Conn, Connector, Dial}; +use super::{Acceptor, Conn, Connector, Dial, Done}; -/// A routed link's [`Connector`]: every open dials one fresh -/// connection to the peer's router and labels it with the link's -/// token. +/// A routed link's [`Connector`]: every open dials one 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. +/// Whether "dials" means a fresh connection or a recovered one is the +/// [`Dial`]'s policy: a dial that pools what [`Dial::recycle`] hands +/// back pays no new connection setup for the next stream. pub struct StreamConnector { dial: D, peer: D::Addr, @@ -43,10 +47,12 @@ impl Clone for StreamConnector { impl Connector for StreamConnector { type Tx = D::Conn; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let mut conn = self.dial.dial(&self.peer).await?; conn.write_all(&header::stream_header(&self.token)).await?; - Ok(conn) + let dial = self.dial.clone(); + let peer = self.peer.clone(); + Ok((conn, Done::new(move |conn| dial.recycle(&peer, conn)))) } } @@ -58,7 +64,7 @@ impl Connector for StreamConnector { /// 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, + streams: mpsc::Receiver<(C, Done)>, /// Revokes this link's token when the acceptor drops. _registration: Registration, } @@ -66,7 +72,10 @@ pub struct StreamAcceptor { 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 { + pub(super) fn new( + streams: mpsc::Receiver<(C, Done)>, + registration: Registration, + ) -> Self { StreamAcceptor { streams, _registration: registration, @@ -77,7 +86,7 @@ impl StreamAcceptor { impl Acceptor for StreamAcceptor { type Rx = C; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { // 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 diff --git a/src/link/routed/tests.rs b/src/link/routed/tests.rs index 6ddd0740..aa11450f 100644 --- a/src/link/routed/tests.rs +++ b/src/link/routed/tests.rs @@ -1,10 +1,13 @@ +use std::collections::HashMap; use std::future::Future; use std::io; use std::pin::pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; use futures::FutureExt; use futures::future::{Either, select, try_join}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; use super::header::{self, Token}; use super::{Config, Dial, Endpoint, EndpointError, Incoming, LinkError, LinkInfo, RoutedLink}; @@ -78,11 +81,11 @@ async fn transfer( Ab: Acceptor, { let open = async { - let mut tx = opener.connector.connect().await.expect("stream opens"); + 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 rx, _) = acceptor.acceptor.accept().await.expect("stream arrives"); let mut received = Vec::new(); rx.read_to_end(&mut received) .await @@ -429,14 +432,14 @@ fn cancelled_opens_leave_the_link_usable() { // 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"); + 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 rx, _) = at_a.acceptor.accept().await.expect("stream arrives"); let mut received = Vec::new(); rx.read_to_end(&mut received) .await @@ -454,6 +457,133 @@ fn cancelled_opens_leave_the_link_usable() { }); } +/// Open one stream from `opener` to `acceptor`, move `payload` across +/// it, and end it by completion on both halves: the receiver reads +/// exactly the payload and completes there, never probing for +/// end-of-stream. +async fn transfer_completed( + opener: &Link, + acceptor: &mut Link, + payload: &[u8], +) where + Ca: Connector, + Ab: Acceptor, +{ + let open = async { + let (mut tx, done) = opener.connector.connect().await.expect("stream opens"); + tx.write_all(payload).await.expect("payload writes"); + tx.flush().await.expect("payload flushes"); + done.complete(tx); + }; + let read = async { + let (mut rx, done) = acceptor.acceptor.accept().await.expect("stream arrives"); + let mut received = vec![0u8; payload.len()]; + rx.read_exact(&mut received) + .await + .expect("the completed stream delivers its bytes"); + done.complete(rx); + received + }; + let ((), received) = futures::join!(open, read); + assert_eq!(received, payload); +} + +/// A dialer that pools recycled connections per peer and counts the +/// fresh dials it performs: how the reuse tests observe which streams +/// paid for a connection. +#[derive(Clone)] +struct PoolingDial { + inner: MemoryDial, + pool: Arc>>>, + fresh: Arc, +} + +impl PoolingDial { + fn new(net: &MemoryNet) -> Self { + PoolingDial { + inner: net.dial(), + pool: Arc::default(), + fresh: Arc::default(), + } + } + + fn fresh_dials(&self) -> usize { + self.fresh.load(Ordering::Relaxed) + } +} + +impl Dial for PoolingDial { + type Addr = MemoryName; + type Conn = DuplexStream; + + async fn dial(&self, addr: &MemoryName) -> io::Result { + let pooled = self + .pool + .lock() + .expect("pool lock") + .get_mut(&addr.0) + .and_then(Vec::pop); + if let Some(conn) = pooled { + return Ok(conn); + } + self.fresh.fetch_add(1, Ordering::Relaxed); + self.inner.dial(addr).await + } + + fn recycle(&self, peer: &MemoryName, conn: DuplexStream) { + self.pool + .lock() + .expect("pool lock") + .entry(peer.0.clone()) + .or_default() + .push(conn); + } +} + +/// A completed stream's connection carries the next stream instead of +/// a fresh dial. +/// +/// The write half hands it back through [`Dial::recycle`], the read +/// half returns it to the router for its next connect header, and the +/// recycled connection routes exactly as a dialed one would — in both +/// directions of the link. +#[test] +fn completed_streams_reuse_their_connection() { + pollster::block_on(async { + let net = MemoryNet::new(); + let dial = PoolingDial::new(&net); + let a_name = MemoryName::new("a"); + let (a, mut a_incoming, a_router) = + Endpoint::new(net.listen(&a_name), a_name, dial.clone(), Config::default()) + .expect("a valid construction"); + let b_name = MemoryName::new("b"); + let (b, _b_incoming, b_router) = + Endpoint::new(net.listen(&b_name), b_name, dial.clone(), Config::default()) + .expect("a valid construction"); + drive(routers(a_router, b_router), async { + let (linked, arrival) = + futures::join!(b.link(MemoryName::new("a")), a_incoming.accept()); + let at_b = linked.expect("establishment succeeds"); + let (_info, mut at_a) = arrival.expect("the router delivers the link"); + let established = dial.fresh_dials(); + + // Forward: the first stream dials, the second rides its + // recycled connection. + transfer_completed(&at_b, &mut at_a, b"paid by a dial").await; + transfer_completed(&at_b, &mut at_a, b"rides recycled").await; + assert_eq!(dial.fresh_dials(), established + 1); + + // Reverse: the accepting side's dial-back pools the same way. + let mut at_b = at_b; + transfer_completed(&at_a, &mut at_b, b"paid by a dial").await; + transfer_completed(&at_a, &mut at_b, b"rides recycled").await; + assert_eq!(dial.fresh_dials(), established + 2); + drop((a, b)); + }) + .await; + }); +} + /// A dialer over socket addresses for construction tests; endpoint /// construction never dials, so its `dial` is unreachable. #[derive(Clone)] diff --git a/src/link/tests.rs b/src/link/tests.rs index 7cdccf9f..b7ef057b 100644 --- a/src/link/tests.rs +++ b/src/link/tests.rs @@ -48,12 +48,12 @@ fn connect_delivers_an_ordered_half_closing_stream() { let (a, mut b) = memory(); run_to_quiescence(async { let send = async { - let mut tx = a.connector.connect().await.unwrap(); + let (mut tx, _) = a.connector.connect().await.unwrap(); tx.write_all(b"hello").await.unwrap(); drop(tx); }; let receive = async { - let mut rx = b.acceptor.accept().await.unwrap(); + let (mut rx, _) = b.acceptor.accept().await.unwrap(); let mut bytes = Vec::new(); rx.read_to_end(&mut bytes).await.unwrap(); assert_eq!(bytes, b"hello"); @@ -70,18 +70,18 @@ fn a_stalled_stream_does_not_couple_its_siblings() { let (a, mut b) = memory_with_capacity(4); run_to_quiescence(async { let send = async { - let mut stalled = a.connector.connect().await.unwrap(); + let (mut stalled, _) = a.connector.connect().await.unwrap(); // Fill the stalled stream's bounded buffer to its brim; its // reader never drains it, so more writes would block. stalled.write_all(&[0u8; 4]).await.unwrap(); - let mut live = a.connector.connect().await.unwrap(); + let (mut live, _) = a.connector.connect().await.unwrap(); live.write_all(b"live").await.unwrap(); drop(live); stalled }; let receive = async { let _stalled = b.acceptor.accept().await.unwrap(); - let mut rx = b.acceptor.accept().await.unwrap(); + let (mut rx, _) = b.acceptor.accept().await.unwrap(); let mut bytes = Vec::new(); rx.read_to_end(&mut bytes).await.unwrap(); assert_eq!(bytes, b"live"); @@ -98,12 +98,12 @@ fn stream_writes_block_on_their_own_reader() { let (a, mut b) = memory_with_capacity(2); run_to_quiescence(async { let send = async { - let mut tx = a.connector.connect().await.unwrap(); + let (mut tx, _) = a.connector.connect().await.unwrap(); tx.write_all(b"abcdef").await.unwrap(); drop(tx); }; let receive = async { - let mut rx = b.acceptor.accept().await.unwrap(); + let (mut rx, _) = b.acceptor.accept().await.unwrap(); let mut bytes = Vec::new(); rx.read_to_end(&mut bytes).await.unwrap(); assert_eq!(bytes, b"abcdef"); diff --git a/src/testing/transport.rs b/src/testing/transport.rs index ada5418b..b24799a8 100644 --- a/src/testing/transport.rs +++ b/src/testing/transport.rs @@ -13,6 +13,8 @@ use std::{ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use crate::link::Done; + /// Which endpoint owns an observed transport operation. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum Side { @@ -51,8 +53,8 @@ pub enum FaultUnit { /// A read fault fires in place of the first *payload-bearing* read beyond /// the prefix: end-of-stream probes pass through untouched. This keeps /// "would the fault fire?" a function of the clean run's successful-read -/// counts alone — the link world checks for end-of-stream after every -/// stream's end control, so attempts-after-last-success are structural and +/// counts alone — the control stream reads for a clean goodbye at every +/// session boundary, so attempts-after-last-success are structural and /// must not trip an operations-counted fault. /// /// Stream-supply faults follow the same discipline, counted in operations @@ -561,7 +563,7 @@ impl Clone for AdversarialConnector { impl crate::link::Connector for AdversarialConnector { type Tx = AdversarialWrite; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { // The fault fires in place of the call: a healthy supply's connects // always succeed, so the clean run's success count is also its call // count and "would the fault fire?" remains a function of it. @@ -573,13 +575,17 @@ impl crate::link::Connector for AdversarialConnector< { return Err(error); } - let tx = self.inner.connect().await?; + let (tx, done) = self.inner.connect().await?; self.state .lock() .expect("transport state lock") .report .connects += 1; - Ok(wrap_write(tx, self.state.clone())) + // Completion unwraps the adversity and passes the half through. + Ok(( + wrap_write(tx, self.state.clone()), + Done::new(move |wrapped: AdversarialWrite| done.complete(wrapped.inner)), + )) } } @@ -593,7 +599,7 @@ pub struct AdversarialAcceptor { impl crate::link::Acceptor for AdversarialAcceptor { type Rx = AdversarialRead; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { // An already-injected accept fault keeps failing without consuming // further arrivals. { @@ -604,7 +610,7 @@ impl crate::link::Acceptor for AdversarialAcceptor return Err(error); } } - let rx = self.inner.accept().await?; + let (rx, done) = self.inner.accept().await?; let mut state = self.state.lock().expect("transport state lock"); // The fault fires in place of a successful accept, so the final // forever-pending accept a session parks on against an honest peer @@ -617,11 +623,15 @@ impl crate::link::Acceptor for AdversarialAcceptor } state.report.accepts += 1; drop(state); - Ok(AdversarialRead { - inner: rx, - state: self.state.clone(), - delay: None, - }) + // Completion unwraps the adversity and passes the half through. + Ok(( + AdversarialRead { + inner: rx, + state: self.state.clone(), + delay: None, + }, + Done::new(move |wrapped: AdversarialRead| done.complete(wrapped.inner)), + )) } } @@ -659,7 +669,7 @@ const REORDER_PATIENCE: u8 = 32; /// does not depend on the public `conformance` feature. pub struct ReorderingAcceptor { inner: A, - held: VecDeque, + held: VecDeque<(A::Rx, Done)>, /// Arrivals buffered before each reversed release. batch: usize, /// Batches of two or more released: genuine inversions. @@ -669,7 +679,7 @@ pub struct ReorderingAcceptor { impl crate::link::Acceptor for ReorderingAcceptor { type Rx = A::Rx; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { if let Some(held) = self.held.pop_front() { return Ok(held); } diff --git a/src/tests.rs b/src/tests.rs index 052203fc..9e6d33ce 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -297,12 +297,15 @@ struct FusedConnector { impl Connector for FusedConnector { type Tx = Fuse; - async fn connect(&self) -> std::io::Result { - let inner = self.inner.connect().await?; - Ok(Fuse { - inner, - remaining: Arc::clone(&self.remaining), - }) + async fn connect(&self) -> std::io::Result<(Self::Tx, crate::link::Done)> { + let (inner, _) = self.inner.connect().await?; + Ok(( + Fuse { + inner, + remaining: Arc::clone(&self.remaining), + }, + crate::link::Done::discard(), + )) } } diff --git a/src/tree/mirror/streaming/remote/codec/decode/async_io.rs b/src/tree/mirror/streaming/remote/codec/decode/async_io.rs index b4cb5f7c..4a0a80f0 100644 --- a/src/tree/mirror/streaming/remote/codec/decode/async_io.rs +++ b/src/tree/mirror/streaming/remote/codec/decode/async_io.rs @@ -28,6 +28,13 @@ impl FrameRead { pub fn new(speaker: Speaker, read: R) -> Self { Self { speaker, read } } + + /// Recover the transport half. The reader buffers nothing (every + /// read is exact), so between frames the half rests exactly at a + /// frame boundary. + pub fn into_inner(self) -> R { + self.read + } } impl FrameRead { diff --git a/src/tree/mirror/streaming/remote/codec/encode/async_io.rs b/src/tree/mirror/streaming/remote/codec/encode/async_io.rs index b7ed223e..f8c5dc4d 100644 --- a/src/tree/mirror/streaming/remote/codec/encode/async_io.rs +++ b/src/tree/mirror/streaming/remote/codec/encode/async_io.rs @@ -25,8 +25,9 @@ impl FrameWrite { Self { speaker, write } } - /// Recover the transport writer without buffered frame state. - #[cfg(test)] + /// Recover the transport writer without buffered frame state. Every + /// frame is flushed as it is written, so between frames the writer + /// rests exactly at a frame boundary. pub fn into_inner(self) -> W { self.write } diff --git a/src/tree/mirror/streaming/remote/proxy/tests/harness.rs b/src/tree/mirror/streaming/remote/proxy/tests/harness.rs index cadbdc8d..4f633a1c 100644 --- a/src/tree/mirror/streaming/remote/proxy/tests/harness.rs +++ b/src/tree/mirror/streaming/remote/proxy/tests/harness.rs @@ -14,7 +14,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::ReadBuf; -use crate::link::{Acceptor, Connector, Link, MemoryLink, memory_with_capacity}; +use crate::link::{Acceptor, Connector, Done, Link, MemoryLink, memory_with_capacity}; use crate::testing::{IoPlan, IoReportHandle, IoSide, wrap_link}; use crate::tree::mirror::framing::{GREETING_WORD_LEN, LENGTH_HEADER_LEN}; use crate::tree::mirror::streaming::window::WindowConfig; @@ -244,9 +244,9 @@ pub struct ScriptedConnector { impl Connector for ScriptedConnector { type Tx = ScriptedWrite; - async fn connect(&self) -> io::Result { - let tx = self.inner.connect().await?; - Ok(ScriptedWrite::new(tx, self.script.clone())) + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let (tx, _) = self.inner.connect().await?; + Ok((ScriptedWrite::new(tx, self.script.clone()), Done::discard())) } } diff --git a/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs b/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs index e961056e..68dd500c 100644 --- a/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs +++ b/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs @@ -182,10 +182,16 @@ fn duplicated_reply_is_rejected_as_unasked() { assert!(right_result.is_err()); } -/// Duplicating a stream-end frame is rejected as traffic after closure rather -/// than being mistaken for a second clean end. +/// Bytes past a stream's end control belong to the transport, not the +/// session: a duplicated stream-end frame is never read, and the session +/// completes as if it were absent. +/// +/// The receiver completes its transport half exactly at the end control +/// (the link contract's completion clause), so trailing bytes are left +/// where they lie — on a reusing link they would be the next stream's +/// connect header — rather than parsed as protocol. #[test] -fn duplicate_stream_end_is_rejected_by_the_session() { +fn bytes_past_the_stream_end_are_never_read() { const STREAM_END_STATE: u8 = 9; for corrupt_left in [false, true] { @@ -200,13 +206,12 @@ fn duplicate_stream_end_is_rejected_by_the_session() { corrupt_left.then(|| script.clone()), (!corrupt_left).then(|| script.clone()), )) - .expect("duplicate stream end must terminate both sessions"); + .expect("sessions complete despite the trailing frame"); assert!(script.fired(), "no stream-end frame reached the mutator"); - assert!(matches!( - receiving_error(corrupt_left, &left_result, &right_result), - RemoteError::Stream(StreamError::AfterEnd { .. }) - )); - assert!(left_result.is_err()); - assert!(right_result.is_err()); + assert!(left_result.is_ok(), "left session failed: {left_result:?}"); + assert!( + right_result.is_ok(), + "right session failed: {right_result:?}" + ); } } diff --git a/src/tree/mirror/streaming/remote/proxy/work/tests.rs b/src/tree/mirror/streaming/remote/proxy/work/tests.rs index e2972a14..c83bf35c 100644 --- a/src/tree/mirror/streaming/remote/proxy/work/tests.rs +++ b/src/tree/mirror/streaming/remote/proxy/work/tests.rs @@ -231,7 +231,8 @@ fn queued_supply_closed_outranks_a_selected_consequence_at_stream_granularity() // A reporter that needed the supply: publishes SupplyClosed at stream // granularity, then parks. Spawned first so it publishes in the same // wave the consequence resolves. - let (claim_send, claim_receive) = oneshot::channel::(); + let (claim_send, claim_receive) = + oneshot::channel::<(DuplexStream, crate::link::Done)>(); drop(claim_send); let mut receiver: StreamReceiver = StreamReceiver::new( claim_receive, @@ -295,7 +296,8 @@ fn published_stream_error_preempts_a_parked_protocol() { // A receiver whose claim sender is already gone: its first poll reports // `SupplyClosed` to the error route and parks, never resolving — the // pump driving it is the parked reporter the select must not wait for. - let (claim_send, claim_receive) = oneshot::channel::(); + let (claim_send, claim_receive) = + oneshot::channel::<(DuplexStream, crate::link::Done)>(); drop(claim_send); let mut receiver: StreamReceiver = StreamReceiver::new( claim_receive, diff --git a/src/tree/mirror/streaming/remote/streams.rs b/src/tree/mirror/streaming/remote/streams.rs index 3bb35001..1b2019a0 100644 --- a/src/tree/mirror/streaming/remote/streams.rs +++ b/src/tree/mirror/streaming/remote/streams.rs @@ -47,7 +47,7 @@ use tokio::{ sync::{mpsc, oneshot}, }; -use crate::link::{Acceptor, Connector}; +use crate::link::{Acceptor, Connector, Done}; use crate::tree::mirror::streaming::stats::{CountedRead, CountedWrite, Recorder}; use crate::tree::mirror::streaming::tasks::cancelled; @@ -126,7 +126,7 @@ pub struct StreamSender { enum SendState { Unopened, - Open(FrameWrite>), + Open(FrameWrite>, Done), } impl StreamSender { @@ -161,13 +161,23 @@ impl StreamSender { /// End this logical stream after all of its replies, if it ever opened. /// - /// Dropping the transport half afterward is the transport-level - /// half-close; the explicit end control before it distinguishes a - /// completed stream from one truncated mid-reply. + /// The explicit end control distinguishes a completed stream from one + /// truncated mid-reply. The transport half is handed to its [`Done`] + /// right behind it, resting at the frame boundary; failure paths drop + /// the half instead, the contract's abort. pub async fn finish(mut self) -> Result<(), SendError> { match self.state { SendState::Unopened => Ok(()), - SendState::Open(_) => self.write(Frame::End(End::Stream)).await, + SendState::Open(..) => { + self.write(Frame::End(End::Stream)).await?; + let SendState::Open(write, done) = + std::mem::replace(&mut self.state, SendState::Unopened) + else { + unreachable!("the open state was just written through"); + }; + done.complete(write.into_inner().into_inner()); + Ok(()) + } } } @@ -175,9 +185,9 @@ impl StreamSender { async fn write(&mut self, frame: Frame) -> Result<(), SendError> { let stream = self.stream; let write = match &mut self.state { - SendState::Open(write) => write, + SendState::Open(write, _) => write, state @ SendState::Unopened => { - let mut tx = + let (mut tx, done) = self.connector .connect() .await @@ -191,11 +201,11 @@ impl StreamSender { origin: Origin::stream(self.speaker, stream), source, })?; - *state = SendState::Open(FrameWrite::new( - self.speaker, - CountedWrite::new(tx, self.stats.clone()), - )); - let SendState::Open(write) = state else { + *state = SendState::Open( + FrameWrite::new(self.speaker, CountedWrite::new(tx, self.stats.clone())), + done, + ); + let SendState::Open(write, _) = state else { unreachable!("the open state was just stored"); }; write @@ -254,9 +264,6 @@ pub enum StreamError { /// The transport stream ended before its explicit end control. #[error("{origin}: transport stream ended before its end control")] Truncated { origin: Origin }, - /// The peer sent more frames after ending its logical stream. - #[error("{origin}: peer sent a frame after ending the logical stream")] - AfterEnd { origin: Origin }, /// The stream supply failed before an awaited stream was delivered. /// /// `source` carries the supply's own transport failure when the session @@ -287,7 +294,7 @@ pub struct StreamReceiver { } struct ReceiverStart { - claim: oneshot::Receiver, + claim: oneshot::Receiver<(Rx, Done)>, /// The remote role whose direction this stream carries. speaker: Speaker, stream: Stream, @@ -306,7 +313,7 @@ where { /// Bind one incoming logical stream to its claim slot. pub fn new( - claim: oneshot::Receiver, + claim: oneshot::Receiver<(Rx, Done)>, speaker: Speaker, stream: Stream, route: ErrorRoute, @@ -384,7 +391,7 @@ where /// Every failure path publishes to the session error route and parks: the /// consumer never observes a truncated stream as a clean end. fn read_frames( - claim: oneshot::Receiver, + claim: oneshot::Receiver<(Rx, Done)>, speaker: Speaker, stream: Stream, route: ErrorRoute, @@ -395,7 +402,7 @@ where T: BorshDeserialize + Send + Sync + 'static, { stream! { - let Ok(rx) = claim.await else { + let Ok((rx, done)) = claim.await else { // The claim slot is gone: the link's stream supply failed before // the peer's stream for this level arrived. This is the one // consumer that provably needed it, so the report comes from @@ -436,26 +443,19 @@ where }; if matches!(frame, Frame::End(End::Stream)) { // The lifecycle control is consumed here; the consumer sees - // only complete replies followed by a clean end. The sender - // half-closes immediately after this control, so requiring - // end-of-stream costs no waiting against an honest peer and - // catches one that keeps talking past its own end. - match read.frame::().await { - Ok(None) => break, - Ok(Some(_)) => { - route.report(StreamError::AfterEnd { - origin: Origin::stream(speaker, stream), - }); - cancelled().await - } - Err(error) => { - route.report(StreamError::Decode(error)); - cancelled().await - } - } + // only complete replies followed by a clean end. + break; } yield frame; } + // The end control is exactly where the data ends, so the transport + // half rests at the link contract's clean boundary: hand it back + // there, never reading past it. The hand-back is a framing + // judgment, not a protocol one: a stream the consumer goes on to + // rule invalid has still completed here. Every failure path above + // parks instead, dropping the half at teardown, which is the + // contract's abort. + done.complete(read.into_inner().into_inner()); } } @@ -565,17 +565,17 @@ pub fn error_route() -> (ErrorRoute, FirstStreamError) { /// The claim slots the accept driver delivers incoming streams into. pub struct ClaimSlots { - slots: [Option>; STREAM_COUNT], + slots: [Option)>>; STREAM_COUNT], } /// The claim receivers the session's typed states take streams from. pub struct Claims { - slots: [Option>; STREAM_COUNT], + slots: [Option)>>; STREAM_COUNT], } impl Claims { /// Take the sole claim for `stream`. - pub fn take(&mut self, stream: Stream) -> oneshot::Receiver { + pub fn take(&mut self, stream: Stream) -> oneshot::Receiver<(Rx, Done)> { self.slots[usize::from(stream.index())] .take() .expect("each incoming logical stream is claimed exactly once") @@ -677,7 +677,7 @@ impl AcceptDriver { /// driver serves belongs to the same peer, so there is no bystander to /// protect with a concurrent label read. async fn accept_one(&mut self) -> Result<(), AcceptFate> { - let mut rx = self + let (mut rx, done) = self .acceptor .accept() .await @@ -705,7 +705,7 @@ impl AcceptDriver { .ok_or(AcceptError::Duplicate { origin: Origin::stream(self.speaker, stream), })?; - slot.send(rx).map_err(|_| { + slot.send((rx, done)).map_err(|_| { // The claim's consumer already finished without asking anything // at this level, so whatever this stream carries was never // asked for. diff --git a/src/tree/mirror/streaming/remote/streams/tests.rs b/src/tree/mirror/streaming/remote/streams/tests.rs index 5453cd4d..f9c94698 100644 --- a/src/tree/mirror/streaming/remote/streams/tests.rs +++ b/src/tree/mirror/streaming/remote/streams/tests.rs @@ -222,7 +222,7 @@ async fn raw_labeled( connector: &crate::link::MemoryConnector, stream: Stream, ) -> FrameWrite { - let mut tx = connector.connect().await.expect("stream opens"); + let (mut tx, _) = connector.connect().await.expect("stream opens"); tx.write_all(&label(EPOCH, stream)) .await .expect("label writes"); @@ -362,44 +362,6 @@ fn truncated_stream_is_reported_not_ended() { ); } -/// A frame arriving after the explicit end control is reported as -/// `AfterEnd`: a peer that keeps talking past its own end is caught. -/// -/// After `End(Stream)` the receiver requires transport end-of-stream -/// before ending cleanly: the double-checked stream end, so trailing -/// frames are a reportable protocol violation, never silently dropped. -#[test] -fn frames_after_the_end_control_are_reported() { - let (a, mut b) = memory(); - let stream = Stream::new(5).expect("stream 5 exists"); - let error = run_to_quiescence(async { - let send = async { - // An honest `StreamSender` half-closes right after its end - // control, so the frame beyond it comes from the raw writer. - let mut write = raw_labeled(&a.connector, stream).await; - write - .frame(&(stream, Frame::::End(End::Stream))) - .await - .expect("the end control writes"); - write - .frame(&(stream, Frame::::End(End::Reply))) - .await - .expect("the frame beyond the end writes"); - }; - let receive = first_reported_error(&mut b.acceptor, stream, &[]); - join(send, receive).await.1.0 - }) - .expect("after-end detection resolves"); - // Pinned in full, origin included, like the truncation test above. - assert!( - matches!( - error, - StreamError::AfterEnd { origin } if origin == Origin::stream(Speaker::Initiator, stream) - ), - "unexpected stream error: {error:?}", - ); -} - /// A second transport stream bearing an already-delivered label is /// rejected by the accept driver as `Duplicate`. /// @@ -448,7 +410,7 @@ fn accept_driver_rejects_unknown_stream_index() { let (a, mut b) = memory(); run_to_quiescence(async { let send = async { - let mut tx = a.connector.connect().await.expect("stream opens"); + let (mut tx, _) = a.connector.connect().await.expect("stream opens"); // One past the last logical stream: no claim slot can exist. tx.write_all(&[EPOCH, Stream::COUNT]) .await diff --git a/src/tree/mirror/streaming/stats.rs b/src/tree/mirror/streaming/stats.rs index 8e2cd71c..b28cf2a2 100644 --- a/src/tree/mirror/streaming/stats.rs +++ b/src/tree/mirror/streaming/stats.rs @@ -247,6 +247,11 @@ impl CountedWrite { pub fn new(inner: W, recorder: Recorder) -> Self { Self { inner, recorder } } + + /// Recover the transport half; the counter holds no bytes. + pub fn into_inner(self) -> W { + self.inner + } } impl AsyncWrite for CountedWrite { @@ -291,6 +296,11 @@ impl CountedRead { pub fn new(inner: R, recorder: Recorder) -> Self { Self { inner, recorder } } + + /// Recover the transport half; the counter holds no bytes. + pub fn into_inner(self) -> R { + self.inner + } } impl AsyncRead for CountedRead { diff --git a/tests/common/fault.rs b/tests/common/fault.rs index 5dfd9cdd..5ab74306 100644 --- a/tests/common/fault.rs +++ b/tests/common/fault.rs @@ -31,7 +31,7 @@ use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use rumors::link::{ - Acceptor, Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink, + Acceptor, Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink, }; use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf}; @@ -149,15 +149,19 @@ impl Clone for FaultConnector { impl Connector for FaultConnector { type Tx = Fuse; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { // A dead write direction cannot open new streams either; this is // what lets a cut exercise `SendError::Connect` deterministically // instead of only through real-transport races. if *self.budget.lock().expect("write budget lock") == 0 { return Err(write_severed()); } - let tx = self.inner.connect().await?; - Ok(Fuse::new(tx, self.budget.clone())) + let (tx, done) = self.inner.connect().await?; + // Completion unwraps the fuse and passes the half through. + Ok(( + Fuse::new(tx, self.budget.clone()), + Done::new(move |fuse: Fuse| done.complete(fuse.inner)), + )) } } @@ -171,15 +175,19 @@ pub struct FaultAcceptor { impl Acceptor for FaultAcceptor { type Rx = Cut; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { // A dead read direction cannot deliver new streams either; this // reaches the session's deferred supply-failure path (the parked // accept driver) deterministically rather than only via races. if *self.budget.lock().expect("read budget lock") == 0 { return Err(read_severed()); } - let rx = self.inner.accept().await?; - Ok(Cut::new(rx, self.budget.clone())) + let (rx, done) = self.inner.accept().await?; + // Completion unwraps the cut and passes the half through. + Ok(( + Cut::new(rx, self.budget.clone()), + Done::new(move |cut: Cut| done.complete(cut.inner)), + )) } } diff --git a/tests/common/gossip_snapshot.rs b/tests/common/gossip_snapshot.rs index 186f50d6..9825c1af 100644 --- a/tests/common/gossip_snapshot.rs +++ b/tests/common/gossip_snapshot.rs @@ -42,7 +42,7 @@ use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use borsh::{BorshDeserialize, BorshSerialize}; -use rumors::link::{Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector}; +use rumors::link::{Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector}; use rumors::{ Rumors, testing::{LinkCapture, render_v2_capture}, @@ -155,11 +155,11 @@ pub struct CaptureConnector { impl Connector for CaptureConnector { type Tx = CaptureWrite; - async fn connect(&self) -> io::Result { - let tx = self.inner.connect().await?; + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let (tx, _) = self.inner.connect().await?; let buffer = StreamBuf::default(); self.streams.lock().unwrap().push(buffer.clone()); - Ok(CaptureWrite { inner: tx, buffer }) + Ok((CaptureWrite { inner: tx, buffer }, Done::discard())) } } diff --git a/tests/common/routed_tcp.rs b/tests/common/routed_tcp.rs index c448cb7e..f2e16e59 100644 --- a/tests/common/routed_tcp.rs +++ b/tests/common/routed_tcp.rs @@ -6,8 +6,10 @@ //! the OS floor. Socket policy stops here: the adapter above sees only //! byte streams. +use std::collections::HashMap; use std::io; use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; use rumors::link::routed::{Dial, Listen}; use tokio::net::{TcpListener, TcpSocket, TcpStream}; @@ -40,6 +42,40 @@ impl Dial for TcpDial { } } +/// A TCP dialer pooling recycled connections per peer, so completed +/// streams ride recycled connections instead of fresh dials. +#[derive(Clone, Default)] +pub struct PoolingTcpDial { + pool: Arc>>>, +} + +impl Dial for PoolingTcpDial { + type Addr = SocketAddr; + type Conn = TcpStream; + + async fn dial(&self, addr: &SocketAddr) -> io::Result { + let pooled = self + .pool + .lock() + .expect("pool lock") + .get_mut(addr) + .and_then(Vec::pop); + match pooled { + Some(conn) => Ok(conn), + None => TcpStream::connect(*addr).await, + } + } + + fn recycle(&self, peer: &SocketAddr, conn: TcpStream) { + self.pool + .lock() + .expect("pool lock") + .entry(*peer) + .or_default() + .push(conn); + } +} + /// Accepts one process's inbound routed-link connections. pub struct TcpListen(TcpListener); diff --git a/tests/common/tcp.rs b/tests/common/tcp.rs index 7ef3cfc4..84756209 100644 --- a/tests/common/tcp.rs +++ b/tests/common/tcp.rs @@ -24,7 +24,7 @@ use std::io; use std::net::SocketAddr; use std::sync::Arc; -use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT}; +use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{ TcpListener, TcpSocket, TcpStream, @@ -108,7 +108,7 @@ pub struct TcpConnector(Arc); impl Connector for TcpConnector { type Tx = OwnedWriteHalf; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let stream = match self.0.send_buffer { None => TcpStream::connect(self.0.peer).await?, Some(send) => { @@ -122,7 +122,7 @@ impl Connector for TcpConnector { // dropping it later half-closes toward the peer. The unread half is // dropped now; the peer never writes on this socket. drop(read); - Ok(write) + Ok((write, Done::discard())) } } @@ -132,12 +132,12 @@ pub struct TcpAcceptor(TcpListener); impl Acceptor for TcpAcceptor { type Rx = OwnedReadHalf; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { let (stream, _) = self.0.accept().await?; let (read, write) = stream.into_split(); // Half-close our unused direction immediately; the peer never reads // this socket, so the shutdown is invisible to it. drop(write); - Ok(read) + Ok((read, Done::discard())) } } diff --git a/tests/dispute_wire.rs b/tests/dispute_wire.rs index 6f8106fb..91744ea8 100644 --- a/tests/dispute_wire.rs +++ b/tests/dispute_wire.rs @@ -32,7 +32,7 @@ use std::task::{Context, Poll}; use borsh::{BorshDeserialize, BorshSerialize}; use rand::rngs::SmallRng; use rand::{RngCore, SeedableRng}; -use rumors::link::{Connector, Link, LinkParts, MemoryLink}; +use rumors::link::{Connector, Done, Link, LinkParts, MemoryLink}; use rumors::testing::{dispute_overhead_bytes, envelope_and_wire_bytes}; use rumors::{Peer, Rumors}; use tokio::io::AsyncWrite; @@ -123,11 +123,15 @@ struct CountingConnector { impl Connector for CountingConnector { type Tx = CountingWrite; - async fn connect(&self) -> io::Result { - Ok(CountingWrite { - inner: self.inner.connect().await?, - written: self.written.clone(), - }) + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { + let (inner, _) = self.inner.connect().await?; + Ok(( + CountingWrite { + inner, + written: self.written.clone(), + }, + Done::discard(), + )) } } diff --git a/tests/hop_trace.rs b/tests/hop_trace.rs index 168bfee1..38b50d4a 100644 --- a/tests/hop_trace.rs +++ b/tests/hop_trace.rs @@ -39,7 +39,7 @@ use std::time::Duration; use rand::rngs::SmallRng; use rand::seq::SliceRandom; use rand::{RngCore, SeedableRng}; -use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT}; +use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT}; use rumors::{DEFAULT_SYNC_MEMORY_BUDGET, Key, Peer, Protocol, Rumors}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::sync::mpsc; @@ -234,7 +234,7 @@ struct TracedConnector { impl Connector for TracedConnector { type Tx = TracedWriter; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { let (tx, rx) = delayed_pipe(CAPACITY, DELAY); let pipe = PipeId { side: self.side, @@ -248,11 +248,14 @@ impl Connector for TracedConnector { }) .await .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?; - Ok(TracedWriter { - inner: tx, - pipe, - trace: self.trace.clone(), - }) + Ok(( + TracedWriter { + inner: tx, + pipe, + trace: self.trace.clone(), + }, + Done::discard(), + )) } } @@ -264,10 +267,11 @@ struct TracedAcceptor { impl Acceptor for TracedAcceptor { type Rx = TracedReader; - async fn accept(&mut self) -> io::Result { + async fn accept(&mut self) -> io::Result<(Self::Rx, Done)> { self.streams .recv() .await + .map(|rx| (rx, Done::discard())) .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone")) } } diff --git a/tests/routed_link.rs b/tests/routed_link.rs index b248f0bb..9e0c192f 100644 --- a/tests/routed_link.rs +++ b/tests/routed_link.rs @@ -28,7 +28,7 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; use tokio::time::timeout; -use crate::common::routed_tcp::{TcpDial, TcpListen}; +use crate::common::routed_tcp::{PoolingTcpDial, TcpDial, TcpListen}; use crate::common::wire::bootstrap_fork_async; /// Bound on one whole suite run; loopback and in-memory checks finish @@ -163,6 +163,58 @@ async fn conforms_over_tcp_at_minimal_buffers_swapped() { tcp_conformance(Some(MINIMAL_BUFFER_REQUEST), false).await; } +/// Mint a routed-link pair whose shared dialer pools recycled +/// connections, so the suite's completed streams ride recycled +/// connections wherever a pooled one is available. +async fn pooled_tcp_pair( + dialer_first: bool, +) -> (RoutedLink, RoutedLink) { + let dial = PoolingTcpDial::default(); + let pooled_endpoint = async |dial: PoolingTcpDial| { + let (listen, addr) = TcpListen::bind(None) + .await + .expect("bind a loopback listener"); + let (endpoint, incoming, router) = Endpoint::new(listen, addr, dial, Config::default()) + .expect("an unscoped loopback name is routable"); + tokio::spawn(router); + (endpoint, incoming, addr) + }; + let (_a, mut a_incoming, a_addr) = pooled_endpoint(dial.clone()).await; + let (b, _b_incoming, _b_addr) = pooled_endpoint(dial).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) + } +} + +/// The suite holds when completed streams ride recycled connections: +/// pooling is invisible to every clause the contract states. +#[tokio::test] +async fn conforms_over_tcp_with_a_pooling_dial() { + timeout( + SUITE_TIMEOUT, + rumors::conformance::link::check(async || pooled_tcp_pair(true).await), + ) + .await + .expect("conformance suite ran past its liveness bound"); +} + +/// The pooling suite with the pair's seats swapped, as for the plain +/// TCP variants. +#[tokio::test] +async fn conforms_over_tcp_with_a_pooling_dial_swapped() { + timeout( + SUITE_TIMEOUT, + rumors::conformance::link::check(async || pooled_tcp_pair(false).await), + ) + .await + .expect("conformance suite ran past its liveness bound"); +} + /// 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. diff --git a/tests/session_stats.rs b/tests/session_stats.rs index 8585141a..ae3dbed1 100644 --- a/tests/session_stats.rs +++ b/tests/session_stats.rs @@ -19,7 +19,7 @@ use std::task::{Context, Poll}; use futures::StreamExt; use proptest::prelude::*; -use rumors::link::{Connector, Link, LinkParts, MemoryLink}; +use rumors::link::{Connector, Done, Link, LinkParts, MemoryLink}; use rumors::{Gossiped, Led, Peer, Rumors, SessionStats}; use tokio::io::AsyncWrite; @@ -195,12 +195,16 @@ struct CountingConnector { impl Connector for CountingConnector { type Tx = CountingWrite; - async fn connect(&self) -> io::Result { + async fn connect(&self) -> io::Result<(Self::Tx, Done)> { self.opens.fetch_add(1, Ordering::Relaxed); - Ok(CountingWrite { - inner: self.inner.connect().await?, - written: self.written.clone(), - }) + let (inner, _) = self.inner.connect().await?; + Ok(( + CountingWrite { + inner, + written: self.written.clone(), + }, + Done::discard(), + )) } } From b2675dbef99b784efb817db01920ee59dac708c3 Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Fri, 14 Aug 2026 06:39:59 +0000 Subject: [PATCH 3/3] Gate reuse on the router's ready byte A recovered connection is not usable until the previous stream's consumer lets it go, and the consumer may be lazy. A pool that hands the connection out earlier couples the next stream's delivery to that consumer's progress, which is the serialization the link contract forbids, and mutual sessions deadlock on the cycle. The router now writes one byte when it re-arms. A reusing dial consumes it off the dialing path and reuses only connections whose byte has arrived; a dial that awaits the byte inline re-creates the coupling at the open. The multi-thread regression test pins the safe discipline. --- src/link/routed.rs | 9 +++++ src/link/routed/header.rs | 11 ++++++ src/link/routed/router.rs | 23 +++++++++--- src/link/routed/tests.rs | 73 +++++++++++++++++++++++++++++++++----- tests/common/routed_tcp.rs | 23 ++++++++---- tests/routed_link.rs | 44 +++++++++++++++++++++++ 6 files changed, 163 insertions(+), 20 deletions(-) diff --git a/src/link/routed.rs b/src/link/routed.rs index c220182f..6cccad04 100644 --- a/src/link/routed.rs +++ b/src/link/routed.rs @@ -248,6 +248,15 @@ pub trait Dial: Clone + Send + Sync + 'static { /// certifies nothing about liveness: the peer's router evicts idle /// connections by count, so a pooled connection may be dead and /// discovered only by the stream that draws it. + /// + /// The peer's router writes one byte on the connection once it is + /// ready for the next stream. A stream sent earlier is not + /// delivered until the previous stream's consumer lets the + /// connection go. Consume the byte off the dialing path and reuse + /// only connections whose byte has arrived, dialing fresh + /// otherwise: a `dial` that waits for it serializes the open + /// behind another stream's progress, which the link contract + /// forbids, and deadlocks a session whose streams cross. fn recycle(&self, _peer: &Self::Addr, conn: Self::Conn) { drop(conn); } diff --git a/src/link/routed/header.rs b/src/link/routed/header.rs index d5d83c03..3b232559 100644 --- a/src/link/routed/header.rs +++ b/src/link/routed/header.rs @@ -63,6 +63,17 @@ pub(super) const PREFIX_LEN: usize = MAGIC.len() + 2 + TOKEN_LEN; /// link is on its way to the application. pub(super) const ACK: u8 = 1; +/// The byte a router writes back on a recovered connection when it is +/// reading for the next header. +/// +/// Until it arrives, the previous stream's consumer still holds the +/// connection, and a stream sent on it would wait on that consumer's +/// progress. A reusing [`Dial`] consumes the byte off the dialing +/// path before the connection is reused. +/// +/// [`Dial`]: super::Dial +pub(super) const READY: u8 = 2; + /// Bytes in a [`Token`]. const TOKEN_LEN: usize = 16; diff --git a/src/link/routed/router.rs b/src/link/routed/router.rs index 62d73de4..75864a33 100644 --- a/src/link/routed/router.rs +++ b/src/link/routed/router.rs @@ -146,9 +146,9 @@ where let mut order: VecDeque<(u64, AbortHandle)> = VecDeque::new(); let mut next_id: u64 = 0; loop { - let conn = tokio::select! { - accepted = listen.accept() => accepted?, - Some(conn) = returned.recv() => conn, + let (conn, recovered) = tokio::select! { + accepted = listen.accept() => (accepted?, false), + Some(conn) = returned.recv() => (conn, true), Some(finished) = pending.next() => { if let Ok(id) = finished { order.retain(|(pending_id, _)| *pending_id != id); @@ -166,7 +166,15 @@ where next_id += 1; order.push_back((id, abort)); pending.push(Abortable::new( - route(conn, dial.clone(), &table, &incoming, &returns, id), + route( + conn, + recovered, + dial.clone(), + &table, + &incoming, + &returns, + id, + ), registration, )); } @@ -179,6 +187,7 @@ where /// handle. async fn route( conn: D::Conn, + recovered: bool, dial: D, table: &Table, incoming: &mpsc::Sender>, @@ -188,18 +197,22 @@ async fn route( // 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, returns).await; + let _ = deliver(conn, recovered, dial, table, incoming, returns).await; id } /// The routing step behind [`route`]: parse, then attach or establish. async fn deliver( mut conn: D::Conn, + recovered: bool, dial: D, table: &Table, incoming: &mpsc::Sender>, returns: &mpsc::Sender, ) -> io::Result<()> { + if recovered { + conn.write_all(&[header::READY]).await?; + } match header::read::(&mut conn).await? { Header::Stream { token } => { let Some(queue) = entries(table).get(&token).cloned() else { diff --git a/src/link/routed/tests.rs b/src/link/routed/tests.rs index aa11450f..92acc755 100644 --- a/src/link/routed/tests.rs +++ b/src/link/routed/tests.rs @@ -1,13 +1,16 @@ use std::collections::HashMap; -use std::future::Future; +use std::future::{Future, poll_fn}; use std::io; -use std::pin::pin; +use std::mem::take; +use std::pin::{Pin, pin}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; use futures::FutureExt; use futures::future::{Either, select, try_join}; -use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; +use futures::task::noop_waker; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream, ReadBuf}; use super::header::{self, Token}; use super::{Config, Dial, Endpoint, EndpointError, Incoming, LinkError, LinkInfo, RoutedLink}; @@ -494,6 +497,9 @@ async fn transfer_completed( #[derive(Clone)] struct PoolingDial { inner: MemoryDial, + /// Recycled, awaiting the router's ready byte. + pending: Arc>>>, + /// Ready for reuse. pool: Arc>>>, fresh: Arc, } @@ -502,6 +508,7 @@ impl PoolingDial { fn new(net: &MemoryNet) -> Self { PoolingDial { inner: net.dial(), + pending: Arc::default(), pool: Arc::default(), fresh: Arc::default(), } @@ -510,6 +517,35 @@ impl PoolingDial { fn fresh_dials(&self) -> usize { self.fresh.load(Ordering::Relaxed) } + + /// Admit pending connections whose ready byte has arrived, polling + /// each read exactly once: the byte is consumed off the dialing + /// path, never awaited. + fn admit(&self, addr: &MemoryName) { + let mut pending = self.pending.lock().expect("pending lock"); + let Some(conns) = pending.get_mut(&addr.0) else { + return; + }; + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + for mut conn in take(conns) { + let mut byte = [0u8; 1]; + let mut buf = ReadBuf::new(&mut byte); + match Pin::new(&mut conn).poll_read(&mut cx, &mut buf) { + Poll::Ready(Ok(())) if buf.filled().len() == 1 => { + self.pool + .lock() + .expect("pool lock") + .entry(addr.0.clone()) + .or_default() + .push(conn); + } + Poll::Pending => conns.push(conn), + // EOF or error: the connection is dead. + _ => {} + } + } + } } impl Dial for PoolingDial { @@ -517,6 +553,7 @@ impl Dial for PoolingDial { type Conn = DuplexStream; async fn dial(&self, addr: &MemoryName) -> io::Result { + self.admit(addr); let pooled = self .pool .lock() @@ -531,15 +568,33 @@ impl Dial for PoolingDial { } fn recycle(&self, peer: &MemoryName, conn: DuplexStream) { - self.pool + self.pending .lock() - .expect("pool lock") + .expect("pending lock") .entry(peer.0.clone()) .or_default() .push(conn); } } +/// Give the routers a few polls, so ready bytes land before the next +/// dial's single-poll admission. +async fn settle() { + for _ in 0..8 { + let mut yielded = false; + poll_fn(|cx| { + if yielded { + Poll::Ready(()) + } else { + yielded = true; + cx.waker().wake_by_ref(); + Poll::Pending + } + }) + .await; + } +} + /// A completed stream's connection carries the next stream instead of /// a fresh dial. /// @@ -567,16 +622,18 @@ fn completed_streams_reuse_their_connection() { let (_info, mut at_a) = arrival.expect("the router delivers the link"); let established = dial.fresh_dials(); - // Forward: the first stream dials, the second rides its + // Forward: the first stream dials, the second reuses its // recycled connection. transfer_completed(&at_b, &mut at_a, b"paid by a dial").await; - transfer_completed(&at_b, &mut at_a, b"rides recycled").await; + settle().await; + transfer_completed(&at_b, &mut at_a, b"reuses recycled").await; assert_eq!(dial.fresh_dials(), established + 1); // Reverse: the accepting side's dial-back pools the same way. let mut at_b = at_b; transfer_completed(&at_a, &mut at_b, b"paid by a dial").await; - transfer_completed(&at_a, &mut at_b, b"rides recycled").await; + settle().await; + transfer_completed(&at_a, &mut at_b, b"reuses recycled").await; assert_eq!(dial.fresh_dials(), established + 2); drop((a, b)); }) diff --git a/tests/common/routed_tcp.rs b/tests/common/routed_tcp.rs index f2e16e59..70f65868 100644 --- a/tests/common/routed_tcp.rs +++ b/tests/common/routed_tcp.rs @@ -12,6 +12,7 @@ use std::net::SocketAddr; use std::sync::{Arc, Mutex}; use rumors::link::routed::{Dial, Listen}; +use tokio::io::AsyncReadExt; use tokio::net::{TcpListener, TcpSocket, TcpStream}; /// Listener backlog: a full session complement of simultaneous stream @@ -66,13 +67,21 @@ impl Dial for PoolingTcpDial { } } - fn recycle(&self, peer: &SocketAddr, conn: TcpStream) { - self.pool - .lock() - .expect("pool lock") - .entry(*peer) - .or_default() - .push(conn); + fn recycle(&self, peer: &SocketAddr, mut conn: TcpStream) { + // Pooled only once the router's ready byte arrives, read off + // the dialing path. + let pool = Arc::clone(&self.pool); + let peer = *peer; + tokio::spawn(async move { + let mut ready = [0u8; 1]; + if conn.read_exact(&mut ready).await.is_ok() { + pool.lock() + .expect("pool lock") + .entry(peer) + .or_default() + .push(conn); + } + }); } } diff --git a/tests/routed_link.rs b/tests/routed_link.rs index 9e0c192f..985d79d3 100644 --- a/tests/routed_link.rs +++ b/tests/routed_link.rs @@ -215,6 +215,50 @@ async fn conforms_over_tcp_with_a_pooling_dial_swapped() { .expect("conformance suite ran past its liveness bound"); } +/// Mutual gossip sessions over a pooling dialer converge under a +/// multi-thread scheduler. +/// +/// The regression this pins: a pool that hands out a recycled +/// connection before the peer router's ready byte couples the next +/// stream's delivery to the previous stream's consumer, and mutual +/// sessions then deadlock. Single-thread schedules rarely close the +/// cycle, so the flavor here is load-bearing. +#[tokio::test(flavor = "multi_thread")] +async fn pooled_mutual_sessions_converge() { + timeout(SUITE_TIMEOUT, async { + let (mut a, mut b) = pooled_tcp_pair(true).await; + let seed: Rumors = Peer::seed().into_rumors(); + let (served, joined) = + tokio::join!(seed.gossip(&mut a), Peer::::bootstrap().join(&mut b)); + served.expect("the bootstrap-serving session completes"); + let newcomer = joined + .expect("the bootstrap session completes") + .expect("the seed serves the bootstrap") + .into_rumors(); + { + let mut batch = seed.batch(); + for payload in 0..48u64 { + batch.send(payload); + } + } + { + let mut batch = newcomer.batch(); + for payload in 48..96u64 { + batch.send(payload); + } + } + for _ in 0..2 { + let (near, far) = tokio::join!(seed.gossip(&mut a), newcomer.gossip(&mut b)); + near.expect("gossip completes over the link"); + far.expect("gossip completes over the link"); + } + assert_eq!(seed.snapshot().len(), 96); + assert_eq!(seed.snapshot(), newcomer.snapshot()); + }) + .await + .expect("pooled mutual sessions timed out"); +} + /// 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.