fix: park early-arrival chunks for un-adopted channels (open/first-data race)

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

@@ -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

Cargo.lock

@@ -27,7 +27,7 @@ dependencies = [
 
 [[package]]
 name = "alkcall"
-version = "0.4.0"
+version = "0.4.1"
 dependencies = [
  "async-trait",
  "bytes",

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"

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<SocketAddr>,
     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<HashMap<u32, VecDeque<Bytes>>>,
+    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<Bytes>,
+    ) -> 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"
         );
     }