feat(client): land the lifecycle primitives — listener handles + the clean-disconnect boundary (w4-lifecycle)

- src/client_accept.rs: the ReverseListener handles (ReverseEndpoint
  carried; cancel() the wire-honest teardown) returned by the -R listen
  rows; the shared cancel paths both the handles and the stateless wire
  forms compose into; the confirmed cancel releases the listener count
  exactly once; no Drop hides a wire op
- src/client.rs: the SshConnection internals behind the Arc
  (ConnectionShared) — the handles' independent cancels compose on the
  same stack; the §Lifecycle posture module docs (primitives-not-policy,
  OQ-12; the abrupt-drop/clean-close facts, row-pinned)
- rows: drop-handle no-wire-state, handle-cancel + budget release,
  cancel-mid-data-leg (tunnel build) + the every-build abrupt-drop row;
  existing reverse rows adapted to the handles; the unix cancel rides
  the handle flavor
- docs/architecture/client.md: the §Lifecycle primitives landing block;
  task file filled

Verification: cargo test (default / session / git / tunnel / all-features)
all green; clippy ×2 -D warnings clean; fmt clean; cargo doc --no-deps clean
This commit is contained in:
glm-5.3-flash committed 2026-10-29 17:42:29 +00:00
1 parent 7e5989f574
commit dec644e653
8 files changed
+762 -106

No files matched your search

+25
View File
@@ -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
+108 -41
View File
@@ -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<SshConnection, DialError> {
}
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<ClientHandler>,
limits: Limits,
address: String,
username: String,
shared: Arc<ConnectionShared>,
}
/// 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<ClientHandler>,
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<AcceptorRegistry>,
pub(crate) acceptors: Arc<AcceptorRegistry>,
/// 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<ChannelCap>,
}
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<ConnectionShared> {
&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<crate::client_handler::ClientHandler> {
&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()
}
}
+187 -42
View File
@@ -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<u32> {
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<crate::client::ConnectionShared>,
}
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<String>,
port: u32,
) -> Result<u32, OpError> {
) -> Result<ReverseListener, OpError> {
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<String>,
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<String>) -> Result<(), OpError> {
pub async fn reverse_listen_unix(
&self,
socket_path: impl Into<String>,
) -> Result<ReverseListener, OpError> {
#[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
}
}
}
+5 -4
View File
@@ -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;
+133 -8
View File
@@ -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
> 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 <each>` / `--all-features`
all green; clippy ×2 (`-D warnings`) clean; `cargo fmt --check`
clean; `cargo doc --no-deps` clean (default build).
+33
View File
@@ -90,6 +90,39 @@ fn postured_key_source(posture: &'static str) -> (Arc<dyn KeySource>, 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<dyn HostKeyPolicy>,
);
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,
+266 -10
View File
@@ -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<_>>(),
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!(
+5 -1
View File
@@ -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"