alktunnels/tasks/tunnels/bindfirst-producer.md

id, name, status, depends_on, scope, risk, impact, level, tags

id	name	status	depends_on	scope	risk	impact	level	tags
tunnels/bindfirst-producer	Bind-first establisher + accept dispatcher + registration (ADR-008 shape 3)	completed	tunnels/bindfirst-params	broad	high	component	implementation	producer establisher bind-first accept-dispatcher forwarded-op

Description

Implement the bind-first establisher (producer.md §The Establisher, shape 3; ADR-008 §1): the establisher binds (via the injected BindFn), mints the correlation token, contributes bound + listen to the reply, and returns the plan; the pump handler is the accept dispatcher: it accepts the channel's BiStream, runs the assembly-provided accept loop, and per accept opens channels/tunnel/forwarded back toward the listen opener and pumps the accepted halves against the adopted channel. Plus register_tunnel_bindfirst_openable wiring it.

The establisher (shape 3, bind-first)

pub type BindFn = Arc<
    dyn Fn(&str) -> BoxFuture<'static, Result<BoundListener, TunnelEstablishError>>
        + Send + Sync,
>;

/// What the assembly's bind produces: the opaque listener (the accept
/// loop's business) + the OS-chosen bound address (the `bound` reply
/// field's source). The protocol never binds (OQ-TN-04).
pub struct BoundListener {
    pub listener: Arc<dyn Any + Send + Sync>,   // assembly-typed
    pub bound: SubstrateAddr,
}

pub fn bindfirst_establisher(
    registry: ResourceRegistry,   // namespace gate only, as listen_establisher
    bind: BindFn,
) -> OpenEstablisher

• Flow: parse TunnelParams → registry namespace gate (unknown_resource on miss, the same posture as listen_establisher) → mint the correlation token (per-open-fresh, per-producer-unique; tunnels/bindfirst-params) → bind(substrate, backing) → contribute reply fields (Establishment::with_reply_field("bound", …).with_reply_field("listen", token)) → Establishment::new(Arc::new(BindFirstPlan { listener, token })).
• Error mapping (producer.md shape-3 table): unbindable/held port → dial_failed (the bind IS the dial here); bind resource exhaustion → resource_shortage; registry miss → unknown_resource; internal → handler_error. The BindFn returns the typed errors; the establisher passes them through.
• The token lives in BOTH the reply and the plan (the dispatcher needs it; the opener needs it via the reply).

The accept dispatcher (the shape-3 pump handler)

pub fn make_bindfirst_accept_handler(
    accept_loop: BindFirstAcceptLoop,
    open_forwarded: ForwardedOpener,
) -> OpenHandler

• Injection seams (the design pin: decide the exact closure signatures during implementation, document in Notes): the dispatcher needs two assembly-provided capabilities the protocol cannot own:
  1. The accept loop, which pops accepted connections from the BindFirstPlan's opaque listener and yields (SubstrateAddr /* peer */, TargetHandle /* accepted halves */). Substrate code lives behind the closure (ADR-004).
  2. The forwarded-open caller: opens OP_TUNNEL_FORWARDED with TunnelForwardedParams { listen: token, substrate, peer } on the channel-0 call plane toward the listen opener, adopts the opener-allocated ID (adopt_channel on the producer-side manager), and returns the streams. This needs the call plane: the per-channel Connection the handler receives is stream-less (open_bi is StreamClosed; ADR-035) and carries no call surface, so the registration must close over the serving session's Arc<CallConnection> (the reverse POC's open_reverse_channel shape: call_with_payload on channel 0 → extract channel_id from the reply → manager.adopt_channel). The registration function takes the call-plane handle (the assembly has it: it constructed the serving session); document the wiring.
• Handler contract (producer.md §The Pump Handler item 5 + R-02): conn.accept_bi() FIRST (the yield-once contract; the wrapper's accepted-flag telemetry requires it), then run the accept loop concurrently with holding the channel's stream. The listen channel's own BiStream stays idle (no pumps on it); its EOF cancels the loop and tears the listener down via select! on the idle stream's EOF vs the accept loop (read from the stream and observe EOF; the zero-byte sentinel is the opener dropping the listener session). Per-accept data planes are pump_bidi (convention 8, never a hand-rolled loop over one stream): each accepted handle pumps against its adopted forwarded channel inside a spawned task tracked by the handler's JoinHandle set.
• Lifetime (R-02): the handler's returned JoinHandle tracks the whole accept loop + every per-accept pump it spawned (collect the pump handles; the handler task awaits them on loop exit, since a forwarded pump outliving its listener is a leak). Handler exit → wrapper teardown (the listen channel reaps) → the far listener tears down (the BindFn's listener drops with the plan/loop).
• Error posture per accept: a forwarded-open failure (opener without the scope → FORBIDDEN; old opener → NOT_FOUND; correlation miss → handler_error-class) logs and drops that accepted connection; the loop continues, and the listener's overall health is not hostage to one failed accept. (Decide: retry/backoff. v1 says no; document.)

Registration

pub fn register_tunnel_bindfirst_openable(
    core: &ChannelCore,
    registry: &ResourceRegistry,
    on_registry: &Arc<OperationRegistry>,
    auth: AuthContext,
    call_plane: Arc<CallConnection>,       // the producer-side call plane for forwarded opens
    manager: &ChannelManager,              // the producer-side manager (adopt the forwarded ids)
    bind: BindFn,
    accept_loop: BindFirstAcceptLoop,
    identity_witness: Option<IdentityWitness>,
) -> Result<(), String>

• Same spec (tunnel_open_spec()) + scope gate (tunnel:open) as the dial/listen variants; the bind-first shape is a per-registration establisher choice over the same op (ADR-008 §5's per-registration posture). One establisher per op id per registry (same note as register_tunnel_listen_openable).
• The forwarded-op SERVING side (opener side) is a separate registration shape (register_openable, with no establisher) and lands with tunnels/forwarded-receipt; this task's tests still need an opener-side stub (see Tests).

Tests (integration: the duplex harness, listen-topology extension)

• The full bind-first flow e2e: opener opens channels/tunnel/sub with the bind-first producer registered → reply carries bound (the real OS/duplex-chosen address) + listen (the token) → the opener serves channels/tunnel/forwarded (scope tunnel:forwarded) with a correlation map keyed on the token → the test accept loop accepts a duplex connection → the producer's dispatcher opens forwarded with {listen: token, substrate, peer} → the opener correlates, adopts, pumps against the "local BIND client" halves → byte round-trip through the forwarded channel. This test IS the ADR-008 acceptance gate.
• Reply-field projection e2e: open_channel_with_reply sees both fields; a dial establisher's open sees neither (the reserved-key guard upstream guarantees channel_id stays wrapper-owned).
• EOF teardown: drop the listener session (close the listen channel's stream) → the accept loop cancels, the far listener is gone (a follow-on accept attempt fails), the forwarded channels already adopted own their teardown independently.
• Per-accept failure isolation: opener rejects one forwarded open (e.g. scope-less identity on a second opener registry) → that accept's connection drops, the loop serves the next accept.
• Establishment error paths: bind failure → dial_failed-class on the wire (channel:open_failed); registry miss → unknown_resource.
• Token discipline: two opens on the same producer mint distinct tokens; each dispatcher stamps ITS OWN plan's token.

Acceptance Criteria

• bindfirst_establisher: registry gate → mint token → bind → bound+listen reply fields → BindFirstPlan plan-flow
• The accept dispatcher: accept_bi first (R-02 telemetry), idle-stream EOF cancels the loop + tears the listener down, per-accept pump_bidi tasks tracked by the handler JoinHandle
• The forwarded-open caller rides the call plane (call_with_payload + adopt, the reverse POC's shape); the registration wiring documents where the call plane comes from
• register_tunnel_bindfirst_openable wires spec + establisher + dispatcher + injections
• Integration tests: the full flow e2e (the acceptance gate), reply-field projection, EOF teardown, per-accept isolation, typed establishment errors, token discipline
• Clippy/fmt clean; wasm32 check passes (closures injected; no substrate code in the default crate)
• cargo test green

References

• docs/architecture/decisions/008-listen-metadata-forwarded-op.md (§1 shape 3, §2 the reply fields, §3 the forwarded op, Amendment 1 on the token; §6 the hub policy sentence)
• docs/architecture/producer.md §The Establisher (shape 3) + §The Pump Handler (item 5) + §The Forwarded Op
• tasks/tunnels/bindfirst-params.md (the types this consumes)
• tasks/tunnels/producer-open-op.md + producer-listen.md (the harness + registration shapes this extends)
• alkcall surface: Establishment::with_reply_field (ADR-049 amendment 3, landed 0.8.0), the OpenHandler signature (no channel ID; Amendment 1's premise), ChannelRelay reply-field pass-through (ADR-051, the relay-transparency property the tests should pin)

Notes

• Closure signatures decided (the design pin): BindFn / BindFirstAcceptLoop / ForwardedOpener follow the type-erased closure precedent:
  • BindFn = Arc<dyn Fn(Substrate, &str) -> BoxFuture<'static, Result<BoundListener, TunnelEstablishError>> + Send + Sync>, same (substrate, backing) form as DialFn (the registry backing string rides the same signature; the bind closure stays substrate-agnostic from the protocol's view).
  • BindFirstAcceptLoop = Arc<dyn Fn(Arc<dyn Any + Send + Sync>) -> BoxFuture<'static, Result<(SubstrateAddr, TargetHandle), TunnelEstablishError>> + Send + Sync>, and the closure receives the plan's OPAQUE listener (Arc<dyn Any + Send + Sync>, no downcast in the protocol: the assembly downcasts to ITS concrete type inside the closure; the same injection pattern as DialFn/AcceptFn). A returned Err is LOOP-CRITICAL (the substrate accept source broke), so the dispatcher ends the loop; per-accept forwarded-open failures never come from this closure.
  • ForwardedOpener = Arc<dyn Fn(TunnelForwardedParams) -> BoxFuture<'static, Result<(MpscSendStream, MpscRecvStream), TunnelEstablishError>> + Send + Sync>, which closes over the serving session's Arc<CallConnection> + the producer-side ChannelManager (the reverse POC's open_reverse_channel shape: call_with_payload on channel 0 → extract channel_id → manager.adopt_channel). Where the call plane comes from: the assembly layer holds it, since it constructed the serving session (the harness's wire_bindfirst shows the wiring: the ForwardedOpener closure closes over the Arc<ChannelClient> and calls call_open_op + adopt_channel; in production the assembly hands its CallConnection the same way).
• Registration signature deviation (documented): the task sketch had 8 parameters (clippy's too_many_arguments); the injected seams bundle into BindFirstInjections {bind, accept_loop, open_forwarded}, one coherent injection point passed beside the core pair + the optional witness. Same posture as DialFn riding the dial registration. The identity_witness seam is wired via a small witness_establisher wrapper (records the per-call opener identity, then delegates; the same CF-006 seam the dial/direct variants inline).
• The token mint (mint_token, private): AtomicU32 counter seeded at 1 inside the establisher's closure state; fetch_add returns the PREVIOUS value, so the minted stream is 1, 2, 3, …; token 0 (an illegal wire value) is never minted, and a u32 wrap that lands on 0 re-increments (the loop skips it). Per-open fresh, per-producer unique; unit-pinned including the wrap shape.
• The idle-stream EOF watch: a 1-byte read on the split idle half inside a select! against the accept loop: Ok(0) (the zero-byte EOF sentinel), a read error, or (defensively) any non-empty read on an idle stream all cancel the loop. The listen channel's stream is never pumped (no pump_bidi on it).
• Per-accept failure posture (v1, documented): a forwarded-open failure (scope-less opener → FORBIDDEN; old opener → NOT_FOUND; correlation miss → handler_error-class) logs a warning and drops that accepted connection; the loop continues, with NO retry/backoff (v1 posture per the task).
• Lifetime (R-02): the handler collects every per-accept pump_bidi JoinHandle in a Vec and awaits them all on loop exit (before the handler task returns); a forwarded pump never outlives its listener. Handler exit → wrapper teardown → the listen channel reaps → the plan's listener drops with the loop.
• Harness changes (test infrastructure): wire_bindfirst (RegistrationMode::BindFirst); the consumer's install hook gained a shared serving registry + the accept-side identity propagation (the transport identity attaches to channel 0, the same posture wire_forward models; needed because the forwarded opens arrive ON the consumer's serving path); Topology.consumer_registry exposed; the forwarded-op serving stub (register_forwarded_stub_on_consumer / make_forwarded_stub_handler: correlate by listen token → accept the BiStream → pump_bidi against the correlated halves; the opener-side registration shape lands for real with tunnels/forwarded-receipt).
• Test-data-plane shape: the e2e round-trip drives the local BIND client end directly (the correlation map's halves are one end of a raw duplex; the stub's pump bridges forwarded channel ↔ those halves; the accepted side's echo task closes the loop). The forwarded channel on the opener side is wrapper-managed (the serving wrapper allocated it, so adopt_channel on it a second time fails ChannelExists), so the test drives the client end, not a re-adopt.
• EOF-teardown observability: the test pins the listen channel reaping on the producer side after the opener drops its session (has_channel false) and the accept loop's death via the queue (a follow-on push is never consumed).

Summary

Implemented the bind-first producer (shape 3, ADR-008 as amended) per the task. src/producer.rs: BindFn + BoundListener (the platform-boundary bind seam); the token mint (non-zero-seeded AtomicU32, per-open fresh, per-producer unique, token 0 never minted, the wrap shape unit-pinned); bindfirst_establisher (registry namespace gate → mint → bind → bound + listen reply fields → the BindFirstPlan plan; the typed errors pass through: bind failure dial_failed, exhaustion resource_shortage, registry miss unknown_resource); BindFirstAcceptLoop (the type-erased accept closure receiving the opaque listener); ForwardedOpener (the call-plane forwarded-open caller: call + adopt); make_bindfirst_accept_handler (the accept dispatcher: accept_bi first, the idle-stream EOF sentinel cancels the loop via select!, per-accept pump_bidi in spawned tasks tracked by the handler's JoinHandle, per-accept failure isolation, loop-critical accept errors end the loop); register_tunnel_bindfirst_openable (same spec + scope gate, BindFirstInjections bundle + witness seam), re-exported through lib.rs. tests/harness.rs: RegistrationMode::BindFirst + wire_bindfirst (the consumer hook gained the accept-side identity propagation + a shared serving registry, Topology.consumer_registry exposed) + the forwarded-op serving stub (make_forwarded_stub_handler). tests/ producer_bindfirst.rs has 6 integration tests: the full flow e2e (the acceptance gate: reply projection → token correlation → forwarded open → byte round-trip through the whole path), the dial-open neither-field projection, EOF teardown (listen reaped, loop gone), per-accept isolation (scope-less opener; the loop serves the next accept), typed establishment errors, and the token discipline (two opens distinct tokens; each dispatcher stamps its own plan's token). Plus 6 unit tests in producer.rs (the mint discipline incl. the wrap shape, the establisher flow at the Establishment surface (plan token == reply token), distinct tokens per open, registry miss, bind-error passthrough, the witness wrapper).

Verified: cargo test green (118 default; 138 --all-features), clippy --all-targets -D warnings clean (native + wasm32), fmt clean, wasm32-unknown-unknown check passes (closures injected; the default crate stays wasm-clean), cargo doc --no-deps clean. CHANGELOG updated (Unreleased → Added).