From ef2c504a6535953174b381a2514dc966ebc09f64 Mon Sep 17 00:00:00 2001 From: yyjeqhc <1772413353@qq.com> Date: Mon, 9 Nov 2026 22:36:43 +0800 Subject: [PATCH] fix(tunnel): preserve owned tasks on heartbeat publication failure --- .github/workflows/ci.yml | 6 + crates/webcodex-environment/src/tunnel.rs | 2 +- .../src/tunnel/health_tests.rs | 43 ++++++ docs/architecture/server-owned-tunnels.md | 10 ++ src/server_tunnels.rs | 36 ++++- src/server_tunnels/tests.rs | 145 ++++++++++++++++++ 6 files changed, 236 insertions(+), 6 deletions(-) create mode 100644 src/server_tunnels/tests.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec97aa7a9..923d5bf1f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -797,6 +797,12 @@ jobs: if ($LASTEXITCODE -ne 0) { throw "Computer helper ownership tests failed" } cargo test --locked -p webcodex-environment --lib windows_helper_task_scheduler_roundtrip_is_owned -- --ignored if ($LASTEXITCODE -ne 0) { throw "Computer helper Task Scheduler roundtrip failed" } + - name: Test Windows Tunnel heartbeat contention and recovery + run: | + cargo test --locked -p webcodex-environment --lib tunnel::health_tests + if ($LASTEXITCODE -ne 0) { throw "Windows Tunnel health file tests failed" } + cargo test --locked -p webcodex --lib server_tunnels + if ($LASTEXITCODE -ne 0) { throw "Windows Tunnel heartbeat supervision tests failed" } - name: Test Windows unified installer routing and Desktop build freshness run: python -m unittest scripts.tests.test_prepare_windows_unified_installer scripts.tests.test_desktop_runtime_manifest scripts.tests.test_desktop_frontend_build - name: Test optional Codex observation adapter on Windows diff --git a/crates/webcodex-environment/src/tunnel.rs b/crates/webcodex-environment/src/tunnel.rs index 32db64748..43dac4465 100644 --- a/crates/webcodex-environment/src/tunnel.rs +++ b/crates/webcodex-environment/src/tunnel.rs @@ -1893,7 +1893,7 @@ pub(crate) fn read_tunnel_health( }) } -#[cfg(all(test, unix))] +#[cfg(test)] mod health_tests; #[cfg(test)] diff --git a/crates/webcodex-environment/src/tunnel/health_tests.rs b/crates/webcodex-environment/src/tunnel/health_tests.rs index 62b90a169..7aaf992c9 100644 --- a/crates/webcodex-environment/src/tunnel/health_tests.rs +++ b/crates/webcodex-environment/src/tunnel/health_tests.rs @@ -1,6 +1,7 @@ use super::*; #[test] +#[cfg(unix)] fn health_locks_release_without_waiting_for_inherited_handles() { for exclusive in [false, true] { for fail_after_lock in [false, true] { @@ -53,3 +54,45 @@ fn health_locks_release_without_waiting_for_inherited_handles() { } } } + +#[test] +fn missed_heartbeats_expire_during_contention_and_recover_after_unlock() { + let temp = crate::test_tempdir().unwrap(); + let store = EnvironmentStore::open(temp.path().join("environment")).unwrap(); + let path = store.root().join("readiness.json"); + let stale = serde_json::json!({ + "schema_version": 1, + "observed_at_ms": 0, + "service_pid": std::process::id(), + "profile_revision": 7, + "state": "running", + "tunnel_ready": true, + "local_mcp_ready": true, + }); + atomic_private_write(&path, stale.to_string().as_bytes()).unwrap(); + let file = std::fs::File::open(&path).unwrap(); + fs2::FileExt::try_lock_shared(&file).unwrap(); + let reader = TunnelHealthFileLock { file }; + for _ in 0..4 { + assert!(write_embedded_tunnel_observation( + &path, + 7, + TunnelState::Running, + true, + true, + None, + ) + .is_err()); + let health = read_tunnel_health(&path).unwrap(); + assert!(!health.fresh); + assert!(!health.tunnel_ready && !health.local_mcp_ready); + assert_eq!(health.state, Some(TunnelState::Running)); + assert_eq!(health.profile_revision, Some(7)); + } + drop(reader); + write_embedded_tunnel_observation(&path, 7, TunnelState::Running, true, true, None).unwrap(); + let health = read_tunnel_health(&path).unwrap(); + assert!(health.fresh && health.tunnel_ready && health.local_mcp_ready); + assert_eq!(health.service_pid, Some(std::process::id())); + assert_eq!(health.profile_revision, Some(7)); +} diff --git a/docs/architecture/server-owned-tunnels.md b/docs/architecture/server-owned-tunnels.md index 36bbfa8fa..6dfcc287a 100644 --- a/docs/architecture/server-owned-tunnels.md +++ b/docs/architecture/server-owned-tunnels.md @@ -106,6 +106,16 @@ existing per-profile service for standalone. Readiness files keep the existing freshness contract. Transport readiness is not evidence of a successful client operation, and process liveness is not either readiness bit. +The initial status write must succeed before a profile launches. Once its Tunnel +task is owned, a heartbeat publication error (including a concurrent reader's +shared file lock) does not end or restart that task. Runtime status keeps the live +readiness bits and reports `tunnel_health_unavailable` until publication succeeds +at a later normal two-second health tick. Only failure/recovery transitions are +logged. A missed write does not refresh the saved observation: readers still +reject it after seven seconds. The writer retains the precreated file, its private +ACL and its nonblocking lock; it never replaces the file or bypasses validation. +Task failure, startup deadlines, shutdown and restart-fence semantics are unchanged. + Server termination runs these phases in order: 1. Stop new Tunnel polling/admission. diff --git a/src/server_tunnels.rs b/src/server_tunnels.rs index 49d7ef8c6..5ab30deb5 100644 --- a/src/server_tunnels.rs +++ b/src/server_tunnels.rs @@ -259,11 +259,7 @@ async fn run_profile( state.state = if reached_ready { TunnelState::Running } else { TunnelState::Starting }; state.tunnel_ready = ready; state.local_mcp_ready = local; }); - if persist_status(&profile, &status, index) - .is_err() - { - break Some("tunnel_health_unavailable".into()); - } + persist_heartbeat(&profile, &status, index); } } }; @@ -294,6 +290,33 @@ fn persist_status( index: usize, ) -> webcodex_environment::SetupResultValue<()> { let state = &status.snapshot()[index]; + write_status(profile, state) +} + +/// A saved heartbeat is an observation, not the owned task's execution state. +/// Retry publication at the next normal tick without restarting or replaying work. +/// Readers still reject stale observations using the existing freshness bound. +fn persist_heartbeat(profile: &EmbeddedTunnelProfile, status: &TunnelStatus, index: usize) { + let mut state = status.snapshot()[index].clone(); + let was_unavailable = state.diagnostic.as_deref() == Some("tunnel_health_unavailable"); + state.diagnostic = None; + let unavailable = write_status(profile, &state).is_err(); + status.update(index, |state| { + state.diagnostic = unavailable.then(|| "tunnel_health_unavailable".into()); + }); + // Log transitions, not every failed tick. Never include filesystem errors or + // private profile bindings in diagnostics. + if unavailable && !was_unavailable { + tracing::warn!(profile = %profile.profile_id, "Tunnel heartbeat unavailable; retrying at the next health tick"); + } else if !unavailable && was_unavailable { + tracing::info!(profile = %profile.profile_id, "Tunnel heartbeat publication recovered"); + } +} + +fn write_status( + profile: &EmbeddedTunnelProfile, + state: &ProfileStatus, +) -> webcodex_environment::SetupResultValue<()> { write_embedded_tunnel_observation( &profile.readiness_path, profile.runtime_revision, @@ -303,3 +326,6 @@ fn persist_status( state.diagnostic.as_deref(), ) } + +#[cfg(test)] +mod tests; diff --git a/src/server_tunnels/tests.rs b/src/server_tunnels/tests.rs new file mode 100644 index 000000000..4db14c2fc --- /dev/null +++ b/src/server_tunnels/tests.rs @@ -0,0 +1,145 @@ +use super::*; +use webcodex_environment::{EnvironmentStore, Secret, TunnelCredentials}; + +fn fixture() -> (tempfile::TempDir, EmbeddedTunnelProfile, TunnelStatus) { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap().join("environment"); + let store = EnvironmentStore::open(root).unwrap(); + let path = store.root().join("readiness.json"); + create_health_file(&path); + let profile = EmbeddedTunnelProfile { + profile_id: "primary".into(), + autostart: true, + runtime_revision: 7, + credentials: TunnelCredentials { + tunnel_id: Secret::new("tunnel_0123456789abcdef0123456789abcdef".into()), + api_key: Secret::new("fixture-control".into()), + }, + local_token: Secret::new("fixture-local".into()), + proxy: None, + readiness_path: path, + }; + let status = TunnelStatus(Arc::new(Mutex::new(vec![ProfileStatus { + profile_id: profile.profile_id.clone(), + host_mode: "embedded", + state: TunnelState::Running, + tunnel_ready: true, + local_mcp_ready: true, + shutdown_outcome: None, + diagnostic: None, + }]))); + (temp, profile, status) +} + +fn create_health_file(path: &Path) { + let mut options = std::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + // On Windows the file inherits the EnvironmentStore directory's private ACL. + options.open(path).unwrap(); +} + +#[test] +fn reader_contention_preserves_live_status_and_publication_recovers() { + let (_temp, profile, status) = fixture(); + persist_status(&profile, &status, 0).unwrap(); + let previous = std::fs::read(&profile.readiness_path).unwrap(); + let reader = std::fs::File::open(&profile.readiness_path).unwrap(); + reader.try_lock_shared().unwrap(); + // A separately opened handle exercises the same lock conflict as a Desktop + // or CLI reader, deterministically and without timing or external services. + assert!(persist_status(&profile, &status, 0).is_err()); + for _ in 0..4 { + persist_heartbeat(&profile, &status, 0); + let state = &status.snapshot()[0]; + assert_eq!(state.state, TunnelState::Running); + assert!(state.tunnel_ready && state.local_mcp_ready); + assert_eq!(state.shutdown_outcome, None); + assert_eq!( + state.diagnostic.as_deref(), + Some("tunnel_health_unavailable") + ); + assert_eq!(std::fs::read(&profile.readiness_path).unwrap(), previous); + } + + // Live observations continue changing while publication is unavailable. + status.update(0, |state| state.local_mcp_ready = false); + persist_heartbeat(&profile, &status, 0); + assert!(!status.snapshot()[0].local_mcp_ready); + reader.unlock().unwrap(); + persist_heartbeat(&profile, &status, 0); + let recovered: serde_json::Value = + serde_json::from_slice(&std::fs::read(&profile.readiness_path).unwrap()).unwrap(); + assert_eq!(recovered["state"], "running"); + assert_eq!(recovered["service_pid"], std::process::id()); + assert_eq!(recovered["profile_revision"], 7); + assert_eq!(recovered["tunnel_ready"], true); + assert_eq!(recovered["local_mcp_ready"], false); + assert!(recovered["diagnostic"].is_null()); + assert!(status.snapshot()[0].diagnostic.is_none()); +} + +#[test] +fn missing_heartbeat_file_preserves_live_status_without_recreating_it() { + let (_temp, profile, status) = fixture(); + std::fs::remove_file(&profile.readiness_path).unwrap(); + persist_heartbeat(&profile, &status, 0); + assert_eq!(status.snapshot()[0].state, TunnelState::Running); + assert_eq!( + status.snapshot()[0].diagnostic.as_deref(), + Some("tunnel_health_unavailable") + ); + assert!(!profile.readiness_path.exists()); + // Only repair of the precreated private file allows publication to recover. + create_health_file(&profile.readiness_path); + persist_heartbeat(&profile, &status, 0); + assert!(status.snapshot()[0].diagnostic.is_none()); + + // A real terminal failure still owns the final diagnostic and lifecycle. + status.update(0, |state| { + state.state = TunnelState::Failed; + state.tunnel_ready = false; + state.local_mcp_ready = false; + state.shutdown_outcome = Some("uncertain"); + state.diagnostic = Some("tunnel_restart_uncertain".into()); + }); + persist_status(&profile, &status, 0).unwrap(); + let terminal: serde_json::Value = + serde_json::from_slice(&std::fs::read(&profile.readiness_path).unwrap()).unwrap(); + assert_eq!(terminal["state"], "failed"); + assert_eq!(terminal["diagnostic"], "tunnel_restart_uncertain"); + assert_eq!(terminal["tunnel_ready"], false); + assert_eq!(terminal["local_mcp_ready"], false); +} + +#[tokio::test] +async fn unavailable_initial_status_still_blocks_profile_launch() { + let (_temp, profile, status) = fixture(); + std::fs::remove_file(&profile.readiness_path).unwrap(); + status.update(0, |state| { + state.state = TunnelState::Starting; + state.tunnel_ready = false; + state.local_mcp_ready = false; + }); + let (_stop, rx) = watch::channel(false); + run_profile( + 0, + profile, + "http://127.0.0.1:9/mcp".into(), + status.clone(), + rx, + ) + .await; + let state = &status.snapshot()[0]; + assert_eq!(state.state, TunnelState::Failed); + assert_eq!( + state.diagnostic.as_deref(), + Some("tunnel_health_unavailable") + ); + assert!(!state.tunnel_ready && !state.local_mcp_ready); + assert_eq!(state.shutdown_outcome, None); +}