gunnar-ssh 4.0.0

//! **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());
    }
}