crates/gunnar-ssh/src/channel.rs
//! An **owned** byte stream over a russh channel that does *not* close the
//! channel when it is dropped.
//!
//! # The bug this module exists to make impossible
//!
//! russh's [`Channel::into_stream`](russh::Channel::into_stream) wraps the
//! channel in `ChannelCloseOnDrop`, whose `Drop` does
//!
//! ```text
//! tokio::spawn(async move { sender.send(ChannelMsg::Close).await })
//! ```
//!
//! (`russh-0.61.2/src/channels/io/mod.rs`). The server hands that stream to the
//! transport, so the `Close` is queued **from a task nobody awaits** the moment
//! the transfer finishes, racing the `exit-status` the server sends next.
//! OpenSSH frees a channel on `Close` and discards anything that arrives after
//! it, so the race decided whether a client saw an exit status at all.
//!
//! Measured 2026-08-03, before this module existed: a raw
//! `ssh … git-upload-pack 'team/project.git'` returned a complete 244-byte
//! advertisement and exit status **255**, and `git push` printed
//! `* [new branch] HEAD -> main` followed by `error: failed to push some refs`
//! for a push that had fully succeeded server-side. `gunnar-cli` carried a
//! 250 ms `linger` timer to invert the race; it is deleted.
//!
//! # The fix, by construction (one implementation)
//!
//! [`Channel::split`](russh::Channel::split) hands back a
//! [`ChannelReadHalf`] and a [`ChannelWriteHalf`], **neither of which closes the
//! channel on drop**. The write half stays with the server, which sends
//! `exit-status` and then `close` through *the same sender the data went
//! through*, so the ordering is a property of one FIFO queue rather than of a
//! sleep. The read half is moved in here and turned into an owned
//! [`AsyncRead`], because russh's own `make_reader` borrows the half and the
//! transport needs a `'static` stream (it runs the blocking git protocol on
//! `spawn_blocking`, which will not accept a borrow).
//!
//! There is no timing dependency left anywhere on this path.
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use russh::{ChannelMsg, ChannelReadHalf};
use tokio::io::{AsyncRead, ReadBuf};
/// The future that owns the read half while it waits for the next message.
type Pending = Pin<Box<dyn Future<Output = (ChannelReadHalf, Option<ChannelMsg>)> + Send>>;
/// Where the reader is between two `poll_read` calls.
enum State {
/// Nothing is in flight; the half is parked here.
Idle(ChannelReadHalf),
/// A `wait()` is in flight, and it owns the half until it resolves.
Waiting(Pending),
/// The peer sent EOF (or the channel went away). Every further read is EOF.
Ended,
}
/// The data band of an SSH channel, as an owned [`AsyncRead`].
///
/// Dropping this drops a [`ChannelReadHalf`], which sends nothing at all. That
/// is the whole point. See the module docs.
pub struct ChannelReader {
state: State,
/// A `ChannelMsg::Data` that did not fit in the last `poll_read`, and how
/// much of it has already been handed out.
partial: Option<(ChannelMsg, usize)>,
}
impl ChannelReader {
/// Take ownership of a channel's read half.
pub fn new(half: ChannelReadHalf) -> Self {
ChannelReader {
state: State::Idle(half),
partial: None,
}
}
}
impl std::fmt::Debug for ChannelReader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChannelReader").finish_non_exhaustive()
}
}
/// The bytes of a `ChannelMsg::Data`, and nothing for anything else.
///
/// [`next_data`] only ever parks a `Data`, so the fallback arm is unreachable
/// in practice; it is a `&[]` rather than a panic because `ChannelMsg` is
/// `#[non_exhaustive]` and a future russh could add a variant here.
fn payload(message: &ChannelMsg) -> &[u8] {
match message {
ChannelMsg::Data { data } => data,
_ => &[],
}
}
/// Wait for the next `Data` message, handing the half back with it.
///
/// Anything that is not `Data` is skipped **inside** the future rather than by
/// returning `Poll::Pending` after a self-wake, so a chatty peer cannot spin
/// the reactor. `Eof`, `Close` and a dead channel all mean end-of-stream.
fn next_data(mut half: ChannelReadHalf) -> Pending {
Box::pin(async move {
loop {
match half.wait().await {
Some(ChannelMsg::Data { data }) => {
return (half, Some(ChannelMsg::Data { data }));
}
Some(ChannelMsg::Eof) | Some(ChannelMsg::Close) | None => return (half, None),
Some(_) => continue,
}
}
})
}
impl AsyncRead for ChannelReader {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let me = self.get_mut();
loop {
if let Some((message, taken)) = me.partial.as_mut() {
let data = payload(message);
// Clamped by `min`, so neither index can be out of range.
let readable = buf.remaining().min(data.len().saturating_sub(*taken));
buf.put_slice(&data[*taken..*taken + readable]);
*taken += readable;
if *taken >= data.len() {
me.partial = None;
}
return Poll::Ready(Ok(()));
}
match std::mem::replace(&mut me.state, State::Ended) {
// Reading past EOF is not an error, it is EOF again: a zero-fill
// return is exactly how `AsyncRead` says so.
State::Ended => return Poll::Ready(Ok(())),
State::Idle(half) => me.state = State::Waiting(next_data(half)),
State::Waiting(mut waiting) => match waiting.as_mut().poll(cx) {
Poll::Pending => {
me.state = State::Waiting(waiting);
return Poll::Pending;
}
Poll::Ready((half, Some(message))) => {
me.state = State::Idle(half);
me.partial = Some((message, 0));
}
Poll::Ready((_, None)) => {
me.state = State::Ended;
return Poll::Ready(Ok(()));
}
},
}
}
}
}
// ── the writer that hears the session end ───────────────────────────────────
/// **The session is over**, as a value every transfer on it can watch.
///
/// The [`crate::server`] connection handler owns the sender and sets it from
/// `Drop` — and russh drops the handler on EVERY way a session ends: the client
/// hung up, its keepalives went unanswered, the inactivity bound, a protocol
/// error, the server's own shutdown.
pub type SessionOver = tokio::sync::watch::Receiver<bool>;
/// Resolve once `over` says the session has ended (or its sender is gone,
/// which is the same thing said by `Drop`).
pub async fn session_over(mut over: SessionOver) {
let _ = over.wait_for(|ended| *ended).await;
}
/// The error a transfer gets when the client's session ended under it.
pub fn session_ended_error() -> io::Error {
io::Error::new(
io::ErrorKind::ConnectionAborted,
"the client's ssh session ended mid-transfer (it hung up, or its link went silent \
and its keepalives went unanswered)",
)
}
/// How long a transfer whose session has ended is given to end by itself.
///
/// The session's end also ends the channel, so a transfer that was only
/// waiting for the client's next request reads EOF at once and ends cleanly;
/// one still working (main building a pack) is cut when this runs out.
pub const SESSION_END_GRACE: std::time::Duration = std::time::Duration::from_secs(2);
/// The grace when the transfer's fan has others queued behind it: a worker a
/// hung-up client holds is a worker a waiting client does not get.
pub const CROWDED_SESSION_END_GRACE: std::time::Duration = std::time::Duration::from_millis(200);
/// **The grace a transfer whose session ended is given**: `calm`, or
/// `crowded` when `queued` operations wait for its fan.
pub fn grace_for(
calm: std::time::Duration,
crowded: std::time::Duration,
queued: usize,
) -> std::time::Duration {
if queued > 0 {
crowded.min(calm)
} else {
calm
}
}
/// **The word a transfer's git work hears when its client is gone.**
///
/// Set by [`serve_until_session_ends`] when the grace runs out, and by a
/// transport whose serve future is dropped. The git work checks it at its safe
/// points — every read from and write to the client, and before it opens the
/// store — and ends there. It is never checked between the ref updates of a
/// receive-pack transaction: those land together or not at all.
///
/// Cheap to clone; every clone is the same flag.
#[derive(Clone, Default)]
pub struct Cancel(std::sync::Arc<CancelInner>);
#[derive(Default)]
struct CancelInner {
set: std::sync::atomic::AtomicBool,
wake: tokio::sync::Notify,
}
impl Cancel {
/// A flag nobody has set.
pub fn new() -> Self {
Self::default()
}
/// Tell the transfer to stop at its next safe point.
pub fn cancel(&self) {
self.0.set.store(true, std::sync::atomic::Ordering::SeqCst);
self.0.wake.notify_waiters();
}
/// Has the transfer been told to stop?
pub fn is_cancelled(&self) -> bool {
self.0.set.load(std::sync::atomic::Ordering::SeqCst)
}
/// Resolves once the transfer has been told to stop.
pub async fn cancelled(&self) {
loop {
let notified = self.0.wake.notified();
if self.is_cancelled() {
return;
}
notified.await;
}
}
}
impl std::fmt::Debug for Cancel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_tuple("Cancel").field(&self.is_cancelled()).finish()
}
}
/// **A transfer, bound to its session**: what `serve` returns, unless the
/// session ends first and the transfer does not end by itself within
/// `grace()` (asked when the session ends; [`grace_for`] — the server gives
/// [`SESSION_END_GRACE`], or [`CROWDED_SESSION_END_GRACE`] when others wait
/// for the fan), in which case it ends with `ended()`.
///
/// The server's transfer task runs every `git-upload-pack` /
/// `git-receive-pack` through this, so a client that went silent while main
/// still builds its pack gives back its fan worker, admission slot and store
/// handle when its session ends (the UpCloud mockup's LD1/LD2, 2026-10-09).
///
/// The grace is what keeps a SERVED client honest: `ls-remote` reads the ref
/// list and hangs up, and its session's end used to win the race against
/// upload-pack's own clean EOF, so every such ls-remote was logged as a failed
/// transfer (test2 install receipt 2026-10-09: WARN spam, REFUSED
/// [log-verdict]). The protocol decides whether the client was served, not
/// which of the two wakes first.
///
/// # The cut waits for the work (stinky row 1, 2026-10-10)
///
/// When the grace runs out, `cancel` is set and `serve` is **awaited to its
/// end**, not dropped. Dropping it dropped only the `JoinHandle` of the
/// transport's blocking task: the git work went on (a pack being built, refs
/// being moved) while the fan worker running this was given back, so the fan
/// bounded nothing. The work ends at its next safe point once it reads the
/// flag ([`Cancel`]); the worker is held until it has.
pub async fn serve_until_session_ends<T, E>(
serve: impl Future<Output = Result<T, E>>,
over: SessionOver,
cancel: &Cancel,
grace: impl FnOnce() -> std::time::Duration,
ended: impl FnOnce() -> E,
) -> Result<T, E> {
let mut serve = std::pin::pin!(serve);
tokio::select! {
served = &mut serve => served,
() = session_over(over) => match tokio::time::timeout(grace(), &mut serve).await {
Ok(served) => served,
Err(_) => {
cancel.cancel();
tracing::debug!(target: "gunnar::ssh", event = "ssh.transfer.told_to_stop", "the session ended and the transfer did not end within the grace; it is told to stop and its worker waits for it");
match serve.await {
Ok(served) => Ok(served),
Err(_) => Err(ended()),
}
}
},
}
}
/// A channel writer that **ends with an error when the session does**.
///
/// # The bug this exists to make impossible (cov-13 finding 4, 2026-10-06)
///
/// russh's channel writer (`ChannelTx`, what `make_writer` returns) waits for
/// SSH window on a `Notify` that only a window-adjust from the client fires. A
/// session that ENDS fires nothing. So a clone or fetch streaming a pack to a
/// client whose link died — the window shut, the session then ended by the
/// keepalive — left the transfer parked forever: its fan worker, its admission
/// slot, its memory-ledger charge, its store handle, all held for good. One
/// dead client per leak, until the fan refuses everyone.
///
/// Every poll here also polls the session's end, so the parked write wakes and
/// fails with [`session_ended_error`] the moment the session is gone.
pub struct SessionBound<W> {
inner: W,
over: SessionOver,
/// The pending "session over" wait, kept across polls so its waker stays
/// registered while the inner write is parked.
ending: Option<Pin<Box<dyn Future<Output = ()> + Send>>>,
}
impl<W> SessionBound<W> {
/// Bind `inner` to the session `over` watches.
pub fn new(inner: W, over: SessionOver) -> Self {
SessionBound {
inner,
over,
ending: None,
}
}
/// `Ready` once the session has ended; registers the waker otherwise.
fn poll_over(&mut self, cx: &mut Context<'_>) -> Poll<()> {
if *self.over.borrow() {
return Poll::Ready(());
}
let over = self.over.clone();
let ending = self
.ending
.get_or_insert_with(|| Box::pin(session_over(over)));
ending.as_mut().poll(cx)
}
}
impl<W> std::fmt::Debug for SessionBound<W> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SessionBound").finish_non_exhaustive()
}
}
impl<W: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for SessionBound<W> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let me = self.get_mut();
match Pin::new(&mut me.inner).poll_write(cx, buf) {
Poll::Ready(r) => Poll::Ready(r),
Poll::Pending => match me.poll_over(cx) {
Poll::Ready(()) => Poll::Ready(Err(session_ended_error())),
Poll::Pending => Poll::Pending,
},
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let me = self.get_mut();
match Pin::new(&mut me.inner).poll_flush(cx) {
Poll::Ready(r) => Poll::Ready(r),
Poll::Pending => match me.poll_over(cx) {
Poll::Ready(()) => Poll::Ready(Err(session_ended_error())),
Poll::Pending => Poll::Pending,
},
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let me = self.get_mut();
match Pin::new(&mut me.inner).poll_shutdown(cx) {
Poll::Ready(r) => Poll::Ready(r),
Poll::Pending => match me.poll_over(cx) {
Poll::Ready(()) => Poll::Ready(Err(session_ended_error())),
Poll::Pending => Poll::Pending,
},
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
fn ended_session() -> (tokio::sync::watch::Sender<bool>, SessionOver) {
let (tx, rx) = tokio::sync::watch::channel(false);
tx.send_replace(true);
(tx, rx)
}
/// ★ **A client that hung up after it was served is no failed transfer**
/// (test2 install receipt 2026-10-09 18:35 UTC, REFUSED [log-verdict]):
/// `ls-remote` reads the ref list and hangs up. Its session ends while
/// upload-pack waits for a next command that never comes, and upload-pack
/// ends cleanly on that EOF a moment later. Every such ls-remote was logged
/// as `git.transfer.failed` + `git.exit_status.undelivered` at WARN, and the
/// kit's functest (7 repositories listed in 38 s) tripped the WARN-spam rule.
#[tokio::test]
async fn a_client_that_hung_up_after_being_served_is_no_failed_transfer() {
let (_tx, over) = ended_session();
let serve = async {
// upload-pack reading the EOF the ended session leaves behind
tokio::time::sleep(Duration::from_millis(50)).await;
Ok::<(), io::Error>(())
};
let served = serve_until_session_ends(
serve,
over,
&Cancel::new(),
|| SESSION_END_GRACE,
session_ended_error,
)
.await;
assert!(
served.is_ok(),
"a transfer that ended cleanly was called failed: {served:?}"
);
}
/// Stinky row 19 (2026-10-10): the 2 s grace was a constant, so every
/// client that hung up held a fan worker up to 2 s more — however many
/// waited behind it, which is what a connect-and-hang-up flood feeds on.
/// The grace is the caller's, and shorter when the fan is crowded.
#[test]
fn the_grace_is_shorter_when_others_wait_for_the_fan() {
let (calm, crowded) = (Duration::from_secs(2), Duration::from_millis(200));
assert_eq!(
grace_for(calm, crowded, 0),
calm,
"nobody waits: the whole grace"
);
assert_eq!(
grace_for(calm, crowded, 1),
crowded,
"one waits: the crowded grace"
);
assert_eq!(
grace_for(Duration::from_millis(100), crowded, 5),
Duration::from_millis(100),
"never longer than the calm grace"
);
}
#[tokio::test]
async fn a_transfer_is_cut_after_the_grace_it_is_given() {
let (_tx, over) = ended_session();
let cancel = Cancel::new();
let heard = cancel.clone();
let serve = async move {
heard.cancelled().await;
Err::<(), io::Error>(io::Error::new(
io::ErrorKind::ConnectionAborted,
"told to stop",
))
};
let started = std::time::Instant::now();
let served = serve_until_session_ends(
serve,
over,
&cancel,
|| Duration::from_millis(50),
session_ended_error,
)
.await;
assert!(served.is_err());
assert!(
started.elapsed() < Duration::from_secs(1),
"cut after {:?}, not the 50 ms it was given",
started.elapsed()
);
}
/// The bug the binding exists for stays fixed: a transfer that is still
/// working (main building the pack) when the session ends is told to stop,
/// and gives back what it holds, within a bound, once it has.
#[tokio::test]
async fn a_transfer_still_working_when_the_session_ends_is_cut() {
let (_tx, over) = ended_session();
let cancel = Cancel::new();
let heard = cancel.clone();
let serve = async move {
heard.cancelled().await;
Err::<(), io::Error>(io::Error::new(
io::ErrorKind::ConnectionAborted,
"told to stop",
))
};
let served = tokio::time::timeout(
Duration::from_secs(10),
serve_until_session_ends(
serve,
over,
&cancel,
|| SESSION_END_GRACE,
session_ended_error,
),
)
.await
.expect("a transfer outlived its session by 10 s");
let err =
served.expect_err("a transfer still working when its session ended was called served");
assert_eq!(err.kind(), io::ErrorKind::ConnectionAborted, "{err}");
}
/// ★ **A hung-up client cannot make blocking work exceed the fan** (stinky
/// row 1, 2026-10-10). The transport's git work runs on tokio's blocking
/// pool, and when the grace ran out only its `JoinHandle` was dropped: the
/// fan worker was released while the work (a pack being built, refs being
/// moved) went on, so the fan bounded nothing and a stream of clients that
/// hung up mid-clone could put up to 512 blocking threads to work. Here a
/// fan of ONE worker serves two transfers whose clients hung up while main
/// still works; at no moment may two pieces of that work run together.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_hung_up_client_cannot_make_blocking_work_exceed_the_fan() {
use crate::governor::OpClass;
use crate::pipeline::{Job, Pipeline};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
// Longer than the grace, so the work is still running when it runs out.
const WORK: Duration = Duration::from_secs(3);
let fan: Pipeline<Job> = Pipeline::start(OpClass::Clone, 1, 4, |job: Job| job());
let live = Arc::new(AtomicUsize::new(0));
let peak = Arc::new(AtomicUsize::new(0));
let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
for _ in 0..2 {
let (live, peak, done) = (Arc::clone(&live), Arc::clone(&peak), done_tx.clone());
let handle = tokio::runtime::Handle::current();
let (session, over) = ended_session();
fan.submit(Box::new(move || {
handle.block_on(async move {
let _session = session;
let serve = async move {
tokio::task::spawn_blocking(move || {
let now = live.fetch_add(1, Ordering::SeqCst) + 1;
peak.fetch_max(now, Ordering::SeqCst);
let until = std::time::Instant::now() + WORK;
while std::time::Instant::now() < until {
std::thread::sleep(Duration::from_millis(10));
}
live.fetch_sub(1, Ordering::SeqCst);
Ok::<(), io::Error>(())
})
.await
.map_err(io::Error::other)?
};
let _ = serve_until_session_ends(
serve,
over,
&Cancel::new(),
|| SESSION_END_GRACE,
session_ended_error,
)
.await;
});
let _ = done.send(());
}))
.expect("the fan takes the transfer");
}
for _ in 0..2 {
done_rx.recv().await.expect("a transfer finished");
}
// Whatever the fan let go of early is still working; let it end so the
// peak is the whole story.
while live.load(Ordering::SeqCst) > 0 {
tokio::time::sleep(Duration::from_millis(10)).await;
}
let peak = peak.load(Ordering::SeqCst);
assert!(
peak <= 1,
"a fan of one worker had {peak} transfers' blocking work running at once: the worker \
was released while the work of a client that hung up went on"
);
}
/// A transfer that ends while its session is alive is what it says.
#[tokio::test]
async fn a_transfer_with_its_session_alive_is_its_own_verdict() {
let (_tx, over) = tokio::sync::watch::channel(false);
let ok = serve_until_session_ends(
async { Ok::<u8, io::Error>(7) },
over.clone(),
&Cancel::new(),
|| SESSION_END_GRACE,
session_ended_error,
)
.await;
assert_eq!(ok.unwrap(), 7);
let err = serve_until_session_ends(
async { Err::<u8, io::Error>(io::Error::other("store")) },
over,
&Cancel::new(),
|| SESSION_END_GRACE,
session_ended_error,
)
.await;
assert_eq!(err.unwrap_err().to_string(), "store");
}
}