RustProxy/src/proxy/forward/route_cache.rs

{
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
    *exchange.route.lock().unwrap_or_else(|e| e.into_inner()) = Some(route.id.clone());
    exchange.set_source(if route.response.is_some() {
        ExchangeSource::LocalResponse
    } else if route.redirect.is_some() {
        ExchangeSource::LocalRedirect
    } else {
        match &route.source {
            crate::config::CompiledSource::WebServer(_) => ExchangeSource::WebServer,
            crate::config::CompiledSource::ReverseProxy => ExchangeSource::ReverseProxy,
        }
    });
}
if let Some(spec) = &route.response {
    return local_location_response(spec, request.method());
}
if let Some(spec) = &route.redirect {
    let location = match route.redirect_location(&path, request.uri()) {
        Ok(Some(location)) => location,
        _ => return error(StatusCode::INTERNAL_SERVER_ERROR, "RustProxy: invalid location redirect\n"),
    };
    let status = StatusCode::from_u16(spec.status).unwrap_or(StatusCode::FOUND);
    let mut response = status.into_response();
    response.headers_mut().insert(http::header::LOCATION, location);
    response.headers_mut().insert("cache-control", HeaderValue::from_static("no-store"));
    return response;
}
if route.unavailable {
    let mut response = error(
        StatusCode::SERVICE_UNAVAILABLE,
        "RustProxy: published Docker service is currently unavailable\n",
    );
    response
        .headers_mut()
        .insert("retry-after", HeaderValue::from_static("1"));
    if let Some(fallback) = execute_error_fallback(
        state.clone(), peer, local, &snapshot, &host, tls, route,
        response.status(), &fallback_request, &fallback_body,
    ).await {
        return fallback;
    }
    return response;
}
if let crate::config::CompiledSource::WebServer(service) = &route.source {
    return include!("web_source.rs");
}
let (backend, backend_lease, affinity_cookie) = if let Some(pool) = &route.pool {
    let selection = match &route.backend_id {
        Some(id) => pool.select_member_wait(id).await,
        None => pool
            .select_wait(context.client_ip, request.headers(), tls)
            .await,
    };
    let selection = match selection {
        Ok(selection) => selection,
        Err(crate::balance::SelectionFailure::Unavailable) => {
            let mut response = error(
                StatusCode::SERVICE_UNAVAILABLE,
                "RustProxy: no healthy backend in this vhost pool\n",
            );
            response
                .headers_mut()
                .insert("retry-after", HeaderValue::from_static("1"));
            if let Some(fallback) = execute_error_fallback(
                state.clone(), peer, local, &snapshot, &host, tls, route,
                response.status(), &fallback_request, &fallback_body,
            ).await {
                return fallback;
            }
            return response;
        }
        Err(crate::balance::SelectionFailure::Capacity) => {
            let mut response = error(
                StatusCode::SERVICE_UNAVAILABLE,
                "RustProxy: backend concurrency limit reached\n",
            );
            response
                .headers_mut()
                .insert("retry-after", HeaderValue::from_static("1"));
            if let Some(fallback) = execute_error_fallback(
                state.clone(), peer, local, &snapshot, &host, tls, route,
                response.status(), &fallback_request, &fallback_body,
            ).await {
                return fallback;
            }
            return response;
        }
    };
    (
        selection.lease.member.backend.clone(),
        Some(selection.lease),
        selection.cookie,
    )
} else {
    let Some(backend) = route.backend.clone() else {
        let response = error(
            StatusCode::INTERNAL_SERVER_ERROR,
            "RustProxy: Reverse Proxy route has no backend\n",
        );
        if let Some(fallback) = execute_error_fallback(
            state.clone(), peer, local, &snapshot, &host, tls, route,
            response.status(), &fallback_request, &fallback_body,
        ).await {
            return fallback;
        }
        return response;
    };
    (backend, None, None)
};
let admin_control_plane = admin_session_passthrough(&snapshot, &host, tls, &backend);
if admin_control_plane {
    headers::restore_admin_cookie(request.headers_mut(), &admin_cookie);
}
let cache_settings = route
    .frontend
    .as_ref()
    .map(|f| &f.cache)
    .unwrap_or(&snapshot.envelope.config.cache);
let transform_settings = route.frontend.as_ref().map(|f| &f.transform);
let scope = route.frontend.as_ref().map(|f| f.id.as_str()).unwrap_or("");
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
    exchange.set_cache_enabled(cache_settings.active_for(route.cache));
    exchange.set_transform_enabled(transform_settings.is_some_and(|settings| settings.enabled));
}
let policy_reason = if admin_control_plane {
    Some("admin_control_plane")
} else if frontend.is_some_and(|f| f.basic_auth.is_some()) {
    Some("basic_auth")
} else if frontend.is_some_and(|f| f.tinyauth.is_some()) {
    Some("tinyauth")
} else if route
    .pool
    .as_ref()
    .is_some_and(|pool| pool.cookie_affinity())
{
    Some("sticky_cookie")
} else if !cache_settings.enabled {
    Some("cache_disabled")
} else if route.cache == Some(false) {
    Some("route_disabled")
} else if handshake.is_some() {
    Some("request_upgrade")
} else {
    None
};
let cache_enabled = policy_reason.is_none();
let body_empty = if cache_enabled
    && handshake.is_none()
    && (request.method() == Method::GET || request.method() == Method::HEAD)
{
    match empty_for_cache(&mut request).await {
        Ok(empty) => empty,
        Err(_) => {
            return error(
                StatusCode::BAD_REQUEST,
                "RustProxy: invalid or incomplete request body\n",
            )
        }
    }
} else {
    cache_enabled && request.body().is_end_stream()
};
let public_suffix: std::borrow::Cow<'_, str> = std::borrow::Cow::Borrowed(
    routing_uri.path_and_query().map(|v| v.as_str()).unwrap_or("/")
);
let cache_plan = cache_plan::CachePlan::new(policy_reason, cache_settings, &path, || {
    cache_settings.active_for(route.cache) && route
        .frontend
        .as_ref()
        .is_some_and(|frontend| frontend.force_cache.matches(&path, &public_suffix))
});
let force_cache = cache_plan.force_cache();
let ignore_static_cookie = cache_plan.ignore_cookie();
let original_policy = cache_plan.request_policy(
    request.method(), request.headers(), body_empty, &path,
    request.uri().query().is_some(), cache_settings,
);

let only_cached =
    cache_settings.active_for(route.cache) && !cache_settings.varnish_compatible
        && crate::cache::only_if_cached(request.headers());
let origin_suffix = match route.rewritten_path(&path) {
    Ok(None) => public_suffix.clone(),
    Ok(Some(rewritten)) => {
        let Some(mut encoded) = encoded_origin_path(&rewritten) else {
            return error(StatusCode::BAD_REQUEST, "RustProxy: invalid rewritten request target\n");
        };
        if let Some(query) = request.uri().query() {
            encoded.push('?');
            encoded.push_str(query);
        }
        std::borrow::Cow::Owned(encoded)
    }
    Err(_) => return error(StatusCode::BAD_REQUEST, "RustProxy: invalid rewritten request target\n"),
};
let Some(uri) = upstream_uri(&backend, &origin_suffix) else {
    return error(
        StatusCode::BAD_REQUEST,
        "RustProxy: invalid request target\n",
    );
};
let exchange = request.extensions().get::<Arc<Exchange>>().cloned();
let lease = request.extensions().get::<Arc<ConnectionLease>>().cloned();
let stop = request.extensions().get::<ServerShutdown>().cloned();
if handshake.is_some() {
    if let Some(exchange) = &exchange {
        exchange.websocket_phase("upstream_handshake");
    }
}
let downstream = if handshake.as_ref().is_some_and(|handshake| handshake.h1()) {
    Some(hyper::upgrade::on(&mut request))
} else {
    None
};
let h2_downstream = if handshake.as_ref().is_some_and(|handshake| handshake.h2())
    && (!matches!(backend.protocol,
        crate::config::UpstreamProtocol::Http2 | crate::config::UpstreamProtocol::H2)
        || websocket_legacy_support)
{
    Some(std::mem::replace(request.body_mut(), Body::empty()))
} else {
    None
};
headers::strip_hop_by_hop(request.headers_mut());
request.headers_mut().insert(
    HOST,
    if websocket_legacy_support {
        backend.host_header.clone()
    } else if backend.preserve_host {
        original_host.clone()
    } else {
        backend.host_header.clone()
    },
);
crate::headers::apply(request.headers_mut(), &snapshot.headers.request);
if let Some(frontend) = &route.frontend {
    crate::headers::apply(request.headers_mut(), &frontend.headers.request);
}
crate::headers::apply(request.headers_mut(), &route.headers.request);
context.apply_upstream(
    request.headers_mut(),
    route
        .frontend
        .as_ref()
        .is_none_or(|frontend| frontend.forward_client_ip),
);
if let Some(auth) = frontend.and_then(|f| f.tinyauth.as_ref()) {
    auth.strip_identity(request.headers_mut());
}
if let Some(identity) = identity {
    identity.apply(request.headers_mut());
}
if websocket_legacy_support {
    if let Some(upstream_host) = request.headers().get(HOST).cloned() {
        if let Some(origin) = legacy_websocket_origin(&backend.request_origin, &upstream_host) {
            request.headers_mut().insert("origin", origin);
        }
    }
}
let effective_policy = cache_plan::combine_policy(original_policy, || {
    cache_plan.request_policy(
        request.method(), request.headers(), body_empty, &path,
        request.uri().query().is_some(), cache_settings,
    )
});
let force_vary_cookie = force_cache && request.headers().contains_key("cookie");
let force_vary_authorization = force_cache && request.headers().contains_key("authorization");
let effective_reason = effective_policy.bypass_reason();
let revalidation_reason = effective_policy.revalidation_reason();
let effective_cacheable = effective_reason.is_none();
let range_selection = cache_plan::range_selection(
    effective_cacheable, request.method(), request.headers(),
);
let if_range = if effective_cacheable {
    request.headers().get("if-range").cloned()
} else { None };
let range_requested = !matches!(range_selection, crate::cache::RangeSelection::None);

let cache_route = route.id.as_str();
let mut cache_key_headers = None;
if effective_cacheable && ignore_static_cookie && request.headers().contains_key("cookie") {
    let mut headers = request.headers().clone();
    headers.remove("cookie");
    cache_key_headers = Some(headers);
}
let key_headers = cache_key_headers
    .as_ref()
    .unwrap_or_else(|| request.headers());
let cache_ticket = if effective_cacheable {
    state.cache.ticket_scoped_for_config(
        snapshot.envelope.revision,
        state.cache.revision(),
        route.frontend.as_ref().map(|f| f.id.as_str()).unwrap_or(""),
        context.scheme,
        original_host.to_str().unwrap_or(""),
        cache_route,
        &public_suffix,
        key_headers,
    ).map(|ticket| ticket.with_varnish_compatible(cache_settings.varnish_compatible))
} else {
    None
};
let client_conditionals = if cache_ticket.is_some() {
    request.headers().clone()
} else { HeaderMap::new() };
let cached_revalidation = if let Some(ticket) = cache_ticket.as_ref() {
    state.cache.revalidation_tiered(ticket).await
} else {
    None
};
let freshness_headers = crate::cache::request_freshness_headers(
    &client_conditionals, cache_settings,
);
let freshness_revalidation_reason = cached_revalidation
    .as_ref()
    .and_then(|cached| cached.request_freshness_revalidation_reason(&freshness_headers));
let pending_revalidation = revalidation_reason.is_some()
    || freshness_revalidation_reason.is_some()
    || cached_revalidation
        .as_ref()
        .is_some_and(|cached| !cached.reusable_for_request(&freshness_headers));
let miss_recorded = false;
    (
        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, h2_downstream,
        range_selection, if_range, range_requested, effective_reason, revalidation_reason,
        cache_ticket, client_conditionals, cached_revalidation, freshness_revalidation_reason,
        pending_revalidation, miss_recorded,
    )
}