From 718ce0847e70c1b66648d36f2af624edc4f2444a Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Mon, 31 Aug 2026 21:39:09 +0300 Subject: [PATCH] WEB Carrier Counters + Status Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/config_edit.rs | 6 +- src/api/web_runtime.rs | 7 +- src/api/web_runtime/observability.rs | 40 ++ src/config/hot_reload.rs | 2 +- src/config/hot_reload/fields.rs | 13 +- src/config/hot_reload/tests.rs | 41 ++ src/config/hot_reload/watcher.rs | 12 +- src/config/load.rs | 5 + src/config/load/validate_web/negotiation.rs | 2 +- src/config/types.rs | 1 + src/config/types/web_carrier.rs | 3 + src/maestro/listeners/control.rs | 12 + src/maestro/reload_supervisor.rs | 15 +- src/maestro/runtime_build.rs | 41 +- src/maestro/runtime_build_tests.rs | 87 +++- src/metrics/web.rs | 189 +++++++- src/web/bridge.rs | 2 + src/web/bridge/document.html | 3 + src/web/bridge/response.js | 40 ++ src/web/bridge/runtime.js | 138 +++--- src/web/bridge/tests.rs | 11 +- src/web/http/carrier_diagnostic_tests.rs | 148 +++++++ src/web/http/request.rs | 2 +- src/web/http/session.rs | 13 +- src/web/http/tests.rs | 3 + src/web/manager.rs | 5 +- src/web/manager/carrier_learning.rs | 217 ++++----- src/web/manager/carrier_learning/status.rs | 140 ++++++ src/web/manager/carrier_learning/tests.rs | 415 ++++++++++++++---- src/web/manager/carrier_outcome.rs | 97 +++- src/web/manager/credentials.rs | 22 +- src/web/manager/lifecycle.rs | 49 ++- src/web/manager/negotiation.rs | 20 + src/web/manager/session_creation.rs | 94 ++-- .../manager/session_creation/replacement.rs | 11 + src/web/manager/state.rs | 3 + src/web/manager/status.rs | 4 + src/web/manager/websocket.rs | 22 + src/web/manager/websocket/tests.rs | 52 +++ src/web/session.rs | 9 +- src/web/session/downlink.rs | 4 +- src/web/session/lane_uplink.rs | 6 +- src/web/session/lanes.rs | 4 +- src/web/session/lifecycle.rs | 20 +- src/web/session/negotiation.rs | 226 +++++++++- src/web/session/status.rs | 14 +- src/web/session/uplink.rs | 6 +- src/web/session/websocket.rs | 10 +- src/web/telemetry.rs | 50 +-- src/web/telemetry/carrier.rs | 310 +++++++++++++ src/web/telemetry/tests.rs | 59 +++ 51 files changed, 2252 insertions(+), 453 deletions(-) create mode 100644 src/web/bridge/response.js create mode 100644 src/web/http/carrier_diagnostic_tests.rs create mode 100644 src/web/manager/carrier_learning/status.rs create mode 100644 src/web/telemetry/carrier.rs create mode 100644 src/web/telemetry/tests.rs diff --git a/src/api/config_edit.rs b/src/api/config_edit.rs index b9a8fd2..0f46236 100644 --- a/src/api/config_edit.rs +++ b/src/api/config_edit.rs @@ -51,7 +51,8 @@ pub(super) async fn patch_config( let active_config = shared.active_runtime.load_full().config(); let mut prepared = prepare_patch_to_path(&shared.config_path, &patch_json, expected_revision).await?; - let resolved = resolve_reload_config(&active_config, &prepared.desired_config); + let resolved = resolve_reload_config(&active_config, &prepared.desired_config) + .map_err(ApiFailure::bad_request)?; prepared.response.runtime_reload_required = resolved.runtime_changed; prepared.response.process_restart_required = !resolved.deferred_process_fields.is_empty(); prepared.response.deferred_process_fields = resolved.deferred_process_fields; @@ -205,7 +206,8 @@ async fn prepare_patch_to_path( let revision = compute_snapshot_revision(&candidate); let new_cfg = candidate.config; let class = classify_config_changes(&old_cfg, &new_cfg); - let deferred_process_fields = deferred_process_fields(&old_cfg, &new_cfg); + let deferred_process_fields = + deferred_process_fields(&old_cfg, &new_cfg).map_err(ApiFailure::bad_request)?; Ok(PreparedConfigPatch { owner_path, diff --git a/src/api/web_runtime.rs b/src/api/web_runtime.rs index b9ccae6..b825f49 100644 --- a/src/api/web_runtime.rs +++ b/src/api/web_runtime.rs @@ -19,7 +19,9 @@ use crate::web::manager::{ControlError, OperatorLifecycleError, SessionDetail, W mod request; // Ingress, capacity, and decoy telemetry remain separate availability planes. mod observability; -use observability::{WebCapacityStatus, WebDecoyUpstreamStatus, WebIngressStatus}; +use observability::{ + WebCapacityStatus, WebCarrierNegotiationStatus, WebDecoyUpstreamStatus, WebIngressStatus, +}; use request::{ CloseRequest, DrainRequest, RuntimeInstanceRequest, parse_session_query, parse_session_ref, valid_runtime_instance, @@ -269,6 +271,7 @@ struct WebStatusData { ingress: WebIngressStatus, capacity: WebCapacityStatus, decoy_upstream: WebDecoyUpstreamStatus, + carrier_negotiation: WebCarrierNegotiationStatus, #[serde(skip_serializing_if = "Option::is_none")] operator_lifecycle: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -302,6 +305,7 @@ impl WebStatusData { let ingress = WebIngressStatus::new(&publication, runtime.is_some()); let capacity = WebCapacityStatus::new(&publication, runtime, config); let decoy_upstream = WebDecoyUpstreamStatus::new(&publication); + let carrier_negotiation = WebCarrierNegotiationStatus::new(&publication); Self { lifecycle: publication.lifecycle.as_str(), lifecycle_epoch: publication.epoch, @@ -317,6 +321,7 @@ impl WebStatusData { ingress, capacity, decoy_upstream, + carrier_negotiation, operator_lifecycle, runtime: runtime.map(WebProcessRuntime::try_status), } diff --git a/src/api/web_runtime/observability.rs b/src/api/web_runtime/observability.rs index 9dad37c..a545ca7 100644 --- a/src/api/web_runtime/observability.rs +++ b/src/api/web_runtime/observability.rs @@ -4,6 +4,9 @@ use crate::config::{ProxyConfig, WebHttpConnectionCapacityAction}; use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; use crate::web::manager::{WebCapacityResourceStatus, WebCapacitySnapshot, WebProcessRuntime}; use crate::web::telemetry::{WebOutcomeCounter, WebRejectionCounter}; +use crate::web::telemetry::{ + WebCarrierFailureCounter, WebCarrierLearningCounter, WebCarrierSelectionCounter, +}; /// Private WEB ingress state owned by this Telemt process. #[derive(Serialize)] @@ -117,6 +120,25 @@ impl WebDecoyUpstreamStatus { } } +/// Fixed-cardinality process-lifetime carrier negotiation counters. +#[derive(Serialize)] +pub(super) struct WebCarrierNegotiationStatus { + selections: Vec, + reported_failures: Vec, + learning_outcomes: Vec, +} + +impl WebCarrierNegotiationStatus { + /// Builds counters from publication ownership even when runtime state is unavailable. + pub(super) fn new(publication: &WebRuntimePublication) -> Self { + Self { + selections: publication.telemetry.carrier_selection_counters(), + reported_failures: publication.telemetry.carrier_failure_counters(), + learning_outcomes: publication.telemetry.carrier_learning_counters(), + } + } +} + #[cfg(test)] mod tests { use crate::config::ProxyConfig; @@ -143,6 +165,8 @@ mod tests { serde_json::to_value(super::WebCapacityStatus::new(&publication, None, &config)) .unwrap(); let decoy = serde_json::to_value(super::WebDecoyUpstreamStatus::new(&publication)).unwrap(); + let carrier = + serde_json::to_value(super::WebCarrierNegotiationStatus::new(&publication)).unwrap(); assert_eq!( capacity["rejections"].as_array().unwrap().len(), @@ -160,5 +184,21 @@ mod tests { crate::web::telemetry::WebDecoyUpstreamOutcome::ALL.len() ); assert_eq!(capacity["partial"][0], "runtime"); + assert_eq!( + carrier["selections"].as_array().unwrap().len(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierSelectionDisposition::ALL.len() + ); + assert_eq!( + carrier["reported_failures"].as_array().unwrap().len(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierFailurePhase::ALL.len() + * crate::web::manager::CarrierFailure::ALL.len() + ); + assert_eq!( + carrier["learning_outcomes"].as_array().unwrap().len(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierLearningOutcome::ALL.len() + ); } } diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 10e3d9d..2c4407b 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -37,7 +37,7 @@ use super::load::{LoadedConfig, ProxyConfig}; #[allow(unused_imports)] use crate::config::{ CidrRateLimitKey, LogLevel, MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy, MeTelemetryLevel, - MeWriterPickMode, WebDebugConfig, web_debug_fits_limits, + MeWriterPickMode, WEB_CARRIER_LEARNING_MIN_ENTRIES, WebDebugConfig, web_debug_fits_limits, }; #[cfg(test)] use crate::config::{ListenerConfig, SynLimitMode}; diff --git a/src/config/hot_reload/fields.rs b/src/config/hot_reload/fields.rs index 372681c..8ca7747 100644 --- a/src/config/hot_reload/fields.rs +++ b/src/config/hot_reload/fields.rs @@ -345,15 +345,22 @@ pub(super) fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyC let process_limits = cfg.web.limits.clone(); cfg.web = new.web.clone(); cfg.web.limits = process_limits; + if cfg.web.carrier_negotiation_enabled() + && cfg.web.carrier_learning + && cfg.web.limits.max_carrier_learning_entries < WEB_CARRIER_LEARNING_MIN_ENTRIES + { + if old.web.carrier_learning != new.web.carrier_learning { + cfg.web.carrier_learning = old.web.carrier_learning; + } else { + cfg.web.carriers = old.web.carriers.clone(); + } + } if !web_debug_fits_limits(&cfg.web.debug, &cfg.web.limits) { cfg.web.debug = old.web.debug.clone(); } if cfg.rebuild_runtime_user_auth().is_err() { cfg.runtime_user_auth = None; } - if cfg.rebuild_runtime_web().is_err() { - cfg.web = old.web.clone(); - } cfg } diff --git a/src/config/hot_reload/tests.rs b/src/config/hot_reload/tests.rs index 606ecd7..4bb6869 100644 --- a/src/config/hot_reload/tests.rs +++ b/src/config/hot_reload/tests.rs @@ -123,6 +123,47 @@ fn web_debug_policy_is_hot_while_debug_capacity_is_process_owned() { ); } +#[test] +fn hot_overlay_defers_learning_that_requires_new_process_capacity() { + let mut old = sample_config(); + old.web.limits.max_carrier_learning_entries = 1; + old.web.carriers = crate::config::WebCarriers::Disabled; + old.web.carrier_learning = false; + let mut new = old.clone(); + new.web.limits.max_carrier_learning_entries = 3; + new.web.carriers = crate::config::WebCarriers::Enabled(vec![ + crate::config::WebCarrier::Websocket, + crate::config::WebCarrier::Https, + ]); + new.web.carrier_learning = true; + + let applied = overlay_hot_fields(&old, &new); + + assert_eq!(applied.web.limits.max_carrier_learning_entries, 1); + assert!(applied.web.carrier_negotiation_enabled()); + assert!(!applied.web.carrier_learning); +} + +#[test] +fn hot_overlay_defers_carriers_for_dormant_learning_with_small_capacity() { + let mut old = sample_config(); + old.web.limits.max_carrier_learning_entries = 1; + old.web.carriers = crate::config::WebCarriers::Disabled; + old.web.carrier_learning = true; + let mut new = old.clone(); + new.web.limits.max_carrier_learning_entries = 3; + new.web.carriers = crate::config::WebCarriers::Enabled(vec![ + crate::config::WebCarrier::Websocket, + crate::config::WebCarrier::Https, + ]); + + let applied = overlay_hot_fields(&old, &new); + + assert_eq!(applied.web.limits.max_carrier_learning_entries, 1); + assert!(!applied.web.carrier_negotiation_enabled()); + assert!(applied.web.carrier_learning); +} + #[test] fn web_debug_prefix_requiring_deferred_capacity_is_not_hot_applied() { let old = sample_config(); diff --git a/src/config/hot_reload/watcher.rs b/src/config/hot_reload/watcher.rs index b0c0538..9197d2b 100644 --- a/src/config/hot_reload/watcher.rs +++ b/src/config/hot_reload/watcher.rs @@ -161,7 +161,17 @@ fn reload_config_with_resolver( } let old_cfg = config_tx.borrow().clone(); - let applied_cfg = overlay_hot_fields(&old_cfg, &new_cfg); + let mut applied_cfg = overlay_hot_fields(&old_cfg, &new_cfg); + if let Err(error) = applied_cfg + .validate_effective_web() + .and_then(|_| applied_cfg.rebuild_runtime_web()) + { + error!( + "config reload: effective WEB validation failed: {}; keeping old config", + error + ); + return Some(next_manifest); + } let old_hot = HotFields::from_config(&old_cfg); let applied_hot = HotFields::from_config(&applied_cfg); let non_hot_changed = !config_equal(&applied_cfg, &new_cfg); diff --git a/src/config/load.rs b/src/config/load.rs index 66876e9..e6560a2 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -216,6 +216,11 @@ impl ProxyConfig { runtime_web::rebuild(self) } + /// Validates the mixed effective WEB snapshot after restart fields are retained. + pub(crate) fn validate_effective_web(&mut self) -> Result<()> { + validate_web::validate(self) + } + /// Revalidates decoy separation after restart-only listener fields are resolved. pub(crate) fn validate_web_decoy_listener_separation(&self) -> Result<()> { validate_web::validate_decoy_listener_separation(self) diff --git a/src/config/load/validate_web/negotiation.rs b/src/config/load/validate_web/negotiation.rs index 3544ac4..d398a87 100644 --- a/src/config/load/validate_web/negotiation.rs +++ b/src/config/load/validate_web/negotiation.rs @@ -16,7 +16,7 @@ pub(super) fn validate(config: &WebConfig) -> Result> { let candidates = config.carrier_candidates(); if config.carrier_negotiation_enabled() && config.carrier_learning - && config.limits.max_carrier_learning_entries < 3 + && config.limits.max_carrier_learning_entries < WEB_CARRIER_LEARNING_MIN_ENTRIES { return config_error( "web.limits.max_carrier_learning_entries must be >= 3 when carrier learning is enabled", diff --git a/src/config/types.rs b/src/config/types.rs index d5f8d61..72813f7 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -63,6 +63,7 @@ pub(crate) use web::{ }; #[allow(unused_imports)] pub use web_carrier::{WebCarrier, WebCarriers}; +pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES; pub(crate) use web_debug::web_debug_fits_limits; pub use web_debug::{WebDebugBodyCapture, WebDebugConfig}; diff --git a/src/config/types/web_carrier.rs b/src/config/types/web_carrier.rs index f3749e2..3dcee56 100644 --- a/src/config/types/web_carrier.rs +++ b/src/config/types/web_carrier.rs @@ -1,5 +1,8 @@ use serde::{Deserialize, Serialize}; +/// Minimum restart-owned entries required for one complete learning sample. +pub(crate) const WEB_CARRIER_LEARNING_MIN_ENTRIES: usize = 3; + /// Carrier selected for one newly issued WEB relay session. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] diff --git a/src/maestro/listeners/control.rs b/src/maestro/listeners/control.rs index b0a1380..56ecd3d 100644 --- a/src/maestro/listeners/control.rs +++ b/src/maestro/listeners/control.rs @@ -248,6 +248,18 @@ impl ListenerManager { ); } + /// Publishes one generation through the process-owned WEB policy fence. + pub(crate) fn activate_runtime_generation( + &self, + generation: Arc, + ) -> Arc { + if let Some(runtime) = &self.web_runtime { + runtime.activate_generation(generation) + } else { + self.active_runtime.swap(generation) + } + } + /// Stops every accept task and applies one deadline to the complete WEB ingress. pub(crate) async fn shutdown(&mut self) -> Result<(), String> { self.web_control.publish( diff --git a/src/maestro/reload_supervisor.rs b/src/maestro/reload_supervisor.rs index 77dcc96..d186745 100644 --- a/src/maestro/reload_supervisor.rs +++ b/src/maestro/reload_supervisor.rs @@ -159,7 +159,13 @@ impl ReloadSupervisor { .mark_phase(command.reload_id, ReloadPhase::Preparing) .await; let old_runtime = self.active_runtime.load_full(); - let resolved = resolve_reload_config(&old_runtime.config(), &command.config); + let resolved = match resolve_reload_config(&old_runtime.config(), &command.config) { + Ok(resolved) => resolved, + Err(error) => { + self.control.fail(command.reload_id, error).await; + return; + } + }; self.control .set_deferred_fields(command.reload_id, resolved.deferred_process_fields.clone()) .await; @@ -292,8 +298,11 @@ impl ReloadSupervisor { } else { None }; - old_runtime.stop_accepting_sessions(); - let replaced = self.active_runtime.swap(new_runtime.clone()); + let replaced = { + let listener_manager = self.listener_manager.lock().await; + old_runtime.stop_accepting_sessions(); + listener_manager.activate_runtime_generation(new_runtime.clone()) + }; self.web_trace .apply_policy(new_runtime.id, &new_runtime.config().web.debug); config_watcher_activation.send_replace(true); diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index 12f9c83..f43b19b 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -5,7 +5,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::{RwLock, Semaphore, watch}; -use crate::config::{ProxyConfig, ServerConfig, web_debug_fits_limits}; +use crate::config::{ + ProxyConfig, ServerConfig, WEB_CARRIER_LEARNING_MIN_ENTRIES, web_debug_fits_limits, +}; use crate::crypto::SecureRandom; use crate::ip_tracker::UserIpTracker; use crate::network::probe::{decide_network_capabilities, run_probe}; @@ -337,7 +339,7 @@ pub(crate) struct ResolvedReloadConfig { pub(crate) fn resolve_reload_config( old: &ProxyConfig, desired: &ProxyConfig, -) -> ResolvedReloadConfig { +) -> Result { let mut effective = desired.clone(); let mut fields = Vec::new(); let listener_identity_matches = listeners_have_same_bind_identity(&old.server, &desired.server); @@ -424,21 +426,39 @@ pub(crate) fn resolve_reload_config( { fields.push("web.limits".to_string()); effective.web.limits = old.web.limits.clone(); - if effective.rebuild_runtime_web().is_err() { - fields.push("web".to_string()); - effective.web = old.web.clone(); + } + if effective.web.carrier_negotiation_enabled() + && effective.web.carrier_learning + && effective.web.limits.max_carrier_learning_entries + < WEB_CARRIER_LEARNING_MIN_ENTRIES + { + if old.web.carrier_learning != desired.web.carrier_learning { + fields.push("web.carrier_learning".to_string()); + effective.web.carrier_learning = old.web.carrier_learning; + } else { + fields.push("web.carriers".to_string()); + effective.web.carriers = old.web.carriers.clone(); } } if !web_debug_fits_limits(&effective.web.debug, &effective.web.limits) { fields.push("web.debug".to_string()); effective.web.debug = old.web.debug.clone(); } + effective + .validate_effective_web() + .map_err(|error| format!("effective WEB configuration is invalid: {error}"))?; + effective + .rebuild_runtime_user_auth() + .map_err(|error| format!("effective user runtime preparation failed: {error}"))?; + effective + .rebuild_runtime_web() + .map_err(|error| format!("effective WEB runtime preparation failed: {error}"))?; let runtime_changed = !configs_equal(old, &effective); - ResolvedReloadConfig { + Ok(ResolvedReloadConfig { effective, deferred_process_fields: fields, runtime_changed, - } + }) } fn listeners_have_same_bind_identity(old: &ServerConfig, desired: &ServerConfig) -> bool { @@ -469,8 +489,11 @@ fn listener_process_fields_equal(old: &ServerConfig, desired: &ServerConfig) -> } /// Returns process-owned fields that cannot change in the current generation. -pub(crate) fn deferred_process_fields(old: &ProxyConfig, new: &ProxyConfig) -> Vec { - resolve_reload_config(old, new).deferred_process_fields +pub(crate) fn deferred_process_fields( + old: &ProxyConfig, + new: &ProxyConfig, +) -> Result, String> { + resolve_reload_config(old, new).map(|resolved| resolved.deferred_process_fields) } fn configs_equal(old: &ProxyConfig, new: &ProxyConfig) -> bool { diff --git a/src/maestro/runtime_build_tests.rs b/src/maestro/runtime_build_tests.rs index 080ccfc..a0db12e 100644 --- a/src/maestro/runtime_build_tests.rs +++ b/src/maestro/runtime_build_tests.rs @@ -31,7 +31,7 @@ fn process_socket_and_logging_changes_are_deferred() { new.server.listen_backlog = new.server.listen_backlog.saturating_add(1); new.general.disable_colors = !new.general.disable_colors; - let fields = deferred_process_fields(&old, &new); + let fields = deferred_process_fields(&old, &new).unwrap(); assert!(fields.contains(&"server.listeners".to_string())); assert!(fields.contains(&"general.disable_colors".to_string())); } @@ -43,7 +43,7 @@ fn global_mss_profiles_are_deferred_with_the_listener_socket_group() { desired.server.client_mss = Some("92".to_string()); desired.server.client_mss_bulk = Some("1400".to_string()); - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!( resolved.deferred_process_fields, @@ -64,7 +64,7 @@ fn mixed_reload_retains_process_state_and_applies_runtime_state() { desired.server.client_mss = Some("92".to_string()); desired.censorship.tls_domain = "reload.example".to_string(); - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!(resolved.effective.server.client_mss, old.server.client_mss); assert_eq!( @@ -105,7 +105,7 @@ fn listener_announcement_is_runtime_owned_when_bind_identity_is_stable() { let mut desired = old.clone(); desired.server.listeners[0].announce = Some("proxy.example".to_string()); - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert!(resolved.deferred_process_fields.is_empty()); assert_eq!( @@ -128,7 +128,7 @@ fn process_field_labels_are_stable_ordered_and_unique() { .saturating_add(1); desired.general.disable_colors = !desired.general.disable_colors; - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!( resolved.deferred_process_fields, @@ -146,7 +146,7 @@ fn runtime_only_change_does_not_require_process_rebind() { let old = ProxyConfig::default(); let mut new = old.clone(); new.censorship.tls_domain = "reload.example".to_string(); - assert!(deferred_process_fields(&old, &new).is_empty()); + assert!(deferred_process_fields(&old, &new).unwrap().is_empty()); } #[test] @@ -157,7 +157,7 @@ fn web_allocation_limits_are_deferred_until_restart() { let mut desired = old.clone(); desired.web.limits.max_sessions_global += 1; - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!( resolved.deferred_process_fields, @@ -170,6 +170,65 @@ fn web_allocation_limits_are_deferred_until_restart() { assert!(!resolved.runtime_changed); } +#[test] +fn enabling_learning_is_deferred_when_retained_capacity_is_too_small() { + let mut old = ProxyConfig::default(); + old.web.limits.max_carrier_learning_entries = 1; + old.web.carriers = crate::config::WebCarriers::Disabled; + old.web.carrier_learning = false; + old.rebuild_runtime_user_auth().unwrap(); + old.rebuild_runtime_web().unwrap(); + let mut desired = old.clone(); + desired.web.limits.max_carrier_learning_entries = 3; + desired.web.carriers = crate::config::WebCarriers::Enabled(vec![ + crate::config::WebCarrier::Websocket, + crate::config::WebCarrier::Https, + ]); + desired.web.carrier_learning = true; + + let resolved = resolve_reload_config(&old, &desired).unwrap(); + + assert_eq!( + resolved.deferred_process_fields, + vec!["web.limits".to_string(), "web.carrier_learning".to_string()] + ); + assert_eq!( + resolved.effective.web.limits.max_carrier_learning_entries, + 1 + ); + assert!(resolved.effective.web.carrier_negotiation_enabled()); + assert!(!resolved.effective.web.carrier_learning); +} + +#[test] +fn enabling_carriers_is_deferred_for_dormant_learning_with_small_capacity() { + let mut old = ProxyConfig::default(); + old.web.limits.max_carrier_learning_entries = 1; + old.web.carriers = crate::config::WebCarriers::Disabled; + old.web.carrier_learning = true; + old.rebuild_runtime_user_auth().unwrap(); + old.rebuild_runtime_web().unwrap(); + let mut desired = old.clone(); + desired.web.limits.max_carrier_learning_entries = 3; + desired.web.carriers = crate::config::WebCarriers::Enabled(vec![ + crate::config::WebCarrier::Websocket, + crate::config::WebCarrier::Https, + ]); + + let resolved = resolve_reload_config(&old, &desired).unwrap(); + + assert_eq!( + resolved.deferred_process_fields, + vec!["web.limits".to_string(), "web.carriers".to_string()] + ); + assert_eq!( + resolved.effective.web.limits.max_carrier_learning_entries, + 1 + ); + assert!(!resolved.effective.web.carrier_negotiation_enabled()); + assert!(resolved.effective.web.carrier_learning); +} + #[test] fn web_debug_prefix_dependent_on_new_capacity_is_deferred_with_limits() { let mut old = ProxyConfig::default(); @@ -179,7 +238,7 @@ fn web_debug_prefix_dependent_on_new_capacity_is_deferred_with_limits() { desired.web.limits.max_body_bytes = 4 * 1024 * 1024; desired.web.debug.body_prefix_bytes = 3 * 1024 * 1024; - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!( resolved.deferred_process_fields, @@ -206,7 +265,7 @@ fn endpoint_only_listener_move_is_runtime_rebindable() { let mut desired = old.clone(); desired.server.listeners[0].port = Some(8443); - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert!(resolved.deferred_process_fields.is_empty()); assert_eq!(resolved.effective.server.listeners[0].port, Some(8443)); @@ -221,7 +280,7 @@ fn synlimited_endpoint_move_remains_restart_only() { let mut desired = old.clone(); desired.server.listeners[0].port = Some(8443); - let resolved = resolve_reload_config(&old, &desired); + let resolved = resolve_reload_config(&old, &desired).unwrap(); assert_eq!( resolved.deferred_process_fields, @@ -253,11 +312,5 @@ fn deferred_listener_identity_cannot_create_an_effective_decoy_loop() { ]; assert!(desired.validate_web_decoy_listener_separation().is_ok()); - let resolved = resolve_reload_config(&old, &desired); - assert!( - resolved - .effective - .validate_web_decoy_listener_separation() - .is_err() - ); + assert!(resolve_reload_config(&old, &desired).is_err()); } diff --git a/src/metrics/web.rs b/src/metrics/web.rs index 23f6b5d..6cdbaa7 100644 --- a/src/metrics/web.rs +++ b/src/metrics/web.rs @@ -1,9 +1,13 @@ use std::fmt::Write; -use crate::config::{ProxyConfig, WebHttpConnectionCapacityAction}; +use crate::config::{ + ProxyConfig, WebCarrier, WebCarrierNegotiationAggressiveness, + WebHttpConnectionCapacityAction, +}; use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; -use crate::web::manager::OperatorLifecycleState; +use crate::web::manager::{CarrierFailure, OperatorLifecycleState}; use crate::web::telemetry::{ + WebCarrierFailurePhase, WebCarrierLearningOutcome, WebCarrierSelectionDisposition, WebDecoyUpstreamOutcome, WebHttpConnectionOverloadOutcome, WebRejectionReason, }; @@ -192,9 +196,158 @@ pub(super) fn render(out: &mut String, publication: &WebRuntimePublication, conf ); } + render_carrier_negotiation(out, publication, runtime.as_deref(), config); render_aggregate_totals(out, publication); } +fn render_carrier_negotiation( + out: &mut String, + publication: &WebRuntimePublication, + runtime: Option<&crate::web::manager::WebProcessRuntime>, + config: &ProxyConfig, +) { + let _ = writeln!( + out, + "# HELP telemt_web_carrier_selections_total Successful carrier selections by learning disposition" + ); + let _ = writeln!(out, "# TYPE telemt_web_carrier_selections_total counter"); + for carrier in WebCarrier::ALL { + for disposition in WebCarrierSelectionDisposition::ALL { + let _ = writeln!( + out, + "telemt_web_carrier_selections_total{{carrier=\"{}\",disposition=\"{}\"}} {}", + carrier.as_str(), + disposition.as_str(), + publication + .telemetry + .carrier_selection_total(carrier, disposition) + ); + } + } + + let _ = writeln!( + out, + "# HELP telemt_web_carrier_reported_failures_total Canonical client-reported failures for authenticated carrier chains" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_carrier_reported_failures_total counter" + ); + for carrier in WebCarrier::ALL { + for phase in WebCarrierFailurePhase::ALL { + for reason in CarrierFailure::ALL { + let _ = writeln!( + out, + "telemt_web_carrier_reported_failures_total{{carrier=\"{}\",phase=\"{}\",reason=\"{}\"}} {}", + carrier.as_str(), + phase.as_str(), + reason.as_str(), + publication + .telemetry + .carrier_failure_total(carrier, phase, reason) + ); + } + } + } + + let _ = writeln!( + out, + "# HELP telemt_web_carrier_learning_outcomes_total Terminal carrier health and evidence publication outcomes" + ); + let _ = writeln!( + out, + "# TYPE telemt_web_carrier_learning_outcomes_total counter" + ); + for carrier in WebCarrier::ALL { + for outcome in WebCarrierLearningOutcome::ALL { + let _ = writeln!( + out, + "telemt_web_carrier_learning_outcomes_total{{carrier=\"{}\",outcome=\"{}\"}} {}", + carrier.as_str(), + outcome.as_str(), + publication + .telemetry + .carrier_learning_total(carrier, outcome) + ); + } + } + + let learning = runtime.and_then(|runtime| runtime.try_carrier_learning_status()); + let policy_matches = learning.as_ref().is_some_and(|status| { + status.policy_generation == runtime.map(|runtime| runtime.active_generation().id) + && status.enabled + == (config.web.carrier_negotiation_enabled() && config.web.carrier_learning) + && status.aggressiveness == config.web.carrier_negotiation_aggressiveness + && status.lifetime_secs == config.web.timeouts.carrier_learning_secs + && status.health_secs == config.web.timeouts.carrier_health_secs + }); + let active_state = if runtime.is_none() { + "unavailable" + } else if learning.is_none() { + "partial" + } else if !policy_matches { + "pending" + } else if learning.as_ref().is_some_and(|status| status.epoch.is_none()) { + "exhausted" + } else if learning.as_ref().is_some_and(|status| status.enabled) { + "enabled" + } else { + "disabled" + }; + let _ = writeln!( + out, + "# HELP telemt_web_carrier_learning_state Current bounded learning policy state" + ); + let _ = writeln!(out, "# TYPE telemt_web_carrier_learning_state gauge"); + for state in LEARNING_STATES { + let _ = writeln!( + out, + "telemt_web_carrier_learning_state{{state=\"{state}\"}} {}", + flag(active_state == state) + ); + } + + let _ = writeln!( + out, + "# HELP telemt_web_carrier_learning_entries Current bounded evidence entries" + ); + let _ = writeln!(out, "# TYPE telemt_web_carrier_learning_entries gauge"); + for (kind, value) in [ + ("used", learning.as_ref().map_or(0, |status| status.entries)), + ("limit", learning.as_ref().map_or(0, |status| status.capacity)), + ] { + let _ = writeln!( + out, + "telemt_web_carrier_learning_entries{{kind=\"{kind}\"}} {value}" + ); + } + + let _ = writeln!( + out, + "# HELP telemt_web_carrier_learning_policy Effective learning aggressiveness" + ); + let _ = writeln!(out, "# TYPE telemt_web_carrier_learning_policy gauge"); + for aggressiveness in [ + WebCarrierNegotiationAggressiveness::Conservative, + WebCarrierNegotiationAggressiveness::Balanced, + WebCarrierNegotiationAggressiveness::Aggressive, + ] { + let token = match aggressiveness { + WebCarrierNegotiationAggressiveness::Conservative => "conservative", + WebCarrierNegotiationAggressiveness::Balanced => "balanced", + WebCarrierNegotiationAggressiveness::Aggressive => "aggressive", + }; + let active = learning + .as_ref() + .is_some_and(|status| status.aggressiveness == aggressiveness); + let _ = writeln!( + out, + "telemt_web_carrier_learning_policy{{aggressiveness=\"{token}\"}} {}", + flag(active) + ); + } +} + fn render_capacity(out: &mut String, snapshot: &crate::web::manager::WebCapacitySnapshot) { let _ = writeln!( out, @@ -328,6 +481,15 @@ const OPERATOR_STATES: [&str; 6] = [ "drained", ]; +const LEARNING_STATES: [&str; 6] = [ + "unavailable", + "partial", + "pending", + "exhausted", + "disabled", + "enabled", +]; + const fn flag(value: bool) -> u8 { if value { 1 } else { 0 } } @@ -361,5 +523,28 @@ mod tests { crate::web::telemetry::WebDecoyUpstreamOutcome::ALL.len() ); assert!(output.contains("telemt_web_ingress_lifecycle_state{state=\"starting\"} 1")); + assert_eq!( + output.matches("telemt_web_carrier_selections_total{").count(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierSelectionDisposition::ALL.len() + ); + assert_eq!( + output + .matches("telemt_web_carrier_reported_failures_total{") + .count(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierFailurePhase::ALL.len() + * crate::web::manager::CarrierFailure::ALL.len() + ); + assert_eq!( + output + .matches("telemt_web_carrier_learning_outcomes_total{") + .count(), + crate::config::WebCarrier::ALL.len() + * crate::web::telemetry::WebCarrierLearningOutcome::ALL.len() + ); + assert!(output.contains( + "telemt_web_carrier_learning_state{state=\"unavailable\"} 1" + )); } } diff --git a/src/web/bridge.rs b/src/web/bridge.rs index 7f3759a..385ab17 100644 --- a/src/web/bridge.rs +++ b/src/web/bridge.rs @@ -34,6 +34,7 @@ pub(crate) fn render( rng.fill(&mut nonce); let nonce = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(nonce); let body = DOCUMENT + .replace("__RESPONSE_RUNTIME__", RESPONSE_RUNTIME) .replace("__RUNTIME__", RUNTIME) .replace("__NONCE__", &nonce) .replace("__HOST__", host) @@ -70,6 +71,7 @@ pub(crate) fn render( } const DOCUMENT: &str = include_str!("bridge/document.html"); +const RESPONSE_RUNTIME: &str = include_str!("bridge/response.js"); const RUNTIME: &str = include_str!("bridge/runtime.js"); // Rendered wire-contract tests remain separate from the embedded document. diff --git a/src/web/bridge/document.html b/src/web/bridge/document.html index a7c25b5..56dab71 100644 --- a/src/web/bridge/document.html +++ b/src/web/bridge/document.html @@ -7,6 +7,9 @@ + diff --git a/src/web/bridge/response.js b/src/web/bridge/response.js new file mode 100644 index 0000000..50681c5 --- /dev/null +++ b/src/web/bridge/response.js @@ -0,0 +1,40 @@ +(()=>{ +'use strict'; +const maxChunks=4096; +function cancel(response){ + if(!response.body)return; + try{const pending=response.body.cancel();if(pending&&typeof pending.catch==='function')pending.catch(()=>{})}catch(error){} +} +function declaredLength(response){ + const value=response.headers.get('Content-Length'); + if(value===null)return null; + if(!/^(0|[1-9]\d*)$/.test(value))throw new Error('invalid response length'); + const parsed=Number(value); + if(!Number.isSafeInteger(parsed))throw new Error('invalid response length'); + return parsed; +} +async function read(response,limit,exact,signal){ + let declared; + try{declared=declaredLength(response)}catch(error){cancel(response);throw error} + if(declared!==null&&(declared>limit||(exact&&declared!==limit))){cancel(response);throw new Error('response body overflow')} + if(limit===0){cancel(response);return new ArrayBuffer(0)} + if(!response.body){if(exact||declared){throw new Error('missing response body')}return new ArrayBuffer(0)} + const reader=response.body.getReader(),chunks=[];let total=0,count=0,failed=false; + try{ + for(;;){ + if(signal&&signal.aborted)throw new Error('response read aborted'); + const part=await reader.read();if(part.done)break; + if(!(part.value instanceof Uint8Array))throw new Error('invalid response chunk'); + count++;total+=part.value.byteLength; + if(count>maxChunks||total>limit||(declared!==null&&total>declared))throw new Error('response body overflow'); + chunks.push(part.value); + } + if((exact&&total!==limit)||(declared!==null&&total!==declared))throw new Error('invalid response length'); + }catch(error){failed=true;try{const pending=reader.cancel();if(pending&&typeof pending.catch==='function')pending.catch(()=>{})}catch(cancelError){}throw error} + finally{try{reader.releaseLock()}catch(error){}if(failed&&signal&&signal.aborted)cancel(response)} + const joined=new Uint8Array(total);let offset=0; + for(const chunk of chunks){joined.set(chunk,offset);offset+=chunk.byteLength} + return joined.buffer; +} +globalThis.TelemtBridgeResponse=Object.freeze({cancel,read}); +})(); diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js index f684017..903fe7c 100644 --- a/src/web/bridge/runtime.js +++ b/src/web/bridge/runtime.js @@ -1,7 +1,7 @@ -(()=>{ -'use strict'; +(()=>{'use strict'; const bootstrap="__BOOTSTRAP__"; const relayOrigin='https://__HOST__',carrierCapabilities='https,https-lanes,websocket,websocket-lanes'; +const responseBody=globalThis.TelemtBridgeResponse;if(!responseBody)throw new Error('missing response runtime'); const negotiationEnabled=__NEGOTIATION_ENABLED__,candidateCount=__CANDIDATE_COUNT__,candidateDeadlines=[__CARRIER_DEADLINES__]; const longPollMs=__LONG_POLL_SECS__*1000,bridgeRequestMs=__BRIDGE_REQUEST_SECS__*1000,bridgeRetryMs=__BRIDGE_RETRY_SECS__*1000; const probeCoalesceMs=__CARRIER_PROBE_COALESCE_MS__; @@ -12,10 +12,19 @@ const fragment=location.hash,androidNonce=/^#android=([A-Za-z0-9_-]{43})$/.exec( history.replaceState(null,'',location.pathname); let initialized=false,closed=false,port=null,sessionToken='',cleanupToken='',createStarted=false,socket=null,socketReady=false,carrier=''; let queuedBytes=0,queuedItems=0,upSequence=1,downCursor='0',upRunning=false,upLease=null,pollController=null; -let helloFrame=null,welcomeSent=false,carrierAttempt=1,carrierFailure='',carrierCommitted=false; +let helloFrame=null,helloTimer=null,welcomeSent=false,carrierAttempt=1,carrierFailure='',carrierCommitted=false,terminalFailure=''; let negotiationStartedAt=0,carrierTimer=null,probeTimer=null,attemptController=null,attemptEpoch=1,candidateRunning=false,switching=false,currentAttempt=null; const pending=[],upPending=[],lanes=new Map(),closedLanes=new Set(),closedLaneOrder=[]; -const status=state=>{if(port&&!closed)port.postMessage({t:'status',state})}; +const canonicalFailures=['timeout','network','upgrade','http','protocol']; +const failure=(reason,message)=>Object.assign(new Error(message||reason),{telemtReason:reason}); +const failureReason=(error,fallback)=>error&&canonicalFailures.includes(error.telemtReason)?error.telemtReason:fallback; +const status=(state,phase,reason,deadlineMs)=>{ + if(!port||closed)return; + const currentPhase=phase||(state==='connected'?'committed':state==='reconnecting'?'retrying':state==='failed'?'terminal':createStarted?'negotiating':'starting'); + let currentDeadline=deadlineMs; + if(currentDeadline===undefined)currentDeadline=negotiationStartedAt?Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()):0; + port.postMessage({t:'status',state,phase:currentPhase,reason:reason||'',deadline_ms:Math.max(0,Math.ceil(currentDeadline))}); +}; const pause=(milliseconds,signal)=>new Promise((resolve,reject)=>{ if(signal&&signal.aborted){reject(new Error('request aborted'));return} const timer=setTimeout(done,milliseconds);function done(){if(signal)signal.removeEventListener('abort',abort);resolve()} @@ -152,8 +161,13 @@ function retryAfterMs(response){ return 0; } function retryableStatus(status){return status===408||status===429||status===502||status===503||status===504} +function responsePolicy(path,status){ + if(path==='/api/v1/session'&&status===200)return {limit:8,exact:true}; + if(path==='/api/v1/down'&&status===200)return {limit:batchLimit,exact:false}; + return {limit:0,exact:true}; +} async function request(path,frozenOptions){ - let delay=250,attempt=0;const deadline=Date.now()+bridgeRetryMs,external=frozenOptions.signal; + let delay=250,attempt=0,lastReason='network';const deadline=Date.now()+bridgeRetryMs,external=frozenOptions.signal; const attemptLimit=path==='/api/v1/down'?longPollMs+bridgeRequestMs:bridgeRequestMs; while(attempt<9){ if(closed||(external&&external.aborted))throw new Error('request aborted'); @@ -164,21 +178,31 @@ async function request(path,frozenOptions){ const timer=setTimeout(abort,Math.max(1,Math.min(attemptLimit,remaining))); let response=null,wait=0; try{ - const fetched=await fetch(relayOrigin+path,requestOptions),body=await fetched.arrayBuffer(); - response={status:fetched.status,headers:fetched.headers,body}; - if(!retryableStatus(response.status))return response; - wait=retryAfterMs(response); + const fetched=await fetch(relayOrigin+path,requestOptions); + if(retryableStatus(fetched.status)){ + lastReason='http';wait=retryAfterMs(fetched);responseBody.cancel(fetched); + }else{ + const policy=responsePolicy(path,fetched.status);let body; + try{body=await responseBody.read(fetched,policy.limit,policy.exact,controller.signal)} + catch(error){controller.abort();throw failure('protocol',error&&error.message)} + response={status:fetched.status,headers:fetched.headers,body};return response; + } }catch(error){ + controller.abort(); if(closed||(external&&external.aborted))throw error; + if(failureReason(error,'')==='protocol')throw error; }finally{clearTimeout(timer);if(external)external.removeEventListener('abort',abort)} const after=deadline-Date.now();if(attempt>=9||after<=0)break; status('reconnecting'); const backoff=wait||delay+Math.floor(Math.random()*Math.max(1,delay/4)); await pause(Math.min(backoff,after),external);delay=Math.min(delay*2,5000); } - throw new Error('carrier retry limit reached'); + throw failure(lastReason,'carrier retry limit reached'); +} +function fail(reason){ + if(closed)return;reason=reason||'protocol';if(canonicalFailures.includes(reason))terminalFailure=reason; + status('failed','terminal',reason,0);if(port)port.postMessage({t:'close'});close(true); } -function fail(){if(closed)return;status('failed');if(port)port.postMessage({t:'close'});close(true)} function knownCarrier(value){return value==='https'||value==='https-lanes'||value==='websocket'||value==='websocket-lanes'} function sessionEcho(response,expectedAttempt,states,exactAttempt){ const selected=response.headers.get('X-Carrier-Mode')||'',echo=response.headers.get('X-Carrier-Attempt')||''; @@ -216,39 +240,39 @@ function resetCandidate(){ function advanceConfirmed(reason,epoch){ if(closed||carrierCommitted||epoch!==attemptEpoch)return; resetCandidate(); - if(carrierAttempt>=negotiatedCandidateCount||Date.now()>=negotiationStartedAt+negotiatedFinalDeadline*1000){switching=false;fail();return} + if(carrierAttempt>=negotiatedCandidateCount||Date.now()>=negotiationStartedAt+negotiatedFinalDeadline*1000){switching=false;fail(reason);return} carrierAttempt++;carrierFailure=reason;attemptEpoch++;const nextEpoch=attemptEpoch;switching=false; status('reconnecting');armCarrierDeadline(nextEpoch);createSession(nextEpoch); } function advanceCarrier(reason,epoch){ if(closed||carrierCommitted||epoch!==attemptEpoch||switching)return; - if(!negotiationEnabled){fail();return} + if(!negotiationEnabled){fail(reason);return} switching=true;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;clearProbeTimer(); const snapshot=currentAttempt;if(attemptController)attemptController.abort();attemptController=null; - if(!snapshot||snapshot.epoch!==epoch){switching=false;fail();return} + if(!snapshot||snapshot.epoch!==epoch){switching=false;fail('protocol');return} if(snapshot.selected){advanceConfirmed(reason,epoch);return} resolveAttempt(reason,epoch,snapshot); } async function resolveAttempt(reason,epoch,snapshot){ const controller=new AbortController();attemptController=controller; const remaining=negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now(); - if(remaining<=0){switching=false;fail();return} + if(remaining<=0){switching=false;fail('timeout');return} const timer=setTimeout(()=>controller.abort(),remaining); try{ const frozen=options('POST',bootstrap,snapshot.hello,attemptHeaders(snapshot.attempt,snapshot.failure),controller.signal); const response=await request('/api/v1/session',frozen); if(closed||epoch!==attemptEpoch)return - if(response.status===409){sessionEcho(response,snapshot.attempt,['committed','healthy'],false);switching=false;fail();return} - if(response.status!==200){switching=false;fail();return} + if(response.status===409){sessionEcho(response,snapshot.attempt,['committed','healthy'],false);switching=false;fail('protocol');return} + if(response.status!==200){switching=false;fail('http');return} const echo=sessionEcho(response,snapshot.attempt,['provisional','committed','healthy'],true); const token=response.headers.get('X-Session-Token')||'',cursor=response.headers.get('X-Down-Cursor')||''; if(!token||cursor!=='0'||(snapshot.selected&&echo.selected!==snapshot.selected))throw new Error('changed carrier replay'); const welcome=response.body;if(closed||epoch!==attemptEpoch)return; cleanupToken=token; - if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome])} - if(echo.state!=='provisional'){switching=false;fail();return} + if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting','provisional','',Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()))} + if(echo.state!=='provisional'){switching=false;fail('protocol');return} advanceConfirmed(reason,epoch); - }catch(error){if(!closed&&epoch===attemptEpoch){switching=false;fail()}} + }catch(error){if(!closed&&epoch===attemptEpoch){switching=false;fail(failureReason(error,'protocol'))}} finally{clearTimeout(timer);if(attemptController===controller)attemptController=null} } function startCandidate(probe,epoch){ @@ -262,10 +286,10 @@ function startCandidate(probe,epoch){ } function maybeStartCandidate(){ if(closed||carrierCommitted||!sessionToken||candidateRunning)return;const epoch=attemptEpoch; - let probe;try{probe=findProbe(probeCoalesceMs>0)}catch(error){fail();return}if(!probe)return; + let probe;try{probe=findProbe(probeCoalesceMs>0)}catch(error){fail('protocol');return}if(!probe)return; if(!probeCoalesceMs||probe.hasData){startCandidate(probe,epoch);return} if(probeTimer)return;const owner={epoch,timer:null}; - owner.timer=setTimeout(()=>{if(probeTimer!==owner||closed||owner.epoch!==attemptEpoch)return;probeTimer=null;let current;try{current=findProbe(false)}catch(error){fail();return}startCandidate(current,owner.epoch)},probeCoalesceMs); + owner.timer=setTimeout(()=>{if(probeTimer!==owner||closed||owner.epoch!==attemptEpoch)return;probeTimer=null;let current;try{current=findProbe(false)}catch(error){fail('protocol');return}startCandidate(current,owner.epoch)},probeCoalesceMs); probeTimer=owner; } async function createSession(epoch){ @@ -276,17 +300,17 @@ async function createSession(epoch){ const frozen=options('POST',bootstrap,snapshot.hello,attemptHeaders(attempt,failure),controller.signal); const response=await request('/api/v1/session',frozen); if(closed||epoch!==attemptEpoch)return - if(response.status===409){sessionEcho(response,attempt,['committed','healthy'],false);fail();return} + if(response.status===409){sessionEcho(response,attempt,['committed','healthy'],false);fail('protocol');return} if(response.status!==200){advanceCarrier('http',epoch);return} const echo=sessionEcho(response,attempt,['provisional'],true),selected=echo.selected;snapshot.selected=selected; const token=response.headers.get('X-Session-Token')||'',cursor=response.headers.get('X-Down-Cursor')||''; if(!token||cursor!=='0'){advanceCarrier('protocol',epoch);return} const welcome=response.body;if(closed||epoch!==attemptEpoch)return; carrier=selected;sessionToken=token;cleanupToken=token;downCursor=cursor; - if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome])} + if(!welcomeSent){welcomeSent=true;port.postMessage(welcome,[welcome]);status('connecting','provisional','',Math.max(0,negotiationStartedAt+negotiatedFinalDeadline*1000-Date.now()))} if(carrier==='websocket')openCandidateSocket(null,null,epoch); maybeStartCandidate(); - }catch(error){if(closed||epoch!==attemptEpoch)return;advanceCarrier('network',epoch)} + }catch(error){if(closed||epoch!==attemptEpoch)return;advanceCarrier(failureReason(error,'network'),epoch)} } async function probeHttp(probe,laneID,epoch){ try{ @@ -297,15 +321,15 @@ async function probeHttp(probe,laneID,epoch){ if(response.headers.get('X-Up-Ack')!=='1'){advanceCarrier('protocol',epoch);return} if(laneID===null)upSequence=2;else ensureLane(laneID).sequence=2; commitCarrier(probe,epoch); - }catch(error){if(!closed&&epoch===attemptEpoch)advanceCarrier('network',epoch)} + }catch(error){if(!closed&&epoch===attemptEpoch)advanceCarrier(failureReason(error,'network'),epoch)} } function commitCarrier(probe,epoch){ if(closed||carrierCommitted||epoch!==attemptEpoch)return; - if(switching){fail();return} - clearProbeTimer();try{consumeProbe(probe)}catch(error){fail();return} + if(switching){fail('protocol');return} + clearProbeTimer();try{consumeProbe(probe)}catch(error){fail('protocol');return} carrierCommitted=true;candidateRunning=false;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null; attemptController=null;currentAttempt=null; - status('connected'); + status('connected','committed','',0); if(carrier==='https')poll(); else if(carrier==='https-lanes'){const lane=lanes.get(probe.id);if(lane&&!lane.polling)pollLane(lane)} for(const data of pending.splice(0)){release(data.byteLength,1,null);queueCarrier(data)} @@ -315,19 +339,20 @@ function queueCarrier(data){ if(carrier==='https')queueUp(data); else if(carrier==='websocket')queueSocket(data); else for(const value of splitFrames(data))queueLane(value); - }catch(error){fail()} + }catch(error){fail('protocol')} } -function queueUp(data){if(!reserve(data,null)){fail();return}upPending.push(data);runUp()} +function queueUp(data){if(!reserve(data,null)){fail('capacity');return}upPending.push(data);runUp()} async function runUp(){ if(upRunning)return;upRunning=true;let lease=null; try{ while(!closed&&sessionToken&&upPending.length){ lease=takeBatch(upPending,null);upLease=lease;lease.controller=new AbortController();const sequence=String(upSequence); const response=await request('/api/v1/up',options('POST',sessionToken,lease.body,{'X-Up-Seq':sequence},lease.controller.signal)); - if(response.status!==204||response.headers.get('X-Up-Ack')!==sequence)throw new Error('uplink rejected'); + if(response.status!==204)throw failure('http','uplink rejected'); + if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','uplink acknowledgement rejected'); if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});upSequence++;lease=null; } - }catch(error){if(!closed&&!(lease&&lease.cancelled))fail()} + }catch(error){if(!closed&&!(lease&&lease.cancelled))fail(failureReason(error,'network'))} finally{upRunning=false;if(!closed&&sessionToken&&upPending.length)runUp()} } function sendCandidateSocket(next){ @@ -352,17 +377,17 @@ function openCandidateSocket(probe,laneID,epoch){ try{ if(state.lane){const values=splitFrames(event.data);for(const value of values)if(value.id!==state.lane.id)throw new Error('cross-lane frame');if(values.some(value=>value.type===3))state.lane.remoteClosed=true} else{const bound=frameBound(event.data,4096,batchLimit);if(bound.bytes!==event.data.byteLength)throw new Error('invalid frame batch')} - }catch(error){if(state.lane)finishLane(state.lane,true);else fail();return} + }catch(error){if(state.lane)finishLane(state.lane,true);else fail('protocol');return} port.postMessage({t:'traffic',up:0,down:event.data.byteLength});port.postMessage(event.data,[event.data]);status('connected'); }; next.onerror=()=>{}; next.onclose=()=>{ const state=next.telemt;if(state.epoch!==attemptEpoch||closed)return; if(!carrierCommitted){advanceCarrier(state.opened?'network':'upgrade',state.epoch);return} - if(state.lane){state.lane.ready=false;state.lane.socket=null;finishLane(state.lane,true)}else{socketReady=false;fail()} + if(state.lane){state.lane.ready=false;state.lane.socket=null;finishLane(state.lane,true)}else{socketReady=false;fail('network')} }; } -function queueSocket(data){if(!reserve(data,null)){fail();return}upPending.push(data);runSocketUp()} +function queueSocket(data){if(!reserve(data,null)){fail('capacity');return}upPending.push(data);runSocketUp()} async function waitSocket(next,size,limit,signal){ while(!closed&&next.readyState===WebSocket.OPEN&&next.bufferedAmount>limit-size)await pause(10,signal); if(closed||(signal&&signal.aborted)||next.readyState!==WebSocket.OPEN)throw new Error('websocket closed'); @@ -375,7 +400,7 @@ async function runSocketUp(){ await waitSocket(socket,lease.total,queueLimit,lease.controller.signal);socket.send(lease.body); if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});lease=null; } - }catch(error){if(!closed&&!(lease&&lease.cancelled))fail()} + }catch(error){if(!closed&&!(lease&&lease.cancelled))fail(failureReason(error,'network'))} finally{upRunning=false;if(!closed&&socketReady&&upPending.length)runSocketUp()} } async function poll(){ @@ -384,12 +409,12 @@ async function poll(){ pollController=new AbortController(); const response=await request('/api/v1/down',options('POST',sessionToken,null,{'X-Down-Cursor':downCursor},pollController.signal)); if(response.status===204){status('connected');continue} - if(response.status!==200)throw new Error('downlink rejected'); + if(response.status!==200)throw failure('http','downlink rejected'); const next=response.headers.get('X-Down-Cursor')||'',data=response.body; - if(!next||!data.byteLength)throw new Error('invalid downlink response'); + if(!next||!data.byteLength)throw failure('protocol','invalid downlink response'); if(closed)return; port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);downCursor=next;status('connected'); - }catch(error){if(!closed)fail();return} + }catch(error){if(!closed)fail(failureReason(error,'network'));return} } } function ensureLane(id){ @@ -416,7 +441,7 @@ function queueLane(value){ if(!lane&&closedLanes.has(value.id))throw new Error('closed lane was reused'); if(!lane&&value.type!==1)throw new Error('lane did not begin with OPEN'); lane=lane||ensureLane(value.id); - if(!reserve(value.data,lane)){fail();return} + if(!reserve(value.data,lane)){fail('capacity');return} lane.pending.push(value.data); if(carrier==='websocket-lanes'){openLaneSocket(lane);runLaneSocketUp(lane)}else runLaneUp(lane); } @@ -449,11 +474,12 @@ async function runLaneUp(lane){ lease=takeBatch(lane.pending,lane);lane.upLease=lease;lease.controller=new AbortController(); const sequence=String(lane.sequence),laneID=String(lane.id); const response=await request('/api/v1/up',options('POST',sessionToken,lease.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID},lease.controller.signal)); - if(response.status!==204||response.headers.get('X-Up-Ack')!==sequence)throw new Error('lane uplink rejected'); + if(response.status!==204)throw failure('http','lane uplink rejected'); + if(response.headers.get('X-Up-Ack')!==sequence)throw failure('protocol','lane uplink acknowledgement rejected'); if(!settleBatch(lease))return;port.postMessage({t:'traffic',up:lease.total,down:0});lane.sequence++;lease=null; if(!lane.polling)pollLane(lane); } - }catch(error){if(!closed&&lanes.get(lane.id)===lane&&!(lease&&lease.cancelled))fail()} + }catch(error){if(!closed&&lanes.get(lane.id)===lane&&!(lease&&lease.cancelled))fail(failureReason(error,'network'))} finally{lane.running=false;if(!closed&&lanes.get(lane.id)===lane&&sessionToken&&lane.pending.length)runLaneUp(lane)} } async function pollLane(lane){ @@ -466,21 +492,22 @@ async function pollLane(lane){ if(response.headers.get('X-Lane-Closed')==='1'){finishLane(lane,false);return} status('connected');continue; } - if(response.status!==200)throw new Error('lane downlink rejected'); + if(response.status!==200)throw failure('http','lane downlink rejected'); const next=response.headers.get('X-Down-Cursor')||'',data=response.body; - if(!next||!data.byteLength)throw new Error('invalid lane downlink response'); + if(!next||!data.byteLength)throw failure('protocol','invalid lane downlink response'); for(const value of splitFrames(data))if(value.id!==lane.id)throw new Error('cross-lane frame'); if(closed)return; port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);lane.cursor=next;status('connected'); } - }catch(error){if(!closed)fail()} + }catch(error){if(!closed)fail(failureReason(error,'network'))} finally{lane.polling=false;lane.controller=null} } function deleteSession(){ - const token=cleanupToken||sessionToken;if(token)fetch(relayOrigin+'/api/v1/session',options('DELETE',token,null,null,undefined,true)).catch(()=>{}); + const token=cleanupToken||sessionToken,headers=canonicalFailures.includes(terminalFailure)?{'X-Carrier-Failure':terminalFailure}:null; + if(token)fetch(relayOrigin+'/api/v1/session',options('DELETE',token,null,headers,undefined,true)).catch(()=>{}); } function close(notifyServer){ - if(closed)return;closed=true;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(attemptController)attemptController.abort();if(pollController)pollController.abort(); + if(closed)return;closed=true;if(helloTimer)clearTimeout(helloTimer);helloTimer=null;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(attemptController)attemptController.abort();if(pollController)pollController.abort(); if(socket)socket.close();cancelBatch(upLease);releasePending(upPending,null); for(const lane of lanes.values()){ if(lane.controller)lane.controller.abort();cancelBatch(lane.upLease);releasePending(lane.pending,lane);if(lane.socket)lane.socket.close(); @@ -492,15 +519,16 @@ function activatePort(nextPort){ initialized=true;port=nextPort; port.onmessage=message=>{ if(message.data instanceof ArrayBuffer){ - if(!createStarted){createStarted=true;helloFrame=message.data;if(negotiationEnabled){negotiationStartedAt=Date.now();armCarrierDeadline(attemptEpoch)}createSession(attemptEpoch)} - else if(!carrierCommitted){if(!reserve(message.data,null)){fail();return}pending.push(message.data);maybeStartCandidate()} + if(!createStarted){createStarted=true;if(helloTimer)clearTimeout(helloTimer);helloTimer=null;helloFrame=message.data;if(negotiationEnabled){negotiationStartedAt=Date.now();armCarrierDeadline(attemptEpoch)}createSession(attemptEpoch)} + else if(!carrierCommitted){if(!reserve(message.data,null)){fail('capacity');return}pending.push(message.data);maybeStartCandidate()} else queueCarrier(message.data); - }else if(message.data&&message.data.t==='close')close(true); + }else if(message.data&&message.data.t==='close'){status('failed','terminal','closed',0);close(true)} }; - port.start();status('connecting'); + port.start();status('connecting','starting','',bridgeRequestMs);helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs); } addEventListener('message',event=>{ - if(initialized||event.source!==parent||event.data===null||typeof event.data!=='object')return; + if(event.source!==parent)return;if(initialized){if(event.ports&&event.ports.length===1)event.ports[0].close();return} + if(event.data===null||typeof event.data!=='object')return; const keys=Object.keys(event.data).sort(); if(keys.length!==2||keys[0]!=='t'||keys[1]!=='v'||event.data.t!=='tproxy-init'||event.data.v!==1||event.ports.length!==1)return; let source;try{source=new URL(event.origin)}catch(error){return} @@ -511,12 +539,12 @@ const androidBridge=globalThis.TelegramWebProxy; if(!initialized&&androidNonce&&androidBridge&&typeof androidBridge.postMessage==='function'){ const androidPort={onmessage:null,start(){},close(){androidBridge.onmessage=null},postMessage(value){ if(value instanceof ArrayBuffer){ - let frames;try{frames=splitFrames(value)}catch(error){fail();return} + let frames;try{frames=splitFrames(value)}catch(error){fail('protocol');return} for(const frame of frames)androidBridge.postMessage(frame.data); }else androidBridge.postMessage(JSON.stringify(value)); }}; androidBridge.onmessage=event=>{let data=event.data;if(typeof data==='string'){try{data=JSON.parse(data)}catch(error){return}}if(androidPort.onmessage)androidPort.onmessage({data})}; activatePort(androidPort);androidBridge.postMessage(JSON.stringify({t:'tproxy-android-init',v:1,nonce:androidNonce})); } -addEventListener('pagehide',()=>close(true),{once:true}); +addEventListener('pagehide',()=>fail('navigation'),{once:true}); })(); diff --git a/src/web/bridge/tests.rs b/src/web/bridge/tests.rs index ddab40f..6f157c9 100644 --- a/src/web/bridge/tests.rs +++ b/src/web/bridge/tests.rs @@ -31,6 +31,10 @@ fn rendered_page_contains_bounded_negotiation_contract() { assert!(page.body.contains("X-Lane-ID")); assert!(page.body.contains("tproxy-auto-v1.")); assert!(page.body.contains("tproxy-auto-lane-v1.")); + assert!(page.body.contains("globalThis.TelemtBridgeResponse")); + assert!(page.body.contains("responseBody.read")); + assert!(!page.body.contains("arrayBuffer()")); + assert!(page.body.contains("maxChunks=4096")); assert!( page.content_security_policy .contains("frame-ancestors http://127.0.0.1:*") @@ -69,6 +73,9 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() { assert!(page.body.contains("bridgeRequestMs=7*1000")); assert!(page.body.contains("bridgeRetryMs=41*1000")); assert!(page.body.contains("const probeCoalesceMs=4")); + assert!(page.body.contains( + "helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs)" + )); } #[test] @@ -140,7 +147,9 @@ fn ambiguous_commit_is_resolved_before_carrier_advance() { )); assert!( page.body - .contains("if(echo.state!=='provisional'){switching=false;fail();return}") + .contains("if(echo.state!=='provisional'){switching=false;fail('protocol');return}") ); assert!(page.body.contains("const token=cleanupToken||sessionToken")); + assert!(page.body.contains("'X-Carrier-Failure':terminalFailure")); + assert!(page.body.contains("addEventListener('pagehide',()=>fail('navigation')")); } diff --git a/src/web/http/carrier_diagnostic_tests.rs b/src/web/http/carrier_diagnostic_tests.rs new file mode 100644 index 0000000..54a389e --- /dev/null +++ b/src/web/http/carrier_diagnostic_tests.rs @@ -0,0 +1,148 @@ +use super::*; + +use sha2::{Digest, Sha256}; + +use crate::web::manager::CarrierFailure; +use crate::web::telemetry::WebCarrierFailurePhase; + +fn issue_bootstrap(runtime: &Arc, client_ip: &str) -> String { + let profile = runtime + .active_generation() + .config() + .web + .runtime + .as_ref() + .unwrap() + .profiles[0] + .clone(); + runtime + .issue_bootstrap(profile, client_ip.parse().unwrap()) + .unwrap() + .token +} + +fn token_hash(token: &str) -> crate::web::manager::TokenHash { + let raw = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(token) + .unwrap(); + Sha256::digest(raw).into() +} + +async fn create_automatic_session( + listener: &TcpListener, + runtime: &Arc, +) -> String { + let bootstrap = issue_bootstrap(runtime, "192.0.2.10"); + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let mut create = format!( + "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nX-Carrier-Capabilities: https\r\nX-Carrier-Attempt: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + create.extend_from_slice(&hello); + let response = request(listener, runtime, create).await; + let (headers, _) = split_response(&response); + assert!(headers.starts_with(b"HTTP/1.1 200")); + response_header(headers, "x-session-token").to_string() +} + +fn delete_request(token: &str, failure_headers: &str) -> Vec { + format!( + "DELETE /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {token}\r\n{failure_headers}Content-Length: 0\r\nConnection: close\r\n\r\n" + ) + .into_bytes() +} + +fn diagnostic_runtime(capability: [u8; 32]) -> ProxyConfig { + negotiation_runtime_config( + capability, + WebCarrier::Https, + false, + Arc::from([WebCarrier::Https]), + ) +} + +#[tokio::test] +async fn delete_failure_is_strict_and_counted_only_for_the_winning_close() { + let generation = test_runtime_generation(1, diagnostic_runtime([61; 32])); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = create_automatic_session(&listener, &runtime).await; + let hash = token_hash(&token); + + let malformed = request( + &listener, + &runtime, + delete_request(&token, "X-Carrier-Failure: invalid\r\n"), + ) + .await; + assert!(!malformed.starts_with(b"HTTP/1.1 204")); + assert!(runtime.get_session(hash, "proxy.example.com").is_ok()); + + let close = delete_request(&token, "X-Carrier-Failure: network\r\n"); + let response = request(&listener, &runtime, close.clone()).await; + assert!(response.starts_with(b"HTTP/1.1 204")); + let retry = request(&listener, &runtime, close).await; + assert!(retry.starts_with(b"HTTP/1.1 204")); + assert_eq!( + runtime.telemetry().carrier_failure_total( + WebCarrier::Https, + WebCarrierFailurePhase::Provisional, + CarrierFailure::Network, + ), + 1 + ); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} + +#[tokio::test] +async fn delete_failure_uses_the_committed_phase_after_real_uplink_progress() { + let generation = test_runtime_generation(1, diagnostic_runtime([62; 32])); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let token = create_automatic_session(&listener, &runtime).await; + let hash = token_hash(&token); + let open = frame::encode(FrameType::Open, 7, &[]); + let data = frame::encode(FrameType::Data, 7, &[1]); + let mut body = Vec::with_capacity(open.len() + data.len()); + body.extend_from_slice(&open); + body.extend_from_slice(&data); + let mut uplink = format!( + "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {token}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .into_bytes(); + uplink.extend_from_slice(&body); + + let accepted = request(&listener, &runtime, uplink).await; + assert!(accepted.starts_with(b"HTTP/1.1 204")); + assert!( + runtime + .get_session(hash, "proxy.example.com") + .unwrap() + .is_carrier_committed() + ); + let response = request( + &listener, + &runtime, + delete_request(&token, "X-Carrier-Failure: protocol\r\n"), + ) + .await; + + assert!(response.starts_with(b"HTTP/1.1 204")); + assert_eq!( + runtime.telemetry().carrier_failure_total( + WebCarrier::Https, + WebCarrierFailurePhase::Committed, + CarrierFailure::Protocol, + ), + 1 + ); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} diff --git a/src/web/http/request.rs b/src/web/http/request.rs index a0a1897..c150096 100644 --- a/src/web/http/request.rs +++ b/src/web/http/request.rs @@ -223,7 +223,7 @@ fn optional_canonical_u8_header(request: &Request, name: &'static str) -> .map(Some) } -fn optional_failure_header(request: &Request) -> Option> { +pub(super) fn optional_failure_header(request: &Request) -> Option> { if !request.headers().contains_key("x-carrier-failure") { return Some(None); } diff --git a/src/web/http/session.rs b/src/web/http/session.rs index a054bae..89d1f21 100644 --- a/src/web/http/session.rs +++ b/src/web/http/session.rs @@ -7,7 +7,9 @@ use hyper::{Method, Request, StatusCode}; use super::body::{CollectBodyError, CollectedBody, RequestBody, collect_body}; use super::decoy::serve_decoy; -use super::request::{binary_content_type, carrier_ip_learning_eligible, carrier_request}; +use super::request::{ + binary_content_type, carrier_ip_learning_eligible, carrier_request, optional_failure_header, +}; use super::response::{ carrier_empty, carrier_headers, full_response, insert_header, service_unavailable, }; @@ -34,6 +36,9 @@ pub(super) async fn handle_session( if request.headers().contains_key(header::CONTENT_TYPE) { return serve_decoy(request, vhost, true, &runtime).await; } + let Some(carrier_failure) = optional_failure_header(&request) else { + return serve_decoy(request, vhost, true, &runtime).await; + }; let session = runtime.get_session(token_hash, &vhost.host).ok(); if let Some(trace) = request_trace(&request) && let Some(session) = &session @@ -56,7 +61,11 @@ pub(super) async fn handle_session( return serve_decoy(request, vhost, true, &runtime).await; } }; - if !body.is_empty() || runtime.close_token(token_hash, &vhost.host).is_err() { + if !body.is_empty() + || runtime + .close_token(token_hash, &vhost.host, carrier_failure) + .is_err() + { return serve_decoy(request, vhost, true, &runtime).await; } return carrier_empty(StatusCode::NO_CONTENT); diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index 01e0330..042231a 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -24,6 +24,9 @@ use crate::web::manager::{ mod legacy_tests; #[path = "negotiation_tests.rs"] mod negotiation_tests; +// Client failure diagnostics remain separate from negotiation state scenarios. +#[path = "carrier_diagnostic_tests.rs"] +mod carrier_diagnostic_tests; // Reload-stability tests for session-owned timeout policy. #[path = "session_policy_tests.rs"] mod session_policy_tests; diff --git a/src/web/manager.rs b/src/web/manager.rs index 584a0ef..674d443 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -207,12 +207,15 @@ impl WebProcessRuntime { let limits = config.web.limits.clone(); let learning_capacity = limits.max_carrier_learning_entries; let mut carrier_learning = learning::CarrierLearning::new(learning_capacity); - let _ = carrier_learning.apply_policy( + let learning_policy = carrier_learning.apply_policy( std::time::Instant::now(), + initial_generation.id, config.web.carrier_negotiation_enabled() && config.web.carrier_learning, config.web.carrier_negotiation_aggressiveness, Duration::from_secs(config.web.timeouts.carrier_learning_secs), + Duration::from_secs(config.web.timeouts.carrier_health_secs), ); + drop(learning_policy.detached); let websocket_connections = limits .max_http_connections .saturating_sub(limits.websocket_http_connection_reserve); diff --git a/src/web/manager/carrier_learning.rs b/src/web/manager/carrier_learning.rs index 233872f..ec63b79 100644 --- a/src/web/manager/carrier_learning.rs +++ b/src/web/manager/carrier_learning.rs @@ -8,6 +8,10 @@ use super::ProfileKey; use super::negotiation::{CarrierClientClass, CarrierLearningContext}; use crate::config::{WebCarrier, WebCarrierNegotiationAggressiveness}; +#[path = "carrier_learning/status.rs"] +mod status; +pub(super) use status::{CarrierLearningEpoch, CarrierLearningRecordOutcome}; + const PROFILE_WEIGHT: i16 = 32; const USER_AGENT_WEIGHT: i16 = 32; const IP_WEIGHT: i16 = 1; @@ -117,6 +121,23 @@ struct LearningPolicy { enabled: bool, aggressiveness: WebCarrierNegotiationAggressiveness, lifetime: Duration, + health_window: Duration, +} + +/// Evidence detached under the policy lock and released by its caller. +pub(super) struct DetachedCarrierEvidence { + entries: HashMap, + insertion_order: VecDeque<(EvidenceKey, u64)>, +} + +/// Result of one generation-fenced policy reconciliation. +pub(super) struct CarrierLearningPolicyOutcome { + /// Current epoch after the reconciliation. + pub(super) epoch: Option, + /// Whether this generation was current enough to apply. + pub(super) applied: bool, + /// Retired evidence whose allocation is released outside the policy lock. + pub(super) detached: Option, } #[derive(Clone, Copy)] @@ -160,37 +181,10 @@ pub(super) struct CarrierLearning { insertion_sequence: u64, epoch: Option, policy: Option, + applied_generation: Option, policy_started_at: Instant, } -/// Bounded control-plane summary of carrier-learning state. -#[derive(Clone, Copy)] -pub(crate) struct CarrierLearningStatus { - /// Whether outcome learning is active in the effective policy. - pub(crate) enabled: bool, - /// Effective evidence thresholds. - pub(crate) aggressiveness: WebCarrierNegotiationAggressiveness, - /// Current evidence epoch, or none after counter exhaustion. - pub(crate) epoch: Option, - /// Retained evidence entries. - pub(crate) entries: usize, - /// Restart-owned evidence ceiling. - pub(crate) capacity: usize, - /// Effective evidence lifetime. - pub(crate) lifetime_secs: u64, - /// Monotonic age of the current policy epoch. - pub(crate) age_ms: u64, -} - -/// Result of one epoch-fenced learning reset. -#[derive(Clone, Copy)] -pub(crate) struct CarrierLearningResetOutcome { - /// Evidence entries detached by the reset. - pub(crate) entries_cleared: usize, - /// New epoch fencing pre-reset outcomes. - pub(crate) epoch: u64, -} - impl CarrierLearning { /// Creates an empty store under the restart-owned capacity ceiling. pub(super) fn new(capacity: usize) -> Self { @@ -201,70 +195,80 @@ impl CarrierLearning { insertion_sequence: 1, epoch: Some(0), policy: None, + applied_generation: None, policy_started_at: Instant::now(), } } - fn status(&self, now: Instant) -> CarrierLearningStatus { - let policy = self.policy.unwrap_or(LearningPolicy { - enabled: false, - aggressiveness: WebCarrierNegotiationAggressiveness::Conservative, - lifetime: Duration::ZERO, - }); - CarrierLearningStatus { - enabled: policy.enabled, - aggressiveness: policy.aggressiveness, - epoch: self.epoch, - entries: self.entries.len(), - capacity: self.capacity, - lifetime_secs: policy.lifetime.as_secs(), - age_ms: millis(now.saturating_duration_since(self.policy_started_at)), - } - } - - /// Applies hot-reloaded learning policy and returns its outcome epoch. + /// Applies one semantic policy unless a newer generation already owns the store. pub(super) fn apply_policy( &mut self, now: Instant, + generation: u64, enabled: bool, aggressiveness: WebCarrierNegotiationAggressiveness, lifetime: Duration, - ) -> Option { + health_window: Duration, + ) -> CarrierLearningPolicyOutcome { + if self + .applied_generation + .is_some_and(|applied| generation < applied) + { + return CarrierLearningPolicyOutcome { + epoch: self.epoch, + applied: false, + detached: None, + }; + } let policy = LearningPolicy { enabled, aggressiveness, lifetime, + health_window, }; + self.applied_generation = Some(generation); + let mut detached = None; if self.policy != Some(policy) { - self.entries.clear(); - self.insertion_order.clear(); - if !enabled { - self.entries.shrink_to_fit(); - self.insertion_order.shrink_to_fit(); - } + detached = Some(DetachedCarrierEvidence { + entries: std::mem::take(&mut self.entries), + insertion_order: std::mem::take(&mut self.insertion_order), + }); self.insertion_sequence = 1; self.epoch = self.epoch.and_then(|epoch| epoch.checked_add(1)); self.policy = Some(policy); self.policy_started_at = now; } - self.epoch + CarrierLearningPolicyOutcome { + epoch: self.epoch, + applied: true, + detached, + } } - /// Returns the current epoch only when the request snapshot matches owner policy. + /// Matches a request to the exact applied generation and semantic policy. pub(super) fn epoch_for_policy( &self, + generation: u64, enabled: bool, aggressiveness: WebCarrierNegotiationAggressiveness, lifetime: Duration, - ) -> Option { - (self.policy - == Some(LearningPolicy { - enabled, - aggressiveness, - lifetime, - })) - .then_some(self.epoch) - .flatten() + health_window: Duration, + ) -> CarrierLearningEpoch { + if self.applied_generation != Some(generation) + || self.policy + != Some(LearningPolicy { + enabled, + aggressiveness, + lifetime, + health_window, + }) + { + return CarrierLearningEpoch::Pending; + } + self.epoch.map_or( + CarrierLearningEpoch::Exhausted, + CarrierLearningEpoch::Ready, + ) } /// Ranks supported configured candidates without scanning the evidence store. @@ -348,12 +352,12 @@ impl CarrierLearning { context: CarrierLearningContext, failures: &[WebCarrier], winner: WebCarrier, - ) { + ) -> CarrierLearningRecordOutcome { let Some(policy) = self.policy.filter(|policy| policy.enabled) else { - return; + return CarrierLearningRecordOutcome::PolicyDisabled; }; if Some(epoch) != self.epoch { - return; + return CarrierLearningRecordOutcome::StaleEpoch; } let mut deltas = [0i8; 4]; let _ = failures; @@ -369,21 +373,39 @@ impl CarrierLearning { (context.ip_learning_eligible && thresholds.ip.is_some()) .then_some(EvidenceKey::Ip(context.profile_key, context.client_ip)), ]; - self.make_room(&keys); let missing = keys .iter() .flatten() .filter(|key| !self.entries.contains_key(key)) .count(); - if self.entries.len().saturating_add(missing) > self.capacity { - return; + if self + .insertion_sequence + .checked_add(missing as u64) + .is_none() + { + return CarrierLearningRecordOutcome::SequenceExhausted; } + let evictions_needed = self + .entries + .len() + .saturating_add(missing) + .saturating_sub(self.capacity); + let evictable = self + .entries + .keys() + .filter(|key| !keys.contains(&Some(**key))) + .count(); + if evictions_needed > evictable { + return CarrierLearningRecordOutcome::CapacityRejected; + } + self.make_room(&keys); let slot = bucket_slot(self.policy_started_at, now, policy.lifetime); let cohort = cohort_hash(context); for (index, key) in keys.into_iter().enumerate() { let Some(key) = key else { continue }; self.update_key(key, slot, deltas, (index == 0).then_some(cohort)); } + CarrierLearningRecordOutcome::Recorded } /// Reclaims a fixed number of entries outside both half-window buckets. @@ -452,9 +474,8 @@ impl CarrierLearning { entry.update(slot, deltas, cohort); return; } - let Some(insertion_sequence) = self.next_insertion_sequence() else { - return; - }; + let insertion_sequence = self.insertion_sequence; + self.insertion_sequence += 1; self.entries.insert(key, Evidence::new(insertion_sequence)); self.insertion_order.push_back((key, insertion_sequence)); if let Some(entry) = self.entries.get_mut(&key) { @@ -462,56 +483,6 @@ impl CarrierLearning { } } - fn next_insertion_sequence(&mut self) -> Option { - let sequence = self.insertion_sequence; - self.insertion_sequence = sequence.checked_add(1)?; - Some(sequence) - } -} - -impl super::WebProcessRuntime { - /// Captures learning state without waiting for a contended evidence lock. - pub(crate) fn try_carrier_learning_status(&self) -> Option { - self.learning - .try_lock() - .map(|learning| learning.status(Instant::now())) - } - - /// Clears all evidence under a new epoch without changing the active policy. - pub(crate) fn reset_carrier_learning( - &self, - ) -> Result { - let control = self - .control_mutation_guard() - .map_err(|_| super::ManagerError::Closed)?; - let (outcome, retired_entries, retired_order) = { - let mut learning = self.learning.lock(); - let epoch = learning - .epoch - .and_then(|epoch| epoch.checked_add(1)) - .ok_or(super::ManagerError::Closed)?; - learning.epoch = Some(epoch); - learning.insertion_sequence = 1; - learning.policy_started_at = Instant::now(); - let retired_entries = std::mem::take(&mut learning.entries); - let retired_order = std::mem::take(&mut learning.insertion_order); - ( - CarrierLearningResetOutcome { - entries_cleared: retired_entries.len(), - epoch, - }, - retired_entries, - retired_order, - ) - }; - drop(control); - drop((retired_entries, retired_order)); - Ok(outcome) - } -} - -fn millis(duration: Duration) -> u64 { - duration.as_millis().min(u128::from(u64::MAX)) as u64 } fn supported(configured: &[WebCarrier], request: super::CarrierRequest) -> Vec { diff --git a/src/web/manager/carrier_learning/status.rs b/src/web/manager/carrier_learning/status.rs new file mode 100644 index 0000000..879453d --- /dev/null +++ b/src/web/manager/carrier_learning/status.rs @@ -0,0 +1,140 @@ +use std::time::{Duration, Instant}; + +use crate::config::WebCarrierNegotiationAggressiveness; + +use super::{CarrierLearning, DetachedCarrierEvidence, LearningPolicy}; + +/// Exact result of matching one request snapshot to the applied learning policy. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(in crate::web::manager) enum CarrierLearningEpoch { + /// The request may use the returned evidence epoch. + Ready(u64), + /// The process store has not applied this exact generation and policy yet. + Pending, + /// The exact policy is active but its monotonic epoch space is exhausted. + Exhausted, +} + +/// Terminal result of applying one complete attempt chain. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(in crate::web::manager) enum CarrierLearningRecordOutcome { + /// Every eligible evidence key received the complete sample. + Recorded, + /// The effective learning policy does not accept evidence. + PolicyDisabled, + /// The result belongs to a superseded or exhausted epoch. + StaleEpoch, + /// The bounded store cannot fit all evidence keys atomically. + CapacityRejected, + /// A unique insertion sequence cannot be assigned to every new key. + SequenceExhausted, +} + +/// Bounded control-plane summary of carrier-learning state. +#[derive(Clone, Copy)] +pub(crate) struct CarrierLearningStatus { + /// Whether outcome learning is active in the effective policy. + pub(crate) enabled: bool, + /// Effective evidence thresholds. + pub(crate) aggressiveness: WebCarrierNegotiationAggressiveness, + /// Generation whose policy is applied to the process-owned store. + pub(crate) policy_generation: Option, + /// Current evidence epoch, or none after counter exhaustion. + pub(crate) epoch: Option, + /// Retained evidence entries. + pub(crate) entries: usize, + /// Restart-owned evidence ceiling. + pub(crate) capacity: usize, + /// Effective evidence lifetime. + pub(crate) lifetime_secs: u64, + /// Effective post-commit health observation window. + pub(crate) health_secs: u64, + /// Monotonic age of the current semantic policy epoch. + pub(crate) age_ms: u64, +} + +/// Result of one epoch-fenced learning reset. +#[derive(Clone, Copy)] +pub(crate) struct CarrierLearningResetOutcome { + /// Evidence entries detached by the reset. + pub(crate) entries_cleared: usize, + /// New epoch fencing pre-reset outcomes. + pub(crate) epoch: u64, +} + +impl CarrierLearning { + pub(super) fn status(&self, now: Instant) -> CarrierLearningStatus { + let policy = self.policy.unwrap_or(LearningPolicy { + enabled: false, + aggressiveness: WebCarrierNegotiationAggressiveness::Conservative, + lifetime: Duration::ZERO, + health_window: Duration::ZERO, + }); + CarrierLearningStatus { + enabled: policy.enabled, + aggressiveness: policy.aggressiveness, + policy_generation: self.applied_generation, + epoch: self.epoch, + entries: self.entries.len(), + capacity: self.capacity, + lifetime_secs: policy.lifetime.as_secs(), + health_secs: policy.health_window.as_secs(), + age_ms: millis(now.saturating_duration_since(self.policy_started_at)), + } + } +} + +impl super::super::WebProcessRuntime { + /// Captures learning state without waiting for a contended evidence lock. + pub(crate) fn try_carrier_learning_status(&self) -> Option { + self.learning + .try_lock() + .map(|learning| learning.status(Instant::now())) + } + + /// Clears all evidence under a new epoch without changing the active policy. + pub(crate) fn reset_carrier_learning( + &self, + ) -> Result { + let control = self + .control_mutation_guard() + .map_err(|_| super::super::ManagerError::Closed)?; + let (outcome, detached) = { + let mut learning = self.learning.lock(); + let epoch = learning + .epoch + .and_then(|epoch| epoch.checked_add(1)) + .ok_or(super::super::ManagerError::Closed)?; + learning.epoch = Some(epoch); + learning.insertion_sequence = 1; + learning.policy_started_at = Instant::now(); + let entries = std::mem::take(&mut learning.entries); + let entries_cleared = entries.len(); + let detached = DetachedCarrierEvidence { + entries, + insertion_order: std::mem::take(&mut learning.insertion_order), + }; + ( + CarrierLearningResetOutcome { + entries_cleared, + epoch, + }, + detached, + ) + }; + drop(control); + drop(detached); + Ok(outcome) + } +} + +impl Drop for DetachedCarrierEvidence { + fn drop(&mut self) { + self.entries.clear(); + self.insertion_order.clear(); + } +} + +fn millis(duration: Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 +} diff --git a/src/web/manager/carrier_learning/tests.rs b/src/web/manager/carrier_learning/tests.rs index 8f5f94f..375ec2d 100644 --- a/src/web/manager/carrier_learning/tests.rs +++ b/src/web/manager/carrier_learning/tests.rs @@ -22,28 +22,45 @@ fn context(hash: u8) -> CarrierLearningContext { } } +fn apply_epoch( + learning: &mut CarrierLearning, + now: Instant, + enabled: bool, + aggressiveness: WebCarrierNegotiationAggressiveness, + lifetime: Duration, +) -> u64 { + let outcome = learning.apply_policy( + now, + 1, + enabled, + aggressiveness, + lifetime, + Duration::from_secs(3), + ); + drop(outcome.detached); + outcome.epoch.unwrap() +} + #[test] fn policy_epoch_rejects_late_outcomes_and_clears_state() { let now = Instant::now(); let mut learning = CarrierLearning::new(6); - let epoch = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let epoch = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); learning.record_chain(now, epoch, context(1), &[], WebCarrier::Websocket); assert_eq!(learning.entries.len(), 3); - let next = learning - .apply_policy( - now, - false, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let next = apply_epoch( + &mut learning, + now, + false, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); assert_ne!(epoch, next); learning.record_chain(now, epoch, context(1), &[], WebCarrier::Https); assert!(learning.entries.is_empty()); @@ -53,14 +70,13 @@ fn policy_epoch_rejects_late_outcomes_and_clears_state() { fn aggressive_policy_ranks_one_atomic_chain_sample() { let now = Instant::now(); let mut learning = CarrierLearning::new(6); - let epoch = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let epoch = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); learning.record_chain(now, epoch, context(2), &[], WebCarrier::Websocket); let (ranked, scores) = learning.rank( now, @@ -89,14 +105,13 @@ fn aggressive_policy_ranks_one_atomic_chain_sample() { fn two_half_windows_expire_without_sliding_updates() { let start = Instant::now(); let mut learning = CarrierLearning::new(6); - let epoch = learning - .apply_policy( - start, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let epoch = apply_epoch( + &mut learning, + start, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); learning.record_chain(start, epoch, context(3), &[], WebCarrier::Websocket); learning.record_chain( start + Duration::from_secs(6), @@ -116,36 +131,43 @@ fn exhausted_epoch_and_insertion_identifiers_fail_closed() { let now = Instant::now(); let mut learning = CarrierLearning::new(3); learning.epoch = Some(u64::MAX); - assert_eq!( - learning.apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ), - None + let exhausted = learning.apply_policy( + now, + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ); + assert_eq!(exhausted.epoch, None); + drop(exhausted.detached); + assert_eq!( + learning.record_chain(now, u64::MAX, context(4), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::StaleEpoch ); - learning.record_chain(now, u64::MAX, context(4), &[], WebCarrier::Https); assert!(learning.entries.is_empty()); learning.epoch = Some(1); learning.insertion_sequence = u64::MAX; - learning.record_chain(now, 1, context(4), &[], WebCarrier::Https); + assert_eq!( + learning.record_chain(now, 1, context(4), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::SequenceExhausted + ); assert!(learning.entries.is_empty()); + assert!(learning.insertion_order.is_empty()); } #[test] fn client_reported_failures_do_not_create_negative_evidence() { let now = Instant::now(); let mut learning = CarrierLearning::new(3); - let epoch = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let epoch = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); learning.record_chain( now, epoch, @@ -168,52 +190,208 @@ fn client_reported_failures_do_not_create_negative_evidence() { fn old_new_old_policy_rejects_both_stale_epochs() { let now = Instant::now(); let mut learning = CarrierLearning::new(3); - let old = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); - let middle = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Balanced, - Duration::from_secs(10), - ) - .unwrap(); - let current = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let old = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); + let middle = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Balanced, + Duration::from_secs(10), + ); + let current = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); assert_ne!(old, middle); assert_ne!(middle, current); assert_ne!(old, current); - learning.record_chain(now, old, context(6), &[], WebCarrier::Https); - learning.record_chain(now, middle, context(6), &[], WebCarrier::Https); + assert_eq!( + learning.record_chain(now, old, context(6), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::StaleEpoch + ); + assert_eq!( + learning.record_chain(now, middle, context(6), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::StaleEpoch + ); assert!(learning.entries.is_empty()); - learning.record_chain(now, current, context(6), &[], WebCarrier::Https); + assert_eq!( + learning.record_chain(now, current, context(6), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::Recorded + ); assert_eq!(learning.entries.len(), 3); } +#[test] +fn generation_rollover_preserves_same_policy_evidence() { + let now = Instant::now(); + let mut learning = CarrierLearning::new(3); + let first = learning.apply_policy( + now, + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ); + let epoch = first.epoch.unwrap(); + assert!(first.detached.is_some()); + drop(first.detached); + assert_eq!( + learning.record_chain(now, epoch, context(8), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::Recorded + ); + + let second = learning.apply_policy( + now, + 2, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ); + + assert!(second.applied); + assert_eq!(second.epoch, Some(epoch)); + assert!(second.detached.is_none()); + assert_eq!(learning.entries.len(), 3); + assert_eq!( + learning.epoch_for_policy( + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ), + CarrierLearningEpoch::Pending + ); + assert_eq!( + learning.epoch_for_policy( + 2, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ), + CarrierLearningEpoch::Ready(epoch) + ); +} + +#[test] +fn stale_generation_cannot_restore_an_old_policy() { + let now = Instant::now(); + let mut learning = CarrierLearning::new(3); + let current = learning.apply_policy( + now, + 2, + true, + WebCarrierNegotiationAggressiveness::Balanced, + Duration::from_secs(10), + Duration::from_secs(3), + ); + let epoch = current.epoch; + drop(current.detached); + + let stale = learning.apply_policy( + now, + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(20), + Duration::from_secs(4), + ); + let status = learning.status(now); + + assert!(!stale.applied); + assert_eq!(stale.epoch, epoch); + assert!(stale.detached.is_none()); + assert_eq!(status.policy_generation, Some(2)); + assert_eq!( + status.aggressiveness, + WebCarrierNegotiationAggressiveness::Balanced + ); + assert_eq!(status.lifetime_secs, 10); + assert_eq!(status.health_secs, 3); +} + +#[test] +fn health_window_change_starts_a_new_empty_epoch() { + let now = Instant::now(); + let mut learning = CarrierLearning::new(3); + let first = learning.apply_policy( + now, + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ); + let epoch = first.epoch.unwrap(); + drop(first.detached); + assert_eq!( + learning.record_chain(now, epoch, context(9), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::Recorded + ); + + let changed = learning.apply_policy( + now, + 2, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(4), + ); + + assert_ne!(changed.epoch, Some(epoch)); + assert_eq!( + changed.detached.as_ref().map(|detached| detached.entries.len()), + Some(3) + ); + assert!(learning.entries.is_empty()); + drop(changed.detached); +} + +#[test] +fn capacity_rejection_does_not_partially_mutate_the_store() { + let now = Instant::now(); + let mut learning = CarrierLearning::new(2); + let epoch = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); + let sequence = learning.insertion_sequence; + + assert_eq!( + learning.record_chain(now, epoch, context(10), &[], WebCarrier::Https), + CarrierLearningRecordOutcome::CapacityRejected + ); + assert!(learning.entries.is_empty()); + assert!(learning.insertion_order.is_empty()); + assert_eq!(learning.insertion_sequence, sequence); +} + #[test] fn fifo_metadata_stays_within_the_entry_capacity() { let now = Instant::now(); let mut learning = CarrierLearning::new(3); - let epoch = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let epoch = apply_epoch( + &mut learning, + now, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + ); for hash in 1..=32 { learning.record_chain(now, epoch, context(hash), &[], WebCarrier::Https); assert!(learning.entries.len() <= 3); @@ -221,6 +399,55 @@ fn fifo_metadata_stays_within_the_entry_capacity() { } } +#[tokio::test] +async fn runtime_activation_publishes_matching_policy_and_generation() { + let mut config = crate::config::ProxyConfig::default(); + config.web.carriers = crate::config::WebCarriers::Enabled(vec![WebCarrier::Websocket]); + config.web.carrier_learning = true; + config.web.carrier_negotiation_aggressiveness = + WebCarrierNegotiationAggressiveness::Aggressive; + config.web.timeouts.carrier_learning_secs = 10; + config.web.timeouts.carrier_health_secs = 3; + let first = crate::maestro::generation::test_runtime_generation(1, config.clone()); + let runtime = crate::web::manager::WebProcessRuntime::start(std::sync::Arc::new( + arc_swap::ArcSwap::from(first.clone()), + )); + let epoch = { + let mut learning = runtime.learning.lock(); + let epoch = learning.status(Instant::now()).epoch.unwrap(); + assert_eq!( + learning.record_chain( + Instant::now(), + epoch, + context(11), + &[], + WebCarrier::Websocket, + ), + CarrierLearningRecordOutcome::Recorded + ); + epoch + }; + let second = crate::maestro::generation::test_runtime_generation(2, config); + + let replaced = runtime.activate_generation(second.clone()); + + assert!(std::sync::Arc::ptr_eq(&replaced, &first)); + assert_eq!(runtime.active_generation().id, 2); + { + let learning = runtime.learning.lock(); + let status = learning.status(Instant::now()); + assert_eq!(status.policy_generation, Some(2)); + assert_eq!(status.epoch, Some(epoch)); + assert_eq!(status.entries, 3); + } + + runtime.shutdown().await; + first.stop_sessions().await; + first.stop_background_tasks().await; + second.stop_sessions().await; + second.stop_background_tasks().await; +} + #[tokio::test] async fn explicit_reset_preserves_policy_and_rejects_old_epoch_outcomes() { let generation = crate::maestro::generation::test_runtime_generation( @@ -233,14 +460,16 @@ async fn explicit_reset_preserves_policy_and_rejects_old_epoch_outcomes() { let now = Instant::now(); let old_epoch = { let mut learning = runtime.learning.lock(); - let epoch = learning - .apply_policy( - now, - true, - WebCarrierNegotiationAggressiveness::Aggressive, - Duration::from_secs(10), - ) - .unwrap(); + let outcome = learning.apply_policy( + now, + 1, + true, + WebCarrierNegotiationAggressiveness::Aggressive, + Duration::from_secs(10), + Duration::from_secs(3), + ); + let epoch = outcome.epoch.unwrap(); + drop(outcome.detached); learning.record_chain(now, epoch, context(7), &[], WebCarrier::Websocket); epoch }; diff --git a/src/web/manager/carrier_outcome.rs b/src/web/manager/carrier_outcome.rs index c77fb64..86a5c23 100644 --- a/src/web/manager/carrier_outcome.rs +++ b/src/web/manager/carrier_outcome.rs @@ -9,8 +9,25 @@ use super::{ }; use crate::config::WebCarrier; use crate::web::session::WebSession; +use crate::web::telemetry::WebCarrierLearningOutcome; use crate::web::trace::{TraceIdentity, TraceLifecycleEvent}; +/// Typed result of publishing one exact session health transition. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum CarrierHealthPublicationOutcome { + /// Manager ownership was confirmed and the attempt became healthy. + Published(WebCarrierLearningOutcome), + /// Manager or transport ownership rejected the publication. + Rejected(WebCarrierLearningOutcome), +} + +impl CarrierHealthPublicationOutcome { + /// Returns whether manager state accepted the health transition. + pub(crate) const fn published(self) -> bool { + matches!(self, Self::Published(_)) + } +} + impl WebProcessRuntime { /// Returns authenticated current chain metadata after a committed retry conflict. pub(crate) fn carrier_echo( @@ -123,30 +140,75 @@ impl WebProcessRuntime { learning_context: Option, client_ip: IpAddr, identity: TraceIdentity, - ) { - let (scores, failures) = { + websocket_owner: Option, + ) -> CarrierHealthPublicationOutcome { + if carrier.uses_websocket() + && websocket_owner.is_none_or(|owner| { + !self.claim_websocket_health(owner, session_hash) + }) + { + let outcome = WebCarrierLearningOutcome::OwnerNotLive; + self.telemetry.record_carrier_learning(carrier, outcome); + return CarrierHealthPublicationOutcome::Rejected(outcome); + } + let scores = { let mut state = self.state.lock(); let Some(entry) = state.bootstraps.get_mut(&bootstrap_hash) else { - return; + let outcome = WebCarrierLearningOutcome::MissingChain; + self.telemetry.record_carrier_learning(carrier, outcome); + return CarrierHealthPublicationOutcome::Rejected(outcome); }; - if entry.carrier_attempt != attempt - || entry.carrier_phase != CarrierChainPhase::CommittedPendingHealth - || entry - .session - .as_ref() - .is_none_or(|session| session.token_hash() != session_hash) - { - return; + if entry.carrier_phase != CarrierChainPhase::CommittedPendingHealth { + let outcome = WebCarrierLearningOutcome::PhaseMismatch; + self.telemetry.record_carrier_learning(carrier, outcome); + return CarrierHealthPublicationOutcome::Rejected(outcome); + } + let Some(session) = entry.session.as_ref().filter(|session| { + entry.carrier_attempt == attempt && session.token_hash() == session_hash + }) else { + let outcome = WebCarrierLearningOutcome::SessionMismatch; + self.telemetry.record_carrier_learning(carrier, outcome); + return CarrierHealthPublicationOutcome::Rejected(outcome); + }; + if !session.publish_carrier_health() { + let outcome = WebCarrierLearningOutcome::ClosedBeforeHealth; + self.telemetry.record_carrier_learning(carrier, outcome); + return CarrierHealthPublicationOutcome::Rejected(outcome); } entry.carrier_phase = CarrierChainPhase::Healthy; - (entry.carrier_scores, entry.carrier_failures) + entry.carrier_scores }; - if let Some(context) = learning_context { + let learning_outcome = if let Some(context) = learning_context { let now = Instant::now(); - let mut learning = self.learning.lock(); - let failures = failures.into_iter().flatten().collect::>(); - learning.record_chain(now, context.epoch, context, &failures, carrier); - } + let outcome = self.learning.lock().record_chain( + now, + context.epoch, + context, + &[], + carrier, + ); + match outcome { + super::learning::CarrierLearningRecordOutcome::Recorded => { + WebCarrierLearningOutcome::Recorded + } + super::learning::CarrierLearningRecordOutcome::PolicyDisabled => { + WebCarrierLearningOutcome::PolicyDisabled + } + super::learning::CarrierLearningRecordOutcome::StaleEpoch => { + WebCarrierLearningOutcome::StaleEpoch + } + super::learning::CarrierLearningRecordOutcome::CapacityRejected => { + WebCarrierLearningOutcome::CapacityRejected + } + super::learning::CarrierLearningRecordOutcome::SequenceExhausted => { + WebCarrierLearningOutcome::SequenceExhausted + } + } + } else { + WebCarrierLearningOutcome::NotEligible + }; + self.telemetry + .record_carrier_learning(carrier, learning_outcome); self.trace.record_carrier_lifecycle( client_ip, identity, @@ -157,5 +219,6 @@ impl WebProcessRuntime { scores, None, ); + CarrierHealthPublicationOutcome::Published(learning_outcome) } } diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index 60e84fb..2b20a7c 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -147,6 +147,8 @@ impl WebProcessRuntime { carrier_deadline_at: None, carrier_failures: [None; 3], carrier_learning_epoch: 0, + carrier_learning_disposition: + crate::web::telemetry::WebCarrierSelectionDisposition::ProfileDisabled, close_requested: false, session_client_ip: None, session_ip_learning_eligible: false, @@ -218,6 +220,7 @@ impl WebProcessRuntime { &self, hash: TokenHash, host: &str, + failure: Option, ) -> std::result::Result<(), ManagerError> { let mut state = self.state.lock(); let session = state @@ -225,7 +228,8 @@ impl WebProcessRuntime { .get(&hash) .filter(|session| session.matches_host(host)) .cloned(); - if session.is_some() { + let mut failure_phase = None; + if let Some(session) = &session { for bootstrap in state.bootstraps.values_mut() { if bootstrap .session @@ -233,6 +237,15 @@ impl WebProcessRuntime { .is_some_and(|current| current.token_hash() == hash) { bootstrap.close_requested = true; + failure_phase = Some(if matches!( + bootstrap.carrier_phase, + CarrierChainPhase::CommittedPendingHealth | CarrierChainPhase::Healthy + ) || session.is_carrier_committed() + { + crate::web::telemetry::WebCarrierFailurePhase::Committed + } else { + crate::web::telemetry::WebCarrierFailurePhase::Provisional + }); break; } } @@ -243,7 +256,12 @@ impl WebProcessRuntime { .is_some_and(|closed| closed.host == host); drop(state); if let Some(session) = session { - session.close(); + if session.close() + && let (Some(failure), Some(phase)) = (failure, failure_phase) + { + self.telemetry + .record_carrier_failure(session.carrier(), phase, failure); + } return Ok(()); } closed.then_some(()).ok_or(ManagerError::Authentication) diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index 6f1cdec..e924901 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -1,4 +1,5 @@ use std::net::IpAddr; +use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::time::Instant as TokioInstant; @@ -9,6 +10,7 @@ use super::state::{ remove_bootstrap_locked, remove_expired_locked, }; use super::{ProfileKey, TokenHash, WebProcessRuntime}; +use crate::maestro::generation::RuntimeGeneration; /// Result of draining all process-owned WEB work under one absolute deadline. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -27,6 +29,30 @@ pub(crate) struct WebShutdownDrain { } impl WebProcessRuntime { + /// Applies learning policy before publishing one new runtime generation. + pub(crate) fn activate_generation( + &self, + generation: Arc, + ) -> Arc { + let config = generation.config(); + let (replaced, detached) = { + let mut learning = self.learning.lock(); + let outcome = learning.apply_policy( + Instant::now(), + generation.id, + config.web.carrier_negotiation_enabled() && config.web.carrier_learning, + config.web.carrier_negotiation_aggressiveness, + Duration::from_secs(config.web.timeouts.carrier_learning_secs), + Duration::from_secs(config.web.timeouts.carrier_health_secs), + ); + debug_assert!(outcome.applied, "runtime generations must increase monotonically"); + let replaced = self.active_runtime.swap(generation); + (replaced, outcome.detached) + }; + drop(detached); + replaced + } + /// Removes one closed session and retains a bounded host-bound replay marker. pub(crate) fn session_finished( &self, @@ -150,15 +176,20 @@ impl WebProcessRuntime { let generation = self.active_generation(); let config = &generation.config().web; let learning_enabled = config.carrier_negotiation_enabled() && config.carrier_learning; - let mut learning = self.learning.lock(); - let _ = learning.apply_policy( - now, - learning_enabled, - config.carrier_negotiation_aggressiveness, - Duration::from_secs(config.timeouts.carrier_learning_secs), - ); - learning.prune(now); - drop(learning); + let detached = { + let mut learning = self.learning.lock(); + let outcome = learning.apply_policy( + now, + generation.id, + learning_enabled, + config.carrier_negotiation_aggressiveness, + Duration::from_secs(config.timeouts.carrier_learning_secs), + Duration::from_secs(config.timeouts.carrier_health_secs), + ); + learning.prune(now); + outcome.detached + }; + drop(detached); let (sessions, expired_chains) = { let mut state = self.state.lock(); state.apply_issuance_policy(generation.id, config.enabled); diff --git a/src/web/manager/negotiation.rs b/src/web/manager/negotiation.rs index 27252bd..1231afb 100644 --- a/src/web/manager/negotiation.rs +++ b/src/web/manager/negotiation.rs @@ -43,6 +43,26 @@ pub(crate) enum CarrierFailure { } impl CarrierFailure { + /// Complete fixed bridge failure set in stable metric order. + pub(crate) const ALL: [Self; 5] = [ + Self::Timeout, + Self::Network, + Self::Upgrade, + Self::Http, + Self::Protocol, + ]; + + /// Returns the stable fixed-slot index used by process telemetry. + pub(crate) const fn index(self) -> usize { + match self { + Self::Timeout => 0, + Self::Network => 1, + Self::Upgrade => 2, + Self::Http => 3, + Self::Protocol => 4, + } + } + /// Parses one canonical bridge failure token. pub(crate) const fn parse(value: &str) -> Option { match value.as_bytes() { diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index 9f9512b..67c3674 100644 --- a/src/web/manager/session_creation.rs +++ b/src/web/manager/session_creation.rs @@ -19,6 +19,7 @@ use super::{ use crate::config::{WebCarrier, WebRuntimeProfile}; use crate::web::frame; use crate::web::session::WebSession; +use crate::web::telemetry::WebCarrierSelectionDisposition; use crate::web::trace::TraceLifecycleEvent; struct Replacement { @@ -31,6 +32,7 @@ struct Replacement { request: CarrierRequest, scores: [i16; 4], learning_epoch: u64, + learning_disposition: WebCarrierSelectionDisposition, ip_learning_eligible: bool, carrier_deadline_at: Instant, } @@ -185,6 +187,7 @@ impl WebProcessRuntime { request: carrier_request, scores: entry.carrier_scores, learning_epoch: entry.carrier_learning_epoch, + learning_disposition: entry.carrier_learning_disposition, ip_learning_eligible, carrier_deadline_at: entry.carrier_deadline_at.ok_or(ManagerError::Protocol)?, }; @@ -223,51 +226,70 @@ impl WebProcessRuntime { config.web.carrier_negotiation_enabled() && config.web.carrier_learning, config.web.carrier_negotiation_aggressiveness, Duration::from_secs(config.web.timeouts.carrier_learning_secs), + Duration::from_secs(config.web.timeouts.carrier_health_secs), ); - let (candidates, scores, learning_epoch) = if capability_selection + let (candidates, scores, learning_epoch, learning_disposition) = if capability_selection && profile.carrier_learning + && learning_policy.0 { let learning = self.learning.lock(); - if let Some(epoch) = - learning.epoch_for_policy(learning_policy.0, learning_policy.1, learning_policy.2) - { - let (candidates, scores) = learning.rank( - now, - &profile.carriers, - carrier_request, - profile_key, - client_ip, - ip_learning_eligible, - ); - (candidates, scores, Some(epoch)) - } else { - ( - profile - .carriers - .iter() - .copied() - .filter(|carrier| carrier_request.supports(*carrier)) - .collect(), + match learning.epoch_for_policy( + generation.id, + learning_policy.0, + learning_policy.1, + learning_policy.2, + learning_policy.3, + ) { + super::learning::CarrierLearningEpoch::Ready(epoch) => { + let (candidates, scores) = learning.rank( + now, + &profile.carriers, + carrier_request, + profile_key, + client_ip, + ip_learning_eligible, + ); + let disposition = if scores.iter().any(|score| *score != 0) { + WebCarrierSelectionDisposition::Applied + } else { + WebCarrierSelectionDisposition::Cold + }; + (candidates, scores, Some(epoch), disposition) + } + super::learning::CarrierLearningEpoch::Pending => ( + supported_candidates(&profile.carriers, carrier_request), [0; 4], None, - ) + WebCarrierSelectionDisposition::PolicyPending, + ), + super::learning::CarrierLearningEpoch::Exhausted => ( + supported_candidates(&profile.carriers, carrier_request), + [0; 4], + None, + WebCarrierSelectionDisposition::EpochExhausted, + ), } } else if capability_selection { ( - profile - .carriers - .iter() - .copied() - .filter(|carrier| carrier_request.supports(*carrier)) - .collect(), + supported_candidates(&profile.carriers, carrier_request), [0; 4], None, + if learning_policy.0 { + WebCarrierSelectionDisposition::ProfileDisabled + } else { + WebCarrierSelectionDisposition::PolicyDisabled + }, ) } else if carrier_request.uses_capabilities() && !carrier_request.supports(profile.carrier) { return Err(ManagerError::Protocol); } else { - (vec![profile.carrier], [0; 4], None) + ( + vec![profile.carrier], + [0; 4], + None, + WebCarrierSelectionDisposition::ProfileDisabled, + ) }; let Some(carrier) = candidates.first().copied() else { return Err(ManagerError::Protocol); @@ -331,6 +353,7 @@ impl WebProcessRuntime { entry.carrier_deadline_at = carrier_deadline_at; entry.carrier_failures = [None; 3]; entry.carrier_learning_epoch = learning_epoch.unwrap_or(0); + entry.carrier_learning_disposition = learning_disposition; entry.expires_at = now + Duration::from_secs(issued_timeouts.bootstrap_lifetime_secs); entry.session_client_ip = Some(client_ip); entry.session_ip_learning_eligible = ip_learning_eligible; @@ -367,6 +390,8 @@ impl WebProcessRuntime { }, ); drop(state); + self.telemetry + .record_carrier_selection(carrier, learning_disposition); self.trace.record_carrier_lifecycle( client_ip, identity.clone(), @@ -399,5 +424,16 @@ impl WebProcessRuntime { } } +fn supported_candidates( + configured: &[WebCarrier], + request: CarrierRequest, +) -> Vec { + configured + .iter() + .copied() + .filter(|carrier| request.supports(*carrier)) + .collect() +} + // Atomic pre-commit carrier replacement and frozen-policy transfer. mod replacement; diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index fdcaab4..985a790 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/src/web/manager/session_creation/replacement.rs @@ -135,6 +135,17 @@ impl WebProcessRuntime { let old_identity = replacement.old_session.trace_identity(); drop(state); supersede.finish(); + self.telemetry.record_carrier_selection( + replacement.carrier, + replacement.learning_disposition, + ); + if let Some(failure) = replacement.request.failure() { + self.telemetry.record_carrier_failure( + replacement.old_session.carrier(), + crate::web::telemetry::WebCarrierFailurePhase::Provisional, + failure, + ); + } self.trace.record_carrier_lifecycle( client_ip, old_identity.clone(), diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index f7ec376..71c84d5 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -11,6 +11,7 @@ use super::{CarrierRequest, ProfileKey, TOKEN_BYTES, TokenHash}; use crate::config::{WebCarrier, WebRuntimeConfig, WebRuntimeProfile, WebTimeoutsConfig}; use crate::maestro::generation::RuntimeGeneration; use crate::web::session::WebSession; +use crate::web::telemetry::WebCarrierSelectionDisposition; const WEB_PROFILE_OWNER_CONTEXT: &[u8] = b"telemt-web-profile-owner-v1\0"; @@ -75,6 +76,8 @@ pub(super) struct Bootstrap { pub(super) carrier_failures: [Option; 3], /// Learning-policy epoch frozen by the first automatic attempt. pub(super) carrier_learning_epoch: u64, + /// Frozen reason learning did or did not influence carrier ordering. + pub(super) carrier_learning_disposition: WebCarrierSelectionDisposition, /// DELETE observed before an in-flight replacement committed its swap. pub(super) close_requested: bool, /// Effective address frozen by the first session-creation request. diff --git a/src/web/manager/status.rs b/src/web/manager/status.rs index 825542c..297e915 100644 --- a/src/web/manager/status.rs +++ b/src/web/manager/status.rs @@ -72,10 +72,12 @@ struct WebSocketStatus { struct LearningStatus { enabled: bool, aggressiveness: WebCarrierNegotiationAggressiveness, + policy_generation: Option, epoch: Option, entries: usize, capacity: usize, lifetime_secs: u64, + health_secs: u64, age_ms: u64, } @@ -267,10 +269,12 @@ impl WebProcessRuntime { .map(|status| LearningStatus { enabled: status.enabled, aggressiveness: status.aggressiveness, + policy_generation: status.policy_generation, epoch: status.epoch, entries: status.entries, capacity: status.capacity, lifetime_secs: status.lifetime_secs, + health_secs: status.health_secs, age_ms: status.age_ms, }); if learning.is_none() { diff --git a/src/web/manager/websocket.rs b/src/web/manager/websocket.rs index 3c7dc28..cdfcacc 100644 --- a/src/web/manager/websocket.rs +++ b/src/web/manager/websocket.rs @@ -50,6 +50,7 @@ pub(super) struct WebSocketEntry { last_progress_tick: AtomicU64, phase: AtomicU8, closing: AtomicBool, + health_claimed: AtomicBool, cancel: CancellationToken, released: CancellationToken, } @@ -345,6 +346,7 @@ fn try_admit( last_progress_tick: AtomicU64::new(now), phase: AtomicU8::new(WebSocketPhase::Claimed as u8), closing: AtomicBool::new(false), + health_claimed: AtomicBool::new(false), cancel: parent_cancellation.child_token(), released: CancellationToken::new(), }); @@ -371,6 +373,26 @@ impl WebProcessRuntime { self.websocket_clock.elapsed().as_millis() as u64 } + /// Claims health from one exact active owner under the eviction registry lock. + pub(super) fn claim_websocket_health( + &self, + owner: u64, + session_hash: super::TokenHash, + ) -> bool { + let registry = self.websockets.lock(); + let Some(entry) = registry.entries.get(&owner) else { + return false; + }; + !registry.closed + && entry.claim.session_hash == session_hash + && entry.phase.load(Ordering::Acquire) == WebSocketPhase::Active as u8 + && !entry.closing.load(Ordering::Acquire) + && entry + .health_claimed + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + } + pub(super) fn cleanup_websockets(&self) { let now = self.websocket_tick(); let mut victims = claim_stale_victims(self, now); diff --git a/src/web/manager/websocket/tests.rs b/src/web/manager/websocket/tests.rs index 7591a60..5a95b6e 100644 --- a/src/web/manager/websocket/tests.rs +++ b/src/web/manager/websocket/tests.rs @@ -36,6 +36,7 @@ fn entry( last_progress_tick: AtomicU64::new(progress_tick), phase: AtomicU8::new(phase as u8), closing: AtomicBool::new(false), + health_claimed: AtomicBool::new(false), cancel: CancellationToken::new(), released: CancellationToken::new(), } @@ -297,3 +298,54 @@ async fn concurrent_victim_claims_stay_bounded_and_return_to_zero() { generation.stop_sessions().await; generation.stop_background_tasks().await; } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_health_claims_publish_one_exact_live_owner() { + let generation = test_runtime_generation(1, ProxyConfig::default()); + let runtime = WebProcessRuntime::start(Arc::new(ArcSwap::from(Arc::clone(&generation)))); + let owner = Arc::new(entry( + 1, + [1; 32], + 1, + "192.0.2.10", + WebSocketKind::Multiplex, + WebSocketPhase::Active, + 1, + 1, + )); + let connection = WebSocketConnection { + runtime: Arc::downgrade(&runtime), + entry: Arc::clone(&owner), + slot: None, + base_budget: None, + }; + { + let mut registry = runtime.websockets.lock(); + registry.claims.insert(owner.claim, owner.id); + registry.entries.insert(owner.id, Arc::clone(&owner)); + } + let successes = Arc::new(AtomicUsize::new(0)); + let mut tasks = Vec::new(); + for _ in 0..64 { + let runtime = Arc::clone(&runtime); + let successes = Arc::clone(&successes); + tasks.push(tokio::spawn(async move { + if runtime.claim_websocket_health(1, [0; 32]) { + successes.fetch_add(1, Ordering::AcqRel); + } + })); + } + for task in tasks { + task.await.unwrap(); + } + + assert_eq!(successes.load(Ordering::Acquire), 1); + assert!(!runtime.claim_websocket_health(1, [0; 32])); + assert!(!runtime.claim_websocket_health(2, [0; 32])); + assert!(!runtime.claim_websocket_health(1, [1; 32])); + + drop(connection); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} diff --git a/src/web/session.rs b/src/web/session.rs index 05486ac..f4b70e5 100644 --- a/src/web/session.rs +++ b/src/web/session.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::io; use std::net::IpAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize}; use std::task::{Context, Poll, Waker}; use std::time::Instant; @@ -39,6 +39,7 @@ pub(crate) use websocket::WebSocketLaneReservation; pub(crate) use websocket::WebSocketProbeReservation; // Carrier commit and health evidence share one session-locked state machine. mod negotiation; +use negotiation::CarrierHealthPublicationState; // Session closure and carrier-attempt transitions share one cancellation boundary. mod lifecycle; // Uplink batches own exactly-once sequencing and client-frame validation. @@ -175,7 +176,6 @@ struct SessionState { carrier_health_uplink: bool, carrier_health_downlink: bool, carrier_commit_published: bool, - carrier_health_reported: bool, websocket_carrier_active: bool, websocket_commit_ack_pending: bool, websocket_commit_ack_owner: Option, @@ -212,6 +212,7 @@ pub(crate) struct WebSession { limits: WebLimitsConfig, timeouts: WebTimeoutsConfig, state: Mutex, + carrier_health_publication: AtomicU8, down_notify: Arc, lane_open_notify: Arc, cancel: CancellationToken, @@ -304,7 +305,6 @@ impl WebSession { carrier_health_uplink: false, carrier_health_downlink: false, carrier_commit_published: false, - carrier_health_reported: false, websocket_carrier_active: false, websocket_commit_ack_pending: false, websocket_commit_ack_owner: None, @@ -313,6 +313,9 @@ impl WebSession { close_requested: false, closed: false, }), + carrier_health_publication: AtomicU8::new( + CarrierHealthPublicationState::Awaiting as u8, + ), down_notify: Arc::new(Notify::new()), lane_open_notify: Arc::new(Notify::new()), cancel: CancellationToken::new(), diff --git a/src/web/session/downlink.rs b/src/web/session/downlink.rs index 45df0c5..b7b5f3a 100644 --- a/src/web/session/downlink.rs +++ b/src/web/session/downlink.rs @@ -55,8 +55,8 @@ impl WebSession { let healthy = self.carrier_health_ready_locked(&mut state, Instant::now()); (state.down_epoch, healthy) }; - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } self.down_notify.notify_waiters(); diff --git a/src/web/session/lane_uplink.rs b/src/web/session/lane_uplink.rs index 157c601..27fd7bc 100644 --- a/src/web/session/lane_uplink.rs +++ b/src/web/session/lane_uplink.rs @@ -39,7 +39,7 @@ impl WebSession { let digest: TokenHash = Sha256::digest(body).into(); let mut opened = Vec::new(); let mut committed = false; - let mut healthy = false; + let mut healthy = None; let result = { let mut state = self.state.lock(); if state.closed { @@ -172,8 +172,8 @@ impl WebSession { if committed { self.finish_carrier_commit(); } - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } self.lane_open_notify.notify_waiters(); for completion in opened { diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs index cc6748a..52f47ac 100644 --- a/src/web/session/lanes.rs +++ b/src/web/session/lanes.rs @@ -142,8 +142,8 @@ impl WebSession { let healthy = self.carrier_health_ready_locked(&mut state, Instant::now()); (instance, epoch, notify, healthy) }; - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } notify.notify_waiters(); diff --git a/src/web/session/lifecycle.rs b/src/web/session/lifecycle.rs index e2fc2b3..b0c221b 100644 --- a/src/web/session/lifecycle.rs +++ b/src/web/session/lifecycle.rs @@ -8,6 +8,7 @@ struct ReleasedQueues { data_items: usize, control_bytes: usize, control_items: usize, + closed_before_health: bool, } /// Deferred queue release after manager publication linearizes a supersede. @@ -26,11 +27,12 @@ impl CarrierSupersedeCompletion<'_> { impl WebSession { /// Closes carrier state while relay tasks retain their admission until exit. - pub(crate) fn close(&self) { + pub(crate) fn close(&self) -> bool { let Some(released) = self.begin_close(false, None) else { - return; + return false; }; self.finish_close(released, false); + true } /// Atomically prevents first-frame commit while one successor is prepared. @@ -97,8 +99,8 @@ impl WebSession { let mut state = self.state.lock(); self.carrier_health_ready_locked(&mut state, now) }; - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } let Some(released) = self.begin_close(false, Some(now)) else { return false; @@ -127,6 +129,9 @@ impl WebSession { state.close_requested = true; return None; } + let closed_before_health = self.automatic_carrier + && state.negotiation_phase == SessionNegotiationPhase::Committed + && self.reject_carrier_health_on_close(); state.closed = true; if superseded { state.negotiation_phase = SessionNegotiationPhase::Superseded; @@ -177,6 +182,7 @@ impl WebSession { data_items, control_bytes, control_items, + closed_before_health, }) } @@ -189,6 +195,12 @@ impl WebSession { self.lane_open_notify.notify_waiters(); } if let Some(manager) = self.manager.upgrade() { + if released.closed_before_health { + manager.telemetry().record_carrier_learning( + self.selected_carrier, + crate::web::telemetry::WebCarrierLearningOutcome::ClosedBeforeHealth, + ); + } manager.release_pending( self.profile_key, released.data_bytes, diff --git a/src/web/session/negotiation.rs b/src/web/session/negotiation.rs index 26e10d6..a213b30 100644 --- a/src/web/session/negotiation.rs +++ b/src/web/session/negotiation.rs @@ -3,6 +3,26 @@ use std::time::{Duration, Instant}; use super::uplink::AppliedProgress; use super::{SessionNegotiationPhase, SessionState, WebSession}; +/// Fixed ownership state for one carrier-health publication attempt. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u8)] +pub(super) enum CarrierHealthPublicationState { + /// No eligible callback has claimed publication. + Awaiting, + /// One callback is validating manager and transport ownership. + Publishing, + /// Manager state accepted the health transition. + Published, + /// Close or ownership validation permanently rejected publication. + Rejected, +} + +/// Transport owner captured by the single publication claimant. +#[derive(Clone, Copy)] +pub(super) struct CarrierHealthClaim { + websocket_owner: Option, +} + impl WebSession { /// Returns whether accepted carrier progress made this attempt immutable. pub(crate) fn is_carrier_committed(&self) -> bool { @@ -48,21 +68,21 @@ impl WebSession { let healthy = { let mut state = self.state.lock(); if state.closed || state.negotiation_phase != SessionNegotiationPhase::Committed { - false + None } else { state.carrier_commit_published = true; self.carrier_health_ready_locked(&mut state, Instant::now()) } }; - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } } /// Publishes complete transport-specific health evidence to process state. - pub(super) fn finish_carrier_health(&self) { + pub(super) fn finish_carrier_health(&self, claim: CarrierHealthClaim) { if let Some(manager) = self.manager.upgrade() { - manager.carrier_became_healthy( + let outcome = manager.carrier_became_healthy( self.bootstrap_hash, self.token_hash, self.carrier_attempt, @@ -71,7 +91,13 @@ impl WebSession { self.learning_context, self.client_ip, self.trace_identity(), + claim.websocket_owner, ); + if !outcome.published() { + self.reject_carrier_health_publication(); + } + } else { + self.reject_carrier_health_publication(); } } @@ -80,9 +106,9 @@ impl WebSession { &self, state: &mut SessionState, progress: AppliedProgress, - ) -> (bool, bool) { + ) -> (bool, Option) { if !self.automatic_carrier || !progress.any() { - return (false, false); + return (false, None); } if self.selected_carrier.uses_websocket() { state.websocket_carrier_active = true; @@ -109,15 +135,14 @@ impl WebSession { &self, state: &mut SessionState, now: Instant, - ) -> bool { + ) -> Option { if !self.automatic_carrier || state.closed || state.negotiation_phase != SessionNegotiationPhase::Committed || !state.carrier_commit_published - || state.carrier_health_reported || state.carrier_health_due_at.is_none_or(|due| now < due) { - return false; + return None; } let evidence = if state.websocket_carrier_active { state.websocket_probe_claimed @@ -132,10 +157,23 @@ impl WebSession { .zip(state.carrier_health_due_at) .is_some_and(|(activity, due)| activity >= due) }; - if evidence { - state.carrier_health_reported = true; + if !evidence { + return None; } - evidence + self.carrier_health_publication + .compare_exchange( + CarrierHealthPublicationState::Awaiting as u8, + CarrierHealthPublicationState::Publishing as u8, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ) + .ok() + .map(|_| CarrierHealthClaim { + websocket_owner: state + .websocket_carrier_active + .then_some(state.websocket_commit_ack_owner) + .flatten(), + }) } /// Returns whether the exact automatic WebSocket owner must receive a commit acknowledgement. @@ -175,17 +213,85 @@ impl WebSession { state.carrier_health_activity_at = Some(now); self.carrier_health_ready_locked(&mut state, now) }; - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } true } + + /// Confirms manager ownership as the health publication linearization point. + pub(crate) fn publish_carrier_health(&self) -> bool { + self.carrier_health_publication + .compare_exchange( + CarrierHealthPublicationState::Publishing as u8, + CarrierHealthPublicationState::Published as u8, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ) + .is_ok() + } + + /// Rejects an in-flight publication after manager validation fails. + pub(super) fn reject_carrier_health_publication(&self) { + let _ = self.carrier_health_publication.compare_exchange( + CarrierHealthPublicationState::Publishing as u8, + CarrierHealthPublicationState::Rejected as u8, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ); + } + + /// Rejects pending health on close and reports whether no callback was in flight. + pub(super) fn reject_carrier_health_on_close(&self) -> bool { + loop { + let current = self + .carrier_health_publication + .load(std::sync::atomic::Ordering::Acquire); + let count_locally = current == CarrierHealthPublicationState::Awaiting as u8; + if current != CarrierHealthPublicationState::Awaiting as u8 + && current != CarrierHealthPublicationState::Publishing as u8 + { + return false; + } + if self + .carrier_health_publication + .compare_exchange( + current, + CarrierHealthPublicationState::Rejected as u8, + std::sync::atomic::Ordering::AcqRel, + std::sync::atomic::Ordering::Acquire, + ) + .is_ok() + { + return count_locally; + } + } + } + + /// Returns the current fixed health-publication state. + pub(super) fn carrier_health_publication_state(&self) -> CarrierHealthPublicationState { + match self + .carrier_health_publication + .load(std::sync::atomic::Ordering::Acquire) + { + value if value == CarrierHealthPublicationState::Awaiting as u8 => { + CarrierHealthPublicationState::Awaiting + } + value if value == CarrierHealthPublicationState::Publishing as u8 => { + CarrierHealthPublicationState::Publishing + } + value if value == CarrierHealthPublicationState::Published as u8 => { + CarrierHealthPublicationState::Published + } + _ => CarrierHealthPublicationState::Rejected, + } + } } #[cfg(test)] mod tests { use std::net::SocketAddr; - use std::sync::Arc; + use std::sync::{Arc, Barrier}; use super::*; use crate::config::{ @@ -229,6 +335,16 @@ mod tests { ) } + fn arm_http_health(session: &WebSession, now: Instant) { + let mut state = session.state.lock(); + state.negotiation_phase = SessionNegotiationPhase::Committed; + state.carrier_commit_published = true; + state.carrier_health_due_at = Some(now - Duration::from_secs(1)); + state.carrier_health_uplink = true; + state.carrier_health_downlink = true; + state.carrier_health_activity_at = Some(now); + } + #[test] fn final_deadline_refuses_uncommitted_progress() { let session = session(WebCarrier::Https, Instant::now() - Duration::from_secs(1)); @@ -254,9 +370,9 @@ mod tests { state.carrier_health_uplink = true; state.carrier_health_downlink = true; state.carrier_health_activity_at = Some(now - Duration::from_secs(2)); - assert!(!session.carrier_health_ready_locked(&mut state, now)); + assert!(session.carrier_health_ready_locked(&mut state, now).is_none()); state.carrier_health_activity_at = Some(now); - assert!(session.carrier_health_ready_locked(&mut state, now)); + assert!(session.carrier_health_ready_locked(&mut state, now).is_some()); } #[test] @@ -274,9 +390,9 @@ mod tests { state.websocket_commit_ack_owner = Some(7); state.websocket_commit_ack_written = true; state.carrier_health_uplink = true; - assert!(!session.carrier_health_ready_locked(&mut state, now)); + assert!(session.carrier_health_ready_locked(&mut state, now).is_none()); state.websocket_probe_claimed = true; - assert!(session.carrier_health_ready_locked(&mut state, now)); + assert!(session.carrier_health_ready_locked(&mut state, now).is_some()); } #[test] @@ -290,8 +406,74 @@ mod tests { state.carrier_health_downlink = true; state.carrier_health_activity_at = Some(now); - assert!(!session.carrier_health_ready_locked(&mut state, now)); - assert!(!state.carrier_health_reported); + assert!(session.carrier_health_ready_locked(&mut state, now).is_none()); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Awaiting + ); + } + + #[test] + fn health_publication_claim_is_single_shot() { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + arm_http_health(&session, now); + let mut state = session.state.lock(); + + assert!(session.carrier_health_ready_locked(&mut state, now).is_some()); + assert!(session.carrier_health_ready_locked(&mut state, now).is_none()); + drop(state); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Publishing + ); + assert!(session.publish_carrier_health()); + assert!(!session.publish_carrier_health()); + assert_eq!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Published + ); + } + + #[test] + fn concurrent_health_and_close_always_reach_one_terminal_state() { + for _ in 0..512 { + let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60)); + let now = Instant::now(); + arm_http_health(&session, now); + let barrier = Arc::new(Barrier::new(3)); + let health_session = Arc::clone(&session); + let health_barrier = Arc::clone(&barrier); + let health = std::thread::spawn(move || { + health_barrier.wait(); + std::thread::yield_now(); + let claim = { + let mut state = health_session.state.lock(); + health_session.carrier_health_ready_locked(&mut state, now) + }; + if claim.is_some() { + health_session.publish_carrier_health(); + } + }); + let close_session = Arc::clone(&session); + let close_barrier = Arc::clone(&barrier); + let close = std::thread::spawn(move || { + close_barrier.wait(); + std::thread::yield_now(); + close_session.close(); + }); + barrier.wait(); + health.join().unwrap(); + close.join().unwrap(); + + assert!(matches!( + session.carrier_health_publication_state(), + CarrierHealthPublicationState::Published + | CarrierHealthPublicationState::Rejected + )); + assert!(!session.publish_carrier_health()); + assert!(!session.close()); + } } #[test] diff --git a/src/web/session/status.rs b/src/web/session/status.rs index 8310085..cd55cda 100644 --- a/src/web/session/status.rs +++ b/src/web/session/status.rs @@ -2,7 +2,7 @@ use std::time::Instant; use serde::Serialize; -use super::{SessionNegotiationPhase, WebSession}; +use super::{CarrierHealthPublicationState, SessionNegotiationPhase, WebSession}; use crate::config::WebCarrier; /// One bounded point-in-time session snapshot without bearer identity. @@ -28,6 +28,8 @@ pub(crate) struct WebSessionStatus { pub(crate) automatic: bool, /// Current session lifecycle token. pub(crate) state: &'static str, + /// Current manager-confirmed health publication phase. + pub(crate) health_publication: &'static str, /// Live logical streams. pub(crate) streams: usize, /// Stream relay tasks that have not exited. @@ -66,7 +68,9 @@ impl WebSession { "closed" } else if state.close_requested { "closing" - } else if state.carrier_health_reported { + } else if self.carrier_health_publication_state() + == CarrierHealthPublicationState::Published + { "healthy" } else { match state.negotiation_phase { @@ -87,6 +91,12 @@ impl WebSession { client_class: self.carrier_class.as_str(), automatic: self.automatic_carrier, state: state_name, + health_publication: match self.carrier_health_publication_state() { + CarrierHealthPublicationState::Awaiting => "awaiting", + CarrierHealthPublicationState::Publishing => "publishing", + CarrierHealthPublicationState::Published => "published", + CarrierHealthPublicationState::Rejected => "rejected", + }, streams: state.streams.len(), tasks: self.tasks_live(), lanes: state.carrier_lanes.len(), diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs index c224f0f..91f0a0d 100644 --- a/src/web/session/uplink.rs +++ b/src/web/session/uplink.rs @@ -85,7 +85,7 @@ impl WebSession { let digest: TokenHash = Sha256::digest(body).into(); let mut opened = Vec::new(); let mut committed = false; - let mut healthy = false; + let mut healthy = None; let result = { let mut state = self.state.lock(); if state.closed { @@ -154,8 +154,8 @@ impl WebSession { if committed { self.finish_carrier_commit(); } - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } for completion in opened { self.spawn_stream(completion, false); diff --git a/src/web/session/websocket.rs b/src/web/session/websocket.rs index 400d389..b1b6b26 100644 --- a/src/web/session/websocket.rs +++ b/src/web/session/websocket.rs @@ -58,7 +58,9 @@ impl Drop for WebSocketProbeReservation { state.websocket_probe_claimed = false; if state.websocket_commit_ack_owner == self.owner { state.websocket_commit_ack_owner = None; - if !state.carrier_health_reported { + if self.session.carrier_health_publication_state() + != super::CarrierHealthPublicationState::Published + { state.websocket_commit_ack_written = false; state.carrier_health_uplink = false; state.carrier_health_activity_at = None; @@ -316,7 +318,7 @@ impl WebSession { let digest = Sha256::digest(body).into(); let mut opened = Vec::new(); let mut committed = false; - let mut healthy = false; + let mut healthy = None; let result = { let mut state = self.state.lock(); if state.closed { @@ -401,8 +403,8 @@ impl WebSession { if committed { self.finish_carrier_commit(); } - if healthy { - self.finish_carrier_health(); + if let Some(claim) = healthy { + self.finish_carrier_health(claim); } for completion in opened { let stream = completion.stream; diff --git a/src/web/telemetry.rs b/src/web/telemetry.rs index d06d6c2..1b3e4d3 100644 --- a/src/web/telemetry.rs +++ b/src/web/telemetry.rs @@ -4,6 +4,13 @@ use std::time::Instant; use serde::Serialize; +mod carrier; +pub(crate) use carrier::{ + WebCarrierFailureCounter, WebCarrierFailurePhase, WebCarrierLearningCounter, + WebCarrierLearningOutcome, WebCarrierSelectionCounter, WebCarrierSelectionDisposition, +}; +use carrier::{CARRIER_FAILURE_SLOTS, CARRIER_LEARNING_SLOTS, CARRIER_SELECTION_SLOTS}; + const LAST_DECOY_OUTCOME_BITS: u32 = 4; const LAST_DECOY_OUTCOME_MASK: u64 = (1 << LAST_DECOY_OUTCOME_BITS) - 1; const LAST_DECOY_ELAPSED_MAX: u64 = u64::MAX >> LAST_DECOY_OUTCOME_BITS; @@ -300,6 +307,9 @@ pub(crate) struct WebTelemetry { rejections: [AtomicU64; WebRejectionReason::ALL.len()], overload_outcomes: [AtomicU64; WebHttpConnectionOverloadOutcome::ALL.len()], decoy_outcomes: [AtomicU64; WebDecoyUpstreamOutcome::ALL.len()], + carrier_selections: [AtomicU64; CARRIER_SELECTION_SLOTS], + carrier_failures: [AtomicU64; CARRIER_FAILURE_SLOTS], + carrier_learning_outcomes: [AtomicU64; CARRIER_LEARNING_SLOTS], last_decoy: AtomicU64, sessions_created: AtomicU64, sessions_closed: AtomicU64, @@ -321,6 +331,9 @@ impl WebTelemetry { rejections: std::array::from_fn(|_| AtomicU64::new(0)), overload_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), decoy_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), + carrier_selections: std::array::from_fn(|_| AtomicU64::new(0)), + carrier_failures: std::array::from_fn(|_| AtomicU64::new(0)), + carrier_learning_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), last_decoy: AtomicU64::new(0), sessions_created: AtomicU64::new(0), sessions_closed: AtomicU64::new(0), @@ -510,38 +523,5 @@ impl Drop for WebAcceptorGuard { } #[cfg(test)] -mod tests { - use super::*; - - #[test] - fn fixed_counter_sets_and_acceptor_guard_are_exact() { - let telemetry = WebTelemetry::new(); - let guard = telemetry.acceptor_guard(); - assert_eq!(telemetry.live_acceptors(), 1); - telemetry.record_rejection(WebRejectionReason::HttpConnectionCapacity); - telemetry.record_overload(WebHttpConnectionOverloadOutcome::Dropped); - telemetry.record_decoy(WebDecoyUpstreamOutcome::ConnectRefused); - assert_eq!( - telemetry.rejection_counters().len(), - WebRejectionReason::ALL.len() - ); - assert_eq!( - telemetry.overload_counters().len(), - WebHttpConnectionOverloadOutcome::ALL.len() - ); - assert_eq!( - telemetry.decoy_counters().len(), - WebDecoyUpstreamOutcome::ALL.len() - ); - assert_eq!( - telemetry.rejection_total(WebRejectionReason::HttpConnectionCapacity), - 1 - ); - assert_eq!( - telemetry.last_decoy().map(|value| value.0), - Some("connect_refused") - ); - drop(guard); - assert_eq!(telemetry.live_acceptors(), 0); - } -} +#[path = "telemetry/tests.rs"] +mod tests; diff --git a/src/web/telemetry/carrier.rs b/src/web/telemetry/carrier.rs new file mode 100644 index 0000000..ccffc2f --- /dev/null +++ b/src/web/telemetry/carrier.rs @@ -0,0 +1,310 @@ +use std::sync::atomic::Ordering; + +use serde::Serialize; + +use super::WebTelemetry; +use crate::config::WebCarrier; +use crate::web::manager::CarrierFailure; + +/// Reason learning did or did not influence one successful initial selection. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebCarrierSelectionDisposition { + /// The selected runtime profile did not enable learning. + ProfileDisabled, + /// The effective process policy disabled learning. + PolicyDisabled, + /// The exact generation policy had not reached the process store. + PolicyPending, + /// The monotonic evidence epoch space was exhausted. + EpochExhausted, + /// Active evidence produced no non-zero candidate score. + Cold, + /// Active evidence contributed to candidate ordering. + Applied, +} + +impl WebCarrierSelectionDisposition { + /// Complete fixed selection disposition set in stable metric order. + pub(crate) const ALL: [Self; 6] = [ + Self::ProfileDisabled, + Self::PolicyDisabled, + Self::PolicyPending, + Self::EpochExhausted, + Self::Cold, + Self::Applied, + ]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::ProfileDisabled => "profile_disabled", + Self::PolicyDisabled => "policy_disabled", + Self::PolicyPending => "policy_pending", + Self::EpochExhausted => "epoch_exhausted", + Self::Cold => "cold", + Self::Applied => "applied", + } + } +} + +/// Attempt-chain phase in which a client reported one carrier failure. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebCarrierFailurePhase { + /// The replaced attempt had not committed carrier progress. + Provisional, + /// The committed attempt failed before its health publication completed. + Committed, +} + +impl WebCarrierFailurePhase { + /// Complete fixed reported-failure phase set. + pub(crate) const ALL: [Self; 2] = [Self::Provisional, Self::Committed]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Provisional => "provisional", + Self::Committed => "committed", + } + } +} + +/// Terminal result of one carrier health or learning publication. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(usize)] +pub(crate) enum WebCarrierLearningOutcome { + /// The complete eligible chain was recorded. + Recorded, + /// The healthy session had no learning context. + NotEligible, + /// The effective learning policy rejected evidence. + PolicyDisabled, + /// The sample belonged to a superseded evidence epoch. + StaleEpoch, + /// The bounded evidence store could not accept the complete sample. + CapacityRejected, + /// Evidence insertion identifiers were exhausted. + SequenceExhausted, + /// The attempt chain no longer existed at publication time. + MissingChain, + /// The attempt chain was not awaiting health publication. + PhaseMismatch, + /// Another session incarnation owned the attempt chain. + SessionMismatch, + /// The exact WebSocket owner was closing or no longer active. + OwnerNotLive, + /// Session closure won the publication race. + ClosedBeforeHealth, +} + +impl WebCarrierLearningOutcome { + /// Complete fixed health and learning outcome set. + pub(crate) const ALL: [Self; 11] = [ + Self::Recorded, + Self::NotEligible, + Self::PolicyDisabled, + Self::StaleEpoch, + Self::CapacityRejected, + Self::SequenceExhausted, + Self::MissingChain, + Self::PhaseMismatch, + Self::SessionMismatch, + Self::OwnerNotLive, + Self::ClosedBeforeHealth, + ]; + + /// Returns the stable API and Prometheus label token. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Recorded => "recorded", + Self::NotEligible => "not_eligible", + Self::PolicyDisabled => "policy_disabled", + Self::StaleEpoch => "stale_epoch", + Self::CapacityRejected => "capacity_rejected", + Self::SequenceExhausted => "sequence_exhausted", + Self::MissingChain => "missing_chain", + Self::PhaseMismatch => "phase_mismatch", + Self::SessionMismatch => "session_mismatch", + Self::OwnerNotLive => "owner_not_live", + Self::ClosedBeforeHealth => "closed_before_health", + } + } +} + +pub(super) const CARRIER_SELECTION_SLOTS: usize = + WebCarrier::ALL.len() * WebCarrierSelectionDisposition::ALL.len(); +pub(super) const CARRIER_FAILURE_SLOTS: usize = WebCarrier::ALL.len() + * WebCarrierFailurePhase::ALL.len() + * CarrierFailure::ALL.len(); +pub(super) const CARRIER_LEARNING_SLOTS: usize = + WebCarrier::ALL.len() * WebCarrierLearningOutcome::ALL.len(); + +/// API-safe fixed carrier selection counter. +#[derive(Clone, Serialize)] +pub(crate) struct WebCarrierSelectionCounter { + /// Stable carrier token. + pub(crate) carrier: &'static str, + /// Stable selection disposition token. + pub(crate) disposition: &'static str, + /// Process-lifetime event count. + pub(crate) total: u64, +} + +/// API-safe fixed client-reported carrier failure counter. +#[derive(Clone, Serialize)] +pub(crate) struct WebCarrierFailureCounter { + /// Stable carrier token. + pub(crate) carrier: &'static str, + /// Attempt-chain phase at failure reporting time. + pub(crate) phase: &'static str, + /// Canonical client-reported failure token. + pub(crate) reason: &'static str, + /// Process-lifetime event count. + pub(crate) total: u64, +} + +/// API-safe fixed carrier health and learning outcome counter. +#[derive(Clone, Serialize)] +pub(crate) struct WebCarrierLearningCounter { + /// Stable carrier token. + pub(crate) carrier: &'static str, + /// Terminal health or learning publication token. + pub(crate) outcome: &'static str, + /// Process-lifetime event count. + pub(crate) total: u64, +} + +impl WebTelemetry { + /// Records why learning did or did not affect one successful selection. + pub(crate) fn record_carrier_selection( + &self, + carrier: WebCarrier, + disposition: WebCarrierSelectionDisposition, + ) { + self.carrier_selections[selection_index(carrier, disposition)] + .fetch_add(1, Ordering::Relaxed); + } + + /// Returns one fixed carrier selection counter. + pub(crate) fn carrier_selection_total( + &self, + carrier: WebCarrier, + disposition: WebCarrierSelectionDisposition, + ) -> u64 { + self.carrier_selections[selection_index(carrier, disposition)].load(Ordering::Relaxed) + } + + /// Captures the complete carrier selection counter set. + pub(crate) fn carrier_selection_counters(&self) -> Vec { + WebCarrier::ALL + .into_iter() + .flat_map(|carrier| { + WebCarrierSelectionDisposition::ALL + .into_iter() + .map(move |disposition| WebCarrierSelectionCounter { + carrier: carrier.as_str(), + disposition: disposition.as_str(), + total: self.carrier_selection_total(carrier, disposition), + }) + }) + .collect() + } + + /// Records one canonical failure reported for an authenticated chain. + pub(crate) fn record_carrier_failure( + &self, + carrier: WebCarrier, + phase: WebCarrierFailurePhase, + reason: CarrierFailure, + ) { + self.carrier_failures[failure_index(carrier, phase, reason)] + .fetch_add(1, Ordering::Relaxed); + } + + /// Returns one fixed client-reported carrier failure counter. + pub(crate) fn carrier_failure_total( + &self, + carrier: WebCarrier, + phase: WebCarrierFailurePhase, + reason: CarrierFailure, + ) -> u64 { + self.carrier_failures[failure_index(carrier, phase, reason)].load(Ordering::Relaxed) + } + + /// Captures the complete client-reported carrier failure counter set. + pub(crate) fn carrier_failure_counters(&self) -> Vec { + WebCarrier::ALL + .into_iter() + .flat_map(|carrier| { + WebCarrierFailurePhase::ALL.into_iter().flat_map(move |phase| { + CarrierFailure::ALL + .into_iter() + .map(move |reason| WebCarrierFailureCounter { + carrier: carrier.as_str(), + phase: phase.as_str(), + reason: reason.as_str(), + total: self.carrier_failure_total(carrier, phase, reason), + }) + }) + }) + .collect() + } + + /// Records one terminal health or evidence publication result. + pub(crate) fn record_carrier_learning( + &self, + carrier: WebCarrier, + outcome: WebCarrierLearningOutcome, + ) { + self.carrier_learning_outcomes[learning_index(carrier, outcome)] + .fetch_add(1, Ordering::Relaxed); + } + + /// Returns one fixed health and learning outcome counter. + pub(crate) fn carrier_learning_total( + &self, + carrier: WebCarrier, + outcome: WebCarrierLearningOutcome, + ) -> u64 { + self.carrier_learning_outcomes[learning_index(carrier, outcome)].load(Ordering::Relaxed) + } + + /// Captures the complete health and learning outcome counter set. + pub(crate) fn carrier_learning_counters(&self) -> Vec { + WebCarrier::ALL + .into_iter() + .flat_map(|carrier| { + WebCarrierLearningOutcome::ALL + .into_iter() + .map(move |outcome| WebCarrierLearningCounter { + carrier: carrier.as_str(), + outcome: outcome.as_str(), + total: self.carrier_learning_total(carrier, outcome), + }) + }) + .collect() + } +} + +const fn selection_index( + carrier: WebCarrier, + disposition: WebCarrierSelectionDisposition, +) -> usize { + carrier.index() * WebCarrierSelectionDisposition::ALL.len() + disposition as usize +} + +const fn failure_index( + carrier: WebCarrier, + phase: WebCarrierFailurePhase, + reason: CarrierFailure, +) -> usize { + (carrier.index() * WebCarrierFailurePhase::ALL.len() + phase as usize) + * CarrierFailure::ALL.len() + + reason.index() +} + +const fn learning_index(carrier: WebCarrier, outcome: WebCarrierLearningOutcome) -> usize { + carrier.index() * WebCarrierLearningOutcome::ALL.len() + outcome as usize +} diff --git a/src/web/telemetry/tests.rs b/src/web/telemetry/tests.rs new file mode 100644 index 0000000..960c34b --- /dev/null +++ b/src/web/telemetry/tests.rs @@ -0,0 +1,59 @@ +use super::*; +use crate::config::WebCarrier; +use crate::web::manager::CarrierFailure; + +#[test] +fn fixed_counter_sets_and_acceptor_guard_are_exact() { + let telemetry = WebTelemetry::new(); + let guard = telemetry.acceptor_guard(); + assert_eq!(telemetry.live_acceptors(), 1); + telemetry.record_rejection(WebRejectionReason::HttpConnectionCapacity); + telemetry.record_overload(WebHttpConnectionOverloadOutcome::Dropped); + telemetry.record_decoy(WebDecoyUpstreamOutcome::ConnectRefused); + telemetry.record_carrier_selection( + WebCarrier::Https, + WebCarrierSelectionDisposition::Cold, + ); + telemetry.record_carrier_failure( + WebCarrier::Https, + WebCarrierFailurePhase::Provisional, + CarrierFailure::Network, + ); + telemetry.record_carrier_learning( + WebCarrier::Https, + WebCarrierLearningOutcome::Recorded, + ); + assert_eq!(telemetry.rejection_counters().len(), WebRejectionReason::ALL.len()); + assert_eq!( + telemetry.overload_counters().len(), + WebHttpConnectionOverloadOutcome::ALL.len() + ); + assert_eq!( + telemetry.decoy_counters().len(), + WebDecoyUpstreamOutcome::ALL.len() + ); + assert_eq!( + telemetry.carrier_selection_counters().len(), + WebCarrier::ALL.len() * WebCarrierSelectionDisposition::ALL.len() + ); + assert_eq!( + telemetry.carrier_failure_counters().len(), + WebCarrier::ALL.len() + * WebCarrierFailurePhase::ALL.len() + * CarrierFailure::ALL.len() + ); + assert_eq!( + telemetry.carrier_learning_counters().len(), + WebCarrier::ALL.len() * WebCarrierLearningOutcome::ALL.len() + ); + assert_eq!( + telemetry.rejection_total(WebRejectionReason::HttpConnectionCapacity), + 1 + ); + assert_eq!( + telemetry.last_decoy().map(|value| value.0), + Some("connect_refused") + ); + drop(guard); + assert_eq!(telemetry.live_acceptors(), 0); +}