3 Commits
Author SHA1 Message Date
glm-5.3-flash a99cc67e76 chore(release): 0.10.1 — the adoption-window ordering fix
Changelog entry for review 013's fix (4dbd4bc); version bump 0.10.0 →
0.10.1 (patch — private fix surface, no API or wire-format change).

Verification: semver-checks 196 pass (no semver update required);
cargo test 700 + 724 (--all-features) pass; clippy --all-targets
--all-features -D warnings, fmt --check, wasm32-unknown-unknown check +
clippy, fuzz corpus replay (5/5), and cargo publish --dry-run clean.
Downstream gate: alkgit's flaking row re-verified green under
multi-process load against 4dbd4bc.
2026-10-22 05:28:12 +00:00
glm-5.3-flash 4dbd4bc583 fix(channels): close the adopt-window ordering race (review 013) — atomic install+park-drain
A chunk routed by the demux inside the adopt_channel install→drain
window was delivered ahead of earlier-parked early arrivals: the mpsc
send order was inverted relative to wire-arrival order, misordering the
channel's byte stream (C-01, critical — downstream alkgit observed two
desync shapes under multi-process load). A related loss variant is also
closed: a chunk whose map miss parked it after the drain had removed
the queue sat parked forever.

The window closes by serialization, not by widening locks across
awaits — the resolution is fully synchronous:

- route_payload holds the channels-map lock across the miss → park
  decision (park_early_arrival runs nested; lock order channels →
  early_arrivals): a park can never straddle the install, and the
  installer that caused the miss always swaps the queue out.
- install_locked (new, shared by open_channel / adopt_channel /
  install_channel_zero) swaps the parked queue, inserts the state, and
  try_sends the parked queue into the fresh, empty receiver before
  releasing the channels lock — a direct route can only begin after
  release, so parked chunks always precede it in send order. No await
  runs inside the section: the effective park cap is now
  min(EARLY_ARRIVAL_CAP, buffer_cap), bounding the parked queue by the
  receiver's capacity so the drain never blocks. drain_early_arrivals
  remains as the defensive full-buffer fallback and is now called by
  all three install paths.

Regression tests: adopt_window_does_not_reorder_chunks (200 iterations
on a 2-worker runtime — a racer routing C the moment the install is
visible, receiver byte stream asserted byte-identical to the producer's;
catches the old window under genuine preemption, the alkgit repro
shape), adopt_drains_parked_and_raced_chunks_in_fifo, and
park_cap_bounded_by_receiver_buffer. rt-multi-thread added as a
dev-dependency for the runtime test.

Semver: patch (0.10.1) — private fix surface; no API, wire-format, or
documented-cap behavior change; behavior only strengthens.

Verification: cargo test 700/724 (--all-features) passed; clippy
--all-targets -D warnings clean; fmt --check clean; cargo doc --no-deps
generated; fuzz corpus replay passed; wasm32-unknown-unknown check
passed.
2026-10-22 04:47:42 +00:00
glm-5.3-flash 56e0f135d7 docs(review): find the adoption-window ordering race — routed from alkgit's wave-2 gate
Review 013 [critical]: a chunk routed in the adopt_channel
install->drain window overtakes earlier-parked early arrivals,
misordering the reassembled byte stream (and, combined with the
EARLY_ARRIVAL_CAP overflow branch, can drop chunks) on a freshly
adopted channel. Diagnosed downstream: alkgit's wave-2 gate caught its
channels open-op test flaking under load and traced both observed
desync shapes (mid-advertisement 'fetc' after 0 lines, '00ev' after 1)
to the window. Includes the fix-shape sketch (per-channel ordered
mailbox recommended), the chunk-granularity regression gate, and the
downstream re-verification requirement. The ordering half of review
006 E-04's adopt-race family; alkgit carries no workaround by design.
2026-10-21 13:51:09 +00:00
7 changed files with 661 additions and 52 deletions

No files matched your search

+55
View File
@@ -4,6 +4,61 @@ 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.10.1] - 2026-10-22
The adoption-window ordering fix (review 013, routed from alkgit's
wave-2 gate): a chunk routed by the demux inside the
`adopt_channel` install→drain window was delivered **ahead of**
earlier-parked early arrivals, inverting the channel's byte-stream
order (observed downstream as two distinct pkt-line desync shapes in
alkgit's open-op integration test under multi-process load — failure
shape 1, the reader starting mid-advertisement at offset ~16; failure
shape 2, a chunk boundary two bytes into a hex-length header). A
related loss variant is also closed: a chunk whose map miss parked it
after the drain had removed the queue sat parked forever, invisible to
`dropped_unknown_chunks`. No API, wire-format (`EventEnvelope`, the
channels 8-byte chunk header), or documented-cap behavior (review 006
E-04) change — `ChannelState` and the fix surface are private; the
behavior only strengthens.
### Fixed
- **The adopt-window ordering race (review 013 C-01).** The install
and its park-drain are now one atomic section. `route_payload`'s
map-miss → early-arrival-park decision runs under the channels-map
lock (a park can never straddle the install — closing the orphaned
chunk loss variant), and the new private `install_locked` helper
(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 lock** — a direct route (map hit)
can only begin after release, so parked chunks always precede it in
mpsc send order and the reassembled byte stream preserves
wire-arrival order through the adoption window. No await runs inside
the section: the effective park cap is
`min(EARLY_ARRIVAL_CAP, buffer_cap)`, bounding the parked queue by
the receiver's capacity so the non-blocking drain always fits
(smaller `buffer_cap` values tighten the park cap; the default
64/64 pairing is unchanged). `drain_early_arrivals` remains as the
defensive full-buffer fallback, now called by all three install
paths. Regression tests:
`adopt_window_does_not_reorder_chunks` (200 iterations on a real
2-worker runtime — a racer routing a chunk the moment the install
becomes visible; the receiver's byte stream must be byte-identical
to the producer's every time, the alkgit repro shape),
`adopt_drains_parked_and_raced_chunks_in_fifo`, and
`park_cap_bounded_by_receiver_buffer`. Downstream gate: alkgit's
flaking row re-verified green under multi-process load against this
fix before the 0.10.1 publish.
### Verified
- 700 tests pass (including the stub guard; 724 with `--all-features`);
`clippy --all-targets -- -D warnings`, `fmt --check`, `cargo doc`,
wasm32-unknown-unknown check, fuzz corpus replay
(`cargo test --manifest-path fuzz/shared/Cargo.toml`),
`cargo semver-checks`, and `cargo publish --dry-run` clean.
## [0.10.0] - 2026-10-17
The gateway ownership seam (review 012, downstream evidence from
Generated
+1 -1
View File
@@ -27,7 +27,7 @@ dependencies = [
[[package]]
name = "alkcall"
version = "0.10.0"
version = "0.10.1"
dependencies = [
"async-trait",
"bytes",
+5 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "alkcall"
version = "0.10.0"
version = "0.10.1"
edition = "2021"
rust-version = "1.88"
license = "MIT OR Apache-2.0"
@@ -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"] }
zeroize = { version = "1", features = ["alloc", "derive"] }
[dev-dependencies]
tokio = { version = "1", features = ["rt-multi-thread"] }
+15 -5
View File
@@ -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
@@ -0,0 +1,264 @@
# Review 013 — The adoption-window ordering race: a chunk routed in the `adopt_channel` install→drain window overtakes earlier-parked early arrivals and misorders the channel's byte stream
## Status
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,
and traced every failure to the window analyzed below. The downstream
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.
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.
## Scope
`src/channels/manager.rs` — the adopt/adopt-equivalent paths and the
demux receive side they race:
- `ChannelManager::adopt_channel` (`manager.rs:338-402`): the
install (`channels.insert`, `manager.rs:389-394`) and the
post-install drain (`drain_early_arrivals`, `manager.rs:402`).
- `ChannelManager::route_payload` (`manager.rs:515-533`): the
channels-map lookup → direct `send` path, and the not-yet-installed
→ `park_early_arrival` fallback.
- `ChannelManager::park_early_arrival` (`manager.rs:539-561`).
- `ChannelManager::drain_early_arrivals` (`manager.rs:567+`).
The open-reply adoption path (`ChannelClient`/open-op consumers
calling `adopt_channel` after the open response) and the
`open_channel` path share the same shape; the observed failures are on
the `adopt_channel` path (the consumer side of the open op).
This is the ordering half of the adopt-race family review 006 E-04
already bounded ("early-arrival park cap (64) interacts with
push-first producers under slow adopters"): E-04 covered the *cap drop*
of chunks arriving before adopt and classed it not-a-correctness-bug
because the cap is documented behavior. This review covers the
*ordering* of chunks around the moment routing state becomes visible —
nothing documented bounds it, and nothing in the implementation
preserves it.
## Severity legend note
Same scale as reviews 004/005/006. Tagged
**[critical]** by E-01's second clause ("corrupts data"): the channel's
reassembled byte stream is delivered with misordered bytes — which, at
the layer every real consumer sits on (a framing codec), is corrupted
data. The mitigating fact (the window is narrow and adoption-time-only)
bounds frequency, not impact; a consumer cannot detect or recover from
a misordered chunk, and the REQ-CH series pins delivery as a byte
stream on a "single ordered, reliable" substrate
(`docs/architecture/channels-wire.md:9`) — reassembly delivering
out-of-order violates the one property the demux exists to preserve.
## The finding
### C-01 [critical] — a chunk arriving between install and drain is delivered ahead of the chunks parked before install
**Sequence** (single channel id, one demux):
1. The demux receives chunk P1 for `channel_id`; `route_payload`'s
map lookup misses (the adopting side has not run
`adopt_channel` yet) → P1 is parked in `early_arrivals` (FIFO).
Possibly P2…
2. The adopting side's `adopt_channel` runs: `channels.insert`
stores the state carrying the new `demux_sender`. **The map
lock is released here.**
3. The demux receives chunk C for the same `channel_id` *now*, inside
the window between the insert (step 2) and the drain
(`drain_early_arrivals`, step 4): `route_payload`'s lookup **hits**
→ C is sent **directly** into the mpsc.
4. `drain_early_arrivals` then pushes P1 (and any other parked
chunks) into the **same** mpsc.
Delivery order on the byte stream: **C, then P1** — with P1 having
arrived on the wire *earlier*. The mpsc preserves send order, and the
send order is inverted relative to wire-arrival order. The receiver
(`MpscRecvStream`) has no way to know; it reassembles faithfully in
send order = wrong order.
The window is the wall-clock span between `manager.rs:394` (insert
completes) and the drain's first take — normally nanoseconds apart,
which is why isolated runs never observe it; under scheduling pressure
(spawned adoption task descheduled immediately after the insert, or
the pending demux `send` completing faster than the adoption task's
next lock acquisition) the window is real and reproducible.
**Observed evidence** (alkgit `tests/native_transport.rs`
`the_channels_open_op_serves_a_fetch_session_on_the_opened_channel`;
11 tests, both processes multi-threaded, ~2 failures in ~10
suite-adjacent runs, 0 in ~15 isolated single-test runs). The test
reads the server's V2 capability advertisement back from the channel
as a raw pkt-line stream (`read_exact` 4-byte hex headers). The
advertisement is a fixed byte sequence (4+10 bytes `version 2`, hex
headers + `ls-refs=unborn` / `fetch=wait-for-done` /
`object-format=sha1` lines), so any desync is attributable to
delivery, not content:
- Failure shape 1 — after **0** lines, the first 4-byte header read
came out `fetc` = `[0x66,0x65,0x74,0x63]`, followed by
`h=wait-for-done`. That is offset ~16 bytes into the advertisement
(inside the `fetch=wait-for-done` line, four bytes past its
header): **the leading chunks were delivered late**; the reader
started mid-stream.
- Failure shape 2 — after **1** line (`version 2` parsed correctly),
the next header read came out `00ev` = `[0x30,0x30,0x65,0x76]`
(the `00` prefix of a hex length straddled with text): a **chunk
boundary two bytes into the second line's header** — the second
chunk was delivered ahead of (or its tail interleaved against) the
chunk carrying the line it belongs to.
Both shapes are exactly the C-befor-P1 inversion at different chunk
granularities (chunk sizes vary with scheduling).
A further aggravation this review flags on the same paths: the window
also interacts with `park_early_arrival`'s
`EARLY_ARRIVAL_CAP` overflow branch (`manager.rs:542-551`) — a burst
that parks *while* the window delays the drain can additionally
**drop** chunks. That combination is loss, not just misorder; any fix
must be ordering-exact *and* keep the drop path unreachable while the
channel is installed-but-undrained.
**Why the consumer cannot fix it** (and why no downstream workaround
exists or should): the reassembled `BiStream` is the promised byte
stream; a framing codec (alkgit's pkt-line reader, a tunnel's inner
protocol) consumes bytes with no awareness that chunk reassembly
happened beneath it. Alkgit carries no compensating logic and must
not — the fix belongs here, in the one place that owns delivery order.
### Fix-shape sketch (for the implementing patch)
The window closes when *all* chunks of a channel are serialized
through one ordering decision. Candidate directions, in increasing
invasiveness:
1. **Drain under the channels-map lock.** Hold the map lock across
the insert *and* the parked-queue swap-out, then push the swapped
queue into the mpsc (the direct `route_payload` paths arriving
behind the lock block until the drain's queue is out; they then
send directly — after the parked bytes). The push of the swapped
queue must not await while the lock is held: `try_send` first,
draining backpressure by holding the *queue*, not the lock, gets
complicated — see direction 3 for a clean cut.
2. **Re-check in `route_payload`'s park branch.** After parking,
re-check the map; if the channel now exists, take the queue back
out (FIFO) and push it through the sender, then the just-parked
chunk. This closes the "parked-after-install" half but leaves the
window's other half (a direct send overtaking parked chunks whose
drain hasn't run yet) — a direct route still sends C before the
adopter pushes P1. Not sufficient alone.
3. **Per-channel ordered mailbox.** Make every chunk for an installed
channel take the channel's own queue-lock, append, and hand off to
a single per-channel drainer (or drain under the queue lock with
`try_send` + a parked-overflow flag the drainer resumes). This
serializes park, install-drain, and direct route on one lock per
channel and closes every interleave, including the push-first
producer shapes of 006 E-04.
Whichever shape lands must add a **chunk-granularity ordering
regression test**: a producer that emits many small chunks with an
open-reply-then-immediate-pump shape, the adopter parked mid-window
under injected scheduling (e.g. a `park` latch between insert and
drain — a test-only seam), asserting the receiver's byte stream is
byte-identical to the producer's. The two observed desync shapes above
are the acceptance shapes. The `route_payload_to_open_channel_succeeds`
single-chunk test does not exercise the ordering and must not be
counted as covering this.
**Downstream re-verification gate**: alkcall's fix must be
re-verified against alkgit's flaking row under multi-process load
(the condition that reproduced it) — the alkgit review task
(`core/review-wave-2`, finding F-2) records this; wave 8's
end-to-end `git push`/`git clone` rows ride channel adoption with
push-first producers and would exercise the window hard without the
fix.
## Non-findings (verified, recorded to bound the re-review)
- The demux/mux wire format is not implicated: the 8-byte chunk header
is per-chunk framing only, and the misorder below it is a manager
scheduling property (ADR-034's wire surface untouched; no ADR-level
wire change is needed for the fix).
- `MpscRecvStream`'s pending-chunk continuation is correct — the
observed corruptions are all *order* corruptions, never
*content* corruption; the reassembly layer faithfully delivered
what it was handed.
- The install-side duplicate/collision checks in `adopt_channel`
(`ChannelExists` + the 0.8.1 regression pin) are unaffected by this
finding; the fix must preserve them and the adopt-parity rules.
## References
- alkgit wave-2 gate review:
`/workspace/@alkdev/alkgit/tasks/implementation/core/review-wave-2.md`
(finding F-2, with the repro data and the fix-urgency routing) and
the diagnostic commit alkgit `b6db513`
(`tests/native_transport.rs` → `read_channel_pkts`'s desync panic).
- Review 006 E-04 (the early-arrival cap half of this family;
`docs/reviews/006-channel-open-establishment-gap-review.md`),
review 007 (the establishment follow-ups).
- `docs/architecture/channels-adapter.md` (the demux invariants
REQ-CH-02/04 this finding sits between),
`docs/architecture/channels-wire.md` (the ordered-substrate premise,
line 9; REQ-CH-01..05 wire invariants),
`docs/architecture/decisions/039-channelsadapter-and-channelmanager.md`
(the substrate-agnostic demux loop contract).
- alkcall `a22b2b8` — the original park-early-arrival fix (the
open/first-data race this review's window is the residue of).
+1 -1
View File
@@ -27,7 +27,7 @@ dependencies = [
[[package]]
name = "alkcall"
version = "0.9.0"
version = "0.10.1"
dependencies = [
"async-trait",
"bytes",
+320 -43
View File
@@ -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<HashMap<u32, ChannelControl>>,
}
/// 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<u32, ChannelState>,
early_arrivals: &Mutex<HashMap<u32, VecDeque<Bytes>>>,
channel_id: u32,
state: ChannelState,
demux_sender: &mpsc::Sender<Bytes>,
) {
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");
}
}