gunnar_ssh/
idle.rs

1//! **A transfer that moves no bytes gives its slot back** (GUN-SSH-SLOT-HANG,
2//! 2026-11-01 on git.gunnar.rs).
3//!
4//! Four parallel clones from one host sat in `git-lfs-transfer` with 0 CPU and held
5//! the key's 32 transfer slots for ~35 minutes: every new clone of that key waited its
6//! keyed 30 s and was refused. Nothing timed the stuck channels out. `inactivity_timeout`
7//! is the CONNECTION's and russh resets it on every packet, keepalive replies included,
8//! so a live client that never sends a byte of git or LFS protocol is never "inactive";
9//! the keepalive itself only catches a dead link.
10//!
11//! [`Idle`] wraps the transfer's byte stream. Every byte read from or written to the
12//! client is activity; when the transfer WAITS on the client (a read, or a write parked
13//! on the SSH window) and nothing has moved for the timeout, the wait fails with
14//! [`std::io::ErrorKind::TimedOut`] and the [`IDLE_CLOSED`] words. The transfer ends, the
15//! channel closes, and its admission slot is free again. Work the server does without
16//! waiting on the client (building a pack) is never timed: no wait is pending.
17
18use std::future::Future;
19use std::io;
20use std::pin::Pin;
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::sync::Arc;
23use std::task::{Context, Poll};
24use std::time::Duration;
25
26use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
27use tokio::time::{Instant, Sleep};
28
29/// What a transfer closed for idleness says (the log, the client's stderr).
30pub const IDLE_CLOSED: &str = "no bytes moved on this git transfer";
31
32/// Whether an [`Idle`] stream ended its transfer for idleness (set once).
33#[derive(Debug, Clone, Default)]
34pub struct IdleFlag(Arc<AtomicBool>);
35
36impl IdleFlag {
37    /// Did the stream time out?
38    pub fn timed_out(&self) -> bool {
39        self.0.load(Ordering::SeqCst)
40    }
41}
42
43/// The transfer's stream, timed out after `timeout` with no bytes either way while it
44/// waits on the client.
45pub struct Idle<S> {
46    inner: S,
47    /// `None`: never timed out (the switch is off).
48    timeout: Option<Duration>,
49    last: Instant,
50    timer: Option<Pin<Box<Sleep>>>,
51    flag: IdleFlag,
52}
53
54impl<S> Idle<S> {
55    /// Wrap `inner`; `flag` says afterwards whether it timed out.
56    pub fn new(inner: S, timeout: Duration) -> (Self, IdleFlag) {
57        Self::maybe(inner, Some(timeout))
58    }
59
60    /// [`Idle::new`], or a plain pass-through when `timeout` is `None`.
61    pub fn maybe(inner: S, timeout: Option<Duration>) -> (Self, IdleFlag) {
62        let flag = IdleFlag::default();
63        let now = Instant::now();
64        (
65            Idle {
66                inner,
67                timeout,
68                last: now,
69                timer: timeout.map(|t| Box::pin(tokio::time::sleep_until(now + t))),
70                flag: flag.clone(),
71            },
72            flag,
73        )
74    }
75
76    /// Bytes moved: the transfer is alive.
77    fn touch(&mut self) {
78        self.last = Instant::now();
79    }
80
81    fn error(&self) -> io::Error {
82        let secs = self.timeout.map(|t| t.as_secs()).unwrap_or_default();
83        io::Error::new(
84            io::ErrorKind::TimedOut,
85            format!("{IDLE_CLOSED} for {secs} s while it waited on the client; it is closed and its transfer slot is free again"),
86        )
87    }
88
89    /// While the transfer waits on the client: `Ready` with the error once nothing has
90    /// moved for the timeout, `Pending` (the timer's waker registered) before that.
91    fn expired(&mut self, cx: &mut Context<'_>) -> Poll<io::Error> {
92        let Some(timeout) = self.timeout else {
93            return Poll::Pending;
94        };
95        let last = self.last;
96        let Some(timer) = self.timer.as_mut() else {
97            return Poll::Pending;
98        };
99        loop {
100            if timer.as_mut().poll(cx).is_pending() {
101                return Poll::Pending;
102            }
103            let deadline = last + timeout;
104            if Instant::now() >= deadline {
105                break;
106            }
107            // Activity since the timer was set: wait out the rest from the last byte.
108            timer.as_mut().reset(deadline);
109        }
110        self.flag.0.store(true, Ordering::SeqCst);
111        Poll::Ready(self.error())
112    }
113}
114
115impl<S> std::fmt::Debug for Idle<S> {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        f.debug_struct("Idle")
118            .field("timeout", &self.timeout)
119            .finish_non_exhaustive()
120    }
121}
122
123impl<S: AsyncRead + Unpin> AsyncRead for Idle<S> {
124    fn poll_read(
125        self: Pin<&mut Self>,
126        cx: &mut Context<'_>,
127        buf: &mut ReadBuf<'_>,
128    ) -> Poll<io::Result<()>> {
129        let me = self.get_mut();
130        if me.flag.timed_out() {
131            return Poll::Ready(Err(me.error()));
132        }
133        let before = buf.filled().len();
134        match Pin::new(&mut me.inner).poll_read(cx, buf) {
135            Poll::Ready(Ok(())) => {
136                if buf.filled().len() > before {
137                    me.touch();
138                }
139                Poll::Ready(Ok(()))
140            }
141            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
142            Poll::Pending => me.expired(cx).map(Err),
143        }
144    }
145}
146
147impl<S: AsyncWrite + Unpin> AsyncWrite for Idle<S> {
148    fn poll_write(
149        self: Pin<&mut Self>,
150        cx: &mut Context<'_>,
151        buf: &[u8],
152    ) -> Poll<io::Result<usize>> {
153        let me = self.get_mut();
154        if me.flag.timed_out() {
155            return Poll::Ready(Err(me.error()));
156        }
157        match Pin::new(&mut me.inner).poll_write(cx, buf) {
158            Poll::Ready(Ok(n)) => {
159                if n > 0 {
160                    me.touch();
161                }
162                Poll::Ready(Ok(n))
163            }
164            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
165            // A write parked on the SSH window is waiting on the client too.
166            Poll::Pending => me.expired(cx).map(Err),
167        }
168    }
169
170    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
171        let me = self.get_mut();
172        match Pin::new(&mut me.inner).poll_flush(cx) {
173            Poll::Pending => me.expired(cx).map(Err),
174            ready => ready,
175        }
176    }
177
178    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
179        Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
180    }
181}
182
183#[cfg(test)]
184mod tests {
185    use super::*;
186    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
187
188    const T: Duration = Duration::from_millis(400);
189
190    /// ★ A client that sends nothing: the read waiting on it fails after the timeout,
191    /// by name, and the flag says so.
192    #[tokio::test]
193    async fn a_read_that_waits_past_the_timeout_fails_by_name() {
194        let (_client, server) = tokio::io::duplex(64);
195        let (mut idle, flag) = Idle::new(server, T);
196        let started = std::time::Instant::now();
197        let mut buf = [0u8; 8];
198        let err = tokio::time::timeout(Duration::from_secs(5), idle.read(&mut buf))
199            .await
200            .expect("the idle read ends")
201            .expect_err("an idle read times out");
202        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
203        assert!(err.to_string().contains(IDLE_CLOSED), "{err}");
204        assert!(flag.timed_out());
205        let waited = started.elapsed();
206        assert!(waited >= T && waited < 4 * T, "{waited:?}");
207    }
208
209    /// ★ Bytes either way are activity: a client that sends a byte every T/2 keeps the
210    /// timeout from firing, and the server's own write counts too.
211    #[tokio::test]
212    async fn bytes_either_way_keep_the_transfer_alive() {
213        let (mut client, server) = tokio::io::duplex(64);
214        let (mut idle, flag) = Idle::new(server, T);
215        let feeder = tokio::spawn(async move {
216            for _ in 0..4 {
217                tokio::time::sleep(T / 2).await;
218                client.write_all(b"x").await.unwrap();
219            }
220            client
221        });
222        let mut buf = [0u8; 1];
223        for _ in 0..4 {
224            idle.read_exact(&mut buf)
225                .await
226                .expect("a live client is not idle");
227        }
228        let mut client = feeder.await.unwrap();
229        tokio::time::sleep(T * 3 / 4).await;
230        idle.write_all(b"y").await.expect("the server writes");
231        let mut got = [0u8; 1];
232        client.read_exact(&mut got).await.unwrap();
233        // 3T/4 + a little since the client's last byte, ~0 since the server's: a read
234        // waiting T/2 more is still not idle.
235        let read = tokio::time::timeout(T / 2, idle.read(&mut buf)).await;
236        assert!(
237            read.is_err(),
238            "the server's own write was activity: {read:?}"
239        );
240        assert!(!flag.timed_out());
241    }
242}
gunnar_ssh/idle.rs

gunnar_ssh/
idle.rs
