diff --git a/Cargo.toml b/Cargo.toml index 1582f0f..8036d3a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,4 +35,7 @@ getrandom = { version = "0.4", features = ["wasm_js"] } futures = "0.3" parking_lot = "0.12" bytes = "1" -zeroize = { version = "1", features = ["alloc", "derive"] } \ No newline at end of file +zeroize = { version = "1", features = ["alloc", "derive"] } + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread"] } diff --git a/docs/architecture/channel-client.md b/docs/architecture/channel-client.md index 61b1ace..8861a13 100644 --- a/docs/architecture/channel-client.md +++ b/docs/architecture/channel-client.md @@ -119,11 +119,12 @@ backend's banner, a sub protocol's greeting) must not lose its first chunks. The park is bounded: **up to 64 chunks per channel** are parked -(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds); chunks arriving past -the cap are dropped silently — at the consumer this presents as a -truncated stream with clean framing everywhere else, not as an error. -Consumer-facing consequences (review 006 E-04, noted for tunnel-style -consumers): +(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds — bounded further by the +receiver's buffer capacity when the connection is constructed with a +smaller `buffer_cap`); chunks arriving past the cap are dropped +silently — at the consumer this presents as a truncated stream with +clean framing everywhere else, not as an error. Consumer-facing +consequences (review 006 E-04, noted for tunnel-style consumers): - Size any pre-adopt buffering math (e.g. a UDP POC's MTU-vs-buffer sizing) against **64 parked chunks as the observable bound**, and @@ -137,6 +138,15 @@ consumers): adopter was too slow for the producer's burst, and the affected streams were truncated at the park boundary. +**Adoption-window ordering (review 013):** the install and its drain of +parked chunks are one atomic section inside `adopt_channel`, and +`route_payload` parks only while that section's lock is free. A parked +chunk is therefore always pushed into the new receiver before any +direct route can begin, and a park can never straddle the install — +the reassembled byte stream preserves wire-arrival order through the +adoption window, whatever scheduling interleaves the demux and the +adopter. + ## Transport-agnostic by construction `ChannelClient` is the client side of the channels protocol. The channels diff --git a/docs/reviews/013-adopt-window-ordering-race.md b/docs/reviews/013-adopt-window-ordering-race.md index 8e0235f..641b35e 100644 --- a/docs/reviews/013-adopt-window-ordering-race.md +++ b/docs/reviews/013-adopt-window-ordering-race.md @@ -2,7 +2,54 @@ ## Status -Open — filed 2026-10-21. Triggered by a downstream finding: alkgit's +Resolved — filed 2026-10-21, resolved 2026-10-22 (same session). C-01 +was re-verified (code walk plus a latch-based repro: holding the old +install→drain window open via a test-only gate makes a receiver read +`CP1P2` where the wire order was P1, P2, C — the deterministically +observed inversion), then closed by making the install and its park +drain **one atomic section** (the re-review's fix direction 1+3 +hybrid: `route_payload`'s map-miss-and-park is now also one lock +section, so nothing can straddle either edge of the window). The +resolution is fully synchronous — no settle flag, no extra task, no +`pending` queue per channel: + +- `route_payload` holds the channels-map lock across the miss → park + decision (`park_early_arrival` runs nested under it) — a park can + never straddle the install, closing the orphaned-chunk loss variant + the re-review added to the finding (a chunk that missed the map + pre-install but parked after the post-install drain would sit + parked forever, invisible to `dropped_unknown_chunks`). +- `install_locked` (new, shared by `open_channel` / + `adopt_channel` / `install_channel_zero`) swaps the parked queue + out, inserts the state, and `try_send`s the parked queue into the + fresh, empty receiver **before releasing the channels lock** — a + direct route (map hit) can only begin after that release, so parked + chunks always precede it in mpsc send order. No await runs inside + the section: the effective park cap is + `min(EARLY_ARRIVAL_CAP, buffer_cap)` so the parked queue always fits + the fresh receiver. `drain_early_arrivals` remains as the defensive + full-buffer fallback (previously-unreachable-by-invariant) and is + now called by all three install paths. +- **Semver: patch release of the 0.10 line** (0.10.1) — `ChannelState` + and the fix surface are private; the API surface, the wire formats, + and the documented cap behavior (review 006 E-04) are unchanged; + behavior only strengthens (in-window drops and the orphan loss + become unreachable). + +Regression tests (src/channels/manager.rs): the chunk-granularity +ordering test the fix sketch called for +(`adopt_window_does_not_reorder_chunks` — 200 iterations on a real +2-worker runtime, a racer routing C the moment the install becomes +visible, receiver byte stream asserted byte-identical to the +producer's — catches the old window probabilistically under genuine +preemption, exactly the alkgit repro shape), the pre-install park +FIFO shape (`adopt_drains_parked_and_raced_chunks_in_fifo`), and the +park-cap normalization (`park_cap_bounded_by_receiver_buffer`). +**Downstream re-verification gate still applies**: alkgit's flaking +row (`tests/native_transport.rs …` under multi-process load, finding +F-2) must pass against the fixed crate before publishing 0.10.1. + +Triggered by a downstream finding: alkgit's wave-2 review gate (`/workspace/@alkdev/alkgit/` — `core/review-wave-2`) caught its channels open-op integration test flaking under multi-process load, instrumented the failing byte stream, @@ -11,10 +58,11 @@ diagnostic data (two distinct desync shapes at two distinct offsets) is reproduced here in full; alkgit carries the panic improvement (alkgit `b6db513`) that makes the shape observable. -No alkcall code change is included in this review — this is the -finding record, the repro evidence, and the fix-shape sketch the -implementing patch builds from. **Recommended urgency: patch release -of the published 0.10 line**, because the failure mode is silent +As filed — before the resolution — this review contained no alkcall +code change: it was the finding record, the repro evidence, and the +fix-shape sketch the implementing patch builds from. **Recommended +urgency: patch release of the published 0.10 line** (the ruling the +Status section implements), because the failure mode is silent byte-stream corruption on a freshly adopted channel and the documented downstream consumers (the git producer half in alkgit; any dial-then-pump tunnel shape) sit exactly on the affected window. diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index 3d6c388..1370d44 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -27,7 +27,7 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.9.0" +version = "0.10.0" dependencies = [ "async-trait", "bytes", diff --git a/src/channels/manager.rs b/src/channels/manager.rs index c1492d7..5baa988 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -23,6 +23,7 @@ use serde_json::Value; use super::mux::{MuxHandle, OpenerLedger}; use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP}; +use tokio::sync::mpsc::error::TrySendError; /// Which side of a channels connection this manager belongs to. /// Determines the ID allocation range to prevent collisions when both @@ -126,21 +127,22 @@ struct Inner { control_handles: Mutex>, } -/// The per-channel cap on parked early-arrival chunks (`route_payload` -/// parks chunks for not-yet-adopted channels instead of dropping them — -/// the open-op response/first-data race). A channel whose adopter never -/// arrives leaks its parked chunks until `clear_all`; the cap bounds -/// that leak per channel. +/// The nominal per-channel cap on parked early-arrival chunks +/// (`route_payload` parks chunks for not-yet-adopted channels instead +/// of dropping them — the open-op response/first-data race). A channel +/// whose adopter never arrives leaks its parked chunks until +/// `clear_all`; the cap bounds that leak per channel. /// /// Consumer-facing bound (review 006 E-04): for a push-first producer -/// (dial-then-pump is the common tunnel shape), up to `EARLY_ARRIVAL_CAP` -/// chunks per channel can be parked before the consumer's -/// `adopt_channel` runs — and chunks past the cap drop silently at the -/// consumer (data loss presents as a truncated stream with clean -/// framing elsewhere; `dropped_unknown_chunks` / -/// `early_arrival_count` are the producer-side observability). A -/// tunnel crate's sizing (e.g. a UDP POC's MTU-vs-buffer math) should -/// treat 64 parked chunks as the observable bound and adopt promptly. +/// (dial-then-pump is the common tunnel shape), up to the effective +/// park cap (see `ChannelManager::early_arrival_cap`) chunks per +/// channel can be parked before the consumer's `adopt_channel` runs — +/// and chunks past the cap drop silently at the consumer (data loss +/// presents as a truncated stream with clean framing elsewhere; +/// `dropped_unknown_chunks` / `early_arrival_count` are the +/// producer-side observability). A tunnel crate's sizing (e.g. a UDP +/// POC's MTU-vs-buffer math) should treat the effective park cap as +/// the observable bound and adopt promptly. const EARLY_ARRIVAL_CAP: usize = 64; impl ChannelManager { @@ -311,11 +313,21 @@ impl ChannelManager { if channels.contains_key(&channel_id) { return Err(ManagerError::ChannelExists(channel_id)); } - channels.insert(channel_id, state); + Self::install_locked( + &mut channels, + &self.inner.early_arrivals, + channel_id, + state, + &demux_sender, + ); } self.inner.opener_ledger.record(channel_id, opener); + // Route any parked chunks that survived the install (the + // defensive re-park path — normally nothing lands here). + self.drain_early_arrivals(channel_id, &demux_sender).await; + Ok((channel_id, send, recv)) } @@ -335,6 +347,14 @@ impl ChannelManager { /// (collision — the remote side allocated an ID this side already /// uses). Returns `TooManyChannels` if the per-connection cap is /// reached. + /// + /// **Adoption-window ordering (review 013):** the install and its + /// drain of parked early arrivals share one channels-lock section + /// (`install_locked`), and `route_payload` parks only while that + /// lock is free. A parked chunk is therefore always pushed into + /// the new receiver before any direct route can begin, and a park + /// can never straddle the install — the channel's byte stream + /// preserves wire-arrival order through the adoption window. pub async fn adopt_channel( &self, channel_id: u32, @@ -394,11 +414,20 @@ impl ChannelManager { if channels.contains_key(&channel_id) { return Err(ManagerError::ChannelExists(channel_id)); } - channels.insert(channel_id, state); + Self::install_locked( + &mut channels, + &self.inner.early_arrivals, + channel_id, + state, + &demux_sender, + ); } - // Route any chunks that arrived between the open-op response and - // this adopt (the producer may have started pumping already). + // Route any parked chunks that survived the install (the + // defensive re-park path — normally the install already pushed + // them). Chunks parked between the open-op response and this + // adopt were pushed under the lock above (review 013); the + // receiver was fresh and empty, so nothing ever lands here. self.drain_early_arrivals(channel_id, &demux_sender).await; Ok((send, recv)) @@ -474,19 +503,27 @@ impl ChannelManager { .map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?; let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap); let state = ChannelState { - demux_sender, + demux_sender: demux_sender.clone(), handler_task, alpn: "alk/call".to_string(), }; { let mut channels = self.inner.channels.lock(); - if channels - .insert(super::wire::CHANNEL_ID_ZERO, state) - .is_some() - { + if channels.contains_key(&super::wire::CHANNEL_ID_ZERO) { return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO)); } + Self::install_locked( + &mut channels, + &self.inner.early_arrivals, + super::wire::CHANNEL_ID_ZERO, + state, + &demux_sender, + ); } + // Route any parked chunks that survived the install (the + // defensive re-park path — normally nothing lands here). + self.drain_early_arrivals(super::wire::CHANNEL_ID_ZERO, &demux_sender) + .await; // Channel 0 is opened by the accept side itself; the opener // ledger records the local peer (or is left empty for channel // 0 — the cap doesn't apply to the pre-negotiated control @@ -498,7 +535,7 @@ impl ChannelManager { /// `channel_id` (the demux's per-chunk route). A zero-length /// payload is the EOF sentinel — the reassembled stream interprets /// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload - /// in the early-arrival buffer (up to `EARLY_ARRIVAL_CAP` payloads + /// in the early-arrival buffer (up to `early_arrival_cap` payloads /// per channel) — the open-op response carrying `channel_id` and /// the producer's first data-plane write race, and a push-first /// producer (a TTY backend's banner, a sub protocol's greeting) @@ -507,6 +544,22 @@ impl ChannelManager { /// drop with a debug log (the adopted-side REQ-CH-04 leniency for /// genuinely unknown channels). /// + /// The park decision is atomic with the channels-map state: the + /// map miss and the park happen under the same channels-lock + /// acquisition, so an installing `adopt_channel` can never straddle + /// them (review 013 — a chunk parked after the install drained the + /// queue would otherwise be orphaned). A park can only be observed + /// by an installer that has not inserted yet, and that installer + /// swaps the queue out when it inserts. + /// + /// **Ordering (review 013):** when the install and its drain of + /// parked chunks are one lock section, a direct route (map hit) + /// can only begin after that section released — the parked chunks + /// are already in the receiver's mpsc by then. Send order on the + /// mpsc therefore matches wire-arrival order for every chunk of a + /// freshly adopted channel; a direct route can never overtake an + /// earlier-parked chunk. + /// /// Awaits the bounded channel sender — if the handler's read half /// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless /// bounded-buffer backpressure). The demux loop is the only caller; @@ -515,31 +568,50 @@ impl ChannelManager { pub async fn route_payload(&self, channel_id: u32, payload: Bytes) { let sender = { let channels = self.inner.channels.lock(); - channels.get(&channel_id).map(|s| s.demux_sender.clone()) + match channels.get(&channel_id) { + Some(state) => state.demux_sender.clone(), + None => return self.park_early_arrival(channel_id, payload), + } }; - match sender { - Some(sender) => { - if sender.send(payload).await.is_err() { - debug!( - channel_id, - "demux: channel receiver dropped, dropping chunk" - ); - } - } - None => { - self.park_early_arrival(channel_id, payload); - } + if sender.send(payload).await.is_err() { + debug!( + channel_id, + "demux: channel receiver dropped, dropping chunk" + ); } } + /// The effective per-channel park cap: the nominal + /// `EARLY_ARRIVAL_CAP` bounded by the receiver's mpsc buffer + /// capacity. The install paths drain the parked queue with + /// `try_send` inside their channels-lock section (review 013), and + /// the drain target is the fresh, empty receiver — bounding the + /// park cap by the buffer capacity guarantees the drain never + /// needs to await (which it cannot, holding a `parking_lot` + /// lock). With the default constructor settings both are 64 and + /// the bound is inactive; smaller `buffer_cap` values tighten the + /// park cap instead of turning the drain into a stall. + fn early_arrival_cap(&self) -> usize { + EARLY_ARRIVAL_CAP.min(self.inner.buffer_cap) + } + /// Park an early-arrival chunk for `channel_id` (the adopter has - /// not run yet). FIFO per channel, bounded by `EARLY_ARRIVAL_CAP`; - /// beyond the cap the chunk is dropped with a debug log and the - /// dropped counter increments (observability only). + /// not run yet). FIFO per channel, bounded by + /// `early_arrival_cap`; beyond the cap the chunk is dropped with a + /// debug log and the dropped counter increments (observability + /// only). + /// + /// Called with the channels lock already held (review 013): the + /// caller (`route_payload`) holds it across the map miss and this + /// park so the decision is atomic with an in-flight install — a + /// parked chunk is always visible to the installer that caused the + /// miss. Lock order is channels → early_arrivals; no other path + /// nests them the other way. fn park_early_arrival(&self, channel_id: u32, payload: Bytes) { + let cap = self.early_arrival_cap(); let mut arrivals = self.inner.early_arrivals.lock(); let queue = arrivals.entry(channel_id).or_default(); - if queue.len() >= EARLY_ARRIVAL_CAP { + if queue.len() >= cap { drop(arrivals); self.inner .dropped_unknown_chunks @@ -561,9 +633,21 @@ impl ChannelManager { } /// Drain the early-arrival buffer for `channel_id` into `recv`'s - /// sender (called by `adopt_channel` after the channel state is - /// installed). Returns the drained payload count. Payloads past the - /// receiver's buffer capacity await normally (backpressure). + /// sender (called after the channel state is installed). Returns + /// the drained payload count. Payloads past the receiver's buffer + /// capacity await normally (backpressure). + /// + /// This is the defensive leftover path only: since review 013 the + /// install paths (`open_channel` / `adopt_channel` / + /// `install_channel_zero`) drain the parked queue under the + /// channels lock via `install_locked`, and the effective park cap + /// is bounded by the receiver's buffer capacity so that drain + /// never needs this. Only chunks re-parked by `install_locked`'s + /// full-buffer fallback (violating the buffer-cap invariant) or + /// chunks parked for an id with no installer pending at all reach + /// this method after the fact. Chunks parked for an id that is + /// never adopted stay parked until `clear_all` (the documented + /// leak bound). async fn drain_early_arrivals( &self, channel_id: u32, @@ -632,6 +716,63 @@ impl ChannelManager { self.inner.opener_ledger.drain() } + /// Core install, called with the channels lock already held: swap + /// out any parked early-arrival queue for `channel_id` and push it + /// into the fresh receiver's mpsc **before the lock is released** + /// (review 013 — the install and its park-drain are one atomic + /// decision, so a direct route (a map hit after release) can never + /// overtake an earlier-parked chunk). + /// + /// The parked queue never exceeds `early_arrival_cap` (bounded by + /// the receiver's buffer capacity) and the receiver is fresh and + /// empty, so each push is a non-blocking `try_send`. Chunks + /// bouncing off a full buffer (possible only if the buffer-cap + /// invariant above is violated) are re-parked for the caller's + /// post-release async drain — the pre-fix behavior, kept as a + /// defensive fallback rather than a correctness path; chunks hit by + /// a dropped receiver are dropped as everywhere else. + fn install_locked( + channels: &mut HashMap, + early_arrivals: &Mutex>>, + channel_id: u32, + state: ChannelState, + demux_sender: &mpsc::Sender, + ) { + let parked = early_arrivals + .lock() + .remove(&channel_id) + .unwrap_or_default(); + channels.insert(channel_id, state); + let mut leftovers = VecDeque::new(); + for payload in parked { + if let Err(error) = demux_sender.try_send(payload) { + match error { + TrySendError::Full(bounced) => { + debug!( + channel_id, + "install: receiver buffer full draining early arrivals" + ); + leftovers.push_back(bounced); + break; + } + TrySendError::Closed(_) => { + debug!( + channel_id, + "install: receiver dropped, dropping early arrival" + ); + } + } + } + } + if !leftovers.is_empty() { + let mut arrivals = early_arrivals.lock(); + let queue = arrivals.entry(channel_id).or_default(); + for payload in leftovers { + queue.push_back(payload); + } + } + } + /// `true` if `channel_id` is currently open. pub fn has_channel(&self, channel_id: u32) -> bool { self.inner.channels.lock().contains_key(&channel_id) @@ -1134,4 +1275,140 @@ mod tests { Ok(_) => panic!("expected TooManyChannels, got Ok"), } } + + /// Review 013 regression: a chunk routed through the adoption + /// window must not overtake earlier-parked chunks on the receiver's + /// byte stream. The racer routes `C` as soon as the install is + /// visible — under the pre-fix code and real preemption that could + /// land in the install→drain window and be delivered first (with + /// P1/P2 parked); the install and its park-drain being one lock + /// section makes P1, P2, C the only reachable order, whichever + /// side of the install the racer's route lands on (pre-install → + /// parked and drained FIFO; post-install → sent after the park + /// queue already went out). 200 iterations to give a real + /// scheduler the chance to catch the old window; the receiver's + /// byte stream must be byte-identical to the producer's every time. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn adopt_window_does_not_reorder_chunks() { + use std::time::Duration; + use tokio::io::AsyncReadExt; + + for iteration in 0..200u32 { + let manager = make_manager_with_runner().await; + let channel_id = 7 + iteration * 2; + + manager + .route_payload(channel_id, Bytes::from_static(b"P1")) + .await; + manager + .route_payload(channel_id, Bytes::from_static(b"P2")) + .await; + + let adopt_manager = manager.clone(); + let adopter = tokio::spawn(async move { + adopt_manager + .adopt_channel(channel_id, "alk/tty", None) + .await + .expect("adopt") + }); + let race_manager = manager.clone(); + let racer = tokio::spawn(async move { + loop { + if race_manager.has_channel(channel_id) { + race_manager + .route_payload(channel_id, Bytes::from_static(b"C")) + .await; + break; + } + tokio::task::yield_now().await; + } + }); + + let (_send, mut recv) = adopter.await.expect("adopt task join"); + racer.await.expect("race task join"); + let mut buf = [0u8; 5]; + tokio::time::timeout(Duration::from_secs(10), recv.read_exact(&mut buf)) + .await + .expect("read within timeout") + .expect("read all"); + assert_eq!( + &buf, b"P1P2C", + "iteration {iteration}: the receiver's byte stream diverged from + the producer's wire order (C overtook parked chunks)" + ); + } + } + + /// Parked chunks routed while the install is in flight (map still + /// empty) are drained in FIFO order after the parked prefix — the + /// park decision is atomic with the map state, so a chunk can only + /// miss the map while the adopter has not inserted yet, and that + /// adopter swaps the whole queue out when it does. + #[tokio::test] + async fn adopt_drains_parked_and_raced_chunks_in_fifo() { + let manager = make_manager_with_runner().await; + + manager.route_payload(7, Bytes::from_static(b"P1")).await; + manager.route_payload(7, Bytes::from_static(b"P2")).await; + + let adopt_manager = manager.clone(); + let adopter = tokio::spawn(async move { + adopt_manager + .adopt_channel(7, "alk/tty", None) + .await + .expect("adopt") + }); + + while !manager.has_channel(7) { + tokio::task::yield_now().await; + } + + let (_send, mut recv) = adopter.await.expect("adopt task join"); + manager.route_payload(7, Bytes::from_static(b"C")).await; + + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 5]; + recv.read_exact(&mut buf).await.expect("read all"); + assert_eq!(&buf, b"P1P2C", "raced chunk must follow the parked prefix"); + } + + /// The effective park cap is the nominal EARLY_ARRIVAL_CAP bounded + /// by the receiver's buffer capacity (review 013 — the install + /// drains the parked queue with non-blocking sends into the fresh + /// receiver, which must always fit). With buffer_cap 4 the fifth + /// parked chunk drops at park time (the documented E-04 cap + /// behavior, tightened) and the first four arrive on adopt. + #[tokio::test] + async fn park_cap_bounded_by_receiver_buffer() { + let (_client, server) = duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, runner) = MuxRunner::new(Box::new(writer)); + tokio::spawn(async move { + let _ = runner.run().await; + }); + let manager = ChannelManager::new(handle, 256, 4, None, ChannelSide::Accept); + + for chunk in b"abcde" { + manager + .route_payload(7, Bytes::copy_from_slice(&[*chunk])) + .await; + } + assert_eq!( + manager.dropped_unknown_chunks(), + 1, + "the fifth parked chunk dropped (park cap = min(64, buffer_cap=4))" + ); + + let (_send, mut recv) = manager + .adopt_channel(7, "alk/tty", None) + .await + .expect("adopt"); + + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 4]; + recv.read_exact(&mut buf) + .await + .expect("read parked quartet"); + assert_eq!(&buf, b"abcd", "parked chunks arrive in FIFO order"); + } }