From 57a728dc4b2633649d6108bca41a1624d7f9b1a5 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 20 Sep 2026 07:00:18 +0000 Subject: [PATCH] fix: bind-first dispatcher reads substrate from the plan (review 002 M-1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BindFirstPlan gains substrate: Substrate, stamped at establisher time from the one successful TunnelParams parse (one parse, one source of truth; the plan-payload change is pre-consumer) - make_bindfirst_accept_handler drops the second params re-parse and the silent unwrap_or(Substrate::Tcp) fallback entirely — the dispatcher reads plan.substrate; no fallback branch remains, so a future drift cannot silently stamp Tcp into a UDP listen tunnel's forwarded opens (the raw-vs-framed data-plane mismatch class) - tests: forwarded_opens_carry_plan_stamped_substrate pins the plan as the substrate's source over a real UDP bind-first wire flow (a substrate-recording forwarded stub observes substrate: udp); the unit downcast + establisher-flow tests pin the stamp - docs: producer.md (the plan carries {listener, token, substrate}; the forwarded open's substrate also comes from the plan), ADR-008 §1 (the stamp + the removed fallback) - task review-002-m1-substrate-fallback: completed (Notes filled) Verification: cargo test (145 pass), cargo test --all-features (153), clippy --all-targets -D warnings (default + all-features), fmt --check, doc --no-deps, wasm32 check + clippy -D warnings. --- CHANGELOG.md | 13 ++ .../008-listen-metadata-forwarded-op.md | 7 +- docs/architecture/producer.md | 12 +- src/producer.rs | 51 +++++-- .../review-002-m1-substrate-fallback.md | 69 ++++++++- tests/producer_bindfirst.rs | 142 ++++++++++++++++++ 6 files changed, 273 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 59f5f99..0e52525 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -164,6 +164,19 @@ dependency bump below is the only behavior-affecting change so far No alktunnels API or wire change in this bump; 71/91 tests pass, wasm-clean verified. +- **`BindFirstPlan` gains `substrate: Substrate` (review 002 M-1 + remediation)** — the accept dispatcher's silent + `unwrap_or(Substrate::Tcp)` fallback (the second + `TunnelParams` re-parse of wire input) is removed: the establisher + stamps the parsed substrate into the plan (one parse, one source + of truth) and the dispatcher reads it from the plan it already + downcast. A drift can no longer silently stamp `Tcp` into a UDP + listen tunnel's forwarded opens (a raw-vs-framed data-plane + mismatch on the opener side). Plan-payload change is pre-consumer + (no external `BindFirstPlan` consumer yet). Pinned by + `forwarded_opens_carry_plan_stamped_substrate` (a UDP bind-first + flow whose forwarded opens carry `substrate: udp` from the plan). + ## [0.1.0] - 2026-09-10 The initial release: the v1 tunnel protocol implementation over diff --git a/docs/architecture/decisions/008-listen-metadata-forwarded-op.md b/docs/architecture/decisions/008-listen-metadata-forwarded-op.md index 3c5f514..32595fc 100644 --- a/docs/architecture/decisions/008-listen-metadata-forwarded-op.md +++ b/docs/architecture/decisions/008-listen-metadata-forwarded-op.md @@ -57,7 +57,12 @@ accept-first shape (which stays — see §5): - **The establisher mints a correlation token** (Amendment 1): a fresh u32 unique per open on this producer, carried twice — inside the plan (the pump handler needs it to correlate per-accept forwarded - opens) and as the additive `listen` reply field (§2). The original + opens) and as the additive `listen` reply field (§2). The plan also + carries the listen tunnel's substrate, stamped at establisher time + from the one successful params parse — the accept dispatcher reads + it from the plan and stamps it into each forwarded open's + `substrate` param; it never re-parses wire input (review 002 M-1: + the removed silent `Tcp` fallback). The original §3 key (the listen channel's ID) is superseded: the serving-side open-op wrapper never exposes the allocated channel ID to the establisher or handler (alkcall 0.8.0's `OpenHandler` surface), and diff --git a/docs/architecture/producer.md b/docs/architecture/producer.md index 5c5320b..627765c 100644 --- a/docs/architecture/producer.md +++ b/docs/architecture/producer.md @@ -158,7 +158,9 @@ dial: reply field (the same projection) and carried in the plan — the accept dispatcher reads it from the plan and stamps it into each forwarded open's `listen` param. The plan carries - `{listener, token}` (a bind-first plan type); the token is + `{listener, token, substrate}` (a bind-first plan type; the + substrate is stamped at establisher time — the dispatcher reads it + from the plan, never re-parses wire input); the token is relay-transparent by construction (reply fields and params both pass hubs untouched — alkcall ADR-051; ADR-008 Amendment 1 has the why-superseded of the channel-ID key). @@ -240,9 +242,11 @@ no establisher, the data plane starts immediately): 1. The producer's per-accept open carries `{listen, substrate, peer}` — the correlation token from the plan (the value the establisher minted; the opener matches it against the `listen` field of its - open listener sessions — ADR-008 Amendment 1), the substrate, the - accepted connection's remote address (reply#2-class fidelity's - carrier). + open listener sessions — ADR-008 Amendment 1), the substrate + (also from the plan — stamped at establisher time from the one + successful params parse; the dispatcher never re-parses wire + input), the accepted connection's remote address (reply#2-class + fidelity's carrier). 2. The opener's registry allocates the `channel_id`; the producer adopts it (`TunnelSession::adopt`) and pumps the accepted halves against it (`pump_against`). diff --git a/src/producer.rs b/src/producer.rs index cb825df..13e9436 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -72,7 +72,8 @@ impl std::fmt::Debug for TargetHandle { } /// The bind-first establisher's plan payload (ADR-008 §1 as amended): -/// the assembly's listener handle + the correlation token. +/// the assembly's listener handle + the correlation token + the listen +/// tunnel's substrate. /// /// **The listener is NOT `TargetHandle` halves** — a listener is /// accepted FROM, not pumped. The plan carries whatever the assembly's @@ -103,6 +104,14 @@ pub struct BindFirstPlan { /// unique per producer, stamped into each forwarded open's /// `listen` param. pub token: u32, + /// The listen tunnel's substrate (ADR-008 §3), stamped here at + /// establisher time (the establisher already parsed the params + /// successfully — one parse, one source of truth). The accept + /// dispatcher reads it from the plan it already downcast; it must + /// NOT re-parse wire input (a silent `Tcp` fallback on a parse + /// failure would mismatch a UDP listen tunnel's data-plane + /// framing — review 002 M-1). + pub substrate: Substrate, } impl std::fmt::Debug for BindFirstPlan { @@ -110,6 +119,7 @@ impl std::fmt::Debug for BindFirstPlan { f.debug_struct("BindFirstPlan") .field("listener", &"") .field("token", &self.token) + .field("substrate", &self.substrate) .finish() } } @@ -644,7 +654,11 @@ pub fn bindfirst_establisher(registry: ResourceRegistry, bind: BindFn) -> OpenEs })?; let token = mint_token(&counter); let BoundListener { listener, bound } = bind(params.substrate, &backing).await?; - let plan = Arc::new(BindFirstPlan { listener, token }) as ChannelPlan; + let plan = Arc::new(BindFirstPlan { + listener, + token, + substrate: params.substrate, + }) as ChannelPlan; let bound = serde_json::to_value(&bound) .map_err(|e| TunnelEstablishError::HandlerError(format!("bound serialize: {e}")))?; Ok(Establishment::new(plan) @@ -743,15 +757,17 @@ pub type ForwardedOpener = Arc< /// down (the plan's listener drops with the loop). /// /// The substrate discriminator for the forwarded opens rides the -/// open-op INPUT params (`TunnelParams.substrate` — the listen -/// tunnel's substrate; ADR-008 §3 "the listen tunnel's substrate -/// shapes the forwarded channel's data-plane framing"). +/// plan ([`BindFirstPlan`].substrate — stamped at establisher time +/// from the one successful `TunnelParams` parse; the listen tunnel's +/// substrate; ADR-008 §3 "the listen tunnel's substrate shapes the +/// forwarded channel's data-plane framing"). The dispatcher never +/// re-parses wire input — no fallback branch exists. pub fn make_bindfirst_accept_handler( accept_loop: BindFirstAcceptLoop, open_forwarded: ForwardedOpener, ) -> OpenHandler { Arc::new( - move |input: Value, plan: Option, conn: Connection, _auth: AuthContext| { + move |_input: Value, plan: Option, conn: Connection, _auth: AuthContext| { let accept_loop = Arc::clone(&accept_loop); let open_forwarded = Arc::clone(&open_forwarded); tokio::spawn(async move { @@ -763,9 +779,12 @@ pub fn make_bindfirst_accept_handler( tracing::error!("tunnel bind-first: no BindFirstPlan in plan"); return; }; - let substrate = serde_json::from_value::(input.clone()) - .map(|p| p.substrate) - .unwrap_or(Substrate::Tcp); + // The substrate rides the plan (stamped at establisher + // time — one parse, one source of truth): the + // dispatcher must not re-parse wire input (a parse + // failure here would have nowhere loud to go; review + // 002 M-1 removed the silent `Tcp` fallback instead). + let substrate = plan.substrate; let (mut idle_read, _idle_write) = tokio::io::split(bidi); // The per-accept pump task set — tracked by the @@ -1063,11 +1082,17 @@ mod tests { let plan: ChannelPlan = Arc::new(BindFirstPlan { listener: Arc::new(TestListener(42)) as Arc, token: 918273, + substrate: Substrate::Udp, }); let unwrapped = plan .downcast::() .expect("BindFirstPlan plan"); assert_eq!(unwrapped.token, 918273); + assert_eq!( + unwrapped.substrate, + Substrate::Udp, + "the plan's substrate rides it" + ); let listener = unwrapped .listener .downcast_ref::() @@ -1153,10 +1178,16 @@ mod tests { .and_then(|v| v.as_u64()) .expect("listen"); assert!(listen > 0, "the minted token is positive"); - // The plan: the BindFirstPlan carries the SAME token. + // The plan: the BindFirstPlan carries the SAME token + the + // parsed substrate. let plan = establishment.plan.expect("plan"); let plan = plan.downcast::().expect("BindFirstPlan"); assert_eq!(plan.token as u64, listen, "plan token == reply token"); + assert_eq!( + plan.substrate, + Substrate::Tcp, + "the plan stamps the parsed substrate" + ); let listener = plan .listener .downcast_ref::() diff --git a/tasks/tunnels/review-002-m1-substrate-fallback.md b/tasks/tunnels/review-002-m1-substrate-fallback.md index cb18d1f..81d0ad7 100644 --- a/tasks/tunnels/review-002-m1-substrate-fallback.md +++ b/tasks/tunnels/review-002-m1-substrate-fallback.md @@ -1,7 +1,7 @@ --- id: tunnels/review-002-m1-substrate-fallback name: "M-1 remediation — the accept dispatcher's silent substrate fallback (stamp the plan, fail loud)" -status: pending +status: completed depends_on: [] scope: narrow risk: low @@ -72,13 +72,13 @@ plan (the dispatcher must not re-parse wire input). ## Acceptance Criteria -- [ ] The silent `unwrap_or(Substrate::Tcp)` fallback is gone from +- [x] The silent `unwrap_or(Substrate::Tcp)` fallback is gone from `make_bindfirst_accept_handler` -- [ ] The chosen shape documented in this task's Notes (plan-stamp vs +- [x] The chosen shape documented in this task's Notes (plan-stamp vs log-and-return, with the rationale) -- [ ] A test pins the forwarded-op substrate source (plan-stamped, or +- [x] A test pins the forwarded-op substrate source (plan-stamped, or the loud teardown on parse failure) -- [ ] `cargo test` green; clippy `--all-targets -D warnings` clean; +- [x] `cargo test` green; clippy `--all-targets -D warnings` clean; fmt clean; wasm32 check + clippy clean ## References @@ -92,4 +92,61 @@ plan (the dispatcher must not re-parse wire input). ## Notes -> Agent fills during implementation. \ No newline at end of file +> Agent fills during implementation. + +**Chosen shape: 2 — stamp the plan** (the review's recommendation). +Rationale: one parse (the establisher's `parse_params` — which +rejects malformed input with `handler_error` before any channel +exists), one source of truth; the substrate is per-open-static and +the plan is already the per-open carrier; the fallback branch is not +just guarded against, it is REMOVED — a future drift cannot degrade +silently because there is no fallback to degrade into. Pre-consumer +window used (`BindFirstPlan` has no external consumer yet — a +plan-payload change is cheap). + +Implemented: + +- `BindFirstPlan` gains `substrate: Substrate` (doc comment states + why it rides the plan — the dispatcher must not re-parse wire + input; the removed fallback's data-plane mismatch risk is cited). + The `Debug` impl stamps it (the opaque listener field stays). +- `bindfirst_establisher` stamps the parsed `params.substrate` into + the plan at construction time. +- `make_bindfirst_accept_handler`: the second + `serde_json::from_value::` parse and the + `unwrap_or(Substrate::Tcp)` fallback are gone; the dispatcher + reads `plan.substrate` from the plan it already downcast. The + handler's `input` param is now unused (`_input`). + +Tests: + +- Existing suite kept green — the two-open token-discipline test + (`token_discipline_two_opens_distinct_tokens`) still pins + per-dispatcher stamping from its own plan. +- `bindfirst_plan_downcasts_through_channel_plan` (unit): the plan's + substrate rides the downcast (`Substrate::Udp` constructed → + recovered). +- `bindfirst_establisher_flow_end_to_end` (unit): the establisher + stamps the parsed substrate into the plan (`Tcp` input → + `plan.substrate == Tcp`). +- **`forwarded_opens_carry_plan_stamped_substrate` (integration, + the M-1 pin):** a UDP bind-first listen over the wire — the + forwarded open the dispatcher makes carries `substrate: udp` (the + plan's stamp), observed through a substrate-recording + forwarded-op stub on the opener side. The removed fallback would + have stamped `Tcp`; the assert is `vec![Substrate::Udp]`. + +Docs synced: `docs/architecture/producer.md` (§3 — the plan carries +`{listener, token, substrate}`; §The Forwarded Op item 1 — the +substrate also comes from the plan, never re-parsed), +`docs/architecture/decisions/008-listen-metadata-forwarded-op.md` +(§1 — the plan's substrate stamp + the removed fallback, review 002 +M-1 referenced). + +Verification (all green): `cargo test` (145 tests across the lib + +10 integration binaries), `cargo test --all-features` (153), +`cargo clippy --all-targets -- -D warnings`, +`cargo clippy --all-targets --all-features -- -D warnings`, +`cargo fmt --check`, `cargo doc --no-deps`, +`cargo check --target wasm32-unknown-unknown` + +`cargo clippy --target wasm32-unknown-unknown -- -D warnings`. \ No newline at end of file diff --git a/tests/producer_bindfirst.rs b/tests/producer_bindfirst.rs index b9cc2df..68ac856 100644 --- a/tests/producer_bindfirst.rs +++ b/tests/producer_bindfirst.rs @@ -539,6 +539,148 @@ async fn token_discipline_two_opens_distinct_tokens() { ); } +/// The forwarded-op serving stub's forwarded-substrate log: the +/// `substrate` param of every forwarded open (the M-1 remediation +/// probe surface — the substrate's SOURCE: the plan, not a re-parse). +pub type SubstrateLog = Arc>>; + +/// The forwarded-op serving stub variant that records each forwarded +/// open's `substrate` param, then pumps against the correlated halves +/// (the same shape as `make_forwarded_stub_handler` + the log). +fn make_substrate_recording_stub_handler( + correlations: CorrelationMap, + substrates: SubstrateLog, +) -> alkcall::channels::operations::OpenHandler { + Arc::new( + move |input: serde_json::Value, + _plan: Option, + conn: alkcall::core::types::Connection, + _auth: alkcall::core::auth::AuthContext| { + let correlations = Arc::clone(&correlations); + let substrates = Arc::clone(&substrates); + tokio::spawn(async move { + let forwarded: alktunnels::params::TunnelForwardedParams = + match serde_json::from_value(input) { + Ok(p) => p, + Err(e) => { + tracing::error!(error = %e, "substrate stub: params parse failed"); + return; + } + }; + substrates.lock().await.push(forwarded.substrate); + let target = correlations.lock().await.remove(&forwarded.listen); + let Some(target) = target else { + tracing::error!( + listen = forwarded.listen, + "substrate stub: correlation miss" + ); + return; + }; + let Some(bidi) = conn.accept_bi().await.ok() else { + tracing::error!("substrate stub: accept_bi failed"); + return; + }; + alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await; + }) + }, + ) +} + +/// The forwarded-op serving stub recording substrates, installed on +/// the opener (consumer) side (the same post-hoc registration posture +/// as `install_forwarded_stub`). +async fn install_substrate_recording_stub( + topo: &Topology, + substrates: SubstrateLog, +) -> CorrelationMap { + let correlations: CorrelationMap = Arc::new(tokio::sync::Mutex::new(Default::default())); + let core = alkcall::channels::operations::ChannelCore::new( + topo.consumer_manager.clone(), + alkcall::channels::policy::default_policy(), + ); + core.register_openable( + alktunnels::params::tunnel_forwarded_spec(), + make_substrate_recording_stub_handler(Arc::clone(&correlations), substrates), + &topo.consumer_registry, + alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"), + ) + .expect("register substrate-recording forwarded stub on the opener"); + correlations +} + +/// M-1 remediation pin: the forwarded opens' `substrate` comes from +/// the PLAN (stamped at establisher time), not a re-parse of wire +/// input — a UDP bind-first listen produces forwarded opens carrying +/// `substrate: udp` even though the re-parse-and-fallback shape would +/// have stamped `tcp` if it ever fell back. (The fallback itself is +/// REMOVED — this test pins the plan as the source by observing the +/// forwarded params through a real dispatcher round trip.) +#[tokio::test] +async fn forwarded_opens_carry_plan_stamped_substrate() { + init_tracing(); + // A UDP-flavored bind-first registry: the open is + // `{resource: "bind", substrate: "udp"}` — the plan must stamp + // Udp (the old fallback shape would have stamped Tcp on any + // parse failure). + let registry = ResourceRegistry::new(); + registry + .register("bind", Substrate::Udp, "assembly-owned") + .await; + let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())); + let notify = Arc::new(tokio::sync::Notify::new()); + let topo = wire_bindfirst( + registry, + test_bind(), + harness_queue_loop(queue.clone(), notify.clone()), + ) + .await; + + let substrates: SubstrateLog = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let correlations = install_substrate_recording_stub(&topo, Arc::clone(&substrates)).await; + + // The listen open over the UDP substrate (the harness's + // open-call helper hardcodes "tcp" — call with the udp input + // directly; the schema gate accepts the registered UDP resource). + let payload = serde_json::json!({ + "operationId": "channels/tunnel/sub", + "input": {"resource": "bind", "substrate": "udp"}, + }); + let response = topo.consumer_call.call_with_payload(payload).await; + let reply = response.result.expect("udp listen open"); + let listen_id = reply["channel_id"].as_u64().expect("channel_id") as u32; + let token = parse_listen_reply(&reply) + .expect("projected reply") + .listen + .expect("the `listen` token"); + + // The opener correlates on the token and adopts the listen + // channel (the idle side). + let (l_read, l_write, _, _) = bind_client_pair(); + correlations.lock().await.insert( + token, + TargetHandle { + read: Box::new(l_read), + write: Box::new(l_write), + }, + ); + topo.consumer_manager + .adopt_channel(listen_id, alktunnels::TUNNEL_ALPN, None) + .await + .expect("adopt listen channel"); + + // One accept: the dispatcher opens forwarded — the forwarded + // params' `substrate` must be the PLAN's (Udp), not a fallback. + push_accept(&queue, ¬ify, (198, 55236)).await; + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + + let logged = substrates.lock().await.clone(); + assert_eq!( + logged, + vec![alktunnels::params::Substrate::Udp], + "the forwarded open's substrate is plan-stamped (Udp), not re-parsed" + ); +} + /// A plain dial open sees NEITHER reply field (the projection /// property — the reserved-key guard upstream guarantees `channel_id` /// stays wrapper-owned; a dial establisher contributes nothing).