plecto-server 0.9.0

//! 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::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};

use bytes::Bytes;
use hyper::body::Body;
use hyper::header::{CONTENT_ENCODING, CONTENT_RANGE, CONTENT_TYPE};
use hyper::{Response, StatusCode};
use plecto_control::otlp::SpanRecord;
use plecto_control::{
    ChainOutcome, ConfigSnapshot, HashInput, HashKeySource, HttpRequest, HttpResponse, OverCapMode,
    RateLimitDecision, RequestBodyOutcome, RequestTrace, ResponseBodyConfig, ResponseBodyOutcome,
    ResponseOutcome, RouteInfo, UninspectableMode,
};
use tokio::sync::OwnedSemaphorePermit;

use crate::body::{
    INBOUND_BODY_READ_TIMEOUT, MAX_REQUEST_BODY_BUFFER, ResponseBufferOutcome,
    UPSTREAM_BODY_READ_TIMEOUT, buffer_request_body, buffer_response_body, hold_budget, prefixed,
};
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::metrics::InspectionSkip;
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,
    mut 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);

    // 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 —
    // 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. The address it settles on is the client every
    // per-client observation point below uses (access log, rate-limit bucket, source-IP hash):
    // the connection peer, or — behind a declared front proxy (ADR 000103) — the client that
    // proxy named.
    let client = set_forwarded(
        &mut parts.headers,
        peer.ip(),
        scheme,
        state.trusted_proxy.as_ref(),
    );

    // 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(),
        )
    });

    // Why this transaction's response body escaped the route's `on-response-body` filters, if it
    // did (ADR 000098). The metric is tallied where the decision is made; this carries the reason
    // out to the access log, so a skip is attributable to a request and not only to a counter.
    let mut inspection_skipped: Option<InspectionSkip> = None;
    let result = proxy_core_inner(
        state.clone(),
        scheme,
        client,
        trace,
        parts,
        body,
        &mut inspection_skipped,
    )
    .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,
            client,
            &access,
            status,
            elapsed,
            &trace,
            inspection_skipped.map(InspectionSkip::as_str),
        );
    }
    // 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,
    client: IpAddr,
    trace: RequestTrace,
    mut parts: hyper::http::request::Parts,
    body: ReqBody,
    inspection_skipped: &mut Option<InspectionSkip>,
) -> Result<Response<ResponseBody>, ServerError> {
    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_with_query) {
        // Borrowed = already normalized (the common case); only a rewritten path is stored back.
        Some(std::borrow::Cow::Owned(path)) => http_req.path_with_query = 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;

    // From here on the route is chosen, so EVERY response this transaction can produce is that
    // route's to answer. They funnel through the single exit below, where the operator's declared
    // `[route.headers]` (ADR 000100) is stamped on — "every route-attributed response" is then a
    // property of the control flow, not a checklist over a dozen scattered `return`s.
    let routed: Result<Routed, ServerError> = async {
        // 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 client-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(client) {
            state.metrics.inc_rate_limited();
            return Ok(Routed::Synthesised(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 mut 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(Routed::Synthesised(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(Routed::Synthesised(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 client's IP (hashed
        // as canonical octets; the resolved client, never a claim Plecto did not authorise).
        // `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(client)),
        });

        // --- forward to a healthy instance, with bounded retry onto ANOTHER instance on a
        // retryable failure (ADR 000019 timeout / 000023 retry). Per-attempt invariants once. ---
        let upstream_path = route.rewrite_path(&forward.path_with_query).into_owned();
        // Both bounds are the ROUTE's (ADR 000102): its `[route.timeouts]` override when it
        // declares one, else the default declared by the upstream this request is going to. Routes
        // with different latency budgets can therefore share one upstream.
        let timeout = route.request_timeout(&group);
        // 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 = route.overall_timeout(&group);
        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 transaction scope so it drops with the transaction.
        let _buf_permit = match request_body_hook(
            &state,
            &chain,
            route.body_hooks.request,
            &mut real_body,
            &mut forward.headers,
        )
        .await?
        {
            BodyHookOutcome::Proceed(permit) => permit,
            BodyHookOutcome::Respond(resp) => return Ok(Routed::Synthesised(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(Routed::Synthesised(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(Routed::Synthesised(synth(
                StatusCode::SERVICE_UNAVAILABLE,
                &fault::NO_HEALTHY_UPSTREAM,
                b"no healthy upstream",
            )));
        };

        // A route that inspects response bodies asks the upstream for something inspectable: the
        // negotiated-encoding and byte-range request headers are stripped from the FORWARDED
        // request (ADR 000098), so an encoded or partial response becomes the anomaly the
        // fail-closed classes are for instead of the normal case. Applied last, after the request
        // body hook, so a filter cannot re-add one. The client's own headers are untouched —
        // `[route.compression]` still negotiates against them on the way back.
        if route.body_hooks.response {
            strip_inspection_hostile_headers(&mut forward.headers);
        }

        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(Routed::Synthesised(synth(
                    StatusCode::GATEWAY_TIMEOUT,
                    &fault::REQUEST_TIMEOUT,
                    b"request timeout",
                )));
            }
            ForwardOutcome::PerTryTimeout => {
                return Ok(Routed::Synthesised(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
            .map(Routed::Synthesised);
        }

        respond_through_chain(
            &state,
            &chain,
            &route,
            forward,
            upstream_resp,
            inspection_skipped,
        )
        .await
    }
    .await;

    // A transport fault raised AFTER the route was chosen is still that route's response — the
    // same 502 the transports synthesise for an escaping error, built here so the declaration
    // below reaches it too.
    let routed = routed.unwrap_or_else(|e| {
        tracing::warn!(error = %e, "fast-path error");
        Routed::Synthesised(synth(
            StatusCode::BAD_GATEWAY,
            &fault::UPSTREAM,
            b"upstream error",
        ))
    });

    // The single exit (ADR 000100 decision 2). The declaration lands AFTER the response chain, so
    // neither a filter's `replace` nor a fail-closed fault can drop it, and BEFORE compression, so
    // the codec's own `Vary` / `Content-Encoding` stay the last word on the representation. Only a
    // forwarded upstream stream is a compression candidate — a synthesised response is host-framed
    // and never transformed (ADR 000075 decision 1).
    Ok(match routed {
        Routed::Synthesised(resp) => with_declared_headers(resp, &route),
        Routed::Forwarded(resp) => {
            crate::compression::apply(with_declared_headers(resp, &route), &route, &parts)
        }
    })
}

/// What the routed half of a transaction settled on. The split is the compression rule (ADR 000075
/// decision 1): only a FORWARDED upstream stream is a candidate for `[route.compression]`, while a
/// response Plecto synthesised is host-framed and never transformed. Both carry the route's
/// declared `[route.headers]` (ADR 000100).
enum Routed {
    /// A response Plecto built for this route: a filter's short-circuit or `replace`, the chain's
    /// fail-closed 5xx, the native 429, a forward-side 5xx, or an upgrade handshake.
    Synthesised(Response<ResponseBody>),
    /// The upstream's response, streamed back through the (possibly empty) response chain.
    Forwarded(Response<ResponseBody>),
}

/// Stamp a route's declared `[route.headers]` (ADR 000100) onto an about-to-be-sent response.
/// A route with no declaration is a no-op.
fn with_declared_headers(
    mut resp: Response<ResponseBody>,
    route: &RouteInfo,
) -> Response<ResponseBody> {
    if let Some(declared) = route.response_headers.as_deref() {
        declared.apply(resp.headers_mut());
    }
    resp
}

/// 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,
    headers: &mut Vec<plecto_control::Header>,
) -> 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): reserve this body's cap out of the shared byte budget 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 budget entirely for this request.
    let permit = match state
        .body_buffer_budget
        .clone()
        .acquire_many_owned(budget_permits(MAX_REQUEST_BODY_BUFFER))
        .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;
    // The chain runs on the blocking pool, so the headers travel there and back by value — a
    // body hook's `modified` decision can edit them (ADR 000098), and the edit has to land on the
    // request this transaction is about to forward.
    let chain_headers = std::mem::take(headers);
    match tokio::task::spawn_blocking(move || {
        snap_body.dispatch_request_body(idx, buffered, chain_headers)
    })
    .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 {
            body: edited,
            headers: edited_headers,
        } => {
            *real_body = ForwardBody::Replayable(bytes::Bytes::from(edited));
            *headers = edited_headers;
            Ok(BodyHookOutcome::Proceed(Some(permit)))
        }
    }
}

/// The request headers a response-inspecting route strips from the request it forwards (ADR
/// 000098): the ones that make the upstream answer with something other than a whole identity
/// representation.
const INSPECTION_HOSTILE_REQUEST_HEADERS: [&str; 3] = ["accept-encoding", "range", "if-range"];

fn strip_inspection_hostile_headers(headers: &mut Vec<plecto_control::Header>) {
    headers.retain(|h| {
        !INSPECTION_HOSTILE_REQUEST_HEADERS
            .iter()
            .any(|n| h.name.eq_ignore_ascii_case(n))
    });
}

/// One body's reservation out of the shared byte budget. Route validation keeps every cap far
/// inside `u32`, so the clamp is a total-function guard, not a reachable case.
fn budget_permits(bytes: usize) -> u32 {
    u32::try_from(bytes).unwrap_or(u32::MAX)
}

/// 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). What comes out is either
/// the forwarded upstream stream — which the caller's single exit then stamps and compresses, after
/// the chain, so filters always see identity (ADR 000074) — or a synthesised response, which is
/// host-framed and never compressed (ADR 000075 decision 1).
/// 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(
    state: &ServerState,
    chain: &ChainRef<'_>,
    route: &RouteInfo,
    forward: HttpRequest,
    upstream_resp: Response<ResponseBody>,
    inspection_skipped: &mut Option<InspectionSkip>,
) -> Result<Routed, ServerError> {
    let (uparts, ubody) = upstream_resp.into_parts();
    if !chain.has_filters {
        return Ok(Routed::Forwarded(stream_response_direct(
            uparts.status,
            &uparts.headers,
            ubody,
        )));
    }
    // Classify BEFORE the chain runs, on the response exactly as the upstream sent it (ADR 000098
    // decision 3). Deciding on a content-type a chain filter could have rewritten would let an
    // earlier filter steer a response past the one that inspects it.
    let inspectability = route
        .body_hooks
        .response
        .then(|| classify_response(uparts.status, &uparts.headers, &route.response_body));

    let http_resp = HttpResponse {
        status: uparts.status.as_u16(),
        headers: headers_to_vec(&uparts.headers),
        body: Vec::new(), // header-only: this hook never sees the 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. It comes back out because the body hook, which
    // runs after this one, is handed the same snapshot.
    let snap_resp = chain.snapshot.clone();
    let idx = chain.idx;
    let (outcome, forward) = tokio::task::spawn_blocking(move || {
        let outcome = snap_resp.dispatch_response(idx, &forward, http_resp);
        (outcome, forward)
    })
    .await?;

    // The typed successor of the old in-band signal (ADR 000073): `Forward` sends the edited
    // status + headers; `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 — and, per ADR 000098, a replaced response has no upstream body left to
    // inspect, so the body hook does not run at all.
    let edited = match outcome {
        ResponseOutcome::Forward(edited) => edited,
        ResponseOutcome::Respond(resp) => {
            discard_upstream_body(ubody);
            return Ok(Routed::Synthesised(http_response(resp)));
        }
    };

    match inspectability {
        // No response-body filter on the route, or a response outside its declared media-type
        // scope: the body streams through untouched, exactly as before this hook existed. The
        // upstream's original headers ride along so framing stays host-owned (the chain cannot
        // desync `Content-Length` from the streamed body).
        None | Some(Inspectability::OutOfScope) => Ok(Routed::Forwarded(stream_response(
            edited.status,
            &edited.headers,
            &uparts.headers,
            ubody,
        ))),
        Some(Inspectability::Uninspectable(reason)) => {
            state.metrics.inc_inspection_skipped(reason);
            *inspection_skipped = Some(reason);
            match route.response_body.uninspectable() {
                UninspectableMode::Passthrough => Ok(Routed::Forwarded(stream_response(
                    edited.status,
                    &edited.headers,
                    &uparts.headers,
                    ubody,
                ))),
                UninspectableMode::Reject => {
                    discard_upstream_body(ubody);
                    Ok(Routed::Synthesised(synth(
                        StatusCode::BAD_GATEWAY,
                        &fault::RESPONSE_BODY_UNINSPECTABLE,
                        b"response body could not be inspected",
                    )))
                }
            }
        }
        Some(Inspectability::Inspect) => {
            inspect_response_body(
                state,
                chain,
                route,
                forward,
                edited,
                uparts.headers,
                ubody,
                inspection_skipped,
            )
            .await
        }
    }
}

/// What the host can do with an upstream response on a route whose chain reads response bodies
/// (ADR 000098 decision 3).
enum Inspectability {
    /// Buffer it (bounded) and run the hooks.
    Inspect,
    /// Outside the route's declared media-type allowlist. Streams through uninspected and
    /// UNCOUNTED: the allowlist states what the operator asked to inspect, so a response outside
    /// it is not a skip, it is the scope.
    OutOfScope,
    /// Inside the scope, but not inspectable at all. The route's `uninspectable` mode decides
    /// whether it is refused or forwarded; either way the reason is recorded.
    Uninspectable(InspectionSkip),
}

/// Decide what can be done with a response, from the AS-RECEIVED status and headers.
fn classify_response(
    status: StatusCode,
    headers: &hyper::HeaderMap,
    cfg: &ResponseBodyConfig,
) -> Inspectability {
    let essence = headers
        .get(CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.split(';').next().unwrap_or(s).trim())
        .unwrap_or_default();
    // A streaming media type is a REFUSED class, not merely an unlisted one: it is absent from the
    // allowlist because buffering it would break the stream, and letting that absence read as
    // "quietly skip" is the silent-skip this design exists to avoid.
    if is_streaming_media_type(essence) {
        return Inspectability::Uninspectable(InspectionSkip::StreamingContentType);
    }
    if !cfg.content_type_eligible(essence) {
        return Inspectability::OutOfScope;
    }
    // A fragment of a representation cannot be inspected as the representation.
    if status == StatusCode::PARTIAL_CONTENT || headers.contains_key(CONTENT_RANGE) {
        return Inspectability::Uninspectable(InspectionSkip::PartialContent);
    }
    // An encoding that survived the request-side `Accept-Encoding` strip: the bytes are not the
    // ones a filter would be reading. Host-side decompression is a later increment.
    if is_content_encoded(headers) {
        return Inspectability::Uninspectable(InspectionSkip::ContentEncoding);
    }
    Inspectability::Inspect
}

fn is_streaming_media_type(essence: &str) -> bool {
    essence.eq_ignore_ascii_case("text/event-stream")
}

fn is_content_encoded(headers: &hyper::HeaderMap) -> bool {
    headers
        .get_all(CONTENT_ENCODING)
        .iter()
        .any(|v| match v.to_str() {
            Ok(s) => s.split(',').any(|token| {
                let token = token.trim();
                !token.is_empty() && !token.eq_ignore_ascii_case("identity")
            }),
            // An unreadable coding is a coding we cannot rule out. Fail closed.
            Err(_) => true,
        })
}

/// Buffer the upstream body (bounded, reserved out of the shared byte budget) and run the route's
/// `on-response-body` hooks over it (ADR 000098). Nothing has been written to the client yet — the
/// whole response is still a value being built — which is what makes a `replace` here honourable
/// and the fail-closed statuses below reachable.
#[allow(clippy::too_many_arguments)]
async fn inspect_response_body(
    state: &ServerState,
    chain: &ChainRef<'_>,
    route: &RouteInfo,
    forward: HttpRequest,
    edited: HttpResponse,
    upstream_headers: hyper::HeaderMap,
    ubody: ResponseBody,
    inspection_skipped: &mut Option<InspectionSkip>,
) -> Result<Routed, ServerError> {
    let cfg = &route.response_body;
    let cap = cfg.max_bytes();
    let status = edited.status;
    let permit = match state
        .body_buffer_budget
        .clone()
        .acquire_many_owned(budget_permits(cap))
        .await
    {
        Ok(permit) => permit,
        Err(_) => {
            discard_upstream_body(ubody);
            return Ok(Routed::Synthesised(synth(
                StatusCode::SERVICE_UNAVAILABLE,
                &fault::RESPONSE_BODY_BUFFER_UNAVAILABLE,
                b"response body buffer unavailable",
            )));
        }
    };
    let buffered =
        match tokio::time::timeout(UPSTREAM_BODY_READ_TIMEOUT, buffer_response_body(ubody, cap))
            .await
        {
            Ok(outcome) => outcome,
            // The headers are being held for this decision, so a trickling upstream would hold them
            // forever. Bound it, and attribute the fault to the upstream (504), not to the client.
            Err(_) => {
                return Ok(Routed::Synthesised(synth(
                    StatusCode::GATEWAY_TIMEOUT,
                    &fault::RESPONSE_BODY_TIMEOUT,
                    b"upstream response body timeout",
                )));
            }
        };
    // `Some(rest)` means only the head was inspected (`process-partial`) and the remainder still
    // has to be forwarded behind it.
    let (body, rest) = match buffered {
        ResponseBufferOutcome::Buffered(body) => (body, None),
        ResponseBufferOutcome::ReadError => {
            return Ok(Routed::Synthesised(synth(
                StatusCode::BAD_GATEWAY,
                &fault::RESPONSE_BODY_READ_ERROR,
                b"upstream response body read error",
            )));
        }
        ResponseBufferOutcome::OverCap { head, rest } => {
            // This is also where an unbounded stream lands: a body with no declared length that
            // never ends is, at the cap, indistinguishable from one that is merely too big — and
            // gets the same recorded reason and the same route-declared treatment.
            state
                .metrics
                .inc_inspection_skipped(InspectionSkip::OverCap);
            *inspection_skipped = Some(InspectionSkip::OverCap);
            match cfg.over_cap() {
                OverCapMode::Reject => {
                    discard_upstream_body(rest);
                    return Ok(Routed::Synthesised(synth(
                        StatusCode::BAD_GATEWAY,
                        &fault::RESPONSE_BODY_TOO_LARGE,
                        b"response body too large to inspect",
                    )));
                }
                OverCapMode::Passthrough => {
                    return Ok(Routed::Forwarded(stream_response(
                        status,
                        &edited.headers,
                        &upstream_headers,
                        hold_budget(prefixed(Bytes::from(head), rest), permit),
                    )));
                }
                OverCapMode::ProcessPartial => (head, Some(rest)),
            }
        }
    };

    let snap = chain.snapshot.clone();
    let idx = chain.idx;
    let outcome = tokio::task::spawn_blocking(move || {
        snap.dispatch_response_body(idx, &forward, edited, body)
    })
    .await?;

    match outcome {
        ResponseBodyOutcome::Respond(resp) => {
            if let Some(rest) = rest {
                discard_upstream_body(rest);
            }
            Ok(Routed::Synthesised(http_response(resp)))
        }
        ResponseBodyOutcome::Forward {
            body,
            headers,
            transformed,
        } => match rest {
            // The whole body was inspected: send exactly the bytes the chain settled on, framed by
            // the host from those bytes (the upstream's own `Content-Length` is deliberately NOT
            // carried over — it describes bytes a transform may have replaced).
            None => {
                let (parts, body) = http_response(HttpResponse {
                    status,
                    headers,
                    body,
                })
                .into_parts();
                Ok(Routed::Forwarded(Response::from_parts(
                    parts,
                    hold_budget(body, permit),
                )))
            }
            // `process-partial`: the hooks saw a prefix. Forwarding a rewrite of a prefix followed
            // by an untouched remainder would send a body no filter ever asked for, so a transform
            // here fails closed; inspection alone passes through.
            Some(rest) if transformed => {
                discard_upstream_body(rest);
                Ok(Routed::Synthesised(synth(
                    StatusCode::BAD_GATEWAY,
                    &fault::RESPONSE_BODY_PARTIAL_MODIFIED,
                    b"filter rewrote a partially inspected response body",
                )))
            }
            Some(rest) => Ok(Routed::Forwarded(stream_response(
                status,
                &headers,
                &upstream_headers,
                hold_budget(prefixed(Bytes::from(body), rest), permit),
            ))),
        },
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hyper::header::HeaderValue;

    fn cfg() -> ResponseBodyConfig {
        ResponseBodyConfig::default()
    }

    fn headers(pairs: &[(&str, &str)]) -> hyper::HeaderMap {
        let mut map = hyper::HeaderMap::new();
        for (name, value) in pairs {
            map.append(
                hyper::header::HeaderName::from_bytes(name.as_bytes()).unwrap(),
                HeaderValue::from_str(value).unwrap(),
            );
        }
        map
    }

    #[test]
    fn the_allowlist_arms_inspection_on_the_essence_only() {
        // Parameters and case are irrelevant to the media type; the allowlist is what the operator
        // asked to inspect, so anything outside it is scope, not a skip.
        assert!(matches!(
            classify_response(
                StatusCode::OK,
                &headers(&[("content-type", "Application/JSON; charset=utf-8")]),
                &cfg()
            ),
            Inspectability::Inspect
        ));
        assert!(matches!(
            classify_response(
                StatusCode::OK,
                &headers(&[("content-type", "image/png")]),
                &cfg()
            ),
            Inspectability::OutOfScope
        ));
        assert!(
            matches!(
                classify_response(StatusCode::OK, &headers(&[]), &cfg()),
                Inspectability::OutOfScope
            ),
            "no declared type, no declared scope"
        );
    }

    #[test]
    fn every_uninspectable_class_is_named_rather_than_skipped() {
        // ADR 000098 decision 3. A streaming media type is refused rather than merely unlisted:
        // buffering it would break the stream, and reading its absence from the allowlist as
        // "quietly skip" is exactly the silent skip this design refuses.
        assert!(matches!(
            classify_response(
                StatusCode::OK,
                &headers(&[("content-type", "text/event-stream")]),
                &cfg()
            ),
            Inspectability::Uninspectable(InspectionSkip::StreamingContentType)
        ));
        assert!(matches!(
            classify_response(
                StatusCode::OK,
                &headers(&[
                    ("content-type", "application/json"),
                    ("content-encoding", "gzip")
                ]),
                &cfg()
            ),
            Inspectability::Uninspectable(InspectionSkip::ContentEncoding)
        ));
        assert!(matches!(
            classify_response(
                StatusCode::PARTIAL_CONTENT,
                &headers(&[("content-type", "application/json")]),
                &cfg()
            ),
            Inspectability::Uninspectable(InspectionSkip::PartialContent)
        ));
        assert!(
            matches!(
                classify_response(
                    StatusCode::OK,
                    &headers(&[
                        ("content-type", "application/json"),
                        ("content-range", "bytes 0-99/1000")
                    ]),
                    &cfg()
                ),
                Inspectability::Uninspectable(InspectionSkip::PartialContent)
            ),
            "a fragment is a fragment whatever status carries it"
        );
        assert!(
            matches!(
                classify_response(
                    StatusCode::OK,
                    &headers(&[
                        ("content-type", "application/json"),
                        ("content-encoding", "identity")
                    ]),
                    &cfg()
                ),
                Inspectability::Inspect
            ),
            "`identity` is not an encoding to see through"
        );
    }

    #[test]
    fn a_response_inspecting_route_asks_the_upstream_for_a_whole_identity_representation() {
        let mut headers = vec![
            plecto_control::Header {
                name: "Accept-Encoding".to_string(),
                value: b"gzip, br".to_vec(),
            },
            plecto_control::Header {
                name: "range".to_string(),
                value: b"bytes=0-99".to_vec(),
            },
            plecto_control::Header {
                name: "If-Range".to_string(),
                value: b"\"v1\"".to_vec(),
            },
            plecto_control::Header {
                name: "x-keep".to_string(),
                value: b"1".to_vec(),
            },
        ];
        strip_inspection_hostile_headers(&mut headers);
        assert_eq!(headers.len(), 1);
        assert_eq!(headers[0].name, "x-keep");
    }
}