gunnar-ssh 3.0.0

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