SQLite Transactions

server/src/error_util.rs

@@ -8,6 +8,7 @@ impl From<StoreError> for tonic::Status {
         let msg = err.to_string();
         match err {
             StoreError::Decode(_) => tonic::Status::data_loss(msg),
+            StoreError::CasConflict(_) => tonic::Status::aborted(msg),
             StoreError::Read(_)
             | StoreError::Write(_)
             | StoreError::Encode(_)

server/src/op_store.rs

@@ -195,7 +195,11 @@ impl OpStoreService for CommitCloudOpStoreService {
 
         let current_heads = self
             .store
-            .update_op_heads(req.repo_id, &req.old_op_head_ids, req.new_op_head_id)
+            .update_op_heads_append_new_remove_old(
+                req.repo_id,
+                &req.old_op_head_ids,
+                req.new_op_head_id,
+            )
             .await?;
 
         Ok(tonic::Response::new(UpdateOpHeadsResponse {

server/src/store/memorystore.rs

@@ -4,7 +4,7 @@ use cc_common::op_store::*;
 use std::collections::{HashMap, HashSet};
 use std::sync::Mutex;
 
-use super::{Store, StoreResult};
+use super::{Store, StoreError, StoreResult};
 
 pub type RepoId = String;
 pub type CommitId = Vec<u8>;
@@ -121,39 +121,64 @@ impl Store for MemoryStore {
         Ok(op_heads.get(repo_id).cloned())
     }
 
-    // Updates op heads by removing old superseded heads and appending new_id, while preserving concurrent sibling heads.
-    // This is required for divergent op head resolution down the road—if we wiped out sibling heads here, client-side reconciliation wouldn't receive the divergent heads it needs to merge them.
-    async fn update_op_heads(
+    async fn update_op_heads_append_new_remove_old(
         &self,
         repo_id: String,
-        old_ids: &[Vec<u8>],
-        new_id: Vec<u8>,
+        ids_to_remove: &[Vec<u8>],
+        id_to_add: Vec<u8>,
     ) -> StoreResult<Vec<OpId>> {
         let mut op_heads = self.op_heads.lock().unwrap();
         let current_heads = op_heads
-            .entry(repo_id)
+            .entry(repo_id.clone())
             .or_insert_with(|| vec![cc_common::ROOT_OPERATION_ID_BYTES.to_vec()]);
 
-        if old_ids.is_empty() {
-            if !current_heads.contains(&new_id) {
-                current_heads.push(new_id);
+        if ids_to_remove.is_empty() {
+            if !current_heads.contains(&id_to_add) {
+                current_heads.push(id_to_add);
             }
         } else {
-            let mut removed_any = false;
-            current_heads.retain(|head| {
-                if old_ids.contains(head) {
-                    removed_any = true;
-                    false
-                } else {
-                    true
+            for old in ids_to_remove {
+                if !current_heads.contains(old) {
+                    return Err(StoreError::CasConflict(format!(
+                        "write race detected on stale old_op_head_ids: head {} no longer active in repo {}",
+                        hex::encode(old),
+                        repo_id
+                    )));
                 }
-            });
+            }
 
-            if removed_any && !current_heads.contains(&new_id) {
-                current_heads.push(new_id);
+            current_heads.retain(|head| !ids_to_remove.contains(head));
+            if !current_heads.contains(&id_to_add) {
+                current_heads.push(id_to_add);
             }
         }
 
         Ok(current_heads.clone())
     }
+
+    async fn update_op_heads_compare_and_swap(
+        &self,
+        repo_id: String,
+        expected_exact_heads: &[Vec<u8>],
+        new_id: Vec<u8>,
+    ) -> StoreResult<Vec<OpId>> {
+        let mut op_heads = self.op_heads.lock().unwrap();
+        let current_heads = op_heads
+            .entry(repo_id.clone())
+            .or_insert_with(|| vec![cc_common::ROOT_OPERATION_ID_BYTES.to_vec()]);
+
+        let current_set: std::collections::HashSet<Vec<u8>> =
+            current_heads.iter().cloned().collect();
+        let expected_set: std::collections::HashSet<Vec<u8>> =
+            expected_exact_heads.iter().cloned().collect();
+
+        if current_set != expected_set {
+            return Err(StoreError::CasConflict(format!(
+                "reconciliation CAS conflict for repo {repo_id}: expected heads {expected_set:?}, found {current_set:?}"
+            )));
+        }
+
+        *current_heads = vec![new_id.clone()];
+        Ok(vec![new_id])
+    }
 }

server/src/store/mod.rs

@@ -18,6 +18,8 @@ pub enum StoreError {
     Read(String),
     // A write to the storage backend failed.
     Write(String),
+    // A Compare-And-Swap (CAS) optimistic concurrency conflict occurred.
+    CasConflict(String),
     // Data failed to be encoded into protobuf format.
     Encode(String),
     // Data was retrieved but could not be decoded, indicating corruption.
@@ -31,6 +33,7 @@ impl std::fmt::Display for StoreError {
         match self {
             StoreError::Read(msg) => write!(f, "storage read error: {msg}"),
             StoreError::Write(msg) => write!(f, "storage write error: {msg}"),
+            StoreError::CasConflict(msg) => write!(f, "CAS conflict: {msg}"),
             StoreError::Encode(msg) => write!(f, "storage encode error: {msg}"),
             StoreError::Decode(msg) => write!(f, "stored data is corrupt: {msg}"),
             StoreError::Task(msg) => write!(f, "storage task error: {msg}"),
@@ -74,10 +77,22 @@ pub trait Store: Send + Sync {
     async fn put_view(&self, repo_id: String, view_id: Vec<u8>, view: View) -> StoreResult<()>;
 
     async fn get_op_heads(&self, repo_id: &str) -> StoreResult<Option<Vec<Vec<u8>>>>;
-    async fn update_op_heads(
+    // Updates op heads by removing `ids_to_remove` (if present) and appending `id_to_add`.
+    // Validates that `ids_to_remove` are currently active to prevent stale client overwrites.
+    async fn update_op_heads_append_new_remove_old(
         &self,
         repo_id: String,
-        old_ids: &[Vec<u8>],
+        ids_to_remove: &[Vec<u8>],
+        id_to_add: Vec<u8>,
+    ) -> StoreResult<Vec<Vec<u8>>>;
+
+    // Atomically compares the full set of currently active op heads against `expected_exact_heads`
+    // and replaces them entirely with `[new_id]`. Fails immediately with `StoreError::CasConflict`
+    // if the active set does not match `expected_exact_heads`. Used during server-side reconciliation.
+    async fn update_op_heads_compare_and_swap(
+        &self,
+        repo_id: String,
+        expected_exact_heads: &[Vec<u8>],
         new_id: Vec<u8>,
     ) -> StoreResult<Vec<Vec<u8>>>;
 }