WEB Debug + Trace

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>

src/web/http.rs

@@ -6,8 +6,7 @@ use std::time::{Duration, Instant};
 
 use bytes::Bytes;
 use http_body_util::combinators::UnsyncBoxBody;
-use http_body_util::{BodyExt, Full};
-use hyper::body::Incoming;
+use http_body_util::BodyExt;
 use hyper::header::{self, HeaderName, HeaderValue};
 use hyper::server::conn::http1;
 use hyper::service::service_fn;
@@ -18,7 +17,7 @@ use parking_lot::Mutex;
 use tokio::net::TcpStream;
 use tokio_util::sync::CancellationToken;
 
-use crate::config::{WebCarrier, WebClientIpSource, WebRuntimeVhost};
+use crate::config::{WebClientIpSource, WebRuntimeVhost};
 use crate::web::bridge;
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::{ManagerError, WebProcessRuntime};
@@ -29,18 +28,34 @@ mod activity;
 mod body;
 // Decoy routing and upstream proxying are isolated from carrier authentication.
 mod decoy;
+// Downlink long-poll handling remains isolated from request routing.
+mod down;
 // Canonical request parsing rejects ambiguous credentials before routing.
 mod request;
+// Carrier response construction and lane-header helpers are shared by handlers.
+mod response;
 #[cfg(test)]
 mod tests;
+// Enabled-debug integration coverage remains separate from carrier behavior tests.
+#[cfg(test)]
+#[path = "http/trace_tests.rs"]
+mod trace_tests;
 
 use activity::{ActivityBody, RequestActivity};
-use body::{CollectBodyError, CollectedBody, collect_body};
+use body::{CollectBodyError, CollectedBody, RequestBody, collect_body};
 use decoy::serve_decoy;
+use down::handle_down;
 use request::{
     bearer_token_hash, binary_content_type, bridge_candidate, canonical_request_host,
     canonical_u64_header, client_ip, compatible_cookie_header, match_profile,
 };
+use response::{
+    bad_gateway, carrier_empty, carrier_headers, carrier_lane, full_response, generic_not_found,
+    insert_header, service_unavailable,
+};
+use crate::web::trace::{
+    HttpTraceExchange, TraceDirection, TraceLifecycleEvent, TraceRoute,
+};
 
 type BoxError = Box<dyn Error + Send + Sync>;
 type HttpBody = UnsyncBoxBody<Bytes, BoxError>;
@@ -65,13 +80,18 @@ pub(crate) async fn serve_connection(
     let idle_timeout = Duration::from_secs(config.web.timeouts.http_idle_secs);
     let last_activity = Arc::new(Mutex::new(Instant::now()));
     let service_last_activity = Arc::clone(&last_activity);
-    let service = service_fn(move |request| {
+    let service = service_fn(move |mut request| {
         let runtime = Arc::clone(&runtime);
         let trusted_proxy_cidrs = Arc::clone(&trusted_proxy_cidrs);
         let last_activity = Arc::clone(&service_last_activity);
         let client_ip_source = client_ip_source;
         async move {
             let activity = RequestActivity::begin(last_activity);
+            let trace = runtime.trace().begin_http(&request, peer.ip());
+            if let Some(trace) = &trace {
+                request.extensions_mut().insert(Arc::clone(trace));
+            }
+            let request = request.map(|body| RequestBody::new(body, trace.clone()));
             let response = if let Some(_handler_permit) = runtime.try_http_handler() {
                 handle_request(
                     request,
@@ -84,7 +104,11 @@ pub(crate) async fn serve_connection(
             } else {
                 service_unavailable()
             };
-            let response = response.map(|body| ActivityBody::new(body, activity).boxed_unsync());
+            if let Some(trace) = &trace {
+                trace.response_ready(&response);
+            }
+            let response = response
+                .map(|body| ActivityBody::new(body, activity, trace).boxed_unsync());
             Ok::<_, Infallible>(response)
         }
     });
@@ -115,12 +139,18 @@ pub(crate) async fn serve_connection(
 }
 
 async fn handle_request(
-    request: Request<Incoming>,
+    request: Request<RequestBody>,
     peer: SocketAddr,
     client_ip_source: WebClientIpSource,
     trusted_proxy_cidrs: &[IpNetwork],
     runtime: Arc<WebProcessRuntime>,
 ) -> HttpResponse {
+    set_trace_route(&request, TraceRoute::Decoy);
+    if let Some(trace) = request_trace(&request)
+        && let Some(client_ip) = client_ip(&request, peer, client_ip_source, trusted_proxy_cidrs)
+    {
+        trace.set_effective_ip(client_ip);
+    }
     let generation = runtime.active_generation();
     let config = generation.config();
     let Some(web_runtime) = config.web.runtime.as_ref() else {
@@ -159,7 +189,7 @@ async fn handle_request(
 }
 
 async fn handle_root(
-    mut request: Request<Incoming>,
+    mut request: Request<RequestBody>,
     peer: SocketAddr,
     client_ip_source: WebClientIpSource,
     trusted_proxy_cidrs: &[IpNetwork],
@@ -175,15 +205,34 @@ async fn handle_root(
         strip_query(&mut request);
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.set_route(TraceRoute::Bridge);
+        trace.set_effective_ip(client_ip);
+    }
     let carrier = profile.carrier;
-    let Ok(bootstrap) = runtime.issue_bootstrap(profile, client_ip) else {
-        strip_query(&mut request);
-        return serve_decoy(request, vhost, true, &runtime).await;
+    let bootstrap = match runtime.issue_bootstrap(Arc::clone(&profile), client_ip) {
+        Ok(bootstrap) => bootstrap,
+        Err(error) => {
+            runtime.trace().record_profile_lifecycle(
+                client_ip,
+                None,
+                &profile,
+                TraceLifecycleEvent::BootstrapRejected,
+                None,
+                Some(manager_error_reason(error)),
+            );
+            strip_query(&mut request);
+            return serve_decoy(request, vhost, true, &runtime).await;
+        }
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.bind_profile(&profile, bootstrap.trace_session_id);
+        trace.register_redaction(bootstrap.token.as_bytes());
+    }
     let generation = runtime.active_generation();
     let page = bridge::render(
         &vhost.host,
-        &bootstrap,
+        &bootstrap.token,
         generation.config().web.limits.carrier_batch_bytes,
         generation.config().web.limits.pending_bytes_per_session,
         generation.config().web.limits.pending_items_per_session,
@@ -224,7 +273,7 @@ async fn handle_root(
 }
 
 async fn handle_api(
-    request: Request<Incoming>,
+    request: Request<RequestBody>,
     peer: SocketAddr,
     client_ip_source: WebClientIpSource,
     trusted_proxy_cidrs: &[IpNetwork],
@@ -237,6 +286,9 @@ async fn handle_api(
     let Some(client_ip) = client_ip(&request, peer, client_ip_source, trusted_proxy_cidrs) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.set_effective_ip(client_ip);
+    }
     let Some(token_hash) = bearer_token_hash(&request) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
@@ -249,7 +301,7 @@ async fn handle_api(
 }
 
 async fn handle_session(
-    request: Request<Incoming>,
+    request: Request<RequestBody>,
     runtime: Arc<WebProcessRuntime>,
     vhost: Arc<WebRuntimeVhost>,
     token_hash: crate::web::manager::TokenHash,
@@ -262,6 +314,12 @@ async fn handle_session(
         if request.headers().contains_key(header::CONTENT_TYPE) {
             return serve_decoy(request, vhost, true, &runtime).await;
         }
+        if let Some(trace) = request_trace(&request)
+            && let Ok(session) = runtime.get_session(token_hash, &vhost.host)
+        {
+            trace.set_route(TraceRoute::Session);
+            trace.bind_identity(session.trace_identity());
+        }
         let CollectedBody {
             request,
             body,
@@ -281,8 +339,14 @@ async fn handle_session(
     if request.method() != Method::POST || !binary_content_type(&request) {
         return serve_decoy(request, vhost, true, &runtime).await;
     }
-    if !runtime.has_bootstrap(token_hash, &vhost.host) {
+    let Some((trace_session_id, profile)) =
+        runtime.bootstrap_trace_identity(token_hash, &vhost.host)
+    else {
         return serve_decoy(request, vhost, true, &runtime).await;
+    };
+    if let Some(trace) = request_trace(&request) {
+        trace.set_route(TraceRoute::Session);
+        trace.bind_profile(&profile, trace_session_id);
     }
     let CollectedBody {
         request,
@@ -295,9 +359,24 @@ async fn handle_session(
             return serve_decoy(request, vhost, true, &runtime).await;
         }
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.record_frames(
+            TraceDirection::Request,
+            &body,
+            &runtime.active_generation().config().web.limits,
+        );
+    }
     match runtime.create_session(token_hash, &vhost.host, client_ip, &body) {
         Ok(result) => {
             let welcome = frame::encode(FrameType::Welcome, 0, &[]);
+            if let Some(trace) = request_trace(&request) {
+                trace.register_redaction(result.token.as_bytes());
+                trace.record_frames(
+                    TraceDirection::Response,
+                    &welcome,
+                    &runtime.active_generation().config().web.limits,
+                );
+            }
             let mut response = full_response(StatusCode::OK, welcome);
             carrier_headers(&mut response);
             insert_header(
@@ -315,15 +394,33 @@ async fn handle_session(
             );
             response
         }
-        Err(ManagerError::Limit | ManagerError::Backpressure | ManagerError::Concurrent) => {
+        Err(error @ (ManagerError::Limit | ManagerError::Backpressure | ManagerError::Concurrent)) => {
+            runtime.trace().record_profile_lifecycle(
+                client_ip,
+                Some(trace_session_id),
+                &profile,
+                TraceLifecycleEvent::SessionRejected,
+                None,
+                Some(manager_error_reason(error)),
+            );
             service_unavailable()
         }
-        Err(_) => serve_decoy(request, vhost, true, &runtime).await,
+        Err(error) => {
+            runtime.trace().record_profile_lifecycle(
+                client_ip,
+                Some(trace_session_id),
+                &profile,
+                TraceLifecycleEvent::SessionRejected,
+                None,
+                Some(manager_error_reason(error)),
+            );
+            serve_decoy(request, vhost, true, &runtime).await
+        }
     }
 }
 
 async fn handle_up(
-    request: Request<Incoming>,
+    request: Request<RequestBody>,
     runtime: Arc<WebProcessRuntime>,
     vhost: Arc<WebRuntimeVhost>,
     token_hash: crate::web::manager::TokenHash,
@@ -338,6 +435,10 @@ async fn handle_up(
     let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.set_route(TraceRoute::Uplink);
+        trace.bind_identity(session.trace_identity());
+    }
     let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
@@ -358,6 +459,13 @@ async fn handle_up(
             return serve_decoy(request, vhost, true, &runtime).await;
         }
     };
+    if let Some(trace) = request_trace(&request) {
+        trace.record_frames(
+            TraceDirection::Request,
+            &body,
+            &runtime.active_generation().config().web.limits,
+        );
+    }
     let result = match lane_id {
         Some(lane_id) => session.process_up_lane(lane_id, sequence, &body),
         None => session.process_up(sequence, &body),
@@ -379,151 +487,6 @@ async fn handle_up(
     }
 }
 
-async fn handle_down(
-    request: Request<Incoming>,
-    runtime: Arc<WebProcessRuntime>,
-    vhost: Arc<WebRuntimeVhost>,
-    token_hash: crate::web::manager::TokenHash,
-) -> HttpResponse {
-    if request.method() != Method::POST || request.headers().contains_key(header::CONTENT_TYPE) {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    }
-    let Some(cursor) = canonical_u64_header(&request, "x-down-cursor") else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    };
-    let CollectedBody {
-        request,
-        body,
-        _body_budget,
-    } = match collect_body(request, &runtime, 1, true).await {
-        Ok(result) => result,
-        Err(CollectBodyError::Limit) => return service_unavailable(),
-        Err(CollectBodyError::Invalid(request)) => {
-            return serve_decoy(request, vhost, true, &runtime).await;
-        }
-    };
-    if !body.is_empty() {
-        return serve_decoy(request, vhost, true, &runtime).await;
-    }
-    let _lane_poll = if lane_id.is_some() {
-        let Some(permit) = runtime.try_lane_poll() else {
-            return service_unavailable();
-        };
-        Some(permit)
-    } else {
-        None
-    };
-    let result = match lane_id {
-        Some(lane_id) => session.poll_down_lane(lane_id, cursor).await,
-        None => session.poll_down(cursor).await,
-    };
-    match result {
-        Ok(result) if result.body.is_empty() => {
-            let mut response = carrier_empty(StatusCode::NO_CONTENT);
-            insert_header(
-                &mut response,
-                HeaderName::from_static("x-down-cursor"),
-                &result.next_cursor.to_string(),
-            );
-            if result.lane_closed {
-                response.headers_mut().insert(
-                    HeaderName::from_static("x-lane-closed"),
-                    HeaderValue::from_static("1"),
-                );
-            }
-            response
-        }
-        Ok(result) => {
-            let mut response = full_response(StatusCode::OK, result.body);
-            carrier_headers(&mut response);
-            insert_header(
-                &mut response,
-                HeaderName::from_static("x-down-cursor"),
-                &result.next_cursor.to_string(),
-            );
-            response
-        }
-        Err(ManagerError::Concurrent | ManagerError::Backpressure | ManagerError::Limit) => {
-            service_unavailable()
-        }
-        Err(_) => serve_decoy(request, vhost, true, &runtime).await,
-    }
-}
-
-fn carrier_lane<B>(request: &Request<B>, carrier: WebCarrier) -> Option<Option<u32>> {
-    match carrier {
-        WebCarrier::Https => (!request.headers().contains_key("x-lane-id")).then_some(None),
-        WebCarrier::HttpsLanes => canonical_u64_header(request, "x-lane-id")
-            .and_then(|value| u32::try_from(value).ok())
-            .filter(|value| *value <= frame::MAX_STREAM_ID)
-            .map(Some),
-    }
-}
-
-fn carrier_headers(response: &mut HttpResponse) {
-    response.headers_mut().insert(
-        header::CONTENT_TYPE,
-        HeaderValue::from_static("application/octet-stream"),
-    );
-    response
-        .headers_mut()
-        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
-}
-
-fn carrier_empty(status: StatusCode) -> HttpResponse {
-    let mut response = empty_response(status);
-    response
-        .headers_mut()
-        .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
-    response
-}
-
-fn service_unavailable() -> HttpResponse {
-    let mut response = carrier_empty(StatusCode::SERVICE_UNAVAILABLE);
-    response
-        .headers_mut()
-        .insert(header::RETRY_AFTER, HeaderValue::from_static("1"));
-    response
-}
-
-fn bad_gateway() -> HttpResponse {
-    full_response(
-        StatusCode::BAD_GATEWAY,
-        Bytes::from_static(b"site unavailable\n"),
-    )
-}
-
-fn generic_not_found() -> HttpResponse {
-    full_response(StatusCode::NOT_FOUND, Bytes::from_static(b"not found\n"))
-}
-
-fn full_response(status: StatusCode, body: Bytes) -> HttpResponse {
-    let length = body.len();
-    let body = Full::new(body)
-        .map_err(|never| -> BoxError { match never {} })
-        .boxed_unsync();
-    let mut response = Response::new(body);
-    *response.status_mut() = status;
-    insert_header(&mut response, header::CONTENT_LENGTH, &length.to_string());
-    response
-}
-
-fn empty_response(status: StatusCode) -> HttpResponse {
-    full_response(status, Bytes::new())
-}
-
-fn insert_header(response: &mut HttpResponse, name: HeaderName, value: &str) {
-    if let Ok(value) = HeaderValue::from_str(value) {
-        response.headers_mut().insert(name, value);
-    }
-}
-
 fn strip_query<B>(request: &mut Request<B>) {
     if request.uri().query().is_some()
         && let Ok(uri) = request.uri().path().parse()
@@ -531,3 +494,24 @@ fn strip_query<B>(request: &mut Request<B>) {
         *request.uri_mut() = uri;
     }
 }
+
+fn request_trace<B>(request: &Request<B>) -> Option<&Arc<HttpTraceExchange>> {
+    request.extensions().get::<Arc<HttpTraceExchange>>()
+}
+
+fn set_trace_route<B>(request: &Request<B>, route: TraceRoute) {
+    if let Some(trace) = request_trace(request) {
+        trace.set_route(route);
+    }
+}
+
+fn manager_error_reason(error: ManagerError) -> &'static str {
+    match error {
+        ManagerError::Authentication => "authentication",
+        ManagerError::Backpressure => "backpressure",
+        ManagerError::Limit => "limit",
+        ManagerError::Protocol => "protocol",
+        ManagerError::Concurrent => "concurrent",
+        ManagerError::Closed => "closed",
+    }
+}