fix: bind-first dispatcher reads substrate from the plan (review 002 M-1)
- 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.
This commit is contained in:
1 parent
3df5c484c6
commit
57a728dc4b
6 files changed
+273
-21
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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`).
|
||||
|
||||
+41
-10
@@ -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", &"<opaque>")
|
||||
.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<ChannelPlan>, conn: Connection, _auth: AuthContext| {
|
||||
move |_input: Value, plan: Option<ChannelPlan>, 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::<TunnelParams>(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<dyn Any + Send + Sync>,
|
||||
token: 918273,
|
||||
substrate: Substrate::Udp,
|
||||
});
|
||||
let unwrapped = plan
|
||||
.downcast::<BindFirstPlan>()
|
||||
.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::<TestListener>()
|
||||
@@ -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::<BindFirstPlan>().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::<TestListener>()
|
||||
|
||||
@@ -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.
|
||||
> 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::<TunnelParams>` 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`.
|
||||
@@ -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<tokio::sync::Mutex<Vec<alktunnels::params::Substrate>>>;
|
||||
|
||||
/// 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<alkcall::channels::operations::ChannelPlan>,
|
||||
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).
|
||||
|
||||
Reference in new issue
Block a user