Prioritize foreground reads during background sync

.changeset/shared-worker-priority-scheduler.md

@@ -0,0 +1,7 @@
+---
+'@treecrdt/interface': patch
+'@treecrdt/wa-sqlite': patch
+'@treecrdt/sync-sqlite': patch
+---
+
+Prioritize foreground shared-worker RPCs ahead of background sync appends between inbound sync batches.

packages/sync-protocol/material/sqlite/src/backend.ts

@@ -1,5 +1,6 @@
 import type { Operation } from '@treecrdt/interface';
 import { bytesToHex, hexToBytes } from '@treecrdt/interface/ids';
+import type { WriteOptions } from '@treecrdt/interface/engine';
 import type { SqliteRunner } from '@treecrdt/interface/sqlite';
 import { deriveOpRefV0 } from '@treecrdt/sync-protocol';
 import type { Filter, OpRef, SyncBackend } from '@treecrdt/sync-protocol';
@@ -18,7 +19,7 @@ export type TreecrdtSyncBackendClient = {
   ops: {
     all?: () => Promise<Operation[]>;
     get: (opRefs: OpRef[]) => Promise<Operation[]>;
-    appendMany: (ops: Operation[]) => Promise<unknown>;
+    appendMany: (ops: Operation[], opts?: WriteOptions) => Promise<unknown>;
   };
 };
 
@@ -169,7 +170,7 @@ export function createTreecrdtSyncBackendFromClient(
 
     applyOps: async (ops) => {
       if (ops.length === 0) return;
-      await client.ops.appendMany(ops);
+      await client.ops.appendMany(ops, { priority: 'background' });
     },
 
     ...(pending

packages/treecrdt-sync/tests/in-memory-sync.test.ts

@@ -112,9 +112,11 @@ test('syncOnce defaults split inbound applies into modest batches', async () =>
   const { client: aClient, getOps: getAllA } = createInMemoryTestClient(docId, []);
   const { client: bClient } = createInMemoryTestClient(docId, remoteOps);
   const appendBatchSizes: number[] = [];
+  const appendPriorities: Array<string | undefined> = [];
   const appendMany = aClient.ops.appendMany.bind(aClient.ops);
   aClient.ops.appendMany = async (ops, writeOpts) => {
     appendBatchSizes.push(ops.length);
+    appendPriorities.push(writeOpts?.priority);
     return appendMany(ops, writeOpts);
   };
 
@@ -142,6 +144,7 @@ test('syncOnce defaults split inbound applies into modest batches', async () =>
   expect(appendBatchSizes.length).toBeGreaterThan(1);
   expect(Math.max(...appendBatchSizes)).toBeLessThanOrEqual(DEFAULT_MAX_OPS_PER_BATCH);
   expect(appendBatchSizes.reduce((sum, size) => sum + size, 0)).toBe(totalOps);
+  expect(new Set(appendPriorities)).toEqual(new Set(['background']));
 });
 
 test('syncOnce pulls insert, move, payload, and delete operations', async () => {

packages/treecrdt-ts/src/engine.ts

@@ -126,6 +126,11 @@ export function addMaterializationWriteId(
 
 export type WriteOptions = {
   writeId?: string;
+  /**
+   * Background writes may yield to foreground operations in runtimes that
+   * serialize requests through a worker.
+   */
+  priority?: 'foreground' | 'background';
 };
 
 export type LocalWriteAuthSession = {

packages/treecrdt-wa-sqlite/e2e/src/runtime-bench.ts

@@ -214,7 +214,7 @@ export async function runRuntimeMixedWriteBench(
 
         const batchStart = performance.now();
         try {
-          await client.ops.appendMany(batch);
+          await client.ops.appendMany(batch, { priority: 'background' });
         } catch (error) {
           throw new Error(`runtime mixed benchmark remote ingest failed: ${errorMessage(error)}`);
         }
@@ -244,7 +244,7 @@ export async function runRuntimeMixedWriteBench(
 
           const batchStart = performance.now();
           try {
-            await client.ops.appendMany(ops);
+            await client.ops.appendMany(ops, { priority: 'background' });
           } catch (error) {
             remoteApplied.reject(error);
             throw error;

packages/treecrdt-wa-sqlite/e2e/src/sync.ts

@@ -38,6 +38,7 @@ export type SyncBenchResult = {
 type StorageKind = 'browser-memory' | 'browser-opfs-coop-sync';
 
 const memoryStorage = { type: 'memory' } as const;
+const rootNode = '0'.repeat(32);
 
 function storageOptionsForMode(mode: 'memory' | 'opfs', filename?: string) {
   return mode === 'opfs'
@@ -62,6 +63,25 @@ function hasOp(ops: Operation[], replica: Uint8Array, counter: number): boolean
   );
 }
 
+function makeRootInsertOps(opts: {
+  replica: Uint8Array;
+  startCounter: number;
+  startLamport: number;
+  startNodeInt: number;
+  startPosition: number;
+  count: number;
+}): Operation[] {
+  return Array.from({ length: opts.count }, (_, index) => {
+    const offset = index;
+    return makeOp(opts.replica, opts.startCounter + offset, opts.startLamport + offset, {
+      type: 'insert',
+      parent: rootNode,
+      node: nodeIdFromInt(opts.startNodeInt + offset),
+      orderKey: orderKeyFromPosition(opts.startPosition + offset),
+    });
+  });
+}
+
 function makeBackend(
   client: TreecrdtClient,
   docId: string,
@@ -394,6 +414,70 @@ export async function runTreecrdtAuthLocalWriteE2E(): Promise<{
   return { ok: true, direct, worker };
 }
 
+export async function runTreecrdtSharedOpfsInterleavedLocalWriteE2E(): Promise<{
+  ok: true;
+  childCount: number;
+  localExists: boolean;
+  opCount: number;
+}> {
+  const docId = `e2e-shared-opfs-interleaved-local-${crypto.randomUUID()}`;
+  const filename = `/interleave-${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}.db`;
+  const client = await createTreecrdtClient({
+    storage: { type: 'opfs', filename, fallback: 'throw' },
+    runtime: { type: 'shared-worker' },
+    docId,
+  });
+  const remoteReplica = replicaFromLabel('interleaved-remote');
+  const localReplica = replicaFromLabel('interleaved-local');
+  const firstCount = 5_500;
+  const secondCount = 500;
+  const localNode = nodeIdFromInt(1_000_000);
+
+  try {
+    await client.ops.appendMany(
+      makeRootInsertOps({
+        replica: remoteReplica,
+        startCounter: 1,
+        startLamport: 1,
+        startNodeInt: 1,
+        startPosition: 0,
+        count: firstCount,
+      }),
+    );
+    await client.local.insert(localReplica, rootNode, localNode, { type: 'last' }, null);
+    await client.ops.appendMany(
+      makeRootInsertOps({
+        replica: remoteReplica,
+        startCounter: firstCount + 1,
+        startLamport: firstCount + 1,
+        startNodeInt: firstCount + 1,
+        startPosition: firstCount,
+        count: secondCount,
+      }),
+    );
+
+    const [children, localExists, ops] = await Promise.all([
+      client.tree.children(rootNode),
+      client.tree.exists(localNode),
+      client.ops.all(),
+    ]);
+    const expectedCount = firstCount + secondCount + 1;
+    if (children.length !== expectedCount) {
+      throw new Error(
+        `interleaved-local-write: expected ${expectedCount} root children, got ${children.length}`,
+      );
+    }
+    if (!localExists) throw new Error('interleaved-local-write: local node is not materialized');
+    if (ops.length !== expectedCount) {
+      throw new Error(`interleaved-local-write: expected ${expectedCount} ops, got ${ops.length}`);
+    }
+
+    return { ok: true, childCount: children.length, localExists, opCount: ops.length };
+  } finally {
+    await client.close();
+  }
+}
+
 export async function runTreecrdtSyncLargeFanoutE2E(): Promise<{ ok: true }> {
   await runLargeFanoutAllE2e();
   return { ok: true };
@@ -853,6 +937,7 @@ declare global {
     runTreecrdtSyncE2E?: typeof runTreecrdtSyncE2E;
     runTreecrdtMaterializationEventE2E?: typeof runTreecrdtMaterializationEventE2E;
     runTreecrdtAuthLocalWriteE2E?: typeof runTreecrdtAuthLocalWriteE2E;
+    runTreecrdtSharedOpfsInterleavedLocalWriteE2E?: typeof runTreecrdtSharedOpfsInterleavedLocalWriteE2E;
     runTreecrdtSyncLargeFanoutE2E?: typeof runTreecrdtSyncLargeFanoutE2E;
     runTreecrdtSyncSubscribeE2E?: typeof runTreecrdtSyncSubscribeE2E;
     runTreecrdtSyncBench?: typeof runTreecrdtSyncBench;
@@ -867,6 +952,8 @@ if (typeof window !== 'undefined') {
   window.runTreecrdtSyncE2E = runTreecrdtSyncE2E;
   window.runTreecrdtMaterializationEventE2E = runTreecrdtMaterializationEventE2E;
   window.runTreecrdtAuthLocalWriteE2E = runTreecrdtAuthLocalWriteE2E;
+  window.runTreecrdtSharedOpfsInterleavedLocalWriteE2E =
+    runTreecrdtSharedOpfsInterleavedLocalWriteE2E;
   window.runTreecrdtSyncLargeFanoutE2E = runTreecrdtSyncLargeFanoutE2E;
   window.runTreecrdtSyncSubscribeE2E = runTreecrdtSyncSubscribeE2E;
   window.runTreecrdtSyncBench = runTreecrdtSyncBench;

packages/treecrdt-wa-sqlite/e2e/tests/sync.spec.ts

@@ -55,3 +55,22 @@ test('auth-aware local writes roll back and defer materialization events e2e', a
     expect(mode.success.authorizedBeforeEvent).toBe(true);
   }
 });
+
+test('shared OPFS applies remote batches around a local write e2e', async ({ page }) => {
+  test.setTimeout(120_000);
+  await page.goto('/');
+  await page.waitForFunction(
+    () => typeof window.runTreecrdtSharedOpfsInterleavedLocalWriteE2E === 'function',
+  );
+
+  const result = await page.evaluate(async () => {
+    const runner = window.runTreecrdtSharedOpfsInterleavedLocalWriteE2E;
+    if (!runner) throw new Error('runTreecrdtSharedOpfsInterleavedLocalWriteE2E not available');
+    return await runner();
+  });
+
+  expect(result.ok).toBe(true);
+  expect(result.childCount).toBe(6001);
+  expect(result.localExists).toBe(true);
+  expect(result.opCount).toBe(6001);
+});