Overlord/porthole
1
1
Fork 0
Code Releases 1 Activity

Standardize error messages, formatting, and code style. Refine --help text for consistency. Update comments for clarity and tone alignment.

Browse Source
This commit is contained in:
Overlord
2026-08-13 17:50:06 +02:00
parent e36360e18c
commit 1b947fec24
7 changed files with 123 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
src/supervisor.rs
View File

@@ -27,9 +27,7 @@ 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);
}
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
@@ -41,7 +39,7 @@ extern "C" fn handle_sigterm(_sig: libc::c_int) {
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);
libc::signal(libc::SIGINT, handle_sigterm as *const () as usize);
}
}
@@ -64,62 +62,70 @@ pub fn run(name: &str) -> Result<()> {
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).
// 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 beat us to it; nothing to do
return Ok(()); // another supervisor; nothing to do
};
let pid = std::process::id() as i32;
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 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 max_delay = (profile.backoff_max as u64).max(base_delay);
let mut delay: u64 = base_delay;
let mut unrecognized_streak: u32 = 0;
loop {
loop
{
let attempt_started = timefmt::now();
match run_ssh_once(name, &profile, &mut inst) {
match run_ssh_once(name, &profile, &mut inst)
{
Outcome::ShutdownRequested => {
instance::delete(name)?;
return Ok(());
}
Outcome::Failed { class, message } => {
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::Unrecognized => unrecognized_streak += 1,
Class::KnownTransient => unrecognized_streak = 0,
Class::Fatal => {}
Class::Fatal => {}
}
inst.last_error =
Some(if message.is_empty() { "ssh exited unexpectedly (no output captured)".to_string() } else { message });
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 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.state = State::Reconnecting;
inst.reconnect_count += 1;
inst.last_reconnect_at = Some(timefmt::now());
inst.connected_at = None;
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);
}
}
@@ -129,14 +135,17 @@ pub fn run(name: &str) -> Result<()> {
/// 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 {
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)
}
@@ -146,37 +155,43 @@ fn sleep_or_shutdown(dur: Duration) -> bool {
/// 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 {
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,
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 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;
let mut marked_up = false;
loop {
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() {
match child.try_wait()
{
Ok(Some(_status)) => break,
Ok(None) => {
if !marked_up && Instant::now() >= grace_deadline {
marked_up = true;
inst.state = State::Up;
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);
let _ = instance::save(inst);
}
std::thread::sleep(POLL_INTERVAL);
}
@@ -187,7 +202,9 @@ fn run_ssh_once(name: &str, profile: &Profile, inst: &mut Instance) -> Outcome {
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 }
}
@@ -201,8 +218,9 @@ fn join_all<const N: usize>(handles: [Option<JoinHandle<()>>; N]) {
/// 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"];
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",
@@ -213,17 +231,15 @@ fn classify(stderr_tail: &str) -> (Class, String) {
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)
}
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) {
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"));
@@ -237,8 +253,10 @@ fn append_log(name: &str, line: &str) {
}
}
fn spawn_log_drain(name: &str, out: std::process::ChildStdout) -> JoinHandle<()> {
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);
@@ -246,11 +264,15 @@ fn spawn_log_drain(name: &str, out: std::process::ChildStdout) -> JoinHandle<()>
})
}
fn spawn_stderr_drain(name: &str, err: std::process::ChildStderr, tail: Arc<Mutex<String>>) -> JoinHandle<()> {
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) {
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');