impl Cache { fn get_hot(&self, ticket: &Ticket, count_miss: bool) -> Option { 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) != ticket.scope_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) { let Some(mut entry) = store.remove(&storage_key) else { return None; }; if self.hot_entry_banned(&storage_key, &entry) { if count_miss { self.counters.misses.fetch_add(1, Ordering::Relaxed); } drop(store); release_allocator_memory(); return None; } let resident_age_seconds = hot_resident_age(&entry).as_secs(); let age = entry .initial_age .saturating_add(entry.received.elapsed()) .as_secs(); let state = cache_state( age, entry.ttl, entry.stale_while_revalidate, entry.must_revalidate, entry.proxy_revalidate, entry.requires_revalidation, &ticket.request_headers, ); if state.servable() { entry.hits = entry.hits.saturating_add(1); entry.last_access = Instant::now(); entry.sequence = store.next_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(), }; if let Some(c) = store.domain(&entry.scope, &entry.domain) { c.saved_body_bytes = c.saved_body_bytes.saturating_add(entry.body.len() as u64); } store.bytes = store.bytes.saturating_add(entry.cost); store.logical_bytes = store.logical_bytes.saturating_add(entry.body.len()); store.lru.insert(entry.sequence, storage_key); store.index_variant(entry.base_key, storage_key); store.entries.insert(storage_key, entry); self.counters.hits.fetch_add(1, Ordering::Relaxed); return Some(result); } store.bytes = store.bytes.saturating_add(entry.cost); store.logical_bytes = store.logical_bytes.saturating_add(entry.body.len()); store.lru.insert(entry.sequence, storage_key); store.index_variant(entry.base_key, storage_key); store.entries.insert(storage_key, entry); } if count_miss { self.counters.misses.fetch_add(1, Ordering::Relaxed); } None } pub fn get(&self, ticket: &Ticket) -> Option { self.get_hot(ticket, true) } async fn get_cold(&self, ticket: &Ticket, settings: &Settings) -> Option { if !self.ticket_current(ticket) { return None; } let banned = { let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner()); matching_cold_key(&cold, ticket).and_then(|storage_key| { let banned = cold .entries .get(&storage_key) .is_some_and(|entry| self.cold_entry_banned(&storage_key, entry)); banned.then(|| cold.remove(&storage_key)).flatten() }) }; if let Some(entry) = banned { self.delete_cold_entry(entry); return None; } let (storage_key, entry) = { let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner()); let storage_key = matching_cold_key(&cold, ticket)?; let old_sequence = { let entry = cold.entries.get(&storage_key)?; if entry.revision != ticket.revision || entry.scope != ticket.scope { return None; } let state = cache_state( cold_age(entry), entry.ttl, entry.stale_while_revalidate, entry.must_revalidate, entry.proxy_revalidate, entry.requires_revalidation, &ticket.request_headers, ); if !state.servable() { return None; } entry.sequence }; let sequence = cold.next_sequence(); cold.lru.remove(&old_sequence); cold.lru.insert(sequence, storage_key); let entry = cold.entries.get_mut(&storage_key)?; entry.sequence = sequence; entry.last_access = SystemTime::now(); entry.pinned_reads = entry.pinned_reads.saturating_add(1); (storage_key, entry.clone()) }; let _pin = ColdReadGuard { cache: self, key: storage_key, generation: entry.generation, }; if entry.body_bytes > settings.max_object_bytes { self.counters .memory_pressure_rejections .fetch_add(1, Ordering::Relaxed); return None; } let Some(_read_lease) = self.try_reserve_transient(entry.body_bytes, settings) else { return None; }; let _cold_read = ByteGaugeGuard::new(&self.cold_read_bytes_active, entry.body_bytes as u64); let body_vec = match tokio::fs::read(&entry.body_path).await { Ok(body) if cold_body_matches(&entry, &body) => body, _ => { self.counters.cold_errors.fetch_add(1, Ordering::Relaxed); if let Some(entry) = self.take_cold(&storage_key) { self.delete_cold_entry(entry); } return None; } }; let body_allocated = body_vec.capacity().max(body_vec.len()); let extra = body_allocated.saturating_sub(entry.body_bytes); let extra_lease = if extra > 0 { self.try_reserve_transient(extra, settings) } else { None }; if extra > 0 && extra_lease.is_none() { return None; } let _cold_read_extra = extra_lease .as_ref() .map(|_| ByteGaugeGuard::new(&self.cold_read_bytes_active, extra as u64)); let body = Bytes::from(body_vec); if !self.ticket_current(ticket) { return None; } let age = cold_age(&entry); let state = cache_state( age, entry.ttl, entry.stale_while_revalidate, entry.must_revalidate, entry.proxy_revalidate, entry.requires_revalidation, &ticket.request_headers, ); if !state.servable() { return None; } let hits = entry.hits.saturating_add(1); let cost = hot_entry_cost( body_allocated, &entry.headers, &entry.vary, &entry.scope, &entry.scheme, &entry.authority, &entry.domain, &entry.path, &entry.target, &entry.tags, entry.origin_etag.as_ref(), entry.origin_last_modified.as_ref(), ); let resident_age_seconds = cold_resident_age(&entry); let hot_entry = Entry { base_key: entry.base_key, vary: entry.vary.clone(), scope: entry.scope.clone(), scheme: entry.scheme.clone(), authority: entry.authority.clone(), domain: entry.domain.clone(), path: entry.path.clone(), target: entry.target.clone(), tags: entry.tags.clone(), created_at: entry.created_at, headers: entry.headers.clone(), body: body.clone(), body_allocated, received: Instant::now(), last_access: Instant::now(), resident_since: Instant::now(), resident_age_base: Duration::from_secs(resident_age_seconds), initial_age: Duration::from_secs(age), ttl: entry.ttl, stale_while_revalidate: entry.stale_while_revalidate, stale_if_error: entry.stale_if_error, must_revalidate: entry.must_revalidate, proxy_revalidate: entry.proxy_revalidate, origin_etag: entry.origin_etag.clone(), origin_last_modified: entry.origin_last_modified.clone(), cost, sequence: 0, hits, transformed: entry.transformed, skip_compression: entry.skip_compression, requires_revalidation: entry.requires_revalidation, }; if !self.insert_inner(ticket.clone(), hot_entry, settings, false) { return None; } let mut response = Response::new(Body::from(body.clone())); *response.headers_mut() = entry.headers; 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 mut store = self.store.lock().unwrap_or_else(|e| e.into_inner()); if let Some(c) = store.domain(&ticket.scope, &ticket.domain) { c.saved_body_bytes = c.saved_body_bytes.saturating_add(entry.body_bytes as u64); } } self.counters.hits.fetch_add(1, Ordering::Relaxed); self.counters.cold_hits.fetch_add(1, Ordering::Relaxed); self.counters .cold_promotions .fetch_add(1, Ordering::Relaxed); Some(Hit { response, body, hits, transformed: entry.transformed, remaining_ttl: entry.ttl.saturating_sub(age), resident_age_seconds, stale: state.stale(), background_revalidate: state.background_revalidate(), }) } pub async fn get_tiered(&self, ticket: &Ticket, settings: &Settings) -> Option { if let Some(hit) = self.get_hot(ticket, false) { return Some(hit); } if let Some(hit) = self.get_cold(ticket, settings).await { return Some(hit); } self.counters.misses.fetch_add(1, Ordering::Relaxed); None } pub fn publish_revalidated( &self, ticket: Ticket, cached: &Revalidation, mut headers: HeaderMap, validation_headers: &HeaderMap, settings: &Settings, response_delay: Duration, ) -> Option { let plan = plan(StatusCode::OK, &headers, settings, response_delay)?; if cached.body.len() > settings.max_object_bytes { return None; } if !headers.contains_key("date") { if let Ok(value) = httpdate::fmt_http_date(SystemTime::now()).parse() { headers.insert("date", value); } } for name in [ "x-request-id", "x-cache", "x-cache-hits", "x-cache-status", "x-cache-reason", "cache-status", ] { headers.remove(name); } let hits = cached.hits.saturating_add(1); let age = plan.initial_age.as_secs(); let remaining_ttl = plan.ttl.saturating_sub(age); let vary = plan.vary.clone(); let tags = cache_tags(&headers); let origin_etag = validation_headers .get("etag") .cloned() .or_else(|| cached.origin_etag.clone()); let origin_last_modified = validation_headers .get("last-modified") .cloned() .or_else(|| cached.origin_last_modified.clone()); let cost = hot_entry_cost( cached.body_allocated, &headers, &vary, &ticket.scope, &ticket.scheme, &ticket.authority, &ticket.domain, &ticket.path, &ticket.target, &tags, origin_etag.as_ref(), origin_last_modified.as_ref(), ); let storage_key = variant_key(&ticket.key, &vary, &ticket.request_headers); let resident_age = cached.resident_age_at_lookup.saturating_add(response_delay); let entry = Entry { base_key: ticket.key, vary, scope: ticket.scope.clone(), scheme: ticket.scheme.clone(), authority: ticket.authority.clone(), domain: ticket.domain.clone(), path: ticket.path.clone(), target: ticket.target.clone(), tags, created_at: ticket.created_at, headers: headers.clone(), body: cached.body.clone(), body_allocated: cached.body_allocated, received: plan.received, last_access: Instant::now(), resident_since: Instant::now(), resident_age_base: resident_age, initial_age: plan.initial_age, ttl: plan.ttl, stale_while_revalidate: plan.stale_while_revalidate, stale_if_error: plan.stale_if_error, must_revalidate: plan.must_revalidate, proxy_revalidate: plan.proxy_revalidate, origin_etag, origin_last_modified, cost, sequence: 0, hits, transformed: cached.transformed, skip_compression: cached.skip_compression, requires_revalidation: plan.requires_revalidation, }; if !self.insert(ticket, entry, settings) { return None; } if cached.storage_key != storage_key { self.discard_storage_key(&cached.storage_key); } self.counters.hits.fetch_add(1, Ordering::Relaxed); let mut response = cached.response(headers); response.headers_mut().insert( "age", HeaderValue::from_str(&age.to_string()).expect("integer header"), ); Some(Hit { response, body: cached.body.clone(), hits, transformed: cached.transformed, remaining_ttl, resident_age_seconds: resident_age.as_secs(), stale: false, background_revalidate: false, }) } pub fn wrap( self: &Arc, response: Response, ticket: Ticket, plan: Plan, settings: &Settings, ) -> Response { self.wrap_with_fill(response, ticket, plan, settings, None) } pub fn wrap_with_fill( self: &Arc, response: Response, ticket: Ticket, plan: Plan, settings: &Settings, fill_leader: Option, ) -> Response { let fill_size = plan.expected_length.unwrap_or(settings.max_object_bytes); let permits = if fill_size > 64 * 1024 * 1024 { FILL_SLOTS as u32 } else if fill_size > 16 * 1024 * 1024 { 4 } else if fill_size > 4 * 1024 * 1024 { 2 } else { 1 }; let permit = self.fills.clone().try_acquire_many_owned(permits); let Ok(permit) = permit else { self.counters.rejected_fills.fetch_add(1, Ordering::Relaxed); return response; }; let (parts, body) = response.into_parts(); let headers = parts.headers.clone(); let origin_validators = parts .extensions .get::() .cloned(); let origin_etag = match origin_validators.as_ref() { Some(validators) => validators.etag.clone(), None => headers.get("etag").cloned(), }; let origin_last_modified = match origin_validators.as_ref() { Some(validators) => validators.last_modified.clone(), None => headers.get("last-modified").cloned(), }; let initial_capacity = fill_size.min(64 * 1024); let Some(mut fill_gauge) = FillGauge::new(self.clone(), initial_capacity, settings) else { self.counters.rejected_fills.fetch_add(1, Ordering::Relaxed); self.record_reason(&ticket.scope, &ticket.domain, "hot_memory_pressure"); return Response::from_parts(parts, body); }; let data = Vec::with_capacity(initial_capacity); let actual_capacity = data.capacity(); if actual_capacity > initial_capacity && !fill_gauge.grow(actual_capacity - initial_capacity, settings) { self.counters.rejected_fills.fetch_add(1, Ordering::Relaxed); self.record_reason(&ticket.scope, &ticket.domain, "fill_allocator_overshoot"); return Response::from_parts(parts, body); } let candidate = Candidate { cache: self.clone(), ticket, plan, headers, data, fill_gauge, settings: settings.clone(), skip_compression: parts.extensions.get::().is_some(), origin_etag, origin_last_modified, transformed: parts.extensions.get::().is_some(), _permit: permit, _fill_leader: fill_leader, }; Response::from_parts( parts, Body::new(FillBody { inner: body, candidate: Some(candidate), done: false, }), ) } fn insert(&self, ticket: Ticket, entry: Entry, settings: &Settings) -> bool { self.insert_inner(ticket, entry, settings, true) } fn insert_inner( &self, ticket: Ticket, mut entry: Entry, settings: &Settings, count_stored: bool, ) -> bool { entry.base_key = ticket.key; let storage_key = variant_key(&ticket.key, &entry.vary, &ticket.request_headers); let vary_schema = entry.vary.clone(); if self.entry_banned_before_insert(&storage_key, &entry) { self.counters.ban_matches.fetch_add(1, Ordering::Relaxed); return false; } if entry.cost > settings.max_bytes || (!entry.requires_revalidation && entry.initial_age.saturating_add(entry.received.elapsed()).as_secs() >= retention_deadline( entry.ttl, entry.stale_while_revalidate, entry.stale_if_error, entry.origin_etag.is_some() || entry.origin_last_modified.is_some(), )) { return false; } let mut store = self.store.lock().unwrap_or_else(|e| e.into_inner()); if store.revision != ticket.revision || store.generation != ticket.generation || 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 { return false; } // Do not keep objects for incompatible Vary schemas under the same // Varnish-style host+URL hash. A newer representation policy replaces // the older schema while preserving sibling values of the same Vary set. for key in incompatible_hot_variants(&store, &ticket.key, &vary_schema) { 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; let mut removed = 0usize; while store.entries.len() >= settings.max_entries || store.bytes.saturating_add(entry.cost) > settings.max_bytes { if removed >= settings.hot_eviction_max_objects { self.counters .hot_eviction_limited .fetch_add(1, Ordering::Relaxed); break; } let Some((_, key)) = store.lru.first_key_value().map(|(seq, key)| (*seq, *key)) else { break; }; if let Some(old) = store.remove(&key) { removed = removed.saturating_add(1); released_hot = true; let scope_generation = store .scope_generations .get(&old.scope) .copied() .unwrap_or(0); let key_generation = store.key_generations.get(&old.base_key).copied().unwrap_or(0); if !(settings.cold_enabled && settings.cold_on_lru && self.enqueue_demotion(key, old, revision, generation, scope_generation, key_generation)) { self.counters.evictions.fetch_add(1, Ordering::Relaxed); } } } if store.entries.len() >= settings.max_entries || store.bytes.saturating_add(entry.cost) > settings.max_bytes { self.counters .memory_pressure_rejections .fetch_add(1, Ordering::Relaxed); drop(store); if released_hot { release_allocator_memory(); } return false; } if count_stored { if let Some(c) = store.domain(&entry.scope, &entry.domain) { c.stored += 1; } } entry.last_access = Instant::now(); entry.sequence = store.next_sequence(); store.bytes = store.bytes.saturating_add(entry.cost); store.logical_bytes = store.logical_bytes.saturating_add(entry.body.len()); store.lru.insert(entry.sequence, storage_key); store.index_variant(entry.base_key, storage_key); store.entries.insert(storage_key, entry); drop(store); if released_hot { release_allocator_memory(); } let mut cold_delete = Vec::new(); { let mut cold = self.cold.lock().unwrap_or_else(|e| e.into_inner()); for key in incompatible_cold_variants(&cold, &ticket.key, &vary_schema) { if let Some(old) = cold.remove(&key) { cold_delete.push(old); } } if let Some(old) = cold.remove(&storage_key) { cold_delete.push(old); } } for old in cold_delete { self.delete_cold_entry(old); } if count_stored { self.counters.stored.fetch_add(1, Ordering::Relaxed); } true } }