The operation head on a store moves by compare-and-swap dd21fd82687d

        let head = match self.read_head() {

            Ok((h, _)) => h,

            Err(RepoError::Msg(m)) if m.starts_with("corrupt") => {

                rep.error(None, ".oxy/ops/head does not hold an object reference");

                return;

            }

/// The operation head's reference record on a store that is not this

/// disk's directory (oxygen#68).

pub const HEAD_RECORD: &str = "ops/head";

/// How often recording an operation reads the head again when other

/// processes keep moving it.

const HEAD_TRIES: usize = 8;

        let (head, version) = self.read_head()?;

        if !self.swap_head(&new_parent.unwrap(), version.as_ref())? {

            return fail("the operation log changed while it was rewritten; run the command again");

        }

        let r = self.head_ref()?;

        // Compare and swap (spec §12.1): when another process recorded an

        // operation since this one read the head, the two are merged

        // instead of one overwriting the other. On this disk the head moves

        // under a short lock; on any other store by its record's version,

        // and a record that moved meanwhile is read again and merged again.

        let _lock = if self.store.is_local() {

            Some(self.lock_ops()?)

        } else {

            None

        };

        let Some(parent) = parents.first() else {

            if !self.swap_head(&r, None)? {

                return fail(format!(

                    "{} already holds a repository's operation log",

                    self.store.backend().describe()

                ));

            }

        let base = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?.view;

        for _ in 0..HEAD_TRIES {

            let (current, version) = self.read_head()?;

            let to = if current == *parent {

                r

            } else {

                let theirs = Op::decode(&self.store.get_typed(&current, ObjectType::Op)?)?.view;

                let merged = Op {

                    parents: vec![current, r],

                    hlc: Hlc::now(),

                    command: format!("{command} (merged with a concurrent operation)"),

                    view: self.merge_views(&base, &op.view, &theirs)?,

                };

                let m = self.store.put(ObjectType::Op, &merged.encode())?;

                self.store.flush()?;

                m

            };

            if self.swap_head(&to, version.as_ref())? {

                return Ok(to);

            }

        }

        fail("other oxy processes kept changing the operation log while this one recorded; run the command again")

        Ok(self.read_head()?.0)

    }

    /// The operation head, and on a store its record's version for

    /// [`Repo::swap_head`]. On this disk the head is `.oxy/ops/head`; on

    /// any other store (a bucket, or memory in tests) it is the store's

    /// [`HEAD_RECORD`] reference record (oxygen#68).

    fn read_head(&self) -> Result<(ObjectRef, Option<crate::strata::Version>)> {

        if self.store.is_local() {

            let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;

            let r = ObjectRef::from_hex(hex.trim())

                .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;

            return Ok((r, None));

        }

        let backend = self.store.backend();

        match backend.read_ref(HEAD_RECORD) {

            Ok(Some((bytes, version))) => {

                let r = ObjectRef::from_hex(String::from_utf8_lossy(&bytes).trim()).ok_or_else(

                    || {

                        RepoError::Msg(format!(

                            "the operation head in {} is corrupt",

                            backend.describe()

                        ))

                    },

                )?;

                Ok((r, Some(version)))

            }

            Ok(None) => fail(format!(

                "{} holds no operation log (no {HEAD_RECORD} record)",

                backend.describe()

            )),

            Err(e) => fail(format!("read the operation head: {e}")),

        }

    /// Move the operation head to `to` from the `expected` version

    /// [`Repo::read_head`] gave (`None`: there is no head yet). On this

    /// disk the caller holds the operations lock and the head always

    /// moves; on a store it moves only if the record is unchanged, and

    /// `false` says another process moved it first.

    fn swap_head(&self, to: &ObjectRef, expected: Option<&crate::strata::Version>) -> Result<bool> {

        if self.store.is_local() {

            write_atomic(&self.oxy.join("ops").join("head"), to.to_hex().as_bytes())?;

            return Ok(true);

        }

        match self

            .store

            .backend()

            .write_ref(HEAD_RECORD, to.to_hex().as_bytes(), expected)

        {

            Ok(_) => Ok(true),

            Err(crate::strata::Error::Conflict(_)) => Ok(false),

            Err(e) => fail(format!("move the operation head: {e}")),

        }

    }

            let _lock = if self.store.is_local() {

                Some(self.lock_ops()?)

            } else {

                None

            };

            let (now, version) = self.read_head()?;

            if now != read_head || !self.swap_head(parent, version.as_ref())? {

/// oxygen#68: on a store that is not this disk's directory, the operation

/// head is the store's reference record, moved by compare-and-swap. Two

/// handles recording from the same head both get through, merged; a swap

/// from a stale version is refused; a second repository cannot start on a

/// store that already holds one.

#[test]

fn over_a_store_the_operation_head_moves_by_compare_and_swap() {

    use crate::strata::{MemoryStore, Store as _};

    let backend = std::sync::Arc::new(MemoryStore::new());

    crate::store::TEST_BACKEND.with(|t| *t.borrow_mut() = Some(backend.clone()));

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

    let a = Repo::init(dir.path()).unwrap();

    assert!(

        !dir.path().join(".oxy/ops/head").exists(),

        "the head is not a file"

    );

    let (h0, op0) = a.head_op().unwrap();

    let record = backend

        .read_ref(HEAD_RECORD)

        .unwrap()

        .expect("the head is the store's record");

    assert_eq!(String::from_utf8(record.0).unwrap(), h0.to_hex());

    let b = Repo::open(dir.path()).unwrap();

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

    let mut va = op0.view.clone();

    va.bookmarks.insert("from-a".into(), state);

    let mut vb = op0.view.clone();

    vb.bookmarks.insert("from-b".into(), state);

    let rb = b.record_op(&[h0], "bookmark from-b", vb).unwrap();

    assert_eq!(b.head_ref().unwrap(), rb);

    // a recorded from h0 too: the head moved, so the two are merged.

    let ra = a.record_op(&[h0], "bookmark from-a", va).unwrap();

    let (head, op) = Repo::open(dir.path()).unwrap().head_op().unwrap();

    assert_eq!(head, ra);

    assert!(

        op.command.ends_with("(merged with a concurrent operation)"),

        "{}",

        op.command

    );

    assert_eq!(op.parents, vec![rb, op.parents[1]]);

    assert!(op.view.bookmarks.contains_key("from-a") && op.view.bookmarks.contains_key("from-b"));

    // A swap from a version that is no longer current moves nothing.

    let (h, v) = a.read_head().unwrap();

    assert!(b.swap_head(&rb, v.as_ref()).unwrap());

    assert!(!a.swap_head(&h, v.as_ref()).unwrap(), "stale");

    assert_eq!(a.head_ref().unwrap(), rb);

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

    let err = Repo::init(other.path())

        .err()

        .expect("the store already holds a log")

        .to_string();

    assert!(

        err.contains("already holds a repository's operation log"),

        "{err}"

    );

    crate::store::TEST_BACKEND.with(|t| *t.borrow_mut() = None);

}

#[cfg(test)]

thread_local! {

    /// A backend every store opened on this thread uses instead of its

    /// directory: tests of a repository kept on a store that is not this

    /// disk's directory, with several handles on one store.

    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };

}

        #[cfg(test)]

        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {

            return Store::open_on(backend, dir);

        }