diff --git a/src/git_map.rs b/src/git_map.rs index ad3284e..41a3747 100644 --- a/src/git_map.rs +++ b/src/git_map.rs @@ -40,6 +40,33 @@ //! evidence, contract-pinned by the denial rows in //! `tests/git_mapping.rs`. //! +//! - **Status-before-close by construction** (the wave-3 gate's +//! booked finding 1, `w4-git-exit-close`): the seam's stream is +//! built from the channel's **split halves**, never +//! `Channel::into_stream()`. `into_stream` wraps the whole channel +//! in russh's `ChannelCloseOnDrop`, whose `Drop` sends +//! server→client `CHANNEL_CLOSE` from a *spawned task* — inside the +//! seam call, racing [`serve_git_row`]'s explicit +//! `exit_status_request` for a queue position on the same bounded +//! `Sender` (russh's `event_buffer_size`, default 10): a full +//! buffer at the serve tail could enqueue the implicit close first, +//! let the loop remove the channel, and silently no-op the report. +//! The split-halves shape removes the competitor instead of racing +//! it: russh 0.65 keeps its `ChannelRx`/`ChannelTx` private, so +//! [`GitChannelDuplex`] is built from the public API — the reader +//! is the module's [`ChannelHalveReader`] (an owned read half +//! repinning the private `ChannelRx`'s observation semantics +//! without the close-on-drop wrapper; its only `Drop` behavior +//! differs from `into_stream`'s), the writer the write half's +//! `ChannelTx` via `make_writer_ext` (its `poll_shutdown` still +//! sends EOF, exactly `into_stream`'s). Nothing the seam drops ever +//! messages the wire, so the report's send is the sole +//! close-contracting sender, and its queue position ahead of the +//! explicit close holds by construction — sequential awaited sends +//! on the one `Handle`, saturated buffer or not. The tty rows are +//! not in this class: their report precedes any close-contracting +//! drop within one task — the wave-3 gate verified. +//! //! - **Exit-status mapping** (POC-3 §A.5, decided): //! [`serve_git_row`] maps the serve outcome onto exactly two //! codes — `Ok(())` → `exit-status(0)` (vanilla git exits 0 on @@ -80,12 +107,16 @@ //! substrate already mapped its own error onto the wire (bands, the //! denial), and the client's observation is the exit-status above. +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; +use std::task::{Context, Poll}; use alkcall::core::auth::Identity; use alkgit::{Service, SubstrateError}; -use russh::Channel; -use russh::server::Handle; +use russh::server::{Handle, Msg}; +use russh::{Channel, ChannelMsg, ChannelReadHalf}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use crate::config::GitSeam; use crate::dispatch::GitServiceRoute; @@ -119,12 +150,161 @@ fn exit_code(outcome: &Result<(), SubstrateError>) -> u32 { } } -/// The spawned git serving path: dispatch to the seam over the channel -/// duplex, then the exit-status-then-close report through the cloned -/// handle. The channel was taken from the handler's store by the -/// dispatch (the store-and-take handoff); the spawned task owns the -/// stream and its report surface. A failing report send means the -/// session is gone — the lifecycle's `warn`-failed records it (the +/// The seam's duplex over the **split** channel halves — the +/// status-before-close shape's carrier (the module doc's +/// §Status-before-close decision). russh 0.65 keeps its `ChannelRx`/ +/// `ChannelTx` types private, so both halves are built from public +/// API: the reader is a [`ChannelHalveReader`] (an owned close-free +/// read half), the writer the write half's `ChannelTx` via +/// `make_writer_ext(None)` — its `poll_shutdown` sends EOF, exactly +/// `into_stream`'s semantics, its `Drop` messages nothing (the +/// `ChannelTx` `Drop` is a writer-notification only). Nothing the +/// seam drops ever messages the wire, so the report's send is the +/// sole close-contracting sender, ordered by construction ahead of +/// the explicit close (sequential awaited sends on the one `Handle`). +struct GitChannelDuplex { + reader: ChannelHalveReader, + writer: Box, +} + +impl AsyncRead for GitChannelDuplex { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.reader).poll_read(cx, buf) + } +} + +impl AsyncWrite for GitChannelDuplex { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.writer).poll_write(cx, buf) + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.writer).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.writer).poll_shutdown(cx) + } +} + +/// The owned close-free read half — the seam-side reader. Semantics +/// repinned from russh's private `ChannelRx` (the `into_stream` +/// shape's reader; identical observable behavior without the +/// `ChannelCloseOnDrop` wrapper): `Data` fills the buffer (partial +/// reads buffered), `ExtendedData` with no sideband and the other +/// non-data messages are swallowed and re-read (the `_` arm), +/// `Eof` ends the stream, and the peer's `Close` leaves the reader +/// to the channel's sender-death EOF (`wait()` → `None` — the +/// channel's internal sender drops at the loop's close processing). +/// +/// The `wait()` access is future-owned: each read round-trips the +/// half through a boxed [`Future`] that parks it inside russh's +/// `ChannelReadHalf::wait` and hands it back with the observed +/// message on completion — the reader is fully owned (usable under +/// the seam's `'static` `T` bound) across every park, buffer and +/// drop state; no `Drop` path ever messages the wire. +struct ChannelHalveReader { + slot: Option, + parked: Option, + buffered: Option<(ChannelMsg, usize)>, + eof: bool, +} + +type PinnedHalfWait = Pin)> + Send>>; + +impl ChannelHalveReader { + fn new(half: ChannelReadHalf) -> Self { + Self { + slot: Some(half), + parked: None, + buffered: None, + eof: false, + } + } +} + +impl AsyncRead for ChannelHalveReader { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + loop { + if self.eof { + return Poll::Ready(Ok(())); + } + match self.buffered.take() { + Some((ChannelMsg::Data { data }, index)) => { + let readable = buf.remaining().min(data.len() - index); + buf.put_slice(&data[index..index + readable]); + if index + readable < data.len() { + self.buffered = Some((ChannelMsg::Data { data }, index + readable)); + } + return Poll::Ready(Ok(())); + } + Some((ChannelMsg::Eof, _)) => self.eof = true, + Some(_) => continue, + None => {} + } + if self.eof { + return Poll::Ready(Ok(())); + } + if let Some(mut parked) = self.parked.take() { + let (half, observed) = match parked.as_mut().poll(cx) { + Poll::Ready((half, observed)) => (half, observed), + Poll::Pending => { + self.parked = Some(parked); + return Poll::Pending; + } + }; + self.slot = Some(half); + match observed { + Some(msg) => self.buffered = Some((msg, 0)), + None => self.eof = true, + } + continue; + } + let Some(mut half) = self.slot.take() else { + self.eof = true; + continue; + }; + self.parked = Some(Box::pin(async move { + let observed = half.wait().await; + (half, observed) + })); + } + } +} + +fn git_channel_duplex(channel: Channel) -> GitChannelDuplex { + let (reader_half, writer_half) = channel.split(); + // The `make_writer_ext` return captures the half's borrow (Rust + // 2024 capture rules) though the value only holds clones; the + // trait-object erasure drops the phantom capture, leaving the + // 'static seam bound satisfiable without parking the half. + let writer = Box::new(writer_half.make_writer_ext(None)) as Box; + let reader = ChannelHalveReader::new(reader_half); + GitChannelDuplex { reader, writer } +} + +/// The spawned git serving path: dispatch to the seam over the +/// [`git_channel_duplex`] (the split halves — the seam contracts +/// nothing on its drops), then the exit-status-then-close report +/// through the cloned handle — the report's send is the sole +/// close-contracting sender and each send awaits before the next is +/// issued, so the status-before-close order holds by construction +/// under any schedule. The channel was taken from the handler's store +/// by the dispatch (the store-and-take handoff); the spawned task owns +/// the halves and their report surface. A failing report send means +/// the session is gone — the lifecycle's `warn`-failed records it (the /// one warn of the vocabulary). pub(crate) async fn serve_git_row( handle: Handle, @@ -152,7 +332,7 @@ pub(crate) async fn serve_git_row( &repo_id, Some(&identity), alkgit_service(route), - channel.into_stream(), + git_channel_duplex(channel), seam.limits, Arc::clone(&seam.registry), Arc::clone(&seam.refs), diff --git a/tasks/implementation/w4-git-exit-close.md b/tasks/implementation/w4-git-exit-close.md index cfbe651..4d0296a 100644 --- a/tasks/implementation/w4-git-exit-close.md +++ b/tasks/implementation/w4-git-exit-close.md @@ -1,7 +1,7 @@ --- id: w4-git-exit-close name: Wave 4 — git row hardening (status-before-close by construction — wave-3 gate booked item 1) -status: pending +status: completed depends_on: [w3-git-mapping] scope: narrow risk: medium @@ -71,8 +71,114 @@ the by-construction property proven on the wire, not by reasoning. ## Notes -> To be filled by implementation agent +### The shape decided (2026-10-29) + +Went with the first named option — **split the channel halves in +`serve_git_row` and pass them into the seam** — refined by one russh +0.65 reality: `russh::channels::io::{ChannelRx, ChannelTx}` are +*private* at this pin, so the split-halves duplex is rebuilt from the +public API in `src/git_map.rs`: + +- **The reader** is `git_map`'s own + [`ChannelHalveReader`]: the owned `ChannelReadHalf` + (`channel.split()`), a `poll_read` state machine whose observation + semantics repin russh's private `ChannelRx` *without* the + `ChannelCloseOnDrop` wrapper — `Data` fills (partial reads + buffered), `ExtendedData`/other non-data messages swallowed and + re-read (the `_`-arm parity), `Eof` ends the stream, the peer's + `Close` leaves the reader to the channel's sender-death EOF + (`wait()` → `None`). The `wait()` access is *future-owned*: each + read round-trips the half through a boxed future that parks it + inside `ChannelReadHalf::wait` and hands it back with the observed + message — fully owned (`'static`-usable, the seam's `T` bound) + across every park/buffer/drop state, no extra task, and **no + `Drop` path ever messages the wire**. +- **The writer** is the write half's `ChannelTx` via + `make_writer_ext(None)`. Its value is `'static`-free of the half + (clones only) but the Rust 2024 capture rules give the opaque a + phantom of the half's borrow; a trait-object erasure + (`Box`) drops the phantom, leaving + the seam's `'static` bound satisfiable without parking the half. + `poll_shutdown` still sends EOF (exactly `into_stream`'s). + +The result: **nothing the seam drops ever messages the wire** — the +report's `exit_status_request` send is the sole close-contracting +sender, and its queue position ahead of the explicit close holds by +construction (sequential awaited sends on the one `Handle`), +saturated buffer or not. The hazard's premise (russh's +`ChannelCloseOnDrop` `Drop` spawning a `CHANNEL_CLOSE` send that can +win a queue position on the bounded `Sender` — +`event_buffer_size`, default 10 — before the report's) is *removed*, +not raced. + +### The wire row (the by-construction property pinned) + +`tests/git_mapping.rs::backlogged_seam_tail_still_reports_exit_status_before_close` +— the finding's failure scenario made real on the wire: the serve's +answer volume (300 one-pkt-line ref answers) exceeds the entire +outbound pipeline via the new client-direction data gate +(`tests/common/harness.rs::DataGate` + the harness handler's `data` +hook, which russh's client loop awaits inline): the armed gate freezes +the client's loop → the transport duplex fills (`DUPLEX_CAPACITY`) → +the server's flush pends → russh's bounded message queue fills → the +serve's remaining answer writes pend (the serve tail blocked). The +gate's release then drives the serve tail through the just-unwinding +contention, and the row pins: the whole backlogged answer drains +**through the same queue the report rides**, then exit-status(0), then +close — order asserted, exit code asserted. + +**The row falsifies the old shape.** With a temporary one-line revert +of `serve_git_row` to `channel.into_stream()` the row FAILED +**15/15** runs — the implicit close won the frozen-drain queue race +("expected exit-status(0), got Some(Close)"), the silent report loss +the gate booked. With the split-halves shape: **10/10 green**. So the +row is a two-sided pin: deterministic red under the hazard, green by +construction without it. + +Test-vehicle additions (test surface, freely versioned): the doubles' +per-repo serve gate (`tests/common/git_doubles.rs`: `install_gate` — +race-safe double-check `MemGate`, gating `GitRefs::list_refs`, so the +row schedules the serve's stages against prepared wire state) and the +harness's `DataGate` + `drive_with_data_gate` (the `drive*` bodies +factored into the shared `mount_and_connect` core; existing wrappers' +signatures unchanged). + +### Constraints check + +- **Every existing git wire row green** — the full git suite + (`tests/git_mapping.rs`, 12 rows) and every other loopback suite: + no behavior change on the happy path (the wire order the rows pin — + data, exit-status, close — is identical under both shapes when the + race doesn't fire). +- **Spawn-and-return intact** (ADR-006 decision 4): the reporting + still rides the spawned serving task through the cloned + `Session::handle()`; `serve.rs`'s mechanics untouched. +- **The tty rows untouched** (not in this class — the gate verified). +- **No API/limits change**: `event_buffer_size` stays russh's default + posture (the adapter docs' recorded default posture preserved); the + row pins the default's value with a note that a russh default + change re-reads the row's freeze-settle yields, not its assertions. ## Summary -> To be filled on completion \ No newline at end of file +**Completed.** The git row's status-before-close now holds **by +construction**: `serve_git_row` no longer hands russh's +close-on-drop `ChannelStream` (`into_stream()`) to the seam — the +seam rides a split-halves duplex (`git_channel_duplex` → +`ChannelHalveReader` over the owned read half + the write half's +`ChannelTx` via the capture-erased trait object) where **no drop +path ever messages the wire**, making the report's send the sole +close-contracting sender and the report-before-close order +unconditional (russh 0.65's private `ChannelRx`/`ChannelTx` noted; +the semantics repinned from the public API). New wire row pins the +by-construction property on a genuinely saturated tail (volume flood +→ frozen client loop → duplex → bounded message queue; gate release +→ the tail through the unwinding contention) — and falsifies the old +shape deterministically (15/15 red under a temporary +`into_stream()` revert, 10/10 green with the fix). Doubles gained a +race-safe per-repo serve gate; the harness gained the client-side +data gate + `drive_with_data_gate` (the `drive*` cores factored, +signatures unchanged). Verification: `cargo test` (default), +`--all-features`, and 6 mixed per-feature builds all green; clippy ×2 +`-D warnings` clean; fmt clean; `cargo doc --no-deps` warning-free; +`taskgraph validate` clean (35 tasks). \ No newline at end of file diff --git a/tests/common/git_doubles.rs b/tests/common/git_doubles.rs index 658ba42..df99c8e 100644 --- a/tests/common/git_doubles.rs +++ b/tests/common/git_doubles.rs @@ -20,6 +20,7 @@ use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; /// The seam's ref-line shape re-exported for the rows (`RefLine` is /// the doubles' `list_refs` payload). @@ -113,6 +114,44 @@ pub struct GeneratedPack { pub boundary_haves: Vec, } +/// One per-repo serve gate (the `w4-git-exit-close` row's +/// serve-timing control): the double's `GitRefs::list_refs` for the +/// gated repo id pends on it until [`MemGate::release`] — the serving +/// task stalls at that call with its channel's reader parked, so the +/// row schedules the serve's later stages (the answer flood, the +/// serve tail) against a prepared wire state. Race-safe by +/// construction (the double-check pattern over the released flag). +pub struct MemGate { + released: AtomicBool, + opened: tokio::sync::Notify, +} + +impl MemGate { + fn new() -> Self { + Self { + released: AtomicBool::new(false), + opened: tokio::sync::Notify::new(), + } + } + + /// Releases the gate's waiting serve. + pub fn release(&self) { + self.released.store(true, Ordering::Release); + self.opened.notify_waiters(); + } + + async fn wait(&self) { + if self.released.load(Ordering::Acquire) { + return; + } + let notified = self.opened.notified(); + if self.released.load(Ordering::Acquire) { + return; + } + notified.await; + } +} + /// The in-memory quartet: one struct, four trait impls — the /// [`GitSeam`](alkssh::GitSeam) takes a separate `Arc` per backend /// trait and the rows share this state. @@ -124,6 +163,7 @@ pub struct MemGit { applied: Mutex>, prepared: Mutex>, generated: Mutex>, + gates: Mutex>>, } impl MemGit { @@ -158,6 +198,18 @@ impl MemGit { pub fn generated(&self) -> Vec { self.generated.lock().clone() } + + /// Installs a serve gate for one repo id — [`GitRefs::list_refs`] + /// calls for it pend until the returned gate's `release`. The + /// rows use the gate to prepare wire state (saturated queues, + /// blocked clients) for the serve's later stages. + pub fn install_gate(&self, repo_id: &str) -> Arc { + let gate = Arc::new(MemGate::new()); + self.gates + .lock() + .insert(repo_id.to_string(), Arc::clone(&gate)); + gate + } } /// The record for a resolved id — the RepoRecord the seam hands the @@ -188,6 +240,10 @@ impl GitRegistry for MemGit { #[async_trait] impl GitRefs for MemGit { async fn list_refs(&self, repo: &RepoRecord) -> Result, StorageError> { + let gate = self.gates.lock().get(&repo.repo_id).map(Arc::clone); + if let Some(gate) = gate { + gate.wait().await; + } Ok(self .refs .read() diff --git a/tests/common/harness.rs b/tests/common/harness.rs index 482a57c..ce90203 100644 --- a/tests/common/harness.rs +++ b/tests/common/harness.rs @@ -72,11 +72,13 @@ #![allow(dead_code)] use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use alkcall::core::auth::IdentityProvider; +use parking_lot::Mutex; use russh::client::{AuthResult, Handle, Msg}; use russh::keys::PrivateKeyWithHashAlg; -use russh::{Channel, ChannelMsg, ChannelOpenFailure, Disconnect, Error, MethodSet}; +use russh::{Channel, ChannelId, ChannelMsg, ChannelOpenFailure, Disconnect, Error, MethodSet}; use alkssh::{ AdapterError, HostKeyPolicy, Limits, PresentedHostKey, ServeConfig, SshAdapter, fingerprint, @@ -161,6 +163,70 @@ impl ForwardedAcceptor { } } +/// The client-direction data gate (`w4-git-exit-close`'s row +/// vehicle): armed on a channel id, the harness client's packet loop +/// pends inside its `data` hook for that channel's packets (the hook +/// is awaited inline in russh's client loop) — the server's outbound +/// writes for the channel back up through the transport's duplex pair +/// into the server's message queue until [`DataGate::release`]. The +/// held counter (`DataGate::held_packets`) lets the row observe that +/// the freeze is real before it releases. +pub struct DataGate { + target: Mutex>, + opened: Mutex>>, + release: Mutex>>, + held: AtomicUsize, +} + +impl DataGate { + /// Builds the inert gate — arm it with [`DataGate::arm`] once the + /// row has the channel id to gate. + pub fn new() -> Self { + Self { + target: Mutex::new(None), + opened: Mutex::new(None), + release: Mutex::new(None), + held: AtomicUsize::new(0), + } + } + + /// Arms the gate on one channel id: the next client-direction + /// data packet for that channel pends the client's packet loop. + pub fn arm(&self, channel: ChannelId) { + let (opened, release) = tokio::sync::oneshot::channel(); + *self.target.lock() = Some(channel); + *self.opened.lock() = Some(opened); + *self.release.lock() = Some(release); + } + + /// How many data packets have hit the armed gate so far — the + /// row's freeze signal (count `>= 1` means the client's packet + /// loop is pended inside the hook). + pub fn held_packets(&self) -> usize { + self.held.load(Ordering::Acquire) + } + + /// Releases the gate — the client's packet loop resumes and the + /// backed-up server queues drain. + pub fn release(&self) { + *self.target.lock() = None; + if let Some(opened) = self.opened.lock().take() { + let _ = opened.send(()); + } + } + + async fn block(&self, channel: ChannelId) { + if *self.target.lock() != Some(channel) { + return; + } + self.held.fetch_add(1, Ordering::Release); + let release = self.release.lock().take(); + if let Some(release) = release { + let _ = release.await; + } + } +} + /// The harness's russh client handler — the [`MemHostKeyPolicy`] /// decision double bridged into russh's `check_server_key` hook, with /// the decision input exactly as the wave-4 dial wiring will shape it @@ -171,6 +237,7 @@ impl ForwardedAcceptor { pub struct HarnessClientHandler { policy: Arc, acceptor: Option>, + data_gate: Option>, } impl HarnessClientHandler { @@ -183,6 +250,7 @@ impl HarnessClientHandler { Self { policy, acceptor: None, + data_gate: None, } } @@ -191,6 +259,17 @@ impl HarnessClientHandler { Self { policy: self.policy, acceptor: Some(acceptor), + data_gate: self.data_gate, + } + } + + /// Installs the client-direction data gate (`w4-git-exit-close`'s + /// row vehicle). Inert until the gate is armed on a channel id. + pub fn with_data_gate(self, gate: Arc) -> Self { + Self { + policy: self.policy, + acceptor: self.acceptor, + data_gate: Some(gate), } } } @@ -198,6 +277,24 @@ impl HarnessClientHandler { impl russh::client::Handler for HarnessClientHandler { type Error = Error; + /// The data gate's hook: when the gate is armed on this channel id + /// and not yet released, the client's packet loop pends inside it + /// (the hook is awaited inline) — the server's outbound writes + /// back up through the transport into its message queue until the + /// row releases. Inert otherwise; also the block counter the row + /// polls to know the freeze is real. + async fn data( + &mut self, + channel: ChannelId, + _data: &[u8], + _session: &mut russh::client::Session, + ) -> Result<(), Self::Error> { + if let Some(gate) = self.data_gate.as_ref() { + gate.block(channel).await; + } + Ok(()) + } + /// The policy decision, verbatim: the presented key's fingerprint /// is computed on the crate's single fingerprint path, the /// endpoint labels are the loopback constants, and the policy's @@ -358,6 +455,29 @@ pub async fn drive_with_acceptor( host_key_policy: Arc, limits: Limits, acceptor: Option>, +) -> Result { + mount_and_connect(adapter, host_key_policy, limits, acceptor, None).await +} + +/// [`drive`] with the client-direction data gate installed — the +/// `w4-git-exit-close` row's vehicle. +pub async fn drive_with_data_gate( + adapter: SshAdapter, + host_key_policy: Arc, + limits: Limits, + gate: Option>, +) -> Result { + mount_and_connect(adapter, host_key_policy, limits, None, gate).await +} + +/// `[drive*]`'s shared core: mount + connect (the doc decisions at +/// [`drive`] cover both; the optional hooks ride the handler). +async fn mount_and_connect( + adapter: SshAdapter, + host_key_policy: Arc, + limits: Limits, + acceptor: Option>, + data_gate: Option>, ) -> Result { let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY); let server_task = tokio::spawn(async move { adapter.handle(server_io).await }); @@ -366,6 +486,9 @@ pub async fn drive_with_acceptor( if let Some(acceptor) = acceptor { handler = handler.with_forwarded_acceptor(acceptor); } + if let Some(gate) = data_gate { + handler = handler.with_data_gate(gate); + } let client = russh::client::connect_stream(client_config(limits), client_io, handler).await?; Ok(LoopbackSession { diff --git a/tests/git_mapping.rs b/tests/git_mapping.rs index 8921d0a..67b34c9 100644 --- a/tests/git_mapping.rs +++ b/tests/git_mapping.rs @@ -23,8 +23,8 @@ mod serving { use super::common::git_doubles as gd; use super::common::harness::{ - LoopbackSession, authenticate_publickey, await_close, drive, next_message, - open_session_channel, request_exec, request_subsystem, + DataGate, LoopbackSession, authenticate_publickey, await_close, drive, + drive_with_data_gate, next_message, open_session_channel, request_exec, request_subsystem, serve_config as harness_serve_config, }; use super::common::{ @@ -686,6 +686,140 @@ mod serving { session.unmount().await; } + /// - *Contract stamp:* the status-before-close hardening + /// (`w4-git-exit-close` — the wave-3 gate's booked finding 1, + /// the saturated-`Sender` tail made real on the wire): the + /// serve's *answer volume* exceeds the whole outbound pipeline + /// (the harness client's packet loop is frozen on the armed + /// data gate → the transport duplex `DUPLEX_CAPACITY` fills → + /// the server's flush pends → russh's bounded message queue — + /// `event_buffer_size`, default 10, pinned below — fills → the + /// serve's remaining answer writes pend). The gate's release + /// then drives the serve tail through the just-unwinding + /// contention: the backlogged answer data drains **through the + /// same queue the report sends ride**, and the exit-status + /// still lands **before** the close — the split-halves + /// construction's property (the report is the only + /// close-contracting sender; sequential awaited sends on the + /// one handle), pinning every wire row of this class against + /// scheduler luck. + #[tokio::test] + async fn backlogged_seam_tail_still_reports_exit_status_before_close() { + assert_eq!( + russh::server::Config::default().event_buffer_size, + 10, + "the row's construction pins russh's default message-queue capacity \ + (the saturated-tail scenario's depth) — a russh default change here \ + re-reads this row's freeze-settle yields, not its assertions" + ); + + let doubles = Arc::new(gd::MemGit::new()); + let ref_oid = gd::oid(0xbb); + let name_padding = "x".repeat(1200); + let flooded_refs: Vec = (0..300) + .map(|n| gd::RefLine { + name: format!("refs/heads/backlog-{n:03}-{name_padding}"), + oid: Some(ref_oid), + peeled: None, + symref_target: None, + }) + .collect(); + doubles.install(gd::RepoSetup::private_granted( + "owner/backlog", + "git-rider", + flooded_refs.clone(), + )); + let serve_gate = doubles.install_gate("owner/backlog"); + + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture_key("alpha").fingerprint, + ident("git-rider", &["git"]), + ); + let mut serve = harness_serve_config(provider, &[fixture_key("bravo")], Limits::default()); + serve.git_seam = Some(gd::git_seam(&doubles)); + let adapter = SshAdapter::new(serve) + .await + .expect("the backlogged tail's adapter constructs"); + let policy = Arc::new(MemHostKeyPolicy::with_default(true)); + let gate = Arc::new(DataGate::new()); + let mut session = drive_with_data_gate( + adapter, + Arc::clone(&policy), + Limits::default(), + Some(Arc::clone(&gate)), + ) + .await + .expect("the gated loopback mounts"); + let result = + authenticate_publickey(&mut session.client, "git-rider", fixture_key("alpha")).await; + assert!(result.success(), "the backlogged tail's row authenticates"); + + let mut channel = route_git(&session, "git-upload-pack '/owner/backlog'").await; + let mut framer = PktFramer::new(); + loop { + match next_packet(&mut channel, &mut framer, "backlog advertisement").await { + Packet::Flush => break, + Packet::Payload(_) => {} + } + } + + let mut request = pkt(b"command=ls-refs\n"); + request.extend_from_slice(DELIM); + request.extend_from_slice(pkt(b"ref-prefix refs/heads/\n").as_slice()); + request.extend_from_slice(FLUSH); + channel + .data_bytes(request) + .await + .expect("the ls-refs request rides"); + channel + .data_bytes(FLUSH) + .await + .expect("the trailing flush rides"); + + gate.arm(channel.id()); + serve_gate.release(); + + while gate.held_packets() < 1 { + tokio::task::yield_now().await; + } + for _ in 0..256 { + tokio::task::yield_now().await; + } + + gate.release(); + + let mut seen = 0; + loop { + match next_packet(&mut channel, &mut framer, "the backlogged answer").await { + Packet::Flush => break, + Packet::Payload(line) => { + let expected = format!( + "{} refs/heads/backlog-{seen:03}-{name_padding}", + hex(ref_oid) + ); + assert_eq!( + String::from_utf8(line).unwrap().trim(), + expected, + "the backlogged answer's line {seen} rides in order" + ); + seen += 1; + } + } + } + assert_eq!( + seen, + flooded_refs.len(), + "the whole 300-line answer drained through the contended queue" + ); + expect_exit_status(&mut channel, &mut framer, 0, "the backlogged tail").await; + assert!( + await_close(&mut channel).await, + "exit-status(0) lands before the close through the contended tail" + ); + session.unmount().await; + } + // ------------------------------------------------------------------ // The store-and-take handoff, observed: two sequential services on // two channels of one session — the store is the session task's,