Strata: a repository kept on a store, as a remote 9f9e72b93e4e

//! A repository kept on a [`Store`] (Oxygen Strata, docs/stores.md): what

//! `oxy push dir://…` or `oxy clone s3://…` talk to.

//!

//! The store holds three kinds of key:

//!

//! - `oxygen.toml`: the layout's format and the repository's id, written

//!   once by the first push;

//! - `bundles/<seq>-<hash>.bundle`: each push's changes, signatures,

//!   seals, objects and policies, as the wire protocol encodes a bundle,

//!   written once and named by content;

//! - `refs`: every bookmark with its heads and attestation, every tag, the

//!   policy and its history, encoded as the protocol's `refs` response,

//!   moved by compare-and-swap.

//!

//! A store cannot run code, so the client does what a hub does on a push,

//! against a **mirror**: a bare repository of its own, kept beside the

//! remote's other state under `.oxy/remotes/<name>/strata/`, brought level

//! with the store before every use (the record read, the bundles it has

//! not seen imported, the view set from the record), and then used as a

//! [`LocalRemote`] would be. A push runs the hub's own checks

//! (`receive_push`) on the mirror, uploads the bundle, and moves `refs`

//! from the version it read; if the record moved meanwhile, the mirror is

//! put back and the push reports it, exactly as a hub reports a stale

//! push. Nothing in the store is trusted: every object is re-hashed as it

//! is imported, and the fetch that follows verifies each bookmark's

//! attestation under the policy, as it does for any remote.

use super::{fail, Repo, RepoError, Result};

use crate::hash::{ChangeHash, ObjectRef, ObjectType};

use crate::objects::StateObject;

use crate::protocol::{

    wire, Bundle, RefResult, RefUpdate, RefsResponse, Remote, RemoteOptions, SignedPolicy, TagRef,

    UNBOUND_ID,

};

use crate::strata::{Error as StoreError, Store, Version};

use std::cell::RefCell;

use std::collections::BTreeSet;

use std::fs;

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

const MARKER_KEY: &str = "oxygen.toml";

const REFS_KEY: &str = "refs";

const BUNDLES_PREFIX: &str = "bundles/";

const FORMAT: u32 = 1;

impl From<StoreError> for RepoError {

    fn from(e: StoreError) -> RepoError {

        RepoError::Msg(e.to_string())

    }

}

/// The `oxygen.toml` marker: two lines, written once.

fn marker(id: [u8; 16]) -> Vec<u8> {

    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()

}

fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {

    let text = String::from_utf8_lossy(bytes);

    let mut format = None;

    let mut repo = None;

    for line in text.lines() {

        let Some((k, v)) = line.split_once('=') else {

            continue;

        };

        let v = v.trim().trim_matches('"');

        match k.trim() {

            "format" => format = v.parse::<u32>().ok(),

            "repo" => {

                repo = hex::decode(v)

                    .ok()

                    .and_then(|b| <[u8; 16]>::try_from(b).ok())

            }

            _ => {}

        }

    }

    match (format, repo) {

        (Some(f), _) if f > FORMAT => fail(format!(

            "{what} holds a repository in Strata format {f}, newer than this oxy understands (format {FORMAT}): upgrade oxy"

        )),

        (Some(_), Some(id)) => Ok(id),

        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),

    }

}

/// A remote kept on a store.

pub struct BucketRemote {

    store: Box<dyn Store>,

    /// Where the mirror lives: the remote's directory under `.oxy/remotes`,

    /// or a temporary directory when there is no repository yet (a clone),

    /// removed on drop.

    dir: PathBuf,

    temporary: bool,

    claimed_id: Option<[u8; 16]>,

    mirror: RefCell<Option<Repo>>,

}

impl Drop for BucketRemote {

    fn drop(&mut self) {

        if self.temporary {

            let _ = fs::remove_dir_all(&self.dir);

        }

    }

}

/// What a sync found: the record's version (to move it from) and the

/// repository id the store holds, if any.

struct Synced {

    version: Option<Version>,

    id: Option<[u8; 16]>,

}

impl BucketRemote {

    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {

        let (dir, temporary) = match &options.cache_dir {

            Some(d) => (d.clone(), false),

            None => {

                let mut nonce = [0u8; 6];

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

                (

                    std::env::temp_dir().join(format!(

                        "oxy-strata-{}-{}",

                        std::process::id(),

                        hex::encode(nonce)

                    )),

                    true,

                )

            }

        };

        Ok(BucketRemote {

            store,

            dir,

            temporary,

            claimed_id: options.repo_id,

            mirror: RefCell::new(None),

        })

    }

    fn what(&self) -> String {

        self.store.describe()

    }

    /// The repository id the store holds: `None` until a first push.

    fn stored_id(&self) -> Result<Option<[u8; 16]>> {

        match self.store.get(MARKER_KEY) {

            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),

            Err(StoreError::NotFound(_)) => Ok(None),

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

        }

    }

    fn mirror_dir(&self) -> PathBuf {

        self.dir.join("mirror")

    }

    fn seen_path(&self) -> PathBuf {

        self.dir.join("seen")

    }

    /// The bundles already imported into the mirror.

    fn seen(&self) -> BTreeSet<String> {

        fs::read_to_string(self.seen_path())

            .map(|s| s.lines().map(str::to_string).collect())

            .unwrap_or_default()

    }

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

        let mut seen = self.seen();

        if seen.insert(key.to_string()) {

            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();

            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;

        }

        Ok(())

    }

    /// Open the mirror for repository `id`, creating it on first use.

    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {

        if self.mirror.borrow().is_some() {

            return Ok(());

        }

        let dir = self.mirror_dir();

        let repo = if dir.join(super::OXY_DIR).is_dir() {

            let repo = Repo::open(&dir)?;

            if repo.id()? != id {

                // The store was emptied and reused for another repository:

                // start the mirror over.

                drop(repo);

                fs::remove_dir_all(&self.dir)?;

                fs::create_dir_all(&self.dir)?;

                Repo::init_bare_with_id(&dir, id)?

            } else {

                repo

            }

        } else {

            fs::create_dir_all(&self.dir)?;

            Repo::init_bare_with_id(&dir, id)?

        };

        *self.mirror.borrow_mut() = Some(repo);

        Ok(())

    }

    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {

        let m = self.mirror.borrow();

        match m.as_ref() {

            Some(repo) => f(repo),

            None => fail(format!(

                "{} holds no repository yet: push one to it first",

                self.what()

            )),

        }

    }

    /// Bring the mirror level with the store.

    fn sync(&self) -> Result<Synced> {

        let Some(id) = self.stored_id()? else {

            return Ok(Synced {

                version: None,

                id: None,

            });

        };

        self.open_mirror(id)?;

        let record = self.store.read_ref(REFS_KEY)?;

        let (resp, version) = match record {

            Some((bytes, version)) => {

                let (rid, resp) = wire::decode_refs(&bytes)?;

                if rid != id {

                    return fail(format!(

                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",

                        self.what()

                    ));

                }

                (resp, Some(version))

            }

            None => (RefsResponse::default(), None),

        };

        self.with_mirror(|mirror| {

            // The policy first: bundles are then imported under it.

            if let Some(sp) = &resp.policy {

                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);

                if have.is_none_or(|s| s < sp.policy.serial) {

                    mirror

                        .adopt_policy_chain(&resp.history, sp, "strata sync")

                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;

                }

            }

            self.import_new_bundles(mirror)?;

            self.set_view(mirror, &resp)?;

            Ok(())

        })?;

        Ok(Synced {

            version,

            id: Some(id),

        })

    }

    /// Import every bundle in the store the mirror has not seen. Bundles

    /// are named in the order they were pushed, but two pushes from the

    /// same base may interleave, so a bundle whose dependencies are not

    /// there yet is tried again after the others.

    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {

        let seen = self.seen();

        let mut pending: Vec<String> = self

            .store

            .list(BUNDLES_PREFIX)?

            .into_iter()

            .map(|e| e.key)

            .filter(|k| !seen.contains(k))

            .collect();

        let strict = mirror.current_policy()?.is_some();

        while !pending.is_empty() {

            let mut deferred = Vec::new();

            let mut last_error = None;

            for key in pending.drain(..) {

                let bytes = self.store.get(&key)?;

                let bundle = wire::decode_bundle(&bytes)

                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;

                match mirror.import_bundle(&bundle, strict) {

                    Ok(_) => self.mark_seen(&key)?,

                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {

                        last_error = Some(m);

                        deferred.push(key);

                    }

                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),

                }

            }

            if deferred.len() == pending.len() + deferred.len() && !deferred.is_empty() {

                // No progress: something those bundles need was never

                // pushed, or a push is still in progress.

                return fail(format!(

                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",

                    self.what(),

                    deferred.join(", "),

                    last_error.unwrap_or_default()

                ));

            }

            pending = deferred;

        }

        Ok(())

    }

    /// Make the mirror's view say what the record says.

    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {

        let (op_ref, op) = mirror.head_op()?;

        let mut view = op.view.clone();

        view.bookmarks.clear();

        for r in &resp.refs {

            for h in &r.heads {

                if !mirror.store.has(h) {

                    return fail(format!(

                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",

                        self.what(),

                        r.name,

                        h.short(12)

                    ));

                }

            }

            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;

            if state != r.state {

                return fail(format!(

                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",

                    self.what(),

                    r.name

                ));

            }

            if let Some(sa) = &r.attestation {

                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);

                if current.as_ref() != Some(sa) {

                    let a = mirror.store.put_attest(&sa.attest)?;

                    for sig in &sa.sigs {

                        mirror.store.put_signature(&a, sig)?;

                    }

                    mirror.set_ref_attest(&r.name, &a)?;

                }

            }

            view.bookmarks.insert(r.name.clone(), state);

        }

        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);

        for t in &resp.tags {

            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;

            mirror

                .adopt_tag(&mut view, t, policy.as_ref())

                .map_err(|e| RepoError::Msg(format!("{}: {e}", self.what())))?;

        }

        if view != op.view {

            mirror.record_op(&[op_ref], "strata sync", view)?;

        }

        Ok(())

    }

    /// The record for the mirror's current references.

    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {

        self.with_mirror(|mirror| {

            let mut resp = mirror.remote_refs()?;

            // A mirror has no checkout; the store's clone picks a bookmark

            // as any other remote's does.

            resp.default_bookmark = None;

            Ok(wire::encode_refs(id, &resp))

        })

    }

    /// Store a bundle under the next sequence number; returns its key.

    fn upload(&self, bundle: &Bundle) -> Result<String> {

        let bytes = wire::encode_bundle(bundle);

        let hash = blake3::hash(&bytes).to_hex();

        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;

        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);

        match self.store.put_new(&key, &bytes) {

            Ok(()) | Err(StoreError::Exists(_)) => {}

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

        }

        self.mark_seen(&key)?;

        Ok(key)

    }

}

impl Remote for BucketRemote {

    fn repo_id(&self) -> Result<[u8; 16]> {

        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))

    }

    fn refs(&self) -> Result<RefsResponse> {

        let synced = self.sync()?;

        if synced.id.is_none() {

            return Ok(RefsResponse::default());

        }

        self.with_mirror(|mirror| {

            let mut resp = mirror.remote_refs()?;

            resp.default_bookmark = None;

            Ok(resp)

        })

    }

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {

        self.with_mirror(|m| m.negotiate(have, want))

    }

    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {

        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))

    }

    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {

        self.with_mirror(|m| m.objects_for(refs))

    }

    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        self.with_mirror(|m| m.checkpoint_bundle_for(heads))

    }

    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {

        self.with_mirror(|m| m.shallow_changes(heads, depth))

    }

    fn push(

        &self,

        bundle: Bundle,

        updates: Vec<RefUpdate>,

        policy: Option<SignedPolicy>,

        tags: Vec<TagRef>,

    ) -> Result<Vec<RefResult>> {

        let mut synced = self.sync()?;

        let id = match synced.id {

            Some(id) => id,

            None => {

                // The first push binds the store to this repository. Two

                // first pushes race on the marker; the loser reads the

                // winner's and goes on only if it is the same repository.

                let Some(id) = self.claimed_id else {

                    return fail(format!(

                        "{} is empty, and this client did not say which repository it is",

                        self.what()

                    ));

                };

                match self.store.put_new(MARKER_KEY, &marker(id)) {

                    Ok(()) => {}

                    Err(StoreError::Exists(_)) => {}

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

                }

                synced = self.sync()?;

                match synced.id {

                    Some(stored) if stored == id => id,

                    _ => {

                        return fail(format!(

                            "{} was just bound to another repository by a concurrent push",

                            self.what()

                        ))

                    }

                }

            }

        };

        if let Some(claimed) = self.claimed_id {

            if claimed != id {

                return fail(format!(

                    "{} holds a different repository (id {}) than this one ({}): a store holds one repository; use another prefix or bucket",

                    self.what(),

                    hex::encode(id),

                    hex::encode(claimed)

                ));

            }

        }

        let (before_ref, before) = self.with_mirror(|m| m.head_op())?;

        let results =

            self.with_mirror(|m| m.receive_push(bundle.clone(), updates, policy, tags))?;

        if !results.iter().any(|r| r.ok) {

            return Ok(results);

        }

        if !bundle.is_empty() {

            self.upload(&bundle)?;

        }

        let record = self.record(id)?;

        match self

            .store

            .write_ref(REFS_KEY, &record, synced.version.as_ref())

        {

            Ok(_) => Ok(results),

            Err(StoreError::Conflict(_)) => {

                // Someone else moved the record meanwhile. The bundle is in

                // the store and harmless; the mirror goes back to what the

                // store says, and the push is reported as a hub would.

                self.with_mirror(|m| {

                    let (now, _) = m.head_op()?;

                    if now != before_ref {

                        m.record_op(

                            &[now],

                            "strata push undone: the remote moved",

                            before.view.clone(),

                        )?;

                    }

                    Ok(())

                })?;

                let _ = self.sync();

                Ok(results

                    .into_iter()

                    .map(|r| RefResult {

                        name: r.name,

                        ok: false,

                        reason: "the remote moved while pushing (another push landed first); pull, then push again".into(),

                    })

                    .collect())

            }

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

        }

    }

}

/// Open a remote on `store` for the local repository described by `options`.

pub fn open_store_remote(

    store: Box<dyn Store>,

    options: &RemoteOptions,

) -> Result<Box<dyn Remote>> {

    Ok(Box::new(BucketRemote::open(store, options)?))

}

/// Whether `path` holds a Strata store rather than a repository.

pub fn is_store_dir(path: &Path) -> bool {

    path.join(MARKER_KEY).is_file() || path.join(REFS_KEY).is_file()

}

#[cfg(test)]

mod tests {

    use super::*;

    use crate::repo::IdentityScope;

    use crate::strata::{Entry, LocalStore, MemoryStore};

    use std::ops::Range;

    use std::sync::Mutex;

    fn hermetic() {

        static ONCE: std::sync::Once = std::sync::Once::new();

        static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

        ONCE.call_once(|| {

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

            std::env::set_var("OXY_HOME", d.path());

            DIR.set(d).ok();

        });

    }

    fn write(root: &Path, rel: &str, content: &[u8]) {

        let p = root.join(rel);

        fs::create_dir_all(p.parent().unwrap()).unwrap();

        fs::write(p, content).unwrap();

    }

    fn add_remote(repo: &mut Repo, name: &str, url: &str) {

        repo.config.set(&format!("remote.{name}.url"), url).unwrap();

        repo.save_config().unwrap();

    }

    fn keys(store: &dyn Store, prefix: &str) -> Vec<String> {

        store

            .list(prefix)

            .unwrap()

            .into_iter()

            .map(|e| e.key)

            .collect()

    }

    #[test]

    fn push_clone_and_pull_through_a_directory_store() {

        hermetic();

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

        let root = dir.path();

        let url = format!("dir://{}", root.join("bucket").display());

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

        // A signs, commits, tags and pushes: the store gets the marker, one

        // bundle, and a record.

        let mut a = Repo::init(&root.join("a"))

            .unwrap()

            .with_home(root.join("home-a"));

        a.create_identity("Ada", "[email protected]", IdentityScope::Repo)

            .unwrap();

        add_remote(&mut a, "bucket", &url);

        write(&root.join("a"), "f.txt", b"a\n");

        let h1 = a.commit("first", &[]).unwrap();

        a.tag_create("v1", &[h1], "one").unwrap();

        let results = a.push("bucket", false).unwrap();

        assert!(results.iter().all(|r| r.ok), "{results:?}");

        assert_eq!(

            keys(&store, ""),

            [

                "bundles/00000001-".to_string() + &keys(&store, "bundles/")[0][17..],

                "oxygen.toml".into(),

                "refs".into()

            ]

        );

        assert_eq!(

            parse_marker(&store.get("oxygen.toml").unwrap(), "t").unwrap(),

            a.id().unwrap()

        );

        let again = a.push("bucket", false).unwrap();

        assert!(

            again[0].ok && again[0].reason.contains("up to date"),

            "{again:?}"

        );

        assert_eq!(

            keys(&store, "bundles/").len(),

            1,

            "a no-op push uploads nothing"

        );

        // B clones from the store: same id, same file, the tag, and the

        // seal on the change.

        let b = Repo::clone(&url, &root.join("b")).unwrap();

        assert_eq!(b.id().unwrap(), a.id().unwrap());

        assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"a\n");

        assert_eq!(b.bookmark_heads().unwrap(), vec![h1]);

        assert_eq!(

            b.tags()

                .unwrap()

                .iter()

                .map(|t| t.name.clone())

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

            ["v1"]

        );

        assert!(matches!(

            b.authenticate_change(&h1).unwrap(),

            crate::repo::SignatureStatus::Valid { sealed: true, .. }

        ));

        assert!(

            root.join("b/.oxy/remotes/origin/strata/mirror/.oxy")

                .is_dir(),

            "the mirror lives with the remote's state"

        );

        assert!(

            !std::env::temp_dir()

                .read_dir()

                .unwrap()

                .flatten()

                .any(|e| e.file_name().to_string_lossy().starts_with("oxy-strata-")),

            "a clone's temporary mirror is removed"

        );

        // B commits and pushes; A pulls it. A then commits and pushes.

        write(&root.join("b"), "f.txt", b"a\nb\n");

        let h2 = b.commit("second", &[]).unwrap();

        assert!(b.push("origin", false).unwrap()[0].ok);

        assert_eq!(keys(&store, "bundles/").len(), 2);

        let pr = a.pull("bucket").unwrap();

        assert!(pr.changed && pr.checked_out);

        assert_eq!(fs::read(root.join("a/f.txt")).unwrap(), b"a\nb\n");

        write(&root.join("a"), "g.txt", b"g\n");

        let h3 = a.commit("third", &[]).unwrap();

        assert!(a.push("bucket", false).unwrap()[0].ok);

        let pr = b.pull("origin").unwrap();

        assert!(pr.changed);

        // g.txt is new, so its change commutes with B's: two heads.

        let mut want = vec![h2, h3];

        want.sort();

        assert_eq!(b.bookmark_heads().unwrap(), want);

        assert_eq!(fs::read(root.join("b/g.txt")).unwrap(), b"g\n");

        // A third clone gets everything from the three bundles.

        let c = Repo::clone(&url, &root.join("c")).unwrap();

        assert_eq!(c.bookmark_heads().unwrap(), want);

        assert_eq!(fs::read(root.join("c/f.txt")).unwrap(), b"a\nb\n");

    }

    #[test]

    fn a_push_behind_the_remote_is_refused_before_anything_is_uploaded() {

        hermetic();

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

        let root = dir.path();

        let url = format!("dir://{}", root.join("bucket").display());

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

        let mut a = Repo::init(&root.join("a")).unwrap();

        add_remote(&mut a, "bucket", &url);

        write(&root.join("a"), "f.txt", b"a\n");

        a.commit("first", &[]).unwrap();

        assert!(a.push("bucket", false).unwrap()[0].ok);

        let b = Repo::clone(&url, &root.join("b")).unwrap();

        write(&root.join("b"), "f.txt", b"b\n");

        b.commit("from b", &[]).unwrap();

        assert!(b.push("origin", false).unwrap()[0].ok);

        write(&root.join("a"), "g.txt", b"g\n");

        a.commit("from a", &[]).unwrap();

        let e = a.push("bucket", false).unwrap_err().to_string();

        assert!(e.contains("pull first"), "{e}");

        assert_eq!(keys(&store, "bundles/").len(), 2);

        assert!(a.pull("bucket").unwrap().changed);

        assert!(a.push("bucket", false).unwrap()[0].ok);

        assert_eq!(keys(&store, "bundles/").len(), 3);

    }

    /// A store whose record moves under a writer the first time it is

    /// written to: what a concurrent push looks like between the read and

    /// the compare-and-swap.

    struct Racing {

        inner: MemoryStore,

        raced: Mutex<bool>,

    }

    impl Store for Racing {

        fn describe(&self) -> String {

            "racing".into()

        }

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

            self.inner.has(key)

        }

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

            self.inner.get(key)

        }

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

            self.inner.get_range(key, range)

        }

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

            self.inner.put_new(key, bytes)

        }

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

            self.inner.list(prefix)

        }

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

            self.inner.delete(key)

        }

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

            self.inner.read_ref(key)

        }

        fn write_ref(

            &self,

            key: &str,

            bytes: &[u8],

            expected: Option<&Version>,

        ) -> crate::strata::Result<Version> {

            let mut raced = self.raced.lock().unwrap();

            if !*raced && expected.is_some() {

                *raced = true;

                // Someone else lands a record: the same references with a

                // different detail, so the version moves.

                let (current, v) = self.inner.read_ref(key)?.unwrap();

                let (id, mut resp) =

                    wire::decode_refs(&current).map_err(|e| StoreError::Backend(e.to_string()))?;

                resp.default_bookmark = Some("raced".into());

                self.inner

                    .write_ref(key, &wire::encode_refs(id, &resp), Some(&v))?;

            }

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

        }

    }

    #[test]

    fn a_record_moved_during_a_push_fails_the_push_and_keeps_the_mirror_honest() {

        hermetic();

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

        let root = dir.path();

        let a = Repo::init(&root.join("a")).unwrap();

        write(&root.join("a"), "f.txt", b"a\n");

        let h1 = a.commit("first", &[]).unwrap();

        let store = Box::new(Racing {

            inner: MemoryStore::new(),

            raced: Mutex::new(false),

        });

        let options = RemoteOptions {

            repo_id: Some(a.id().unwrap()),

            cache_dir: Some(root.join("cache")),

            ..Default::default()

        };

        let remote = BucketRemote::open(store, &options).unwrap();

        // The first push creates the record (expected None: no race yet).

        let state = a.store.put_state(&StateObject::new(vec![h1])).unwrap();

        let empty = a.store.put_state(&StateObject::new(vec![])).unwrap();

        let mut bundle = a.bundle_for(&[h1]).unwrap();

        bundle

            .objects

            .push((ObjectType::State, StateObject::new(vec![h1]).encode()));

        let update = |old, new| RefUpdate {

            name: "main".into(),

            old_state: old,

            new_state: new,

            force: false,

            attestation: None,

        };

        let r = remote

            .push(bundle, vec![update(empty, state)], None, vec![])

            .unwrap();

        assert!(r[0].ok, "{r:?}");

        // The second push meets the race: refused, nothing moved.

        write(&root.join("a"), "f.txt", b"ab\n");

        let h2 = a.commit("second", &[]).unwrap();

        let state2 = a.store.put_state(&StateObject::new(vec![h2])).unwrap();

        let mut bundle = a.bundle_for(&[h2]).unwrap();

        bundle

            .objects

            .push((ObjectType::State, StateObject::new(vec![h2]).encode()));

        let r = remote

            .push(bundle, vec![update(state, state2)], None, vec![])

            .unwrap();

        assert!(

            !r[0].ok && r[0].reason.contains("moved while pushing"),

            "{r:?}"

        );

        let refs = remote.refs().unwrap();

        assert_eq!(

            refs.refs[0].heads,

            vec![h1],

            "the mirror says what the store says"

        );

    }

    #[test]

    fn a_store_holds_one_repository() {

        hermetic();

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

        let root = dir.path();

        let url = format!("dir://{}", root.join("bucket").display());

        let mut a = Repo::init(&root.join("a")).unwrap();

        add_remote(&mut a, "bucket", &url);

        write(&root.join("a"), "f.txt", b"a\n");

        a.commit("first", &[]).unwrap();

        assert!(a.push("bucket", false).unwrap()[0].ok);

        let mut other = Repo::init(&root.join("other")).unwrap();

        add_remote(&mut other, "bucket", &url);

        write(&root.join("other"), "f.txt", b"o\n");

        other.commit("mine", &[]).unwrap();

        let e = other.push("bucket", false).unwrap_err().to_string();

        assert!(

            e.contains("different repository") || e.contains("another repository"),

            "{e}"

        );

        assert_eq!(

            keys(&LocalStore::open(root.join("bucket")).unwrap(), "bundles/").len(),

            1

        );

    }

    #[test]

    fn a_damaged_bundle_is_caught_and_an_empty_store_cannot_be_cloned() {

        hermetic();

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

        let root = dir.path();

        let url = format!("dir://{}", root.join("bucket").display());

        let e = Repo::clone(&url, &root.join("nothing"))

            .err()

            .expect("refused")

            .to_string();

        assert!(e.contains("empty and unbound"), "{e}");

        let mut a = Repo::init(&root.join("a")).unwrap();

        add_remote(&mut a, "bucket", &url);

        write(&root.join("a"), "f.txt", b"a\n");

        a.commit("first", &[]).unwrap();

        assert!(a.push("bucket", false).unwrap()[0].ok);

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

        let key = keys(&store, "bundles/").remove(0);

        let mut bytes = store.get(&key).unwrap();

        let mid = bytes.len() / 2;

        bytes[mid] ^= 0xff;

        store.delete(&key).unwrap();

        store.put_new(&key, &bytes).unwrap();

        let e = Repo::clone(&url, &root.join("b"))

            .err()

            .expect("refused")

            .to_string();

        assert!(

            e.contains(&key) || e.contains("bundle") || e.contains("hash"),

            "{e}"

        );

        assert!(!root.join("b/f.txt").exists());

    }

    #[test]

    fn markers_parse_and_refuse_newer_formats() {

        let id = [7u8; 16];

        assert_eq!(parse_marker(&marker(id), "s").unwrap(), id);

        let e = parse_marker(

            b"format = 9\nrepo = \"07070707070707070707070707070707\"\n",

            "s",

        )

        .unwrap_err()

        .to_string();

        assert!(e.contains("format 9"), "{e}");

        assert!(parse_marker(b"hello", "s").is_err());

    }

}

            cache_dir: None,

use std::sync::{Arc, Mutex};

    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its

    /// mirror and what it has seen: the remote's directory under

    /// `.oxy/remotes`. None (a clone, before there is a repository) means

    /// a temporary directory.

    pub cache_dir: Option<PathBuf>,

/// Opens a store for a URL it understands, or passes (`None`). Stores that

/// need more than the standard library (the cloud ones) are added by the

/// program that links them, with [`register_store_opener`].

pub type StoreOpener =

    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;

static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());

pub fn register_store_opener(opener: StoreOpener) {

    STORE_OPENERS

        .lock()

        .unwrap_or_else(|p| p.into_inner())

        .push(opener);

}

/// A store for `url`, from `dir://` or any registered opener.

pub fn open_store(

    url: &str,

    options: &RemoteOptions,

) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {

    if let Some(path) = url.strip_prefix("dir://") {

        return Some(

            crate::strata::LocalStore::open(path)

                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),

        );

    }

    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());

    openers.iter().find_map(|open| open(url, options))

}

    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {

        (Some(p), _) => ("file://", p),

        (_, Some(p)) => ("dir://", p),

        _ => ("", url),

/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`

/// (a hub), or a store (`dir://`, and whatever openers are registered).

    }

    if let Some(store) = open_store(url, options) {

        return crate::repo::strata::open_store_remote(store?, options);

pub mod strata;

    pub(crate) fn set_ref_attest(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {

            cache_dir: Some(self.oxy.join("remotes").join(name).join("strata")),

            if !matches!(scheme, "http" | "https") {

                return None;

            }

                cache_dir: None,

        // Asked of the remote as now configured, so a store-backed one

        // answers from the mirror the fetch just filled.

        let advertised = repo

            .open_named_remote("origin")

            .ok()

            .and_then(|r| r.refs().ok())

            .and_then(|r| r.default_bookmark);

## 4. Layout in the bucket (as built, 2026-10-03)

  oxygen.toml                      format version and repository id, written once by the first push

  bundles/<seq>-<hash>.bundle      one per push: changes, signatures, seals, objects, policies

                                   (the wire protocol's bundle encoding), written once, named by content

  refs                             every bookmark with its heads and attestation, every tag, the policy

                                   and its history (the protocol's refs response), moved by compare-and-swap

- **Pushing** uploads one bundle, then updates `refs` from the version

  it read. A bundle named in `refs` is complete, because the bundle is

  written before `refs` moves.

- **One `refs` record** so that a push moving several bookmarks is

  atomic. Per-reference records would scale to more writers but lose

- **Bundles are whole-file transfers.** Each push is one bundle, so a

  pull downloads only the bundles it has not seen. Packs with indexes

  for range reads, as first planned, are an optimisation for later

  (`oxy gc` against a remote merges bundles, and could write a pack).

- **The client does what a hub does.** A store cannot run code, so each

  client keeps a **mirror**, a bare repository of its own under

  `.oxy/remotes/<name>/strata/`, brought level with the store before

  every use: the record read, the bundles it has not seen imported (each

  object re-hashed), the view set from the record. A push runs the hub's

  own checks (`receive_push`: policy, attestations, fast-forward) on the

  mirror, uploads the bundle, then moves `refs`. A fetch verifies each

  bookmark's attestation under the policy, as for any remote. The

  mirror costs a second copy of the objects on disk; sharing them with

  the repository's own store is a later optimisation.

- **The record moved meanwhile:** the mirror is put back to what the

  store says and the push is reported as a hub reports a stale push:

  pull, then push again. Merging two pushes to different bookmarks

  without a round trip is ST5's work.

- **A directory is a store too:** `oxy remote add usb dir:///media/usb/repo`

  (`dir://`, since a plain path or `file://` is another repository).

  Tests run the whole flow through it with no network.

        cache_dir: None,

            cache_dir: None,

                cache_dir: None,