fix(rust): import_batch in multithread env (#796)

* fix(rust): import_batch in multithread env

Batch import may cause the doc enter a temporary detached state. If
there are local edits happen during this phase, they may end up with
errors like 'AutoCommitNotStarted' or read-only errors.

This temporary phase should not be visible to users. This fix will lock
the transaction object during the temporary detached state so it's not
visible to another thread.

* docs: add comment to code

Author: zxch3n

crates/loro-internal/src/loro.rs

@@ -208,7 +208,12 @@ impl LoroDoc {
     /// It only returns Some(options_of_the_empty_txn) when the txn is empty
     #[inline]
     #[must_use]
-    pub fn commit_then_stop(&self) -> (Option<CommitOptions>, LoroMutexGuard<'_, Option<Transaction>>) {
+    pub fn commit_then_stop(
+        &self,
+    ) -> (
+        Option<CommitOptions>,
+        LoroMutexGuard<'_, Option<Transaction>>,
+    ) {
         let (a, b) = self.commit_with(CommitOptions::new().immediate_renew(false));
         (a, b.unwrap())
     }
@@ -1176,13 +1181,35 @@ impl LoroDoc {
         });
 
         let (options, txn) = self.commit_then_stop();
-        drop(txn);
+        // Why we should keep locking `txn` here
+        //
+        // In a multi-threaded environment, `import_batch` used to drop the txn lock
+        // (via `commit_then_stop` + `drop(txn)`) and call `detach()`/`checkout_to_latest()`
+        // around the batch import. That created a race where another thread could
+        // start or renew the auto-commit txn and perform local edits while we were
+        // importing and temporarily detached. Those interleaved local edits could
+        // violate invariants between `OpLog` and `DocState` (e.g., state being
+        // updated when we expect it not to, missed events, or inconsistent
+        // frontiers), as exposed by the loom test `local_edits_during_batch_import`.
+        //
+        // The fix is to hold the txn mutex for the entire critical section:
+        // - Stop the current txn and keep the mutex guard.
+        // - Force-detach with `set_detached(true)` (avoids `detach()` side effects),
+        //   then run each `_import_with(...)` while detached so imports only touch
+        //   the `OpLog`.
+        // - After importing, reattach by checking out to latest and renew the txn
+        //   using `_checkout_to_latest_with_guard`, which keeps the mutex held while
+        //   (re)starting the auto-commit txn.
+        //
+        // Holding the lock ensures no concurrent thread can create/renew a txn and
+        // do local edits in the middle of the batch import, making the whole
+        // operation atomic with respect to local edits.
         let is_detached = self.is_detached();
-        self.detach();
+        self.set_detached(true);
         self.oplog.lock().unwrap().batch_importing = true;
         let mut err = None;
         for (_meta, data) in meta_arr {
-            match self.import(data) {
+            match self._import_with(data, Default::default()) {
                 Ok(s) => {
                     for (peer, (start, end)) in s.success.iter() {
                         match success.0.entry(*peer) {
@@ -1218,9 +1245,10 @@ impl LoroDoc {
         let mut oplog = self.oplog.lock().unwrap();
         oplog.batch_importing = false;
         drop(oplog);
-
         if !is_detached {
-            self.checkout_to_latest();
+            self._checkout_to_latest_with_guard(txn);
+        } else {
+            drop(txn);
         }
 
         self.renew_txn_if_auto_commit(options);
@@ -1269,6 +1297,16 @@ impl LoroDoc {
         self.renew_txn_if_auto_commit(options);
     }
 
+    fn _checkout_to_latest_with_guard(&self, guard: LoroMutexGuard<Option<Transaction>>) {
+        if !self.is_detached() {
+            self._renew_txn_if_auto_commit_with_guard(None, guard);
+            return;
+        }
+
+        self._checkout_to_latest_without_commit(true);
+        self._renew_txn_if_auto_commit_with_guard(None, guard);
+    }
+
     /// NOTE: The caller of this method should ensure the txn is locked and set to None
     pub(crate) fn _checkout_to_latest_without_commit(&self, to_commit_then_renew: bool) {
         tracing::info_span!("CheckoutToLatest", peer = self.peer_id()).in_scope(|| {

crates/loro-internal/src/txn.rs

@@ -26,7 +26,7 @@ use crate::{
     event::{Diff, ListDeltaMeta, TextDiff},
     handler::{Handler, ValueOrHandler},
     id::{Counter, PeerID, ID},
-    lock::LoroMutex,
+    lock::{LoroMutex, LoroMutexGuard},
     loro::CommitOptions,
     op::{Op, RawOp, RawOpContent},
     pre_commit::{ChangeModifier, PreCommitCallbackPayload},
@@ -120,6 +120,25 @@ impl crate::LoroDoc {
             self_txn.replace(txn);
         }
     }
+
+    #[inline]
+    pub(crate) fn _renew_txn_if_auto_commit_with_guard(
+        &self,
+        options: Option<CommitOptions>,
+        mut guard: LoroMutexGuard<Option<Transaction>>,
+    ) {
+        if self.auto_commit.load(std::sync::atomic::Ordering::Acquire) && self.can_edit() {
+            if guard.is_some() {
+                return;
+            }
+
+            let mut txn = self.txn().unwrap();
+            if let Some(options) = options {
+                txn.set_options(options);
+            }
+            guard.replace(txn);
+        }
+    }
 }
 
 pub(crate) type OnCommitFn =

crates/loro/src/lib.rs

@@ -112,6 +112,10 @@ impl Default for LoroDoc {
 }
 
 impl Clone for LoroDoc {
+    /// This creates a reference clone, not a deep clone. The cloned doc will share the same
+    /// underlying doc as the original one.
+    ///
+    /// For deep clone, please use the `.fork()` method.
     fn clone(&self) -> Self {
         let doc = self.doc.clone();
         LoroDoc::_new(doc)

crates/loro/tests/multi_thread_test.rs

@@ -2,7 +2,7 @@
 mod loom_test {
 
     use loom::{explore, stop_exploring, sync::atomic::AtomicUsize, thread::yield_now};
-    use loro::{ExportMode, LoroDoc};
+    use loro::{ExportMode, LoroDoc, UpdateOptions};
     use std::{sync::atomic::Ordering, thread::sleep, time::Duration};
 
     #[test]
@@ -264,4 +264,37 @@ mod loom_test {
             }
         });
     }
+
+    #[test]
+    fn local_edits_during_batch_import() {
+        let mut builder = loom::model::Builder::new();
+        builder.max_branches = 3000;
+        builder.check(move || {
+            let doc = LoroDoc::new();
+            doc.get_text("text").insert(0, "hello").unwrap();
+            let update_a = doc.export(ExportMode::all_updates()).unwrap();
+            doc.get_text("text")
+                .update("yo! hello", UpdateOptions::default())
+                .unwrap();
+            let update_b = doc.export(ExportMode::all_updates()).unwrap();
+
+            let mut handlers = vec![];
+            let doc = LoroDoc::new();
+            let doc_clone = doc.clone();
+
+            handlers.push(loom::thread::spawn(move || {
+                doc_clone.get_text("text").insert(0, "1").unwrap();
+                doc_clone.commit();
+                doc_clone.get_text("text").insert(0, "1").unwrap();
+                doc_clone.commit();
+            }));
+
+            handlers.push(loom::thread::spawn(move || {
+                doc.import_batch(&[update_a, update_b]).unwrap();
+            }));
+            for h in handlers {
+                h.join().unwrap();
+            }
+        });
+    }
 }