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

v0.31.8

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-12 11:21:50 +02:00
1 parent 427d23cb6e
commit 68ada6680d
28 files changed
-8669
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/cache/tests/cold_tier_tests/part_01.rs
Vendored
-416
View File
@@ -1,416 +0,0 @@
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();
}
src/cache/tests/cold_tier_tests/part_02.rs
Vendored
-327
View File
@@ -1,327 +0,0 @@
#[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/tests/part_01.rs
Vendored
-404
View File
@@ -1,404 +0,0 @@
use super::*;
use http_body_util::BodyExt;
fn headers(value: &str) -> HeaderMap {
let mut h = HeaderMap::new();
h.insert("cache-control", value.parse().unwrap());
h
}
fn settings() -> Settings {
Settings {
enabled: true,
..Default::default()
}
}
fn ticket(cache: &Cache, headers: &HeaderMap) -> Ticket {
cache
.ticket(1, "http", "example.test", "app", "/", headers)
.unwrap()
}
#[test]
fn ttl_safety_and_shared_directives() {
let s = settings();
for control in [
"private, max-age=60",
"no-store, max-age=60",
"max-age=0",
"max-age=invalid",
"max-age=60,max-age=90",
"public",
"must-understand, max-age=60",
"max-age=\"60",
"max-age=60\"",
"public=invalid,max-age=60",
] {
assert!(
plan(StatusCode::OK, &headers(control), &s, Duration::ZERO).is_none(),
"{control}"
);
}
assert!(plan(
StatusCode::OK,
&headers("public,max-age=60"),
&s,
Duration::ZERO
)
.is_some());
let mut revalidate_headers = headers("no-cache, max-age=60");
revalidate_headers.insert("etag", HeaderValue::from_static("\"validator\""));
let revalidate = plan(StatusCode::OK, &revalidate_headers, &s, Duration::ZERO)
.expect("no-cache with a validator permits storage for validation");
assert!(revalidate.requires_revalidation);
assert!(plan(StatusCode::OK, &headers("no-cache"), &s, Duration::ZERO).is_none());
let mut validator_only = headers("no-cache");
validator_only.insert("etag", HeaderValue::from_static("\"validator\""));
assert!(plan(StatusCode::OK, &validator_only, &s, Duration::ZERO).is_some());
assert!(plan(
StatusCode::OK,
&headers("max-age=60,s-maxage=0"),
&s,
Duration::ZERO
)
.is_none());
let shared = plan(
StatusCode::OK,
&headers("public, max-age=600, s-maxage=30"),
&s,
Duration::ZERO,
)
.unwrap();
assert_eq!(shared.ttl, 30);
let mut h = headers("max-age=60");
h.insert("age", "61".parse().unwrap());
assert!(plan(StatusCode::OK, &h, &s, Duration::ZERO).is_some());
h.remove("age");
h.insert("set-cookie", "session=secret".parse().unwrap());
assert!(plan(StatusCode::OK, &h, &s, Duration::ZERO).is_none());
}
#[test]
fn request_policy_uses_conditional_cache_and_revalidation() {
for name in [
"authorization",
"cookie",
"if-match",
"if-unmodified-since",
"upgrade",
] {
let mut h = HeaderMap::new();
h.insert(
http::HeaderName::from_bytes(name.as_bytes()).unwrap(),
"value".parse().unwrap(),
);
assert!(
!request_cacheable(&Method::GET, &h, true, "/", false, &settings()),
"{name}"
);
}
for name in ["if-none-match", "if-modified-since"] {
let mut h = HeaderMap::new();
h.insert(
http::HeaderName::from_bytes(name.as_bytes()).unwrap(),
"value".parse().unwrap(),
);
assert!(
request_cacheable(&Method::GET, &h, true, "/", false, &settings()),
"{name}"
);
}
for (value, reason) in [
("no-cache", "request_no_cache"),
("max-age=0", "request_max_age"),
] {
let h = headers(value);
assert_eq!(
request_policy(&Method::GET, &h, true, "/", false, &settings()),
RequestPolicy::Revalidate(reason)
);
}
let min_fresh = headers("min-fresh=1");
assert_eq!(
request_policy(&Method::GET, &min_fresh, true, "/", false, &settings()),
RequestPolicy::Use
);
let immutable = headers("immutable");
assert_eq!(
request_policy(&Method::GET, &immutable, true, "/", false, &settings()),
RequestPolicy::Use
);
let h = headers("no-store");
assert_eq!(
request_policy(&Method::GET, &h, true, "/", false, &settings()),
RequestPolicy::Bypass("request_no_store")
);
let mut pragma = HeaderMap::new();
pragma.insert("pragma", HeaderValue::from_static("no-cache"));
assert_eq!(
request_policy(&Method::GET, &pragma, true, "/", false, &settings()),
RequestPolicy::Revalidate("request_pragma_no_cache")
);
assert!(!request_cacheable(
&Method::POST,
&HeaderMap::new(),
true,
"/",
false,
&settings()
));
assert!(!request_cacheable(
&Method::GET,
&HeaderMap::new(),
false,
"/",
false,
&settings()
));
assert!(request_cacheable(
&Method::GET,
&HeaderMap::new(),
true,
"/",
false,
&settings()
));
assert!(request_cacheable(
&Method::HEAD,
&HeaderMap::new(),
true,
"/",
false,
&settings()
));
let mut single_range = HeaderMap::new();
single_range.insert("range", HeaderValue::from_static("bytes=0-3"));
assert_eq!(
request_policy(&Method::GET, &single_range, true, "/asset.bin", false, &settings()),
RequestPolicy::Use
);
assert!(request_cacheable(
&Method::GET,
&single_range,
true,
"/asset.bin",
false,
&settings()
));
single_range.insert("range", HeaderValue::from_static("bytes=0-3,7-9"));
assert_eq!(
request_policy(&Method::GET, &single_range, true, "/asset.bin", false, &settings()),
RequestPolicy::Bypass("request_range_unsupported")
);
}
#[test]
fn rfc5861_windows_and_revalidation_flags_are_separate_metadata() {
let s = settings();
let plan = plan(
StatusCode::OK,
&headers("public, max-age=60, stale-while-revalidate=30, stale-if-error=90, must-revalidate, proxy-revalidate"),
&s,
Duration::ZERO,
)
.expect("revalidation flags must not prevent storage");
assert_eq!(plan.ttl, 60);
assert_eq!(plan.stale_while_revalidate, 30);
assert_eq!(plan.stale_if_error, 90);
assert!(plan.must_revalidate);
assert!(plan.proxy_revalidate);
}
#[test]
fn request_max_stale_is_bounded_and_respects_revalidation_flags() {
let unbounded = headers("max-stale");
let bounded = headers("max-stale=15");
assert_eq!(stale_mode(120, 60, 0, false, false, false, &unbounded), Some(false));
assert_eq!(stale_mode(70, 60, 0, false, false, false, &bounded), Some(false));
assert_eq!(stale_mode(80, 60, 0, false, false, false, &bounded), None);
assert_eq!(stale_mode(70, 60, 30, false, false, false, &HeaderMap::new()), Some(true));
assert_eq!(stale_mode(100, 60, 30, false, false, false, &HeaderMap::new()), None);
assert_eq!(stale_mode(70, 60, 30, true, false, false, &unbounded), None);
assert_eq!(stale_mode(70, 60, 30, false, true, false, &unbounded), None);
}
#[test]
fn cache_state_distinguishes_fresh_stale_and_expired() {
let none = HeaderMap::new();
assert_eq!(
cache_state(30, 60, 20, false, false, false, &none),
CacheState::Fresh
);
assert_eq!(
cache_state(70, 60, 20, false, false, false, &none),
CacheState::StaleButServable {
background_revalidate: true
}
);
assert_eq!(
cache_state(90, 60, 20, false, false, false, &none),
CacheState::Expired
);
}
#[test]
fn stale_if_error_is_only_available_inside_its_window() {
let base = Revalidation {
headers: HeaderMap::new(),
body: Bytes::from_static(b"body"),
body_allocated: 4,
_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: 75,
ttl: 60,
stale_if_error: 30,
must_revalidate: false,
proxy_revalidate: false,
origin_etag: None,
origin_last_modified: None,
};
assert!(base.can_stale_if_error());
assert!(!Revalidation { age: 91, ..base.clone() }.can_stale_if_error());
assert!(!Revalidation { must_revalidate: true, ..base.clone() }.can_stale_if_error());
assert!(!Revalidation { proxy_revalidate: true, ..base }.can_stale_if_error());
}
#[test]
fn freshness_expiry_does_not_force_physical_eviction() {
assert_eq!(retention_deadline(60, 0, 0, false), u64::MAX);
assert_eq!(retention_deadline(60, 30, 0, false), u64::MAX);
assert_eq!(retention_deadline(60, 0, 45, false), u64::MAX);
assert_eq!(retention_deadline(60, 0, 0, true), u64::MAX);
}
#[test]
fn request_max_age_only_revalidates_when_cached_age_exceeds_limit() {
let fresh = headers("max-age=30");
assert!(request_freshness_allows(10, 60, &fresh));
assert!(!request_freshness_allows(31, 60, &fresh));
assert_eq!(
request_policy(&Method::GET, &fresh, true, "/", false, &settings()),
RequestPolicy::Use
);
let zero = headers("max-age=0");
assert_eq!(
request_policy(&Method::GET, &zero, true, "/", false, &settings()),
RequestPolicy::Revalidate("request_max_age")
);
let constrained_stale = headers("max-age=30, max-stale");
assert_eq!(
stale_mode(70, 60, 0, false, false, false, &constrained_stale),
None
);
}
#[test]
fn duplicate_and_argument_immutable_do_not_invalidate_cache_control() {
let duplicate = headers("public, max-age=60, immutable, immutable");
assert!(plan(StatusCode::OK, &duplicate, &settings(), Duration::ZERO).is_some());
let argument = headers("public, max-age=60, immutable=ignored");
assert!(plan(StatusCode::OK, &argument, &settings(), Duration::ZERO).is_some());
}
#[test]
fn no_transform_does_not_disable_shared_storage() {
let protected = headers("public, max-age=60, no-transform");
assert!(plan(StatusCode::OK, &protected, &settings(), Duration::ZERO).is_some());
}
#[test]
fn transformed_plan_uses_transformed_content_length() {
let s = settings();
let origin = plan(
StatusCode::OK,
&{
let mut h = headers("public, max-age=60");
h.insert("content-length", HeaderValue::from_static("95"));
h
},
&s,
Duration::ZERO,
)
.unwrap();
let transformed = plan(
StatusCode::OK,
&{
let mut h = headers("public, max-age=60");
h.insert("content-length", HeaderValue::from_static("76"));
h
},
&s,
Duration::ZERO,
)
.unwrap();
assert_eq!(origin.restrict(transformed).expected_length, Some(76));
}
#[tokio::test]
async fn transformed_representation_is_stored_and_reused() {
let cache = Cache::new(1);
let cache_settings = settings();
let request_headers = HeaderMap::new();
let ticket = ticket(&cache, &request_headers);
let origin_body = br#"{ "message": "hello" }"#;
let mut response = Response::new(Body::from(origin_body.as_slice()));
response.headers_mut().insert(
"cache-control",
HeaderValue::from_static("public, max-age=60"),
);
response
.headers_mut()
.insert("content-type", HeaderValue::from_static("application/json"));
response.headers_mut().insert(
"content-length",
HeaderValue::from_str(&origin_body.len().to_string()).unwrap(),
);
response
.headers_mut()
.insert("etag", HeaderValue::from_static("\"origin\""));
let origin_plan = plan(
StatusCode::OK,
response.headers(),
&cache_settings,
Duration::ZERO,
)
.unwrap();
let transform_settings = crate::transform::Settings {
enabled: true,
json: true,
..Default::default()
};
let transformed = crate::transform::apply(
response,
"/data.json",
&transform_settings,
false,
)
.await
.unwrap();
let transformed_plan = plan(
StatusCode::OK,
transformed.headers(),
&cache_settings,
Duration::ZERO,
)
.unwrap();
let expected_etag = transformed.headers()["etag"].clone();
let wrapped = cache.wrap(
transformed,
ticket.clone(),
origin_plan.restrict(transformed_plan),
&cache_settings,
);
let first = wrapped.into_body().collect().await.unwrap().to_bytes();
assert_eq!(&first[..], br#"{"message":"hello"}"#);
let hit = cache.get(&ticket).expect("transformed representation should be cached");
assert_eq!(hit.response.headers()["etag"], expected_etag);
assert_eq!(
hit.response.headers()["content-length"].to_str().unwrap(),
first.len().to_string()
);
let second = hit.response.into_body().collect().await.unwrap().to_bytes();
assert_eq!(second, first);
}
src/cache/tests/tests/part_02.rs
Vendored
-411
View File
@@ -1,411 +0,0 @@
#[tokio::test]
async fn every_enabled_transform_kind_can_fill_and_reuse_the_cached_representation() {
let cases: [(&str, &str, &[u8], &str); 5] = [
(
"/app.js",
"application/javascript",
&b"const value = 1;\n"[..],
"javascript",
),
(
"/app.css",
"text/css",
&b"body { color: red; }\n"[..],
"css",
),
(
"/page.html",
"text/html",
&b"<div><!--drop--><b>x</b></div>"[..],
"html",
),
(
"/icon.svg",
"image/svg+xml",
&b"<svg><!--drop--><path d=\"M0 0\"/></svg>"[..],
"svg",
),
(
"/data.json",
"application/json",
&br#"{ "a": 1 }"#[..],
"json",
),
];
for (path, content_type, origin_body, kind) in cases {
let cache = Cache::new(1);
let cache_settings = settings();
let request_headers = HeaderMap::new();
let ticket = cache
.ticket(1, "http", "example.test", "app", path, &request_headers)
.unwrap();
let mut response = Response::new(Body::from(origin_body));
response.headers_mut().insert(
"cache-control",
HeaderValue::from_static("public, max-age=60"),
);
response.headers_mut().insert(
"content-type",
HeaderValue::from_str(content_type).unwrap(),
);
response.headers_mut().insert(
"content-length",
HeaderValue::from_str(&origin_body.len().to_string()).unwrap(),
);
response
.headers_mut()
.insert("etag", HeaderValue::from_static("\"origin\""));
let origin_plan = plan(
StatusCode::OK,
response.headers(),
&cache_settings,
Duration::ZERO,
)
.unwrap();
let mut transform_settings = crate::transform::Settings {
enabled: true,
javascript: false,
css: false,
html: false,
svg: false,
json: false,
..Default::default()
};
match kind {
"javascript" => transform_settings.javascript = true,
"css" => transform_settings.css = true,
"html" => transform_settings.html = true,
"svg" => transform_settings.svg = true,
"json" => transform_settings.json = true,
_ => unreachable!(),
}
let transformed = crate::transform::apply(
response,
path,
&transform_settings,
false,
)
.await
.unwrap();
let transformed_plan = plan(
StatusCode::OK,
transformed.headers(),
&cache_settings,
Duration::ZERO,
)
.unwrap();
let expected_etag = transformed.headers()["etag"].clone();
let expected_length = transformed.headers()["content-length"].clone();
let wrapped = cache.wrap(
transformed,
ticket.clone(),
origin_plan.restrict(transformed_plan),
&cache_settings,
);
let first = wrapped.into_body().collect().await.unwrap().to_bytes();
assert!(!first.is_empty(), "{kind}");
assert_ne!(first.as_ref(), origin_body, "{kind} should change the chosen fixture");
let hit = cache.get(&ticket).expect("transformed representation should be cached");
assert_eq!(hit.response.headers()["etag"], expected_etag, "{kind}");
assert_eq!(
hit.response.headers()["content-length"],
expected_length,
"{kind}"
);
let second = hit.response.into_body().collect().await.unwrap().to_bytes();
assert_eq!(second, first, "{kind}");
}
}
#[test]
fn conditional_hit_uses_weak_etag_and_last_modified() {
let mut response = HeaderMap::new();
response.insert("etag", HeaderValue::from_static("W/\"abc\""));
response.insert(
"last-modified",
HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT"),
);
let mut request = HeaderMap::new();
request.insert("if-none-match", HeaderValue::from_static("\"abc\""));
assert!(conditional_not_modified(&request, &response));
request.insert("if-none-match", HeaderValue::from_static("\"other\""));
assert!(!conditional_not_modified(&request, &response));
request.remove("if-none-match");
request.insert(
"if-modified-since",
HeaderValue::from_static("Tue, 27 Jan 2026 00:00:00 GMT"),
);
assert!(conditional_not_modified(&request, &response));
}
#[test]
fn query_strings_remain_distinct_and_bypass_query_is_only_a_request_policy() {
let cache = Cache::new(1);
let headers = HeaderMap::new();
let v1 = cache
.ticket(1, "http", "example.test", "app", "/file.css?v=1", &headers)
.unwrap();
let v2 = cache
.ticket(1, "http", "example.test", "app", "/file.css?v=2", &headers)
.unwrap();
let flag = cache
.ticket(1, "http", "example.test", "app", "/file.css?F", &headers)
.unwrap();
assert_ne!(v1.key, v2.key);
assert_ne!(v1.key, flag.key);
assert_ne!(v2.key, flag.key);
let mut cache_settings = settings();
assert_eq!(
request_policy(
&Method::GET,
&headers,
true,
"/file.css",
true,
&cache_settings,
),
RequestPolicy::Use
);
cache_settings.bypass_query = true;
assert_eq!(
request_policy(
&Method::GET,
&headers,
true,
"/file.css",
true,
&cache_settings,
),
RequestPolicy::Bypass("query_excluded")
);
}
#[tokio::test]
async fn transformed_client_conditionals_use_the_representation_etag_not_the_origin_etag() {
let transform_settings = crate::transform::Settings {
enabled: true,
json: true,
..Default::default()
};
let body = br#"{ "a": 1 }"#;
let mut response = Response::new(Body::from(body.as_slice()));
response
.headers_mut()
.insert("content-type", HeaderValue::from_static("application/json"));
response.headers_mut().insert(
"content-length",
HeaderValue::from_str(&body.len().to_string()).unwrap(),
);
response
.headers_mut()
.insert("etag", HeaderValue::from_static("\"origin\""));
let response = crate::transform::apply(response, "/data.json", &transform_settings, false)
.await
.unwrap();
let representation_etag = response.headers()["etag"].clone();
assert_ne!(representation_etag, HeaderValue::from_static("\"origin\""));
let mut request = HeaderMap::new();
request.insert("if-none-match", representation_etag.clone());
assert!(conditional_not_modified(&request, response.headers()));
request.insert("if-none-match", HeaderValue::from_static("\"origin\""));
assert!(!conditional_not_modified(&request, response.headers()));
let validators = response
.extensions()
.get::<crate::transform::OriginValidators>()
.unwrap();
assert_eq!(validators.etag.as_ref().unwrap(), "\"origin\"");
}
#[test]
fn cache_key_separates_frontend_scheme() {
let cache = Cache::new(1);
let headers = HeaderMap::new();
let http = cache
.ticket(1, "http", "example.test", "app", "/asset", &headers)
.unwrap();
let https = cache
.ticket(1, "https", "example.test", "app", "/asset", &headers)
.unwrap();
assert_ne!(http.key, https.key);
}
#[test]
fn cache_key_ignores_revalidation_headers() {
let cache = Cache::new(1);
let base = ticket(&cache, &HeaderMap::new());
for (name, value) in [
("cache-control", "no-cache"),
("pragma", "no-cache"),
("if-none-match", "\"origin\""),
("if-modified-since", "Mon, 26 Jan 2026 23:11:40 GMT"),
] {
let mut headers = HeaderMap::new();
headers.insert(
http::HeaderName::from_bytes(name.as_bytes()).unwrap(),
HeaderValue::from_str(value).unwrap(),
);
assert_eq!(base.key, ticket(&cache, &headers).key, "{name}");
}
}
#[test]
fn cache_memory_defaults_are_bounded_and_sane() {
let s = Settings::default();
assert_eq!(s.memory_preset, None);
assert_eq!(s.cold_preset, None);
assert_eq!(s.max_bytes, 256 * 1024 * 1024);
assert_eq!(s.max_object_bytes, 16 * 1024 * 1024);
assert_eq!(s.transient_max_bytes, 64 * 1024 * 1024);
assert_eq!(s.cold_queue_max_bytes, 32 * 1024 * 1024);
assert_eq!(s.hot_eviction_max_objects, 64);
assert!(s.validate().is_ok());
assert!(Cache::fill_headroom_bytes(8 * 1024 * 1024) <= 128 * 1024);
assert!(Cache::fill_headroom_bytes(s.max_bytes) <= 1024 * 1024);
}
#[test]
fn cache_preset_intent_round_trips_with_configuration() {
let mut settings = Settings::default();
settings.memory_preset = Some(MemoryPreset::Recommended);
settings.cold_preset = Some(ColdPreset::Recommended);
let encoded = serde_json::to_value(&settings).unwrap();
assert_eq!(encoded["memory_preset"], "recommended");
assert_eq!(encoded["cold_preset"], "recommended");
let decoded: Settings = serde_json::from_value(encoded).unwrap();
assert_eq!(decoded.memory_preset, Some(MemoryPreset::Recommended));
assert_eq!(decoded.cold_preset, Some(ColdPreset::Recommended));
}
#[test]
fn transient_budget_is_hard_and_released_with_lease() {
let cache = Cache::new(1);
let mut s = Settings::default();
s.transient_max_bytes = 128 * 1024;
s.cold_queue_max_bytes = 64 * 1024;
let first = cache.try_reserve_transient(96 * 1024, &s).expect("first reservation");
assert!(cache.try_reserve_transient(40 * 1024, &s).is_none());
assert_eq!(cache.view().transient_bytes, 96 * 1024);
drop(first);
assert_eq!(cache.view().transient_bytes, 0);
assert!(cache.try_reserve_transient(128 * 1024, &s).is_some());
}
#[test]
fn cold_queue_budget_cannot_exceed_transient_budget() {
let mut s = Settings::default();
s.transient_max_bytes = 8 * 1024 * 1024;
s.cold_queue_max_bytes = 16 * 1024 * 1024;
assert!(s.validate().is_err());
}
#[cfg(target_pointer_width = "64")]
#[test]
fn cache_limits_allow_explicit_multi_gibibyte_budgets() {
let mut s = Settings::default();
s.max_bytes = 4usize * 1024 * 1024 * 1024;
s.max_object_bytes = 2usize * 1024 * 1024 * 1024;
s.transient_max_bytes = 512usize * 1024 * 1024;
s.cold_queue_max_bytes = 256usize * 1024 * 1024;
assert!(s.validate().is_ok());
s.max_object_bytes = s.max_bytes.saturating_add(1);
assert!(s.validate().is_err());
}
#[test]
fn revalidation_only_reuses_matching_client_validator() {
let cached = Revalidation {
headers: {
let mut h = HeaderMap::new();
h.insert("etag", HeaderValue::from_static("W/\"origin\""));
h.insert(
"last-modified",
HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT"),
);
h
},
body: Bytes::from_static(b"body"),
body_allocated: 4,
_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: 61,
ttl: 60,
stale_if_error: 30,
must_revalidate: false,
proxy_revalidate: false,
origin_etag: Some(HeaderValue::from_static("W/\"origin\"")),
origin_last_modified: Some(HeaderValue::from_static(
"Mon, 26 Jan 2026 23:11:40 GMT",
)),
};
let mut matching = HeaderMap::new();
matching.insert("if-none-match", HeaderValue::from_static("\"origin\""));
assert!(cached.apply_validators(&mut matching));
let mut other = HeaderMap::new();
other.insert("if-none-match", HeaderValue::from_static("\"other\""));
assert!(!cached.apply_validators(&mut other));
assert_eq!(other["if-none-match"], "\"other\"");
let mut empty = HeaderMap::new();
assert!(cached.apply_validators(&mut empty));
assert_eq!(empty["if-none-match"], "W/\"origin\"");
}
#[tokio::test]
async fn transformed_etag_is_never_used_as_an_origin_validator() {
let cache = Cache::new(1);
let ticket = ticket(&cache, &HeaderMap::new());
let settings = settings();
let modified = HeaderValue::from_static("Mon, 26 Jan 2026 23:11:40 GMT");
let public_etag = HeaderValue::from_static("\"rustproxy-deadbeef\"");
let mut response_headers = headers("public, max-age=60");
response_headers.insert("etag", public_etag.clone());
response_headers.insert("last-modified", modified.clone());
let plan = plan(
StatusCode::OK,
&response_headers,
&settings,
Duration::ZERO,
)
.unwrap();
let mut response = Response::new(Body::from("transformed"));
*response.headers_mut() = response_headers;
response
.extensions_mut()
.insert(crate::transform::OriginValidators {
etag: None,
last_modified: Some(modified.clone()),
});
cache
.wrap(response, ticket.clone(), plan, &settings)
.into_body()
.collect()
.await
.unwrap();
let cached = cache.revalidation(&ticket).expect("stored response");
assert!(cached.origin_etag.is_none());
assert_eq!(cached.origin_last_modified.as_ref(), Some(&modified));
let mut outbound = HeaderMap::new();
assert!(cached.apply_validators(&mut outbound));
assert!(!outbound.contains_key("if-none-match"));
assert_eq!(outbound["if-modified-since"], modified);
let mut client_conditional = HeaderMap::new();
client_conditional.insert("if-none-match", public_etag.clone());
assert!(cached.apply_validators(&mut client_conditional));
assert!(!client_conditional.contains_key("if-none-match"));
assert_eq!(client_conditional["if-modified-since"], modified);
let mut validation = HeaderMap::new();
validation.insert("cache-control", HeaderValue::from_static("public, max-age=60"));
validation.insert("etag", HeaderValue::from_static("\"origin-new\""));
validation.insert("last-modified", HeaderValue::from_static("Mon, 26 Jan 2026 23:12:40 GMT"));
let merged = cached.merge_validation_headers(&validation);
assert_eq!(merged["etag"], public_etag);
}
src/cache/tests/tests/part_03.rs
Vendored
-441
View File
@@ -1,441 +0,0 @@
#[tokio::test]
async fn vary_selects_variants_without_fragmenting_the_primary_hash() {
let cache = Cache::new(1);
let s = settings();
let mut en = HeaderMap::new();
en.insert("accept-language", "en".parse().unwrap());
let mut pl = en.clone();
pl.insert("accept-language", "pl".parse().unwrap());
let en_ticket = ticket(&cache, &en);
let pl_ticket = ticket(&cache, &pl);
assert_eq!(en_ticket.key, pl_ticket.key);
let mut h = headers("max-age=60");
h.insert("vary", HeaderValue::from_static("Accept-Language"));
let p = plan(StatusCode::OK, &h, &s, Duration::ZERO).unwrap();
let mut response = Response::new(Body::from("hello"));
*response.headers_mut() = h.clone();
cache
.wrap(response, en_ticket.clone(), p, &s)
.into_body()
.collect()
.await
.unwrap();
assert!(cache.get(&en_ticket).is_some());
assert!(cache.get(&pl_ticket).is_none());
let p = plan(StatusCode::OK, &h, &s, Duration::ZERO).unwrap();
let mut response = Response::new(Body::from("czesc"));
*response.headers_mut() = h;
cache
.wrap(response, pl_ticket.clone(), p, &s)
.into_body()
.collect()
.await
.unwrap();
let en_hit = cache.get(&en_ticket).expect("English Vary variant");
let pl_hit = cache.get(&pl_ticket).expect("Polish Vary variant");
assert_eq!(
en_hit
.response
.into_body()
.collect()
.await
.unwrap()
.to_bytes(),
Bytes::from_static(b"hello")
);
assert_eq!(
pl_hit
.response
.into_body()
.collect()
.await
.unwrap()
.to_bytes(),
Bytes::from_static(b"czesc")
);
let mut star = headers("max-age=60");
star.insert("vary", HeaderValue::from_static("*"));
assert!(plan(StatusCode::OK, &star, &s, Duration::ZERO).is_none());
for vary in [
"x-request-id",
"cache-control",
"pragma",
"if-none-match",
"if-modified-since",
] {
let mut h = headers("max-age=60");
h.insert("vary", vary.parse().unwrap());
assert!(
plan(StatusCode::OK, &h, &s, Duration::ZERO).is_some(),
"valid Vary field must remain cacheable: {vary}"
);
}
}
#[tokio::test]
async fn canonical_identity_drops_accept_encoding_as_a_cache_variant() {
let cache = Cache::new(1);
let s = settings();
let mut gzip = HeaderMap::new();
gzip.insert("accept-encoding", HeaderValue::from_static("gzip"));
let mut brotli = HeaderMap::new();
brotli.insert("accept-encoding", HeaderValue::from_static("br"));
let gzip_ticket = ticket(&cache, &gzip);
let brotli_ticket = ticket(&cache, &brotli);
let mut origin_headers = headers("public, max-age=60");
origin_headers.insert("vary", HeaderValue::from_static("Accept-Encoding"));
origin_headers.insert("content-length", HeaderValue::from_static("7"));
let origin_plan = plan(StatusCode::OK, &origin_headers, &s, Duration::ZERO)
.expect("origin plan");
let mut transformed_headers = headers("public, max-age=60");
transformed_headers.insert("content-length", HeaderValue::from_static("7"));
let transformed_plan = plan(
StatusCode::OK,
&transformed_headers,
&s,
Duration::ZERO,
)
.expect("transformed plan");
let cache_plan = origin_plan
.without_vary("accept-encoding")
.restrict(transformed_plan);
assert!(cache_plan.vary.is_empty());
let mut response = Response::new(Body::from("payload"));
*response.headers_mut() = transformed_headers;
cache
.wrap(response, gzip_ticket.clone(), cache_plan, &s)
.into_body()
.collect()
.await
.unwrap();
let hit = cache
.get(&brotli_ticket)
.expect("canonical identity must be shared across client encodings");
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"payload")
);
}
#[tokio::test]
async fn canonical_identity_preserves_non_encoding_vary_dimensions() {
let cache = Cache::new(1);
let s = settings();
let mut light = HeaderMap::new();
light.insert("accept-encoding", HeaderValue::from_static("gzip"));
light.insert("x-theme", HeaderValue::from_static("light"));
let mut dark = HeaderMap::new();
dark.insert("accept-encoding", HeaderValue::from_static("br"));
dark.insert("x-theme", HeaderValue::from_static("dark"));
let mut origin_headers = headers("public, max-age=60");
origin_headers.insert(
"vary",
HeaderValue::from_static("Accept-Encoding, X-Theme"),
);
let origin_plan = plan(StatusCode::OK, &origin_headers, &s, Duration::ZERO)
.expect("origin plan");
let mut transformed_headers = headers("public, max-age=60");
transformed_headers.insert("vary", HeaderValue::from_static("X-Theme"));
let transformed_plan = plan(
StatusCode::OK,
&transformed_headers,
&s,
Duration::ZERO,
)
.expect("transformed plan");
let cache_plan = origin_plan
.without_vary("accept-encoding")
.restrict(transformed_plan);
assert_eq!(cache_plan.vary, vec!["x-theme".to_owned()]);
let mut response = Response::new(Body::from("light"));
*response.headers_mut() = transformed_headers;
cache
.wrap(response, ticket(&cache, &light), cache_plan, &s)
.into_body()
.collect()
.await
.unwrap();
assert!(cache.get(&ticket(&cache, &light)).is_some());
assert!(cache.get(&ticket(&cache, &dark)).is_none());
}
#[tokio::test]
async fn vary_schema_change_removes_incompatible_fallbacks() {
let cache = Cache::new(1);
let s = settings();
let mut en = HeaderMap::new();
en.insert("accept-language", HeaderValue::from_static("en"));
let mut pl = HeaderMap::new();
pl.insert("accept-language", HeaderValue::from_static("pl"));
let en_ticket = ticket(&cache, &en);
let pl_ticket = ticket(&cache, &pl);
fill(&cache, en_ticket.clone(), "shared-old", &s).await;
assert!(cache.get(&pl_ticket).is_some());
let mut varied_headers = headers("max-age=60");
varied_headers.insert("vary", HeaderValue::from_static("Accept-Language"));
let varied_plan = plan(StatusCode::OK, &varied_headers, &s, Duration::ZERO).unwrap();
let mut varied = Response::new(Body::from("english-new"));
*varied.headers_mut() = varied_headers;
cache
.wrap(varied, en_ticket.clone(), varied_plan, &s)
.into_body()
.collect()
.await
.unwrap();
assert!(cache.get(&en_ticket).is_some());
assert!(
cache.get(&pl_ticket).is_none(),
"an older no-Vary object must not shadow a newer Vary schema"
);
fill(&cache, pl_ticket.clone(), "shared-new", &s).await;
let hit = cache
.get(&en_ticket)
.expect("new no-Vary schema should replace old variants");
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"shared-new")
);
}
async fn fill(cache: &Arc<Cache>, ticket: Ticket, data: &'static str, settings: &Settings) {
let h = headers("max-age=60");
let p = plan(StatusCode::OK, &h, settings, Duration::ZERO).unwrap();
let mut r = Response::new(Body::from(data));
*r.headers_mut() = h;
cache
.wrap(r, ticket, p, settings)
.into_body()
.collect()
.await
.unwrap();
}
#[tokio::test]
async fn max_stale_can_reuse_an_object_after_normal_ttl_expiry() {
let cache = Cache::new(1);
let s = settings();
let initial = ticket(&cache, &HeaderMap::new());
let response_headers = headers("public, max-age=1");
let p = plan(StatusCode::OK, &response_headers, &s, Duration::ZERO).unwrap();
let mut response = Response::new(Body::from("stale-payload"));
*response.headers_mut() = response_headers;
cache
.wrap(response, initial.clone(), p, &s)
.into_body()
.collect()
.await
.unwrap();
{
let mut store = cache.store.lock().unwrap();
let entry = store.entries.values_mut().next().expect("stored object");
entry.received = Instant::now() - Duration::from_secs(2);
}
assert!(cache.get(&initial).is_none(), "normal request must not receive stale data");
assert_eq!(cache.view().entries, 1, "freshness expiry must not physically evict the object");
let stale = ticket(&cache, &headers("max-stale"));
let hit = cache.get(&stale).expect("explicit max-stale may reuse retained stale data");
assert!(hit.stale);
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"stale-payload")
);
}
#[tokio::test]
async fn response_no_cache_is_stored_only_for_validation() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
let s = settings();
let mut h = headers("no-cache");
h.insert("etag", HeaderValue::from_static("\"origin\""));
let p = plan(StatusCode::OK, &h, &s, Duration::ZERO).unwrap();
assert!(p.requires_revalidation);
let mut response = Response::new(Body::from("payload"));
*response.headers_mut() = h;
cache
.wrap(response, t.clone(), p, &s)
.into_body()
.collect()
.await
.unwrap();
assert!(
cache.get(&t).is_none(),
"no-cache must not be reused without validation"
);
let candidate = cache.revalidation(&t).expect("stored validator/body");
assert!(!candidate.reusable_without_validation());
let mut validation = HeaderMap::new();
validation.insert("cache-control", HeaderValue::from_static("no-cache"));
validation.insert("etag", HeaderValue::from_static("\"origin\""));
let merged = candidate.merge_validation_headers(&validation);
let hit = cache
.publish_revalidated(t.clone(), &candidate, merged, &validation, &s, Duration::ZERO)
.expect("validated body can be delivered");
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"payload")
);
assert!(
cache.get(&t).is_none(),
"stored no-cache response still requires validation next time"
);
}
#[tokio::test]
async fn resident_age_survives_successful_revalidation() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
let s = settings();
let mut h = headers("no-cache");
h.insert("etag", HeaderValue::from_static("\"origin\""));
let p = plan(StatusCode::OK, &h, &s, Duration::ZERO).unwrap();
let mut response = Response::new(Body::from("payload"));
*response.headers_mut() = h;
cache
.wrap(response, t.clone(), p, &s)
.into_body()
.collect()
.await
.unwrap();
{
let mut store = cache.store.lock().unwrap();
let entry = store.entries.values_mut().next().expect("stored object");
entry.resident_age_base = Duration::from_secs(30);
entry.resident_since = Instant::now();
}
let candidate = cache.revalidation(&t).expect("revalidation candidate");
assert!(candidate.resident_age_at_lookup >= Duration::from_secs(30));
let mut validation = HeaderMap::new();
validation.insert(
"cache-control",
HeaderValue::from_static("public, max-age=60"),
);
validation.insert("etag", HeaderValue::from_static("\"origin\""));
let merged = candidate.merge_validation_headers(&validation);
let hit = cache
.publish_revalidated(t, &candidate, merged, &validation, &s, Duration::from_secs(2))
.expect("validated object");
assert!(hit.resident_age_seconds >= 32);
}
#[tokio::test]
async fn complete_body_hit_age_and_purge() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
assert!(cache.get(&t).is_none());
fill(&cache, t.clone(), "payload", &settings()).await;
let view = cache.view();
assert_eq!(view.hot_logical_bytes, b"payload".len());
assert_eq!(view.hot_allocated_bytes, view.resident_bytes);
assert!(view.hot_allocated_bytes > view.hot_logical_bytes);
assert!(view.hot_allocated_bytes <= view.hot_limit_bytes);
let hit = cache.get(&t).unwrap();
assert_eq!(hit.hits, 1);
assert!(hit.response.headers().contains_key("age"));
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"payload")
);
assert_eq!(cache.purge(None), 1);
assert!(cache.get(&t).is_none());
}
#[tokio::test]
async fn cache_hit_and_dashboard_age_advance_from_storage_clock() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
fill(&cache, t.clone(), "payload", &settings()).await;
{
let mut store = cache.store.lock().unwrap_or_else(|e| e.into_inner());
let entry = store.entries.values_mut().next().expect("stored entry");
entry.received = Instant::now() - Duration::from_secs(5);
}
let hit = cache.get(&t).expect("fresh cached response");
assert!(
hit.response.headers()["age"]
.to_str()
.unwrap()
.parse::<u64>()
.unwrap()
>= 5
);
assert!(cache.dashboard(None, 0, 50).objects[0].age_seconds >= 5);
}
#[test]
fn defaults_enable_static_cookie_boost_for_common_assets() {
let settings = Settings::default();
assert!(settings.enabled);
assert!(settings.static_ignore_cookies);
for path in [
"/asset.jpg",
"/asset.jpeg",
"/asset.png",
"/app.css",
"/app.js",
"/font.woff2",
"/audio.mp3",
"/note.txt",
] {
assert!(settings.ignore_cookie_for_path(path), "{path}");
}
assert!(!settings.ignore_cookie_for_path("/account"));
assert!(!settings.ignore_cookie_for_path("/api/profile.json"));
}
#[test]
fn static_cookie_bypass_is_limited_to_configured_extensions() {
let mut settings = settings();
settings.static_ignore_cookies = true;
let mut h = HeaderMap::new();
h.insert("cookie", HeaderValue::from_static("session=secret"));
assert_eq!(
request_reason(&Method::GET, &h, true, "/asset.jpg", false, &settings),
None
);
assert_eq!(
request_reason(&Method::GET, &h, true, "/account", false, &settings),
Some("request_cookie")
);
}
#[test]
fn age_uses_corrected_received_age_plus_response_delay_without_rounding_up() {
let settings = settings();
let mut h = headers("max-age=120");
h.insert(
"date",
httpdate::fmt_http_date(SystemTime::now() - Duration::from_secs(10))
.parse()
.unwrap(),
);
h.insert("age", HeaderValue::from_static("3"));
let age_plan = plan(StatusCode::OK, &h, &settings, Duration::from_millis(250)).unwrap();
assert!(age_plan.initial_age >= Duration::from_secs(10));
assert!(age_plan.initial_age < Duration::from_secs(12));
let fresh = plan(
StatusCode::OK,
&headers("max-age=60"),
&settings,
Duration::from_millis(250),
)
.unwrap();
assert_eq!(fresh.initial_age.as_secs(), 0);
}
#[tokio::test]
async fn purged_inflight_fill_cannot_repopulate() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
cache.purge(None);
fill(&cache, t, "old", &settings()).await;
assert_eq!(cache.view().entries, 0);
}
src/cache/tests/tests/part_04.rs
Vendored
-221
View File
@@ -1,221 +0,0 @@
#[tokio::test]
async fn revision_isolation_and_capacity() {
let cache = Cache::new(1);
let s = Settings {
max_entries: 1,
..settings()
};
let a = ticket(&cache, &HeaderMap::new());
fill(&cache, a.clone(), "a", &s).await;
let h = HeaderMap::new();
let b = cache
.ticket(1, "http", "example.test", "app", "/other", &h)
.unwrap();
fill(&cache, b.clone(), "b", &s).await;
assert!(cache.get(&a).is_none());
assert!(cache.get(&b).is_some());
assert_eq!(cache.view().evictions, 1);
cache.purge(Some(2));
assert!(cache
.ticket(1, "http", "example.test", "app", "/", &h)
.is_none());
}
#[tokio::test]
async fn hot_byte_budget_evicts_before_another_fill_grows() {
let cache = Cache::new(1);
let s = Settings {
max_bytes: 96 * 1024,
max_object_bytes: 64 * 1024,
max_entries: 16,
transient_max_bytes: 128 * 1024,
cold_queue_max_bytes: 64 * 1024,
..settings()
};
let headers = HeaderMap::new();
let first = cache
.ticket(1, "http", "example.test", "app", "/first", &headers)
.unwrap();
let second = cache
.ticket(1, "http", "example.test", "app", "/second", &headers)
.unwrap();
fill(&cache, first.clone(), "first", &s).await;
assert!(cache.get(&first).is_some());
let before = cache.view();
assert!(before.hot_allocated_bytes <= s.max_bytes);
fill(&cache, second.clone(), "second", &s).await;
let after = cache.view();
assert!(after.hot_allocated_bytes <= s.max_bytes);
assert_eq!(after.fill_bytes_active, 0);
assert_eq!(after.transient_bytes, 0);
assert!(cache.get(&first).is_none(), "LRU entry should make room before the next fill allocation");
assert!(cache.get(&second).is_some());
assert!(after.evictions >= 1);
}
#[tokio::test]
async fn dropped_partial_bodies_and_oversize_are_never_cached() {
let cache = Cache::new(1);
let t = ticket(&cache, &HeaderMap::new());
let s = settings();
let h = headers("max-age=60");
let p = plan(StatusCode::OK, &h, &s, Duration::ZERO).unwrap();
let r = Response::new(Body::from("not-consumed"));
drop(cache.wrap(r, t.clone(), p, &s));
assert!(cache.get(&t).is_none());
assert_eq!(cache.view().fills_active, 0);
let small = Settings {
max_object_bytes: 2,
..s
};
let h = headers("max-age=60");
let p = plan(StatusCode::OK, &h, &small, Duration::ZERO).unwrap();
let mut response = Response::new(Body::from("oversize"));
*response.headers_mut() = h;
let streamed = cache
.wrap(response, t.clone(), p, &small)
.into_body()
.collect()
.await
.unwrap()
.to_bytes();
assert_eq!(streamed, Bytes::from_static(b"oversize"));
assert!(cache.get(&t).is_none());
assert_eq!(cache.view().fill_bytes_active, 0);
assert_eq!(cache.view().transient_bytes, 0);
}
#[test]
fn varnish_primary_hash_uses_host_and_url_not_unrelated_request_headers() {
let cache = Cache::new(1);
let mut a = HeaderMap::new();
a.append("user-agent", "browser-a".parse().unwrap());
a.append("sec-ch-ua", "brand-a".parse().unwrap());
let mut b = HeaderMap::new();
b.append("user-agent", "browser-b".parse().unwrap());
b.append("x-forwarded-for", "192.0.2.8".parse().unwrap());
let base = cache
.ticket(
1,
"http",
"Example.Test",
"pool/member-a",
"/asset.js?x=1",
&a,
)
.unwrap();
let other_transport = cache
.ticket(
1,
"https",
"example.test",
"pool/member-b",
"/asset.js?x=1",
&b,
)
.unwrap();
assert_eq!(base.key, other_transport.key);
let other_url = cache
.ticket(
1,
"https",
"example.test",
"pool/member-b",
"/asset.js?x=2",
&b,
)
.unwrap();
assert_ne!(base.key, other_url.key);
let explicit_port = cache
.ticket(
1,
"http",
"example.test:8080",
"pool/member-a",
"/asset.js?x=1",
&a,
)
.unwrap();
assert_ne!(base.key, explicit_port.key);
}
#[tokio::test]
async fn only_if_cached_can_reuse_a_normal_get() {
let cache = Cache::new(1);
let plain = HeaderMap::new();
fill(&cache, ticket(&cache, &plain), "payload", &settings()).await;
let only = headers("only-if-cached");
assert!(request_cacheable(
&Method::GET,
&only,
true,
"/",
false,
&settings()
));
assert!(only_if_cached(&only));
assert!(cache.get(&ticket(&cache, &only)).is_some());
}
#[tokio::test]
async fn dropped_partial_fill_is_not_published_and_a_later_get_can_fill() {
use futures_util::stream;
use http_body_util::StreamBody;
let cache = Cache::new(1);
let s = settings();
let t = ticket(&cache, &HeaderMap::new());
let frames = vec![
Ok::<_, std::io::Error>(Frame::data(Bytes::from_static(b"first-"))),
Ok(Frame::data(Bytes::from_static(b"second"))),
];
let mut response = Response::new(Body::new(StreamBody::new(stream::iter(frames))));
*response.headers_mut() = headers("max-age=60");
response
.headers_mut()
.insert("content-length", HeaderValue::from_static("12"));
let p = plan(StatusCode::OK, response.headers(), &s, Duration::ZERO).unwrap();
let mut body = cache.wrap(response, t.clone(), p, &s).into_body();
let first = body
.frame()
.await
.expect("first frame")
.expect("successful first frame")
.into_data()
.expect("data frame");
assert_eq!(first, Bytes::from_static(b"first-"));
drop(body);
assert!(cache.get(&t).is_none(), "partial body must never become a HIT");
assert_eq!(cache.view().fills_active, 0);
assert_eq!(cache.view().rejected_fills, 1);
fill(&cache, t.clone(), "complete", &s).await;
let hit = cache.get(&t).expect("a later complete GET can fill the cache");
assert_eq!(
hit.response.into_body().collect().await.unwrap().to_bytes(),
Bytes::from_static(b"complete")
);
}
#[tokio::test]
async fn failed_streams_and_trailers_cannot_publish_entries() {
use futures_util::stream;
use http_body_util::StreamBody;
for fail in [true, false] {
let cache = Cache::new(1);
let s = settings();
let t = ticket(&cache, &HeaderMap::new());
let frames = vec![
Ok(Frame::data(Bytes::from_static(b"partial"))),
if fail {
Err(std::io::Error::other("broken origin"))
} else {
Ok(Frame::trailers(HeaderMap::new()))
},
];
let mut r = Response::new(Body::new(StreamBody::new(stream::iter(frames))));
*r.headers_mut() = headers("max-age=60");
let p = plan(StatusCode::OK, r.headers(), &s, Duration::ZERO).unwrap();
let result = cache.wrap(r, t.clone(), p, &s).into_body().collect().await;
assert_eq!(result.is_err(), fail);
assert!(cache.get(&t).is_none());
assert_eq!(cache.view().fills_active, 0);
}
}
src/cache/tests/v022_cache_tests/part_01.rs
Vendored
-421
View File
@@ -1,421 +0,0 @@
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 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);
}
src/cache/tests/v022_cache_tests/part_02.rs
Vendored
-471
View File
@@ -1,471 +0,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");
}