perf: rm thin wrapper for ops that will not trigger event (#894)

* perf: rm the event calling wrapper for ops that will not trigger events

* chore: changeset

* fix: revertTo and applyDiff

* fix: getCursorPos

* fix: add setDetachedEditing

Author: zxch3n

.changeset/mighty-ligers-dance.md

@@ -0,0 +1,6 @@
+---
+"loro-crdt": patch
+"loro-crdt-map": patch
+---
+
+perf: rm the event calling wrapper for ops that will not trigger events

AGENTS.md

@@ -0,0 +1,39 @@
+# Agent Notes (Loro)
+
+## Invariant: Flush Pending Events In `loro-wasm`
+
+In `crates/loro-wasm/src/lib.rs`, subscription callbacks (`subscribe*`, container `subscribe`, etc.)
+do not call user JS immediately. Instead, the binding enqueues JS calls into a global pending queue,
+and schedules a microtask check. If the microtask runs before `callPendingEvents()` flushes the
+queue, it will log:
+
+- `[LORO_INTERNAL_ERROR] Event not called`
+
+This creates a strict invariant:
+
+- **Any WASM-exposed API that can enqueue subscription events must flush pending events before
+  returning control back to JS.**
+
+To avoid adding overhead to every single op, we only wrap (decorate) a small allowlist of
+methods on the JS side. The wrapper calls `callPendingEvents()` in a `finally` block.
+
+### How To Maintain
+
+- When adding or changing a `#[wasm_bindgen]` API in `crates/loro-wasm/src/lib.rs` that can
+  *mutate document state*:
+  - If it can trigger an implicit commit / barrier (`commit`, `with_barrier` /
+    `implicit_commit_then_stop`), emit events (`emit_events`), or applies diffs (e.g. `revertTo`,
+    `applyDiff`), it typically **must** flush pending events.
+  - Add its JS name to the allowlist in `crates/loro-wasm/index.ts` near the bottom:
+    `decorateMethods(LoroDoc.prototype, [...])` (or the relevant prototype allowlist).
+  - If it is a pure read/query API (no state mutation, no commit/barrier, no event emission),
+    do **not** decorate it, to avoid unnecessary per-call cost.
+
+### Quick Check
+
+With active subscriptions (`doc.subscribe(...)` / container `subscribe(...)`), calling mutating APIs
+should not produce the error above. A recommended local check is:
+
+```sh
+pnpm -C crates/loro-wasm build-release
+```

crates/loro-wasm/index.ts

@@ -491,7 +491,36 @@ function decorateAllPrototypeMethods(prototype: object) {
   }
 }
 
-decorateAllPrototypeMethods(LoroDoc.prototype);
-decorateAllPrototypeMethods(EphemeralStoreWasm.prototype);
-decorateAllPrototypeMethods(AwarenessWasm.prototype);
+decorateMethods(LoroDoc.prototype, [
+  "setDetachedEditing",
+  "attach",
+  "detach",
+  "fork",
+  "forkAt",
+  "checkoutToLatest",
+  "checkout",
+  "commit",
+  "getCursorPos",
+  "revertTo",
+  "export",
+  "exportJsonUpdates",
+  "exportJsonInIdSpan",
+  "importJsonUpdates",
+  "import",
+  "importUpdateBatch",
+  "importBatch",
+  "travelChangeAncestors",
+  "getChangedContainersIn",
+  "diff",
+  "applyDiff",
+  "setPeerId",
+]);
+
+decorateMethods(EphemeralStoreWasm.prototype, [
+  "set",
+  "delete",
+  "apply",
+  "removeOutdated",
+]);
+
 decorateMethods(UndoManager.prototype, ["undo", "redo"]);

crates/loro-wasm/src/lib.rs

@@ -35,7 +35,7 @@ use parking_lot::lock_api::ReentrantMutex;
 use rle::HasLength;
 use serde::{Deserialize, Serialize};
 use std::{
-    cell::RefCell,
+    cell::{Cell, RefCell},
     cmp::Ordering,
     collections::VecDeque,
     ops::ControlFlow,
@@ -89,6 +89,10 @@ pub fn set_debug() {
 type JsResult<T> = Result<T, JsValue>;
 type EventCallback = Box<dyn Fn(&SafeJsValue) -> bool + Send + Sync + 'static>;
 
+thread_local! {
+    static IN_PRE_COMMIT_CALLBACK: Cell<bool> = Cell::new(false);
+}
+
 /// The CRDTs document. Loro supports different CRDTs include [**List**](LoroList),
 /// [**RichText**](LoroText), [**Map**](LoroMap) and [**Movable Tree**](LoroTree),
 /// you could build all kind of applications by these.
@@ -1508,7 +1512,15 @@ impl LoroDoc {
     #[wasm_bindgen(js_name = "exportJsonInIdSpan", skip_typescript)]
     pub fn exportJsonInIdSpan(&self, idSpan: JsIdSpan) -> JsResult<JsValue> {
         let id_span = js_to_id_span(idSpan)?;
-        let json = self.doc.export_json_in_id_span(id_span);
+        // Most LoroDoc reads run in an implicit-commit barrier (export/checkout/etc.).
+        // `exportJsonInIdSpan` is special: it is often called from `subscribePreCommit`,
+        // where the txn lock is already held. In that case, triggering another implicit
+        // commit would deadlock/panic. We skip the barrier while inside pre-commit.
+        let json = if IN_PRE_COMMIT_CALLBACK.with(|f| f.get()) {
+            self.doc.export_json_in_id_span(id_span)
+        } else {
+            self.doc.with_barrier(|| self.doc.export_json_in_id_span(id_span))
+        };
         let s = serde_wasm_bindgen::Serializer::new().serialize_maps_as_objects(true);
         let v = json
             .serialize(&s)
@@ -2235,7 +2247,13 @@ impl LoroDoc {
                 &ChangeModifier(e.modifier.clone()).into(),
             )
             .unwrap();
-            if let Err(e) = observer.call1(&obj.into()) {
+            let res = IN_PRE_COMMIT_CALLBACK.with(|f| {
+                let prev = f.replace(true);
+                let res = observer.call1(&obj.into());
+                f.set(prev);
+                res
+            });
+            if let Err(e) = res {
                 console_error!("Error: {:?}", e);
             }
             true
@@ -6626,7 +6644,9 @@ interface LoroDoc<T extends Record<string, Container> = Record<string, Container
      *
      * This method can also export pending changes from the uncommitted transaction that have not yet been applied to the OpLog.
      *
-     * This method will NOT trigger a new commit implicitly.
+     * This method will implicitly commit pending local operations (like `export(...)`) so callers can
+     * observe the latest local edits. When called inside `subscribePreCommit(...)`, it will NOT trigger
+     * an additional implicit commit.
      *
      * @param idSpan - The id span to export.
      * @returns The changes in the given id span.

crates/loro-wasm/tests/basic.test.ts

@@ -1244,6 +1244,41 @@ it("export json in id span #602", () => {
   }
 });
 
+it("export json in id span implicitly commits pending ops", () => {
+  const doc = new LoroDoc();
+  doc.setPeerId("1");
+  doc.getText("text").insert(0, "Hello");
+  expect(doc.getPendingTxnLength()).toBe(5);
+
+  const changes = doc.exportJsonInIdSpan({
+    peer: "1",
+    counter: 0,
+    length: 5,
+  });
+  expect(changes).toStrictEqual([
+    {
+      id: "0@1",
+      timestamp: expect.any(Number),
+      deps: [],
+      lamport: 0,
+      msg: undefined,
+      ops: [
+        {
+          container: "cid:root-text:Text",
+          counter: 0,
+          content: {
+            type: "insert",
+            pos: 0,
+            text: "Hello",
+          },
+        },
+      ],
+    },
+  ]);
+
+  expect(doc.getPendingTxnLength()).toBe(0);
+});
+
 it("find spans between versions", () => {
   const doc = new LoroDoc();
   doc.setPeerId("1");

crates/loro-wasm/tests/event.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import crypto from "crypto";
 import {
   Delta,
@@ -537,6 +537,65 @@ describe("event", () => {
       expect(text2.toString()).toBe("1. Hello World!");
     });
   });
+
+  it("revertTo flushes pending events", async () => {
+    const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+    try {
+      const doc = new LoroDoc();
+      doc.setPeerId("1");
+
+      let called = 0;
+      doc.subscribe(() => {
+        called += 1;
+      });
+
+      doc.getText("text").update("Hello");
+      doc.revertTo([]);
+      await Promise.resolve();
+
+      expect(called).toBeGreaterThan(0);
+      expect(
+        errorSpy.mock.calls.some((args) =>
+          args.some((arg) =>
+            String(arg).includes("[LORO_INTERNAL_ERROR] Event not called"),
+          ),
+        ),
+      ).toBe(false);
+    } finally {
+      errorSpy.mockRestore();
+    }
+  });
+
+  it("setDetachedEditing flushes pending events", async () => {
+    const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+    try {
+      const doc = new LoroDoc();
+      doc.setPeerId("1");
+
+      let called = 0;
+      doc.subscribe(() => {
+        called += 1;
+      });
+
+      doc.detach();
+      doc.setDetachedEditing(true);
+
+      doc.getText("text").update("Hello");
+      doc.setDetachedEditing(true);
+      await Promise.resolve();
+
+      expect(called).toBeGreaterThan(0);
+      expect(
+        errorSpy.mock.calls.some((args) =>
+          args.some((arg) =>
+            String(arg).includes("[LORO_INTERNAL_ERROR] Event not called"),
+          ),
+        ),
+      ).toBe(false);
+    } finally {
+      errorSpy.mockRestore();
+    }
+  });
 });
 
 it.skip("subscription works after timeout", async () => {