Add per-crate variant_ctx() helper methods to StorageError, BlobCacheError, ProtocolError, and BridgeError. Replace 78 .map_err(|e| format!(...)) closures with concise context calls. Identical error messages preserved.
369 lines
12 KiB
Rust
369 lines
12 KiB
Rust
//! Disposable content-addressed blob cache for avatar and icon files.
|
|
//!
|
|
//! Wraps [`cacache`] for production-tested crash safety and integrity
|
|
//! verification. The on-disk layout is managed by cacache (content-v2,
|
|
//! index-v2). Chanora maps protocol keys (`av_<md5>`, `ic_<crc32>`) to
|
|
//! cacache string keys.
|
|
//!
|
|
//! This crate is intentionally separate from `chanora_storage`:
|
|
//! storage owns persistent identity/bookmark data; cache owns
|
|
//! reconstructible, disposable blob data with different durability
|
|
//! and backup semantics.
|
|
|
|
#![forbid(unsafe_code)]
|
|
#![warn(missing_docs)]
|
|
|
|
use std::path::{Path, PathBuf};
|
|
|
|
/// Errors raised by the blob cache.
|
|
// TODO(refactor): Io(String) variant is duplicated across chanora_cache,
|
|
// chanora_storage, and chanora_diagnostics. Could use a shared error type
|
|
// or derive From<std::io::Error> instead of manually wrapping.
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum BlobCacheError {
|
|
/// Filesystem I/O error.
|
|
#[error("io: {0}")]
|
|
Io(String),
|
|
/// Key validation error.
|
|
#[error("invalid key: {0}")]
|
|
InvalidKey(String),
|
|
}
|
|
|
|
impl BlobCacheError {
|
|
fn io_ctx(ctx: impl std::fmt::Display, e: impl std::fmt::Display) -> Self {
|
|
BlobCacheError::Io(format!("{ctx}: {e}"))
|
|
}
|
|
}
|
|
|
|
/// Content-addressed blob cache backed by cacache.
|
|
pub struct BlobCache {
|
|
cache_dir: PathBuf,
|
|
/// Maximum total cache size in bytes. 0 = no limit.
|
|
max_bytes: u64,
|
|
}
|
|
|
|
/// Avatar blob prefix.
|
|
pub const PREFIX_AVATAR: &str = "av_";
|
|
/// Icon blob prefix.
|
|
pub const PREFIX_ICON: &str = "ic_";
|
|
|
|
impl BlobCache {
|
|
/// Create or open a [`BlobCache`] rooted at `cache_dir/chanora/`.
|
|
///
|
|
/// Creates the cacache directory. `max_bytes` sets the eviction
|
|
/// threshold; 0 means no automatic eviction.
|
|
pub fn new(cache_dir: impl AsRef<Path>, max_bytes: u64) -> Result<Self, BlobCacheError> {
|
|
let cache_dir = cache_dir.as_ref().join("chanora").join("blobs");
|
|
// cacache creates the directory on first write, but we create
|
|
// it eagerly so total_size() works before any writes.
|
|
std::fs::create_dir_all(&cache_dir)
|
|
.map_err(|e| BlobCacheError::io_ctx("mkdir cache", e))?;
|
|
Ok(Self {
|
|
cache_dir,
|
|
max_bytes,
|
|
})
|
|
}
|
|
|
|
/// Store a blob. `prefix` is [`PREFIX_AVATAR`] or [`PREFIX_ICON`].
|
|
/// `key` is the content hash (MD5 hex for avatars, unsigned
|
|
/// decimal CRC32 for icons).
|
|
///
|
|
/// Cacache handles dedup and integrity internally.
|
|
pub async fn put(
|
|
&self,
|
|
prefix: &str,
|
|
key: &str,
|
|
data: &[u8],
|
|
) -> Result<(), BlobCacheError> {
|
|
validate_key(prefix, key)?;
|
|
let cache_key = format!("{prefix}{key}");
|
|
cacache::write(&self.cache_dir, &cache_key, data)
|
|
.await
|
|
.map_err(|e| BlobCacheError::io_ctx("cacache write", e))?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Read a blob. Returns `None` if not cached.
|
|
///
|
|
/// Cacache verifies SSRI integrity on every read.
|
|
pub async fn get(&self, prefix: &str, key: &str) -> Result<Option<Vec<u8>>, BlobCacheError> {
|
|
validate_key(prefix, key)?;
|
|
let cache_key = format!("{prefix}{key}");
|
|
match cacache::read(&self.cache_dir, &cache_key).await {
|
|
Ok(data) => Ok(Some(data)),
|
|
Err(cacache::Error::EntryNotFound(_, _)) => Ok(None),
|
|
Err(e) => {
|
|
// Integrity failure or I/O error — remove corrupt entry.
|
|
tracing::warn!(
|
|
target: "chanora_cache",
|
|
key = %cache_key,
|
|
error = %e,
|
|
"cache read failed; removing entry"
|
|
);
|
|
let _ = cacache::remove(&self.cache_dir, &cache_key).await;
|
|
Ok(None)
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Delete a specific blob.
|
|
pub async fn remove(&self, prefix: &str, key: &str) -> Result<(), BlobCacheError> {
|
|
validate_key(prefix, key)?;
|
|
let cache_key = format!("{prefix}{key}");
|
|
cacache::remove(&self.cache_dir, &cache_key)
|
|
.await
|
|
.map_err(|e| BlobCacheError::io_ctx("cacache remove", e))?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Delete all blobs.
|
|
pub async fn clear(&self) -> Result<(), BlobCacheError> {
|
|
let path = self.cache_dir.clone();
|
|
tokio::task::spawn_blocking(move || {
|
|
if path.exists() {
|
|
std::fs::remove_dir_all(&path)
|
|
.map_err(|e| BlobCacheError::io_ctx("clear cache", e))?;
|
|
std::fs::create_dir_all(&path)
|
|
.map_err(|e| BlobCacheError::io_ctx("recreate cache dir", e))?;
|
|
}
|
|
Ok(())
|
|
})
|
|
.await
|
|
.map_err(|e| BlobCacheError::io_ctx("clear task", e))?
|
|
}
|
|
|
|
/// Return total bytes used by all blobs.
|
|
///
|
|
/// Walks cacache entries and sums sizes.
|
|
pub async fn total_size(&self) -> Result<u64, BlobCacheError> {
|
|
let cache_dir = self.cache_dir.clone();
|
|
tokio::task::spawn_blocking(move || {
|
|
let mut total: u64 = 0;
|
|
for entry in cacache::list_sync(&cache_dir) {
|
|
match entry {
|
|
Ok(meta) => total += meta.size as u64,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
target: "chanora_cache",
|
|
error = %e,
|
|
"skipping bad entry during size scan"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
Ok(total)
|
|
})
|
|
.await
|
|
.map_err(|e| BlobCacheError::io_ctx("total_size task", e))?
|
|
}
|
|
|
|
/// Evict oldest entries by timestamp until total size is under
|
|
/// `max_bytes`. Call on startup or periodically. No-op if
|
|
/// `max_bytes` is 0.
|
|
pub async fn evict(&self) -> Result<(), BlobCacheError> {
|
|
if self.max_bytes == 0 {
|
|
return Ok(());
|
|
}
|
|
let cache_dir = self.cache_dir.clone();
|
|
let max_bytes = self.max_bytes;
|
|
tokio::task::spawn_blocking(move || {
|
|
let mut entries: Vec<(String, usize, u128)> = Vec::new();
|
|
for entry in cacache::list_sync(&cache_dir) {
|
|
match entry {
|
|
Ok(meta) => {
|
|
entries.push((meta.key, meta.size, meta.time));
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
target: "chanora_cache",
|
|
error = %e,
|
|
"skipping bad entry during eviction scan"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
let total: usize = entries.iter().map(|(_, s, _)| *s).sum();
|
|
if total as u64 <= max_bytes {
|
|
return Ok(());
|
|
}
|
|
entries.sort_by_key(|(_, _, t)| *t);
|
|
let mut freed: usize = 0;
|
|
let target = total - max_bytes as usize;
|
|
for (key, size, _) in entries {
|
|
if freed >= target {
|
|
break;
|
|
}
|
|
let _ = cacache::remove_sync(&cache_dir, &key);
|
|
freed += size;
|
|
}
|
|
tracing::info!(
|
|
target: "chanora_cache",
|
|
freed_bytes = freed,
|
|
"evicted oldest blobs"
|
|
);
|
|
Ok(())
|
|
})
|
|
.await
|
|
.map_err(|e| BlobCacheError::io_ctx("evict task", e))?
|
|
}
|
|
}
|
|
|
|
/// Validate key format to prevent malformed entries.
|
|
fn validate_key(prefix: &str, key: &str) -> Result<(), BlobCacheError> {
|
|
if !matches!(prefix, PREFIX_AVATAR | PREFIX_ICON) {
|
|
return Err(BlobCacheError::InvalidKey(format!("bad prefix: {prefix}")));
|
|
}
|
|
match prefix {
|
|
PREFIX_AVATAR => {
|
|
// MD5 hex = exactly 32 hex chars.
|
|
if key.len() != 32 || !key.chars().all(|c| c.is_ascii_hexdigit()) {
|
|
return Err(BlobCacheError::InvalidKey(format!(
|
|
"avatar key must be 32 hex chars, got: {key}"
|
|
)));
|
|
}
|
|
}
|
|
PREFIX_ICON => {
|
|
// Unsigned CRC32 = decimal digits.
|
|
if key.is_empty() || !key.chars().all(|c| c.is_ascii_digit()) {
|
|
return Err(BlobCacheError::InvalidKey(format!(
|
|
"icon key must be decimal digits, got: {key}"
|
|
)));
|
|
}
|
|
}
|
|
_ => unreachable!(),
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn tempdir() -> tempfile::TempDir {
|
|
tempfile::Builder::new()
|
|
.prefix("chanora_cache_test_")
|
|
.tempdir()
|
|
.unwrap()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn put_get_roundtrip() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
assert!(cache
|
|
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
|
.await
|
|
.unwrap()
|
|
.is_none());
|
|
cache
|
|
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"avatar-bytes")
|
|
.await
|
|
.unwrap();
|
|
let data = cache
|
|
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(data.as_deref(), Some(b"avatar-bytes".as_slice()));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn get_missing_returns_none() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
assert!(cache
|
|
.get(PREFIX_AVATAR, "00000000000000000000000000000000")
|
|
.await
|
|
.unwrap()
|
|
.is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn clear_removes_all() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
cache
|
|
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"data")
|
|
.await
|
|
.unwrap();
|
|
cache.put(PREFIX_ICON, "12345", b"icon").await.unwrap();
|
|
cache.clear().await.unwrap();
|
|
assert_eq!(cache.total_size().await.unwrap(), 0);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn total_size_accounts_for_all_entries() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
cache
|
|
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"12345")
|
|
.await
|
|
.unwrap();
|
|
cache.put(PREFIX_ICON, "99999", b"12").await.unwrap();
|
|
assert_eq!(cache.total_size().await.unwrap(), 5 + 2);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn invalid_key_rejected() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
// Too short for MD5.
|
|
assert!(cache.put(PREFIX_AVATAR, "abc", b"data").await.is_err());
|
|
// Non-hex in MD5.
|
|
assert!(cache
|
|
.put(PREFIX_AVATAR, "g".repeat(32).as_str(), b"data")
|
|
.await
|
|
.is_err());
|
|
// Non-digit in icon key.
|
|
assert!(cache.put(PREFIX_ICON, "12a45", b"data").await.is_err());
|
|
// Bad prefix.
|
|
assert!(cache.put("xx_", "abc", b"data").await.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn evict_deletes_oldest_until_under_cap() {
|
|
let tmp = tempdir();
|
|
// 10 byte cap.
|
|
let cache = BlobCache::new(&tmp, 10).unwrap();
|
|
cache
|
|
.put(
|
|
PREFIX_AVATAR,
|
|
"a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6",
|
|
b"12345678",
|
|
)
|
|
.await
|
|
.unwrap(); // 8 bytes
|
|
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
|
cache
|
|
.put(PREFIX_ICON, "11111", b"12345")
|
|
.await
|
|
.unwrap(); // 5 bytes → total 13, over cap
|
|
cache.evict().await.unwrap();
|
|
// Oldest (avatar) should be evicted.
|
|
assert!(cache
|
|
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
|
.await
|
|
.unwrap()
|
|
.is_none());
|
|
assert!(cache.get(PREFIX_ICON, "11111").await.unwrap().is_some());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn remove_deletes_entry() {
|
|
let tmp = tempdir();
|
|
let cache = BlobCache::new(&tmp, 0).unwrap();
|
|
cache
|
|
.put(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", b"data")
|
|
.await
|
|
.unwrap();
|
|
cache
|
|
.remove(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
|
.await
|
|
.unwrap();
|
|
assert!(cache
|
|
.get(PREFIX_AVATAR, "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")
|
|
.await
|
|
.unwrap()
|
|
.is_none());
|
|
}
|
|
}
|