v0.31.8

src/cache/fill.rs

@@ -63,6 +63,10 @@ struct Candidate {
     plan: Plan,
     headers: HeaderMap,
     data: Vec<u8>,
+    // A transformed response with one uniquely owned Bytes frame can share its
+    // allocation with the downstream response, avoiding an extra Vec copy.
+    shared: Option<Bytes>,
+    shared_allocated: usize,
     fill_gauge: FillGauge,
     settings: Settings,
     skip_compression: bool,
@@ -77,7 +81,7 @@ impl Candidate {
         if self
             .plan
             .expected_length
-            .is_some_and(|length| self.data.len() != length)
+            .is_some_and(|length| self.shared.as_ref().map_or(self.data.len(), Bytes::len) != length)
         {
             let reason = if self.transformed {
                 "transform_length_mismatch"
@@ -112,8 +116,12 @@ impl Candidate {
         // Keep the actual backing capacity as the HOT charge. Vec::reserve_exact
         // avoids geometric multi-MiB overshoot while Bytes::from(Vec) transfers
         // the allocation without copying.
-        let body_allocation = self.data.capacity().max(self.data.len());
-        let body = Bytes::from(std::mem::take(&mut self.data));
+        let (body, body_allocation) = if let Some(body) = self.shared.take() {
+            (body, self.shared_allocated)
+        } else {
+            let allocated = self.data.capacity().max(self.data.len());
+            (Bytes::from(std::mem::take(&mut self.data)), allocated)
+        };
         let scope = self.ticket.scope.clone();
         let scheme = self.ticket.scheme.clone();
         let authority = self.ticket.authority.clone();
@@ -192,8 +200,51 @@ impl HttpBody for FillBody {
         let this = self.get_mut();
         match Pin::new(&mut this.inner).poll_frame(cx) {
             Poll::Ready(Some(Ok(frame))) => {
+                let mut frame = frame;
+                // Only the transformed single-frame case is eligible. Taking
+                // unique ownership lets Bytes reveal its true allocation
+                // capacity; we must never undercharge an arbitrary shared
+                // Bytes slice that could retain a much larger parent buffer.
+                if this.inner.is_end_stream() {
+                    if let Some(candidate) = &mut this.candidate {
+                        let single_frame = candidate.transformed
+                            && candidate.data.is_empty()
+                            && candidate.shared.is_none()
+                            && frame.data_ref().is_some_and(|data| {
+                                candidate.plan.expected_length == Some(data.len())
+                                    && data.len() <= candidate.settings.max_object_bytes
+                            });
+                        if single_frame {
+                            match frame.into_data() {
+                                Ok(bytes) => {
+                                    match bytes.try_into_mut() {
+                                        Ok(mutable) => {
+                                            let allocation = mutable.capacity();
+                                            let bytes = mutable.freeze();
+                                            // A very oversized backing Vec is cheaper to
+                                            // copy into a smaller HOT allocation instead.
+                                            let acceptable_spare = bytes.len().saturating_div(8).min(64 * 1024);
+                                            if allocation <= bytes.len().saturating_add(acceptable_spare)
+                                                && allocation <= candidate.settings.max_object_bytes
+                                                && candidate.fill_gauge.grow(allocation, &candidate.settings)
+                                            {
+                                                candidate.shared = Some(bytes.clone());
+                                                candidate.shared_allocated = allocation;
+                                            }
+                                            frame = Frame::data(bytes);
+                                        }
+                                        Err(bytes) => frame = Frame::data(bytes),
+                                    }
+                                }
+                                Err(original) => frame = original,
+                            }
+                        }
+                    }
+                }
                 if let Some(candidate) = &mut this.candidate {
-                    if let Some(data) = frame.data_ref() {
+                    if candidate.shared.is_some() {
+                        // The cache now owns the very same backing allocation.
+                    } else if let Some(data) = frame.data_ref() {
                         let needed = candidate.data.len().saturating_add(data.len());
                         if needed <= candidate.settings.max_object_bytes {
                             let before = candidate.data.capacity();

src/cache/lookup/tiered.rs

@@ -170,7 +170,10 @@ impl Cache {
             Some(validators) => validators.last_modified.clone(),
             None => headers.get("last-modified").cloned(),
         };
-        let initial_capacity = fill_size.min(64 * 1024);
+        let transformed = parts.extensions.get::<crate::transform::Applied>().is_some();
+        // A transformed, fully buffered body normally arrives as one owned
+        // frame. Do not allocate a second fill buffer before inspecting it.
+        let initial_capacity = if transformed { 0 } else { 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");
@@ -191,15 +194,14 @@ impl Cache {
             plan,
             headers,
             data,
+            shared: None,
+            shared_allocated: 0,
             fill_gauge,
             settings: settings.clone(),
             skip_compression: parts.extensions.get::<crate::compression::Skip>().is_some(),
             origin_etag,
             origin_last_modified,
-            transformed: parts
-                .extensions
-                .get::<crate::transform::Applied>()
-                .is_some(),
+            transformed,
             _permit: permit,
             _fill_leader: fill_leader,
         };

src/cache/tests/cold_tier_tests.rs

@@ -1,4 +1,746 @@
 mod cold_tier_tests {
-    include!("cold_tier_tests/part_01.rs");
-    include!("cold_tier_tests/part_02.rs");
+use super::*;
+use http_body_util::BodyExt;
+
+fn cold_settings() -> Settings {
+    Settings {
+        enabled: true,
+        max_entries: 1,
+        max_bytes: 256 * 1024,
+        max_object_bytes: 64 * 1024,
+        cold_enabled: true,
+        cold_path: "cache/cold-test".into(),
+        cold_max_bytes: 2 * 1024 * 1024,
+        cold_max_entries: 16,
+        cold_after_seconds: 300,
+        ..Settings::default()
+    }
+}
+
+async fn fill(cache: &Arc<Cache>, settings: &Settings, path: &str, body: &'static str) -> Ticket {
+    let ticket = cache
+        .ticket_scoped(
+            11,
+            "cold-test",
+            "http",
+            "cold.example",
+            "origin@127.0.0.1:8080",
+            path,
+            &HeaderMap::new(),
+        )
+        .unwrap();
+    let mut response = Response::new(Body::from(body));
+    response.headers_mut().insert(
+        "cache-control",
+        HeaderValue::from_static("public, max-age=120"),
+    );
+    response
+        .headers_mut()
+        .insert("etag", HeaderValue::from_static("\"cold-v1\""));
+    let plan = plan(StatusCode::OK, response.headers(), settings, Duration::ZERO).unwrap();
+    cache
+        .wrap(response, ticket.clone(), plan, settings)
+        .into_body()
+        .collect()
+        .await
+        .unwrap();
+    ticket
+}
+
+async fn wait_for_cold(cache: &Cache, expected: usize) {
+    for _ in 0..100 {
+        if cache.view().cold_entries >= expected {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("cold tier did not persist in time");
+}
+
+#[tokio::test]
+async fn selective_purge_removes_matching_cold_object() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = cold_settings();
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    let first = fill(&cache, &settings, "/one?v=1", "first-payload").await;
+    let _second = fill(&cache, &settings, "/two", "second-payload").await;
+    wait_for_cold(&cache, 1).await;
+    assert_eq!(cache.view().cold_entries, 1);
+    let result = cache
+        .invalidate(InvalidationRequest {
+            operation: InvalidationOperation::Purge,
+            selector: InvalidationSelector::Exact,
+            value: "/one?v=1".into(),
+            host: Some("cold.example".into()),
+            vhost: Some("cold-test".into()),
+            ttl_seconds: None,
+            confirm_wide: false,
+        })
+        .unwrap();
+    assert_eq!(result.affected_entries, 1);
+    assert_eq!(cache.view().cold_entries, 0);
+    assert!(cache.get_tiered(&first, &settings).await.is_none());
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn lru_demotion_persists_and_survives_restart() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = cold_settings();
+    settings.validate().unwrap();
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    let first = fill(&cache, &settings, "/one", "first-payload").await;
+    let _second = fill(&cache, &settings, "/two", "second-payload").await;
+    wait_for_cold(&cache, 1).await;
+    assert_eq!(cache.view().entries, 1);
+    assert_eq!(cache.view().cold_entries, 1);
+    let cold = cache.dashboard(Some("cold-test"), 0, 50);
+    assert!(cold
+        .objects
+        .iter()
+        .any(|row| row.path == "/one" && row.tier == "cold"));
+
+    shutdown.cancel();
+    drop(cache);
+
+    let restart_shutdown = CancellationToken::new();
+    let restarted = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        restart_shutdown.clone(),
+    );
+    assert_eq!(restarted.view().cold_entries, 1);
+    let hit = restarted
+        .get_tiered(&first, &settings)
+        .await
+        .expect("cold object should be readable after restart");
+    let payload = hit.response.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(payload, Bytes::from_static(b"first-payload"));
+    assert_eq!(restarted.view().cold_hits, 1);
+    assert_eq!(restarted.view().cold_promotions, 1);
+    restart_shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_restart_preserves_transformed_validators_stale_metadata_and_no_transform() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+
+    let transformed_ticket = cache
+        .ticket_scoped(
+            11,
+            "cold-test",
+            "http",
+            "cold.example",
+            "origin@127.0.0.1:8080",
+            "/transformed.json",
+            &HeaderMap::new(),
+        )
+        .unwrap();
+    let mut transformed = Response::new(Body::from(r#"{"a":1}"#));
+    transformed.headers_mut().insert(
+        "cache-control",
+        HeaderValue::from_static(
+            "public, max-age=120, stale-while-revalidate=30, stale-if-error=90",
+        ),
+    );
+    transformed.headers_mut().insert(
+        "etag",
+        HeaderValue::from_static("\"rustproxy-representation\""),
+    );
+    transformed
+        .headers_mut()
+        .insert("content-length", HeaderValue::from_static("7"));
+    transformed
+        .extensions_mut()
+        .insert(crate::transform::OriginValidators {
+            etag: Some(HeaderValue::from_static("\"origin-v1\"")),
+            last_modified: Some(HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT")),
+        });
+    transformed
+        .extensions_mut()
+        .insert(crate::transform::Applied {
+            kind: crate::transform::Kind::Json,
+            original_bytes: 9,
+            transformed_bytes: 7,
+        });
+    let transformed_plan = plan(
+        StatusCode::OK,
+        transformed.headers(),
+        &settings,
+        Duration::ZERO,
+    )
+    .unwrap();
+    cache
+        .wrap(
+            transformed,
+            transformed_ticket.clone(),
+            transformed_plan,
+            &settings,
+        )
+        .into_body()
+        .collect()
+        .await
+        .unwrap();
+
+    let protected_ticket = cache
+        .ticket_scoped(
+            11,
+            "cold-test",
+            "http",
+            "cold.example",
+            "origin@127.0.0.1:8080",
+            "/protected.css",
+            &HeaderMap::new(),
+        )
+        .unwrap();
+    let original_body = "body { color: red; }";
+    let mut protected = Response::new(Body::from(original_body));
+    protected.headers_mut().insert(
+        "cache-control",
+        HeaderValue::from_static("public, max-age=120, immutable, no-transform"),
+    );
+    protected
+        .headers_mut()
+        .insert("etag", HeaderValue::from_static("\"origin-css\""));
+    protected.extensions_mut().insert(crate::compression::Skip);
+    let protected_plan = plan(
+        StatusCode::OK,
+        protected.headers(),
+        &settings,
+        Duration::ZERO,
+    )
+    .unwrap();
+    cache
+        .wrap(
+            protected,
+            protected_ticket.clone(),
+            protected_plan,
+            &settings,
+        )
+        .into_body()
+        .collect()
+        .await
+        .unwrap();
+
+    assert_eq!(cache.flush_hot_to_cold().await, 2);
+    wait_for_cold(&cache, 2).await;
+    shutdown.cancel();
+    drop(cache);
+
+    let restart_shutdown = CancellationToken::new();
+    let restarted = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        restart_shutdown.clone(),
+    );
+    assert_eq!(restarted.view().cold_entries, 2);
+    // COLD entries intentionally keep validators and headers on disk to bound RAM.
+    // Inspect them through the same verified metadata loader used on COLD hits.
+    let (key, transformed) = {
+        let cold = restarted.cold.lock().unwrap();
+        cold.entries
+            .iter()
+            .find(|(_, entry)| entry.path == "/transformed.json")
+            .map(|(key, entry)| (*key, entry.clone()))
+            .expect("transformed cold entry")
+    };
+    assert_eq!(transformed.stale_while_revalidate, 30);
+    assert_eq!(transformed.stale_if_error, 90);
+    assert!(transformed.has_validator);
+    assert!(transformed.transformed);
+    let metadata = read_cold_metadata(&key, &transformed)
+        .await
+        .expect("persisted transformed metadata");
+    assert_eq!(metadata.origin_etag.as_ref().unwrap(), "\"origin-v1\"");
+    assert_eq!(
+        metadata.origin_last_modified.as_ref().unwrap(),
+        "Mon, 26 Jan 2026 23:11:40 GMT"
+    );
+    assert_eq!(metadata.headers["etag"], "\"rustproxy-representation\"");
+
+    let transformed_hit = restarted
+        .get_tiered(&transformed_ticket, &settings)
+        .await
+        .expect("transformed object after restart");
+    assert_eq!(
+        transformed_hit.response.headers()["etag"],
+        "\"rustproxy-representation\""
+    );
+    assert!(transformed_hit.transformed);
+    assert!(transformed_hit
+        .response
+        .extensions()
+        .get::<crate::transform::CachedTransformed>()
+        .is_some());
+    assert_eq!(
+        transformed_hit
+            .response
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes(),
+        Bytes::from_static(br#"{"a":1}"#)
+    );
+
+    let protected_hit = restarted
+        .get_tiered(&protected_ticket, &settings)
+        .await
+        .expect("no-transform object after restart");
+    assert_eq!(protected_hit.response.headers()["etag"], "\"origin-css\"");
+    assert!(crate::transform::no_transform(
+        protected_hit.response.headers()
+    ));
+    assert!(protected_hit
+        .response
+        .extensions()
+        .get::<crate::compression::Skip>()
+        .is_some());
+    assert_eq!(
+        protected_hit
+            .response
+            .into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes(),
+        Bytes::from_static(b"body { color: red; }")
+    );
+    restart_shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_policy_can_disable_lru_demotion() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.cold_on_lru = false;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    fill(&cache, &settings, "/two", "second-payload").await;
+    tokio::time::sleep(Duration::from_millis(30)).await;
+    let view = cache.view();
+    assert_eq!(view.entries, 1);
+    assert_eq!(view.cold_entries, 0);
+    assert_eq!(view.evictions, 1);
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_policy_can_disable_idle_demotion() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    settings.cold_on_idle = false;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/idle", "idle-payload").await;
+    {
+        let mut store = cache.store.lock().unwrap();
+        for entry in store.entries.values_mut() {
+            entry.last_access = Instant::now() - Duration::from_secs(600);
+        }
+    }
+    cache.demote_idle();
+    assert_eq!(cache.view().entries, 1);
+    assert_eq!(cache.view().cold_entries, 0);
+
+    settings.cold_on_idle = true;
+    cache.configure(settings.clone());
+    cache.demote_idle();
+    wait_for_cold(&cache, 1).await;
+    assert_eq!(cache.view().entries, 0);
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_policy_can_demote_by_total_hot_residency_age() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    settings.cold_on_idle = false;
+    settings.cold_on_age = true;
+    settings.cold_max_hot_seconds = 5;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/aged", "aged-payload").await;
+    {
+        let mut store = cache.store.lock().unwrap();
+        for entry in store.entries.values_mut() {
+            entry.resident_since = Instant::now() - Duration::from_secs(60);
+            entry.last_access = Instant::now();
+        }
+    }
+    cache.demote_idle();
+    wait_for_cold(&cache, 1).await;
+    assert_eq!(cache.view().entries, 0);
+    assert_eq!(cache.view().cold_entries, 1);
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn graceful_shutdown_flush_moves_hot_objects_and_releases_resident_bytes() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    fill(&cache, &settings, "/two", "second-payload").await;
+    assert_eq!(cache.view().entries, 2);
+    assert!(cache.view().resident_bytes > 0);
+    assert_eq!(cache.view().cold_entries, 0);
+
+    assert_eq!(cache.flush_hot_to_cold().await, 2);
+    let view = cache.view();
+    assert_eq!(view.entries, 0);
+    assert_eq!(view.resident_bytes, 0);
+    assert_eq!(view.hot_logical_bytes, 0);
+    assert_eq!(
+        view.transient_bytes, 0,
+        "persisted ColdEntry must not retain the body allocation"
+    );
+    assert_eq!(view.cold_entries, 2);
+    shutdown.cancel();
+
+    let restart_shutdown = CancellationToken::new();
+    let restarted = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        restart_shutdown.clone(),
+    );
+    assert_eq!(restarted.view().cold_entries, 2);
+    restart_shutdown.cancel();
+}
+
+#[tokio::test]
+async fn managed_update_flush_ignores_shutdown_toggle_when_cold_tier_is_enabled() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    settings.cold_on_shutdown = false;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/update", "update-payload").await;
+    assert_eq!(cache.flush_hot_to_cold().await, 0);
+    assert_eq!(cache.view().entries, 1);
+    assert_eq!(cache.flush_hot_to_cold_for_update().await, 1);
+    let view = cache.view();
+    assert_eq!(view.entries, 0);
+    assert_eq!(view.resident_bytes, 0);
+    assert_eq!(view.cold_entries, 1);
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn graceful_shutdown_flush_can_be_disabled() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 8;
+    settings.cold_on_shutdown = false;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    assert_eq!(cache.flush_hot_to_cold().await, 0);
+    assert_eq!(cache.view().entries, 1);
+    assert_eq!(cache.view().cold_entries, 0);
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_cache_uses_two_hex_character_shards() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = cold_settings();
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    fill(&cache, &settings, "/two", "second-payload").await;
+    wait_for_cold(&cache, 1).await;
+    let root = dir.path().join(&settings.cold_path);
+    let mut shards = Vec::new();
+    for item in std::fs::read_dir(&root).unwrap() {
+        let item = item.unwrap();
+        let name = item.file_name().to_string_lossy().to_string();
+        assert!(
+            item.path().is_dir(),
+            "cold root must contain shard directories only"
+        );
+        assert_eq!(name.len(), 2);
+        assert!(name.bytes().all(|b| b.is_ascii_hexdigit()));
+        shards.push(item.path());
+    }
+    assert!(!shards.is_empty());
+    assert!(shards.iter().any(|shard| {
+        std::fs::read_dir(shard)
+            .unwrap()
+            .filter_map(|entry| entry.ok())
+            .any(|entry| {
+                let name = entry.file_name().to_string_lossy().to_string();
+                name.ends_with(".json") || name.ends_with(".body")
+            })
+    }));
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn legacy_flat_cold_cache_is_not_migrated() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = cold_settings();
+    let root = dir.path().join(&settings.cold_path);
+    std::fs::create_dir_all(&root).unwrap();
+    std::fs::write(root.join("legacy.body"), b"old").unwrap();
+    std::fs::write(root.join("legacy.json"), b"{}").unwrap();
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings,
+        shutdown.clone(),
+    );
+    assert_eq!(cache.view().cold_entries, 0);
+    assert!(!root.join("legacy.body").exists());
+    assert!(!root.join("legacy.json").exists());
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn manual_cold_cleanup_reports_removed_objects_and_freed_bytes() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 1;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    fill(&cache, &settings, "/two", "second-payload").await;
+    fill(&cache, &settings, "/three", "third-payload").await;
+    wait_for_cold(&cache, 2).await;
+    let job = cache
+        .start_cold_cleanup(ColdCleanupRequest {
+            unused_for_seconds: None,
+            oldest: Some(1),
+            hits_zero: false,
+        })
+        .unwrap();
+    for _ in 0..100 {
+        let current = cache.cold_cleanup_job(job.id).unwrap();
+        if current.status == "done" {
+            assert_eq!(current.removed_entries, 1);
+            assert!(current.freed_bytes > 0);
+            assert_eq!(cache.view().cold_entries, 1);
+            shutdown.cancel();
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("cold cleanup did not finish");
+}
+
+#[tokio::test]
+async fn manual_cold_cleanup_is_single_flight() {
+    let dir = tempfile::tempdir().unwrap();
+    let settings = cold_settings();
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings,
+        shutdown.clone(),
+    );
+    let _held = cache.cleanup_slots.clone().try_acquire_owned().unwrap();
+    let result = cache.start_cold_cleanup(ColdCleanupRequest {
+        unused_for_seconds: None,
+        oldest: Some(1),
+        hits_zero: false,
+    });
+    assert!(result.is_err());
+    shutdown.cancel();
+}
+
+#[tokio::test]
+async fn cold_cleanup_skips_an_entry_while_a_read_is_pinned() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut settings = cold_settings();
+    settings.max_entries = 1;
+    let shutdown = CancellationToken::new();
+    let cache = Cache::start(
+        11,
+        Some(dir.path().to_path_buf()),
+        settings.clone(),
+        shutdown.clone(),
+    );
+    fill(&cache, &settings, "/one", "first-payload").await;
+    fill(&cache, &settings, "/two", "second-payload").await;
+    wait_for_cold(&cache, 1).await;
+
+    let key = {
+        let mut cold = cache.cold.lock().unwrap();
+        let key = *cold.entries.keys().next().unwrap();
+        cold.entries.get_mut(&key).unwrap().pinned_reads = 1;
+        key
+    };
+    let request = ColdCleanupRequest {
+        unused_for_seconds: None,
+        oldest: Some(1),
+        hits_zero: false,
+    };
+    let result = cache
+        .remove_cold_candidates(Some(&request), None, None, 1)
+        .await;
+    assert_eq!(result.removed_entries, 0);
+    assert!(cache.cold.lock().unwrap().entries.contains_key(&key));
+
+    cache
+        .cold
+        .lock()
+        .unwrap()
+        .entries
+        .get_mut(&key)
+        .unwrap()
+        .pinned_reads = 0;
+    let result = cache
+        .remove_cold_candidates(Some(&request), None, None, 1)
+        .await;
+    assert_eq!(result.removed_entries, 1);
+    assert!(!cache.cold.lock().unwrap().entries.contains_key(&key));
+    shutdown.cancel();
+}
+
+#[test]
+fn old_cold_read_guard_does_not_unpin_a_replacement_with_the_same_key() {
+    let cache = Cache::new(1);
+    let key = [7_u8; 32];
+    let template = ColdEntry {
+        base_key: key,
+        vary: Vec::new(),
+        revision: 1,
+        scope: "app".to_owned(),
+        scheme: "https".to_owned(),
+        authority: "example.test".to_owned(),
+        domain: "example.test".to_owned(),
+        path: "/asset".to_owned(),
+        target: "https://example.test/asset".to_owned(),
+        tags: Vec::new(),
+        created_at: SystemTime::now(),
+        body_path: PathBuf::from("old.body"),
+        meta_path: PathBuf::from("old.json"),
+        body_bytes: 1,
+        body_sha256: [0u8; 32],
+        disk_bytes: 2,
+        stored_at: SystemTime::now(),
+        last_access: SystemTime::now(),
+        age_at_store_seconds: 0,
+        resident_age_at_store_seconds: 0,
+        ttl: 60,
+        stale_while_revalidate: 0,
+        stale_if_error: 0,
+        must_revalidate: false,
+        proxy_revalidate: false,
+        has_validator: false,
+        hits: 0,
+        transformed: false,
+        sequence: 1,
+        generation: 1,
+        skip_compression: false,
+        requires_revalidation: false,
+        pinned_reads: 1,
+    };
+    cache
+        .cold
+        .lock()
+        .unwrap()
+        .entries
+        .insert(key, template.clone());
+    let old_guard = ColdReadGuard {
+        cache: &cache,
+        key,
+        generation: 1,
+    };
+
+    let mut replacement = template;
+    replacement.sequence = 2;
+    replacement.generation = 2;
+    replacement.pinned_reads = 1;
+    cache.cold.lock().unwrap().entries.insert(key, replacement);
+    drop(old_guard);
+
+    let cold = cache.cold.lock().unwrap();
+    let replacement = cold.entries.get(&key).unwrap();
+    assert_eq!(replacement.generation, 2);
+    assert_eq!(replacement.pinned_reads, 1);
+}
+
+#[test]
+fn cold_path_cannot_escape_data_directory() {
+    let mut settings = Settings::default();
+    settings.cold_path = "../outside".into();
+    assert!(settings.validate().is_err());
+    settings.cold_path = "/tmp/outside".into();
+    assert!(settings.validate().is_err());
+    settings.cold_path = "cache/cold".into();
+    assert!(settings.validate().is_ok());
+}
 }

src/cache/tests/v022_cache_tests.rs

@@ -1,4 +1,950 @@
 mod v022_cache_tests {
-    include!("v022_cache_tests/part_01.rs");
-    include!("v022_cache_tests/part_02.rs");
+    use super::*;
+    use http_body_util::BodyExt;
+
+    fn active_settings() -> Settings {
+        Settings {
+            enabled: true,
+            ..Settings::default()
+        }
+    }
+
+    async fn store(
+        cache: &Arc<Cache>,
+        scope: &str,
+        host: &str,
+        target: &str,
+        request_headers: HeaderMap,
+        extra_response_headers: &[(&str, &str)],
+        body: &'static str,
+    ) -> Ticket {
+        let ticket = cache
+            .ticket_scoped(
+                22,
+                scope,
+                "http",
+                host,
+                "origin@127.0.0.1:8080",
+                target,
+                &request_headers,
+            )
+            .unwrap();
+        let settings = active_settings();
+        let mut response = Response::new(Body::from(body));
+        response.headers_mut().insert(
+            "cache-control",
+            HeaderValue::from_static("public, max-age=120, stale-if-error=60"),
+        );
+        for (name, value) in extra_response_headers {
+            response.headers_mut().insert(
+                http::HeaderName::from_bytes(name.as_bytes()).unwrap(),
+                HeaderValue::from_str(value).unwrap(),
+            );
+        }
+        let plan = plan(
+            StatusCode::OK,
+            response.headers(),
+            &settings,
+            Duration::ZERO,
+        )
+        .unwrap();
+        cache
+            .wrap(response, ticket.clone(), plan, &settings)
+            .into_body()
+            .collect()
+            .await
+            .unwrap();
+        ticket
+    }
+
+    fn invalidation(
+        operation: InvalidationOperation,
+        selector: InvalidationSelector,
+        value: &str,
+        host: Option<&str>,
+        vhost: Option<&str>,
+    ) -> InvalidationRequest {
+        InvalidationRequest {
+            operation,
+            selector,
+            value: value.to_owned(),
+            host: host.map(str::to_owned),
+            vhost: vhost.map(str::to_owned),
+            ttl_seconds: None,
+            confirm_wide: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn transformed_fill_retains_identical_content_and_releases_fill_gauge() {
+        for body in [
+            Bytes::from(vec![b'x'; 4096]),
+            Bytes::from_static(b"unchanged static bytes"),
+        ] {
+            let cache = Cache::new(22);
+            let settings = active_settings();
+            let ticket = cache.ticket(22, "http", "example.test", "origin", "/transform.js", &HeaderMap::new()).unwrap();
+            let mut response = Response::new(Body::from(body.clone()));
+            response.headers_mut().insert("cache-control", HeaderValue::from_static("public, max-age=120"));
+            response.headers_mut().insert("content-length", HeaderValue::from_str(&body.len().to_string()).unwrap());
+            response.extensions_mut().insert(crate::transform::Applied {
+                kind: crate::transform::Kind::Javascript,
+                original_bytes: body.len(),
+                transformed_bytes: body.len(),
+            });
+            let policy = plan(StatusCode::OK, response.headers(), &settings, Duration::ZERO).unwrap();
+            assert_eq!(policy.expected_length, Some(body.len()));
+            let downstream = cache.wrap(response, ticket.clone(), policy, &settings)
+                .into_body().collect().await.unwrap().to_bytes();
+            assert_eq!(downstream, body);
+            let cached = cache.get(&ticket).unwrap().response.into_body().collect().await.unwrap().to_bytes();
+            assert_eq!(cached, body);
+            assert_eq!(cache.fill_bytes_active.load(Ordering::Relaxed), 0);
+        }
+    }
+
+    #[tokio::test]
+    async fn uniquely_owned_transformed_frame_is_cached_without_body_copy() {
+        let cache = Cache::new(22);
+        let settings = active_settings();
+        let ticket = cache.ticket(22, "http", "example.test", "origin", "/owned.js", &HeaderMap::new()).unwrap();
+        let payload = Bytes::from(vec![b'z'; 8192]);
+        let original_allocation = payload.as_ptr();
+        let mut response = Response::new(Body::from(payload));
+        response.headers_mut().insert("cache-control", HeaderValue::from_static("public, max-age=120"));
+        response.headers_mut().insert("content-length", HeaderValue::from_static("8192"));
+        response.extensions_mut().insert(crate::transform::Applied {
+            kind: crate::transform::Kind::Javascript,
+            original_bytes: 8192,
+            transformed_bytes: 8192,
+        });
+        let policy = plan(StatusCode::OK, response.headers(), &settings, Duration::ZERO).unwrap();
+        let downstream = cache.wrap(response, ticket.clone(), policy, &settings)
+            .into_body().collect().await.unwrap().to_bytes();
+        assert_eq!(downstream.len(), 8192);
+        assert_eq!(downstream.as_ptr(), original_allocation);
+        let hit = cache.get(&ticket).unwrap();
+        let cached = hit.response.into_body().collect().await.unwrap().to_bytes();
+        assert_eq!(cached, downstream);
+        assert_eq!(cached.as_ptr(), original_allocation, "cache fill copied the transformed body");
+        assert_eq!(cache.fill_bytes_active.load(Ordering::Relaxed), 0);
+    }
+
+    #[tokio::test]
+    async fn cached_single_range_returns_206_and_preserves_validators() {
+        let cache = Cache::new(22);
+        let ticket = store(
+            &cache,
+            "range",
+            "range.example",
+            "/asset.bin",
+            HeaderMap::new(),
+            &[
+                ("etag", "\"asset-v1\""),
+                ("last-modified", "Mon, 26 Jan 2026 23:11:40 GMT"),
+            ],
+            "abcdefghij",
+        )
+        .await;
+        let mut request_headers = HeaderMap::new();
+        request_headers.insert("range", HeaderValue::from_static("bytes=2-5"));
+        let range = match request_range(&Method::GET, &request_headers) {
+            RangeSelection::Single(range) => range,
+            other => panic!("unexpected range selection: {other:?}"),
+        };
+        let hit = cache.get(&ticket).unwrap().apply_range(range, None);
+        assert_eq!(hit.response.status(), StatusCode::PARTIAL_CONTENT);
+        assert_eq!(hit.response.headers()["content-range"], "bytes 2-5/10");
+        assert_eq!(hit.response.headers()["content-length"], "4");
+        assert_eq!(hit.response.headers()["accept-ranges"], "bytes");
+        assert_eq!(hit.response.headers()["etag"], "\"asset-v1\"");
+        assert!(hit.response.headers().contains_key("age"));
+        let body = hit.response.into_body().collect().await.unwrap().to_bytes();
+        assert_eq!(body, Bytes::from_static(b"cdef"));
+    }
+
+    #[tokio::test]
+    async fn cached_range_supports_suffix_open_unsatisfied_and_if_range() {
+        let cache = Cache::new(22);
+        let ticket = store(
+            &cache,
+            "range",
+            "range.example",
+            "/asset.bin",
+            HeaderMap::new(),
+            &[
+                ("etag", "\"asset-v1\""),
+                ("last-modified", "Mon, 26 Jan 2026 23:11:40 GMT"),
+            ],
+            "abcdefghij",
+        )
+        .await;
+        let cases = [
+            (ByteRange { start: None, end: Some(3) }, "hij", "bytes 7-9/10"),
+            (ByteRange { start: Some(7), end: None }, "hij", "bytes 7-9/10"),
+        ];
+        for (range, expected, content_range) in cases {
+            let hit = cache.get(&ticket).unwrap().apply_range(range, None);
+            assert_eq!(hit.response.status(), StatusCode::PARTIAL_CONTENT);
+            assert_eq!(hit.response.headers()["content-range"], content_range);
+            let body = hit.response.into_body().collect().await.unwrap().to_bytes();
+            assert_eq!(body, Bytes::copy_from_slice(expected.as_bytes()));
+        }
+
+        let unsatisfied = cache.get(&ticket).unwrap().apply_range(
+            ByteRange { start: Some(50), end: None },
+            None,
+        );
+        assert_eq!(unsatisfied.response.status(), StatusCode::RANGE_NOT_SATISFIABLE);
+        assert_eq!(unsatisfied.response.headers()["content-range"], "bytes */10");
+        assert_eq!(
+            unsatisfied.response.into_body().collect().await.unwrap().to_bytes().len(),
+            0
+        );
+
+        let mismatch = HeaderValue::from_static("\"asset-v2\"");
+        let full = cache.get(&ticket).unwrap().apply_range(
+            ByteRange { start: Some(0), end: Some(2) },
+            Some(&mismatch),
+        );
+        assert_eq!(full.response.status(), StatusCode::OK);
+        assert!(!full.response.headers().contains_key("content-range"));
+        assert_eq!(
+            full.response.into_body().collect().await.unwrap().to_bytes(),
+            Bytes::from_static(b"abcdefghij")
+        );
+
+        let matched = HeaderValue::from_static("\"asset-v1\"");
+        let partial = cache.get(&ticket).unwrap().apply_range(
+            ByteRange { start: Some(0), end: Some(2) },
+            Some(&matched),
+        );
+        assert_eq!(partial.response.status(), StatusCode::PARTIAL_CONTENT);
+    }
+
+    #[test]
+    fn range_parser_rejects_multipart_and_duplicate_fields_and_if_range_date_allows_older_representation() {
+        let mut multipart = HeaderMap::new();
+        multipart.insert("range", HeaderValue::from_static("bytes=0-1,4-5"));
+        assert_eq!(
+            request_range(&Method::GET, &multipart),
+            RangeSelection::Unsupported
+        );
+
+        let mut duplicate = HeaderMap::new();
+        duplicate.append("range", HeaderValue::from_static("bytes=0-1"));
+        duplicate.append("range", HeaderValue::from_static("bytes=4-5"));
+        assert_eq!(
+            request_range(&Method::GET, &duplicate),
+            RangeSelection::Unsupported
+        );
+
+        let mut response_headers = HeaderMap::new();
+        response_headers.insert(
+            "last-modified",
+            HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT"),
+        );
+        let newer_if_range = HeaderValue::from_static("Tue, 27 Jan 2026 23:11:40 GMT");
+        assert!(if_range_matches(&response_headers, Some(&newer_if_range)));
+    }
+
+    #[tokio::test]
+    async fn concurrent_miss_single_flight_waits_until_fill_is_published() {
+        let cache = Cache::new(22);
+        let mut settings = active_settings();
+        settings.collapse_wait_timeout_ms = 500;
+        let ticket = cache
+            .ticket_scoped(
+                22,
+                "collapse",
+                "http",
+                "collapse.example",
+                "origin@127.0.0.1:8080",
+                "/same",
+                &HeaderMap::new(),
+            )
+            .unwrap();
+        let leader = match cache.claim_fill(&ticket, &settings).await {
+            FillClaim::Leader(guard) => guard,
+            _ => panic!("first request must become fill leader"),
+        };
+        let follower_cache = cache.clone();
+        let follower_ticket = ticket.clone();
+        let follower_settings = settings.clone();
+        let follower = tokio::spawn(async move {
+            match follower_cache
+                .claim_fill(&follower_ticket, &follower_settings)
+                .await
+            {
+                FillClaim::Waited => follower_cache.get(&follower_ticket).is_some(),
+                _ => false,
+            }
+        });
+        for _ in 0..50 {
+            if cache.view().collapse_waiters == 1 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(2)).await;
+        }
+        assert_eq!(cache.view().collapse_waiters, 1);
+        assert!(!follower.is_finished());
+
+        let mut response = Response::new(Body::from("collapsed-payload"));
+        response.headers_mut().insert(
+            "cache-control",
+            HeaderValue::from_static("public, max-age=60"),
+        );
+        let plan = plan(
+            StatusCode::OK,
+            response.headers(),
+            &settings,
+            Duration::ZERO,
+        )
+        .unwrap();
+        cache
+            .wrap_with_fill(response, ticket.clone(), plan, &settings, Some(leader))
+            .into_body()
+            .collect()
+            .await
+            .unwrap();
+        assert!(follower.await.unwrap());
+        assert_eq!(cache.view().collapsed_fills, 1);
+        assert_eq!(cache.view().collapse_waiters, 0);
+    }
+
+    #[tokio::test]
+    async fn single_flight_timeout_and_leader_drop_release_the_key() {
+        let cache = Cache::new(22);
+        let mut settings = active_settings();
+        settings.collapse_wait_timeout_ms = 10;
+        let ticket = cache
+            .ticket(22, "http", "collapse.example", "origin", "/timeout", &HeaderMap::new())
+            .unwrap();
+        let leader = match cache.claim_fill(&ticket, &settings).await {
+            FillClaim::Leader(guard) => guard,
+            _ => panic!("missing leader"),
+        };
+        assert!(matches!(cache.claim_fill(&ticket, &settings).await, FillClaim::Timeout));
+        assert_eq!(cache.view().collapse_timeouts, 1);
+        drop(leader);
+        assert!(matches!(
+            cache.claim_fill(&ticket, &settings).await,
+            FillClaim::Leader(_)
+        ));
+    }
+
+    #[tokio::test]
+    async fn cancelled_single_flight_waiter_releases_waiter_metric() {
+        let cache = Cache::new(22);
+        let mut settings = active_settings();
+        settings.collapse_wait_timeout_ms = 5_000;
+        let ticket = cache
+            .ticket(22, "http", "collapse.example", "origin", "/cancel", &HeaderMap::new())
+            .unwrap();
+        let leader = match cache.claim_fill(&ticket, &settings).await {
+            FillClaim::Leader(guard) => guard,
+            _ => panic!("missing leader"),
+        };
+        let follower_cache = cache.clone();
+        let follower_ticket = ticket.clone();
+        let follower_settings = settings.clone();
+        let follower = tokio::spawn(async move {
+            follower_cache
+                .claim_fill(&follower_ticket, &follower_settings)
+                .await
+        });
+        for _ in 0..50 {
+            if cache.view().collapse_waiters == 1 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(2)).await;
+        }
+        assert_eq!(cache.view().collapse_waiters, 1);
+        follower.abort();
+        let _ = follower.await;
+        tokio::task::yield_now().await;
+        assert_eq!(cache.view().collapse_waiters, 0);
+        drop(leader);
+    }
+
+    #[tokio::test]
+    async fn single_flight_retries_after_another_vary_representation_is_published() {
+        let cache = Cache::new(22);
+        let mut settings = active_settings();
+        settings.collapse_wait_timeout_ms = 500;
+        let mut en_headers = HeaderMap::new();
+        en_headers.insert("accept-language", HeaderValue::from_static("en"));
+        let mut pl_headers = HeaderMap::new();
+        pl_headers.insert("accept-language", HeaderValue::from_static("pl"));
+        let en = cache
+            .ticket(22, "http", "vary-collapse.example", "origin", "/page", &en_headers)
+            .unwrap();
+        let pl = cache
+            .ticket(22, "http", "vary-collapse.example", "origin", "/page", &pl_headers)
+            .unwrap();
+        assert_eq!(en.key, pl.key, "Vary is learned from the response");
+        let leader = match cache.claim_fill(&en, &settings).await {
+            FillClaim::Leader(guard) => guard,
+            _ => panic!("first representation must become leader"),
+        };
+        let follower_cache = cache.clone();
+        let follower_ticket = pl.clone();
+        let follower_settings = settings.clone();
+        let follower = tokio::spawn(async move {
+            assert!(matches!(
+                follower_cache
+                    .claim_fill(&follower_ticket, &follower_settings)
+                    .await,
+                FillClaim::Waited
+            ));
+            assert!(follower_cache.get(&follower_ticket).is_none());
+            matches!(
+                follower_cache
+                    .claim_fill(&follower_ticket, &follower_settings)
+                    .await,
+                FillClaim::Leader(_)
+            )
+        });
+        for _ in 0..50 {
+            if cache.view().collapse_waiters == 1 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(2)).await;
+        }
+
+        let mut response = Response::new(Body::from("hello"));
+        response.headers_mut().insert(
+            "cache-control",
+            HeaderValue::from_static("public, max-age=60"),
+        );
+        response
+            .headers_mut()
+            .insert("vary", HeaderValue::from_static("Accept-Language"));
+        let plan = plan(
+            StatusCode::OK,
+            response.headers(),
+            &settings,
+            Duration::ZERO,
+        )
+        .unwrap();
+        cache
+            .wrap_with_fill(response, en, plan, &settings, Some(leader))
+            .into_body()
+            .collect()
+            .await
+            .unwrap();
+        assert!(follower.await.unwrap());
+    }
+
+    #[test]
+    fn stale_if_error_policy_is_configurable_by_error_class_and_status() {
+        let mut settings = Settings::default();
+        settings.stale_if_error_timeout = false;
+        settings.stale_if_error_dns = false;
+        settings.stale_if_error_statuses = vec![500, 503];
+        assert!(!settings.allows_stale_error_class(StaleErrorClass::Timeout));
+        assert!(!settings.allows_stale_error_class(StaleErrorClass::Dns));
+        assert!(settings.allows_stale_error_class(StaleErrorClass::Connection));
+        assert!(settings.allows_stale_error_class(StaleErrorClass::Tls));
+        assert!(settings.allows_stale_error_status(StatusCode::INTERNAL_SERVER_ERROR));
+        assert!(settings.allows_stale_error_status(StatusCode::SERVICE_UNAVAILABLE));
+        assert!(!settings.allows_stale_error_status(StatusCode::BAD_GATEWAY));
+    }
+
+    #[test]
+    fn proactive_refresh_guards_deduplicate_and_bound_parallel_work() {
+        let cache = Cache::new(22);
+        let ticket = cache
+            .ticket(22, "http", "refresh.example", "origin", "/asset", &HeaderMap::new())
+            .unwrap();
+        let mut settings = active_settings();
+        settings.proactive_refresh_enabled = true;
+        settings.proactive_refresh_percent = 100;
+        settings.proactive_refresh_max_inflight = 1;
+        let scheduled = cache
+            .begin_proactive_schedule(&ticket, &settings)
+            .expect("first schedule");
+        assert!(cache.begin_proactive_schedule(&ticket, &settings).is_none());
+        let running = cache
+            .try_begin_proactive_refresh(&settings)
+            .expect("first inflight refresh");
+        assert!(cache.try_begin_proactive_refresh(&settings).is_none());
+        assert_eq!(cache.view().proactive_inflight, 1);
+        drop(running);
+        drop(scheduled);
+        assert!(cache.begin_proactive_schedule(&ticket, &settings).is_some());
+        assert_eq!(cache.view().proactive_inflight, 0);
+    }
+
+    #[test]
+    fn proactive_refresh_pending_tasks_are_bounded() {
+        let cache = Cache::new(22);
+        let mut settings = active_settings();
+        settings.proactive_refresh_enabled = true;
+        settings.proactive_refresh_percent = 100;
+        settings.proactive_refresh_max_inflight = 1;
+        let mut guards = Vec::new();
+        for index in 0..64 {
+            let ticket = cache
+                .ticket(
+                    22,
+                    "http",
+                    "refresh.example",
+                    "origin",
+                    &format!("/asset-{index}"),
+                    &HeaderMap::new(),
+                )
+                .unwrap();
+            guards.push(
+                cache
+                    .begin_proactive_schedule(&ticket, &settings)
+                    .expect("schedule below pending limit"),
+            );
+        }
+        let overflow = cache
+            .ticket(
+                22,
+                "http",
+                "refresh.example",
+                "origin",
+                "/overflow",
+                &HeaderMap::new(),
+            )
+            .unwrap();
+        assert!(cache.begin_proactive_schedule(&overflow, &settings).is_none());
+        assert!(cache.view().proactive_skipped > 0);
+        drop(guards);
+    }
+
+    #[test]
+    fn background_revalidation_set_has_a_hard_limit() {
+        let cache = Cache::new(22);
+        let base = Revalidation {
+            headers: HeaderMap::new(),
+            body: Bytes::new(),
+            body_allocated: 0,
+            _transient_leases: Vec::new(),
+            hits: 0,
+            transformed: false,
+            storage_key: [0; 32],
+            resident_age_at_lookup: Duration::ZERO,
+            skip_compression: false,
+            reusable_without_validation: false,
+            age: 0,
+            ttl: 0,
+            stale_if_error: 0,
+            must_revalidate: false,
+            proxy_revalidate: false,
+            origin_etag: None,
+            origin_last_modified: None,
+        };
+        let mut guards = Vec::new();
+        for index in 0..BACKGROUND_REVALIDATION_LIMIT {
+            let mut cached = base.clone();
+            cached.storage_key[0..8].copy_from_slice(&(index as u64).to_be_bytes());
+            guards.push(
+                cache
+                    .begin_background_revalidation(&cached)
+                    .expect("revalidation below hard limit"),
+            );
+        }
+        let mut overflow = base;
+        overflow.storage_key = [0xff; 32];
+        assert!(cache.begin_background_revalidation(&overflow).is_none());
+        drop(guards);
+    }
+
+    #[tokio::test]
+    async fn selective_purge_exact_prefix_tag_cache_key_and_query_are_precise() {
+        let cache = Cache::new(22);
+        let one = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/asset.js?v=1",
+            HeaderMap::new(),
+            &[("cache-tag", "release:one")],
+            "one",
+        )
+        .await;
+        let two = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/asset.js?v=2",
+            HeaderMap::new(),
+            &[("cache-tag", "release:two")],
+            "two",
+        )
+        .await;
+        let result = cache
+            .invalidate(invalidation(
+                InvalidationOperation::Purge,
+                InvalidationSelector::Exact,
+                "/asset.js?v=1",
+                Some("purge.example"),
+                Some("app"),
+            ))
+            .unwrap();
+        assert_eq!(result.affected_entries, 1);
+        assert!(cache.get(&one).is_none());
+        assert!(cache.get(&two).is_some());
+
+        let three = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/assets/a.css",
+            HeaderMap::new(),
+            &[("cache-tag", "release:three")],
+            "three",
+        )
+        .await;
+        let four = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/assets/b.css",
+            HeaderMap::new(),
+            &[("cache-tag", "release:four")],
+            "four",
+        )
+        .await;
+        assert_eq!(
+            cache
+                .invalidate(invalidation(
+                    InvalidationOperation::Purge,
+                    InvalidationSelector::Prefix,
+                    "/assets/",
+                    Some("purge.example"),
+                    Some("app"),
+                ))
+                .unwrap()
+                .affected_entries,
+            2
+        );
+        assert!(cache.get(&three).is_none());
+        assert!(cache.get(&four).is_none());
+
+        let tagged = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/tagged",
+            HeaderMap::new(),
+            &[("cache-tag", "release:tagged")],
+            "tagged",
+        )
+        .await;
+        assert_eq!(
+            cache
+                .invalidate(invalidation(
+                    InvalidationOperation::Purge,
+                    InvalidationSelector::Tag,
+                    "release:tagged",
+                    Some("purge.example"),
+                    Some("app"),
+                ))
+                .unwrap()
+                .affected_entries,
+            1
+        );
+        assert!(cache.get(&tagged).is_none());
+
+        let keyed = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/by-key",
+            HeaderMap::new(),
+            &[],
+            "keyed",
+        )
+        .await;
+        let key = key_hex(&keyed.key);
+        assert_eq!(
+            cache
+                .invalidate(invalidation(
+                    InvalidationOperation::Purge,
+                    InvalidationSelector::CacheKey,
+                    &key,
+                    Some("purge.example"),
+                    Some("app"),
+                ))
+                .unwrap()
+                .affected_entries,
+            1
+        );
+        assert!(cache.get(&keyed).is_none());
+    }
+
+    #[tokio::test]
+    async fn selective_purge_is_a_barrier_for_a_fill_started_before_the_purge() {
+        let cache = Cache::new(22);
+        let settings = active_settings();
+        let old_ticket = cache
+            .ticket_scoped(
+                22,
+                "app",
+                "http",
+                "purge.example",
+                "origin@127.0.0.1:8080",
+                "/late",
+                &HeaderMap::new(),
+            )
+            .unwrap();
+        let result = cache
+            .invalidate(invalidation(
+                InvalidationOperation::Purge,
+                InvalidationSelector::Tag,
+                "release:late",
+                Some("purge.example"),
+                Some("app"),
+            ))
+            .unwrap();
+        assert_eq!(result.affected_entries, 0);
+
+        let mut response = Response::new(Body::from("old-fill"));
+        response.headers_mut().insert(
+            "cache-control",
+            HeaderValue::from_static("public, max-age=60"),
+        );
+        response
+            .headers_mut()
+            .insert("cache-tag", HeaderValue::from_static("release:late"));
+        let plan = plan(
+            StatusCode::OK,
+            response.headers(),
+            &settings,
+            Duration::ZERO,
+        )
+        .unwrap();
+        cache
+            .wrap(response, old_ticket, plan, &settings)
+            .into_body()
+            .collect()
+            .await
+            .unwrap();
+        assert_eq!(cache.view().entries, 0, "pre-purge fill must not republish");
+
+        let new_ticket = store(
+            &cache,
+            "app",
+            "purge.example",
+            "/late",
+            HeaderMap::new(),
+            &[("cache-tag", "release:late")],
+            "new-fill",
+        )
+        .await;
+        assert!(cache.get(&new_ticket).is_some());
+    }
+
+    #[tokio::test]
+    async fn purge_of_one_vary_variant_removes_the_whole_variant_family() {
+        let cache = Cache::new(22);
+        let mut en_headers = HeaderMap::new();
+        en_headers.insert("accept-language", HeaderValue::from_static("en"));
+        let mut pl_headers = HeaderMap::new();
+        pl_headers.insert("accept-language", HeaderValue::from_static("pl"));
+        let en = store(
+            &cache,
+            "app",
+            "vary.example",
+            "/localized",
+            en_headers,
+            &[("vary", "Accept-Language"), ("cache-tag", "variant:en")],
+            "hello",
+        )
+        .await;
+        let pl = store(
+            &cache,
+            "app",
+            "vary.example",
+            "/localized",
+            pl_headers,
+            &[("vary", "Accept-Language"), ("cache-tag", "variant:pl")],
+            "czesc",
+        )
+        .await;
+        let result = cache
+            .invalidate(invalidation(
+                InvalidationOperation::Purge,
+                InvalidationSelector::Tag,
+                "variant:en",
+                Some("vary.example"),
+                Some("app"),
+            ))
+            .unwrap();
+        assert_eq!(result.affected_entries, 2);
+        assert!(cache.get(&en).is_none());
+        assert!(cache.get(&pl).is_none());
+    }
+
+    #[tokio::test]
+    async fn ban_is_lazy_blocks_old_object_allows_new_fill_and_can_be_removed() {
+        let cache = Cache::new(22);
+        let old = store(
+            &cache,
+            "app",
+            "ban.example",
+            "/page",
+            HeaderMap::new(),
+            &[],
+            "old",
+        )
+        .await;
+        assert_eq!(cache.view().entries, 1);
+        let result = cache
+            .invalidate(invalidation(
+                InvalidationOperation::Ban,
+                InvalidationSelector::Exact,
+                "/page",
+                Some("ban.example"),
+                Some("app"),
+            ))
+            .unwrap();
+        let ban_id = result.ban_id.unwrap();
+        assert_eq!(result.affected_entries, 0);
+        assert_eq!(cache.view().entries, 1, "BAN must not scan storage eagerly");
+        assert!(cache.get(&old).is_none());
+        assert_eq!(cache.view().entries, 0);
+
+        tokio::time::sleep(Duration::from_millis(2)).await;
+        let fresh = store(
+            &cache,
+            "app",
+            "ban.example",
+            "/page",
+            HeaderMap::new(),
+            &[],
+            "fresh",
+        )
+        .await;
+        assert!(cache.get(&fresh).is_some(), "old BAN must not reject newer objects");
+        assert_eq!(cache.bans().len(), 1);
+        assert!(cache.remove_ban(ban_id));
+        assert!(cache.bans().is_empty());
+        assert!(!cache.remove_ban(ban_id));
+    }
+
+    #[tokio::test]
+    async fn regex_and_root_prefix_require_explicit_wide_confirmation() {
+        let cache = Cache::new(22);
+        store(
+            &cache,
+            "app",
+            "wide.example",
+            "/one",
+            HeaderMap::new(),
+            &[],
+            "one",
+        )
+        .await;
+        let regex = invalidation(
+            InvalidationOperation::Purge,
+            InvalidationSelector::Regex,
+            ".*",
+            None,
+            None,
+        );
+        assert!(cache.invalidate(regex.clone()).is_err());
+        let scoped_regex = invalidation(
+            InvalidationOperation::Purge,
+            InvalidationSelector::Regex,
+            "^/one$",
+            Some("wide.example"),
+            Some("app"),
+        );
+        assert!(cache.invalidate(scoped_regex).is_err());
+        let mut confirmed = regex;
+        confirmed.confirm_wide = true;
+        assert_eq!(cache.invalidate(confirmed).unwrap().affected_entries, 1);
+
+        let prefix = invalidation(
+            InvalidationOperation::Ban,
+            InvalidationSelector::Prefix,
+            "/",
+            Some("wide.example"),
+            Some("app"),
+        );
+        assert!(cache.invalidate(prefix.clone()).is_err());
+        let mut confirmed_prefix = prefix;
+        confirmed_prefix.confirm_wide = true;
+        assert!(cache.invalidate(confirmed_prefix).unwrap().ban_id.is_some());
+    }
+
+    #[tokio::test]
+    async fn many_distinct_selective_purges_do_not_retain_tombstones_forever() {
+        let cache = Cache::new(22);
+        store(
+            &cache,
+            "app",
+            "memory.example",
+            "/stable",
+            HeaderMap::new(),
+            &[],
+            "unchanged",
+        )
+        .await;
+        // A production cache may see an unbounded stream of PURGEs for URLs
+        // that will never be cached again. The invalidated-key index must
+        // not remember all of those removed objects indefinitely.
+        for index in 0..512 {
+            let target = format!("/unique-{index}");
+            let old = store(
+                &cache,
+                "app",
+                "memory.example",
+                &target,
+                HeaderMap::new(),
+                &[],
+                "x",
+            )
+            .await;
+            assert!(cache.ticket_current(&old));
+            let result = cache
+                .invalidate(invalidation(
+                    InvalidationOperation::Purge,
+                    InvalidationSelector::Exact,
+                    &target,
+                    Some("memory.example"),
+                    Some("app"),
+                ))
+                .unwrap();
+            assert_eq!(result.affected_entries, 1);
+            assert!(!cache.ticket_current(&old), "pre-purge ticket must remain stale");
+            let state = cache.store.lock().unwrap_or_else(|e| e.into_inner());
+            assert!(
+                state.key_generations.len() <= 1,
+                "key generations must not accumulate for already purged URLs"
+            );
+        }
+        // A new ticket after the invalidation barrier is valid and can fill
+        // the same URL again, preserving ordinary cache behavior.
+        let new = store(
+            &cache,
+            "app",
+            "memory.example",
+            "/unique-0",
+            HeaderMap::new(),
+            &[],
+            "new",
+        )
+        .await;
+        assert!(cache.ticket_current(&new));
+        assert!(cache.get(&new).is_some());
+        let stable = cache
+            .ticket_scoped(
+                22,
+                "app",
+                "http",
+                "memory.example",
+                "origin@127.0.0.1:8080",
+                "/stable",
+                &HeaderMap::new(),
+            )
+            .unwrap();
+        assert!(cache.get(&stable).is_some(), "unrelated entries must survive");
+    }
 }