use std::fmt;
use std::fs::{self, File, FileTimes, OpenOptions};
use std::io::{self, Read, Write};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
pub const PACK_L2_CACHE_ENTRY_SCHEMA_V1: &str = "ee.pack.l2_cache.entry.v1";
pub const PACK_L2_CACHE_ENTRY_SCHEMA_V2: &str = "ee.pack.l2_cache.entry.v2";
pub const DEFAULT_MAX_BYTES: u64 = 256 * 1024 * 1024;
pub const DEFAULT_MAX_ENTRY_BYTES: u64 = 1024 * 1024;
const PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1: &str = "zstd_frame_v1";
const PACK_L2_COMPRESSION_LEVEL: i32 = 3;
const DEFAULT_MAX_AGE: Duration = Duration::from_secs(30 * 24 * 60 * 60);
static TEMP_FILE_COUNTER: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CacheOptions {
pub max_bytes: u64,
pub max_entry_bytes: u64,
pub max_age: Duration,
}
impl PackL2CacheOptions {
#[must_use]
pub const fn new(max_bytes: u64, max_age: Duration) -> Self {
Self {
max_bytes,
max_entry_bytes: DEFAULT_MAX_ENTRY_BYTES,
max_age,
}
}
#[must_use]
pub const fn with_max_entry_bytes(mut self, max_entry_bytes: u64) -> Self {
self.max_entry_bytes = max_entry_bytes;
self
}
}
impl Default for PackL2CacheOptions {
fn default() -> Self {
Self {
max_bytes: DEFAULT_MAX_BYTES,
max_entry_bytes: DEFAULT_MAX_ENTRY_BYTES,
max_age: DEFAULT_MAX_AGE,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2Cache {
root: PathBuf,
options: PackL2CacheOptions,
}
impl PackL2Cache {
#[must_use]
pub fn new(root: impl Into<PathBuf>, options: PackL2CacheOptions) -> Self {
Self {
root: root.into(),
options,
}
}
#[must_use]
pub fn root(&self) -> &Path {
&self.root
}
#[must_use]
pub fn options(&self) -> &PackL2CacheOptions {
&self.options
}
#[must_use]
pub fn entry_path(&self, key: &str) -> PathBuf {
self.root.join(cache_file_name(key))
}
#[must_use]
pub fn entry_path_for_body_hash(&self, key: &str, body_hash_prefix: &str) -> PathBuf {
self.root
.join(cache_file_name_with_body_hash(key, body_hash_prefix))
}
pub fn get(&self, key: &str) -> Result<PackL2CacheLookup, PackL2CacheError> {
self.get_at(key, system_time_seconds(SystemTime::now())?)
}
pub fn get_at(
&self,
key: &str,
now_epoch_seconds: u64,
) -> Result<PackL2CacheLookup, PackL2CacheError> {
let fallback_path = self.entry_path(key);
let candidates = self.entry_candidates(key)?;
if candidates.is_empty() {
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path: fallback_path,
reason: PackL2CacheMissReason::NotFound,
}));
}
let mut last_miss = None;
for path in candidates {
match self.get_candidate_at(key, path, now_epoch_seconds)? {
PackL2CacheLookup::Hit(hit) => return Ok(PackL2CacheLookup::Hit(hit)),
PackL2CacheLookup::Miss(miss) => {
last_miss = Some(miss);
}
}
}
Ok(PackL2CacheLookup::Miss(last_miss.unwrap_or(
PackL2CacheMiss {
key: key.to_owned(),
path: fallback_path,
reason: PackL2CacheMissReason::NotFound,
},
)))
}
fn get_candidate_at(
&self,
key: &str,
path: PathBuf,
now_epoch_seconds: u64,
) -> Result<PackL2CacheLookup, PackL2CacheError> {
ensure_no_symlink_components(&path, "inspect_entry")?;
let bytes = match read_cache_entry_file(&path, self.options.max_entry_bytes) {
Ok(bytes) => bytes,
Err(error) if error.kind() == io::ErrorKind::NotFound => {
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason: PackL2CacheMissReason::NotFound,
}));
}
Err(error) => {
return Err(PackL2CacheError::Io {
path,
operation: "read",
source: error,
});
}
};
if let Some(expected_body_hash_prefix) = body_hash_prefix_from_path(&path) {
let actual_body_hash_prefix = body_hash_prefix(&bytes);
if actual_body_hash_prefix != expected_body_hash_prefix {
remove_cache_entry_best_effort(&path);
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason: PackL2CacheMissReason::BodyHashMismatch {
expected: expected_body_hash_prefix,
actual: actual_body_hash_prefix,
},
}));
}
}
let byte_len = bytes.len() as u64;
if byte_len > self.options.max_entry_bytes {
remove_cache_entry_best_effort(&path);
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason: PackL2CacheMissReason::TooLarge {
byte_len,
max_entry_bytes: self.options.max_entry_bytes,
},
}));
}
let entry = match decode_pack_l2_cache_entry(&bytes, self.options.max_entry_bytes) {
Ok(entry) => entry,
Err(reason) => {
remove_cache_entry_best_effort(&path);
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason,
}));
}
};
if entry.key != key {
remove_cache_entry_best_effort(&path);
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason: PackL2CacheMissReason::KeyMismatch {
stored_key: entry.key,
},
}));
}
if is_expired(
entry.stored_at_epoch_seconds,
now_epoch_seconds,
self.options.max_age,
) {
return Ok(PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path,
reason: PackL2CacheMissReason::Expired {
stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
},
}));
}
touch_cache_entry_mtime_best_effort(&path, now_epoch_seconds);
Ok(PackL2CacheLookup::Hit(PackL2CacheHit {
key: entry.key,
path,
stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
pack_json: entry.pack_json,
byte_len,
compression: entry.compression,
}))
}
pub fn put(
&self,
key: &str,
pack_json: &JsonValue,
) -> Result<PackL2WriteReport, PackL2CacheError> {
self.put_at(key, pack_json, system_time_seconds(SystemTime::now())?)
}
pub fn put_at(
&self,
key: &str,
pack_json: &JsonValue,
stored_at_epoch_seconds: u64,
) -> Result<PackL2WriteReport, PackL2CacheError> {
let path = self.entry_path(key);
let entry = PackL2CacheEntry {
schema: PACK_L2_CACHE_ENTRY_SCHEMA_V1.to_owned(),
key: key.to_owned(),
stored_at_epoch_seconds,
pack_json: pack_json.clone(),
};
let bytes = serde_json::to_vec(&entry).map_err(|source| PackL2CacheError::Json {
path: path.clone(),
operation: "serialize",
source,
})?;
let byte_len = bytes.len() as u64;
let body_hash_prefix = body_hash_prefix(&bytes);
let path = self.entry_path_for_body_hash(key, &body_hash_prefix);
if byte_len > self.options.max_entry_bytes {
return Ok(PackL2WriteReport {
key: key.to_owned(),
path,
byte_len,
uncompressed_byte_len: byte_len,
compression: None,
outcome: PackL2WriteOutcome::SkippedTooLarge {
max_entry_bytes: self.options.max_entry_bytes,
},
eviction: PackL2EvictionReport::default(),
});
}
ensure_cache_dir(&self.root)?;
let temp_path = self.temp_path(key, &body_hash_prefix, stored_at_epoch_seconds);
ensure_no_symlink_components(&path, "inspect_entry")?;
ensure_no_symlink_components(&temp_path, "inspect_temp")?;
write_synced_file(&temp_path, &bytes)?;
publish_cache_entry_temp_file(&temp_path, &path)?;
touch_cache_entry_mtime_best_effort(&path, stored_at_epoch_seconds);
sync_directory(&self.root)?;
let duplicate_cleanup = self.prune_duplicate_key_entries_best_effort(key, &path)?;
let eviction = merge_cache_cleanup_reports(
duplicate_cleanup,
self.evict_best_effort_at(stored_at_epoch_seconds)?,
);
Ok(PackL2WriteReport {
key: key.to_owned(),
path,
byte_len,
uncompressed_byte_len: byte_len,
compression: None,
outcome: PackL2WriteOutcome::Stored,
eviction,
})
}
pub fn put_compressed(
&self,
key: &str,
pack_json: &JsonValue,
) -> Result<PackL2WriteReport, PackL2CacheError> {
self.put_compressed_with_dictionary_at(
key,
pack_json,
None,
system_time_seconds(SystemTime::now())?,
)
}
pub fn put_compressed_at(
&self,
key: &str,
pack_json: &JsonValue,
stored_at_epoch_seconds: u64,
) -> Result<PackL2WriteReport, PackL2CacheError> {
self.put_compressed_with_dictionary_at(key, pack_json, None, stored_at_epoch_seconds)
}
pub fn put_compressed_with_dictionary_at(
&self,
key: &str,
pack_json: &JsonValue,
dictionary: Option<&PackL2CompressionDictionary>,
stored_at_epoch_seconds: u64,
) -> Result<PackL2WriteReport, PackL2CacheError> {
let path = self.entry_path(key);
let uncompressed =
serde_json::to_vec(pack_json).map_err(|source| PackL2CacheError::Json {
path: path.clone(),
operation: "serialize_uncompressed",
source,
})?;
let uncompressed_byte_len = uncompressed.len() as u64;
let compression_start = Instant::now();
let compressed = zstd_compress(&uncompressed, dictionary)?;
let compression_latency_ms = elapsed_millis(compression_start.elapsed());
let compressed_byte_len = compressed.len() as u64;
let entry = PackL2CacheEntryV2 {
schema: PACK_L2_CACHE_ENTRY_SCHEMA_V2.to_owned(),
key: key.to_owned(),
stored_at_epoch_seconds,
compression: PackL2CacheCompressionPayload {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
compressed_payload_base64: BASE64_STANDARD.encode(&compressed),
compressed_byte_len,
uncompressed_byte_len,
uncompressed_hash: blake3_hash(&uncompressed),
dictionary: dictionary.map(PackL2CacheCompressionDictionaryRef::from_dictionary),
},
};
let bytes = serde_json::to_vec(&entry).map_err(|source| PackL2CacheError::Json {
path: path.clone(),
operation: "serialize_compressed",
source,
})?;
let byte_len = bytes.len() as u64;
let body_hash_prefix = body_hash_prefix(&bytes);
let path = self.entry_path_for_body_hash(key, &body_hash_prefix);
let compression_report = PackL2CompressionWriteReport {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
dictionary_id: dictionary.map(|dictionary| dictionary.id.clone()),
compressed_bytes: compressed_byte_len,
uncompressed_bytes: uncompressed_byte_len,
compression_latency_ms,
};
if byte_len > self.options.max_entry_bytes {
return Ok(PackL2WriteReport {
key: key.to_owned(),
path,
byte_len,
uncompressed_byte_len,
compression: Some(compression_report),
outcome: PackL2WriteOutcome::SkippedTooLarge {
max_entry_bytes: self.options.max_entry_bytes,
},
eviction: PackL2EvictionReport::default(),
});
}
ensure_cache_dir(&self.root)?;
let temp_path = self.temp_path(key, &body_hash_prefix, stored_at_epoch_seconds);
ensure_no_symlink_components(&path, "inspect_entry")?;
ensure_no_symlink_components(&temp_path, "inspect_temp")?;
write_synced_file(&temp_path, &bytes)?;
publish_cache_entry_temp_file(&temp_path, &path)?;
touch_cache_entry_mtime_best_effort(&path, stored_at_epoch_seconds);
sync_directory(&self.root)?;
let duplicate_cleanup = self.prune_duplicate_key_entries_best_effort(key, &path)?;
let eviction = merge_cache_cleanup_reports(
duplicate_cleanup,
self.evict_best_effort_at(stored_at_epoch_seconds)?,
);
Ok(PackL2WriteReport {
key: key.to_owned(),
path,
byte_len,
uncompressed_byte_len,
compression: Some(compression_report),
outcome: PackL2WriteOutcome::Stored,
eviction,
})
}
pub fn evict_best_effort(&self) -> Result<PackL2EvictionReport, PackL2CacheError> {
self.evict_best_effort_at(system_time_seconds(SystemTime::now())?)
}
pub fn evict_best_effort_at(
&self,
now_epoch_seconds: u64,
) -> Result<PackL2EvictionReport, PackL2CacheError> {
ensure_no_symlink_components(&self.root, "inspect_root")?;
let mut report = PackL2EvictionReport::default();
let mut candidates = Vec::new();
let entries = match fs::read_dir(&self.root) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(report),
Err(error) => {
return Err(PackL2CacheError::Io {
path: self.root.clone(),
operation: "read_dir",
source: error,
});
}
};
for entry in entries {
let Ok(entry) = entry else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
let path = entry.path();
if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
continue;
}
let Ok(file_type) = entry.file_type() else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
if file_type.is_symlink() {
report.skipped = report.skipped.saturating_add(1);
continue;
}
if !file_type.is_file() {
continue;
}
let Ok(metadata) = fs::symlink_metadata(&path) else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
let byte_len = metadata.len();
report.bytes_before = report.bytes_before.saturating_add(byte_len);
let fallback_epoch_seconds = metadata
.modified()
.ok()
.and_then(|modified| system_time_seconds(modified).ok())
.unwrap_or(0);
let stored_epoch_seconds = cache_entry_stored_at(&path, self.options.max_entry_bytes)
.unwrap_or(fallback_epoch_seconds);
let last_used_epoch_seconds = fallback_epoch_seconds;
let expired = stored_epoch_seconds == 0
|| is_expired(
stored_epoch_seconds,
now_epoch_seconds,
self.options.max_age,
);
candidates.push(EvictionCandidate {
path,
byte_len,
stored_epoch_seconds,
last_used_epoch_seconds,
expired,
});
}
candidates.sort_by(|left, right| {
left.expired
.cmp(&right.expired)
.reverse()
.then_with(|| {
left.last_used_epoch_seconds
.cmp(&right.last_used_epoch_seconds)
})
.then_with(|| left.stored_epoch_seconds.cmp(&right.stored_epoch_seconds))
.then_with(|| left.path.cmp(&right.path))
});
let mut bytes_current = report.bytes_before;
for candidate in candidates {
if !candidate.expired && bytes_current <= self.options.max_bytes {
break;
}
remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);
}
report.bytes_after = bytes_current;
Ok(report)
}
fn entry_candidates(&self, key: &str) -> Result<Vec<PathBuf>, PackL2CacheError> {
ensure_no_symlink_components(&self.root, "inspect_root")?;
let key_stem = cache_file_stem(key);
let mut candidates = Vec::new();
let entries = match fs::read_dir(&self.root) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => {
return Err(PackL2CacheError::Io {
path: self.root.clone(),
operation: "read_dir",
source: error,
});
}
};
for entry in entries {
let entry = entry.map_err(|source| PackL2CacheError::Io {
path: self.root.clone(),
operation: "read_dir_entry",
source,
})?;
let path = entry.path();
let Some(file_name) = path.file_name().and_then(|file_name| file_name.to_str()) else {
continue;
};
if file_name == cache_file_name(key)
|| body_hashed_file_name_matches(file_name, &key_stem)
{
let preference_epoch =
cache_entry_preference_epoch_seconds(&path, self.options.max_entry_bytes);
candidates.push((path, preference_epoch));
}
}
candidates.sort_by(|(left_path, left_epoch), (right_path, right_epoch)| {
right_epoch
.cmp(left_epoch)
.then_with(|| left_path.cmp(right_path))
});
Ok(candidates.into_iter().map(|(path, _)| path).collect())
}
fn prune_duplicate_key_entries_best_effort(
&self,
key: &str,
retained_path: &Path,
) -> Result<PackL2EvictionReport, PackL2CacheError> {
ensure_no_symlink_components(&self.root, "inspect_root")?;
let key_stem = cache_file_stem(key);
let mut report = PackL2EvictionReport::default();
let mut bytes_current = 0_u64;
let entries = match fs::read_dir(&self.root) {
Ok(entries) => entries,
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(report),
Err(error) => {
return Err(PackL2CacheError::Io {
path: self.root.clone(),
operation: "read_dir",
source: error,
});
}
};
for entry in entries {
let Ok(entry) = entry else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
let path = entry.path();
if path == retained_path {
continue;
}
let Some(file_name) = path.file_name().and_then(|file_name| file_name.to_str()) else {
continue;
};
if file_name != cache_file_name(key)
&& !body_hashed_file_name_matches(file_name, &key_stem)
{
continue;
}
let Ok(file_type) = entry.file_type() else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
if file_type.is_symlink() {
report.skipped = report.skipped.saturating_add(1);
continue;
}
if !file_type.is_file() {
continue;
}
let Ok(metadata) = fs::symlink_metadata(&path) else {
report.skipped = report.skipped.saturating_add(1);
continue;
};
let byte_len = metadata.len();
report.bytes_before = report.bytes_before.saturating_add(byte_len);
bytes_current = bytes_current.saturating_add(byte_len);
let candidate = EvictionCandidate {
path,
byte_len,
stored_epoch_seconds: 0,
last_used_epoch_seconds: 0,
expired: false,
};
remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);
}
report.bytes_after = bytes_current;
Ok(report)
}
fn temp_path(
&self,
key: &str,
body_hash_prefix: &str,
stored_at_epoch_seconds: u64,
) -> PathBuf {
let process_id = std::process::id();
let temp_counter = TEMP_FILE_COUNTER.fetch_add(1, Ordering::Relaxed);
self.root.join(format!(
".{}.{}.{}.{}.{}.tmp",
cache_file_stem(key),
body_hash_prefix,
process_id,
stored_at_epoch_seconds,
temp_counter
))
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum PackL2CacheLookup {
Hit(PackL2CacheHit),
Miss(PackL2CacheMiss),
}
impl PackL2CacheLookup {
#[must_use]
pub const fn is_hit(&self) -> bool {
matches!(self, Self::Hit(_))
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct PackL2CacheHit {
pub key: String,
pub path: PathBuf,
pub stored_at_epoch_seconds: u64,
pub pack_json: JsonValue,
pub byte_len: u64,
pub compression: Option<PackL2CompressionHit>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionHit {
pub algorithm: String,
pub dictionary_id: Option<String>,
pub compressed_bytes: u64,
pub uncompressed_bytes: u64,
pub decompression_latency_ms: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CacheMiss {
pub key: String,
pub path: PathBuf,
pub reason: PackL2CacheMissReason,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PackL2CacheMissReason {
NotFound,
Expired {
stored_at_epoch_seconds: u64,
},
Corrupt(String),
BodyHashMismatch {
expected: String,
actual: String,
},
KeyMismatch {
stored_key: String,
},
TooLarge {
byte_len: u64,
max_entry_bytes: u64,
},
CompressionDictionaryMissing {
dictionary_id: String,
},
CompressionDictionaryCorrupt {
dictionary_id: String,
message: String,
},
CompressionDecode {
message: String,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2WriteReport {
pub key: String,
pub path: PathBuf,
pub byte_len: u64,
pub uncompressed_byte_len: u64,
pub compression: Option<PackL2CompressionWriteReport>,
pub outcome: PackL2WriteOutcome,
pub eviction: PackL2EvictionReport,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionWriteReport {
pub algorithm: String,
pub dictionary_id: Option<String>,
pub compressed_bytes: u64,
pub uncompressed_bytes: u64,
pub compression_latency_ms: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PackL2CompressionDictionary {
pub id: String,
pub byte_hash: String,
pub bytes: Vec<u8>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PackL2WriteOutcome {
Stored,
SkippedTooLarge { max_entry_bytes: u64 },
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct PackL2EvictionReport {
pub removed: u64,
pub skipped: u64,
pub bytes_before: u64,
pub bytes_removed: u64,
pub bytes_after: u64,
}
#[derive(Debug)]
pub enum PackL2CacheError {
Io {
path: PathBuf,
operation: &'static str,
source: io::Error,
},
Json {
path: PathBuf,
operation: &'static str,
source: serde_json::Error,
},
Compression {
operation: &'static str,
source: io::Error,
},
TimeBeforeUnixEpoch {
source: std::time::SystemTimeError,
},
}
impl fmt::Display for PackL2CacheError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Io {
path,
operation,
source,
} => write!(
formatter,
"failed to {operation} pack L2 cache path {}: {source}",
path.display()
),
Self::Json {
path,
operation,
source,
} => write!(
formatter,
"failed to {operation} pack L2 cache JSON at {}: {source}",
path.display()
),
Self::Compression { operation, source } => {
write!(
formatter,
"failed to {operation} pack L2 cache entry: {source}"
)
}
Self::TimeBeforeUnixEpoch { source } => {
write!(formatter, "system time predates Unix epoch: {source}")
}
}
}
}
impl std::error::Error for PackL2CacheError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io { source, .. } => Some(source),
Self::Json { source, .. } => Some(source),
Self::Compression { source, .. } => Some(source),
Self::TimeBeforeUnixEpoch { source } => Some(source),
}
}
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntryEnvelope {
schema: String,
stored_at_epoch_seconds: u64,
}
#[derive(Debug)]
struct DecodedPackL2CacheEntry {
key: String,
stored_at_epoch_seconds: u64,
pack_json: JsonValue,
compression: Option<PackL2CompressionHit>,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntry {
schema: String,
key: String,
stored_at_epoch_seconds: u64,
pack_json: JsonValue,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheEntryV2 {
schema: String,
key: String,
stored_at_epoch_seconds: u64,
compression: PackL2CacheCompressionPayload,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheCompressionPayload {
algorithm: String,
compressed_payload_base64: String,
compressed_byte_len: u64,
uncompressed_byte_len: u64,
uncompressed_hash: String,
#[serde(skip_serializing_if = "Option::is_none")]
dictionary: Option<PackL2CacheCompressionDictionaryRef>,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct PackL2CacheCompressionDictionaryRef {
dictionary_id: String,
dictionary_byte_hash: String,
#[serde(skip_serializing_if = "Option::is_none")]
dictionary_bytes_base64: Option<String>,
}
impl PackL2CacheCompressionDictionaryRef {
fn from_dictionary(dictionary: &PackL2CompressionDictionary) -> Self {
Self {
dictionary_id: dictionary.id.clone(),
dictionary_byte_hash: dictionary.byte_hash.clone(),
dictionary_bytes_base64: Some(BASE64_STANDARD.encode(&dictionary.bytes)),
}
}
}
#[derive(Debug)]
struct EvictionCandidate {
path: PathBuf,
byte_len: u64,
stored_epoch_seconds: u64,
last_used_epoch_seconds: u64,
expired: bool,
}
fn remove_cache_entry_best_effort(path: &Path) {
match fs::remove_file(path) {
Ok(()) => {}
Err(error) if error.kind() == io::ErrorKind::NotFound => {}
Err(_) => {}
}
}
fn remove_eviction_candidate_file(
candidate: &EvictionCandidate,
report: &mut PackL2EvictionReport,
bytes_current: &mut u64,
) {
match fs::remove_file(&candidate.path) {
Ok(()) => record_eviction_candidate_removed(candidate, report, bytes_current),
Err(error) if error.kind() == io::ErrorKind::NotFound => {
record_eviction_candidate_removed(candidate, report, bytes_current);
}
Err(_) => {
report.skipped = report.skipped.saturating_add(1);
}
}
}
fn record_eviction_candidate_removed(
candidate: &EvictionCandidate,
report: &mut PackL2EvictionReport,
bytes_current: &mut u64,
) {
report.removed = report.removed.saturating_add(1);
report.bytes_removed = report.bytes_removed.saturating_add(candidate.byte_len);
*bytes_current = bytes_current.saturating_sub(candidate.byte_len);
}
fn merge_cache_cleanup_reports(
duplicate_cleanup: PackL2EvictionReport,
eviction: PackL2EvictionReport,
) -> PackL2EvictionReport {
PackL2EvictionReport {
removed: duplicate_cleanup.removed.saturating_add(eviction.removed),
skipped: duplicate_cleanup.skipped.saturating_add(eviction.skipped),
bytes_before: eviction
.bytes_before
.saturating_add(duplicate_cleanup.bytes_removed),
bytes_removed: duplicate_cleanup
.bytes_removed
.saturating_add(eviction.bytes_removed),
bytes_after: eviction.bytes_after,
}
}
fn decode_pack_l2_cache_entry(
bytes: &[u8],
max_decompressed_entry_bytes: u64,
) -> Result<DecodedPackL2CacheEntry, PackL2CacheMissReason> {
let envelope = serde_json::from_slice::<PackL2CacheEntryEnvelope>(bytes)
.map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
match envelope.schema.as_str() {
PACK_L2_CACHE_ENTRY_SCHEMA_V1 => {
let entry = serde_json::from_slice::<PackL2CacheEntry>(bytes)
.map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
Ok(DecodedPackL2CacheEntry {
key: entry.key,
stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
pack_json: entry.pack_json,
compression: None,
})
}
PACK_L2_CACHE_ENTRY_SCHEMA_V2 => {
decode_compressed_pack_l2_cache_entry(bytes, max_decompressed_entry_bytes)
}
schema => Err(PackL2CacheMissReason::Corrupt(format!(
"unexpected schema {schema}"
))),
}
}
fn decode_compressed_pack_l2_cache_entry(
bytes: &[u8],
max_decompressed_entry_bytes: u64,
) -> Result<DecodedPackL2CacheEntry, PackL2CacheMissReason> {
let entry = serde_json::from_slice::<PackL2CacheEntryV2>(bytes)
.map_err(|error| PackL2CacheMissReason::Corrupt(error.to_string()))?;
if entry.compression.algorithm != PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1 {
return Err(PackL2CacheMissReason::Corrupt(format!(
"unsupported compression algorithm {}",
entry.compression.algorithm
)));
}
let compressed = BASE64_STANDARD
.decode(&entry.compression.compressed_payload_base64)
.map_err(|error| PackL2CacheMissReason::CompressionDecode {
message: format!("compressed payload is not base64: {error}"),
})?;
if compressed.len() as u64 != entry.compression.compressed_byte_len {
return Err(PackL2CacheMissReason::CompressionDecode {
message: format!(
"compressed byte length mismatch expected={} actual={}",
entry.compression.compressed_byte_len,
compressed.len()
),
});
}
let dictionary_bytes = decode_pack_l2_dictionary_bytes(entry.compression.dictionary.as_ref())?;
if entry.compression.uncompressed_byte_len > max_decompressed_entry_bytes {
return Err(PackL2CacheMissReason::CompressionDecode {
message: format!(
"uncompressed byte length {} exceeds the {max_decompressed_entry_bytes}-byte decompression cap",
entry.compression.uncompressed_byte_len
),
});
}
let capacity = usize::try_from(entry.compression.uncompressed_byte_len).map_err(|_| {
PackL2CacheMissReason::CompressionDecode {
message: format!(
"uncompressed byte length does not fit usize: {}",
entry.compression.uncompressed_byte_len
),
}
})?;
let decompress_start = Instant::now();
let uncompressed = zstd_decompress(&compressed, capacity, dictionary_bytes.as_deref())
.map_err(|source| PackL2CacheMissReason::CompressionDecode {
message: source.to_string(),
})?;
let decompression_latency_ms = elapsed_millis(decompress_start.elapsed());
if uncompressed.len() as u64 != entry.compression.uncompressed_byte_len {
return Err(PackL2CacheMissReason::CompressionDecode {
message: format!(
"uncompressed byte length mismatch expected={} actual={}",
entry.compression.uncompressed_byte_len,
uncompressed.len()
),
});
}
let actual_hash = blake3_hash(&uncompressed);
if actual_hash != entry.compression.uncompressed_hash {
return Err(PackL2CacheMissReason::CompressionDecode {
message: format!(
"uncompressed hash mismatch expected={} actual={actual_hash}",
entry.compression.uncompressed_hash
),
});
}
let pack_json = serde_json::from_slice::<JsonValue>(&uncompressed).map_err(|error| {
PackL2CacheMissReason::CompressionDecode {
message: format!("decompressed pack JSON is malformed: {error}"),
}
})?;
Ok(DecodedPackL2CacheEntry {
key: entry.key,
stored_at_epoch_seconds: entry.stored_at_epoch_seconds,
pack_json,
compression: Some(PackL2CompressionHit {
algorithm: entry.compression.algorithm,
dictionary_id: entry
.compression
.dictionary
.map(|dictionary| dictionary.dictionary_id),
compressed_bytes: compressed.len() as u64,
uncompressed_bytes: uncompressed.len() as u64,
decompression_latency_ms,
}),
})
}
fn decode_pack_l2_dictionary_bytes(
dictionary: Option<&PackL2CacheCompressionDictionaryRef>,
) -> Result<Option<Vec<u8>>, PackL2CacheMissReason> {
let Some(dictionary) = dictionary else {
return Ok(None);
};
let Some(encoded) = &dictionary.dictionary_bytes_base64 else {
return Err(PackL2CacheMissReason::CompressionDictionaryMissing {
dictionary_id: dictionary.dictionary_id.clone(),
});
};
let bytes = BASE64_STANDARD.decode(encoded).map_err(|error| {
PackL2CacheMissReason::CompressionDictionaryCorrupt {
dictionary_id: dictionary.dictionary_id.clone(),
message: format!("dictionary bytes are not base64: {error}"),
}
})?;
let actual_hash = blake3_hash(&bytes);
if actual_hash != dictionary.dictionary_byte_hash {
return Err(PackL2CacheMissReason::CompressionDictionaryCorrupt {
dictionary_id: dictionary.dictionary_id.clone(),
message: format!(
"dictionary byte hash mismatch expected={} actual={actual_hash}",
dictionary.dictionary_byte_hash
),
});
}
Ok(Some(bytes))
}
fn zstd_compress(
payload: &[u8],
dictionary: Option<&PackL2CompressionDictionary>,
) -> Result<Vec<u8>, PackL2CacheError> {
let mut compressor = match dictionary {
Some(dictionary) => {
zstd::bulk::Compressor::with_dictionary(PACK_L2_COMPRESSION_LEVEL, &dictionary.bytes)
}
None => zstd::bulk::Compressor::new(PACK_L2_COMPRESSION_LEVEL),
}
.map_err(|source| PackL2CacheError::Compression {
operation: "initialize_compressor",
source,
})?;
compressor
.compress(payload)
.map_err(|source| PackL2CacheError::Compression {
operation: "compress",
source,
})
}
fn zstd_decompress(
payload: &[u8],
capacity: usize,
dictionary: Option<&[u8]>,
) -> io::Result<Vec<u8>> {
let mut decompressor = match dictionary {
Some(dictionary) => zstd::bulk::Decompressor::with_dictionary(dictionary)?,
None => zstd::bulk::Decompressor::new()?,
};
decompressor.decompress(payload, capacity)
}
fn elapsed_millis(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
fn cache_file_name(key: &str) -> String {
format!("{}.json", cache_file_stem(key))
}
fn cache_file_name_with_body_hash(key: &str, body_hash_prefix: &str) -> String {
format!("{}.{}.json", cache_file_stem(key), body_hash_prefix)
}
fn cache_file_stem(key: &str) -> String {
blake3::hash(key.as_bytes()).to_hex().to_string()
}
fn body_hash_prefix(bytes: &[u8]) -> String {
blake3::hash(bytes).to_hex()[..16].to_owned()
}
fn blake3_hash(bytes: &[u8]) -> String {
format!("blake3:{}", blake3::hash(bytes).to_hex())
}
fn body_hashed_file_name_matches(file_name: &str, key_stem: &str) -> bool {
let Some(rest) = file_name.strip_prefix(key_stem) else {
return false;
};
let Some(body_hash_prefix) = rest
.strip_prefix('.')
.and_then(|rest| rest.strip_suffix(".json"))
else {
return false;
};
body_hash_prefix.len() == 16
&& body_hash_prefix
.bytes()
.all(|byte| byte.is_ascii_hexdigit())
}
fn body_hash_prefix_from_path(path: &Path) -> Option<String> {
let file_name = path.file_name()?.to_str()?;
let body_hash_prefix = file_name
.strip_suffix(".json")?
.rsplit_once('.')?
.1
.to_owned();
(body_hash_prefix.len() == 16
&& body_hash_prefix
.bytes()
.all(|byte| byte.is_ascii_hexdigit()))
.then_some(body_hash_prefix)
}
fn is_expired(stored_at_epoch_seconds: u64, now_epoch_seconds: u64, max_age: Duration) -> bool {
now_epoch_seconds.saturating_sub(stored_at_epoch_seconds) > max_age.as_secs()
}
fn system_time_seconds(time: SystemTime) -> Result<u64, PackL2CacheError> {
time.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.map_err(|source| PackL2CacheError::TimeBeforeUnixEpoch { source })
}
fn touch_cache_entry_mtime_best_effort(path: &Path, epoch_seconds: u64) {
let modified_at = UNIX_EPOCH + Duration::from_secs(epoch_seconds);
let times = FileTimes::new().set_modified(modified_at);
if let Ok(file) = open_cache_entry_file_for_touch(path) {
let _ = file.set_times(times);
}
}
fn cache_entry_stored_at(path: &Path, max_entry_bytes: u64) -> Option<u64> {
if first_existing_symlink_component(path)
.ok()
.flatten()
.is_some()
{
return None;
}
let bytes = read_cache_entry_file(path, max_entry_bytes).ok()?;
serde_json::from_slice::<PackL2CacheEntryEnvelope>(&bytes)
.ok()
.map(|entry| entry.stored_at_epoch_seconds)
}
fn cache_entry_preference_epoch_seconds(path: &Path, max_entry_bytes: u64) -> u64 {
cache_entry_stored_at(path, max_entry_bytes)
.or_else(|| {
fs::symlink_metadata(path)
.ok()
.and_then(|metadata| metadata.modified().ok())
.and_then(|modified| system_time_seconds(modified).ok())
})
.unwrap_or(0)
}
fn ensure_cache_dir(path: &Path) -> Result<(), PackL2CacheError> {
ensure_no_symlink_components(path, "inspect_root")?;
fs::create_dir_all(path).map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "create_dir_all",
source,
})?;
ensure_no_symlink_components(path, "inspect_root")?;
#[cfg(unix)]
fs::set_permissions(path, fs::Permissions::from_mode(0o700)).map_err(|source| {
PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "set_permissions",
source,
}
})?;
Ok(())
}
fn ensure_no_symlink_components(
path: &Path,
operation: &'static str,
) -> Result<(), PackL2CacheError> {
if let Some(symlink_path) =
first_existing_symlink_component(path).map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation,
source,
})?
{
return Err(PackL2CacheError::Io {
path: path.to_path_buf(),
operation,
source: io::Error::new(
io::ErrorKind::PermissionDenied,
format!(
"pack L2 cache path traverses symbolic link {}",
symlink_path.display()
),
),
});
}
Ok(())
}
fn first_existing_symlink_component(path: &Path) -> io::Result<Option<PathBuf>> {
match crate::core::path_safety::first_existing_symlink_component(path) {
Err(error) if error.kind() == io::ErrorKind::NotADirectory => Ok(None),
other => other,
}
}
struct TempPackL2FileGuard<'a> {
path: &'a Path,
armed: bool,
}
impl<'a> TempPackL2FileGuard<'a> {
fn disarmed(path: &'a Path) -> Self {
Self { path, armed: false }
}
fn arm(&mut self) {
self.armed = true;
}
fn disarm(&mut self) {
self.armed = false;
}
}
impl Drop for TempPackL2FileGuard<'_> {
fn drop(&mut self) {
if self.armed {
let _ = fs::remove_file(self.path);
}
}
}
fn write_synced_file(path: &Path, bytes: &[u8]) -> Result<(), PackL2CacheError> {
let mut cleanup_guard = TempPackL2FileGuard::disarmed(path);
let mut file =
open_cache_temp_file_for_create(path).map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "open_temp",
source,
})?;
cleanup_guard.arm();
file.write_all(bytes)
.and_then(|()| file.sync_all())
.map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "write_sync",
source,
})?;
#[cfg(unix)]
file.set_permissions(fs::Permissions::from_mode(0o600))
.map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "set_file_permissions",
source,
})?;
cleanup_guard.disarm();
Ok(())
}
fn publish_cache_entry_temp_file(temp_path: &Path, path: &Path) -> Result<(), PackL2CacheError> {
ensure_no_symlink_components(path, "inspect_entry")?;
ensure_no_symlink_components(temp_path, "inspect_temp")?;
ensure_cache_temp_path_is_regular(temp_path)?;
fs::rename(temp_path, path).map_err(|source| {
let _ = fs::remove_file(temp_path);
PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "rename",
source,
}
})
}
fn ensure_cache_temp_path_is_regular(path: &Path) -> Result<(), PackL2CacheError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_file() => Ok(()),
Ok(_) => Err(PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "inspect_temp",
source: io::Error::new(
io::ErrorKind::InvalidInput,
"pack L2 cache temp path is not a regular file",
),
}),
Err(source) => Err(PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "inspect_temp",
source,
}),
}
}
fn sync_directory(path: &Path) -> Result<(), PackL2CacheError> {
open_cache_directory_for_sync(path)
.and_then(|directory| directory.sync_all())
.map_err(|source| PackL2CacheError::Io {
path: path.to_path_buf(),
operation: "sync_dir",
source,
})
}
fn read_cache_entry_file(path: &Path, max_entry_bytes: u64) -> io::Result<Vec<u8>> {
let file = open_cache_entry_file_for_read(path)?;
let mut bytes = Vec::new();
file.take(max_entry_bytes.saturating_add(1))
.read_to_end(&mut bytes)?;
Ok(bytes)
}
fn open_cache_entry_file_for_read(path: &Path) -> io::Result<File> {
let mut options = OpenOptions::new();
options.read(true);
configure_pack_l2_open_no_follow(&mut options);
options.open(path)
}
fn open_cache_temp_file_for_create(path: &Path) -> io::Result<File> {
let mut options = OpenOptions::new();
options.write(true).create_new(true);
configure_pack_l2_open_no_follow(&mut options);
options.open(path)
}
fn open_cache_entry_file_for_touch(path: &Path) -> io::Result<File> {
let mut options = OpenOptions::new();
options.write(true);
configure_pack_l2_open_no_follow(&mut options);
options.open(path)
}
fn open_cache_directory_for_sync(path: &Path) -> io::Result<File> {
let mut options = OpenOptions::new();
options.read(true);
configure_pack_l2_open_no_follow(&mut options);
options.open(path)
}
#[cfg(all(unix, not(any(target_os = "espidf", target_os = "horizon"))))]
fn configure_pack_l2_open_no_follow(options: &mut OpenOptions) {
use std::os::unix::fs::OpenOptionsExt;
options.custom_flags(rustix::fs::OFlags::NOFOLLOW.bits() as i32);
}
#[cfg(not(all(unix, not(any(target_os = "espidf", target_os = "horizon")))))]
fn configure_pack_l2_open_no_follow(_options: &mut OpenOptions) {}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
type TestResult = Result<(), String>;
fn cache(
max_bytes: u64,
max_age: Duration,
) -> Result<(tempfile::TempDir, PackL2Cache), String> {
cache_with_options(PackL2CacheOptions::new(max_bytes, max_age))
}
fn cache_with_options(
options: PackL2CacheOptions,
) -> Result<(tempfile::TempDir, PackL2Cache), String> {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(temp.path().join("pack-l2"), options);
Ok((temp, cache))
}
fn hit_json(lookup: PackL2CacheLookup) -> Result<JsonValue, String> {
match lookup {
PackL2CacheLookup::Hit(hit) => Ok(hit.pack_json),
PackL2CacheLookup::Miss(miss) => {
Err(format!("expected hit, got miss: {:?}", miss.reason))
}
}
}
fn raw_entry_bytes(
key: &str,
pack_json: JsonValue,
stored_at_epoch_seconds: u64,
) -> Result<Vec<u8>, String> {
serde_json::to_vec(&PackL2CacheEntry {
schema: PACK_L2_CACHE_ENTRY_SCHEMA_V1.to_owned(),
key: key.to_owned(),
stored_at_epoch_seconds,
pack_json,
})
.map_err(|error| error.to_string())
}
fn write_raw_entry(cache: &PackL2Cache, key: &str, bytes: &[u8]) -> Result<PathBuf, String> {
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let path = cache.entry_path_for_body_hash(key, &body_hash_prefix(bytes));
fs::write(&path, bytes).map_err(|error| error.to_string())?;
Ok(path)
}
fn raw_compressed_entry_bytes(
key: &str,
payload: PackL2CacheCompressionPayload,
stored_at_epoch_seconds: u64,
) -> Result<Vec<u8>, String> {
serde_json::to_vec(&PackL2CacheEntryV2 {
schema: PACK_L2_CACHE_ENTRY_SCHEMA_V2.to_owned(),
key: key.to_owned(),
stored_at_epoch_seconds,
compression: payload,
})
.map_err(|error| error.to_string())
}
fn modified_epoch_seconds(path: &Path) -> Result<u64, String> {
let modified = fs::metadata(path)
.map_err(|error| error.to_string())?
.modified()
.map_err(|error| error.to_string())?;
system_time_seconds(modified).map_err(|error| error.to_string())
}
#[test]
fn temp_pack_l2_file_guard_removes_armed_orphan() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let path = temp.path().join("entry.tmp");
fs::write(&path, b"orphan").map_err(|error| error.to_string())?;
{
let mut guard = TempPackL2FileGuard::disarmed(&path);
guard.arm();
}
assert!(
!path.exists(),
"armed pack L2 temp guard should remove an owned orphan"
);
Ok(())
}
#[test]
fn temp_pack_l2_file_guard_disarm_preserves_success_temp() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let path = temp.path().join("entry.tmp");
fs::write(&path, b"ready-to-publish").map_err(|error| error.to_string())?;
{
let mut guard = TempPackL2FileGuard::disarmed(&path);
guard.arm();
guard.disarm();
}
assert!(
path.exists(),
"disarmed pack L2 temp guard must preserve the successful temp file"
);
assert_eq!(
fs::read(&path).map_err(|error| error.to_string())?,
b"ready-to-publish",
"disarmed guard must not change temp file contents"
);
Ok(())
}
#[test]
fn temp_pack_l2_file_guard_unarmed_ignores_missing_path() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let path = temp.path().join("never-created.tmp");
{
let _guard = TempPackL2FileGuard::disarmed(&path);
}
assert!(
!path.exists(),
"unarmed pack L2 temp guard must not create or remove a missing path"
);
Ok(())
}
#[test]
fn default_options_use_pass2_size_limits() {
let options = PackL2CacheOptions::default();
assert_eq!(
options.max_bytes, DEFAULT_MAX_BYTES,
"default cache cap should stay at the pass-2 256 MiB budget"
);
assert_eq!(
options.max_entry_bytes, DEFAULT_MAX_ENTRY_BYTES,
"default per-entry cap should keep pathological packs out of L2"
);
}
#[test]
fn happy_path_roundtrip_returns_stored_pack_json() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let pack = json!({"hash": "blake3:test", "items": [{"id": "mem_1"}]});
let report = cache
.put_at("blake3:key-a", &pack, 100)
.map_err(|error| error.to_string())?;
assert!(
report.path.exists(),
"write should publish final cache file"
);
let file_name = report
.path
.file_name()
.and_then(|file_name| file_name.to_str())
.ok_or_else(|| "cache path should have a UTF-8 file name".to_owned())?;
let parts = file_name
.strip_suffix(".json")
.ok_or_else(|| format!("cache file should end in .json: {file_name}"))?
.split('.')
.collect::<Vec<_>>();
assert_eq!(parts.len(), 2, "cache file should have key and body hash");
assert_eq!(parts[0].len(), 64, "key hash should be full BLAKE3 hex");
assert_eq!(parts[1].len(), 16, "body hash should be truncated hex");
assert!(
body_hashed_file_name_matches(file_name, &cache_file_stem("blake3:key-a")),
"cache file name should bind key hash and body hash"
);
let stored = hit_json(
cache
.get_at("blake3:key-a", 120)
.map_err(|error| error.to_string())?,
)?;
assert_eq!(stored, pack, "cache hit should preserve pack JSON exactly");
Ok(())
}
#[test]
fn repeated_writes_return_newest_stored_entry_not_lexical_first() -> TestResult {
let (_temp, cache) = cache(10_000, Duration::from_secs(10_000))?;
let key = "blake3:repeat-shadow";
let old_pack = json!({"payload": "old"});
let new_pack = json!({"payload": "new"});
let old_report = cache
.put_at(key, &old_pack, 100)
.map_err(|error| error.to_string())?;
let new_report = cache
.put_at(key, &new_pack, 200)
.map_err(|error| error.to_string())?;
assert!(
old_report.path < new_report.path,
"fixture should cover the old filename-ordering bug"
);
let lookup = cache.get_at(key, 210).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Hit(hit) => {
assert_eq!(hit.path, new_report.path);
assert_eq!(hit.stored_at_epoch_seconds, 200);
assert_eq!(hit.pack_json, new_pack);
}
PackL2CacheLookup::Miss(miss) => {
return Err(format!("newest duplicate-key entry should hit: {miss:?}"));
}
}
Ok(())
}
#[test]
fn same_second_compressed_rewrite_prunes_stale_same_key_entry() -> TestResult {
let (_temp, cache) = cache(10_000, Duration::from_secs(10_000))?;
let key = "blake3:same-second-repeat";
let old_pack = json!({"payload": "old"});
let new_pack = json!({"payload": "new"});
let old_report = cache
.put_compressed_at(key, &old_pack, 100)
.map_err(|error| error.to_string())?;
let new_report = cache
.put_compressed_at(key, &new_pack, 100)
.map_err(|error| error.to_string())?;
assert_ne!(
old_report.path, new_report.path,
"different same-second payloads should publish distinct body-hashed entries"
);
assert!(
!old_report.path.exists(),
"a same-second rewrite must remove the prior same-key entry"
);
assert!(
new_report.path.exists(),
"same-second rewrite must retain the newly published entry"
);
assert_eq!(
new_report.eviction.removed, 1,
"same-key cleanup should be counted in the write cleanup report"
);
let lookup = cache.get_at(key, 100).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Hit(hit) => {
assert_eq!(hit.path, new_report.path);
assert_eq!(hit.stored_at_epoch_seconds, 100);
assert_eq!(hit.pack_json, new_pack);
}
PackL2CacheLookup::Miss(miss) => {
return Err(format!(
"same-second rewrite should hit the retained entry: {miss:?}"
));
}
}
Ok(())
}
#[test]
fn compressed_v2_roundtrip_returns_stored_pack_json() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let pack = json!({
"schema": "ee.pack_l2.test_payload.v1",
"responseJson": "{\"schema\":\"ee.response.v2\",\"success\":true,\"data\":{\"items\":[\"mem_1\"]}}"
});
let report = cache
.put_compressed_at("blake3:compressed-key", &pack, 100)
.map_err(|error| error.to_string())?;
assert_eq!(report.outcome, PackL2WriteOutcome::Stored);
let compression = report
.compression
.as_ref()
.ok_or_else(|| "compressed write should report compression metadata".to_owned())?;
assert_eq!(compression.algorithm, PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1);
assert!(compression.compressed_bytes > 0);
assert!(compression.uncompressed_bytes > 0);
assert_eq!(report.uncompressed_byte_len, compression.uncompressed_bytes);
let lookup = cache
.get_at("blake3:compressed-key", 120)
.map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Hit(hit) => {
assert_eq!(hit.pack_json, pack);
let hit_compression = hit.compression.ok_or_else(|| {
"compressed hit should report compression metadata".to_owned()
})?;
assert_eq!(
hit_compression.algorithm,
PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1
);
assert_eq!(
hit_compression.compressed_bytes,
compression.compressed_bytes
);
assert_eq!(
hit_compression.uncompressed_bytes,
compression.uncompressed_bytes
);
}
PackL2CacheLookup::Miss(miss) => {
return Err(format!("compressed v2 entry should hit: {miss:?}"));
}
}
Ok(())
}
#[test]
fn empty_or_boundary_entry_at_exactly_max_entry_bytes_is_cached() -> TestResult {
let key = "blake3:exact-entry-cap";
let stored_at_epoch_seconds = 100;
let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_exact"}]});
let entry_len = raw_entry_bytes(key, pack.clone(), stored_at_epoch_seconds)?.len() as u64;
let (_temp, cache) = cache_with_options(
PackL2CacheOptions::new(4096, Duration::from_secs(60)).with_max_entry_bytes(entry_len),
)?;
let report = cache
.put_at(key, &pack, stored_at_epoch_seconds)
.map_err(|error| error.to_string())?;
assert_eq!(report.byte_len, entry_len);
assert_eq!(report.outcome, PackL2WriteOutcome::Stored);
assert!(
report.path.exists(),
"entry exactly at max_entry_bytes should be cached"
);
assert_eq!(
hit_json(cache.get_at(key, 120).map_err(|error| error.to_string())?)?,
pack
);
Ok(())
}
#[test]
fn compressed_v2_entry_at_max_entry_bytes_plus_one_is_skipped_with_event() -> TestResult {
let key = "blake3:compressed-oversized-entry";
let stored_at_epoch_seconds = 100;
let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]});
let (_temp, baseline_cache) =
cache_with_options(PackL2CacheOptions::new(4096, Duration::from_secs(60)))?;
let baseline_report = baseline_cache
.put_compressed_at(key, &pack, stored_at_epoch_seconds)
.map_err(|error| error.to_string())?;
let max_entry_bytes = baseline_report
.byte_len
.checked_sub(1)
.ok_or_else(|| "compressed test entry should have non-zero length".to_owned())?;
let (_temp, cache) = cache_with_options(
PackL2CacheOptions::new(4096, Duration::from_secs(60))
.with_max_entry_bytes(max_entry_bytes),
)?;
let report = cache
.put_compressed_at(key, &pack, stored_at_epoch_seconds)
.map_err(|error| error.to_string())?;
assert_eq!(report.byte_len, baseline_report.byte_len);
assert_eq!(
report.outcome,
PackL2WriteOutcome::SkippedTooLarge { max_entry_bytes }
);
assert!(
report.compression.is_some(),
"skipped compressed entries should still report compression accounting"
);
assert!(
!report.path.exists(),
"oversized compressed entries should not publish a cache file"
);
Ok(())
}
#[test]
fn empty_or_boundary_entry_at_max_entry_bytes_plus_one_is_skipped_with_event() -> TestResult {
let key = "blake3:oversized-entry";
let stored_at_epoch_seconds = 100;
let pack = json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]});
let entry_len = raw_entry_bytes(key, pack, stored_at_epoch_seconds)?.len() as u64;
let max_entry_bytes = entry_len
.checked_sub(1)
.ok_or_else(|| "test entry should have non-zero serialized length".to_owned())?;
let (_temp, cache) = cache_with_options(
PackL2CacheOptions::new(4096, Duration::from_secs(60))
.with_max_entry_bytes(max_entry_bytes),
)?;
let report = cache
.put_at(
key,
&json!({"hash": "entry-cap", "items": [{"id": "mem_oversized"}]}),
stored_at_epoch_seconds,
)
.map_err(|error| error.to_string())?;
assert_eq!(report.byte_len, entry_len);
assert_eq!(
report.outcome,
PackL2WriteOutcome::SkippedTooLarge { max_entry_bytes }
);
assert_eq!(
report.eviction,
PackL2EvictionReport::default(),
"skipped entries should not run write-through eviction"
);
assert!(
!report.path.exists(),
"oversized entries should not publish a cache file"
);
assert!(
matches!(
cache.get_at(key, 120).map_err(|error| error.to_string())?,
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason: PackL2CacheMissReason::NotFound,
..
})
),
"skipped oversized entries should behave like cold misses"
);
Ok(())
}
#[test]
fn happy_path_touch_on_read_advances_mtime() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let report = cache
.put_at("blake3:touch", &json!({"hash": "mtime"}), 100)
.map_err(|error| error.to_string())?;
let mtime_before = modified_epoch_seconds(&report.path)?;
assert_eq!(
mtime_before, 100,
"write path should seed mtime from the stored-at timestamp"
);
let lookup = cache
.get_at("blake3:touch", 150)
.map_err(|error| error.to_string())?;
assert!(
lookup.is_hit(),
"fresh entry should hit before touching mtime"
);
assert!(
modified_epoch_seconds(&report.path)? >= 150,
"read path should advance mtime for portable LRU accounting"
);
Ok(())
}
#[test]
fn empty_or_boundary_missing_key_returns_not_found_miss() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let lookup = cache
.get_at("blake3:missing", 100)
.map_err(|error| error.to_string())?;
assert_eq!(
lookup,
PackL2CacheLookup::Miss(PackL2CacheMiss {
key: "blake3:missing".to_owned(),
path: cache.entry_path("blake3:missing"),
reason: PackL2CacheMissReason::NotFound,
})
);
Ok(())
}
#[test]
fn empty_or_boundary_expired_entry_returns_expired_miss() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(10))?;
let report = cache
.put_at("blake3:key-expired", &json!({"hash": "old"}), 100)
.map_err(|error| error.to_string())?;
let lookup = cache
.get_at("blake3:key-expired", 111)
.map_err(|error| error.to_string())?;
assert_eq!(
lookup,
PackL2CacheLookup::Miss(PackL2CacheMiss {
key: "blake3:key-expired".to_owned(),
path: report.path,
reason: PackL2CacheMissReason::Expired {
stored_at_epoch_seconds: 100,
},
})
);
Ok(())
}
#[test]
fn error_or_invalid_oversized_existing_entry_is_removed_on_read() -> TestResult {
let key = "blake3:old-oversized-entry";
let pack = json!({"hash": "old-entry", "items": [{"id": "mem_old_oversized"}]});
let bytes = raw_entry_bytes(key, pack, 100)?;
let max_entry_bytes = (bytes.len() as u64)
.checked_sub(1)
.ok_or_else(|| "test entry should have non-zero serialized length".to_owned())?;
let (_temp, cache) = cache_with_options(
PackL2CacheOptions::new(4096, Duration::from_secs(60))
.with_max_entry_bytes(max_entry_bytes),
)?;
let path = write_raw_entry(&cache, key, &bytes)?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
assert_eq!(
lookup,
PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path: path.clone(),
reason: PackL2CacheMissReason::TooLarge {
byte_len: bytes.len() as u64,
max_entry_bytes,
},
})
);
assert!(
!path.exists(),
"oversized legacy entries should be invalidated under the current cap"
);
Ok(())
}
#[test]
fn error_or_invalid_corrupt_entry_returns_corrupt_miss() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let path = write_raw_entry(&cache, "blake3:corrupt", b"{not-json")?;
let lookup = cache
.get_at("blake3:corrupt", 100)
.map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Miss(miss) => {
assert!(
matches!(miss.reason, PackL2CacheMissReason::Corrupt(_)),
"corrupt JSON should be a typed miss"
);
}
PackL2CacheLookup::Hit(_) => return Err("corrupt entry must not hit".to_owned()),
}
assert!(
!path.exists(),
"corrupt cache entry should be invalidated after a typed miss"
);
Ok(())
}
#[test]
fn error_or_invalid_corrupt_candidate_does_not_mask_valid_fallback() -> TestResult {
let key = "blake3:multi-candidate";
let pack = json!({"hash": "valid-fallback", "items": [{"id": "mem_valid"}]});
let valid_bytes = raw_entry_bytes(key, pack.clone(), 100)?;
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let corrupt_bytes = b"{not-json";
let corrupt_path = cache.entry_path_for_body_hash(key, &body_hash_prefix(corrupt_bytes));
fs::write(&corrupt_path, corrupt_bytes).map_err(|error| error.to_string())?;
let valid_path = cache.entry_path(key);
fs::write(&valid_path, valid_bytes).map_err(|error| error.to_string())?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Hit(hit) => {
assert_eq!(hit.path, valid_path);
assert_eq!(hit.pack_json, pack);
}
PackL2CacheLookup::Miss(miss) => {
return Err(format!(
"valid fallback should hit after bad candidate: {miss:?}"
));
}
}
assert!(
!corrupt_path.exists(),
"bad body-hash candidate should be invalidated before trying the valid fallback"
);
Ok(())
}
#[test]
fn compressed_v2_missing_dictionary_returns_typed_miss_and_removes_entry() -> TestResult {
let key = "blake3:missing-dictionary";
let payload = PackL2CacheCompressionPayload {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
compressed_payload_base64: BASE64_STANDARD.encode(b"not-used-before-dictionary-check"),
compressed_byte_len: b"not-used-before-dictionary-check".len() as u64,
uncompressed_byte_len: 128,
uncompressed_hash: blake3_hash(b"not-present"),
dictionary: Some(PackL2CacheCompressionDictionaryRef {
dictionary_id: "zstd_dict_missing".to_owned(),
dictionary_byte_hash: "blake3:missing".to_owned(),
dictionary_bytes_base64: None,
}),
};
let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let path = write_raw_entry(&cache, key, &bytes)?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
assert_eq!(
lookup,
PackL2CacheLookup::Miss(PackL2CacheMiss {
key: key.to_owned(),
path: path.clone(),
reason: PackL2CacheMissReason::CompressionDictionaryMissing {
dictionary_id: "zstd_dict_missing".to_owned()
},
})
);
assert!(
!path.exists(),
"missing-dictionary compressed entries should be invalidated"
);
Ok(())
}
#[test]
fn compressed_v2_corrupt_dictionary_returns_typed_miss_and_removes_entry() -> TestResult {
let key = "blake3:corrupt-dictionary";
let dictionary_bytes = b"dictionary bytes with the wrong recorded hash";
let payload = PackL2CacheCompressionPayload {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
compressed_payload_base64: BASE64_STANDARD.encode(b"not-used-before-dictionary-check"),
compressed_byte_len: b"not-used-before-dictionary-check".len() as u64,
uncompressed_byte_len: 128,
uncompressed_hash: blake3_hash(b"not-present"),
dictionary: Some(PackL2CacheCompressionDictionaryRef {
dictionary_id: "zstd_dict_corrupt".to_owned(),
dictionary_byte_hash: blake3_hash(b"different dictionary bytes"),
dictionary_bytes_base64: Some(BASE64_STANDARD.encode(dictionary_bytes)),
}),
};
let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let path = write_raw_entry(&cache, key, &bytes)?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason:
PackL2CacheMissReason::CompressionDictionaryCorrupt {
dictionary_id,
message,
},
..
}) => {
assert_eq!(dictionary_id, "zstd_dict_corrupt");
assert!(
message.contains("dictionary byte hash mismatch"),
"corrupt dictionary miss should explain the hash mismatch: {message}"
);
}
other => {
return Err(format!(
"corrupt dictionary should return a typed miss; got {other:?}"
));
}
}
assert!(
!path.exists(),
"corrupt-dictionary compressed entries should be invalidated"
);
Ok(())
}
#[test]
fn compressed_v2_oversized_uncompressed_length_uses_configured_entry_cap() -> TestResult {
let key = "blake3:oversized-uncompressed";
let max_entry_bytes = 1024_u64;
let payload = PackL2CacheCompressionPayload {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
compressed_payload_base64: BASE64_STANDARD.encode(b"not-a-zstd-frame"),
compressed_byte_len: b"not-a-zstd-frame".len() as u64,
uncompressed_byte_len: max_entry_bytes.saturating_add(1),
uncompressed_hash: blake3_hash(b"not-present"),
dictionary: None,
};
let bytes = raw_compressed_entry_bytes(key, payload, 100)?;
assert!(
bytes.len() as u64 <= max_entry_bytes,
"test fixture envelope must fit under max_entry_bytes so the decompression cap is exercised"
);
let (_temp, cache) = cache_with_options(
PackL2CacheOptions::new(4096, Duration::from_secs(60))
.with_max_entry_bytes(max_entry_bytes),
)?;
let path = write_raw_entry(&cache, key, &bytes)?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason: PackL2CacheMissReason::CompressionDecode { message },
..
}) => {
assert!(
message.contains("decompression cap"),
"oversized compressed miss should cite decompression cap: {message}"
);
}
other => {
return Err(format!(
"oversized compressed entry should return a typed miss; got {other:?}"
));
}
}
assert!(
!path.exists(),
"oversized compressed entries should be invalidated"
);
Ok(())
}
#[test]
fn compressed_v2_corrupt_body_does_not_mask_valid_fallback() -> TestResult {
let key = "blake3:compressed-multi-candidate";
let pack = json!({"hash": "valid-fallback", "items": [{"id": "mem_valid"}]});
let valid_bytes = raw_entry_bytes(key, pack.clone(), 100)?;
let payload = PackL2CacheCompressionPayload {
algorithm: PACK_L2_COMPRESSION_ALGORITHM_ZSTD_V1.to_owned(),
compressed_payload_base64: BASE64_STANDARD.encode(b"not-a-zstd-frame"),
compressed_byte_len: b"not-a-zstd-frame".len() as u64,
uncompressed_byte_len: 64,
uncompressed_hash: blake3_hash(b"not-a-json-payload"),
dictionary: None,
};
let corrupt_bytes = raw_compressed_entry_bytes(key, payload, 100)?;
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let corrupt_path = cache.entry_path_for_body_hash(key, &body_hash_prefix(&corrupt_bytes));
fs::write(&corrupt_path, corrupt_bytes).map_err(|error| error.to_string())?;
let valid_path = cache.entry_path(key);
fs::write(&valid_path, valid_bytes).map_err(|error| error.to_string())?;
let lookup = cache.get_at(key, 120).map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Hit(hit) => {
assert_eq!(hit.path, valid_path);
assert_eq!(hit.pack_json, pack);
assert!(
hit.compression.is_none(),
"valid fallback in this fixture is a legacy v1 entry"
);
}
PackL2CacheLookup::Miss(miss) => {
return Err(format!(
"valid fallback should hit after bad compressed candidate: {miss:?}"
));
}
}
assert!(
!corrupt_path.exists(),
"bad compressed candidate should be invalidated before trying the valid fallback"
);
Ok(())
}
#[test]
fn error_or_invalid_body_hash_mismatch_removes_entry() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let path = cache.entry_path_for_body_hash("blake3:tampered", "0000000000000000");
fs::write(&path, b"{\"schema\":\"tampered\"}").map_err(|error| error.to_string())?;
let lookup = cache
.get_at("blake3:tampered", 100)
.map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Miss(miss) => {
assert_eq!(
miss.reason,
PackL2CacheMissReason::BodyHashMismatch {
expected: "0000000000000000".to_owned(),
actual: body_hash_prefix(b"{\"schema\":\"tampered\"}"),
}
);
assert_eq!(miss.path, path);
}
PackL2CacheLookup::Hit(_) => {
return Err("body-hash mismatch must not hit".to_owned());
}
}
assert!(
!path.exists(),
"body-hash mismatch should remove the corrupted entry"
);
Ok(())
}
#[test]
fn error_or_invalid_key_mismatch_returns_miss() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
let original = raw_entry_bytes("blake3:original", json!({"hash": "mismatch"}), 100)?;
let path = write_raw_entry(&cache, "blake3:other", &original)?;
let lookup = cache
.get_at("blake3:other", 100)
.map_err(|error| error.to_string())?;
match lookup {
PackL2CacheLookup::Miss(miss) => assert_eq!(
miss.reason,
PackL2CacheMissReason::KeyMismatch {
stored_key: "blake3:original".to_owned()
}
),
PackL2CacheLookup::Hit(_) => return Err("mismatched key must not hit".to_owned()),
}
assert!(
!path.exists(),
"key-mismatched cache entry should be invalidated"
);
Ok(())
}
#[test]
fn error_or_invalid_unwritable_root_reports_io_error() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let file_root = temp.path().join("not-a-directory");
fs::write(&file_root, b"already a file").map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(file_root.clone(), PackL2CacheOptions::default());
let error = cache
.put_at("blake3:key", &json!({"hash": "nope"}), 100)
.expect_err("file root should not be writable as a cache directory");
match error {
PackL2CacheError::Io {
path,
operation: "create_dir_all",
..
} => assert_eq!(path, file_root),
other => return Err(format!("unexpected error: {other}")),
}
Ok(())
}
#[test]
fn error_or_invalid_symlink_scan_stops_cleanly_at_file_component() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let file_component = temp.path().join("not-a-directory");
fs::write(&file_component, b"already a file").map_err(|error| error.to_string())?;
let path_below_file = file_component.join("child.json");
let symlink = first_existing_symlink_component(&path_below_file)
.map_err(|error| error.to_string())?;
assert_eq!(
symlink, None,
"a non-directory component should stop the existing-prefix scan, not become an IO failure"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn error_or_invalid_put_rejects_symlinked_cache_root() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let real_root = temp.path().join("real-pack-l2");
fs::create_dir_all(&real_root).map_err(|error| error.to_string())?;
let linked_root = temp.path().join("pack-l2");
symlink(&real_root, &linked_root).map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(linked_root.clone(), PackL2CacheOptions::default());
let error = cache
.put_at("blake3:symlink-root", &json!({"hash": "unsafe"}), 100)
.expect_err("symlinked cache root should be rejected");
match error {
PackL2CacheError::Io {
path,
operation: "inspect_root",
source,
} => {
assert_eq!(path, linked_root);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
other => return Err(format!("unexpected error: {other}")),
}
assert!(
fs::read_dir(&real_root)
.map_err(|error| error.to_string())?
.next()
.is_none(),
"cache write must not publish through symlinked root"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn error_or_invalid_get_rejects_symlinked_cache_root() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let real_root = temp.path().join("real-pack-l2");
fs::create_dir_all(&real_root).map_err(|error| error.to_string())?;
let linked_root = temp.path().join("pack-l2");
symlink(&real_root, &linked_root).map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(linked_root.clone(), PackL2CacheOptions::default());
let error = cache
.get_at("blake3:symlink-root", 100)
.expect_err("symlinked cache root should be rejected before lookup");
match error {
PackL2CacheError::Io {
path,
operation: "inspect_root",
source,
} => {
assert_eq!(path, linked_root);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
other => return Err(format!("unexpected error: {other}")),
}
Ok(())
}
#[cfg(unix)]
#[test]
fn error_or_invalid_get_and_put_reject_symlinked_cache_entry() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(
temp.path().join("pack-l2"),
PackL2CacheOptions::new(4096, Duration::from_secs(60)),
);
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let outside_entry = temp.path().join("outside-entry.json");
fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
let linked_entry =
cache.entry_path_for_body_hash("blake3:linked-entry", "0000000000000000");
symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;
let get_error = cache
.get_at("blake3:linked-entry", 100)
.expect_err("symlinked final cache entry should not be read");
match get_error {
PackL2CacheError::Io {
path,
operation: "inspect_entry",
source,
} => {
assert_eq!(path, linked_entry);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
other => return Err(format!("unexpected get error: {other}")),
}
let overwrite_pack = json!({"hash": "overwrite"});
let overwrite_bytes = raw_entry_bytes("blake3:linked-entry", overwrite_pack.clone(), 100)?;
let linked_write_entry = cache
.entry_path_for_body_hash("blake3:linked-entry", &body_hash_prefix(&overwrite_bytes));
symlink(&outside_entry, &linked_write_entry).map_err(|error| error.to_string())?;
let put_error = cache
.put_at("blake3:linked-entry", &overwrite_pack, 100)
.expect_err("symlinked final cache entry should not be overwritten");
match put_error {
PackL2CacheError::Io {
path,
operation: "inspect_entry",
source,
} => {
assert_eq!(path, linked_write_entry);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
other => return Err(format!("unexpected put error: {other}")),
}
assert_eq!(
fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
r#"{"schema":"outside"}"#,
"cache write must not overwrite a symlink target"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn cache_entry_final_read_open_rejects_symlinked_entry_path() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let outside_entry = temp.path().join("outside-entry.json");
fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
let linked_entry = temp.path().join("linked-entry.json");
symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;
let error = open_cache_entry_file_for_read(&linked_entry)
.expect_err("final cache entry read open must reject symlinks");
assert_ne!(
error.kind(),
io::ErrorKind::NotFound,
"final symlink read should fail because the path is a symlink"
);
assert_eq!(
fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
r#"{"schema":"outside"}"#,
"cache read helper must not follow the symlink target"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn cache_temp_final_create_open_rejects_symlinked_temp_path() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let outside_entry = temp.path().join("outside-temp.json");
fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
let linked_temp = temp.path().join("entry.tmp");
symlink(&outside_entry, &linked_temp).map_err(|error| error.to_string())?;
let error = open_cache_temp_file_for_create(&linked_temp)
.expect_err("final cache temp create open must reject symlinks");
assert_ne!(
error.kind(),
io::ErrorKind::NotFound,
"final symlink create should fail because the path is a symlink"
);
assert_eq!(
fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
r#"{"schema":"outside"}"#,
"cache temp create helper must not follow or truncate the symlink target"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn error_or_invalid_publish_rechecks_symlinked_final_entry() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(
temp.path().join("pack-l2"),
PackL2CacheOptions::new(4096, Duration::from_secs(60)),
);
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let pack_json = json!({"hash": "publish-recheck"});
let bytes = raw_entry_bytes("blake3:publish-recheck", pack_json, 100)?;
let body_hash = body_hash_prefix(&bytes);
let entry_path = cache.entry_path_for_body_hash("blake3:publish-recheck", &body_hash);
let temp_path = cache.temp_path("blake3:publish-recheck", &body_hash, 100);
write_synced_file(&temp_path, &bytes).map_err(|error| error.to_string())?;
let outside_entry = temp.path().join("outside-entry.json");
fs::write(&outside_entry, br#"{"schema":"outside"}"#).map_err(|error| error.to_string())?;
symlink(&outside_entry, &entry_path).map_err(|error| error.to_string())?;
let error = publish_cache_entry_temp_file(&temp_path, &entry_path)
.expect_err("symlinked final entry should be rejected before publish");
match error {
PackL2CacheError::Io {
path,
operation: "inspect_entry",
source,
} => {
assert_eq!(path, entry_path);
assert_eq!(source.kind(), io::ErrorKind::PermissionDenied);
}
other => return Err(format!("unexpected publish error: {other}")),
}
assert_eq!(
fs::read_to_string(&outside_entry).map_err(|error| error.to_string())?,
r#"{"schema":"outside"}"#,
"cache publish must not mutate the symlink target"
);
assert!(
temp_path.exists(),
"cache temp entry should remain available after rejected publish"
);
Ok(())
}
#[cfg(unix)]
#[test]
fn eviction_skips_symlinked_json_entries_without_following_targets() -> TestResult {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let cache = PackL2Cache::new(
temp.path().join("pack-l2"),
PackL2CacheOptions::new(0, Duration::from_secs(0)),
);
ensure_cache_dir(cache.root()).map_err(|error| error.to_string())?;
let outside_entry = temp.path().join("outside-entry.json");
fs::write(&outside_entry, br#"{"storedAtEpochSeconds":0}"#)
.map_err(|error| error.to_string())?;
let linked_entry = cache.root().join("linked.json");
symlink(&outside_entry, &linked_entry).map_err(|error| error.to_string())?;
let report = cache
.evict_best_effort_at(100)
.map_err(|error| error.to_string())?;
assert_eq!(report.skipped, 1, "symlink entries should be skipped");
assert_eq!(report.removed, 0, "symlink entries should not be removed");
assert!(
fs::symlink_metadata(&linked_entry)
.map_err(|error| error.to_string())?
.file_type()
.is_symlink(),
"cache eviction should leave the symlink entry untouched"
);
assert!(
outside_entry.exists(),
"cache eviction must not follow and remove a symlink target"
);
Ok(())
}
#[test]
fn eviction_removes_expired_entries_before_fresh_entries() -> TestResult {
let (_temp, cache) = cache(10_000, Duration::from_secs(10))?;
cache
.put_at("blake3:old", &json!({"payload": "old"}), 100)
.map_err(|error| error.to_string())?;
let fresh_report = cache
.put_at("blake3:fresh", &json!({"payload": "fresh"}), 120)
.map_err(|error| error.to_string())?;
assert_eq!(
fresh_report.eviction.removed, 1,
"one expired entry should be removed during the next write"
);
assert!(
matches!(
cache
.get_at("blake3:old", 120)
.map_err(|error| error.to_string())?,
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason: PackL2CacheMissReason::NotFound,
..
})
),
"old entry should be gone"
);
assert!(
cache
.get_at("blake3:fresh", 120)
.map_err(|error| error.to_string())?
.is_hit(),
"fresh entry should remain"
);
Ok(())
}
#[test]
fn eviction_reduces_cache_to_byte_cap_by_oldest_first() -> TestResult {
let (_temp, cache) = cache(170, Duration::from_secs(10_000))?;
cache
.put_at(
"blake3:first",
&json!({"payload": "aaaaaaaaaaaaaaaaaaaaaaaa"}),
100,
)
.map_err(|error| error.to_string())?;
cache
.put_at(
"blake3:second",
&json!({"payload": "bbbbbbbbbbbbbbbbbbbbbbbb"}),
200,
)
.map_err(|error| error.to_string())?;
let third_report = cache
.put_at(
"blake3:third",
&json!({"payload": "cccccccccccccccccccccccc"}),
300,
)
.map_err(|error| error.to_string())?;
let report = cache
.evict_best_effort_at(300)
.map_err(|error| error.to_string())?;
let removed_total = third_report.eviction.removed.saturating_add(report.removed);
assert!(
report.bytes_after <= cache.options().max_bytes,
"eviction should reduce byte usage below the configured cap"
);
assert!(
removed_total >= 1,
"at least one entry should be evicted by write-through or explicit eviction"
);
assert!(
matches!(
cache
.get_at("blake3:first", 300)
.map_err(|error| error.to_string())?,
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason: PackL2CacheMissReason::NotFound,
..
})
),
"oldest entry should be evicted first"
);
Ok(())
}
#[test]
fn eviction_uses_touched_mtime_for_lru_order() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let root = temp.path().join("pack-l2");
let writer = PackL2Cache::new(
root.clone(),
PackL2CacheOptions::new(u64::MAX, Duration::from_secs(10_000)),
);
let first = writer
.put_at("blake3:first", &json!({"payload": "first"}), 100)
.map_err(|error| error.to_string())?;
let _second = writer
.put_at("blake3:second", &json!({"payload": "second"}), 200)
.map_err(|error| error.to_string())?;
assert!(
writer
.get_at("blake3:first", 300)
.map_err(|error| error.to_string())?
.is_hit(),
"read should touch the first entry before size eviction"
);
let third = writer
.put_at("blake3:third", &json!({"payload": "third"}), 250)
.map_err(|error| error.to_string())?;
let evicting = PackL2Cache::new(
root,
PackL2CacheOptions::new(first.byte_len + third.byte_len, Duration::from_secs(10_000)),
);
let report = evicting
.evict_best_effort_at(300)
.map_err(|error| error.to_string())?;
assert_eq!(
report.removed, 1,
"size eviction should remove exactly one oldest LRU entry"
);
assert!(
matches!(
evicting
.get_at("blake3:second", 300)
.map_err(|error| error.to_string())?,
PackL2CacheLookup::Miss(PackL2CacheMiss {
reason: PackL2CacheMissReason::NotFound,
..
})
),
"untouched second entry should be evicted before the touched first entry"
);
assert!(
evicting
.get_at("blake3:first", 300)
.map_err(|error| error.to_string())?
.is_hit(),
"touched first entry should survive LRU eviction"
);
assert!(
evicting
.get_at("blake3:third", 300)
.map_err(|error| error.to_string())?
.is_hit(),
"newest third entry should survive LRU eviction"
);
Ok(())
}
#[test]
fn concurrent_eviction_enoent_treated_as_success() -> TestResult {
let temp = tempfile::tempdir().map_err(|error| error.to_string())?;
let candidate = EvictionCandidate {
path: temp.path().join("already-evicted.json"),
byte_len: 128,
stored_epoch_seconds: 100,
last_used_epoch_seconds: 100,
expired: true,
};
let mut report = PackL2EvictionReport {
bytes_before: 256,
..PackL2EvictionReport::default()
};
let mut bytes_current = report.bytes_before;
remove_eviction_candidate_file(&candidate, &mut report, &mut bytes_current);
assert_eq!(
report.skipped, 0,
"peer-removed cache entries should not count as skipped"
);
assert_eq!(
report.removed, 1,
"peer-removed cache entries count as logically removed"
);
assert_eq!(
report.bytes_removed, candidate.byte_len,
"logical byte accounting should include the raced entry"
);
assert_eq!(
bytes_current, 128,
"current byte estimate should shrink after ENOENT"
);
Ok(())
}
#[test]
fn happy_path_cache_directory_uses_private_permissions() -> TestResult {
let (_temp, cache) = cache(4096, Duration::from_secs(60))?;
cache
.put_at("blake3:key-a", &json!({"hash": "perms"}), 100)
.map_err(|error| error.to_string())?;
#[cfg(unix)]
{
let mode = fs::metadata(cache.root())
.map_err(|error| error.to_string())?
.permissions()
.mode()
& 0o777;
assert_eq!(mode, 0o700, "cache directory should be owner-only");
}
Ok(())
}
}