consumer: the BIND reply#2 re-entry guard + session hygiene (review-002 C-1, D-1, N-6, S-2)

- C-1: BindSession::await_inbound is take-guarded — the first read
  takes the session's stream; the returned BindDataPlane owns it (the
  TakenInner pairing posture; the session stays the teardown owner),
  so a second call returns the family's spent shape Ok(None) and can
  never re-read the live data plane. A failed read restores the inner
  (the phase is only consumed on success; the producer ends the
  session on every post-reply#1 failure exit, so a re-call EOFs —
  never a live data-plane re-read either way). Pinned by the
  double-await_inbound integration test (the spent shape + the
  data-plane bytes asserted untouched) and the spent-BIND unit test.
- D-1: one normative reply#2 reader (option a) — the session's
  await_inbound; BindReply1::await_inbound is cut and reply1() is
  documented as a view (bind_addr + raw_mut), matching its actual
  usage; the reply1()-twice re-entry path is closed (pinned by the
  view test).
- N-6: the three session constructors' prologues collapsed into one
  private open_and_negotiate helper (teardown-at-birth in one place);
  zero behavior change.
- S-2: the EOF-propagation doc line on all three reap sites (the
  dropped teardown_channel handle is deliberate — the graceful shape;
  clear_all aborts instead).
- BindDataPlane is un-lifetimed (owns the stream); into_inner returns
  the owned Socks5Stream<BiStream>; docs synced (module + session
  docs, consumer.md §The BIND client).

Verification: cargo test 352 green (all-features too), clippy
-D warnings clean, fmt clean, cargo doc clean, both wasm checks clean.

src/consumer.rs

@@ -15,6 +15,12 @@
 //! then the pass-through data plane. [`AssociateSession`] is the UDP
 //! ASSOCIATE session (POC #2's shape — see its docs).
 //!
+//! The BIND session's reply#2 phase is take-guarded (review 002
+//! C-1): the first `await_inbound` takes the stream — the plane owns
+//! it, the session stays the teardown owner — and a second call
+//! returns the family's spent shape (`Ok(None)`); the reply#2 reader
+//! is that one method alone (`reply1()` is a view, D-1).
+//!
 //! Target construction rides [`ToTargetAddr`]: IP-string forms parse
 //! to IP targets; the domain form passes through unresolved (ADR-012
 //! — resolution is the egress's job, never the protocol's).
@@ -67,6 +73,40 @@ use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr};
 use crate::rfc1928::{ReplyError, Socks5Command, SocksError, UdpHeaderError};
 use crate::wire::{frame, DatagramReader, WireError};
 
+/// The three session constructors' shared prologue (review 002 N-6):
+/// open the channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)`
+/// — the trivial params, ADR-001), join the halves into the channel's
+/// `BiStream`, run the command negotiation over the forked client
+/// typestate, and on any negotiation failure reap the adopted channel
+/// before the error returns (teardown-at-birth — the phantom-channel
+/// property, in one place instead of three). `Ok` yields the channel
+/// id, the negotiated stream, and the cloned manager for the session
+/// to own.
+async fn open_and_negotiate<T, F, Fut>(
+    client: &ChannelClient,
+    negotiate: F,
+) -> Result<(u32, T, ChannelManager), SocksSessionError>
+where
+    F: FnOnce(Socks5Stream<BiStream>) -> Fut,
+    Fut: std::future::Future<Output = Result<T, SocksSessionError>>,
+{
+    let (channel_id, send, recv) = client
+        .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
+        .await
+        .map_err(SocksOpenError::from)?;
+
+    let bidi = BiStream::from_joined(recv, send);
+    let negotiated = negotiate(Socks5Stream::use_stream(bidi, None, Config::default()).await?);
+
+    match negotiated.await {
+        Ok(inner) => Ok((channel_id, inner, client.manager().clone())),
+        Err(err) => {
+            let _ = client.manager().teardown_channel(channel_id);
+            Err(err)
+        }
+    }
+}
+
 /// The spent-session marker (review 002 R-2): [`Socks5ChannelSession
 /// ::into_inner`] / [`AssociateSession::into_inner`] return it when
 /// the pipe/stream was already taken — reachable only by calling
@@ -123,29 +163,17 @@ impl Socks5ChannelSession {
         target: impl ToTargetAddr,
     ) -> Result<Self, SocksSessionError> {
         let target: TargetAddr = target.to_target_addr()?;
-        let (channel_id, send, recv) = client
-            .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
-            .await
-            .map_err(SocksOpenError::from)?;
-
-        let bidi = BiStream::from_joined(recv, send);
-        let negotiated = async {
-            let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?;
+        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
             inner.request(Socks5Command::TCPConnect, target).await?;
-            Result::<Socks5Stream<BiStream>, SocksSessionError>::Ok(inner)
-        };
+            Ok(inner)
+        })
+        .await?;
 
-        match negotiated.await {
-            Ok(inner) => Ok(Self {
-                channel_id,
-                manager: Some(client.manager().clone()),
-                inner: Some(inner),
-            }),
-            Err(err) => {
-                let _ = client.manager().teardown_channel(channel_id);
-                Err(err)
-            }
-        }
+        Ok(Self {
+            channel_id,
+            manager: Some(manager),
+            inner: Some(inner),
+        })
     }
 
     /// The adopted channel's id (observability + correlation; one
@@ -194,6 +222,13 @@ impl Socks5ChannelSession {
         self.reap()
     }
 
+    /// The adopted channel's teardown (`teardown_channel` returns a
+    /// producer-side pump-task handle the caller may abort after the
+    /// drain — the channels-call path's shape). The handle is
+    /// deliberately dropped here: the producer-side pump finishes via
+    /// EOF propagation (the graceful shape — aborting would cut
+    /// mid-flight data); `clear_all` aborts instead (the substrate's
+    /// abrupt path).
     fn reap(&mut self) -> bool {
         self.manager
             .take()
@@ -288,39 +323,26 @@ impl AssociateSession {
     /// adopted channel is torn down before the error returns (no
     /// phantom channel).
     pub async fn associate(client: &ChannelClient) -> Result<Self, SocksSessionError> {
-        let (channel_id, send, recv) = client
-            .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
-            .await
-            .map_err(SocksOpenError::from)?;
-
-        let bidi = BiStream::from_joined(recv, send);
-        let negotiated = async {
-            let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?;
+        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
             let relay_addr = inner
                 .request(
                     Socks5Command::UDPAssociate,
                     TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)),
                 )
                 .await?;
-            Result::<(TargetAddr, Socks5Stream<BiStream>), SocksSessionError>::Ok((
-                relay_addr, inner,
-            ))
-        };
+            Ok((relay_addr, inner))
+        })
+        .await?;
+        let (relay_addr, inner) = inner;
 
-        match negotiated.await {
-            Ok((relay_addr, inner)) => Ok(Self {
-                channel_id,
-                manager: Some(client.manager().clone()),
-                inner: Some(inner),
-                reader: DatagramReader::new(),
-                pending: std::collections::VecDeque::new(),
-                relay_addr: Some(relay_addr),
-            }),
-            Err(err) => {
-                let _ = client.manager().teardown_channel(channel_id);
-                Err(err)
-            }
-        }
+        Ok(Self {
+            channel_id,
+            manager: Some(manager),
+            inner: Some(inner),
+            reader: DatagramReader::new(),
+            pending: std::collections::VecDeque::new(),
+            relay_addr: Some(relay_addr),
+        })
     }
 
     /// The adopted channel's id (observability + correlation; one
@@ -487,6 +509,13 @@ impl AssociateSession {
         self.reap()
     }
 
+    /// The adopted channel's teardown (`teardown_channel` returns a
+    /// producer-side pump-task handle the caller may abort after the
+    /// drain — the channels-call path's shape). The handle is
+    /// deliberately dropped here: the producer-side pump finishes via
+    /// EOF propagation (the graceful shape — aborting would cut
+    /// mid-flight data); `clear_all` aborts instead (the substrate's
+    /// abrupt path).
     fn reap(&mut self) -> bool {
         self.manager
             .take()
@@ -540,32 +569,25 @@ pub type SplitPipe<'s> = (
 /// real address on the channels path: the local variant's bound
 /// listener; the composed variant's expected far-side address; the
 /// wrapper-aware sentinel `0.0.0.0:0` on the in-band default). The
-/// next phase is [`Self::await_inbound`].
+/// next phase is the session's [`BindSession::await_inbound`].
 ///
-/// The session owns the adopted channel — teardown stays here
-/// (reap via the paired `BindSession`; this handle borrows into it).
-/// No `Clone` (the compile-assert rides [`BindSession`]).
+/// The handle is a **view**, not a reader (review 002 D-1's
+/// one-reader decision): it carries the advertised address and the
+/// raw pre-reply#2 stream ([`Self::raw_mut`], the front-door bridge
+/// seam); the one normative reply#2 reader is
+/// [`BindSession::await_inbound`] alone, take-guarded there — a
+/// fresh handle per call cannot double-read anything.
+///
+/// The session owns the adopted channel — teardown stays there
+/// (reap via the session; this handle borrows into it). No `Clone`
+/// (the compile-assert rides [`BindSession`]).
 pub struct BindReply1<'s> {
     inner: &'s mut Socks5Stream<BiStream>,
     /// Reply#1's BND.ADDR — the advertised address.
     pub bind_addr: TargetAddr,
 }
 
-impl<'s> BindReply1<'s> {
-    /// Await reply#2 (the inbound accept) and take the pass-through
-    /// data plane. Reply#2 is hand-read with the forked `read_address`
-    /// codec — reply#1's codec applied again, no second codec (POC #5's
-    /// `read_reply2` sketch; ADR-011's rationale). A non-success
-    /// reply#2 errors with the mapped RFC reply code (accept-stage
-    /// failures surface here — no data plane).
-    pub async fn await_inbound(self) -> Result<BindDataPlane<'s>, SocksSessionError> {
-        let peer = read_reply2(self.inner).await?;
-        Ok(BindDataPlane {
-            inner: self.inner,
-            peer,
-        })
-    }
-
+impl BindReply1<'_> {
     /// The raw stream (the front-door bridge seam: a raw pipe to a
     /// vanilla client, pre-reply#2).
     pub fn raw_mut(&mut self) -> &mut Socks5Stream<BiStream> {
@@ -573,27 +595,29 @@ impl<'s> BindReply1<'s> {
     }
 }
 
-/// The post-reply#2 data plane (the `BindDataPlane`-shaped pass
-/// through): the proxied connection to the accepted inbound peer, 0 B
-/// framing — reads/writes ride the channel directly. Accessors:
-/// [`Self::into_inner`] (owned) / [`Self::get_socket_mut`] (borrowed;
-/// the POC shape). The teardown owner is the [`BindSession`] the
-/// `into_inner`/`get_socket_mut` borrows came from — drop it (or
-/// `close` it) when done; dropping the data plane EOFs the pipe
-/// structurally.
-pub struct BindDataPlane<'s> {
-    inner: &'s mut Socks5Stream<BiStream>,
+/// The post-reply#2 data plane (the pass-through to the accepted
+/// inbound peer): 0 B framing — reads/writes ride the channel
+/// directly. The plane **owns** the stream (review 002 C-1's
+/// take-shape — the taken inner pairs with the session that stays
+/// the teardown owner, the `TakenInner` pairing posture); the
+/// [`BindSession`] keeps the channel's teardown — drop it (or
+/// `close` it) when done; dropping the plane EOFs the pipe
+/// structurally (the mux's implicit-EOF sentinel). Accessors:
+/// [`Self::into_inner`] (owned) / [`Self::get_socket_mut`]
+/// (borrowed; the POC shape).
+pub struct BindDataPlane {
+    inner: Socks5Stream<BiStream>,
     /// Reply#2's peer address (the RFC's "the address of the incoming
     /// connection" — the attribution surface, in-band).
     pub peer: TargetAddr,
 }
 
-impl<'s> BindDataPlane<'s> {
+impl BindDataPlane {
     /// Take the underlying [`Socks5Stream<BiStream>`] out of the data
     /// plane (the owned pipe; the forked client type). The session
     /// rides along still owning the teardown (`BindSession`'s
     /// `close`/`Drop`).
-    pub fn into_inner(self) -> &'s mut Socks5Stream<BiStream> {
+    pub fn into_inner(self) -> Socks5Stream<BiStream> {
         self.inner
     }
 
@@ -604,41 +628,37 @@ impl<'s> BindDataPlane<'s> {
     }
 }
 
-impl tokio::io::AsyncRead for BindDataPlane<'_> {
+impl tokio::io::AsyncRead for BindDataPlane {
     fn poll_read(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
         buf: &mut tokio::io::ReadBuf<'_>,
     ) -> std::task::Poll<std::io::Result<()>> {
-        let inner = &mut *self.inner;
-        std::pin::Pin::new(inner).poll_read(cx, buf)
+        std::pin::Pin::new(&mut self.inner).poll_read(cx, buf)
     }
 }
 
-impl tokio::io::AsyncWrite for BindDataPlane<'_> {
+impl tokio::io::AsyncWrite for BindDataPlane {
     fn poll_write(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
         buf: &[u8],
     ) -> std::task::Poll<std::io::Result<usize>> {
-        let inner = &mut *self.inner;
-        std::pin::Pin::new(inner).poll_write(cx, buf)
+        std::pin::Pin::new(&mut self.inner).poll_write(cx, buf)
     }
 
     fn poll_flush(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<std::io::Result<()>> {
-        let inner = &mut *self.inner;
-        std::pin::Pin::new(inner).poll_flush(cx)
+        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
     }
 
     fn poll_shutdown(
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<std::io::Result<()>> {
-        let inner = &mut *self.inner;
-        std::pin::Pin::new(inner).poll_shutdown(cx)
+        std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
     }
 }
 
@@ -662,7 +682,18 @@ impl tokio::io::AsyncWrite for BindDataPlane<'_> {
 /// property); accept-stage failures surface at `await_inbound` as
 /// reply#2's mapped code ([`SocksSessionError::Rfc`]) — the channel
 /// is still open there (the listener existed; the session lives until
-/// the app closes it).
+/// the app closes it; the inner is restored, see the take-guard
+/// below).
+///
+/// The reply#2 phase is take-guarded (review 002
+/// C-1): the first `await_inbound` takes the inner stream — the
+/// returned plane owns it (the `TakenInner` pairing posture; the
+/// session stays the teardown owner) — so a second call hits the
+/// `None` arm and returns `Ok(None)`: the session family's spent
+/// shape, with no re-read of the live data plane possible. A failed
+/// read restores the inner (the phase is only consumed on success;
+/// the producer ends the session on every post-reply#1 failure exit,
+/// so a re-call EOFs — never a live data-plane re-read either way).
 pub struct BindSession {
     channel_id: u32,
     manager: Option<ChannelManager>,
@@ -704,32 +735,19 @@ impl BindSession {
         target: impl ToTargetAddr,
     ) -> Result<Self, SocksSessionError> {
         let target: TargetAddr = target.to_target_addr()?;
-        let (channel_id, send, recv) = client
-            .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
-            .await
-            .map_err(SocksOpenError::from)?;
-
-        let bidi = BiStream::from_joined(recv, send);
-        let negotiated = async {
-            let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?;
+        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
             let bind_addr = inner.request(Socks5Command::TCPBind, target).await?;
-            Result::<(TargetAddr, Socks5Stream<BiStream>), SocksSessionError>::Ok((
-                bind_addr, inner,
-            ))
-        };
+            Ok((bind_addr, inner))
+        })
+        .await?;
+        let (bind_addr, inner) = inner;
 
-        match negotiated.await {
-            Ok((bind_addr, inner)) => Ok(Self {
-                channel_id,
-                manager: Some(client.manager().clone()),
-                inner: Some(inner),
-                bind_addr: Some(bind_addr),
-            }),
-            Err(err) => {
-                let _ = client.manager().teardown_channel(channel_id);
-                Err(err)
-            }
-        }
+        Ok(Self {
+            channel_id,
+            manager: Some(manager),
+            inner: Some(inner),
+            bind_addr: Some(bind_addr),
+        })
     }
 
     /// The adopted channel's id (observability + correlation; one
@@ -739,9 +757,12 @@ impl BindSession {
     }
 
     /// Reply#1's view: the advertised BND.ADDR (the app server dials
-    /// it out-of-band — the RFC's original role for the address). The
-    /// reply#2 reader is [`BindReply1::await_inbound`]; the returned
-    /// handle borrows the session (teardown stays owned).
+    /// it out-of-band — the RFC's original role for the address) and
+    /// the raw pre-reply#2 stream ([`BindReply1::raw_mut`], the
+    /// front-door bridge seam). A **view**, not a reader (review 002
+    /// D-1's one-reader decision): the one normative reply#2 reader
+    /// is [`Self::await_inbound`]; the handle can be re-borrowed
+    /// freely — it carries no read phase.
     pub fn reply1(&mut self) -> Option<BindReply1<'_>> {
         let inner = self.inner.as_mut()?;
         Some(BindReply1 {
@@ -751,23 +772,43 @@ impl BindSession {
     }
 
     /// Await the inbound accept (reply#2) and take the pass-through
-    /// data plane — [`BindReply1::await_inbound`] reached through the
-    /// session (the borrow rides the session; teardown stays owned).
-    /// `None` after a taken plane (the spent-session posture).
-    pub async fn await_inbound(&mut self) -> Result<Option<BindDataPlane<'_>>, SocksSessionError> {
-        let Some(inner) = self.inner.as_mut() else {
+    /// data plane. The one normative reply#2 reader (review 002 C-1 +
+    /// D-1): the phase is **take-guarded** — the read consumes the
+    /// session's inner stream (the plane owns it; the session stays
+    /// the teardown owner) — so a second call cannot re-read the
+    /// live data plane. Reply#2 is hand-read with the forked
+    /// `read_address` codec — reply#1's codec applied again, no
+    /// second codec (POC #5's `read_reply2` sketch; ADR-011's
+    /// rationale).
+    ///
+    /// `Ok(None)` after the plane was taken (the spent-session
+    /// posture — the doc claim this guard makes true). A failed read
+    /// restores the inner (the phase is not consumed on failure): the
+    /// channel stays owned (reply#1 succeeded — the listener existed;
+    /// reap via `close`/`Drop`), the pipe stays reachable
+    /// (`split`), and a re-call re-reads against the already-ended
+    /// session (the producer ends it on every post-reply#1 failure
+    /// exit — the read EOFs; never a live data-plane re-read).
+    pub async fn await_inbound(&mut self) -> Result<Option<BindDataPlane>, SocksSessionError> {
+        let Some(mut inner) = self.inner.take() else {
             return Ok(None);
         };
-        let peer = read_reply2(inner).await?;
-        Ok(Some(BindDataPlane { inner, peer }))
+        match read_reply2(&mut inner).await {
+            Ok(peer) => Ok(Some(BindDataPlane { inner, peer })),
+            Err(err) => {
+                self.inner = Some(inner);
+                Err(err)
+            }
+        }
     }
 
     /// The pass-through pipe as split read/write halves (0 B framing
-    /// — raw bytes to/from the accepted peer). Valid both in the
-    /// reply#1 phase (pre-reply#2 bytes ride the same channel — the
-    /// front-door bridge seam) and post-[`Self::await_inbound`] (the
-    /// data plane proper). `None` after a graceful `close` (the pipe
-    /// is dropped with the reap).
+    /// — raw bytes to/from the accepted peer). Valid in the reply#1
+    /// phase (pre-reply#2 bytes ride the same channel — the
+    /// front-door bridge seam); post-[`Self::await_inbound`] the
+    /// plane owns the stream (the take-guard), so the data plane's
+    /// halves ride `BindDataPlane`/its `into_inner` instead. `None`
+    /// after a graceful `close` (the pipe is dropped with the reap).
     pub fn split(&mut self) -> Option<SplitPipe<'_>> {
         let inner = self.inner.as_mut()?;
         Some(tokio::io::split(inner))
@@ -784,6 +825,13 @@ impl BindSession {
         self.reap()
     }
 
+    /// The adopted channel's teardown (`teardown_channel` returns a
+    /// producer-side pump-task handle the caller may abort after the
+    /// drain — the channels-call path's shape). The handle is
+    /// deliberately dropped here: the producer-side pump finishes via
+    /// EOF propagation (the graceful shape — aborting would cut
+    /// mid-flight data); `clear_all` aborts instead (the substrate's
+    /// abrupt path).
     fn reap(&mut self) -> bool {
         self.manager
             .take()
@@ -865,6 +913,33 @@ mod tests {
         assert_send::<TakenInner>();
     }
 
+    #[test]
+    fn bind_session_shape_is_send_and_not_clone() {
+        assert_send::<BindSession>();
+        assert_send::<BindDataPlane>();
+    }
+
+    #[tokio::test]
+    async fn spent_bind_await_inbound_returns_the_spent_shape() {
+        // The C-1 take-guard's spent arm: a session whose inner is
+        // gone (taken at the first read / the teardown) returns the
+        // family's `Ok(None)` — no re-read, no panic.
+        let mut session = BindSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+            bind_addr: None,
+        };
+        let spent = session.await_inbound().await;
+        assert!(
+            matches!(spent, Ok(None)),
+            "the spent shape: {:?}",
+            spent.as_ref().err().map(|e| e.to_string())
+        );
+        assert!(session.reply1().is_none(), "no reply#1 view either");
+        assert!(session.split().is_none(), "no pipe after teardown");
+    }
+
     #[test]
     fn debug_impl_names_the_channel_and_liveness() {
         let session = Socks5ChannelSession {