From cf0cd08387f5ff07a46f786739f1ee913495af94 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:42:14 +0300 Subject: [PATCH] WEB Knobs in API Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/config_edit/tests.rs | 58 ++ src/api/config_store.rs | 13 +- src/api/mod.rs | 45 +- src/api/web_runtime.rs | 444 ++++++++++++++ src/api/web_runtime/request.rs | 391 ++++++++++++ src/api/web_status.rs | 6 +- src/api/web_status/tests.rs | 27 +- src/config/hot_reload/tests.rs | 40 ++ src/config/hot_reload/watcher.rs | 52 +- src/config/load/validate_web.rs | 5 + .../tests/load_basic_tests/web_tests.rs | 11 + src/maestro/listeners/control.rs | 59 +- src/maestro/orchestrator.rs | 12 +- src/maestro/reload_supervisor.rs | 12 +- src/maestro/reload_supervisor_tests.rs | 29 +- src/maestro/runtime_build.rs | 8 + src/maestro/runtime_startup.rs | 1 + src/maestro/runtime_tasks.rs | 18 +- src/web/control.rs | 153 +++++ src/web/http.rs | 15 +- src/web/http/activity.rs | 30 +- src/web/http/control_tests.rs | 103 ++++ src/web/http/session.rs | 5 +- src/web/http/session_policy_tests.rs | 55 ++ src/web/http/tests.rs | 28 +- src/web/manager.rs | 32 +- src/web/manager/budget.rs | 26 + src/web/manager/carrier_learning.rs | 90 +++ src/web/manager/carrier_learning/tests.rs | 49 ++ src/web/manager/control.rs | 441 ++++++++++++++ src/web/manager/credentials.rs | 82 ++- src/web/manager/lifecycle.rs | 63 +- src/web/manager/session_creation.rs | 74 ++- .../manager/session_creation/replacement.rs | 10 +- src/web/manager/state.rs | 116 +++- src/web/manager/status.rs | 560 ++++++++++++++++++ src/web/manager/websocket.rs | 20 + src/web/mod.rs | 2 + src/web/session.rs | 10 + src/web/session/status.rs | 135 +++++ src/web/trace/mod.rs | 4 +- src/web/trace/store.rs | 150 ++--- src/web/trace/store/tests.rs | 111 ++++ 43 files changed, 3399 insertions(+), 196 deletions(-) create mode 100644 src/api/web_runtime.rs create mode 100644 src/api/web_runtime/request.rs create mode 100644 src/web/control.rs create mode 100644 src/web/http/control_tests.rs create mode 100644 src/web/manager/control.rs create mode 100644 src/web/manager/status.rs create mode 100644 src/web/session/status.rs create mode 100644 src/web/trace/store/tests.rs diff --git a/src/api/config_edit/tests.rs b/src/api/config_edit/tests.rs index f738449..816718b 100644 --- a/src/api/config_edit/tests.rs +++ b/src/api/config_edit/tests.rs @@ -87,6 +87,64 @@ async fn read_managed_config_strips_access() { ); } +#[tokio::test] +async fn read_managed_config_exposes_web_without_runtime_or_access_secrets() { + let (path, _directory) = temp_config(concat!( + "[web]\nenabled = false\ncarrier = \"https\"\n", + "[web.debug]\nenabled = true\ndefault_window_secs = 180\n", + "[access.users]\nbob = \"00000000000000000000000000000000\"\n", + )); + + let (value, _revision) = read_managed_config(&path).await.unwrap(); + let table = value.as_table().unwrap(); + + assert!(table.contains_key("web")); + assert!(table["web"].get("debug").is_some()); + assert!(table["web"].get("runtime").is_none()); + assert!(!table.contains_key("access")); +} + +#[tokio::test] +async fn patch_web_debug_is_hot_and_limits_are_process_deferred() { + let (path, _directory) = temp_config("[web]\nenabled = false\n"); + let debug_patch: Json = serde_json::json!({ + "web": {"debug": {"enabled": true, "capture_headers": false}} + }); + let debug = apply_patch_to_path(&path, &debug_patch, None) + .await + .unwrap(); + assert!(!debug.process_restart_required); + assert!(debug.changed.iter().any(|section| section == "web")); + + let limits_patch: Json = serde_json::json!({ + "web": {"limits": {"max_http_connections": 2049}} + }); + let limits = apply_patch_to_path(&path, &limits_patch, None) + .await + .unwrap(); + assert!(limits.process_restart_required); + assert!( + limits + .deferred_process_fields + .iter() + .any(|field| field == "web.limits") + ); +} + +#[tokio::test] +async fn invalid_web_patch_does_not_modify_the_source() { + let (path, _directory) = temp_config("[web]\nenabled = false\n"); + let original = tokio::fs::read_to_string(&path).await.unwrap(); + let patch: Json = serde_json::json!({ + "web": {"debug": {"default_window_secs": 181, "max_window_secs": 180}} + }); + + let error = apply_patch_to_path(&path, &patch, None).await.unwrap_err(); + + assert_eq!(error.status, hyper::StatusCode::BAD_REQUEST); + assert_eq!(tokio::fs::read_to_string(&path).await.unwrap(), original); +} + #[tokio::test] async fn read_managed_config_returns_only_editable_sections() { // Full server (api/port) and network must not leak. Listeners-only server diff --git a/src/api/config_store.rs b/src/api/config_store.rs index a034882..7e97ec7 100644 --- a/src/api/config_store.rs +++ b/src/api/config_store.rs @@ -262,12 +262,12 @@ pub(super) async fn load_config_from_disk(config_path: &Path) -> Result Result { - let config_path = config_path.to_path_buf(); - tokio::task::spawn_blocking(move || ProxyConfig::load(config_path)) - .await - .map_err(|error| ApiFailure::internal(format!("failed to join config loader: {}", error)))? - .map_err(|error| ApiFailure::bad_request(format!("invalid runtime config: {}", error))) +pub(super) async fn load_config_for_reload( + config_path: &Path, +) -> Result<(ProxyConfig, String), ApiFailure> { + let loaded = load_config_snapshot(config_path, true).await?; + let revision = compute_snapshot_revision(&loaded); + Ok((loaded.config, revision)) } #[allow(dead_code)] @@ -307,6 +307,7 @@ pub(super) const EDITABLE_SECTIONS: &[&str] = &[ "censorship", "upstreams", "dc_overrides", + "web", ]; /// Nested fields under `[server]` that may be read/patched via the config API. diff --git a/src/api/mod.rs b/src/api/mod.rs index 7126f4f..af70f42 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -30,6 +30,7 @@ use crate::startup::StartupTracker; use crate::stats::Stats; use crate::transport::UpstreamManager; use crate::transport::middle_proxy::MePool; +use crate::web::control::WebRuntimePublication; use crate::web::trace::WebTraceStore; mod config_edit; @@ -48,6 +49,8 @@ mod runtime_stats; mod runtime_watch; mod runtime_zero; mod users; +// WEB runtime status and bounded controls remain separate from general API DTOs. +mod web_runtime; mod web_status; use config_store::{ @@ -125,6 +128,7 @@ pub(super) struct ApiShared { pub(super) reload_control: ReloadControl, pub(super) active_runtime: Arc>, pub(super) web_trace: Arc, + pub(super) web_runtime_rx: watch::Receiver, } impl ApiShared { @@ -159,6 +163,7 @@ impl ApiShared { reload_control: self.reload_control.clone(), active_runtime: self.active_runtime.clone(), web_trace: self.web_trace.clone(), + web_runtime_rx: self.web_runtime_rx.clone(), } } } @@ -196,9 +201,15 @@ async fn submit_reload_from_disk( request: ReloadRequest, ) -> Result<(ReloadAccepted, String), ApiFailure> { let _guard = mutation_lock.lock().await; - ensure_expected_revision(config_path, expected_revision).await?; - let revision = current_revision(config_path).await?; - let config = Arc::new(load_config_for_reload(config_path).await?); + let (config, revision) = load_config_for_reload(config_path).await?; + if expected_revision.is_some_and(|expected| expected != revision) { + return Err(ApiFailure::new( + StatusCode::CONFLICT, + "revision_conflict", + "Config revision mismatch", + )); + } + let config = Arc::new(config); let accepted = reload_control .submit(config, revision.clone(), request) .await @@ -218,6 +229,9 @@ async fn submit_reload_from_disk( } fn allowed_methods_for_path(path: &str) -> Option<&'static str> { + if let Some(allow) = web_runtime::allowed_methods(path) { + return Some(allow); + } match path { "/v1/health" | "/v1/health/ready" @@ -285,6 +299,7 @@ pub async fn serve( mut active_runtime_rx: watch::Receiver>>>, mut runtime_watch_rx: watch::Receiver>, web_trace: Arc, + web_runtime_rx: watch::Receiver, ) { let active_runtime = loop { if let Some(active_runtime) = active_runtime_rx.borrow().clone() { @@ -351,6 +366,7 @@ pub async fn serve( reload_control, active_runtime, web_trace, + web_runtime_rx, }); spawn_runtime_watchers( @@ -498,9 +514,30 @@ async fn handle( let body_limit = api_cfg.request_body_limit_bytes; let result: Result>, ApiFailure> = async { + if web_runtime::is_route(normalized_path) { + let web_mutation = method == Method::POST; + let result = web_runtime::handle( + method, + normalized_path, + query.as_deref(), + req, + shared.as_ref(), + cfg.as_ref(), + request_id, + body_limit, + ) + .await; + if web_mutation && let Err(error) = &result { + shared.runtime_events.record( + "api.web.control.failed", + format!("path={} code={}", normalized_path, error.code), + ); + } + return result; + } match (method.as_str(), normalized_path) { ("GET", "/web-status") => { - Ok(web_status::render(query.as_deref(), &shared.web_trace, &cfg.web.debug).await) + Ok(web_status::render(query.as_deref(), &shared.web_trace).await) } ("GET", "/v1/health") => { let revision = current_revision(&shared.config_path).await?; diff --git a/src/api/web_runtime.rs b/src/api/web_runtime.rs new file mode 100644 index 0000000..b9a25be --- /dev/null +++ b/src/api/web_runtime.rs @@ -0,0 +1,444 @@ +use std::sync::Arc; +use std::time::Instant; + +use http_body_util::Full; +use hyper::body::{Bytes, Incoming}; +use hyper::header::CONTENT_TYPE; +use hyper::{Method, Request, Response, StatusCode}; +use serde::Serialize; + +use super::config_store::current_revision; +use super::http_utils::{read_json, success_response}; +use super::model::ApiFailure; +use super::{ALLOW_GET, ALLOW_POST, ApiShared}; +use crate::config::ProxyConfig; +use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication}; +use crate::web::manager::{ControlError, SessionDetail, WebProcessRuntime}; + +// Exact JSON DTOs and strict query parsing stay independent from route dispatch. +mod request; +use request::{ + CloseRequest, RuntimeInstanceRequest, parse_session_query, parse_session_ref, + valid_runtime_instance, +}; + +const STATUS_PATH: &str = "/v1/runtime/web/status"; +const SESSIONS_PATH: &str = "/v1/runtime/web/sessions"; +const CLOSE_PATH: &str = "/v1/runtime/web/sessions/close"; +const DEBUG_CLEAR_PATH: &str = "/v1/runtime/web/debug/clear"; +const LEARNING_RESET_PATH: &str = "/v1/runtime/web/carrier-learning/reset"; +const SESSION_DETAIL_PREFIX: &str = "/v1/runtime/web/sessions/"; +const OPERATION_PREFIX: &str = "/v1/runtime/web/operations/"; +const MAX_CONTROL_BODY_BYTES: usize = 64 * 1024; + +/// Returns the exact allowed method set for a WEB runtime route. +pub(super) fn allowed_methods(path: &str) -> Option<&'static str> { + match path { + STATUS_PATH | SESSIONS_PATH => Some(ALLOW_GET), + CLOSE_PATH | DEBUG_CLEAR_PATH | LEARNING_RESET_PATH => Some(ALLOW_POST), + _ if detail_ref(path).is_some() || operation_ref(path).is_some() => Some(ALLOW_GET), + _ => None, + } +} + +/// Returns whether the normalized API path belongs to WEB runtime control. +pub(super) fn is_route(path: &str) -> bool { + allowed_methods(path).is_some() +} + +/// Dispatches one authenticated WEB runtime status or control request. +pub(super) async fn handle( + method: Method, + path: &str, + query: Option<&str>, + request: Request, + shared: &ApiShared, + config: &ProxyConfig, + _request_id: u64, + body_limit: usize, +) -> Result>, ApiFailure> { + let revision = current_revision(&shared.config_path).await?; + match (method.as_str(), path) { + ("GET", STATUS_PATH) => { + reject_query(query)?; + let publication = shared.web_runtime_rx.borrow().clone(); + let runtime = publication.runtime.upgrade(); + let data = WebStatusData::new(publication, runtime.as_deref(), config.web.enabled); + Ok(success_response(StatusCode::OK, data, revision)) + } + ("GET", SESSIONS_PATH) => { + let runtime = readable_runtime(shared)?; + let request = parse_session_query(&runtime, query)?; + let page = runtime.list_sessions(request); + Ok(success_response(StatusCode::OK, page, revision)) + } + ("GET", _) if detail_ref(path).is_some() => { + reject_query(query)?; + let runtime = readable_runtime(shared)?; + let session_ref = detail_ref(path).expect("route guard checked detail reference"); + let trace_session_id = parse_session_ref(&runtime, session_ref)?; + match runtime.session_detail(trace_session_id) { + SessionDetail::Active(row) => Ok(success_response(StatusCode::OK, row, revision)), + SessionDetail::Gone { attempt } => Ok(success_response( + StatusCode::GONE, + GoneSessionData { + session_ref: session_ref.to_string(), + state: "closed", + attempt, + }, + revision, + )), + SessionDetail::Busy => Err(snapshot_busy()), + SessionDetail::NotFound => Err(ApiFailure::new( + StatusCode::NOT_FOUND, + "web_session_not_found", + "WEB session was not found", + )), + } + } + ("GET", _) if operation_ref(path).is_some() => { + reject_query(query)?; + let runtime = readable_runtime(shared)?; + let operation_id = operation_ref(path).expect("route guard checked operation id"); + let status = runtime + .control_operation(operation_id) + .map_err(control_failure)?; + Ok(success_response(StatusCode::OK, status, revision)) + } + ("POST", CLOSE_PATH) => { + require_mutable(config)?; + reject_query(query)?; + require_json_content_type(&request)?; + let request = read_json::( + request.into_body(), + body_limit.min(MAX_CONTROL_BODY_BYTES), + ) + .await?; + let runtime = control_runtime(shared)?; + require_runtime_instance(&runtime, &request.runtime_instance)?; + let selector = request.selector.resolve(&runtime)?; + let status = runtime + .start_close_operation(&request.runtime_instance, selector) + .map_err(control_failure)?; + shared.runtime_events.record( + "api.web.sessions.close.accepted", + format!( + "operation_id={} requested={}", + status.operation_id, status.requested + ), + ); + Ok(success_response(StatusCode::ACCEPTED, status, revision)) + } + ("POST", DEBUG_CLEAR_PATH) => { + require_mutable(config)?; + reject_query(query)?; + require_json_content_type(&request)?; + let request = read_json::( + request.into_body(), + body_limit.min(MAX_CONTROL_BODY_BYTES), + ) + .await?; + let runtime = control_runtime(shared)?; + require_runtime_instance(&runtime, &request.runtime_instance)?; + let outcome = runtime.clear_debug().map_err(control_failure)?; + let data = DebugClearData { + runtime_instance: runtime.runtime_instance().to_string(), + records_cleared: outcome.records_cleared, + leased_bytes: outcome.leased_bytes, + epoch: outcome.epoch, + }; + shared.runtime_events.record( + "api.web.debug.clear.ok", + format!("records={} epoch={}", data.records_cleared, data.epoch), + ); + Ok(success_response(StatusCode::OK, data, revision)) + } + ("POST", LEARNING_RESET_PATH) => { + require_mutable(config)?; + reject_query(query)?; + require_json_content_type(&request)?; + let request = read_json::( + request.into_body(), + body_limit.min(MAX_CONTROL_BODY_BYTES), + ) + .await?; + let runtime = control_runtime(shared)?; + require_runtime_instance(&runtime, &request.runtime_instance)?; + let outcome = runtime + .reset_carrier_learning() + .map_err(|_| runtime_unavailable(WebRuntimeLifecycle::Draining))?; + let data = LearningResetData { + runtime_instance: runtime.runtime_instance().to_string(), + entries_cleared: outcome.entries_cleared, + epoch: outcome.epoch, + }; + shared.runtime_events.record( + "api.web.carrier_learning.reset.ok", + format!("entries={} epoch={}", data.entries_cleared, data.epoch), + ); + Ok(success_response(StatusCode::OK, data, revision)) + } + _ => Err(ApiFailure::method_not_allowed( + allowed_methods(path).unwrap_or(ALLOW_GET), + )), + } +} + +#[derive(Serialize)] +struct WebStatusData { + lifecycle: &'static str, + lifecycle_epoch: u64, + lifecycle_age_ms: u64, + available: bool, + #[serde(skip_serializing_if = "Option::is_none")] + reason: Option<&'static str>, + listeners: Vec, + effective_config_enabled: bool, + #[serde(skip_serializing_if = "Option::is_none")] + runtime: Option, +} + +impl WebStatusData { + fn new( + publication: WebRuntimePublication, + runtime: Option<&WebProcessRuntime>, + effective_config_enabled: bool, + ) -> Self { + let available = runtime.is_some() + && matches!( + publication.lifecycle, + WebRuntimeLifecycle::Running | WebRuntimeLifecycle::Draining + ); + let reason = if available { + None + } else { + Some(match publication.lifecycle { + WebRuntimeLifecycle::Starting => "starting", + WebRuntimeLifecycle::NoWebListener => "no_web_listener", + WebRuntimeLifecycle::Running => "runtime_released", + WebRuntimeLifecycle::Draining => "runtime_released", + WebRuntimeLifecycle::Drained => "drained", + WebRuntimeLifecycle::DeadlineExceeded => "deadline_exceeded", + }) + }; + Self { + lifecycle: publication.lifecycle.as_str(), + lifecycle_epoch: publication.epoch, + lifecycle_age_ms: millis(Instant::now().saturating_duration_since(publication.since)), + available, + reason, + listeners: publication + .listeners + .iter() + .map(ToString::to_string) + .collect(), + effective_config_enabled, + runtime: runtime.map(WebProcessRuntime::try_status), + } + } +} + +#[derive(Serialize)] +struct GoneSessionData { + session_ref: String, + state: &'static str, + attempt: u8, +} + +#[derive(Serialize)] +struct DebugClearData { + runtime_instance: String, + records_cleared: usize, + leased_bytes: usize, + epoch: u64, +} + +#[derive(Serialize)] +struct LearningResetData { + runtime_instance: String, + entries_cleared: usize, + epoch: u64, +} + +fn readable_runtime(shared: &ApiShared) -> Result, ApiFailure> { + let publication = shared.web_runtime_rx.borrow().clone(); + if !matches!( + publication.lifecycle, + WebRuntimeLifecycle::Running | WebRuntimeLifecycle::Draining + ) { + return Err(runtime_unavailable(publication.lifecycle)); + } + publication + .runtime + .upgrade() + .ok_or_else(|| runtime_unavailable(publication.lifecycle)) +} + +fn control_runtime(shared: &ApiShared) -> Result, ApiFailure> { + let publication = shared.web_runtime_rx.borrow().clone(); + if publication.lifecycle != WebRuntimeLifecycle::Running { + return Err(runtime_unavailable(publication.lifecycle)); + } + publication + .runtime + .upgrade() + .ok_or_else(|| runtime_unavailable(publication.lifecycle)) +} + +fn runtime_unavailable(lifecycle: WebRuntimeLifecycle) -> ApiFailure { + ApiFailure::new( + StatusCode::SERVICE_UNAVAILABLE, + "web_runtime_unavailable", + format!("WEB runtime is unavailable: {}", lifecycle.as_str()), + ) +} + +fn require_mutable(config: &ProxyConfig) -> Result<(), ApiFailure> { + if config.server.api.read_only { + return Err(ApiFailure::new( + StatusCode::FORBIDDEN, + "read_only", + "API runs in read-only mode", + )); + } + Ok(()) +} + +fn require_json_content_type(request: &Request) -> Result<(), ApiFailure> { + let mut values = request.headers().get_all(CONTENT_TYPE).iter(); + let exact = values + .next() + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value == "application/json") + && values.next().is_none(); + if !exact { + return Err(ApiFailure::new( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "unsupported_media_type", + "Content-Type must be exactly application/json", + )); + } + Ok(()) +} + +fn require_runtime_instance( + runtime: &WebProcessRuntime, + runtime_instance: &str, +) -> Result<(), ApiFailure> { + if !valid_runtime_instance(runtime_instance) { + return Err(ApiFailure::bad_request( + "runtime_instance must be 32 lowercase hexadecimal characters", + )); + } + if runtime.runtime_instance() != runtime_instance { + return Err(ApiFailure::new( + StatusCode::CONFLICT, + "web_runtime_mismatch", + "WEB runtime instance no longer matches", + )); + } + Ok(()) +} + +fn control_failure(error: ControlError) -> ApiFailure { + match error { + ControlError::StaleInstance => ApiFailure::new( + StatusCode::CONFLICT, + "web_runtime_mismatch", + "WEB runtime instance no longer matches", + ), + ControlError::InvalidSelector | ControlError::InvalidOperation => { + ApiFailure::bad_request("Invalid WEB control request") + } + ControlError::IssuanceEnabled => ApiFailure::new( + StatusCode::CONFLICT, + "web_issuance_enabled", + "Close-all requires effective WEB issuance to be disabled", + ), + ControlError::OperationInProgress => ApiFailure::new( + StatusCode::CONFLICT, + "web_operation_in_progress", + "Another WEB close operation is active", + ), + ControlError::OperationNotFound => ApiFailure::new( + StatusCode::NOT_FOUND, + "web_operation_not_found", + "WEB control operation was not found", + ), + ControlError::Closed => runtime_unavailable(WebRuntimeLifecycle::Draining), + } +} + +fn snapshot_busy() -> ApiFailure { + ApiFailure::new( + StatusCode::SERVICE_UNAVAILABLE, + "web_snapshot_busy", + "WEB runtime snapshot is temporarily busy", + ) +} + +fn reject_query(query: Option<&str>) -> Result<(), ApiFailure> { + if query.is_some_and(|query| !query.is_empty()) { + return Err(ApiFailure::bad_request( + "This endpoint does not accept query parameters", + )); + } + Ok(()) +} + +fn detail_ref(path: &str) -> Option<&str> { + path.strip_prefix(SESSION_DETAIL_PREFIX) + .filter(|value| !value.is_empty() && !value.contains('/') && *value != "close") +} + +fn operation_ref(path: &str) -> Option<&str> { + path.strip_prefix(OPERATION_PREFIX) + .filter(|value| !value.is_empty() && !value.contains('/')) +} + +fn millis(duration: std::time::Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 +} + +#[cfg(test)] +mod tests { + use super::*; + use hyper::header::HeaderValue; + + #[test] + fn route_table_keeps_status_read_only_and_controls_post_only() { + assert_eq!(allowed_methods(STATUS_PATH), Some(ALLOW_GET)); + assert_eq!(allowed_methods(SESSIONS_PATH), Some(ALLOW_GET)); + assert_eq!(allowed_methods(CLOSE_PATH), Some(ALLOW_POST)); + assert_eq!(allowed_methods(DEBUG_CLEAR_PATH), Some(ALLOW_POST)); + assert_eq!(allowed_methods(LEARNING_RESET_PATH), Some(ALLOW_POST)); + assert_eq!( + allowed_methods("/v1/runtime/web/sessions/ws1.instance.0000000000000001"), + Some(ALLOW_GET) + ); + } + + #[test] + fn control_content_type_is_exact_and_single() { + let exact = Request::builder() + .header(CONTENT_TYPE, "application/json") + .body(()) + .unwrap(); + assert!(require_json_content_type(&exact).is_ok()); + + let parameterized = Request::builder() + .header(CONTENT_TYPE, "application/json; charset=utf-8") + .body(()) + .unwrap(); + assert!(require_json_content_type(¶meterized).is_err()); + + let mut duplicated = Request::builder() + .header(CONTENT_TYPE, "application/json") + .body(()) + .unwrap(); + duplicated + .headers_mut() + .append(CONTENT_TYPE, HeaderValue::from_static("application/json")); + assert!(require_json_content_type(&duplicated).is_err()); + } +} diff --git a/src/api/web_runtime/request.rs b/src/api/web_runtime/request.rs new file mode 100644 index 0000000..8a2f011 --- /dev/null +++ b/src/api/web_runtime/request.rs @@ -0,0 +1,391 @@ +use std::collections::BTreeSet; +use std::net::IpAddr; + +use hyper::StatusCode; +use serde::Deserialize; + +use crate::config::WebCarrier; +use crate::web::manager::{ + CloseOperationSelector, SessionFilter, SessionListRequest, SessionRefError, WebProcessRuntime, +}; + +use super::super::model::ApiFailure; + +const DEFAULT_SESSION_LIMIT: usize = 50; +const MAX_SESSION_LIMIT: usize = 200; + +/// Exact process-instance fence for one runtime mutation. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct RuntimeInstanceRequest { + /// Random process identifier copied from WEB runtime status. + pub(super) runtime_instance: String, +} + +/// One process-fenced asynchronous close request. +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct CloseRequest { + /// Random process identifier copied from WEB runtime status. + pub(super) runtime_instance: String, + /// Exact point-in-time close selector. + pub(super) selector: CloseSelectorRequest, +} + +/// Strict tagged selector accepted by the WEB close endpoint. +#[derive(Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub(super) enum CloseSelectorRequest { + /// Closes an explicit bounded set of logical sessions. + Refs { + /// Unique current-instance opaque session references. + session_refs: Vec, + }, + /// Closes the point-in-time sessions matching every supplied field. + Filter { + #[serde(default)] + session_ref: Option, + #[serde(default)] + ip: Option, + #[serde(default)] + host: Option, + #[serde(default)] + user: Option, + #[serde(default)] + user_agent_id: Option, + #[serde(default)] + key_id: Option, + #[serde(default)] + carrier: Option, + #[serde(default)] + state: Option, + }, + /// Closes every point-in-time session below the submission high-water mark. + All {}, +} + +impl CloseSelectorRequest { + /// Validates external identifiers and resolves them to manager-owned values. + pub(super) fn resolve( + self, + runtime: &WebProcessRuntime, + ) -> Result { + match self { + Self::Refs { session_refs } => resolve_refs(runtime, session_refs), + Self::Filter { + session_ref, + ip, + host, + user, + user_agent_id, + key_id, + carrier, + state, + } => { + validate_filter_strings( + host.as_deref(), + user.as_deref(), + key_id.as_deref(), + state.as_deref(), + )?; + let trace_session_id = session_ref + .as_deref() + .map(|value| parse_session_ref(runtime, value)) + .transpose()?; + let client_ip = ip.as_deref().map(parse_canonical_ip).transpose()?; + let filter = SessionFilter { + trace_session_id, + client_ip, + host, + user, + user_agent_id: user_agent_id + .as_deref() + .map(parse_user_agent_id) + .transpose()?, + key_id, + carrier, + state, + }; + if filter.is_empty() { + return Err(ApiFailure::bad_request( + "filter selector requires at least one filter", + )); + } + Ok(CloseOperationSelector::Filter(filter)) + } + Self::All {} => Ok(CloseOperationSelector::All), + } + } +} + +fn resolve_refs( + runtime: &WebProcessRuntime, + session_refs: Vec, +) -> Result { + if session_refs.is_empty() || session_refs.len() > 200 { + return Err(ApiFailure::bad_request( + "session_refs must contain 1..200 references", + )); + } + let mut resolved = Vec::with_capacity(session_refs.len()); + let mut unique = BTreeSet::new(); + for session_ref in session_refs { + let id = parse_session_ref(runtime, &session_ref)?; + if !unique.insert(id) { + return Err(ApiFailure::bad_request( + "session_refs must not contain duplicates", + )); + } + resolved.push(id); + } + Ok(CloseOperationSelector::Refs(resolved)) +} + +/// Parses one duplicate-free bounded session-list query. +pub(super) fn parse_session_query( + runtime: &WebProcessRuntime, + raw: Option<&str>, +) -> Result { + let mut limit = DEFAULT_SESSION_LIMIT; + let mut cursor = None; + let mut filter = SessionFilter::default(); + let mut seen = BTreeSet::new(); + for (name, value) in url::form_urlencoded::parse(raw.unwrap_or_default().as_bytes()) { + if !seen.insert(name.to_string()) { + return Err(ApiFailure::bad_request(format!("{} must not repeat", name))); + } + match name.as_ref() { + "limit" => { + limit = value + .parse::() + .ok() + .filter(|value| (1..=MAX_SESSION_LIMIT).contains(value)) + .ok_or_else(|| ApiFailure::bad_request("limit must be within 1..200"))?; + } + "cursor" => cursor = Some(parse_session_ref(runtime, &value)?), + "session_ref" => { + let id = parse_session_ref(runtime, &value)?; + filter.trace_session_id = Some(id); + cursor = id.checked_sub(1); + limit = 1; + } + "ip" => { + filter.client_ip = Some(parse_canonical_ip(&value)?); + } + "host" => filter.host = Some(value.into_owned()), + "user" => filter.user = Some(value.into_owned()), + "user_agent_id" => filter.user_agent_id = Some(parse_user_agent_id(&value)?), + "key_id" => filter.key_id = Some(value.into_owned()), + "carrier" => filter.carrier = Some(parse_carrier(&value)?), + "state" => filter.state = Some(value.into_owned()), + _ => { + return Err(ApiFailure::bad_request(format!( + "unknown query field `{}`", + name + ))); + } + } + } + if filter.trace_session_id.is_some() && (seen.contains("cursor") || seen.contains("limit")) { + return Err(ApiFailure::bad_request( + "session_ref must not be combined with cursor or limit", + )); + } + validate_filter_strings( + filter.host.as_deref(), + filter.user.as_deref(), + filter.key_id.as_deref(), + filter.state.as_deref(), + )?; + Ok(SessionListRequest { + limit, + cursor, + filter, + }) +} + +/// Maps one opaque session-reference failure to the stable API error contract. +pub(super) fn parse_session_ref( + runtime: &WebProcessRuntime, + session_ref: &str, +) -> Result { + runtime + .parse_session_ref(session_ref) + .map_err(|error| match error { + SessionRefError::Invalid => ApiFailure::bad_request("Invalid WEB session reference"), + SessionRefError::StaleInstance => ApiFailure::new( + StatusCode::CONFLICT, + "web_runtime_mismatch", + "WEB session reference belongs to another runtime instance", + ), + }) +} + +fn parse_carrier(value: &str) -> Result { + WebCarrier::ALL + .into_iter() + .find(|carrier| carrier.as_str() == value) + .ok_or_else(|| { + ApiFailure::bad_request( + "carrier must be https, https-lanes, websocket, or websocket-lanes", + ) + }) +} + +fn validate_filter_strings( + host: Option<&str>, + user: Option<&str>, + key_id: Option<&str>, + state: Option<&str>, +) -> Result<(), ApiFailure> { + if host.is_some_and(|value| value.is_empty() || value.len() > 253) { + return Err(ApiFailure::bad_request("host must contain 1..253 bytes")); + } + if user.is_some_and(|value| value.is_empty() || value.len() > 64) { + return Err(ApiFailure::bad_request("user must contain 1..64 bytes")); + } + if key_id.is_some_and(|value| !lower_hex(value, 16)) { + return Err(ApiFailure::bad_request( + "key_id must be 16 lowercase hexadecimal characters", + )); + } + if state.is_some_and(|value| { + !matches!( + value, + "provisional" + | "replacing" + | "committed" + | "healthy" + | "closing" + | "superseded" + | "closed" + ) + }) { + return Err(ApiFailure::bad_request("Invalid WEB session state")); + } + Ok(()) +} + +fn parse_user_agent_id(value: &str) -> Result<[u8; 16], ApiFailure> { + if !lower_hex(value, 32) { + return Err(ApiFailure::bad_request( + "user_agent_id must be 32 lowercase hexadecimal characters", + )); + } + let mut id = [0; 16]; + hex::decode_to_slice(value, &mut id).map_err(|_| { + ApiFailure::bad_request("user_agent_id must be 32 lowercase hexadecimal characters") + })?; + Ok(id) +} + +fn lower_hex(value: &str, length: usize) -> bool { + value.len() == length + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) +} + +/// Returns whether a process instance uses its canonical lowercase form. +pub(super) fn valid_runtime_instance(value: &str) -> bool { + lower_hex(value, 32) +} + +fn parse_canonical_ip(value: &str) -> Result { + let ip = value + .parse::() + .map_err(|_| ApiFailure::bad_request("ip must be a canonical IP address"))?; + if ip.to_string() != value { + return Err(ApiFailure::bad_request("ip must use canonical formatting")); + } + Ok(ip) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn filter_identifiers_are_exact_lowercase_hex() { + assert!(lower_hex("0123456789abcdef", 16)); + assert!(!lower_hex("0123456789ABCDEF", 16)); + assert!(!lower_hex("0123", 16)); + assert!(valid_runtime_instance("0123456789abcdef0123456789abcdef")); + assert!(!valid_runtime_instance("0123456789ABCDEF0123456789ABCDEF")); + } + + #[test] + fn mutation_dtos_reject_unknown_fields() { + let runtime_instance = "0123456789abcdef0123456789abcdef"; + assert!( + serde_json::from_value::(serde_json::json!({ + "runtime_instance": runtime_instance, + "extra": true, + })) + .is_err() + ); + assert!( + serde_json::from_value::(serde_json::json!({ + "runtime_instance": runtime_instance, + "selector": {"kind": "all", "extra": true}, + })) + .is_err() + ); + } + + #[test] + fn filter_ips_require_canonical_text() { + assert!(parse_canonical_ip("2001:db8::1").is_ok()); + assert!(parse_canonical_ip("2001:0db8::1").is_err()); + } + + #[test] + fn state_filter_accepts_every_emitted_session_state() { + for state in [ + "provisional", + "replacing", + "committed", + "healthy", + "closing", + "superseded", + "closed", + ] { + assert!( + validate_filter_strings(None, None, None, Some(state)).is_ok(), + "state {state} must be accepted" + ); + } + } + + #[tokio::test] + async fn exact_session_query_rejects_pagination_fields_in_any_order() { + let generation = crate::maestro::generation::test_runtime_generation( + 1, + crate::config::ProxyConfig::default(), + ); + let runtime = WebProcessRuntime::start(std::sync::Arc::new(arc_swap::ArcSwap::from( + generation.clone(), + ))); + let session_ref = runtime.session_ref(1); + let cursor = runtime.session_ref(2); + + assert!( + parse_session_query( + &runtime, + Some(&format!("session_ref={session_ref}&cursor={cursor}")), + ) + .is_err() + ); + assert!( + parse_session_query( + &runtime, + Some(&format!("limit=2&session_ref={session_ref}")), + ) + .is_err() + ); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} diff --git a/src/api/web_status.rs b/src/api/web_status.rs index 0914a97..d162e0d 100644 --- a/src/api/web_status.rs +++ b/src/api/web_status.rs @@ -41,10 +41,9 @@ impl AsRef<[u8]> for RenderedPage { pub(super) async fn render( raw_query: Option<&str>, store: &Arc, - policy: &WebDebugConfig, ) -> Response> { - store.apply_policy(policy); - let query = match parse_query(raw_query, policy) { + let status = store.status(); + let query = match parse_query(raw_query, &status.policy) { Ok(query) => query, Err(error) => return html_error(StatusCode::BAD_REQUEST, "Invalid query", &error), }; @@ -62,7 +61,6 @@ pub(super) async fn render( 0 }; let records = store.snapshot_matching(|record| record_matches(record, &query, since_millis)); - let status = store.status(); let mut html = String::with_capacity(MAX_PAGE_BYTES); push_page_start(&mut html); html.push_str("

WEB status

"); diff --git a/src/api/web_status/tests.rs b/src/api/web_status/tests.rs index b2480a3..52ef582 100644 --- a/src/api/web_status/tests.rs +++ b/src/api/web_status/tests.rs @@ -46,7 +46,6 @@ async fn renderer_filters_groups_and_sets_control_plane_security_headers() { let response = render( Some("ip=192.0.2.40&session=42&key=0123456789abcdef&group_by=ip&group_by=key"), &store, - &policy, ) .await; assert_eq!(response.status(), StatusCode::OK); @@ -69,18 +68,36 @@ async fn render_permits_remain_owned_by_inflight_response_bodies() { let limits = crate::config::WebLimitsConfig::default(); let store = WebTraceStore::new(policy.clone(), &limits); - let first = render(None, &store, &policy).await; - let second = render(None, &store, &policy).await; - let busy = render(None, &store, &policy).await; + let first = render(None, &store).await; + let second = render(None, &store).await; + let busy = render(None, &store).await; assert_eq!(busy.status(), StatusCode::SERVICE_UNAVAILABLE); drop(first); - let admitted = render(None, &store, &policy).await; + let admitted = render(None, &store).await; assert_eq!(admitted.status(), StatusCode::OK); drop(second); drop(admitted); } +#[tokio::test] +async fn stale_renderer_cannot_restore_an_old_debug_policy() { + let stale_policy = WebDebugConfig::default(); + let active_policy = WebDebugConfig { + enabled: true, + capture_headers: false, + ..Default::default() + }; + let limits = crate::config::WebLimitsConfig::default(); + let store = WebTraceStore::new(stale_policy.clone(), &limits); + store.apply_policy(2, &active_policy); + + let response = render(None, &store).await; + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(store.status().policy.as_ref(), &active_policy); +} + #[test] fn page_truncation_preserves_utf8_boundary_and_cap() { let mut html = "\u{044f}".repeat(MAX_PAGE_BYTES); diff --git a/src/config/hot_reload/tests.rs b/src/config/hot_reload/tests.rs index 8a5a99b..2d95608 100644 --- a/src/config/hot_reload/tests.rs +++ b/src/config/hot_reload/tests.rs @@ -256,6 +256,46 @@ fn reload_applies_hot_change_on_first_observed_snapshot() { let _ = std::fs::remove_file(path); } +#[tokio::test] +async fn candidate_watcher_waits_for_activation_and_reconciles_disk() { + let initial_tag = "10101010101010101010101010101010"; + let disk_tag = "20202020202020202020202020202020"; + let path = temp_config_path("telemt_hot_reload_activation_gate"); + write_reload_config(&path, Some(initial_tag), None); + let initial = Arc::new(ProxyConfig::load(&path).unwrap()); + write_reload_config(&path, Some(disk_tag), None); + let cancellation = tokio_util::sync::CancellationToken::new(); + let (activation_tx, activation_rx) = watch::channel(false); + let (mut config_rx, _log_rx, watcher) = spawn_config_watcher( + path.clone(), + initial, + None, + None, + cancellation.clone(), + Some(activation_rx), + ); + let watcher = tokio::spawn(watcher); + + tokio::task::yield_now().await; + assert_eq!( + config_rx.borrow().general.ad_tag.as_deref(), + Some(initial_tag) + ); + activation_tx.send_replace(true); + tokio::time::timeout(Duration::from_secs(2), config_rx.changed()) + .await + .unwrap() + .unwrap(); + assert_eq!( + config_rx.borrow_and_update().general.ad_tag.as_deref(), + Some(disk_tag) + ); + + cancellation.cancel(); + watcher.await.unwrap(); + let _ = std::fs::remove_file(path); +} + #[test] fn reload_keeps_hot_apply_when_non_hot_fields_change() { let initial_tag = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; diff --git a/src/config/hot_reload/watcher.rs b/src/config/hot_reload/watcher.rs index 838988d..0770f2d 100644 --- a/src/config/hot_reload/watcher.rs +++ b/src/config/hot_reload/watcher.rs @@ -213,20 +213,47 @@ pub fn spawn_config_watcher( detected_ip_v4: Option, detected_ip_v6: Option, cancellation: tokio_util::sync::CancellationToken, -) -> (watch::Receiver>, watch::Receiver) { + mut activation: Option>, +) -> ( + watch::Receiver>, + watch::Receiver, + impl std::future::Future + Send + 'static, +) { let initial_level = initial.general.log_level.clone(); let (config_tx, config_rx) = watch::channel(initial); let (log_tx, log_rx) = watch::channel(initial_level); let config_path = normalize_watch_path(&config_path); - let initial_loaded = ProxyConfig::load_with_metadata(&config_path).ok(); - let initial_manifest = initial_loaded - .as_ref() - .map(|loaded| WatchManifest::from_source_files(&loaded.source_files)) - .unwrap_or_else(|| WatchManifest::from_source_files(std::slice::from_ref(&config_path))); - let initial_snapshot_hash = initial_loaded.as_ref().map(|loaded| loaded.rendered_hash); - - tokio::spawn(async move { + let task = async move { + if let Some(activation) = activation.as_mut() { + loop { + if *activation.borrow_and_update() { + break; + } + tokio::select! { + result = activation.changed() => { + if result.is_err() { + return; + } + } + _ = cancellation.cancelled() => return, + } + } + } + let initial_loaded = ProxyConfig::load_with_metadata(&config_path).ok(); + let initial_manifest = initial_loaded + .as_ref() + .map(|loaded| WatchManifest::from_source_files(&loaded.source_files)) + .unwrap_or_else(|| { + WatchManifest::from_source_files(std::slice::from_ref(&config_path)) + }); + let initial_matches_disk = initial_loaded + .as_ref() + .is_some_and(|loaded| config_equal(config_tx.borrow().as_ref(), &loaded.config)); + let initial_snapshot_hash = initial_loaded + .as_ref() + .filter(|_| initial_matches_disk) + .map(|loaded| loaded.rendered_hash); let (notify_tx, mut notify_rx) = mpsc::channel::<()>(4); let manifest_state = Arc::new(StdRwLock::new(WatchManifest::default())); let mut reload_state = ReloadState::new(initial_snapshot_hash); @@ -304,6 +331,9 @@ pub fn spawn_config_watcher( if poll_watcher.is_some() { info!("config watcher: poll watcher active (Docker/NFS safe)"); } + if initial_loaded.is_some() && !initial_matches_disk { + let _ = notify_tx.try_send(()); + } #[cfg(unix)] let mut sighup = { @@ -364,7 +394,7 @@ pub fn spawn_config_watcher( ); } } - }); + }; - (config_rx, log_rx) + (config_rx, log_rx, task) } diff --git a/src/config/load/validate_web.rs b/src/config/load/validate_web.rs index 3ea8824..0863298 100644 --- a/src/config/load/validate_web.rs +++ b/src/config/load/validate_web.rs @@ -381,6 +381,11 @@ fn validate_vhosts(config: &mut ProxyConfig) -> Result<()> { validate_decoy(vhost_idx, &vhost.decoy)?; let mut profiles = HashSet::with_capacity(vhost.profiles.len()); for (profile_idx, profile) in vhost.profiles.iter().enumerate() { + if profile.user.is_empty() || profile.user.len() > 64 { + return config_error(&format!( + "web.vhosts[{vhost_idx}].profiles[{profile_idx}].user must contain 1..64 bytes" + )); + } if !config.access.users.contains_key(&profile.user) { return config_error(&format!( "web.vhosts[{vhost_idx}].profiles[{profile_idx}].user references unknown access user `{}`", diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs index a223666..17a5c05 100644 --- a/src/config/tests/load_basic_tests/web_tests.rs +++ b/src/config/tests/load_basic_tests/web_tests.rs @@ -56,6 +56,17 @@ fn web_config_builds_canonical_runtime_snapshot() { ); } +#[test] +fn web_profile_user_labels_are_bounded_for_runtime_status() { + let user = "a".repeat(65); + let invalid = WEB_CONFIG.replace("alice", &user); + + assert!( + load_config_error_from_temp_toml(&invalid) + .contains("web.vhosts[0].profiles[0].user must contain 1..64 bytes") + ); +} + #[test] fn web_carriers_missing_or_false_disable_negotiation() { let missing = load_config_from_temp_toml(WEB_CONFIG); diff --git a/src/maestro/listeners/control.rs b/src/maestro/listeners/control.rs index a8fc0fb..46cfd17 100644 --- a/src/maestro/listeners/control.rs +++ b/src/maestro/listeners/control.rs @@ -14,6 +14,7 @@ use super::bind::{BoundListeners, BoundTcpListener, PreparedTcpListener, prepare use super::plan::{ListenerBindSpec, listener_bind_plan}; #[cfg(unix)] use super::unix::UnixAcceptHandle; +use crate::web::control::{WebRuntimeControl, WebRuntimeLifecycle}; use crate::web::manager::{WebProcessRuntime, WebShutdownOutcome}; use crate::web::trace::WebTraceStore; @@ -22,6 +23,8 @@ pub(crate) struct ListenerManager { active_runtime: Arc>, slots: BTreeMap, web_runtime: Option>, + web_control: WebRuntimeControl, + web_listeners: Arc<[SocketAddr]>, #[cfg(unix)] unix: Option, } @@ -46,11 +49,15 @@ impl ListenerManager { bound: BoundListeners, active_runtime: Arc>, trace: Arc, + web_control: WebRuntimeControl, ) -> Self { - let has_web = bound + let web_listeners: Arc<[SocketAddr]> = bound .listeners .iter() - .any(|listener| listener.spec.transport == ListenerTransport::Web); + .filter(|listener| listener.spec.transport == ListenerTransport::Web) + .map(|listener| listener.spec.addr) + .collect(); + let has_web = !web_listeners.is_empty(); let web_runtime = has_web.then(|| WebProcessRuntime::start_with_trace(active_runtime.clone(), trace)); let mut slots = BTreeMap::new(); @@ -65,10 +72,23 @@ impl ListenerManager { let unix = bound .unix_listener .map(|listener| UnixAcceptHandle::start(listener, active_runtime.clone())); + web_control.publish( + if has_web { + WebRuntimeLifecycle::Running + } else { + WebRuntimeLifecycle::NoWebListener + }, + Arc::clone(&web_listeners), + web_runtime + .as_ref() + .map_or_else(std::sync::Weak::new, Arc::downgrade), + ); Self { active_runtime, slots, web_runtime, + web_control, + web_listeners, #[cfg(unix)] unix, } @@ -76,10 +96,18 @@ impl ListenerManager { #[cfg(test)] pub(crate) fn empty(active_runtime: Arc>) -> Self { + let web_control = WebRuntimeControl::new(); + web_control.publish( + WebRuntimeLifecycle::NoWebListener, + Arc::from([]), + std::sync::Weak::new(), + ); Self { active_runtime, slots: BTreeMap::new(), web_runtime: None, + web_control, + web_listeners: Arc::from([]), #[cfg(unix)] unix: None, } @@ -217,6 +245,13 @@ impl ListenerManager { /// 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( + WebRuntimeLifecycle::Draining, + Arc::clone(&self.web_listeners), + self.web_runtime + .as_ref() + .map_or_else(std::sync::Weak::new, Arc::downgrade), + ); if self.web_runtime.is_none() { let mut errors = Vec::new(); for slot in self.slots.values_mut() { @@ -235,6 +270,11 @@ impl ListenerManager { { self.unix = None; } + self.web_control.publish( + WebRuntimeLifecycle::Drained, + Arc::clone(&self.web_listeners), + std::sync::Weak::new(), + ); return if errors.is_empty() { Ok(()) } else { @@ -289,6 +329,15 @@ impl ListenerManager { { self.unix = None; } + self.web_control.publish( + if web_outcome == WebShutdownOutcome::DeadlineExceeded { + WebRuntimeLifecycle::DeadlineExceeded + } else { + WebRuntimeLifecycle::Drained + }, + Arc::clone(&self.web_listeners), + std::sync::Weak::new(), + ); if errors.is_empty() { Ok(()) } else { @@ -367,7 +416,8 @@ mod tests { runtime.config().web.debug.clone(), &runtime.config().web.limits, ); - let mut manager = ListenerManager::start(bound, active_runtime, trace); + let mut manager = + ListenerManager::start(bound, active_runtime, trace, WebRuntimeControl::new()); let blocker = TcpListener::bind("127.0.0.1:0").await.unwrap(); let blocked_addr = blocker.local_addr().unwrap(); let mut desired = ProxyConfig::default(); @@ -394,7 +444,8 @@ mod tests { runtime.config().web.debug.clone(), &runtime.config().web.limits, ); - let mut manager = ListenerManager::start(bound, active_runtime, trace); + let mut manager = + ListenerManager::start(bound, active_runtime, trace, WebRuntimeControl::new()); let reservation = TcpListener::bind("127.0.0.1:0").await.unwrap(); let new_addr = reservation.local_addr().unwrap(); drop(reservation); diff --git a/src/maestro/orchestrator.rs b/src/maestro/orchestrator.rs index 387a299..2e7a4cd 100644 --- a/src/maestro/orchestrator.rs +++ b/src/maestro/orchestrator.rs @@ -17,6 +17,7 @@ use crate::stats::{QuotaStore, Stats}; use crate::synlimit_control; use crate::transport::UpstreamManager; use crate::transport::middle_proxy::MePool; +use crate::web::control::WebRuntimeControl; use crate::web::trace::WebTraceStore; use super::{ @@ -106,6 +107,7 @@ pub(super) async fn run_telemt_core( config.access.cidr_rate_limits.clone(), ); let web_trace = WebTraceStore::new(config.web.debug.clone(), &config.web.limits); + let web_runtime_control = WebRuntimeControl::new(); let (detected_ips_tx, detected_ips_rx) = watch::channel((None::, None::)); let initial_direct_first = config.general.use_middle_proxy && config.general.me2dc_fallback; @@ -157,6 +159,7 @@ pub(super) async fn run_telemt_core( let active_runtime_rx_api = active_runtime_rx.clone(); let runtime_watch_rx_api = runtime_watch_rx.clone(); let web_trace_api = web_trace.clone(); + let web_runtime_rx_api = web_runtime_control.subscribe(); tokio::spawn(async move { api::serve( listen, @@ -175,6 +178,7 @@ pub(super) async fn run_telemt_core( active_runtime_rx_api, runtime_watch_rx_api, web_trace_api, + web_runtime_rx_api, ) .await; }); @@ -318,8 +322,12 @@ pub(super) async fn run_telemt_core( active_runtime_tx.send_replace(Some(active_runtime.clone())); runtime_tasks::mark_runtime_ready(&startup_tracker).await; - let listener_manager = - listeners::ListenerManager::start(bound, active_runtime.clone(), web_trace.clone()); + let listener_manager = listeners::ListenerManager::start( + bound, + active_runtime.clone(), + web_trace.clone(), + web_runtime_control, + ); let reload_supervisor = reload_supervisor::ReloadSupervisor::spawn( active_runtime.clone(), reload_control, diff --git a/src/maestro/reload_supervisor.rs b/src/maestro/reload_supervisor.rs index 1ffa580..21c2ea5 100644 --- a/src/maestro/reload_supervisor.rs +++ b/src/maestro/reload_supervisor.rs @@ -278,7 +278,11 @@ impl ReloadSupervisor { self.control .mark_phase(command.reload_id, ReloadPhase::Activating) .await; - let new_runtime = prepared.generation; + let PreparedRuntime { + generation: new_runtime, + detected_ips, + config_watcher_activation, + } = prepared; if let Err(error) = install_dns(&new_runtime.config().network.dns_overrides) { let message = format!("runtime DNS activation failed: {}", error); if command.request.failure_policy == ReloadFailurePolicy::Rollback { @@ -313,14 +317,16 @@ impl ReloadSupervisor { }; old_runtime.stop_accepting_sessions(); let replaced = self.active_runtime.swap(new_runtime.clone()); - self.web_trace.apply_policy(&new_runtime.config().web.debug); + self.web_trace + .apply_policy(new_runtime.id, &new_runtime.config().web.debug); + config_watcher_activation.send_replace(true); if let Some(pending) = pending_listener_transition { self.listener_manager .lock() .await .finish_transition(pending); } - self.detected_ips_tx.send_replace(prepared.detected_ips); + self.detected_ips_tx.send_replace(detected_ips); self.runtime_log_filter .apply_reload(&new_runtime.config().general.log_level); self.runtime_watch_tx diff --git a/src/maestro/reload_supervisor_tests.rs b/src/maestro/reload_supervisor_tests.rs index 99285c0..d742760 100644 --- a/src/maestro/reload_supervisor_tests.rs +++ b/src/maestro/reload_supervisor_tests.rs @@ -21,6 +21,15 @@ fn runtime_log_filter() -> RuntimeLogFilter { RuntimeLogFilter::new(handle) } +fn prepared_runtime(generation: Arc) -> PreparedRuntime { + let (config_watcher_activation, _activation_rx) = watch::channel(false); + PreparedRuntime { + generation, + detected_ips: (None, None), + config_watcher_activation, + } +} + async fn fixture(request: ReloadRequest) -> ReloadFixture { let old_runtime = test_runtime_generation(1, ProxyConfig::default()); let new_config = Arc::new(ProxyConfig::default()); @@ -120,10 +129,7 @@ async fn revision_rollback_keeps_old_generation_and_cleans_candidate() { .activate_prepared( fixture.command, fixture.old_runtime.clone(), - PreparedRuntime { - generation: fixture.new_runtime, - detected_ips: (None, None), - }, + prepared_runtime(fixture.new_runtime), RevisionGateAction::Rollback("revision changed".to_string()), |_| -> Result<(), String> { panic!("DNS activation must not run on rollback") }, ) @@ -161,10 +167,7 @@ async fn dns_failure_policy_controls_rollback_or_keep_new() { .activate_prepared( fixture.command, fixture.old_runtime.clone(), - PreparedRuntime { - generation: fixture.new_runtime.clone(), - detected_ips: (None, None), - }, + prepared_runtime(fixture.new_runtime.clone()), RevisionGateAction::Proceed, |_| Err("invalid DNS entry".to_string()), ) @@ -215,10 +218,7 @@ async fn drain_publishes_new_generation_before_old_sessions_finish() { .activate_prepared( fixture.command, old_runtime, - PreparedRuntime { - generation: new_runtime, - detected_ips: (None, None), - }, + prepared_runtime(new_runtime), RevisionGateAction::Proceed, |_| Ok(()), ) @@ -271,10 +271,7 @@ async fn drain_timeout_cancels_old_sessions_and_records_one_warning() { .activate_prepared( fixture.command, old_runtime, - PreparedRuntime { - generation: new_runtime, - detected_ips: (None, None), - }, + prepared_runtime(new_runtime), RevisionGateAction::Proceed, |_| Ok(()), ) diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index 7f1c37d..8593487 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -28,9 +28,14 @@ use super::listeners::listener_rebind_supported; use super::runtime_tasks::RuntimeLogFilter; use super::{me_startup, runtime_tasks, tls_bootstrap}; +/// Fully prepared candidate runtime and its activation-gated config watcher. pub(crate) struct PreparedRuntime { + /// Candidate generation ready for publication. pub(crate) generation: Arc, + /// Detected public addresses associated with the candidate. pub(crate) detected_ips: (Option, Option), + /// Gate opened only after the candidate becomes the active generation. + pub(crate) config_watcher_activation: watch::Sender, } pub(crate) async fn prepare_runtime( @@ -171,6 +176,7 @@ pub(crate) async fn prepare_runtime( config.server.max_connections as usize }; let max_connections = Arc::new(Semaphore::new(max_connections_limit)); + let (config_watcher_activation, config_watcher_activation_rx) = watch::channel(false); let watches = runtime_tasks::spawn_runtime_tasks( &config, config_path, @@ -190,6 +196,7 @@ pub(crate) async fn prepare_runtime( proxy_shared.clone(), me_ready_tx.clone(), task_scope.clone(), + Some(config_watcher_activation_rx), ) .await; let config_rx = watches.config_rx; @@ -295,6 +302,7 @@ pub(crate) async fn prepare_runtime( Ok(PreparedRuntime { generation, + config_watcher_activation, detected_ips: ( probe.detected_ipv4.map(IpAddr::V4), probe.detected_ipv6.map(IpAddr::V6), diff --git a/src/maestro/runtime_startup.rs b/src/maestro/runtime_startup.rs index e23d65a..bfa4485 100644 --- a/src/maestro/runtime_startup.rs +++ b/src/maestro/runtime_startup.rs @@ -247,6 +247,7 @@ pub(super) async fn prepare_runtime( shared_state.clone(), me_ready_tx.clone(), runtime_task_scope.clone(), + None, ) .await; let config_rx = runtime_watches.config_rx; diff --git a/src/maestro/runtime_tasks.rs b/src/maestro/runtime_tasks.rs index 1b896e6..632da2c 100644 --- a/src/maestro/runtime_tasks.rs +++ b/src/maestro/runtime_tasks.rs @@ -107,6 +107,7 @@ pub(crate) async fn spawn_runtime_tasks( shared_state: Arc, me_ready_tx: watch::Sender, task_scope: RuntimeTaskScope, + config_watcher_activation: Option>, ) -> RuntimeWatches { let um_clone = upstream_manager.clone(); let dc_overrides_for_health = config.dc_overrides.clone(); @@ -151,14 +152,15 @@ pub(crate) async fn spawn_runtime_tasks( Some("spawn config hot-reload watcher".to_string()), ) .await; - let (config_rx, log_level_rx): (watch::Receiver>, watch::Receiver) = - spawn_config_watcher( - config_path.to_path_buf(), - config.clone(), - detected_ip_v4, - detected_ip_v6, - task_scope.cancellation_token(), - ); + let (config_rx, log_level_rx, config_watcher_task) = spawn_config_watcher( + config_path.to_path_buf(), + config.clone(), + detected_ip_v4, + detected_ip_v6, + task_scope.cancellation_token(), + config_watcher_activation, + ); + task_scope.spawn(config_watcher_task); startup_tracker .complete_component( COMPONENT_CONFIG_WATCHER_START, diff --git a/src/web/control.rs b/src/web/control.rs new file mode 100644 index 0000000..fa7b9cf --- /dev/null +++ b/src/web/control.rs @@ -0,0 +1,153 @@ +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::Weak; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +use tokio::sync::watch; + +use super::manager::WebProcessRuntime; + +/// Process-owned WEB ingress lifecycle state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum WebRuntimeLifecycle { + /// Listener orchestration has not completed. + Starting, + /// This process has no WEB listener. + NoWebListener, + /// WEB admission and request handling are active. + Running, + /// WEB admission is closed while owned work drains. + Draining, + /// All WEB ingress work drained within the deadline. + Drained, + /// The bounded shutdown deadline expired. + DeadlineExceeded, +} + +impl WebRuntimeLifecycle { + /// Returns the stable API token for this lifecycle state. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Starting => "starting", + Self::NoWebListener => "no_web_listener", + Self::Running => "running", + Self::Draining => "draining", + Self::Drained => "drained", + Self::DeadlineExceeded => "deadline_exceeded", + } + } +} + +/// One immutable lifecycle publication consumed by the control plane. +#[derive(Clone)] +pub(crate) struct WebRuntimePublication { + /// Monotonic process-local lifecycle transition number. + pub(crate) epoch: u64, + /// Current lifecycle state. + pub(crate) lifecycle: WebRuntimeLifecycle, + /// Monotonic transition time used only for relative age. + pub(crate) since: Instant, + /// Actual WEB listener addresses frozen for this process. + pub(crate) listeners: Arc<[SocketAddr]>, + /// Weak runtime access that never extends data-plane ownership. + pub(crate) runtime: Weak, +} + +/// Single-writer process lifecycle publisher for WEB ingress. +#[derive(Clone)] +pub(crate) struct WebRuntimeControl { + epoch: Arc, + tx: watch::Sender, +} + +impl WebRuntimeControl { + /// Creates the process channel in the pre-listener `starting` state. + pub(crate) fn new() -> Self { + let publication = WebRuntimePublication { + epoch: 1, + lifecycle: WebRuntimeLifecycle::Starting, + since: Instant::now(), + listeners: Arc::from([]), + runtime: Weak::new(), + }; + let (tx, _rx) = watch::channel(publication); + Self { + epoch: Arc::new(AtomicU64::new(1)), + tx, + } + } + + /// Subscribes without transferring runtime ownership to the receiver. + pub(crate) fn subscribe(&self) -> watch::Receiver { + self.tx.subscribe() + } + + /// Publishes one lifecycle transition and optional weak runtime reference. + pub(crate) fn publish( + &self, + lifecycle: WebRuntimeLifecycle, + listeners: Arc<[SocketAddr]>, + runtime: Weak, + ) { + let epoch = self.epoch.fetch_add(1, Ordering::AcqRel).saturating_add(1); + self.tx.send_replace(WebRuntimePublication { + epoch, + lifecycle, + since: Instant::now(), + listeners, + runtime, + }); + } +} + +impl Default for WebRuntimeControl { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn publication_is_monotonic_and_does_not_require_a_runtime_owner() { + let control = WebRuntimeControl::new(); + let receiver = control.subscribe(); + control.publish( + WebRuntimeLifecycle::NoWebListener, + Arc::from([]), + Weak::new(), + ); + + let publication = receiver.borrow().clone(); + assert_eq!(publication.epoch, 2); + assert_eq!(publication.lifecycle, WebRuntimeLifecycle::NoWebListener); + assert!(publication.runtime.upgrade().is_none()); + } + + #[tokio::test] + async fn publication_keeps_only_weak_runtime_ownership() { + let generation = crate::maestro::generation::test_runtime_generation( + 1, + crate::config::ProxyConfig::default(), + ); + let runtime = + WebProcessRuntime::start(Arc::new(arc_swap::ArcSwap::from(generation.clone()))); + let strong_before = Arc::strong_count(&runtime); + let control = WebRuntimeControl::new(); + control.publish( + WebRuntimeLifecycle::Running, + Arc::from([]), + Arc::downgrade(&runtime), + ); + + assert_eq!(Arc::strong_count(&runtime), strong_before); + runtime.shutdown().await; + drop(runtime); + assert!(control.subscribe().borrow().runtime.upgrade().is_none()); + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } +} diff --git a/src/web/http.rs b/src/web/http.rs index 1810074..20f86d4 100644 --- a/src/web/http.rs +++ b/src/web/http.rs @@ -17,6 +17,7 @@ use tokio::net::TcpStream; use tokio_util::sync::CancellationToken; use crate::config::{WebClientIpSource, WebRuntimeVhost}; +use crate::maestro::generation::RuntimeGeneration; use crate::web::bridge; use crate::web::manager::{ManagerError, WebProcessRuntime}; @@ -201,6 +202,7 @@ async fn handle_request( client_ip_source, trusted_proxy_cidrs, runtime, + generation, vhost, ) .await; @@ -214,6 +216,7 @@ async fn handle_root( client_ip_source: WebClientIpSource, trusted_proxy_cidrs: &[IpNetwork], runtime: Arc, + generation: Arc, vhost: Arc, ) -> HttpResponse { let (candidate, canonical) = bridge_candidate(request.uri().query()); @@ -229,7 +232,16 @@ async fn handle_root( trace.set_route(TraceRoute::Bridge); trace.set_effective_ip(client_ip); } - let bootstrap = match runtime.issue_bootstrap(Arc::clone(&profile), client_ip) { + let user_agent = request + .headers() + .get(header::USER_AGENT) + .and_then(|value| value.to_str().ok()); + let bootstrap = match runtime.issue_bootstrap_for_request( + &generation, + Arc::clone(&profile), + client_ip, + user_agent, + ) { Ok(bootstrap) => bootstrap, Err(error) => { runtime.trace().record_profile_lifecycle( @@ -248,7 +260,6 @@ async fn handle_root( trace.bind_profile(&profile, bootstrap.trace_session_id); trace.register_redaction(bootstrap.token.as_bytes()); } - let generation = runtime.active_generation(); let config = generation.config(); let page = bridge::render( &vhost.host, diff --git a/src/web/http/activity.rs b/src/web/http/activity.rs index c84d9e9..377c733 100644 --- a/src/web/http/activity.rs +++ b/src/web/http/activity.rs @@ -63,18 +63,20 @@ impl ConnectionActivity { if state.failed { return true; } - let request_protected = state + let request_deadline = state .request .as_ref() .and_then(|request| request.deadline) - .is_some_and(|deadline| now <= deadline.deadline); - let upgrade_protected = state + .map(|deadline| deadline.deadline); + let upgrade_deadline = state .upgrade .as_ref() - .is_some_and(|upgrade| now <= upgrade.deadline.deadline); - !request_protected - && !upgrade_protected - && now.saturating_duration_since(state.last_progress) >= idle + .map(|upgrade| upgrade.deadline.deadline); + let protected_until = request_deadline.into_iter().chain(upgrade_deadline).max(); + let idle_since = protected_until + .filter(|deadline| *deadline > state.last_progress) + .unwrap_or(state.last_progress); + now.saturating_duration_since(idle_since) >= idle } fn fail(&self) { @@ -390,6 +392,20 @@ mod tests { assert!(!activity.should_close(Instant::now(), Duration::from_secs(1))); } + #[test] + fn expired_operation_lease_gets_one_idle_interval_to_publish_its_result() { + let activity = ConnectionActivity::new(); + let request = RequestActivity::begin(activity.clone()).unwrap(); + let now = Instant::now(); + let _lease = request + .deadline_handle() + .lease_until(now + Duration::from_secs(1)) + .unwrap(); + + assert!(!activity.should_close(now + Duration::from_millis(1500), Duration::from_secs(1))); + assert!(activity.should_close(now + Duration::from_secs(2), Duration::from_secs(1))); + } + #[test] fn stale_request_lease_cannot_clear_a_new_request_deadline() { let activity = ConnectionActivity::new(); diff --git a/src/web/http/control_tests.rs b/src/web/http/control_tests.rs new file mode 100644 index 0000000..b3e2e4e --- /dev/null +++ b/src/web/http/control_tests.rs @@ -0,0 +1,103 @@ +use super::*; + +#[tokio::test] +async fn runtime_index_lists_and_asynchronously_closes_by_opaque_reference() { + let capability = [19u8; 32]; + let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); + let runtime = WebProcessRuntime::start(active_runtime); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let root = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.19\r\nUser-Agent: Telemt-Control-Test/1\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + let root_response = request(&listener, &runtime, root).await; + let (_, root_body) = split_response(&root_response); + let bootstrap = std::str::from_utf8(root_body) + .unwrap() + .split_once("bootstrap=\"") + .and_then(|(_, suffix)| suffix.split_once('"')) + .map(|(token, _)| token) + .unwrap(); + 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.19\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + create.extend_from_slice(&hello); + let create_response = request(&listener, &runtime, create).await; + assert!(create_response.starts_with(b"HTTP/1.1 200")); + + let page = runtime.list_sessions(SessionListRequest { + limit: 50, + cursor: None, + filter: SessionFilter::default(), + }); + assert_eq!(page.sessions.len(), 1); + assert_eq!( + page.sessions[0].user_agent.as_deref(), + Some("Telemt-Control-Test/1") + ); + let session_ref = page.sessions[0].session_ref.clone(); + let trace_session_id = runtime.parse_session_ref(&session_ref).unwrap(); + let noncanonical_session_ref = format!("ws1.{}.000000000000000A", runtime.runtime_instance()); + assert_eq!( + runtime.parse_session_ref(&noncanonical_session_ref), + Err(SessionRefError::Invalid) + ); + let noncanonical_operation_id = format!("wo1.{}.000000000000000A", runtime.runtime_instance()); + assert!(matches!( + runtime.control_operation(&noncanonical_operation_id), + Err(ControlError::InvalidOperation) + )); + let nonmatching = runtime + .start_close_operation( + runtime.runtime_instance(), + CloseOperationSelector::Filter(SessionFilter { + state: Some("healthy".to_string()), + ..SessionFilter::default() + }), + ) + .unwrap(); + let mut nonmatching_status = None; + for _ in 0..32 { + let status = runtime + .control_operation(&nonmatching.operation_id) + .unwrap(); + if serde_json::to_value(&status).unwrap()["state"] == "completed" { + nonmatching_status = Some(status); + break; + } + tokio::task::yield_now().await; + } + let nonmatching_status = nonmatching_status.expect("filtered close operation completed"); + assert_eq!(nonmatching_status.matched, 0); + assert_eq!(nonmatching_status.close_signalled, 0); + assert!(matches!( + runtime.session_detail(trace_session_id), + SessionDetail::Active(_) + )); + let operation = runtime + .start_close_operation( + runtime.runtime_instance(), + CloseOperationSelector::Refs(vec![trace_session_id]), + ) + .unwrap(); + for _ in 0..32 { + let status = runtime.control_operation(&operation.operation_id).unwrap(); + if serde_json::to_value(&status).unwrap()["state"] == "completed" { + break; + } + tokio::task::yield_now().await; + } + + assert!(matches!( + runtime.session_detail(trace_session_id), + SessionDetail::Gone { .. } + )); + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} diff --git a/src/web/http/session.rs b/src/web/http/session.rs index 9712e1f..41fcbb6 100644 --- a/src/web/http/session.rs +++ b/src/web/http/session.rs @@ -68,7 +68,7 @@ pub(super) async fn handle_session( return serve_decoy(request, vhost, true, &runtime).await; }; let ip_learning_eligible = carrier_ip_learning_eligible(&request, client_ip); - let Some((trace_session_id, profile, frozen_body_timeout)) = + let Some((trace_session_id, profile, body_timeout)) = runtime.bootstrap_trace_identity(token_hash, &vhost.host) else { return serve_decoy(request, vhost, true, &runtime).await; @@ -77,9 +77,6 @@ pub(super) async fn handle_session( trace.set_route(TraceRoute::Session); trace.bind_profile(&profile, trace_session_id); } - let body_timeout = frozen_body_timeout.unwrap_or_else(|| { - Duration::from_secs(runtime.active_generation().config().web.timeouts.body_secs) - }); let CollectedBody { request, body, diff --git a/src/web/http/session_policy_tests.rs b/src/web/http/session_policy_tests.rs index 24176ff..5d52b6e 100644 --- a/src/web/http/session_policy_tests.rs +++ b/src/web/http/session_policy_tests.rs @@ -75,6 +75,61 @@ async fn request_with_body_delay( response } +#[tokio::test] +async fn issued_bootstrap_timeouts_survive_reload_before_session_creation() { + let capability = [20u8; 32]; + let mut initial_config = runtime_config(capability, WebCarrier::Https); + initial_config.web.timeouts.body_secs = 3; + initial_config.web.timeouts.long_poll_secs = 3; + initial_config.web.timeouts.bootstrap_lifetime_secs = 5; + let generation = test_runtime_generation(1, initial_config); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability); + let root = format!( + "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n" + ) + .into_bytes(); + let root_response = request(&listener, &runtime, root).await; + let (_, root_body) = split_response(&root_response); + let bootstrap = std::str::from_utf8(root_body) + .unwrap() + .split_once("bootstrap=\"") + .and_then(|(_, suffix)| suffix.split_once('"')) + .map(|(token, _)| token.to_string()) + .unwrap(); + + let mut replacement_config = runtime_config(capability, WebCarrier::Https); + replacement_config.web.timeouts.body_secs = 1; + replacement_config.web.timeouts.long_poll_secs = 1; + replacement_config.web.timeouts.bootstrap_lifetime_secs = 1; + let replacement = test_runtime_generation(2, replacement_config); + active_runtime.store(Arc::clone(&replacement)); + + let hello = frame::encode(FrameType::Hello, 0, &[1]); + let create_head = 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\nContent-Length: {}\r\nConnection: close\r\n\r\n", + hello.len() + ) + .into_bytes(); + let create_response = request_with_body_delay( + &listener, + &runtime, + create_head, + &hello, + std::time::Duration::from_millis(1200), + ) + .await; + assert!(create_response.starts_with(b"HTTP/1.1 200")); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} + #[tokio::test] async fn live_session_body_and_closed_token_timeouts_survive_reload() { let capability = [21u8; 32]; diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs index 2be00ab..f8414c2 100644 --- a/src/web/http/tests.rs +++ b/src/web/http/tests.rs @@ -15,7 +15,10 @@ use crate::config::{ }; use crate::maestro::generation::test_runtime_generation; use crate::web::frame::{self, FrameType}; -use crate::web::manager::WebProcessRuntime; +use crate::web::manager::{ + CloseOperationSelector, ControlError, SessionDetail, SessionFilter, SessionListRequest, + SessionRefError, WebProcessRuntime, +}; #[path = "legacy_tests.rs"] mod legacy_tests; @@ -24,6 +27,9 @@ mod negotiation_tests; // Reload-stability tests for session-owned timeout policy. #[path = "session_policy_tests.rs"] mod session_policy_tests; +// Runtime control integration stays separate from carrier protocol scenarios. +#[path = "control_tests.rs"] +mod control_tests; const TEST_CARRIER_DEADLINES_SECS: [u64; 4] = [3, 5, 8, 12]; @@ -394,6 +400,26 @@ async fn unused_bootstrap_survives_equivalent_runtime_generation_swap() { replacement.stop_background_tasks().await; } +#[tokio::test] +async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() { + let initial = test_runtime_generation(1, runtime_config([21; 32], WebCarrier::Https)); + let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial))); + let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime)); + let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone(); + let replacement = test_runtime_generation(2, runtime_config([22; 32], WebCarrier::HttpsLanes)); + active_runtime.store(Arc::clone(&replacement)); + + let result = + runtime.issue_bootstrap_for_generation(&initial, profile, "192.0.2.10".parse().unwrap()); + + assert!(result.is_ok()); + runtime.shutdown().await; + initial.stop_sessions().await; + initial.stop_background_tasks().await; + replacement.stop_sessions().await; + replacement.stop_background_tasks().await; +} + #[tokio::test] async fn unused_bootstrap_is_rejected_after_profile_identity_change() { let capability = [11u8; 32]; diff --git a/src/web/manager.rs b/src/web/manager.rs index a8ec896..e249858 100644 --- a/src/web/manager.rs +++ b/src/web/manager.rs @@ -38,8 +38,16 @@ pub(crate) use lifecycle::WebShutdownOutcome; mod budget; // WebSocket admission, replacement, and liveness are process-scoped. mod websocket; +// Bounded read-only snapshots and opaque session references serve the API. +mod status; +pub(crate) use status::{ + SessionDetail, SessionFilter, SessionListRequest, SessionRefError, WebRuntimeStatus, +}; +// Asynchronous bounded close operations isolate mutation lifecycle from HTTP requests. +mod control; pub(crate) use budget::WebSocketBudgetLease; use budget::{WebDataBudget, WebSocketBudgetClass}; +pub(crate) use control::{CloseOperationSelector, ControlError}; pub(crate) use negotiation::{ CarrierCapabilities, CarrierClientClass, CarrierFailure, CarrierLearningContext, CarrierRequest, }; @@ -128,6 +136,7 @@ pub(crate) struct BootstrapResult { /// Process-owned bounded WEB credential, session, and memory coordinator. pub(crate) struct WebProcessRuntime { + runtime_instance: Arc, active_runtime: Arc>, trace: Arc, limits: WebLimitsConfig, @@ -147,6 +156,8 @@ pub(crate) struct WebProcessRuntime { websocket_clock: std::time::Instant, websocket_notify: Arc, data_budget: Arc, + control_operations: Mutex, + next_control_operation_id: AtomicU64, shutdown: CancellationToken, tasks: TaskTracker, sessions_created: AtomicU64, @@ -172,7 +183,9 @@ impl WebProcessRuntime { active_runtime: Arc>, trace: Arc, ) -> Arc { - let config = active_runtime.load().config(); + let initial_generation = active_runtime.load_full(); + let config = initial_generation.config(); + trace.apply_policy(initial_generation.id, &config.web.debug); let limits = config.web.limits.clone(); let learning_capacity = limits.max_carrier_learning_entries; let mut carrier_learning = learning::CarrierLearning::new(learning_capacity); @@ -188,6 +201,7 @@ impl WebProcessRuntime { let lane_poll_limit = limits.max_http_handlers / 2; let lane_aux_poll_limit = (lane_poll_limit / 2).max(1); let runtime = Arc::new(Self { + runtime_instance: Arc::from(format!("{:032x}", rand::random::())), active_runtime, trace, http_connections: Arc::new(Semaphore::new(limits.max_http_connections)), @@ -203,8 +217,10 @@ impl WebProcessRuntime { websocket_clock: std::time::Instant::now(), websocket_notify: Arc::new(Notify::new()), data_budget: WebDataBudget::new(limits.clone()), + control_operations: Mutex::new(control::ControlOperationRegistry::default()), + next_control_operation_id: AtomicU64::new(1), limits, - state: Mutex::new(ManagerState::default()), + state: Mutex::new(ManagerState::new(initial_generation.id, config.web.enabled)), stream_admission: Mutex::new(StreamAdmissionState::default()), learning: Mutex::new(carrier_learning), shutdown: CancellationToken::new(), @@ -229,8 +245,11 @@ impl WebProcessRuntime { let Some(runtime) = weak.upgrade() else { break; }; - let policy = runtime.active_generation().config().web.debug.clone(); - runtime.trace.apply_policy(&policy); + let generation = runtime.active_generation(); + let policy = generation.config().web.debug.clone(); + runtime + .trace + .apply_policy(generation.id, &policy); runtime.cleanup(); } } @@ -244,6 +263,11 @@ impl WebProcessRuntime { self.active_runtime.load_full() } + /// Returns the random process-instance fence used by control-plane references. + pub(crate) fn runtime_instance(&self) -> &str { + &self.runtime_instance + } + /// Returns the process-owned WEB debug trace store. pub(crate) fn trace(&self) -> &Arc { &self.trace diff --git a/src/web/manager/budget.rs b/src/web/manager/budget.rs index 3c979c6..149dd05 100644 --- a/src/web/manager/budget.rs +++ b/src/web/manager/budget.rs @@ -52,10 +52,18 @@ pub(crate) struct WebDataBudgetSnapshot { pub(crate) queue_bytes: usize, /// Total queue items currently retained. pub(crate) queue_items: usize, + /// Control bytes included in the queue total. + pub(crate) queue_control_bytes: usize, + /// Control items included in the queue total. + pub(crate) queue_control_items: usize, /// Total WebSocket bytes currently retained. pub(crate) websocket_bytes: usize, /// Largest combined byte usage observed since process start. pub(crate) high_water_bytes: usize, + /// Distinct profile owners currently charged. + pub(crate) owners: usize, + /// Whether shutdown closed this allocation authority. + pub(crate) closed: bool, } /// Bounded owner-usage view captured before WebSocket registry selection. @@ -259,11 +267,29 @@ impl WebDataBudget { WebDataBudgetSnapshot { queue_bytes: state.queue_bytes, queue_items: state.queue_items, + queue_control_bytes: state.queue_control_bytes, + queue_control_items: state.queue_control_items, websocket_bytes: state.websocket_bytes, high_water_bytes: state.high_water_bytes, + owners: state.owner_bytes.len(), + closed: state.closed, } } + pub(super) fn try_snapshot(&self) -> Option { + let state = self.state.try_lock()?; + Some(WebDataBudgetSnapshot { + queue_bytes: state.queue_bytes, + queue_items: state.queue_items, + queue_control_bytes: state.queue_control_bytes, + queue_control_items: state.queue_control_items, + websocket_bytes: state.websocket_bytes, + high_water_bytes: state.high_water_bytes, + owners: state.owner_bytes.len(), + closed: state.closed, + }) + } + pub(super) fn close(&self) { self.state.lock().closed = true; self.notify.notify_waiters(); diff --git a/src/web/manager/carrier_learning.rs b/src/web/manager/carrier_learning.rs index 991c92d..233872f 100644 --- a/src/web/manager/carrier_learning.rs +++ b/src/web/manager/carrier_learning.rs @@ -163,6 +163,34 @@ pub(super) struct CarrierLearning { 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 { @@ -177,6 +205,23 @@ impl CarrierLearning { } } + 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. pub(super) fn apply_policy( &mut self, @@ -424,6 +469,51 @@ impl CarrierLearning { } } +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 { configured .iter() diff --git a/src/web/manager/carrier_learning/tests.rs b/src/web/manager/carrier_learning/tests.rs index 8fb83d7..8f5f94f 100644 --- a/src/web/manager/carrier_learning/tests.rs +++ b/src/web/manager/carrier_learning/tests.rs @@ -220,3 +220,52 @@ fn fifo_metadata_stays_within_the_entry_capacity() { assert!(learning.insertion_order.len() <= 3); } } + +#[tokio::test] +async fn explicit_reset_preserves_policy_and_rejects_old_epoch_outcomes() { + let generation = crate::maestro::generation::test_runtime_generation( + 1, + crate::config::ProxyConfig::default(), + ); + let runtime = crate::web::manager::WebProcessRuntime::start(std::sync::Arc::new( + arc_swap::ArcSwap::from(generation.clone()), + )); + 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(); + learning.record_chain(now, epoch, context(7), &[], WebCarrier::Websocket); + epoch + }; + + let outcome = runtime.reset_carrier_learning().unwrap(); + { + let mut learning = runtime.learning.lock(); + learning.record_chain( + Instant::now(), + old_epoch, + context(7), + &[], + WebCarrier::Https, + ); + let status = learning.status(Instant::now()); + assert!(status.enabled); + assert_eq!( + status.aggressiveness, + WebCarrierNegotiationAggressiveness::Aggressive + ); + assert_eq!(status.entries, 0); + assert_eq!(status.epoch, Some(outcome.epoch)); + } + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; +} diff --git a/src/web/manager/control.rs b/src/web/manager/control.rs new file mode 100644 index 0000000..d28a54f --- /dev/null +++ b/src/web/manager/control.rs @@ -0,0 +1,441 @@ +use std::collections::{BTreeSet, VecDeque}; +use std::sync::Arc; +use std::sync::atomic::Ordering; +use std::time::Instant; + +use serde::Serialize; + +use super::status::immutable_matches; +use super::{SessionFilter, WebProcessRuntime}; + +const OPERATION_REF_VERSION: &str = "wo1"; +const OPERATION_RETENTION: usize = 32; +const CLOSE_CHUNK: usize = 128; + +/// Validated bulk-close selector. +#[derive(Clone)] +pub(crate) enum CloseOperationSelector { + /// Exact logical session references resolved by the API. + Refs(Vec), + /// Point-in-time sessions matching a bounded filter. + Filter(SessionFilter), + /// Every logical session at or below the submission high-water mark. + All, +} + +/// Stable close-operation state. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ControlOperationState { + /// Accepted but not yet executing. + Queued, + /// Scanning the bounded point-in-time registry. + Running, + /// Finished the complete bounded scan. + Completed, + /// Stopped because process shutdown began. + Cancelled, + /// Stopped on one sanitized internal failure. + Failed, +} + +/// Retained status for one asynchronous close operation. +#[derive(Clone, Serialize)] +pub(crate) struct ControlOperationStatus { + /// Opaque process-fenced operation reference. + pub(crate) operation_id: String, + /// Current lifecycle state. + pub(crate) state: ControlOperationState, + /// Highest logical session eligible for this point-in-time operation. + pub(crate) high_water_session_ref: Option, + /// Exact submitted reference count, or zero for filter/all selectors. + pub(crate) requested: usize, + /// Registry candidates visited so far. + pub(crate) scanned: usize, + /// Candidates that matched the complete selector. + pub(crate) matched: usize, + /// Matching session incarnations sent a close signal. + pub(crate) close_signalled: usize, + /// Matching candidates replaced or closed before signalling. + pub(crate) conflicted: usize, + /// Wall-clock creation timestamp for operator correlation. + pub(crate) created_epoch_millis: u64, + /// Wall-clock timestamp of the latest status mutation. + pub(crate) updated_epoch_millis: u64, + /// Stable sanitized failure token. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) failure: Option<&'static str>, +} + +/// Runtime control validation failure. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ControlError { + /// The supplied process-instance fence is stale. + StaleInstance, + /// The selector is empty or exceeds its bound. + InvalidSelector, + /// Close-all was requested before issuance stopped. + IssuanceEnabled, + /// The single operation slot is occupied. + OperationInProgress, + /// The operation reference is not canonical. + InvalidOperation, + /// The canonical operation is outside retained history. + OperationNotFound, + /// Process shutdown has closed the mutation gate. + Closed, +} + +/// Single-slot execution gate with bounded terminal history. +#[derive(Default)] +pub(super) struct ControlOperationRegistry { + active: Option, + retained: VecDeque<(u64, ControlOperationStatus)>, + closed: bool, +} + +struct WorkCandidate { + trace_session_id: u64, + session: Arc, + bootstrap_hash: super::TokenHash, +} + +impl WebProcessRuntime { + /// Acquires the process mutation gate without permitting post-shutdown work. + pub(super) fn control_mutation_guard( + &self, + ) -> Result, ControlError> { + let operations = self.control_operations.lock(); + if operations.closed || self.shutdown.is_cancelled() { + return Err(ControlError::Closed); + } + Ok(operations) + } + + /// Clears debug records under the process mutation gate. + pub(crate) fn clear_debug(&self) -> Result { + let _control = self.control_mutation_guard()?; + Ok(self.trace.clear()) + } + + /// Starts one bounded point-in-time close sweep. + pub(crate) fn start_close_operation( + self: &Arc, + runtime_instance: &str, + selector: CloseOperationSelector, + ) -> Result { + if runtime_instance != self.runtime_instance() { + return Err(ControlError::StaleInstance); + } + if matches!(&selector, CloseOperationSelector::Refs(refs) if refs.is_empty() || refs.len() > 200) + || matches!(&selector, CloseOperationSelector::Filter(filter) if filter.is_empty()) + { + return Err(ControlError::InvalidSelector); + } + let generation = self.active_generation(); + let web_enabled = generation.config().web.enabled; + let (high_water, issuance_enabled) = { + let mut state = self.state.lock(); + state.apply_issuance_policy(generation.id, web_enabled); + ( + state.session_index.last_key_value().map(|(id, _)| *id), + state.issuance_enabled, + ) + }; + if matches!(selector, CloseOperationSelector::All) && issuance_enabled { + return Err(ControlError::IssuanceEnabled); + } + if self.shutdown.is_cancelled() { + return Err(ControlError::Closed); + } + let sequence = self + .next_control_operation_id + .fetch_add(1, Ordering::Relaxed); + let operation_id = self.operation_ref(sequence); + let now = crate::web::trace::store_epoch_millis(); + let requested = match &selector { + CloseOperationSelector::Refs(refs) => refs.len(), + CloseOperationSelector::Filter(_) | CloseOperationSelector::All => 0, + }; + let status = ControlOperationStatus { + operation_id, + state: ControlOperationState::Queued, + high_water_session_ref: high_water.map(|id| self.session_ref(id)), + requested, + scanned: 0, + matched: 0, + close_signalled: 0, + conflicted: 0, + created_epoch_millis: now, + updated_epoch_millis: now, + failure: None, + }; + let tracked = { + let mut operations = self.control_operations.lock(); + if operations.closed || self.shutdown.is_cancelled() { + return Err(ControlError::Closed); + } + if operations.active.is_some() { + return Err(ControlError::OperationInProgress); + } + operations.active = Some(sequence); + operations.retained.push_back((sequence, status.clone())); + trim_operations(&mut operations); + let weak = Arc::downgrade(self); + self.tasks.track_future(async move { + if let Some(runtime) = weak.upgrade() { + runtime + .run_close_operation(sequence, high_water, selector) + .await; + } + }) + }; + drop(tokio::spawn(tracked)); + Ok(status) + } + + /// Returns one retained operation status under the process-instance fence. + pub(crate) fn control_operation( + &self, + operation_id: &str, + ) -> Result { + let sequence = self.parse_operation_ref(operation_id)?; + self.control_operations + .lock() + .retained + .iter() + .find_map(|(id, status)| (*id == sequence).then(|| status.clone())) + .ok_or(ControlError::OperationNotFound) + } + + /// Prevents new control work from racing process-runtime shutdown. + pub(super) fn close_control_submission_gate(&self) { + self.control_operations.lock().closed = true; + } + + async fn run_close_operation( + self: &Arc, + sequence: u64, + high_water: Option, + selector: CloseOperationSelector, + ) { + self.update_operation(sequence, |status| { + status.state = ControlOperationState::Running + }); + let refs = match &selector { + CloseOperationSelector::Refs(refs) => { + Some(refs.iter().copied().collect::>()) + } + CloseOperationSelector::Filter(_) | CloseOperationSelector::All => None, + }; + let filter = match &selector { + CloseOperationSelector::Filter(filter) => Some(filter), + CloseOperationSelector::Refs(_) | CloseOperationSelector::All => None, + }; + let mut cursor = None; + loop { + if self.shutdown.is_cancelled() { + self.finish_operation(sequence, ControlOperationState::Cancelled, None); + return; + } + let mut direct = Vec::new(); + let mut state_filtered = Vec::new(); + let (scanned, next_cursor, reached_end) = { + let mut state = self.state.lock(); + let mut scanned = 0usize; + let mut next_cursor = cursor; + let mut reached_end = true; + let ids = state + .session_index + .range(( + cursor.map_or(std::ops::Bound::Unbounded, std::ops::Bound::Excluded), + std::ops::Bound::Unbounded, + )) + .take(CLOSE_CHUNK) + .map(|(id, _)| *id) + .collect::>(); + for trace_session_id in ids { + if high_water.is_some_and(|high_water| trace_session_id > high_water) { + break; + } + scanned += 1; + next_cursor = Some(trace_session_id); + let Some(index) = state.session_index.get(&trace_session_id) else { + continue; + }; + let selected = refs + .as_ref() + .is_none_or(|refs| refs.contains(&trace_session_id)); + let Some(session) = selected + .then(|| state.sessions.get(&index.session_hash).cloned()) + .flatten() + else { + continue; + }; + if filter.is_some_and(|filter| !immutable_matches(&session, index, filter)) { + continue; + } + let candidate = WorkCandidate { + trace_session_id, + session, + bootstrap_hash: index.bootstrap_hash, + }; + if filter.and_then(|filter| filter.state.as_ref()).is_some() { + state_filtered.push(candidate); + } else { + mark_close_locked(&mut state, &candidate); + direct.push(candidate); + } + } + if scanned == CLOSE_CHUNK { + reached_end = false; + } + (scanned, next_cursor, reached_end) + }; + self.update_operation(sequence, |status| { + status.scanned = status.scanned.saturating_add(scanned); + status.matched = status.matched.saturating_add(direct.len()); + }); + for candidate in direct { + candidate.session.close(); + self.update_operation(sequence, |status| { + status.close_signalled = status.close_signalled.saturating_add(1) + }); + } + if let Some(expected_state) = filter.and_then(|filter| filter.state.as_deref()) { + for candidate in state_filtered { + let matches_state = candidate + .session + .try_status(Instant::now()) + .is_some_and(|status| status.state == expected_state); + if !matches_state { + continue; + } + let session = { + let mut state = self.state.lock(); + let current = state + .session_index + .get(&candidate.trace_session_id) + .and_then(|index| state.sessions.get(&index.session_hash)) + .filter(|current| Arc::ptr_eq(current, &candidate.session)) + .cloned(); + if current.is_some() { + mark_close_locked(&mut state, &candidate); + } + current + }; + if let Some(session) = session { + session.close(); + self.update_operation(sequence, |status| { + status.matched = status.matched.saturating_add(1); + status.close_signalled = status.close_signalled.saturating_add(1); + }); + } else { + self.update_operation(sequence, |status| { + status.matched = status.matched.saturating_add(1); + status.conflicted = status.conflicted.saturating_add(1); + }); + } + } + } + cursor = next_cursor; + if reached_end || cursor.is_none() { + break; + } + tokio::task::yield_now().await; + } + self.finish_operation(sequence, ControlOperationState::Completed, None); + } + + fn update_operation(&self, sequence: u64, update: impl FnOnce(&mut ControlOperationStatus)) { + let mut operations = self.control_operations.lock(); + if let Some((_, status)) = operations + .retained + .iter_mut() + .find(|(id, _)| *id == sequence) + { + update(status); + status.updated_epoch_millis = crate::web::trace::store_epoch_millis(); + } + } + + fn finish_operation( + &self, + sequence: u64, + state: ControlOperationState, + failure: Option<&'static str>, + ) { + let mut operations = self.control_operations.lock(); + if let Some((_, status)) = operations + .retained + .iter_mut() + .find(|(id, _)| *id == sequence) + { + status.state = failure.map_or(state, |_| ControlOperationState::Failed); + status.failure = failure; + status.updated_epoch_millis = crate::web::trace::store_epoch_millis(); + } + if operations.active == Some(sequence) { + operations.active = None; + } + } + + fn operation_ref(&self, sequence: u64) -> String { + format!( + "{OPERATION_REF_VERSION}.{}.{sequence:016x}", + self.runtime_instance() + ) + } + + fn parse_operation_ref(&self, value: &str) -> Result { + let mut parts = value.split('.'); + if parts.next() != Some(OPERATION_REF_VERSION) { + return Err(ControlError::InvalidOperation); + } + let instance = parts.next().ok_or(ControlError::InvalidOperation)?; + if instance.len() != 32 + || !instance + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + return Err(ControlError::InvalidOperation); + } + let sequence = parts + .next() + .filter(|value| { + value.len() == 16 + && value + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + }) + .and_then(|value| u64::from_str_radix(value, 16).ok()) + .filter(|value| *value != 0) + .ok_or(ControlError::InvalidOperation)?; + if parts.next().is_some() { + return Err(ControlError::InvalidOperation); + } + if instance != self.runtime_instance() { + return Err(ControlError::StaleInstance); + } + Ok(sequence) + } +} + +fn mark_close_locked(state: &mut super::state::ManagerState, candidate: &WorkCandidate) { + if let Some(bootstrap) = state.bootstraps.get_mut(&candidate.bootstrap_hash) { + bootstrap.close_requested = true; + } +} + +fn trim_operations(operations: &mut ControlOperationRegistry) { + while operations.retained.len() > OPERATION_RETENTION { + if operations + .retained + .front() + .is_some_and(|(id, _)| Some(*id) == operations.active) + { + break; + } + operations.retained.pop_front(); + } +} diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs index cdfdb4b..ecc3482 100644 --- a/src/web/manager/credentials.rs +++ b/src/web/manager/credentials.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; +use sha2::{Digest, Sha256}; use zeroize::Zeroizing; use super::state::{ @@ -11,16 +12,50 @@ use super::state::{ }; use super::{BootstrapResult, ManagerError, TOKEN_BYTES, TokenHash, WebProcessRuntime}; use crate::config::WebRuntimeProfile; +use crate::maestro::generation::RuntimeGeneration; use crate::web::session::WebSession; impl WebProcessRuntime { /// Issues a one-use bootstrap credential for an active compatible profile. + #[cfg(test)] pub(crate) fn issue_bootstrap( &self, profile: Arc, client_ip: IpAddr, ) -> std::result::Result { let generation = self.active_generation(); + self.issue_bootstrap_inner(&generation, profile, client_ip, None) + } + + /// Issues one bootstrap against the generation that selected the bridge profile. + #[cfg(test)] + pub(crate) fn issue_bootstrap_for_generation( + &self, + generation: &Arc, + profile: Arc, + client_ip: IpAddr, + ) -> std::result::Result { + self.issue_bootstrap_inner(generation, profile, client_ip, None) + } + + /// Issues one bridge bootstrap with bounded non-secret request metadata. + pub(crate) fn issue_bootstrap_for_request( + &self, + generation: &Arc, + profile: Arc, + client_ip: IpAddr, + user_agent: Option<&str>, + ) -> std::result::Result { + self.issue_bootstrap_inner(generation, profile, client_ip, user_agent) + } + + fn issue_bootstrap_inner( + &self, + generation: &Arc, + profile: Arc, + client_ip: IpAddr, + user_agent: Option<&str>, + ) -> std::result::Result { let config = generation.config(); let profile = config .web @@ -34,7 +69,9 @@ impl WebProcessRuntime { let now = Instant::now(); let mut state = self.state.lock(); remove_expired_locked(&mut state, now); + state.apply_issuance_policy(generation.id, config.web.enabled); if state.closed + || !state.issuance_enabled || state .bootstraps_per_ip .get(&client_ip) @@ -57,11 +94,12 @@ impl WebProcessRuntime { self.limit_hits.fetch_add(1, Ordering::Relaxed); return Err(ManagerError::Limit); } - let Some((token, hash)) = new_unique_token(&generation, &state) else { + let Some((token, hash)) = new_unique_token(generation, &state) else { self.limit_hits.fetch_add(1, Ordering::Relaxed); return Err(ManagerError::Limit); }; let trace_session_id = self.trace.next_session_id(); + let (user_agent, user_agent_id) = bounded_user_agent(user_agent); state.bootstraps.insert( hash, Bootstrap { @@ -69,7 +107,10 @@ impl WebProcessRuntime { issued_at: now, issuance_ip: client_ip, profile, + timeouts: config.web.timeouts.clone(), trace_session_id, + user_agent, + user_agent_id, body_digest: [0; TOKEN_BYTES], session_token: Zeroizing::new(String::new()), session: None, @@ -110,12 +151,12 @@ impl WebProcessRuntime { }) } - /// Resolves bootstrap trace identity and the frozen live-session body timeout. + /// Resolves bootstrap trace identity and its issuance-frozen body timeout. pub(crate) fn bootstrap_trace_identity( &self, hash: TokenHash, host: &str, - ) -> Option<(u64, Arc, Option)> { + ) -> Option<(u64, Arc, Duration)> { let now = Instant::now(); self.state .lock() @@ -126,10 +167,10 @@ impl WebProcessRuntime { ( entry.trace_session_id, Arc::clone(&entry.profile), - entry - .session - .as_ref() - .map(|session| Duration::from_secs(session.timeouts().body_secs)), + entry.session.as_ref().map_or_else( + || Duration::from_secs(entry.timeouts.body_secs), + |session| Duration::from_secs(session.timeouts().body_secs), + ), ) }) } @@ -185,3 +226,30 @@ impl WebProcessRuntime { closed.then_some(()).ok_or(ManagerError::Authentication) } } + +fn bounded_user_agent(value: Option<&str>) -> (Option>, Option<[u8; 16]>) { + const DISPLAY_BYTES: usize = 256; + const HASH_CONTEXT: &[u8] = b"telemt-web-user-agent-v1\0"; + let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else { + return (None, None); + }; + let mut digest = Sha256::new(); + digest.update(HASH_CONTEXT); + digest.update(value.as_bytes()); + let digest = digest.finalize(); + let mut id = [0; 16]; + id.copy_from_slice(&digest[..16]); + let mut display = String::with_capacity(value.len().min(DISPLAY_BYTES)); + for character in value.chars() { + let character = if character.is_control() { + '\u{fffd}' + } else { + character + }; + if display.len().saturating_add(character.len_utf8()) > DISPLAY_BYTES { + break; + } + display.push(character); + } + (Some(Arc::from(display)), Some(id)) +} diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs index c10e7b1..4985495 100644 --- a/src/web/manager/lifecycle.rs +++ b/src/web/manager/lifecycle.rs @@ -6,7 +6,8 @@ use tokio::time::Instant as TokioInstant; use tracing::{info, warn}; use super::state::{ - decrement_map, remember_closed_token_locked, remove_bootstrap_locked, remove_expired_locked, + decrement_map, remember_closed_session_locked, remember_closed_token_locked, + remove_bootstrap_locked, remove_expired_locked, }; use super::{ProfileKey, TokenHash, WebProcessRuntime}; @@ -37,9 +38,9 @@ impl WebProcessRuntime { closed_token_lifetime: Duration, ) { let mut state = self.state.lock(); - if state.sessions.remove(&hash).is_none() { + let Some(session) = state.sessions.remove(&hash) else { return; - } + }; decrement_map(&mut state.sessions_per_ip, &client_ip); decrement_map(&mut state.sessions_per_profile, &profile_key); remember_closed_token_locked( @@ -49,6 +50,21 @@ impl WebProcessRuntime { closed_token_lifetime, self.limits.max_sessions_global.saturating_mul(16), ); + let trace_session_id = session.trace_session_id(); + if state + .session_index + .get(&trace_session_id) + .is_some_and(|index| index.session_hash == hash) + { + state.session_index.remove(&trace_session_id); + remember_closed_session_locked( + &mut state, + trace_session_id, + session.carrier_attempt(), + closed_token_lifetime, + self.limits.max_sessions_global, + ); + } let bootstrap_hashes = state .bootstraps .iter() @@ -70,6 +86,7 @@ impl WebProcessRuntime { pub(crate) fn begin_shutdown(self: &std::sync::Arc) -> WebShutdownDrain { let started = TokioInstant::now(); self.shutdown.cancel(); + self.close_control_submission_gate(); self.close_websockets(); self.data_budget.close(); self.http_connections.close(); @@ -141,6 +158,7 @@ impl WebProcessRuntime { drop(learning); let (sessions, expired_chains) = { let mut state = self.state.lock(); + state.apply_issuance_policy(generation.id, config.enabled); let expired = state .bootstraps .iter() @@ -293,6 +311,7 @@ mod tests { use super::*; use crate::config::ProxyConfig; use crate::maestro::generation::test_runtime_generation; + use crate::web::manager::{CloseOperationSelector, ControlError}; struct DropProbe { polls: Arc, @@ -375,6 +394,44 @@ mod tests { generation.stop_background_tasks().await; } + #[tokio::test] + async fn initial_trace_policy_is_attributed_to_active_generation() { + let (runtime, generation) = runtime(); + + assert_eq!(runtime.trace().status().policy_generation, generation.id); + + runtime.shutdown().await; + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } + + #[tokio::test] + async fn shutdown_closes_the_control_submission_gate() { + let (runtime, generation) = runtime(); + let drain = runtime.begin_shutdown(); + + assert!(matches!( + runtime.start_close_operation( + runtime.runtime_instance(), + CloseOperationSelector::Refs(vec![1]), + ), + Err(ControlError::Closed) + )); + assert!(matches!( + runtime.reset_carrier_learning(), + Err(crate::web::manager::ManagerError::Closed) + )); + assert!(matches!(runtime.clear_debug(), Err(ControlError::Closed))); + assert_eq!( + drain + .wait_until(TokioInstant::now() + Duration::from_secs(1)) + .await, + WebShutdownOutcome::Drained + ); + generation.stop_sessions().await; + generation.stop_background_tasks().await; + } + #[tokio::test(start_paused = true)] async fn expired_deadline_still_closes_every_runtime_gate() { let (runtime, generation) = runtime(); diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs index 0c41b5f..4e068f9 100644 --- a/src/web/manager/session_creation.rs +++ b/src/web/manager/session_creation.rs @@ -10,8 +10,8 @@ use zeroize::Zeroizing; use super::negotiation::carrier_attempt_deadline_index; use super::session_admission::admit_initial; use super::state::{ - CarrierChainPhase, decrement_map, matching_profile, new_unique_token, profile_key, - remember_closed_token_locked, remove_expired_locked, + CarrierChainPhase, LiveSessionIndex, decrement_map, matching_profile, new_unique_token, + profile_key, remember_closed_token_locked, remove_expired_locked, }; use super::{ CarrierLearningContext, CarrierRequest, CreateResult, ManagerError, TokenHash, @@ -56,6 +56,10 @@ impl WebProcessRuntime { let now = Instant::now(); let mut state = self.state.lock(); remove_expired_locked(&mut state, now); + state.apply_issuance_policy(generation.id, config.web.enabled); + if state.closed || !state.issuance_enabled { + return Err(ManagerError::Closed); + } let Some(entry) = state.bootstraps.get(&bootstrap_hash) else { return Err(ManagerError::Authentication); }; @@ -201,9 +205,7 @@ impl WebProcessRuntime { } let trace_session_id = entry.trace_session_id; let issued_profile = Arc::clone(&entry.profile); - if state.closed || !config.web.enabled { - return Err(ManagerError::Closed); - } + let issued_timeouts = entry.timeouts.clone(); let profile = config .web .runtime @@ -303,33 +305,39 @@ impl WebProcessRuntime { learning_context, carrier_request.is_automatic(), self.limits.clone(), - config.web.timeouts.clone(), + issued_timeouts.clone(), ); state.sessions.insert(session_hash, Arc::clone(&session)); *state.sessions_per_ip.entry(client_ip).or_insert(0) += 1; *state.sessions_per_profile.entry(profile_key).or_insert(0) += 1; - let entry = state - .bootstraps - .get_mut(&bootstrap_hash) - .ok_or(ManagerError::Authentication)?; - entry.used = true; - entry.body_digest = body_digest; - entry.session_token = Zeroizing::new(session_token.clone()); - entry.session = Some(Arc::clone(&session)); - entry.carrier_request = Some(carrier_request); - entry.carrier_candidates = candidates.into(); - entry.carrier_scores = scores; - entry.carrier_attempt = 1; - entry.carrier_phase = CarrierChainPhase::Provisional; - entry.carrier_started_at = carrier_request.is_automatic().then_some(now); - entry.carrier_deadline_at = carrier_deadline_at; - entry.carrier_failures = [None; 3]; - entry.carrier_learning_epoch = learning_epoch.unwrap_or(0); - entry.expires_at = now + Duration::from_secs(config.web.timeouts.bootstrap_lifetime_secs); - entry.session_client_ip = Some(client_ip); - entry.session_ip_learning_eligible = ip_learning_eligible; - let issuance_ip = entry.issuance_ip; - let candidate_count = u8::try_from(entry.carrier_candidates.len()).unwrap_or(4); + let (issuance_ip, candidate_count, user_agent, user_agent_id) = { + let entry = state + .bootstraps + .get_mut(&bootstrap_hash) + .ok_or(ManagerError::Authentication)?; + entry.used = true; + entry.body_digest = body_digest; + entry.session_token = Zeroizing::new(session_token.clone()); + entry.session = Some(Arc::clone(&session)); + entry.carrier_request = Some(carrier_request); + entry.carrier_candidates = candidates.into(); + entry.carrier_scores = scores; + entry.carrier_attempt = 1; + entry.carrier_phase = CarrierChainPhase::Provisional; + entry.carrier_started_at = carrier_request.is_automatic().then_some(now); + entry.carrier_deadline_at = carrier_deadline_at; + entry.carrier_failures = [None; 3]; + entry.carrier_learning_epoch = learning_epoch.unwrap_or(0); + 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; + ( + entry.issuance_ip, + u8::try_from(entry.carrier_candidates.len()).unwrap_or(4), + entry.user_agent.clone(), + entry.user_agent_id, + ) + }; decrement_map(&mut state.bootstraps_per_ip, &issuance_ip); self.sessions_created.fetch_add(1, Ordering::Relaxed); let identity = session.trace_identity(); @@ -345,6 +353,16 @@ impl WebProcessRuntime { .is_automatic() .then_some(CarrierChainPhase::Provisional.as_str()), }; + state.session_index.insert( + trace_session_id, + LiveSessionIndex { + session_hash, + bootstrap_hash, + attempt: 1, + user_agent, + user_agent_id, + }, + ); drop(state); self.trace.record_carrier_lifecycle( client_ip, diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs index b7483db..6f73020 100644 --- a/src/web/manager/session_creation/replacement.rs +++ b/src/web/manager/session_creation/replacement.rs @@ -21,6 +21,7 @@ impl WebProcessRuntime { let now = Instant::now(); let mut state = self.state.lock(); remove_expired_locked(&mut state, now); + state.apply_issuance_policy(generation.id, config.web.enabled); let valid = state.bootstraps.get(&bootstrap_hash).is_some_and(|entry| { entry.carrier_transitioning && entry.carrier_phase == CarrierChainPhase::Provisional @@ -37,7 +38,7 @@ impl WebProcessRuntime { .is_some_and(|session| Arc::ptr_eq(session, &replacement.old_session)); if !valid || state.closed - || !config.web.enabled + || !state.issuance_enabled || !generation .proxy_shared .is_user_enabled(&replacement.profile.user) @@ -121,6 +122,13 @@ impl WebProcessRuntime { deadline_secs: Some(entry.profile.carrier_negotiation_deadlines_secs[3]), carrier_state: Some(CarrierChainPhase::Provisional.as_str()), }; + if let Some(index) = state.session_index.get_mut(&replacement.trace_session_id) + && index.session_hash == old_hash + { + index.session_hash = session_hash; + index.bootstrap_hash = bootstrap_hash; + index.attempt = replacement.attempt; + } let identity = session.trace_identity(); let old_identity = replacement.old_session.trace_identity(); drop(state); diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs index 29fb268..f7ec376 100644 --- a/src/web/manager/state.rs +++ b/src/web/manager/state.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -8,7 +8,7 @@ use sha2::{Digest, Sha256}; use zeroize::Zeroizing; use super::{CarrierRequest, ProfileKey, TOKEN_BYTES, TokenHash}; -use crate::config::{WebCarrier, WebRuntimeConfig, WebRuntimeProfile}; +use crate::config::{WebCarrier, WebRuntimeConfig, WebRuntimeProfile, WebTimeoutsConfig}; use crate::maestro::generation::RuntimeGeneration; use crate::web::session::WebSession; @@ -41,8 +41,14 @@ pub(super) struct Bootstrap { pub(super) issuance_ip: IpAddr, /// Immutable profile selected during capability validation. pub(super) profile: Arc, + /// Request and session deadlines frozen with the generated bridge. + pub(super) timeouts: WebTimeoutsConfig, /// Process-unique non-secret identifier shared by bootstrap and session traces. pub(super) trace_session_id: u64, + /// Bounded display form of the issuing User-Agent. + pub(super) user_agent: Option>, + /// Opaque non-secret identifier used for exact User-Agent filtering. + pub(super) user_agent_id: Option<[u8; 16]>, /// Digest of the accepted HELLO body for idempotent retry matching. pub(super) body_digest: TokenHash, /// Zeroizing copy returned only for an exact session-creation retry. @@ -87,6 +93,28 @@ pub(super) struct ClosedToken { pub(super) host: String, } +/// Current logical-session owner stored without exposing bearer credentials. +pub(super) struct LiveSessionIndex { + /// Current bearer hash used only for internal pointer revalidation. + pub(super) session_hash: TokenHash, + /// Bootstrap chain that owns carrier replacement and close intent. + pub(super) bootstrap_hash: TokenHash, + /// Current carrier incarnation in the logical trace session. + pub(super) attempt: u8, + /// Bounded display form of the issuing User-Agent. + pub(super) user_agent: Option>, + /// Opaque non-secret identifier used for exact User-Agent filtering. + pub(super) user_agent_id: Option<[u8; 16]>, +} + +/// Bounded logical-session tombstone used for exact detail semantics. +pub(super) struct ClosedSession { + /// Tombstone expiry deadline. + pub(super) expires_at: Instant, + /// Last carrier incarnation closed for this logical session. + pub(super) attempt: u8, +} + /// Token-bucket state for one process-wide creation class. #[derive(Default)] pub(super) struct RateState { @@ -114,7 +142,6 @@ pub(super) struct StreamAdmissionState { } /// Process-wide WEB registries and quota accounting protected by one short lock. -#[derive(Default)] pub(super) struct ManagerState { /// Bootstrap credentials indexed by their SHA-256 token hash. pub(super) bootstraps: HashMap, @@ -122,6 +149,12 @@ pub(super) struct ManagerState { pub(super) bootstraps_per_ip: HashMap, /// Live sessions indexed by bearer-token hash. pub(super) sessions: HashMap>, + /// Stable ordered logical-session lookup independent from bearer hashes. + pub(super) session_index: BTreeMap, + /// Recently closed logical sessions retained for exact detail responses. + pub(super) closed_sessions: HashMap, + /// Insertion order for bounded logical-session tombstones. + pub(super) closed_session_order: VecDeque, /// Recently closed token hashes retained for idempotent DELETE semantics. pub(super) closed_tokens: HashMap, /// Live session counts by forwarded client address. @@ -132,10 +165,48 @@ pub(super) struct ManagerState { pub(super) bootstrap_rate: RateState, /// Session creation rate limiter. pub(super) session_rate: RateState, + /// Generation-fenced issuance gate mirrored from the effective WEB policy. + pub(super) issuance_enabled: bool, + /// Generation that last authored `issuance_enabled`. + pub(super) issuance_generation: u64, /// Process shutdown admission latch. pub(super) closed: bool, } +impl ManagerState { + pub(super) fn new(issuance_generation: u64, issuance_enabled: bool) -> Self { + Self { + bootstraps: HashMap::new(), + bootstraps_per_ip: HashMap::new(), + sessions: HashMap::new(), + session_index: BTreeMap::new(), + closed_sessions: HashMap::new(), + closed_session_order: VecDeque::new(), + closed_tokens: HashMap::new(), + sessions_per_ip: HashMap::new(), + sessions_per_profile: HashMap::new(), + bootstrap_rate: RateState::default(), + session_rate: RateState::default(), + issuance_enabled, + issuance_generation, + closed: false, + } + } + + pub(super) fn apply_issuance_policy(&mut self, generation: u64, enabled: bool) { + if generation >= self.issuance_generation { + self.issuance_generation = generation; + self.issuance_enabled = enabled; + } + } +} + +impl Default for ManagerState { + fn default() -> Self { + Self::new(0, false) + } +} + /// Generates one collision-checked credential and its stable hash key. pub(super) fn new_unique_token( generation: &RuntimeGeneration, @@ -241,6 +312,16 @@ pub(super) fn remove_expired_locked(state: &mut ManagerState, now: Instant) { state .closed_tokens .retain(|_, closed| now <= closed.expires_at); + while state + .closed_session_order + .front() + .and_then(|trace_session_id| state.closed_sessions.get(trace_session_id)) + .is_some_and(|closed| now > closed.expires_at) + { + if let Some(trace_session_id) = state.closed_session_order.pop_front() { + state.closed_sessions.remove(&trace_session_id); + } + } } /// Removes one bootstrap and releases its per-address issuance quota when unused. @@ -281,6 +362,35 @@ pub(super) fn remember_closed_token_locked( } } +/// Retains one bounded logical-session marker without storing its bearer identity. +pub(super) fn remember_closed_session_locked( + state: &mut ManagerState, + trace_session_id: u64, + attempt: u8, + lifetime: Duration, + capacity: usize, +) { + if state + .closed_sessions + .insert( + trace_session_id, + ClosedSession { + expires_at: Instant::now() + lifetime, + attempt, + }, + ) + .is_none() + { + state.closed_session_order.push_back(trace_session_id); + } + while state.closed_sessions.len() > capacity { + let Some(oldest) = state.closed_session_order.pop_front() else { + break; + }; + state.closed_sessions.remove(&oldest); + } +} + /// Decrements one counted owner and removes its map entry at zero. pub(super) fn decrement_map(values: &mut HashMap, key: &Q) where diff --git a/src/web/manager/status.rs b/src/web/manager/status.rs new file mode 100644 index 0000000..45991a7 --- /dev/null +++ b/src/web/manager/status.rs @@ -0,0 +1,560 @@ +use std::net::IpAddr; +use std::ops::Bound::{Excluded, Unbounded}; +use std::sync::Arc; +use std::sync::atomic::Ordering; +use std::time::Instant; + +use serde::Serialize; + +use super::WebProcessRuntime; +use crate::config::{ + WebCarrier, WebCarrierNegotiationAggressiveness, WebDebugConfig, WebLimitsConfig, +}; +use crate::web::session::WebSessionStatus; + +const SESSION_REF_VERSION: &str = "ws1"; +const MAX_SESSION_SCAN: usize = 1000; + +/// Current usage of one process-owned semaphore. +#[derive(Clone, Serialize)] +struct PermitStatus { + used: usize, + available: usize, + capacity: usize, + closed: bool, +} + +/// Short-lock manager registry counts. +#[derive(Clone, Serialize)] +struct ManagerStatus { + issuance_enabled: bool, + issuance_generation: u64, + shutdown: bool, + bootstraps: usize, + sessions: usize, + closed_tokens: usize, + closed_sessions: usize, + client_ips: usize, + profiles: usize, +} + +/// Logical-stream admission counters. +#[derive(Clone, Serialize)] +struct StreamStatus { + live: usize, + profiles: usize, + closed: bool, +} + +/// Shared queue and WebSocket byte-budget counters. +#[derive(Clone, Serialize)] +struct BudgetStatus { + queue_bytes: usize, + queue_items: usize, + control_bytes: usize, + control_items: usize, + websocket_bytes: usize, + high_water_bytes: usize, + owners: usize, + closed: bool, +} + +/// Process WebSocket registry counters. +#[derive(Clone, Serialize)] +struct WebSocketStatus { + entries: usize, + claims: usize, + evictions_in_flight: usize, + closed: bool, +} + +/// Process-local carrier-learning summary. +#[derive(Clone, Serialize)] +struct LearningStatus { + enabled: bool, + aggressiveness: WebCarrierNegotiationAggressiveness, + epoch: Option, + entries: usize, + capacity: usize, + lifetime_secs: u64, + age_ms: u64, +} + +/// Effective trace policy and bounded ring counters. +#[derive(Clone, Serialize)] +struct DebugStatus { + policy: WebDebugConfig, + policy_generation: u64, + epoch: u64, + records: usize, + records_capacity: usize, + used_bytes: usize, + bytes_capacity: usize, + contention_drops: u64, + evictions: u64, + byte_truncations: u64, + earliest_seq: Option, + latest_seq: Option, +} + +/// One non-blocking multi-plane WEB runtime snapshot. +#[derive(Clone, Serialize)] +pub(crate) struct WebRuntimeStatus { + runtime_instance: String, + generation_id: u64, + limits: WebLimitsConfig, + manager: Option, + streams: Option, + budget: Option, + websockets: Option, + learning: Option, + debug: Option, + permits: Vec<(&'static str, PermitStatus)>, + auxiliary_tasks: usize, + session_incarnations_created: u64, + session_incarnations_closed: u64, + streams_opened: u64, + streams_rejected: u64, + bytes_up: u64, + bytes_down: u64, + limit_hits: u64, + partial: Vec<&'static str>, +} + +/// Strict bounded filters for session enumeration and bulk close. +#[derive(Clone, Default)] +pub(crate) struct SessionFilter { + /// Exact process-local logical session identifier. + pub(crate) trace_session_id: Option, + /// Exact forwarded client address. + pub(crate) client_ip: Option, + /// Exact canonical virtual host. + pub(crate) host: Option, + /// Exact configured user label. + pub(crate) user: Option, + /// Exact non-secret User-Agent identifier. + pub(crate) user_agent_id: Option<[u8; 16]>, + /// Exact non-secret profile-key fingerprint. + pub(crate) key_id: Option, + /// Exact current carrier. + pub(crate) carrier: Option, + /// Exact point-in-time lifecycle token. + pub(crate) state: Option, +} + +impl SessionFilter { + /// Returns whether the selector would match every live session. + pub(crate) fn is_empty(&self) -> bool { + self.trace_session_id.is_none() + && self.client_ip.is_none() + && self.host.is_none() + && self.user.is_none() + && self.user_agent_id.is_none() + && self.key_id.is_none() + && self.carrier.is_none() + && self.state.is_none() + } +} + +/// Validated bounded list request. +pub(crate) struct SessionListRequest { + /// Maximum returned rows. + pub(crate) limit: usize, + /// Exclusive ordered logical-session cursor. + pub(crate) cursor: Option, + /// Exact bounded filters. + pub(crate) filter: SessionFilter, +} + +/// One API-safe session row with optional User-Agent metadata. +#[derive(Clone, Serialize)] +pub(crate) struct SessionRow { + /// Opaque process-fenced logical-session reference. + pub(crate) session_ref: String, + /// Bounded sanitized User-Agent display value. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) user_agent: Option, + #[serde(skip_serializing_if = "Option::is_none")] + user_agent_id: Option, + #[serde(flatten)] + status: WebSessionStatus, +} + +/// Bounded session page and continuation metadata. +#[derive(Clone, Serialize)] +pub(crate) struct SessionPage { + /// Ordered live-session rows captured without blocking. + pub(crate) sessions: Vec, + next_cursor: Option, + scanned: usize, + scan_truncated: bool, + partial_sessions: usize, + partial: Vec<&'static str>, +} + +/// Exact detail lookup outcome. +pub(crate) enum SessionDetail { + /// One exact live-session snapshot. + Active(Box), + /// One bounded retained closed-session tombstone. + Gone { attempt: u8 }, + /// A required short lock was contended. + Busy, + /// Neither a live session nor a retained tombstone exists. + NotFound, +} + +struct Candidate { + trace_session_id: u64, + session: Arc, + user_agent: Option>, + user_agent_id: Option<[u8; 16]>, +} + +impl WebProcessRuntime { + /// Captures every independent plane without blocking on a contended lock. + pub(crate) fn try_status(&self) -> WebRuntimeStatus { + let generation_id = self.active_generation().id; + let mut partial = Vec::new(); + let manager = self.state.try_lock().map(|state| ManagerStatus { + issuance_enabled: state.issuance_enabled, + issuance_generation: state.issuance_generation, + shutdown: state.closed, + bootstraps: state.bootstraps.len(), + sessions: state.sessions.len(), + closed_tokens: state.closed_tokens.len(), + closed_sessions: state.closed_sessions.len(), + client_ips: state.sessions_per_ip.len(), + profiles: state.sessions_per_profile.len(), + }); + if manager.is_none() { + partial.push("manager"); + } + let streams = self.stream_admission.try_lock().map(|state| StreamStatus { + live: state.streams_live, + profiles: state.streams_per_profile.len(), + closed: state.closed, + }); + if streams.is_none() { + partial.push("streams"); + } + let budget = self.data_budget.try_snapshot().map(|status| BudgetStatus { + queue_bytes: status.queue_bytes, + queue_items: status.queue_items, + control_bytes: status.queue_control_bytes, + control_items: status.queue_control_items, + websocket_bytes: status.websocket_bytes, + high_water_bytes: status.high_water_bytes, + owners: status.owners, + closed: status.closed, + }); + if budget.is_none() { + partial.push("budget"); + } + let websockets = self.websockets.try_lock().map(|registry| { + let status = registry.status(); + WebSocketStatus { + entries: status.entries, + claims: status.claims, + evictions_in_flight: status.evictions_in_flight, + closed: status.closed, + } + }); + if websockets.is_none() { + partial.push("websockets"); + } + let learning = self + .try_carrier_learning_status() + .map(|status| LearningStatus { + enabled: status.enabled, + aggressiveness: status.aggressiveness, + epoch: status.epoch, + entries: status.entries, + capacity: status.capacity, + lifetime_secs: status.lifetime_secs, + age_ms: status.age_ms, + }); + if learning.is_none() { + partial.push("learning"); + } + let debug = self.trace.try_status().map(|status| DebugStatus { + policy: status.policy.as_ref().clone(), + policy_generation: status.policy_generation, + epoch: status.epoch, + records: status.records, + records_capacity: status.records_capacity, + used_bytes: status.used_bytes, + bytes_capacity: status.bytes_capacity, + contention_drops: status.contention_drops, + evictions: status.evictions, + byte_truncations: status.byte_truncations, + earliest_seq: status.earliest_seq, + latest_seq: status.latest_seq, + }); + if debug.is_none() { + partial.push("debug"); + } + let websocket_capacity = self + .limits + .max_http_connections + .saturating_sub(self.limits.websocket_http_connection_reserve); + WebRuntimeStatus { + runtime_instance: self.runtime_instance().to_string(), + generation_id, + limits: self.limits.clone(), + manager, + streams, + budget, + websockets, + learning, + debug, + permits: vec![ + ( + "http_connections", + permits(&self.http_connections, self.limits.max_http_connections), + ), + ( + "http_handlers", + permits(&self.http_handlers, self.limits.max_http_handlers), + ), + ( + "lane_polls", + permits(&self.lane_polls, self.limits.max_http_handlers / 2), + ), + ( + "lane_aux_polls", + permits( + &self.lane_aux_polls, + (self.limits.max_http_handlers / 4).max(1), + ), + ), + ( + "body_readers", + permits(&self.body_readers, self.limits.max_body_readers), + ), + ( + "body_bytes", + permits(&self.body_bytes, self.limits.max_body_bytes_global), + ), + ( + "stream_handshakes", + permits(&self.stream_handshakes, self.limits.max_stream_handshakes), + ), + ( + "websocket_connections", + permits(&self.websocket_connections, websocket_capacity), + ), + ], + auxiliary_tasks: self.tasks.len(), + session_incarnations_created: self.sessions_created.load(Ordering::Relaxed), + session_incarnations_closed: self.sessions_closed.load(Ordering::Relaxed), + streams_opened: self.streams_opened.load(Ordering::Relaxed), + streams_rejected: self.streams_rejected.load(Ordering::Relaxed), + bytes_up: self.bytes_up.load(Ordering::Relaxed), + bytes_down: self.bytes_down.load(Ordering::Relaxed), + limit_hits: self.limit_hits.load(Ordering::Relaxed), + partial, + } + } + + /// Formats an opaque process-fenced session reference. + pub(crate) fn session_ref(&self, trace_session_id: u64) -> String { + format!( + "{SESSION_REF_VERSION}.{}.{trace_session_id:016x}", + self.runtime_instance() + ) + } + + /// Parses an exact reference and distinguishes stale process instances. + pub(crate) fn parse_session_ref(&self, value: &str) -> Result { + let mut parts = value.split('.'); + let version = parts.next(); + let instance = parts.next(); + let id = parts.next(); + if version != Some(SESSION_REF_VERSION) || parts.next().is_some() { + return Err(SessionRefError::Invalid); + } + if !instance.is_some_and(|instance| { + instance.len() == 32 + && instance + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + }) { + return Err(SessionRefError::Invalid); + } + let id = id + .filter(|id| { + id.len() == 16 + && id + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + }) + .and_then(|id| u64::from_str_radix(id, 16).ok()) + .filter(|id| *id != 0) + .ok_or(SessionRefError::Invalid)?; + if instance != Some(self.runtime_instance()) { + return Err(SessionRefError::StaleInstance); + } + Ok(id) + } + + /// Lists live sessions under an ordered bounded scan. + pub(crate) fn list_sessions(&self, request: SessionListRequest) -> SessionPage { + let Some(state) = self.state.try_lock() else { + return SessionPage { + sessions: Vec::new(), + next_cursor: request.cursor.map(|id| self.session_ref(id)), + scanned: 0, + scan_truncated: false, + partial_sessions: 0, + partial: vec!["manager"], + }; + }; + let mut candidates = Vec::with_capacity(request.limit); + let mut scanned = 0usize; + let mut last_scanned = request.cursor; + for (&trace_session_id, index) in state + .session_index + .range((request.cursor.map_or(Unbounded, Excluded), Unbounded)) + { + if scanned >= MAX_SESSION_SCAN || candidates.len() >= request.limit { + break; + } + scanned += 1; + last_scanned = Some(trace_session_id); + let Some(session) = state.sessions.get(&index.session_hash).cloned() else { + continue; + }; + if !immutable_matches(&session, index, &request.filter) { + continue; + } + candidates.push(Candidate { + trace_session_id, + session, + user_agent: index.user_agent.clone(), + user_agent_id: index.user_agent_id, + }); + } + drop(state); + let mut rows = Vec::with_capacity(candidates.len()); + let mut partial_sessions = 0usize; + let now = Instant::now(); + for candidate in candidates { + let Some(status) = candidate.session.try_status(now) else { + partial_sessions += 1; + continue; + }; + if request + .filter + .state + .as_deref() + .is_some_and(|expected| expected != status.state) + { + continue; + } + rows.push(self.row(candidate, status)); + } + let scan_truncated = scanned >= MAX_SESSION_SCAN; + SessionPage { + sessions: rows, + next_cursor: (scan_truncated || scanned >= request.limit) + .then(|| last_scanned.map(|id| self.session_ref(id))) + .flatten(), + scanned, + scan_truncated, + partial_sessions, + partial: Vec::new(), + } + } + + /// Resolves one active or recently closed logical session. + pub(crate) fn session_detail(&self, trace_session_id: u64) -> SessionDetail { + let Some(state) = self.state.try_lock() else { + return SessionDetail::Busy; + }; + if let Some(index) = state.session_index.get(&trace_session_id) { + let Some(session) = state.sessions.get(&index.session_hash).cloned() else { + return SessionDetail::Busy; + }; + let candidate = Candidate { + trace_session_id, + session, + user_agent: index.user_agent.clone(), + user_agent_id: index.user_agent_id, + }; + drop(state); + return candidate + .session + .try_status(Instant::now()) + .map(|status| SessionDetail::Active(Box::new(self.row(candidate, status)))) + .unwrap_or(SessionDetail::Busy); + } + let closed = state + .closed_sessions + .get(&trace_session_id) + .map(|closed| closed.attempt); + closed.map_or(SessionDetail::NotFound, |attempt| SessionDetail::Gone { + attempt, + }) + } + + fn row(&self, candidate: Candidate, status: WebSessionStatus) -> SessionRow { + SessionRow { + session_ref: self.session_ref(candidate.trace_session_id), + user_agent: candidate.user_agent.map(|value| value.to_string()), + user_agent_id: candidate.user_agent_id.map(hex::encode), + status, + } + } +} + +/// Opaque session-reference validation failure. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum SessionRefError { + /// The reference does not use the canonical versioned shape. + Invalid, + /// The reference belongs to another process runtime. + StaleInstance, +} + +/// Tests immutable candidate fields before any optional state-lock read. +pub(super) fn immutable_matches( + session: &crate::web::session::WebSession, + index: &super::state::LiveSessionIndex, + filter: &SessionFilter, +) -> bool { + filter + .trace_session_id + .is_none_or(|value| value == session.trace_session_id()) + && filter + .client_ip + .is_none_or(|value| value == session.client_ip()) + && filter + .host + .as_deref() + .is_none_or(|value| value == session.profile_host()) + && filter + .user + .as_deref() + .is_none_or(|value| value == session.profile_user()) + && filter + .key_id + .as_deref() + .is_none_or(|value| session.key_id() == value) + && filter + .carrier + .is_none_or(|value| value == session.carrier()) + && filter + .user_agent_id + .is_none_or(|value| index.user_agent_id == Some(value)) +} + +fn permits(semaphore: &Arc, capacity: usize) -> PermitStatus { + let available = semaphore.available_permits().min(capacity); + PermitStatus { + used: capacity.saturating_sub(available), + available, + capacity, + closed: semaphore.is_closed(), + } +} diff --git a/src/web/manager/websocket.rs b/src/web/manager/websocket.rs index ebec5d7..a8abed1 100644 --- a/src/web/manager/websocket.rs +++ b/src/web/manager/websocket.rs @@ -61,6 +61,26 @@ pub(super) struct WebSocketRegistry { closed: bool, } +/// Point-in-time WebSocket registry counters. +#[derive(Clone, Copy)] +pub(super) struct WebSocketRegistryStatus { + pub(super) entries: usize, + pub(super) claims: usize, + pub(super) evictions_in_flight: usize, + pub(super) closed: bool, +} + +impl WebSocketRegistry { + pub(super) fn status(&self) -> WebSocketRegistryStatus { + WebSocketRegistryStatus { + entries: self.entries.len(), + claims: self.claims.len(), + evictions_in_flight: self.evictions_in_flight, + closed: self.closed, + } + } +} + /// Exact process-owned admission retained through the upgraded socket lifetime. pub(crate) struct WebSocketConnection { runtime: std::sync::Weak, diff --git a/src/web/mod.rs b/src/web/mod.rs index 9e77363..e149e0b 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -2,6 +2,8 @@ /// Browser bridge generation for the serialized HTTPS carrier. pub(crate) mod bridge; +/// Process lifecycle publication shared with the control plane. +pub(crate) mod control; /// Shared binary frame codec and protocol constants. pub(crate) mod frame; /// Plain HTTP ingress and decoy routing behind external TLS termination. diff --git a/src/web/session.rs b/src/web/session.rs index 7790413..05486ac 100644 --- a/src/web/session.rs +++ b/src/web/session.rs @@ -24,6 +24,9 @@ mod backend; mod downlink; // Response ownership keeps detached batches charged until the last body clone drops. mod resident; +// Read-only control-plane snapshots stay isolated from carrier operations. +mod status; +pub(crate) use status::WebSessionStatus; // Lane carrier state isolates request sequencing and downlink replay per logical stream. mod lanes; // Lane batch staging transfers queue ownership without escaping process budgets. @@ -205,6 +208,7 @@ pub(crate) struct WebSession { carrier_class: CarrierClientClass, learning_context: Option, automatic_carrier: bool, + created_at: Instant, limits: WebLimitsConfig, timeouts: WebTimeoutsConfig, state: Mutex, @@ -268,6 +272,7 @@ impl WebSession { carrier_class, learning_context, automatic_carrier, + created_at: Instant::now(), limits, timeouts, state: Mutex::new(SessionState { @@ -344,6 +349,11 @@ impl WebSession { self.trace_session_id } + /// Returns the immutable carrier-attempt incarnation number. + pub(crate) fn carrier_attempt(&self) -> u8 { + self.carrier_attempt + } + /// Creates a child cancellation boundary for one owned carrier task. pub(crate) fn carrier_cancellation(&self) -> CancellationToken { self.cancel.child_token() diff --git a/src/web/session/status.rs b/src/web/session/status.rs new file mode 100644 index 0000000..8310085 --- /dev/null +++ b/src/web/session/status.rs @@ -0,0 +1,135 @@ +use std::time::Instant; + +use serde::Serialize; + +use super::{SessionNegotiationPhase, WebSession}; +use crate::config::WebCarrier; + +/// One bounded point-in-time session snapshot without bearer identity. +#[derive(Clone, Serialize)] +pub(crate) struct WebSessionStatus { + /// Stable trace identifier within this process. + pub(crate) trace_session_id: u64, + /// Forwarded client address frozen at session creation. + pub(crate) client_ip: std::net::IpAddr, + /// Canonical WEB virtual host. + pub(crate) host: String, + /// Configured non-secret user label. + pub(crate) user: String, + /// Non-secret configured key fingerprint. + pub(crate) key_id: String, + /// Current carrier incarnation. + pub(crate) carrier: WebCarrier, + /// One-based carrier attempt. + pub(crate) attempt: u8, + /// Stable client classification token. + pub(crate) client_class: &'static str, + /// Whether server-side carrier negotiation owns this chain. + pub(crate) automatic: bool, + /// Current session lifecycle token. + pub(crate) state: &'static str, + /// Live logical streams. + pub(crate) streams: usize, + /// Stream relay tasks that have not exited. + pub(crate) tasks: usize, + /// Carrier lanes currently retained. + pub(crate) lanes: usize, + /// Lane OPEN polls currently waiting. + pub(crate) lane_open_waits: usize, + /// WebSocket lane slots reserved before ownership transfer. + pub(crate) websocket_lane_reservations: usize, + /// Whether the multiplexed WebSocket carrier is active. + pub(crate) websocket_active: bool, + /// Queued and response-resident bytes charged to this session. + pub(crate) pending_bytes: usize, + /// Queued and response-resident items charged to this session. + pub(crate) pending_items: usize, + /// Control bytes included in the pending total. + pub(crate) control_bytes: usize, + /// Control items included in the pending total. + pub(crate) control_items: usize, + /// Monotonic age since session creation. + pub(crate) age_ms: u64, + /// Monotonic age since the latest carrier activity. + pub(crate) idle_ms: u64, + /// Remaining automatic negotiation deadline. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) negotiation_remaining_ms: Option, +} + +impl WebSession { + /// Captures one short-lock session snapshot or reports lock contention. + pub(crate) fn try_status(&self, now: Instant) -> Option { + let state = self.state.try_lock()?; + let resident = self.resident.snapshot(); + let state_name = if state.closed { + "closed" + } else if state.close_requested { + "closing" + } else if state.carrier_health_reported { + "healthy" + } else { + match state.negotiation_phase { + SessionNegotiationPhase::Uncommitted => "provisional", + SessionNegotiationPhase::Replacing => "replacing", + SessionNegotiationPhase::Committed => "committed", + SessionNegotiationPhase::Superseded => "superseded", + } + }; + Some(WebSessionStatus { + trace_session_id: self.trace_session_id, + client_ip: self.client_ip, + host: self.profile.host.clone(), + user: self.profile.user.clone(), + key_id: self.profile.key_fingerprint.clone(), + carrier: self.selected_carrier, + attempt: self.carrier_attempt, + client_class: self.carrier_class.as_str(), + automatic: self.automatic_carrier, + state: state_name, + streams: state.streams.len(), + tasks: self.tasks_live(), + lanes: state.carrier_lanes.len(), + lane_open_waits: state.lane_open_waits, + websocket_lane_reservations: state.websocket_lane_reservations.len(), + websocket_active: state.websocket_carrier_active, + pending_bytes: state.pending_bytes.saturating_add(resident.bytes()), + pending_items: state.pending_items.saturating_add(resident.items()), + control_bytes: state + .pending_control_bytes + .saturating_add(resident.control_bytes), + control_items: state + .pending_control_items + .saturating_add(resident.control_items), + age_ms: millis(now.saturating_duration_since(self.created_at)), + idle_ms: millis(now.saturating_duration_since(state.last_activity)), + negotiation_remaining_ms: self + .carrier_deadline_at + .map(|deadline| millis(deadline.saturating_duration_since(now))), + }) + } + + /// Returns the forwarded client address frozen at session creation. + pub(crate) fn client_ip(&self) -> std::net::IpAddr { + self.client_ip + } + + /// Returns the canonical profile host frozen into this session. + pub(crate) fn profile_host(&self) -> &str { + &self.profile.host + } + + /// Returns the configured non-secret user label. + pub(crate) fn profile_user(&self) -> &str { + &self.profile.user + } + + /// Returns the configured non-secret key fingerprint. + pub(crate) fn key_id(&self) -> &str { + &self.profile.key_fingerprint + } +} + +fn millis(duration: std::time::Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 +} diff --git a/src/web/trace/mod.rs b/src/web/trace/mod.rs index b3c77c3..205a2f5 100644 --- a/src/web/trace/mod.rs +++ b/src/web/trace/mod.rs @@ -10,7 +10,9 @@ mod store; mod types; pub(crate) use exchange::HttpTraceExchange; -pub(crate) use store::{StoredTraceRecord, WebTraceStore, epoch_millis as store_epoch_millis}; +pub(crate) use store::{ + StoredTraceRecord, TraceClearOutcome, WebTraceStore, epoch_millis as store_epoch_millis, +}; pub(crate) use types::{ TraceBodySnapshot, TraceBodyState, TraceDirection, TraceFrame, TraceHeader, TraceIdentity, TraceLifecycleEvent, TraceLifecycleRecord, TraceRecord, TraceRecordKind, TraceRoute, diff --git a/src/web/trace/store.rs b/src/web/trace/store.rs index 2ce56fe..e50044f 100644 --- a/src/web/trace/store.rs +++ b/src/web/trace/store.rs @@ -41,6 +41,10 @@ impl Drop for StoredTraceRecord { pub(crate) struct TraceStoreStatus { /// Current debug policy. pub(crate) policy: Arc, + /// Runtime generation that last authored the effective policy. + pub(crate) policy_generation: u64, + /// Epoch fencing in-flight records across policy changes and clears. + pub(crate) epoch: u64, /// Retained record count. pub(crate) records: usize, /// Configured record capacity. @@ -61,11 +65,22 @@ pub(crate) struct TraceStoreStatus { pub(crate) latest_seq: Option, } +/// Result of one constant-time logical trace clear. +pub(crate) struct TraceClearOutcome { + /// Records detached from the ring. + pub(crate) records_cleared: usize, + /// Bytes still retained by in-flight snapshots after detached records drop. + pub(crate) leased_bytes: usize, + /// New epoch rejecting commits started before the clear. + pub(crate) epoch: u64, +} + /// Process-owned bounded WEB debug trace store. pub(crate) struct WebTraceStore { policy: ArcSwap, policy_update: Mutex<()>, enabled: AtomicBool, + policy_generation: AtomicU64, epoch: AtomicU64, records_capacity: usize, bytes_capacity: usize, @@ -88,6 +103,7 @@ impl WebTraceStore { enabled: AtomicBool::new(policy.enabled), policy: ArcSwap::from_pointee(policy), policy_update: Mutex::new(()), + policy_generation: AtomicU64::new(0), epoch: AtomicU64::new(1), records_capacity: limits.debug_records_capacity, bytes_capacity: limits.debug_bytes_global, @@ -106,11 +122,16 @@ impl WebTraceStore { }) } - /// Applies one hot policy and clears incompatible retained records. - pub(crate) fn apply_policy(&self, policy: &WebDebugConfig) { + /// Applies one generation-authored policy and rejects stale generation writers. + pub(crate) fn apply_policy(&self, generation: u64, policy: &WebDebugConfig) { let _policy_update = self.policy_update.lock(); + let current_generation = self.policy_generation.load(Ordering::Acquire); + if generation < current_generation { + return; + } let current = self.policy.load_full(); if current.as_ref() == policy { + self.policy_generation.store(generation, Ordering::Release); return; } let capture_changed = current.enabled != policy.enabled @@ -122,10 +143,40 @@ impl WebTraceStore { || current.body_prefix_bytes != policy.body_prefix_bytes || current.decoy_body_prefix_bytes != policy.decoy_body_prefix_bytes; self.policy.store(Arc::new(policy.clone())); + self.policy_generation.store(generation, Ordering::Release); self.enabled.store(policy.enabled, Ordering::Release); if capture_changed { self.epoch.fetch_add(1, Ordering::AcqRel); - self.ring.lock().records.clear(); + let detached = { + let mut ring = self.ring.lock(); + std::mem::replace( + &mut ring.records, + VecDeque::with_capacity(self.records_capacity), + ) + }; + drop(_policy_update); + drop(detached); + } + } + + /// Clears retained records while fencing all in-flight pre-clear commits. + pub(crate) fn clear(&self) -> TraceClearOutcome { + let _policy_update = self.policy_update.lock(); + let epoch = self.epoch.fetch_add(1, Ordering::AcqRel).saturating_add(1); + let detached = { + let mut ring = self.ring.lock(); + std::mem::replace( + &mut ring.records, + VecDeque::with_capacity(self.records_capacity), + ) + }; + let records_cleared = detached.len(); + drop(_policy_update); + drop(detached); + TraceClearOutcome { + records_cleared, + leased_bytes: self.used_bytes.load(Ordering::Acquire), + epoch, } } @@ -297,9 +348,12 @@ impl WebTraceStore { /// Returns current bounds, counters, and retained sequence range. pub(crate) fn status(&self) -> TraceStoreStatus { + let _policy_update = self.policy_update.lock(); let ring = self.ring.lock(); TraceStoreStatus { policy: self.policy.load_full(), + policy_generation: self.policy_generation.load(Ordering::Acquire), + epoch: self.epoch.load(Ordering::Acquire), records: ring.records.len(), records_capacity: self.records_capacity, used_bytes: self.used_bytes.load(Ordering::Acquire), @@ -312,6 +366,26 @@ impl WebTraceStore { } } + /// Returns a non-blocking status snapshot or `None` on trace-store contention. + pub(crate) fn try_status(&self) -> Option { + let _policy_update = self.policy_update.try_lock()?; + let ring = self.ring.try_lock()?; + Some(TraceStoreStatus { + policy: self.policy.load_full(), + policy_generation: self.policy_generation.load(Ordering::Acquire), + epoch: self.epoch.load(Ordering::Acquire), + records: ring.records.len(), + records_capacity: self.records_capacity, + used_bytes: self.used_bytes.load(Ordering::Acquire), + bytes_capacity: self.bytes_capacity, + contention_drops: self.contention_drops.load(Ordering::Relaxed), + evictions: self.evictions.load(Ordering::Relaxed), + byte_truncations: self.byte_truncations.load(Ordering::Relaxed), + earliest_seq: ring.records.front().map(|record| record.record.seq), + latest_seq: ring.records.back().map(|record| record.record.seq), + }) + } + /// Reserves one of two bounded concurrent status-page render slots. pub(crate) fn try_render_permit(&self) -> Option { Arc::clone(&self.renders).try_acquire_owned().ok() @@ -427,72 +501,4 @@ pub(crate) fn epoch_millis() -> u64 { } #[cfg(test)] -mod tests { - use super::*; - - fn store(records_capacity: usize, bytes_capacity: usize) -> Arc { - let policy = WebDebugConfig { - enabled: true, - ..Default::default() - }; - let limits = WebLimitsConfig { - debug_records_capacity: records_capacity, - debug_bytes_global: bytes_capacity, - ..Default::default() - }; - WebTraceStore::new(policy, &limits) - } - - #[test] - fn ring_evicts_oldest_records_and_snapshot_leases_survive_clear() { - let store = store(2, 4 * BASE_RECORD_RESERVATION); - for _ in 0..3 { - store.record_lifecycle( - None, - Some("192.0.2.10".parse().unwrap()), - TraceIdentity::default(), - TraceLifecycleEvent::BridgeIssued, - None, - None, - ); - } - - let snapshot = store.snapshot_matching(|_| true); - assert_eq!( - snapshot - .iter() - .map(|record| record.record.seq) - .collect::>(), - vec![3, 2] - ); - assert_eq!(store.status().evictions, 1); - assert_eq!(store.status().used_bytes, 2 * BASE_RECORD_RESERVATION); - - let policy = WebDebugConfig::default(); - store.apply_policy(&policy); - assert_eq!(store.status().records, 0); - assert_eq!(store.status().used_bytes, 2 * BASE_RECORD_RESERVATION); - drop(snapshot); - assert_eq!(store.status().used_bytes, 0); - } - - #[test] - fn capture_policy_epoch_rejects_an_inflight_old_policy_record() { - let store = store(4, 8 * BASE_RECORD_RESERVATION); - let request = hyper::Request::builder().uri("/").body(()).unwrap(); - let exchange = store - .begin_http(&request, "192.0.2.20".parse().unwrap()) - .unwrap(); - - let changed = WebDebugConfig { - enabled: true, - capture_headers: false, - ..Default::default() - }; - store.apply_policy(&changed); - exchange.commit(); - - assert_eq!(store.status().records, 0); - assert_eq!(store.status().used_bytes, 0); - } -} +mod tests; diff --git a/src/web/trace/store/tests.rs b/src/web/trace/store/tests.rs new file mode 100644 index 0000000..79f0434 --- /dev/null +++ b/src/web/trace/store/tests.rs @@ -0,0 +1,111 @@ +use super::*; + +fn store(records_capacity: usize, bytes_capacity: usize) -> Arc { + let policy = WebDebugConfig { + enabled: true, + ..Default::default() + }; + let limits = WebLimitsConfig { + debug_records_capacity: records_capacity, + debug_bytes_global: bytes_capacity, + ..Default::default() + }; + WebTraceStore::new(policy, &limits) +} + +#[test] +fn ring_evicts_oldest_records_and_snapshot_leases_survive_clear() { + let store = store(2, 4 * BASE_RECORD_RESERVATION); + for _ in 0..3 { + store.record_lifecycle( + None, + Some("192.0.2.10".parse().unwrap()), + TraceIdentity::default(), + TraceLifecycleEvent::BridgeIssued, + None, + None, + ); + } + + let snapshot = store.snapshot_matching(|_| true); + assert_eq!( + snapshot + .iter() + .map(|record| record.record.seq) + .collect::>(), + vec![3, 2] + ); + assert_eq!(store.status().evictions, 1); + assert_eq!(store.status().used_bytes, 2 * BASE_RECORD_RESERVATION); + + let policy = WebDebugConfig::default(); + store.apply_policy(2, &policy); + assert_eq!(store.status().records, 0); + assert_eq!(store.status().used_bytes, 2 * BASE_RECORD_RESERVATION); + drop(snapshot); + assert_eq!(store.status().used_bytes, 0); +} + +#[test] +fn capture_policy_epoch_rejects_an_inflight_old_policy_record() { + let store = store(4, 8 * BASE_RECORD_RESERVATION); + let request = hyper::Request::builder().uri("/").body(()).unwrap(); + let exchange = store + .begin_http(&request, "192.0.2.20".parse().unwrap()) + .unwrap(); + + let changed = WebDebugConfig { + enabled: true, + capture_headers: false, + ..Default::default() + }; + store.apply_policy(2, &changed); + exchange.commit(); + + assert_eq!(store.status().records, 0); + assert_eq!(store.status().used_bytes, 0); +} + +#[test] +fn stale_generation_cannot_restore_an_old_policy() { + let store = store(4, 8 * BASE_RECORD_RESERVATION); + let current = WebDebugConfig { + enabled: true, + capture_headers: false, + ..Default::default() + }; + store.apply_policy(3, ¤t); + + store.apply_policy(2, &WebDebugConfig::default()); + + let status = store.status(); + assert_eq!(status.policy_generation, 3); + assert_eq!(status.policy.as_ref(), ¤t); +} + +#[test] +fn explicit_clear_fences_inflight_commits_and_preserves_snapshot_leases() { + let store = store(4, 8 * BASE_RECORD_RESERVATION); + store.record_lifecycle( + None, + Some("192.0.2.30".parse().unwrap()), + TraceIdentity::default(), + TraceLifecycleEvent::BridgeIssued, + None, + None, + ); + let snapshot = store.snapshot_matching(|_| true); + let request = hyper::Request::builder().uri("/").body(()).unwrap(); + let exchange = store + .begin_http(&request, "192.0.2.30".parse().unwrap()) + .unwrap(); + + let cleared = store.clear(); + exchange.commit(); + + assert_eq!(cleared.records_cleared, 1); + assert_eq!(store.status().records, 0); + assert_eq!(store.status().used_bytes, BASE_RECORD_RESERVATION); + drop(snapshot); + assert_eq!(store.status().used_bytes, 0); +}