plecto-server 0.4.2

//! The transport-agnostic transaction core: route → chain (request side) → forward → chain
//! (response side). HTTP/1.1, HTTP/2 and HTTP/3 all funnel through `proxy_core`; only the body
//! adapters differ. Bounded retry onto another instance (ADR 000023, hardened with jittered
//! backoff + retriable-5xx retry in ADR 000030) and the `on-request-body` hook (ADR 000025) live in
//! `forward`/`retry`; this module wires routing, rate-limiting, and the chain around that call.

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};

use hyper::body::Body;
use hyper::{Response, StatusCode};
use plecto_control::otlp::SpanRecord;
use plecto_control::{
    ChainOutcome, ConfigSnapshot, HashInput, HashKeySource, HttpRequest, HttpResponse,
    RateLimitDecision, RequestBodyOutcome, RequestTrace, ResponseOutcome, RouteInfo,
};
use tokio::sync::OwnedSemaphorePermit;

use crate::body::{INBOUND_BODY_READ_TIMEOUT, MAX_REQUEST_BODY_BUFFER, buffer_request_body};
use crate::error::ServerError;
use crate::forward::{ForwardBody, ForwardOutcome, ForwardRequest, forward_with_retry};
use crate::headers::{
    copy_headers, copy_headers_direct, headers_to_vec, set_forwarded, to_http_request,
};
use crate::respond::{
    discard_upstream_body, fault, http_response, stream_response, stream_response_direct, synth,
    synth_retry_after, with_error_code,
};
use crate::{ReqBody, ResponseBody, ServerState, access_log};

/// The transport-agnostic transaction core (Stage A observability wrapper, ADR 000009). Every
/// transport funnels through here, so it is the one place to tally per-request metrics and emit the
/// access log: it times `proxy_core_inner`, records the RED signals, and — when enabled — logs the
/// request, then returns the inner result unchanged.
pub(crate) async fn proxy_core(
    state: Arc<ServerState>,
    scheme: &'static str,
    peer: SocketAddr,
    parts: hyper::http::request::Parts,
    body: ReqBody,
) -> Result<Response<ResponseBody>, ServerError> {
    /// Decrements the in-flight gauge on drop: hyper drops this future when the client
    /// connection dies mid-request (h2 RST_STREAM, disconnect), so a plain post-`.await`
    /// decrement would leak and the gauge would drift upward monotonically. The RED tally
    /// (`record_request`) is deliberately still skipped on cancellation — no response was sent.
    struct InFlight<'a>(&'a crate::metrics::ServerMetrics);
    impl Drop for InFlight<'_> {
        fn drop(&mut self) {
            self.0.dec_in_flight();
        }
    }

    let start = Instant::now();
    state.metrics.inc_in_flight();
    let in_flight = InFlight(&state.metrics);

    // Capture the access-log fields BEFORE the core consumes `parts`, and only when logging is on —
    // a disabled access log allocates nothing on the hot path.
    let access = state
        .control
        .access_log_enabled()
        .then(|| access_log::Access {
            method: parts.method.as_str().to_string(),
            authority: crate::headers::request_authority(&parts),
            path: parts.uri.path().to_string(),
        });

    // Continue an inbound distributed trace (ADR 000009): if the caller sent a W3C `traceparent`,
    // parse it and pin the transaction to it so Plecto's spans JOIN the caller's trace — the
    // inbound span becomes the REMOTE PARENT of a locally-minted request span (ADR 000040).
    // Fail-soft: a missing / malformed header falls back to a new root, never a panic on
    // untrusted input (review f000005 P1#2; `from_traceparent` is the fail-soft parser).
    let trace = parts
        .headers
        .get("traceparent")
        .and_then(|v| v.to_str().ok())
        .and_then(RequestTrace::from_traceparent)
        .unwrap_or_else(RequestTrace::root);

    // Request-span fields for OTLP export (ADR 000040), captured BEFORE the core consumes
    // `parts` and only when a configured exporter will actually see them — like the access log,
    // a disabled (or unsampled) transaction allocates nothing on the hot path.
    let otlp_request = state.otlp.as_ref().filter(|_| trace.is_sampled()).map(|_| {
        (
            parts.method.as_str().to_string(),
            parts.uri.path().to_string(),
            SystemTime::now(),
        )
    });

    let result = proxy_core_inner(state.clone(), scheme, peer, trace, parts, body).await;

    drop(in_flight);
    let status = match &result {
        Ok(resp) => resp.status().as_u16(),
        // an inner error is mapped to 502 by the caller (`dispatch::handle`), so record it as such.
        Err(_) => StatusCode::BAD_GATEWAY.as_u16(),
    };
    let elapsed = start.elapsed();
    state.metrics.record_request(status, elapsed);
    if let Some(access) = access {
        access_log::record(scheme, peer, &access, status, elapsed);
    }
    // One SERVER span per sampled transaction (ADR 000040): the root the filter spans (and the
    // upstream's own trace, via the propagated traceparent) nest under. Push is a bounded-queue
    // append — a slow collector can never back-pressure this path.
    if let (Some(buffer), Some((method, path, started))) = (state.otlp.as_ref(), otlp_request) {
        buffer.push(SpanRecord::request_span(
            &trace, &method, &path, scheme, status, started, elapsed,
        ));
    }
    result
}

/// The transaction core proper: route → chain (request side) → forward → chain (response side).
/// Takes the request head + a boxed body (so HTTP/1.1, HTTP/2 and HTTP/3 all share it) and returns
/// the client-visible response. Errors here are upstream/transport failures the caller maps to 502.
async fn proxy_core_inner(
    state: Arc<ServerState>,
    scheme: &'static str,
    peer: SocketAddr,
    trace: RequestTrace,
    mut parts: hyper::http::request::Parts,
    body: ReqBody,
) -> Result<Response<ResponseBody>, ServerError> {
    // Client-IP propagation, edge model (ADR 000018 / review f000005 P2#3): strip any inbound
    // `X-Forwarded-*` / `Forwarded` (which an untrusted client can forge) and set them afresh from
    // the connection's real peer + scheme — directly on the inbound `HeaderMap`, BEFORE the
    // contract projection, so the chain (IP-based rate-limit / auth filters), route matching, the
    // filterless direct forward path, and the upstream all see the same trusted values.
    set_forwarded(&mut parts.headers, peer.ip(), scheme);

    let mut http_req = to_http_request(&parts, scheme);
    // `exact() == Some(0)` is hyper's framing-accurate "no body", computed up front before the
    // body moves: only a bodyless request can be an Upgrade handshake (ADR 000048), and bodyless
    // maps to the trivially replayable `ForwardBody::Bodyless` retry contract (ADR 000058) below.
    let bodyless = body.size_hint().exact() == Some(0);

    // Normalize the request path once at ingress (CWE-22 Path Traversal / CWE-436
    // Interpretation Conflict): route selection, the filter chain, and the forwarded path then all
    // use the SAME normalized path, so the upstream cannot re-derive a stricter path than the
    // (possibly laxer, unfiltered) route we selected — closing the per-route-filter bypass. An
    // ambiguous (encoded-separator) or root-escaping path is rejected fail-closed.
    match plecto_control::normalize_path(&http_req.path) {
        // Borrowed = already normalized (the common case); only a rewritten path is stored back.
        Some(std::borrow::Cow::Owned(path)) => http_req.path = path,
        Some(std::borrow::Cow::Borrowed(_)) => {}
        None => {
            return Ok(with_error_code(
                synth(
                    StatusCode::BAD_REQUEST,
                    &fault::BAD_PATH,
                    b"bad request path",
                ),
                &plecto_control::PATH_NORMALIZATION_REJECTED,
            ));
        }
    }

    // One snapshot pins config + trace for the whole transaction (a concurrent reload cannot
    // desync the request and response halves); cloning it is a cheap Arc + trace-id clone. The
    // trace itself was resolved by `proxy_core`, which also emits the request span (ADR 000040).
    let snapshot = state.control.snapshot_with_trace(trace);
    // Match against the full request — host, path, method, headers, query (ADR 000034); the most
    // specific route wins. `http_req` carries the forwarded-header-corrected inbound request.
    let Some(route) = snapshot.find_route(&http_req) else {
        return Ok(synth(StatusCode::NOT_FOUND, &fault::NO_ROUTE, b"no route"));
    };
    let idx = route.index;

    // Native rate limit (ADR 000033): a coarse token-bucket baseline consulted at the front door —
    // BEFORE the route's filter chain — so a flood is shed without spending any WASM CPU. The
    // peer-keyed (or route-keyed) bucket math is host-native and never crosses the WASM boundary.
    // Over the limit fails closed with 429 + `Retry-After`, distinct from the breaker's 503
    // (`circuit-open`, upstream saturated): this is the client over its inbound rate floor. The
    // per-filter `host-ratelimit` capability (ADR 000026) is a separate, policy-shaped limiter.
    if let RateLimitDecision::Limit { retry_after_ms } = route.check_rate_limit(peer.ip()) {
        state.metrics.inc_rate_limited();
        return Ok(with_error_code(
            synth_retry_after(
                StatusCode::TOO_MANY_REQUESTS,
                &fault::RATE_LIMITED,
                b"rate limit exceeded",
                retry_after_ms.div_ceil(1000),
            ),
            &plecto_control::QUOTA_EXCEEDED,
        ));
    }

    let upgrade = upgrade_intent(&route, &mut parts, bodyless);

    // The chain view the remaining phases share (request body hook, upgrade switch, response
    // side): the pinned snapshot + the selected route's chain identity.
    let chain = ChainRef {
        snapshot: &snapshot,
        idx,
        has_filters: route.has_filters,
    };

    // --- request side: the route's chain on the blocking pool (sync wasmtime, !Send Store).
    // A route with no filters skips the hop entirely — an empty chain is the identity, and the
    // blocking-pool handoff (~µs each way) would be the pure-proxy path's single largest tax. ---
    let forward = if route.has_filters {
        let snap_req = snapshot.clone();
        match tokio::task::spawn_blocking(move || snap_req.dispatch_request(idx, http_req)).await? {
            ChainOutcome::Respond(resp) => return Ok(http_response(resp)),
            ChainOutcome::Forward(req) => req,
        }
    } else {
        http_req
    };

    // Weighted traffic split (ADR 000034): pick which backend upstream group to forward to, in the
    // route's weighted proportion, skipping any backend with no eligible instance (renormalize over
    // healthy). `None` = no backend has a healthy instance → fail closed 503 (the no-healthy fault,
    // ADR 000017 / 000024). The chosen group then drives the existing instance LB / retry / breaker /
    // health below; a single-upstream route is a one-element split, so this is uniform.
    let Some(group) = route.pick_upstream() else {
        return Ok(synth(
            StatusCode::SERVICE_UNAVAILABLE,
            &fault::NO_HEALTHY_UPSTREAM,
            b"no healthy upstream",
        ));
    };

    // Maglev consistent-hashing key (ADR 000035): for a `maglev` upstream, project the request's
    // hash key — a named header's value (borrowed bytes) or the connection peer's IP (hashed as
    // canonical octets, NOT a spoofable forwarding header). `None` for the other algorithms, or when
    // the configured header is absent (the group then falls back to round-robin). Borrowed from
    // `parts`, which outlives the retry loop, and `Copy`, so each attempt reuses it unchanged.
    let hash_key: Option<HashInput<'_>> = group.hash_key_source().and_then(|src| match src {
        HashKeySource::Header(name) => parts
            .headers
            .get(name)
            .map(|v| HashInput::Bytes(v.as_bytes())),
        HashKeySource::SourceIp => Some(HashInput::Ip(peer.ip())),
    });

    // --- forward to a healthy instance, with bounded retry onto ANOTHER instance on a retryable
    // failure (ADR 000019 timeout / 000023 retry). The per-attempt invariants are computed once. ---
    let upstream_path = route.rewrite_path(&forward.path);
    let timeout = group.request_timeout();
    // Overall request deadline across all attempts + backoff (ADR 000031); `None` = no overall bound
    // (only the per-try `timeout` applies). Pinned once, so the budget shrinks as retries consume it.
    let overall = group.overall_timeout();
    let overall_deadline = (!overall.is_zero()).then(|| Instant::now() + overall);
    // The retry loop's view of the body (ADR 000058): bodyless and buffered bodies are
    // replayable, an opaque streamed body (ADR 000013) moves into its single attempt.
    let mut real_body = if bodyless {
        ForwardBody::Bodyless
    } else {
        ForwardBody::OneShot(body)
    };

    // The buffer permit must outlive the BUFFER, not just the read: the buffered bytes live on in
    // `ForwardBody::Replayable` through the whole forward/retry/response phase, so a permit
    // dropped inside the hook would bound concurrent *reads* while resident buffered-body memory
    // grew past `MAX_INFLIGHT_BODY_BUFFERS × MAX_REQUEST_BODY_BUFFER` under slow upstreams.
    // Bound at function scope so it drops with the transaction.
    let _buf_permit =
        match request_body_hook(&state, &chain, route.reads_body, &mut real_body).await? {
            BodyHookOutcome::Proceed(permit) => permit,
            BodyHookOutcome::Respond(resp) => return Ok(resp),
        };

    // Circuit breaker (ADR 000028): take an in-flight slot under this upstream's `max_requests` cap
    // before forwarding. At the cap, shed load with a fast-fail 503 rather than queueing work onto a
    // saturated backend. One slot per request, held across the retry loop and released by RAII on
    // every return path; an unlimited breaker (the default) is a zero-cost no-op permit. The breaker
    // is overload protection, NOT a health signal — a shed request never demotes an instance.
    let permit = match group.try_acquire() {
        Some(permit) => permit,
        None => {
            state.metrics.inc_circuit_open();
            return Ok(synth(
                StatusCode::SERVICE_UNAVAILABLE,
                &fault::CIRCUIT_OPEN,
                b"upstream overloaded",
            ));
        }
    };

    // First pick per the upstream's LB algorithm (ADR 000035): round-robin, least-request (P2C), or
    // maglev (by `hash_key`). Fail closed (503) if no instance is eligible (ADR 000017). The `Pick`
    // carries the least-request load guard (a no-op otherwise) and is held across the retry loop, so
    // an instance's active-request count is decremented on every exit and on each retry hand-off.
    let Some(pick) = group.pick(hash_key) else {
        return Ok(synth(
            StatusCode::SERVICE_UNAVAILABLE,
            &fault::NO_HEALTHY_UPSTREAM,
            b"no healthy upstream",
        ));
    };

    let forward_req = ForwardRequest {
        method: forward.method.as_str(),
        // The filterless fast path forwards the inbound headers directly (no contract re-parse
        // per attempt) — `set_forwarded` already corrected them on `parts.headers`, and the
        // hop-by-hop strip applies at copy time either way. A filtered route forwards the
        // chain-edited contract bytes.
        headers: if route.has_filters {
            crate::forward::AttemptHeaders::Chain(&forward.headers)
        } else {
            crate::forward::AttemptHeaders::Direct(&parts.headers)
        },
        original_headers: &parts.headers,
        authority: &forward.authority,
        upstream_path: &upstream_path,
        traceparent: &snapshot.traceparent(),
        upgrade_token: upgrade.as_ref().map(|(t, _, _)| t.as_str()),
    };
    // The client for this group's security context (ADR 000042): the shared plain client, or the
    // pooled TLS client for its `[upstream.tls]` config. A cheap clone (shared pool inside).
    let client = state.clients.for_group(&group);
    let (upstream_resp, upstream_pick) = match forward_with_retry(
        &client,
        &state.metrics,
        &group,
        pick,
        hash_key,
        forward_req,
        real_body,
        timeout,
        overall_deadline,
        group.max_retries(),
    )
    .await
    {
        ForwardOutcome::Response(resp, pick) => (resp, pick),
        ForwardOutcome::OverallTimeout => {
            return Ok(synth(
                StatusCode::GATEWAY_TIMEOUT,
                &fault::REQUEST_TIMEOUT,
                b"request timeout",
            ));
        }
        ForwardOutcome::PerTryTimeout => {
            return Ok(synth(
                StatusCode::GATEWAY_TIMEOUT,
                &fault::UPSTREAM_TIMEOUT,
                b"upstream timeout",
            ));
        }
        ForwardOutcome::SendFailed(e) => return Err(e.into()),
        ForwardOutcome::BuildFailed(e) => return Err(e.into()),
    };

    // --- upgrade switch (ADR 000048): a verified 101 splices the two connections into an opaque
    // tunnel; any other status falls through to the normal response path (the handshake was
    // simply refused — the upstream's answer is a legitimate response). ---
    if upstream_resp.status() == StatusCode::SWITCHING_PROTOCOLS {
        return upgrade_switch(
            &state,
            &chain,
            forward,
            upgrade,
            upstream_resp,
            permit,
            upstream_pick,
        )
        .await;
    }

    respond_through_chain(&chain, &route, &parts, forward, upstream_resp).await
}

/// One transaction's pinned view of the filter chain — the config snapshot plus the selected
/// route's chain identity — shared by every phase that may consult the chain (request body
/// hook, upgrade switch, response side), and nothing else.
struct ChainRef<'a> {
    snapshot: &'a ConfigSnapshot,
    idx: usize,
    has_filters: bool,
}

/// What a granted HTTP/1.1 Upgrade opt-in (ADR 000048) carries to the switch: the allowlisted
/// token, the client connection's upgrade handle, and the tunnel's idle timeout (`None` = no
/// idle bound).
type UpgradeIntent = (String, hyper::upgrade::OnUpgrade, Option<Duration>);

/// HTTP/1.1 Upgrade opt-in (ADR 000048): tunnel only when the route declares the token, the
/// client genuinely asked (`Connection: upgrade` + an allowlisted `Upgrade`), the transport
/// can hand its connection over (h1 — hyper left an `OnUpgrade` on the request; h2/h3 never
/// do), and the handshake is bodyless. Anything else stays a plain HTTP request with the
/// Upgrade/Connection pair stripped as hop-by-hop, exactly as before.
fn upgrade_intent(
    route: &RouteInfo,
    parts: &mut hyper::http::request::Parts,
    bodyless: bool,
) -> Option<UpgradeIntent> {
    if !bodyless {
        return None;
    }
    let cfg = route.upgrade.as_ref()?;
    let header = crate::headers::upgrade_request_header(&parts.headers)?;
    let token = cfg.allowed_token(header)?.to_string();
    let on_upgrade = parts.extensions.remove::<hyper::upgrade::OnUpgrade>()?;
    Some((token, on_upgrade, cfg.idle_timeout()))
}

/// What the request-side body hook settled on: proceed to forwarding (carrying the buffer
/// permit for as long as the buffered bytes stay resident), or answer the client now — a
/// filter's short-circuit or one of the buffering path's fail-closed statuses.
enum BodyHookOutcome {
    Proceed(Option<OwnedSemaphorePermit>),
    Respond(Response<ResponseBody>),
}

/// Request-side body hook (ADR 000025): buffer the body (bounded) ONLY when a filter on the
/// route actually reads it — i.e. exports `on-request-body` (`reads_body`, ADR 000038). A route
/// with no body-reading filter (or a bodyless request) skips this entirely and keeps the body on
/// the zero-copy streaming path — the real fix for the body-tax (docs/servey). The chain runs on
/// the blocking pool (sync wasmtime, !Send Store), like the header chain.
async fn request_body_hook(
    state: &ServerState,
    chain: &ChainRef<'_>,
    reads_body: bool,
    real_body: &mut ForwardBody,
) -> Result<BodyHookOutcome, ServerError> {
    if !reads_body {
        return Ok(BodyHookOutcome::Proceed(None));
    }
    let Some(b) = real_body.take_oneshot() else {
        return Ok(BodyHookOutcome::Proceed(None));
    };
    // Bound concurrent buffered-body memory and the time spent reading one body
    // (slow-body slowloris): hold a buffer permit and read under a deadline. Over the
    // size cap → 413, over the time budget → 408 — both fail closed (never an unbounded buffer).
    // An acquire error (the semaphore closed) must fail closed too, not silently proceed
    // without a permit — that would bypass the concurrency cap entirely for this request.
    let permit = match state.body_buffer_limit.clone().acquire_owned().await {
        Ok(permit) => permit,
        Err(_) => {
            return Ok(BodyHookOutcome::Respond(synth(
                StatusCode::SERVICE_UNAVAILABLE,
                &fault::BODY_BUFFER_UNAVAILABLE,
                b"body buffer unavailable",
            )));
        }
    };
    let buffered = match tokio::time::timeout(
        INBOUND_BODY_READ_TIMEOUT,
        buffer_request_body(b, MAX_REQUEST_BODY_BUFFER),
    )
    .await
    {
        Ok(crate::body::BufferOutcome::Buffered(buf)) => buf,
        Ok(crate::body::BufferOutcome::TooLarge) => {
            return Ok(BodyHookOutcome::Respond(synth(
                StatusCode::PAYLOAD_TOO_LARGE,
                &fault::BODY_TOO_LARGE,
                b"request body too large",
            )));
        }
        // A client abort / transport fault mid-body is the CLIENT's fault, not an oversized
        // body — 400, not 413 (the two were previously conflated into one `None`).
        Ok(crate::body::BufferOutcome::ReadError) => {
            return Ok(BodyHookOutcome::Respond(synth(
                StatusCode::BAD_REQUEST,
                &fault::BODY_READ_ERROR,
                b"request body read error",
            )));
        }
        Err(_) => {
            return Ok(BodyHookOutcome::Respond(synth(
                StatusCode::REQUEST_TIMEOUT,
                &fault::BODY_TIMEOUT,
                b"request body read timeout",
            )));
        }
    };
    let snap_body = chain.snapshot.clone();
    let idx = chain.idx;
    match tokio::task::spawn_blocking(move || snap_body.dispatch_request_body(idx, buffered))
        .await?
    {
        RequestBodyOutcome::Respond(resp) => Ok(BodyHookOutcome::Respond(http_response(resp))),
        // The buffered, filter-edited bytes are replayable by definition (ADR 000058):
        // `Vec<u8>` → `Bytes` is a move, and each retry attempt shares it by reference count.
        RequestBodyOutcome::Forward(edited) => {
            *real_body = ForwardBody::Replayable(bytes::Bytes::from(edited));
            Ok(BodyHookOutcome::Proceed(Some(permit)))
        }
    }
}

/// True when the upstream's 101 names the token we offered (case-insensitive, RFC 9110 §7.8).
fn upstream_switched_to(headers: &hyper::HeaderMap, token: &str) -> bool {
    headers
        .get(hyper::header::UPGRADE)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|v| v.split(',').any(|t| t.trim().eq_ignore_ascii_case(token)))
}

/// Upgrade switch (ADR 000048): validate the upstream's 101 against the client's opt-in, relay
/// the handshake back (through the response-side chain when the route has filters — which may
/// veto the switch), then splice the two connections into an opaque tunnel. Only called for a
/// `101 Switching Protocols` upstream response; every arm is terminal for the transaction.
/// `permit` is the breaker slot (ADR 000028) — nameable only as `impl Send` because the breaker
/// module keeps its permit type private — and `pick` the LB in-flight guard; both move into the
/// tunnel task so they live exactly as long as the tunnel.
async fn upgrade_switch(
    state: &Arc<ServerState>,
    chain: &ChainRef<'_>,
    forward: HttpRequest,
    upgrade: Option<UpgradeIntent>,
    mut upstream_resp: Response<ResponseBody>,
    permit: impl Send + 'static,
    pick: plecto_control::Pick,
) -> Result<Response<ResponseBody>, ServerError> {
    let Some((token, downstream_on, idle)) = upgrade else {
        // A 101 the client never solicited (or an unlisted token): relaying it would hand
        // the upstream a raw byte stream the client cannot parse as HTTP — response
        // smuggling. Fail closed (RFC 9110 §7.8: no unrequested switch).
        return Ok(synth(
            StatusCode::BAD_GATEWAY,
            &fault::BAD_UPGRADE,
            b"unsolicited upgrade",
        ));
    };
    // The upstream must switch to the token we offered (case-insensitive, RFC 9110 §7.8).
    if !upstream_switched_to(upstream_resp.headers(), &token) {
        return Ok(synth(
            StatusCode::BAD_GATEWAY,
            &fault::BAD_UPGRADE,
            b"upgrade token mismatch",
        ));
    }
    let upstream_on = hyper::upgrade::on(&mut upstream_resp);
    // The 101 back to the client: the response-side chain still sees the handshake headers
    // (status/headers only), end-to-end headers (e.g. Sec-WebSocket-Accept) pass the
    // hop-by-hop strip, then the switch signal itself is re-issued deliberately.
    let mut builder = Response::builder().status(StatusCode::SWITCHING_PROTOCOLS);
    if chain.has_filters {
        let http_resp = HttpResponse {
            status: StatusCode::SWITCHING_PROTOCOLS.as_u16(),
            headers: headers_to_vec(upstream_resp.headers()),
            body: Vec::new(),
        };
        let snap_resp = chain.snapshot.clone();
        let idx = chain.idx;
        let outcome = tokio::task::spawn_blocking(move || {
            snap_resp.dispatch_response(idx, &forward, http_resp)
        })
        .await?;
        let edited = match outcome {
            // A response filter replaced the handshake (or trapped fail-closed): honour
            // its response and never splice — the upstream connection drops with
            // `upstream_on`.
            ResponseOutcome::Respond(resp) => return Ok(http_response(resp)),
            ResponseOutcome::Forward(edited) => edited,
        };
        if edited.status != StatusCode::SWITCHING_PROTOCOLS.as_u16() {
            // A response filter vetoed the switch by demoting the status: honour that
            // response (header-only) and never splice.
            return Ok(http_response(edited));
        }
        copy_headers(builder.headers_mut(), &edited.headers);
    } else {
        copy_headers_direct(builder.headers_mut(), upstream_resp.headers());
    }
    // Unreachable for a validated allowlist token — but if it were ever reached, silently
    // omitting `Upgrade` while still sending `Connection: upgrade` and splicing would relay a
    // malformed 101. Fail closed instead (bp-rust: no implicit fail-open).
    let Ok(upgrade_value) = hyper::header::HeaderValue::from_str(&token) else {
        return Ok(synth(
            StatusCode::BAD_GATEWAY,
            &fault::BAD_UPGRADE,
            b"bad upgrade token",
        ));
    };
    if let Some(h) = builder.headers_mut() {
        h.insert(hyper::header::UPGRADE, upgrade_value);
        h.insert(
            hyper::header::CONNECTION,
            hyper::header::HeaderValue::from_static("upgrade"),
        );
    }
    let resp101 = builder
        .body(crate::body::full(Vec::new()))
        .map_err(ServerError::from)?;
    let drain = state.drain.clone();
    let metrics = state.metrics.clone();
    let tunnel_active = crate::metrics::TunnelActive::new(metrics.clone());
    tokio::spawn(async move {
        // Long-lived tunnels stay inside the existing resource accounting (ADR 000048): the
        // breaker permit (ADR 000028), the LB pick guard (least-request in-flight) and the
        // `tunnels_active` gauge guard (ADR 000059) live exactly as long as the tunnel, and
        // the drain flag closes it at shutdown. The byte totals are recorded once, at close.
        let _permit = permit;
        let _pick = pick;
        let _active = tunnel_active;
        let (down, up) = crate::tunnel::run(downstream_on, upstream_on, idle, drain).await;
        metrics.add_tunnel_bytes(down, up);
    });
    Ok(resp101)
}

/// Response side: the route's chain in reverse (status / headers only), then compression LAST
/// (ADR 000074 — filters always see identity, and a synthesised `Respond` is deliberately never
/// compressed). A filterless route skips the blocking-pool hop and the contract projection;
/// the hop-by-hop strip still applies, directly on the original header bytes.
async fn respond_through_chain(
    chain: &ChainRef<'_>,
    route: &RouteInfo,
    parts: &hyper::http::request::Parts,
    forward: HttpRequest,
    upstream_resp: Response<ResponseBody>,
) -> Result<Response<ResponseBody>, ServerError> {
    let (uparts, ubody) = upstream_resp.into_parts();
    if !chain.has_filters {
        let resp = stream_response_direct(uparts.status, &uparts.headers, ubody);
        return Ok(crate::compression::apply(resp, route, parts));
    }
    let http_resp = HttpResponse {
        status: uparts.status.as_u16(),
        headers: headers_to_vec(&uparts.headers),
        body: Vec::new(), // header-only: filters never see the streamed body
    };
    // The response chain sees the AS-FORWARDED request snapshot (ADR 000073): `forward` is the
    // request exactly as it left the request-side chain (filter edits applied, before the
    // egress hop-by-hop strip / path rewrite / traceparent injection), moved here for free —
    // no per-request copy is added to hold it.
    let snap_resp = chain.snapshot.clone();
    let idx = chain.idx;
    let outcome =
        tokio::task::spawn_blocking(move || snap_resp.dispatch_response(idx, &forward, http_resp))
            .await?;

    // The typed successor of the old in-band signal (ADR 000073): `Forward` sends the edited
    // status + headers and streams the upstream body through; `Respond` is a synthesised
    // response — a filter's `replace` or the chain's fail-closed 5xx. The upstream body is
    // discarded without blocking the client (background drain up to a cap, else socket close)
    // so a replace does not permanently poison the upstream connection pool.
    match outcome {
        ResponseOutcome::Forward(edited) => {
            // The upstream's original headers ride along so framing stays host-owned (the chain
            // cannot desync `Content-Length` from the streamed body).
            let resp = stream_response(edited.status, &edited.headers, &uparts.headers, ubody);
            Ok(crate::compression::apply(resp, route, parts))
        }
        ResponseOutcome::Respond(resp) => {
            discard_upstream_body(ubody);
            Ok(http_response(resp))
        }
    }
}