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

---
id: implementation/core-consumer-associate
name: "Core: consumer ASSOCIATE, AssociateSession + e2e tests"
status: completed
depends_on: [implementation/core-consumer-connect, implementation/core-producer-associate]
scope: moderate
risk: medium
impact: component
level: implementation
---

## Description

Implement the UDP ASSOCIATE consumer session in `src/consumer.rs`
per consumer.md §The associate session (POC #2's shape): the typed
client that opens the channel, receives the sentinel, and speaks the
framed datagram stage.

Deliverables (port from `/workspace/alksocks-udp-poc/src/consumer.rs`,
adapted to the crate's wire codec):

1. **`AssociateSession::associate(client)`**: opens the channel,
   negotiates (NoAuth), sends ASSOCIATE, reads the reply;
   **receives the sentinel BND.ADDR and never uses it for
   addressing** (the handle is the channel; ADR-003).
2. **`send_datagram(dst, bytes)`**: frames once
   (`[len: u16 BE][addr + data]` via the crate's wire codec, the
   framing-once mirror) and writes the frame.
3. **`recv_datagram()`** returns `(source_addr, bytes)` per
   datagram, and **the incremental reader state (mid-frame position +
   batched-datagram pending queue) persists across calls**: a
   datagram split across channel chunks survives a caller-driven
   receive loop (POC #2's finding: the first cut dropped the reader
   state per call; the 2-byte-prefix chunk test exposed it). Write
   the persistence test (the POC's chunk-splitting echo shape).
4. **`shutdown()`** half-closes the stream: the mux's implicit-EOF
   sentinel reaches the producer's stage as EOF and the association
   ends (association lifetime = channel lifetime).
5. **Teardown ownership**: the session owns the channel; on Drop it
   closes/joins (alktunnels ADR-005 posture). No `Clone`
   (compile-asserted).
6. **Tests** (end-to-end through the harness): associate →
   send/recv round-trips (multiple in-order datagrams, one
   association); the split-datagram persistence test; empty-payload
   datagram round-trip; shutdown → the producer's stage ends and a
   fresh associate succeeds immediately (the POC's test 4, from the
   consumer side); Drop teardown (drop mid-stream → the producer's
   stage sees EOF, the connection stays healthy); CONNECT +
   ASSOCIATE on the same registration (the regression guard
   extends).

## Acceptance Criteria

- [x] `AssociateSession` with `associate`/`send_datagram`/
      `recv_datagram`/`shutdown` per the POC shape
- [x] Reader state persists across `recv_datagram` calls (the
      split-datagram test)
- [x] Framing-once mirror (send frames exactly once; recv decodes
      incrementally)
- [x] Teardown ownership on Drop; no `Clone` (compile-asserted)
- [x] E2E tests through the harness (round-trips, shutdown, Drop
      teardown, regression guard)
- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean

## References

- docs/architecture/consumer.md §The associate session
- docs/architecture/decisions/003-udp-handle-swap.md (the consumer
  mirror invariants)
- POC source: `/workspace/alksocks-udp-poc/src/consumer.rs`
  (`UdpAssociation`, the reader-state finding's origin)

## Notes

- **`send_datagram` takes `impl ToTargetAddr`** (the CONNECT
  session's target-construction posture; under ADR-012, IP-string forms
  parse, domains pass through unresolved), not the POC's
  `&TargetAddr`: the same entry shape as `connect`. A
  `ToTargetAddr for &TargetAddr` impl was added to the forked
  address codec (our owned code, ADR-013) so callers can reuse a
  `TargetAddr` value across sends.
- **Teardown ownership matches the CONNECT/BIND sessions** (the
  crate's posture, strengthened from the POC's "close/join"): the
  session owns the adopted channel entry; Drop sync-reaps
  (`teardown_channel`, best-effort; `Drop` cannot await) + the data
  plane EOFs structurally (the mux's implicit-EOF sentinel);
  `shutdown()` half-closes (the graceful path, returning the reap
  result); `close()` is the `shutdown`+reap alias-shaped variant
  kept for the connect/bind symmetry; teardown-at-birth covers the
  in-stream refusal path (a refused ASSOCIATE never leaves a
  phantom channel).
- **`recv_datagram` returns `Result<Option<(TargetAddr, Bytes)>>`**
  (the POC's shape): `Ok(None)` = channel EOF (the association's
  end); a stream end mid-frame is fail-loud (an `UnexpectedEof`
  I/O error through `SocksSessionError::Rfc`; never a silent
  partial datagram); an undecodable frame (garbage ATYP) errors
  loud through the wire codec's `WireError` → RFC mapping
  (`AddrError` class for address-encoding refusals,
  `UdpHeaderError` for the garbage shapes, the consumer-side
  mirror of the producer stage's `UndecodableFrame` exit).
- **The pending queue stores decoded datagrams** (`(TargetAddr,
  Bytes)`) rather than the POC's pre-decode inner bytes; decoded
  storage keeps the delivery shape uniform (decode once per frame
  either way; no wire.rs change needed, since `decode_inner` stays
  private).
- **Unit tests ride a duplex-backed session** (`skip_auth`'s
  handshake-free `use_stream` path over a `BiStream::from_joined`
  duplex, no channels needed) for the reader-internals probes:
  reader-state persistence (the 2-byte-prefix chunk shape, the POC
  finding's trigger), mid-frame EOF fail-loud, garbage-ATYP fail-
  loud, and the wire-error mapping table.
- **The e2e split-datagram test uses a 60 KiB payload** (60007
  inner bytes, legal under the u16 joint bound, well over the 16
  KB read buffer): first → big → third, in order, with the big
  datagram reassembling across the call boundary and the third
  drained through the pending queue; both halves of the
  persistence invariant pinned end to end.
- **No harness changes**: the consumer-side topology's
  `wire_consumer_side_with` + the producer-associate task's
  `echo_egress`/`refusing_egress` closures cover every test.
- The lib.rs re-exports gained `AssociateSession` +
  `TakenAssociateInner`; the forked `target_addr.rs` gained the
  `&TargetAddr` `ToTargetAddr` impl (ergonomics; owned code,
  ADR-013).

## Summary

Implemented the UDP ASSOCIATE consumer session in `src/consumer.rs`
(POC #2's shape, adapted to the crate's teardown-owning session
posture): `AssociateSession::associate` (open → no-auth → ASSOCIATE
→ reply, capturing the sentinel BND.ADDR as `relay_addr()`,
received but never used for addressing; the ASSOCIATE DST.ADDR is the
vanilla-mirror unspecified address), `send_datagram` (frames once
via the crate's wire `frame()` (the framing-once mirror, the joint
bound enforced at frame time), `recv_datagram` (the incremental
`DatagramReader` + batched-datagram pending queue, **state that
persists across calls**; that is the POC's reader-state finding, with the
fail-loud shapes: mid-frame EOF and undecodable frames), `shutdown`
(the half-close; the mux's implicit-EOF sentinel reaches the
producer's stage as EOF; association lifetime = channel lifetime),
plus `split`/`into_inner` accessors and the CONNECT sessions'
teardown ownership (Drop sync-reap + structural EOF, `close()`
graceful, teardown-at-birth on in-stream refusals, no `Clone`,
since `compile_fail` is asserted there). Typed error surface throughout: open
failures are `SocksSessionError::Open` (ADR-049 reasons); the
missing-egress 0x07 and factory-failure codes surface from
`associate` as `Rfc`. E2E suite (`tests/consumer-associate.rs`, 8
tests over the harness): the round-trip (sentinel relay address →
multiple in-order datagrams, one association), the split-datagram
persistence test (60 KiB datagram across the call boundary +
pending-queue drain), empty-payload round-trip (F-2, never EOF),
shutdown → fresh associate succeeds immediately (the POC's test 4),
Drop teardown + connection health, the missing-egress refusal
(typed 0x07, no phantom channel), the egress-factory failure
mapping, and the CONNECT+ASSOCIATE regression guard on one
registration (interleaved, both serving). Verification: `cargo
test` 305 (incl. the 3 doctests; the compile-fail assertions ride
among them), clippy `-D warnings` (all targets), fmt, `cargo doc
--no-deps`, and both wasm checks clean.