Implement porthole v0.1: profiles, supervised open/close, reconnect
Full CLI per spec v0.2 - add/open/close/edit/status/list/remove/wipe/ completions, plus a hidden `__supervise` subcommand that IS the supervisor process. - profile.rs: TOML-backed profiles at ~/.config/porthole/profiles/, validated -l/-r/-d mapping + --via grammar, atomic writes. - instance.rs: JSON runtime state at ~/.local/state/porthole/, an flock-based lock file that's the source of truth for "is this open" (survives a crash/kill -9 without stale-lock cleanup), pid liveness checked against /proc rather than trusted from disk. - supervisor.rs: the __supervise loop - spawns ssh, traps SIGTERM/SIGINT into a flag (rather than inferring intent from ssh's exit status), classifies failures as fatal/known-transient/unrecognized, backs off with a stability-reset, rotates its log. - ssh.rs: builds the ssh invocation, including splitting --via into a -J jump chain plus the mandatory positional target. - open.rs: the detach/re-exec dance (setsid via pre_exec) and a bounded wait for the supervisor to reach Up/Error before open returns, so an immediate failure surfaces as a non-zero exit instead of a false "opened" - this took a real bug fix during smoke testing, since the instance file's initial state (Reconnecting, meaning "attempt in flight") was indistinguishable from "already failed once" by state alone. - close.rs: SIGTERM+wait, or SIGKILL the whole process group with --force so ssh can't be left orphaned. Smoke-tested against invalid/unreachable hosts (no real infrastructure touched): CLI surface, validation errors, add/edit/list/status/remove, the reconnect/backoff loop with live state transitions, close mid-retry, edit-while-running's warning, open --all, and wipe. cargo test: 14/14. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
283
src/supervisor.rs
Normal file
283
src/supervisor.rs
Normal file
@@ -0,0 +1,283 @@
|
||||
//! The `__supervise` loop - spec §3/§4. Runs as a detached, re-exec'd copy
|
||||
//! of this same binary (`porthole __supervise <name>`, see `main.rs`); owns
|
||||
//! the `ssh` child process for one profile's entire supervised lifetime.
|
||||
|
||||
use crate::error::Result;
|
||||
use crate::instance::{self, Instance, Lock, State};
|
||||
use crate::profile::{self, Profile};
|
||||
use crate::{ssh, timefmt};
|
||||
use std::fs::OpenOptions;
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// How long a connection must survive before its uptime resets the backoff
|
||||
/// counter back to the base delay - spec §4.1.
|
||||
const STABLE_THRESHOLD_SECS: i64 = 60;
|
||||
/// Consecutive unrecognized (not pattern-matched) failures before porthole
|
||||
/// gives up on an apparently-permanently-broken profile - spec §4.2.
|
||||
const MAX_UNRECOGNIZED_STREAK: u32 = 10;
|
||||
/// How long `ssh` must stay alive before porthole calls it "connected" -
|
||||
/// see `run_ssh_once`'s doc comment for why this heuristic is used at all.
|
||||
const CONNECT_GRACE: Duration = Duration::from_secs(2);
|
||||
const POLL_INTERVAL: Duration = Duration::from_millis(200);
|
||||
const LOG_ROTATE_BYTES: u64 = 10 * 1024 * 1024;
|
||||
|
||||
static SHUTDOWN: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
extern "C" fn handle_sigterm(_sig: libc::c_int) {
|
||||
SHUTDOWN.store(true, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
/// Traps SIGTERM (and SIGINT, for `-f/--foreground`'s Ctrl-C - spec §5.2)
|
||||
/// into a flag instead of the default terminate-immediately behavior, so
|
||||
/// `close` is distinguished from a dropped `ssh` connection by *why* the
|
||||
/// loop is unwinding, not by guessing from `ssh`'s exit status - which is
|
||||
/// not a reliable signal either way. In foreground mode this function runs
|
||||
/// in the same process the terminal sends Ctrl-C's SIGINT to, since
|
||||
/// `commands::open` calls `supervisor::run` inline rather than detaching.
|
||||
fn install_signal_handler() {
|
||||
unsafe {
|
||||
libc::signal(libc::SIGTERM, handle_sigterm as *const () as usize);
|
||||
libc::signal(libc::SIGINT, handle_sigterm as *const () as usize);
|
||||
}
|
||||
}
|
||||
|
||||
enum Class {
|
||||
Fatal,
|
||||
KnownTransient,
|
||||
Unrecognized,
|
||||
}
|
||||
|
||||
enum Outcome {
|
||||
ShutdownRequested,
|
||||
Failed { class: Class, message: String },
|
||||
}
|
||||
|
||||
/// Entry point for `porthole __supervise <name>`. Runs until told to stop
|
||||
/// (SIGTERM) or gives up per §4 - this *is* the supervisor process.
|
||||
pub fn run(name: &str) -> Result<()> {
|
||||
install_signal_handler();
|
||||
|
||||
let profile = profile::load(name)?;
|
||||
|
||||
// Holding this for our entire lifetime is what makes "is <name>
|
||||
// already open" a reliable, race-free check for `open` (spec §2.3/§3).
|
||||
let Some(_lock) = Lock::try_acquire(name)? else {
|
||||
return Ok(()); // another supervisor beat us to it; nothing to do
|
||||
};
|
||||
|
||||
let pid = std::process::id() as i32;
|
||||
let mut inst = Instance::new(name.to_string(), pid);
|
||||
instance::save(&inst)?;
|
||||
|
||||
let once = std::env::var_os("PORTHOLE_SUPERVISE_ONCE").is_some();
|
||||
let base_delay = profile.retry_interval.max(1) as u64;
|
||||
let max_delay = (profile.backoff_max as u64).max(base_delay);
|
||||
let mut delay = base_delay;
|
||||
let mut unrecognized_streak: u32 = 0;
|
||||
|
||||
loop {
|
||||
let attempt_started = timefmt::now();
|
||||
match run_ssh_once(name, &profile, &mut inst) {
|
||||
Outcome::ShutdownRequested => {
|
||||
instance::delete(name)?;
|
||||
return Ok(());
|
||||
}
|
||||
Outcome::Failed { class, message } => {
|
||||
let uptime = timefmt::now() - attempt_started;
|
||||
if uptime >= STABLE_THRESHOLD_SECS {
|
||||
delay = base_delay;
|
||||
unrecognized_streak = 0;
|
||||
}
|
||||
match class {
|
||||
Class::Unrecognized => unrecognized_streak += 1,
|
||||
Class::KnownTransient => unrecognized_streak = 0,
|
||||
Class::Fatal => {}
|
||||
}
|
||||
|
||||
inst.last_error =
|
||||
Some(if message.is_empty() { "ssh exited unexpectedly (no output captured)".to_string() } else { message });
|
||||
|
||||
let fatal = matches!(class, Class::Fatal);
|
||||
let give_up = fatal || !profile.reconnect || once || unrecognized_streak > MAX_UNRECOGNIZED_STREAK;
|
||||
if give_up {
|
||||
inst.state = State::Error;
|
||||
instance::save(&inst)?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
inst.state = State::Reconnecting;
|
||||
inst.reconnect_count += 1;
|
||||
inst.last_reconnect_at = Some(timefmt::now());
|
||||
inst.connected_at = None;
|
||||
instance::save(&inst)?;
|
||||
|
||||
if sleep_or_shutdown(Duration::from_secs(delay)) {
|
||||
instance::delete(name)?;
|
||||
return Ok(());
|
||||
}
|
||||
delay = (delay * 2).min(max_delay);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Sleeps for `dur`, polling `SHUTDOWN` periodically so a `close` that
|
||||
/// arrives during a reconnect backoff window is honored promptly instead
|
||||
/// of waiting out the full delay. Returns `true` if shutdown was requested.
|
||||
fn sleep_or_shutdown(dur: Duration) -> bool {
|
||||
let deadline = Instant::now() + dur;
|
||||
while Instant::now() < deadline {
|
||||
if SHUTDOWN.load(Ordering::SeqCst) {
|
||||
return true;
|
||||
}
|
||||
std::thread::sleep(POLL_INTERVAL.min(dur));
|
||||
}
|
||||
SHUTDOWN.load(Ordering::SeqCst)
|
||||
}
|
||||
|
||||
/// Spawns one `ssh` attempt and supervises it until it exits or shutdown is
|
||||
/// requested. Marks `inst` as `State::Up` once the process has survived
|
||||
/// `CONNECT_GRACE` - `ssh` gives no more reliable "the forward is actually
|
||||
/// bound" signal than that without parsing `-v` debug output, and a real
|
||||
/// failure exits near-instantly under `ExitOnForwardFailure=yes` (§3.1), so
|
||||
/// staying alive past the grace window is a reasonable proxy for connected.
|
||||
fn run_ssh_once(name: &str, profile: &Profile, inst: &mut Instance) -> Outcome {
|
||||
rotate_log_if_large(name);
|
||||
|
||||
let mut cmd = ssh::build(profile);
|
||||
let mut child = match cmd.spawn() {
|
||||
Ok(c) => c,
|
||||
Err(e) => return Outcome::Failed { class: Class::Unrecognized, message: format!("failed to spawn ssh: {e}") },
|
||||
};
|
||||
|
||||
let stderr_tail = Arc::new(Mutex::new(String::new()));
|
||||
let stdout_thread = child.stdout.take().map(|out| spawn_log_drain(name, out));
|
||||
let stderr_thread = child.stderr.take().map(|err| spawn_stderr_drain(name, err, stderr_tail.clone()));
|
||||
|
||||
let grace_deadline = Instant::now() + CONNECT_GRACE;
|
||||
let mut marked_up = false;
|
||||
|
||||
loop {
|
||||
if SHUTDOWN.load(Ordering::SeqCst) {
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
join_all([stdout_thread, stderr_thread]);
|
||||
return Outcome::ShutdownRequested;
|
||||
}
|
||||
match child.try_wait() {
|
||||
Ok(Some(_status)) => break,
|
||||
Ok(None) => {
|
||||
if !marked_up && Instant::now() >= grace_deadline {
|
||||
marked_up = true;
|
||||
inst.state = State::Up;
|
||||
inst.connected_at = Some(timefmt::now());
|
||||
let _ = instance::save(inst);
|
||||
}
|
||||
std::thread::sleep(POLL_INTERVAL);
|
||||
}
|
||||
Err(_) => break, // process table race (should not happen on unix); treat as exited
|
||||
}
|
||||
}
|
||||
|
||||
join_all([stdout_thread, stderr_thread]);
|
||||
|
||||
let tail = stderr_tail.lock().map(|s| s.clone()).unwrap_or_default();
|
||||
let (class, message) = classify(&tail);
|
||||
Outcome::Failed { class, message }
|
||||
}
|
||||
|
||||
fn join_all<const N: usize>(handles: [Option<JoinHandle<()>>; N]) {
|
||||
for h in handles.into_iter().flatten() {
|
||||
let _ = h.join();
|
||||
}
|
||||
}
|
||||
|
||||
/// Classifies `ssh`'s captured stderr per spec §4.2. Fatal patterns stop
|
||||
/// the reconnect loop outright; known-transient patterns retry without
|
||||
/// counting toward the unrecognized-failure escalation; anything else
|
||||
/// still retries, but does count toward it.
|
||||
fn classify(stderr_tail: &str) -> (Class, String) {
|
||||
const FATAL: &[&str] = &["Permission denied", "Host key verification failed", "bind: Address already in use"];
|
||||
const KNOWN_TRANSIENT: &[&str] = &[
|
||||
"Connection refused",
|
||||
"No route to host",
|
||||
"Could not resolve hostname",
|
||||
"Connection timed out",
|
||||
"Operation timed out",
|
||||
];
|
||||
|
||||
let message = stderr_tail.lines().rev().find(|l| !l.trim().is_empty()).unwrap_or("").trim().to_string();
|
||||
|
||||
if FATAL.iter().any(|p| stderr_tail.contains(p)) {
|
||||
(Class::Fatal, message)
|
||||
} else if KNOWN_TRANSIENT.iter().any(|p| stderr_tail.contains(p)) {
|
||||
(Class::KnownTransient, message)
|
||||
} else {
|
||||
(Class::Unrecognized, message)
|
||||
}
|
||||
}
|
||||
|
||||
fn rotate_log_if_large(name: &str) {
|
||||
let path = instance::log_path(name);
|
||||
if let Ok(meta) = std::fs::metadata(&path) {
|
||||
if meta.len() > LOG_ROTATE_BYTES {
|
||||
let _ = std::fs::rename(&path, path.with_extension("log.1"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn append_log(name: &str, line: &str) {
|
||||
if let Ok(mut f) = OpenOptions::new().create(true).append(true).open(instance::log_path(name)) {
|
||||
let _ = writeln!(f, "{line}");
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn_log_drain(name: &str, out: std::process::ChildStdout) -> JoinHandle<()> {
|
||||
let name = name.to_string();
|
||||
std::thread::spawn(move || {
|
||||
for line in BufReader::new(out).lines().map_while(std::result::Result::ok) {
|
||||
append_log(&name, &line);
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn spawn_stderr_drain(name: &str, err: std::process::ChildStderr, tail: Arc<Mutex<String>>) -> JoinHandle<()> {
|
||||
let name = name.to_string();
|
||||
std::thread::spawn(move || {
|
||||
for line in BufReader::new(err).lines().map_while(std::result::Result::ok) {
|
||||
append_log(&name, &line);
|
||||
if let Ok(mut t) = tail.lock() {
|
||||
if !t.is_empty() {
|
||||
t.push('\n');
|
||||
}
|
||||
t.push_str(&line);
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn classifies_fatal_patterns() {
|
||||
assert!(matches!(classify("foo\nPermission denied (publickey).").0, Class::Fatal));
|
||||
assert!(matches!(classify("bind: Address already in use").0, Class::Fatal));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classifies_known_transient_patterns() {
|
||||
assert!(matches!(classify("ssh: connect to host x port 22: Connection refused").0, Class::KnownTransient));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classifies_unrecognized_as_transient() {
|
||||
assert!(matches!(classify("something completely unexpected").0, Class::Unrecognized));
|
||||
assert!(matches!(classify("").0, Class::Unrecognized));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user