alksocks/tasks/implementation/core-consumer-bind.md

---
id: implementation/core-consumer-bind
name: "Core: consumer BIND, BindSession (reply#2 reader) + e2e tests"
status: completed
depends_on: [implementation/core-consumer-connect, implementation/core-producer-bind]
scope: narrow
risk: medium
impact: component
level: implementation
---

## Description

Implement the BIND consumer session in `src/consumer.rs` per
consumer.md §The BIND client (POC #5's `read_reply2` sketch): the
wrapper-aware client that opens the channel, sends BIND, reads
reply#1, awaits the inbound accept, and reads reply#2 before riding
the pass-through data plane.

Deliverables (port from `/workspace/alksocks-bind-poc/src/consumer.rs`,
adapted to the crate's error surface):

1. **`BindSession::bind(client, target)`**: opens the channel,
   negotiates, sends the BIND command: the forked client
   `request()` reads reply#1 (carrying BND.ADDR).
2. **The reply#2 reader (`await_inbound`)**: reply#2 is hand-read
   with the forked public `read_address` codec (reply#1's codec
   applied again with **no second codec**, ADR-011's rationale). The
   session then rides the pass-through data plane
   (`BindDataPlane`-shaped: `into_inner()`/`get_socket_mut()`).
3. **Error surface**: a refused BIND (the command gate's 0x02, or a
   factory failure's reply#1 code) surfaces as a typed session
   error carrying the RFC reply code; accept-stage failures surface
   at `await_inbound` as reply#2's mapped code.
4. **Teardown ownership**: the session owns the channel; on Drop it
   closes/joins. No `Clone` (compile-asserted).
5. **Tests** (end-to-end through the harness): bind → reply#1
   (BND.ADDR as the harness accept advertises) → reply#2 (the peer
   address) → pass-through echo both directions; the command-gate
   refusal surfaced as a typed error (a `socks5:open`-only identity
   cannot bind); factory-failure and accept-stage-failure mappings
   surfaced; quiet-wait garbage from the client side ends the
   session; the three-command regression guard from the consumer
   side.

## Acceptance Criteria

- [x] `BindSession` with `bind`/`await_inbound`/data-plane accessors
      per the POC shape
- [x] Reply#2 read via the forked `read_address` codec (no second
      codec)
- [x] Typed error surface (reply-code-carrying session errors)
- [x] Teardown ownership on Drop; no `Clone` (compile-asserted)
- [x] E2E tests through the harness (round-trip, gate refusal,
      failure mappings, regression guard)
- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean

## References

- docs/architecture/consumer.md §The BIND client
- docs/architecture/decisions/007-bind-in-base.md (the client-side
  phase reading)
- POC source: `/workspace/alksocks-bind-poc/src/consumer.rs`
  (`BindReply1`/`BindDataPlane`/`read_reply2`)

## Notes

> Filled by the implementation agent (2026-09-24):

- The POC's `BindReply1` handle is absorbed into the teardown-owning
  session (the alktunnels ADR-005 posture the CONNECT session
  pinned): `BindSession` captures reply#1's BND.ADDR at `bind` (the
  forked client `request()` returns it once) and exposes the phase
  through borrows: `reply1()` yields the `BindReply1` handle
  (`bind_addr` + `raw_mut`), `await_inbound(&mut self)` reads reply#2
  and returns the borrowed `BindDataPlane` (`Ok(None)` on a spent
  session). The session stays the single teardown owner for every
  shape; the POC's `(session, channel_id)` tuple and the
  `BindDataPlane` ownership split collapse into one owner.
- `read_reply2` is the POC's port: 4-byte head → version pin →
  `ReplyError::from_u8` → the forked public `read_address` codec (no
  second codec, ADR-011). Errors ride `SocksSessionError` (the
  `SocksError` conversions routed through the forked vocabulary, per
  the crate's error-surface convention (`AddrError`/`ReplyError` are
  wrapped into `SocksError` first because the session error type
  exposes `From<SocksError>`, not the RFC inner enums directly).
- `BindDataPlane<'s>` borrows the session's stream (the CONNECT
  session's `SplitPipe` shape, generalized): `into_inner()` yields
  the `&mut Socks5Stream<BiStream>` (the teardown owner rides the
  session), `get_socket_mut()` is the forked client's socket accessor
  (the POC shape), and `AsyncRead`/`AsyncWrite` are forwarded so the
  plane is a plain pipe (`pump_bidi`-shaped consumers work without
  the accessors). `BiStream` is not `Sync`, so the plane pins `Send`
  (runtime probe), the same shape the forked client type has.
- Teardown: Drop = sync reap + structural EOF (the CONNECT posture);
  `close()` = half-close (the EOF sentinel reaches the producer's
  pump) then reap. Teardown-at-birth on the in-stream refusal path
  (command-gate 0x02, factory-failure reply#1): the adopted channel
  is reaped before the error returns. Post-reply#1 failures
  (accept-stage, quiet-wait garbage) keep the channel: reply#1
  succeeded, the listener existed; `close`/`Drop` reaps.
- The quiet-wait garbage path surfaces at `await_inbound` as an
  `Io`-shaped error (the producer ends the session with no reply#2;
  the consumer's reply#2 read EOFs), the client-side mirror of the
  producer's `UnexpectedUdpControlGarbage` exit, pinned by test.
- Harness reuse: no harness changes. The consumer-side topology's
  `wire_consumer_side_with` + the producer-bind task's
  `echo_accept`/`refusing_accept*` closures cover every test; the
  parked-accept factory for the quiet-wait garbage test is
  test-local (deterministic garbage-vs-accept branch).

## Summary

Implemented the BIND consumer session in `src/consumer.rs` (POC #5's
shape, adapted to the crate's teardown-owning session posture):
`BindSession::bind` (open → no-auth → BIND → reply#1, capturing
BND.ADDR), `reply1()` (the `BindReply1` handle: the app server's
out-of-band dial address + `raw_mut`), `await_inbound` (the POC's
`read_reply2`, hand-read with the forked `read_address` codec, with no
second codec) returning the `BindDataPlane` (`into_inner`/
`get_socket_mut` + `AsyncRead`/`AsyncWrite` forwards), and the
CONNECT session's teardown ownership (Drop sync-reap + structural
EOF, `close()` graceful, teardown-at-birth on in-stream refusals, no
`Clone`, with `compile_fail` asserting it). Typed error surface throughout:
open failures are `SocksSessionError::Open` (ADR-049 reasons); the
command gate's 0x02 and factory-failure reply#1 codes surface from
`bind` as `Rfc`; accept-stage reply#2 codes surface at
`await_inbound`. E2E suite (`tests/consumer-bind.rs`, 9 tests over
the harness): the round-trip (reply#1 BND.ADDR → reply#2 peer →
pass-through echo both directions), the data-plane accessors
(`into_inner`/`get_socket_mut`/`raw_mut` on one pipe), the
command-gate refusal (typed 0x02, no phantom channel), the
factory-failure (reply#1 code) and accept-stage (reply#2 code)
mappings, quiet-wait garbage ending the session, drop teardown +
connection health, the CONNECT+BIND regression guard on one
registration, and the `Send` compile-shape probe. Verification:
`cargo test` (259 incl. the doctest), clippy `-D warnings` (all
targets), fmt, and both wasm checks clean.