diff --git a/CHANGELOG.md b/CHANGELOG.md index 41bfe39..021f9e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,20 @@ plan payload lands (R-01), the `OpenHandler` lifetime contract is documented (R-02), and the two-pump helper is extracted (R-03). One breaking change (below); the wire surface is unchanged. +### Added + +- **`pump_bidi` (review 007 R-03 / ADR-050).** The two-pump helper — + `channels::pump_bidi(channel, peer_read, peer_write)` — pumps a + channel stream (`BiStream`-shaped) against a peer's split + read/write halves: one pump per direction, each shutting down the + opposite sink on completion (alknet ADR-078's shutdown-on- + completion), joined, returning `(u64, u64)` copy counts. Errors + are EOF-shaped by design (the POC + ADR-078 semantics — no `Err` + state; the review sketch's `io::Result` return was dead code). + The helper pins the contract in one place instead of three + (alktunnels POC `pump_halves`, alktty's channels session, the next + consumer). Purely additive. + ### Changed - **`Establishment` carries the channel plan (review 007 R-01 — diff --git a/docs/architecture/decisions/050-pump-bidi-two-pump-helper.md b/docs/architecture/decisions/050-pump-bidi-two-pump-helper.md new file mode 100644 index 0000000..e1a5620 --- /dev/null +++ b/docs/architecture/decisions/050-pump-bidi-two-pump-helper.md @@ -0,0 +1,105 @@ +# ADR-050: `pump_bidi` — the two-pump helper, extracted upstream + +## Status + +Accepted — implemented in alkcall 0.6.0 (review 007 Unit 3 / R-03). +Pins alknet ADR-078's two-pump contract in one place. + +## Context + +alknet ADR-078 defined the two-pump data-plane contract for forwarding +channels (one pump per direction; each pump shuts the opposite sink +down on completion — `try_join!` alone deadlocks) and deferred helper +extraction until a second two-pump consumer existed and the shapes +converged. Review 007 (R-03) supplied both halves of that test from +the alktunnels UDP POC: + +- **Producer side:** the POC's `pump_halves` — two `tokio::io::copy` + pumps over split channel-vs-substrate halves, each shutting down + the opposite sink on completion, joined. +- **Consumer side:** `TunnelSession::take_halves` + the assembly + layer's copy — the same shape modulo channel side. + +alktty's channels session already implements the shape (its +`drive_session_pre_negotiated` awaited inline); alkhttp's ferry does +not pump. The shapes converged; the review asked this sweep to either +extract the helper or record the decision not to — an un-extracted +helper would surface during alktunnels Phase 1 as the same +fix→publish→update treadmill for a purely additive change. + +## Decision + +Extract the helper as `channels::pump_bidi`: + +```rust +pub async fn pump_bidi(channel: C, peer_read: R, mut peer_write: W) -> (u64, u64) +where + C: AsyncRead + AsyncWrite + Unpin, + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +``` + +Two pumps, joined: `channel → peer_write` and `peer_read → channel`. +On each pump's completion the opposite sink is shut down +(shutdown-on-completion — the half-close the contract specifies). +Returns the two copy counts for observability. + +Three deliberate deviations from the review's sketch, all +semantics-first: + +1. **Return `(u64, u64)`, not `io::Result<(u64, u64)>`.** There is no + meaningful `Err` state: both pumps swallow copy errors by contract + — a mid-stream error is indistinguishable from an abrupt close + (no error channel exists mid-stream, per ADR-049 §6's pinned + EOF-shaped semantics), and shutdown-of-the-opposite-sink runs + either way. A returned `Err` would be dead code; the counts + (with `unwrap_or(0)` on error) are the observability. +2. **Peer side takes split halves (`peer_read: R`, `peer_write: W`), + not one `AsyncRead + AsyncWrite` value.** The dominant producer + shape is the establisher's dial result: `TcpStream::into_split` + halves (tunnels), pty pair halves (TTY). Taking halves matches the + POC (`pump_halves`) and avoids forcing consumers to re-join halves + just to satisfy a bound. The channel side stays a single value — + that is what the pump handler's `accept_bi` yields. +3. **Not `async fn` over `AsyncWriteExt` bounds on both sides.** The + channel side needs `tokio::io::split` (one value, two pumps), so + its bound is `AsyncRead + AsyncWrite + Unpin`; the `AsyncWriteExt` + bound the sketch had on every parameter is the extension trait + only where shutdown is called. + +alktty's three-pump session does not fit the helper (the exit future +is a third signal) and stays as-is — the helper serves the two-pump +shape, exactly as the review scoped. + +## Compatibility + +Purely additive: a new pub fn in `channels::pump`, re-exported in the +module docs; no existing surface changes. Consumers may adopt it at +their own pace (alktunnels Phase 1 gets it for free; alktty's +channels session can migrate later if it wants; the POC's shape is +now canonical). + +## Door type + +**Two-way.** A helper function is trivially removable or re-shapable; +no wire surface, no trait, no state. + +## Verification gate + +The POC's two-pump semantics reproduced through the helper, as unit +tests in `src/channels/pump.rs`: data flows both directions with +exact copy counts; EOF from one side completes the other's +shutdown-on-completion (the far end observes clean EOF, not an +error); a dead source is EOF-shaped teardown, not an `Err`. + +## References + +- Review 007 R-03 (the extraction ask + the convergence evidence) +- alknet ADR-078 (the two-pump contract; shutdown-on-completion) +- alktunnels `poc-summary.md` §Issues Surfaced #4 (the producer-side + shape; `pump_halves` in the POC's producer.rs) +- alktty `src/channels.rs` (the channels session shape; three-pump + carve-out) +- ADR-049 amendment 2 (R-02 — the lifetime contract the helper's + callers must respect: await `pump_bidi(..).await` inline inside the + handler task) \ No newline at end of file diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 5d8ff9e..2b2383b 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -25,6 +25,8 @@ //! ALPN crates via `ChannelCore` (ADR-047 §3). //! - [`policy`]: `ChannelLifecyclePolicy` + `PerIdentityChannelPolicy` //! (ADR-041, amended by ADR-047 §7 — opener ledger). +//! - [`pump`]: `pump_bidi` — the two-pump data-plane helper (alknet +//! ADR-078, pinned upstream by ADR-050). //! - [`client`]: `ChannelClient` — transport-agnostic //! `from_connection` (ADR-043). //! - [`self::env`]: `ChannelOperationEnv` extension trait (ADR-047 §4 — @@ -39,6 +41,7 @@ pub mod manager; pub mod mux; pub mod operations; pub mod policy; +pub mod pump; pub mod reassembly; pub mod source; pub mod wire; diff --git a/src/channels/pump.rs b/src/channels/pump.rs new file mode 100644 index 0000000..2129801 --- /dev/null +++ b/src/channels/pump.rs @@ -0,0 +1,204 @@ +//! `pump_bidi` — the two-pump data-plane helper (alknet ADR-078, as +//! pinned upstream by ADR-050 — review 007 R-03). +//! +//! A forwarding channel's data plane is two unidirectional pumps: one +//! copies channel→peer, the other peer→channel. The contract is +//! **shutdown-on-completion**: when one pump's copy source EOFs, it +//! shuts down the opposite sink so the peer sees a clean half-close; +//! the helper completes when both pumps finish. Copy errors are +//! EOF-shaped by design (the POC and ADR-078 semantics — there is no +//! error channel mid-stream; a pump error is an abrupt-close signal, +//! and shutdown-of-the-opposite-sink is the same either way). The +//! helper returns the two copy counts for observability. +//! +//! The shapes converged across three consumers (the extraction test +//! ADR-078 set): the alktunnels POC's `pump_halves`, alktty's channels +//! session, and the assembly-layer copies. This helper pins the +//! two-pump shape in one place; alktty's three-pump session (an exit +//! future as a third signal) does not fit and stays as-is. + +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; + +/// Pump a channel stream against a peer's split halves (ADR-078, as +/// pinned by ADR-050). Two pumps, joined: +/// +/// - `channel → peer_write`: copy until the channel EOFs, then +/// `shutdown()` the peer write half. +/// - `peer_read → channel`: copy until the peer EOFs, then +/// `shutdown()` the channel. +/// +/// Returns `(channel_to_peer, peer_to_channel)` copy counts. +/// Copy errors are treated as EOF (shutdown still runs) — the +/// EOF-shaped teardown the ADR-078 contract specifies; there is no +/// `Err` state because mid-stream errors are indistinguishable from +/// abrupt closes by design. +/// +/// The generic bounds follow the POC: the channel side is a single +/// `AsyncRead + AsyncWrite` value (the `BiStream` the pump handler +/// received), the peer side arrives as split read/write halves (a +/// dialed socket, a TTY pty pair — `into_split` is the natural +/// establisher result). +/// +/// Await the returned future inline inside the `OpenHandler`'s task — +/// the handler's returned `JoinHandle` must track the data-plane +/// lifetime (ADR-049 amendment 2, R-02). +pub async fn pump_bidi(channel: C, peer_read: R, mut peer_write: W) -> (u64, u64) +where + C: AsyncRead + AsyncWrite + Unpin, + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let (mut c_read, mut c_write) = tokio::io::split(channel); + let mut peer_read = peer_read; + + let c2p = async { + let n = tokio::io::copy(&mut c_read, &mut peer_write) + .await + .unwrap_or(0); + // Shutdown-on-completion: the channel side is done writing; + // the peer must see the half-close. + let _ = peer_write.shutdown().await; + n + }; + let p2c = async { + let n = tokio::io::copy(&mut peer_read, &mut c_write) + .await + .unwrap_or(0); + let _ = c_write.shutdown().await; + n + }; + tokio::join!(c2p, p2c) +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; + + /// Model the topology precisely: the helper gets the channel + /// (`BiStream`-shaped) and the peer\'s split halves. The test + /// drives the *other* ends: `channel_peer` (what the channel\'s + /// remote writes/reads) and the substrate far ends. `duplex` + /// pairs: a write on one half arrives on its counterpart. + struct Topology { + /// The channel value handed to the helper. + channel: DuplexStream, + /// The channel\'s remote end (test-driven). + channel_peer: DuplexStream, + /// The peer read half handed to the helper (helper reads it). + peer_read: DuplexStream, + /// The peer write half handed to the helper (helper writes it). + peer_write: DuplexStream, + /// The peer write half\'s remote (test reads what the helper + /// copied). + peer_write_far: DuplexStream, + /// The peer read half\'s remote (test writes what the helper + /// will copy). + peer_read_far: DuplexStream, + } + + fn topology() -> Topology { + // The channel\'s two ends. + let (channel, channel_peer) = tokio::io::duplex(64); + // The substrate: peer_read\'s far end, peer_write\'s far end. + let (peer_read, peer_read_far) = tokio::io::duplex(64); + let (peer_write_far, peer_write) = tokio::io::duplex(64); + Topology { + channel, + channel_peer, + peer_read, + peer_write, + peer_write_far, + peer_read_far, + } + } + + /// The two-pump semantics test (review 007 Unit 3 gate): the + /// POC\'s `pump_halves` behavior through the helper — data flows + /// both directions, shutdown-on-completion on EOF, exact counts. + #[tokio::test] + async fn pump_bidi_moves_data_both_ways_and_shuts_down_on_completion() { + let mut topo = topology(); + + let pumped = tokio::spawn(pump_bidi(topo.channel, topo.peer_read, topo.peer_write)); + + // channel -> substrate: write at the channel\'s remote, read + // at the substrate\'s far end. + topo.channel_peer + .write_all(b"outbound") + .await + .expect("write"); + let mut buf = [0u8; 8]; + topo.peer_write_far + .read_exact(&mut buf) + .await + .expect("helper copied channel->peer"); + assert_eq!(&buf, b"outbound"); + + // substrate -> channel: write at the substrate\'s far end, + // read at the channel\'s remote. + topo.peer_read_far + .write_all(b"inbound!") + .await + .expect("write"); + let mut buf2 = [0u8; 8]; + topo.channel_peer + .read_exact(&mut buf2) + .await + .expect("helper copied peer->channel"); + assert_eq!(&buf2, b"inbound!"); + + // EOF from the substrate\'s far end: the helper shuts the + // channel down (shutdown-on-completion). + topo.peer_read_far.shutdown().await.expect("shutdown"); + let mut eof_buf = Vec::new(); + let n = topo + .channel_peer + .read_to_end(&mut eof_buf) + .await + .expect("clean EOF at the channel remote"); + assert_eq!(n, 0, "the channel sees EOF after the substrate finished"); + + // Finish the other pump (channel remote drops) and collect + // the counts. + drop(topo.channel_peer); + let (c2p, p2c) = pumped + .await + .expect("helper completes when both pumps finish"); + assert_eq!(c2p, 8, "channel->peer count"); + assert_eq!(p2c, 8, "peer->channel count"); + + // The substrate\'s far write end observes the helper\'s + // shutdown: read returns 0 (EOF), not an error. + let mut eof_buf2 = Vec::new(); + let n2 = topo + .peer_write_far + .read_to_end(&mut eof_buf2) + .await + .expect("eof read"); + assert_eq!(n2, 0, "peer sees clean EOF after the channel side finished"); + } + + /// Copy errors are EOF-shaped: a pump whose source dies at birth + /// still lets the helper complete and shut the opposite sink down + /// (no Err state — ADR-078\'s abrupt-close semantics). + #[tokio::test] + async fn pump_bidi_treats_errors_as_eof_shaped_teardown() { + let topo = topology(); + + // Drop the substrate\'s far write end: the helper\'s peer_read + // source EOFs immediately; the channel side must still be shut + // down cleanly once the channel itself finishes. + drop(topo.peer_read_far); + + let pumped = tokio::spawn(pump_bidi(topo.channel, topo.peer_read, topo.peer_write)); + + // The channel remote first receives what nothing sends — EOF + // after the p2c pump finishes; drop the remote to finish the + // c2p pump too. + drop(topo.channel_peer); + let _counts = pumped + .await + .expect("helper completes despite the dead source"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 7c4447b..6e30267 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,7 +28,8 @@ //! - **Channels** ([`channels`]): the channels protocol — 8-byte chunk //! wire format, demux/mux, `ChannelManager`, `ChannelsAdapter`, //! `ChannelBidiStreamSource`, `ChannelOperations`, -//! `ChannelLifecyclePolicy`, `ChannelClient`. Channel 0 is +//! `ChannelLifecyclePolicy`, `ChannelClient`, `pump_bidi` (the +//! two-pump helper, ADR-050). Channel 0 is //! pre-negotiated as `alk/call`; channels 1..N are opened via //! per-ALPN open ops (`channels//sub`, `channels//pub`) //! on channel 0 (ADR-047).