crates/nornir-git/tests/ssh_liveness.rs
//! **A dropped or silent link must END a transfer, with an error that says why.**
//!
//! The defect (cov-13 finding 4, UpCloud mockup, 2026-10-06 18:33 UTC): the
//! kit's WireGuard tunnel went silent for six minutes and six `gunnar_client::push`
//! calls — all through this crate's russh bridge — hung FOREVER: TCP in
//! CLOSE-WAIT with 245 bytes unread, every thread parked, no error, no
//! recovery. gunnar-cli (customers' pushes) and gunnar's replication ride the
//! same bridge.
//!
//! Every test here dials a real russh server through a proxy that can go
//! [`Mode::Silent`] or [`Mode::Close`] mid-transfer, and runs the client under a
//! watchdog: a client that parks is a FAILED test (`HUNG: …`), never a hung CI.
//!
//! The bounds the tests use are the short ones from [`fast`]; the shipped
//! defaults are asserted in `the_shipped_bounds_are_a_decision`.
mod liveness_common;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
use liveness_common::*;
use nornir_git::ssh::{connect_exec_with, ls_remote_blocking_auth, parse_ssh_url, SshAuth, SshLiveness};
/// Past this the client is parked: the watchdog fails the test by name.
const WATCHDOG: Duration = Duration::from_secs(25);
/// The client's auth with the SHORT test bounds: connect 3 s, keepalive
/// 1 s x 3, inactivity 5 s, no progress 8 s.
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),
})
}
/// The client's auth with a keepalive far too slow to matter, so a test can
/// prove a hang-up is noticed by itself and not by the keepalive.
fn slow_keepalive(key: &std::path::Path) -> SshAuth {
SshAuth::key(key).with_liveness(SshLiveness {
connect: Duration::from_secs(3),
keepalive_interval: Duration::from_secs(60),
keepalive_max: 3,
inactivity: Duration::from_secs(120),
stall: Duration::from_secs(120),
})
}
struct Rig {
_td: tempfile::TempDir,
srv: TestServer,
proxy: Proxy,
key: std::path::PathBuf,
}
fn rig(answer_after: Option<Duration>) -> Rig {
let td = tempfile::tempdir().expect("tempdir");
let srv = TestServer::start(td.path(), ServerPlan { answer_after });
let proxy = Proxy::start(srv.addr);
let key = mint_key(td.path());
Rig { _td: td, srv, proxy, key }
}
/// The far end, as every error must name it.
fn far(proxy: &Proxy) -> String {
format!("127.0.0.1:{}", proxy.port)
}
/// Open `sink`, read its `ready`, then push until the link fails.
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()),
}
}
/// Open `sink`, send a short request ending in `END\n`, tell the test it is now
/// WAITING, and read the 3-byte answer.
fn ask_and_wait(
url: String,
auth: SshAuth,
waiting: std::sync::mpsc::Sender<()>,
) -> Result<Vec<u8>, 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}"))?;
w.write_all(&vec![0x5au8; 128 * 1024]).map_err(|e| format!("request: {e}"))?;
w.write_all(b"END\n").map_err(|e| format!("request end: {e}"))?;
w.flush().map_err(|e| format!("flush: {e}"))?;
let _ = waiting.send(());
read_n(r, 3).map_err(|e| e.to_string())
}
use std::io::Write as _;
// ── a peer that HANGS UP ────────────────────────────────────────────────────
/// The peer hangs up while the push streams: the push ends at once, and the
/// error names the far end — not a bare "channel closed".
#[test]
fn peer_hangs_up_mid_push_ends_it_at_once_naming_the_far_end() {
let rig = rig(None);
rig.proxy.arm(1 << 20, Mode::Close);
let (url, auth, at) = (rig.proxy.url("/x"), fast(&rig.key), far(&rig.proxy));
let (out, took) = within(WATCHDOG, "push while the peer hangs up", move || push_into(url, auth));
let msg = out.expect("the push must fail, not succeed");
assert!(took < Duration::from_secs(5), "a hang-up must end the push at once; took {took:?}");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
}
/// THE INCIDENT'S PARK: the far end stops taking data (the SSH window runs dry
/// and the writer parks waiting for more), then hangs up. russh never wakes a
/// writer parked on the window when its session ends; the push must still end
/// at once. A slow keepalive here proves the HANG-UP ended it.
#[test]
fn peer_hangs_up_while_the_push_is_parked_on_the_window_ends_it_at_once() {
let rig = rig(None);
rig.proxy.arm(1 << 20, Mode::Silent);
let (url, auth, at) = (rig.proxy.url("/x"), slow_keepalive(&rig.key), far(&rig.proxy));
let proxy = rig.proxy;
let (out, _) = within(WATCHDOG, "push parked on the window, then hung up", move || {
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = tx.send(push_into(url, auth));
});
// Let the writer drain the window and park.
assert!(settle(Duration::from_secs(10), || proxy.up_bytes() >= 1 << 20), "the push never started");
std::thread::sleep(Duration::from_millis(1500));
let hung_up = Instant::now();
proxy.set_all(Mode::Close);
let out = rx.recv().expect("push thread");
(out, hung_up.elapsed(), proxy)
});
let (out, after_hangup, _proxy) = out;
let msg = out.expect("the push must fail, not succeed");
assert!(after_hangup < Duration::from_secs(3), "the hang-up must end a parked push at once; took {after_hangup:?}");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
}
/// The peer hangs up while the client waits for the answer: the read ends with
/// an error naming the far end, not an anonymous "failed to fill whole buffer".
#[test]
fn peer_hangs_up_while_waiting_for_the_answer_names_the_far_end() {
let rig = rig(None);
let (url, auth, at) = (rig.proxy.url("/x"), fast(&rig.key), far(&rig.proxy));
let (tx, waiting) = std::sync::mpsc::channel();
let proxy = rig.proxy;
let seen = rig.srv.seen.clone();
let (out, _) = within(WATCHDOG, "read while the peer hangs up", move || {
let (otx, orx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = otx.send(ask_and_wait(url, auth, tx));
});
waiting.recv().expect("the request went out");
assert!(settle(Duration::from_secs(10), || seen.sunk.load(Ordering::SeqCst) >= 128 * 1024 + 4));
proxy.set_all(Mode::Close);
(orx.recv().expect("client thread"), proxy)
});
let msg = out.0.expect_err("the read must fail");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
}
// ── a link that goes SILENT ─────────────────────────────────────────────────
/// The link goes silent mid-push (the proxy stops forwarding both ways and
/// keeps the sockets open — TCP alone never times this out): the push ends
/// within the keepalive bound, and the error says it was the keepalive.
#[test]
fn silent_link_mid_push_ends_within_the_keepalive_bound() {
let rig = rig(None);
rig.proxy.arm(1 << 20, Mode::Silent);
let (url, auth, at) = (rig.proxy.url("/x"), fast(&rig.key), far(&rig.proxy));
let (out, took) = within(WATCHDOG, "push over a silent link", move || push_into(url, auth));
let msg = out.expect("the push must fail, not succeed");
assert!(took < Duration::from_secs(12), "1 s x 3 keepalive (5 s inactivity) must end it; took {took:?}");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
assert!(
msg.contains("keepalive") || msg.contains("inactiv") || msg.contains("silent"),
"the error must say the link went silent: {msg:?}"
);
}
/// The link goes silent while the client waits for the answer.
#[test]
fn silent_link_while_waiting_for_the_answer_ends_within_the_keepalive_bound() {
let rig = rig(None);
let (url, auth, at) = (rig.proxy.url("/x"), fast(&rig.key), far(&rig.proxy));
let (tx, waiting) = std::sync::mpsc::channel();
let proxy = rig.proxy;
let seen = rig.srv.seen.clone();
let (out, _) = within(WATCHDOG, "read over a silent link", move || {
let (otx, orx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = otx.send(ask_and_wait(url, auth, tx));
});
waiting.recv().expect("the request went out");
assert!(settle(Duration::from_secs(10), || seen.sunk.load(Ordering::SeqCst) >= 128 * 1024 + 4));
let silent = Instant::now();
proxy.set_all(Mode::Silent);
let out = orx.recv().expect("client thread");
(out, silent.elapsed(), proxy)
});
let (out, after_silence, _proxy) = out;
let msg = out.expect_err("the read must fail");
assert!(after_silence < Duration::from_secs(7), "the keepalive (1 s x 3) must end it before the 8 s no-progress bound; took {after_silence:?}");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
assert!(msg.contains("keepalive"), "the error must say the keepalive went unanswered: {msg:?}");
}
// ── a far end that is ALIVE but makes no progress ───────────────────────────
/// The far end answers keepalives but never answers the request: the read gives
/// up after the no-progress bound, saying so.
#[test]
fn a_live_far_end_that_never_answers_times_out_on_no_progress() {
let rig = rig(None);
let (url, auth, at) = (rig.proxy.url("/x"), fast(&rig.key), far(&rig.proxy));
let (tx, _waiting) = std::sync::mpsc::channel();
let (out, took) = within(WATCHDOG, "read from a far end that never answers", move || ask_and_wait(url, auth, tx));
let msg = out.expect_err("the read must fail");
assert!(took >= Duration::from_secs(7) && took < Duration::from_secs(16), "the 8 s no-progress bound; took {took:?}");
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
assert!(msg.contains("no progress"), "the error must say nothing moved: {msg:?}");
}
/// The guard on the other side: a far end that is merely SLOW (answers after
/// 6 s — longer than three missed keepalives and the 5 s inactivity bound) but
/// alive is NOT cut off. Keepalive replies are what keep it.
#[test]
fn a_slow_but_live_far_end_is_not_cut_off() {
let rig = rig(Some(Duration::from_secs(6)));
let (url, auth) = (rig.proxy.url("/x"), fast(&rig.key));
let (tx, _waiting) = std::sync::mpsc::channel();
let (out, took) = within(WATCHDOG, "read from a slow far end", move || ask_and_wait(url, auth, tx));
assert_eq!(out.expect("a slow live far end must be waited for"), b"ok\n");
assert!(took >= Duration::from_secs(6), "it really waited: {took:?}");
}
// ── ls-remote ───────────────────────────────────────────────────────────────
/// ls-remote against a far end that sends half an advertisement and then
/// nothing: an error after the no-progress bound, not a hang.
#[test]
fn ls_remote_against_a_stuck_far_end_errors_instead_of_hanging() {
let rig = rig(None);
let (url, auth, at) = (rig.proxy.url("/repo"), fast(&rig.key), far(&rig.proxy));
let (out, _) = within(WATCHDOG, "ls-remote against a stuck far end", move || ls_remote_blocking_auth(&url, &auth));
let msg = format!("{:#}", out.expect_err("ls-remote must fail"));
assert!(msg.contains(&at), "the error must name the far end {at}: {msg:?}");
}
/// ls-remote cut mid-advertisement must be an ERROR — never the refs that
/// happened to arrive, returned as if they were the list.
#[test]
fn ls_remote_cut_mid_advertisement_is_an_error_not_a_short_list() {
let rig = rig(None);
let (url, auth) = (rig.proxy.url("/repo"), fast(&rig.key));
let proxy = rig.proxy;
let seen = rig.srv.seen.clone();
let (out, _) = within(WATCHDOG, "ls-remote cut mid-advertisement", move || {
let (otx, orx) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let _ = otx.send(ls_remote_blocking_auth(&url, &auth));
});
assert!(settle(Duration::from_secs(10), || seen.partial_sent.load(Ordering::SeqCst)));
std::thread::sleep(Duration::from_millis(300));
proxy.set_all(Mode::Close);
(orx.recv().expect("ls-remote thread"), proxy)
});
match out.0 {
Ok(refs) => panic!("a cut advertisement came back as a {}-ref list: {refs:?}", refs.len()),
Err(e) => assert!(format!("{e:#}").contains("127.0.0.1"), "names the far end: {e:#}"),
}
}
// ── connect ─────────────────────────────────────────────────────────────────
/// A far end that accepts TCP and never says a word (not even the SSH banner):
/// connect gives up after the connect bound.
#[test]
fn connect_to_a_far_end_that_never_speaks_times_out() {
let td = tempfile::tempdir().expect("tempdir");
let key = mint_key(td.path());
let port = mute_listener();
let auth = fast(&key);
let (out, took) = within(WATCHDOG, "connect to a mute far end", move || {
let loc = parse_ssh_url(&format!("ssh://git@127.0.0.1:{port}/x")).expect("url");
connect_exec_with(&loc, &auth, "sink", &[]).map(|_| ()).map_err(|e| format!("{e:#}"))
});
let msg = out.expect_err("connect must fail");
assert!(took < Duration::from_secs(8), "the 3 s connect bound; took {took:?}");
assert!(msg.contains(&format!("127.0.0.1:{port}")), "names the far end: {msg:?}");
}