rskit_cache/adapters/fs/
mod.rs

1//! Filesystem cache adapter.
2
3use std::{
4    path::{Path, PathBuf},
5    sync::Arc,
6    time::{Duration, SystemTime, UNIX_EPOCH},
7};
8
9use rskit_errors::{AppError, AppResult, ErrorCode};
10use serde::{Deserialize, Serialize};
11use tokio::sync::Mutex;
12
13use crate::{CacheConfig, CacheRegistry, CacheStore, CacheStoreFactory};
14
15const DEFAULT_MAX_ENTRY_BYTES: u64 = 16 * 1024 * 1024;
16
17/// Filesystem cache configuration.
18///
19/// The root path is supplied when registering the adapter because it is a
20/// deployment concern owned by the composition boundary, not common cache
21/// selection configuration.
22#[derive(Debug, Clone, Deserialize, Serialize)]
23pub struct FileCacheConfig {
24    /// Root directory used to store cache entries.
25    pub root: PathBuf,
26    /// Optional prefix prepended to every key.
27    pub key_prefix: Option<String>,
28    /// Maximum serialized cache-entry size accepted by reads and writes.
29    #[serde(default = "default_max_entry_bytes")]
30    pub max_entry_bytes: u64,
31}
32
33impl FileCacheConfig {
34    /// Create filesystem cache configuration rooted at `root`.
35    #[must_use]
36    pub fn new(root: impl Into<PathBuf>) -> Self {
37        Self {
38            root: root.into(),
39            key_prefix: None,
40            max_entry_bytes: DEFAULT_MAX_ENTRY_BYTES,
41        }
42    }
43}
44
45/// Persistent filesystem cache adapter.
46pub struct FileCache {
47    config: FileCacheConfig,
48    mutation_lock: Arc<Mutex<()>>,
49}
50
51impl FileCache {
52    /// Create a filesystem cache adapter.
53    #[must_use]
54    pub fn new(config: FileCacheConfig) -> Self {
55        Self {
56            config,
57            mutation_lock: Arc::new(Mutex::new(())),
58        }
59    }
60
61    fn prefixed_key(&self, key: &str) -> String {
62        self.config
63            .key_prefix
64            .as_ref()
65            .map_or_else(|| key.to_owned(), |prefix| format!("{prefix}:{key}"))
66    }
67
68    fn entry_path(&self, key: &str) -> PathBuf {
69        let hash = blake3::hash(key.as_bytes()).to_hex().to_string();
70        self.config.root.join(&hash[..2]).join(hash)
71    }
72
73    async fn read_entry(&self, path: &Path, expected_key: &str) -> AppResult<Option<Entry>> {
74        let Some(entry) = self.read_entry_file(path).await? else {
75            return Ok(None);
76        };
77        if entry.key != expected_key {
78            return Err(AppError::new(
79                ErrorCode::Conflict,
80                format!("cache key collision for '{}'", path.display()),
81            ));
82        }
83        if entry.is_expired()? {
84            return Ok(None);
85        }
86        Ok(Some(entry))
87    }
88
89    async fn read_entry_file(&self, path: &Path) -> AppResult<Option<Entry>> {
90        let bytes =
91            match rskit_fs::async_io::file::read_bounded(path, self.config.max_entry_bytes).await {
92                Ok(bytes) => bytes,
93                Err(error) if is_not_found_error(&error) => return Ok(None),
94                Err(error) => return Err(error),
95            };
96        let entry: Entry = serde_json::from_slice(&bytes).map_err(|error| {
97            AppError::new(
98                ErrorCode::Internal,
99                format!("failed to decode cache entry '{}'", path.display()),
100            )
101            .with_cause(error)
102        })?;
103        Ok(Some(entry))
104    }
105
106    /// Remove expired cache entries, checking at most `max_entries` files.
107    ///
108    /// Reads stay non-destructive to avoid unlinking a concurrent fresh write.
109    /// Call this method from application-owned maintenance code when filesystem
110    /// cache entries use TTLs and the cache directory needs bounded cleanup.
111    pub async fn cleanup_expired(&self, max_entries: usize) -> AppResult<usize> {
112        if max_entries == 0 {
113            return Ok(0);
114        }
115
116        let _guard = self.mutation_lock.lock().await;
117        let mut checked = 0;
118        let mut removed = 0;
119        if !rskit_fs::async_io::dir::exists(&self.config.root).await? {
120            return Ok(0);
121        }
122        let shard_dirs = rskit_fs::async_io::dir::list(&self.config.root).await?;
123
124        for shard in shard_dirs.into_iter().filter(|entry| entry.is_dir) {
125            let entries = match rskit_fs::async_io::dir::list(&shard.path).await {
126                Ok(entries) => entries,
127                Err(error) if is_not_found_error(&error) => continue,
128                Err(error) => return Err(error),
129            };
130            for entry in entries.into_iter().filter(|entry| entry.is_file) {
131                if checked == max_entries {
132                    return Ok(removed);
133                }
134                checked += 1;
135                let Some(cache_entry) = self.read_entry_file(&entry.path).await? else {
136                    continue;
137                };
138                if cache_entry.is_expired()? {
139                    match tokio::fs::remove_file(&entry.path).await {
140                        Ok(()) => removed += 1,
141                        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
142                        Err(error) => {
143                            return Err(AppError::new(
144                                ErrorCode::Internal,
145                                format!(
146                                    "failed to delete expired cache entry '{}'",
147                                    entry.path.display()
148                                ),
149                            )
150                            .with_cause(error));
151                        }
152                    }
153                }
154            }
155        }
156
157        Ok(removed)
158    }
159}
160
161#[async_trait::async_trait]
162impl CacheStore for FileCache {
163    async fn get(&self, key: &str) -> AppResult<Option<String>> {
164        let key = self.prefixed_key(key);
165        let path = self.entry_path(&key);
166        Ok(self.read_entry(&path, &key).await?.map(|entry| entry.value))
167    }
168
169    async fn set(&self, key: &str, val: &str, ttl: Option<Duration>) -> AppResult<()> {
170        if ttl.is_some_and(|ttl| ttl.is_zero()) {
171            return Err(AppError::new(
172                ErrorCode::InvalidInput,
173                "cache TTL must be greater than zero",
174            ));
175        }
176        let key = self.prefixed_key(key);
177        let path = self.entry_path(&key);
178        let entry = Entry {
179            key,
180            value: val.to_owned(),
181            expires_at_millis: expires_at_millis(ttl)?,
182        };
183        let json = serde_json::to_vec(&entry).map_err(|error| {
184            AppError::new(ErrorCode::Internal, "failed to encode cache entry").with_cause(error)
185        })?;
186        if json.len() as u64 > self.config.max_entry_bytes {
187            return Err(cache_entry_too_large_error(
188                &path,
189                json.len() as u64,
190                self.config.max_entry_bytes,
191            ));
192        }
193        let _guard = self.mutation_lock.lock().await;
194        rskit_fs::async_io::file::write_atomic_replace(&path, json, "rskit-cache").await
195    }
196
197    async fn delete(&self, key: &str) -> AppResult<bool> {
198        let key = self.prefixed_key(key);
199        let path = self.entry_path(&key);
200        let _guard = self.mutation_lock.lock().await;
201        match tokio::fs::remove_file(&path).await {
202            Ok(()) => Ok(true),
203            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
204            Err(error) => Err(AppError::new(
205                ErrorCode::Internal,
206                format!("failed to delete cache entry '{}'", path.display()),
207            )
208            .with_cause(error)),
209        }
210    }
211
212    async fn exists(&self, key: &str) -> AppResult<bool> {
213        self.get(key).await.map(|value| value.is_some())
214    }
215}
216
217struct FileFactory {
218    config: FileCacheConfig,
219}
220
221#[async_trait::async_trait]
222impl CacheStoreFactory for FileFactory {
223    async fn create(&self, config: &CacheConfig) -> AppResult<Arc<dyn CacheStore>> {
224        let mut fs = self.config.clone();
225        if fs.key_prefix.is_none() {
226            fs.key_prefix.clone_from(&config.key_prefix);
227        }
228        Ok(Arc::new(FileCache::new(fs)))
229    }
230}
231
232/// Explicitly register the filesystem adapter.
233pub fn register_file_cache(registry: &mut CacheRegistry, config: FileCacheConfig) -> AppResult<()> {
234    registry.register("fs", Arc::new(FileFactory { config }))
235}
236
237#[derive(Deserialize, Serialize)]
238struct Entry {
239    key: String,
240    value: String,
241    expires_at_millis: Option<u128>,
242}
243
244impl Entry {
245    fn is_expired(&self) -> AppResult<bool> {
246        self.expires_at_millis
247            .map(|expires_at| now_millis().map(|now| expires_at <= now))
248            .transpose()
249            .map(|expired| expired.unwrap_or(false))
250    }
251}
252
253fn expires_at_millis(ttl: Option<Duration>) -> AppResult<Option<u128>> {
254    ttl.map(ttl_millis)
255        .transpose()?
256        .map(|ttl| now_millis().and_then(|now| now.checked_add(ttl).ok_or_else(ttl_error)))
257        .transpose()
258}
259
260fn ttl_millis(ttl: Duration) -> AppResult<u128> {
261    if ttl.is_zero() {
262        return Err(AppError::new(
263            ErrorCode::InvalidInput,
264            "cache TTL must be greater than zero",
265        ));
266    }
267    Ok(ttl.as_millis().max(1))
268}
269
270fn now_millis() -> AppResult<u128> {
271    SystemTime::now()
272        .duration_since(UNIX_EPOCH)
273        .map(|duration| duration.as_millis())
274        .map_err(|error| {
275            AppError::new(ErrorCode::Internal, "system clock is before UNIX_EPOCH")
276                .with_cause(error)
277        })
278}
279
280fn ttl_error() -> AppError {
281    AppError::new(
282        ErrorCode::InvalidInput,
283        "cache TTL is too large to represent safely for filesystem cache",
284    )
285}
286
287const fn default_max_entry_bytes() -> u64 {
288    DEFAULT_MAX_ENTRY_BYTES
289}
290
291fn is_not_found_error(error: &AppError) -> bool {
292    error
293        .cause()
294        .and_then(|cause| cause.downcast_ref::<std::io::Error>())
295        .is_some_and(|cause| cause.kind() == std::io::ErrorKind::NotFound)
296}
297
298fn cache_entry_too_large_error(path: &Path, actual: u64, limit: u64) -> AppError {
299    AppError::new(
300        ErrorCode::InvalidInput,
301        format!(
302            "cache entry '{}' is {actual} bytes, exceeding limit {limit} bytes",
303            path.display()
304        ),
305    )
306    .with_detail("rskit_cache_error", "entry_too_large")
307    .with_detail("actual_bytes", actual)
308    .with_detail("limit_bytes", limit)
309}
310
311#[cfg(test)]
312mod tests {
313    use std::sync::atomic::{AtomicU64, Ordering};
314
315    use super::*;
316
317    #[tokio::test]
318    async fn stores_and_reads_values() {
319        let root = temp_root();
320        let cache = FileCache::new(FileCacheConfig::new(&root));
321
322        cache.set("key", "value", None).await.unwrap();
323
324        assert_eq!(cache.get("key").await.unwrap().as_deref(), Some("value"));
325        assert!(cache.exists("key").await.unwrap());
326        assert!(cache.delete("key").await.unwrap());
327        assert!(!cache.exists("key").await.unwrap());
328        let _ = tokio::fs::remove_dir_all(root).await;
329    }
330
331    #[tokio::test]
332    async fn updates_existing_values() {
333        let root = temp_root();
334        let cache = FileCache::new(FileCacheConfig::new(&root));
335
336        cache.set("key", "old", None).await.unwrap();
337        cache.set("key", "new", None).await.unwrap();
338
339        assert_eq!(cache.get("key").await.unwrap().as_deref(), Some("new"));
340        let _ = tokio::fs::remove_dir_all(root).await;
341    }
342
343    #[tokio::test]
344    async fn registry_build_inherits_global_key_prefix() {
345        let root = temp_root();
346        let mut registry = CacheRegistry::new();
347        register_file_cache(&mut registry, FileCacheConfig::new(&root)).unwrap();
348        let cache = registry
349            .build(&cache_config_with_prefix("global"))
350            .await
351            .unwrap();
352
353        cache.set("key", "value", None).await.unwrap();
354
355        let mut global_config = FileCacheConfig::new(&root);
356        global_config.key_prefix = Some("global".to_owned());
357        assert_eq!(
358            FileCache::new(global_config)
359                .get("key")
360                .await
361                .unwrap()
362                .as_deref(),
363            Some("value")
364        );
365        assert_eq!(
366            FileCache::new(FileCacheConfig::new(&root))
367                .get("key")
368                .await
369                .unwrap(),
370            None
371        );
372        let _ = tokio::fs::remove_dir_all(root).await;
373    }
374
375    #[tokio::test]
376    async fn registry_build_preserves_adapter_key_prefix_override() {
377        let root = temp_root();
378        let mut registry = CacheRegistry::new();
379        let mut adapter_config = FileCacheConfig::new(&root);
380        adapter_config.key_prefix = Some("adapter".to_owned());
381        register_file_cache(&mut registry, adapter_config).unwrap();
382        let cache = registry
383            .build(&cache_config_with_prefix("global"))
384            .await
385            .unwrap();
386
387        cache.set("key", "value", None).await.unwrap();
388
389        let mut adapter_reader_config = FileCacheConfig::new(&root);
390        adapter_reader_config.key_prefix = Some("adapter".to_owned());
391        let mut global_reader_config = FileCacheConfig::new(&root);
392        global_reader_config.key_prefix = Some("global".to_owned());
393        assert_eq!(
394            FileCache::new(adapter_reader_config)
395                .get("key")
396                .await
397                .unwrap()
398                .as_deref(),
399            Some("value")
400        );
401        assert_eq!(
402            FileCache::new(global_reader_config)
403                .get("key")
404                .await
405                .unwrap(),
406            None
407        );
408        let _ = tokio::fs::remove_dir_all(root).await;
409    }
410
411    #[tokio::test]
412    async fn rejects_entries_exceeding_configured_size() {
413        let root = temp_root();
414        let mut config = FileCacheConfig::new(&root);
415        config.max_entry_bytes = 8;
416        let cache = FileCache::new(config);
417
418        let err = cache
419            .set("key", "value larger than limit", None)
420            .await
421            .expect_err("oversized entries must be rejected");
422
423        assert_eq!(err.code(), ErrorCode::InvalidInput);
424        let _ = tokio::fs::remove_dir_all(root).await;
425    }
426
427    #[tokio::test]
428    async fn rejects_oversized_entry_files_before_decoding() {
429        let root = temp_root();
430        let mut config = FileCacheConfig::new(&root);
431        config.max_entry_bytes = 8;
432        let cache = FileCache::new(config);
433        let key = cache.prefixed_key("key");
434        let path = cache.entry_path(&key);
435
436        rskit_fs::async_io::file::write_atomic_replace(&path, b"012345678", "rskit-cache-test")
437            .await
438            .unwrap();
439
440        let err = cache
441            .get("key")
442            .await
443            .expect_err("oversized entry files must be rejected");
444
445        assert_eq!(err.code(), ErrorCode::InvalidInput);
446        let _ = tokio::fs::remove_dir_all(root).await;
447    }
448
449    #[tokio::test]
450    async fn rejects_zero_ttl() {
451        let root = temp_root();
452        let cache = FileCache::new(FileCacheConfig::new(&root));
453
454        let err = cache
455            .set("key", "value", Some(Duration::ZERO))
456            .await
457            .expect_err("zero TTL must be rejected");
458
459        assert_eq!(err.code(), ErrorCode::InvalidInput);
460        let _ = tokio::fs::remove_dir_all(root).await;
461    }
462
463    #[tokio::test]
464    async fn expired_entries_miss_without_deleting_from_read_path() {
465        let root = temp_root();
466        let cache = FileCache::new(FileCacheConfig::new(&root));
467        let key = cache.prefixed_key("key");
468        let path = cache.entry_path(&key);
469        let entry = expired_entry(key, "value");
470
471        write_entry(&path, &entry).await;
472
473        assert_eq!(cache.get("key").await.unwrap(), None);
474        assert!(!cache.exists("key").await.unwrap());
475        assert!(tokio::fs::metadata(path).await.is_ok());
476        let _ = tokio::fs::remove_dir_all(root).await;
477    }
478
479    #[tokio::test]
480    async fn cleanup_expired_removes_expired_entries_without_deleting_live_entries() {
481        let root = temp_root();
482        let cache = FileCache::new(FileCacheConfig::new(&root));
483        let expired_key = cache.prefixed_key("expired");
484        let expired_path = cache.entry_path(&expired_key);
485        let live_key = cache.prefixed_key("live");
486        let live_path = cache.entry_path(&live_key);
487
488        write_entry(&expired_path, &expired_entry(expired_key, "expired")).await;
489        write_entry(
490            &live_path,
491            &Entry {
492                key: live_key,
493                value: "live".to_owned(),
494                expires_at_millis: Some(now_millis().unwrap() + 60_000),
495            },
496        )
497        .await;
498
499        assert_eq!(cache.cleanup_expired(16).await.unwrap(), 1);
500        assert!(tokio::fs::metadata(expired_path).await.is_err());
501        assert!(tokio::fs::metadata(live_path).await.is_ok());
502        assert_eq!(cache.get("live").await.unwrap().as_deref(), Some("live"));
503        let _ = tokio::fs::remove_dir_all(root).await;
504    }
505
506    #[tokio::test]
507    async fn cleanup_expired_checks_at_most_requested_entries() {
508        let root = temp_root();
509        let cache = FileCache::new(FileCacheConfig::new(&root));
510        let first_key = cache.prefixed_key("first");
511        let first_path = cache.entry_path(&first_key);
512        let second_key = cache.prefixed_key("second");
513        let second_path = cache.entry_path(&second_key);
514
515        write_entry(&first_path, &expired_entry(first_key, "first")).await;
516        write_entry(&second_path, &expired_entry(second_key, "second")).await;
517
518        assert_eq!(cache.cleanup_expired(1).await.unwrap(), 1);
519        let remaining = [first_path, second_path]
520            .into_iter()
521            .filter(|path| path.exists())
522            .count();
523        assert_eq!(remaining, 1);
524        let _ = tokio::fs::remove_dir_all(root).await;
525    }
526
527    #[tokio::test]
528    async fn cleanup_expired_returns_zero_when_cache_root_is_missing() {
529        let root = temp_root();
530        let cache = FileCache::new(FileCacheConfig::new(&root));
531
532        assert_eq!(cache.cleanup_expired(16).await.unwrap(), 0);
533    }
534
535    #[test]
536    fn positive_sub_millisecond_ttl_rounds_up() {
537        assert_eq!(ttl_millis(Duration::from_nanos(1)).unwrap(), 1);
538    }
539
540    #[test]
541    fn new_config_uses_default_entry_size_limit() {
542        assert_eq!(
543            FileCacheConfig::new("cache").max_entry_bytes,
544            DEFAULT_MAX_ENTRY_BYTES
545        );
546    }
547
548    fn cache_config_with_prefix(prefix: &str) -> CacheConfig {
549        CacheConfig {
550            store: "fs".to_owned(),
551            key_prefix: Some(prefix.to_owned()),
552            ..CacheConfig::default()
553        }
554    }
555
556    fn expired_entry(key: String, value: &str) -> Entry {
557        Entry {
558            key,
559            value: value.to_owned(),
560            expires_at_millis: Some(now_millis().unwrap().saturating_sub(1)),
561        }
562    }
563
564    async fn write_entry(path: &Path, entry: &Entry) {
565        rskit_fs::async_io::file::write_atomic_replace(
566            path,
567            serde_json::to_vec(entry).unwrap(),
568            "rskit-cache-test",
569        )
570        .await
571        .unwrap();
572    }
573
574    fn temp_root() -> PathBuf {
575        static COUNTER: AtomicU64 = AtomicU64::new(0);
576        std::env::temp_dir().join(format!(
577            "rskit-cache-fs-{}-{}-{}",
578            std::process::id(),
579            now_millis().unwrap(),
580            COUNTER.fetch_add(1, Ordering::Relaxed)
581        ))
582    }
583}
rskit_cache/adapters/fs/mod.rs

rskit_cache/adapters/fs/
mod.rs
