async fn forward(
    state: Arc<App>,
    peer: PeerAddr,
    local: LocalAddr,
    mut request: Request,
) -> Response {
    let cookies = crate::auth_provider::CookieSink::default();
    request.extensions_mut().insert(cookies.clone());
    let mut response = forward_request(state, peer, local, request).await;
    cookies.apply(&mut response);
    response
}

async fn forward_request(
    state: Arc<App>,
    peer: PeerAddr,
    local: LocalAddr,
    mut request: Request,
) -> Response {
    if request.method() == Method::CONNECT
        && !matches!(request.version(), Version::HTTP_2 | Version::HTTP_3)
    {
        return error(
            StatusCode::NOT_IMPLEMENTED,
            "RustProxy: CONNECT is not supported\n",
        );
    }
    let tls = request
        .extensions()
        .get::<Transport>()
        .copied()
        .unwrap_or_default()
        .tls;
    let (selected_context, original_host, host, decoded_path, host_already_validated) =
        match take_routing_identity(&mut request, tls) {
            Ok(identity) => identity,
            Err(response) => return response,
        };
    let handshake = if request.method() == Method::CONNECT {
        let protocol = match request.version() {
            Version::HTTP_2 => websocket::h2_protocol(&request),
            Version::HTTP_3 => websocket::h3_protocol(&request),
            _ => None,
        };
        match (request.version(), protocol) {
            (Version::HTTP_2, Some("websocket")) => match websocket::validate_h2(&request) {
                Ok(handshake) => Some(handshake),
                Err(_) => {
                    return error(
                        StatusCode::BAD_REQUEST,
                        "RustProxy: invalid HTTP/2 WebSocket Extended CONNECT\n",
                    )
                }
            },
            (Version::HTTP_3, Some("websocket")) => match websocket::validate_h3(&request) {
                Ok(handshake) => Some(handshake),
                Err(_) => {
                    return error(
                        StatusCode::BAD_REQUEST,
                        "RustProxy: invalid HTTP/3 WebSocket Extended CONNECT\n",
                    )
                }
            },
            (_, Some(_)) => {
                return error(
                    StatusCode::NOT_IMPLEMENTED,
                    "RustProxy: unsupported Extended CONNECT protocol\n",
                )
            }
            _ => {
                return error(
                    StatusCode::NOT_IMPLEMENTED,
                    "RustProxy: CONNECT is not supported\n",
                )
            }
        }
    } else if headers::requests_upgrade(request.headers()) {
        match websocket::validate(&request) {
            Ok(handshake) => Some(handshake),
            Err(_) => {
                return error(
                    StatusCode::BAD_REQUEST,
                    "RustProxy: invalid or unsupported WebSocket upgrade\n",
                )
            }
        }
    } else {
        None
    };
    let snapshot = request
        .extensions()
        .get::<Arc<crate::config::Snapshot>>()
        .cloned()
        .unwrap_or_else(|| state.config.load_full());
    let internal_dispatch = request.extensions().get::<InternalFallbackRoute>().cloned();
    let routing_uri = request.uri().clone();
    let path = if let Some(path) = decoded_path.as_deref() {
        std::borrow::Cow::Borrowed(path)
    } else {
        let Ok(path) = crate::acl::routing_path_view(routing_uri.path()) else {
            return error(
                StatusCode::BAD_REQUEST,
                "RustProxy: ambiguous request path\n",
            );
        };
        path
    };
    let frontend = if let Some(selection) = &selected_context {
        selection.frontend.as_ref()
    } else {
        snapshot.frontend(&host, local.0, tls)
    };
    if snapshot
        .global_security
        .content_type_policy
        .blocks(request.headers())
    {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: request Content-Type denied by global policy\n",
        );
    }
    if frontend.is_some_and(|item| item.content_type_policy.blocks(request.headers())) {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: request Content-Type denied by vhost policy\n",
        );
    }
    if snapshot
        .global_security
        .user_agent_blocking
        .blocks(request.headers())
        || frontend.is_some_and(|frontend| frontend.user_agent_blocking.blocks(request.headers()))
    {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: denied by User-Agent policy\n",
        );
    }
    let websocket_legacy_support =
        handshake.is_some() && frontend.is_some_and(|frontend| frontend.websocket_legacy_support);
    // The most specific host name wins: an exact vhost such as autodiscover.example.com must not be
    // shadowed by a wildcard redirect or 404 host (*.example.com) that happens to be evaluated first.
    let frontend_rank = frontend.and_then(|_| snapshot.frontend_rank(&host, local.0, tls));
    let redirect = snapshot
        .redirect_on(&host, &path, local.0, tls)
        .filter(|_| match (frontend_rank, snapshot.redirect_rank(&host, &path, local.0, tls)) {
            (Some(vhost), Some(redirect)) => redirect >= vhost,
            _ => true,
        });
    let local_host = snapshot
        .not_found_host_on(&host, local.0, tls)
        .filter(|_| match (frontend_rank, snapshot.not_found_rank(&host, local.0, tls)) {
            (Some(vhost), Some(local_host)) => local_host >= vhost,
            _ => true,
        });
    if request.version() == Version::HTTP_2
        && (frontend.is_some_and(|frontend| !frontend.http2_enabled)
            || redirect.is_some_and(|(redirect, _)| !redirect.http2_enabled)
            || local_host.is_some_and(|item| !item.http2_enabled))
    {
        return error(
            StatusCode::MISDIRECTED_REQUEST,
            "RustProxy: HTTP/2 is disabled for this vhost\n",
        );
    }
    if request.version() == Version::HTTP_3
        && (frontend.is_some_and(|frontend| !frontend.http3_enabled)
            || redirect.is_some_and(|(redirect, _)| !redirect.http3_enabled)
            || local_host.is_some_and(|item| !item.http3_enabled))
    {
        return error(
            StatusCode::MISDIRECTED_REQUEST,
            "RustProxy: HTTP/3 is disabled for this vhost\n",
        );
    }
    if request.version() == Version::HTTP_2
        && request.method() == Method::CONNECT
        && websocket::h2_protocol(&request) == Some("websocket")
        && frontend.is_none_or(|frontend| !frontend.websocket_rfc8441)
    {
        return error(
            StatusCode::MISDIRECTED_REQUEST,
            "RustProxy: RFC 8441 WebSocket transport is not enabled for this vhost\n",
        );
    }
    if request.version() == Version::HTTP_3
        && request.method() == Method::CONNECT
        && websocket::h3_protocol(&request) == Some("websocket")
        && frontend.is_none_or(|frontend| !frontend.websocket_rfc9220)
    {
        return error(
            StatusCode::MISDIRECTED_REQUEST,
            "RustProxy: RFC 9220 WebSocket transport is not enabled for this vhost\n",
        );
    }
    if let Some(response) =
        method_policy_denial(&snapshot.global_security.method_policy, request.method())
    {
        return response;
    }
    if !cache_http_invalidation_method(request.method()) {
        if let Some(response) = frontend
            .and_then(|frontend| method_policy_denial(&frontend.method_policy, request.method()))
        {
            return response;
        }
        // Redirects are evaluated after global and per-vhost country/IP/rate checks below.
    }
    let mut context = match snapshot.forwarding.context_with_peer(
        request.headers(),
        peer.0,
        local.0,
        tls,
        &original_host,
        request.uri(),
        frontend.is_some_and(|frontend| frontend.auth_provider.is_some()),
        request
            .extensions()
            .get::<Arc<crate::forwarding::PeerFormatting>>()
            .map(Arc::as_ref),
        host_already_validated,
    ) {
        Ok(context) => context,
        Err(_) => {
            return error(
                StatusCode::BAD_REQUEST,
                "RustProxy: invalid forwarding context\n",
            )
        }
    };
    let mut cloudflare_country = None;
    if frontend.is_some_and(|frontend| frontend.cloudflare_client_ip) {
        if let Some(client) = crate::forwarding::cloudflare_client_ip(request.headers(), peer.0) {
            if context.override_client_ip(client).is_err() {
                return error(
                    StatusCode::BAD_REQUEST,
                    "RustProxy: invalid Cloudflare client address\n",
                );
            }
            cloudflare_country = crate::forwarding::cloudflare_country(request.headers(), peer.0);
        }
    }
    if context.client_ip != peer.0.ip() {
        if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
            *exchange.client_ip.lock().unwrap_or_else(|e| e.into_inner()) =
                Some(context.client_ip.to_string());
        }
    }
    let capture_country = state.logs.capture_country();
    let mut country_resolved = false;
    let mut country: Option<String> = None;
    if capture_country
        || snapshot.global_security.country_access.needs_geoip()
        || snapshot.global_security.rate_limit.needs_geoip()
    {
        country = cloudflare_country
            .clone()
            .or_else(|| state.geoip.load().country(context.client_ip));
        country_resolved = true;
        if capture_country {
            if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
                *exchange.country.lock().unwrap_or_else(|e| e.into_inner()) = country.clone();
            }
        }
    }
    // Global rules are cumulative: an explicit IP allow cannot bypass a country deny/allowlist.
    if !snapshot
        .global_security
        .country_access
        .allows(country.as_deref())
        || !snapshot
            .global_security
            .rate_limit
            .country_allowed(country.as_deref())
        || snapshot
            .global_security
            .acl
            .as_ref()
            .is_some_and(|acl| !acl.allows(context.client_ip))
    {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: denied by global access policy\n",
        );
    }
    if internal_dispatch.is_none() {
        if let Err(limited) = state.rate_limiter.check_with_user_agent(
            snapshot.envelope.revision,
            "\u{1f}rustproxy-global-security",
            context.client_ip,
            country.as_deref(),
            request
                .headers()
                .get(::http::header::USER_AGENT)
                .and_then(|value| value.to_str().ok()),
            &snapshot.global_security.rate_limit,
        ) {
            let mut response = error(
                StatusCode::TOO_MANY_REQUESTS,
                "RustProxy: global request rate limit exceeded\n",
            );
            if let Ok(value) = HeaderValue::from_str(&limited.retry_after_seconds.to_string()) {
                response.headers_mut().insert(::http::header::RETRY_AFTER, value);
            }
            return response;
        }
    }
    if let Some(frontend) = frontend {
        if !country_resolved
            && (capture_country
                || frontend
                    .profile_policies
                    .iter()
                    .any(|profile| profile.countries.needs_geoip())
                || frontend.rate_limit.needs_geoip())
        {
            country = cloudflare_country
                .clone()
                .or_else(|| state.geoip.load().country(context.client_ip));
            country_resolved = true;
        }
        if capture_country {
            if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
                *exchange.country.lock().unwrap_or_else(|e| e.into_inner()) = country.clone();
            }
        }
        if !crate::acl::allows_profiles_with_country(
            &frontend.profile_policies,
            context.client_ip,
            country.as_deref(),
        ) || !crate::acl::allows_with_country(
            frontend.acl.as_ref(),
            &frontend.rate_limit.country_access,
            context.client_ip,
            country.as_deref(),
        ) {
            return error(
                StatusCode::FORBIDDEN,
                "RustProxy: denied by access policy\n",
            );
        }
        if internal_dispatch.is_none() {
            if let Err(limited) = state.rate_limiter.check_with_user_agent(
                snapshot.envelope.revision,
                &frontend.id,
                context.client_ip,
                country.as_deref(),
                request
                    .headers()
                    .get(::http::header::USER_AGENT)
                    .and_then(|value| value.to_str().ok()),
                &frontend.rate_limit,
            ) {
                let mut response = error(
                    StatusCode::TOO_MANY_REQUESTS,
                    "RustProxy: request rate limit exceeded\n",
                );
                if let Ok(value) = HeaderValue::from_str(&limited.retry_after_seconds.to_string()) {
                    response.headers_mut().insert(::http::header::RETRY_AFTER, value);
                }
                return response;
            }
        }
    }
    if !snapshot.host_allowed(&host, context.client_ip) {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: denied by access policy\n",
        );
    }
    // Canonical / HTTPS redirects must not bypass global or per-vhost access policy.
    if !cache_http_invalidation_method(request.method()) {
        if let Some(mut response) = frontend.and_then(|frontend| {
            canonical_host_redirect(frontend, &host, tls, local.0.port(), request.uri())
        }) {
            snapshot
                .global_security
                .headers
                .apply(response.headers_mut(), tls);
            return response;
        }
        if !tls {
            if let Some(mut response) =
                frontend.and_then(|frontend| force_ssl_redirect(frontend, &host, request.uri()))
            {
                snapshot
                    .global_security
                    .headers
                    .apply(response.headers_mut(), tls);
                return response;
            }
        }
    }
    if cache_http_invalidation_method(request.method()) {
        return cache_http_invalidation(
            &state,
            &snapshot,
            frontend,
            &host,
            request,
            context.client_ip,
            peer.0.ip(),
            context.scheme,
        )
        .await;
    }
    if let Some((_redirect, rule)) = redirect {
        let authority = original_host.to_str().unwrap_or(&host);
        let Some(mut response) = rule.response(authority, tls, request.uri()) else {
            return error(
                StatusCode::INTERNAL_SERVER_ERROR,
                "RustProxy: redirect target could not be rendered\n",
            );
        };
        snapshot
            .global_security
            .headers
            .apply(response.headers_mut(), tls);
        response.extensions_mut().insert(RedirectResponse {
            http3_enabled: _redirect.http3_enabled,
        });
        return response;
    }
    if let Some(item) = local_host {
        let mut response = item.page.serve(request.method(), StatusCode::NOT_FOUND);
        snapshot
            .global_security
            .headers
            .apply(response.headers_mut(), tls);
        response.extensions_mut().insert(LocalHostResponse {
            http3_enabled: item.http3_enabled,
        });
        return response;
    }
    let selected = if let Some(dispatch) = &internal_dispatch {
        if frontend.is_none_or(|frontend| frontend.id != dispatch.vhost_id) {
            None
        } else {
            snapshot.select_internal_on(&host, &dispatch.vhost_id, &dispatch.target, local.0, tls)
        }
    } else if let Some(selection) = &selected_context {
        selection.route.as_deref()
    } else {
        snapshot.select_on(&host, &path, local.0, tls)
    };
    if let Some(route) = selected {
        if let Some(response) = method_policy_denial(&route.method_policy, request.method()) {
            return response;
        }
        if (route.profile_country_access.needs_geoip() || route.country_access.needs_geoip())
            && !country_resolved
        {
            country = cloudflare_country
                .clone()
                .or_else(|| state.geoip.load().country(context.client_ip));
            if capture_country {
                if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
                    *exchange.country.lock().unwrap_or_else(|e| e.into_inner()) = country.clone();
                }
            }
        }
        if !crate::acl::allows_with_country(
            route.profile_acl.as_ref(),
            &route.profile_country_access,
            context.client_ip,
            country.as_deref(),
        ) || !crate::acl::allows_with_country(
            route.acl.as_ref(),
            &route.country_access,
            context.client_ip,
            country.as_deref(),
        ) {
            return error(
                StatusCode::FORBIDDEN,
                "RustProxy: denied by location access policy\n",
            );
        }
    }
    let fallback_request = selected
        .filter(|route| route.error_fallback.is_some())
        .map(|_| ErrorFallbackRequest::capture(&request, &snapshot));
    let admin_cookie = headers::strip_admin_cookie(request.headers_mut());

    if let Some(auth) = frontend.and_then(|frontend| frontend.auth_provider.as_ref()) {
        if auth.reserved_path(&path) {
            return match auth.serve_reserved(request, &context, &state.shutdown) {
                Some(response) => response.await,
                None => error(
                    StatusCode::BAD_GATEWAY,
                    "RustProxy: authentication route unavailable\n",
                ),
            };
        }
    }
    let basic_identity = if let Some(auth) = frontend.and_then(|f| f.basic_auth.as_ref()) {
        match auth
            .authorize(
                request.headers_mut(),
                tls,
                context.client_ip,
                state.basic_workers.clone(),
                &state.basic_limits,
            )
            .await
        {
            Ok(identity) => identity,
            Err(response) => return response,
        }
    } else {
        None
    };
    if handshake.is_some() && frontend.is_some_and(|f| !f.websockets) {
        return error(
            StatusCode::FORBIDDEN,
            "RustProxy: WebSocket access is disabled for this host\n",
        );
    }
    let identity = if let Some(auth) = frontend.and_then(|f| f.auth_provider.as_ref()) {
        auth.strip_identity(request.headers_mut());
        let cookies = request
            .extensions()
            .get::<crate::auth_provider::CookieSink>()
            .cloned()
            .unwrap_or_default();
        match auth
            .authorize(
                request.headers(),
                request.method(),
                &context,
                &state.shutdown,
                &cookies,
            )
            .await
        {
            Ok(identity) => Some(identity),
            Err(response) => return response,
        }
    } else {
        None
    };
    if crate::pages::literal_ip_authority(original_host.to_str().unwrap_or("")) {
        if let Some(page) = &snapshot.envelope.config.pages.direct_ip {
            return page.serve(request.method(), StatusCode::OK);
        }
    }
    if let Some(page) = frontend.and_then(|f| f.pages.response.as_ref()) {
        return page.serve(request.method(), StatusCode::OK);
    }
    let Some(route) = selected else {
        let page = match frontend {
            Some(frontend) => frontend.pages.fallback.as_ref(),
            None => snapshot.envelope.config.pages.unknown_host.as_ref(),
        };
        return page
            .map(|page| page.serve(request.method(), StatusCode::NOT_FOUND))
            .unwrap_or_else(|| {
                error(
                    StatusCode::NOT_FOUND,
                    "RustProxy: no route for this host/path\n",
                )
            });
    };
    let fallback_body =
        match prepare_error_fallback_body(&mut request, route.error_fallback.is_some()).await {
            Ok(body) => body,
            Err(response) => return response,
        };
    let (
        backend,
        backend_lease,
        affinity_cookie,
        admin_control_plane,
        cache_settings,
        transform_settings,
        scope,
        ignore_static_cookie,
        force_cache,
        force_vary_cookie,
        force_vary_authorization,
        only_cached,
        public_suffix,
        origin_suffix,
        uri,
        exchange,
        lease,
        stop,
        downstream,
        mut h2_downstream,
        range_selection,
        if_range,
        range_requested,
        effective_reason,
        revalidation_reason,
        cache_ticket,
        client_conditionals,
        cached_revalidation,
        freshness_revalidation_reason,
        pending_revalidation,
        mut miss_recorded,
    ) = include!("forward/route_cache.rs");

    let (serve_stale, mut fill_leader) = include!("forward/cache_lookup.rs");

    let (
        websocket_upstream_h2,
        revalidation_sent,
        mut h2_direct_session,
        request_body_empty,
        proactive_base_headers,
        upstream_started,
        active_backend,
        active_lease,
        active_affinity_cookie,
        active_uri,
        retries,
        upstream_result,
        final_attempt_elapsed,
    ) = include!("forward/upstream.rs");

    include!("forward/response.rs")
}