Concurrent commands no longer lose operations 19555025a2ac

pub use workspace::{FileLock, SnapshotLock, WorkspaceInfo};

        // Compare and swap under a short lock (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.

        let _lock = self.lock_ops()?;

        let current = match parents.first() {

            // The first operation (init): there is no head yet.

            None => None,

            Some(_) => Some(self.head_ref()?),

        };

        let Some(current) = current.filter(|c| Some(c) != parents.first()) else {

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

            return Ok(r);

        };

        let base = match parents.first() {

            Some(p) => Op::decode(&self.store.get_typed(p, ObjectType::Op)?)?.view,

            None => View::default(),

        };

        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()?;

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

        Ok(m)

    }

    /// The operation head's lock, held only while the head pointer moves.

    fn lock_ops(&self) -> Result<FileLock> {

        fs::create_dir_all(self.oxy.join("ops"))?;

        FileLock::acquire(

            self.oxy.join("ops").join("lock"),

            "another oxy process holds the operation log",

        )

    }

    fn head_ref(&self) -> Result<ObjectRef> {

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

        ObjectRef::from_hex(hex.trim())

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

    }

    /// Three-way merge of two views that both started from `base` (spec

    /// §12.1). A reference only one side changed takes that side's value.

    /// When both changed it differently: a bookmark keeps both sides'

    /// heads (it may then have several), a change beats a deletion, and

    /// otherwise this process's value wins for workspaces, remote

    /// references and the policy, the first value for a tag (tags never

    /// move).

    fn merge_views(&self, base: &View, ours: &View, theirs: &View) -> Result<View> {

        fn pick<K: Ord + Clone, V: Clone + PartialEq>(

            b: &BTreeMap<K, V>,

            o: &BTreeMap<K, V>,

            t: &BTreeMap<K, V>,

            mut both: impl FnMut(&V, &V) -> Result<V>,

        ) -> Result<BTreeMap<K, V>> {

            let keys: BTreeSet<&K> = b.keys().chain(o.keys()).chain(t.keys()).collect();

            let mut out = BTreeMap::new();

            for k in keys {

                let (bv, ov, tv) = (b.get(k), o.get(k), t.get(k));

                let v = if ov == tv || tv == bv {

                    ov.cloned()

                } else if ov == bv {

                    tv.cloned()

                } else {

                    match (ov, tv) {

                        (Some(x), Some(y)) => Some(both(x, y)?),

                        (Some(x), None) | (None, Some(x)) => Some(x.clone()),

                        (None, None) => None,

                    }

                };

                if let Some(v) = v {

                    out.insert(k.clone(), v);

                }

            }

            Ok(out)

        }

        let bookmarks = pick(

            &base.bookmarks,

            &ours.bookmarks,

            &theirs.bookmarks,

            |x, y| {

                let mut heads = self.store.get_state(x)?.heads;

                heads.extend(self.store.get_state(y)?.heads);

                let heads = self.load_state(&heads)?.maximal(&heads);

                Ok(self.store.put_state(&StateObject::new(heads))?)

            },

        )?;

        let workspaces = pick(

            &base.workspaces,

            &ours.workspaces,

            &theirs.workspaces,

            |x, _| Ok(*x),

        )?;

        let tags = pick(&base.tags, &ours.tags, &theirs.tags, |_, y| Ok(*y))?;

        let empty = BTreeMap::new();

        let names: BTreeSet<&String> = base

            .remotes

            .keys()

            .chain(ours.remotes.keys())

            .chain(theirs.remotes.keys())

            .collect();

        let mut remotes = BTreeMap::new();

        for n in names {

            let m = pick(

                base.remotes.get(n).unwrap_or(&empty),

                ours.remotes.get(n).unwrap_or(&empty),

                theirs.remotes.get(n).unwrap_or(&empty),

                |x, _| Ok(*x),

            )?;

            if !m.is_empty() {

                remotes.insert(n.clone(), m);

            }

        }

        let policy = if ours.policy == theirs.policy || theirs.policy == base.policy {

            ours.policy

        } else if ours.policy == base.policy {

            theirs.policy

        } else {

            ours.policy.or(theirs.policy)

        };

        self.store.flush()?;

        Ok(View {

            bookmarks,

            workspaces,

            remotes,

            policy,

            tags,

        })

        let (read_head, mut op) = self.head_op()?;

        let _lock = self.lock_ops()?;

        if self.head_ref()? != read_head {

            return fail("another oxy command ran while undoing; run undo again");

        }

/// Spec §12.1: commands in two workspaces of one repository, and pushes,

/// running at the same time lose nothing: every commit ends up on the

/// bookmark, locally and on the remote, and the repository passes fsck.

#[test]

fn concurrent_commits_in_two_workspaces_lose_nothing() {

    hermetic();

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

    let root = dir.path();

    let origin = root.join("origin");

    Repo::init_bare_unbound(&origin).unwrap();

    let main = root.join("main");

    let mut repo = Repo::init(&main).unwrap();

    repo.config

        .set("remote.origin.url", &origin.to_string_lossy())

        .unwrap();

    repo.save_config().unwrap();

    write(&main, "seed.txt", b"seed\n");

    repo.commit("seed", &[]).unwrap();

    let second = root.join("second");

    repo.workspace_add(&second, None).unwrap();

    drop(repo);

    const EACH: usize = 15;

    std::thread::scope(|s| {

        for (tag, dir, push) in [("a", main.clone(), true), ("b", second.clone(), false)] {

            s.spawn(move || {

                let r = Repo::open(&dir).unwrap();

                for i in 0..EACH {

                    write(

                        &dir,

                        &format!("{tag}-{i}.txt"),

                        format!("{tag} {i}\n").as_bytes(),

                    );

                    r.commit(&format!("{tag} {i}"), &[]).unwrap();

                    if push && i % 5 == 4 {

                        // A push may be refused while the other workspace moves the bookmark; nothing may be lost either way.

                        let _ = r.push("origin", false);

                    }

                }

            });

        }

    });

    let repo = Repo::open(&main).unwrap();

    let heads = repo.bookmark_heads().unwrap();

    let state = repo.load_state(&heads).unwrap();

    let messages: BTreeSet<String> = state.changes().map(|(_, m)| m.message.clone()).collect();

    for tag in ["a", "b"] {

        for i in 0..EACH {

            assert!(

                messages.contains(&format!("{tag} {i}")),

                "lost commit {tag} {i}; have {messages:?}"

            );

        }

    }

    let rep = repo.fsck().unwrap();

    assert_eq!(rep.errors(), 0, "{:?}", rep.problems);

    assert!(repo.push("origin", false).unwrap().iter().all(|r| r.ok));

    let hosted = Repo::open(&origin).unwrap();

    let hosted_heads = hosted.bookmark_heads().unwrap();

    assert_eq!(

        hosted.load_state(&hosted_heads).unwrap().len(),

        state.len(),

        "the remote has every commit"

    );

}

/// Two operations recorded from the same parent (what two processes do

/// when they race) are merged: both bookmarks' heads survive, and a

/// change made on one side only is kept.

#[test]

fn operations_from_the_same_parent_are_merged() {

    hermetic();

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

    let root = dir.path();

    let repo = Repo::init(root).unwrap();

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

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

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

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

    let (x, op) = repo.head_op().unwrap();

    let state = |heads: Vec<ChangeHash>| repo.store().put_state(&StateObject::new(heads)).unwrap();

    // Side one moves main to just `a` and adds a bookmark; side two moves main to just `b`.

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

    one.bookmarks.insert("main".into(), state(vec![a]));

    one.bookmarks.insert("topic".into(), state(vec![a]));

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

    two.bookmarks.insert("main".into(), state(vec![b]));

    repo.record_op(&[x], "one", one).unwrap();

    let merged = repo.record_op(&[x], "two", two).unwrap();

    let (head, op) = repo.head_op().unwrap();

    assert_eq!(head, merged);

    assert_eq!(op.parents.len(), 2, "a merge of the two operations");

    assert!(op.command.contains("merged with a concurrent operation"));

    let mut main = repo

        .store()

        .get_state(&op.view.bookmarks["main"])

        .unwrap()

        .heads;

    main.sort();

    let mut want = vec![a, b];

    want.sort();

    assert_eq!(main, want, "both sides' heads are kept");

    assert!(

        op.view.bookmarks.contains_key("topic"),

        "a one-sided change survives"

    );

    assert_eq!(repo.fsck().unwrap().errors(), 0);

}

/// A short-lived lock file, taken by creating it exclusively and

/// released on drop: a workspace's snapshot lock, the operation head's

/// lock. A stale one (older than a minute, its process gone) is taken

/// over.

pub struct FileLock {

/// Held while a workspace is snapshotted.

pub type SnapshotLock = FileLock;

impl Drop for FileLock {

impl FileLock {

    /// Wait up to about five seconds for `path`; `busy` is the error after that.

    pub(crate) fn acquire(path: PathBuf, busy: &str) -> Result<FileLock> {

        for _ in 0..500 {

            match fs::OpenOptions::new()

                .write(true)

                .create_new(true)

                .open(&path)

            {

                Ok(mut f) => {

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

                    return Ok(FileLock { path });

                }

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

                    let stale = fs::metadata(&path)

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

                        .map(|t| t.elapsed().map(|d| d.as_secs() > 60).unwrap_or(false))

                        .unwrap_or(true);

                    if stale {

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

                        continue;

                    }

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

                }

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

            }

        }

        fail(busy)

    }

}

        FileLock::acquire(

            dir.join("lock"),

            "another oxy process is snapshotting this workspace",

        )

`view` is the complete state of all references and workspaces after the operation. The operation log is a DAG. The head (`.oxy/ops/head`) moves by compare-and-swap under a short lock (`.oxy/ops/lock`), held only while the pointer is written: a process that finds the head is no longer the parent it built on records a merge operation with both as parents, whose view is the three-way merge of the two views against that parent. A reference changed on one side takes that side's value; changed differently on both, a bookmark keeps both sides' heads (it then has several), a change beats a deletion, a tag keeps its first value, and workspaces, remote references and the policy take the later process's value. `oxy undo` refuses when the head moved while it ran.

Because all mutation goes through the operation log and object writes are idempotent, no long-held repository lock is needed: the operation head's lock covers one pointer write, and a workspace's snapshot lock covers one snapshot of that workspace.

*Changelog.* Draft 0.8h (2026-09-28): `oxy fsck` (§21.1); durability rules and the operation head as commit point (§13.6); the operation head moves by compare-and-swap and concurrent operations are merged (§12.1). Draft 0.8g (2026-09-27): delegates of a rule's identities may sign reference attestations (§14.5); local-path pushes bind unbound repositories like HTTP ones. Draft 0.8f (2026-09-27): challenge signatures (§14.3) for hosts. Draft 0.8e (2026-09-27): anonymous changes (zero author) are authenticated by a seal from any key, so imported and pre-identity history can be pushed under a policy (§14.4). Draft 0.8d (2026-09-27): unbound hosted repositories adopt the first pusher's id (§16.1). Draft 0.8c (2026-09-27): bare repositories (`oxy init --bare`, §12.2: no workspace, no working copy; hosting) and the reference server as an embeddable library (§15.1). Draft 0.8b (2026-09-27): the machine interface for editors (§24): `root`, `cat`, `version`, `hunks`, `discard`, and `--json` on the reporting commands.