Strata: the Store trait, a directory store, a memory store, and the conformance check a5f76c79666c

//! Oxygen Strata (docs/stores.md): the storage a repository can be kept

//! on, as a trait, so that a bucket, a directory or a hub's disk all look

//! the same to the code that pushes to and fetches from them.

//!

//! A store has two halves. **Objects** are written once and never

//! change: packs and their indexes, named by content, which any store

//! can hold safely. **References** are the one mutable record (bookmarks,

//! tags, the policy), updated only by compare-and-swap on a version the

//! store hands out: an ETag on S3, a generation on GCS, a content hash on

//! a directory. Two writers can therefore share a store with no lock

//! server: the loser of a race reads again, merges, and tries again.

//!

//! The working repository (`.oxy/`) is not a `Store`; it is read and

//! written directly, as before. A store is what a *remote* is kept on.

//! This module holds the trait, a directory store, an in-memory store

//! for tests, and the conformance check every store must pass. Cloud

//! stores live in a crate of their own, because of what they pull in.

use std::collections::BTreeMap;

use std::fs;

use std::io::{self, Read, Seek, SeekFrom, Write};

use std::ops::Range;

use std::path::{Path, PathBuf};

use std::sync::Mutex;

use std::time::{Duration, Instant};

/// The version of a reference record as the store reports it: opaque,

/// compared only for equality, and only ever valid for the store that

/// issued it. Writing the same bytes again may keep the version (S3's

/// ETag and a directory's content hash are both of the content): a

/// compare-and-swap guards against losing a *different* record, which is

/// all it needs to.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Version(pub String);

/// One key in a listing.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Entry {

    pub key: String,

    pub size: u64,

}

#[derive(Debug, thiserror::Error)]

pub enum Error {

    #[error("{0}: not in the store")]

    NotFound(String),

    #[error("{0}: already in the store")]

    Exists(String),

    #[error("{0}: changed since it was read")]

    Conflict(String),

    #[error("bad store key {0:?}: {1}")]

    Key(String, &'static str),

    #[error("{0}: busy (another writer holds it)")]

    Busy(String),

    #[error(transparent)]

    Io(#[from] io::Error),

    #[error("{0}")]

    Backend(String),

}

pub type Result<T> = std::result::Result<T, Error>;

/// Longest key a store accepts; S3 allows 1024 bytes.

pub const MAX_KEY_LEN: usize = 1024;

/// Refuse a key that is not a plain relative path: `/`-separated

/// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same

/// rule on every store, so a key written to one can be read from another,

/// and so a key can never escape a directory store's root.

pub fn check_key(key: &str) -> Result<()> {

    let bad = |why| Error::Key(key.to_string(), why);

    if key.is_empty() {

        return Err(bad("empty"));

    }

    if key.len() > MAX_KEY_LEN {

        return Err(bad("longer than 1024 bytes"));

    }

    for seg in key.split('/') {

        if seg.is_empty() {

            return Err(bad(

                "an empty segment (a leading, trailing or doubled slash)",

            ));

        }

        if seg == "." || seg == ".." {

            return Err(bad("a . or .. segment"));

        }

        if !seg

            .bytes()

            .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-')

        {

            return Err(bad("a character outside A-Z a-z 0-9 . _ -"));

        }

    }

    Ok(())

}

/// Where a repository's packs and references are kept. Every method is

/// synchronous and may be called from several threads at once.

pub trait Store: Send + Sync {

    /// A name for messages: the URL, or the directory.

    fn describe(&self) -> String;

    // ---- objects: written once, never changed

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

    /// The whole object; `NotFound` when there is none.

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

    /// Bytes `range` of an object; an error when the range runs past its end.

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

    /// Write an object that does not exist yet; `Exists` when it does, and

    /// then nothing is changed. Readers never see a partial object.

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

    /// Every key under `prefix` (a key prefix, not necessarily a whole

    /// segment), sorted, with sizes.

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

    /// Remove an object; fine when it is already gone.

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

    // ---- references: one small record, updated by compare-and-swap

    /// The record and its current version, or `None` when it was never

    /// written.

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

    /// Replace the record only if its version is still `expected`

    /// (`None`: only if it does not exist yet). `Conflict` otherwise, and

    /// then nothing is changed. Returns the new version.

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

}

/// A reference record's version on stores that have no native one: the

/// hash of its bytes, which two different records never share.

fn content_version(bytes: &[u8]) -> Version {

    Version(blake3::hash(bytes).to_hex().to_string())

}

fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> {

    if current == expected {

        Ok(())

    } else {

        Err(Error::Conflict(key.to_string()))

    }

}

// ---------------------------------------------------------------- directory

/// A store in a directory: each key a file under the root. Create-only

/// writes go through a hard link, which the file system refuses when the

/// name is taken, and a reference is updated under a lock file beside it.

/// Several processes on one machine can share it; it is also what the

/// tests and `oxy remote check` use.

pub struct LocalStore {

    root: PathBuf,

}

/// How long a reference's lock may be held before it counts as abandoned

/// by a process that died, and how long a writer waits for it.

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

const LOCK_WAIT: Duration = Duration::from_secs(10);

impl LocalStore {

    /// Open the store at `root`, creating the directory if needed.

    pub fn open(root: impl Into<PathBuf>) -> Result<LocalStore> {

        let root = root.into();

        fs::create_dir_all(&root)?;

        Ok(LocalStore { root })

    }

    pub fn root(&self) -> &Path {

        &self.root

    }

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

        check_key(key)?;

        Ok(self.root.join(key))

    }

    fn temp_beside(&self, path: &Path) -> PathBuf {

        let mut nonce = [0u8; 6];

        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);

        let name = path

            .file_name()

            .map(|n| n.to_string_lossy().into_owned())

            .unwrap_or_default();

        path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce)))

    }

    fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec<Entry>) -> Result<()> {

        let entries = match fs::read_dir(dir) {

            Ok(e) => e,

            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),

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

        };

        for entry in entries {

            let entry = entry?;

            let name = entry.file_name().to_string_lossy().into_owned();

            // Temporary and lock files are not objects.

            if name.starts_with('.') {

                continue;

            }

            let key = if rel.is_empty() {

                name.clone()

            } else {

                format!("{rel}/{name}")

            };

            let ty = entry.file_type()?;

            if ty.is_dir() {

                // Descend only where the prefix could still match.

                let dir_prefix = format!("{key}/");

                if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) {

                    self.list_dir(&entry.path(), &key, prefix, out)?;

                }

            } else if ty.is_file() && key.starts_with(prefix) {

                out.push(Entry {

                    key,

                    size: entry.metadata()?.len(),

                });

            }

        }

        Ok(())

    }

    /// Take `path.lock`, waiting for another writer, and removing a lock

    /// a dead process left behind.

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

        let lock = path.with_file_name(format!(

            ".{}.lock",

            path.file_name()

                .map(|n| n.to_string_lossy().into_owned())

                .unwrap_or_default()

        ));

        let started = Instant::now();

        loop {

            match fs::OpenOptions::new()

                .write(true)

                .create_new(true)

                .open(&lock)

            {

                Ok(mut f) => {

                    let _ = writeln!(f, "{}", std::process::id());

                    return Ok(lock);

                }

                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {

                    if let Ok(m) = fs::metadata(&lock) {

                        let age = m.modified().ok().and_then(|t| t.elapsed().ok());

                        if age.is_some_and(|a| a > LOCK_STALE) {

                            let _ = fs::remove_file(&lock);

                            continue;

                        }

                    }

                    if started.elapsed() > LOCK_WAIT {

                        return Err(Error::Busy(key.to_string()));

                    }

                    std::thread::sleep(Duration::from_millis(10));

                }

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

            }

        }

    }

}

impl Store for LocalStore {

    fn describe(&self) -> String {

        self.root.display().to_string()

    }

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

        Ok(self.path(key)?.is_file())

    }

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

        let path = self.path(key)?;

        match fs::read(&path) {

            Ok(b) => Ok(b),

            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())),

            Err(e) => Err(e.into()),

        }

    }

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

        let path = self.path(key)?;

        let mut f = match fs::File::open(&path) {

            Ok(f) => f,

            Err(e) if e.kind() == io::ErrorKind::NotFound => {

                return Err(Error::NotFound(key.to_string()))

            }

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

        };

        let len = f.metadata()?.len();

        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

            )));

        }

        f.seek(SeekFrom::Start(range.start))?;

        let mut out = vec![0u8; (range.end - range.start) as usize];

        f.read_exact(&mut out)?;

        Ok(out)

    }

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

        let path = self.path(key)?;

        if path.exists() {

            return Err(Error::Exists(key.to_string()));

        }

        if let Some(dir) = path.parent() {

            fs::create_dir_all(dir)?;

        }

        // Written beside its name, then linked into it: the link fails if

        // the name was taken meanwhile, and a reader never sees a part.

        let tmp = self.temp_beside(&path);

        let written = (|| {

            let mut f = fs::File::create(&tmp)?;

            f.write_all(bytes)?;

            f.sync_all()

        })();

        let linked = written.and_then(|()| fs::hard_link(&tmp, &path));

        let _ = fs::remove_file(&tmp);

        match linked {

            Ok(()) => {

                if let Some(dir) = path.parent() {

                    crate::durable::sync_dir(dir)?;

                }

                Ok(())

            }

            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {

                Err(Error::Exists(key.to_string()))

            }

            Err(e) => Err(e.into()),

        }

    }

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

        let mut out = Vec::new();

        self.list_dir(&self.root, "", prefix, &mut out)?;

        out.sort_by(|a, b| a.key.cmp(&b.key));

        Ok(out)

    }

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

        let path = self.path(key)?;

        match fs::remove_file(&path) {

            Ok(()) => Ok(()),

            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),

            Err(e) => Err(e.into()),

        }

    }

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

        let path = self.path(key)?;

        match fs::read(&path) {

            Ok(b) => {

                let v = content_version(&b);

                Ok(Some((b, v)))

            }

            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),

            Err(e) => Err(e.into()),

        }

    }

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

        let path = self.path(key)?;

        if let Some(dir) = path.parent() {

            fs::create_dir_all(dir)?;

        }

        let lock = self.lock(key, &path)?;

        let result = (|| {

            let current = self.read_ref(key)?.map(|(_, v)| v);

            check_expected(key, current.as_ref(), expected)?;

            crate::durable::write_atomic(&path, bytes)?;

            Ok(content_version(bytes))

        })();

        let _ = fs::remove_file(&lock);

        result

    }

}

// ---------------------------------------------------------------- memory

/// A store in memory, for tests: the conditional writes behave as the

/// cloud stores' do.

#[derive(Default)]

pub struct MemoryStore {

    objects: Mutex<BTreeMap<String, Vec<u8>>>,

}

impl MemoryStore {

    pub fn new() -> MemoryStore {

        MemoryStore::default()

    }

    fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> {

        self.objects.lock().unwrap_or_else(|p| p.into_inner())

    }

}

impl Store for MemoryStore {

    fn describe(&self) -> String {

        "memory".into()

    }

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

        check_key(key)?;

        Ok(self.map().contains_key(key))

    }

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

        check_key(key)?;

        self.map()

            .get(key)

            .cloned()

            .ok_or_else(|| Error::NotFound(key.to_string()))

    }

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

        let all = self.get(key)?;

        let len = all.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(all[range.start as usize..range.end as usize].to_vec())

    }

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

        check_key(key)?;

        let mut map = self.map();

        if map.contains_key(key) {

            return Err(Error::Exists(key.to_string()));

        }

        map.insert(key.to_string(), bytes.to_vec());

        Ok(())

    }

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

        Ok(self

            .map()

            .range(prefix.to_string()..)

            .take_while(|(k, _)| k.starts_with(prefix))

            .map(|(k, v)| Entry {

                key: k.clone(),

                size: v.len() as u64,

            })

            .collect())

    }

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

        check_key(key)?;

        self.map().remove(key);

        Ok(())

    }

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

        check_key(key)?;

        Ok(self.map().get(key).map(|b| (b.clone(), content_version(b))))

    }

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

        check_key(key)?;

        let mut map = self.map();

        let current = map.get(key).map(|b| content_version(b));

        check_expected(key, current.as_ref(), expected)?;

        map.insert(key.to_string(), bytes.to_vec());

        Ok(content_version(bytes))

    }

}

// ---------------------------------------------------------------- conformance

/// Exercise a store under `prefix` (which must be free, and is left

/// empty) and report what passed: object round trips, ranges, create-only

/// refusal, listing, deletion, and compare-and-swap on a reference. Every

/// store runs this in its tests, and `oxy remote check` runs it against a

/// remote before trusting it with a repository.

pub fn conformance(store: &dyn Store, prefix: &str) -> Result<Vec<&'static str>> {

    let mut passed = Vec::new();

    let key = |name: &str| format!("{prefix}/{name}");

    let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe()));

    let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) };

    expect(

        store.list(prefix)?.is_empty(),

        "the probe prefix is not empty",

    )?;

    expect(

        !store.has(&key("a"))?,

        "has() of a missing object says true",

    )?;

    expect(

        matches!(store.get(&key("a")), Err(Error::NotFound(_))),

        "get() of a missing object is not NotFound",

    )?;

    passed.push("missing objects are reported missing");

    let a: Vec<u8> = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect();

    store.put_new(&key("a"), &a)?;

    expect(store.has(&key("a"))?, "has() after put_new says false")?;

    expect(

        store.get(&key("a"))? == a,

        "get() returns other bytes than put_new wrote",

    )?;

    passed.push("an object round-trips");

    expect(

        store.get_range(&key("a"), 1000..1500)? == a[1000..1500],

        "get_range() returns the wrong bytes",

    )?;

    expect(

        store.get_range(&key("a"), 0..0)?.is_empty(),

        "an empty range is not empty",

    )?;

    expect(

        store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(),

        "a range past the end is not refused",

    )?;

    passed.push("ranges read correctly and are bounded");

    expect(

        matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))),

        "put_new() over an existing object is not refused",

    )?;

    expect(

        store.get(&key("a"))? == a,

        "a refused put_new() changed the object",

    )?;

    passed.push("objects are written once");

    store.put_new(&key("dir/b"), b"bb")?;

    let listed = store.list(prefix)?;

    expect(

        listed

            == vec![

                Entry {

                    key: key("a"),

                    size: a.len() as u64,

                },

                Entry {

                    key: key("dir/b"),

                    size: 2,

                },

            ],

        "list() does not return the two objects, sorted, with their sizes",

    )?;

    expect(

        store.list(&key("dir/"))?

            == vec![Entry {

                key: key("dir/b"),

                size: 2,

            }],

        "list() of a sub-prefix is wrong",

    )?;

    expect(

        store.list(&key("nothing/"))?.is_empty(),

        "list() of an unused prefix is not empty",

    )?;

    passed.push("listing by prefix");

    store.delete(&key("dir/b"))?;

    expect(

        !store.has(&key("dir/b"))?,

        "an object is still there after delete()",

    )?;

    store.delete(&key("dir/b"))?;

    passed.push("deletion, and deleting again, succeed");

    let r = key("refs");

    expect(

        store.read_ref(&r)?.is_none(),

        "an unwritten reference reads as something",

    )?;

    let v1 = store.write_ref(&r, b"one", None)?;

    expect(

        matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))),

        "a second create-only write of the reference is not refused",

    )?;

    let (bytes, seen) = store

        .read_ref(&r)?

        .ok_or_else(|| fail("the reference vanished"))?;

    expect(

        bytes == b"one" && seen == v1,

        "the reference reads back wrong",

    )?;

    let v2 = store.write_ref(&r, b"two", Some(&v1))?;

    expect(v2 != v1, "the version did not change on write")?;

    expect(

        matches!(

            store.write_ref(&r, b"three", Some(&v1)),

            Err(Error::Conflict(_))

        ),

        "a write with a stale version is not refused",

    )?;

    let (bytes, seen) = store

        .read_ref(&r)?

        .ok_or_else(|| fail("the reference vanished"))?;

    expect(

        bytes == b"two" && seen == v2,

        "a refused write changed the reference",

    )?;

    passed.push("references update by compare-and-swap");

    store.delete(&key("a"))?;

    store.delete(&r)?;

    expect(

        store.list(prefix)?.is_empty(),

        "the probe prefix is not empty after cleanup",

    )?;

    passed.push("cleanup");

    Ok(passed)

}

#[cfg(test)]

mod tests {

    use super::*;

    use std::sync::Arc;

    #[test]

    fn keys_are_plain_relative_paths() {

        for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] {

            check_key(ok).unwrap();

        }

        for bad in [

            "", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "ä",

        ] {

            assert!(check_key(bad).is_err(), "{bad:?}");

        }

        assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err());

    }

    #[test]

    fn the_memory_store_conforms() {

        let store = MemoryStore::new();

        let passed = conformance(&store, "probe").unwrap();

        assert_eq!(passed.len(), 8, "{passed:?}");

    }

    #[test]

    fn the_directory_store_conforms() {

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

        let store = LocalStore::open(dir.path().join("bucket")).unwrap();

        let passed = conformance(&store, "repos/one").unwrap();

        assert_eq!(passed.len(), 8, "{passed:?}");

        // Nothing but the (empty) directories and no stray temporaries.

        let stray: Vec<_> = walkdir(dir.path())

            .into_iter()

            .filter(|p| p.is_file())

            .collect();

        assert!(stray.is_empty(), "{stray:?}");

    }

    fn walkdir(dir: &Path) -> Vec<PathBuf> {

        let mut out = vec![];

        if let Ok(rd) = fs::read_dir(dir) {

            for e in rd.flatten() {

                let p = e.path();

                if p.is_dir() {

                    out.extend(walkdir(&p));

                } else {

                    out.push(p);

                }

            }

        }

        out

    }

    #[test]

    fn a_directory_store_refuses_keys_that_leave_its_root() {

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

        let store = LocalStore::open(dir.path().join("bucket")).unwrap();

        assert!(matches!(

            store.put_new("../escape", b"x"),

            Err(Error::Key(..))

        ));

        assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..))));

        assert!(!dir.path().join("escape").exists());

    }

    /// Many threads race to move a reference from the same version: exactly

    /// one wins each round, on both stores.

    #[test]

    fn compare_and_swap_admits_one_writer_per_round() {

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

        let local: Arc<dyn Store> = Arc::new(LocalStore::open(dir.path()).unwrap());

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

        for store in [local, memory] {

            let v0 = store.write_ref("refs", b"0", None).unwrap();

            let wins: Vec<bool> = (0..8)

                .map(|i| {

                    let store = store.clone();

                    let v0 = v0.clone();

                    std::thread::spawn(move || {

                        store

                            .write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0))

                            .is_ok()

                    })

                })

                .collect::<Vec<_>>()

                .into_iter()

                .map(|h| h.join().unwrap())

                .collect();

            assert_eq!(

                wins.iter().filter(|w| **w).count(),

                1,

                "{}: {wins:?}",

                store.describe()

            );

            let (bytes, _) = store.read_ref("refs").unwrap().unwrap();

            assert_ne!(bytes, b"0");

        }

    }

    #[test]

    fn a_stale_lock_is_broken_and_a_live_one_waited_for() {

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

        let store = LocalStore::open(dir.path()).unwrap();

        let lock = dir.path().join(".refs.lock");

        fs::write(&lock, "dead").unwrap();

        let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5);

        fs::File::open(&lock).unwrap().set_modified(old).unwrap();

        store.write_ref("refs", b"after a crash", None).unwrap();

        assert!(!lock.exists());

    }

    #[test]

    fn temporaries_and_locks_are_not_listed() {

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

        let store = LocalStore::open(dir.path()).unwrap();

        store.put_new("packs/a", b"a").unwrap();

        fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap();

        fs::write(dir.path().join(".refs.lock"), "1").unwrap();

        let keys: Vec<String> = store.list("").unwrap().into_iter().map(|e| e.key).collect();

        assert_eq!(keys, ["packs/a"]);

    }

}

pub mod strata;

- **The `Store` trait** (`oxy_core::strata`, built 2026-10-03). It has

  two halves:

  - **objects**: `has`, `get`, `get_range`, `put_new` (create only),

    `list` by prefix, and `delete`;

  - **references**: `read_ref() -> (bytes, version)` and

    `write_ref(bytes, expected_version) -> Ok | Conflict`. A version is

    opaque and compared for equality only. Writing identical bytes may

    keep the version, as S3's content-based ETag does; compare-and-swap

    guards against losing a *different* record.

  - Keys are plain relative paths (`[A-Za-z0-9._-]` segments, no `.` or

    `..`), the same rule on every store, so a key can never escape a

    directory store.

  - Implementations: `LocalStore` (a directory: create-only through a

    hard link, references under a lock file beside them, so processes on

    one machine can share it), `MemoryStore` (tests), then S3 and GCS in

    a crate of their own.

  - `strata::conformance(store, prefix)` exercises any store under a

    probe prefix and leaves it empty: every implementation runs it in its

    tests, and `oxy remote check` runs it against a remote.

  - The working repository's own `.oxy/` store is **not** behind the

    trait: it is read and written directly, as before, so nothing on the

    local path changes or slows. A store is what a remote is kept on.