fix(persistence): always route mutations through coordinator to prevent seq collisions

kevin-dp · claude · kevin-dp · commit 533be706ebd0 · 2026-03-05T13:25:33.000+01:00
The leader tab had two mutation paths: a "direct" path (write to SQLite
and broadcast) and an RPC path (through the coordinator). Previously,
only follower tabs used the RPC path — the leader bypassed the
coordinator and wrote directly.

This caused a seq collision: the leader's direct writes incremented the
runtime's `localSeq` but left the coordinator's `state.latestSeq` at 0.
When a follower later sent an RPC, the coordinator assigned seq starting
from 1 again, producing duplicate seq numbers. The leader then skipped
these "already-seen" tx:committed messages, causing follower mutations
to silently disappear.

Fix: Always route through `requestApplyLocalMutations` when available,
regardless of leader/follower status. This keeps the coordinator's seq
counter in sync with all writes.

Also removes `requestApplyLocalMutations` from `SingleProcessCoordinator`
— it was a stub that returned success without persisting, which would
break now that the leader uses this path. Single-process mode correctly
falls back to the direct path since it has no multi-tab coordination.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/packages/db-sqlite-persisted-collection-core/src/persisted.ts b/packages/db-sqlite-persisted-collection-core/src/persisted.ts
@@ -406,21 +406,6 @@ export class SingleProcessCoordinator implements PersistedCollectionCoordinator
 
   public async requestEnsurePersistedIndex(): Promise<void> {}
 
-  public requestApplyLocalMutations(
-    _collectionId: string,
-    mutations: Array<PersistedMutationEnvelope>,
-  ): Promise<ApplyLocalMutationsResponse> {
-    return Promise.resolve({
-      type: `rpc:applyLocalMutations:res`,
-      rpcId: crypto.randomUUID(),
-      ok: true,
-      term: 1,
-      seq: mutations.length,
-      latestRowVersion: mutations.length,
-      acceptedMutationIds: mutations.map((mutation) => mutation.mutationId),
-    })
-  }
-
   public pullSince(): Promise<PullSinceResponse> {
     return Promise.resolve({
       type: `rpc:pullSince:res`,
@@ -1321,10 +1306,13 @@ class PersistedCollectionRuntime<
   private async persistCollectionMutationsUnsafe(
     mutations: Array<PendingMutation<T>>,
   ): Promise<Array<string>> {
-    if (
-      this.persistence.coordinator.requestApplyLocalMutations &&
-      !this.persistence.coordinator.isLeader(this.collectionId)
-    ) {
+    // When a coordinator with requestApplyLocalMutations is available, always
+    // route through it — even on the leader tab. This ensures the coordinator's
+    // seq/rowVersion counters stay in sync with actual writes. Without this,
+    // the leader's direct-path writes would increment the runtime's localSeq
+    // but leave the coordinator's state.latestSeq stale, causing seq collisions
+    // when follower RPCs later arrive.
+    if (this.persistence.coordinator.requestApplyLocalMutations) {
       const envelopeMutations = mutations.map((mutation) =>
         toPersistedMutationEnvelope(
           mutation as unknown as PendingMutation<Record<string, unknown>>,
@@ -1368,6 +1356,8 @@ class PersistedCollectionRuntime<
       return uniqueAcceptedMutationIds
     }
 
+    // Fallback: no coordinator with requestApplyLocalMutations (e.g.
+    // SingleProcessCoordinator). Apply directly and broadcast.
     const streamPosition = this.nextLocalStreamPosition()
     const tx = this.createPersistedTxFromMutations(mutations, streamPosition)
     await this.persistence.adapter.applyCommittedTx(this.collectionId, tx)
