diff --git a/src-tauri/src/commands/remote_backend.rs b/src-tauri/src/commands/remote_backend.rs index 424eccd2a..4138618dc 100644 --- a/src-tauri/src/commands/remote_backend.rs +++ b/src-tauri/src/commands/remote_backend.rs @@ -9,7 +9,7 @@ use tauri::{AppHandle, State}; use crate::services::remote_backend::{ self, daemon::RemoteDirListing, daemon::RemoteToolProbe, RemoteBackendConnection, - RemoteBackendError, RemoteBackendRegistry, RemoteBackendStatus, + RemoteBackendError, RemoteBackendErrorKind, RemoteBackendRegistry, RemoteBackendStatus, }; #[tauri::command] @@ -41,8 +41,30 @@ pub async fn remote_backend_disconnect( host: String, expected_generation: Option, ) -> Result<(), RemoteBackendError> { - remote_backend::disconnect_generation(&app, ®istry, &host, expected_generation); - Ok(()) + if remote_backend::disconnect_generation(&app, ®istry, &host, expected_generation) { + Ok(()) + } else if expected_generation.is_some() { + Err(RemoteBackendError::new( + RemoteBackendErrorKind::DaemonChanged, + "Remote backend changed before disconnect completed", + )) + } else { + Ok(()) + } +} + +#[tauri::command] +pub async fn remote_backend_forget( + registry: State<'_, RemoteBackendRegistry>, + host: String, +) -> Result<(), RemoteBackendError> { + if registry.forget(&host).await { + Ok(()) + } else { + Err(RemoteBackendError::internal( + "Cannot forget an active remote backend", + )) + } } #[tauri::command] @@ -51,8 +73,16 @@ pub async fn remote_backend_shutdown( registry: State<'_, RemoteBackendRegistry>, host: String, expected_instance_token: Option, + expected_generation: Option, ) -> Result<(), RemoteBackendError> { - remote_backend::shutdown(&app, ®istry, &host, expected_instance_token.as_deref()).await + remote_backend::shutdown( + &app, + ®istry, + &host, + expected_instance_token.as_deref(), + expected_generation, + ) + .await } #[tauri::command] diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 7736446a4..43a2ff65b 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -553,6 +553,7 @@ pub fn run() { commands::remote_backend::list_ssh_config_hosts, commands::remote_backend::remote_backend_connect, commands::remote_backend::remote_backend_disconnect, + commands::remote_backend::remote_backend_forget, commands::remote_backend::remote_backend_shutdown, commands::remote_backend::list_remote_backends, commands::remote_backend::check_remote_host, diff --git a/src-tauri/src/services/remote_backend/daemon.rs b/src-tauri/src/services/remote_backend/daemon.rs index e0b1a49e3..937feb429 100644 --- a/src-tauri/src/services/remote_backend/daemon.rs +++ b/src-tauri/src/services/remote_backend/daemon.rs @@ -759,6 +759,15 @@ mod tests { args: &[&str], path_override: Option<&str>, home: &std::path::Path, + ) -> std::process::Child { + spawn_script_source(args, path_override, home, BOOTSTRAP_SCRIPT) + } + + fn spawn_script_source( + args: &[&str], + path_override: Option<&str>, + home: &std::path::Path, + script: &str, ) -> std::process::Child { let nonce = "berd-test-nonce"; let mut command = StdCommand::new("bash"); @@ -781,7 +790,7 @@ mod tests { .stdin .take() .unwrap() - .write_all(BOOTSTRAP_SCRIPT.as_bytes()) + .write_all(script.as_bytes()) .unwrap(); child } @@ -797,6 +806,46 @@ mod tests { (lines, out.status.code()) } + fn legacy_lock_holder_script(marker: &str, release: &str) -> String { + format!( + r#"#!/usr/bin/env bash +set -u +STATE_DIR="${{XDG_STATE_HOME:-$HOME/.local/state}}/berd/remote" +LOCK_DIR="$STATE_DIR/daemon.lock" +LOCK_OWNER="$LOCK_DIR/owner" +b64() {{ printf %s "$1" | base64 | tr -d '\n'; }} +process_identity() {{ + if [ -r "/proc/$$/stat" ]; then + start="$(sed 's/^.*) //' "/proc/$$/stat" | awk '{{print $20}}')" + printf 'proc:%s' "$start" + else + ps -p "$$" -o lstart= -o command= | sed 's/^/ps:/' + fi +}} +mkdir -p "$STATE_DIR" +while ! mkdir "$LOCK_DIR" 2>/dev/null; do sleep 0.01; done +owner="$$ $(b64 "$(process_identity)")" +printf '%s\n' "$owner" >"$LOCK_OWNER" +: >"$STATE_DIR/{marker}" +while [ ! -f "$STATE_DIR/{release}" ]; do sleep 0.01; done +if [ "$(cat "$LOCK_OWNER" 2>/dev/null)" = "$owner" ]; then + rm -f "$LOCK_OWNER" + rmdir "$LOCK_DIR" 2>/dev/null || true +fi +"# + ) + } + + fn wait_for_path(path: &std::path::Path, message: &str) { + for _ in 0..500 { + if path.exists() { + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("{message}"); + } + fn b64_arg(value: &str) -> String { base64::engine::general_purpose::STANDARD.encode(value) } @@ -858,7 +907,9 @@ if [ "$1" = "serve" ]; then port="" while [ $# -gt 0 ]; do [ "$1" = "--port" ] && port="$2"; shift; done exec python3 -c 'import socket,sys,time -sys.stderr.write("x" * (5 * 1024 * 1024) + "\nBOUNDED-TAIL\n"); sys.stderr.flush() +for start in range(0, 80000, 1000): + sys.stderr.write("".join(f"{i:08d} " + "x" * 54 + "\n" for i in range(start, start + 1000))) +sys.stderr.flush() s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) s.bind(("127.0.0.1",int(sys.argv[1]))); s.listen(5); time.sleep(120)' "$port" fi @@ -895,6 +946,28 @@ fi path } + fn write_diagnostic_goose_shim(dir: &std::path::Path) -> std::path::PathBuf { + let path = dir.join("goose-diagnostic"); + std::fs::write( + &path, + r#"#!/usr/bin/env bash +if [ "$1" = "--version" ]; then echo "goose diagnostic"; exit 0; fi +if [ "$1" = "serve" ]; then + port="" + while [ $# -gt 0 ]; do [ "$1" = "--port" ] && port="$2"; shift; done + exec python3 -c 'import socket,sys,time +sys.stderr.write("IMPORTANT-STARTUP-DIAGNOSTIC\n"); sys.stderr.flush() +s=socket.socket(); s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) +s.bind(("127.0.0.1",int(sys.argv[1]))); s.listen(5); time.sleep(120)' "$port" +fi +"#, + ) + .unwrap(); + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + fn read_record_fields(home: &std::path::Path) -> Vec { let record = std::fs::read_to_string( home.join(".state") @@ -1144,26 +1217,174 @@ fi } #[test] - fn synchronized_reclaimers_do_not_remove_a_successor_lock() { + fn a_crashed_legacy_reclaimer_does_not_block_future_daemon_operations() { let home = tempfile::tempdir().unwrap(); let state_dir = home.path().join(".state/berd/remote"); let lock_dir = state_dir.join("daemon.lock"); std::fs::create_dir_all(&lock_dir).unwrap(); std::fs::write(lock_dir.join("owner"), "999999 invalid-identity").unwrap(); - let reclaimers = (0..6) + let paused_marker = state_dir.join("reclaim-paused"); + let paused_script = BOOTSTRAP_SCRIPT.replace( + " if mv \"$LEGACY_LOCK_DIR\" \"$legacy_claim\" 2>/dev/null; then", + " : > \"$STATE_DIR/reclaim-paused\"\n while [ ! -f \"$STATE_DIR/reclaim-continue\" ]; do sleep 0.01; done\n if mv \"$LEGACY_LOCK_DIR\" \"$legacy_claim\" 2>/dev/null; then", + ); + assert_ne!(paused_script, BOOTSTRAP_SCRIPT); + let mut reclaimer = + spawn_script_source(&["shutdown"], None, home.path(), &paused_script); + for _ in 0..500 { + if paused_marker.exists() { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + paused_marker.exists(), + "reclaimer never reached the stale legacy generation" + ); + assert!(StdCommand::new("kill") + .arg("-KILL") + .arg(reclaimer.id().to_string()) + .status() + .unwrap() + .success()); + assert_eq!(reclaimer.wait().unwrap().code(), None); + + // Neither the stale legacy lock nor the ownerless reclaim mutex + // used by the previous implementation participates in the unique + // ticket protocol. + std::fs::create_dir(state_dir.join("daemon.lock.reclaim")).unwrap(); + let (lines, code) = run_script(&["shutdown"], None, home.path()); + assert_eq!(code, Some(0), "lines: {lines:?}"); + assert!(lines.iter().any(|line| line == "STOPPED")); + } + + #[test] + fn legacy_reclaimer_does_not_claim_a_live_successor_generation() { + let home = tempfile::tempdir().unwrap(); + let state_dir = home.path().join(".state/berd/remote"); + let lock_dir = state_dir.join("daemon.lock"); + std::fs::create_dir_all(&lock_dir).unwrap(); + std::fs::write(lock_dir.join("owner"), "999999 invalid-identity").unwrap(); + + let paused_script = BOOTSTRAP_SCRIPT.replace( + " legacy_guard=\"$LEGACY_LOCK_DIR/.berd-reclaim.$NONCE.$$.$compat_attempt\"", + " : > \"$STATE_DIR/reclaim-paused\"\n while [ ! -f \"$STATE_DIR/reclaim-continue\" ]; do sleep 0.01; done\n legacy_guard=\"$LEGACY_LOCK_DIR/.berd-reclaim.$NONCE.$$.$compat_attempt\"", + ); + assert_ne!(paused_script, BOOTSTRAP_SCRIPT); + let reclaimer = spawn_script_source(&["shutdown"], None, home.path(), &paused_script); + wait_for_path( + &state_dir.join("reclaim-paused"), + "reclaimer never paused before generation claim", + ); + + std::fs::remove_file(lock_dir.join("owner")).unwrap(); + std::fs::remove_dir(&lock_dir).unwrap(); + let successor_source = legacy_lock_holder_script("successor-held", "successor-release"); + let successor = + spawn_script_source(&["shutdown"], None, home.path(), &successor_source); + wait_for_path( + &state_dir.join("successor-held"), + "live successor never acquired legacy lock", + ); + + std::fs::write(state_dir.join("reclaim-continue"), "").unwrap(); + std::thread::sleep(Duration::from_millis(250)); + assert!( + lock_dir.join("owner").exists(), + "successor owner was reclaimed" + ); + + std::fs::write(state_dir.join("successor-release"), "").unwrap(); + assert_eq!(successor.wait_with_output().unwrap().status.code(), Some(0)); + let (lines, code) = collect_script(reclaimer); + assert_eq!(code, Some(0), "lines: {lines:?}"); + assert!(lines.iter().any(|line| line == "STOPPED")); + } + + #[test] + fn legacy_holder_blocks_ticket_client_until_release() { + let home = tempfile::tempdir().unwrap(); + let state_dir = home.path().join(".state/berd/remote"); + let legacy_source = legacy_lock_holder_script("legacy-held", "legacy-release"); + let legacy = spawn_script_source(&["shutdown"], None, home.path(), &legacy_source); + wait_for_path( + &state_dir.join("legacy-held"), + "legacy client never acquired", + ); + + let ticket_source = BOOTSTRAP_SCRIPT.replace( + " lock_held=1\n return 0", + " lock_held=1\n : > \"$STATE_DIR/ticket-held\"\n while [ ! -f \"$STATE_DIR/ticket-release\" ]; do sleep 0.01; done\n return 0", + ); + assert_ne!(ticket_source, BOOTSTRAP_SCRIPT); + let ticket = spawn_script_source(&["shutdown"], None, home.path(), &ticket_source); + std::thread::sleep(Duration::from_millis(250)); + assert!( + !state_dir.join("ticket-held").exists(), + "ticket client overlapped a live legacy holder" + ); + std::fs::write(state_dir.join("legacy-release"), "").unwrap(); + assert_eq!(legacy.wait_with_output().unwrap().status.code(), Some(0)); + wait_for_path( + &state_dir.join("ticket-held"), + "ticket client never acquired", + ); + std::fs::write(state_dir.join("ticket-release"), "").unwrap(); + let (lines, code) = collect_script(ticket); + assert_eq!(code, Some(0), "lines: {lines:?}"); + } + + #[test] + fn ticket_holder_blocks_legacy_client_until_release() { + let home = tempfile::tempdir().unwrap(); + let state_dir = home.path().join(".state/berd/remote"); + let ticket_source = BOOTSTRAP_SCRIPT.replace( + " lock_held=1\n return 0", + " lock_held=1\n : > \"$STATE_DIR/ticket-held\"\n while [ ! -f \"$STATE_DIR/ticket-release\" ]; do sleep 0.01; done\n return 0", + ); + assert_ne!(ticket_source, BOOTSTRAP_SCRIPT); + let ticket = spawn_script_source(&["shutdown"], None, home.path(), &ticket_source); + wait_for_path( + &state_dir.join("ticket-held"), + "ticket client never acquired", + ); + + let legacy_source = legacy_lock_holder_script("legacy-held", "legacy-release"); + let legacy = spawn_script_source(&["shutdown"], None, home.path(), &legacy_source); + std::thread::sleep(Duration::from_millis(250)); + assert!( + !state_dir.join("legacy-held").exists(), + "legacy client overlapped a live ticket holder" + ); + + std::fs::write(state_dir.join("ticket-release"), "").unwrap(); + let (lines, code) = collect_script(ticket); + assert_eq!(code, Some(0), "lines: {lines:?}"); + wait_for_path( + &state_dir.join("legacy-held"), + "legacy client never acquired", + ); + std::fs::write(state_dir.join("legacy-release"), "").unwrap(); + assert_eq!(legacy.wait_with_output().unwrap().status.code(), Some(0)); + } + + #[test] + fn synchronized_ticket_holders_serialize_daemon_mutations() { + let home = tempfile::tempdir().unwrap(); + let state_dir = home.path().join(".state/berd/remote"); + + let holders = (0..6) .map(|_| spawn_script(&["shutdown"], None, home.path())) .collect::>(); - for child in reclaimers { + for child in holders { let (lines, code) = collect_script(child); assert_eq!(code, Some(0), "lines: {lines:?}"); } - assert!(!lock_dir.exists(), "lock remained after all shutdowns"); - let leftovers = std::fs::read_dir(&state_dir) + let leftovers = std::fs::read_dir(state_dir.join("daemon.locks")) .unwrap() .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) - .filter(|name| name.contains("daemon.lock")) .collect::>(); assert!( leftovers.is_empty(), @@ -1223,11 +1444,25 @@ fi let bin = tempfile::tempdir().unwrap(); let goose = write_noisy_goose_shim(bin.path()); let goose_arg = b64_arg(&goose.to_string_lossy()); + let state_dir = home.path().join(".state/berd/remote"); + let log_path = state_dir.join("goose-serve.log"); + std::fs::create_dir_all(log_path.parent().unwrap()).unwrap(); + std::fs::write(&log_path, b"!").unwrap(); - let (lines, code) = run_script(&["ensure", "-", &goose_arg], None, home.path()); + let instrumented_script = BOOTSTRAP_SCRIPT.replace( + " if ! tail -c \"$LOG_RETAIN_BYTES\" \"$writer_log\" >\"$writer_tmp\" 2>/dev/null; then", + " printf x >>\"$STATE_DIR/log-rotations\"\n if ! tail -c \"$LOG_RETAIN_BYTES\" \"$writer_log\" >\"$writer_tmp\" 2>/dev/null; then", + ); + assert_ne!(instrumented_script, BOOTSTRAP_SCRIPT); + let (lines, code) = collect_script(spawn_script_source( + &["ensure", "-", &goose_arg], + None, + home.path(), + &instrumented_script, + )); assert_eq!(code, Some(0), "lines: {lines:?}"); - let log_path = home.path().join(".state/berd/remote/goose-serve.log"); let mut log_len = 0; + let mut retained_ids = Vec::new(); for _ in 0..100 { let contents = std::fs::read(&log_path).unwrap_or_default(); log_len = contents.len() as u64; @@ -1235,7 +1470,12 @@ fi log_len <= 4 * 1024 * 1024, "log transiently grew to {log_len} bytes" ); - if contents.ends_with(b"BOUNDED-TAIL\n") { + retained_ids = contents + .split(|byte| *byte == b'\n') + .filter(|line| line.len() == 63 && line[8] == b' ') + .filter_map(|line| std::str::from_utf8(&line[..8]).ok()?.parse::().ok()) + .collect(); + if retained_ids.last() == Some(&79_999) { break; } std::thread::sleep(Duration::from_millis(20)); @@ -1243,10 +1483,24 @@ fi assert!(log_len > 0); assert!(log_len <= 4 * 1024 * 1024, "log grew to {log_len} bytes"); assert!( - std::fs::read(&log_path) - .unwrap_or_default() - .ends_with(b"BOUNDED-TAIL\n"), - "bounded writer did not retain the final record tail" + retained_ids.len() > 1, + "bounded writer did not retain complete sequence records" + ); + assert_eq!(retained_ids.last(), Some(&79_999)); + let first_gap = retained_ids + .windows(2) + .find(|pair| pair[1] != pair[0] + 1) + .map(|pair| (pair[0], pair[1])); + assert!( + first_gap.is_none(), + "bounded writer dropped a sequence record at {first_gap:?}" + ); + let rotations = std::fs::read(state_dir.join("log-rotations")) + .map(|bytes| bytes.len()) + .unwrap_or(0); + assert!( + rotations <= 2, + "5 MiB of output triggered {rotations} full-tail rewrites" ); let (_, code) = run_script(&["shutdown"], None, home.path()); @@ -1291,6 +1545,38 @@ fi assert_eq!(code, Some(0)); } + #[test] + fn daemon_log_publishes_a_short_diagnostic_while_producer_is_alive() { + if !python3_available() { + eprintln!("skipping: python3 unavailable for the goose serve shim"); + return; + } + let home = tempfile::tempdir().unwrap(); + let bin = tempfile::tempdir().unwrap(); + let goose = write_diagnostic_goose_shim(bin.path()); + let goose_arg = b64_arg(&goose.to_string_lossy()); + + let (lines, code) = run_script(&["ensure", "-", &goose_arg], None, home.path()); + assert_eq!(code, Some(0), "lines: {lines:?}"); + let log_path = home.path().join(".state/berd/remote/goose-serve.log"); + let mut visible = false; + for _ in 0..100 { + let contents = std::fs::read_to_string(&log_path).unwrap_or_default(); + if contents.contains("IMPORTANT-STARTUP-DIAGNOSTIC") { + visible = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + visible, + "live producer's startup diagnostic stayed buffered" + ); + + let (_, code) = run_script(&["shutdown"], None, home.path()); + assert_eq!(code, Some(0)); + } + #[test] fn shutdown_refuses_a_changed_daemon_generation() { if !python3_available() { diff --git a/src-tauri/src/services/remote_backend/mod.rs b/src-tauri/src/services/remote_backend/mod.rs index 324675c22..f4b9c016e 100644 --- a/src-tauri/src/services/remote_backend/mod.rs +++ b/src-tauri/src/services/remote_backend/mod.rs @@ -84,6 +84,8 @@ pub enum RemoteBackendState { #[serde(rename_all = "camelCase")] pub struct RemoteBackendStatus { pub host: String, + pub incarnation: String, + pub generation: u64, #[serde(flatten)] pub state: RemoteBackendState, } @@ -97,6 +99,9 @@ pub struct RemoteBackendConnection { pub local_port: u16, pub goose_version: String, pub daemon_reused: bool, + /// Unique identity for this registry slot. A forgotten host receives a + /// new incarnation even when its per-slot generation restarts. + pub incarnation: String, /// Slot generation that owns this tunnel. Callers use it to invalidate /// only the connection they established, never a newer replacement. pub generation: u64, @@ -110,6 +115,7 @@ pub struct RemoteBackendRegistry { struct HostSlot { key: String, spec: RemoteHostSpec, + incarnation: String, /// Serializes establish attempts (user connects and supervisor /// reconnects) per host. connect_lock: tokio::sync::Mutex<()>, @@ -118,6 +124,9 @@ struct HostSlot { struct SlotShared { state: RemoteBackendState, + /// Set before an inactive slot leaves the registry. Connect attempts that + /// already retained its Arc must observe this tombstone after admission. + forgotten: bool, /// Monotonic ownership token: each successful establish bumps it, and a /// supervisor only acts while its own generation is current. Explicit /// disconnects bump it to strand any racing supervisor. @@ -138,9 +147,11 @@ impl RemoteBackendRegistry { Arc::new(HostSlot { key: spec.key(), spec: spec.clone(), + incarnation: uuid::Uuid::new_v4().to_string(), connect_lock: tokio::sync::Mutex::new(()), shared: Mutex::new(SlotShared { state: RemoteBackendState::Disconnected, + forgotten: false, generation: 0, daemon: None, local_port: None, @@ -163,13 +174,54 @@ impl RemoteBackendRegistry { let slots = self.slots.lock().expect("remote backend registry poisoned"); slots .values() - .map(|slot| RemoteBackendStatus { - host: slot.key.clone(), - state: slot.shared.lock().expect("slot poisoned").state.clone(), + .map(|slot| { + let shared = slot.shared.lock().expect("slot poisoned"); + RemoteBackendStatus { + host: slot.key.clone(), + incarnation: slot.incarnation.clone(), + generation: shared.generation, + state: shared.state.clone(), + } }) .collect() } + /// Remove an inactive host slot from the registry. The exact key is used + /// deliberately so malformed inputs from a failed connect can still be + /// forgotten instead of having to pass host parsing again. + pub async fn forget(&self, key: &str) -> bool { + let Some(slot) = self.existing_slot(key) else { + return true; + }; + + // Invalidate queued connection admission before waiting on the lock. + // A current establish already holding the lock publishes Connecting, + // so only an inactive slot can reach this point; a waiter that retained + // the Arc must observe the tombstone when it eventually acquires. + { + let mut shared = slot.shared.lock().expect("slot poisoned"); + if !matches!( + shared.state, + RemoteBackendState::Disconnected | RemoteBackendState::Failed { .. } + ) { + return false; + } + shared.forgotten = true; + shared.generation += 1; + } + + let _guard = slot.connect_lock.lock().await; + let mut slots = self.slots.lock().expect("remote backend registry poisoned"); + let Some(registered) = slots.get(key) else { + return true; + }; + if !Arc::ptr_eq(registered, &slot) { + return false; + } + slots.remove(key); + true + } + /// Best-effort synchronous tunnel teardown for app exit. Daemons are left /// running deliberately: surviving the client is the feature. pub fn kill_all_tunnels(&self) { @@ -208,11 +260,12 @@ fn kill_tunnel_pid(pid: u32) { } fn set_state(app: &AppHandle, slot: &HostSlot, state: RemoteBackendState) { - { + let generation = { let mut shared = slot.shared.lock().expect("slot poisoned"); shared.state = state.clone(); - } - emit_status(app, &slot.key, &state); + shared.generation + }; + emit_status(app, slot, generation, &state); } fn update_state_if_current( @@ -240,13 +293,15 @@ fn set_state_if_current( if !update_state_if_current(&mut shared, generation, state.clone()) { return false; } - emit_status(app, &slot.key, &state); + emit_status(app, slot, generation, &state); true } -fn emit_status(app: &AppHandle, host: &str, state: &RemoteBackendState) { +fn emit_status(app: &AppHandle, slot: &HostSlot, generation: u64, state: &RemoteBackendState) { let payload = RemoteBackendStatus { - host: host.to_string(), + host: slot.key.clone(), + incarnation: slot.incarnation.clone(), + generation, state: state.clone(), }; if let Err(error) = app.emit(REMOTE_BACKEND_STATUS_EVENT, &payload) { @@ -281,7 +336,10 @@ fn extra_serve_args() -> Vec { } } -fn connection_from_shared(shared: &SlotShared) -> Option { +fn connection_from_shared( + shared: &SlotShared, + incarnation: &str, +) -> Option { let daemon = shared.daemon.as_ref()?; let local_port = shared.local_port?; if let RemoteBackendState::Ready { @@ -297,6 +355,7 @@ fn connection_from_shared(shared: &SlotShared) -> Option Option, + incarnation: &str, ) -> Option { if shared.goose_path.as_deref() != requested_goose_path { return None; } - connection_from_shared(shared) + connection_from_shared(shared, incarnation) +} + +async fn acquire_connect_admission( + slot: &HostSlot, +) -> Result, RemoteBackendError> { + let guard = slot.connect_lock.lock().await; + if slot.shared.lock().expect("slot poisoned").forgotten { + return Err(RemoteBackendError::internal( + "remote connection attempt was superseded", + )); + } + Ok(guard) } fn advance_generation_if_current(shared: &mut SlotShared, expected: u64) -> Option { @@ -346,11 +418,19 @@ pub async fn connect( let goose_path = goose_path.map(daemon::normalize_goose_path).transpose()?; let slot = registry.slot(&spec); - let _guard = slot.connect_lock.lock().await; + let _guard = acquire_connect_admission(&slot).await?; { let mut shared = slot.shared.lock().expect("slot poisoned"); - if let Some(existing) = cached_connection(&shared, goose_path.as_deref()) { + // Forget can tombstone an inactive slot after this connect acquires + // admission but before it publishes Connecting. + if shared.forgotten { + return Err(RemoteBackendError::internal( + "remote connection attempt was superseded", + )); + } + if let Some(existing) = cached_connection(&shared, goose_path.as_deref(), &slot.incarnation) + { return Ok(existing); } if shared.goose_path.as_deref() != goose_path.as_deref() { @@ -460,7 +540,7 @@ async fn establish( let _ = tunnel.child.wait().await; return Ok(None); }; - emit_status(app, &slot.key, &state); + emit_status(app, slot, generation, &state); let connection = RemoteBackendConnection { ws_url, @@ -469,6 +549,7 @@ async fn establish( local_port, goose_version: daemon_info.goose_version.clone(), daemon_reused: daemon_info.reused, + incarnation: slot.incarnation.clone(), generation, }; @@ -600,11 +681,6 @@ fn spawn_supervisor( }); } -/// Kill the tunnel; the remote daemon keeps running. -pub fn disconnect(app: &AppHandle, registry: &RemoteBackendRegistry, host_input: &str) { - disconnect_generation(app, registry, host_input, None); -} - /// Disconnect only when `expected_generation` still owns the host slot. This /// lets an initializer clean up work superseded while it was awaiting without /// tearing down a newer connection that won the race. @@ -622,7 +698,7 @@ pub fn disconnect_generation( let Some(slot) = registry.existing_slot(&host_key) else { return false; }; - { + let disconnected_generation = { let mut shared = slot.shared.lock().expect("slot poisoned"); if expected_generation.is_some_and(|expected| shared.generation != expected) { return false; @@ -632,31 +708,58 @@ pub fn disconnect_generation( kill_tunnel_pid(pid); } shared.local_port = None; - } - set_state(app, &slot, RemoteBackendState::Disconnected); + shared.generation + }; + set_state_if_current( + app, + &slot, + disconnected_generation, + RemoteBackendState::Disconnected, + ); record_diagnostic(DiagnosticLevel::Info, "disconnected", &host_key, None); true } +fn validate_shutdown_generation( + slot: &HostSlot, + expected_generation: Option, +) -> Result<(), RemoteBackendError> { + let shared = slot.shared.lock().expect("slot poisoned"); + if expected_generation.is_some_and(|expected| shared.generation != expected) { + return Err(RemoteBackendError::new( + RemoteBackendErrorKind::DaemonChanged, + "Remote backend changed before shutdown completed", + )); + } + Ok(()) +} + /// Stop the remote daemon, then drop the tunnel. pub async fn shutdown( app: &AppHandle, registry: &RemoteBackendRegistry, host_input: &str, expected_instance_token: Option<&str>, + expected_generation: Option, ) -> Result<(), RemoteBackendError> { let aliases = ssh_config::load_ssh_config_hosts(); let spec = RemoteHostSpec::parse(host_input, &aliases)?; let slot = registry.slot(&spec); let shell_env = dir_env::capture_home_interactive_env().await; - // Wait out any in-flight establish, then invalidate its supervisor and - // drop its tunnel before touching the daemon. A new connect cannot start - // until shutdown releases this lock, so ensure_daemon cannot recreate the - // daemon after shutdown_daemon stops it. + // Wait out any in-flight establish and validate the local owner before + // touching either resource. Keep the tunnel intact until remote shutdown + // succeeds: a daemon identity mismatch must not leave a dead local path + // represented by the still-current Ready state. let _guard = slot.connect_lock.lock().await; - disconnect(app, registry, &spec.key()); + validate_shutdown_generation(&slot, expected_generation)?; daemon::shutdown_daemon(&spec, &shell_env, expected_instance_token).await?; + if !disconnect_generation(app, registry, &spec.key(), expected_generation) { + return Err(RemoteBackendError::new( + RemoteBackendErrorKind::DaemonChanged, + "Remote backend changed before shutdown completed", + )); + } record_diagnostic(DiagnosticLevel::Info, "daemon_shutdown", &spec.key(), None); Ok(()) } @@ -714,10 +817,14 @@ mod tests { fn status_payload_flattens_state() { let status = RemoteBackendStatus { host: "devbox".to_string(), + incarnation: "slot-1".to_string(), + generation: 7, state: RemoteBackendState::Connecting, }; let json = serde_json::to_value(&status).unwrap(); assert_eq!(json["host"], "devbox"); + assert_eq!(json["incarnation"], "slot-1"); + assert_eq!(json["generation"], 7); assert_eq!(json["state"], "connecting"); } @@ -728,6 +835,7 @@ mod tests { http_base_url: "http://127.0.0.1:5000".to_string(), local_port: 5000, }, + forgotten: false, generation: 1, daemon: Some(RemoteDaemonInfo { pid: 10, @@ -745,10 +853,11 @@ mod tests { #[test] fn cached_connection_is_reused_for_the_same_goose_path() { - assert!(cached_connection(&ready_shared(None), None).is_some()); + assert!(cached_connection(&ready_shared(None), None, "slot-1").is_some()); assert!(cached_connection( &ready_shared(Some("/opt/goose/bin/goose")), - Some("/opt/goose/bin/goose") + Some("/opt/goose/bin/goose"), + "slot-1" ) .is_some()); } @@ -757,11 +866,18 @@ mod tests { fn cached_connection_is_refused_for_a_different_goose_path() { // Adding, removing, or swapping an override all force a fresh connect // because the bootstrap restarts the remote daemon. - assert!(cached_connection(&ready_shared(None), Some("/opt/goose/bin/goose")).is_none()); - assert!(cached_connection(&ready_shared(Some("/opt/goose/bin/goose")), None).is_none()); + assert!( + cached_connection(&ready_shared(None), Some("/opt/goose/bin/goose"), "slot-1") + .is_none() + ); + assert!( + cached_connection(&ready_shared(Some("/opt/goose/bin/goose")), None, "slot-1") + .is_none() + ); assert!(cached_connection( &ready_shared(Some("/opt/goose/bin/goose")), - Some("~/src/goose/target/release/goose") + Some("~/src/goose/target/release/goose"), + "slot-1" ) .is_none()); } @@ -770,7 +886,7 @@ mod tests { fn cached_connection_is_none_while_not_ready() { let mut shared = ready_shared(None); shared.state = RemoteBackendState::Connecting; - assert!(cached_connection(&shared, None).is_none()); + assert!(cached_connection(&shared, None, "slot-1").is_none()); } #[test] @@ -827,6 +943,25 @@ mod tests { assert!(matches!(shared.state, RemoteBackendState::Disconnected)); } + #[test] + fn shutdown_generation_mismatch_preserves_the_ready_tunnel() { + let spec = RemoteHostSpec::parse("devbox", &[]).unwrap(); + let slot = HostSlot { + key: spec.key(), + spec, + incarnation: "slot-1".to_string(), + connect_lock: tokio::sync::Mutex::new(()), + shared: Mutex::new(ready_shared(None)), + }; + + let error = validate_shutdown_generation(&slot, Some(2)).unwrap_err(); + assert_eq!(error.kind, RemoteBackendErrorKind::DaemonChanged); + let shared = slot.shared.lock().expect("slot poisoned"); + assert_eq!(shared.generation, 1); + assert_eq!(shared.tunnel_pid, Some(99)); + assert!(matches!(shared.state, RemoteBackendState::Ready { .. })); + } + #[test] fn snapshot_reports_registered_slots() { let registry = RemoteBackendRegistry::default(); @@ -836,4 +971,72 @@ mod tests { assert_eq!(snapshot.len(), 1); assert_eq!(snapshot[0].host, "devbox"); } + + #[tokio::test] + async fn forget_removes_inactive_slot_by_its_exact_key() { + let registry = RemoteBackendRegistry::default(); + let spec = RemoteHostSpec::parse("devbox", &[]).unwrap(); + let slot = registry.slot(&spec); + slot.shared.lock().expect("slot poisoned").state = RemoteBackendState::Failed { + error: RemoteBackendError::internal("failed"), + }; + + assert!(registry.forget("devbox").await); + assert!(registry.snapshot().is_empty()); + } + + #[tokio::test] + async fn forget_preserves_active_slot() { + let registry = RemoteBackendRegistry::default(); + let spec = RemoteHostSpec::parse("devbox", &[]).unwrap(); + let slot = registry.slot(&spec); + slot.shared.lock().expect("slot poisoned").state = RemoteBackendState::Connecting; + + assert!(!registry.forget("devbox").await); + assert_eq!(registry.snapshot().len(), 1); + } + + #[tokio::test] + async fn pending_connect_is_superseded_before_establishment_after_forget() { + let registry = Arc::new(RemoteBackendRegistry::default()); + let spec = RemoteHostSpec::parse("devbox", &[]).unwrap(); + let admitted = registry.slot(&spec); + let held = admitted.connect_lock.lock().await; + + let side_effect_reached = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let connect_slot = Arc::clone(&admitted); + let connect_side_effect = Arc::clone(&side_effect_reached); + let pending_connect = tokio::spawn(async move { + let admission = acquire_connect_admission(&connect_slot).await; + if admission.is_ok() { + connect_side_effect.store(true, std::sync::atomic::Ordering::SeqCst); + } + admission.map(|_| ()) + }); + tokio::task::yield_now().await; + + let forget_registry = Arc::clone(®istry); + let forgetting = tokio::spawn(async move { forget_registry.forget("devbox").await }); + for _ in 0..100 { + if admitted.shared.lock().expect("slot poisoned").forgotten { + break; + } + tokio::task::yield_now().await; + } + assert!(admitted.shared.lock().expect("slot poisoned").forgotten); + + drop(held); + let connect_error = pending_connect.await.unwrap().unwrap_err(); + assert_eq!( + connect_error.message, + "remote connection attempt was superseded" + ); + assert!(!side_effect_reached.load(std::sync::atomic::Ordering::SeqCst)); + assert!(forgetting.await.unwrap()); + assert!(registry.snapshot().is_empty()); + + let replacement = registry.slot(&spec); + assert!(!Arc::ptr_eq(&admitted, &replacement)); + assert!(!replacement.shared.lock().expect("slot poisoned").forgotten); + } } diff --git a/src-tauri/src/services/remote_backend/remote_daemon.sh b/src-tauri/src/services/remote_backend/remote_daemon.sh index b439a3ee6..65927a4b8 100644 --- a/src-tauri/src/services/remote_backend/remote_daemon.sh +++ b/src-tauri/src/services/remote_backend/remote_daemon.sh @@ -39,13 +39,13 @@ GOOSE_ARG="${4:--}" STATE_DIR="${XDG_STATE_HOME:-$HOME/.local/state}/berd/remote" RECORD="$STATE_DIR/daemon.record" LOG="$STATE_DIR/goose-serve.log" -LOCK_DIR="$STATE_DIR/daemon.lock" -LOCK_OWNER="$LOCK_DIR/owner" -LOCK_RECLAIM_DIR="$STATE_DIR/daemon.lock.reclaim" +LOCK_ROOT="$STATE_DIR/daemon.locks" +LEGACY_LOCK_DIR="$STATE_DIR/daemon.lock" +LEGACY_LOCK_OWNER="$LEGACY_LOCK_DIR/owner" RECORD_FORMAT="v4" LOG_MAX_BYTES=$((4 * 1024 * 1024)) LOG_RETAIN_BYTES=$((2 * 1024 * 1024)) -LOG_WRITE_CHUNK_BYTES=$((64 * 1024)) +LOG_SEGMENT_BYTES=$((LOG_RETAIN_BYTES - 64 * 1024)) emit() { printf '%s %s\n' "$NONCE" "$*"; } @@ -105,7 +105,8 @@ process_identity() { } lock_owner_is_current() { - IFS=' ' read -r lock_pid lock_b64identity lock_extra <"$LOCK_OWNER" 2>/dev/null || return 1 + owner_path="$1" + IFS=' ' read -r lock_pid lock_b64identity lock_extra <"$owner_path" 2>/dev/null || return 1 case "$lock_pid" in '' | *[!0-9]*) return 1 ;; esac @@ -117,21 +118,21 @@ lock_owner_is_current() { } release_daemon_lock() { - [ "${lock_held:-0}" = "1" ] || return 0 - if IFS= read -r current_lock_owner <"$LOCK_OWNER" 2>/dev/null && + if [ "${compat_lock_held:-0}" = "1" ] && + IFS= read -r current_compat_owner <"$LEGACY_LOCK_OWNER" 2>/dev/null && + [ "$current_compat_owner" = "$our_lock_owner" ]; then + rm -f "$LEGACY_LOCK_OWNER" + rmdir "$LEGACY_LOCK_DIR" 2>/dev/null || true + fi + compat_lock_held=0 + if IFS= read -r current_lock_owner <"$our_ticket/owner" 2>/dev/null && [ "$current_lock_owner" = "$our_lock_owner" ]; then - rm -f "$LOCK_OWNER" - rmdir "$LOCK_DIR" 2>/dev/null || true + rm -rf -- "$our_ticket" fi + [ -z "${pending_ticket:-}" ] || rm -rf -- "$pending_ticket" lock_held=0 } -release_reclaim_lock() { - [ "${reclaim_held:-0}" = "1" ] || return 0 - rmdir "$LOCK_RECLAIM_DIR" 2>/dev/null || true - reclaim_held=0 -} - terminate_uncommitted_daemon() { case "${uncommitted_pid:-}" in '' | *[!0-9]*) ;; @@ -166,45 +167,90 @@ terminate_uncommitted_daemon() { cleanup_daemon_mutation() { terminate_uncommitted_daemon release_daemon_lock - release_reclaim_lock } -# Claim one stale lock generation before deleting it. The separate reclaim -# mutex serializes observers: after its final owner re-check, no peer can -# remove the old directory and let a new generation appear before this process -# atomically renames it. Once renamed, cleanup is confined to the claimed path, -# so partial `.owner.*` publications are removed without touching a successor. -reclaim_stale_daemon_lock() { - if ! mkdir "$LOCK_RECLAIM_DIR" 2>/dev/null; then - return 1 - fi - reclaim_held=1 - - if lock_owner_is_current; then - release_reclaim_lock - return 1 - fi +# Ticket clients also hold daemon.lock for the full mutation. Older Berd +# clients know only this path, so retaining it as a compatibility gate makes +# mixed-version rollouts mutually exclusive in both acquisition orders. The +# ticket remains the generation-safe ordering protocol for current clients. +acquire_legacy_compat_lock() { + compat_attempt=0 + stale_observations=0 + while [ "$compat_attempt" -lt 1200 ]; do + compat_attempt=$((compat_attempt + 1)) + if mkdir "$LEGACY_LOCK_DIR" 2>/dev/null; then + compat_owner_tmp="$LEGACY_LOCK_DIR/.owner.$$" + if ! printf '%s\n' "$our_lock_owner" >"$compat_owner_tmp" || + ! mv -f "$compat_owner_tmp" "$LEGACY_LOCK_OWNER"; then + rm -f "$compat_owner_tmp" + rmdir "$LEGACY_LOCK_DIR" 2>/dev/null || true + stale_observations=0 + sleep 0.01 + continue + fi + # Let an observer that saw the directory before owner publication finish + # its claim, then prove this exact generation still occupies the path. + sleep 0.01 + if ! IFS= read -r current_compat_owner <"$LEGACY_LOCK_OWNER" 2>/dev/null || + [ "$current_compat_owner" != "$our_lock_owner" ]; then + stale_observations=0 + sleep 0.01 + continue + fi + compat_lock_held=1 + return 0 + fi - claimed_lock="$STATE_DIR/.daemon.lock.reclaimed.$$.$lock_attempt" - if mv "$LOCK_DIR" "$claimed_lock" 2>/dev/null; then - rm -rf -- "$claimed_lock" - fi - release_reclaim_lock - return 0 + if lock_owner_is_current "$LEGACY_LOCK_OWNER"; then + stale_observations=0 + else + stale_observations=$((stale_observations + 1)) + # A mkdir winner publishes its owner immediately. Repeated observations + # distinguish that window from a process that died before publication. + if [ "$stale_observations" -ge 20 ]; then + # Pin the directory generation before acting on it. Older clients + # release with rmdir, so this guard prevents the observed directory + # from disappearing and a live successor reusing the shared pathname. + # If replacement won before the guard, the owner recheck detects it. + legacy_guard="$LEGACY_LOCK_DIR/.berd-reclaim.$NONCE.$$.$compat_attempt" + if mkdir "$legacy_guard" 2>/dev/null; then + if lock_owner_is_current "$LEGACY_LOCK_OWNER"; then + rmdir "$legacy_guard" 2>/dev/null || true + else + legacy_claim="$STATE_DIR/.daemon.lock.legacy.$NONCE.$$.$compat_attempt" + if mv "$LEGACY_LOCK_DIR" "$legacy_claim" 2>/dev/null; then + rm -rf -- "$legacy_claim" + else + rmdir "$legacy_guard" 2>/dev/null || true + fi + fi + fi + stale_observations=0 + fi + fi + sleep 0.1 + done + emit "ERR state-dir" + exit 46 } -# `mkdir` is the portable cross-process atomic primitive available on both -# Linux and macOS remotes. The lock covers every daemon.record read/mutation, -# including shutdown from another Berd client using the same remote account. +# Each invocation owns a path that is never reused (`NONCE` is a UUID). Stale +# cleanup therefore removes the exact generation whose owner was inspected, +# never a successor at a shared pathname. A Lamport bakery ticket orders +# simultaneous contenders without introducing another reclaimable mutex: a +# process first publishes `choosing`, then a number, and may enter only after +# every live lower `(number, ticket-id)` pair has departed. acquire_daemon_lock() { umask 077 - mkdir -p "$STATE_DIR" || { + mkdir -p "$STATE_DIR" "$LOCK_ROOT" || { emit "ERR state-dir" exit 46 } lock_held=0 - reclaim_held=0 + compat_lock_held=0 + our_ticket="" + pending_ticket="" uncommitted_pid="" uncommitted_logger_pid="" uncommitted_log_pipe="" @@ -212,46 +258,101 @@ acquire_daemon_lock() { trap 'exit 129' HUP trap 'exit 130' INT trap 'exit 143' TERM - stale_observations=0 + case "$NONCE" in + '' | *[!A-Za-z0-9._-]*) + emit "ERR state-dir" + exit 46 + ;; + esac + our_identity="$(process_identity "$$")" || { + emit "ERR state-dir" + exit 46 + } + our_lock_owner="$$ $(b64 "$our_identity")" + + our_ticket_id="$NONCE.$$" + pending_ticket="$LOCK_ROOT/.pending.$our_ticket_id" + our_ticket="$LOCK_ROOT/ticket.$our_ticket_id" + rm -rf -- "$pending_ticket" + if ! mkdir "$pending_ticket" 2>/dev/null || + ! printf '%s\n' "$our_lock_owner" >"$pending_ticket/owner" || + ! : >"$pending_ticket/choosing" || + ! mv "$pending_ticket" "$our_ticket" 2>/dev/null; then + rm -rf -- "$pending_ticket" + emit "ERR state-dir" + exit 46 + fi + + max_ticket_number=0 + for observed_ticket in "$LOCK_ROOT"/ticket.*; do + [ -d "$observed_ticket" ] || continue + if ! lock_owner_is_current "$observed_ticket/owner"; then + rm -rf -- "$observed_ticket" + continue + fi + if IFS= read -r observed_number <"$observed_ticket/number" 2>/dev/null; then + case "$observed_number" in + '' | *[!0-9]*) ;; + *) + if [ "$observed_number" -gt "$max_ticket_number" ]; then + max_ticket_number="$observed_number" + fi + ;; + esac + fi + done + our_ticket_number=$((max_ticket_number + 1)) + if ! printf '%s\n' "$our_ticket_number" >"$our_ticket/.number.$$" || + ! mv "$our_ticket/.number.$$" "$our_ticket/number" 2>/dev/null || + ! rm -f "$our_ticket/choosing"; then + rm -rf -- "$our_ticket" + emit "ERR state-dir" + exit 46 + fi + lock_attempt=0 # A lock holder can spend about 80 seconds across five readiness attempts # and bounded cleanup. Wait longer than that critical section so a healthy # concurrent ensure cannot be mistaken for a wedged state-dir operation. while [ "$lock_attempt" -lt 1200 ]; do lock_attempt=$((lock_attempt + 1)) - if mkdir "$LOCK_DIR" 2>/dev/null; then - our_identity="$(process_identity "$$")" || { - rmdir "$LOCK_DIR" 2>/dev/null || true - emit "ERR state-dir" - exit 46 - } - our_lock_owner="$$ $(b64 "$our_identity")" - lock_owner_tmp="$LOCK_DIR/.owner.$$" - if ! printf '%s\n' "$our_lock_owner" >"$lock_owner_tmp" || - ! mv -f "$lock_owner_tmp" "$LOCK_OWNER"; then - rm -f "$lock_owner_tmp" - rmdir "$LOCK_DIR" 2>/dev/null || true - emit "ERR state-dir" - exit 46 + ticket_blocked=0 + for observed_ticket in "$LOCK_ROOT"/ticket.*; do + [ -d "$observed_ticket" ] || continue + [ "$observed_ticket" != "$our_ticket" ] || continue + if ! lock_owner_is_current "$observed_ticket/owner"; then + rm -rf -- "$observed_ticket" + continue fi + if [ -e "$observed_ticket/choosing" ]; then + ticket_blocked=1 + continue + fi + if ! IFS= read -r observed_number <"$observed_ticket/number" 2>/dev/null; then + ticket_blocked=1 + continue + fi + case "$observed_number" in + '' | *[!0-9]*) + ticket_blocked=1 + continue + ;; + esac + observed_id="${observed_ticket##*/ticket.}" + if [ "$observed_number" -lt "$our_ticket_number" ] || + { [ "$observed_number" -eq "$our_ticket_number" ] && [[ "$observed_id" < "$our_ticket_id" ]]; }; then + ticket_blocked=1 + fi + done + if [ "$ticket_blocked" = "0" ]; then + acquire_legacy_compat_lock lock_held=1 return 0 fi - - if lock_owner_is_current; then - stale_observations=0 - else - stale_observations=$((stale_observations + 1)) - # Allow the mkdir winner time to publish its owner before treating a - # missing/malformed owner as stale after an interrupted invocation. - if [ "$stale_observations" -ge 20 ]; then - reclaim_stale_daemon_lock || true - stale_observations=0 - fi - fi sleep 0.1 done + rm -rf -- "$our_ticket" emit "ERR state-dir" exit 46 } @@ -325,45 +426,52 @@ trim_log_if_needed() { fi } -prepare_log_for_append() { - append_log="$1" - append_bytes="$2" - [ -f "$append_log" ] || return 0 - current_bytes="$(wc -c <"$append_log" 2>/dev/null | tr -d ' ')" - case "$current_bytes" in - '' | *[!0-9]*) return 0 ;; - esac - if [ $((current_bytes + append_bytes)) -gt "$LOG_MAX_BYTES" ]; then - log_tmp="$STATE_DIR/.goose-serve.log.$$" - if tail -c "$LOG_RETAIN_BYTES" "$append_log" >"$log_tmp" 2>/dev/null; then - mv -f "$log_tmp" "$append_log" - else - rm -f "$log_tmp" - fi - fi -} - -# Consume fixed-size byte chunks rather than newline records. This bounds shell -# memory for a huge line and continues rotating a producer that never emits a -# newline. Each append reopens the path so atomic trims take effect. +# `tee` writes every partial pipe read to its named output (the visible log) +# before stdout. `head` closes that stdout counter after a roughly 2 MiB +# segment, ending this tee only after the same bytes are visible. The 64 KiB reserve +# covers the final in-flight pipe read, so a segment starting from the retained +# 2 MiB tail cannot exceed the 4 MiB cap. Rotation happens per segment rather +# than per small FIFO read. bounded_log_writer() { writer_log="$1" - writer_chunk="$STATE_DIR/.goose-serve.log.chunk.$$" + writer_tmp="$STATE_DIR/.goose-serve.log.rotate.$$" + writer_head_pid="" + # Preserve the Goose FIFO before backgrounding each counter pipeline; + # non-interactive Bash otherwise redirects an asynchronous stdin to null. + exec 3<&0 + trap 'kill -TERM "$writer_head_pid" 2>/dev/null || true; wait "$writer_head_pid" 2>/dev/null || true; rm -f "$writer_tmp"; exit 143' TERM + trap 'kill -TERM "$writer_head_pid" 2>/dev/null || true; wait "$writer_head_pid" 2>/dev/null || true; rm -f "$writer_tmp"; exit 129' HUP + trap 'kill -TERM "$writer_head_pid" 2>/dev/null || true; wait "$writer_head_pid" 2>/dev/null || true; rm -f "$writer_tmp"; exit 130' INT + while :; do - rm -f "$writer_chunk" - dd bs="$LOG_WRITE_CHUNK_BYTES" count=1 of="$writer_chunk" 2>/dev/null || true - writer_bytes="$(wc -c <"$writer_chunk" 2>/dev/null | tr -d ' ')" + writer_bytes="$(wc -c <"$writer_log" 2>/dev/null | tr -d ' ')" case "$writer_bytes" in '' | *[!0-9]*) writer_bytes=0 ;; esac - if [ "$writer_bytes" -eq 0 ]; then - rm -f "$writer_chunk" - break + if [ "$writer_bytes" -gt "$LOG_RETAIN_BYTES" ]; then + if ! tail -c "$LOG_RETAIN_BYTES" "$writer_log" >"$writer_tmp" 2>/dev/null; then + rm -f "$writer_tmp" + break + fi + if ! mv -f "$writer_tmp" "$writer_log"; then + rm -f "$writer_tmp" + break + fi + writer_bytes="$LOG_RETAIN_BYTES" fi - prepare_log_for_append "$writer_log" "$writer_bytes" - cat "$writer_chunk" >>"$writer_log" + + tee -a "$writer_log" <&3 2>/dev/null | head -c "$LOG_SEGMENT_BYTES" >/dev/null 2>&1 & + writer_head_pid=$! + wait "$writer_head_pid" 2>/dev/null || true + writer_head_pid="" + writer_after="$(wc -c <"$writer_log" 2>/dev/null | tr -d ' ')" + case "$writer_after" in + '' | *[!0-9]*) writer_after="$writer_bytes" ;; + esac + [ "$writer_after" -gt "$writer_bytes" ] || break done - rm -f "$writer_chunk" + trim_log_if_needed "$writer_log" + rm -f "$writer_tmp" } # Confirms the PID is still the exact process that wrote this record. `kill -0` diff --git a/src/features/remoteHosts/stores/remoteHostStore.test.ts b/src/features/remoteHosts/stores/remoteHostStore.test.ts index 224962fe8..3e14743db 100644 --- a/src/features/remoteHosts/stores/remoteHostStore.test.ts +++ b/src/features/remoteHosts/stores/remoteHostStore.test.ts @@ -1,6 +1,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { RemoteBackendConnection, + RemoteBackendSnapshotEntry, RemoteToolProbe, } from "@/shared/api/remoteHosts"; @@ -8,6 +9,7 @@ const mocks = vi.hoisted(() => ({ listSshConfigHosts: vi.fn(), connectRemoteHost: vi.fn(), disconnectRemoteHost: vi.fn(), + forgetRemoteHost: vi.fn(), shutdownRemoteHost: vi.fn(), listRemoteBackends: vi.fn(), checkRemoteHost: vi.fn(), @@ -41,9 +43,15 @@ const connection: RemoteBackendConnection = { localPort: 4001, gooseVersion: "1.2.3", daemonReused: false, + incarnation: "slot-1", generation: 1, }; +const backendIdentity = { + incarnation: connection.incarnation, + generation: connection.generation, +}; + function resetStore(): void { useRemoteHostStore.setState({ configHosts: [], @@ -52,6 +60,12 @@ function resetStore(): void { doctorByHost: {}, doctorPendingByHost: {}, doctorErrorByHost: {}, + forgottenHosts: {}, + lifecycleByHost: {}, + connectPendingLifecycleByHost: {}, + retiredIncarnationsByHost: {}, + forgetPendingByHost: {}, + forgetErrorByHost: {}, recentDirsByHost: {}, goosePathByHost: {}, }); @@ -61,17 +75,20 @@ beforeEach(() => { vi.clearAllMocks(); window.localStorage.clear(); resetStore(); + mocks.forgetRemoteHost.mockResolvedValue(undefined); }); describe("applyStatusEvent", () => { it("updates statusByHost from status events", () => { useRemoteHostStore.getState().applyStatusEvent({ host: "devbox", + ...backendIdentity, state: "reconnecting", attempt: 2, }); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "reconnecting", attempt: 2, }); @@ -80,6 +97,7 @@ describe("applyStatusEvent", () => { it("clears a previous error when a ready event arrives", () => { useRemoteHostStore.getState().applyStatusEvent({ host: "devbox", + ...backendIdentity, state: "failed", error: { kind: "host-unreachable", message: "no route" }, }); @@ -89,12 +107,14 @@ describe("applyStatusEvent", () => { useRemoteHostStore.getState().applyStatusEvent({ host: "devbox", + ...backendIdentity, state: "ready", wsUrl: connection.wsUrl, localPort: connection.localPort, }); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "ready", }); }); @@ -103,9 +123,11 @@ describe("applyStatusEvent", () => { describe("syncBackendSnapshot", () => { it("copies snapshot entries into statusByHost", async () => { mocks.listRemoteBackends.mockResolvedValue([ - { host: "devbox", state: "ready" }, + { host: "devbox", ...backendIdentity, state: "ready" }, { host: "broken", + incarnation: "broken-slot", + generation: 3, state: "failed", error: { kind: "auth-failed", message: "denied" }, }, @@ -114,8 +136,13 @@ describe("syncBackendSnapshot", () => { await useRemoteHostStore.getState().syncBackendSnapshot(); const { statusByHost } = useRemoteHostStore.getState(); - expect(statusByHost.devbox).toEqual({ state: "ready" }); + expect(statusByHost.devbox).toEqual({ + ...backendIdentity, + state: "ready", + }); expect(statusByHost.broken).toEqual({ + incarnation: "broken-slot", + generation: 3, state: "failed", error: { kind: "auth-failed", message: "denied" }, }); @@ -124,14 +151,71 @@ describe("syncBackendSnapshot", () => { it("keeps the previous statuses when the snapshot fails", async () => { useRemoteHostStore .getState() - .applyStatusEvent({ host: "devbox", state: "ready" }); + .applyStatusEvent({ host: "devbox", ...backendIdentity, state: "ready" }); mocks.listRemoteBackends.mockRejectedValue(new Error("ipc down")); await useRemoteHostStore.getState().syncBackendSnapshot(); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, + state: "ready", + }); + }); + + it("does not restore a forgotten host from an older in-flight snapshot", async () => { + const host = "broken.blox"; + let resolveSnapshot: (snapshot: RemoteBackendSnapshotEntry[]) => void = + () => {}; + mocks.listRemoteBackends.mockImplementation( + () => + new Promise((resolve) => { + resolveSnapshot = resolve; + }), + ); + useRemoteHostStore.setState({ + statusByHost: { [host]: { ...backendIdentity, state: "failed" } }, + }); + + const syncing = useRemoteHostStore.getState().syncBackendSnapshot(); + await useRemoteHostStore.getState().forgetHost(host); + resolveSnapshot([{ host, ...backendIdentity, state: "failed" }]); + await syncing; + + expect(useRemoteHostStore.getState().statusByHost).not.toHaveProperty(host); + expect(useRemoteHostStore.getState().forgottenHosts[host]).toBe(true); + }); + + it("rejects a pre-forget snapshot after an intentional reconnect", async () => { + const host = "broken.blox"; + let resolveSnapshot: (snapshot: RemoteBackendSnapshotEntry[]) => void = + () => {}; + mocks.listRemoteBackends.mockImplementation( + () => + new Promise((resolve) => { + resolveSnapshot = resolve; + }), + ); + const replacementConnection = { + ...connection, + incarnation: "slot-2", + }; + mocks.connectRemoteHost.mockResolvedValue(replacementConnection); + useRemoteHostStore.setState({ + statusByHost: { [host]: { ...backendIdentity, state: "failed" } }, + }); + + const syncing = useRemoteHostStore.getState().syncBackendSnapshot(); + await useRemoteHostStore.getState().forgetHost(host); + await useRemoteHostStore.getState().ensureHostConnected(host); + resolveSnapshot([{ host, ...backendIdentity, state: "disconnected" }]); + await syncing; + + expect(useRemoteHostStore.getState().statusByHost[host]).toEqual({ + incarnation: replacementConnection.incarnation, + generation: replacementConnection.generation, state: "ready", }); + expect(useRemoteHostStore.getState().forgottenHosts[host]).toBeUndefined(); }); }); @@ -160,7 +244,7 @@ describe("ensureHostConnected", () => { it("resolves without invoking connect when the host is already ready", async () => { useRemoteHostStore .getState() - .applyStatusEvent({ host: "devbox", state: "ready" }); + .applyStatusEvent({ host: "devbox", ...backendIdentity, state: "ready" }); await useRemoteHostStore.getState().ensureHostConnected("devbox"); @@ -186,10 +270,60 @@ describe("ensureHostConnected", () => { expect(mocks.connectRemoteHost).toHaveBeenCalledWith("devbox"); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "ready", }); }); + it("keeps the newest connection when two lifecycles complete out of order", async () => { + const resolvers: Array<(value: RemoteBackendConnection) => void> = []; + mocks.connectRemoteHost.mockImplementation( + () => + new Promise((resolve) => { + resolvers.push(resolve); + }), + ); + + const older = useRemoteHostStore.getState().ensureHostConnected("devbox"); + const newer = useRemoteHostStore.getState().ensureHostConnected("devbox"); + const newestConnection = { + ...connection, + incarnation: "slot-new", + generation: 4, + }; + resolvers[1]?.(newestConnection); + await newer; + resolvers[0]?.({ ...connection, incarnation: "slot-old", generation: 9 }); + await older; + + expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + state: "ready", + incarnation: newestConnection.incarnation, + generation: newestConnection.generation, + }); + }); + + it("does not publish a connection that completes after Forget", async () => { + const host = "forgotten.blox"; + let resolveConnect: (value: RemoteBackendConnection) => void = () => {}; + mocks.connectRemoteHost.mockImplementation( + () => + new Promise((resolve) => { + resolveConnect = resolve; + }), + ); + + const pending = useRemoteHostStore.getState().ensureHostConnected(host); + await useRemoteHostStore.getState().forgetHost(host); + resolveConnect(connection); + await pending; + + const state = useRemoteHostStore.getState(); + expect(state.statusByHost).not.toHaveProperty(host); + expect(state.manualHosts).not.toContain(host); + expect(state.forgottenHosts[host]).toBe(true); + }); + it("marks the host failed with the typed error and rethrows", async () => { const error = { kind: "auth-failed", message: "permission denied" }; mocks.connectRemoteHost.mockRejectedValue(error); @@ -223,12 +357,16 @@ describe("disconnect and shutdownHost", () => { mocks.disconnectRemoteHost.mockResolvedValue(undefined); useRemoteHostStore .getState() - .applyStatusEvent({ host: "devbox", state: "ready" }); + .applyStatusEvent({ host: "devbox", ...backendIdentity, state: "ready" }); await useRemoteHostStore.getState().disconnect("devbox"); - expect(mocks.disconnectRemoteHost).toHaveBeenCalledWith("devbox"); + expect(mocks.disconnectRemoteHost).toHaveBeenCalledWith( + "devbox", + backendIdentity.generation, + ); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "disconnected", }); }); @@ -237,16 +375,41 @@ describe("disconnect and shutdownHost", () => { mocks.shutdownRemoteHost.mockResolvedValue(undefined); useRemoteHostStore .getState() - .applyStatusEvent({ host: "devbox", state: "ready" }); + .applyStatusEvent({ host: "devbox", ...backendIdentity, state: "ready" }); await useRemoteHostStore.getState().shutdownHost("devbox"); - expect(mocks.shutdownRemoteHost).toHaveBeenCalledWith("devbox"); + expect(mocks.shutdownRemoteHost).toHaveBeenCalledWith( + "devbox", + undefined, + backendIdentity.generation, + ); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "disconnected", }); }); + it("keeps an authoritative ready state when shutdown rejects before tunnel teardown", async () => { + const daemonChanged = { + kind: "daemon-changed", + message: "remote daemon changed", + }; + mocks.shutdownRemoteHost.mockRejectedValue(daemonChanged); + useRemoteHostStore + .getState() + .applyStatusEvent({ host: "devbox", ...backendIdentity, state: "ready" }); + + await expect( + useRemoteHostStore.getState().shutdownHost("devbox"), + ).rejects.toBe(daemonChanged); + + expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, + state: "ready", + }); + }); + it("passes a conflict generation token to shutdown", async () => { mocks.shutdownRemoteHost.mockResolvedValue(undefined); @@ -257,8 +420,75 @@ describe("disconnect and shutdownHost", () => { expect(mocks.shutdownRemoteHost).toHaveBeenCalledWith( "devbox", "opaque-generation", + undefined, ); }); + + it("does not publish a stale disconnect completion", async () => { + let resolveDisconnect: () => void = () => {}; + mocks.disconnectRemoteHost.mockImplementation( + () => + new Promise((resolve) => { + resolveDisconnect = resolve; + }), + ); + useRemoteHostStore.setState({ + lifecycleByHost: { devbox: 1 }, + statusByHost: { devbox: { ...backendIdentity, state: "ready" } }, + }); + + const pending = useRemoteHostStore.getState().disconnect("devbox"); + useRemoteHostStore.setState({ + lifecycleByHost: { devbox: 2 }, + statusByHost: { + devbox: { incarnation: "slot-2", generation: 2, state: "ready" }, + }, + }); + resolveDisconnect(); + await pending; + + expect(mocks.disconnectRemoteHost).toHaveBeenCalledWith("devbox", 1); + expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + incarnation: "slot-2", + generation: 2, + state: "ready", + }); + }); + + it("does not publish a stale shutdown completion", async () => { + let resolveShutdown: () => void = () => {}; + mocks.shutdownRemoteHost.mockImplementation( + () => + new Promise((resolve) => { + resolveShutdown = resolve; + }), + ); + useRemoteHostStore.setState({ + lifecycleByHost: { devbox: 1 }, + statusByHost: { devbox: { ...backendIdentity, state: "ready" } }, + }); + + const pending = useRemoteHostStore.getState().shutdownHost("devbox"); + useRemoteHostStore.setState({ + lifecycleByHost: { devbox: 2 }, + statusByHost: { + devbox: { incarnation: "slot-2", generation: 2, state: "ready" }, + }, + }); + resolveShutdown(); + await pending; + + expect(mocks.shutdownRemoteHost).toHaveBeenCalledWith( + "devbox", + undefined, + 1, + ); + expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + incarnation: "slot-2", + generation: 2, + state: "ready", + }); + }); }); describe("runDoctor", () => { @@ -440,14 +670,19 @@ describe("initRemoteHostStore", () => { it("subscribes to status events, seeds state, and returns unsubscribe", async () => { const unlisten = vi.fn(); let statusHandler: - | ((payload: { host: string; state: string }) => void) + | ((payload: { + host: string; + incarnation: string; + generation: number; + state: string; + }) => void) | undefined; mocks.listenRemoteBackendStatus.mockImplementation((handler) => { statusHandler = handler; return Promise.resolve(unlisten); }); mocks.listRemoteBackends.mockResolvedValue([ - { host: "devbox", state: "ready" }, + { host: "devbox", ...backendIdentity, state: "ready" }, ]); mocks.listSshConfigHosts.mockResolvedValue(["devbox"]); @@ -455,11 +690,19 @@ describe("initRemoteHostStore", () => { expect(useRemoteHostStore.getState().configHosts).toEqual(["devbox"]); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, state: "ready", }); - statusHandler?.({ host: "devbox", state: "reconnecting" }); + statusHandler?.({ + host: "devbox", + ...backendIdentity, + generation: backendIdentity.generation + 1, + state: "reconnecting", + }); expect(useRemoteHostStore.getState().statusByHost.devbox).toEqual({ + ...backendIdentity, + generation: backendIdentity.generation + 1, state: "reconnecting", }); @@ -510,14 +753,205 @@ describe("manual host persistence", () => { expect(useRemoteHostStore.getState().manualHosts).toEqual(["adhoc.blox"]); }); - it("forgets a manual host and persists the removal", async () => { - mocks.connectRemoteHost.mockResolvedValue(connection); - await useRemoteHostStore.getState().ensureHostConnected("adhoc.blox"); + it("forgets a broken host while preserving reusable preferences", async () => { + const host = "ssh broken.blox"; + useRemoteHostStore.setState({ + manualHosts: [host, "keep.blox"], + statusByHost: { + [host]: { + state: "failed", + error: { kind: "invalid-host", message: "invalid host" }, + }, + }, + doctorByHost: { [host]: [] }, + doctorPendingByHost: { [host]: false }, + doctorErrorByHost: { + [host]: { kind: "invalid-host", message: "invalid host" }, + }, + recentDirsByHost: { [host]: ["~/src"] }, + goosePathByHost: { [host]: "~/bin/goose" }, + }); + window.localStorage.setItem( + REMOTE_HOST_MANUAL_HOSTS_STORAGE_KEY, + JSON.stringify([host, "keep.blox"]), + ); + window.localStorage.setItem( + REMOTE_HOST_RECENT_DIRS_STORAGE_KEY, + JSON.stringify({ [host]: ["~/src"] }), + ); + window.localStorage.setItem( + REMOTE_HOST_GOOSE_PATH_STORAGE_KEY, + JSON.stringify({ [host]: "~/bin/goose" }), + ); - useRemoteHostStore.getState().removeManualHost("adhoc.blox"); + await useRemoteHostStore.getState().forgetHost(host); - expect(useRemoteHostStore.getState().manualHosts).toEqual([]); - expect(loadPersistedManualHosts()).toEqual([]); + expect(mocks.forgetRemoteHost).toHaveBeenCalledWith(host); + const state = useRemoteHostStore.getState(); + expect(state.manualHosts).toEqual(["keep.blox"]); + expect(state.statusByHost).not.toHaveProperty(host); + expect(state.doctorByHost).not.toHaveProperty(host); + expect(state.doctorPendingByHost).not.toHaveProperty(host); + expect(state.doctorErrorByHost).not.toHaveProperty(host); + expect(state.recentDirsByHost[host]).toEqual(["~/src"]); + expect(state.goosePathByHost[host]).toBe("~/bin/goose"); + expect(state.forgottenHosts[host]).toBe(true); + expect(loadPersistedManualHosts()).toEqual(["keep.blox"]); + expect(loadPersistedRecentDirs()).toEqual({ [host]: ["~/src"] }); + expect(loadPersistedGoosePaths()).toEqual({ [host]: "~/bin/goose" }); + }); + + it("does not let a late Forget completion erase a newer connection", async () => { + const host = "adhoc.blox"; + let resolveForget: () => void = () => {}; + mocks.forgetRemoteHost.mockImplementation( + () => + new Promise((resolve) => { + resolveForget = resolve; + }), + ); + useRemoteHostStore.setState({ + lifecycleByHost: { [host]: 1 }, + manualHosts: [host], + statusByHost: { [host]: { ...backendIdentity, state: "disconnected" } }, + }); + + const forgetting = useRemoteHostStore.getState().forgetHost(host); + const replacement = { + ...connection, + incarnation: "slot-replacement", + generation: 7, + }; + mocks.connectRemoteHost.mockResolvedValue(replacement); + await useRemoteHostStore.getState().ensureHostConnected(host); + resolveForget(); + await forgetting; + + const state = useRemoteHostStore.getState(); + expect(state.statusByHost[host]).toEqual({ + state: "ready", + incarnation: replacement.incarnation, + generation: replacement.generation, + }); + expect(state.manualHosts).toContain(host); + expect(state.forgottenHosts[host]).toBeUndefined(); + expect(state.forgetPendingByHost[host]).toBeUndefined(); + }); + + it("keeps local state when the backend refuses to forget an active host", async () => { + mocks.forgetRemoteHost.mockRejectedValueOnce(new Error("active")); + useRemoteHostStore.setState({ + manualHosts: ["adhoc.blox"], + statusByHost: { "adhoc.blox": { state: "ready" } }, + }); + + await expect( + useRemoteHostStore.getState().forgetHost("adhoc.blox"), + ).rejects.toThrow("active"); + + expect(useRemoteHostStore.getState().manualHosts).toEqual(["adhoc.blox"]); + expect(useRemoteHostStore.getState().statusByHost).toHaveProperty( + "adhoc.blox", + ); + expect( + useRemoteHostStore.getState().forgetPendingByHost["adhoc.blox"], + ).toBe(false); + expect( + useRemoteHostStore.getState().forgetErrorByHost["adhoc.blox"], + ).toEqual({ kind: "internal", message: "active" }); + }); + + it("ignores late status events until an intentional reconnect", async () => { + const host = "broken.blox"; + useRemoteHostStore.setState({ + statusByHost: { [host]: { ...backendIdentity, state: "failed" } }, + }); + + await useRemoteHostStore.getState().forgetHost(host); + useRemoteHostStore.getState().applyStatusEvent({ + host, + ...backendIdentity, + state: "disconnected", + }); + expect(useRemoteHostStore.getState().statusByHost).not.toHaveProperty(host); + + const replacementConnection = { + ...connection, + incarnation: "slot-2", + }; + mocks.connectRemoteHost.mockResolvedValue(replacementConnection); + await useRemoteHostStore.getState().ensureHostConnected(host); + useRemoteHostStore.getState().applyStatusEvent({ + host, + incarnation: "slot-2", + generation: 2, + state: "reconnecting", + attempt: 1, + }); + expect(useRemoteHostStore.getState().statusByHost[host]).toEqual({ + incarnation: "slot-2", + generation: 2, + state: "reconnecting", + attempt: 1, + }); + }); + + it("ignores a retired incarnation after a replacement reconnects", async () => { + const host = "broken.blox"; + useRemoteHostStore.setState({ + statusByHost: { [host]: { ...backendIdentity, state: "failed" } }, + }); + await useRemoteHostStore.getState().forgetHost(host); + mocks.connectRemoteHost.mockResolvedValue({ + ...connection, + incarnation: "slot-2", + }); + await useRemoteHostStore.getState().ensureHostConnected(host); + + useRemoteHostStore.getState().applyStatusEvent({ + host, + incarnation: connection.incarnation, + generation: 99, + state: "disconnected", + }); + + expect(useRemoteHostStore.getState().statusByHost[host]).toEqual({ + state: "ready", + incarnation: "slot-2", + generation: connection.generation, + }); + }); + + it("exposes forget pending and failure state and deduplicates submissions", async () => { + const host = "broken.blox"; + let rejectForget: (error: unknown) => void = () => {}; + mocks.forgetRemoteHost.mockImplementation( + () => + new Promise((_resolve, reject) => { + rejectForget = reject; + }), + ); + useRemoteHostStore.setState({ + statusByHost: { [host]: { state: "failed" } }, + }); + + const first = useRemoteHostStore.getState().forgetHost(host); + const firstResult = expect(first).rejects.toEqual({ + kind: "internal", + message: "still connecting", + }); + expect(useRemoteHostStore.getState().forgetPendingByHost[host]).toBe(true); + const duplicate = useRemoteHostStore.getState().forgetHost(host); + expect(mocks.forgetRemoteHost).toHaveBeenCalledTimes(1); + + rejectForget({ kind: "internal", message: "still connecting" }); + await firstResult; + await duplicate; + expect(useRemoteHostStore.getState().forgetPendingByHost[host]).toBe(false); + expect(useRemoteHostStore.getState().forgetErrorByHost[host]).toEqual({ + kind: "internal", + message: "still connecting", + }); }); it("tolerates corrupted storage when loading manual hosts", () => { diff --git a/src/features/remoteHosts/stores/remoteHostStore.ts b/src/features/remoteHosts/stores/remoteHostStore.ts index 0f630896e..88b1c245a 100644 --- a/src/features/remoteHosts/stores/remoteHostStore.ts +++ b/src/features/remoteHosts/stores/remoteHostStore.ts @@ -8,12 +8,14 @@ import { checkRemoteHost, connectRemoteHost, disconnectRemoteHost, + forgetRemoteHost, isRemoteBackendError, listenRemoteBackendStatus, listRemoteBackends, listSshConfigHosts, shutdownRemoteHost, type RemoteBackendErrorLike, + type RemoteBackendSnapshotEntry, type RemoteBackendState, type RemoteBackendStatusPayload, type RemoteToolProbe, @@ -26,13 +28,41 @@ export const REMOTE_HOST_MANUAL_HOSTS_STORAGE_KEY = const MAX_RECENT_DIRS_PER_HOST = 8; const MAX_MANUAL_HOSTS = 16; +const MAX_RETIRED_INCARNATIONS_PER_HOST = 8; export interface RemoteHostStatus { state: RemoteBackendState; + incarnation?: string; + generation?: number; attempt?: number; error?: RemoteBackendErrorLike; } +function backendStatus( + payload: RemoteBackendStatusPayload | RemoteBackendSnapshotEntry, +): RemoteHostStatus { + return { + state: payload.state, + incarnation: payload.incarnation, + generation: payload.generation, + ...(payload.attempt !== undefined ? { attempt: payload.attempt } : {}), + ...(payload.error ? { error: payload.error } : {}), + }; +} + +function acceptsBackendStatus( + current: RemoteHostStatus | undefined, + retiredIncarnations: string[] | undefined, + payload: RemoteBackendStatusPayload | RemoteBackendSnapshotEntry, +): boolean { + if (retiredIncarnations?.includes(payload.incarnation)) return false; + if (!current?.incarnation) return true; + return ( + current.incarnation === payload.incarnation && + (current.generation ?? 0) <= payload.generation + ); +} + function toRemoteBackendError(error: unknown): RemoteBackendErrorLike { if (isRemoteBackendError(error)) return error; return { @@ -125,6 +155,16 @@ export interface RemoteHostStore { doctorByHost: Record; doctorPendingByHost: Record; doctorErrorByHost: Record; + /** Successful Forget tombstones, cleared only by an explicit new connect. */ + forgottenHosts: Record; + /** Monotonic local lifecycle used to reject snapshots admitted before a change. */ + lifecycleByHost: Record; + /** Explicit connect lifecycle currently awaiting its backend result. */ + connectPendingLifecycleByHost: Record; + /** Forgotten backend slot identities that must never be admitted again. */ + retiredIncarnationsByHost: Record; + forgetPendingByHost: Record; + forgetErrorByHost: Record; recentDirsByHost: Record; /** Per-host goose binary override; absent means the remote login PATH. */ goosePathByHost: Record; @@ -138,7 +178,7 @@ export interface RemoteHostStore { shutdownHost: (host: string, expectedInstanceToken?: string) => Promise; runDoctor: (host: string) => Promise; recordRecentDir: (host: string, dir: string) => void; - removeManualHost: (host: string) => void; + forgetHost: (host: string) => Promise; /** * Set (or clear, with `null`) the goose binary a host's remote backend * should run. Returns false for a path the remote script could not resolve. @@ -154,6 +194,12 @@ export const useRemoteHostStore = create((set, get) => ({ doctorByHost: {}, doctorPendingByHost: {}, doctorErrorByHost: {}, + forgottenHosts: {}, + lifecycleByHost: {}, + connectPendingLifecycleByHost: {}, + retiredIncarnationsByHost: {}, + forgetPendingByHost: {}, + forgetErrorByHost: {}, recentDirsByHost: loadPersistedRecentDirs(), goosePathByHost: loadPersistedGoosePaths(), @@ -168,16 +214,28 @@ export const useRemoteHostStore = create((set, get) => ({ }, syncBackendSnapshot: async () => { + // Keep the lifecycle object from admission time. Store updates replace it, + // so a Forget or explicit reconnect while IPC is in flight is observable. + const lifecycleAtStart = get().lifecycleByHost; try { const snapshot = await listRemoteBackends(); set((state) => { const statusByHost = { ...state.statusByHost }; for (const entry of snapshot) { - statusByHost[entry.host] = { - state: entry.state, - ...(entry.attempt !== undefined ? { attempt: entry.attempt } : {}), - ...(entry.error ? { error: entry.error } : {}), - }; + if ( + state.forgottenHosts[entry.host] || + state.connectPendingLifecycleByHost[entry.host] !== undefined || + (state.lifecycleByHost[entry.host] ?? 0) !== + (lifecycleAtStart[entry.host] ?? 0) || + !acceptsBackendStatus( + statusByHost[entry.host], + state.retiredIncarnationsByHost[entry.host], + entry, + ) + ) { + continue; + } + statusByHost[entry.host] = backendStatus(entry); } return { statusByHost }; }); @@ -187,34 +245,82 @@ export const useRemoteHostStore = create((set, get) => ({ }, applyStatusEvent: (payload) => { - set((state) => ({ - statusByHost: { - ...state.statusByHost, - [payload.host]: { - state: payload.state, - ...(payload.attempt !== undefined - ? { attempt: payload.attempt } - : {}), - ...(payload.error ? { error: payload.error } : {}), + set((state) => { + if ( + state.forgottenHosts[payload.host] || + state.connectPendingLifecycleByHost[payload.host] !== undefined || + !acceptsBackendStatus( + state.statusByHost[payload.host], + state.retiredIncarnationsByHost[payload.host], + payload, + ) + ) { + return state; + } + return { + statusByHost: { + ...state.statusByHost, + [payload.host]: backendStatus(payload), }, - }, - })); + }; + }); }, ensureHostConnected: async (host) => { - if (get().statusByHost[host]?.state === "ready") return; + const current = get(); + if ( + current.statusByHost[host]?.state === "ready" && + !current.forgottenHosts[host] + ) { + return; + } - // Optimistic: the Rust side serializes concurrent connects per host and - // emits status events, but reflect intent immediately in the UI. - set((state) => ({ - statusByHost: { - ...state.statusByHost, - [host]: { state: "connecting" }, - }, - })); + // An explicit connection starts a new local lifecycle. This is the only + // operation that clears a successful Forget tombstone. + const lifecycle = (current.lifecycleByHost[host] ?? 0) + 1; + set((state) => { + const forgottenHosts = { ...state.forgottenHosts }; + const forgetErrorByHost = { ...state.forgetErrorByHost }; + const currentStatus = state.statusByHost[host]; + delete forgottenHosts[host]; + delete forgetErrorByHost[host]; + return { + forgottenHosts, + forgetErrorByHost, + lifecycleByHost: { + ...state.lifecycleByHost, + [host]: lifecycle, + }, + connectPendingLifecycleByHost: { + ...state.connectPendingLifecycleByHost, + [host]: lifecycle, + }, + // Optimistic: Rust serializes concurrent connects per host, but the UI + // should reflect the user's new lifecycle immediately. + statusByHost: { + ...state.statusByHost, + [host]: { + state: "connecting", + ...(currentStatus?.incarnation + ? { + incarnation: currentStatus.incarnation, + generation: currentStatus.generation, + } + : {}), + }, + }, + }; + }); try { - await connectRemoteHost(host); + const connection = await connectRemoteHost(host); set((state) => { + if ( + state.forgottenHosts[host] || + state.lifecycleByHost[host] !== lifecycle || + state.connectPendingLifecycleByHost[host] !== lifecycle + ) { + return state; + } // A host that connected but isn't in ~/.ssh/config was typed in // manually; remember it across restarts. const isKnown = @@ -225,47 +331,105 @@ export const useRemoteHostStore = create((set, get) => ({ if (!isKnown) { persistManualHosts(manualHosts); } + const connectPendingLifecycleByHost = { + ...state.connectPendingLifecycleByHost, + }; + delete connectPendingLifecycleByHost[host]; return { manualHosts, + connectPendingLifecycleByHost, statusByHost: { ...state.statusByHost, - [host]: { state: "ready" }, + [host]: { + state: "ready", + incarnation: connection.incarnation, + generation: connection.generation, + }, }, }; }); } catch (error) { - set((state) => ({ - statusByHost: { - ...state.statusByHost, - [host]: { state: "failed", error: toRemoteBackendError(error) }, - }, - })); + set((state) => { + if ( + state.forgottenHosts[host] || + state.lifecycleByHost[host] !== lifecycle || + state.connectPendingLifecycleByHost[host] !== lifecycle + ) { + return state; + } + const connectPendingLifecycleByHost = { + ...state.connectPendingLifecycleByHost, + }; + delete connectPendingLifecycleByHost[host]; + return { + connectPendingLifecycleByHost, + statusByHost: { + ...state.statusByHost, + [host]: { + state: "failed", + ...(state.statusByHost[host]?.incarnation + ? { + incarnation: state.statusByHost[host].incarnation, + generation: state.statusByHost[host].generation, + } + : {}), + error: toRemoteBackendError(error), + }, + }, + }; + }); throw error; } }, disconnect: async (host) => { - await disconnectRemoteHost(host); - set((state) => ({ - statusByHost: { - ...state.statusByHost, - [host]: { state: "disconnected" }, - }, - })); + const admitted = get(); + const admittedLifecycle = admitted.lifecycleByHost[host] ?? 0; + const admittedStatus = admitted.statusByHost[host]; + await disconnectRemoteHost(host, admittedStatus?.generation); + set((state) => { + const currentStatus = state.statusByHost[host]; + if ( + (state.lifecycleByHost[host] ?? 0) !== admittedLifecycle || + currentStatus?.incarnation !== admittedStatus?.incarnation || + currentStatus?.generation !== admittedStatus?.generation + ) { + return state; + } + return { + statusByHost: { + ...state.statusByHost, + [host]: { ...currentStatus, state: "disconnected" }, + }, + }; + }); }, shutdownHost: async (host, expectedInstanceToken) => { - if (expectedInstanceToken) { - await shutdownRemoteHost(host, expectedInstanceToken); - } else { - await shutdownRemoteHost(host); - } - set((state) => ({ - statusByHost: { - ...state.statusByHost, - [host]: { state: "disconnected" }, - }, - })); + const admitted = get(); + const admittedLifecycle = admitted.lifecycleByHost[host] ?? 0; + const admittedStatus = admitted.statusByHost[host]; + await shutdownRemoteHost( + host, + expectedInstanceToken, + admittedStatus?.generation, + ); + set((state) => { + const currentStatus = state.statusByHost[host]; + if ( + (state.lifecycleByHost[host] ?? 0) !== admittedLifecycle || + currentStatus?.incarnation !== admittedStatus?.incarnation || + currentStatus?.generation !== admittedStatus?.generation + ) { + return state; + } + return { + statusByHost: { + ...state.statusByHost, + [host]: { ...currentStatus, state: "disconnected" }, + }, + }; + }); }, runDoctor: async (host) => { @@ -290,14 +454,92 @@ export const useRemoteHostStore = create((set, get) => ({ } }, - removeManualHost: (host) => { + forgetHost: async (host) => { + if (get().forgetPendingByHost[host]) return; + const admittedLifecycle = get().lifecycleByHost[host] ?? 0; + set((state) => ({ + forgetPendingByHost: { + ...state.forgetPendingByHost, + [host]: true, + }, + forgetErrorByHost: { + ...state.forgetErrorByHost, + [host]: undefined, + }, + })); + try { + await forgetRemoteHost(host); + } catch (error) { + set((state) => { + const forgetPendingByHost = { + ...state.forgetPendingByHost, + [host]: false, + }; + const forgetErrorByHost = { ...state.forgetErrorByHost }; + if ((state.lifecycleByHost[host] ?? 0) === admittedLifecycle) { + forgetErrorByHost[host] = toRemoteBackendError(error); + } else { + delete forgetErrorByHost[host]; + } + return { forgetPendingByHost, forgetErrorByHost }; + }); + throw error; + } set((state) => { - if (!state.manualHosts.includes(host)) return state; + const forgetPendingByHost = { ...state.forgetPendingByHost }; + delete forgetPendingByHost[host]; + if ((state.lifecycleByHost[host] ?? 0) !== admittedLifecycle) { + // A newer explicit Connect owns this row. The old Forget result may + // clear its own pending marker, but must not erase the replacement. + return { forgetPendingByHost }; + } const manualHosts = state.manualHosts.filter( (candidate) => candidate !== host, ); + const statusByHost = { ...state.statusByHost }; + const doctorByHost = { ...state.doctorByHost }; + const doctorPendingByHost = { ...state.doctorPendingByHost }; + const doctorErrorByHost = { ...state.doctorErrorByHost }; + const forgottenHosts = { ...state.forgottenHosts, [host]: true as const }; + const connectPendingLifecycleByHost = { + ...state.connectPendingLifecycleByHost, + }; + const retiredIncarnationsByHost = { + ...state.retiredIncarnationsByHost, + }; + const forgetErrorByHost = { ...state.forgetErrorByHost }; + const forgottenIncarnation = statusByHost[host]?.incarnation; + if (forgottenIncarnation) { + retiredIncarnationsByHost[host] = [ + forgottenIncarnation, + ...(retiredIncarnationsByHost[host] ?? []).filter( + (candidate) => candidate !== forgottenIncarnation, + ), + ].slice(0, MAX_RETIRED_INCARNATIONS_PER_HOST); + } + delete statusByHost[host]; + delete doctorByHost[host]; + delete doctorPendingByHost[host]; + delete doctorErrorByHost[host]; + delete forgetErrorByHost[host]; + delete connectPendingLifecycleByHost[host]; persistManualHosts(manualHosts); - return { manualHosts }; + return { + manualHosts, + statusByHost, + doctorByHost, + doctorPendingByHost, + doctorErrorByHost, + forgottenHosts, + lifecycleByHost: { + ...state.lifecycleByHost, + [host]: (state.lifecycleByHost[host] ?? 0) + 1, + }, + connectPendingLifecycleByHost, + retiredIncarnationsByHost, + forgetPendingByHost, + forgetErrorByHost, + }; }); }, diff --git a/src/features/remoteHosts/ui/RemoteHostsSettings.tsx b/src/features/remoteHosts/ui/RemoteHostsSettings.tsx index e67bbcc8a..174937f00 100644 --- a/src/features/remoteHosts/ui/RemoteHostsSettings.tsx +++ b/src/features/remoteHosts/ui/RemoteHostsSettings.tsx @@ -195,11 +195,15 @@ function RemoteHostRow({ ); const disconnect = useRemoteHostStore((state) => state.disconnect); const runDoctor = useRemoteHostStore((state) => state.runDoctor); - const isManualHost = useRemoteHostStore((state) => - state.manualHosts.includes(host), + const isConfigHost = useRemoteHostStore((state) => + state.configHosts.includes(host), ); - const removeManualHost = useRemoteHostStore( - (state) => state.removeManualHost, + const forgetHost = useRemoteHostStore((state) => state.forgetHost); + const forgetPending = useRemoteHostStore( + (state) => state.forgetPendingByHost[host] === true, + ); + const forgetError = useRemoteHostStore( + (state) => state.forgetErrorByHost[host], ); const state = status?.state ?? "disconnected"; @@ -282,12 +286,18 @@ function RemoteHostRow({ > {t("remoteHosts.actions.check")} - {isManualHost && !isConnected ? ( + {!isConfigHost && (state === "failed" || state === "disconnected") ? ( @@ -296,6 +306,11 @@ function RemoteHostRow({ } details={
+ {forgetError ? ( +

+ {t("remoteHosts.forget.error")} +

+ ) : null} {showDoctor ? ( {}); const runDoctor = vi.fn(async () => {}); const refreshConfigHosts = vi.fn(async () => {}); const syncBackendSnapshot = vi.fn(async () => {}); +const forgetHost = vi.fn(async () => {}); const setGoosePath = vi.fn((_host: string, _path: string | null) => true); function seedStore(overrides?: Partial>) { @@ -26,10 +27,15 @@ function seedStore(overrides?: Partial>) { function baseState() { return { configHosts: [] as string[], + manualHosts: [] as string[], statusByHost: {}, doctorByHost: {}, doctorPendingByHost: {}, doctorErrorByHost: {}, + forgottenHosts: {}, + lifecycleByHost: {}, + forgetPendingByHost: {}, + forgetErrorByHost: {}, recentDirsByHost: {}, goosePathByHost: {} as Record, ensureHostConnected, @@ -38,6 +44,7 @@ function baseState() { runDoctor, refreshConfigHosts, syncBackendSnapshot, + forgetHost, setGoosePath, }; } @@ -97,6 +104,99 @@ describe("RemoteHostsSettings", () => { expect(screen.getByText("user@adhoc")).toBeInTheDocument(); }); + it("can forget a failed free-form host", async () => { + const user = userEvent.setup(); + const host = "ssh broken.blox"; + seedStore({ + statusByHost: { + [host]: { + state: "failed", + error: { + kind: "invalid-host", + message: "host must not contain whitespace or control characters", + }, + }, + }, + }); + renderWithProviders(); + + await user.click( + screen.getByRole("button", { + name: enSettings.remoteHosts.actions.forget, + }), + ); + + expect(forgetHost).toHaveBeenCalledWith(host); + }); + + it("disables Forget while pending", () => { + const host = "broken.blox"; + seedStore({ + statusByHost: { [host]: { state: "failed" } }, + forgetPendingByHost: { [host]: true }, + }); + renderWithProviders(); + + const button = screen.getByRole("button", { + name: enSettings.remoteHosts.actions.forgetting, + }); + expect(button).toBeDisabled(); + expect(button).toHaveAttribute("data-feedback-state", "loading"); + expect(button).toHaveAttribute("aria-busy", "true"); + expect( + within(button) + .getAllByText(enSettings.remoteHosts.actions.forget) + .every((label) => label.getAttribute("aria-hidden") === "true"), + ).toBe(true); + }); + + it("keeps the row and explains a rejected Forget", async () => { + const user = userEvent.setup(); + const host = "broken.blox"; + forgetHost.mockImplementationOnce(async () => { + useRemoteHostStore.setState((state) => ({ + forgetPendingByHost: { + ...state.forgetPendingByHost, + [host]: false, + }, + forgetErrorByHost: { + ...state.forgetErrorByHost, + [host]: { kind: "internal", message: "still connecting" }, + }, + })); + throw new Error("still connecting"); + }); + seedStore({ + statusByHost: { [host]: { state: "failed" } }, + }); + renderWithProviders(); + + await user.click( + screen.getByRole("button", { + name: enSettings.remoteHosts.actions.forget, + }), + ); + + expect(await screen.findByRole("alert")).toHaveTextContent( + enSettings.remoteHosts.forget.error, + ); + expect(screen.getByText(host)).toBeInTheDocument(); + }); + + it("does not offer Forget for an ssh-config host", () => { + seedStore({ + configHosts: ["configured"], + statusByHost: { configured: { state: "failed" } }, + }); + renderWithProviders(); + + expect( + screen.queryByRole("button", { + name: enSettings.remoteHosts.actions.forget, + }), + ).not.toBeInTheDocument(); + }); + it("connects a disconnected host via the row's Connect button", async () => { const user = userEvent.setup(); seedStore({ configHosts: ["alpha"] }); diff --git a/src/shared/api/__tests__/remoteHosts.test.ts b/src/shared/api/__tests__/remoteHosts.test.ts index c0f0c522c..affe36a1f 100644 --- a/src/shared/api/__tests__/remoteHosts.test.ts +++ b/src/shared/api/__tests__/remoteHosts.test.ts @@ -102,11 +102,12 @@ describe("remote host goose binary override", () => { }); it("scopes a confirmed takeover to the inspected daemon instance", async () => { - await shutdownRemoteHost("devbox", "instance-token"); + await shutdownRemoteHost("devbox", "instance-token", 7); expect(mockedInvoke).toHaveBeenCalledWith("remote_backend_shutdown", { host: "devbox", expectedInstanceToken: "instance-token", + expectedGeneration: 7, }); }); }); diff --git a/src/shared/api/remoteHosts.ts b/src/shared/api/remoteHosts.ts index bc49001fa..b6163b169 100644 --- a/src/shared/api/remoteHosts.ts +++ b/src/shared/api/remoteHosts.ts @@ -37,6 +37,7 @@ export interface RemoteBackendConnection { localPort: number; gooseVersion: string; daemonReused: boolean; + incarnation: string; generation: number; } @@ -53,6 +54,8 @@ export const REMOTE_BACKEND_STATUS_EVENT = "berd:remote-backend-status"; export interface RemoteBackendStatusPayload { host: string; + incarnation: string; + generation: number; state: RemoteBackendState; wsUrl?: string; httpBaseUrl?: string; @@ -64,6 +67,8 @@ export interface RemoteBackendStatusPayload { /** One entry from the `list_remote_backends` snapshot. */ export interface RemoteBackendSnapshotEntry { host: string; + incarnation: string; + generation: number; state: RemoteBackendState; wsUrl?: string; httpBaseUrl?: string; @@ -124,14 +129,21 @@ export async function disconnectRemoteHost( }); } +/** Remove an inactive host from the backend registry. */ +export async function forgetRemoteHost(host: string): Promise { + await invoke("remote_backend_forget", { host }); +} + /** Stop the remote daemon on `host` and tear down the tunnel. */ export async function shutdownRemoteHost( host: string, expectedInstanceToken?: string, + expectedGeneration?: number, ): Promise { await invoke("remote_backend_shutdown", { host, expectedInstanceToken: expectedInstanceToken ?? null, + expectedGeneration: expectedGeneration ?? null, }); } diff --git a/src/shared/i18n/locales/en/settings.json b/src/shared/i18n/locales/en/settings.json index 916912aca..f663baca6 100644 --- a/src/shared/i18n/locales/en/settings.json +++ b/src/shared/i18n/locales/en/settings.json @@ -700,6 +700,7 @@ "connect": "Connect", "disconnect": "Disconnect", "forget": "Forget", + "forgetting": "Forgetting...", "shutdown": "Stop remote backend", "takeover": "Stop and reconnect" }, @@ -721,6 +722,9 @@ "notFound": "Not found" }, "empty": "No hosts found in ~/.ssh/config.", + "forget": { + "error": "Couldn't forget this host. Disconnect it or wait for the current connection attempt to finish, then try again." + }, "gooseBinary": { "clear": "Clear", "hint": "Overrides the PATH lookup for goose on this host. Changing it restarts the remote backend on the next connect.",