A caching store in front of any store 454236d1108d

- **A caching store, and the operation head on a store (#68).** `strata::CachingStore` puts a local directory in front of any store: reads are kept, writes go to the remote first, the cache stays under a size limit by removing what was used least recently, and references always come from the remote. A repository kept on a store other than this disk's directory now holds its operation head there, moved by compare-and-swap with merge and retry.

// ---------------------------------------------------------------- caching

/// A store in front of another (oxygen#68): objects are read through a

/// local directory and written through it, to the remote first, so the

/// remote always holds everything and the cache can be dropped at any

/// time. Objects never change, so a cached one is always right (and every

/// read is hashed again above this layer). The cache is bounded: past its

/// limit the objects used least recently are removed. An object larger

/// than an eighth of the limit is not cached; it is read by range from

/// the remote. References are never cached: they are read from and

/// written to the remote every time, so compare-and-swap is the remote's.

pub struct CachingStore {

    cache: LocalStore,

    remote: std::sync::Arc<dyn Store>,

    limit: u64,

    /// Bytes in the cache, counted on first need and kept up to date.

    used: Mutex<Option<u64>>,

}

/// How stale a cached object's use time may get before a read renews it.

const TOUCH_AFTER: Duration = Duration::from_secs(60);

impl CachingStore {

    /// `remote` behind a cache in the directory `dir` of at most `limit`

    /// bytes.

    pub fn open(

        remote: std::sync::Arc<dyn Store>,

        dir: impl Into<PathBuf>,

        limit: u64,

    ) -> Result<CachingStore> {

        Ok(CachingStore {

            cache: LocalStore::open(dir)?,

            remote,

            limit,

            used: Mutex::new(None),

        })

    }

    pub fn remote(&self) -> &std::sync::Arc<dyn Store> {

        &self.remote

    }

    /// Bytes the cache holds now.

    pub fn cached_bytes(&self) -> Result<u64> {

        let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());

        if used.is_none() {

            *used = Some(self.cache.list("")?.iter().map(|e| e.size).sum());

        }

        Ok(used.unwrap_or(0))

    }

    /// Whether an object of `size` bytes is kept in the cache at all.

    fn cacheable(&self, size: u64) -> bool {

        size <= self.limit / 8

    }

    /// Keep a copy of an object the remote holds.

    fn keep(&self, key: &str, bytes: &[u8]) -> Result<()> {

        if !self.cacheable(bytes.len() as u64) {

            return Ok(());

        }

        match self.cache.put_new(key, bytes) {

            Ok(()) => self.added(bytes.len() as u64),

            Err(Error::Exists(_)) => Ok(()),

            Err(e) => Err(e),

        }

    }

    /// Count `n` new bytes, and remove the least recently used objects

    /// once the cache is past its limit, down to nine tenths of it.

    fn added(&self, n: u64) -> Result<()> {

        let total = {

            let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());

            let total = match *used {

                Some(u) => u + n,

                // Counted now, with the new object already in place.

                None => self.cache.list("")?.iter().map(|e| e.size).sum(),

            };

            *used = Some(total);

            total

        };

        if total <= self.limit {

            return Ok(());

        }

        let mut entries = self.cache.list("")?;

        entries.sort_by_key(|e| e.modified_ms.unwrap_or(0));

        let mut left: u64 = entries.iter().map(|e| e.size).sum();

        let target = self.limit / 10 * 9;

        for e in entries {

            if left <= target {

                break;

            }

            self.cache.delete(&e.key)?;

            left -= e.size;

        }

        *self.used.lock().unwrap_or_else(|p| p.into_inner()) = Some(left);

        Ok(())

    }

    /// Mark a cached object used now, at most once a minute.

    fn touch(&self, key: &str) {

        let Ok(path) = self.cache.path(key) else {

            return;

        };

        let stale = fs::metadata(&path)

            .and_then(|m| m.modified())

            .ok()

            .and_then(|t| t.elapsed().ok())

            .is_some_and(|age| age > TOUCH_AFTER);

        if stale {

            if let Ok(f) = fs::File::options().append(true).open(&path) {

                let _ = f.set_modified(std::time::SystemTime::now());

            }

        }

    }

}

impl Store for CachingStore {

    fn describe(&self) -> String {

        format!(

            "{} (cached in {})",

            self.remote.describe(),

            self.cache.root().display()

        )

    }

    fn has(&self, key: &str) -> Result<bool> {

        Ok(self.cache.has(key)? || self.remote.has(key)?)

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        match self.cache.get(key) {

            Ok(bytes) => {

                self.touch(key);

                return Ok(bytes);

            }

            // Not cached, or removed by another process's eviction meanwhile.

            Err(Error::NotFound(_)) => {}

            Err(e) => return Err(e),

        }

        let bytes = self.remote.get(key)?;

        self.keep(key, &bytes)?;

        Ok(bytes)

    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {

        match self.cache.get_range(key, range.clone()) {

            Ok(bytes) => {

                self.touch(key);

                return Ok(bytes);

            }

            Err(Error::NotFound(_)) => {}

            Err(e) if self.cache.has(key)? => return Err(e),

            Err(_) => {}

        }

        // A small object is fetched whole and kept; a large one is read by

        // range, every time.

        match self.remote.stat(key)? {

            Some(e) if self.cacheable(e.size) => {

                let bytes = self.get(key)?;

                let len = bytes.len() as u64;

                if range.end > len || range.start > range.end {

                    return Err(Error::Backend(format!(

                        "{key}: bytes {}..{} asked of an object of {len} bytes",

                        range.start, range.end

                    )));

                }

                Ok(bytes[range.start as usize..range.end as usize].to_vec())

            }

            Some(_) => self.remote.get_range(key, range),

            None => Err(Error::NotFound(key.to_string())),

        }

    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {

        self.remote.put_new(key, bytes)?;

        self.keep(key, bytes)

    }

    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {

        self.remote.put_many_new(objects)?;

        for (key, bytes) in objects {

            self.keep(key, bytes)?;

        }

        Ok(())

    }

    fn put_new_file(&self, key: &str, path: &Path) -> Result<()> {

        let size = fs::metadata(path)?.len();

        let copy = if self.cacheable(size) {

            Some(fs::read(path)?)

        } else {

            None

        };

        self.remote.put_new_file(key, path)?;

        match copy {

            Some(bytes) => self.keep(key, &bytes),

            None => Ok(()),

        }

    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {

        self.remote.list(prefix)

    }

    fn stat(&self, key: &str) -> Result<Option<Entry>> {

        self.remote.stat(key)

    }

    fn delete(&self, key: &str) -> Result<()> {

        self.remote.delete(key)?;

        let size = self.cache.stat(key)?.map(|e| e.size);

        self.cache.delete(key)?;

        if let Some(n) = size {

            let mut used = self.used.lock().unwrap_or_else(|p| p.into_inner());

            if let Some(u) = used.as_mut() {

                *u = u.saturating_sub(n);

            }

        }

        Ok(())

    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {

        self.remote.read_ref(key)

    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {

        self.remote.write_ref(key, bytes, expected)

    }

}

    }

    #[test]

    fn a_caching_store_conforms_and_reads_and_writes_through() {

        let dir = tempfile::tempdir().unwrap();

        let remote: Arc<dyn Store> = Arc::new(MemoryStore::new());

        let store = CachingStore::open(remote.clone(), dir.path().join("cache"), 8 << 20).unwrap();

        assert_eq!(conformance(&store, "probe").unwrap().len(), 8);

        assert_eq!(

            store.cached_bytes().unwrap(),

            0,

            "the probe left nothing cached"

        );

        // Written through: the remote has it, and so does the cache.

        store.put_new("loose/aa", b"written").unwrap();

        assert_eq!(remote.get("loose/aa").unwrap(), b"written");

        assert!(store.cache.has("loose/aa").unwrap());

        // Read through: an object only the remote has is kept on first read.

        remote.put_new("loose/bb", b"remote only").unwrap();

        assert!(!store.cache.has("loose/bb").unwrap());

        assert_eq!(store.get_range("loose/bb", 0..6).unwrap(), b"remote");

        assert!(store.cache.has("loose/bb").unwrap());

        // A cached object read after the remote lost it is still served.

        remote.delete("loose/bb").unwrap();

        assert_eq!(store.get("loose/bb").unwrap(), b"remote only");

        // References are always the remote's.

        let v = remote.write_ref("refs", b"one", None).unwrap();

        assert_eq!(

            store.read_ref("refs").unwrap(),

            Some((b"one".to_vec(), v.clone()))

        );

        remote.write_ref("refs", b"two", Some(&v)).unwrap();

        assert_eq!(store.read_ref("refs").unwrap().unwrap().0, b"two");

        assert!(matches!(

            store.write_ref("refs", b"three", Some(&v)),

            Err(Error::Conflict(_))

        ));

        assert!(!store.cache.has("refs").unwrap());

    }

    #[test]

    fn a_caching_store_stays_within_its_limit() {

        let dir = tempfile::tempdir().unwrap();

        let remote: Arc<dyn Store> = Arc::new(MemoryStore::new());

        let store = CachingStore::open(remote.clone(), dir.path().join("cache"), 8000).unwrap();

        for i in 0..20 {

            store

                .put_new(&format!("loose/{i:02}"), &[i as u8; 900])

                .unwrap();

            assert!(store.cached_bytes().unwrap() <= 8000);

        }

        assert_eq!(

            remote.list("loose/").unwrap().len(),

            20,

            "the remote holds everything"

        );

        let cached = store.cache.list("loose/").unwrap();

        assert!(cached.len() < 20 && cached.len() >= 7, "{}", cached.len());

        assert!(

            cached.iter().any(|e| e.key == "loose/19"),

            "the newest is kept"

        );

        for i in 0..20 {

            assert_eq!(

                store.get(&format!("loose/{i:02}")).unwrap(),

                vec![i as u8; 900]

            );

        }

        // An object over an eighth of the limit is read by range, not kept.

        remote.put_new("pack-big", &[7u8; 2000]).unwrap();

        assert_eq!(store.get_range("pack-big", 10..20).unwrap(), vec![7u8; 10]);

        assert_eq!(store.get("pack-big").unwrap().len(), 2000);

        assert!(!store.cache.has("pack-big").unwrap());

        // Counting from scratch agrees with what was kept track of.

        let fresh = CachingStore::open(remote, dir.path().join("cache"), 8000).unwrap();

        assert_eq!(fresh.cached_bytes().unwrap(), store.cached_bytes().unwrap());

- **The operation head on a store** (B2, 2026-10-05): a repository whose

  objects are on any store but this disk's directory keeps its

  operation head in that store's `ops/head` reference record. Recording

  an operation reads the record and its version, and moves it only if

  the version is unchanged. When another process moved it first, the

  two operations are merged by the same rules as on disk, and the write

  is tried again (eight times at most). A repository on this disk keeps

  `.oxy/ops/head` under its lock file, as before, so older oxy builds

  share it safely.

- **The caching store** (B2, 2026-10-05): `strata::CachingStore` puts a

  directory store in front of any other. Reads go through it: a missing

  object is fetched from the remote and kept. Writes go to the remote

  first and then the cache, so the remote always holds everything and

  the cache can be deleted at any time. The cache has a size limit; past

  it the objects used least recently are removed, down to nine tenths of

  the limit. An object larger than an eighth of the limit is not kept:

  ranges of it are read from the remote each time. References are never

  cached, so compare-and-swap is always the remote's.