diff --git a/docs/architecture/client.md b/docs/architecture/client.md index 264e341..e21cca4 100644 --- a/docs/architecture/client.md +++ b/docs/architecture/client.md @@ -113,6 +113,31 @@ half-capability, which the family forbids (ADR-003 symmetry + primitives (bound ports, handles, cancellation, clean close) to make that possible. +> **The landed lifecycle surface (`w4-lifecycle` — recorded at +> landing):** the `-R` listens return **`ReverseListener` handles** — +> the `ReverseEndpoint` carried (the *actually bound* values: +> the wire address + the bound port / the socket path — the +> assembly's own record for a rebind policy it may compose), the +> explicit `cancel()` the wire-honest teardown (the confirmed cancel +> also releases the connection's listener count; a repeat rides the +> wire and hits the unknown-listener refusal). The **stateless** +> cancel forms (`reverse_listen_cancel` + the unix twin) ride beside +> the handles 1:1 (sshd's global-request cancels are stateless — a +> cancel by the raw wire strings works from anywhere, including +> after the handle itself is gone: the handle shares the +> connection's internals through the arc, so its `cancel()` is the +> same wire op). **No `Drop` hides a wire op** on any of these +> handles: a cancellation/bundle of the wire ops is async and cannot +> run in a sync `Drop`, and at this russh pin dropping the last +> wire-stack handle ends the client runner — the transport *breaks* +> (the peer observes an error-class session end, never the clean +> disconnect message). The primitives-not-policy posture is the +> module-docs contract: `client.rs` §Lifecycle posture (the +> primitives list + the abrupt-drop/clean-close boundary, +> row-pinned both ways) and `client_accept`'s recorded decisions; +> the alkminer reconnect-and-restore flow stays assembly +> composition (reverse-runner fact 4). + ## What the crate does *not* include (client direction) - **Job supervision** — background-job lifecycle (the alkminer diff --git a/src/client.rs b/src/client.rs index 6a309d6..9ca5622 100644 --- a/src/client.rs +++ b/src/client.rs @@ -43,14 +43,39 @@ //! everything else in the wave rides (the op rows open channels on //! it; the `-R` leg installs the acceptor registry over it) — the //! russh `client::Handle` wrapped with the crate's surface, the -//! policy, limits and endpoint labels carried. The full -//! cancellation/clean-close lifecycle surface is `w4-lifecycle`'s; -//! this task lands the raw material (the explicit [`SshConnection:: -//! close`] path; the handle primitives below it). +//! policy, limits and endpoint labels carried. The connection handle's +//! clean-close path is [`SshConnection::close`]; the listener handles' +//! surface is [`crate::client_accept`]'s. //! -//! No reconnect or restore policy is owned by the crate (OQ-12): -//! failures are failures ([`DialError`]); reconnection strategy is -//! assembly composition over these primitives. +//! ## Lifecycle posture (`w4-lifecycle` — the OQ-12 resolution in code) +//! +//! The handle surface is **primitives, not policy** (reverse-runner +//! fact 4): reconnect-and-restore (rebinding persisted tunnel ports — +//! the alkminer flow) is *assembly* composition over these +//! primitives; the crate provides bound ports, handles, cancellation, +//! clean close, and never owns a reconnect/restore policy. Two +//! recorded facts pin the explicit-vs-drop boundary: +//! +//! - **Drop is never the clean path.** At this russh pin dropping the +//! last wire-stack handle ends the runner and the transport goes +//! down **abruptly** — the peer observes a broken session (its +//! serving outcome is an error wire class), never the clean +//! disconnect message. So the surface never relies on drop timing +//! for clean state: only [`SshConnection::close`] sends +//! `Disconnect::ByApplication` — the peer observes a clean +//! disconnect, no half-open channels (row-pinned through the +//! server double, both directions of the boundary: abrupt-on-drop, +//! clean-on-close). +//! And a cancellation is async — it cannot hide inside a sync +//! `Drop` hook, so no handle hides a `Drop` side effect: the +//! listener handles' drop frees *nothing* on their wire state. +//! - **Every listener/handle is explicitly cancellable** — the `-R` +//! listens return the [`ReverseListener`](crate::client_accept:: +//! ReverseListener) handles (the actually-bound endpoint carried; +//! `cancel()` the wire-honest teardown), the connection handle +//! carries [`SshConnection::close`]; the stateless wire forms +//! ([`SshConnection::reverse_listen_cancel`] + the unix twin) +//! mirror sshd's global-request cancels 1:1. //! //! ## Test surface //! @@ -111,14 +136,16 @@ pub async fn dial(config: DialConfig) -> Result { } Ok(SshConnection { - handle, - limits: config.limits, - address: config.address, - username: config.username, - #[cfg(feature = "tunnel")] - acceptors, - #[cfg(feature = "tunnel")] - listener_budget: ChannelCap::default(), + shared: Arc::new(ConnectionShared { + handle, + limits: config.limits, + address: config.address, + username: config.username, + #[cfg(feature = "tunnel")] + acceptors, + #[cfg(feature = "tunnel")] + listener_budget: Arc::new(ChannelCap::default()), + }), }) } @@ -178,41 +205,68 @@ pub fn russh_client_config(limits: &Limits) -> russh::client::Config { /// surface. The remote end is authenticated (publickey-only, single /// attempt); the dial key is gone (its lifetime is the dial's — /// everything after rides the wire, never the key). +/// +/// The russh handle, budgets and registry ride one shared inner +/// (behind the `Arc`) — the sharing the `client_accept` module's +/// listener handles need for their independent `cancel()` calls; the +/// wire-visible teardowns stay explicit calls on top of that shared +/// stack (the module docs' §Lifecycle posture: drop is never a +/// wire-visible act). pub struct SshConnection { - handle: russh::client::Handle, - limits: Limits, - address: String, - username: String, + shared: Arc, +} + +/// The connection handle's internals — the one behind the +/// [`SshConnection`]'s `Arc`, every field private to the crate: the +/// wire surface is the wrapped methods, never the russh stack itself +/// (`serve.rs`'s wrapped-surface pattern mirrored client-side). +/// +/// The arc's meaning for the lifecycle boundary: the russh stack +/// (the runner) lives while *any* handle holds the arc — the +/// connection handle's drop is the abrupt teardown only when it is +/// the last holder (row-pinned), and a live `ReverseListener` (its +/// `cancel()` composes on this stack) keeps the wire serving until +/// its own explicit cancel. The clean teardown is +/// [`SshConnection::close`]'s, whatever holders exist. +pub(crate) struct ConnectionShared { + /// The russh handle the connection and the listener handles + /// compose into (its own request methods take `&self` at this + /// pin — the shared arc is the concurrent-sharing shape). + pub(crate) handle: russh::client::Handle, + pub(crate) limits: Limits, + pub(crate) address: String, + pub(crate) username: String, /// The `-R` leg's acceptor registry (the dial constructs it empty; /// the assembly installs/uninstalls across the live connection — /// see [`SshConnection::acceptors`] and /// [`crate::client_accept`]`'s recorded decisions. #[cfg(feature = "tunnel")] - acceptors: Arc, + pub(crate) acceptors: Arc, /// The reverse-listen count's dedicated bound — admitted at the /// listen rows, released at the confirmed cancels and the refused/ /// wire-error paths (the cap relation recorded in /// [`crate::client_accept`]; [`crate::channel_cap`]'s mutex makes - /// the counts sound under the caller's task). + /// the counts sound under the caller's task). Behind the arc so + /// the listener handles release it from their own tasks too. #[cfg(feature = "tunnel")] - listener_budget: ChannelCap, + pub(crate) listener_budget: Arc, } impl SshConnection { /// The bounded-resource budget the dial ran under (carried; the /// later op rows' admission surface). pub fn limits(&self) -> &Limits { - &self.limits + &self.shared.limits } /// The dialed address as the config carried it. pub fn address(&self) -> &str { - &self.address + &self.shared.address } /// The user the connection authenticated as. pub fn username(&self) -> &str { - &self.username + &self.shared.username } /// The `-R` leg's acceptor registry (the @@ -224,19 +278,22 @@ impl SshConnection { /// arrival time. #[cfg(feature = "tunnel")] pub fn acceptors(&self) -> &AcceptorRegistry { - &self.acceptors + &self.shared.acceptors } /// The reverse-listen count's shared handle — the op rows /// ([`crate::client_accept`]) read it; private to the crate. #[cfg(feature = "tunnel")] pub(crate) fn listener_budget(&self) -> &ChannelCap { - &self.listener_budget + &self.shared.listener_budget } - /// Whether the underlying russh session is closed. - pub fn is_closed(&self) -> bool { - self.handle.is_closed() + /// The shared inner — the listener handles' compose-into point + /// (their cancels run against the same russh stack); private to + /// the crate. + #[cfg(feature = "tunnel")] + pub(crate) fn shared(&self) -> &Arc { + &self.shared } /// The russh handle the op rows compose into (`w4-op-rows` — the @@ -246,7 +303,7 @@ impl SshConnection { pub(crate) fn russh_handle( &self, ) -> &russh::client::Handle { - &self.handle + &self.shared.handle } /// The dial's bounded-resource budget, verbatim — the op rows @@ -254,15 +311,25 @@ impl SshConnection { /// drives the command-carrying rows' client-mirror gate). #[cfg(any(feature = "session", feature = "git"))] pub(crate) fn budgets(&self) -> &Limits { - &self.limits + &self.shared.limits } - /// The explicit clean-close path (OQ-12's raw material; the full - /// cancellable-listener lifecycle surface is `w4-lifecycle`'s): - /// `Disconnect::ByApplication`, the honest close — the peer - /// observes a clean disconnect, no half-open channels. + /// Whether the underlying russh session is closed. + pub fn is_closed(&self) -> bool { + self.shared.handle.is_closed() + } + + /// The explicit clean-close path (the lifecycle primitive, + /// [`russh::Disconnect::ByApplication`] — the honest close): the + /// peer observes a clean disconnect, no half-open channels — + /// every open channel is torn down with the session (row-pinned + /// through the server double). **Drop is not this path** (the + /// module docs' §Lifecycle posture): a dropped handle releases + /// nothing on the wire — the connection outlives the handle + /// until the peer or the wall clock ends it. pub async fn close(self) { let _ = self + .shared .handle .disconnect(Disconnect::ByApplication, "closed by application", "en") .await; @@ -273,12 +340,12 @@ impl std::fmt::Debug for SshConnection { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let mut debug = f.debug_struct("SshConnection"); debug - .field("address", &self.address) - .field("username", &self.username) - .field("limits", &self.limits) - .field("closed", &self.handle.is_closed()); + .field("address", &self.shared.address) + .field("username", &self.shared.username) + .field("limits", &self.shared.limits) + .field("closed", &self.shared.handle.is_closed()); #[cfg(feature = "tunnel")] - debug.field("acceptors", &self.acceptors); + debug.field("acceptors", &self.shared.acceptors); debug.finish() } } diff --git a/src/client_accept.rs b/src/client_accept.rs index 78ca5c3..e26b8e1 100644 --- a/src/client_accept.rs +++ b/src/client_accept.rs @@ -6,9 +6,11 @@ //! / [`SshConnection::reverse_listen_cancel`] / the unix pair //! (`reverse_listen_unix` / `reverse_listen_unix_cancel`): the //! vanilla-server listen requests, the lifecycle primitives -//! (`client.md` §Lifecycle — the actually-bound port surfaced, -//! every listener explicitly cancellable; the *handles*' surface is -//! `w4-lifecycle`'s growth). +//! (`client.md` §Lifecycle). The listens return the +//! **[`ReverseListener`] handles** (`w4-lifecycle` — the handles' +//! surface: the actually-bound endpoint carried, the explicit +//! `cancel()`); the stateless wire cancel forms ride beside them +//! (sshd's global-request cancels mirrored 1:1). //! 2. **The acceptor registry** ([`AcceptorRegistry`]) — the //! assembly-mutable, per-channel-type registry the dial surface's //! russh handler routes **server-initiated `forwarded-*` opens** @@ -79,6 +81,26 @@ //! else: the pump is the acceptor's from the handoff on, the //! channel's transport bounds are russh's windows, and the slot read //! is the arrival admission). +//! - **The listener handles (`w4-lifecycle` — the handles' +//! surface)**: the listen rows return **[`ReverseListener`]** +//! handles — the endpoint (the actually-bound values, the cancel +//! match keys) carried, the explicit `cancel()` the wire-honest +//! teardown. Decisions recorded: +//! (a) **no `Drop` hides a wire op** — a cancellation is async and +//! cannot run in a sync `Drop`, so no `Drop` hook exists: the +//! listen **outlives a dropped handle until an explicit cancel or +//! the connection's end** (no drop-timing-only wire state — +//! row-pinned); +//! (b) the handle **shares the connection's inner stack** (the +//! `SshConnection`'s `Arc`-held internals — the russh request +//! methods take `&self` at this pin), so the handle's cancel is the +//! same wire op the stateless forms issue; +//! (c) the **stateless cancel forms ride beside** the handles +//! (sshd's global-request cancels are stateless — a cancel by the +//! raw strings works from anywhere); every form releases the +//! listener count exactly once on a confirmed cancel (the wire +//! state machine governs: the second cancel is the unknown-listener +//! refusal, no release). //! - **The runtime-Unix gate (OQ-11 symmetry)**: the streamlocal //! *listen* rows (`reverse_listen_unix` / `reverse_listen_unix_cancel`) //! compile everywhere and @@ -377,6 +399,132 @@ pub(crate) async fn handoff_forwarded_streamlocal( observe::serving_path_ended("forwarded-streamlocal", channel_number, None); } +// --------------------------------------------------------------------------- +// The listener handles (w4-lifecycle — the handles' surface) +// --------------------------------------------------------------------------- + +/// The endpoint a reverse listener is bound to — what a +/// [`ReverseListener`] carries for its assembly-side use (the +/// alkminer restore flow: the *actually bound* values to persist and +/// rebind) and the cancel forms' raw wire match key (sshd parity: a +/// cancel matches by the wire strings). +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReverseEndpoint { + /// A tcp forward: the wire address verbatim + the actually-bound + /// port (for a `port = 0` mint the peer-picked number, never `0`). + Tcp { address: String, port: u32 }, + /// A unix forward: the wire socket path verbatim. + Unix { path: String }, +} + +impl ReverseEndpoint { + /// The actually-bound port — the tcp shape's; `None` for the unix + /// shape (no port semantics — the cancel matches by the path). + pub fn bound_port(&self) -> Option { + match self { + Self::Tcp { port, .. } => Some(*port), + Self::Unix { .. } => None, + } + } +} + +/// A returned forward listener — the `-R` leg's **listener handle** +/// (the lifecycle primitive: every listener explicitly cancellable): +/// the bound endpoint carried (see [`ReverseEndpoint`]; the +/// assembly's own record for a restore/rebind policy — the crate +/// ships none, OQ-12), and the explicit `cancel()`. +/// +/// **Drop frees nothing on the wire** (the recorded decision in this +/// module's docs): there is no `Drop` hook — a dropped handle leaves +/// the peer's listen serving until an explicit `cancel()` or the +/// connection's end removes it. The handle does not borrow the +/// connection: the cancel composes on the shared wire stack (the +/// connection's `Arc`-held internals, kept alive by the handle's +/// arc) and works after the connection handle itself was dropped +/// (the stateless-form posture). +pub struct ReverseListener { + endpoint: ReverseEndpoint, + shared: Arc, +} + +impl ReverseListener { + /// The bound endpoint — the actually-bound address/port or socket + /// path (never the requested `0` for a mint: the peer's answer is + /// the surfaced number). + pub fn endpoint(&self) -> &ReverseEndpoint { + &self.endpoint + } + + /// The explicit cancellation — the wire teardown: the peer + /// removes the matching listen (the endpoint's raw wire strings, + /// sshd parity; the accept loop ends, the bound port frees) and + /// the confirmed cancel releases the connection's listener count. + /// The peer's refusal (an unknown listener — e.g. the already- + /// cancelled, or a cancelled by the connection's end itself) is + /// [`OpError::ListenRefused`](crate::OpError). The handle is + /// consumed (one cancel through it); a later cancel by the same + /// listener is the stateless wire forms' business. + pub async fn cancel(self) -> Result<(), OpError> { + match &self.endpoint { + ReverseEndpoint::Tcp { address, port } => { + cancel_reverse_tcp(&self.shared, address.clone(), *port).await + } + ReverseEndpoint::Unix { path } => { + #[cfg(unix)] + return cancel_reverse_unix(&self.shared, path.clone()).await; + #[cfg(not(unix))] + { + let _ = path; + Err(OpError::UnixUnavailable) + } + } + } + } +} + +impl std::fmt::Debug for ReverseListener { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ReverseListener") + .field("endpoint", &self.endpoint) + .finish() + } +} + +/// The tcp cancel's shared path — the stateless wire form and the +/// listener handle's `cancel` compose into it: the wire op, then the +/// confirmed release (exactly once: a second cancel hits the wire's +/// unknown-listener refusal, no release — budget-safe by the wire's +/// own state machine). +async fn cancel_reverse_tcp( + shared: &crate::client::ConnectionShared, + address: String, + port: u32, +) -> Result<(), OpError> { + match shared.handle.cancel_tcpip_forward(address, port).await { + Ok(()) => { + shared.listener_budget.release(); + Ok(()) + } + Err(error) => Err(OpError::from(error)), + } +} + +/// The unix cancel's shared path (the twin; the runtime-Unix gate +/// answers before the wire — see the module docs' gate record). +#[cfg(unix)] +async fn cancel_reverse_unix( + shared: &crate::client::ConnectionShared, + path: String, +) -> Result<(), OpError> { + match shared.handle.cancel_streamlocal_forward(path).await { + Ok(()) => { + shared.listener_budget.release(); + Ok(()) + } + Err(error) => Err(OpError::from(error)), + } +} + // --------------------------------------------------------------------------- // The reverse-listen rows (under `tunnel`) // --------------------------------------------------------------------------- @@ -402,14 +550,14 @@ impl SshConnection { /// [`OpError::ListenRefused`](crate::OpError) (the wire's /// `RequestDenied`) and releases the count; the connection stays /// usable after every refusal class. The listener is explicitly - /// cancellable ([`SshConnection::reverse_listen_cancel`]; it is - /// not implicitly unlistened by anything but the connection's - /// end). + /// cancellable — the returned [`ReverseListener`] handle's + /// `cancel()` (the wire-honest teardown; a cancel is never + /// triggered by a drop — the module docs' recorded decision). pub async fn reverse_listen( &self, address: impl Into, port: u32, - ) -> Result { + ) -> Result { if self.acceptors().tcpip_acceptor().is_none() { return Err(OpError::NoAcceptor); } @@ -419,7 +567,8 @@ impl SshConnection { { return Err(OpError::ListenCapExceeded); } - let bound = match self.russh_handle().tcpip_forward(address, port).await { + let address = address.into(); + let bound = match self.russh_handle().tcpip_forward(&address, port).await { Ok(minted) if port == 0 => minted, Ok(_) => port, Err(error) => { @@ -427,45 +576,46 @@ impl SshConnection { return Err(OpError::from(error)); } }; - Ok(bound) + Ok(ReverseListener { + endpoint: ReverseEndpoint::Tcp { + address, + port: bound, + }, + shared: Arc::clone(self.shared()), + }) } /// The reverse-listen cancel row (`-R` teardown): the explicit - /// cancellation of one bound listener (the lifecycle primitive: - /// every listener explicitly cancellable, the wire-honest path — - /// the peer tears its listen down, the bound port frees). Matches - /// by the raw wire strings (sshd parity). The peer's refusal - /// (unknown listener) is [`OpError::ListenRefused`](crate::OpError); - /// a confirmed cancel releases the listener count. + /// cancellation of one bound listener by the raw wire strings + /// (sshd parity) — the **stateless** wire form beside the + /// [`ReverseListener`] handles' `cancel()` (compose into the same + /// shared path; a confirmed cancel releases the listener count — + /// never twice, the wire's unknown-listener refusal governs the + /// repeats). The peer's refusal (unknown listener) is + /// [`OpError::ListenRefused`](crate::OpError). pub async fn reverse_listen_cancel( &self, address: impl Into, port: u32, ) -> Result<(), OpError> { - match self - .russh_handle() - .cancel_tcpip_forward(address, port) - .await - { - Ok(()) => { - self.listener_budget().release(); - Ok(()) - } - Err(error) => Err(OpError::from(error)), - } + cancel_reverse_tcp(self.shared(), address.into(), port).await } /// The streamlocal reverse-listen row (the unix `-R`): asks the /// vanilla server to listen on the unix socket at `socket_path`, /// minting no port (the global request's success is the outcome; - /// the accepted opens' echo carries the path). Same gates as + /// the accepted opens' echo carries the path). Returns the + /// [`ReverseListener`] handle like the tcp row. Same gates as /// [`SshConnection::reverse_listen`] for the /// `forwarded-streamlocal` slot, plus the **runtime-Unix gate** /// (OQ-11: off-Unix, [`OpError::UnixUnavailable`](crate::OpError) /// before any wire op — the bind is the platform-bound operation; /// the channel ingestion is platform-neutral). #[cfg_attr(not(unix), allow(unused_variables))] - pub async fn reverse_listen_unix(&self, socket_path: impl Into) -> Result<(), OpError> { + pub async fn reverse_listen_unix( + &self, + socket_path: impl Into, + ) -> Result { #[cfg(not(unix))] { return Err(OpError::UnixUnavailable); @@ -481,8 +631,12 @@ impl SshConnection { { return Err(OpError::ListenCapExceeded); } - match self.russh_handle().streamlocal_forward(socket_path).await { - Ok(()) => Ok(()), + let socket_path = socket_path.into(); + match self.russh_handle().streamlocal_forward(&socket_path).await { + Ok(()) => Ok(ReverseListener { + endpoint: ReverseEndpoint::Unix { path: socket_path }, + shared: Arc::clone(self.shared()), + }), Err(error) => { self.listener_budget().release(); Err(OpError::from(error)) @@ -492,7 +646,8 @@ impl SshConnection { } /// The streamlocal cancel row (matches by the socket path, sshd - /// parity — see [`SshConnection::reverse_listen_cancel`]). + /// parity — see [`SshConnection::reverse_listen_cancel`]; composes + /// into the same shared path as the unix handles' `cancel()`). #[cfg_attr(not(unix), allow(unused_variables))] pub async fn reverse_listen_unix_cancel( &self, @@ -504,17 +659,7 @@ impl SshConnection { } #[cfg(unix)] { - match self - .russh_handle() - .cancel_streamlocal_forward(socket_path) - .await - { - Ok(()) => { - self.listener_budget().release(); - Ok(()) - } - Err(error) => Err(OpError::from(error)), - } + cancel_reverse_unix(self.shared(), socket_path.into()).await } } } diff --git a/src/lib.rs b/src/lib.rs index 694d8a4..0b58003 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -60,12 +60,13 @@ pub use client::dial; #[doc(hidden)] pub use client::russh_client_config; /// The `-R` leg's acceptor registry + the reverse-listen rows' -/// payload/acceptor types (`w4-reverse-listen` — under `tunnel`; the -/// recorded decisions live in the `client_accept` module docs). +/// payload/acceptor types + the listener handles (`w4-reverse-listen` +/// / `w4-lifecycle` — under `tunnel`; the recorded decisions live in +/// the `client_accept` module docs). #[cfg(feature = "tunnel")] pub use client_accept::{ - AcceptorHeld, AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip, - StreamlocalForwardedAcceptor, TcpipForwardedAcceptor, + AcceptorHeld, AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip, ReverseEndpoint, + ReverseListener, StreamlocalForwardedAcceptor, TcpipForwardedAcceptor, }; #[doc(hidden)] pub use client_handler::ClientHandler; diff --git a/tasks/implementation/w4-lifecycle.md b/tasks/implementation/w4-lifecycle.md index 001e13e..20b14cc 100644 --- a/tasks/implementation/w4-lifecycle.md +++ b/tasks/implementation/w4-lifecycle.md @@ -1,7 +1,7 @@ --- id: w4-lifecycle name: Wave 4 — lifecycle primitives (cancellable handles + clean disconnect) -status: pending +status: completed depends_on: [w4-op-rows, w4-reverse-listen] scope: narrow risk: low @@ -38,14 +38,14 @@ policy *possible* — and nothing more: ## Acceptance Criteria -- [ ] Every client-direction listener/handle is explicitly +- [x] Every client-direction listener/handle is explicitly cancellable; no drop-timing-only wire state -- [ ] `Disconnect::ByApplication` clean-close lands; the peer +- [x] `Disconnect::ByApplication` clean-close lands; the peer observes a clean disconnect, no half-open channels (row-pinned) -- [ ] The no-reconnect-policy boundary stated in module docs (the +- [x] The no-reconnect-policy boundary stated in module docs (the primitives-not-policy posture) -- [ ] Contract rows green through the server double -- [ ] `cargo test` (+ per-feature + `--all-features`), clippy ×2 +- [x] Contract rows green through the server double +- [x] `cargo test` (+ per-feature + `--all-features`), clippy ×2 (`-D warnings`), fmt, `cargo doc --no-deps` pass ## References @@ -62,8 +62,133 @@ policy *possible* — and nothing more: ## Notes -> To be filled by implementation agent +### What landed (2026-10-29) + +- `src/client_accept.rs`: the **listener handles** — + [`ReverseListener`] + [`ReverseEndpoint`] (the handles' surface the + `w4-reverse-listen` block deferred to this task): `reverse_listen` / + `reverse_listen_unix` return a handle carrying the *actually-bound* + endpoint (the `bound_port()` accessor, `None` for the unix shape) + with the explicit `cancel()`; the shared cancel helpers + (`cancel_reverse_tcp` / `cancel_reverse_unix`) — both the handles' + `cancel()` and the stateless wire forms compose into the same + paths; the confirmed cancel releases the connection's listener + count exactly once (the repeat is the wire's unknown-listener + refusal, no release). The recorded decisions (module docs): no + `Drop` hides a wire op (async cancel cannot ride a sync `Drop`); + the handle shares the connection's internals through the arc so + its `cancel()` works against a live stack regardless of the + connection handle's own holders; the stateless forms ride beside + 1:1. +- `src/client.rs`: the **`SshConnection` restructure** — the russh + handle, limits/labels, acceptor registry and listener budget ride + one `ConnectionShared` behind the `Arc` (the concurrency-safe + sharing the handles' independent `cancel()`s need; the russh client + request methods all take `&self` at this pin, so the arc is the + sharing shape — no mutex), `SshConnection`'s public surface + unchanged (`close()` consumes the handle and sends + `Disconnect::ByApplication`), and the module docs' new **§Lifecycle + posture** (the primitives-not-policy boundary, reverse-runner fact + 4; the explicit-vs-drop facts, row-pinned). +- `src/lib.rs`: the `ReverseListener` / `ReverseEndpoint` exports + (under `tunnel`). +- `docs/architecture/client.md`: the §Lifecycle primitives landing + block (the handle shape + the abrupt-drop/clean-close boundary + + the modules-docs posture anchors). +- `tests/client_dial.rs`: the every-build row + `dropping_the_connection_is_an_abrupt_teardown_close_is_the_only_clean_path` + (the `next_session_outcome` observable: drop → the peer's session + ends with an **error wire class, never the clean `Ok`** — the + boundary pinned both ways: the clean `Ok` is only `close()`'s). +- `tests/client_reverse_listen.rs`: three new lifecycle rows — + `dropping_the_listener_handle_leaves_no_drop_timing_wire_state` + (drop → the peer's listen still serves a round-tripping data leg; + only the explicit cancel tears down), + `the_listener_handle_cancel_tears_the_forward_down_and_releases_the_budget` + (ceiling → refuse pre-wire; the handle's cancel confirms and + releases; the next listen admits; the repeat rides the wire and + refuses — no budget manufactured), + `cancel_mid_data_leg_tears_the_forward_down_mid_connection` (the + cancel lands while a data leg is mid-pump; the forward tears down, + the connection stays usable, ends clean); the existing rows + adapted to the handle returns (the endpoint assertions ride the + handles now); the unix row's cancel exercises the handle's + `cancel()` (the streamlocal flavor). + +### Decided + recorded + +1. **The handles' shape — single `ReverseListener`, one + `ReverseEndpoint` enum, not two handle types.** The endpoint enum + mirrors the wire's two bind shapes (an actually-bound + address+port / a socket path); the tcp-flavored `bound_port()` + accessor returns `Option` (the unix shape carries no port — + honest, no panic path). A second handle type would have duplicated + the cancel plumbing for a shape the cancel forms already treat as + raw strings. +2. **The abrupt-drop fact (row-pinned, corrected mid-task):** at this + russh pin dropping the *last* wire-stack handle closes the + `Handle`'s sender; the client runner's `receiver.recv()` → `None` + sets `disconnected` and the transport tears down abruptly — the + peer's session outcome is an error wire class (`Err(Disconnect)` + observed through the double). So the honest boundary is not + "drop does nothing" but **"drop is never the clean path"** — the + clean `Ok` session end is only `close()`'s + (`Disconnect::ByApplication`), row-pinned in both directions + (abrupt-on-drop, clean-on-close) through the server double. No + handle relies on drop timing for *clean* wire state, and the + listener handles' drop is a pure local no-op (no `Drop` hook; a + cancellation is async and cannot hide in one — the listen lives at + the peer until an explicit cancel or the connection's end). +3. **The `SshConnection` restructure is a shape change, not a + surface change** — the public surface (the wrapped methods, + `limits()`/`address()`/`username()`/`is_closed()`/`acceptors()`/ + `close()`) is unchanged; only the internals moved behind the arc + (the handles' independent cancel). No `Clone` was added — not in + the contract, and the arc means a cloned handle's `close()` + semantics would need their own contract; wave-5 review can add it + if an assembly shape needs it. +4. **A live `ReverseListener` keeps the wire stack alive after the + connection handle's drop** — the arc's ownership side effect, + recorded (the alkminer flow's shape: the listener handle outlives + the connection slot; its `cancel()` composes on the still-live + stack). + +### Verification + +`cargo test` (default), `--features session`, `--features git`, +`--features tunnel`, `--all-features` — all green (16 test binaries +each); `cargo clippy --all-targets -- -D warnings` and +`cargo clippy --all-features --all-targets -- -D warnings` clean; +`cargo fmt --check` clean; `cargo doc --no-deps` warning-free in the +default build (the pre-existing `client_ops.rs` private-link warnings +remain under `--all-features`, unchanged). ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +Landed (`feat(client)`: w4-lifecycle): + +- **`src/client_accept.rs`** — the listener handles: `ReverseListener` + (+ `ReverseEndpoint`) — `reverse_listen` / `reverse_listen_unix` + now return them; the actually-bound endpoint carried, the explicit + `cancel()` the wire-honest teardown, the listener count released + exactly once per confirmed cancel through the shared cancel paths + the stateless wire forms compose into too; no `Drop` hides a wire + op (recorded, row-pinned no-drop-timing wire state). +- **`src/client.rs`** — the `SshConnection` internals behind the arc + (`ConnectionShared`) + the §Lifecycle posture module docs (the + primitives-not-policy boundary, the abrupt-drop/clean-close facts); + the public surface unchanged (`close()` is the clean path). +- **`src/lib.rs`** — the handle exports; **`docs/architecture/ + client.md`** — the §Lifecycle primitives landing block. +- **Rows** — 3 new tunnel-build rows through the server double + (`tests/client_reverse_listen.rs`: the drop-handle no-wire-state + row, the handle-cancel + budget-release row, the cancel-mid-data-leg + row) + 1 new every-build row (`tests/client_dial.rs`: the abrupt- + drop row); the existing reverse rows adapted to the handles; the + unix cancel row exercises the handle flavor. + +Verification: `cargo test` / `--features ` / `--all-features` +all green; clippy ×2 (`-D warnings`) clean; `cargo fmt --check` +clean; `cargo doc --no-deps` clean (default build). \ No newline at end of file diff --git a/tests/client_dial.rs b/tests/client_dial.rs index 909fc22..c9f7f3a 100644 --- a/tests/client_dial.rs +++ b/tests/client_dial.rs @@ -90,6 +90,39 @@ fn postured_key_source(posture: &'static str) -> (Arc, KeySourceE } } +/// - *Contract stamp:* `client.md` §Lifecycle primitives (the +/// explicit-vs-drop boundary, the OQ-12 primitives-only posture) — +/// **drop is never the clean path**: dropping the (feature-free, +/// last) wire-stack handle ends the runner and the transport goes +/// down abruptly — the double's serving outcome is the error wire +/// class, never the clean session end only `close()` produces (see +/// `dial_success_lands_an_authenticated_connection`'s `Ok`). +#[tokio::test] +async fn dropping_the_connection_is_an_abrupt_teardown_close_is_the_only_clean_path() { + let (mut double, accepts) = spawn_double().await; + accepts.register(fixture("alpha").fingerprint); + + let policy = Arc::new(MemHostKeyPolicy::new()); + policy.pin("127.0.0.1", fixture("bravo").fingerprint); + let config = DialConfig::new( + double.address.clone(), + "runner", + alpha_key_source(), + Arc::clone(&policy) as Arc, + ); + let connection = dial(config).await.expect("the dial succeeds"); + + drop(connection); + + let outcome = double.next_session_outcome().await; + assert!( + matches!(outcome, Some(Err(_))), + "the un-closed drop tears the peer's session down abruptly — \ + never the clean `Ok`, got {outcome:?}" + ); + double.shutdown(); +} + /// - *Contract stamp:* the dial surface's success row: decrypt-at-dial /// → the policy accept → the single publickey attempt — an /// authenticated connection, the dial data carried on its surface, diff --git a/tests/client_reverse_listen.rs b/tests/client_reverse_listen.rs index ba93459..853f501 100644 --- a/tests/client_reverse_listen.rs +++ b/tests/client_reverse_listen.rs @@ -1,5 +1,6 @@ -//! The `-R` leg's contract rows (`w4-reverse-listen`) — driven through -//! the in-process fake `sshd` grown with the reverse-forward surfaces +//! The `-R` leg's contract rows (`w4-reverse-listen` + the listener +//! handles' lifecycle rows — `w4-lifecycle`) — driven through the +//! in-process fake `sshd` grown with the reverse-forward surfaces //! ([`sshd_double`](common::sshd_double): the `tcpip-forward` / //! `streamlocal-forward` hooks that bind **real** sockets and open the //! `forwarded-*` channels toward the client per accepted connection): @@ -9,8 +10,13 @@ //! live connection — decline-before-install, serve-after; uninstall; //! the data-leg pump round trip), the off-Unix streamlocal rows //! (OQ-11), and the connection-stays-usable row after every decline -//! class. The row/stamp discipline (`tests/common/mod.rs`'s doc) -//! applies verbatim. +//! class. The lifecycle rows pin the handles' posture: the actually- +//! bound endpoint is carried by the handle, dropping a handle leaves +//! no wire state (no drop-timing teardown), the handle's `cancel()` is +//! the wire-honest teardown that releases the listener count, and a +//! cancel mid-data-leg tears the forward down mid-connection. The +//! row/stamp discipline (`tests/common/mod.rs`'s doc) applies +//! verbatim. //! //! **Per-build matrix note (recorded):** this file is the //! `tunnel`-featured build's rows (the only build whose surface can @@ -206,11 +212,15 @@ async fn reverse_listen_serves_and_the_data_leg_pumps() { .expect("the slot installs"); let port = u32::from(pick_port().await); - let bound = connection + let listener = connection .reverse_listen("127.0.0.1", port) .await .expect("the featured + registered listen serves"); - assert_eq!(bound, port, "an explicit port binds exactly that port"); + assert_eq!( + listener.endpoint().bound_port(), + Some(port), + "an explicit port binds exactly that port — the handle carries it" + ); let mut conn = tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) @@ -266,10 +276,14 @@ async fn port_zero_mint_surfaces_the_actually_bound_port() { })) .expect("the slot installs"); - let bound = connection + let port_zero_mint = connection .reverse_listen("127.0.0.1", 0) .await .expect("the mint serves"); + let bound = port_zero_mint + .endpoint() + .bound_port() + .expect("the tcp mint's shape carries its bound port"); assert!( bound > 0, "the served mint is the actually-bound port, never the requested 0" @@ -564,6 +578,248 @@ async fn cancel_tears_the_forward_down_and_unknown_cancels_refuse() { expect_clean_close(connection, double).await; } +/// - *Contract stamp:* `client.md` §Lifecycle primitives (the +/// explicit-vs-drop boundary, `w4-lifecycle`) — **no drop-timing +/// wire state**: dropping the listener handle without a cancel +/// frees nothing — the peer's listen keeps serving (the arrival +/// still hands off through the registered acceptor); only the +/// explicit cancel tears the forward down (the bound port then +/// refuses a reconnect). The `ReverseListener` carries no `Drop` +/// hook on purpose (async cancellation cannot hide in one). +#[tokio::test] +async fn dropping_the_listener_handle_leaves_no_drop_timing_wire_state() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: true, + })) + .expect("the slot installs"); + + let port = u32::from(pick_port().await); + let listener = connection + .reverse_listen("127.0.0.1", port) + .await + .expect("the listen binds"); + drop(listener); + + let mut conn = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await + .expect("the dropped handle left the peer's listen serving"); + conn.write_all(b"after-drop") + .await + .expect("the still-live forward carries the bytes"); + let mut echoed = [0u8; 10]; + tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed)) + .await + .expect("the round trip completes in time") + .expect("the echo reads"); + assert_eq!( + &echoed, b"after-drop", + "the un-cancelled listen still serves data legs" + ); + let record = next_open(&mut opens_rx, "the after-drop row").await; + assert_eq!( + record.connected_port, port, + "the arrival matched the listen" + ); + + connection + .reverse_listen_cancel("127.0.0.1", port) + .await + .expect("the explicit cancel is the teardown path"); + assert_eq!( + observed.cancels(), + vec![("cancel-tcpip-forward", "127.0.0.1".to_string(), port)], + "only the explicit cancel rode the wire" + ); + + let mut refused = false; + for _ in 0..200 { + match tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await + { + Ok(_) => tokio::time::sleep(Duration::from_millis(25)).await, + Err(err) if err.kind() == std::io::ErrorKind::ConnectionRefused => { + refused = true; + break; + } + Err(err) => panic!("the bound port errors, not refuses, after the cancel: {err:?}"), + } + } + assert!(refused, "the explicit cancel tore the forward down"); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* `client.md` §Lifecycle primitives (the +/// cancel-release relation) + the cap relation with the +/// dedicated listener counter — the handle's `cancel()` is the +/// confirmed teardown **which releases the listener count**: beyond +/// the ceiling the listens refuse pre-wire, the handle's cancel +/// (not the stateless form) recovers the budget, the next listen +/// admits again. The handle is consumed by its cancel — the one +/// cancel through it; a second cancel rides the connection row and +/// refuses (the unknown-listener wire answer, no release). +#[tokio::test] +async fn the_listener_handle_cancel_tears_the_forward_down_and_releases_the_budget() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let limits = alkssh::Limits { + max_concurrent_channels: 1, + ..alkssh::Limits::default() + }; + let (connection, double, observed) = dial_scripted_limits(scripted, limits).await; + + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the slot installs"); + + let first = u32::from(pick_port().await); + let listener = connection + .reverse_listen("127.0.0.1", first) + .await + .expect("the first listen admits"); + assert_eq!(listener.endpoint().bound_port(), Some(first)); + + let second = connection + .reverse_listen("127.0.0.1", u32::from(pick_port().await)) + .await; + assert!( + matches!(second, Err(OpError::ListenCapExceeded)), + "beyond the ceiling the listen refuses pre-wire, got {second:?}" + ); + + listener + .cancel() + .await + .expect("the handle's cancel confirms"); + assert_eq!( + observed.cancels(), + vec![("cancel-tcpip-forward", "127.0.0.1".to_string(), first)], + "the handle's cancel rode the wire verbatim" + ); + + let third = u32::from(pick_port().await); + let again = connection + .reverse_listen("127.0.0.1", third) + .await + .expect("the released budget admits again"); + assert_eq!(again.endpoint().bound_port(), Some(third)); + + again + .cancel() + .await + .expect("the handle's second cancel confirms and releases again"); + assert_eq!( + observed + .cancels() + .iter() + .map(|(_, _, port)| *port) + .collect::>(), + vec![first, third], + "each handle's cancel rode the wire — one confirmed cancel per \ + admitted listener" + ); + + assert!( + matches!( + connection.reverse_listen_cancel("127.0.0.1", third).await, + Err(OpError::ListenRefused) + ), + "the stateless form's repeat is the unknown-listener refusal — \ + the confirmed release ran exactly once (the budgets never \ + manufactured out of a repeated cancel)" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* `client.md` §Lifecycle primitives (the row's +/// recorded contract) + the teardown-mid-connection posture — the +/// cancel lands **while a data leg is live**: the confirmed cancel +/// tears the forward down (the bound port refuses a reconnect), the +/// connection stays usable (the `cancel` row's channel-scoped +/// honesty) and ends clean despite the live leg (the peer observes +/// the clean session end — no half-open channels). +#[tokio::test] +async fn cancel_mid_data_leg_tears_the_forward_down_mid_connection() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, _observed) = dial_scripted(scripted).await; + + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: true, + })) + .expect("the slot installs"); + + let port = u32::from(pick_port().await); + connection + .reverse_listen("127.0.0.1", port) + .await + .expect("the listen binds"); + + let mut conn = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await + .expect("the forward accepts the data leg"); + conn.write_all(b"leg-live") + .await + .expect("the leg's bytes ride"); + let mut echoed = [0u8; 8]; + tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed)) + .await + .expect("the echo arrives") + .expect("the echo reads"); + assert_eq!(&echoed, b"leg-live", "the leg is live mid-pump"); + + connection + .reverse_listen_cancel("127.0.0.1", port) + .await + .expect("the cancel confirms mid-connection"); + + let mut refused = false; + for _ in 0..200 { + match tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await + { + Ok(_) => tokio::time::sleep(Duration::from_millis(25)).await, + Err(err) if err.kind() == std::io::ErrorKind::ConnectionRefused => { + refused = true; + break; + } + Err(err) => panic!("the bound port errors, not refuses, after the cancel: {err:?}"), + } + } + assert!( + refused, + "the forward is torn down while the leg was live — no more arrivals" + ); + + expect_clean_close(connection, double).await; +} + /// - *Contract stamp:* the module docs' recorded cap relation (the /// server-side listener-counter posture mirrored) — the dedicated /// listener count against the `max_concurrent_channels` ceiling: @@ -698,7 +954,7 @@ async fn streamlocal_listen_serves_arrivals_and_cancels_through_the_unix_gate() let socket_path = socket_path.to_string_lossy().to_string(); let _ = std::fs::remove_file(&socket_path); - connection + let unix_listener = connection .reverse_listen_unix(&socket_path) .await .expect("the unix listen serves"); @@ -733,8 +989,8 @@ async fn streamlocal_listen_serves_arrivals_and_cancels_through_the_unix_gate() "the open echoes the listen's socket path verbatim" ); - connection - .reverse_listen_unix_cancel(&socket_path) + unix_listener + .cancel() .await .expect("the unix cancel confirms"); assert_eq!( diff --git a/tests/tunnel_mapping.rs b/tests/tunnel_mapping.rs index de51053..cb9f2c9 100644 --- a/tests/tunnel_mapping.rs +++ b/tests/tunnel_mapping.rs @@ -2000,10 +2000,14 @@ mod client_smoke { .install_tcpip(Arc::new(EchoAcceptor { opens: opens_tx })) .expect("the acceptor installs"); - let bound = connection + let minted_listener = connection .reverse_listen("127.0.0.1", 0) .await .expect("the full-circle mint serves"); + let bound = minted_listener + .endpoint() + .bound_port() + .expect("the tcp mint's handle carries the actually-bound port"); assert!( bound > 0, "the server direction's fabric minted a real bound port"