diff --git a/src/api/config_store/atomic.rs b/src/api/config_store/atomic.rs index 6d546b6..a238d26 100644 --- a/src/api/config_store/atomic.rs +++ b/src/api/config_store/atomic.rs @@ -11,6 +11,8 @@ use nix::fcntl::{Flock, FlockArg, OFlag, openat, renameat}; use nix::sys::stat::Mode; #[cfg(unix)] use nix::unistd::{UnlinkatFlags, fsync, unlinkat}; +#[cfg(unix)] +use tracing::warn; use super::compute_source_revision; use crate::api::model::ApiFailure; @@ -75,7 +77,7 @@ impl ConfigWriteLock { } } -/// Replaces one config source through a durable same-directory rename. +/// Replaces one config source through a same-directory rename after syncing file data. pub(in crate::api) async fn write_atomic( path: PathBuf, contents: String, @@ -362,7 +364,15 @@ fn write_and_publish( anchored.name(), ) .map_err(errno_to_io)?; - fsync(anchored.parent()).map_err(errno_to_io)?; + // Rename is the commit boundary. A later directory-sync error cannot be reported as an + // uncommitted mutation because mandatory in-process publication must still run. + if let Err(error) = fsync(anchored.parent()).map_err(errno_to_io) { + warn!( + path = %path.display(), + %error, + "Config rename committed but directory durability sync failed" + ); + } Ok(committed_revision) } diff --git a/src/api/config_store/tests.rs b/src/api/config_store/tests.rs index 5589002..dd7ff98 100644 --- a/src/api/config_store/tests.rs +++ b/src/api/config_store/tests.rs @@ -346,13 +346,13 @@ async fn config_sidecar_lock_serializes_competing_revision_writers() { let second = second.await.unwrap(); assert_ne!(first.is_ok(), second.is_ok()); - let conflict = if let Err(error) = first { - error - } else { - second.unwrap_err() + let (winner_revision, conflict) = match (first, second) { + (Ok(revision), Err(error)) | (Err(error), Ok(revision)) => (revision, error), + _ => unreachable!("exactly one cooperative writer must commit"), }; assert_eq!(conflict.code, "revision_conflict"); - let persisted = tokio::fs::read_to_string(path).await.unwrap(); + assert_eq!(winner_revision, current_revision(&path).await.unwrap()); + let persisted = tokio::fs::read_to_string(&path).await.unwrap(); assert!(persisted.contains("first.example") || persisted.contains("second.example")); } diff --git a/src/proxy/traffic_limiter/tests.rs b/src/proxy/traffic_limiter/tests.rs index ad352c1..b1a1e2d 100644 --- a/src/proxy/traffic_limiter/tests.rs +++ b/src/proxy/traffic_limiter/tests.rs @@ -281,3 +281,47 @@ fn active_lease_observes_policy_removal() { assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 1); } + +#[test] +fn active_lease_observes_policy_addition() { + let limiter = TrafficLimiter::new(); + let lease = limiter + .acquire_lease("alice", "203.0.113.7".parse().unwrap()) + .unwrap(); + assert_eq!(lease.try_consume(RateDirection::Up, 2).granted, 2); + + let mut user_limits = HashMap::new(); + user_limits.insert("alice".to_string(), rate(1, 0)); + limiter.apply_policy(user_limits, HashMap::new()); + + assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 1); + assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 0); +} + +#[test] +fn reservation_refund_stays_with_retired_binding() { + let limiter = TrafficLimiter::new(); + let mut user_limits = HashMap::new(); + user_limits.insert("alice".to_string(), rate(400_000, 400_000)); + limiter.apply_policy(user_limits, HashMap::new()); + let lease = limiter + .acquire_lease("alice", "203.0.113.7".parse().unwrap()) + .unwrap(); + + let reservation = lease.try_reserve(RateDirection::Down, 800); + let old_bucket = Arc::clone( + reservation + ._binding + .user_bucket + .as_ref() + .expect("the original policy must bind a user bucket"), + ); + let epoch = reservation.user.as_ref().unwrap().epoch; + limiter.apply_policy(HashMap::new(), HashMap::new()); + assert_eq!(lease.try_consume(RateDirection::Down, 1).granted, 1); + assert!(lease.binding.load().user_bucket.is_none()); + + drop(reservation); + + assert_eq!(old_bucket.down.used_at(epoch), Some(0)); +} diff --git a/src/proxy/user_admission.rs b/src/proxy/user_admission.rs index 32a3b3e..3c835af 100644 --- a/src/proxy/user_admission.rs +++ b/src/proxy/user_admission.rs @@ -1,8 +1,8 @@ use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::atomic::AtomicU8; -use parking_lot::{Mutex, MutexGuard}; +use parking_lot::Mutex; use tokio_util::sync::CancellationToken; use crate::crypto::sha256; @@ -12,6 +12,10 @@ const REGISTRATION_PENDING: u8 = 0; const REGISTRATION_ACTIVE: u8 = 1; const REGISTRATION_DROPPED: u8 = 2; +// Authenticated-owner publication and RAII deregistration. +mod registration; +pub(crate) use registration::{UserAdmissionPublication, UserSessionRegistration}; + /// Stable secret identity used to fence authentication across runtime generations. pub(crate) type UserCredentialId = [u8; 16]; @@ -501,96 +505,6 @@ impl UserAdmissionAuthority { } } -/// Authority lock retained until the caller publishes its owned object. -pub(crate) struct UserAdmissionPublication<'a> { - state: MutexGuard<'a, UserAdmissionState>, - authority: Arc, - user: String, - registration_id: u64, - incarnation: UserIncarnation, - token: CancellationToken, - active: Arc, - registration_taken: bool, -} - -impl UserAdmissionPublication<'_> { - /// Moves the registered owner out while retaining the authority lock. - pub(crate) fn take_registration(&mut self) -> Option { - if self.registration_taken { - return None; - } - self.registration_taken = true; - Some(UserSessionRegistration { - authority: Arc::clone(&self.authority), - user: self.user.clone(), - registration_id: self.registration_id, - incarnation: self.incarnation, - token: self.token.clone(), - active: Arc::clone(&self.active), - }) - } - - /// Commits the owner record after the caller publishes its lifecycle object. - pub(crate) fn commit(mut self) { - if !self.registration_taken { - return; - } - if self.active.compare_exchange( - REGISTRATION_PENDING, REGISTRATION_ACTIVE, Ordering::AcqRel, Ordering::Acquire, - ).is_err() { - return; - } - self.state - .owners_by_user - .entry(self.user.clone()) - .or_default() - .insert( - self.registration_id, - RegisteredOwner { - token: self.token.clone(), - incarnation: self.incarnation, - }, - ); - } -} - -/// RAII ownership registered against one user incarnation. -#[must_use = "registered user ownership must be retained until lifecycle completion"] -pub(crate) struct UserSessionRegistration { - authority: Arc, - user: String, - registration_id: u64, - incarnation: UserIncarnation, - token: CancellationToken, - active: Arc, -} - -impl UserSessionRegistration { - /// Returns the cancellation signal for revocation or credential replacement. - pub(crate) fn token(&self) -> CancellationToken { - self.token.clone() - } - - /// Returns the immutable user incarnation owned by this registration. - pub(crate) fn incarnation(&self) -> UserIncarnation { - self.incarnation - } - - /// Returns whether revocation has cancelled this ownership. - pub(crate) fn is_cancelled(&self) -> bool { - self.token.is_cancelled() - } -} - -impl Drop for UserSessionRegistration { - fn drop(&mut self) { - if self.active.swap(REGISTRATION_DROPPED, Ordering::AcqRel) == REGISTRATION_ACTIVE { - self.authority - .unregister(&self.user, self.registration_id, self.incarnation); - } - } -} - /// Derives the stable credential identity from one decoded MTProxy secret. pub(crate) fn credential_id(secret: &[u8; 16]) -> UserCredentialId { let digest = sha256(secret); diff --git a/src/proxy/user_admission/registration.rs b/src/proxy/user_admission/registration.rs new file mode 100644 index 0000000..b69a61d --- /dev/null +++ b/src/proxy/user_admission/registration.rs @@ -0,0 +1,112 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicU8, Ordering}; + +use parking_lot::MutexGuard; +use tokio_util::sync::CancellationToken; + +use super::*; + +/// Authority lock retained until the caller publishes its owned object. +pub(crate) struct UserAdmissionPublication<'a> { + /// Locked authority state that linearizes publication with policy mutation. + pub(super) state: MutexGuard<'a, UserAdmissionState>, + /// Process authority used by the eventual registration guard. + pub(super) authority: Arc, + /// Username owning the published lifecycle object. + pub(super) user: String, + /// Unique registration identity within the process authority. + pub(super) registration_id: u64, + /// User incarnation authenticated by this publication. + pub(super) incarnation: UserIncarnation, + /// Revocation signal shared with the lifecycle owner. + pub(super) token: CancellationToken, + /// Atomic publication state shared with the registration guard. + pub(super) active: Arc, + /// Whether ownership has already moved into a registration guard. + pub(super) registration_taken: bool, +} + +impl UserAdmissionPublication<'_> { + /// Moves the registered owner out while retaining the authority lock. + pub(crate) fn take_registration(&mut self) -> Option { + if self.registration_taken { + return None; + } + self.registration_taken = true; + Some(UserSessionRegistration { + authority: Arc::clone(&self.authority), + user: self.user.clone(), + registration_id: self.registration_id, + incarnation: self.incarnation, + token: self.token.clone(), + active: Arc::clone(&self.active), + }) + } + + /// Commits the owner record after the caller publishes its lifecycle object. + pub(crate) fn commit(mut self) { + if !self.registration_taken { + return; + } + if self + .active + .compare_exchange( + REGISTRATION_PENDING, + REGISTRATION_ACTIVE, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_err() + { + return; + } + self.state + .owners_by_user + .entry(self.user.clone()) + .or_default() + .insert( + self.registration_id, + RegisteredOwner { + token: self.token.clone(), + incarnation: self.incarnation, + }, + ); + } +} + +/// RAII ownership registered against one user incarnation. +#[must_use = "registered user ownership must be retained until lifecycle completion"] +pub(crate) struct UserSessionRegistration { + authority: Arc, + user: String, + registration_id: u64, + incarnation: UserIncarnation, + token: CancellationToken, + active: Arc, +} + +impl UserSessionRegistration { + /// Returns the cancellation signal for revocation or credential replacement. + pub(crate) fn token(&self) -> CancellationToken { + self.token.clone() + } + + /// Returns the immutable user incarnation owned by this registration. + pub(crate) fn incarnation(&self) -> UserIncarnation { + self.incarnation + } + + /// Returns whether revocation has cancelled this ownership. + pub(crate) fn is_cancelled(&self) -> bool { + self.token.is_cancelled() + } +} + +impl Drop for UserSessionRegistration { + fn drop(&mut self) { + if self.active.swap(REGISTRATION_DROPPED, Ordering::AcqRel) == REGISTRATION_ACTIVE { + self.authority + .unregister(&self.user, self.registration_id, self.incarnation); + } + } +} diff --git a/src/proxy/user_admission/tests.rs b/src/proxy/user_admission/tests.rs index 0743078..c5fb769 100644 --- a/src/proxy/user_admission/tests.rs +++ b/src/proxy/user_admission/tests.rs @@ -60,6 +60,19 @@ fn stale_candidate_cannot_overwrite_newer_mutation() { .is_none() ); assert!(!authority.is_user_enabled("alice")); + + let disabled = HashMap::from([("alice".to_string(), false)]); + assert!( + authority + .apply_config_from_source(1, &users(secret), &disabled) + .is_none() + ); + assert!( + authority + .apply_config_from_source(2, &users(secret), &disabled) + .is_some() + ); + assert!(!authority.is_user_enabled("alice")); } #[test] diff --git a/src/stats/replay.rs b/src/stats/replay.rs index a8cc204..47f6113 100644 --- a/src/stats/replay.rs +++ b/src/stats/replay.rs @@ -90,6 +90,13 @@ struct ReplayShard { capacity: usize, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ReplayClaimResult { + Claimed, + Duplicate, + Capacity, +} + impl ReplayShard { fn new(cap: NonZeroUsize) -> Self { Self { @@ -169,22 +176,22 @@ impl ReplayShard { now: Instant, window: Duration, token: u64, - ) -> bool { + ) -> ReplayClaimResult { if window.is_zero() { - return true; + return ReplayClaimResult::Claimed; } self.cleanup(now, window); if self.cache.peek(key.as_slice()).is_some() || self.pending.contains_key(key.as_slice()) { - return false; + return ReplayClaimResult::Duplicate; } while self.cache.len().saturating_add(self.pending.len()) >= self.capacity { if self.queue.is_empty() { - return false; + return ReplayClaimResult::Capacity; } self.evict_queue_front(); } self.pending.insert(key, token); - true + ReplayClaimResult::Claimed } fn remove_pending(&mut self, key: &[u8], token: u64) -> bool { @@ -377,9 +384,16 @@ impl ReplayChecker { } let token = self.reserve_claim_token()?; let mut shard = self.tls_shards[shard_idx].lock(); - if !shard.claim_owned(key.clone(), Instant::now(), self.tls_window, token) { - self.hits.fetch_add(1, Ordering::Relaxed); - return None; + match shard.claim_owned(key.clone(), Instant::now(), self.tls_window, token) { + ReplayClaimResult::Claimed => {} + ReplayClaimResult::Duplicate => { + self.hits.fetch_add(1, Ordering::Relaxed); + return None; + } + ReplayClaimResult::Capacity => { + self.capacity_rejections.fetch_add(1, Ordering::Relaxed); + return None; + } } drop(shard); Some(TlsReplayClaim { @@ -478,6 +492,7 @@ pub struct ReplayStats { pub total_checks: u64, pub total_hits: u64, pub total_additions: u64, + /// Claims rejected because committed and pending entries exhausted a shard. pub total_capacity_rejections: u64, pub total_cleanups: u64, pub num_shards: usize, @@ -503,22 +518,5 @@ impl ReplayStats { } #[cfg(test)] -mod capacity_tests { - use super::*; - - #[test] - fn committed_and_pending_entries_share_one_shard_capacity() { - let capacity = NonZeroUsize::new(2).unwrap(); - let mut shard = ReplayShard::new(capacity); - let now = Instant::now(); - let window = Duration::from_secs(60); - assert!(shard.claim_owned(ReplayKey::from_slice(b"pending-a"), now, window, 1)); - assert!(shard.claim_owned(ReplayKey::from_slice(b"pending-b"), now, window, 2)); - - shard.add_owned(ReplayKey::from_slice(b"committed"), now, window); - - assert!(shard.len() <= capacity.get()); - assert!(shard.pending.contains_key(b"pending-a".as_slice())); - assert!(shard.pending.contains_key(b"pending-b".as_slice())); - } -} +#[path = "replay/tests.rs"] +mod capacity_tests; diff --git a/src/stats/replay/tests.rs b/src/stats/replay/tests.rs new file mode 100644 index 0000000..c1eb199 --- /dev/null +++ b/src/stats/replay/tests.rs @@ -0,0 +1,47 @@ +use super::*; + +#[test] +fn committed_and_pending_entries_share_one_shard_capacity() { + let capacity = NonZeroUsize::new(2).unwrap(); + let mut shard = ReplayShard::new(capacity); + let now = Instant::now(); + let window = Duration::from_secs(60); + assert_eq!( + shard.claim_owned(ReplayKey::from_slice(b"pending-a"), now, window, 1), + ReplayClaimResult::Claimed + ); + assert_eq!( + shard.claim_owned(ReplayKey::from_slice(b"pending-b"), now, window, 2), + ReplayClaimResult::Claimed + ); + + assert!(!shard.add_owned(ReplayKey::from_slice(b"committed"), now, window)); + + assert_eq!(shard.len(), capacity.get()); + assert!(shard.pending.contains_key(b"pending-a".as_slice())); + assert!(shard.pending.contains_key(b"pending-b".as_slice())); +} + +#[test] +fn pending_capacity_rejection_is_not_reported_as_a_replay_hit() { + let checker = ReplayChecker::new(64, Duration::from_secs(60)); + let first = checker + .claim_tls_digest(b"same-shard-key") + .expect("first claim must reserve the shard"); + let shard_idx = checker.get_shard_idx(b"same-shard-key"); + let rejected = (0..10_000u64) + .map(u64::to_le_bytes) + .find(|candidate| checker.get_shard_idx(candidate) == shard_idx) + .expect("bounded search must find a second key in the selected shard"); + let before = checker.stats(); + + assert!(checker.claim_tls_digest(&rejected).is_none()); + + let after = checker.stats(); + assert_eq!(after.total_hits, before.total_hits); + assert_eq!( + after.total_capacity_rejections, + before.total_capacity_rejections + 1 + ); + drop(first); +} diff --git a/src/transport/middle_proxy/pool_reinit.rs b/src/transport/middle_proxy/pool_reinit.rs index 9c9ff07..177561c 100644 --- a/src/transport/middle_proxy/pool_reinit.rs +++ b/src/transport/middle_proxy/pool_reinit.rs @@ -62,15 +62,22 @@ struct ReinitCommitOutcome { } #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -struct DcFamilyGroup { - dc: i32, - family: IpFamily, +/// One independently enforced hardswap writer-floor group. +pub(in crate::transport::middle_proxy) struct DcFamilyGroup { + /// Telegram DC owning the group. + pub(in crate::transport::middle_proxy) dc: i32, + /// Address family whose floor is evaluated independently. + pub(in crate::transport::middle_proxy) family: IpFamily, } -struct HardswapCoverage { - ratio: f32, - missing_groups: Vec, - writer_deficit: usize, +/// Complete floor-coverage result for one candidate hardswap generation. +pub(in crate::transport::middle_proxy) struct HardswapCoverage { + /// Fraction of configured DC-family groups that reached their full floor. + pub(in crate::transport::middle_proxy) ratio: f32, + /// Stable list of DC-family groups that remain below their floor. + pub(in crate::transport::middle_proxy) missing_groups: Vec, + /// Total writer count still required across all missing groups. + pub(in crate::transport::middle_proxy) writer_deficit: usize, } #[derive(Debug)] diff --git a/src/transport/middle_proxy/pool_reinit/coordination.rs b/src/transport/middle_proxy/pool_reinit/coordination.rs index 736c798..0b270eb 100644 --- a/src/transport/middle_proxy/pool_reinit/coordination.rs +++ b/src/transport/middle_proxy/pool_reinit/coordination.rs @@ -287,7 +287,7 @@ impl MePool { } /// Evaluates full writer-floor coverage independently for every DC and address family. - pub(super) fn hardswap_coverage( + pub(in crate::transport::middle_proxy) fn hardswap_coverage( &self, desired_by_dc: &HashMap>, writer_addrs: &[(i32, SocketAddr)], diff --git a/src/transport/middle_proxy/pool_reinit/tests.rs b/src/transport/middle_proxy/pool_reinit/tests.rs index 25cb58f..a0abe67 100644 --- a/src/transport/middle_proxy/pool_reinit/tests.rs +++ b/src/transport/middle_proxy/pool_reinit/tests.rs @@ -537,7 +537,12 @@ async fn generation_role_reconciliation_promotes_active_warm_and_drains_orphans( assert!(orphan_warm.draining.load(Ordering::Acquire)); assert!(!orphan_warm.allow_drain_fallback.load(Ordering::Acquire)); let snapshot = pool.api_hardswap_snapshot().await; + let desired = pool.desired_dc_endpoints().await; + let coverage = pool.hardswap_coverage(&desired, &[(1, endpoint)]); assert_eq!(snapshot.orphan_warm_writers_current, 0); - assert_eq!(snapshot.pending_writer_deficit, 2); - assert_eq!(snapshot.pending_missing_dc_groups, 1); + assert_eq!(snapshot.pending_writer_deficit, coverage.writer_deficit); + assert_eq!( + snapshot.pending_missing_dc_groups, + coverage.missing_groups.len() + ); } diff --git a/src/transport/middle_proxy/pool_status.rs b/src/transport/middle_proxy/pool_status.rs index d19a859..32fd56f 100644 --- a/src/transport/middle_proxy/pool_status.rs +++ b/src/transport/middle_proxy/pool_status.rs @@ -6,7 +6,6 @@ use std::time::Instant; use super::pool::{MePool, ReinitStatusSnapshot, WriterContour}; use crate::config::{MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy}; -use crate::network::IpFamily; use crate::transport::upstream::IpPreference; // ME writer and DC coverage snapshots. diff --git a/src/transport/middle_proxy/pool_status/hardswap_snapshot.rs b/src/transport/middle_proxy/pool_status/hardswap_snapshot.rs index 26aa8cd..1185c12 100644 --- a/src/transport/middle_proxy/pool_status/hardswap_snapshot.rs +++ b/src/transport/middle_proxy/pool_status/hardswap_snapshot.rs @@ -41,7 +41,7 @@ impl MePool { let pending_generation = reinit.pending_hardswap_generation; let pending = pending_generation != 0; let mut pending_writers_current = 0usize; - let mut pending_by_group = HashMap::<(i32, IpFamily), usize>::new(); + let mut pending_writer_addrs = Vec::<(i32, SocketAddr)>::new(); let mut orphan_warm_writers_current = 0usize; for writer in writers.iter() { @@ -60,40 +60,18 @@ impl MePool { .is_some_and(|endpoints| endpoints.contains(&writer.addr)) { pending_writers_current = pending_writers_current.saturating_add(1); - let family = if writer.addr.is_ipv4() { - IpFamily::V4 - } else { - IpFamily::V6 - }; - *pending_by_group - .entry((writer.writer_dc, family)) - .or_insert(0) += 1; + pending_writer_addrs.push((writer.writer_dc, writer.addr)); } } - let mut pending_writer_deficit = 0usize; - let mut pending_missing_dc_groups = 0usize; - if pending { - for (dc, endpoints) in &desired_by_dc { - for family in [IpFamily::V4, IpFamily::V6] { - let endpoint_count = endpoints - .iter() - .filter(|endpoint| endpoint.is_ipv4() == (family == IpFamily::V4)) - .count(); - if endpoint_count == 0 { - continue; - } - let alive = pending_by_group.get(&(*dc, family)).copied().unwrap_or(0); - let required = self.required_writers_for_dc(endpoint_count); - let deficit = required.saturating_sub(alive); - pending_writer_deficit = pending_writer_deficit.saturating_add(deficit); - if deficit > 0 { - pending_missing_dc_groups = - pending_missing_dc_groups.saturating_add(1); - } - } - } - } + let pending_coverage = pending + .then(|| self.hardswap_coverage(&desired_by_dc, &pending_writer_addrs)); + let pending_writer_deficit = pending_coverage + .as_ref() + .map_or(0, |coverage| coverage.writer_deficit); + let pending_missing_dc_groups = pending_coverage + .as_ref() + .map_or(0, |coverage| coverage.missing_groups.len()); let (replacement_preparing_current, replacement_retiring_current) = self.registry.writer_replacement_counts(); let pending_age_secs = pending.then(|| { diff --git a/src/web/http/generation_fence_tests.rs b/src/web/http/generation_fence_tests.rs new file mode 100644 index 0000000..053a3d5 --- /dev/null +++ b/src/web/http/generation_fence_tests.rs @@ -0,0 +1,51 @@ +use super::*; +use crate::web::manager::ManagerError; + +#[tokio::test] +async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() { + let initial = test_runtime_generation(1, runtime_config([21; 32], WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let replacement = test_runtime_generation(2, runtime_config([22; 32], WebCarrier::HttpsLanes)); + active_runtime.store(Arc::clone(&replacement)); + + let result = + runtime.issue_bootstrap_for_generation(&initial, profile, "192.0.2.10".parse().unwrap()); + + assert!(result.is_ok()); + runtime.shutdown().await; + initial.stop_sessions().await; + initial.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} + +#[tokio::test] +async fn stale_generation_cannot_publish_bootstrap_after_disabled_cutover() { + let capability = [23u8; 32]; + let initial = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); + let runtime = WebProcessRuntime::start(active_runtime); + let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let mut disabled_config = runtime_config(capability, WebCarrier::Https); + disabled_config.web.enabled = false; + let disabled = test_runtime_generation(2, disabled_config); + + runtime.activate_generation(Arc::clone(&disabled)); + let result = runtime.issue_bootstrap_for_generation( + &initial, + profile, + "192.0.2.10".parse().unwrap(), + ); + + assert!(matches!(result, Err(ManagerError::Closed))); + let status = serde_json::to_value(runtime.try_status()).unwrap(); + assert_eq!(status["manager"]["bootstraps"], 0); + + runtime.shutdown().await; + initial.stop_sessions().await; + initial.stop_background_tasks().await; + disabled.stop_sessions().await; + disabled.stop_background_tasks().await; +} diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index 1c80f28..1b1d2f3 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -17,8 +17,8 @@ use crate::config::{ use crate::maestro::generation::test_runtime_generation; use crate::web::frame::{self, FrameType}; use crate::web::manager::{ - CloseOperationSelector, ControlError, ManagerError, SessionDetail, SessionFilter, - SessionListRequest, SessionRefError, WebProcessRuntime, + CloseOperationSelector, ControlError, SessionDetail, SessionFilter, SessionListRequest, + SessionRefError, WebProcessRuntime, }; #[path = "legacy_tests.rs"] @@ -34,6 +34,9 @@ mod diagnostic_tests; // Reload-stability tests for session-owned timeout policy. #[path = "session_policy_tests.rs"] mod session_policy_tests; +// Runtime-generation authority fences for WEB bootstrap publication. +#[path = "generation_fence_tests.rs"] +mod generation_fence_tests; // Runtime control integration stays separate from carrier protocol scenarios. #[path = "control_tests.rs"] mod control_tests; @@ -413,55 +416,6 @@ async fn unused_bootstrap_survives_equivalent_runtime_generation_swap() { replacement.stop_background_tasks().await; } -#[tokio::test] -async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() { - let initial = test_runtime_generation(1, runtime_config([21; 32], WebCarrier::Https)); - let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); - let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); - let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); - let replacement = test_runtime_generation(2, runtime_config([22; 32], WebCarrier::HttpsLanes)); - active_runtime.store(Arc::clone(&replacement)); - - let result = - runtime.issue_bootstrap_for_generation(&initial, profile, "192.0.2.10".parse().unwrap()); - - assert!(result.is_ok()); - runtime.shutdown().await; - initial.stop_sessions().await; - initial.stop_background_tasks().await; - replacement.stop_sessions().await; - replacement.stop_background_tasks().await; -} - -#[tokio::test] -async fn stale_generation_cannot_publish_bootstrap_after_disabled_cutover() { - let capability = [23u8; 32]; - let initial = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); - let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); - let runtime = WebProcessRuntime::start(active_runtime); - let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); - let mut disabled_config = runtime_config(capability, WebCarrier::Https); - disabled_config.web.enabled = false; - let disabled = test_runtime_generation(2, disabled_config); - - runtime.activate_generation(Arc::clone(&disabled)); - let result = runtime.issue_bootstrap_for_generation( - &initial, - profile, - "192.0.2.10".parse().unwrap(), - ); - - assert!(matches!(result, Err(ManagerError::Closed))); - let status = serde_json::to_value(runtime.try_status()).unwrap(); - assert_eq!(status["manager"]["bootstraps"], 0); - - runtime.shutdown().await; - initial.stop_sessions().await; - initial.stop_background_tasks().await; - disabled.stop_sessions().await; - disabled.stop_background_tasks().await; -} - #[tokio::test] async fn unused_bootstrap_is_rejected_after_profile_identity_change() { let capability = [11u8; 32]; diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs index 295a3bb..ab3fc43 100644 --- a/src/web/session/lanes.rs +++ b/src/web/session/lanes.rs @@ -2,7 +2,6 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use bytes::{BufMut, Bytes, BytesMut}; -use tokio::sync::OwnedSemaphorePermit; use super::lane_downlink::take_lane_down_batch; use super::{ @@ -13,6 +12,9 @@ use crate::web::frame::{self, FrameType}; use crate::web::manager::ManagerError; use crate::web::telemetry::WebSessionLifecycleObservation; +// Bounded lane-open admission and cancellation-safe wait lifecycle. +mod lane_open_wait; + impl WebSession { /// Polls one lane with independent cursor replay and newest-poll-wins semantics. pub(crate) async fn poll_down_lane( @@ -311,97 +313,6 @@ impl WebSession { } } - async fn wait_for_lane_open(&self, lane_id: u32, cursor: u64) -> Result { - if self.close_if_cancelled() { - return Err(ManagerError::Closed); - } - let wait = { - let mut state = self.state.lock(); - if state.closed || self.cancel.is_cancelled() { - drop(state); - self.close_if_cancelled(); - return Err(ManagerError::Closed); - } - if state.carrier_lanes.contains_key(&lane_id) { - return Ok(true); - } - if cursor != 0 || lane_id == 0 { - drop(state); - self.close(SessionCloseReason::Protocol); - return Err(ManagerError::Protocol); - } - if state.closed_streams.contains(&lane_id) - || state.closing_streams.contains_key(&lane_id) - { - return Ok(true); - } - if state.lane_open_waits >= self.limits.max_lane_open_waits_per_session { - return Err(ManagerError::Limit); - } - let Some(manager) = self.manager.upgrade() else { - return Err(ManagerError::Closed); - }; - let Some(auxiliary) = manager.try_lane_poll(true) else { - return Err(ManagerError::Limit); - }; - state.lane_open_waits += 1; - let observation = WebSessionLifecycleObservation::HttpActivityAfterGap; - self.touch_peer_locked(&mut state, Instant::now(), observation); - LaneOpenWaitGuard { - session: self, - _auxiliary: auxiliary, - } - }; - let deadline = Duration::from_secs(self.timeouts.lane_open_wait_secs); - let opened = tokio::time::timeout(deadline, async { - loop { - let notified = self.lane_open_notify.notified(); - tokio::pin!(notified); - notified.as_mut().enable(); - { - let state = self.state.lock(); - if state.closed || self.cancel.is_cancelled() { - drop(state); - self.close_if_cancelled(); - return Err(ManagerError::Closed); - } - if state.carrier_lanes.contains_key(&lane_id) - || state.closed_streams.contains(&lane_id) - || state.closing_streams.contains_key(&lane_id) - { - return Ok(true); - } - } - notified.await; - } - }); - let opened = tokio::select! { - biased; - _ = self.cancel.cancelled() => { - drop(wait); - self.close_if_cancelled(); - return Err(ManagerError::Closed); - } - opened = opened => opened, - }; - drop(wait); - match opened { - Ok(result) => result, - Err(_) => { - let state = self.state.lock(); - if state.closed || self.cancel.is_cancelled() { - drop(state); - self.close_if_cancelled(); - Err(ManagerError::Closed) - } else { - Ok(state.carrier_lanes.contains_key(&lane_id) - || state.closed_streams.contains(&lane_id) - || state.closing_streams.contains_key(&lane_id)) - } - } - } - } - pub(super) fn queue_lane_frame_locked( &self, state: &mut SessionState, @@ -571,18 +482,6 @@ impl WebSession { } } -struct LaneOpenWaitGuard<'a> { - session: &'a WebSession, - _auxiliary: OwnedSemaphorePermit, -} - -impl Drop for LaneOpenWaitGuard<'_> { - fn drop(&mut self) { - let mut state = self.session.state.lock(); - state.lane_open_waits = state.lane_open_waits.saturating_sub(1); - } -} - // Lane-specific protocol, replay, and lifecycle tests. #[cfg(test)] mod tests; diff --git a/src/web/session/lanes/lane_open_wait.rs b/src/web/session/lanes/lane_open_wait.rs new file mode 100644 index 0000000..6a8a89d --- /dev/null +++ b/src/web/session/lanes/lane_open_wait.rs @@ -0,0 +1,118 @@ +use std::time::{Duration, Instant}; + +use tokio::sync::OwnedSemaphorePermit; + +use super::WebSession; +use crate::web::manager::ManagerError; +use crate::web::session::SessionCloseReason; +use crate::web::telemetry::WebSessionLifecycleObservation; + +impl WebSession { + /// Waits for a non-control lane to become observable within the bounded admission budget. + pub(super) async fn wait_for_lane_open( + &self, + lane_id: u32, + cursor: u64, + ) -> Result { + if self.close_if_cancelled() { + return Err(ManagerError::Closed); + } + let wait = { + let mut state = self.state.lock(); + if state.closed || self.cancel.is_cancelled() { + drop(state); + self.close_if_cancelled(); + return Err(ManagerError::Closed); + } + if state.carrier_lanes.contains_key(&lane_id) { + return Ok(true); + } + if cursor != 0 || lane_id == 0 { + drop(state); + self.close(SessionCloseReason::Protocol); + return Err(ManagerError::Protocol); + } + if state.closed_streams.contains(&lane_id) + || state.closing_streams.contains_key(&lane_id) + { + return Ok(true); + } + if state.lane_open_waits >= self.limits.max_lane_open_waits_per_session { + return Err(ManagerError::Limit); + } + let Some(manager) = self.manager.upgrade() else { + return Err(ManagerError::Closed); + }; + let Some(auxiliary) = manager.try_lane_poll(true) else { + return Err(ManagerError::Limit); + }; + state.lane_open_waits += 1; + let observation = WebSessionLifecycleObservation::HttpActivityAfterGap; + self.touch_peer_locked(&mut state, Instant::now(), observation); + LaneOpenWaitGuard { + session: self, + _auxiliary: auxiliary, + } + }; + let deadline = Duration::from_secs(self.timeouts.lane_open_wait_secs); + let opened = tokio::time::timeout(deadline, async { + loop { + let notified = self.lane_open_notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + { + let state = self.state.lock(); + if state.closed || self.cancel.is_cancelled() { + drop(state); + self.close_if_cancelled(); + return Err(ManagerError::Closed); + } + if state.carrier_lanes.contains_key(&lane_id) + || state.closed_streams.contains(&lane_id) + || state.closing_streams.contains_key(&lane_id) + { + return Ok(true); + } + } + notified.await; + } + }); + let opened = tokio::select! { + biased; + _ = self.cancel.cancelled() => { + drop(wait); + self.close_if_cancelled(); + return Err(ManagerError::Closed); + } + opened = opened => opened, + }; + drop(wait); + match opened { + Ok(result) => result, + Err(_) => { + let state = self.state.lock(); + if state.closed || self.cancel.is_cancelled() { + drop(state); + self.close_if_cancelled(); + Err(ManagerError::Closed) + } else { + Ok(state.carrier_lanes.contains_key(&lane_id) + || state.closed_streams.contains(&lane_id) + || state.closing_streams.contains_key(&lane_id)) + } + } + } + } +} + +struct LaneOpenWaitGuard<'a> { + session: &'a WebSession, + _auxiliary: OwnedSemaphorePermit, +} + +impl Drop for LaneOpenWaitGuard<'_> { + fn drop(&mut self) { + let mut state = self.session.state.lock(); + state.lane_open_waits = state.lane_open_waits.saturating_sub(1); + } +} diff --git a/src/web/session/negotiation.rs b/src/web/session/negotiation.rs index 9a2e539..8385ec0 100644 --- a/src/web/session/negotiation.rs +++ b/src/web/session/negotiation.rs @@ -299,258 +299,6 @@ impl WebSession { } } +// Carrier health publication and negotiation race tests. #[cfg(test)] -mod tests { - use std::net::SocketAddr; - use std::sync::{Arc, Barrier}; - - use super::*; - use crate::config::{ - WebCarrier, WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig, - }; - use crate::web::manager::{CarrierClientClass, WebProcessRuntime}; - use crate::web::session::{SessionCloseOutcome, SessionCloseReason}; - - fn session(carrier: WebCarrier, deadline: Instant) -> Arc { - let profile = Arc::new(WebRuntimeProfile { - host: "proxy.example.com".to_string(), - public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), - user: "alice".to_string(), - secret_mode: WebSecretMode::Plain, - carrier, - carrier_negotiation_enabled: true, - carrier_learning: false, - carriers: Arc::from([carrier]), - carrier_negotiation_deadlines_secs: [3, 5, 8, 12], - capability: [0; 32], - credential_id: [0; 16], - key_fingerprint: "0000000000000000".to_string(), - max_sessions: 1, - max_streams: 1, - max_streams_per_session: 1, - }); - WebSession::new( - std::sync::Weak::::new(), - [1; 32], - "192.0.2.10".parse().unwrap(), - 1, - profile, - [2; 32], - carrier, - 1, - [3; 32], - Some(deadline), - CarrierClientClass::Bridge, - None, - true, - false, - WebLimitsConfig::default(), - WebTimeoutsConfig::default(), - None, - ) - } - - fn arm_http_health(session: &WebSession, now: Instant) { - let mut state = session.state.lock(); - state.negotiation_phase = SessionNegotiationPhase::Committed; - state.carrier_commit_published = true; - state.carrier_health_due_at = Some(now - Duration::from_secs(1)); - state.carrier_health_uplink = true; - state.carrier_health_downlink = true; - state.carrier_health_activity_at = Some(now); - } - - #[test] - fn final_deadline_refuses_uncommitted_progress() { - let session = session(WebCarrier::Https, Instant::now() - Duration::from_secs(1)); - let state = session.state.lock(); - assert_eq!( - session.ensure_carrier_active_locked(&state), - Err(crate::web::manager::ManagerError::Closed) - ); - assert!(matches!( - state.negotiation_phase, - SessionNegotiationPhase::Uncommitted - )); - } - - #[test] - fn http_health_requires_authenticated_activity_after_the_window() { - let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - let now = Instant::now(); - let mut state = session.state.lock(); - state.negotiation_phase = SessionNegotiationPhase::Committed; - state.carrier_commit_published = true; - state.carrier_health_due_at = Some(now - Duration::from_secs(1)); - state.carrier_health_uplink = true; - state.carrier_health_downlink = true; - state.carrier_health_activity_at = Some(now - Duration::from_secs(2)); - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_none() - ); - state.carrier_health_activity_at = Some(now); - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_some() - ); - } - - #[test] - fn websocket_health_requires_the_exact_live_probe_owner() { - let session = session( - WebCarrier::Websocket, - Instant::now() + Duration::from_secs(60), - ); - let now = Instant::now(); - let mut state = session.state.lock(); - state.negotiation_phase = SessionNegotiationPhase::Committed; - state.carrier_commit_published = true; - state.carrier_health_due_at = Some(now - Duration::from_secs(1)); - state.websocket_carrier_active = true; - state.websocket_commit_ack_owner = Some(7); - state.websocket_commit_ack_written = true; - state.carrier_health_uplink = true; - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_none() - ); - state.websocket_probe_claimed = true; - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_some() - ); - } - - #[test] - fn health_waits_for_manager_commit_publication() { - let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - let now = Instant::now(); - let mut state = session.state.lock(); - state.negotiation_phase = SessionNegotiationPhase::Committed; - state.carrier_health_due_at = Some(now - Duration::from_secs(1)); - state.carrier_health_uplink = true; - state.carrier_health_downlink = true; - state.carrier_health_activity_at = Some(now); - - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_none() - ); - assert_eq!( - session.carrier_health_publication_state(), - CarrierHealthPublicationState::Awaiting - ); - } - - #[test] - fn health_publication_claim_is_single_shot() { - let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - let now = Instant::now(); - arm_http_health(&session, now); - let mut state = session.state.lock(); - - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_some() - ); - assert!( - session - .carrier_health_ready_locked(&mut state, now) - .is_none() - ); - drop(state); - assert_eq!( - session.carrier_health_publication_state(), - CarrierHealthPublicationState::Publishing - ); - assert!(session.publish_carrier_health()); - assert!(!session.publish_carrier_health()); - assert_eq!( - session.carrier_health_publication_state(), - CarrierHealthPublicationState::Published - ); - } - - #[test] - fn concurrent_health_and_close_always_reach_one_terminal_state() { - for _ in 0..512 { - let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - let now = Instant::now(); - arm_http_health(&session, now); - let barrier = Arc::new(Barrier::new(3)); - let health_session = Arc::clone(&session); - let health_barrier = Arc::clone(&barrier); - let health = std::thread::spawn(move || { - health_barrier.wait(); - std::thread::yield_now(); - let claim = { - let mut state = health_session.state.lock(); - health_session.carrier_health_ready_locked(&mut state, now) - }; - if claim.is_some() { - health_session.publish_carrier_health(); - } - }); - let close_session = Arc::clone(&session); - let close_barrier = Arc::clone(&barrier); - let close = std::thread::spawn(move || { - close_barrier.wait(); - std::thread::yield_now(); - close_session.close(SessionCloseReason::ApiClose); - }); - barrier.wait(); - health.join().unwrap(); - close.join().unwrap(); - - assert!(matches!( - session.carrier_health_publication_state(), - CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected - )); - assert!(!session.publish_carrier_health()); - assert_eq!( - session.close(SessionCloseReason::ApiClose), - SessionCloseOutcome::AlreadyClosing - ); - } - } - - #[test] - fn commit_and_supersede_have_one_session_lock_winner() { - let committed = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - { - let mut state = committed.state.lock(); - assert!( - committed - .record_uplink_progress_locked( - &mut state, - AppliedProgress { - accepted_open: true, - accepted_data: true, - }, - ) - .0 - ); - } - assert!(!committed.begin_carrier_supersede()); - - let replacing = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); - assert!(replacing.begin_carrier_supersede()); - assert_eq!( - replacing.ensure_carrier_active_locked(&replacing.state.lock()), - Err(crate::web::manager::ManagerError::Closed) - ); - replacing.cancel_carrier_supersede(); - assert!( - replacing - .ensure_carrier_active_locked(&replacing.state.lock()) - .is_ok() - ); - } -} +mod tests; diff --git a/src/web/session/negotiation/tests.rs b/src/web/session/negotiation/tests.rs new file mode 100644 index 0000000..2dae145 --- /dev/null +++ b/src/web/session/negotiation/tests.rs @@ -0,0 +1,252 @@ +use std::net::SocketAddr; +use std::sync::{Arc, Barrier}; + +use super::*; +use crate::config::{ + WebCarrier, WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig, +}; +use crate::web::manager::{CarrierClientClass, WebProcessRuntime}; +use crate::web::session::{SessionCloseOutcome, SessionCloseReason}; + +fn session(carrier: WebCarrier, deadline: Instant) -> Arc { + let profile = Arc::new(WebRuntimeProfile { + host: "proxy.example.com".to_string(), + public_addr: SocketAddr::from(([203, 0, 113, 10], 443)), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier, + carrier_negotiation_enabled: true, + carrier_learning: false, + carriers: Arc::from([carrier]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + capability: [0; 32], + credential_id: [0; 16], + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1, + max_streams: 1, + max_streams_per_session: 1, + }); + WebSession::new( + std::sync::Weak::::new(), + [1; 32], + "192.0.2.10".parse().unwrap(), + 1, + profile, + [2; 32], + carrier, + 1, + [3; 32], + Some(deadline), + CarrierClientClass::Bridge, + None, + true, + false, + WebLimitsConfig::default(), + WebTimeoutsConfig::default(), + None, + ) +} + +fn arm_http_health(session: &WebSession, now: Instant) { + let mut state = session.state.lock(); + state.negotiation_phase = SessionNegotiationPhase::Committed; + state.carrier_commit_published = true; + state.carrier_health_due_at = Some(now - Duration::from_secs(1)); + state.carrier_health_uplink = true; + state.carrier_health_downlink = true; + state.carrier_health_activity_at = Some(now); +} + +#[test] +fn final_deadline_refuses_uncommitted_progress() { + let session = session(WebCarrier::Https, Instant::now() - Duration::from_secs(1)); + let state = session.state.lock(); + assert_eq!( + session.ensure_carrier_active_locked(&state), + Err(crate::web::manager::ManagerError::Closed) + ); + assert!(matches!( + state.negotiation_phase, + SessionNegotiationPhase::Uncommitted + )); +} + +#[test] +fn http_health_requires_authenticated_activity_after_the_window() { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + let mut state = session.state.lock(); + state.negotiation_phase = SessionNegotiationPhase::Committed; + state.carrier_commit_published = true; + state.carrier_health_due_at = Some(now - Duration::from_secs(1)); + state.carrier_health_uplink = true; + state.carrier_health_downlink = true; + state.carrier_health_activity_at = Some(now - Duration::from_secs(2)); + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_none() + ); + state.carrier_health_activity_at = Some(now); + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_some() + ); +} + +#[test] +fn websocket_health_requires_the_exact_live_probe_owner() { + let session = session( + WebCarrier::Websocket, + Instant::now() + Duration::from_secs(60), + ); + let now = Instant::now(); + let mut state = session.state.lock(); + state.negotiation_phase = SessionNegotiationPhase::Committed; + state.carrier_commit_published = true; + state.carrier_health_due_at = Some(now - Duration::from_secs(1)); + state.websocket_carrier_active = true; + state.websocket_commit_ack_owner = Some(7); + state.websocket_commit_ack_written = true; + state.carrier_health_uplink = true; + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_none() + ); + state.websocket_probe_claimed = true; + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_some() + ); +} + +#[test] +fn health_waits_for_manager_commit_publication() { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + let mut state = session.state.lock(); + state.negotiation_phase = SessionNegotiationPhase::Committed; + state.carrier_health_due_at = Some(now - Duration::from_secs(1)); + state.carrier_health_uplink = true; + state.carrier_health_downlink = true; + state.carrier_health_activity_at = Some(now); + + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_none() + ); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Awaiting + ); +} + +#[test] +fn health_publication_claim_is_single_shot() { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + arm_http_health(&session, now); + let mut state = session.state.lock(); + + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_some() + ); + assert!( + session + .carrier_health_ready_locked(&mut state, now) + .is_none() + ); + drop(state); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Publishing + ); + assert!(session.publish_carrier_health()); + assert!(!session.publish_carrier_health()); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Published + ); +} + +#[test] +fn concurrent_health_and_close_always_reach_one_terminal_state() { + for _ in 0..512 { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + arm_http_health(&session, now); + let barrier = Arc::new(Barrier::new(3)); + let health_session = Arc::clone(&session); + let health_barrier = Arc::clone(&barrier); + let health = std::thread::spawn(move || { + health_barrier.wait(); + std::thread::yield_now(); + let claim = { + let mut state = health_session.state.lock(); + health_session.carrier_health_ready_locked(&mut state, now) + }; + if claim.is_some() { + health_session.publish_carrier_health(); + } + }); + let close_session = Arc::clone(&session); + let close_barrier = Arc::clone(&barrier); + let close = std::thread::spawn(move || { + close_barrier.wait(); + std::thread::yield_now(); + close_session.close(SessionCloseReason::ApiClose); + }); + barrier.wait(); + health.join().unwrap(); + close.join().unwrap(); + + assert!(matches!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected + )); + assert!(!session.publish_carrier_health()); + assert_eq!( + session.close(SessionCloseReason::ApiClose), + SessionCloseOutcome::AlreadyClosing + ); + } +} + +#[test] +fn commit_and_supersede_have_one_session_lock_winner() { + let committed = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + { + let mut state = committed.state.lock(); + assert!( + committed + .record_uplink_progress_locked( + &mut state, + AppliedProgress { + accepted_open: true, + accepted_data: true, + }, + ) + .0 + ); + } + assert!(!committed.begin_carrier_supersede()); + + let replacing = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + assert!(replacing.begin_carrier_supersede()); + assert_eq!( + replacing.ensure_carrier_active_locked(&replacing.state.lock()), + Err(crate::web::manager::ManagerError::Closed) + ); + replacing.cancel_carrier_supersede(); + assert!( + replacing + .ensure_carrier_active_locked(&replacing.state.lock()) + .is_ok() + ); +}