From e3220da152f46ff4fd791069d4f3b3dc16c8219c Mon Sep 17 00:00:00 2001 From: Ogulcan Celik Date: Mon, 14 Sep 2026 02:10:31 +0300 Subject: [PATCH] fix: clean up idle ssh bridges and back off flapping connections --- .../src/content/docs/connecting-machines.mdx | 4 +- justfile | 4 + scripts/smoke_ssh_bridge_liveness.mjs | 160 +++++++++++++++++ src/cli/status.rs | 2 + src/client/endpoint/supervisor.rs | 58 +++++- src/main.rs | 2 +- src/platform/fallback.rs | 8 +- src/platform/linux.rs | 2 + src/platform/macos.rs | 2 + src/platform/mod.rs | 7 + src/platform/remote_bridge.rs | 135 ++++++++++++++ src/platform/remote_bridge_tests.rs | 168 ++++++++++++++++++ src/platform/unix_common.rs | 28 ++- src/platform/windows.rs | 5 +- src/remote.rs | 2 +- src/remote/attach.rs | 52 +++++- src/remote/host.rs | 19 +- tests/client_mode.rs | 11 +- 18 files changed, 645 insertions(+), 24 deletions(-) create mode 100644 scripts/smoke_ssh_bridge_liveness.mjs create mode 100644 src/platform/remote_bridge.rs create mode 100644 src/platform/remote_bridge_tests.rs diff --git a/docs/next/website/src/content/docs/connecting-machines.mdx b/docs/next/website/src/content/docs/connecting-machines.mdx index a8497270..6c9e7a55 100644 --- a/docs/next/website/src/content/docs/connecting-machines.mdx +++ b/docs/next/website/src/content/docs/connecting-machines.mdx @@ -73,10 +73,12 @@ Removing or disabling the machine you are viewing returns you to Local. If Local ## Connection problems -- **Reconnecting:** Herdr retries with bounded backoff after a network interruption, sleep, or SSH failure. SSH connections are checked for application-level activity and probed when quiet, so a broken connection does not stay Online indefinitely. Local detects native connection closure or failure instead of using remote health probes. +- **Reconnecting:** Herdr retries automatically after a network interruption, sleep, or SSH failure. Repeated failures increase the delay up to two minutes; brief successful connections do not reset it. A connection must remain healthy for a minute before the next interruption gets a fast retry. SSH connections are checked for application-level activity and probed when quiet, so a broken connection does not stay Online indefinitely. Local detects native connection closure or failure instead of using remote health probes. - **Attention:** The target needs an action that cannot be completed in the background, such as host-key approval, authentication, or a compatible server. Other machines remain usable. - **Saved-machine file error:** An unreadable or invalid catalog leaves current connections unchanged. Herdr shows a notice and automatically retries reading it. +When both installations support bridge idle cleanup, saved-machine connections to Linux and macOS also close their remote bridge after a minute without traffic in either direction. Sleep counts toward that deadline, which is checked when the host wakes. Quiet healthy connections exchange health checks; watching continuous output does not require typing. Cleanup leaves the remote Herdr server, sessions, and pane processes running. Older installations remain compatible without this optional cleanup. + Background connections never answer prompts or install, update, restart, or hand off a server. For Attention, run the standalone setup command shown by Herdr in an interactive terminal, for example: ```bash diff --git a/justfile b/justfile index 77bf48d5..8102454c 100644 --- a/justfile +++ b/justfile @@ -19,6 +19,10 @@ maintenance-test: test-one filter: cargo nextest run --locked "{{filter}}" --status-level fail --final-status-level fail --failure-output final --success-output never +# Rootless Linux SSH teardown smoke, comparing an old and a candidate binary +smoke-ssh-bridge-liveness before after: + bun scripts/smoke_ssh_bridge_liveness.mjs "{{before}}" "{{after}}" + # Enforce deterministic UI hot-path architecture boundaries ui-hot-path-architecture-test: {{python}} -m unittest scripts.test_ui_hot_path_architecture diff --git a/scripts/smoke_ssh_bridge_liveness.mjs b/scripts/smoke_ssh_bridge_liveness.mjs new file mode 100644 index 00000000..71922af6 --- /dev/null +++ b/scripts/smoke_ssh_bridge_liveness.mjs @@ -0,0 +1,160 @@ +#!/usr/bin/env bun +// Linux-only, rootless network-namespace smoke. No host firewall or SSH configuration changes. +import { spawn, spawnSync } from 'node:child_process'; +import { mkdirSync, mkdtempSync, readlinkSync, rmSync, writeFileSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { userInfo } from 'node:os'; +import { fileURLToPath } from 'node:url'; + +const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); +function run(program, args, env = process.env) { + const result = spawnSync(program, args, { encoding: 'utf8', env }); + if (result.status !== 0) throw new Error(`${program}: ${result.stderr || result.error || result.status}`); + return result.stdout; +} +async function until(predicate, milliseconds, label) { + const deadline = Date.now() + milliseconds; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error(`timed out: ${label}`); + await delay(50); + } +} +function string(value) { + const bytes = Buffer.from(value); + const length = bytes.length < 251 ? Buffer.from([bytes.length]) : Buffer.from([251, bytes.length & 255, bytes.length >> 8]); + return Buffer.concat([length, bytes]); +} +function control(kind, data) { + // Generation-1 EndpointControl tag and length framing, also exercised in tests/support/mod.rs. + const payload = Buffer.concat([Buffer.from([20]), string(kind), string(data)]); + const size = Buffer.alloc(4); + size.writeUInt32LE(payload.length); + return Buffer.concat([size, payload]); +} +const quote = (text) => `'${text.replaceAll("'", "'\\''")}'`; +function cleanEnvironment(root) { + const env = Object.fromEntries(Object.entries(process.env).filter(([key]) => !key.startsWith('HERDR_'))); + Object.assign(env, { + XDG_CONFIG_HOME: `${root}/config`, XDG_STATE_HOME: `${root}/state`, XDG_RUNTIME_DIR: `${root}/run`, + HERDR_CONFIG_PATH: `${root}/config.toml`, HERDR_SOCKET_PATH: `${root}/api.sock`, HERDR_CLIENT_SOCKET_PATH: `${root}/client.sock`, + HERDR_DISABLE_SOUND: '1', + }); + for (const path of [env.XDG_CONFIG_HOME, env.XDG_STATE_HOME, env.XDG_RUNTIME_DIR]) mkdirSync(path, { recursive: true }); + writeFileSync(env.HERDR_CONFIG_PATH, '[experimental]\nallow_nested = true\n'); + return env; +} + +async function smoke(before, after) { + const root = mkdtempSync('/var/tmp/herdr-ssh-liveness-'); + const children = []; + const bridges = new Set(); + const launch = (program, args, options = {}) => { + const child = spawn(program, args, { stdio: ['ignore', 'inherit', 'inherit'], ...options }); + children.push(child); + return child; + }; + let success = false; + try { + run('ip', ['link', 'set', 'lo', 'up']); + run('ip', ['addr', 'add', '10.77.0.2/32', 'dev', 'lo']); + run('nft', ['add', 'table', 'inet', 'herdr_smoke']); + run('nft', ['add', 'chain', 'inet', 'herdr_smoke', 'output', '{ type filter hook output priority 0; policy accept; }']); + for (const key of ['host', 'client']) run('ssh-keygen', ['-q', '-t', 'ed25519', '-N', '', '-f', `${root}/${key}`]); + writeFileSync(`${root}/sshd_config`, `Port 22222\nListenAddress 10.77.0.2\nHostKey ${root}/host\nAuthorizedKeysFile ${root}/client.pub\nPidFile ${root}/sshd.pid\nStrictModes no\nPasswordAuthentication no\nKbdInteractiveAuthentication no\nUsePAM no\nClientAliveInterval 0\nTCPKeepAlive no\nLogLevel ERROR\n`); + const sshd = launch(Bun.which('sshd'), ['-D', '-e', '-f', `${root}/sshd_config`]); + await delay(400); + if (sshd.exitCode !== null) throw new Error('disposable sshd failed'); + const sockets = () => run('ss', ['-Hnt', 'sport', '=', '22222']).split('\n').filter((line) => line.startsWith('ESTAB')); + const alive = (pid) => { try { process.kill(pid, 0); return true; } catch { return false; } }; + + for (const [label, binary, expectedCleanup] of [['before', before, false], ['after', after, true]]) { + const phase = `${root}/${label}`; + mkdirSync(phase); + const env = cleanEnvironment(phase); + const server = launch(binary, ['--session', 'ssh-smoke', 'server'], { env }); + await until(() => { + try { return JSON.parse(run(binary, ['--session', 'ssh-smoke', 'status', 'server', '--json'], env)).running; } + catch { return false; } + }, 10000, 'isolated Herdr server'); + const connections = []; + for (let index = 0; index < 3; index++) { + const remoteEnv = Object.entries(env).filter(([key]) => key.startsWith('HERDR_') || key.startsWith('XDG_')).map(([key, value]) => `${key}=${quote(value)}`).join(' '); + const command = `printf 'BRIDGE %s %s\\n' "$$" "$PPID"; exec env ${remoteEnv} ${quote(binary)} --session ssh-smoke remote-client-bridge --idle-timeout-v1`; + const ssh = launch('ssh', ['-F', '/dev/null', '-o', 'StrictHostKeyChecking=no', '-o', 'UserKnownHostsFile=/dev/null', '-o', 'BatchMode=yes', '-o', 'IdentitiesOnly=yes', '-o', 'IdentityAgent=none', '-i', `${root}/client`, '-p', '22222', `${userInfo().username}@10.77.0.2`, command], { stdio: ['pipe', 'pipe', 'inherit'] }); + let output = Buffer.alloc(0); + let welcome = false; + let ids; + ssh.stdout.on('data', (chunk) => { + output = Buffer.concat([output, chunk]); + if (!ids) { + const end = output.indexOf(10); + if (end < 0) return; + const match = output.subarray(0, end).toString().match(/^BRIDGE (\d+) (\d+)$/); + if (!match) throw new Error('invalid bridge marker'); + ids = match.slice(1).map(Number); + bridges.add(ids[0]); + output = output.subarray(end + 1); + } + while (output.length >= 4 && output.length >= 4 + output.readUInt32LE(0)) { + const size = output.readUInt32LE(0); + const frame = output.subarray(4, 4 + size); + if (frame.includes(Buffer.from('endpoint.welcome.v1'))) welcome = true; + output = output.subarray(4 + size); + } + }); + await until(() => ids, 5000, 'bridge marker'); + ssh.stdin.write(control('endpoint.hello.v1', JSON.stringify({ generation: 1, cell_width_px: 8, cell_height_px: 16, surface_size: { cols: 80, rows: 24 }, pixel_mouse: false, direct_graphics: false, endpoint_keybindings: false, mouse_capture: false, surface_active: false, snapshot_codecs: ['shell.snapshot.v1'], surface_codecs: ['shell.surface.v1'], input_codecs: ['shell.input.semantic.v1'], blob_codecs: ['shell.blob.v1'] }))); + await until(() => welcome, 5000, 'real endpoint handshake'); + connections.push({ ssh, ids }); + } + await delay(1000); + console.log(`${label}: three real Herdr handshakes; suppressing disconnect packets`); + run('nft', ['add', 'rule', 'inet', 'herdr_smoke', 'output', 'tcp', 'dport', '22222', 'drop']); + for (const { ssh } of connections) ssh.kill('SIGKILL'); + await delay(200); + run('ss', ['-K', 'dst', '10.77.0.2', 'dport', '=', '22222']); + if (run('ss', ['-Hnt', 'dst', '10.77.0.2', 'dport', '=', '22222']).trim()) throw new Error('desktop sockets remain'); + if (sockets().length !== 3) throw new Error('failed to reproduce three abandoned remote SSH transports'); + run('nft', ['flush', 'chain', 'inet', 'herdr_smoke', 'output']); + const started = Date.now(); + if (expectedCleanup) { + await until(() => sockets().length === 0 && connections.every(({ ids }) => ids.every((pid) => !alive(pid))), 75000, 'all bridges AND SSH sessions reclaimed'); + console.log(`PASS after: 3/3 bridges and SSH sessions reclaimed in ${((Date.now() - started) / 1000).toFixed(1)}s`); + } else { + await delay(65000); + if (sockets().length !== 3 || !connections.every(({ ids }) => ids.every(alive))) throw new Error('baseline did not retain the abandoned connections'); + console.log('RED before: 3/3 bridges and SSH sessions still retained after 65s'); + // Only these recorded disposable bridge PIDs are terminated, never sshd or a main server. + for (const { ids } of connections) process.kill(ids[0], 'SIGTERM'); + await until(() => sockets().length === 0, 5000, 'baseline cleanup'); + } + for (const { ids } of connections) bridges.delete(ids[0]); + if (server.exitCode !== null || sshd.exitCode !== null) throw new Error('server died'); + run(binary, ['--session', 'ssh-smoke', 'api', 'snapshot'], env); + console.log(`${label}: isolated Herdr server still responds`); + server.kill('SIGTERM'); + await until(() => server.exitCode !== null, 5000, 'isolated server shutdown'); + } + success = true; + console.log('PASS: real SSH + Herdr, lost teardown, three concurrent connections, server survival'); + } finally { + for (const pid of bridges) { try { process.kill(pid, 'SIGTERM'); } catch {} } + for (const child of children.reverse()) { if (child.exitCode === null && child.signalCode === null) child.kill('SIGTERM'); } + await delay(300); + for (const child of children) { if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL'); } + if (success) rmSync(root, { recursive: true, force: true }); + else console.error(`Smoke evidence retained at ${root}`); + } +} + +const args = process.argv.slice(2); +if (args[0] !== '--isolated') { + if (args.length !== 2 || process.platform !== 'linux') { + throw new Error('usage: bun scripts/smoke_ssh_bridge_liveness.mjs (Linux; requires unshare, nft, ss, sshd, ssh-keygen)'); + } + const child = spawn('unshare', ['-Ucn', '--keep-caps', process.execPath, fileURLToPath(import.meta.url), '--isolated', ...args.map((path) => resolve(path))], { stdio: 'inherit' }); + child.on('exit', (code) => process.exit(code ?? 1)); +} else { + if (readlinkSync('/proc/self/ns/net') === readlinkSync('/proc/1/ns/net')) throw new Error('refusing host network namespace'); + await smoke(args[1], args[2]); +} diff --git a/src/cli/status.rs b/src/cli/status.rs index 99f278cf..d9c0be79 100644 --- a/src/cli/status.rs +++ b/src/cli/status.rs @@ -253,6 +253,7 @@ struct ClientStatusJson { endpoint_protocol_generation: u32, endpoint_capabilities: Vec<&'static str>, remote_host_bridge: bool, + remote_bridge_idle_timeout: bool, binary: String, session: Option, } @@ -299,6 +300,7 @@ fn client_status_json() -> ClientStatusJson { crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY, ], remote_host_bridge: true, + remote_bridge_idle_timeout: crate::platform::REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED, binary: current_exe_label(), session: crate::session::active_name(), } diff --git a/src/client/endpoint/supervisor.rs b/src/client/endpoint/supervisor.rs index 458ce927..a5154156 100644 --- a/src/client/endpoint/supervisor.rs +++ b/src/client/endpoint/supervisor.rs @@ -9,7 +9,9 @@ use crate::protocol::{ClientSurfaceSize, RenderEncoding}; use interprocess::TryClone as _; const INITIAL_RETRY_DELAY: Duration = Duration::from_millis(500); -const MAX_RETRY_DELAY: Duration = Duration::from_secs(30); +const MAX_RETRY_DELAY: Duration = Duration::from_secs(120); +const MAX_LOCAL_RETRY_DELAY: Duration = Duration::from_secs(30); +const STABLE_CONNECTION_PERIOD: Duration = Duration::from_secs(60); #[derive(Clone, Copy)] pub(crate) struct EndpointConnectOptions { @@ -51,6 +53,7 @@ struct ReconnectState { next_attempt: Option, in_flight: bool, generation: Option, + online_since: Option, } impl ReconnectState { @@ -61,6 +64,7 @@ impl ReconnectState { next_attempt: Some(now), in_flight: false, generation: None, + online_since: None, } } } @@ -200,15 +204,32 @@ impl EndpointSupervisors { state.in_flight = false; match status { ClientEndpointStatus::Online => { - state.attempts = 0; + if endpoint_id.is_local() { + state.attempts = 0; + } + state.online_since.get_or_insert(now); state.next_attempt = None; } ClientEndpointStatus::Attention | ClientEndpointStatus::Disabled => { - state.next_attempt = None + state.online_since = None; + state.next_attempt = None; } ClientEndpointStatus::Connecting | ClientEndpointStatus::Reconnecting => { + // A brief maintenance wake can complete a handshake without restoring the link. + if state.online_since.take().is_some_and(|connected| { + now.saturating_duration_since(connected) >= STABLE_CONNECTION_PERIOD + }) { + state.attempts = 0; + } state.attempts = state.attempts.saturating_add(1); - state.next_attempt = Some(now + retry_delay(state.attempts)); + let delay = retry_delay(state.attempts); + state.next_attempt = Some( + now + if endpoint_id.is_local() { + delay.min(MAX_LOCAL_RETRY_DELAY) + } else { + delay + }, + ); } } true @@ -334,7 +355,7 @@ fn retry_delay(attempt: u32) -> Duration { INITIAL_RETRY_DELAY .saturating_mul( 1_u32 - .checked_shl(attempt.saturating_sub(1).min(6)) + .checked_shl(attempt.saturating_sub(1).min(8)) .unwrap_or(u32::MAX), ) .min(MAX_RETRY_DELAY) @@ -425,6 +446,33 @@ mod tests { assert_eq!(supervisors.endpoints[&other_id].generation, Some(3)); } + #[test] + fn brief_ssh_reconnections_do_not_reset_backoff() { + let now = Instant::now(); + let profile = profile(); + let id = ClientEndpointId::Ssh(profile.id.clone()); + let mut supervisors = EndpointSupervisors::new(&[profile], now); + supervisors.endpoints.get_mut(&id).unwrap().generation = Some(2); + for attempt in 1..=5 { + let connected = now + Duration::from_secs(attempt * 20); + assert!(supervisors.record_status(&id, 2, ClientEndpointStatus::Online, connected)); + let failed = connected + Duration::from_secs(15); + assert!(supervisors.disconnected(&id, 2, failed)); + assert_eq!( + supervisors.endpoints[&id].next_attempt, + Some(failed + INITIAL_RETRY_DELAY * (1 << (attempt - 1))) + ); + } + let connected = now + Duration::from_secs(200); + assert!(supervisors.record_status(&id, 2, ClientEndpointStatus::Online, connected)); + let failed = connected + Duration::from_secs(60); + assert!(supervisors.disconnected(&id, 2, failed)); + assert_eq!( + supervisors.endpoints[&id].next_attempt, + Some(failed + INITIAL_RETRY_DELAY) + ); + } + #[test] fn retry_backoff_is_bounded() { assert_eq!(retry_delay(1), INITIAL_RETRY_DELAY); diff --git a/src/main.rs b/src/main.rs index 848c4e43..8cfe9b0b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -551,7 +551,7 @@ fn main() -> io::Result<()> { // Subcommands and flags (no TUI, no logging needed) if args.get(1).map(|s| s.as_str()) == Some("remote-client-bridge") { - return remote::run_remote_client_bridge(); + return remote::run_remote_client_bridge(&args[2..]); } if args.get(1).map(|s| s.as_str()) == Some("server") { diff --git a/src/platform/fallback.rs b/src/platform/fallback.rs index a533796f..21d13ba0 100644 --- a/src/platform/fallback.rs +++ b/src/platform/fallback.rs @@ -13,7 +13,13 @@ pub(crate) fn set_default_plugin_pane_pwd( ) { } -pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> { +#[cfg(unix)] +pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_MONOTONIC; + +pub(crate) fn forward_remote_bridge_stdio( + stream: crate::ipc::LocalStream, + _idle_timeout: bool, +) -> std::io::Result<()> { use interprocess::TryClone as _; let mut stdout = std::io::stdout().lock(); diff --git a/src/platform/linux.rs b/src/platform/linux.rs index d563512d..7ea0a2c6 100644 --- a/src/platform/linux.rs +++ b/src/platform/linux.rs @@ -7,6 +7,8 @@ use std::{ sync::OnceLock, }; +pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_BOOTTIME; + use super::{ read_limited_reader, ClipboardCommand, ClipboardImage, ForegroundJob, ForegroundProcess, LimitedRead, Signal, diff --git a/src/platform/macos.rs b/src/platform/macos.rs index 422bbfca..4e86deab 100644 --- a/src/platform/macos.rs +++ b/src/platform/macos.rs @@ -7,6 +7,8 @@ use std::process::{Command, Stdio}; use std::ptr::NonNull; use std::sync::OnceLock; +pub(super) const REMOTE_BRIDGE_CLOCK: libc::clockid_t = libc::CLOCK_MONOTONIC; + use super::{ read_limited_reader, ClipboardCommand, ClipboardImage, ForegroundJob, ForegroundProcess, LimitedRead, Signal, diff --git a/src/platform/mod.rs b/src/platform/mod.rs index 3efad45e..a578b42d 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -266,6 +266,13 @@ pub(crate) struct RemoteSshConfigPaths { pub(crate) multiplexing: bool, } +pub(crate) const REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED: bool = + cfg!(any(target_os = "linux", target_os = "macos")); + +#[cfg(unix)] +mod remote_bridge; +#[cfg(all(test, unix))] +mod remote_bridge_tests; #[cfg(unix)] mod unix_common; #[cfg(unix)] diff --git a/src/platform/remote_bridge.rs b/src/platform/remote_bridge.rs new file mode 100644 index 00000000..a8db4d29 --- /dev/null +++ b/src/platform/remote_bridge.rs @@ -0,0 +1,135 @@ +use std::io::{self, Read, Write}; +use std::sync::{ + atomic::{AtomicU64, Ordering}, + mpsc, Arc, +}; +use std::time::Duration; + +pub(crate) const IDLE_TIMEOUT: Duration = Duration::from_secs(60); + +fn now() -> io::Result { + let mut time = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // This clock includes suspend, so short maintenance wakes can reap old bridges. + if unsafe { libc::clock_gettime(super::REMOTE_BRIDGE_CLOCK, &mut time) } != 0 { + return Err(io::Error::last_os_error()); + } + Ok((time.tv_sec as u64) * 1_000_000_000 + time.tv_nsec as u64) +} + +#[derive(Clone)] +pub(super) struct Activity { + last: Arc, + _stop: mpsc::Sender<()>, +} + +impl Activity { + pub(super) fn start(timeout: Duration) -> io::Result { + let last = Arc::new(AtomicU64::new(now()?)); + let watched = Arc::clone(&last); + let (stop, stopped) = mpsc::channel(); + std::thread::Builder::new() + .name("ssh-bridge-liveness".into()) + .spawn(move || { + loop { + match stopped.recv_timeout(timeout.min(Duration::from_secs(1))) { + Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return, + Err(mpsc::RecvTimeoutError::Timeout) => {} + } + let expired = now().map_or(true, |now| idle_expired(&watched, now, timeout)); + if expired { + // Only the dedicated bridge process owns this watchdog. Returning from a + // blocked copy cannot guarantee shutdown, and joining it could hang forever. + std::process::exit(1); + } + } + })?; + Ok(Self { last, _stop: stop }) + } + + fn record(&self) -> io::Result<()> { + self.last.fetch_max(now()?, Ordering::Relaxed); + Ok(()) + } +} + +fn idle_expired(last: &AtomicU64, now: u64, timeout: Duration) -> bool { + Duration::from_nanos(now.saturating_sub(last.load(Ordering::Relaxed))) >= timeout +} + +pub(super) struct TrackedIo { + inner: T, + activity: Option, +} + +impl TrackedIo { + pub(super) fn new(inner: T, activity: Option) -> Self { + Self { inner, activity } + } + + fn progressed(&self, count: usize) -> io::Result<()> { + if count > 0 { + if let Some(activity) = &self.activity { + activity.record()?; + } + } + Ok(()) + } +} + +impl Read for TrackedIo { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + let count = self.inner.read(buffer)?; + self.progressed(count)?; + Ok(count) + } +} + +impl Write for TrackedIo { + fn write(&mut self, buffer: &[u8]) -> io::Result { + let count = self.inner.write(buffer)?; + self.progressed(count)?; + Ok(count) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn idle_deadline_counts_elapsed_sleep_and_only_positive_io_renews_it() { + let (stop, _stopped) = mpsc::channel(); + let last = Arc::new(AtomicU64::new(0)); + let activity = Activity { + last: Arc::clone(&last), + _stop: stop, + }; + assert!(idle_expired( + &last, + IDLE_TIMEOUT.as_nanos() as u64, + IDLE_TIMEOUT + )); + let mut reader = TrackedIo::new(&b"output"[..], Some(activity.clone())); + reader.read_exact(&mut [0; 6]).unwrap(); + let after_read = last.load(Ordering::Relaxed); + assert!(after_read > 0); + assert!(!idle_expired(&last, after_read, IDLE_TIMEOUT)); + assert_eq!(reader.read(&mut [0; 1]).unwrap(), 0); + assert_eq!(last.load(Ordering::Relaxed), after_read); + let mut writer = TrackedIo::new(Vec::new(), Some(activity)); + writer.write_all(b"ping").unwrap(); + assert!(last.load(Ordering::Relaxed) >= after_read); + assert!(idle_expired( + &last, + last.load(Ordering::Relaxed) + 120_000_000_000, + IDLE_TIMEOUT + )); + } +} diff --git a/src/platform/remote_bridge_tests.rs b/src/platform/remote_bridge_tests.rs new file mode 100644 index 00000000..ef4aade8 --- /dev/null +++ b/src/platform/remote_bridge_tests.rs @@ -0,0 +1,168 @@ +use std::io::{Read, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; +use std::process::{Child, Command, ExitStatus, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +const TIMEOUT: Duration = Duration::from_millis(300); + +#[test] +fn bridge_child() { + let Some(path) = std::env::var_os("HERDR_BRIDGE_TEST_SOCKET") else { + return; + }; + let stream = crate::ipc::connect_local_stream(&PathBuf::from(path)).unwrap(); + let timeout = (std::env::var_os("HERDR_BRIDGE_TEST_LEGACY").is_none()).then_some(TIMEOUT); + super::unix_common::forward_remote_bridge_stdio_with_timeout(stream, timeout).unwrap(); +} + +struct Bridge { + child: Child, + stream: UnixStream, + path: PathBuf, +} + +impl Bridge { + fn start(legacy: bool) -> Self { + static NEXT: AtomicU64 = AtomicU64::new(0); + let path = std::env::temp_dir().join(format!( + "hbl-{}-{}.sock", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + let listener = UnixListener::bind(&path).unwrap(); + listener.set_nonblocking(true).unwrap(); + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args([ + "--exact", + "platform::remote_bridge_tests::bridge_child", + "--nocapture", + ]) + .env("HERDR_BRIDGE_TEST_SOCKET", &path) + .env_remove("HERDR_BRIDGE_TEST_LEGACY") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + if legacy { + command.env("HERDR_BRIDGE_TEST_LEGACY", "1"); + } + let mut child = command.spawn().unwrap(); + let deadline = Instant::now() + Duration::from_secs(3); + let stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) + if error.kind() == std::io::ErrorKind::WouldBlock + && Instant::now() < deadline => + { + std::thread::sleep(Duration::from_millis(5)); + } + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + let _ = std::fs::remove_file(&path); + panic!("bridge did not connect: {error}"); + } + } + }; + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + stream + .set_write_timeout(Some(Duration::from_millis(100))) + .unwrap(); + Self { + child, + stream, + path, + } + } + + fn wait(&mut self) -> ExitStatus { + let deadline = Instant::now() + Duration::from_secs(3); + loop { + if let Some(status) = self.child.try_wait().unwrap() { + return status; + } + assert!(Instant::now() < deadline, "bridge did not exit"); + std::thread::sleep(Duration::from_millis(10)); + } + } + + fn finish(&mut self) -> String { + drop(self.child.stdin.take()); + let mut input = Vec::new(); + self.stream.read_to_end(&mut input).unwrap(); + self.stream + .write_all(b"final-output-after-stdin-eof") + .unwrap(); + self.stream.shutdown(std::net::Shutdown::Write).unwrap(); + assert!(self.wait().success()); + let mut output = String::new(); + self.child + .stdout + .take() + .unwrap() + .read_to_string(&mut output) + .unwrap(); + output + } +} + +impl Drop for Bridge { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_file(&self.path); + } +} + +#[test] +fn bridge_expires_when_silent_or_stdout_is_blocked() { + for blocked in [false, true] { + let mut bridge = Bridge::start(false); + if blocked { + let buffer = vec![b'x'; 64 * 1024]; + let deadline = Instant::now() + Duration::from_secs(2); + while bridge.stream.write_all(&buffer).is_ok() { + assert!(Instant::now() < deadline, "failed to fill bridge stdout"); + } + } + assert_eq!(bridge.wait().code(), Some(1)); + } +} + +#[test] +fn bridge_preserves_one_way_progress_and_drains_after_stdin_eof() { + for upload in [false, true] { + let mut bridge = Bridge::start(false); + for _ in 0..12 { + if upload { + bridge + .child + .stdin + .as_mut() + .unwrap() + .write_all(b"ping") + .unwrap(); + bridge.stream.read_exact(&mut [0; 4]).unwrap(); + } else { + bridge.stream.write_all(b"output").unwrap(); + } + std::thread::sleep(Duration::from_millis(60)); + assert!(bridge.child.try_wait().unwrap().is_none()); + } + assert!(bridge.finish().contains("final-output-after-stdin-eof")); + } +} + +#[test] +fn legacy_bridge_has_no_idle_deadline() { + let mut bridge = Bridge::start(true); + std::thread::sleep(TIMEOUT * 2); + assert!(bridge.child.try_wait().unwrap().is_none()); + assert!(bridge.finish().contains("final-output-after-stdin-eof")); +} diff --git a/src/platform/unix_common.rs b/src/platform/unix_common.rs index a633f283..a937b863 100644 --- a/src/platform/unix_common.rs +++ b/src/platform/unix_common.rs @@ -121,15 +121,33 @@ pub(crate) fn wait_client_stream_readable(stream: &crate::ipc::LocalStream) -> s Ok(()) } -pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> { +pub(crate) fn forward_remote_bridge_stdio( + stream: crate::ipc::LocalStream, + idle_timeout: bool, +) -> std::io::Result<()> { + forward_remote_bridge_stdio_with_timeout( + stream, + idle_timeout.then_some(super::remote_bridge::IDLE_TIMEOUT), + ) +} + +pub(super) fn forward_remote_bridge_stdio_with_timeout( + stream: crate::ipc::LocalStream, + idle_timeout: Option, +) -> std::io::Result<()> { + use super::remote_bridge::{Activity, TrackedIo}; use interprocess::TryClone as _; - let mut stdout = std::io::stdout().lock(); - let mut socket_to_stdout = stream.try_clone()?; + let activity = idle_timeout.map(Activity::start).transpose()?; + let mut stdout = TrackedIo::new(std::io::stdout().lock(), activity.clone()); + let mut socket_to_stdout = TrackedIo::new(stream.try_clone()?, activity.clone()); let mut stdin_to_socket = stream; let _upload = std::thread::spawn(move || { - let mut stdin = std::io::stdin(); - let _ = copy_flush(&mut stdin, &mut stdin_to_socket); + let mut stdin = TrackedIo::new(std::io::stdin(), activity.clone()); + let _ = copy_flush( + &mut stdin, + &mut TrackedIo::new(&mut stdin_to_socket, activity), + ); let crate::ipc::LocalStream::UdSocket(stream) = stdin_to_socket; let _ = stream.inner().shutdown(std::net::Shutdown::Write); }); diff --git a/src/platform/windows.rs b/src/platform/windows.rs index 8344be2d..879778cc 100644 --- a/src/platform/windows.rs +++ b/src/platform/windows.rs @@ -52,7 +52,10 @@ pub(crate) fn wait_client_stream_readable( Ok(()) } -pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> { +pub(crate) fn forward_remote_bridge_stdio( + stream: crate::ipc::LocalStream, + _idle_timeout: bool, +) -> std::io::Result<()> { use interprocess::TryClone as _; use std::sync::atomic::{AtomicBool, Ordering}; diff --git a/src/remote.rs b/src/remote.rs index c1c0f65f..a79ac774 100644 --- a/src/remote.rs +++ b/src/remote.rs @@ -23,7 +23,7 @@ pub(crate) fn run_remote_api_bridge(args: &[String]) -> std::io::Result<()> { ), ) })?; - crate::platform::forward_remote_bridge_stdio(stream) + crate::platform::forward_remote_bridge_stdio(stream, false) } [flag] if flag == "--check" => { println!("herdr-api-bridge-v1"); diff --git a/src/remote/attach.rs b/src/remote/attach.rs index ace72f84..6a1ee6f9 100644 --- a/src/remote/attach.rs +++ b/src/remote/attach.rs @@ -258,7 +258,16 @@ impl RemoteExecutable { } fn bridge_command(&self, session_name: &str) -> String { - let args = Self::session_args(session_name, &["remote-client-bridge"]); + self.bridge_command_with_idle_timeout(session_name, false) + } + + fn bridge_command_with_idle_timeout(&self, session_name: &str, idle_timeout: bool) -> String { + let command = if idle_timeout { + &["remote-client-bridge", "--idle-timeout-v1"][..] + } else { + &["remote-client-bridge"][..] + }; + let args = Self::session_args(session_name, command); match self { Self::PosixShellPath(_) => { posix_remote_output_command(&format!("exec {}", self.command(&args))) @@ -310,6 +319,7 @@ pub(super) struct RemoteHerdr { install_suffix: String, executable: RemoteExecutable, platform: RemotePlatform, + bridge_idle_timeout: bool, } impl RemoteHerdr { @@ -328,6 +338,7 @@ impl RemoteHerdr { install_suffix, executable, platform, + bridge_idle_timeout: false, } } @@ -1150,9 +1161,12 @@ pub(super) fn find_installed_remote_herdr(ssh: &RemoteSsh) -> io::Result, #[serde(default)] remote_host_bridge: bool, + #[serde(default)] + remote_bridge_idle_timeout: bool, } impl RemoteClientStatusJson { @@ -2438,7 +2454,13 @@ impl SshStdioBridge { ) -> io::Result { Self::start_command( target, - remote_herdr.executable.bridge_command(&session_name), + if noninteractive && remote_herdr.bridge_idle_timeout { + remote_herdr + .executable + .bridge_command_with_idle_timeout(&session_name, true) + } else { + remote_herdr.executable.bridge_command(&session_name) + }, local_socket, ssh_options, noninteractive, @@ -3669,6 +3691,7 @@ mod tests { crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY.into(), ], remote_host_bridge: false, + remote_bridge_idle_timeout: false, }; assert!(status.supports_endpoint_requirement(&linux, true)); for index in 0..status.endpoint_capabilities.len() { @@ -4300,6 +4323,25 @@ mod tests { } } + #[test] + fn remote_bridge_idle_timeout_requires_explicit_support_and_opt_in() { + let legacy = parse_client_status_json(r#"{"endpoint_protocol_generation":1}"#).unwrap(); + assert!(!legacy.remote_bridge_idle_timeout); + let current = parse_client_status_json( + r#"{"endpoint_protocol_generation":1,"remote_bridge_idle_timeout":true}"#, + ) + .unwrap(); + assert!(current.remote_bridge_idle_timeout); + let remote = RemoteHerdr::for_platform(RemotePlatform { + os: "linux", + arch: "x86_64", + }); + assert!(remote + .executable + .bridge_command_with_idle_timeout("agents", true) + .ends_with(" --session agents remote-client-bridge --idle-timeout-v1")); + } + #[test] fn remote_bridge_command_uses_installed_binary() { let remote_herdr = RemoteHerdr::for_platform(RemotePlatform { diff --git a/src/remote/host.rs b/src/remote/host.rs index 1ff6d1e5..08f5d122 100644 --- a/src/remote/host.rs +++ b/src/remote/host.rs @@ -3,7 +3,22 @@ use std::io; use std::time::Duration; -pub(crate) fn run_remote_client_bridge() -> io::Result<()> { +pub(crate) fn run_remote_client_bridge(args: &[String]) -> io::Result<()> { + let idle_timeout = match args { + [] => false, + [option] + if option == "--idle-timeout-v1" + && crate::platform::REMOTE_BRIDGE_IDLE_TIMEOUT_SUPPORTED => + { + true + } + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "unsupported remote client bridge option", + )) + } + }; ensure_remote_server_running()?; let socket_path = crate::server::socket_paths::client_socket_path(); @@ -17,7 +32,7 @@ pub(crate) fn run_remote_client_bridge() -> io::Result<()> { ) })?; - crate::platform::forward_remote_bridge_stdio(stream) + crate::platform::forward_remote_bridge_stdio(stream, idle_timeout) } fn ensure_remote_server_running() -> io::Result<()> { diff --git a/tests/client_mode.rs b/tests/client_mode.rs index 8f326484..d3205e5a 100644 --- a/tests/client_mode.rs +++ b/tests/client_mode.rs @@ -889,9 +889,10 @@ fn federated_client_starts_without_local_and_survives_its_restart() { std::os::unix::fs::symlink(env!("CARGO_BIN_EXE_herdr"), bin.join("herdr")).unwrap(); let quote = |path: &std::path::Path| format!("'{}'", path.display().to_string().replace('\'', "'\\''")); + let ssh_commands = base.join("ssh-commands"); fs::write(bin.join("ssh"), format!( - "#!/bin/sh\nexport HOME={} XDG_CONFIG_HOME={} XDG_RUNTIME_DIR={} HERDR_SOCKET_PATH={}\nunset HERDR_CLIENT_SOCKET_PATH HERDR_SESSION\nfor arg do last=\"$arg\"; done\nexec /bin/sh -c \"$last\"\n", - quote(&base.join("home")), quote(&remote_config), quote(&remote_runtime), quote(&remote_api), + "#!/bin/sh\nexport HOME={} XDG_CONFIG_HOME={} XDG_RUNTIME_DIR={} HERDR_SOCKET_PATH={}\nunset HERDR_CLIENT_SOCKET_PATH HERDR_SESSION\nfor arg do last=\"$arg\"; done\nprintf '%s\\n' \"$last\" >> {}\nexec /bin/sh -c \"$last\"\n", + quote(&base.join("home")), quote(&remote_config), quote(&remote_runtime), quote(&remote_api), quote(&ssh_commands), )).unwrap(); fs::set_permissions(bin.join("ssh"), fs::Permissions::from_mode(0o700)).unwrap(); let path = format!( @@ -914,6 +915,12 @@ fn federated_client_starts_without_local_and_survives_its_restart() { "remote must be usable before Local exists: {}", read_output(&output) ); + assert!( + fs::read_to_string(&ssh_commands) + .unwrap() + .contains("remote-client-bridge --idle-timeout-v1"), + "saved machine discovery must opt into the advertised bridge idle timeout" + ); let mut local = spawn_server(&config_home, &runtime_dir, &api_socket, &client_socket); wait_for_socket(&api_socket, Duration::from_secs(10));