From e25c6bdd0ca0156c8ae6e11b1d470c7091d3325f Mon Sep 17 00:00:00 2001 From: KKRainbow <5665404+KKRainbow@users.noreply.github.com> Date: Sat, 17 Oct 2026 11:44:48 +0800 Subject: [PATCH] fix(web): prevent heartbeat replay and unsafe receiver takeover (#2718) * fix(web): stop replaying heartbeats after token validation Wake runtime reconciliation once when authentication establishes the session. Repeated successful validation, including a binding refresh, must not replay a heartbeat whose instance list predates a completed Run. Use existing notifications as wakeups and read the latest session heartbeat when reconciling, so delayed validation and config callbacks cannot replace a newer runtime observation with a cached payload. Cover one initial wakeup, 99 successful validations without replay, and a subsequent real heartbeat that still wakes runtime reconciliation. * fix(web): wait for authoritative config during receiver takeover Keep the desired webhook revision on the authenticated session. Gate runtime reconciliation, mutations and applied acknowledgements until the receiver has that exact revision, preserving running Core instances while its local configuration is missing or stale. After committed Full/Patch updates, request current validation authority without letting delayed publication callbacks or validation responses restore an older target. After the connected ACK establishes the route, wake existing validation once when configuration is still unavailable. Cover missing and obsolete state, authoritative empty sets, stale rounds, callback races, live Core handoff and an explicit 1000-connection takeover exercise with an independent mount. --- easytier-web/src/client_manager/mod.rs | 305 ++++++++++++++++- easytier-web/src/client_manager/session.rs | 75 ++++- .../session/runtime_revision.rs | 316 +++++++++++++++++- .../session/webhook_validation.rs | 97 +++++- 4 files changed, 758 insertions(+), 35 deletions(-) diff --git a/easytier-web/src/client_manager/mod.rs b/easytier-web/src/client_manager/mod.rs index a9df1b17..71ee2456 100644 --- a/easytier-web/src/client_manager/mod.rs +++ b/easytier-web/src/client_manager/mod.rs @@ -347,7 +347,11 @@ impl ClientManager { ) && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id) { session - .notify_managed_runtime_state_changed(user_id, machine_id) + .notify_managed_runtime_state_changed( + user_id, + machine_id, + config_revision.as_deref(), + ) .await; } Ok(()) @@ -392,13 +396,17 @@ impl ClientManager { machine_id, ManagedConfigPersistedChange { expected_revision: expected_config_revision, - target_revision: config_revision, + target_revision: config_revision.clone(), dirty_instance_ids, }, ); if changed && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id) { session - .notify_managed_runtime_state_changed(user_id, machine_id) + .notify_managed_runtime_state_changed( + user_id, + machine_id, + Some(&config_revision), + ) .await; } } @@ -416,7 +424,7 @@ impl ClientManager { && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id) { session - .notify_managed_runtime_state_changed(user_id, machine_id) + .notify_managed_runtime_state_changed(user_id, machine_id, None) .await; } } @@ -651,6 +659,7 @@ mod tests { #[derive(Debug, Clone)] struct TestWebhookState { validate_responses: Arc>>, + config_revision: Arc>, validate_count: Arc, block_second_validate: Arc, allow_second_validate: Arc, @@ -665,6 +674,7 @@ mod tests { validate_responses: Arc::new(tokio::sync::Mutex::new( validate_responses.into_iter().collect(), )), + config_revision: Arc::new(tokio::sync::RwLock::new("rev-initial".to_owned())), validate_count: Arc::new(AtomicUsize::new(0)), block_second_validate: Arc::new(AtomicBool::new(false)), allow_second_validate: Arc::new(AtomicBool::new(true)), @@ -728,8 +738,8 @@ mod tests { Json(json!({ "valid": true, - "binding_version": count, - "config_revision": format!("validated-rev-{count}") + "binding_version": 1, + "config_revision": state.config_revision.read().await.clone() })) } @@ -1270,7 +1280,8 @@ mod tests { #[tokio::test] async fn compact_runtime_preserves_unsupported_web_config_during_hot_patch() { - let (webhook_config, webhook_server, _) = test_webhook_config().await; + let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await; + *webhook_state.config_revision.write().await = "compact-initial".to_owned(); let mut mgr = ClientManager::new( Db::memory_db().await, None, @@ -1307,6 +1318,7 @@ mod tests { }) .await; + *webhook_state.config_revision.write().await = "compact-patched".to_owned(); mgr.reconcile_managed_network_configs( user_id, machine_id, @@ -1334,9 +1346,284 @@ mod tests { webhook_server.abort(); } + #[tokio::test] + async fn receiver_takeover_preserves_running_core_until_config_arrives() { + for stale_config in [false, true] { + let (old_webhook, old_server, _) = test_webhook_config().await; + let mut old = ClientManager::new( + Db::memory_db().await, + None, + HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(), + Arc::new(FeatureFlags::default()), + old_webhook, + ); + let old_addr = add_random_udp_listener(&mut old).await; + let machine_id = uuid::Uuid::new_v4(); + let instance_id = uuid::Uuid::new_v4(); + let old_user = old + .db() + .auto_create_user(MANAGED_CONFIG_TOKEN) + .await + .unwrap() + .id; + let desired = initial_managed_network_config(instance_id); + old.reconcile_managed_network_configs( + old_user, + machine_id, + vec![managed_config(instance_id, desired.clone())], + Some("rev-initial".to_owned()), + None, + ) + .await + .unwrap(); + let core = Arc::new(native_instance_manager()); + let old_client = start_web_client_for_test(old_addr, machine_id, core.clone()).await; + wait_for_applied_revision(&old, old_user, machine_id, "rev-initial").await; + let original = core.instance(instance_id).unwrap(); + drop(old_client); + + let state = TestWebhookState::with_blocked_connected([true]); + let (webhook, server, state) = test_webhook_config_with_state(state).await; + let mut replacement = ClientManager::new( + Db::memory_db().await, + None, + HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(), + Arc::new(FeatureFlags::default()), + webhook, + ); + let addr = add_random_udp_listener(&mut replacement).await; + let user_id = replacement + .db() + .auto_create_user(MANAGED_CONFIG_TOKEN) + .await + .unwrap() + .id; + if stale_config { + let obsolete_id = uuid::Uuid::new_v4(); + replacement + .reconcile_managed_network_configs( + user_id, + machine_id, + vec![managed_config( + obsolete_id, + initial_managed_network_config(obsolete_id), + )], + Some("outdated".to_owned()), + None, + ) + .await + .unwrap(); + } + let _client = start_web_client_for_test(addr, machine_id, core.clone()).await; + wait_for_condition( + || async { state.connected_count() == 1 }, + Duration::from_secs(5), + ) + .await; + let session = replacement + .get_session_by_machine_id(user_id, &machine_id) + .unwrap(); + + // Authentication and real heartbeats continue while publication is + // withheld. Neither an empty DB nor an obsolete Exact Set owns the + // already running Core instance. + for _ in 0..3 { + let before = session.get_heartbeat_req().await.unwrap().report_time; + wait_for_condition( + || async { session.get_heartbeat_req().await.unwrap().report_time != before }, + Duration::from_secs(3), + ) + .await; + assert_eq!(core.instance_ids(), vec![instance_id]); + assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap())); + assert!(session.applied_config_revision().await.is_none()); + } + + state.allow_connected(); + // Registration must wake validation before its ten-heartbeat + // periodic cadence, without repeatedly registering the same binding. + wait_for_condition( + || async { state.validate_count() == 2 }, + Duration::from_secs(2), + ) + .await; + assert_eq!(state.connected_count(), 1); + assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap())); + + replacement + .reconcile_managed_network_configs( + user_id, + machine_id, + vec![managed_config(instance_id, desired)], + Some("rev-initial".to_owned()), + None, + ) + .await + .unwrap(); + wait_for_applied_revision(&replacement, user_id, machine_id, "rev-initial").await; + assert_eq!(core.instance_ids(), vec![instance_id]); + assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap())); + + // A versioned authoritative empty set still removes web-owned + // instances and keeps the authenticated management session alive. + *state.config_revision.write().await = "empty".to_owned(); + replacement + .reconcile_managed_network_configs( + user_id, + machine_id, + Vec::new(), + Some("empty".to_owned()), + Some("rev-initial".to_owned()), + ) + .await + .unwrap(); + wait_for_applied_revision(&replacement, user_id, machine_id, "empty").await; + assert!(core.instance_ids().is_empty()); + assert!( + replacement + .get_session_by_machine_id(user_id, &machine_id) + .is_some() + ); + server.abort(); + old_server.abort(); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + #[ignore = "1000 real management connections; run explicitly with a raised file limit"] + async fn receiver_takeover_wave_allows_an_independent_mount() { + use futures::StreamExt; + + let (old_webhook, old_server, _) = test_webhook_config().await; + let mut old = ClientManager::new( + Db::memory_db().await, + None, + HeartbeatPolicy::default(), + Arc::new(FeatureFlags::default()), + old_webhook, + ); + let old_addr = add_random_udp_listener(&mut old).await; + let (webhook, server, state) = test_webhook_config().await; + let mut replacement = ClientManager::new( + Db::memory_db().await, + None, + HeartbeatPolicy::default(), + Arc::new(FeatureFlags::default()), + webhook, + ); + let addr = add_random_udp_listener(&mut replacement).await; + // Provision the account once: password hashing is unrelated to a wave + // of existing devices reconnecting to another receiver. + let old_user = old + .db() + .auto_create_user(MANAGED_CONFIG_TOKEN) + .await + .unwrap() + .id; + let user_id = replacement + .db() + .auto_create_user(MANAGED_CONFIG_TOKEN) + .await + .unwrap() + .id; + let devices: Vec<_> = (0..1_000) + .map(|_| { + ( + uuid::Uuid::new_v4(), + Arc::new(native_compact_instance_manager_with_runtime( + tokio::runtime::Handle::current(), + )), + ) + }) + .collect(); + let old_clients: Vec<_> = futures::stream::iter(&devices) + .map(|(machine_id, core)| async { + old.reconcile_managed_network_configs( + old_user, + *machine_id, + Vec::new(), + Some("rev-initial".to_owned()), + None, + ) + .await + .unwrap(); + let client = start_web_client_for_test(old_addr, *machine_id, core.clone()).await; + wait_for_applied_revision(&old, old_user, *machine_id, "rev-initial").await; + client + }) + .buffer_unordered(64) + .collect() + .await; + + let mount_machine = uuid::Uuid::new_v4(); + let mount_core = Arc::new(native_compact_instance_manager_with_runtime( + tokio::runtime::Handle::current(), + )); + let _mount_client = + start_web_client_for_test(addr, mount_machine, mount_core.clone()).await; + wait_for_validated_user(&replacement, mount_machine).await; + drop(old_clients); + let started = tokio::time::Instant::now(); + let takeover = futures::stream::iter(&devices) + .map(|(machine_id, core)| async { + let client = start_web_client_for_test(addr, *machine_id, core.clone()).await; + wait_for_validated_user(&replacement, *machine_id).await; + replacement + .reconcile_managed_network_configs( + user_id, + *machine_id, + Vec::new(), + Some("rev-initial".to_owned()), + None, + ) + .await + .unwrap(); + wait_for_applied_revision(&replacement, user_id, *machine_id, "rev-initial").await; + client + }) + .buffer_unordered(64) + .collect::>(); + let mount = async { + wait_for_condition( + || async { state.connected_count() >= 11 }, + Duration::from_secs(12), + ) + .await; + let started = tokio::time::Instant::now(); + let instance_id = uuid::Uuid::new_v4(); + replacement + .reconcile_managed_network_configs( + user_id, + mount_machine, + vec![managed_config( + instance_id, + initial_managed_network_config(instance_id), + )], + Some("rev-initial".to_owned()), + None, + ) + .await + .unwrap(); + wait_for_applied_revision(&replacement, user_id, mount_machine, "rev-initial").await; + assert_eq!(mount_core.instance_ids(), vec![instance_id]); + started.elapsed() + }; + let (clients, mount_elapsed) = tokio::join!(takeover, mount); + assert_eq!(clients.len(), 1_000); + assert_eq!(replacement.list_sessions().await.len(), 1_001); + println!( + "1000 receiver takeovers: {:?}; independent mount: {:?}; validations: {}", + started.elapsed(), + mount_elapsed, + state.validate_count() + ); + server.abort(); + old_server.abort(); + } + #[tokio::test] async fn managed_web_config_revision_updates_running_core_config() { - let (webhook_config, webhook_server, _) = test_webhook_config().await; + let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await; let mut mgr = ClientManager::new( Db::memory_db().await, None, @@ -1400,6 +1687,7 @@ mod tests { // Online revision update: web-owned running config is fully overwritten // when non-hot-patch flags such as enable_kcp_proxy change. + *webhook_state.config_revision.write().await = "rev-updated".to_owned(); mgr.reconcile_managed_network_configs( user_id, machine_id, @@ -1454,6 +1742,7 @@ mod tests { .is_none() ); + *webhook_state.config_revision.write().await = "rev-webhook-redelivery".to_owned(); mgr.reconcile_managed_network_configs( user_id, machine_id, diff --git a/easytier-web/src/client_manager/session.rs b/easytier-web/src/client_manager/session.rs index 0dabffbe..4af5ca77 100644 --- a/easytier-web/src/client_manager/session.rs +++ b/easytier-web/src/client_manager/session.rs @@ -133,9 +133,13 @@ pub struct SessionData { storage_token: Option, binding_version: Option, + // Authentication does not prove that this receiver has the desired Exact Set. + // Keep that authority on the Session, not in runtime continuity shared with + // an earlier connection. + webhook_config_revision: Option, managed_runtime: SharedManagedRuntimeState, direct_run_failed_instance_ids: HashSet, - notifier: broadcast::Sender, + notifier: broadcast::Sender<()>, req: Option, location: Option, heartbeat_count: std::sync::atomic::AtomicU32, @@ -165,6 +169,7 @@ impl SessionData { client_url, storage_token: None, binding_version: None, + webhook_config_revision: None, managed_runtime: Arc::new(Mutex::new(ManagedRuntimeState::default())), direct_run_failed_instance_ids: HashSet::new(), notifier: tx, @@ -185,10 +190,22 @@ impl SessionData { self.req.clone() } - pub fn heartbeat_waiter(&self) -> broadcast::Receiver { + pub fn heartbeat_waiter(&self) -> broadcast::Receiver<()> { self.notifier.subscribe() } + fn runtime_config_is_ready(&self, persisted_revision: Option<&str>) -> bool { + if !self.webhook_config.is_enabled() { + return true; + } + match self.webhook_config_revision.as_deref() { + // Standalone webhook integrations can omit both managed fields. + Some("") if self.binding_version == Some(0) => true, + Some(revision) if !revision.is_empty() => persisted_revision == Some(revision), + _ => false, + } + } + pub fn location(&self) -> Option<&Location> { self.location.as_ref() } @@ -302,6 +319,11 @@ async fn record_webhook_connected_binding_if_current( binding_version: u64, ) -> Option { let session_data = session_data.upgrade()?; + let storage = Storage::try_from(session_data.read().await.storage.clone()).ok()?; + let persisted_revision = storage + .db() + .get_managed_config_revision((storage_token.user_id, storage_token.machine_id)) + .await; let mut data = session_data.write().await; if !connection_state_matches(&data, storage_token, binding_version) { return Some(ConnectedBindingRecord::IdentityStale); @@ -313,7 +335,18 @@ async fn record_webhook_connected_binding_if_current( { return Some(ConnectedBindingRecord::OwnershipLost); } + let newly_connected = data.webhook_connected_binding_version != Some(binding_version); data.webhook_connected_binding_version = Some(binding_version); + if newly_connected + && data.webhook_config.is_enabled() + && !persisted_revision + .as_ref() + .is_ok_and(|revision| data.runtime_config_is_ready(revision.as_deref())) + { + // The connected ACK makes the receiver route publishable. Retry the + // existing validation path now, rather than waiting ten heartbeats. + SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data).notify_one(); + } Some(ConnectedBindingRecord::Recorded) } @@ -748,13 +781,13 @@ impl SessionRpcService { data.session_epoch, ); } - let runtime_notify = (authorized && data.storage_token.is_some()) - .then(|| (data.notifier.clone(), runtime_req)); + let runtime_notify = + (authorized && data.storage_token.is_some()).then(|| data.notifier.clone()); (notify, runtime_notify) }; - if let Some((notifier, runtime_req)) = runtime_notify { - let _ = notifier.send(runtime_req); + if let Some(notifier) = runtime_notify { + let _ = notifier.send(()); } if let Some(notify) = notify { notify.notify_one(); @@ -859,7 +892,7 @@ impl SessionRpcService { let report_time = Self::heartbeat_report_timestamp(&runtime_req); storage.update_session_client(storage_token, report_time, true, session_epoch); - let _ = notifier.send(runtime_req); + let _ = notifier.send(()); if let Some(notify) = validation_notify { notify.notify_one(); } @@ -1048,9 +1081,10 @@ impl Session { &self, user_id: i32, machine_id: uuid::Uuid, + published_revision: Option<&str>, ) { let notify = { - let data = self.data.read().await; + let mut data = self.data.write().await; if !data.auth_state.is_authorized() { return; } @@ -1061,10 +1095,21 @@ impl Session { { return; } - data.req.clone().map(|req| (data.notifier.clone(), req)) + if let Some(revision) = published_revision + && data.webhook_config.is_enabled() + && data.webhook_config_revision.as_deref() != Some(revision) + { + // Only validation establishes the desired revision. A delayed + // publication callback must not replace newer authority, even + // when its revision is still the receiver's persisted value. + // Fence validation already in flight and request current authority. + SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data) + .notify_one(); + } + data.req.as_ref().map(|_| data.notifier.clone()) }; - if let Some((notifier, req)) = notify { - let _ = notifier.send(req); + if let Some(notifier) = notify { + let _ = notifier.send(()); } } @@ -1082,10 +1127,10 @@ impl Session { runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1); runtime.runtime_config_cache_epoch = runtime.runtime_config_cache_epoch.wrapping_add(1); drop(runtime); - data.req.clone().map(|req| (data.notifier.clone(), req)) + data.req.as_ref().map(|_| data.notifier.clone()) }; - if let Some((notifier, req)) = notify { - let _ = notifier.send(req); + if let Some(notifier) = notify { + let _ = notifier.send(()); } } @@ -1805,7 +1850,7 @@ mod tests { assert_eq!(data.binding_version, Some(1)); assert_eq!(data.webhook_connected_binding_version, Some(1)); drop(data); - assert_eq!(heartbeat_waiter.recv().await.unwrap().user_token, "token"); + heartbeat_waiter.recv().await.unwrap(); assert!( storage .db() diff --git a/easytier-web/src/client_manager/session/runtime_revision.rs b/easytier-web/src/client_manager/session/runtime_revision.rs index 889808fd..fd9ce08a 100644 --- a/easytier-web/src/client_manager/session/runtime_revision.rs +++ b/easytier-web/src/client_manager/session/runtime_revision.rs @@ -24,11 +24,12 @@ use crate::client_manager::{ }; async fn recv_latest_heartbeat( - heartbeat_waiter: &mut broadcast::Receiver, + session_data: &std::sync::Weak>, + heartbeat_waiter: &mut broadcast::Receiver<()>, ) -> Option { - let mut req = loop { + loop { match heartbeat_waiter.recv().await { - Ok(req) => break req, + Ok(()) => break, Err(broadcast::error::RecvError::Lagged(skipped)) => { tracing::warn!( skipped, @@ -40,21 +41,23 @@ async fn recv_latest_heartbeat( return None; } } - }; + } // Drop any heartbeat backlog accumulated while the previous reconcile // round was doing DB/RPC IO. The newest heartbeat has the freshest // runtime instance list, which is all this task needs. loop { match heartbeat_waiter.try_recv() { - Ok(next_req) => req = next_req, + Ok(()) => {} Err(broadcast::error::TryRecvError::Empty) => break, Err(broadcast::error::TryRecvError::Lagged(_)) => continue, Err(broadcast::error::TryRecvError::Closed) => return None, } } - Some(req) + // Notifications can be delayed by validation or publication IO. Read the + // current heartbeat instead of replaying the snapshot taken before that IO. + session_data.upgrade()?.read().await.req() } struct SessionClients { @@ -65,7 +68,7 @@ struct SessionClients { pub(super) async fn reconcile_network_configs_on_heartbeat( session_data: std::sync::Weak>, - mut heartbeat_waiter: broadcast::Receiver, + mut heartbeat_waiter: broadcast::Receiver<()>, storage: WeakRefStorage, rpc_client: SessionRpcClient, config_client: SessionConfigClient, @@ -78,7 +81,7 @@ pub(super) async fn reconcile_network_configs_on_heartbeat( }; let mut cache = ReconcileCache::default(); loop { - let Some(req) = recv_latest_heartbeat(&mut heartbeat_waiter).await else { + let Some(req) = recv_latest_heartbeat(&session_data, &mut heartbeat_waiter).await else { return; }; let Some(storage) = storage.upgrade() else { @@ -404,6 +407,15 @@ async fn prepare_reconcile_round( RoundStatus::Skip => return RoundStatus::Skip, RoundStatus::Stop => return RoundStatus::Stop, }; + { + let Some(data) = session_data.upgrade() else { + return RoundStatus::Stop; + }; + let data = data.read().await; + if !data.runtime_config_is_ready(target_config_revision.as_deref()) { + return RoundStatus::Skip; + } + } let should_apply_runtime_revision = target_config_revision.is_some() && target_config_revision != applied_config_revision; let mut scope = if should_apply_runtime_revision { @@ -886,7 +898,9 @@ async fn begin_managed_runtime_mutation( return false; }; let data = data.write().await; - if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) { + if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) + || !data.runtime_config_is_ready(round.target_config_revision.as_deref()) + { return false; } let managed_runtime = data.managed_runtime.clone(); @@ -1265,7 +1279,9 @@ async fn mark_config_revision_applied_if_current( }; let notify = { let mut data = data.write().await; - if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) { + if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) + || !data.runtime_config_is_ready(round.target_config_revision.as_deref()) + { return RoundStatus::Ready(()); } record_applied_config_revision( @@ -1319,6 +1335,286 @@ mod tests { use super::*; + async fn takeover_session( + persisted_revision: Option<&str>, + ) -> ( + crate::client_manager::storage::Storage, + std::sync::Arc>, + ReconcileRound, + ) { + use super::super::{SessionAuthState, webhook_validation}; + use crate::client_manager::storage::{Storage, StorageToken}; + + let storage = Storage::new(crate::db::Db::memory_db().await); + let user_id = storage.db().auto_create_user("token").await.unwrap().id; + let machine_id = uuid::Uuid::new_v4(); + if let Some(revision) = persisted_revision { + storage + .db() + .set_managed_config_revision((user_id, machine_id), revision) + .await + .unwrap(); + } + let req = HeartbeatRequest { + user_token: "token".to_owned(), + machine_id: Some(machine_id.into()), + running_network_instances: vec![uuid::Uuid::new_v4().into()], + ..Default::default() + }; + let client_url = "tcp://127.0.0.1:1001".parse().unwrap(); + let webhook = std::sync::Arc::new(crate::webhook::WebhookConfig::new( + Some("http://127.0.0.1:1".to_owned()), + None, + None, + None, + None, + )); + let mut data = SessionData::new( + storage.weak_ref(), + client_url, + None, + std::sync::Arc::new(crate::FeatureFlags::default()), + webhook.clone(), + ); + data.req = Some(req.clone()); + data.auth_state = SessionAuthState::Authorized; + data.binding_version = Some(1); + data.webhook_connected_binding_version = Some(1); + data.storage_token = Some(StorageToken { + token: req.user_token.clone(), + client_url: data.client_url.clone(), + machine_id, + user_id, + }); + data.managed_runtime = storage.bind_managed_runtime_state(user_id, machine_id, None, 0); + data.managed_runtime().applied_config_revision = persisted_revision.map(str::to_owned); + let session_data = std::sync::Arc::new(RwLock::new(data)); + let client_url = session_data.read().await.client_url.clone(); + webhook_validation::apply_success( + &std::sync::Arc::downgrade(&session_data), + webhook_validation::WebhookValidationInput { + storage: storage.clone(), + webhook_config: webhook, + client_url, + applied_config_revision: persisted_revision.map(str::to_owned), + applied_config_revision_known: false, + failed_instance_ids: Vec::new(), + req: req.clone(), + machine_id, + }, + webhook_validation::WebhookHeartbeatValidation { + config_revision: "current".to_owned(), + binding_version: 1, + }, + user_id, + 0, + ) + .await; + let mut round = round_with_revision_state(persisted_revision, false); + round.req = req; + round.machine_id = machine_id; + round.user_id = user_id; + round.runtime_config_epoch = session_data + .read() + .await + .managed_runtime() + .runtime_config_epoch; + (storage, session_data, round) + } + + #[tokio::test] + async fn takeover_skips_missing_and_outdated_receiver_config() { + for revision in [None, Some("outdated")] { + let (storage, data, round) = takeover_session(revision).await; + let rpc = easytier::proto::rpc::bidirect::BidirectRpcManager::new(); + let mut client: SessionRpcClient = rpc.rpc_client().scoped_client::< + easytier::proto::api::manage::WebClientServiceClientFactory, + >(1, 1, String::new()); + + assert!(matches!( + prepare_reconcile_round( + &std::sync::Arc::downgrade(&data), + &storage.weak_ref().upgrade().unwrap(), + &mut client, + round.req, + ) + .await, + RoundStatus::Skip + )); + } + } + + #[tokio::test] + async fn takeover_fences_mutations_and_applied_reports_from_unready_rounds() { + for revision in [None, Some("outdated")] { + let (storage, data, mut round) = takeover_session(revision).await; + let weak = std::sync::Arc::downgrade(&data); + let mut mutation = RuntimeMutationFence::default(); + assert!(!begin_managed_runtime_mutation(&weak, &round, &mut mutation).await); + assert!(!mutation.started); + + round.should_apply_runtime_revision = true; + mark_config_revision_applied_if_current( + &weak, + &storage.weak_ref().upgrade().unwrap(), + &round, + &ReconcileOutcome::default(), + &mutation, + ) + .await; + let data = data.read().await; + let runtime = data.managed_runtime(); + assert!(!runtime.applied_config_revision_known); + assert_eq!(runtime.applied_config_revision.as_deref(), revision); + } + } + + async fn validate_takeover_session( + storage: &crate::client_manager::storage::Storage, + data: &std::sync::Arc>, + round: &ReconcileRound, + revision: &str, + epoch: u64, + ) { + use super::super::webhook_validation; + + let (webhook_config, client_url) = { + let data = data.read().await; + (data.webhook_config.clone(), data.client_url.clone()) + }; + webhook_validation::apply_success( + &std::sync::Arc::downgrade(data), + webhook_validation::WebhookValidationInput { + storage: storage.clone(), + webhook_config, + client_url, + applied_config_revision: None, + applied_config_revision_known: false, + failed_instance_ids: Vec::new(), + req: round.req.clone(), + machine_id: round.machine_id, + }, + webhook_validation::WebhookHeartbeatValidation { + config_revision: revision.to_owned(), + binding_version: 1, + }, + round.user_id, + epoch, + ) + .await; + } + + #[tokio::test] + async fn takeover_new_publication_fences_late_validation_and_publication_callbacks() { + use super::super::{HeartbeatPolicy, Session}; + use std::sync::Arc; + + let (storage, data, round) = takeover_session(Some("current")).await; + let mut session = Session::new( + storage.weak_ref(), + "tcp://127.0.0.1:1001".parse().unwrap(), + None, + HeartbeatPolicy::default(), + Arc::new(crate::FeatureFlags::default()), + data.read().await.webhook_config.clone(), + 0, + ); + session.data = data.clone(); + let old_epoch = data.read().await.webhook_validation_change_epoch; + storage + .db() + .set_managed_config_revision((round.user_id, round.machine_id), "newer") + .await + .unwrap(); + storage.record_full_managed_config_change(round.user_id, round.machine_id, Some("newer")); + session + .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer")) + .await; + + validate_takeover_session(&storage, &data, &round, "current", old_epoch).await; + session + .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("current")) + .await; + let current_epoch = { + let data = data.read().await; + assert_eq!(data.webhook_config_revision.as_deref(), Some("current")); + assert!(!data.runtime_config_is_ready(Some("newer"))); + assert!(data.webhook_validation_dirty); + data.webhook_validation_change_epoch + }; + assert!( + !begin_managed_runtime_mutation( + &Arc::downgrade(&data), + &round, + &mut RuntimeMutationFence::default() + ) + .await + ); + + validate_takeover_session(&storage, &data, &round, "newer", current_epoch).await; + assert!(data.read().await.runtime_config_is_ready(Some("newer"))); + + // A newer Console revision may be observed after a publication commits + // but before its callback runs. SQLite still containing that publication + // does not allow the late callback to replace the newer authority. + validate_takeover_session(&storage, &data, &round, "newest", current_epoch).await; + session + .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer")) + .await; + let data = data.read().await; + assert_eq!(data.webhook_config_revision.as_deref(), Some("newest")); + assert!(!data.runtime_config_is_ready(Some("newer"))); + assert!(data.webhook_validation_dirty); + } + + #[tokio::test] + async fn unchanged_validation_does_not_replay_an_old_heartbeat() { + let (storage, data, round) = takeover_session(Some("current")).await; + let mut notifications = data.read().await.heartbeat_waiter(); + let epoch = data.read().await.webhook_validation_change_epoch; + validate_takeover_session(&storage, &data, &round, "current", epoch).await; + assert!(matches!( + notifications.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + )); + + let newest_instance = uuid::Uuid::new_v4(); + { + let mut data = data.write().await; + data.notifier.send(()).unwrap(); + data.req.as_mut().unwrap().running_network_instances = vec![newest_instance.into()]; + } + let req = recv_latest_heartbeat(&std::sync::Arc::downgrade(&data), &mut notifications) + .await + .unwrap(); + assert_eq!(req.running_network_instances, vec![newest_instance.into()]); + } + + #[tokio::test] + async fn authoritative_empty_config_remains_ready_but_missing_revision_does_not() { + let (storage, data, round) = takeover_session(Some("current")).await; + let mut mutation = RuntimeMutationFence::default(); + assert!(round.local_configs.is_empty()); + assert!( + begin_managed_runtime_mutation( + &std::sync::Arc::downgrade(&data), + &round, + &mut mutation + ) + .await + ); + let epoch = data.read().await.webhook_validation_change_epoch; + validate_takeover_session(&storage, &data, &round, "", epoch).await; + let mut data = data.write().await; + assert!(!data.runtime_config_is_ready(None)); + assert!(!data.runtime_config_is_ready(Some("current"))); + data.binding_version = Some(0); + assert!( + data.runtime_config_is_ready(None), + "legacy webhook without managed fields" + ); + } + fn config_with_port_forwards(port_forwards: Vec) -> NetworkConfig { NetworkConfig { instance_id: Some("11111111-1111-1111-1111-111111111111".to_string()), diff --git a/easytier-web/src/client_manager/session/webhook_validation.rs b/easytier-web/src/client_manager/session/webhook_validation.rs index 407a44ea..484a5b13 100644 --- a/easytier-web/src/client_manager/session/webhook_validation.rs +++ b/easytier-web/src/client_manager/session/webhook_validation.rs @@ -393,7 +393,7 @@ pub(super) async fn apply_success( validation_change_epoch: u64, ) { let WebhookHeartbeatValidation { - config_revision: _, + config_revision, binding_version, } = validation; @@ -408,6 +408,7 @@ pub(super) async fn apply_success( validation_notify, runtime_req, session_epoch, + notify_runtime, ) = { let mut data = session_data.write().await; let Some(runtime_req) = data.req.clone() else { @@ -436,6 +437,9 @@ pub(super) async fn apply_success( return; } + let notify_runtime = !data.auth_state.is_authorized() + || data.webhook_config_revision.as_deref() != Some(config_revision.as_str()); + let runtime_changed = notify_runtime || data.binding_version != Some(binding_version); let previous_connected_binding_version = data.webhook_connected_binding_version; let client_url = data.client_url.clone(); let is_new_storage_token = data.storage_token.is_none(); @@ -461,6 +465,11 @@ pub(super) async fn apply_success( let storage_token = storage_token.clone(); data.auth_state = SessionAuthState::Authorized; data.binding_version = Some(binding_version); + data.webhook_config_revision = Some(config_revision); + if runtime_changed { + let mut runtime = data.managed_runtime(); + runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1); + } if is_new_storage_token { tracing::info!( machine_id = %input.machine_id, @@ -520,6 +529,7 @@ pub(super) async fn apply_success( validation_notify, runtime_req, data.session_epoch, + notify_runtime, ) }; @@ -541,7 +551,9 @@ pub(super) async fn apply_success( .await; } - let _ = notifier.send(runtime_req); + if notify_runtime { + let _ = notifier.send(()); + } } async fn wait_webhook_connection_transition( @@ -563,6 +575,87 @@ async fn wait_webhook_connection_transition( mod tests { use super::*; + #[tokio::test] + async fn unchanged_validation_does_not_repeat_runtime_notifications() { + let storage = Storage::new(crate::db::Db::memory_db().await); + let user_id = storage.db().auto_create_user("token").await.unwrap().id; + let machine_id = uuid::Uuid::new_v4(); + let session_data = validation_session(machine_id).await; + { + let mut data = session_data.write().await; + data.storage = storage.weak_ref(); + data.auth_state = SessionAuthState::Init; + } + let mut notifications = session_data.read().await.heartbeat_waiter(); + for attempt in 0..100 { + let (input, epoch) = { + let data = session_data.read().await; + ( + WebhookValidationInput { + storage: storage.clone(), + webhook_config: data.webhook_config.clone(), + client_url: data.client_url.clone(), + applied_config_revision: (attempt > 0).then(|| "current".to_owned()), + applied_config_revision_known: attempt > 0, + failed_instance_ids: Vec::new(), + req: data.req.clone().unwrap(), + machine_id, + }, + data.webhook_validation_change_epoch, + ) + }; + apply_success( + &Arc::downgrade(&session_data), + input, + WebhookHeartbeatValidation { + config_revision: "current".to_owned(), + binding_version: if attempt == 99 { 2 } else { 1 }, + }, + user_id, + epoch, + ) + .await; + if attempt == 0 { + notifications + .try_recv() + .expect("initial authentication wakes runtime"); + let data = session_data.read().await; + let mut runtime = data.managed_runtime(); + runtime.applied_config_revision = Some("current".to_owned()); + runtime.applied_config_revision_known = true; + } else { + assert!(matches!( + notifications.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + } + // A fresh heartbeat still wakes the reconciler after repeated success. + let instance_id = uuid::Uuid::new_v4(); + let mut fresh = session_data.read().await.req.clone().unwrap(); + fresh.running_network_instances = vec![instance_id.into()]; + SessionRpcService { + data: session_data.clone(), + heartbeat_policy: super::super::HeartbeatPolicy::default(), + } + .handle_heartbeat(fresh) + .await + .unwrap(); + notifications + .try_recv() + .expect("fresh heartbeat wakes runtime"); + assert_eq!( + session_data + .read() + .await + .req + .as_ref() + .unwrap() + .running_network_instances, + vec![instance_id.into()] + ); + } + async fn validation_session(machine_id: uuid::Uuid) -> Arc> { let storage = Storage::new(crate::db::Db::memory_db().await); let mut data = SessionData::new(