crates/gunnar-ssh/src/idle.rs
//! **A transfer that moves no bytes gives its slot back** (GUN-SSH-SLOT-HANG,
//! 2026-11-01 on git.gunnar.rs).
//!
//! Four parallel clones from one host sat in `git-lfs-transfer` with 0 CPU and held
//! the key's 32 transfer slots for ~35 minutes: every new clone of that key waited its
//! keyed 30 s and was refused. Nothing timed the stuck channels out. `inactivity_timeout`
//! is the CONNECTION's and russh resets it on every packet, keepalive replies included,
//! so a live client that never sends a byte of git or LFS protocol is never "inactive";
//! the keepalive itself only catches a dead link.
//!
//! [`Idle`] wraps the transfer's byte stream. Every byte read from or written to the
//! client is activity; when the transfer WAITS on the client (a read, or a write parked
//! on the SSH window) and nothing has moved for the timeout, the wait fails with
//! [`std::io::ErrorKind::TimedOut`] and the [`IDLE_CLOSED`] words. The transfer ends, the
//! channel closes, and its admission slot is free again. Work the server does without
//! waiting on the client (building a pack) is never timed: no wait is pending.
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::time::{Instant, Sleep};
/// What a transfer closed for idleness says (the log, the client's stderr).
pub const IDLE_CLOSED: &str = "no bytes moved on this git transfer";
/// Whether an [`Idle`] stream ended its transfer for idleness (set once).
#[derive(Debug, Clone, Default)]
pub struct IdleFlag(Arc<AtomicBool>);
impl IdleFlag {
/// Did the stream time out?
pub fn timed_out(&self) -> bool {
self.0.load(Ordering::SeqCst)
}
}
/// The transfer's stream, timed out after `timeout` with no bytes either way while it
/// waits on the client.
pub struct Idle<S> {
inner: S,
/// `None`: never timed out (the switch is off).
timeout: Option<Duration>,
last: Instant,
timer: Option<Pin<Box<Sleep>>>,
flag: IdleFlag,
}
impl<S> Idle<S> {
/// Wrap `inner`; `flag` says afterwards whether it timed out.
pub fn new(inner: S, timeout: Duration) -> (Self, IdleFlag) {
Self::maybe(inner, Some(timeout))
}
/// [`Idle::new`], or a plain pass-through when `timeout` is `None`.
pub fn maybe(inner: S, timeout: Option<Duration>) -> (Self, IdleFlag) {
let flag = IdleFlag::default();
let now = Instant::now();
(
Idle {
inner,
timeout,
last: now,
timer: timeout.map(|t| Box::pin(tokio::time::sleep_until(now + t))),
flag: flag.clone(),
},
flag,
)
}
/// Bytes moved: the transfer is alive.
fn touch(&mut self) {
self.last = Instant::now();
}
fn error(&self) -> io::Error {
let secs = self.timeout.map(|t| t.as_secs()).unwrap_or_default();
io::Error::new(
io::ErrorKind::TimedOut,
format!("{IDLE_CLOSED} for {secs} s while it waited on the client; it is closed and its transfer slot is free again"),
)
}
/// While the transfer waits on the client: `Ready` with the error once nothing has
/// moved for the timeout, `Pending` (the timer's waker registered) before that.
fn expired(&mut self, cx: &mut Context<'_>) -> Poll<io::Error> {
let Some(timeout) = self.timeout else {
return Poll::Pending;
};
let last = self.last;
let Some(timer) = self.timer.as_mut() else {
return Poll::Pending;
};
loop {
if timer.as_mut().poll(cx).is_pending() {
return Poll::Pending;
}
let deadline = last + timeout;
if Instant::now() >= deadline {
break;
}
// Activity since the timer was set: wait out the rest from the last byte.
timer.as_mut().reset(deadline);
}
self.flag.0.store(true, Ordering::SeqCst);
Poll::Ready(self.error())
}
}
impl<S> std::fmt::Debug for Idle<S> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Idle")
.field("timeout", &self.timeout)
.finish_non_exhaustive()
}
}
impl<S: AsyncRead + Unpin> AsyncRead for Idle<S> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
let me = self.get_mut();
if me.flag.timed_out() {
return Poll::Ready(Err(me.error()));
}
let before = buf.filled().len();
match Pin::new(&mut me.inner).poll_read(cx, buf) {
Poll::Ready(Ok(())) => {
if buf.filled().len() > before {
me.touch();
}
Poll::Ready(Ok(()))
}
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => me.expired(cx).map(Err),
}
}
}
impl<S: AsyncWrite + Unpin> AsyncWrite for Idle<S> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let me = self.get_mut();
if me.flag.timed_out() {
return Poll::Ready(Err(me.error()));
}
match Pin::new(&mut me.inner).poll_write(cx, buf) {
Poll::Ready(Ok(n)) => {
if n > 0 {
me.touch();
}
Poll::Ready(Ok(n))
}
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
// A write parked on the SSH window is waiting on the client too.
Poll::Pending => me.expired(cx).map(Err),
}
}
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::Pending => me.expired(cx).map(Err),
ready => ready,
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
const T: Duration = Duration::from_millis(400);
/// ★ A client that sends nothing: the read waiting on it fails after the timeout,
/// by name, and the flag says so.
#[tokio::test]
async fn a_read_that_waits_past_the_timeout_fails_by_name() {
let (_client, server) = tokio::io::duplex(64);
let (mut idle, flag) = Idle::new(server, T);
let started = std::time::Instant::now();
let mut buf = [0u8; 8];
let err = tokio::time::timeout(Duration::from_secs(5), idle.read(&mut buf))
.await
.expect("the idle read ends")
.expect_err("an idle read times out");
assert_eq!(err.kind(), io::ErrorKind::TimedOut);
assert!(err.to_string().contains(IDLE_CLOSED), "{err}");
assert!(flag.timed_out());
let waited = started.elapsed();
assert!(waited >= T && waited < 4 * T, "{waited:?}");
}
/// ★ Bytes either way are activity: a client that sends a byte every T/2 keeps the
/// timeout from firing, and the server's own write counts too.
#[tokio::test]
async fn bytes_either_way_keep_the_transfer_alive() {
let (mut client, server) = tokio::io::duplex(64);
let (mut idle, flag) = Idle::new(server, T);
let feeder = tokio::spawn(async move {
for _ in 0..4 {
tokio::time::sleep(T / 2).await;
client.write_all(b"x").await.unwrap();
}
client
});
let mut buf = [0u8; 1];
for _ in 0..4 {
idle.read_exact(&mut buf)
.await
.expect("a live client is not idle");
}
let mut client = feeder.await.unwrap();
tokio::time::sleep(T * 3 / 4).await;
idle.write_all(b"y").await.expect("the server writes");
let mut got = [0u8; 1];
client.read_exact(&mut got).await.unwrap();
// 3T/4 + a little since the client's last byte, ~0 since the server's: a read
// waiting T/2 more is still not idle.
let read = tokio::time::timeout(T / 2, idle.read(&mut buf)).await;
assert!(
read.is_err(),
"the server's own write was activity: {read:?}"
);
assert!(!flag.timed_out());
}
}