From 20a4d505240ee5e31f749e56d7f08a435dafc3aa Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sun, 6 Sep 2026 18:51:13 +0300 Subject: [PATCH] Restart-gated Decoy Fast-Track Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/config_edit.rs | 46 ++++-- src/api/config_edit/tests.rs | 5 +- src/api/web_runtime/observability.rs | 13 +- src/config/hot_reload.rs | 2 + src/config/hot_reload/tests.rs | 75 ++++++++- .../tests/load_basic_tests/web_tests.rs | 17 +- src/config/types/web.rs | 2 +- src/maestro/runtime_build_tests.rs | 50 +++++- src/metrics/web/fasttrack.rs | 24 +-- src/web/http.rs | 26 ++-- src/web/http/capability.rs | 5 +- src/web/http/decoy_fasttrack_tests.rs | 146 ++++++++++++++---- src/web/http/request.rs | 7 +- src/web/http/request/tests.rs | 14 ++ src/web/http/response_test_support.rs | 18 +++ src/web/http/tests.rs | 47 ++---- src/web/telemetry.rs | 3 +- src/web/telemetry/fasttrack.rs | 13 +- src/web/telemetry/tests.rs | 1 - 19 files changed, 343 insertions(+), 171 deletions(-) create mode 100644 src/web/http/response_test_support.rs diff --git a/src/api/config_edit.rs b/src/api/config_edit.rs index cb9dfeb..7b087c6 100644 --- a/src/api/config_edit.rs +++ b/src/api/config_edit.rs @@ -15,20 +15,30 @@ use super::model::ApiFailure; use crate::config::ProxyConfig; use crate::config::hot_reload::classify_config_changes; use crate::maestro::reload::{ReloadAccepted, ReloadRequest, ReloadSubmitError}; -use crate::maestro::runtime_build::resolve_reload_config; +use crate::maestro::runtime_build::{ + ResolvedReloadConfig, deferred_process_fields, resolve_reload_config, +}; use serde::Serialize; use std::path::{Path, PathBuf}; use std::sync::Arc; +/// Result of one validated managed-config mutation. #[derive(Debug, Serialize)] pub(super) struct PatchConfigResponse { + /// Revision of the persisted desired configuration. pub revision: String, + /// Whether any changed field is not hot-reloadable. pub restart_required: bool, + /// Whether the effective runtime snapshot must be reloaded. pub runtime_reload_required: bool, + /// Whether any desired field remains deferred until process restart. pub process_restart_required: bool, + /// Stable paths of desired fields retained from the active process. pub deferred_process_fields: Vec, + /// Top-level managed sections changed by the mutation. pub changed: Vec, #[serde(skip_serializing_if = "Option::is_none")] + /// Accepted runtime reload when one was requested and required. pub reload: Option, } @@ -51,11 +61,11 @@ 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) - .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; + let resolved = reconcile_runtime_effect( + &mut prepared.response, + &active_config, + &prepared.desired_config, + )?; let reservation = if let Some(request) = reload_request.filter(|_| resolved.runtime_changed) { Some( shared @@ -79,6 +89,19 @@ pub(super) async fn patch_config( Ok(resp) } +fn reconcile_runtime_effect( + response: &mut PatchConfigResponse, + active_config: &ProxyConfig, + desired_config: &ProxyConfig, +) -> Result { + let resolved = + resolve_reload_config(active_config, desired_config).map_err(ApiFailure::bad_request)?; + response.runtime_reload_required = resolved.runtime_changed; + response.process_restart_required = !resolved.deferred_process_fields.is_empty(); + response.deferred_process_fields = resolved.deferred_process_fields.clone(); + Ok(resolved) +} + /// Core patch logic, decoupled from hyper/shared-state so it is unit-testable /// against a temp file. The route handler holds `mutation_lock` while calling this. #[cfg(test)] @@ -206,7 +229,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 resolved = resolve_reload_config(&old_cfg, &new_cfg).map_err(ApiFailure::bad_request)?; + let deferred_process_fields = + deferred_process_fields(&old_cfg, &new_cfg).map_err(ApiFailure::bad_request)?; Ok(PreparedConfigPatch { owner_path, @@ -215,9 +239,9 @@ async fn prepare_patch_to_path( response: PatchConfigResponse { revision, restart_required: class.restart_required, - runtime_reload_required: resolved.runtime_changed, - process_restart_required: !resolved.deferred_process_fields.is_empty(), - deferred_process_fields: resolved.deferred_process_fields, + runtime_reload_required: class.restart_required, + process_restart_required: !deferred_process_fields.is_empty(), + deferred_process_fields, changed: class.changed, reload: None, }, @@ -239,7 +263,7 @@ fn reload_submit_failure(error: ReloadSubmitError) -> ApiFailure { } } -/// Return only the editable config sections + current revision. +/// Returns only the editable config sections and current revision. pub(super) async fn read_managed_config(config_path: &Path) -> Result<(Toml, String), ApiFailure> { let loaded = load_config_snapshot(config_path, false).await?; let revision = compute_snapshot_revision(&loaded); diff --git a/src/api/config_edit/tests.rs b/src/api/config_edit/tests.rs index 4bbba8f..1143a31 100644 --- a/src/api/config_edit/tests.rs +++ b/src/api/config_edit/tests.rs @@ -134,11 +134,14 @@ async fn patch_web_debug_is_hot_and_limits_are_process_deferred() { #[tokio::test] async fn patch_web_decoy_fasttrack_requires_only_process_restart() { let (path, _directory) = temp_config("[web]\nenabled = false\n"); + let active = ProxyConfig::load(&path).unwrap(); let patch: Json = serde_json::json!({ "web": {"decoy_fasttrack_mode": "shadow"} }); - let response = apply_patch_to_path(&path, &patch, None).await.unwrap(); + let mut prepared = prepare_patch_to_path(&path, &patch, None).await.unwrap(); + reconcile_runtime_effect(&mut prepared.response, &active, &prepared.desired_config).unwrap(); + let response = prepared.response; assert!(response.restart_required); assert!(!response.runtime_reload_required); diff --git a/src/api/web_runtime/observability.rs b/src/api/web_runtime/observability.rs index a675b07..a571c78 100644 --- a/src/api/web_runtime/observability.rs +++ b/src/api/web_runtime/observability.rs @@ -127,7 +127,6 @@ impl WebDecoyUpstreamStatus { pub(super) struct WebDecoyFastTrackStatus { mode: WebDecoyFastTrackMode, requests: Vec, - shadow_mismatches_total: u64, } impl WebDecoyFastTrackStatus { @@ -136,9 +135,6 @@ impl WebDecoyFastTrackStatus { Self { mode: config.web.decoy_fasttrack_mode, requests: publication.telemetry.decoy_fasttrack_counters(), - shadow_mismatches_total: publication - .telemetry - .decoy_fasttrack_shadow_mismatches(), } } } @@ -209,11 +205,9 @@ 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 fasttrack = serde_json::to_value(super::WebDecoyFastTrackStatus::new( - &publication, - &config, - )) - .unwrap(); + let fasttrack = + serde_json::to_value(super::WebDecoyFastTrackStatus::new(&publication, &config)) + .unwrap(); let carrier = serde_json::to_value(super::WebCarrierNegotiationStatus::new(&publication)).unwrap(); let lifecycle = serde_json::to_value(super::WebLifecycleCountersStatus::new( @@ -242,7 +236,6 @@ mod tests { fasttrack["requests"].as_array().unwrap().len(), crate::web::telemetry::WebDecoyFastTrackDisposition::ALL.len() ); - assert_eq!(fasttrack["shadow_mismatches_total"], 0); assert_eq!(capacity["partial"][0], "runtime"); assert_eq!( carrier["selections"].as_array().unwrap().len(), diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 2c4407b..1a5f511 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -21,6 +21,8 @@ //! `server.port`, `censorship.*`, `network.*`, `use_middle_proxy`) are **not** //! applied; a warning is emitted. SYN limiter rules are process-owned and are //! reconciled only during privileged startup. +//! `web.decoy_fasttrack_mode` is also restart-only so one process never mixes +//! capability timing policies or process-lifetime counter semantics. //! Non-hot changes are never mixed into the runtime config snapshot. use std::collections::BTreeSet; diff --git a/src/config/hot_reload/tests.rs b/src/config/hot_reload/tests.rs index 802935b..4741f10 100644 --- a/src/config/hot_reload/tests.rs +++ b/src/config/hot_reload/tests.rs @@ -47,6 +47,43 @@ fn write_web_reload_config(path: &Path, carriers: &str, carrier_learning: bool) std::fs::write(path, config).unwrap(); } +fn write_web_fasttrack_reload_config(path: &Path, mode: &str, ad_tag: &str) { + let config = format!( + r#" + [general] + ad_tag = "{ad_tag}" + + [access.users] + alice = "000102030405060708090a0b0c0d0e0f" + + [[server.listeners]] + ip = "127.0.0.1" + port = 18080 + transport = "web" + proxy_protocol = false + web_client_ip_source = "x_forwarded_for" + web_trusted_proxy_cidrs = ["127.0.0.1/32"] + + [web] + enabled = true + decoy_fasttrack_mode = "{mode}" + + [[web.vhosts]] + host = "proxy.example.com" + public_addr = "203.0.113.10:443" + + [web.vhosts.decoy] + mode = "http_upstream" + upstream = "http://127.0.0.1:18081" + + [[web.vhosts.profiles]] + user = "alice" + secret_mode = "plain" + "#, + ); + std::fs::write(path, config).unwrap(); +} + fn temp_config_path(prefix: &str) -> PathBuf { let nonce = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -135,7 +172,10 @@ fn decoy_fasttrack_mode_is_deferred_until_restart() { applied.web.decoy_fasttrack_mode, old.web.decoy_fasttrack_mode ); - assert_eq!(HotFields::from_config(&old), HotFields::from_config(&applied)); + assert_eq!( + HotFields::from_config(&old), + HotFields::from_config(&applied) + ); } #[test] @@ -378,6 +418,39 @@ fn reload_keeps_hot_apply_when_non_hot_fields_change() { let _ = std::fs::remove_file(path); } +#[test] +fn reload_rebuilds_vhosts_with_the_effective_fasttrack_mode() { + let initial_tag = "abababababababababababababababab"; + let final_tag = "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd"; + let path = temp_config_path("telemt_web_fasttrack_reload"); + + write_web_fasttrack_reload_config(&path, "off", initial_tag); + let initial_cfg = Arc::new(ProxyConfig::load(&path).unwrap()); + let initial_hash = ProxyConfig::load_with_metadata(&path) + .unwrap() + .rendered_hash; + let (config_tx, _config_rx) = watch::channel(Arc::clone(&initial_cfg)); + let (log_tx, _log_rx) = watch::channel(initial_cfg.general.log_level.clone()); + let mut reload_state = ReloadState::new(Some(initial_hash)); + + write_web_fasttrack_reload_config(&path, "enforce", final_tag); + reload_config(&path, &config_tx, &log_tx, None, None, &mut reload_state).unwrap(); + + let applied = config_tx.borrow().clone(); + assert_eq!(applied.general.ad_tag.as_deref(), Some(final_tag)); + assert_eq!( + applied.web.decoy_fasttrack_mode, + crate::config::WebDecoyFastTrackMode::Off + ); + let runtime = applied.web.runtime.as_ref().unwrap(); + assert_eq!( + runtime.vhosts["proxy.example.com"].decoy_fasttrack_mode, + crate::config::WebDecoyFastTrackMode::Off + ); + + let _ = std::fs::remove_file(path); +} + #[test] fn reload_publishes_web_negotiation_policy_outside_hot_field_reporting() { let path = temp_config_path("telemt_web_negotiation_reload"); diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index 665b953..b18d004 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -43,10 +43,7 @@ fn web_config_builds_canonical_runtime_snapshot() { assert_eq!(vhost.profiles.len(), 1); assert_eq!(vhost.capabilities.len(), vhost.profiles.len()); assert_eq!(vhost.capabilities[0], vhost.profiles[0].capability); - assert_eq!( - vhost.decoy_fasttrack_mode, - WebDecoyFastTrackMode::Off - ); + assert_eq!(vhost.decoy_fasttrack_mode, WebDecoyFastTrackMode::Off); assert_eq!(vhost.profiles[0].user, "alice"); assert_eq!(vhost.profiles[0].secret_mode, WebSecretMode::Dd); assert_eq!(vhost.profiles[0].carrier, WebCarrier::HttpsLanes); @@ -76,20 +73,12 @@ fn web_decoy_fasttrack_mode_is_typed_and_defaults_off() { ] { let configured = WEB_CONFIG.replace( "carrier = \"https-lanes\"", - &format!( - "carrier = \"https-lanes\"\ndecoy_fasttrack_mode = \"{token}\"" - ), + &format!("carrier = \"https-lanes\"\ndecoy_fasttrack_mode = \"{token}\""), ); let config = load_config_from_temp_toml(&configured); assert_eq!(config.web.decoy_fasttrack_mode, expected); assert_eq!( - config - .web - .runtime - .as_ref() - .unwrap() - .vhosts["proxy.example.com"] - .decoy_fasttrack_mode, + config.web.runtime.as_ref().unwrap().vhosts["proxy.example.com"].decoy_fasttrack_mode, expected ); } diff --git a/src/config/types/web.rs b/src/config/types/web.rs index 49592e1..70fb062 100644 --- a/src/config/types/web.rs +++ b/src/config/types/web.rs @@ -203,7 +203,7 @@ pub struct WebLimitsConfig { /// Process-wide retained and in-flight WEB debug byte ceiling. #[serde(default = "default_web_debug_bytes_global")] pub debug_bytes_global: usize, - /// Declared process envelope for HTTP, queues, lane state, learning, and static snapshots. + /// Declared process envelope for HTTP, queues, capabilities, learning, and static snapshots. #[serde(default = "default_web_memory_envelope_bytes")] pub memory_envelope_bytes: usize, /// Sustained process-wide bootstrap issuance rate. diff --git a/src/maestro/runtime_build_tests.rs b/src/maestro/runtime_build_tests.rs index 7fe965b..babaa54 100644 --- a/src/maestro/runtime_build_tests.rs +++ b/src/maestro/runtime_build_tests.rs @@ -24,6 +24,43 @@ fn test_listener(port: u16) -> crate::config::ListenerConfig { } } +fn web_config_with_fasttrack(mode: &str) -> ProxyConfig { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.toml"); + let config = format!( + r#" +[access.users] +alice = "000102030405060708090a0b0c0d0e0f" + +[[server.listeners]] +ip = "127.0.0.1" +port = 18080 +transport = "web" +proxy_protocol = false +web_client_ip_source = "x_forwarded_for" +web_trusted_proxy_cidrs = ["127.0.0.1/32"] + +[web] +enabled = true +decoy_fasttrack_mode = "{mode}" + +[[web.vhosts]] +host = "proxy.example.com" +public_addr = "203.0.113.10:443" + +[web.vhosts.decoy] +mode = "http_upstream" +upstream = "http://127.0.0.1:18081" + +[[web.vhosts.profiles]] +user = "alice" +secret_mode = "plain" +"#, + ); + std::fs::write(&path, config).unwrap(); + ProxyConfig::load(path).unwrap() +} + #[test] fn process_socket_and_logging_changes_are_deferred() { let old = ProxyConfig::default(); @@ -172,11 +209,8 @@ fn web_allocation_limits_are_deferred_until_restart() { #[test] fn web_decoy_fasttrack_mode_is_deferred_without_runtime_publication() { - let mut old = ProxyConfig::default(); - old.rebuild_runtime_user_auth().unwrap(); - old.rebuild_runtime_web().unwrap(); - let mut desired = old.clone(); - desired.web.decoy_fasttrack_mode = crate::config::WebDecoyFastTrackMode::Enforce; + let old = web_config_with_fasttrack("off"); + let desired = web_config_with_fasttrack("enforce"); let resolved = resolve_reload_config(&old, &desired).unwrap(); @@ -188,6 +222,12 @@ fn web_decoy_fasttrack_mode_is_deferred_without_runtime_publication() { resolved.effective.web.decoy_fasttrack_mode, old.web.decoy_fasttrack_mode ); + let effective_runtime = resolved.effective.web.runtime.as_ref().unwrap(); + let effective_vhost = &effective_runtime.vhosts["proxy.example.com"]; + assert_eq!( + effective_vhost.decoy_fasttrack_mode, + old.web.decoy_fasttrack_mode + ); assert!(!resolved.runtime_changed); } diff --git a/src/metrics/web/fasttrack.rs b/src/metrics/web/fasttrack.rs index 77525a0..966b602 100644 --- a/src/metrics/web/fasttrack.rs +++ b/src/metrics/web/fasttrack.rs @@ -5,11 +5,7 @@ use crate::web::control::WebRuntimePublication; use crate::web::telemetry::WebDecoyFastTrackDisposition; /// Renders fixed-cardinality decoy capability-routing metrics. -pub(super) fn render( - out: &mut String, - publication: &WebRuntimePublication, - config: &ProxyConfig, -) { +pub(super) fn render(out: &mut String, publication: &WebRuntimePublication, config: &ProxyConfig) { let _ = writeln!( out, "# HELP telemt_web_decoy_fasttrack_mode Effective restart-frozen decoy fast-track mode" @@ -40,20 +36,6 @@ pub(super) fn render( publication.telemetry.decoy_fasttrack_total(disposition) ); } - - let _ = writeln!( - out, - "# HELP telemt_web_decoy_fasttrack_shadow_mismatches_total Shadow decisions that disagreed with legacy bridge eligibility" - ); - let _ = writeln!( - out, - "# TYPE telemt_web_decoy_fasttrack_shadow_mismatches_total counter" - ); - let _ = writeln!( - out, - "telemt_web_decoy_fasttrack_shadow_mismatches_total {}", - publication.telemetry.decoy_fasttrack_shadow_mismatches() - ); } #[cfg(test)] @@ -67,9 +49,6 @@ mod tests { control .telemetry() .record_decoy_fasttrack(WebDecoyFastTrackDisposition::EnforceFastTrack); - control - .telemetry() - .record_decoy_fasttrack_shadow_mismatch(); let publication = control.subscribe().borrow().clone(); let mut config = ProxyConfig::default(); config.web.decoy_fasttrack_mode = WebDecoyFastTrackMode::Enforce; @@ -88,6 +67,5 @@ mod tests { assert!(output.contains( "telemt_web_decoy_fasttrack_requests_total{disposition=\"enforce_fasttrack\"} 1" )); - assert!(output.contains("telemt_web_decoy_fasttrack_shadow_mismatches_total 1")); } } diff --git a/src/web/http.rs b/src/web/http.rs index 69f95a1..7a383e2 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -241,11 +241,13 @@ async fn handle_root( match fasttrack_mode { WebDecoyFastTrackMode::Off => {} WebDecoyFastTrackMode::Shadow => { - runtime.telemetry().record_decoy_fasttrack(if plausible_candidate { - WebDecoyFastTrackDisposition::ShadowCandidateFullScan - } else { - WebDecoyFastTrackDisposition::ShadowWouldFastTrack - }); + runtime + .telemetry() + .record_decoy_fasttrack(if plausible_candidate { + WebDecoyFastTrackDisposition::ShadowCandidateFullScan + } else { + WebDecoyFastTrackDisposition::ShadowWouldFastTrack + }); } WebDecoyFastTrackMode::Enforce if !plausible_candidate => { runtime @@ -259,22 +261,14 @@ async fn handle_root( return serve_decoy(request, vhost, recovery_requested, &runtime).await; } WebDecoyFastTrackMode::Enforce => { - runtime.telemetry().record_decoy_fasttrack( - WebDecoyFastTrackDisposition::EnforceCandidateFullScan, - ); + runtime + .telemetry() + .record_decoy_fasttrack(WebDecoyFastTrackDisposition::EnforceCandidateFullScan); } } let matched_profile = match_profile(&vhost, candidate.scan_bytes()); let recovery_requested = matches!(representation, recovery::RootRepresentation::Recovery(_)); let profile = matched_profile.filter(|_| canonical && request.method() == Method::GET); - if fasttrack_mode == WebDecoyFastTrackMode::Shadow - && !plausible_candidate - && profile.is_some() - { - runtime - .telemetry() - .record_decoy_fasttrack_shadow_mismatch(); - } let Some(profile) = profile else { if recovery_requested { strip_query(&mut request); diff --git a/src/web/http/capability.rs b/src/web/http/capability.rs index f8fc353..b781e2f 100644 --- a/src/web/http/capability.rs +++ b/src/web/http/capability.rs @@ -68,10 +68,7 @@ pub(super) struct CapabilityScan { } /// Scans every configured capability without candidate-dependent control flow. -pub(super) fn scan_capabilities( - capabilities: &[[u8; 32]], - candidate: &[u8; 32], -) -> CapabilityScan { +pub(super) fn scan_capabilities(capabilities: &[[u8; 32]], candidate: &[u8; 32]) -> CapabilityScan { let mut matched = Choice::from(0); let mut matched_index = 0u64; #[cfg(test)] diff --git a/src/web/http/decoy_fasttrack_tests.rs b/src/web/http/decoy_fasttrack_tests.rs index daefbcd..038eaa5 100644 --- a/src/web/http/decoy_fasttrack_tests.rs +++ b/src/web/http/decoy_fasttrack_tests.rs @@ -8,7 +8,20 @@ const RECOVERY_TYPE: &str = "application/vnd.telemt.web-recovery+json"; struct Observation { response: Vec, counters: [u64; WebDecoyFastTrackDisposition::ALL.len()], - shadow_mismatches: u64, +} + +fn runtime_config_with_fasttrack( + capability: [u8; 32], + carrier: WebCarrier, + mode: WebDecoyFastTrackMode, +) -> ProxyConfig { + let mut config = runtime_config(capability, carrier); + config.web.decoy_fasttrack_mode = mode; + let runtime = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); + for vhost in runtime.vhosts.values_mut() { + Arc::get_mut(vhost).unwrap().decoy_fasttrack_mode = mode; + } + config } async fn capture_origin_request(listener: &TcpListener) -> Vec { @@ -17,10 +30,12 @@ async fn capture_origin_request(listener: &TcpListener) -> Vec { let mut buffer = [0u8; 4096]; loop { let read = stream.read(&mut buffer).await.unwrap(); - assert_ne!(read, 0, "decoy origin connection closed before the request completed"); + assert_ne!( + read, 0, + "decoy origin connection closed before the request completed" + ); captured.extend_from_slice(&buffer[..read]); - let Some(header_end) = captured.windows(4).position(|window| window == b"\r\n\r\n") - else { + let Some(header_end) = captured.windows(4).position(|window| window == b"\r\n\r\n") else { continue; }; let headers = std::str::from_utf8(&captured[..header_end]).unwrap(); @@ -51,13 +66,7 @@ async fn observe_http_origin( ) -> (Observation, Vec) { let mut config = runtime_config_with_fasttrack(capability, WebCarrier::Https, mode); let runtime_config = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); - let vhost = Arc::get_mut( - runtime_config - .vhosts - .get_mut("proxy.example.com") - .unwrap(), - ) - .unwrap(); + let vhost = Arc::get_mut(runtime_config.vhosts.get_mut("proxy.example.com").unwrap()).unwrap(); vhost.decoy = WebRuntimeDecoy::HttpUpstream { addr: origin.local_addr().unwrap(), authority: "decoy.example".to_string(), @@ -72,23 +81,19 @@ async fn observe_http_origin( ); let counters = WebDecoyFastTrackDisposition::ALL .map(|disposition| runtime.telemetry().decoy_fasttrack_total(disposition)); - let shadow_mismatches = runtime.telemetry().decoy_fasttrack_shadow_mismatches(); runtime.shutdown().await; generation.stop_sessions().await; generation.stop_background_tasks().await; - ( - Observation { - response, - counters, - shadow_mismatches, - }, - captured, - ) + (Observation { response, counters }, captured) } -async fn observe(mode: WebDecoyFastTrackMode, capability: [u8; 32], request_bytes: Vec) -> Observation { +async fn observe( + mode: WebDecoyFastTrackMode, + capability: [u8; 32], + request_bytes: Vec, +) -> Observation { let generation = test_runtime_generation( 1, runtime_config_with_fasttrack(capability, WebCarrier::Https, mode), @@ -99,17 +104,12 @@ async fn observe(mode: WebDecoyFastTrackMode, capability: [u8; 32], request_byte let response = request(&listener, &runtime, request_bytes).await; let counters = WebDecoyFastTrackDisposition::ALL .map(|disposition| runtime.telemetry().decoy_fasttrack_total(disposition)); - let shadow_mismatches = runtime.telemetry().decoy_fasttrack_shadow_mismatches(); runtime.shutdown().await; generation.stop_sessions().await; generation.stop_background_tasks().await; - Observation { - response, - counters, - shadow_mismatches, - } + Observation { response, counters } } fn root_request(method: &str, query: &str) -> Vec { @@ -128,7 +128,12 @@ async fn impossible_root_shapes_preserve_decoy_bytes_and_follow_the_selected_mod root_request("GET", "?bridge=not-canonical"), root_request("HEAD", &format!("?bridge={encoded}")), ] { - let off = observe(WebDecoyFastTrackMode::Off, capability, request_bytes.clone()).await; + let off = observe( + WebDecoyFastTrackMode::Off, + capability, + request_bytes.clone(), + ) + .await; let shadow = observe( WebDecoyFastTrackMode::Shadow, capability, @@ -142,7 +147,6 @@ async fn impossible_root_shapes_preserve_decoy_bytes_and_follow_the_selected_mod assert_eq!(off.counters, [0, 0, 0, 0]); assert_eq!(shadow.counters, [1, 0, 0, 0]); assert_eq!(enforce.counters, [0, 0, 1, 0]); - assert_eq!(shadow.shadow_mismatches, 0); } } @@ -161,7 +165,6 @@ async fn canonical_hit_and_miss_always_retain_the_full_scan() { let observation = observe(mode, capability, request_bytes.clone()).await; assert!(observation.response.starts_with(b"HTTP/1.1 200")); assert_eq!(observation.counters, expected_counters); - assert_eq!(observation.shadow_mismatches, 0); if candidate == capability { assert!( observation @@ -186,7 +189,12 @@ async fn recovery_sanitization_precedes_fasttrack_classification() { ) .into_bytes(); - let off = observe(WebDecoyFastTrackMode::Off, capability, request_bytes.clone()).await; + let off = observe( + WebDecoyFastTrackMode::Off, + capability, + request_bytes.clone(), + ) + .await; let shadow = observe( WebDecoyFastTrackMode::Shadow, capability, @@ -197,11 +205,42 @@ async fn recovery_sanitization_precedes_fasttrack_classification() { assert_eq!(shadow.response, off.response); assert_eq!(enforce.response, off.response); - assert_eq!(response_header(split_response(&off.response).0, "cache-control"), "no-store"); + assert_eq!( + response_header(split_response(&off.response).0, "cache-control"), + "no-store" + ); assert_eq!(off.counters, [0, 0, 0, 0]); assert_eq!(shadow.counters, [1, 0, 0, 0]); assert_eq!(enforce.counters, [0, 0, 1, 0]); - assert_eq!(shadow.shadow_mismatches, 0); +} + +#[tokio::test] +async fn canonical_recovery_hit_and_miss_always_retain_the_full_scan() { + let capability = [77u8; 32]; + let bearer = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode([78u8; 32]); + for candidate in [capability, [79u8; 32]] { + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(candidate); + let request_bytes = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.90\r\nAccept: {RECOVERY_TYPE}\r\nAuthorization: Bearer {bearer}\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + + for (mode, expected_counters) in [ + (WebDecoyFastTrackMode::Off, [0, 0, 0, 0]), + (WebDecoyFastTrackMode::Shadow, [0, 1, 0, 0]), + (WebDecoyFastTrackMode::Enforce, [0, 0, 0, 1]), + ] { + let observation = observe(mode, capability, request_bytes.clone()).await; + let (headers, body) = split_response(&observation.response); + assert_eq!(observation.counters, expected_counters); + if candidate == capability { + assert_eq!(response_header(headers, "content-type"), RECOVERY_TYPE); + assert!(serde_json::from_slice::(body).is_ok()); + } else { + assert_eq!(body, b"decoy"); + } + } + } } #[tokio::test] @@ -238,7 +277,11 @@ async fn http_origin_forwarding_is_identical_across_modes() { assert_eq!(shadow_upstream, off_upstream); assert_eq!(enforce_upstream, off_upstream); assert!(off_upstream.starts_with(b"GET /?bridge=not-canonical HTTP/1.1\r\n")); - assert!(off_upstream.windows(21).any(|window| window == b"x-ordinary: preserved")); + assert!( + off_upstream + .windows(21) + .any(|window| window == b"x-ordinary: preserved") + ); assert!(off_upstream.ends_with(b"\r\n\r\nbody")); } @@ -291,3 +334,38 @@ async fn http_origin_recovery_sanitization_is_identical_across_modes() { } assert!(off_upstream.ends_with(b"\r\n\r\n")); } + +#[tokio::test] +async fn fasttrack_counters_remain_process_owned_across_generation_swap() { + let capability = [76u8; 32]; + let initial = test_runtime_generation( + 1, + runtime_config_with_fasttrack(capability, WebCarrier::Https, WebDecoyFastTrackMode::Shadow), + ); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + + let first = request(&listener, &runtime, root_request("GET", "")).await; + assert!(first.starts_with(b"HTTP/1.1 200")); + let replacement = test_runtime_generation( + 2, + runtime_config_with_fasttrack(capability, WebCarrier::Https, WebDecoyFastTrackMode::Shadow), + ); + active_runtime.store(Arc::clone(&replacement)); + let second = request(&listener, &runtime, root_request("GET", "")).await; + assert!(second.starts_with(b"HTTP/1.1 200")); + + assert_eq!( + runtime + .telemetry() + .decoy_fasttrack_total(WebDecoyFastTrackDisposition::ShadowWouldFastTrack), + 2 + ); + + runtime.shutdown().await; + initial.stop_sessions().await; + initial.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} diff --git a/src/web/http/request.rs b/src/web/http/request.rs index 9e2a1c5..aa09587 100644 --- a/src/web/http/request.rs +++ b/src/web/http/request.rs @@ -24,12 +24,17 @@ pub(super) fn match_profile( vhost: &WebRuntimeVhost, candidate: &[u8; 32], ) -> Option> { - debug_assert_eq!(vhost.capabilities.len(), vhost.profiles.len()); + // Runtime construction owns alignment; any future constructor drift must fail closed. + if vhost.capabilities.len() != vhost.profiles.len() { + return None; + } let scan = scan_capabilities(&vhost.capabilities, candidate); if bool::from(scan.matched) { usize::try_from(scan.matched_index) .ok() .and_then(|index| vhost.profiles.get(index)) + // Revalidate the selected identity after the index-only complete scan. + .filter(|profile| bool::from(profile.capability.ct_eq(candidate))) .map(Arc::clone) } else { None diff --git a/src/web/http/request/tests.rs b/src/web/http/request/tests.rs index e43665d..97a5a88 100644 --- a/src/web/http/request/tests.rs +++ b/src/web/http/request/tests.rs @@ -32,6 +32,20 @@ fn capability_scan_checks_every_entry_independent_of_match_position() { assert_eq!(empty.comparisons, 0); } +#[test] +fn capability_profile_table_drift_fails_closed() { + let capability = [5u8; 32]; + let mut config = crate::web::http::tests::runtime_config(capability, WebCarrier::Https); + let runtime = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); + let vhost = Arc::get_mut(runtime.vhosts.get_mut("proxy.example.com").unwrap()).unwrap(); + + assert!(match_profile(vhost, &capability).is_some()); + vhost.capabilities[0] = [6u8; 32]; + assert!(match_profile(vhost, &[6u8; 32]).is_none()); + vhost.capabilities = Vec::new().into_boxed_slice(); + assert!(match_profile(vhost, &capability).is_none()); +} + proptest! { #[test] fn every_capability_has_one_canonical_bridge_query(capability in any::<[u8; 32]>()) { diff --git a/src/web/http/response_test_support.rs b/src/web/http/response_test_support.rs new file mode 100644 index 0000000..ef6bc41 --- /dev/null +++ b/src/web/http/response_test_support.rs @@ -0,0 +1,18 @@ +/// Splits one complete raw HTTP response into head and body slices. +pub(in crate::web::http) fn split_response(response: &[u8]) -> (&[u8], &[u8]) { + let separator = response + .windows(4) + .position(|window| window == b"\r\n\r\n") + .unwrap(); + (&response[..separator], &response[separator + 4..]) +} + +/// Returns one case-insensitive header value from a raw HTTP response head. +pub(in crate::web::http) fn response_header<'a>(headers: &'a [u8], name: &str) -> &'a str { + std::str::from_utf8(headers) + .unwrap() + .lines() + .filter_map(|line| line.split_once(':')) + .find_map(|(header, value)| header.eq_ignore_ascii_case(name).then_some(value.trim())) + .unwrap() +} diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index c9545e2..387d200 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -10,9 +10,9 @@ use tokio_util::sync::CancellationToken; use super::serve_connection; use crate::config::{ - ProxyConfig, WebCarrier, WebCarriers, WebClientIpSource, WebRuntimeConfig, WebRuntimeDecoy, - WebDecoyFastTrackMode, WebRuntimeProfile, WebRuntimeVhost, WebSecretMode, WebStaticAsset, - WebStaticSite, + ProxyConfig, WebCarrier, WebCarriers, WebClientIpSource, WebDecoyFastTrackMode, + WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeVhost, WebSecretMode, + WebStaticAsset, WebStaticSite, }; use crate::maestro::generation::test_runtime_generation; use crate::web::frame::{self, FrameType}; @@ -43,28 +43,20 @@ mod recovery_tests; // Decoy fast-track routing and telemetry remain isolated from carrier protocol scenarios. #[path = "decoy_fasttrack_tests.rs"] mod decoy_fasttrack_tests; +// Raw response parsing helpers are shared by the HTTP integration test modules. +#[path = "response_test_support.rs"] +mod response_test_support; + +pub(super) use response_test_support::{response_header, split_response}; const TEST_CARRIER_DEADLINES_SECS: [u64; 4] = [3, 5, 8, 12]; +/// Builds the default static-decoy runtime used by WEB HTTP tests. pub(super) fn runtime_config(capability: [u8; 32], carrier: WebCarrier) -> ProxyConfig { runtime_config_with_carriers(capability, carrier, false, true, Arc::from([carrier])) } -/// Builds a static-decoy runtime with one restart-frozen fast-track mode. -pub(super) fn runtime_config_with_fasttrack( - capability: [u8; 32], - carrier: WebCarrier, - mode: WebDecoyFastTrackMode, -) -> ProxyConfig { - let mut config = runtime_config(capability, carrier); - config.web.decoy_fasttrack_mode = mode; - let runtime = Arc::get_mut(config.web.runtime.as_mut().unwrap()).unwrap(); - for vhost in runtime.vhosts.values_mut() { - Arc::get_mut(vhost).unwrap().decoy_fasttrack_mode = mode; - } - config -} - +/// Builds a negotiation-enabled static-decoy runtime for carrier tests. pub(super) fn negotiation_runtime_config( capability: [u8; 32], carrier: WebCarrier, @@ -91,6 +83,7 @@ fn runtime_config_with_carriers( ) } +/// Builds a negotiation runtime with explicit cumulative carrier deadlines. pub(super) fn negotiation_runtime_config_with_deadlines( capability: [u8; 32], carrier: WebCarrier, @@ -185,6 +178,7 @@ fn runtime_config_with_carriers_and_deadlines( config } +/// Serves one raw HTTP request through the private WEB listener test harness. pub(super) async fn request( listener: &TcpListener, runtime: &Arc, @@ -211,23 +205,6 @@ pub(super) async fn request( response } -pub(super) fn split_response(response: &[u8]) -> (&[u8], &[u8]) { - let separator = response - .windows(4) - .position(|window| window == b"\r\n\r\n") - .unwrap(); - (&response[..separator], &response[separator + 4..]) -} - -pub(super) fn response_header<'a>(headers: &'a [u8], name: &str) -> &'a str { - std::str::from_utf8(headers) - .unwrap() - .lines() - .filter_map(|line| line.split_once(':')) - .find_map(|(header, value)| header.eq_ignore_ascii_case(name).then_some(value.trim())) - .unwrap() -} - #[tokio::test] async fn https_carrier_bootstraps_and_closes_one_session() { let capability = [7u8; 32]; diff --git a/src/web/telemetry.rs b/src/web/telemetry.rs index 0145382..2648cf7 100644 --- a/src/web/telemetry.rs +++ b/src/web/telemetry.rs @@ -10,6 +10,7 @@ pub(crate) use carrier::{ WebCarrierFailureCounter, WebCarrierFailurePhase, WebCarrierLearningCounter, WebCarrierLearningOutcome, WebCarrierSelectionCounter, WebCarrierSelectionDisposition, }; +// Decoy fast-track counters retain a fixed process-owned disposition set. mod fasttrack; use fasttrack::DECOY_FASTTRACK_SLOTS; pub(crate) use fasttrack::{WebDecoyFastTrackCounter, WebDecoyFastTrackDisposition}; @@ -320,7 +321,6 @@ pub(crate) struct WebTelemetry { carrier_failures: [AtomicU64; CARRIER_FAILURE_SLOTS], carrier_learning_outcomes: [AtomicU64; CARRIER_LEARNING_SLOTS], decoy_fasttrack_requests: [AtomicU64; DECOY_FASTTRACK_SLOTS], - decoy_fasttrack_shadow_mismatches: AtomicU64, session_closures: [AtomicU64; SESSION_CLOSE_SLOTS], session_observations: [AtomicU64; SESSION_OBSERVATION_SLOTS], bridge_recovery_events: [AtomicU64; WebBridgeRecoveryEvent::ALL.len()], @@ -349,7 +349,6 @@ impl WebTelemetry { carrier_failures: std::array::from_fn(|_| AtomicU64::new(0)), carrier_learning_outcomes: std::array::from_fn(|_| AtomicU64::new(0)), decoy_fasttrack_requests: std::array::from_fn(|_| AtomicU64::new(0)), - decoy_fasttrack_shadow_mismatches: AtomicU64::new(0), session_closures: std::array::from_fn(|_| AtomicU64::new(0)), session_observations: std::array::from_fn(|_| AtomicU64::new(0)), bridge_recovery_events: std::array::from_fn(|_| AtomicU64::new(0)), diff --git a/src/web/telemetry/fasttrack.rs b/src/web/telemetry/fasttrack.rs index 3b1f776..a84cf14 100644 --- a/src/web/telemetry/fasttrack.rs +++ b/src/web/telemetry/fasttrack.rs @@ -38,6 +38,7 @@ impl WebDecoyFastTrackDisposition { } } +/// Fixed storage width for process-owned decoy fast-track counters. pub(super) const DECOY_FASTTRACK_SLOTS: usize = WebDecoyFastTrackDisposition::ALL.len(); /// API-safe fixed decoy fast-track counter. @@ -70,16 +71,4 @@ impl WebTelemetry { }) .collect() } - - /// Records a shadow decision that disagreed with legacy bridge eligibility. - pub(crate) fn record_decoy_fasttrack_shadow_mismatch(&self) { - self.decoy_fasttrack_shadow_mismatches - .fetch_add(1, Ordering::Relaxed); - } - - /// Returns shadow decisions that disagreed with legacy bridge eligibility. - pub(crate) fn decoy_fasttrack_shadow_mismatches(&self) -> u64 { - self.decoy_fasttrack_shadow_mismatches - .load(Ordering::Relaxed) - } } diff --git a/src/web/telemetry/tests.rs b/src/web/telemetry/tests.rs index 613f6b4..406dba3 100644 --- a/src/web/telemetry/tests.rs +++ b/src/web/telemetry/tests.rs @@ -57,7 +57,6 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() { telemetry.decoy_fasttrack_total(WebDecoyFastTrackDisposition::ShadowWouldFastTrack), 1 ); - assert_eq!(telemetry.decoy_fasttrack_shadow_mismatches(), 0); assert_eq!( telemetry.session_close_counters().len(), WebCarrier::ALL.len() * SessionCloseReason::ALL.len()