somatize_runtime/cache/
fs_store.rs

1//! `FsActionStore` — the persistent two-table cache (action records + CAS).
2//!
3//! Layout under the store root:
4//!
5//! ```text
6//! format.json                      {"version": 2}
7//! actions/<aa>/<key-hex>.json      ActionResult records (small)
8//! cas/<algo>/<aa>/<hash-hex>.bin   SOMA1-encoded blobs (deduplicated)
9//! pins/<name>                      GC roots: files containing an action key hex
10//! ```
11//!
12//! Commit protocol (crash-safe by construction):
13//! 1. blob written first (idempotent temp+fsync+rename — concurrent
14//!    same-content writers race benignly),
15//! 2. action record renamed into place **last** — the record is the
16//!    commit point. A crash in between leaves an orphan blob, never a
17//!    record pointing at missing required state.
18//!
19//! Eviction deletes blobs only; records are retained so an evicted
20//! entry is recomputed and re-fills the same content address
21//! (Nectar: eviction degrades performance, never correctness).
22
23use chrono::Utc;
24use somatize_core::action::{ActionCache, ActionResult, BlobStore, ContentHash};
25use somatize_core::cache::{CacheKey, CacheStore, EntryMeta, Origin};
26use somatize_core::codec::{decode_value, encode_and_hash};
27use somatize_core::error::{Result, SomaError};
28use somatize_core::value::Value;
29use std::collections::BTreeMap;
30use std::fs;
31use std::path::{Path, PathBuf};
32
33/// On-disk layout version, recorded in `format.json`. A mismatched dir is
34/// refused with a pointer to `soma cache purge-v1` — silently reading an
35/// old layout would corrupt it.
36pub const FORMAT_VERSION: u32 = 2;
37
38/// The persistent two-table store: action records + CAS blobs. See the
39/// module docs for the layout and the crash-safe commit protocol.
40pub struct FsActionStore {
41    root: PathBuf,
42}
43
44impl FsActionStore {
45    /// Open (or initialize) a store rooted at `root`.
46    ///
47    /// Creates the directory skeleton and stamps `format.json` on first
48    /// use; refuses a root whose recorded version is not
49    /// [`FORMAT_VERSION`].
50    pub fn new(root: impl Into<PathBuf>) -> Result<Self> {
51        let root = root.into();
52        fs::create_dir_all(&root)?;
53
54        let format_path = root.join("format.json");
55        if format_path.exists() {
56            let raw = fs::read_to_string(&format_path)?;
57            let format: serde_json::Value = serde_json::from_str(&raw)
58                .map_err(|e| SomaError::Cache(format!("unreadable cache format.json: {e}")))?;
59            let version = format.get("version").and_then(|v| v.as_u64()).unwrap_or(0);
60            if version != FORMAT_VERSION as u64 {
61                return Err(SomaError::Cache(format!(
62                    "cache dir {} has format version {version}, expected {FORMAT_VERSION}; \
63                     run `soma cache purge-v1` or point SOMA_CACHE_DIR elsewhere",
64                    root.display()
65                )));
66            }
67        } else {
68            write_atomic(
69                &format_path,
70                serde_json::json!({ "version": FORMAT_VERSION })
71                    .to_string()
72                    .as_bytes(),
73            )?;
74        }
75        fs::create_dir_all(root.join("actions"))?;
76        fs::create_dir_all(root.join("cas"))?;
77        fs::create_dir_all(root.join("pins"))?;
78        Ok(Self { root })
79    }
80
81    /// The store's root directory.
82    pub fn root(&self) -> &Path {
83        &self.root
84    }
85
86    fn action_path(&self, key: &CacheKey) -> PathBuf {
87        let hex = key.to_hex();
88        self.root
89            .join("actions")
90            .join(&hex[..2])
91            .join(format!("{hex}.json"))
92    }
93
94    fn blob_path(&self, hash: &ContentHash) -> PathBuf {
95        let hex = hash.to_hex();
96        self.root
97            .join("cas")
98            .join(hash.algo.prefix())
99            .join(&hex[..2])
100            .join(format!("{hex}.bin"))
101    }
102
103    /// Pin an action as a GC root under a human-readable name.
104    pub fn pin(&self, name: &str, key: &CacheKey) -> Result<()> {
105        if name.contains(['/', '\\']) || name.starts_with('.') {
106            return Err(SomaError::Cache(format!("invalid pin name: {name:?}")));
107        }
108        write_atomic(&self.root.join("pins").join(name), key.to_hex().as_bytes())
109    }
110
111    /// All pinned action keys.
112    pub fn pinned(&self) -> Result<Vec<CacheKey>> {
113        let mut keys = Vec::new();
114        let pins = self.root.join("pins");
115        if !pins.exists() {
116            return Ok(keys);
117        }
118        for entry in fs::read_dir(&pins)? {
119            let entry = entry?;
120            let hex = fs::read_to_string(entry.path())?;
121            if let Some(key) = key_from_hex(hex.trim()) {
122                keys.push(key);
123            }
124        }
125        Ok(keys)
126    }
127
128    /// Iterate every action record in the store.
129    pub fn actions(&self) -> Result<Vec<ActionResult>> {
130        let mut out = Vec::new();
131        let actions = self.root.join("actions");
132        if !actions.exists() {
133            return Ok(out);
134        }
135        for shard in fs::read_dir(&actions)? {
136            let shard = shard?.path();
137            if !shard.is_dir() {
138                continue;
139            }
140            for entry in fs::read_dir(&shard)? {
141                let path = entry?.path();
142                if path.extension().is_some_and(|e| e == "json")
143                    && let Ok(raw) = fs::read_to_string(&path)
144                    && let Ok(record) = serde_json::from_str::<ActionResult>(&raw)
145                {
146                    out.push(record);
147                }
148            }
149        }
150        Ok(out)
151    }
152
153    /// Remove a blob (eviction). The action records naming it stay —
154    /// the value is regenerable by re-running the computation.
155    pub fn evict_blob(&self, hash: &ContentHash) -> Result<()> {
156        let path = self.blob_path(hash);
157        if path.exists() {
158            fs::remove_file(&path)?;
159        }
160        Ok(())
161    }
162
163    /// Total bytes currently stored in the CAS.
164    pub fn cas_bytes(&self) -> Result<u64> {
165        fn dir_size(dir: &Path) -> u64 {
166            let Ok(entries) = fs::read_dir(dir) else {
167                return 0;
168            };
169            entries
170                .filter_map(|e| e.ok())
171                .map(|e| {
172                    let p = e.path();
173                    if p.is_dir() {
174                        dir_size(&p)
175                    } else {
176                        e.metadata().map(|m| m.len()).unwrap_or(0)
177                    }
178                })
179                .sum()
180        }
181        Ok(dir_size(&self.root.join("cas")))
182    }
183
184    fn store_computed(
185        &self,
186        key: &CacheKey,
187        value: &Value,
188        origin: &Origin,
189        compute: std::time::Duration,
190        deterministic: bool,
191    ) -> Result<()> {
192        let (bytes, hash) = encode_and_hash(value)?;
193        let output_bytes = bytes.len() as u64;
194        // Blob first, record last: the record is the commit point.
195        self.put_bytes_prehashed(&bytes, &hash)?;
196        let mut outputs = BTreeMap::new();
197        outputs.insert("output".to_string(), hash);
198        self.put_action(&ActionResult {
199            key: key.clone(),
200            outputs,
201            output_bytes,
202            compute_ms: compute.as_millis() as u64,
203            deterministic,
204            origin: origin.clone(),
205            created_at: Utc::now(),
206            last_accessed: Utc::now(),
207        })
208    }
209
210    fn put_bytes_prehashed(&self, bytes: &[u8], hash: &ContentHash) -> Result<()> {
211        let path = self.blob_path(hash);
212        if path.exists() {
213            return Ok(()); // dedup: identical content already stored
214        }
215        write_atomic(&path, bytes)
216    }
217}
218
219fn key_from_hex(hex: &str) -> Option<CacheKey> {
220    if hex.len() != 64 {
221        return None;
222    }
223    let mut digest = [0u8; 32];
224    for (i, byte) in digest.iter_mut().enumerate() {
225        *byte = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?;
226    }
227    Some(CacheKey(digest))
228}
229
230/// Temp file in the same directory → fsync → rename. Entry existence is
231/// the commit point; renames are idempotent for concurrent writers.
232fn write_atomic(path: &Path, data: &[u8]) -> Result<()> {
233    use std::io::Write;
234    use std::sync::atomic::{AtomicU64, Ordering};
235    static WRITE_SEQ: AtomicU64 = AtomicU64::new(0);
236
237    let parent = path
238        .parent()
239        .ok_or_else(|| SomaError::Cache("store path has no parent".into()))?;
240    fs::create_dir_all(parent)?;
241    let seq = WRITE_SEQ.fetch_add(1, Ordering::Relaxed);
242    let tmp = path.with_extension(format!("tmp-{}-{seq}", std::process::id()));
243    {
244        let mut f = fs::File::create(&tmp)?;
245        f.write_all(data)?;
246        f.sync_all()?;
247    }
248    if let Err(e) = fs::rename(&tmp, path) {
249        let _ = fs::remove_file(&tmp);
250        return Err(e.into());
251    }
252    Ok(())
253}
254
255impl BlobStore for FsActionStore {
256    fn put_bytes(&self, bytes: &[u8]) -> Result<ContentHash> {
257        let hash = ContentHash::blake3(bytes);
258        self.put_bytes_prehashed(bytes, &hash)?;
259        Ok(hash)
260    }
261
262    fn get_bytes(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
263        let path = self.blob_path(hash);
264        if !path.exists() {
265            return Ok(None);
266        }
267        let bytes = fs::read(&path)?;
268        if !hash.verify(&bytes) {
269            // Torn or corrupted blob: treat as absent (regenerable).
270            tracing::warn!(hash = %hash.to_hex(), "corrupt CAS blob, ignoring");
271            return Ok(None);
272        }
273        Ok(Some(bytes))
274    }
275
276    fn contains(&self, hash: &ContentHash) -> Result<bool> {
277        Ok(self.blob_path(hash).exists())
278    }
279}
280
281impl ActionCache for FsActionStore {
282    fn get_action(&self, key: &CacheKey) -> Result<Option<ActionResult>> {
283        let path = self.action_path(key);
284        if !path.exists() {
285            return Ok(None);
286        }
287        let raw = fs::read_to_string(&path)?;
288        match serde_json::from_str(&raw) {
289            Ok(record) => Ok(Some(record)),
290            Err(e) => {
291                tracing::warn!(key = %key, error = %e, "corrupt action record, ignoring");
292                Ok(None)
293            }
294        }
295    }
296
297    fn put_action(&self, result: &ActionResult) -> Result<()> {
298        let raw = serde_json::to_string(result)
299            .map_err(|e| SomaError::Cache(format!("action record encode: {e}")))?;
300        write_atomic(&self.action_path(&result.key), raw.as_bytes())
301    }
302}
303
304impl CacheStore for FsActionStore {
305    fn get(&self, key: &CacheKey) -> Result<Option<Value>> {
306        let Some(record) = self.get_action(key)? else {
307            return Ok(None);
308        };
309        let Some(hash) = record.outputs.get("output") else {
310            return Ok(None);
311        };
312        let Some(bytes) = self.get_bytes(hash)? else {
313            return Ok(None); // blob evicted → miss, caller recomputes
314        };
315        decode_value(&bytes).map(Some)
316    }
317
318    fn put(&self, key: &CacheKey, value: &Value) -> Result<()> {
319        self.store_computed(
320            key,
321            value,
322            &Origin::Ingested {
323                source: "unknown".into(),
324            },
325            std::time::Duration::ZERO,
326            true,
327        )
328    }
329
330    fn put_with_origin(&self, key: &CacheKey, value: &Value, origin: &Origin) -> Result<()> {
331        self.store_computed(key, value, origin, std::time::Duration::ZERO, true)
332    }
333
334    fn put_computed(
335        &self,
336        key: &CacheKey,
337        value: &Value,
338        origin: &Origin,
339        compute: std::time::Duration,
340        deterministic: bool,
341    ) -> Result<()> {
342        self.store_computed(key, value, origin, compute, deterministic)
343    }
344
345    fn exists(&self, key: &CacheKey) -> Result<bool> {
346        let Some(record) = self.get_action(key)? else {
347            return Ok(false);
348        };
349        match record.outputs.get("output") {
350            Some(hash) => self.contains(hash),
351            None => Ok(false),
352        }
353    }
354
355    fn remove(&self, key: &CacheKey) -> Result<()> {
356        // Removes the record only; shared blobs are left for GC
357        // (another record may name the same content).
358        let path = self.action_path(key);
359        if path.exists() {
360            fs::remove_file(&path)?;
361        }
362        Ok(())
363    }
364
365    fn metadata(&self, key: &CacheKey) -> Result<Option<EntryMeta>> {
366        Ok(self.get_action(key)?.map(|record| EntryMeta {
367            key: record.key,
368            size_bytes: record.output_bytes,
369            created_at: record.created_at,
370            last_accessed: record.last_accessed,
371            ttl: None,
372            origin: record.origin,
373        }))
374    }
375}
376
377#[cfg(test)]
378mod tests {
379    use super::*;
380    use std::sync::Arc;
381    use std::sync::atomic::{AtomicU64, Ordering};
382
383    static COUNTER: AtomicU64 = AtomicU64::new(0);
384
385    fn temp_root() -> PathBuf {
386        let id = COUNTER.fetch_add(1, Ordering::Relaxed);
387        let dir = std::env::temp_dir().join(format!("soma_fs_store_{}_{id}", std::process::id()));
388        let _ = fs::remove_dir_all(&dir);
389        dir
390    }
391
392    #[test]
393    fn roundtrip_and_dedup() {
394        let root = temp_root();
395        let store = FsActionStore::new(&root).unwrap();
396        let value = Value::tensor(vec![1.0; 1000], vec![1000]);
397
398        let k1 = CacheKey::hash_data(b"action-1");
399        let k2 = CacheKey::hash_data(b"action-2");
400        store.put(&k1, &value).unwrap();
401        store.put(&k2, &value).unwrap(); // same content, different action
402
403        assert_eq!(store.get(&k1).unwrap().unwrap(), value);
404        assert_eq!(store.get(&k2).unwrap().unwrap(), value);
405
406        // Both records point at ONE deduplicated blob.
407        let r1 = store.get_action(&k1).unwrap().unwrap();
408        let r2 = store.get_action(&k2).unwrap().unwrap();
409        assert_eq!(r1.outputs["output"], r2.outputs["output"]);
410        let blob_count = walk_count(&root.join("cas"), "bin");
411        assert_eq!(blob_count, 1, "identical outputs must share one blob");
412
413        let _ = fs::remove_dir_all(&root);
414    }
415
416    #[test]
417    fn eviction_keeps_record_and_refills() {
418        let root = temp_root();
419        let store = FsActionStore::new(&root).unwrap();
420        let key = CacheKey::hash_data(b"expensive");
421        let value = Value::tensor(vec![2.0; 64], vec![64]);
422        store
423            .put_computed(
424                &key,
425                &value,
426                &Origin::Computed {
427                    node_id: "n".into(),
428                    run_id: "r".into(),
429                },
430                std::time::Duration::from_secs(3600),
431                true,
432            )
433            .unwrap();
434
435        let record = store.get_action(&key).unwrap().unwrap();
436        assert_eq!(record.compute_ms, 3_600_000);
437        let hash = record.outputs["output"];
438
439        // Evict the blob: record survives, lookup misses (recompute path).
440        store.evict_blob(&hash).unwrap();
441        assert!(store.get_action(&key).unwrap().is_some());
442        assert!(store.get(&key).unwrap().is_none());
443        assert!(!store.exists(&key).unwrap());
444
445        // Recompute re-fills the SAME content address; entry live again.
446        store.put(&key, &value).unwrap();
447        assert_eq!(store.get(&key).unwrap().unwrap(), value);
448
449        let _ = fs::remove_dir_all(&root);
450    }
451
452    #[test]
453    fn corrupt_blob_is_a_miss_not_an_error() {
454        let root = temp_root();
455        let store = FsActionStore::new(&root).unwrap();
456        let key = CacheKey::hash_data(b"c");
457        store.put(&key, &Value::tensor(vec![1.0], vec![1])).unwrap();
458
459        let hash = store.get_action(&key).unwrap().unwrap().outputs["output"];
460        fs::write(store.blob_path(&hash), b"garbage").unwrap();
461
462        assert!(store.get(&key).unwrap().is_none());
463        let _ = fs::remove_dir_all(&root);
464    }
465
466    #[test]
467    fn format_version_guard() {
468        let root = temp_root();
469        fs::create_dir_all(&root).unwrap();
470        fs::write(root.join("format.json"), r#"{"version": 99}"#).unwrap();
471        assert!(FsActionStore::new(&root).is_err());
472        let _ = fs::remove_dir_all(&root);
473    }
474
475    #[test]
476    fn pins_roundtrip() {
477        let root = temp_root();
478        let store = FsActionStore::new(&root).unwrap();
479        let key = CacheKey::hash_data(b"best-model");
480        store.pin("best-model", &key).unwrap();
481        assert_eq!(store.pinned().unwrap(), vec![key.clone()]);
482        assert!(store.pin("../escape", &key).is_err());
483        let _ = fs::remove_dir_all(&root);
484    }
485
486    #[test]
487    fn concurrent_writers_same_key() {
488        let root = temp_root();
489        let store = Arc::new(FsActionStore::new(&root).unwrap());
490        let key = CacheKey::hash_data(b"contended");
491        let value = Value::tensor(vec![7.0; 512], vec![512]);
492
493        std::thread::scope(|s| {
494            for _ in 0..8 {
495                let store = store.clone();
496                let key = key.clone();
497                let value = value.clone();
498                s.spawn(move || store.put(&key, &value).unwrap());
499            }
500        });
501        assert_eq!(store.get(&key).unwrap().unwrap(), value);
502        let _ = fs::remove_dir_all(&root);
503    }
504
505    #[test]
506    fn survives_restart() {
507        let root = temp_root();
508        let key = CacheKey::hash_data(b"persist");
509        let value = Value::json(serde_json::json!({"w": [1, 2, 3]}));
510        {
511            let store = FsActionStore::new(&root).unwrap();
512            store.put(&key, &value).unwrap();
513        }
514        {
515            let store = FsActionStore::new(&root).unwrap();
516            assert_eq!(store.get(&key).unwrap().unwrap(), value);
517        }
518        let _ = fs::remove_dir_all(&root);
519    }
520
521    fn walk_count(dir: &Path, ext: &str) -> usize {
522        let Ok(entries) = fs::read_dir(dir) else {
523            return 0;
524        };
525        entries
526            .filter_map(|e| e.ok())
527            .map(|e| {
528                let p = e.path();
529                if p.is_dir() {
530                    walk_count(&p, ext)
531                } else if p.extension().is_some_and(|x| x == ext) {
532                    1
533                } else {
534                    0
535                }
536            })
537            .sum()
538    }
539}
somatize_runtime/cache/fs_store.rs

somatize_runtime/cache/
fs_store.rs
