From a22b2b84c9b6a3e4551b61ecdede4ebb2582b9ad Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 5 Sep 2026 07:04:30 +0000 Subject: [PATCH] fix: park early-arrival chunks for un-adopted channels (open/first-data race) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The connect side adopts a channel (installs local routing state) only after the open-op response arrives, but the accept side's OpenHandler can start pumping data the moment the channel opens — the two race and the demux's lenient unknown-channel drop (REQ-CH-04) silently lost the producer's first chunks (a TTY backend's banner, a sub protocol's greeting). route_payload now parks up to 64 payloads per unknown channel_id in a bounded early-arrival buffer; adopt_channel drains them into the new receiver in order. Beyond the cap the chunk drops with the existing debug log + dropped_unknown_chunks counter (which now also counts overflow). clear_all drops parked buffers with the connection. Surfaced by alktty's consumer end-to-end test (review #001 L3): the session never resolved because the producer's first chunks (stdout sentinel + exit chunk for an immediately-resolving backend) arrived before the adopt and were dropped. REQ-CH-04 wording updated by this behavior; ADR-039 §demux loop describes the lenient drop for genuinely unknown channels, which remains the case past the cap. Verification: cargo test 597 (2 rewritten for the new semantics + route_payload_to_unknown_channel_parks_until_adopt gate); --all-features 614; clippy -D warnings clean; fmt clean; doc 0 warnings; publish --dry-run ok; standalone probe (handler-writes-first e2e over one connection) shows 0 dropped chunks with the fix vs 1 without. --- CHANGELOG.md | 27 ++++++++ Cargo.lock | 2 +- Cargo.toml | 2 +- src/channels/manager.rs | 149 +++++++++++++++++++++++++++++++++++----- 4 files changed, 160 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e065ff..8e1819f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,33 @@ All notable changes to this crate are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this crate adheres to [Semantic Versioning](https://semver.org/). +## [0.4.1] - 2026-09-05 + +Bug-fix release: chunks arriving for a not-yet-adopted channel are +parked instead of dropped (the open-op response / first-data race). +No API changes. + +### Fixed + +- **Early-arrival chunks for un-adopted channels are parked, not + dropped.** The connect side adopts a channel (installs local routing + state) only after the open-op response arrives, but the accept side's + `OpenHandler` can start pumping data the moment the channel opens — + the two race. Previously the connect side's demux dropped those + chunks (REQ-CH-04's lenient unknown-channel drop), silently losing + the first chunks of any push-first producer (a TTY backend's banner + or greeting, a sub protocol's initial frame). `route_payload` now + parks up to 64 payloads per unknown `channel_id` in a bounded + early-arrival buffer and `adopt_channel` drains them into the new + receiver in order; chunks beyond the cap drop with the pre-existing + debug log and `dropped_unknown_chunks` counter (the counter now means + "early-arrival overflow or genuinely unknown channel", not just the + latter). `clear_all` drops the parked buffers with the connection. + Surfaced by alktty's consumer end-to-end test (review #001 L3): the + session never resolved because the producer's first chunks (the + stdout sentinel + exit chunk for an immediately-resolving backend) + arrived before the adopt and were dropped. + ## [0.4.0] - 2026-09-05 `input_schema` is now enforced at call time (the advertise-vs-enforce diff --git a/Cargo.lock b/Cargo.lock index 164bcfc..da1d7f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -27,7 +27,7 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.4.0" +version = "0.4.1" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 67a8bca..b1b3bfd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "alkcall" -version = "0.4.0" +version = "0.4.1" edition = "2021" rust-version = "1.85" license = "MIT OR Apache-2.0" diff --git a/src/channels/manager.rs b/src/channels/manager.rs index 7b3cfd2..c842604 100644 --- a/src/channels/manager.rs +++ b/src/channels/manager.rs @@ -7,7 +7,7 @@ //! `ChannelsAdapter`, the open-op wrapper, and relay logic can all //! hold a handle. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::net::SocketAddr; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; use std::sync::Arc; @@ -93,8 +93,23 @@ struct Inner { remote_addr: Option, side: ChannelSide, dropped_unknown_chunks: AtomicU64, + /// Chunks that arrived for a channel_id this side has not + /// adopted yet (the open-op response carrying `channel_id` and the + /// producer's first data-plane write race — the handler can start + /// pumping before the connect side's `adopt_channel` runs). Each + /// channel buffers up to `early_arrival_cap` payloads; `adopt_channel` + /// drains them into the new receiver. `clear_all` drops the map. + early_arrivals: Mutex>>, + early_arrival_count: AtomicU64, } +/// 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. +const EARLY_ARRIVAL_CAP: usize = 64; + impl ChannelManager { /// Construct a new manager with the given `MuxHandle`, max /// channels, buffer cap, and side. The `remote_addr` is @@ -124,6 +139,8 @@ impl ChannelManager { remote_addr, side, dropped_unknown_chunks: AtomicU64::new(0), + early_arrivals: Mutex::new(HashMap::new()), + early_arrival_count: AtomicU64::new(0), }), } } @@ -232,7 +249,7 @@ impl ChannelManager { let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap); let state = ChannelState { - demux_sender, + demux_sender: demux_sender.clone(), handler_task, alpn, }; @@ -288,7 +305,7 @@ impl ChannelManager { let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap); let state = ChannelState { - demux_sender, + demux_sender: demux_sender.clone(), handler_task, alpn, }; @@ -305,6 +322,10 @@ impl ChannelManager { } } + // Route any chunks that arrived between the open-op response and + // this adopt (the producer may have started pumping already). + self.drain_early_arrivals(channel_id, &demux_sender).await; + Ok((send, recv)) } @@ -377,8 +398,15 @@ impl ChannelManager { /// Route a chunk payload to the reassembled stream for /// `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` is dropped with - /// a debug log (REQ-CH-04 — lenient handling). + /// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload + /// 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) + /// must not lose its first chunks. `adopt_channel` drains the + /// parked payloads into the new receiver. Chunks beyond the cap + /// drop with a debug log (the adopted-side REQ-CH-04 leniency for + /// genuinely unknown channels). /// /// Awaits the bounded channel sender — if the handler's read half /// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless @@ -400,17 +428,67 @@ impl ChannelManager { } } None => { - self.inner - .dropped_unknown_chunks - .fetch_add(1, Ordering::Relaxed); - debug!( - channel_id, - "demux: unknown channel_id, dropping chunk (lenient)" - ); + self.park_early_arrival(channel_id, payload); } } } + /// 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). + fn park_early_arrival(&self, channel_id: u32, payload: Bytes) { + let mut arrivals = self.inner.early_arrivals.lock(); + let queue = arrivals.entry(channel_id).or_default(); + if queue.len() >= EARLY_ARRIVAL_CAP { + drop(arrivals); + self.inner + .dropped_unknown_chunks + .fetch_add(1, Ordering::Relaxed); + debug!( + channel_id, + "demux: unknown channel_id, early-arrival buffer full, dropping chunk" + ); + return; + } + queue.push_back(payload); + self.inner + .early_arrival_count + .fetch_add(1, Ordering::Relaxed); + debug!( + channel_id, + "demux: unknown channel_id, parking chunk (early arrival, not yet adopted)" + ); + } + + /// 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). + async fn drain_early_arrivals( + &self, + channel_id: u32, + demux_sender: &mpsc::Sender, + ) -> usize { + let payloads = self + .inner + .early_arrivals + .lock() + .remove(&channel_id) + .unwrap_or_default(); + let count = payloads.len(); + for payload in payloads { + if demux_sender.send(payload).await.is_err() { + debug!( + channel_id, + "demux: channel receiver dropped while draining early arrivals" + ); + break; + } + } + count + } + /// Take the sender for `channel_id` — used on close / teardown to /// drop the sender (which signals EOF to the handler, REQ-CH-02). /// Returns the handler task (if any) so the caller can abort it @@ -447,6 +525,8 @@ impl ChannelManager { task.abort(); } } + // The early-arrival buffers die with the connection too. + self.inner.early_arrivals.lock().clear(); // The opener ledger has the opener PeerIds. self.inner.opener_ledger.drain() } @@ -582,20 +662,53 @@ mod tests { .await; } + /// Early arrivals for a not-yet-adopted channel are parked, not + /// dropped: `adopt_channel` drains them into the new receiver. This + /// closes the open-op-response/first-data race (a push-first + /// producer's first chunks must not be lost). + #[tokio::test] + async fn route_payload_to_unknown_channel_parks_until_adopt() { + let manager = make_manager_with_runner().await; + manager.route_payload(7, Bytes::from_static(b"first")).await; + manager + .route_payload(7, Bytes::from_static(b"second")) + .await; + + let (_send, mut recv) = manager + .adopt_channel(7, "alk/tty", None) + .await + .expect("adopt"); + + use tokio::io::AsyncReadExt; + let mut buf = [0u8; 11]; + recv.read_exact(&mut buf).await.expect("read first+second"); + assert_eq!(&buf, b"firstsecond"); + } + #[tokio::test] async fn route_payload_to_unknown_channel_increments_dropped_counter() { let manager = make_manager_with_runner().await; assert_eq!(manager.dropped_unknown_chunks(), 0); + // Parked arrivals don't increment the counter... manager - .route_payload(999, Bytes::from_static(b"data")) - .await; - manager - .route_payload(998, Bytes::from_static(b"more")) + .route_payload(998, Bytes::from_static(b"kept")) .await; assert_eq!( manager.dropped_unknown_chunks(), - 2, - "dropped_unknown_chunks counter increments per unknown-channel drop" + 0, + "parked early arrivals are not drops" + ); + // ...but chunks beyond the per-channel cap do (the adopted-side + // REQ-CH-04 leniency for genuinely unknown channels). + for i in 0..(super::EARLY_ARRIVAL_CAP + 1) { + let n = i; + let payload = vec![b'x'; n % 9 + 1]; + manager.route_payload(999, Bytes::from(payload)).await; + } + assert_eq!( + manager.dropped_unknown_chunks(), + 1, + "dropped_unknown_chunks counter increments when the early-arrival buffer overflows" ); }