astelm/telemt
1
0
Fork 0
mirror of https://github.com/telemt/telemt.git synced 2026-11-01 13:21:58 +03:00
Code Issues Packages Projects Releases Wiki Activity

WEB: skip registry sweeps before the earliest possible expiry

Browse Source
This commit is contained in:
Alexey
2026-10-06 22:00:13 +03:00
parent 9d5b896bb6
commit deafa4a6cf
14 changed files with 1382 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/bridge.rs
+73 -46
View File
@@ -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::<Vec<_>>()
.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<String>; 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!("<script nonce=\"__NONCE__\">\n{DIAGNOSTIC_RUNTIME}\n</script>\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::<Vec<_>>()
.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;
src/web/bridge/render_benchmarks.rs
+85
View File
@@ -0,0 +1,85 @@
use super::*;
use std::hint::black_box;
use std::time::Instant;
fn report(label: &str, mut samples: Vec<u128>) {
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::<Vec<_>>();
handles
.into_iter()
.flat_map(|h| h.join().unwrap())
.collect()
});
report(
&format!(
"bridge repeat={repeat} diagnostics={diagnostics} workers={workers} cached={cached}"
),
samples,
);
}
}
}
}
}
src/web/bridge/render_reference.rs
+155
View File
@@ -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!("<script nonce=\"__NONCE__\">\n{DIAGNOSTIC_RUNTIME}\n</script>\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::<Vec<_>>()
.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"
),
}
}
src/web/bridge/render_tests.rs
+225
View File
@@ -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::<Vec<_>>()
})
})
.collect::<Vec<_>>();
handles
.into_iter()
.flat_map(|handle| handle.join().unwrap())
.collect::<Vec<_>>()
});
assert_eq!(nonces.iter().collect::<HashSet<_>>().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__"));
}
}
src/web/manager.rs
+2
View File
@@ -265,6 +265,8 @@ impl WebProcessRuntime {
) -> Arc<Self> {
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();
src/web/manager/credentials.rs
+3 -1
View File
@@ -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,
src/web/manager/lifecycle.rs
+1
View File
@@ -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::<Vec<_>>()
};
src/web/manager/state.rs
+25 -33
View File
@@ -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<TokenHash, Bootstrap>,
/// Conservative earliest expiry; deletion and frozen-TTL renewal may leave it early.
pub(super) bootstrap_expiry: Option<Instant>,
/// Unused bootstrap ownership counts by forwarded client address.
pub(super) bootstraps_per_ip: HashMap<IpAddr, usize>,
/// Live sessions indexed by bearer-token hash.
@@ -179,6 +190,8 @@ pub(super) struct ManagerState {
pub(super) closed_session_order: VecDeque<u64>,
/// Recently closed token hashes retained for idempotent DELETE semantics.
pub(super) closed_tokens: HashMap<TokenHash, ClosedToken>,
/// Conservative earliest expiry, lowered on every insertion or overwrite.
pub(super) closed_token_expiry: Option<Instant>,
/// Live session counts by forwarded client address.
pub(super) sessions_per_ip: HashMap<IpAddr, usize>,
/// 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::<Vec<_>>();
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,
},
src/web/manager/state/expiry.rs
+67
View File
@@ -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<Instant> = 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::<Vec<_>>();
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<Instant> = 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;
src/web/manager/state/expiry/benchmarks.rs
+132
View File
@@ -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<u128>) {
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::<Vec<_>>();
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,
);
}
}
}
}
src/web/manager/state/expiry/fixtures.rs
+94
View File
@@ -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<UserAdmissionAuthority>,
profile: Arc<WebRuntimeProfile>,
}
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,
});
}
src/web/manager/state/expiry/production_tests.rs
+184
View File
@@ -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<RuntimeGeneration> {
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;
}
}
src/web/manager/state/expiry/reference.rs
+34
View File
@@ -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::<Vec<_>>();
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
}
src/web/manager/state/expiry/tests.rs
+302
View File
@@ -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::<BTreeMap<_, _>>()
};
let closed = |state: &ManagerState| {
state
.closed_tokens
.iter()
.map(|(key, c)| (*key, (c.expires_at, c.host.clone(), c.carrier)))
.collect::<BTreeMap<_, _>>()
};
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::<BTreeMap<_, _>>()
};
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::<u8>(), 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);
}
}
}