From deafa4a6cf9a6ae9a01f8427b10a6447f00a7062 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:00:13 +0300 Subject: [PATCH] WEB: skip registry sweeps before the earliest possible expiry --- src/web/bridge.rs | 119 ++++--- src/web/bridge/render_benchmarks.rs | 85 +++++ src/web/bridge/render_reference.rs | 155 +++++++++ src/web/bridge/render_tests.rs | 225 +++++++++++++ src/web/manager.rs | 2 + src/web/manager/credentials.rs | 4 +- src/web/manager/lifecycle.rs | 1 + src/web/manager/state.rs | 58 ++-- src/web/manager/state/expiry.rs | 67 ++++ src/web/manager/state/expiry/benchmarks.rs | 132 ++++++++ src/web/manager/state/expiry/fixtures.rs | 94 ++++++ .../manager/state/expiry/production_tests.rs | 184 +++++++++++ src/web/manager/state/expiry/reference.rs | 34 ++ src/web/manager/state/expiry/tests.rs | 302 ++++++++++++++++++ 14 files changed, 1382 insertions(+), 80 deletions(-) create mode 100644 src/web/bridge/render_benchmarks.rs create mode 100644 src/web/bridge/render_reference.rs create mode 100644 src/web/bridge/render_tests.rs create mode 100644 src/web/manager/state/expiry.rs create mode 100644 src/web/manager/state/expiry/benchmarks.rs create mode 100644 src/web/manager/state/expiry/fixtures.rs create mode 100644 src/web/manager/state/expiry/production_tests.rs create mode 100644 src/web/manager/state/expiry/reference.rs create mode 100644 src/web/manager/state/expiry/tests.rs diff --git a/src/web/bridge.rs b/src/web/bridge.rs index c0a485a..b057072 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -1,3 +1,5 @@ +use std::sync::OnceLock; + use base64::Engine as _; use crate::config::WebCarrierMethod; @@ -41,6 +43,68 @@ pub(crate) fn render( let mut nonce = [0u8; 18]; rng.fill(&mut nonce); let nonce = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(nonce); + let base_prefix = base.strip_suffix('/').unwrap_or(base); + let body = template(bridge_diagnostics_enabled) + .replace("__NONCE__", &nonce) + .replace("__HOST__", host) + .replace("__BASE_PREFIX__", base_prefix) + .replace("__BOOTSTRAP__", bootstrap) + .replace("__CARRIER_METHOD__", carrier_method.as_str()) + .replace("__BATCH_LIMIT__", &batch_limit.to_string()) + .replace("__QUEUE_LIMIT__", &queue_limit.to_string()) + .replace("__QUEUE_ITEMS__", &queue_items.to_string()) + .replace("__MAX_STREAMS__", &max_streams.to_string()) + .replace( + "__NEGOTIATION_ENABLED__", + if negotiation_enabled { "true" } else { "false" }, + ) + .replace("__CANDIDATE_COUNT__", &candidate_count.to_string()) + .replace("__LONG_POLL_SECS__", &long_poll_secs.to_string()) + .replace("__BRIDGE_REQUEST_SECS__", &bridge_request_secs.to_string()) + .replace("__BRIDGE_RETRY_SECS__", &bridge_retry_secs.to_string()) + .replace( + "__BRIDGE_RECOVERY_SECS__", + &bridge_recovery_secs.to_string(), + ) + .replace("__WEBSOCKET_OPEN_SECS__", &websocket_open_secs.to_string()) + .replace( + "__RECONNECT_GRACE_SECS__", + &reconnect_grace_secs.to_string(), + ) + .replace( + "__CARRIER_PROBE_COALESCE_MS__", + &carrier_probe_coalesce_ms.to_string(), + ) + .replace( + "__CARRIER_DEADLINES__", + &carrier_deadlines + .iter() + .map(u64::to_string) + .collect::>() + .join(","), + ); + BridgePage { + body, + content_security_policy: format!( + "default-src 'none'; base-uri 'none'; child-src 'none'; connect-src 'self' wss://{host}; font-src 'none'; form-action 'none'; frame-ancestors http://127.0.0.1:*; frame-src 'none'; img-src 'none'; manifest-src 'none'; media-src 'none'; object-src 'none'; script-src 'nonce-{nonce}'; style-src 'none'; worker-src 'none'; sandbox allow-same-origin allow-scripts" + ), + } +} + +// Only immutable embedded source is cached; credentials and CSP nonces stay response-owned. +static TEMPLATES: [OnceLock; 2] = [OnceLock::new(), OnceLock::new()]; + +/// Prepares both static variants before a WEB runtime accepts requests. +pub(super) fn prepare_templates() { + template(false); + template(true); +} + +fn template(diagnostics: bool) -> &'static str { + TEMPLATES[usize::from(diagnostics)].get_or_init(|| assemble_template(diagnostics)) +} + +fn assemble_template(bridge_diagnostics_enabled: bool) -> String { let diagnostic_script = if bridge_diagnostics_enabled { format!("\n") } else { @@ -58,8 +122,7 @@ pub(crate) fn render( } else { String::new() }; - let base_prefix = base.strip_suffix('/').unwrap_or(base); - let body = DOCUMENT + DOCUMENT .replace("__DIAGNOSTIC_RUNTIME__\n", &diagnostic_script) .replace("__RESPONSE_RUNTIME__", RESPONSE_RUNTIME) .replace("__REQUEST_RUNTIME__", REQUEST_RUNTIME) @@ -124,50 +187,6 @@ pub(crate) fn render( "" }, ) - .replace("__NONCE__", &nonce) - .replace("__HOST__", host) - .replace("__BASE_PREFIX__", base_prefix) - .replace("__BOOTSTRAP__", bootstrap) - .replace("__CARRIER_METHOD__", carrier_method.as_str()) - .replace("__BATCH_LIMIT__", &batch_limit.to_string()) - .replace("__QUEUE_LIMIT__", &queue_limit.to_string()) - .replace("__QUEUE_ITEMS__", &queue_items.to_string()) - .replace("__MAX_STREAMS__", &max_streams.to_string()) - .replace( - "__NEGOTIATION_ENABLED__", - if negotiation_enabled { "true" } else { "false" }, - ) - .replace("__CANDIDATE_COUNT__", &candidate_count.to_string()) - .replace("__LONG_POLL_SECS__", &long_poll_secs.to_string()) - .replace("__BRIDGE_REQUEST_SECS__", &bridge_request_secs.to_string()) - .replace("__BRIDGE_RETRY_SECS__", &bridge_retry_secs.to_string()) - .replace( - "__BRIDGE_RECOVERY_SECS__", - &bridge_recovery_secs.to_string(), - ) - .replace("__WEBSOCKET_OPEN_SECS__", &websocket_open_secs.to_string()) - .replace( - "__RECONNECT_GRACE_SECS__", - &reconnect_grace_secs.to_string(), - ) - .replace( - "__CARRIER_PROBE_COALESCE_MS__", - &carrier_probe_coalesce_ms.to_string(), - ) - .replace( - "__CARRIER_DEADLINES__", - &carrier_deadlines - .iter() - .map(u64::to_string) - .collect::>() - .join(","), - ); - BridgePage { - body, - content_security_policy: format!( - "default-src 'none'; base-uri 'none'; child-src 'none'; connect-src 'self' wss://{host}; font-src 'none'; form-action 'none'; frame-ancestors http://127.0.0.1:*; frame-src 'none'; img-src 'none'; manifest-src 'none'; media-src 'none'; object-src 'none'; script-src 'nonce-{nonce}'; style-src 'none'; worker-src 'none'; sandbox allow-same-origin allow-scripts" - ), - } } const DOCUMENT: &str = include_str!("bridge/document.html"); @@ -183,3 +202,11 @@ const RUNTIME: &str = include_str!("bridge/runtime.js"); // Rendered wire-contract tests remain separate from the embedded document. #[cfg(test)] mod tests; + +// Differential tests retain the original renderer as an independent wire-contract oracle. +#[cfg(test)] +#[path = "bridge/render_reference.rs"] +mod render_reference; +#[cfg(test)] +#[path = "bridge/render_tests.rs"] +mod render_tests; diff --git a/src/web/bridge/render_benchmarks.rs b/src/web/bridge/render_benchmarks.rs new file mode 100644 index 0000000..1a6aa76 --- /dev/null +++ b/src/web/bridge/render_benchmarks.rs @@ -0,0 +1,85 @@ +use super::*; +use std::hint::black_box; +use std::time::Instant; + +fn report(label: &str, mut samples: Vec) { + samples.sort_unstable(); + eprintln!( + "{label}: n={} p50={}ns p95={}ns p99={}ns", + samples.len(), + samples[samples.len() / 2], + samples[samples.len() * 95 / 100], + samples[samples.len() * 99 / 100] + ); +} + +#[test] +#[ignore = "manual optimized-profile latency comparison; no timing assertions"] +fn compare_bridge_render_latency() { + for diagnostics in [false, true] { + let samples = (0..100) + .map(|_| { + let start = Instant::now(); + drop(black_box(assemble_template(diagnostics))); + start.elapsed().as_nanos() + }) + .collect(); + report( + &format!("bridge template initialization diagnostics={diagnostics}"), + samples, + ); + } + prepare_templates(); + for repeat in 0..3 { + for diagnostics in [false, true] { + for workers in [1, 32] { + for cached in [false, true] { + let barrier = Barrier::new(workers); + let samples = std::thread::scope(|scope| { + let handles = (0..workers) + .map(|worker| { + let barrier = &barrier; + scope.spawn(move || { + let rng = SecureRandom::new(); + let mut case = Case::new(worker); + case.diagnostics = diagnostics; + for _ in 0..20 { + drop(case.render(&rng)); + } + let mut samples = Vec::with_capacity(300); + barrier.wait(); + for _ in 0..300 { + let start = Instant::now(); + let page = if cached { + case.render(&rng) + } else { + let mut nonce = [0; 18]; + rng.fill(&mut nonce); + let nonce = + base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(nonce); + case.reference(&nonce) + }; + drop(black_box(page)); + samples.push(start.elapsed().as_nanos()); + } + samples + }) + }) + .collect::>(); + handles + .into_iter() + .flat_map(|h| h.join().unwrap()) + .collect() + }); + report( + &format!( + "bridge repeat={repeat} diagnostics={diagnostics} workers={workers} cached={cached}" + ), + samples, + ); + } + } + } + } +} diff --git a/src/web/bridge/render_reference.rs b/src/web/bridge/render_reference.rs new file mode 100644 index 0000000..2629100 --- /dev/null +++ b/src/web/bridge/render_reference.rs @@ -0,0 +1,155 @@ +// Frozen pre-cache renderer; callers supply the response nonce for exact comparisons. +use super::*; + +/// Replays the original substitution chain using the supplied response nonce. +#[allow(clippy::too_many_arguments)] +pub(super) fn render( + host: &str, + base: &str, + bootstrap: &str, + batch_limit: usize, + queue_limit: usize, + queue_items: usize, + max_streams: usize, + negotiation_enabled: bool, + candidate_count: usize, + carrier_deadlines: [u64; 4], + long_poll_secs: u64, + bridge_request_secs: u64, + bridge_retry_secs: u64, + bridge_recovery_secs: u64, + websocket_open_secs: u64, + reconnect_grace_secs: u64, + carrier_probe_coalesce_ms: u64, + bridge_diagnostics_enabled: bool, + carrier_method: WebCarrierMethod, + nonce: &str, +) -> BridgePage { + let diagnostic_script = if bridge_diagnostics_enabled { + format!("\n") + } else { + String::new() + }; + let diagnostic_hook = |method: &str| { + bridge_diagnostics_enabled + .then(|| { + format!("if(clientDiagnostics)try{{clientDiagnostics.{method}()}}catch(error){{}}") + }) + .unwrap_or_default() + }; + let diagnostic_runtime_started = if bridge_diagnostics_enabled { + format!("{}\n", diagnostic_hook("runtimeStarted")) + } else { + String::new() + }; + let base_prefix = base.strip_suffix('/').unwrap_or(base); + let body = DOCUMENT + .replace("__DIAGNOSTIC_RUNTIME__\n", &diagnostic_script) + .replace("__RESPONSE_RUNTIME__", RESPONSE_RUNTIME) + .replace("__REQUEST_RUNTIME__", REQUEST_RUNTIME) + .replace("__BUFFER_RUNTIME__", BUFFER_RUNTIME) + .replace("__RECOVERY_RUNTIME__", RECOVERY_RUNTIME) + .replace("__DOWNLINK_RUNTIME__", DOWNLINK_RUNTIME) + .replace("__CONVEYOR_RUNTIME__", CONVEYOR_RUNTIME) + .replace("__RUNTIME__", RUNTIME) + .replace( + "__DIAGNOSTIC_BINDING__;\n", + if bridge_diagnostics_enabled { + "const clientDiagnostics=globalThis.TelemtBridgeDiagnostics;\n" + } else { + "" + }, + ) + .replace( + "__DIAGNOSTIC_RUNTIME_STARTED__;\n", + &diagnostic_runtime_started, + ) + .replace( + "__DIAGNOSTIC_BOUNDARY_ACTIVATED__;", + &diagnostic_hook("boundaryActivated"), + ) + .replace( + "__STATUS_FUNCTION__", + if bridge_diagnostics_enabled { + "state=>{if(port&&!closed){port.postMessage({t:'status',state});if(clientDiagnostics)try{clientDiagnostics.statusPosted()}catch(error){}}}" + } else { + "state=>{if(port&&!closed)port.postMessage({t:'status',state})}" + }, + ) + .replace( + "__DIAGNOSTIC_HELLO_RECEIVED__;", + &diagnostic_hook("helloReceived"), + ) + .replace( + "__HELLO_TIMEOUT_CALLBACK__", + if bridge_diagnostics_enabled { + "()=>{if(clientDiagnostics)try{clientDiagnostics.helloTimeout()}catch(error){}fail('timeout')}" + } else { + "()=>fail('timeout')" + }, + ) + .replace( + "__DIAGNOSTIC_CLIENT_CLOSE__;", + &diagnostic_hook("clientCloseBeforeHello"), + ) + .replace( + "__PAGEHIDE_CALLBACK__", + if bridge_diagnostics_enabled { + "()=>{if(clientDiagnostics)try{clientDiagnostics.documentUnloadedBeforeHello()}catch(error){}fail('navigation')}" + } else { + "()=>fail('navigation')" + }, + ) + .replace( + "__DIAGNOSTIC_BOOTSTRAP_REPLACED__;", + if bridge_diagnostics_enabled { + "if(clientDiagnostics)try{clientDiagnostics.setBootstrap(bootstrap)}catch(error){}" + } else { + "" + }, + ) + .replace("__NONCE__", &nonce) + .replace("__HOST__", host) + .replace("__BASE_PREFIX__", base_prefix) + .replace("__BOOTSTRAP__", bootstrap) + .replace("__CARRIER_METHOD__", carrier_method.as_str()) + .replace("__BATCH_LIMIT__", &batch_limit.to_string()) + .replace("__QUEUE_LIMIT__", &queue_limit.to_string()) + .replace("__QUEUE_ITEMS__", &queue_items.to_string()) + .replace("__MAX_STREAMS__", &max_streams.to_string()) + .replace( + "__NEGOTIATION_ENABLED__", + if negotiation_enabled { "true" } else { "false" }, + ) + .replace("__CANDIDATE_COUNT__", &candidate_count.to_string()) + .replace("__LONG_POLL_SECS__", &long_poll_secs.to_string()) + .replace("__BRIDGE_REQUEST_SECS__", &bridge_request_secs.to_string()) + .replace("__BRIDGE_RETRY_SECS__", &bridge_retry_secs.to_string()) + .replace( + "__BRIDGE_RECOVERY_SECS__", + &bridge_recovery_secs.to_string(), + ) + .replace("__WEBSOCKET_OPEN_SECS__", &websocket_open_secs.to_string()) + .replace( + "__RECONNECT_GRACE_SECS__", + &reconnect_grace_secs.to_string(), + ) + .replace( + "__CARRIER_PROBE_COALESCE_MS__", + &carrier_probe_coalesce_ms.to_string(), + ) + .replace( + "__CARRIER_DEADLINES__", + &carrier_deadlines + .iter() + .map(u64::to_string) + .collect::>() + .join(","), + ); + BridgePage { + body, + content_security_policy: format!( + "default-src 'none'; base-uri 'none'; child-src 'none'; connect-src 'self' wss://{host}; font-src 'none'; form-action 'none'; frame-ancestors http://127.0.0.1:*; frame-src 'none'; img-src 'none'; manifest-src 'none'; media-src 'none'; object-src 'none'; script-src 'nonce-{nonce}'; style-src 'none'; worker-src 'none'; sandbox allow-same-origin allow-scripts" + ), + } +} diff --git a/src/web/bridge/render_tests.rs b/src/web/bridge/render_tests.rs new file mode 100644 index 0000000..f04d976 --- /dev/null +++ b/src/web/bridge/render_tests.rs @@ -0,0 +1,225 @@ +use super::*; +use std::collections::HashSet; +use std::sync::{Arc, Barrier}; + +#[path = "render_benchmarks.rs"] +mod benchmarks; + +#[test] +fn disabled_runtime_prewarms_templates_before_hot_enable() { + const CHILD: &str = "TELEMT_TEST_BRIDGE_PREWARM_CHILD"; + if std::env::var_os(CHILD).is_none() { + // A fresh process prevents unrelated render tests from satisfying the warmup assertion. + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "web::bridge::render_tests::disabled_runtime_prewarms_templates_before_hot_enable", + ]) + .env(CHILD, "1") + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + output.status.success() && stdout.contains("1 passed; 0 failed"), + "{}\n{}", + stdout, + String::from_utf8_lossy(&output.stderr) + ); + return; + } + assert!(TEMPLATES.iter().all(|template| template.get().is_none())); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let mut config = crate::config::ProxyConfig::default(); + config.web.enabled = false; + let generation = crate::maestro::generation::test_runtime_generation(1, config); + let runtime = crate::web::manager::WebProcessRuntime::start(Arc::new( + arc_swap::ArcSwap::from(Arc::clone(&generation)), + )); + assert!(TEMPLATES.iter().all(|template| template.get().is_some())); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + }); +} + +struct Case { + host: String, + base: String, + bootstrap: String, + diagnostics: bool, + method: WebCarrierMethod, + negotiation: bool, + candidates: usize, + value: usize, +} + +impl Case { + fn new(index: usize) -> Self { + Self { + host: format!("proxy-{index}.example.com"), + base: "/telegram/web/".to_string(), + bootstrap: format!("{index:043}"), + diagnostics: index % 2 != 0, + method: if index % 3 == 0 { + WebCarrierMethod::Put + } else { + WebCarrierMethod::Post + }, + negotiation: index % 2 == 0, + candidates: index % 4 + 1, + value: index + 1, + } + } + + fn render(&self, rng: &SecureRandom) -> BridgePage { + render( + &self.host, + &self.base, + &self.bootstrap, + self.value * 1024, + self.value * 4096, + self.value * 4, + self.value, + self.negotiation, + self.candidates, + [3, 5, 8, 12], + self.value as u64, + self.value as u64 + 1, + self.value as u64 + 2, + self.value as u64 + 3, + self.value as u64 + 4, + self.value as u64 + 5, + self.value as u64 - 1, + self.diagnostics, + self.method, + rng, + ) + } + + fn reference(&self, nonce: &str) -> BridgePage { + render_reference::render( + &self.host, + &self.base, + &self.bootstrap, + self.value * 1024, + self.value * 4096, + self.value * 4, + self.value, + self.negotiation, + self.candidates, + [3, 5, 8, 12], + self.value as u64, + self.value as u64 + 1, + self.value as u64 + 2, + self.value as u64 + 3, + self.value as u64 + 4, + self.value as u64 + 5, + self.value as u64 - 1, + self.diagnostics, + self.method, + nonce, + ) + } + + fn assert_page(&self, page: &BridgePage) -> String { + let nonce = page + .content_security_policy + .split_once("script-src 'nonce-") + .unwrap() + .1 + .split_once('\'') + .unwrap() + .0; + assert_eq!(nonce.len(), 24); + let expected = self.reference(nonce); + assert_eq!(page.body, expected.body); + assert_eq!( + page.content_security_policy, + expected.content_security_policy + ); + assert_eq!( + page.body.matches(&format!("nonce=\"{nonce}\"")).count(), + if self.diagnostics { 8 } else { 7 } + ); + nonce.to_string() + } +} + +#[test] +fn cached_templates_preserve_exact_dynamic_substitution_order() { + prepare_templates(); + let rng = SecureRandom::new(); + let long_base = format!("/{}/", "a".repeat(126)); + let bases = [ + "/", + "/telegram/web/", + "/a__BOOTSTRAP__/", + "/a__CARRIER_DEADLINES__/", + long_base.as_str(), + ]; + for diagnostics in [false, true] { + for method in [WebCarrierMethod::Post, WebCarrierMethod::Put] { + for negotiation in [false, true] { + for candidates in 1..=4 { + for base in bases { + let mut case = Case::new(candidates); + case.diagnostics = diagnostics; + case.method = method; + case.negotiation = negotiation; + case.candidates = candidates; + case.base = base.to_string(); + case.assert_page(&case.render(&rng)); + } + } + } + } + } +} + +#[test] +fn concurrent_responses_keep_nonces_and_configuration_response_owned() { + prepare_templates(); + let pointers = [ + template(false).as_ptr() as usize, + template(true).as_ptr() as usize, + ]; + let barrier = Arc::new(Barrier::new(32)); + let nonces = std::thread::scope(|scope| { + let handles = (0..32) + .map(|worker| { + let barrier = Arc::clone(&barrier); + scope.spawn(move || { + let rng = SecureRandom::new(); + barrier.wait(); + (0..8) + .map(|iteration| { + let case = Case::new(worker * 8 + iteration); + case.assert_page(&case.render(&rng)) + }) + .collect::>() + }) + }) + .collect::>(); + handles + .into_iter() + .flat_map(|handle| handle.join().unwrap()) + .collect::>() + }); + assert_eq!(nonces.iter().collect::>().len(), nonces.len()); + assert_eq!( + pointers, + [ + template(false).as_ptr() as usize, + template(true).as_ptr() as usize + ] + ); + for diagnostics in [false, true] { + assert!(template(diagnostics).contains("__NONCE__")); + assert!(template(diagnostics).contains("__BOOTSTRAP__")); + assert!(template(diagnostics).contains("__BASE_PREFIX__")); + } +} diff --git a/src/web/manager.rs b/src/web/manager.rs index bd9bba5..7f84490 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -265,6 +265,8 @@ impl WebProcessRuntime { ) -> Arc { let initial_generation = active_runtime.load_full(); let config = initial_generation.config(); + // Hot reload can enable WEB without replacing the runtime generation. + crate::web::bridge::prepare_templates(); let token_authenticator = TokenAuthenticator::new(&initial_generation.rng); trace.apply_policy(initial_generation.id, &config.web.debug); let limits = config.web.limits.clone(); diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index dca907a..db6bbf0 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -166,11 +166,13 @@ impl WebProcessRuntime { let bridge_diagnostics_enabled = config.web.debug.bridge_diagnostics_enabled(); let (user_agent, user_agent_id) = bounded_user_agent(user_agent); let issued_profile = Arc::clone(&profile); + let expires_at = now + Duration::from_secs(config.web.timeouts.bootstrap_lifetime_secs); + state.bootstrap_expiry = Some(state.bootstrap_expiry.unwrap_or(expires_at).min(expires_at)); state.bootstraps.insert( hash, Bootstrap { user_registration, - expires_at: now + Duration::from_secs(config.web.timeouts.bootstrap_lifetime_secs), + expires_at, issued_at: now, issuance_ip: client_ip, profile, diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index dabd3eb..510d6c2 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -143,6 +143,7 @@ impl WebProcessRuntime { let mut state = self.state.lock(); state.closed = true; state.bootstraps.clear(); + state.bootstrap_expiry = None; state.bootstraps_per_ip.clear(); state.sessions.values().cloned().collect::>() }; diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index 439c0e3..509de4b 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -14,16 +14,25 @@ use crate::proxy::user_admission::UserSessionRegistration; use crate::web::session::WebSession; use crate::web::telemetry::WebCarrierSelectionDisposition; +// Expiry sweeps preserve admission accounting while skipping maps with no due entries. +mod expiry; +pub(super) use expiry::remove_expired_locked; + const WEB_PROFILE_OWNER_CONTEXT: &[u8] = b"telemt-web-profile-owner-v1\0"; +/// Manager-owned carrier progress used to fence provisional replacement. #[derive(Clone, Copy, PartialEq, Eq)] pub(super) enum CarrierChainPhase { + /// No carrier has committed verified application traffic yet. Provisional, + /// A committed carrier still awaits the health observation window. CommittedPendingHealth, + /// The committed carrier has satisfied the health observation window. Healthy, } impl CarrierChainPhase { + /// Returns the stable carrier-state token exposed in negotiation responses. pub(super) const fn as_str(self) -> &'static str { match self { Self::Provisional => "provisional", @@ -167,6 +176,8 @@ pub(super) struct StreamAdmissionState { pub(super) struct ManagerState { /// Bootstrap credentials indexed by their SHA-256 token hash. pub(super) bootstraps: HashMap, + /// Conservative earliest expiry; deletion and frozen-TTL renewal may leave it early. + pub(super) bootstrap_expiry: Option, /// Unused bootstrap ownership counts by forwarded client address. pub(super) bootstraps_per_ip: HashMap, /// Live sessions indexed by bearer-token hash. @@ -179,6 +190,8 @@ pub(super) struct ManagerState { pub(super) closed_session_order: VecDeque, /// Recently closed token hashes retained for idempotent DELETE semantics. pub(super) closed_tokens: HashMap, + /// Conservative earliest expiry, lowered on every insertion or overwrite. + pub(super) closed_token_expiry: Option, /// Live session counts by forwarded client address. pub(super) sessions_per_ip: HashMap, /// Live session counts by stable profile key. @@ -196,15 +209,18 @@ pub(super) struct ManagerState { } impl ManagerState { + /// Initializes empty process-owned registries with a generation-fenced issuance policy. pub(super) fn new(issuance_generation: u64, issuance_enabled: bool) -> Self { Self { bootstraps: HashMap::new(), + bootstrap_expiry: None, bootstraps_per_ip: HashMap::new(), sessions: HashMap::new(), session_index: BTreeMap::new(), closed_sessions: HashMap::new(), closed_session_order: VecDeque::new(), closed_tokens: HashMap::new(), + closed_token_expiry: None, sessions_per_ip: HashMap::new(), sessions_per_profile: HashMap::new(), bootstrap_rate: RateState::default(), @@ -215,6 +231,7 @@ impl ManagerState { } } + /// Applies only policies at least as recent as the last published generation. pub(super) fn apply_issuance_policy(&mut self, generation: u64, enabled: bool) { if generation >= self.issuance_generation { self.issuance_generation = generation; @@ -326,38 +343,6 @@ pub(super) fn evict_oldest_unused_bootstrap(state: &mut ManagerState) -> Option< Some(bootstrap) } -/// Removes expired bootstrap and closed-token entries while the manager lock is held. -pub(super) fn remove_expired_locked(state: &mut ManagerState, now: Instant) -> usize { - let expired = state - .bootstraps - .iter() - .filter_map(|(hash, bootstrap)| { - (now > bootstrap.expires_at).then_some((*hash, bootstrap.recovery && !bootstrap.used)) - }) - .collect::>(); - let recovery_unused = expired - .iter() - .filter(|(_, recovery_unused)| *recovery_unused) - .count(); - for (hash, _) in expired { - remove_bootstrap_locked(state, hash); - } - state - .closed_tokens - .retain(|_, closed| now <= closed.expires_at); - while state - .closed_session_order - .front() - .and_then(|trace_session_id| state.closed_sessions.get(trace_session_id)) - .is_some_and(|closed| now > closed.expires_at) - { - if let Some(trace_session_id) = state.closed_session_order.pop_front() { - state.closed_sessions.remove(&trace_session_id); - } - } - recovery_unused -} - /// Removes one bootstrap and releases its per-address issuance quota when unused. pub(super) fn remove_bootstrap_locked(state: &mut ManagerState, hash: TokenHash) { let Some(bootstrap) = state.bootstraps.remove(&hash) else { @@ -377,10 +362,17 @@ pub(super) fn remember_closed_token_locked( lifetime: Duration, capacity: usize, ) { + let expires_at = Instant::now() + lifetime; + state.closed_token_expiry = Some( + state + .closed_token_expiry + .unwrap_or(expires_at) + .min(expires_at), + ); state.closed_tokens.insert( hash, ClosedToken { - expires_at: Instant::now() + lifetime, + expires_at, host: host.to_string(), carrier, }, diff --git a/src/web/manager/state/expiry.rs b/src/web/manager/state/expiry.rs new file mode 100644 index 0000000..e9592f8 --- /dev/null +++ b/src/web/manager/state/expiry.rs @@ -0,0 +1,67 @@ +use std::time::Instant; + +use super::{ManagerState, remove_bootstrap_locked}; + +/// Removes expired credentials without changing the caller's expiry boundary. +pub(in crate::web::manager) fn remove_expired_locked( + state: &mut ManagerState, + now: Instant, +) -> usize { + // None implies an empty map; Some may be stale, but never later than a live deadline. + // Deletion and bootstrap renewal only leave early hints, so one sweep repairs them. + let mut recovery_unused = 0; + if state.bootstrap_expiry.is_some_and(|expiry| now > expiry) { + let mut next_expiry: Option = None; + let expired = state + .bootstraps + .iter() + .filter_map(|(hash, bootstrap)| { + if now <= bootstrap.expires_at { + if next_expiry.is_none_or(|expiry| bootstrap.expires_at < expiry) { + next_expiry = Some(bootstrap.expires_at); + } + None + } else { + Some((*hash, bootstrap.recovery && !bootstrap.used)) + } + }) + .collect::>(); + recovery_unused = expired + .iter() + .filter(|(_, recovery_unused)| *recovery_unused) + .count(); + for (hash, _) in expired { + remove_bootstrap_locked(state, hash); + } + state.bootstrap_expiry = next_expiry; + } + if state.closed_token_expiry.is_some_and(|expiry| now > expiry) { + let mut next_expiry: Option = None; + state.closed_tokens.retain(|_, closed| { + if now > closed.expires_at { + false + } else { + if next_expiry.is_none_or(|expiry| closed.expires_at < expiry) { + next_expiry = Some(closed.expires_at); + } + true + } + }); + state.closed_token_expiry = next_expiry; + } + // Logical-session tombstones have their own FIFO lifetime, independent of both maps. + while state + .closed_session_order + .front() + .and_then(|trace_session_id| state.closed_sessions.get(trace_session_id)) + .is_some_and(|closed| now > closed.expires_at) + { + if let Some(trace_session_id) = state.closed_session_order.pop_front() { + state.closed_sessions.remove(&trace_session_id); + } + } + recovery_unused +} + +#[cfg(test)] +mod tests; diff --git a/src/web/manager/state/expiry/benchmarks.rs b/src/web/manager/state/expiry/benchmarks.rs new file mode 100644 index 0000000..f3bb922 --- /dev/null +++ b/src/web/manager/state/expiry/benchmarks.rs @@ -0,0 +1,132 @@ +use super::*; +use parking_lot::Mutex; +use std::hint::black_box; +use std::sync::Barrier; + +fn hash(index: usize) -> TokenHash { + let mut hash = [0; 32]; + hash[..8].copy_from_slice(&(index as u64).to_le_bytes()); + hash +} + +fn populated(fixture: &Fixture, entries: usize, now: Instant) -> ManagerState { + let mut state = ManagerState::default(); + for index in 0..entries { + insert_bootstrap( + &mut state, + hash(index), + fixture.bootstrap( + now, + now + Duration::from_secs(60 + index as u64 % 120), + (index % 4) as u8, + ), + ); + } + for index in 0..entries * 4 { + insert_closed_token( + &mut state, + hash(index), + now + Duration::from_secs(60 + index as u64 % 120), + ); + } + state +} + +fn report(label: &str, mut samples: Vec) { + samples.sort_unstable(); + eprintln!( + "{label}: n={} p50={}ns p95={}ns p99={}ns", + samples.len(), + samples[samples.len() / 2], + samples[samples.len() * 95 / 100], + samples[samples.len() * 99 / 100] + ); +} + +#[test] +#[ignore = "manual optimized-profile latency comparison; no timing assertions"] +fn compare_registry_sweep_latency() { + let fixture = Fixture::new(); + let now = Instant::now(); + for repeat in 0..3 { + for entries in [0, 512, 8192] { + for workers in [1, 32] { + for hinted in [false, true] { + let state = Mutex::new(populated(&fixture, entries, now)); + let barrier = Barrier::new(workers); + let sweep = if hinted { + remove_expired_locked + } else { + reference::exhaustive + }; + let samples = std::thread::scope(|scope| { + let handles = (0..workers) + .map(|_| { + let state = &state; + let barrier = &barrier; + scope.spawn(move || { + let mut samples = Vec::with_capacity(300); + barrier.wait(); + for _ in 0..300 { + let start = Instant::now(); + { + let mut state = state.lock(); + black_box(sweep(black_box(&mut state), black_box(now))); + } + samples.push(start.elapsed().as_nanos()); + } + samples + }) + }) + .collect::>(); + handles + .into_iter() + .flat_map(|h| h.join().unwrap()) + .collect() + }); + report( + &format!( + "registry no-expiry repeat={repeat} entries={entries} workers={workers} hinted={hinted}" + ), + samples, + ); + } + } + for hinted in [false, true] { + let sweep = if hinted { + remove_expired_locked + } else { + reference::exhaustive + }; + let mut state = populated(&fixture, entries, now); + let mut samples = Vec::with_capacity(300); + for index in 0..300 { + // Force both gates due while most entries retain mixed future deadlines. + insert_bootstrap( + &mut state, + hash(entries + index), + fixture.bootstrap( + now - Duration::from_secs(2), + now - Duration::from_secs(1), + 2, + ), + ); + insert_closed_token( + &mut state, + hash(entries * 4 + index), + now - Duration::from_secs(1), + ); + let start = Instant::now(); + assert_eq!(black_box(sweep(black_box(&mut state), black_box(now))), 1); + samples.push(start.elapsed().as_nanos()); + } + report( + &format!( + "registry expiry-churn repeat={repeat} entries={entries} workers=1 hinted={hinted}" + ), + samples, + ); + } + } + } +} diff --git a/src/web/manager/state/expiry/fixtures.rs b/src/web/manager/state/expiry/fixtures.rs new file mode 100644 index 0000000..dbcd9e5 --- /dev/null +++ b/src/web/manager/state/expiry/fixtures.rs @@ -0,0 +1,94 @@ +use crate::config::WebSecretMode; +use crate::proxy::user_admission::{UserAdmissionAuthority, credential_id_from_hex}; +use crate::web::telemetry::WebCarrierSelectionDisposition; + +struct Fixture { + authority: Arc, + profile: Arc, +} + +impl Fixture { + fn new() -> Self { + let secret = "0123456789abcdef0123456789abcdef"; + let authority = UserAdmissionAuthority::new(); + authority.apply_config( + &HashMap::from([("alice".to_string(), secret.to_string())]), + &HashMap::new(), + ); + Self { + authority, + profile: Arc::new(WebRuntimeProfile { + host: "proxy.example.com".to_string(), + public_addr: "203.0.113.10:443".parse().unwrap(), + user: "alice".to_string(), + secret_mode: WebSecretMode::Plain, + carrier: WebCarrier::Https, + carrier_negotiation_enabled: false, + carrier_learning: true, + carriers: Arc::from([WebCarrier::Https]), + carrier_negotiation_deadlines_secs: [3, 5, 8, 12], + capability: [7; 32], + credential_id: credential_id_from_hex(secret).unwrap(), + key_fingerprint: "0000000000000000".to_string(), + max_sessions: 1024, + max_streams: 1024, + max_streams_per_session: 1, + }), + } + } + + fn bootstrap(&self, issued_at: Instant, expires_at: Instant, flags: u8) -> Bootstrap { + Bootstrap { + user_registration: self.authority.register_legacy("alice").unwrap(), + expires_at, + issued_at, + issuance_ip: "192.0.2.10".parse().unwrap(), + profile: Arc::clone(&self.profile), + timeouts: WebTimeoutsConfig::default(), + trace_session_id: 1, + bridge_diagnostics_enabled: false, + bridge_diagnostic_events: 0, + user_agent: None, + user_agent_id: None, + body_digest: [0; 32], + session_token: Zeroizing::new(String::new()), + session: None, + carrier_request: None, + carrier_candidates: Arc::from([]), + carrier_scores: [0; 4], + carrier_attempt: 0, + carrier_transitioning: false, + carrier_phase: CarrierChainPhase::Provisional, + carrier_started_at: None, + carrier_deadline_at: None, + carrier_failures: [None; 3], + carrier_learning_epoch: 0, + carrier_learning_disposition: WebCarrierSelectionDisposition::ProfileDisabled, + close_requested: false, + session_client_ip: None, + session_ip_learning_eligible: false, + used: flags & 1 != 0, + recovery: flags & 2 != 0, + predecessor_session_id: None, + } + } +} + +fn insert_bootstrap(state: &mut ManagerState, hash: TokenHash, bootstrap: Bootstrap) { + remove_bootstrap_locked(state, hash); + let expires_at = bootstrap.expires_at; + state.bootstrap_expiry = Some(state.bootstrap_expiry.unwrap_or(expires_at).min(expires_at)); + if !bootstrap.used { + *state.bootstraps_per_ip.entry(bootstrap.issuance_ip).or_insert(0) += 1; + } + state.bootstraps.insert(hash, bootstrap); +} + +fn insert_closed_token(state: &mut ManagerState, hash: TokenHash, expires_at: Instant) { + state.closed_token_expiry = Some(state.closed_token_expiry.unwrap_or(expires_at).min(expires_at)); + state.closed_tokens.insert(hash, ClosedToken { + expires_at, + host: "proxy.example.com".to_string(), + carrier: WebCarrier::Https, + }); +} diff --git a/src/web/manager/state/expiry/production_tests.rs b/src/web/manager/state/expiry/production_tests.rs new file mode 100644 index 0000000..51bdab1 --- /dev/null +++ b/src/web/manager/state/expiry/production_tests.rs @@ -0,0 +1,184 @@ +use super::*; +use crate::config::ProxyConfig; +use crate::maestro::generation::test_runtime_generation; +use crate::web::frame::{self, FrameType}; +use crate::web::manager::{CarrierRequest, ManagerError, WebProcessRuntime}; +use arc_swap::ArcSwap; + +fn generation(id: u64, lifetime: u64) -> Arc { + let fixture = Fixture::new(); + let mut config = ProxyConfig::default(); + config.web.enabled = true; + config.web.timeouts.bootstrap_lifetime_secs = lifetime; + config.web.timeouts.shutdown_secs = 1; + config.web.limits.max_bootstraps_global = 1024; + config.web.limits.max_bootstraps_per_ip = 512; + config.web.limits.max_sessions_global = 1024; + config.web.limits.max_sessions_per_ip = 512; + config.web.limits.new_bootstraps_burst = 4096; + config.web.limits.new_sessions_burst = 4096; + config.web.runtime = Some(Arc::new(WebRuntimeConfig { + vhosts: BTreeMap::new(), + profiles: vec![Arc::clone(&fixture.profile)], + capabilities: vec![fixture.profile.capability].into_boxed_slice(), + })); + let generation = test_runtime_generation(id, config); + generation.proxy_shared.apply_user_config( + &HashMap::from([( + "alice".to_string(), + "0123456789abcdef0123456789abcdef".to_string(), + )]), + &HashMap::new(), + ); + generation +} + +fn token_hash(token: &str) -> TokenHash { + Sha256::digest( + base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(token) + .unwrap(), + ) + .into() +} + +fn assert_live_invariants(runtime: &WebProcessRuntime) { + let state = runtime.state.lock(); + let mut per_ip = HashMap::new(); + for bootstrap in state.bootstraps.values() { + assert!(state.bootstrap_expiry.unwrap() <= bootstrap.expires_at); + if !bootstrap.used { + *per_ip.entry(bootstrap.issuance_ip).or_insert(0) += 1; + } + } + assert_eq!(state.bootstraps_per_ip, per_ip); + for closed in state.closed_tokens.values() { + assert!(state.closed_token_expiry.unwrap() <= closed.expires_at); + } +} + +#[tokio::test] +async fn real_issuance_reload_renewal_and_shutdown_preserve_expiry_hints() { + let initial = generation(1, 120); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&initial)))); + let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let ip = "192.0.2.10".parse().unwrap(); + let issued = runtime.issue_bootstrap(Arc::clone(&profile), ip).unwrap(); + let hash = token_hash(&issued.token); + let first_expiry = runtime.state.lock().bootstraps[&hash].expires_at; + assert_eq!(runtime.state.lock().bootstrap_expiry, Some(first_expiry)); + + let replacement = generation(2, 60); + runtime.activate_generation(Arc::clone(&replacement)); + let newer = runtime.issue_bootstrap(Arc::clone(&profile), ip).unwrap(); + let newer_hash = token_hash(&newer.token); + let shorter_expiry = runtime.state.lock().bootstraps[&newer_hash].expires_at; + assert!(shorter_expiry < first_expiry); + assert_eq!(runtime.state.lock().bootstrap_expiry, Some(shorter_expiry)); + + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let create = runtime + .create_session( + hash, + &profile.host, + ip, + &hello, + CarrierRequest::legacy([0; 32]), + false, + ) + .unwrap(); + let replay = runtime + .create_session( + hash, + &profile.host, + ip, + &hello, + CarrierRequest::legacy([0; 32]), + false, + ) + .unwrap(); + assert_eq!(create.token, replay.token); + { + let state = runtime.state.lock(); + assert!(state.bootstraps[&hash].expires_at >= first_expiry); + assert_eq!( + state.bootstraps[&hash].timeouts.bootstrap_lifetime_secs, + 120 + ); + assert_eq!(state.bootstrap_expiry, Some(shorter_expiry)); + } + assert_live_invariants(&runtime); + runtime.shutdown().await; + assert_eq!(runtime.state.lock().bootstrap_expiry, None); + assert!(matches!( + runtime.issue_bootstrap(profile, ip), + Err(ManagerError::AdmissionPaused) + )); + for generation in [initial, replacement] { + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_issuance_creation_close_and_reload_preserve_quota_and_hint_invariants() { + let initial = generation(1, 120); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&initial)))); + let barrier = Arc::new(tokio::sync::Barrier::new(33)); + let mut workers = Vec::new(); + for worker in 0..32 { + let runtime = Arc::clone(&runtime); + let barrier = Arc::clone(&barrier); + workers.push(tokio::spawn(async move { + barrier.wait().await; + for _ in 0..8 { + let active = runtime.active_generation(); + let profile = active.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let ip = IpAddr::from([192, 0, 2, worker + 1]); + let issued = runtime.issue_bootstrap(profile.clone(), ip).unwrap(); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let created = runtime + .create_session( + token_hash(&issued.token), + &profile.host, + ip, + &hello, + CarrierRequest::legacy([0; 32]), + false, + ) + .unwrap(); + let hash = token_hash(&created.token); + let session = runtime.get_session(hash, &profile.host).unwrap(); + runtime.close_token(hash, &profile.host, None).unwrap(); + session.wait().await; + runtime.close_token(hash, &profile.host, None).unwrap(); + runtime.cleanup(); + assert_live_invariants(&runtime); + } + })); + } + barrier.wait().await; + let mut generations = vec![initial]; + for id in 2..=8 { + let next = generation(id, if id % 2 == 0 { 60 } else { 120 }); + runtime.activate_generation(Arc::clone(&next)); + generations.push(next); + tokio::task::yield_now().await; + } + for worker in workers { + worker.await.unwrap(); + } + assert_live_invariants(&runtime); + { + let state = runtime.state.lock(); + assert!(state.sessions.is_empty()); + assert!(state.sessions_per_ip.is_empty()); + assert!(state.sessions_per_profile.is_empty()); + assert!(state.bootstraps_per_ip.is_empty()); + } + runtime.shutdown().await; + for generation in generations { + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} diff --git a/src/web/manager/state/expiry/reference.rs b/src/web/manager/state/expiry/reference.rs new file mode 100644 index 0000000..975dc7d --- /dev/null +++ b/src/web/manager/state/expiry/reference.rs @@ -0,0 +1,34 @@ +// Frozen exhaustive sweep from the pre-optimization implementation. +use super::*; + +/// Removes expired bootstrap and closed-token entries while the manager lock is held. +pub(super) fn exhaustive(state: &mut ManagerState, now: Instant) -> usize { + let expired = state + .bootstraps + .iter() + .filter_map(|(hash, bootstrap)| { + (now > bootstrap.expires_at).then_some((*hash, bootstrap.recovery && !bootstrap.used)) + }) + .collect::>(); + let recovery_unused = expired + .iter() + .filter(|(_, recovery_unused)| *recovery_unused) + .count(); + for (hash, _) in expired { + remove_bootstrap_locked(state, hash); + } + state + .closed_tokens + .retain(|_, closed| now <= closed.expires_at); + while state + .closed_session_order + .front() + .and_then(|trace_session_id| state.closed_sessions.get(trace_session_id)) + .is_some_and(|closed| now > closed.expires_at) + { + if let Some(trace_session_id) = state.closed_session_order.pop_front() { + state.closed_sessions.remove(&trace_session_id); + } + } + recovery_unused +} diff --git a/src/web/manager/state/expiry/tests.rs b/src/web/manager/state/expiry/tests.rs new file mode 100644 index 0000000..aa0e636 --- /dev/null +++ b/src/web/manager/state/expiry/tests.rs @@ -0,0 +1,302 @@ +use super::super::*; +use proptest::prelude::*; + +// Keep synthetic ownership fixtures separate from the frozen exhaustive algorithm. +include!("fixtures.rs"); +#[path = "benchmarks.rs"] +mod benchmarks; +#[path = "production_tests.rs"] +mod production_tests; +#[path = "reference.rs"] +mod reference; + +fn assert_equivalent(actual: &ManagerState, expected: &ManagerState) { + let bootstraps = |state: &ManagerState| { + state + .bootstraps + .iter() + .map(|(key, b)| { + ( + *key, + (b.expires_at, b.issued_at, b.issuance_ip, b.used, b.recovery), + ) + }) + .collect::>() + }; + let closed = |state: &ManagerState| { + state + .closed_tokens + .iter() + .map(|(key, c)| (*key, (c.expires_at, c.host.clone(), c.carrier))) + .collect::>() + }; + let sessions = |state: &ManagerState| { + state + .closed_sessions + .iter() + .map(|(key, c)| { + ( + *key, + (c.expires_at, c.closed_at, c.attempt, c.carrier, c.reason), + ) + }) + .collect::>() + }; + assert_eq!(bootstraps(actual), bootstraps(expected)); + assert_eq!(closed(actual), closed(expected)); + assert_eq!(sessions(actual), sessions(expected)); + assert_eq!(actual.closed_session_order, expected.closed_session_order); + assert_eq!(actual.bootstraps_per_ip, expected.bootstraps_per_ip); + for b in actual.bootstraps.values() { + assert!( + actual + .bootstrap_expiry + .is_some_and(|expiry| expiry <= b.expires_at) + ); + assert!(!b.user_registration.is_cancelled()); + } + for c in actual.closed_tokens.values() { + assert!( + actual + .closed_token_expiry + .is_some_and(|expiry| expiry <= c.expires_at) + ); + } +} + +#[test] +fn exact_expiry_boundary_and_recovery_accounting_match_exhaustive_sweep() { + let fixture = Fixture::new(); + let now = Instant::now(); + let deadline = now + Duration::from_secs(1); + let mut actual = ManagerState::default(); + let mut expected = ManagerState::default(); + for state in [&mut actual, &mut expected] { + for flags in 0..4 { + insert_bootstrap(state, [flags; 32], fixture.bootstrap(now, deadline, flags)); + } + insert_closed_token(state, [8; 32], deadline); + insert_bootstrap( + state, + [6; 32], + fixture.bootstrap(now, deadline + Duration::from_secs(2), 0), + ); + insert_closed_token(state, [9; 32], deadline + Duration::from_secs(2)); + } + for (time, count) in [ + (deadline - Duration::from_nanos(1), 0), + (deadline, 0), + (deadline + Duration::from_nanos(1), 1), + // A caller may have captured an older time before waiting for the manager lock. + (deadline - Duration::from_nanos(1), 0), + (deadline + Duration::from_secs(1), 0), + (deadline + Duration::from_secs(3), 0), + ] { + assert_eq!(remove_expired_locked(&mut actual, time), count); + assert_eq!(reference::exhaustive(&mut expected, time), count); + assert_equivalent(&actual, &expected); + } + assert!(actual.bootstraps_per_ip.is_empty()); + assert_eq!(actual.bootstrap_expiry, None); + assert_eq!(actual.closed_token_expiry, None); +} + +#[test] +fn independent_hints_do_not_skip_logical_session_tombstones() { + let fixture = Fixture::new(); + let now = Instant::now(); + for bootstrap_due in [false, true] { + let mut state = ManagerState::default(); + let early = now - Duration::from_secs(1); + let late = now + Duration::from_secs(1); + insert_bootstrap( + &mut state, + [1; 32], + fixture.bootstrap(early, if bootstrap_due { early } else { late }, 0), + ); + insert_closed_token( + &mut state, + [2; 32], + if bootstrap_due { late } else { early }, + ); + state.closed_sessions.insert( + 3, + ClosedSession { + expires_at: early, + attempt: 1, + carrier: WebCarrier::Https, + reason: crate::web::session::SessionCloseReason::ApiClose, + closed_at: early, + }, + ); + state.closed_session_order.push_back(3); + remove_expired_locked(&mut state, now); + assert_eq!(state.bootstraps.is_empty(), bootstrap_due); + assert_eq!(state.closed_tokens.is_empty(), !bootstrap_due); + assert!(state.closed_sessions.is_empty()); + assert!(state.closed_session_order.is_empty()); + } + let mut state = ManagerState::default(); + remember_closed_session_locked( + &mut state, + 1, + 1, + WebCarrier::Https, + crate::web::session::SessionCloseReason::ApiClose, + Duration::ZERO, + 1, + ); + let time = state.closed_sessions[&1].expires_at + Duration::from_nanos(1); + remove_expired_locked(&mut state, time); + assert!(state.closed_sessions.is_empty()); +} + +#[test] +fn closed_token_overwrite_and_capacity_eviction_keep_a_conservative_hint() { + let mut state = ManagerState::default(); + for (lifetime, capacity) in [(60, 2), (1, 2), (120, 2), (1, 0), (60, 1)] { + remember_closed_token_locked( + &mut state, + [1; 32], + "proxy.example.com", + WebCarrier::Https, + Duration::from_secs(lifetime), + capacity, + ); + assert!(state.closed_tokens.len() <= capacity); + for c in state.closed_tokens.values() { + assert!(state.closed_token_expiry.unwrap() <= c.expires_at); + } + } + let deadline = state.closed_tokens[&[1; 32]].expires_at; + remove_expired_locked(&mut state, deadline); + assert_eq!(state.closed_tokens.len(), 1); + remove_expired_locked(&mut state, deadline + Duration::from_nanos(1)); + assert!(state.closed_tokens.is_empty()); + + for (key, lifetime) in [(1, 1), (2, 60), (3, 120)] { + remember_closed_token_locked( + &mut state, + [key; 32], + "proxy.example.com", + WebCarrier::Https, + Duration::from_secs(lifetime), + 2, + ); + } + let stale_hint = state.closed_token_expiry.unwrap(); + assert!(!state.closed_tokens.contains_key(&[1; 32])); + assert!(state.closed_tokens.contains_key(&[2; 32])); + assert!(state.closed_tokens.contains_key(&[3; 32])); + let survivor_min = state + .closed_tokens + .values() + .map(|c| c.expires_at) + .min() + .unwrap(); + assert!(stale_hint < survivor_min); + remove_expired_locked(&mut state, stale_hint + Duration::from_nanos(1)); + assert_eq!(state.closed_tokens.len(), 2); + assert_eq!(state.closed_token_expiry, Some(survivor_min)); +} + +#[test] +fn concurrent_due_sweeps_match_exhaustive_accounting() { + let fixture = Fixture::new(); + let now = Instant::now(); + let states = parking_lot::Mutex::new((ManagerState::default(), ManagerState::default())); + let barrier = std::sync::Barrier::new(16); + std::thread::scope(|scope| { + for worker in 0..16 { + let fixture = &fixture; + let states = &states; + let barrier = &barrier; + scope.spawn(move || { + barrier.wait(); + for iteration in 0..64 { + let flags = (iteration % 4) as u8; + let expired = now - Duration::from_nanos(1); + let first = fixture.bootstrap(expired, expired, flags); + let second = fixture.bootstrap(expired, expired, flags); + let mut states = states.lock(); + let (actual, expected) = &mut *states; + insert_bootstrap(actual, [worker; 32], first); + insert_bootstrap(expected, [worker; 32], second); + insert_closed_token(actual, [worker; 32], expired); + insert_closed_token(expected, [worker; 32], expired); + assert_eq!( + remove_expired_locked(actual, now), + reference::exhaustive(expected, now), + ); + assert_equivalent(actual, expected); + assert!(actual.bootstraps_per_ip.is_empty()); + assert!(actual.closed_tokens.is_empty()); + } + }); + } + }); +} + +proptest! { + #![proptest_config(ProptestConfig::with_cases(128))] + #[test] + fn mixed_registry_operations_match_exhaustive_sweep( + actions in prop::collection::vec((0u8..9, any::(), 0u16..100), 1..300) + ) { + let fixture = Fixture::new(); + let mut actual = ManagerState::default(); + let mut expected = ManagerState::default(); + let start = Instant::now(); + let mut now = start; + for (step, (operation, key, ttl)) in actions.into_iter().enumerate() { + let hash = [key; 32]; + let expiry = now + Duration::from_millis(u64::from(ttl)); + for state in [&mut actual, &mut expected] { + match operation { + 0 => insert_bootstrap(state, hash, fixture.bootstrap( + start + Duration::from_nanos(step as u64), expiry, key % 4, + )), + 1 => remove_bootstrap_locked(state, hash), + 2 => { + if let Some(b) = state.bootstraps.get_mut(&hash) { + if !b.used { + decrement_map(&mut state.bootstraps_per_ip, &b.issuance_ip); + b.used = true; + } + b.expires_at += Duration::from_millis(u64::from(ttl)); + } + } + 3 => { drop(evict_oldest_unused_bootstrap(state)); } + 4 => insert_closed_token(state, hash, expiry), + 5 => { state.closed_tokens.remove(&hash); } + 6 => { + state.bootstraps.clear(); + state.bootstraps_per_ip.clear(); + state.bootstrap_expiry = None; + } + 7 => { + let id = u64::from(key); + if state.closed_sessions.insert(id, ClosedSession { + expires_at: expiry, + attempt: 1, + carrier: WebCarrier::Https, + reason: crate::web::session::SessionCloseReason::ApiClose, + closed_at: now, + }).is_none() { + state.closed_session_order.push_back(id); + } + } + _ => {} + } + } + if operation == 8 { + now += Duration::from_millis(u64::from(ttl)); + } + assert_eq!( + remove_expired_locked(&mut actual, now), + reference::exhaustive(&mut expected, now), + ); + assert_equivalent(&actual, &expected); + } + } +}