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

v0.34.0_2

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-17 23:37:43 +02:00
1 parent 0a8bc6f040
commit 434e789144
38 files changed
+899 -238
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/balance/lease_runtime.rs
+6 -2
View File
@@ -296,7 +296,7 @@ impl Pool {
text
}
}
async fn probe(member: Arc<Member>, settings: HealthCheck) {
async fn probe(member: Arc<Member>, settings: Arc<HealthCheck>) {
let start = Instant::now();
let result = tokio::time::timeout(Duration::from_millis(settings.timeout_ms), async {
match settings.mode {
@@ -388,8 +388,12 @@ pub async fn health_loop(app: Weak<crate::runtime::App>, shutdown: CancellationT
let snapshot = app.config.load_full();
// Own an Arc for each candidate. JoinSet tasks require an
// owned, 'static member rather than a reference into snapshot.
// Health settings were already compiled into each member.
// Reuse the immutable Arc instead of cloning each HealthCheck
// and recalculating per-member overrides on every 250ms tick.
let members: Vec<_> = snapshot.pools.iter().flat_map(|pool|
pool.members.iter().map(move |member| (Arc::clone(member), pool.settings.health_for(&member.definition)))).collect();
pool.members.iter().map(|member|
(Arc::clone(member), Arc::clone(&member.health_settings)))).collect();
for (member, _) in &members {
member.state.sample_traffic();
// Time-based pool recycling must not depend on receiving a
src/balance/pool_core.rs
+7 -3
View File
@@ -279,9 +279,13 @@ impl Pool {
let weight = member
.state
.weight(member.definition.weight, self.settings.slow_start_seconds);
let current = scheduling
.entry(member.definition.id.clone())
.or_insert(0.0);
let current = if let Some(current) =
scheduling.get_mut(member.definition.id.as_str())
{
current
} else {
scheduling.entry(member.definition.id.clone()).or_insert(0.0)
};
*current += weight;
total += weight;
if *current > best {
src/cache/dashboard.rs
Vendored
+14 -13
View File
@@ -6,9 +6,6 @@ impl Cache {
.unwrap_or_else(|e| e.into_inner())
.clone();
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let n = |value: &AtomicU64| value.load(Ordering::Relaxed);
let hits = n(&self.counters.hits);
let misses = n(&self.counters.misses);
@@ -180,6 +177,7 @@ impl Cache {
store.revision = revision;
self.revision.store(revision, Ordering::Relaxed);
store.domains.clear();
store.domain_count = 0;
}
drop(store);
self.cancel_all_fill_flights();
@@ -247,9 +245,6 @@ impl Cache {
}
pub fn scope_view(&self, scope: &str) -> ScopeView {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let mut view = ScopeView {
entries: 0,
resident_bytes: 0,
@@ -276,26 +271,32 @@ impl Cache {
"MISS" => c.misses += 1,
_ => c.bypasses += 1,
}
*c.reasons.entry(reason.into()).or_default() += 1;
if let Some(count) = c.reasons.get_mut(reason) {
*count = count.saturating_add(1);
} else {
c.reasons.insert(reason.to_owned(), 1);
}
}
}
pub fn record_reason(&self, scope: &str, domain: &str, reason: &'static str) {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
if let Some(c) = store.domain(scope, domain) {
*c.reasons.entry(reason.into()).or_default() += 1;
if let Some(count) = c.reasons.get_mut(reason) {
*count = count.saturating_add(1);
} else {
c.reasons.insert(reason.to_owned(), 1);
}
}
}
pub fn dashboard(&self, scope: Option<&str>, offset: usize, limit: usize) -> Dashboard {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let revision = store.revision;
let mut groups: BTreeMap<(String, String), DomainView> = store
.domains
.iter()
.filter(|((vhost, _), _)| scope.is_none_or(|s| s == vhost))
.map(|((vhost, domain), counters)| {
.filter(|(vhost, _)| scope.is_none_or(|s| s == vhost.as_str()))
.flat_map(|(vhost, domains)| domains.iter().map(move |(domain, counters)| (vhost, domain, counters)))
.map(|(vhost, domain, counters)| {
(
(vhost.clone(), domain.clone()),
DomainView {
src/cache/fill.rs
Vendored
+1 -1
View File
@@ -160,7 +160,7 @@ impl Candidate {
created_at,
body,
body_allocated: body_allocation,
headers: self.headers,
headers: Arc::new(self.headers),
received: self.plan.received,
last_access: Instant::now(),
resident_since: Instant::now(),
src/cache/invalidation/invalidate.rs
Vendored
+27 -11
View File
@@ -1,17 +1,24 @@
impl Cache {
fn active_bans(&self) -> Vec<BanRule> {
if self.ban_rule_count.load(Ordering::Acquire) == 0 {
return Vec::new();
}
let now = SystemTime::now();
let mut bans = self.bans.lock().unwrap_or_else(|e| e.into_inner());
bans.retain(|rule| !rule.expired(now));
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
bans.clone()
}
fn hot_entry_banned(&self, key: &Key, entry: &Entry) -> bool {
if self.ban_rule_count.load(Ordering::Acquire) == 0 {
return false;
}
let now = SystemTime::now();
let banned = self
.active_bans()
.iter()
.any(|rule| rule.matches_hot(key, entry, now));
let mut bans = self.bans.lock().unwrap_or_else(|e| e.into_inner());
bans.retain(|rule| !rule.expired(now));
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
let banned = bans.iter().any(|rule| rule.matches_hot(key, entry, now));
if banned {
self.counters.ban_matches.fetch_add(1, Ordering::Relaxed);
}
@@ -19,11 +26,14 @@ impl Cache {
}
fn cold_entry_banned(&self, key: &Key, entry: &ColdEntry) -> bool {
if self.ban_rule_count.load(Ordering::Acquire) == 0 {
return false;
}
let now = SystemTime::now();
let banned = self
.active_bans()
.iter()
.any(|rule| rule.matches_cold(key, entry, now));
let mut bans = self.bans.lock().unwrap_or_else(|e| e.into_inner());
bans.retain(|rule| !rule.expired(now));
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
let banned = bans.iter().any(|rule| rule.matches_cold(key, entry, now));
if banned {
self.counters.ban_matches.fetch_add(1, Ordering::Relaxed);
}
@@ -31,10 +41,14 @@ impl Cache {
}
fn entry_banned_before_insert(&self, key: &Key, entry: &Entry) -> bool {
if self.ban_rule_count.load(Ordering::Acquire) == 0 {
return false;
}
let now = SystemTime::now();
self.active_bans()
.iter()
.any(|rule| rule.matches_hot(key, entry, now))
let mut bans = self.bans.lock().unwrap_or_else(|e| e.into_inner());
bans.retain(|rule| !rule.expired(now));
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
bans.iter().any(|rule| rule.matches_hot(key, entry, now))
}
pub fn bans(&self) -> Vec<BanView> {
@@ -46,6 +60,7 @@ impl Cache {
let before = bans.len();
bans.retain(|rule| rule.id != id);
let removed = bans.len() != before;
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
if removed {
self.counters.bans_removed.fetch_add(1, Ordering::Relaxed);
}
@@ -82,6 +97,7 @@ impl Cache {
created_at: now,
expires_at: now.checked_add(Duration::from_secs(ttl)).unwrap_or(now),
});
self.ban_rule_count.store(bans.len() as u64, Ordering::Release);
self.counters.bans_created.fetch_add(1, Ordering::Relaxed);
Ok(InvalidationResult {
operation: request.operation,
src/cache/lifecycle.rs
Vendored
+2
View File
@@ -25,6 +25,7 @@ impl Cache {
fill_flight_sequence: AtomicU64::new(0),
collapse_waiters: AtomicU64::new(0),
bans: Mutex::new(Vec::new()),
ban_rule_count: AtomicU64::new(0),
ban_sequence: AtomicU64::new(0),
proactive_scheduled_keys: Mutex::new(HashSet::new()),
proactive_inflight: AtomicU64::new(0),
@@ -75,6 +76,7 @@ impl Cache {
fill_flight_sequence: AtomicU64::new(0),
collapse_waiters: AtomicU64::new(0),
bans: Mutex::new(Vec::new()),
ban_rule_count: AtomicU64::new(0),
ban_sequence: AtomicU64::new(0),
proactive_scheduled_keys: Mutex::new(HashSet::new()),
proactive_inflight: AtomicU64::new(0),
src/cache/lookup/hot.rs
Vendored
+49 -57
View File
@@ -1,29 +1,20 @@
impl Cache {
fn get_hot(&self, ticket: &Ticket, count_miss: bool) -> Option<Hit> {
// Only metadata, hit counters and exact LRU ordering require the store
// mutex. Materializing an HTTP response may allocate (headers, Body,
// Age formatting) and must be done *after* dropping that mutex.
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
let expired = store.sweep();
self.counters
.expired
.fetch_add(expired as u64, Ordering::Relaxed);
if store.revision != ticket.revision
|| store.generation != ticket.generation
|| store
.scope_generations
.get(&ticket.scope)
.copied()
.unwrap_or(0)
|| store.scope_generations.get(&ticket.scope).copied().unwrap_or(0)
!= ticket.scope_generation
|| store.key_generations.get(&ticket.key).copied().unwrap_or(0) != ticket.key_generation
|| store.key_generations.get(&ticket.key).copied().unwrap_or(0)
!= ticket.key_generation
{
return None;
}
if let Some(storage_key) = matching_hot_key(&store, ticket) {
// Don't remove and reinsert the entire Entry + variant index on
// every HOT hit. A cache hit changes only the LRU sequence, hits
// counter, and last-access clock; body/headers remain immutable.
let banned = store
.entries
.get(&storage_key)
let banned = store.entries.get(&storage_key)
.is_some_and(|entry| self.hot_entry_banned(&storage_key, entry));
if banned {
store.remove(&storage_key);
@@ -34,14 +25,9 @@ impl Cache {
release_allocator_memory();
return None;
}
let Some(entry) = store.entries.get(&storage_key) else {
return None;
};
let entry = store.entries.get(&storage_key)?;
let resident_age_seconds = hot_resident_age(entry).as_secs();
let age = entry
.initial_age
.saturating_add(entry.received.elapsed())
.as_secs();
let age = entry.initial_age.saturating_add(entry.received.elapsed()).as_secs();
let state = cache_state(
age,
entry.ttl,
@@ -53,48 +39,54 @@ impl Cache {
);
if state.servable() {
let sequence = store.next_sequence();
let (old_sequence, scope, domain, body_bytes, result) = {
let (old_sequence, headers, body, hits, transformed, skip_compression,
remaining_ttl, body_bytes) = {
let entry = store.entries.get_mut(&storage_key)?;
let old_sequence = entry.sequence;
entry.hits = entry.hits.saturating_add(1);
entry.last_access = Instant::now();
entry.sequence = sequence;
let mut response = Response::new(Body::from(entry.body.clone()));
*response.headers_mut() = entry.headers.clone();
if entry.transformed {
response
.extensions_mut()
.insert(crate::transform::CachedTransformed);
}
if entry.skip_compression {
response.extensions_mut().insert(crate::compression::Skip);
}
response.headers_mut().insert(
"age",
HeaderValue::from_str(&age.to_string()).expect("integer header"),
);
let result = Hit {
response,
body: entry.body.clone(),
hits: entry.hits,
transformed: entry.transformed,
remaining_ttl: entry.ttl.saturating_sub(age),
resident_age_seconds,
stale: state.stale(),
background_revalidate: state.background_revalidate(),
};
(old_sequence, entry.scope.clone(), entry.domain.clone(), entry.body.len(), result)
(
old_sequence,
Arc::clone(&entry.headers),
entry.body.clone(),
entry.hits,
entry.transformed,
entry.skip_compression,
entry.ttl.saturating_sub(age),
entry.body.len(),
)
};
store.lru.remove(&old_sequence);
store.lru.insert(sequence, storage_key);
// `store.bytes`, `store.logical_bytes` and the Vary schema index
// did not change. In particular, this path does not allocate a
// replacement variants Vec for every response.
if let Some(counter) = store.domain(&scope, &domain) {
counter.saved_body_bytes = counter.saved_body_bytes.saturating_add(body_bytes as u64);
}
// Preserve the exact LRU and domain-byte counters, without
// allocating a replacement variant index or String keys.
store.saved_body_hit(&ticket.scope, &ticket.domain, body_bytes);
self.counters.hits.fetch_add(1, Ordering::Relaxed);
return Some(result);
drop(store);
let mut response = Response::new(Body::from(body.clone()));
*response.headers_mut() = headers.as_ref().clone();
if transformed {
response.extensions_mut().insert(crate::transform::CachedTransformed);
}
if skip_compression {
response.extensions_mut().insert(crate::compression::Skip);
}
response.headers_mut().insert(
"age",
HeaderValue::from_str(&age.to_string()).expect("integer header"),
);
return Some(Hit {
response,
body,
hits,
transformed,
remaining_ttl,
resident_age_seconds,
stale: state.stale(),
background_revalidate: state.background_revalidate(),
});
}
}
if count_miss {
@@ -243,7 +235,7 @@ impl Cache {
target: entry.target.clone(),
tags: entry.tags.clone(),
created_at: entry.created_at,
headers: metadata.headers.clone(),
headers: Arc::new(metadata.headers.clone()),
body: body.clone(),
body_allocated,
received: Instant::now(),
src/cache/lookup/insert.rs
Vendored
-3
View File
@@ -51,9 +51,6 @@ impl Cache {
store.remove(&key);
}
store.remove(&storage_key);
self.counters
.expired
.fetch_add(store.sweep() as u64, Ordering::Relaxed);
let revision = store.revision;
let generation = store.generation;
let mut released_hot = false;
src/cache/lookup/tiered.rs
Vendored
+1 -1
View File
@@ -77,7 +77,7 @@ impl Cache {
target: ticket.target.clone(),
tags,
created_at: ticket.created_at,
headers: headers.clone(),
headers: Arc::new(headers.clone()),
body: cached.body.clone(),
body_allocated: cached.body_allocated,
received: plan.received,
src/cache/model/cache.rs
Vendored
+20 -32
View File
@@ -111,7 +111,7 @@ struct Entry {
target: String,
tags: Vec<String>,
created_at: SystemTime,
headers: HeaderMap,
headers: Arc<HeaderMap>,
body: Bytes,
/// Backing allocation retained by `body`; can exceed `body.len()`.
body_allocated: usize,
@@ -181,6 +181,9 @@ fn hot_entry_cost(
.sum::<usize>();
body_allocated
.saturating_add(std::mem::size_of::<Entry>())
// Arc<HeaderMap> owns the actual map on a separate allocation.
.saturating_add(std::mem::size_of::<HeaderMap>())
.saturating_add(2 * std::mem::size_of::<usize>())
.saturating_add(index_heap)
.saturating_add(header_heap)
.saturating_add(string_vec_heap(vary))
@@ -333,8 +336,8 @@ struct Store {
control_revision: u64,
scope_generations: HashMap<String, u64>,
key_generations: HashMap<Key, u64>,
domains: BTreeMap<(String, String), DomainCounters>,
last_sweep: Option<Instant>,
domains: BTreeMap<String, BTreeMap<String, DomainCounters>>,
domain_count: usize,
}
impl Store {
fn index_variant(&mut self, base_key: Key, key: Key) {
@@ -360,39 +363,24 @@ impl Store {
Some(entry)
}
fn domain(&mut self, scope: &str, domain: &str) -> Option<&mut DomainCounters> {
let key = (scope.to_owned(), domain.to_owned());
// Cardinality is bounded independently of resident cache capacity.
if self.domains.len() >= 4096 && !self.domains.contains_key(&key) {
// The common path is a borrowed lookup: no allocation of scope/domain
// strings for every cache hit or log-reason counter update.
if self.domains.get(scope).is_some_and(|d| d.contains_key(domain)) {
return self.domains.get_mut(scope)?.get_mut(domain);
}
if self.domain_count >= 4096 {
return None;
}
Some(self.domains.entry(key).or_default())
self.domain_count += 1;
Some(self.domains.entry(scope.to_owned()).or_default()
.entry(domain.to_owned()).or_default())
}
fn sweep(&mut self) -> usize {
if self
.last_sweep
.is_some_and(|v| v.elapsed() < Duration::from_secs(1))
{
return 0;
fn saved_body_hit(&mut self, scope: &str, domain: &str, bytes: usize) {
// Every admitted HOT insertion registered its domain. If the bounded
// stats table is full, skip accounting for an unregistered domain.
if let Some(c) = self.domains.get_mut(scope).and_then(|d| d.get_mut(domain)) {
c.saved_body_bytes = c.saved_body_bytes.saturating_add(bytes as u64);
}
self.last_sweep = Some(Instant::now());
let stale: Vec<_> = self
.entries
.iter()
.filter(|(_, e)| {
let age = e.initial_age.saturating_add(e.received.elapsed()).as_secs();
age >= retention_deadline(
e.ttl,
e.stale_while_revalidate,
e.stale_if_error,
e.origin_etag.is_some() || e.origin_last_modified.is_some(),
)
})
.map(|(k, _)| *k)
.collect();
for key in &stale {
self.remove(key);
}
stale.len()
}
fn next_sequence(&mut self) -> u64 {
// No practical wrapping, but a wrap must not corrupt LRU bookkeeping.
src/cache/model/maintenance.rs
Vendored
+4
View File
@@ -73,6 +73,7 @@ pub struct Cache {
fill_flight_sequence: AtomicU64,
collapse_waiters: AtomicU64,
bans: Mutex<Vec<BanRule>>,
ban_rule_count: AtomicU64,
ban_sequence: AtomicU64,
proactive_scheduled_keys: Mutex<HashSet<Key>>,
proactive_inflight: AtomicU64,
@@ -347,6 +348,9 @@ fn hot_resident_age(entry: &Entry) -> Duration {
}
fn matching_hot_key(store: &Store, ticket: &Ticket) -> Option<Key> {
if store.entries.get(&ticket.key).is_some_and(|entry| entry.vary.is_empty()) {
return Some(ticket.key);
}
store
.variants
.get(&ticket.key)?
src/cache/tests.rs
Vendored
+3
View File
@@ -19,3 +19,6 @@ include!("tests/cold_tier_tests.rs");
#[cfg(test)]
include!("tests/http_invalidation_tests.rs");
#[cfg(test)]
include!("tests/hot_hit_034_tests.rs");
src/cache/tests/dashboard_tests.rs
Vendored
-1
View File
@@ -84,7 +84,6 @@ mod dashboard_tests {
for entry in store.entries.values_mut() {
entry.received = Instant::now() - Duration::from_secs(2 * 60 * 60);
}
store.last_sweep = None;
}
let dashboard = cache.dashboard(None, 0, 50);
assert_eq!(dashboard.matching_objects, 1);
src/cache/tests/hot_hit_034_tests.rs
Vendored
+172
View File
@@ -0,0 +1,172 @@
mod hot_hit_034_tests {
use super::*;
use http_body_util::BodyExt;
fn settings() -> Settings {
Settings { enabled: true, ..Settings::default() }
}
async fn publish(
cache: &Arc<Cache>,
settings: &Settings,
path: &str,
body: &'static str,
transformed: bool,
) -> Ticket {
let ticket = cache.ticket_scoped(
34, "app", "http", "cache.example", "origin@127.0.0.1:8080",
path, &HeaderMap::new(),
).expect("cache ticket");
let mut response = Response::new(Body::from(body));
response.headers_mut().insert(
"cache-control", HeaderValue::from_static("public, max-age=31536000"),
);
response.headers_mut().insert(
"etag", HeaderValue::from_static("\"cached-representation\""),
);
if transformed {
response.extensions_mut().insert(crate::transform::Applied {
kind: crate::transform::Kind::Css,
original_bytes: body.len(),
transformed_bytes: body.len(),
});
}
let policy = plan(StatusCode::OK, response.headers(), settings, Duration::ZERO)
.expect("public static object cacheable");
cache.wrap(response, ticket.clone(), policy, settings)
.into_body().collect().await.unwrap();
ticket
}
#[tokio::test]
async fn hot_hit_headers_are_independent_and_lru_counters_remain_exact() {
let cache = Cache::new(34);
let settings = settings();
let ticket = publish(&cache, &settings, "/theme.css", "body{color:red}", true).await;
let initial = cache.view();
let before = cache.dashboard(Some("app"), 0, 10);
assert_eq!(before.matching_objects, 1);
assert!(before.objects[0].transformed);
for i in 1..=500 {
let hit = cache.get(&ticket).expect("HOT cache hit");
assert_eq!(hit.hits, i);
assert!(hit.transformed);
assert!(hit.response.extensions().get::<crate::transform::CachedTransformed>().is_some());
assert_eq!(hit.response.headers()["etag"], "\"cached-representation\"");
let mut independent = hit.response;
independent.headers_mut().insert("etag", HeaderValue::from_static("\"client-only\""));
if i == 500 {
assert_eq!(independent.into_body().collect().await.unwrap().to_bytes(), Bytes::from_static(b"body{color:red}"));
}
}
let after = cache.view();
assert_eq!(after.entries, 1);
assert_eq!(after.hot_allocated_bytes, initial.hot_allocated_bytes);
assert_eq!(after.hot_logical_bytes, initial.hot_logical_bytes);
assert_eq!(after.hits.saturating_sub(initial.hits), 500);
let store = cache.store.lock().unwrap();
assert_eq!(store.entries.len(), 1);
assert_eq!(store.lru.len(), 1);
assert_eq!(store.variants.len(), 1);
drop(store);
let dashboard = cache.dashboard(Some("app"), 0, 10);
assert_eq!(dashboard.objects[0].hits, 500);
assert_eq!(dashboard.domains[0].counters.saved_body_bytes, 500 * 15);
assert_eq!(cache.get(&ticket).unwrap().response.headers()["etag"], "\"cached-representation\"");
}
#[tokio::test]
async fn concurrent_hot_hits_have_unique_monotonic_counts_and_bounded_indexes() {
let cache = Cache::new(34);
let ticket = publish(&cache, &settings(), "/shared.css", "cachebody", false).await;
let workers = 6;
let per_worker = 300;
let mut threads = Vec::new();
for _ in 0..workers {
let cache = Arc::clone(&cache);
let ticket = ticket.clone();
threads.push(std::thread::spawn(move || {
let mut hits = Vec::with_capacity(per_worker);
for _ in 0..per_worker {
let hit = cache.get(&ticket).expect("concurrent HOT hit");
assert_eq!(hit.response.headers()["etag"], "\"cached-representation\"");
hits.push(hit.hits);
}
hits
}));
}
let mut counts: Vec<_> = threads.into_iter().flat_map(|t| t.join().unwrap()).collect();
counts.sort_unstable();
assert_eq!(counts, (1..=(workers * per_worker) as u64).collect::<Vec<_>>());
assert_eq!(cache.view().hits, (workers * per_worker) as u64);
let store = cache.store.lock().unwrap();
assert_eq!(store.lru.len(), 1);
assert_eq!(store.entries.len(), 1);
}
#[tokio::test]
async fn hot_lru_eviction_and_ban_are_preserved() {
let cache = Cache::new(34);
let settings = Settings { max_entries: 2, ..settings() };
let a = publish(&cache, &settings, "/a.css", "AAAA", false).await;
let b = publish(&cache, &settings, "/b.css", "BBBB", false).await;
assert!(cache.get(&a).is_some(), "A becomes most recently used");
let c = publish(&cache, &settings, "/c.css", "CCCC", false).await;
assert!(cache.get(&a).is_some());
assert!(cache.get(&b).is_none(), "B is least recently used");
assert!(cache.get(&c).is_some());
let ban = cache.invalidate(InvalidationRequest {
operation: InvalidationOperation::Ban,
selector: InvalidationSelector::Exact,
value: "/a.css".into(),
host: Some("cache.example".into()),
vhost: Some("app".into()),
ttl_seconds: Some(3600),
confirm_wide: false,
}).unwrap();
assert_eq!(cache.ban_rule_count.load(Ordering::Acquire), 1);
assert!(cache.get(&a).is_none(), "BAN must block cached A");
assert!(cache.remove_ban(ban.ban_id.unwrap()));
assert_eq!(cache.ban_rule_count.load(Ordering::Acquire), 0);
assert!(cache.get(&c).is_some(), "unrelated C must remain cached");
}
#[tokio::test]
#[ignore = "manual in-process timing only; NOT HTTP RPS"]
async fn manual_hot_lookup_microbenchmark() {
let cache = Cache::new(34);
let ticket = publish(&cache, &settings(), "/bench.css", "smallbody", true).await;
for _ in 0..1_000 {
std::hint::black_box(cache.get(&ticket).unwrap());
}
let iterations = 100_000;
let start = Instant::now();
for _ in 0..iterations {
std::hint::black_box(cache.get(&ticket).unwrap());
}
let duration = start.elapsed();
eprintln!("HOT lookup incl. response materialization: {:.1} ns/op ({} hits); NOT network RPS",
duration.as_secs_f64() * 1e9 / iterations as f64, iterations);
assert_eq!(cache.view().hits, iterations + 1_000);
}
#[tokio::test]
async fn stale_object_remains_available_for_explicit_max_stale() {
let cache = Cache::new(34);
let ticket = publish(&cache, &settings(), "/old.css", "oldbody", false).await;
{
let mut store = cache.store.lock().unwrap();
store.entries.values_mut().next().unwrap().received =
Instant::now() - Duration::from_secs(700_000);
}
assert!(cache.get(&ticket).is_none(), "expired freshness is not an unconditional HIT");
assert_eq!(cache.view().entries, 1, "stale representation must survive for revalidation");
let mut stale_ticket = ticket;
stale_ticket.request_headers.insert(
"cache-control", HeaderValue::from_static("max-stale"),
);
assert!(cache.get(&stale_ticket).unwrap().stale);
}
}
src/cache/tickets.rs
Vendored
+1 -5
View File
@@ -109,10 +109,6 @@ impl Cache {
}
pub fn revalidation(&self, ticket: &Ticket) -> Option<Revalidation> {
let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner());
let expired = store.sweep();
self.counters
.expired
.fetch_add(expired as u64, Ordering::Relaxed);
if store.revision != ticket.revision
|| store.generation != ticket.generation
|| store
@@ -142,7 +138,7 @@ impl Cache {
.saturating_add(entry.received.elapsed())
.as_secs();
Some(Revalidation {
headers: entry.headers.clone(),
headers: entry.headers.as_ref().clone(),
body: entry.body.clone(),
body_allocated: entry.body_allocated,
_transient_leases: Vec::new(),
src/config/backend.rs
+4 -1
View File
@@ -94,11 +94,14 @@ pub(crate) fn compile_backend_with_timeout(
connect_timeout,
)?
};
let request_origin = format!("{}://{request_authority}", url.scheme());
let request_origin_uri: http::Uri = request_origin.parse()?;
Ok(CompiledBackend {
address_policy,
id: backend.id.clone(),
origin: format!("{}://{authority}", url.scheme()),
request_origin: format!("{}://{request_authority}", url.scheme()),
request_origin,
request_origin_uri,
host_header: authority.parse()?,
preserve_host: backend.preserve_host,
protocol: backend.protocol,
src/config/model.rs
+2
View File
@@ -174,6 +174,8 @@ pub struct CompiledBackend {
pub origin: String,
/// HTTP request origin. For HTTPS this may carry a separate verified TLS name.
pub request_origin: String,
/// Parsed once at configuration publish; reused for every upstream request.
pub request_origin_uri: http::Uri,
pub host_header: HeaderValue,
pub preserve_host: bool,
pub protocol: UpstreamProtocol,
src/logging/format.rs
+21 -7
View File
@@ -136,8 +136,17 @@ impl Logs {
self.settings.load().capture_country
}
pub fn push(&self, entry: LogEntry) {
// String cloning and eviction cleanup do not need the history mutex.
let history_entry = entry.clone();
let has_viewers = self.live.receiver_count() != 0;
let has_sink = self.settings.load().enabled();
// The usual history-only configuration needs no deep copy at all.
// When external consumers are present, clone exactly for history and
// keep the original entry available for streaming/sink delivery.
let mut external = Some(entry);
let history_entry = if has_viewers || has_sink {
external.as_ref().expect("entry").clone()
} else {
external.take().expect("entry")
};
let evicted = if let Ok(mut entries) = self.entries.try_lock() {
let evicted = if entries.len() == HISTORY_CAPACITY {
entries.pop_front()
@@ -150,13 +159,18 @@ impl Logs {
self.dropped.fetch_add(1, Ordering::Relaxed);
None
};
// Destroy evicted strings outside of the history lock.
drop(evicted);
// No live viewers means no reason to allocate/clone a second LogEntry.
// History retention and external sinks remain unchanged.
if self.live.receiver_count() != 0 {
let _ = self.live.send(entry.clone());
if let Some(entry) = external {
if has_viewers {
let _ = self.live.send(entry.clone());
}
// Snapshot the sink choice once. This preserves sink drop
// accounting and avoids a second ArcSwap load per request.
if has_sink && self.sink.try_send(entry).is_err() {
self.sink_dropped.fetch_add(1, Ordering::Relaxed);
}
}
self.push_sink(entry);
}
fn push_sink(&self, entry: LogEntry) {
if self.settings.load().enabled() && self.sink.try_send(entry).is_err() {
src/logging/tests.rs
+36
View File
@@ -422,4 +422,40 @@ mod tests {
assert_eq!(stats.sent, 1);
assert_eq!(stats.errors, 0);
}
#[test]
fn history_only_mode_preserves_exact_fields_without_any_live_or_sink_copies() {
let logs = Logs::default();
let item = entry();
let expected = serde_json::to_value(&item).unwrap();
logs.push(item);
let saved = logs.recent();
assert_eq!(saved.len(), 1);
assert_eq!(serde_json::to_value(&saved[0]).unwrap(), expected);
assert_eq!(logs.stats().queue_dropped, 0);
}
#[test]
fn live_viewers_receive_same_event_as_persisted_history() {
let logs = Logs::default();
let mut live = logs.subscribe();
let item = entry();
let expected = serde_json::to_value(&item).unwrap();
logs.push(item);
assert_eq!(serde_json::to_value(live.try_recv().unwrap()).unwrap(), expected);
assert_eq!(serde_json::to_value(&logs.recent()[0]).unwrap(), expected);
}
#[test]
fn enabled_external_sink_retains_existing_queue_drop_accounting() {
// The default test Logs has a closed sink receiver. Delivering to a
// configured external sink must still account for its failure.
let logs = Logs::default();
let mut settings = Settings::default();
settings.file.enabled = true;
logs.configure(settings);
logs.push(entry());
assert_eq!(logs.recent().len(), 1);
assert_eq!(logs.stats().queue_dropped, 1);
}
}
src/proxy/body.rs
+14 -1
View File
@@ -301,6 +301,7 @@ pub struct MeteredBody {
/// For a response, backend selection is final before this wrapper is made.
/// Keep the selected lease alive and avoid locking Exchange.backend per frame.
response_backend: Option<Arc<crate::balance::Lease>>,
request_backend: Option<Arc<crate::balance::Lease>>,
direction: Direction,
timer: Option<Pin<Box<Sleep>>>,
deadline: tokio::time::Instant,
@@ -317,6 +318,7 @@ impl MeteredBody {
inner,
exchange: Some(exchange),
response_backend,
request_backend: None,
direction,
// Ready/empty bodies never need a timer allocation. Keep the
// original deadline so a delayed first poll still times out.
@@ -330,6 +332,7 @@ impl MeteredBody {
self.finished = true;
self.timer.take();
self.response_backend.take();
self.request_backend.take();
self.exchange.take();
}
}
@@ -350,7 +353,17 @@ impl HttpBody for MeteredBody {
if let (Some(data), Some(exchange)) = (frame.data_ref(), this.exchange.as_ref()) {
let count = data.len() as u64;
match this.direction {
Direction::Request => exchange.backend_bytes(count, true),
Direction::Request => {
// Backend is selected before a non-empty upstream
// upload is polled. No retry of a streamed write.
if this.request_backend.is_none() {
this.request_backend = exchange.backend
.lock().unwrap_or_else(|e| e.into_inner()).clone();
}
if let Some(lease) = &this.request_backend {
lease.bytes(count, true);
}
}
Direction::Response => {
if let Some(lease) = &this.response_backend {
lease.bytes(count, false);
src/proxy/forward/route_cache.rs
+3 -1
View File
@@ -125,7 +125,9 @@ let cache_settings = route
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.route.lock().unwrap_or_else(|e| e.into_inner()) = Some(route.id.clone());
// The resolved route was already recorded before local response, LB and
// ACL decisions. Do not acquire the route mutex and copy its ID twice on
// every reverse-proxy request.
exchange.set_cache_enabled(cache_settings.active_for(route.cache));
exchange.set_transform_enabled(transform_settings.is_some_and(|settings| settings.enabled));
}
src/proxy/request_id.rs
+4 -5
View File
@@ -1,11 +1,10 @@
fn request_id_hash(sequence: u64) -> String {
use rand::TryRng;
use sha2::{Digest, Sha256};
let mut nonce = [0u8; 32];
let mut rng = rand::rngs::SysRng;
rng.try_fill_bytes(&mut nonce)
.expect("secure operating system random source unavailable");
// Thread-local CSPRNG is seeded (and periodically reseeded) from the OS.
// Do not make a getrandom syscall on the hot path for every HTTP request.
// Keep hashing the per-process sequence as before for the same ID format.
let nonce: [u8; 32] = rand::random();
let mut hash = Sha256::new();
hash.update(sequence.to_be_bytes());
hash.update(nonce);
src/proxy/request_id/tests.rs
+1 -1
View File
@@ -25,7 +25,7 @@ mod request_id_tests {
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)));
assert!(
seen.insert(id),
"fixed sequence must still use independent OS entropy"
"fixed sequence must still use independent CSPRNG output"
);
}
}
src/proxy/tests/authority_tests.rs
+19
View File
@@ -595,4 +595,23 @@ mod graceful_h2_tests {
assert!(!is_graceful_h2_goaway(&protocol));
assert!(!is_graceful_h2_goaway(&local));
}
#[test]
fn compiled_request_origin_matches_previous_uri_parser() {
let mut config = crate::config::demo_config("127.0.0.1:9000".parse().unwrap());
config.vhosts[0].backends[0].url = "https://127.0.0.1:9000".into();
config.vhosts[0].backends[0].tls_server_name = Some("backend.example".into());
let snapshot = crate::config::compile(config, 1, &validation_policy()).unwrap();
let backend = &snapshot.pools[0].members[0].backend;
assert_eq!(backend.request_origin_uri.to_string(), backend.request_origin);
for suffix in [
"/", "/static/main.css", "/path?x=1&y=2", "/p%2Fq?url=https%3A//example.org/a",
"/?q=%C5%82&empty=", "/double//slash", "/item#frag",
] {
let previous = format!("{}{suffix}", backend.request_origin).parse::<Uri>().ok();
assert_eq!(upstream_uri(backend, suffix), previous, "{suffix}");
}
assert!(upstream_uri(backend, "http://attacker.invalid/").is_none());
assert_eq!(backend.origin, "https://127.0.0.1:9000");
}
}
src/proxy/types.rs
+7 -1
View File
@@ -152,7 +152,13 @@ fn retry_safe_method(method: &Method) -> bool {
}
fn upstream_uri(backend: &crate::config::CompiledBackend, suffix: &str) -> Option<Uri> {
format!("{}{suffix}", backend.request_origin).parse().ok()
// Reuse the validated scheme and authority from the immutable compiled
// backend. Only the untrusted path/query is parsed per request. This also
// prevents accidental authority changes from rewritten paths.
let path: http::uri::PathAndQuery = suffix.parse().ok()?;
let mut parts = backend.request_origin_uri.clone().into_parts();
parts.path_and_query = Some(path);
Uri::from_parts(parts).ok()
}
fn legacy_websocket_origin(
src/proxy/upstream_helpers.rs
+5 -8
View File
@@ -56,14 +56,11 @@ async fn observe(State(state): State<Arc<App>>, mut request: Request, next: Next
|| (request.version() == Version::HTTP_3
&& request.method() == Method::CONNECT
&& websocket::h3_protocol(&request) == Some("websocket"));
if request.version() == Version::HTTP_11 {
state.metrics.http1_requests.fetch_add(1, Ordering::Relaxed);
}
if request.version() == Version::HTTP_2 {
state.metrics.http2_requests.fetch_add(1, Ordering::Relaxed);
}
if request.version() == Version::HTTP_3 {
state.metrics.http3_requests.fetch_add(1, Ordering::Relaxed);
match request.version() {
Version::HTTP_11 => { state.metrics.http1_requests.fetch_add(1, Ordering::Relaxed); }
Version::HTTP_2 => { state.metrics.http2_requests.fetch_add(1, Ordering::Relaxed); }
Version::HTTP_3 => { state.metrics.http3_requests.fetch_add(1, Ordering::Relaxed); }
_ => {}
}
let id = state.metrics.requests.fetch_add(1, Ordering::Relaxed) + 1;
let request_id = request_id_hash(id);
src/upstream/client_tests.rs
+16
View File
@@ -96,5 +96,21 @@ mod tests {
}
assert_eq!(client.rotation_stats().rotations, 0);
}
#[test]
fn request_recycling_still_has_exact_threshold_after_clock_sampling() {
let backend = backend("requests-boundary", 18085);
let mut keepalive = crate::balance::Keepalive::default();
keepalive.pool_recycle_requests = 1_000;
keepalive.pool_recycle_seconds = 0;
let client = client_with_keepalive(
&backend, AddressPolicy::new(&backend).unwrap(), false, &keepalive,
).unwrap();
for _ in 0..1_000 { client.inner.client_for_request(); }
assert_eq!(client.rotation_stats().rotations, 0);
client.inner.client_for_request();
assert_eq!(client.rotation_stats().rotations, 1);
assert_eq!(client.rotation_stats().generation_requests, 1);
}
}