gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.29.4

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-09 13:11:33 +02:00
1 parent f87eec7287
commit dd8c195600
38 files changed
+5491 -4682
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/proxy/forward/route_cache.rs
+1 -54
View File
@@ -48,60 +48,7 @@ if route.unavailable {
return response;
}
if let crate::config::CompiledSource::WebServer(service) = &route.source {
if handshake.is_some() {
return error(
StatusCode::BAD_REQUEST,
"RustProxy: Web Server source does not support WebSocket upgrades\n",
);
}
let service_path = match route.rewritten_path(&path) {
Ok(Some(path)) => path,
Ok(None) => path.clone(),
Err(_) => {
return error(
StatusCode::BAD_REQUEST,
"RustProxy: invalid Web Server rewrite path\n",
)
}
};
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
*exchange.route.lock().unwrap_or_else(|e| e.into_inner()) = Some(route.id.clone());
*exchange.cache_effective_enabled.lock().unwrap_or_else(|e| e.into_inner()) = Some(false);
*exchange.transform_effective_enabled.lock().unwrap_or_else(|e| e.into_inner()) = Some(false);
}
// Do not hold a shared reference to Request<Body> across await. Axum handlers
// must return Send futures, while Body is not Sync. Clone only static-request
// metadata needed by the filesystem source before entering the async server.
let static_method = request.method().clone();
let static_headers = request.headers().clone();
let static_uri = request.uri().clone();
let static_version = request.version();
let static_body = std::mem::replace(request.body_mut(), Body::empty());
let mut response = crate::web_server::serve(
service,
static_method,
static_headers,
static_uri,
&service_path,
static_body,
crate::web_server::RequestContext {
client_ip: context.client_ip,
client_port: Some(peer.0.port()),
server_name: host.clone(),
server_port: local.0.port(),
tls,
version: static_version,
},
)
.await;
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await {
return fallback;
}
response.extensions_mut().insert(StaticSourceResponse);
return response;
return include!("web_source.rs");
}
let (backend, backend_lease, affinity_cookie) = if let Some(pool) = &route.pool {
let selection = match &route.backend_id {
src/proxy/forward/web_source.rs
+219
View File
@@ -0,0 +1,219 @@
{
// Web Server is an origin just like a reverse-proxy backend. It must pass
// through the same vhost cache policy, transform stage and response rules.
if handshake.is_some() {
return error(StatusCode::BAD_REQUEST,
"RustProxy: Web Server source does not support WebSocket upgrades\n");
}
let service_path = match route.rewritten_path(&path) {
Ok(Some(value)) => value,
Ok(None) => path.clone(),
Err(_) => return error(StatusCode::BAD_REQUEST,
"RustProxy: invalid Web Server rewrite path\n"),
};
let cache_settings = route.frontend.as_ref()
.map(|frontend| &frontend.cache)
.unwrap_or(&snapshot.envelope.config.cache);
let transform_settings = route.frontend.as_ref().map(|frontend| &frontend.transform);
let scope = route.frontend.as_ref().map(|frontend| frontend.id.as_str()).unwrap_or("");
let enabled = cache_settings.active_for(route.cache)
&& frontend.is_none_or(|frontend| frontend.basic_auth.is_none() && frontend.tinyauth.is_none());
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
*exchange.cache_effective_enabled.lock().unwrap_or_else(|e| e.into_inner()) = Some(enabled);
*exchange.transform_effective_enabled.lock().unwrap_or_else(|e| e.into_inner()) =
Some(transform_settings.is_some_and(|settings| settings.enabled));
}
// The authority is already validated by effective_host. HTTP/2 and HTTP/3
// requests commonly have :authority but no Host field, which otherwise
// makes PHP-FPM omit HTTP_HOST, breaking WordPress.
request.headers_mut().insert(HOST, original_host.clone());
let request_method = request.method().clone();
let request_headers = request.headers().clone();
let request_uri = request.uri().clone();
let request_version = request.version();
let public_suffix = request_uri.path_and_query()
.map(|item| item.as_str()).unwrap_or("/").to_owned();
// H2/H3 may expose an unknown-length empty stream before its final frame.
// Reuse the bounded probe that preserves any nonempty streaming body.
let empty_body = if enabled && (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 { request.body().is_end_stream() };
let force_cache = enabled && route.frontend.as_ref().is_some_and(|frontend|
frontend.force_cache.matches(&path, &public_suffix));
let policy = if enabled {
if force_cache {
crate::cache::forced_request_policy(&request_method, &request_headers, empty_body)
} else {
crate::cache::request_policy(
&request_method, &request_headers, empty_body, &path,
request_uri.query().is_some(), cache_settings)
}
} else {
crate::cache::RequestPolicy::Bypass("cache_disabled")
};
// HEAD can use an existing GET representation but may not fill a cache
// entry with its empty downstream body.
let eligible = enabled && policy.bypass_reason().is_none();
let mut key_headers = request_headers.clone();
let ignore_cookie = !force_cache && cache_settings.ignore_cookie_for_path(&path);
if ignore_cookie { key_headers.remove("cookie"); }
let cache_ticket = if eligible {
state.cache.ticket_scoped(state.cache.revision(), scope, context.scheme,
original_host.to_str().unwrap_or(""), &route.id, &public_suffix, &key_headers)
} else { None };
let range = crate::cache::request_range(&request_method, &request_headers);
let if_range = request_headers.get("if-range").cloned();
let range_requested = !matches!(range, crate::cache::RangeSelection::None);
let needs_validation = policy.revalidation_reason().is_some();
if let Some(ticket) = &cache_ticket {
if !needs_validation {
if let Some(mut hit) = state.cache.get_tiered(ticket, cache_settings).await {
// Filesystem/PHP does not have the reverse-proxy background
// revalidator. Refresh stale representations synchronously.
if hit.stale {
// Fall through and fetch from the local origin.
} else {
let conditional = crate::cache::conditional_not_modified(
&request_headers, hit.response.headers());
if !conditional {
if let crate::cache::RangeSelection::Single(requested) = range {
hit = hit.apply_range(requested, if_range.as_ref());
}
}
let mut response = if conditional {
crate::cache::make_not_modified(hit.response)
} else { hit.response };
if request_method == Method::HEAD {
*response.body_mut() = Body::empty();
}
cache_settings.annotate_reason(response.headers_mut(), "HIT", hit.hits,
hit.remaining_ttl, "cache_hit");
cache_settings.annotate_transformed(response.headers_mut(), hit.transformed);
cache_settings.annotate_resident_age(response.headers_mut(), hit.resident_age_seconds);
state.cache.record(scope, &host, "HIT", "cache_hit");
if hit.transformed { state.transform_metrics.record_cache_hit(scope); }
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
exchange.cache_status.store(1, Ordering::Relaxed);
*exchange.cache_reason.lock().unwrap_or_else(|e| e.into_inner()) = "cache_hit";
}
response.extensions_mut().insert(ResponseRulesApplied);
response.extensions_mut().insert(StaticSourceResponse);
return response;
}
}
}
}
if cache_ticket.is_some() && crate::cache::only_if_cached(&request_headers) {
return error(StatusCode::GATEWAY_TIMEOUT,
"RustProxy: only-if-cached request has no usable Web Server representation\n");
}
// Neither the Request nor the streaming Body is held across a borrowed
// await: the filesystem/PHP source owns the body for its full lifetime.
let static_body = std::mem::replace(request.body_mut(), Body::empty());
let started = Instant::now();
let mut response = crate::web_server::serve(service, request_method.clone(),
request_headers.clone(), request_uri, &service_path, static_body,
crate::web_server::RequestContext {
client_ip: context.client_ip,
client_port: Some(peer.0.port()),
server_name: host.clone(),
server_port: local.0.port(),
tls,
version: request_version,
}).await;
if let Some(fallback) = execute_error_fallback(
state.clone(), peer, local, &snapshot, &host, tls, route,
response.status(), &fallback_request, &fallback_body,
).await { return fallback; }
let origin_headers = response.headers().clone();
let original_plan = crate::cache::plan(response.status(), &origin_headers,
cache_settings, started.elapsed());
// Do not allow a later response-header rewrite to override a private,
// explicitly uncacheable PHP/CGI origin response.
let explicit_origin_policy = origin_headers.contains_key("cache-control")
|| origin_headers.contains_key("expires")
|| origin_headers.contains_key("set-cookie")
|| origin_headers.contains_key("vary");
let origin_safe = !explicit_origin_policy || original_plan.is_some();
let no_transform = crate::transform::no_transform(&origin_headers);
let skip_compression = crate::compression::no_transform(&origin_headers);
if skip_compression || range_requested {
response.extensions_mut().insert(crate::compression::Skip);
}
response.extensions_mut().insert(crate::transform::OriginValidators {
etag: origin_headers.get("etag").cloned(),
last_modified: origin_headers.get("last-modified").cloned(),
});
apply_response_rules(response.headers_mut(), &snapshot, route.frontend.as_deref(),
Some(route), false);
response.extensions_mut().insert(ResponseRulesApplied);
response.extensions_mut().insert(StaticSourceResponse);
// Cache representation must reflect vhost response rules and transforms,
// not solely the Web Service's Cache-Control header.
let origin_eligible_plan = if origin_safe
&& !(ignore_cookie && crate::cache::varies_on(response.headers(), "cookie"))
{
crate::cache::plan(response.status(), response.headers(),
cache_settings, started.elapsed())
} else { None };
if request_method == Method::GET && !range_requested {
if let Some(settings) = transform_settings.filter(|settings| settings.enabled) {
let transform_start = Instant::now();
response = match crate::transform::apply(response, &public_suffix,
settings, no_transform).await {
Ok(value) => {
state.transform_metrics.record_response(scope, &value, transform_start.elapsed());
value
}
Err(cause) => {
state.transform_metrics.record_error(scope, transform_start.elapsed());
tracing::debug!(error = %cause, "Web Server transform failed");
return error(StatusCode::BAD_GATEWAY,
"RustProxy: Web Server response could not be transformed\n");
}
};
}
}
let fill_plan = if origin_eligible_plan.is_some() {
crate::cache::plan(response.status(), response.headers(),
cache_settings, started.elapsed())
} else { None };
let reason = if !enabled { "cache_disabled" }
else if let Some(reason) = policy.bypass_reason() { reason }
else if needs_validation { "request_revalidated" }
else if cache_ticket.is_some() && fill_plan.is_none() { "response_headers_policy" }
else if cache_ticket.is_some() { "cache_miss" }
else { "cache_unavailable" };
let fillable = request_method == Method::GET && !range_requested;
let outcome = if cache_ticket.is_some() { "MISS" } else { "BYPASS" };
if let (Some(ticket), Some(origin), Some(transformed)) = (
cache_ticket, origin_eligible_plan, fill_plan,
) {
if fillable {
let canonical = response.extensions().get::<crate::transform::CanonicalIdentity>().is_some();
let mut plan = if canonical { origin.without_vary("accept-encoding") }
else { origin };
if force_cache {
if request_headers.contains_key("cookie") { plan = plan.with_vary("cookie"); }
if request_headers.contains_key("authorization") { plan = plan.with_vary("authorization"); }
}
response = state.cache.wrap_with_fill(response, ticket,
plan.restrict(transformed), cache_settings, None);
}
}
if let Some(exchange) = request.extensions().get::<Arc<Exchange>>() {
exchange.cache_status.store(if outcome == "MISS" { 2 } else { 3 }, Ordering::Relaxed);
*exchange.cache_reason.lock().unwrap_or_else(|e| e.into_inner()) = reason;
}
state.cache.record(scope, &host, outcome, reason);
cache_settings.annotate_reason(response.headers_mut(), outcome, 0, 0, reason);
let transformed = response.extensions().get::<crate::transform::Applied>().is_some();
cache_settings.annotate_transformed(response.headers_mut(), transformed);
response
}