nordisk/edda

crates/nornir-git/tests/ssh_liveness_leaks.rs

//! **Nothing is left behind** when a dropped or silent link ends a transfer: no
//! parked thread, no open descriptor, no socket sitting in CLOSE-WAIT.
//!
//! The incident (cov-13 finding 4) left all three: six pushes' threads parked
//! forever, their sockets in CLOSE-WAIT with 245 bytes unread. An error that
//! surfaces while the bridge's runtime threads or its socket live on is half a
//! fix — a long-lived caller (gunnar's replication, the kit) would still
//! accumulate them, one per bad link.
//!
//! ONE test in its own binary on purpose: thread and descriptor counts are
//! per-PROCESS, and a sibling test running alongside would move them.

mod liveness_common;

use std::time::Duration;

use liveness_common::*;
use nornir_git::ssh::{connect_exec_with, parse_ssh_url, SshAuth, SshLiveness};

const WATCHDOG: Duration = Duration::from_secs(25);

fn fast(key: &std::path::Path) -> SshAuth {
    SshAuth::key(key).with_liveness(SshLiveness {
        connect: Duration::from_secs(3),
        keepalive_interval: Duration::from_secs(1),
        keepalive_max: 3,
        inactivity: Duration::from_secs(5),
        stall: Duration::from_secs(8),
    })
}

fn push_into(url: String, auth: SshAuth) -> Result<String, String> {
    let loc = parse_ssh_url(&url).map_err(|e| format!("{e:#}"))?;
    let mut ch = connect_exec_with(&loc, &auth, "sink", &[]).map_err(|e| format!("connect: {e:#}"))?;
    let (r, w) = ch.io();
    read_n(r, 6).map_err(|e| format!("ready: {e}"))?;
    match write_until_error(w, 1 << 30) {
        Ok(n) => Err(format!("wrote {n} bytes into a dead link and was told nothing")),
        Err(e) => Ok(e.to_string()),
    }
    // `ch` drops here: the bridge's runtime, its threads and its socket go with it.
}

/// One healthy round trip, so everything a first connection initialises lazily
/// is in the baseline rather than counted as a leak.
fn warm_up(url: String, auth: SshAuth) {
    let loc = parse_ssh_url(&url).expect("url");
    let mut ch = connect_exec_with(&loc, &auth, "sink", &[]).expect("warm-up connect");
    let (r, _) = ch.io();
    read_n(r, 6).expect("warm-up ready");
}

#[test]
fn a_dead_link_leaves_no_thread_no_descriptor_and_no_close_wait_socket() {
    let td = tempfile::tempdir().expect("tempdir");
    let srv = TestServer::start(td.path(), ServerPlan { answer_after: None });
    let proxy = std::sync::Arc::new(Proxy::start(srv.addr));
    let key = mint_key(td.path());

    warm_up(proxy.url("/x"), fast(&key));
    proxy.close_all();
    // Let the warm-up's sockets on all three parties close.
    std::thread::sleep(Duration::from_millis(500));
    let (t0, f0) = (threads(), fds());

    for (name, mode) in [("silent link mid-push", Mode::Silent), ("hang-up mid-push", Mode::Close)] {
        proxy.arm(1 << 20, mode);
        let (url, auth) = (proxy.url("/x"), fast(&key));
        let (out, took) = within(WATCHDOG, name, move || push_into(url, auth));
        let msg = out.unwrap_or_else(|e| panic!("{name}: the push must fail, not succeed: {e}"));
        eprintln!("{name}: ended after {took:?}: {msg}");
        proxy.disarm();
        // The CLIENT must already be clean: its socket is not left in
        // CLOSE-WAIT. (The proxy still holds its side of a silent link here.)
        assert!(
            settle(Duration::from_secs(3), || close_wait_to(proxy.port) == 0),
            "{name}: {} client socket(s) to the proxy left in CLOSE-WAIT",
            close_wait_to(proxy.port)
        );
        // Now let the proxy and the server drop their halves, and count.
        proxy.close_all();
        let clean = settle(Duration::from_secs(5), || threads() <= t0 && fds() <= f0);
        assert!(
            clean,
            "{name}: leaked — threads {} (baseline {t0}), descriptors {} (baseline {f0})",
            threads(),
            fds()
        );
    }
}