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(