porthole/src/supervisor.rs

//! The `__supervise` loop 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 std::fs::OpenOptions;
use std::sync::{ Arc, Mutex };
use std::thread::JoinHandle;
use std::time::{ Instant, Duration };
use std::io::{ Write, BufRead, BufReader };
use std::sync::atomic::{ Ordering, AtomicBool };

use crate::error::Result;
use crate::{ ssh, timefmt };
use crate::profile::{ self, Profile };
use crate::instance::{ self, Lock, State, Instance };

/// How long a connection must survive before its uptime resets the backoff
/// counter back to the base delay.
const STABLE_THRESHOLD_SECS: i64 = 60;
/// Consecutive unrecognized (not pattern-matched) failures before porthole
/// gives up on an apparently-permanently-broken profile.
const MAX_UNRECOGNIZED_STREAK: u32 = 10;
/// How long `ssh` must stay alive before porthole treats it as connected;
/// see `run_ssh_once` for the heuristic this backs.
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 into a flag instead of the default
/// terminate-immediately behavior. This is how `close` (SIGTERM) and
/// `-f/--foreground`'s Ctrl-C (SIGINT) are distinguished from a
/// dropped `ssh` connection: by which signal arrived, not by inferring
/// intent from `ssh`'s exit status. In foreground mode this function runs
/// in the process the terminal sends Ctrl-C to directly, 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>`. This function is the
/// supervisor process: it runs until told to stop (SIGTERM/SIGINT) or
/// gives up per §4.
pub fn run(name: &str) -> Result<()> {
    install_signal_handler();

    let profile = profile::load(name)?;

    // Holding this for our entire lifetime is what makes
    // for a reliable, race-free check for `open`.
    let Some(_lock) = Lock::try_acquire(name)? else {
        return Ok(()); // another supervisor; 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: u64               = 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` does not report "the forward is bound" directly
/// without parsing `-v` debug output; a real failure exits near-instantly
/// under `ExitOnForwardFailure=yes` (§3.1), so staying alive past the grace
/// window is used as a 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. 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));
    }
}