use std::collections::VecDeque;
use std::fs;
use std::io::Read;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::RwLock;
use rustc_hash::FxHasher;
use std::hash::Hasher;
use crate::normalize::NormalizedSchema;
static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
const CACHE_VERSION: u32 = 2;
const MAX_MEMORY_ENTRIES: usize = 1000;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct CacheEntry {
schema: NormalizedSchema,
original_bytes: Vec<u8>,
}
#[derive(Debug, Default)]
pub struct Cache {
inner: std::collections::HashMap<u64, CacheEntry>,
order: VecDeque<u64>,
}
impl Cache {
pub fn new() -> Self {
Self::default()
}
pub fn get(&self, hash: u64, bytes: &[u8]) -> Option<&NormalizedSchema> {
let entry = self.inner.get(&hash)?;
if entry.original_bytes == bytes {
Some(&entry.schema)
} else {
None
}
}
pub fn insert(&mut self, hash: u64, bytes: Vec<u8>, schema: NormalizedSchema) {
let is_new = !self.inner.contains_key(&hash);
if is_new && self.inner.len() >= MAX_MEMORY_ENTRIES {
if let Some(oldest) = self.order.pop_front() {
self.inner.remove(&oldest);
}
}
self.inner.insert(
hash,
CacheEntry {
schema,
original_bytes: bytes,
},
);
if is_new {
self.order.push_back(hash);
}
}
pub fn clear(&mut self) {
self.inner.clear();
self.order.clear();
}
}
#[derive(Debug)]
pub struct DiskCache {
memory: RwLock<Cache>,
cache_dir: Option<PathBuf>,
evict_lock: std::sync::Mutex<()>,
}
impl Default for DiskCache {
fn default() -> Self {
Self::new()
}
}
impl DiskCache {
pub fn with_cache_dir(cache_dir: PathBuf) -> Self {
let isolated = cache_dir.join(format!("pid-{}", std::process::id()));
if let Err(e) = fs::create_dir_all(&isolated) {
eprintln!(
"warning: failed to create cache directory '{}': {}",
isolated.display(),
e
);
return Self {
memory: RwLock::new(Cache::new()),
cache_dir: None,
evict_lock: std::sync::Mutex::new(()),
};
}
Self {
memory: RwLock::new(Cache::new()),
cache_dir: Some(isolated),
evict_lock: std::sync::Mutex::new(()),
}
}
pub fn new() -> Self {
let candidate =
dirs::cache_dir().map(|d| d.join(format!("schemalint-{}", std::process::id())));
match candidate {
Some(ref dir) => {
if let Err(e) = fs::create_dir_all(dir) {
eprintln!(
"warning: failed to create cache directory '{}': {}",
dir.display(),
e
);
return Self {
memory: RwLock::new(Cache::new()),
cache_dir: None,
evict_lock: std::sync::Mutex::new(()),
};
}
Self {
memory: RwLock::new(Cache::new()),
cache_dir: candidate,
evict_lock: std::sync::Mutex::new(()),
}
}
None => Self {
memory: RwLock::new(Cache::new()),
cache_dir: None,
evict_lock: std::sync::Mutex::new(()),
},
}
}
pub fn get(&self, hash: u64, bytes: &[u8]) -> Option<NormalizedSchema> {
{
let memory = self.memory.read().unwrap();
if let Some(cached) = memory.get(hash, bytes) {
return Some(cached.clone());
}
}
let dir = self.cache_dir.as_ref()?;
let path = dir.join(format!("{:016x}.bin", hash));
let mut file = fs::File::open(&path).ok()?;
let mut buf = Vec::new();
file.read_to_end(&mut buf).ok()?;
if buf.len() < 4 {
return None;
}
let version = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if version != CACHE_VERSION {
return None;
}
let entry: CacheEntry = serde_json::from_slice(&buf[4..]).ok()?;
if entry.original_bytes != bytes {
return None;
}
self.memory.write().unwrap().insert(
hash,
entry.original_bytes.clone(),
entry.schema.clone(),
);
Some(entry.schema)
}
pub fn insert(&self, hash: u64, bytes: Vec<u8>, schema: NormalizedSchema) {
self.memory
.write()
.unwrap()
.insert(hash, bytes.clone(), schema.clone());
if let Some(ref dir) = self.cache_dir {
let final_path = dir.join(format!("{:016x}.bin", hash));
let counter = TMP_COUNTER.fetch_add(1, Ordering::Relaxed);
let tmp_name = format!("{:016x}.bin.tmp.{}.{}", hash, std::process::id(), counter);
let tmp_path = dir.join(&tmp_name);
let mut buf = Vec::new();
buf.extend_from_slice(&CACHE_VERSION.to_le_bytes());
let entry = CacheEntry {
schema,
original_bytes: bytes,
};
match serde_json::to_vec(&entry) {
Ok(serialized) => {
buf.extend_from_slice(&serialized);
if let Err(e) = fs::write(&tmp_path, &buf) {
eprintln!(
"warning: failed to write cache temp file '{}': {}",
tmp_path.display(),
e
);
let _ = fs::remove_file(&tmp_path);
} else {
if let Err(e) = fs::rename(&tmp_path, &final_path) {
eprintln!(
"warning: failed to rename cache file '{}' -> '{}': {}",
tmp_path.display(),
final_path.display(),
e
);
let _ = fs::remove_file(&tmp_path);
}
}
}
Err(e) => {
eprintln!("warning: failed to serialize schema for cache: {}", e);
}
}
self.evict_if_needed(dir);
}
}
#[cfg(test)]
pub(crate) fn cache_dir(&self) -> Option<&PathBuf> {
self.cache_dir.as_ref()
}
fn evict_if_needed(&self, dir: &PathBuf) {
let _guard = self.evict_lock.lock().unwrap_or_else(|e| e.into_inner());
let entries = match fs::read_dir(dir) {
Ok(e) => e,
Err(e) => {
eprintln!(
"warning: failed to read cache directory '{}': {}",
dir.display(),
e
);
return;
}
};
let mut files: Vec<(fs::DirEntry, std::time::SystemTime)> = Vec::new();
for entry in entries.filter_map(|e| e.ok()) {
let name = entry.file_name();
let name_str = name.to_string_lossy();
if !name_str.ends_with(".bin") || name_str.contains(".tmp.") {
continue;
}
let Ok(meta) = entry.metadata() else { continue };
let Ok(mtime) = meta.modified() else { continue };
files.push((entry, mtime));
}
if files.len() > 1000 {
files.sort_by_key(|a| a.1);
let to_remove = files.len() - 1000;
for (entry, _) in files.into_iter().take(to_remove) {
if let Err(e) = fs::remove_file(entry.path()) {
eprintln!("warning: failed to remove stale cache file: {}", e);
}
}
}
}
}
pub fn hash_bytes(bytes: &[u8]) -> u64 {
let mut hasher = FxHasher::default();
hasher.write(bytes);
hasher.finish()
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn make_schema() -> NormalizedSchema {
crate::normalize::normalize(serde_json::Value::Bool(true))
.expect("normalizing `true` must succeed")
}
fn schema_bytes(val: &serde_json::Value) -> Vec<u8> {
serde_json::to_vec(val).expect("serialization of test value must succeed")
}
#[test]
fn test_disk_round_trip() {
let tmp = TempDir::new().unwrap();
let raw = serde_json::json!({"type": "string"});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let schema = make_schema();
let cache1 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
cache1.insert(hash, bytes.clone(), schema);
let cache2 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let result = cache2.get(hash, &bytes);
assert!(
result.is_some(),
"disk round-trip: expected a cache hit on the second DiskCache instance"
);
}
#[test]
fn test_disk_round_trip_collision_rejected() {
let tmp = TempDir::new().unwrap();
let raw = serde_json::json!({"type": "string"});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let schema = make_schema();
let cache1 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
cache1.insert(hash, bytes.clone(), schema);
let different_bytes = schema_bytes(&serde_json::json!({"type": "integer"}));
let cache2 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let result = cache2.get(hash, &different_bytes);
assert!(
result.is_none(),
"collision guard: bytes mismatch must result in a cache miss"
);
}
#[test]
fn test_atomic_write_no_tmp_files_remain() {
let tmp = TempDir::new().unwrap();
let raw = serde_json::json!({"type": "object"});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let schema = make_schema();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
cache.insert(hash, bytes, schema);
let cache_dir = cache.cache_dir().expect("cache dir must be set");
let entries: Vec<_> = fs::read_dir(cache_dir)
.expect("read_dir must succeed")
.filter_map(|e| e.ok())
.collect();
let tmp_files: Vec<_> = entries
.iter()
.filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
.collect();
assert!(
tmp_files.is_empty(),
"expected no leftover .tmp files after successful insert, found: {:?}",
tmp_files.iter().map(|e| e.file_name()).collect::<Vec<_>>()
);
let bin_files: Vec<_> = entries
.iter()
.filter(|e| e.file_name().to_string_lossy().ends_with(".bin"))
.collect();
assert_eq!(
bin_files.len(),
1,
"expected exactly one .bin cache file, found {:?}",
bin_files.iter().map(|e| e.file_name()).collect::<Vec<_>>()
);
}
#[test]
fn test_eviction_trims_to_limit() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let cache_dir = cache.cache_dir().expect("cache dir must be set").clone();
let total = 1_001usize;
for i in 0..total {
let raw = serde_json::json!({"__test_index": i});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let schema = make_schema();
cache.insert(hash, bytes, schema);
}
let file_count = fs::read_dir(&cache_dir)
.expect("read_dir must succeed")
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().ends_with(".bin"))
.count();
assert_eq!(
file_count, 1_000,
"eviction must trim disk cache to exactly 1 000 entries, found {}",
file_count
);
}
#[test]
fn test_memory_cache_get_miss_absent_hash() {
let cache = Cache::new();
let bytes = b"hello";
let result = cache.get(0xdeadbeef, bytes);
assert!(result.is_none(), "get on absent hash must return None");
}
#[test]
fn test_memory_cache_get_miss_bytes_mismatch() {
let mut cache = Cache::new();
let bytes_a = b"schema-a".to_vec();
let hash = hash_bytes(&bytes_a);
cache.insert(hash, bytes_a, make_schema());
let result = cache.get(hash, b"schema-b");
assert!(
result.is_none(),
"bytes mismatch must result in None even when hash matches"
);
}
#[test]
fn test_memory_cache_get_hit() {
let mut cache = Cache::new();
let bytes = b"schema-x".to_vec();
let hash = hash_bytes(&bytes);
cache.insert(hash, bytes.clone(), make_schema());
let result = cache.get(hash, &bytes);
assert!(result.is_some(), "get with matching hash+bytes must hit");
}
#[test]
fn test_memory_cache_reinsertion_does_not_grow_order() {
let mut cache = Cache::new();
let bytes = b"same".to_vec();
let hash = hash_bytes(&bytes);
cache.insert(hash, bytes.clone(), make_schema());
let order_len_after_first = cache.order.len();
cache.insert(hash, bytes.clone(), make_schema());
assert_eq!(
cache.order.len(),
order_len_after_first,
"re-inserting an existing hash must not append to the order deque"
);
assert!(cache.get(hash, &bytes).is_some());
}
#[test]
fn test_memory_cache_lru_eviction() {
let mut cache = Cache::new();
let sentinel_bytes = b"sentinel".to_vec();
let sentinel_hash = hash_bytes(&sentinel_bytes);
cache.insert(sentinel_hash, sentinel_bytes.clone(), make_schema());
for i in 0..MAX_MEMORY_ENTRIES {
let bytes = format!("entry-{}", i).into_bytes();
let hash = hash_bytes(&bytes);
cache.insert(hash, bytes, make_schema());
}
assert!(
cache.get(sentinel_hash, &sentinel_bytes).is_none(),
"sentinel entry must have been evicted after MAX_MEMORY_ENTRIES overflow"
);
assert_eq!(
cache.inner.len(),
MAX_MEMORY_ENTRIES,
"inner map size must equal MAX_MEMORY_ENTRIES after eviction"
);
}
#[test]
fn test_memory_cache_clear() {
let mut cache = Cache::new();
let bytes = b"to-be-cleared".to_vec();
let hash = hash_bytes(&bytes);
cache.insert(hash, bytes.clone(), make_schema());
assert!(
cache.get(hash, &bytes).is_some(),
"pre-clear: must be a hit"
);
cache.clear();
assert!(
cache.get(hash, &bytes).is_none(),
"post-clear: must be a miss"
);
assert!(
cache.inner.is_empty(),
"inner map must be empty after clear"
);
assert!(
cache.order.is_empty(),
"order deque must be empty after clear"
);
}
#[test]
fn test_with_cache_dir_fallback_when_path_is_file() {
let tmp = TempDir::new().unwrap();
let file_path = tmp.path().join("not_a_dir");
fs::write(&file_path, b"I am a file").expect("write must succeed");
let cache = DiskCache::with_cache_dir(file_path);
assert!(
cache.cache_dir().is_none(),
"cache_dir must be None when directory creation fails"
);
let bytes = b"mem-only".to_vec();
let hash = hash_bytes(&bytes);
cache.insert(hash, bytes.clone(), make_schema());
assert!(
cache.get(hash, &bytes).is_some(),
"memory-only fallback must still return hits"
);
}
#[test]
fn test_disk_get_version_mismatch_is_miss() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let cache_dir = cache.cache_dir().expect("cache dir must be set").clone();
let raw = serde_json::json!({"type": "boolean"});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let bad_version: u32 = 99;
let mut buf = bad_version.to_le_bytes().to_vec();
let schema = make_schema();
let entry = CacheEntry {
schema,
original_bytes: bytes.clone(),
};
buf.extend_from_slice(&serde_json::to_vec(&entry).unwrap());
let path = cache_dir.join(format!("{:016x}.bin", hash));
fs::write(&path, &buf).expect("manual write must succeed");
let cache2 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let result = cache2.get(hash, &bytes);
assert!(
result.is_none(),
"a disk entry with a wrong CACHE_VERSION must be treated as a miss"
);
}
#[test]
fn test_disk_get_truncated_file_is_miss() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let cache_dir = cache.cache_dir().expect("cache dir must be set").clone();
let bytes = b"truncated-file-test".to_vec();
let hash = hash_bytes(&bytes);
let path = cache_dir.join(format!("{:016x}.bin", hash));
fs::write(&path, &[0x01u8, 0x02, 0x03]).expect("write must succeed");
let cache2 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let result = cache2.get(hash, &bytes);
assert!(
result.is_none(),
"a file shorter than 4 bytes must be treated as a miss"
);
}
#[test]
fn test_disk_get_corrupt_json_is_miss() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let cache_dir = cache.cache_dir().expect("cache dir must be set").clone();
let bytes = b"corrupt-json-test".to_vec();
let hash = hash_bytes(&bytes);
let path = cache_dir.join(format!("{:016x}.bin", hash));
let mut buf = CACHE_VERSION.to_le_bytes().to_vec();
buf.extend_from_slice(b"not valid json !!!");
fs::write(&path, &buf).expect("write must succeed");
let cache2 = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let result = cache2.get(hash, &bytes);
assert!(
result.is_none(),
"a disk entry with corrupt JSON must be treated as a miss"
);
}
#[test]
fn test_disk_get_absent_entry_is_miss() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let bytes = b"never-written".to_vec();
let hash = hash_bytes(&bytes);
let result = cache.get(hash, &bytes);
assert!(
result.is_none(),
"a hash that was never inserted must be a miss"
);
}
#[test]
fn test_disk_insert_rename_failure_cleans_up_tmp() {
let tmp = TempDir::new().unwrap();
let cache = DiskCache::with_cache_dir(tmp.path().to_path_buf());
let cache_dir = cache.cache_dir().expect("cache dir must be set").clone();
let raw = serde_json::json!({"type": "null"});
let bytes = schema_bytes(&raw);
let hash = hash_bytes(&bytes);
let final_path = cache_dir.join(format!("{:016x}.bin", hash));
fs::create_dir_all(&final_path).expect("creating dir as rename target must succeed");
cache.insert(hash, bytes, make_schema());
let tmp_count = fs::read_dir(&cache_dir)
.expect("read_dir must succeed")
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
.count();
assert_eq!(
tmp_count, 0,
"rename failure must not leave orphaned .tmp files"
);
assert!(
final_path.is_dir(),
"pre-existing directory at final path must still exist"
);
}
}