perf: speed up concurrent imports with large peer histories (#1222 + clippy fix)

.changeset/concurrent-import-frontier-replay.md

@@ -0,0 +1,7 @@
+---
+"loro-crdt": patch
+---
+
+Speed up concurrent imports by reusing the current frontier, reducing repeated
+ancestor traversal, and filtering unrelated container history before materializing
+replay version vectors.

crates/loro-internal/docs/critical-version-spec.md

@@ -620,6 +620,19 @@ criss-cross ladder 测试钉"tips 去重后复杂度线性"（性能声明，不
 多头基准命中最后同步点、单头扫描停在初始分岔、不相交历史归零、
 切口落在同一 change 内部（style 对之间）仍收敛。
 
+前沿与重放性能优化路径的补充回归覆盖：
+
+- `src/tests/replay_base.rs::filtered_replay_preserves_full_causal_context`
+  以作者当时看到的版本和 change 的依赖前沿为对照检查 L13，覆盖跳过无关
+  历史、同一 change 混合多个容器的操作、并发分支、历史目标版本，以及
+  从快照恢复的历史。
+- `filtered_replay_preserves_critical_base_and_checkout_contract` 检查过滤在
+  向前导入与向后 checkout 时，保持重放基准、critical 认证标记、模式、
+  重放范围和上下文不变。
+- `src/version.rs::tests::frontier_contract_with_tied_heads_and_same_peer_prefixes`
+  以明确给出的预期前沿检查 D6/D7 和 L5，覆盖 lamport 相等的并发头、
+  同 peer 的冗余前缀，以及当前版本和历史版本的前沿转换。
+
 ---
 
 ## 6. 待定问题（需要维护者拍板）

crates/loro-internal/src/diff_calc.rs

@@ -118,7 +118,7 @@ pub(crate) struct DiffCalcVersionInfo<'a> {
     replay_base_vv: &'a VersionVector,
     /// The replay base is a critical version of `ancestry(from) ∪ ancestry(to)`:
     /// nothing above it is concurrent with anything below it. See
-    /// `OpLog::iter_from_replay_base_causally`.
+    /// `OpLog::try_iter_from_replay_base_causally`.
     replay_base_is_critical: bool,
 }
 
@@ -596,8 +596,13 @@ fn replay_container_ops_from_empty(
     let empty_vv = VersionVector::default();
     let empty_frontiers = Frontiers::default();
     let target_frontiers = oplog.dag.vv_to_frontiers(vv);
-    let (_, iter) =
-        oplog.iter_from_replay_base_causally(&empty_vv, &empty_frontiers, vv, &target_frontiers);
+    let (_, iter) = oplog.iter_from_replay_base_causally_filtered(
+        &empty_vv,
+        &empty_frontiers,
+        vv,
+        &target_frontiers,
+        Some(idx),
+    );
 
     for (change, (start_counter, end_counter), vv) in iter {
         let iter_start = change

crates/loro-internal/src/oplog.rs

@@ -133,7 +133,7 @@ pub(crate) enum CriticalVersionSearch {
     Unknown,
 }
 
-/// The replay base [`OpLog::iter_from_replay_base_causally`] chose.
+/// The replay base [`OpLog::try_iter_from_replay_base_causally`] chose.
 #[derive(Debug)]
 pub(crate) struct ReplayBase {
     pub vv: VersionVector,
@@ -1115,6 +1115,8 @@ impl OpLog {
     /// every container with an op in the concurrent old history: the diff
     /// calculator resolves those from history and treats them as "source not
     /// in op context", and may trust `from` as the base for everything else.
+    // Production replays go through the fallible or filtered entry points.
+    #[cfg(test)]
     #[allow(clippy::type_complexity)]
     pub(crate) fn iter_from_replay_base_causally(
         &self,
@@ -1138,6 +1140,37 @@ impl OpLog {
         (base, iter.map(Result::unwrap))
     }
 
+    /// Filter unrelated changes before materializing their version vectors.
+    #[allow(clippy::type_complexity)]
+    pub(crate) fn iter_from_replay_base_causally_filtered(
+        &self,
+        from: &VersionVector,
+        from_frontiers: &Frontiers,
+        to: &VersionVector,
+        to_frontiers: &Frontiers,
+        container_filter: Option<ContainerIdx>,
+    ) -> (
+        ReplayBase,
+        impl Iterator<
+                Item = (
+                    BlockChangeRef,
+                    (Counter, Counter),
+                    Rc<RefCell<VersionVector>>,
+                ),
+            > + '_,
+    ) {
+        let (base, iter) = self
+            .try_iter_from_replay_base_causally_filtered(
+                from,
+                from_frontiers,
+                to,
+                to_frontiers,
+                container_filter,
+            )
+            .unwrap();
+        (base, iter.map(Result::unwrap))
+    }
+
     #[allow(clippy::type_complexity)]
     pub(crate) fn try_iter_from_replay_base_causally(
         &self,
@@ -1160,6 +1193,39 @@ impl OpLog {
                 > + '_,
         ),
         LoroError,
+    > {
+        self.try_iter_from_replay_base_causally_filtered(
+            from,
+            from_frontiers,
+            to,
+            to_frontiers,
+            None,
+        )
+    }
+
+    #[allow(clippy::type_complexity)]
+    pub(crate) fn try_iter_from_replay_base_causally_filtered(
+        &self,
+        from: &VersionVector,
+        from_frontiers: &Frontiers,
+        to: &VersionVector,
+        to_frontiers: &Frontiers,
+        container_filter: Option<ContainerIdx>,
+    ) -> Result<
+        (
+            ReplayBase,
+            impl Iterator<
+                    Item = Result<
+                        (
+                            BlockChangeRef,
+                            (Counter, Counter),
+                            Rc<RefCell<VersionVector>>,
+                        ),
+                        LoroError,
+                    >,
+                > + '_,
+        ),
+        LoroError,
     > {
         let mut merged_vv = from.clone();
         merged_vv.merge(to);
@@ -1279,40 +1345,47 @@ impl OpLog {
             },
             std::iter::from_fn(move || {
                 let result = (|| -> Result<_, LoroError> {
-                    if let Some(inner) = &node {
-                        let mut inner_vv = vv.borrow_mut();
-                        // FIXME: PERF: it looks slow for large vv, like 10000+ entries
-                        inner_vv.clear();
-                        self.dag.try_ensure_vv_for(&inner.data)?;
-                        inner_vv.extend_to_include_vv(inner.data.vv.get().unwrap().iter());
-                        let peer = inner.data.peer;
-                        let cnt = inner
-                            .data
-                            .cnt
-                            .max(cur_cnt)
-                            .max(replay_base_vv.get(&peer).copied().unwrap_or(0));
-                        let dag_node_end = (inner.data.cnt + inner.data.len as Counter)
-                            .min(merged_vv.get(&peer).copied().unwrap_or(0));
-                        let change = match self.change_store.get_change(ID::new(peer, cnt)) {
-                            Some(change) => change,
-                            None => {
-                                self.check_history_parsable()?;
-                                panic!("DAG node must have a change");
+                    loop {
+                        if let Some(inner) = &node {
+                            let dag_node = inner.data.clone();
+                            let peer = inner.data.peer;
+                            let cnt = inner
+                                .data
+                                .cnt
+                                .max(cur_cnt)
+                                .max(replay_base_vv.get(&peer).copied().unwrap_or(0));
+                            let dag_node_end = (inner.data.cnt + inner.data.len as Counter)
+                                .min(merged_vv.get(&peer).copied().unwrap_or(0));
+                            let change = match self.change_store.get_change(ID::new(peer, cnt)) {
+                                Some(change) => change,
+                                None => {
+                                    self.check_history_parsable()?;
+                                    panic!("DAG node must have a change");
+                                }
+                            };
+
+                            if change.ctr_end() < dag_node_end {
+                                cur_cnt = change.ctr_end();
+                            } else {
+                                node = iter.next();
+                                cur_cnt = 0;
+                            }
+
+                            if container_filter
+                                .is_some_and(|idx| !change.ops.iter().any(|op| op.container == idx))
+                            {
+                                continue;
                             }
-                        };
+                            let mut inner_vv = vv.borrow_mut();
+                            inner_vv.clear();
+                            self.dag.try_ensure_vv_for(&dag_node)?;
+                            inner_vv.extend_to_include_vv(dag_node.vv.get().unwrap().iter());
+                            inner_vv.extend_to_include_end_id(change.id);
 
-                        if change.ctr_end() < dag_node_end {
-                            cur_cnt = change.ctr_end();
+                            return Ok(Some((change, (cnt, dag_node_end), vv.clone())));
                         } else {
-                            node = iter.next();
-                            cur_cnt = 0;
+                            return Ok(None);
                         }
-
-                        inner_vv.extend_to_include_end_id(change.id);
-
-                        Ok(Some((change, (cnt, dag_node_end), vv.clone())))
-                    } else {
-                        Ok(None)
                     }
                 })();
                 match result {

crates/loro-internal/src/oplog/loro_dag.rs

@@ -1348,6 +1348,10 @@ impl AppDag {
     }
 
     pub fn try_vv_to_frontiers(&self, vv: &VersionVector) -> LoroResult<Frontiers> {
+        // The current version already has a maintained frontier.
+        if vv == &self.vv {
+            return Ok(self.frontiers.clone());
+        }
         if vv.is_empty() {
             return Ok(Default::default());
         }

crates/loro-internal/src/tests/replay_base.rs

@@ -721,3 +721,145 @@ fn persisted_walks_record_each_movable_list_value_once() {
     });
     assert_eq!(len, 2 * 2 * rounds);
 }
+
+/// L13: filtering is a projection of yielded changes, never of their causal
+/// context. Record the author's actual version before each retained op as an
+/// oracle independent of the replay iterator (both iterator entry points share
+/// an implementation).
+#[test]
+fn filtered_replay_preserves_full_causal_context() {
+    use crate::handler::HandlerTrait;
+    use loro_common::ID;
+
+    let a = LoroDoc::new_auto_commit();
+    a.set_peer_id(1).unwrap();
+    a.get_map("noise").insert("seed", 0).unwrap();
+    a.commit_then_renew();
+    let b = a.fork();
+    b.set_peer_id(2).unwrap();
+    let c = a.fork();
+    c.set_peer_id(3).unwrap();
+    // A transitive dependency containing no target-container operations.
+    b.get_map("noise").insert("remote", 1).unwrap();
+    b.commit_then_renew();
+    a.import(&b.export(ExportMode::all_updates()).unwrap())
+        .unwrap();
+    a.get_map("noise").insert("local", 2).unwrap();
+    a.commit_then_renew();
+    let first_context = a.oplog_vv();
+    let first_id = ID::new(1, first_context.get(&1).copied().unwrap());
+    a.get_text("target").insert_unicode(0, "x").unwrap();
+    a.commit_then_renew();
+    // An unrelated op inside the retained change must still count toward the
+    // author's prefix when its later target op is handed to the tracker.
+    let second_change_context = a.oplog_vv();
+    a.get_map("noise").insert("mixed", 3).unwrap();
+    let mut second_context = second_change_context.clone();
+    let second_id = ID::new(1, second_context.get(&1).copied().unwrap() + 1);
+    second_context.extend_to_include_end_id(second_id);
+    a.get_text("target").insert_unicode(1, "y").unwrap();
+    a.commit_then_renew();
+    let historical = a.oplog_vv();
+    // Concurrent retained and skipped changes must not leak into A's context.
+    let concurrent_context = c.oplog_vv();
+    c.get_text("target").insert_unicode(0, "z").unwrap();
+    c.commit_then_renew();
+    c.get_map("noise").insert("concurrent", 4).unwrap();
+    c.commit_then_renew();
+    a.import(&c.export(ExportMode::all_updates()).unwrap())
+        .unwrap();
+
+    for cold in [false, true] {
+        let restored = LoroDoc::new_auto_commit();
+        let doc = if cold {
+            restored
+                .import(&a.export(ExportMode::Snapshot).unwrap())
+                .unwrap();
+            &restored
+        } else {
+            &a
+        };
+        let idx = doc.get_text("target").idx();
+        let oplog = doc.oplog().lock();
+        for target in [&historical, oplog.dag.vv()] {
+            let heads = oplog.dag.vv_to_frontiers(target);
+            let (_, iter) = oplog.iter_from_replay_base_causally_filtered(
+                &VersionVector::default(),
+                &Default::default(),
+                target,
+                &heads,
+                Some(idx),
+            );
+            let mut seen = std::collections::HashMap::new();
+            for (change, (start, end), context) in iter {
+                assert!(change.ops.iter().any(|op| op.container == idx));
+                let mut expected_change_context = oplog.dag.frontiers_to_vv(&change.deps).unwrap();
+                expected_change_context.extend_to_include_end_id(change.id);
+                assert_eq!(*context.borrow(), expected_change_context);
+                for op in change
+                    .ops
+                    .iter()
+                    .filter(|op| op.container == idx && op.counter >= start && op.counter < end)
+                {
+                    let id = ID::new(change.peer(), op.counter);
+                    let mut actual = context.borrow().clone();
+                    actual.extend_to_include_end_id(id);
+                    assert!(seen.insert(id, actual).is_none());
+                }
+            }
+            assert_eq!(seen.get(&first_id), Some(&first_context));
+            assert_eq!(seen.get(&second_id), Some(&second_context));
+            if target == &historical {
+                assert_eq!(seen.len(), 2);
+            } else {
+                assert_eq!(seen.len(), 3);
+                assert_eq!(seen.get(&ID::new(3, 0)), Some(&concurrent_context));
+            }
+        }
+    }
+}
+
+/// D8b / section 4.1: filtering must not change the selected base, its
+/// criticality certificate, the mode, or retained changes' ranges/contexts.
+#[test]
+fn filtered_replay_preserves_critical_base_and_checkout_contract() {
+    use crate::handler::HandlerTrait;
+    let (doc, sync, before, after) = one_concurrent_import(4);
+    let idx = doc.get_text("t").idx();
+    let absent = doc.get_text("absent").idx();
+    let oplog = doc.oplog().lock();
+    for (from, to) in [(&before, &after), (&after, &before), (&sync, &after)] {
+        let from_heads = oplog.dag.vv_to_frontiers(from);
+        let to_heads = oplog.dag.vv_to_frontiers(to);
+        let (expected_base, iter) =
+            oplog.iter_from_replay_base_causally(from, &from_heads, to, &to_heads);
+        let expected: Vec<_> = iter
+            .map(|(c, range, vv)| (c.id, range, vv.borrow().clone()))
+            .collect();
+        assert!(!expected.is_empty());
+        for filter in [idx, absent] {
+            let (base, iter) = oplog.iter_from_replay_base_causally_filtered(
+                from,
+                &from_heads,
+                to,
+                &to_heads,
+                Some(filter),
+            );
+            assert_eq!(base.vv, expected_base.vv);
+            assert_eq!(base.is_critical, expected_base.is_critical);
+            assert_eq!(base.diff_mode, expected_base.diff_mode);
+            assert_eq!(
+                base.concurrent_containers,
+                expected_base.concurrent_containers
+            );
+            let actual: Vec<_> = iter
+                .map(|(c, range, vv)| (c.id, range, vv.borrow().clone()))
+                .collect();
+            if filter == idx {
+                assert_eq!(actual, expected);
+            } else {
+                assert!(actual.is_empty());
+            }
+        }
+    }
+}