OXYGENHUB

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

A repository whose objects are on any store but this disk's directory
keeps its operation head in the store's ops/head reference record.
Recording an operation reads the record's version and moves it only if
unchanged; when another process moved it first the two operations are
merged as on disk and the write is tried again. Undo and the format-1
migration go through the same swap, and fsck reads the head the same
way. A local repository keeps .oxy/ops/head under its lock. refs #68
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-10-05 01:08 UTC
Signature
✔ signed by Andy McCall
Hash
dd21fd82687d832b61f53f9e9b50ad6ad31c21b25854ffec1a11ba2130bfac1a
Depends on
192fad1e46a2 19555025a2ac 236f35c0baba 33be24095b4b 387335ff7f4d 648189388f1a 75167672ebcd 982864068a59 98a1620a3840 ff0af1ee299d
Depended on by
767ac6cd0545 530a0fc650a2 d069701a4997
In bookmarks
bare-hashes carried-not-removed carry-identical doctor-trust identical-sides keyless-tags main merge-remote-bookmark only-help-103 paths-for-resolutions-and-amend push-notes show-stat subrepo-pin-state switch-says
In tags
v0.10.0 v0.11.0 v0.4.1 v0.4.2 v0.4.3 v0.4.4 v0.4.5 v0.4.6 v0.5.0 v0.5.1 v0.6.0 v0.7.0 v0.8.0 v0.9.0
Edits
16 atoms, 0 tree operations, 4 files

Browse the tree at this change

modified crates/oxy-core/src/repo/fsck.rs +6 −6
…
159 unchanged lines
160 160
161 161
    /// Pass 2.
162 162
    fn fsck_ops(&self, rep: &mut FsckReport, bad: &HashSet<ObjectRef>) {
163 -
        let head = match fs::read_to_string(self.oxy.join("ops").join("head")) {
164 -
            Ok(h) => h,
163 +
        let head = match self.read_head() {
164 +
            Ok((h, _)) => h,
165 +
            Err(RepoError::Msg(m)) if m.starts_with("corrupt") => {
166 +
                rep.error(None, ".oxy/ops/head does not hold an object reference");
167 +
                return;
168 +
            }
165 169
            Err(e) => {
166 170
                rep.error(None, format!("cannot read .oxy/ops/head: {e}"));
167 171
                return;
168 172
            }
169 -
        };
170 -
        let Some(head) = ObjectRef::from_hex(head.trim()) else {
171 -
            rep.error(None, ".oxy/ops/head does not hold an object reference");
172 -
            return;
173 173
        };
174 174
        let mut seen = HashSet::new();
175 175
        let mut queue: VecDeque<(ObjectRef, Option<(ObjectRef, u64)>)> =
…
337 unchanged lines
modified crates/oxy-core/src/repo/mod.rs +113 −44
…
59 unchanged lines
60 60
61 61
pub type Result<T> = std::result::Result<T, RepoError>;
62 62
63 +
/// The operation head's reference record on a store that is not this
64 +
/// disk's directory (oxygen#68).
65 +
pub const HEAD_RECORD: &str = "ops/head";
66 +
67 +
/// How often recording an operation reads the head again when other
68 +
/// processes keep moving it.
69 +
const HEAD_TRIES: usize = 8;
70 +
63 71
fn fail<T>(s: impl Into<String>) -> Result<T> {
64 72
    Err(RepoError::Msg(s.into()))
65 73
}
…
450 unchanged lines
516 524
    /// Rewrite a format-1 operation log, whose bookmarks were head lists,
517 525
    /// into format 2, where each bookmark names a `state` object.
518 526
    fn migrate_format1(&self) -> Result<()> {
519 -
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
520 -
        let head = ObjectRef::from_hex(hex.trim())
521 -
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;
527 +
        let (head, version) = self.read_head()?;
522 528
        // Collect the chain root-first.
523 529
        let mut chain: Vec<Format1Op> = Vec::new();
524 530
        let mut cur = Some(head);
…
21 unchanged lines
546 552
            };
547 553
            new_parent = Some(self.store.put(ObjectType::Op, &op.encode())?);
548 554
        }
549 -
        write_atomic(
550 -
            &self.oxy.join("ops").join("head"),
551 -
            new_parent.unwrap().to_hex().as_bytes(),
552 -
        )?;
555 +
        if !self.swap_head(&new_parent.unwrap(), version.as_ref())? {
556 +
            return fail("the operation log changed while it was rewritten; run the command again");
557 +
        }
553 558
        write_atomic(&self.oxy.join("format"), FORMAT_LINE.as_bytes())?;
554 559
        Ok(())
555 560
    }
…
121 unchanged lines
677 682
        if let Some(r) = self.at_op {
678 683
            return Ok((r, Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?));
679 684
        }
680 -
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
681 -
        let r = ObjectRef::from_hex(hex.trim())
682 -
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;
685 +
        let r = self.head_ref()?;
683 686
        let op = Op::decode(&self.store.get_typed(&r, ObjectType::Op)?)?;
684 687
        Ok((r, op))
685 688
    }
…
19 unchanged lines
705 708
        let flushing = crate::timing::phase("op: flush the store");
706 709
        self.store.flush()?;
707 710
        drop(flushing);
708 -
        // Compare and swap under a short lock (spec §12.1): when another
709 -
        // process recorded an operation since this one read the head, the
710 -
        // two are merged instead of one overwriting the other.
711 -
        let _lock = self.lock_ops()?;
712 -
        let current = match parents.first() {
711 +
        // Compare and swap (spec §12.1): when another process recorded an
712 +
        // operation since this one read the head, the two are merged
713 +
        // instead of one overwriting the other. On this disk the head moves
714 +
        // under a short lock; on any other store by its record's version,
715 +
        // and a record that moved meanwhile is read again and merged again.
716 +
        let _lock = if self.store.is_local() {
717 +
            Some(self.lock_ops()?)
718 +
        } else {
719 +
            None
720 +
        };
721 +
        let Some(parent) = parents.first() else {
713 722
            // The first operation (init): there is no head yet.
714 -
            None => None,
715 -
            Some(_) => Some(self.head_ref()?),
716 -
        };
717 -
        let Some(current) = current.filter(|c| Some(c) != parents.first()) else {
718 -
            write_atomic(&self.oxy.join("ops").join("head"), r.to_hex().as_bytes())?;
723 +
            if !self.swap_head(&r, None)? {
724 +
                return fail(format!(
725 +
                    "{} already holds a repository's operation log",
726 +
                    self.store.backend().describe()
727 +
                ));
728 +
            }
719 729
            return Ok(r);
720 730
        };
721 -
        let base = match parents.first() {
722 -
            Some(p) => Op::decode(&self.store.get_typed(p, ObjectType::Op)?)?.view,
723 -
            None => View::default(),
724 -
        };
725 -
        let theirs = Op::decode(&self.store.get_typed(&current, ObjectType::Op)?)?.view;
726 -
        let merged = Op {
727 -
            parents: vec![current, r],
728 -
            hlc: Hlc::now(),
729 -
            command: format!("{command} (merged with a concurrent operation)"),
730 -
            view: self.merge_views(&base, &op.view, &theirs)?,
731 -
        };
732 -
        let m = self.store.put(ObjectType::Op, &merged.encode())?;
733 -
        self.store.flush()?;
734 -
        write_atomic(&self.oxy.join("ops").join("head"), m.to_hex().as_bytes())?;
735 -
        Ok(m)
731 +
        let base = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?.view;
732 +
        for _ in 0..HEAD_TRIES {
733 +
            let (current, version) = self.read_head()?;
734 +
            let to = if current == *parent {
735 +
                r
736 +
            } else {
737 +
                let theirs = Op::decode(&self.store.get_typed(&current, ObjectType::Op)?)?.view;
738 +
                let merged = Op {
739 +
                    parents: vec![current, r],
740 +
                    hlc: Hlc::now(),
741 +
                    command: format!("{command} (merged with a concurrent operation)"),
742 +
                    view: self.merge_views(&base, &op.view, &theirs)?,
743 +
                };
744 +
                let m = self.store.put(ObjectType::Op, &merged.encode())?;
745 +
                self.store.flush()?;
746 +
                m
747 +
            };
748 +
            if self.swap_head(&to, version.as_ref())? {
749 +
                return Ok(to);
750 +
            }
751 +
        }
752 +
        fail("other oxy processes kept changing the operation log while this one recorded; run the command again")
736 753
    }
737 754
738 755
    /// The operation head's lock, held only while the head pointer moves.
…
6 unchanged lines
745 762
    }
746 763
747 764
    fn head_ref(&self) -> Result<ObjectRef> {
748 -
        let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
749 -
        ObjectRef::from_hex(hex.trim())
750 -
            .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))
765 +
        Ok(self.read_head()?.0)
766 +
    }
767 +
768 +
    /// The operation head, and on a store its record's version for
769 +
    /// [`Repo::swap_head`]. On this disk the head is `.oxy/ops/head`; on
770 +
    /// any other store (a bucket, or memory in tests) it is the store's
771 +
    /// [`HEAD_RECORD`] reference record (oxygen#68).
772 +
    fn read_head(&self) -> Result<(ObjectRef, Option<crate::strata::Version>)> {
773 +
        if self.store.is_local() {
774 +
            let hex = fs::read_to_string(self.oxy.join("ops").join("head"))?;
775 +
            let r = ObjectRef::from_hex(hex.trim())
776 +
                .ok_or_else(|| RepoError::Msg("corrupt .oxy/ops/head".into()))?;
777 +
            return Ok((r, None));
778 +
        }
779 +
        let backend = self.store.backend();
780 +
        match backend.read_ref(HEAD_RECORD) {
781 +
            Ok(Some((bytes, version))) => {
782 +
                let r = ObjectRef::from_hex(String::from_utf8_lossy(&bytes).trim()).ok_or_else(
783 +
                    || {
784 +
                        RepoError::Msg(format!(
785 +
                            "the operation head in {} is corrupt",
786 +
                            backend.describe()
787 +
                        ))
788 +
                    },
789 +
                )?;
790 +
                Ok((r, Some(version)))
791 +
            }
792 +
            Ok(None) => fail(format!(
793 +
                "{} holds no operation log (no {HEAD_RECORD} record)",
794 +
                backend.describe()
795 +
            )),
796 +
            Err(e) => fail(format!("read the operation head: {e}")),
797 +
        }
751 798
    }
752 799
800 +
    /// Move the operation head to `to` from the `expected` version
801 +
    /// [`Repo::read_head`] gave (`None`: there is no head yet). On this
802 +
    /// disk the caller holds the operations lock and the head always
803 +
    /// moves; on a store it moves only if the record is unchanged, and
804 +
    /// `false` says another process moved it first.
805 +
    fn swap_head(&self, to: &ObjectRef, expected: Option<&crate::strata::Version>) -> Result<bool> {
806 +
        if self.store.is_local() {
807 +
            write_atomic(&self.oxy.join("ops").join("head"), to.to_hex().as_bytes())?;
808 +
            return Ok(true);
809 +
        }
810 +
        match self
811 +
            .store
812 +
            .backend()
813 +
            .write_ref(HEAD_RECORD, to.to_hex().as_bytes(), expected)
814 +
        {
815 +
            Ok(_) => Ok(true),
816 +
            Err(crate::strata::Error::Conflict(_)) => Ok(false),
817 +
            Err(e) => fail(format!("move the operation head: {e}")),
818 +
        }
819 +
    }
820 +
753 821
    /// Three-way merge of two views that both started from `base` (spec
754 822
    /// §12.1). A reference only one side changed takes that side's value.
755 823
    /// When both changed it differently: a bookmark keeps both sides'
…
149 unchanged lines
905 973
        };
906 974
        let parent_op = Op::decode(&self.store.get_typed(parent, ObjectType::Op)?)?;
907 975
        {
908 -
            let _lock = self.lock_ops()?;
909 -
            if self.head_ref()? != read_head {
976 +
            let _lock = if self.store.is_local() {
977 +
                Some(self.lock_ops()?)
978 +
            } else {
979 +
                None
980 +
            };
981 +
            let (now, version) = self.read_head()?;
982 +
            if now != read_head || !self.swap_head(parent, version.as_ref())? {
910 983
                return fail("another oxy command ran while undoing; run undo again");
911 984
            }
912 -
            write_atomic(
913 -
                &self.oxy.join("ops").join("head"),
914 -
                parent.to_hex().as_bytes(),
915 -
            )?;
916 985
        }
917 986
        self.drop_manifest();
918 987
        if let Some(then) = parent_op.view.workspaces.get(&self.workspace) {
…
1239 unchanged lines
modified crates/oxy-core/src/repo/tests.rs +61 −0
…
7530 unchanged lines
7531 7531
    let v = repo.verify_git_import(&g, "one").unwrap();
7532 7532
    assert!(v.mismatches.is_empty(), "{:?}", v.mismatches);
7533 7533
}
7534 +
7535 +
/// oxygen#68: on a store that is not this disk's directory, the operation
7536 +
/// head is the store's reference record, moved by compare-and-swap. Two
7537 +
/// handles recording from the same head both get through, merged; a swap
7538 +
/// from a stale version is refused; a second repository cannot start on a
7539 +
/// store that already holds one.
7540 +
#[test]
7541 +
fn over_a_store_the_operation_head_moves_by_compare_and_swap() {
7542 +
    use crate::strata::{MemoryStore, Store as _};
7543 +
    let backend = std::sync::Arc::new(MemoryStore::new());
7544 +
    crate::store::TEST_BACKEND.with(|t| *t.borrow_mut() = Some(backend.clone()));
7545 +
    let dir = tempfile::tempdir().unwrap();
7546 +
    let a = Repo::init(dir.path()).unwrap();
7547 +
    assert!(
7548 +
        !dir.path().join(".oxy/ops/head").exists(),
7549 +
        "the head is not a file"
7550 +
    );
7551 +
    let (h0, op0) = a.head_op().unwrap();
7552 +
    let record = backend
7553 +
        .read_ref(HEAD_RECORD)
7554 +
        .unwrap()
7555 +
        .expect("the head is the store's record");
7556 +
    assert_eq!(String::from_utf8(record.0).unwrap(), h0.to_hex());
7557 +
7558 +
    let b = Repo::open(dir.path()).unwrap();
7559 +
    let state = a.store.put_state(&StateObject::new(vec![])).unwrap();
7560 +
    let mut va = op0.view.clone();
7561 +
    va.bookmarks.insert("from-a".into(), state);
7562 +
    let mut vb = op0.view.clone();
7563 +
    vb.bookmarks.insert("from-b".into(), state);
7564 +
    let rb = b.record_op(&[h0], "bookmark from-b", vb).unwrap();
7565 +
    assert_eq!(b.head_ref().unwrap(), rb);
7566 +
    // a recorded from h0 too: the head moved, so the two are merged.
7567 +
    let ra = a.record_op(&[h0], "bookmark from-a", va).unwrap();
7568 +
    let (head, op) = Repo::open(dir.path()).unwrap().head_op().unwrap();
7569 +
    assert_eq!(head, ra);
7570 +
    assert!(
7571 +
        op.command.ends_with("(merged with a concurrent operation)"),
7572 +
        "{}",
7573 +
        op.command
7574 +
    );
7575 +
    assert_eq!(op.parents, vec![rb, op.parents[1]]);
7576 +
    assert!(op.view.bookmarks.contains_key("from-a") && op.view.bookmarks.contains_key("from-b"));
7577 +
7578 +
    // A swap from a version that is no longer current moves nothing.
7579 +
    let (h, v) = a.read_head().unwrap();
7580 +
    assert!(b.swap_head(&rb, v.as_ref()).unwrap());
7581 +
    assert!(!a.swap_head(&h, v.as_ref()).unwrap(), "stale");
7582 +
    assert_eq!(a.head_ref().unwrap(), rb);
7583 +
7584 +
    let other = tempfile::tempdir().unwrap();
7585 +
    let err = Repo::init(other.path())
7586 +
        .err()
7587 +
        .expect("the store already holds a log")
7588 +
        .to_string();
7589 +
    assert!(
7590 +
        err.contains("already holds a repository's operation log"),
7591 +
        "{err}"
7592 +
    );
7593 +
    crate::store::TEST_BACKEND.with(|t| *t.borrow_mut() = None);
7594 +
}
modified crates/oxy-core/src/store.rs +12 −0
…
458 unchanged lines
459 459
    Some(store)
460 460
}
461 461
462 +
#[cfg(test)]
463 +
thread_local! {
464 +
    /// A backend every store opened on this thread uses instead of its
465 +
    /// directory: tests of a repository kept on a store that is not this
466 +
    /// disk's directory, with several handles on one store.
467 +
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
468 +
}
469 +
462 470
impl Store {
463 471
    /// Open the store in `dir`, which is also its scratch directory.
464 472
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
465 473
        let dir = dir.into();
466 474
        fs::create_dir_all(dir.join("loose"))?;
475 +
        #[cfg(test)]
476 +
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
477 +
            return Store::open_on(backend, dir);
478 +
        }
467 479
        if let Some(backend) = memory_backend_for(&dir) {
468 480
            return Store::open_on(backend, dir);
469 481
        }
…
2014 unchanged lines