From 53739b20cbde2d0d6672124e028f8f4402346cfb Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Thu, 24 Sep 2026 11:58:55 +0000 Subject: [PATCH] consumer: the BIND reply#2 re-entry guard + session hygiene (review-002 C-1, D-1, N-6, S-2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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; 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. --- docs/architecture/consumer.md | 20 +- src/consumer.rs | 331 +++++++++++------- .../implementation/fix-bind-reply2-reentry.md | 92 ++++- tests/consumer-bind.rs | 155 +++++++- 4 files changed, 459 insertions(+), 139 deletions(-) diff --git a/docs/architecture/consumer.md b/docs/architecture/consumer.md index 9f8c6fe..f1d8372 100644 --- a/docs/architecture/consumer.md +++ b/docs/architecture/consumer.md @@ -1,8 +1,8 @@ --- status: draft -last_updated: 2026-09-24 (the landed session signatures reconciled — -`&ChannelClient` constructors, `impl ToTargetAddr` datagrams, the -`Option`-shaped `recv_datagram`; Phase 1 initial draft 2026-09-20) +last_updated: 2026-09-24 (the BIND spent posture reconciled — the +take-guarded reply#2 reader, review 002 C-1/D-1; the landed session +signatures reconciled 2026-09-24; Phase 1 initial draft 2026-09-20) --- # alksocks — The Consumer Half @@ -69,8 +69,18 @@ The BIND reply#2 reader (POC #5's `read_reply2` sketch): the forked client `request()` reads reply#1; reply#2 is hand-read with the forked public `read_address` codec (reply#1's codec applied again; no second codec). The session then rides the pass-through data -plane. The BIND client is a wrapper-aware-consumer convenience — -vanilla clients reach BIND through the front door instead. +plane. The reply#2 phase is take-guarded: the one normative reader +is `await_inbound` — the first read consumes the session's stream +(the returned plane owns it; the session stays the teardown owner), +so a second call returns the family's spent shape (`Ok(None)`, the +same posture the sibling sessions pin) and cannot re-read the live +data plane; a failed read restores the inner (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). The +`reply1()` handle is a view (the advertised BND.ADDR + the raw +pre-reply#2 stream), not a reader. The BIND client is a +wrapper-aware-consumer convenience — vanilla clients reach BIND +through the front door instead. ### Error surface diff --git a/src/consumer.rs b/src/consumer.rs index 5126c91..841319a 100644 --- a/src/consumer.rs +++ b/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( + client: &ChannelClient, + negotiate: F, +) -> Result<(u32, T, ChannelManager), SocksSessionError> +where + F: FnOnce(Socks5Stream) -> Fut, + Fut: std::future::Future>, +{ + 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 { 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::, 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 { - 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), 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, /// 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, 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 { @@ -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, +/// 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, /// 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`] 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 { + pub fn into_inner(self) -> Socks5Stream { 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> { - 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> { - 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> { - 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> { - 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, @@ -704,32 +735,19 @@ impl BindSession { target: impl ToTargetAddr, ) -> Result { 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), 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> { 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>, 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, 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> { 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::(); } + #[test] + fn bind_session_shape_is_send_and_not_clone() { + assert_send::(); + assert_send::(); + } + + #[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 { diff --git a/tasks/implementation/fix-bind-reply2-reentry.md b/tasks/implementation/fix-bind-reply2-reentry.md index 3c1aa3b..5d66b33 100644 --- a/tasks/implementation/fix-bind-reply2-reentry.md +++ b/tasks/implementation/fix-bind-reply2-reentry.md @@ -1,7 +1,7 @@ --- id: implementation/fix-bind-reply2-reentry name: "Fix — BIND reply#2 re-entry guard + consumer session hygiene (review-002 C-1, D-1, N-6, S-2)" -status: pending +status: completed depends_on: [] scope: narrow risk: medium @@ -104,8 +104,94 @@ accept-failure mappings unaffected). ## Notes -> Agent fills this during implementation. +> Filled by the implementation agent (2026-09-24): + +- **C-1 landed as the take-shape** (the review's first sketch): `BindSession + ::await_inbound` takes the session's inner stream before reading reply#2 + (`inner.take()`); the returned `BindDataPlane` **owns** the + `Socks5Stream` while the session stays the teardown owner — the + `TakenInner` pairing posture `Socks5ChannelSession::into_inner` pinned, now + carried by `BindDataPlane` itself (no new pairing type — the plane is the + pair's inner half; the session is the pair's teardown owner). A second + call hits the `None` arm → the family's spent shape `Ok(None)` (the + method's doc claim "`None` after a taken plane" is now true). The spent + shape chosen is **`Ok(None)`** (not a typed `SpentSessionError`): the + method's existing return type `Result>` already carried the + documented spent arm, and the sibling read-shaped surfaces + (`recv_datagram`'s spent stream) return the EOF-shape `Ok(None)` — the + session family's read-side posture; `SpentSessionError` stays the + taking-methods' (`into_inner`) shape. +- **A failed reply#2 read restores the inner** (the phase is *not* + consumed on failure): the taken inner is returned to the session on + error — the channel stays owned (reply#1 succeeded, the listener + existed, reap via `close`/`Drop`, the error-surface contract the + existing tests pin), and the session's `split()`/`close()` keep the + pipe. The corruption path is the *post-success* re-entry, and that + is exactly what the take-guard closes: after success the plane is + taken, the spent shape holds. A post-failure re-call re-reads + reply#2 against the (already-ended) session — the producer ends the + session on every post-reply#1 failure exit, so the read EOFs the + same way (the quiet-wait garbage test pins this shape, green + unchanged). +- **D-1 decision: option (a)** — the session's `await_inbound` is the one + normative reply#2 reader. `BindReply1::await_inbound` is **cut** (not + private: the take-shape makes the handle's borrowed-plane reader + impossible — the plane now owns the stream); `reply1()` is documented as + a **view** (`bind_addr` + `raw_mut`, the front-door bridge seam), which + is exactly how the four existing call sites use it. `BindReply1` stays + as the typed view (its `Send` probe pinned). +- **`BindDataPlane` is un-lifetimed** (`BindDataPlane`, not + `BindDataPlane<'s>`): it owns the stream, so the `'static` lifetime + probe became a plain `assert_send::()`; `into_inner` + returns the owned `Socks5Stream` (the test call site gained a + `mut` + drop-before-reap, the CONNECT `TakenInner` tests' shape); the + `get_socket_mut` accessor is unchanged (borrowed socket access through + the owned inner). The `split()` doc notes the take-guard's consequence + (post-reply#2 halves ride the plane, not the session). +- **N-6 landed** as the private `open_and_negotiate(client, negotiate)` + helper: open → `BiStream::from_joined` → the forked typestate's + `use_stream` → the negotiation closure → teardown-on-err, one place for + the teardown-at-birth invariant. The three constructors are one-liners + over it (CONNECT returns the inner; ASSOCIATE/BIND return + `(addr, inner)` tuples). Zero behavior change — the full consumer + + integration suites stayed green unchanged (the only touched call sites + were the two test-side ownership consequences above). +- **S-2 landed**: the same doc line on all three `reap` sites (CONNECT, + ASSOCIATE, BIND) — the dropped `teardown_channel` `JoinHandle` is + deliberate (the producer-side pump finishes via EOF propagation, the + graceful shape; `clear_all` aborts instead, the substrate's abrupt path). +- **Test round**: the pinning integration test + (`double_await_inbound_returns_the_spent_shape_and_bytes_survive` — the + second call returns `Ok(None)` and the first plane's exchange completed + uncorrupted, a fresh bind round-trips), the `reply1()`-twice view test + (`reply1_twice_is_a_view_and_the_reader_yields_one_plane` — repeated + views carry the same address, the reader yields exactly one plane, the + second reader call is the spent shape), the spent-BIND unit test + (`spent_bind_await_inbound_returns_the_spent_shape` — the `None`-inner + arm + `reply1`/`split` spent probes), and the `Send` probe updated + (`BindSession` + owned `BindDataPlane`). The existing BIND suite stayed + green unchanged apart from the one `into_inner` ownership call site + (the round-trip, gate-refusal, factory/accept-failure mappings all + untouched). ## Summary -> Agent fills this on completion. \ No newline at end of file +Closed review 002's C-1 (the one reachable silent-corruption path): the +BIND reply#2 phase is take-guarded — `BindSession::await_inbound` takes +the session's stream, the returned `BindDataPlane` owns it (the +`TakenInner` pairing posture; the session stays the teardown owner), and +a second call returns the family's spent shape `Ok(None)` with no re-read +of the live data plane possible. D-1 resolved as option (a): the +session's `await_inbound` is the one normative reply#2 reader; +`BindReply1::await_inbound` is cut and `reply1()` is documented as a view +(`bind_addr` + `raw_mut`), matching its actual usage. N-6 collapsed the +three session constructors' prologues into one private +`open_and_negotiate` helper (teardown-at-birth in one place). S-2 added +the EOF-propagation doc line to all three `reap` sites. Docs synced +(module docs + the session docs + consumer.md §The BIND client; the +frontmatter `last_updated` records the reconciliation). Tests: the +double-read pinning test (spent shape + bytes untouched), the +`reply1()`-twice view test, the spent-BIND unit test, and the existing +suite green (352 total). Verification: `cargo test` + `--all-features` +green, clippy `-D warnings` clean, fmt clean, `cargo doc` clean, both +wasm checks clean. \ No newline at end of file diff --git a/tests/consumer-bind.rs b/tests/consumer-bind.rs index df7fcd8..d69dc5e 100644 --- a/tests/consumer-bind.rs +++ b/tests/consumer-bind.rs @@ -173,7 +173,7 @@ async fn bind_data_plane_accessors_into_inner_and_socket_mut() { .write_all(b"raw") .await .expect("write"); - let inner = plane.into_inner(); + let mut inner = plane.into_inner(); let mut buf = [0u8; 6]; tokio::time::timeout( std::time::Duration::from_secs(10), @@ -183,7 +183,7 @@ async fn bind_data_plane_accessors_into_inner_and_socket_mut() { .expect("echo timed out") .expect("read the echo"); assert_eq!(&buf, b"preraw", "both write paths rode the same pipe"); - let _unused = inner; + drop(inner); assert!(session.close().await, "the entry existed"); no_leaked_channels(&topo).await; @@ -484,6 +484,155 @@ async fn bind_drop_teardown_reaps_and_connection_stays_healthy() { fn bind_session_type_is_send() { fn assert_send() {} assert_send::(); + assert_send::(); assert_send::>(); - assert_send::>(); +} + +/// The C-1 re-entry guard (review 002): a second `await_inbound` on +/// the same session cannot re-read the data plane — it returns the +/// spent shape (`Ok(None)`), and the data-plane bytes survive +/// untouched (the first plane's echo round-trip is intact after the +/// second call; the pinning test for the silent-corruption path). +#[tokio::test] +async fn double_await_inbound_returns_the_spent_shape_and_bytes_survive() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("reentry.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + + // The first read: the plane exists (and consumes the phase — + // the plane owns the stream; the session stays the teardown + // owner). + let mut plane = session + .await_inbound() + .await + .expect("reply#2 read") + .expect("the plane exists before reply#2"); + plane.write_all(b"ping").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"ping", "the first plane's round-trip"); + drop(plane); + + // The second call: the spent shape — `Ok(None)`, never a + // re-read of the live data plane. + let second = session + .await_inbound() + .await + .expect("the spent shape is an Ok, not an error"); + assert!(second.is_none(), "the phase was consumed"); + + // The data-plane bytes survive: the first plane's echo + // round-trip was complete and no re-read corrupted the pipe — + // the channel is closed with the reap (the session owner's + // `close`), and a fresh bind on the same registration + // round-trips with its own bytes intact. + assert!(session.close().await, "the entry existed"); + no_leaked_channels(&topo).await; + + let mut fresh = BindSession::bind( + topo.consumer.as_ref(), + ("after.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("fresh bind after the spent call"); + let mut plane = fresh + .await_inbound() + .await + .expect("reply#2") + .expect("the plane"); + plane.write_all(b"again").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 5]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the fresh echo"); + assert_eq!(&buf, b"again", "the fresh session's bytes are intact"); + drop(plane); + fresh.close().await; + no_leaked_channels(&topo).await; +} + +/// The `reply1()`-twice re-entry path (C-1's other door, closed by +/// the D-1 one-reader decision): `reply1()` is a view — fresh handles +/// per call, no read phase — so repeated `reply1()` calls carry the +/// same advertised address without consuming anything, and the one +/// normative reader (`await_inbound`) still yields exactly one plane. +#[tokio::test] +async fn reply1_twice_is_a_view_and_the_reader_yields_one_plane() { + let topo = wire_consumer_side_with( + harness::echo_dial(), + None, + Some(harness::echo_accept()), + Some(bind_scoped()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("view.test", ECHO_BIND_PORT).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + + let expected = TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))); + + // Repeated views: every handle carries the same advertised + // address; nothing is consumed. + let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle"); + assert_eq!(bind_addr, expected); + let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle again"); + assert_eq!(bind_addr, expected, "a view, not a reader"); + + // The one normative reader: exactly one plane. + let mut plane = session + .await_inbound() + .await + .expect("reply#2 read") + .expect("the plane"); + plane.write_all(b"once").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + plane.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"once"); + drop(plane); + + // The re-entry is closed on this door too. + assert!( + session + .await_inbound() + .await + .expect("spent shape") + .is_none(), + "no second plane after the reader ran" + ); + session.close().await; + no_leaked_channels(&topo).await; }