rskit_cache/adapters/fs/
store.rs

1use std::{
2    path::{Path, PathBuf},
3    sync::Arc,
4    time::{Duration, SystemTime, UNIX_EPOCH},
5};
6
7use rskit_errors::{AppError, AppResult, ErrorCode};
8use serde::{Deserialize, Serialize};
9use tokio::sync::Mutex;
10
11use crate::CacheStore;
12
13use super::FileCacheConfig;
14
15/// Persistent filesystem cache adapter.
16pub struct FileCache {
17    config: FileCacheConfig,
18    mutation_lock: Arc<Mutex<()>>,
19}
20
21impl FileCache {
22    /// Create a filesystem cache adapter.
23    #[must_use]
24    pub fn new(config: FileCacheConfig) -> Self {
25        Self {
26            config,
27            mutation_lock: Arc::new(Mutex::new(())),
28        }
29    }
30
31    pub(crate) fn prefixed_key(&self, key: &str) -> String {
32        self.config
33            .key_prefix
34            .as_ref()
35            .map_or_else(|| key.to_owned(), |prefix| format!("{prefix}:{key}"))
36    }
37
38    pub(crate) fn entry_path(&self, key: &str) -> PathBuf {
39        let hash = rskit_util::hash::hash_hex(key.as_bytes());
40        self.config.root.join(&hash[..2]).join(hash)
41    }
42
43    async fn read_entry(&self, path: &Path, expected_key: &str) -> AppResult<Option<Entry>> {
44        let Some(entry) = self.read_entry_file(path).await? else {
45            return Ok(None);
46        };
47        if entry.key != expected_key {
48            return Err(AppError::new(
49                ErrorCode::Conflict,
50                format!("cache key collision for '{}'", path.display()),
51            ));
52        }
53        if entry.is_expired()? {
54            return Ok(None);
55        }
56        Ok(Some(entry))
57    }
58
59    async fn read_entry_file(&self, path: &Path) -> AppResult<Option<Entry>> {
60        let bytes =
61            match rskit_fs::async_io::file::read_bounded(path, self.config.max_entry_bytes).await {
62                Ok(bytes) => bytes,
63                Err(error) if is_not_found_error(&error) => return Ok(None),
64                Err(error) => return Err(error),
65            };
66        let entry: Entry = serde_json::from_slice(&bytes).map_err(|error| {
67            AppError::new(
68                ErrorCode::Internal,
69                format!("failed to decode cache entry '{}'", path.display()),
70            )
71            .with_cause(error)
72        })?;
73        Ok(Some(entry))
74    }
75
76    /// Remove expired cache entries, checking at most `max_entries` files.
77    ///
78    /// Reads stay non-destructive to avoid unlinking a concurrent fresh write.
79    /// Call this method from application-owned maintenance code when filesystem cache entries use TTLs
80    /// and the cache directory needs bounded cleanup.
81    pub async fn cleanup_expired(&self, max_entries: usize) -> AppResult<usize> {
82        if max_entries == 0 {
83            return Ok(0);
84        }
85
86        let _guard = self.mutation_lock.lock().await;
87        let mut checked = 0;
88        let mut removed = 0;
89        if !rskit_fs::async_io::dir::exists(&self.config.root).await? {
90            return Ok(0);
91        }
92        let shard_dirs = rskit_fs::async_io::dir::list(&self.config.root).await?;
93
94        for shard in shard_dirs.into_iter().filter(|entry| entry.is_dir) {
95            let entries = match rskit_fs::async_io::dir::list(&shard.path).await {
96                Ok(entries) => entries,
97                Err(error) if is_not_found_error(&error) => continue,
98                Err(error) => return Err(error),
99            };
100            for entry in entries.into_iter().filter(|entry| entry.is_file) {
101                if checked == max_entries {
102                    return Ok(removed);
103                }
104                checked += 1;
105                let Some(cache_entry) = self.read_entry_file(&entry.path).await? else {
106                    continue;
107                };
108                if cache_entry.is_expired()? {
109                    match tokio::fs::remove_file(&entry.path).await {
110                        Ok(()) => removed += 1,
111                        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
112                        Err(error) => {
113                            return Err(AppError::new(
114                                ErrorCode::Internal,
115                                format!(
116                                    "failed to delete expired cache entry '{}'",
117                                    entry.path.display()
118                                ),
119                            )
120                            .with_cause(error));
121                        }
122                    }
123                }
124            }
125        }
126
127        Ok(removed)
128    }
129}
130
131#[async_trait::async_trait]
132impl CacheStore for FileCache {
133    async fn get(&self, key: &str) -> AppResult<Option<String>> {
134        let key = self.prefixed_key(key);
135        let path = self.entry_path(&key);
136        Ok(self.read_entry(&path, &key).await?.map(|entry| entry.value))
137    }
138
139    async fn set(&self, key: &str, val: &str, ttl: Option<Duration>) -> AppResult<()> {
140        if ttl.is_some_and(|ttl| ttl.is_zero()) {
141            return Err(AppError::new(
142                ErrorCode::InvalidInput,
143                "cache TTL must be greater than zero",
144            ));
145        }
146        let key = self.prefixed_key(key);
147        let path = self.entry_path(&key);
148        let entry = Entry {
149            key,
150            value: val.to_owned(),
151            expires_at_millis: expires_at_millis(ttl)?,
152        };
153        let json = serde_json::to_vec(&entry).map_err(|error| {
154            AppError::new(ErrorCode::Internal, "failed to encode cache entry").with_cause(error)
155        })?;
156        if json.len() as u64 > self.config.max_entry_bytes {
157            return Err(cache_entry_too_large_error(
158                &path,
159                json.len() as u64,
160                self.config.max_entry_bytes,
161            ));
162        }
163        let _guard = self.mutation_lock.lock().await;
164        rskit_fs::async_io::file::write_atomic_replace(&path, json, "rskit-cache").await
165    }
166
167    async fn delete(&self, key: &str) -> AppResult<bool> {
168        let key = self.prefixed_key(key);
169        let path = self.entry_path(&key);
170        let _guard = self.mutation_lock.lock().await;
171        match tokio::fs::remove_file(&path).await {
172            Ok(()) => Ok(true),
173            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
174            Err(error) => Err(AppError::new(
175                ErrorCode::Internal,
176                format!("failed to delete cache entry '{}'", path.display()),
177            )
178            .with_cause(error)),
179        }
180    }
181
182    async fn exists(&self, key: &str) -> AppResult<bool> {
183        self.get(key).await.map(|value| value.is_some())
184    }
185}
186
187#[derive(Deserialize, Serialize)]
188pub(crate) struct Entry {
189    pub(crate) key: String,
190    pub(crate) value: String,
191    pub(crate) expires_at_millis: Option<u128>,
192}
193
194impl Entry {
195    fn is_expired(&self) -> AppResult<bool> {
196        self.expires_at_millis
197            .map(|expires_at| now_millis().map(|now| expires_at <= now))
198            .transpose()
199            .map(|expired| expired.unwrap_or(false))
200    }
201}
202
203fn expires_at_millis(ttl: Option<Duration>) -> AppResult<Option<u128>> {
204    ttl.map(ttl_millis)
205        .transpose()?
206        .map(|ttl| now_millis().and_then(|now| now.checked_add(ttl).ok_or_else(ttl_error)))
207        .transpose()
208}
209
210pub(crate) fn ttl_millis(ttl: Duration) -> AppResult<u128> {
211    if ttl.is_zero() {
212        return Err(AppError::new(
213            ErrorCode::InvalidInput,
214            "cache TTL must be greater than zero",
215        ));
216    }
217    Ok(ttl.as_millis().max(1))
218}
219
220pub(crate) fn now_millis() -> AppResult<u128> {
221    SystemTime::now()
222        .duration_since(UNIX_EPOCH)
223        .map(|duration| duration.as_millis())
224        .map_err(|error| {
225            AppError::new(ErrorCode::Internal, "system clock is before UNIX_EPOCH")
226                .with_cause(error)
227        })
228}
229
230fn ttl_error() -> AppError {
231    AppError::new(
232        ErrorCode::InvalidInput,
233        "cache TTL is too large to represent safely for filesystem cache",
234    )
235}
236
237fn is_not_found_error(error: &AppError) -> bool {
238    error
239        .cause()
240        .and_then(|cause| cause.downcast_ref::<std::io::Error>())
241        .is_some_and(|cause| cause.kind() == std::io::ErrorKind::NotFound)
242}
243
244fn cache_entry_too_large_error(path: &Path, actual: u64, limit: u64) -> AppError {
245    AppError::new(
246        ErrorCode::InvalidInput,
247        format!(
248            "cache entry '{}' is {actual} bytes, exceeding limit {limit} bytes",
249            path.display()
250        ),
251    )
252    .with_detail("rskit_cache_error", "entry_too_large")
253    .with_detail("actual_bytes", actual)
254    .with_detail("limit_bytes", limit)
255}
rskit_cache/adapters/fs/store.rs

rskit_cache/adapters/fs/
store.rs
