gunnar-ssh 3.0.0

//! **A transfer to a client that died must END**, and give back what it held.
//!
//! cov-13 finding 4 (UpCloud mockup, 2026-10-06): a silent WireGuard link hung
//! the kit's pushes forever. The CLIENT half is fixed in edda's nornir-git; this
//! is the SERVER half of the same defect. russh's channel writer parks on the
//! SSH window behind a `Notify` that only the client's window-adjust fires, and
//! a session that ends fires nothing. So a clone or fetch streaming a pack to a
//! client whose link died stayed parked for good, holding its fan worker, its
//! admission slot, its memory-ledger charge and its store handle. That is one
//! leak per dead client, until the fan refuses everyone.
//!
//! The client here dials through a loopback proxy and never reads, so the
//! server's window shuts while it streams. Then the proxy either hangs up or
//! goes SILENT (forwards nothing, keeps both sockets open, so TCP never times
//! out and only the server's keepalive can tell). Either way the transfer must
//! leave `transfers_in_flight` within the bound.

mod common;

use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, Instant};

use common::*;
use gunnar_ssh::seam::{DuplexStream, GitTransport, ServiceRequest, TransportError};
use gunnar_ssh::{Access, GunnarSsh, SshConfig};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::watch;

/// A transport that streams 256 MiB at the client: far past any SSH window, so
/// a client that stops reading leaves it parked on the window.
struct Flood;

impl GitTransport for Flood {
    fn serve<'a>(
        &'a self,
        _request: ServiceRequest,
        mut stream: Box<dyn DuplexStream>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move {
            let chunk = vec![0x5au8; 64 * 1024];
            for _ in 0..4096 {
                stream.write_all(&chunk).await?;
            }
            stream.shutdown().await?;
            Ok(())
        })
    }
}

/// A transport that is still BUILDING the pack when the client goes silent: a long
/// silence with not one byte written (the UpCloud mockup's LD1/LD2, 2026-10-09: main
/// spent 2 min 10 s on the reachability table of a 512 MiB entry on TCG, the client
/// was frozen before its first byte, and main never ended the transfer), then the flood.
struct Building;

impl GitTransport for Building {
    fn serve<'a>(
        &'a self,
        request: ServiceRequest,
        mut stream: Box<dyn DuplexStream>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move {
            // The build, with the safe point the real transport has: told to
            // stop (the client gone past the grace), it stops — and only then
            // is its fan worker given back (stinky row 1, 2026-10-10).
            tokio::select! {
                () = tokio::time::sleep(Duration::from_secs(120)) => {}
                () = request.cancel.cancelled() => {
                    return Err(TransportError::Io(std::io::Error::new(std::io::ErrorKind::ConnectionAborted, "told to stop mid-build")));
                }
            }
            let chunk = vec![0x5au8; 64 * 1024];
            for _ in 0..4096 {
                stream.write_all(&chunk).await?;
            }
            stream.shutdown().await?;
            Ok(())
        })
    }
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum Mode {
    Forward,
    Silent,
    Close,
}

/// A loopback proxy in front of `upstream`. All of its connections follow ONE
/// mode switch.
async fn proxy(upstream: SocketAddr) -> (u16, watch::Sender<Mode>) {
    let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
        .await
        .expect("bind proxy");
    let port = listener.local_addr().expect("addr").port();
    let (tx, rx) = watch::channel(Mode::Forward);
    tokio::spawn(async move {
        loop {
            let Ok((client, _)) = listener.accept().await else {
                continue;
            };
            let Ok(server) = tokio::net::TcpStream::connect(upstream).await else {
                continue;
            };
            let (cr, cw) = client.into_split();
            let (sr, sw) = server.into_split();
            tokio::spawn(pump(cr, sw, rx.clone()));
            tokio::spawn(pump(sr, cw, rx.clone()));
        }
    });
    (port, tx)
}

async fn pump(
    mut src: tokio::net::tcp::OwnedReadHalf,
    mut dst: tokio::net::tcp::OwnedWriteHalf,
    mut mode: watch::Receiver<Mode>,
) {
    let mut buf = vec![0u8; 64 * 1024];
    loop {
        let now = *mode.borrow_and_update();
        match now {
            Mode::Forward => {}
            Mode::Close => return,
            Mode::Silent => loop {
                if mode.changed().await.is_err() {
                    return;
                }
                let next = *mode.borrow_and_update();
                if next == Mode::Close {
                    return;
                }
            },
        }
        tokio::select! {
            r = src.read(&mut buf) => match r {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    if *mode.borrow() == Mode::Forward && dst.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            },
            _ = mode.changed() => {}
        }
    }
}

/// A client that keeps no timer of its own: the SERVER must notice.
struct Patient;
impl russh::client::Handler for Patient {
    type Error = russh::Error;
    async fn check_server_key(
        &mut self,
        _: &russh::keys::PublicKeyOrCertificate,
    ) -> Result<bool, Self::Error> {
        Ok(true)
    }
}

async fn wait_for(bound: Duration, mut f: impl FnMut() -> bool) -> bool {
    let until = Instant::now() + bound;
    while Instant::now() < until {
        if f() {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    f()
}

/// Start a flooding fetch through the proxy and leave it parked on the window.
/// Returns the server and the client handle; the caller then kills the link.
async fn parked_fetch(
    port: u16,
    key: &russh::keys::ssh_key::PrivateKey,
) -> russh::client::Handle<Patient> {
    let config = Arc::new(russh::client::Config::default());
    let mut session = russh::client::connect(config, ("127.0.0.1", port), Patient)
        .await
        .expect("connect");
    let ok = session
        .authenticate_publickey(
            "git",
            russh::keys::PrivateKeyWithHashAlg::new(Arc::new(key.clone()), None),
        )
        .await
        .expect("auth");
    assert!(ok.success(), "the key is registered");
    let channel = session.channel_open_session().await.expect("channel");
    channel
        .exec(true, "git-upload-pack 'nordisk/osm-katana.git'")
        .await
        .expect("exec");
    // The channel is never read: its buffer fills, the client's session stops
    // granting window, and the server's writer parks.
    std::mem::forget(channel);
    session
}

async fn a_dead_client_releases_its_transfer(kill: Mode, bound: Duration) {
    let key = mint_key();
    let mut config = SshConfig::ephemeral_loopback();
    // Short, so a test waits seconds: 1 s x 3 (russh's keepalive_max).
    config.keepalive_interval = Some(Duration::from_secs(1));
    let server = GunnarSsh::new(
        config,
        plane_with_key(&key, Access::Read).control,
        Arc::new(Flood),
    )
    .bind(mint_key())
    .await
    .expect("bind");
    let (port, mode) = proxy(server.local_addr()).await;

    let _client = parked_fetch(port, &key).await;
    assert!(
        wait_for(Duration::from_secs(10), || server
            .metrics()
            .transfers_in_flight
            == 1)
        .await,
        "the flood never started"
    );
    if kill == Mode::Close {
        // Let the window run dry and the writer park, then hang up — well
        // inside the 1 s x 3 keepalive, so it is the HANG-UP that must end it.
        tokio::time::sleep(Duration::from_millis(1000)).await;
        assert_eq!(
            server.metrics().transfers_in_flight,
            1,
            "parked, still in flight"
        );
    }

    let killed = Instant::now();
    mode.send_replace(kill);
    let released = wait_for(bound, || server.metrics().transfers_in_flight == 0).await;
    assert!(
        released,
        "{kill:?}: the transfer to a dead client was still in flight {:?} after the link died \
         (fan worker, admission slot and store handle held for good)",
        killed.elapsed()
    );
    eprintln!(
        "{kill:?}: transfer released {:?} after the link died",
        killed.elapsed()
    );
    mode.send_replace(Mode::Close);
}

/// The client's link hangs up while the pack streams.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_client_that_hangs_up_mid_clone_releases_its_transfer_at_once() {
    a_dead_client_releases_its_transfer(Mode::Close, Duration::from_millis(2500)).await;
}

/// The client's link goes silent while the pack streams: the server's own
/// keepalive (1 s x 3 here; 15 s x 3 shipped) must end it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_client_whose_link_goes_silent_mid_clone_releases_its_transfer() {
    // 1 s x 3 keepalive; 20 s (`ephemeral_loopback`'s inactivity) if the
    // server's own socket write is what blocks.
    a_dead_client_releases_its_transfer(Mode::Silent, Duration::from_secs(25)).await;
}

/// ★ **A client that goes silent while main is still BUILDING the pack** — before a single
/// byte was written, so no write is parked on the window for `SessionBound` to wake — must
/// be cut by the server's keepalive as well (the UpCloud mockup's LD1/LD2, 2026-10-09). The
/// transfer is released within the keepalive bound, not when the build ends two minutes on.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_client_that_goes_silent_while_the_pack_is_built_releases_its_transfer() {
    let key = mint_key();
    let mut config = SshConfig::ephemeral_loopback();
    config.keepalive_interval = Some(Duration::from_secs(1));
    let server = GunnarSsh::new(
        config,
        plane_with_key(&key, Access::Read).control,
        Arc::new(Building),
    )
    .bind(mint_key())
    .await
    .expect("bind");
    let (port, mode) = proxy(server.local_addr()).await;
    let _client = parked_fetch(port, &key).await;
    assert!(
        wait_for(Duration::from_secs(10), || server
            .metrics()
            .transfers_in_flight
            == 1)
        .await,
        "the build never started"
    );
    let killed = Instant::now();
    mode.send_replace(Mode::Silent);
    // 1 s x 3 keepalives; well inside the build's 120 s, and inside `ephemeral_loopback`'s
    // 20 s inactivity bound too.
    let released = wait_for(Duration::from_secs(15), || {
        server.metrics().transfers_in_flight == 0
    })
    .await;
    assert!(
        released,
        "a client silent while the pack is still being built kept its transfer {:?} after the link died",
        killed.elapsed()
    );
    mode.send_replace(Mode::Close);
}

/// The shipped keepalive is a decision: a dead client is called dead within a
/// minute (15 s x russh's 3), not four.
#[test]
fn the_shipped_server_keepalive_is_fifteen_seconds() {
    assert_eq!(
        SshConfig::default().keepalive_interval,
        Some(Duration::from_secs(15))
    );
}