diff --git a/src/atomic.rs b/src/atomic.rs index e660b6d..f00c406 100644 --- a/src/atomic.rs +++ b/src/atomic.rs @@ -14,10 +14,12 @@ pub fn write(path: &Path, contents: &[u8]) -> std::io::Result<()> path.extension().and_then(|e| e.to_str()).unwrap_or(""), std::process::id() )); + { let mut f = std::fs::File::create(&tmp)?; f.write_all(contents)?; f.sync_all()?; } + std::fs::rename(&tmp, path) } diff --git a/src/cli.rs b/src/cli.rs index 73870bc..fee8d08 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -11,7 +11,7 @@ pub struct Cli { #[derive(Subcommand)] pub enum Commands { - /// Save a new forward profile (doesn't open it). + /// Save a new forward profile. #[command(visible_alias = "create", visible_alias = "new")] Add(AddArgs), @@ -54,11 +54,11 @@ pub enum Commands { /// struct (`#[command(flatten)]`ed into both) so the two can never drift. #[derive(Args, Default)] pub struct MappingArgs { - /// Local forward: your machine -> remote. + /// Local forward (current machine -> remote machine). #[arg(short, long, value_name = "[BIND:]PORT:HOST:PORT")] pub local: Option, - /// Remote forward: remote -> your machine. + /// Remote forward (remote machine -> current machine). #[arg(short, long, value_name = "[BIND:]PORT:HOST:PORT")] pub remote: Option, @@ -78,23 +78,23 @@ pub struct MappingArgs { #[arg(short, long, value_name = "PATH")] pub identity: Option, - /// SSH port on the final target only. + /// SSH port of the final target. #[arg(short, long, value_name = "PORT")] pub port: Option, - /// Auto-reconnect on drop. + /// Auto-reconnect on connection drop. #[arg(long, num_args = 0..=1, default_missing_value = "true", value_name = "BOOL")] pub reconnect: Option, - /// Base delay between reconnect attempts, in seconds. + /// Base delay between reconnection attempts, in seconds. #[arg(long = "retry-interval", value_name = "SECONDS")] pub retry_interval: Option, - /// Cap on the doubling reconnect delay, in seconds. + /// Cap on the doubling reconnection delay, in seconds. #[arg(long = "backoff-max", value_name = "SECONDS")] pub backoff_max: Option, - /// ServerAliveInterval, in seconds. + /// SSH ServerAliveInterval, in seconds\. #[arg(long, value_name = "SECONDS")] pub keepalive: Option, } diff --git a/src/error.rs b/src/error.rs index e623b12..db64960 100644 --- a/src/error.rs +++ b/src/error.rs @@ -20,13 +20,13 @@ pub enum PortholeError { #[error("only one of -l/--local, -r/--remote, -d/--dynamic may be given")] MultipleMappingKinds, - #[error("invalid forward spec '{0}': expected [bind:]port:host:hostport (or [bind:]port for -d)")] + #[error("invalid forward spec '{0}': expected [BIND:]PORT:HOST:PORT (or [BIND:]PORT for -d)")] InvalidMapping(String), - #[error("invalid --via hop '{0}': expected [user@]host[:port]")] + #[error("invalid --via hop '{0}': expected [USER@]HOST[:PORT]")] InvalidVia(String), - #[error("--via is required: at least one hop (its last entry is the ssh connection target)")] + #[error("--via is required: at least one hop (connection target)")] NoViaHosts, #[error("nothing to do: {0}")] diff --git a/src/ssh.rs b/src/ssh.rs index 104fae2..14eebc0 100644 --- a/src/ssh.rs +++ b/src/ssh.rs @@ -10,7 +10,7 @@ pub fn build(profile: &Profile) -> Command { let mut cmd = Command::new("ssh"); cmd.stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped()); - // Forced flags - see spec §3.1 for why each of these is non-negotiable. + // Forced flags, see spec §3.1. cmd.args([ "-o", "BatchMode=yes", @@ -47,12 +47,13 @@ mod tests { use super::*; use crate::profile::{Kind, ProfileEdits}; - fn profile_with(via: Vec<&str>) -> Profile { + fn profile_with(via: Vec<&str>) -> Profile + { Profile::new( "t".into(), &ProfileEdits { local: Some("5432:db.internal:5432".into()), - via: Some(via.into_iter().map(String::from).collect()), + via: Some(via.into_iter().map(String::from).collect()), ..Default::default() }, ) diff --git a/src/supervisor.rs b/src/supervisor.rs index 18a3815..90aad4f 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -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 - // 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(handles: [Option>; 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>) -> JoinHandle<()> { +fn spawn_stderr_drain(name: &str, err: std::process::ChildStderr, tail: Arc>) -> 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'); diff --git a/src/timefmt.rs b/src/timefmt.rs index 2be2854..9728563 100644 --- a/src/timefmt.rs +++ b/src/timefmt.rs @@ -13,40 +13,43 @@ pub fn now() -> i64 { /// (public domain, http://howardhinnant.github.io/date_algorithms.html) - /// avoids pulling in a whole calendar/timezone crate for what's otherwise a /// handful of integer operations. -pub fn fmt_timestamp(unix_secs: i64) -> String { - let days = unix_secs.div_euclid(86_400); +pub fn fmt_timestamp(unix_secs: i64) -> String +{ + let days = unix_secs.div_euclid(86_400); let secs_of_day = unix_secs.rem_euclid(86_400); - let (h, m, s) = (secs_of_day / 3600, (secs_of_day / 60) % 60, secs_of_day % 60); + let (h, m, s) = (secs_of_day / 3600, (secs_of_day / 60) % 60, secs_of_day % 60); - let z = days + 719_468; + let z = days + 719_468; let era = if z >= 0 { z } else { z - 146_096 } / 146_097; - let doe = (z - era * 146_097) as i64; // [0, 146096] + + let doe = (z - era * 146_097) as i64; // [0, 146096] let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365; // [0, 399] + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365] - let mp = (5 * doy + 2) / 153; // [0, 11] - let d = doy - (153 * mp + 2) / 5 + 1; // [1, 31] + let mp = (5 * doy + 2) / 153; // [0, 11] + let d = doy - (153 * mp + 2) / 5 + 1; // [1, 31] let m_num = if mp < 10 { mp + 3 } else { mp - 9 }; // [1, 12] + let y = if m_num <= 2 { y + 1 } else { y }; format!("{y:04}-{m_num:02}-{d:02} {h:02}:{m:02}:{s:02} UTC") } /// Compact `1d 02h 03m 04s`-style duration, dropping leading zero units. -pub fn fmt_duration(secs: i64) -> String { - let secs = secs.max(0); +pub fn fmt_duration(secs: i64) -> String +{ + let secs = secs.max(0); let (d, rem) = (secs / 86_400, secs % 86_400); let (h, rem) = (rem / 3600, rem % 3600); - let (m, s) = (rem / 60, rem % 60); + let (m, s) = (rem / 60, rem % 60); - if d > 0 { - format!("{d}d {h:02}h {m:02}m {s:02}s") - } else if h > 0 { - format!("{h}h {m:02}m {s:02}s") - } else if m > 0 { - format!("{m}m {s:02}s") - } else { - format!("{s}s") + match (d, h, m) { + (d, _, _) if d > 0 => format!("{d}d {h:02}h {m:02}m {s:02}s"), + (_, h, _) if h > 0 => format!("{h}h {m:02}m {s:02}s"), + (_, _, m) if m > 0 => format!("{m}m {s:02}s"), + _ => format!("{s}s"), } } diff --git a/src/ui.rs b/src/ui.rs index 3d86fb0..16af074 100644 --- a/src/ui.rs +++ b/src/ui.rs @@ -1,9 +1,10 @@ -//! Colorized status output, honoring `NO_COLOR` and terminal detection. +//! Colorized status output, honouring `NO_COLOR` and terminal detection. use std::io::IsTerminal; use std::sync::OnceLock; -fn color_enabled() -> bool { +fn color_enabled() -> bool +{ static ENABLED: OnceLock = OnceLock::new(); *ENABLED.get_or_init(|| { std::env::var_os("NO_COLOR").is_none() @@ -13,46 +14,19 @@ fn color_enabled() -> bool { }) } -fn paint(code: &str, s: &str) -> String { - if color_enabled() { - format!("\x1b[{code}m{s}\x1b[0m") - } else { - s.to_string() - } +fn paint(code: &str, s: &str) -> String +{ + if color_enabled() { format!("\x1b[{code}m{s}\x1b[0m") } + else { s.to_string() } } -pub fn red(s: &str) -> String { - paint("31", s) -} +pub fn red(s: &str) -> String { paint("31", s) } +pub fn yellow(s: &str) -> String { paint("33", s) } +pub fn green(s: &str) -> String { paint("32", s) } +pub fn blue(s: &str) -> String { paint("34", s) } +pub fn cyan(s: &str) -> String { paint("36", s) } -pub fn yellow(s: &str) -> String { - paint("33", s) -} - -pub fn green(s: &str) -> String { - paint("32", s) -} - -pub fn blue(s: &str) -> String { - paint("34", s) -} - -pub fn cyan(s: &str) -> String { - paint("36", s) -} - -pub fn err(msg: &str) { - eprintln!("{}", red(&format!("Error: {msg}"))); -} - -pub fn warn(msg: &str) { - eprintln!("{}", yellow(&format!("Warning: {msg}"))); -} - -pub fn info(msg: &str) { - println!("{}", blue(msg)); -} - -pub fn ok(msg: &str) { - println!("{}", green(msg)); -} +pub fn err(msg: &str) { eprintln!("{}", red(&format!("Error: {msg}"))); } +pub fn warn(msg: &str) { eprintln!("{}", yellow(&format!("Warning: {msg}"))); } +pub fn info(msg: &str) { println!("{}", blue(msg)); } +pub fn ok(msg: &str) { println!("{}", green(msg)); }