perf(wa-sqlite): prioritize reads during background sync

Author: Marcus Pousette

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

@@ -0,0 +1,8 @@
+---
+'@treecrdt/interface': patch
+'@treecrdt/wa-sqlite': patch
+'@treecrdt/sync-sqlite': patch
+---
+
+Let dedicated-worker clients prioritize engine reads between background sync append batches while
+preserving normal call ordering and bounding foreground starvation.

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;
+  /**
+   * Scheduling hint for clients that serialize requests through a dedicated worker.
+   * Background writes may yield to foreground operations.
+   */
+  priority?: 'background';
 };
 
 export type LocalWriteAuthSession = {

packages/treecrdt-wa-sqlite/src/client.ts

@@ -39,6 +39,7 @@ import {
   type RpcResponse,
   type RpcResult,
 } from './rpc.js';
+import { createPrioritizedRpcCall } from './rpc-scheduler.js';
 import { openTreecrdtDb, type OpenTreecrdtDbOptions, type OpenTreecrdtDbResult } from './open.js';
 import type {
   ClientMaterializationDispatcher,
@@ -52,6 +53,7 @@ import type {
   NormalizedRuntimeOptions,
   NormalizedStorageOptions,
   RpcCall,
+  RpcCallOptions,
   RuntimeMode,
   SharedWorkerFactory,
   StorageMode,
@@ -324,15 +326,8 @@ async function createWorkerClient(opts: {
   >();
   let terminalError: Error | null = null;
   let closed = false;
-  let callQueue: Promise<void> = Promise.resolve();
 
   const closedError = new Error(CLIENT_CLOSED_ERROR);
-  const settleQueue = <T>(promise: Promise<T>): Promise<void> =>
-    promise.then(
-      () => undefined,
-      () => undefined,
-    );
-
   const callRaw = <M extends RpcMethod>(method: M, params: RpcParams<M>): Promise<RpcResult<M>> => {
     if (closed) return Promise.reject(closedError);
     const id = nextId++;
@@ -342,11 +337,7 @@ async function createWorkerClient(opts: {
       worker.postMessage({ id, method, params } satisfies RpcRequest<M>);
     });
   };
-  const call = <M extends RpcMethod>(method: M, params: RpcParams<M>): Promise<RpcResult<M>> => {
-    const run = callQueue.then(() => callRaw(method, params));
-    callQueue = settleQueue(run);
-    return run;
-  };
+  const call = createPrioritizedRpcCall(callRaw);
 
   const onMessage = (ev: MessageEvent<RpcResponse | RpcPushMessage>) => {
     const data = ev.data;
@@ -744,7 +735,7 @@ export async function buildDirectClient(
       throw wrapError(method, err);
     }
   };
-  const call: RpcCall = (method, params) => {
+  const call: RpcCall = (method, params, _options) => {
     const run = callQueue.then(() => runDirectCall(method, params));
     callQueue = settleQueue(run);
     return run;
@@ -775,7 +766,8 @@ export async function buildDirectClient(
 
 // --- helpers
 
-function makeTreecrdtClientFromCall(opts: {
+/** @internal Exported for deterministic client scheduling tests. */
+export function makeTreecrdtClientFromCall(opts: {
   mode: ClientMode;
   runtime: RuntimeMode;
   storage: StorageMode;
@@ -806,18 +798,20 @@ function makeTreecrdtClientFromCall(opts: {
     localWriters.set(key, next);
     return next;
   };
+  const foregroundCall: RpcCall = (method, params) =>
+    call(method, params, { priority: 'foreground' });
 
   const opsSinceImpl = async (lamport: number, root?: string) => {
-    const rows = await call('opsSince', [lamport, root]);
+    const rows = await foregroundCall('opsSince', [lamport, root]);
     return decodeSqliteOps(rows);
   };
-  const opRefsAllImpl = async () => decodeSqliteOpRefs(await call('opRefsAll', []));
+  const opRefsAllImpl = async () => decodeSqliteOpRefs(await foregroundCall('opRefsAll', []));
   const opRefsChildrenImpl = async (parent: string) =>
-    decodeSqliteOpRefs(await call('opRefsChildren', [parent]));
+    decodeSqliteOpRefs(await foregroundCall('opRefsChildren', [parent]));
   const opsByOpRefsImpl = async (opRefs: Uint8Array[]) =>
-    decodeSqliteOps(await call('opsByOpRefs', [opRefs.map((r) => Array.from(r))]));
+    decodeSqliteOps(await foregroundCall('opsByOpRefs', [opRefs.map((r) => Array.from(r))]));
   const treeChildrenImpl = async (parent: string) =>
-    decodeSqliteNodeIds(await call('treeChildren', [parent]));
+    decodeSqliteNodeIds(await foregroundCall('treeChildren', [parent]));
   const treeChildrenPageImpl = async (
     parent: string,
     cursor: { orderKey: Uint8Array; node: Uint8Array } | null,
@@ -826,34 +820,42 @@ function makeTreecrdtClientFromCall(opts: {
     const rpcCursor = cursor
       ? { orderKey: Array.from(cursor.orderKey), node: Array.from(cursor.node) }
       : null;
-    return decodeSqliteTreeChildRows(await call('treeChildrenPage', [parent, rpcCursor, limit]));
+    return decodeSqliteTreeChildRows(
+      await foregroundCall('treeChildrenPage', [parent, rpcCursor, limit]),
+    );
   };
-  const treeDumpImpl = async () => decodeSqliteTreeRows(await call('treeDump', []));
-  const treeNodeCountImpl = async () => Number(await call('treeNodeCount', []));
+  const treeDumpImpl = async () => decodeSqliteTreeRows(await foregroundCall('treeDump', []));
+  const treeNodeCountImpl = async () => Number(await foregroundCall('treeNodeCount', []));
   const treeParentImpl = async (node: string) => {
-    const result = await call('treeParent', [node]);
+    const result = await foregroundCall('treeParent', [node]);
     if (result === null) return null;
     return nodeIdFromBytes16(toRpcBytes(result));
   };
-  const treeExistsImpl = async (node: string) => Boolean(await call('treeExists', [node]));
+  const treeExistsImpl = async (node: string) =>
+    Boolean(await foregroundCall('treeExists', [node]));
   const treeGetPayloadImpl = async (node: string) => {
-    const result = await call('treePayload', [node]);
+    const result = await foregroundCall('treePayload', [node]);
     return result === null ? null : toRpcBytes(result);
   };
-  const headLamportImpl = async () => Number(await call('headLamport', []));
+  const headLamportImpl = async () => Number(await foregroundCall('headLamport', []));
   const replicaMaxCounterImpl = async (replica: Operation['meta']['id']['replica']) =>
-    Number(await call('replicaMaxCounter', [Array.from(encodeReplica(replica))]));
+    Number(await foregroundCall('replicaMaxCounter', [Array.from(encodeReplica(replica))]));
   const appendManyImpl = async (operations: Operation[], writeOpts?: WriteOptions) => {
+    const callOptions = writeOpts?.priority ? { priority: writeOpts.priority } : undefined;
     if (operations.length <= APPEND_MANY_RPC_CHUNK_SIZE) {
-      const outcome = await call('appendMany', [operations]);
+      const outcome = await call('appendMany', [operations], callOptions);
       materialized.emitOutcome(outcome, writeOpts?.writeId);
       return;
     }
 
     const outcomes: MaterializationOutcome[] = [];
     for (let start = 0; start < operations.length; start += APPEND_MANY_RPC_CHUNK_SIZE) {
       outcomes.push(
-        await call('appendMany', [operations.slice(start, start + APPEND_MANY_RPC_CHUNK_SIZE)]),
+        await call(
+          'appendMany',
+          [operations.slice(start, start + APPEND_MANY_RPC_CHUNK_SIZE)],
+          callOptions,
+        ),
       );
     }
     materialized.emitOutcome(mergeMaterializationOutcomes(outcomes), writeOpts?.writeId);
@@ -928,7 +930,8 @@ function makeTreecrdtClientFromCall(opts: {
     runner,
     ops: {
       append: async (op, writeOpts?: WriteOptions) => {
-        const outcome = await call('append', [op]);
+        const callOptions = writeOpts?.priority ? { priority: writeOpts.priority } : undefined;
+        const outcome = await call('append', [op], callOptions);
         materialized.emitOutcome(outcome, writeOpts?.writeId);
       },
       appendMany: appendManyImpl,

packages/treecrdt-wa-sqlite/src/rpc-scheduler.ts

@@ -0,0 +1,70 @@
+import type { RpcCall, RpcCallOptions } from './types.js';
+
+export type RpcSchedulePriority = NonNullable<RpcCallOptions['priority']> | 'normal';
+
+type ScheduledJob = {
+  priority: RpcSchedulePriority;
+  run: () => Promise<unknown>;
+  resolve: (value: unknown) => void;
+  reject: (reason: unknown) => void;
+};
+
+// Bound foreground bypasses so a steady read stream cannot starve sync forever.
+const MAX_FOREGROUND_BURST = 8;
+
+/** Serializes RPC work while allowing reads to bypass only explicitly-background work. */
+export function createRpcScheduler() {
+  const queue: ScheduledJob[] = [];
+  let running = false;
+  let foregroundBurst = 0;
+
+  const nextJobIndex = (): number => {
+    // A normal call is an ordering barrier: a later read must not jump an earlier local write.
+    const normalBarrier = queue.findIndex((job) => job.priority === 'normal');
+    const foreground = queue.findIndex(
+      (job, index) =>
+        job.priority === 'foreground' && (normalBarrier === -1 || index < normalBarrier),
+    );
+    if (
+      foreground !== -1 &&
+      (foregroundBurst < MAX_FOREGROUND_BURST || queue[0]?.priority === 'foreground')
+    ) {
+      return foreground;
+    }
+    return 0;
+  };
+
+  const drain = () => {
+    if (running || queue.length === 0) return;
+    const [job] = queue.splice(nextJobIndex(), 1);
+    if (!job) return;
+    running = true;
+    foregroundBurst = job.priority === 'foreground' ? foregroundBurst + 1 : 0;
+    void Promise.resolve()
+      .then(job.run)
+      .then(job.resolve, job.reject)
+      .finally(() => {
+        running = false;
+        drain();
+      });
+  };
+
+  return <T>(priority: RpcSchedulePriority, run: () => Promise<T>): Promise<T> =>
+    new Promise<T>((resolve, reject) => {
+      queue.push({
+        priority,
+        run,
+        resolve: resolve as (value: unknown) => void,
+        reject,
+      });
+      drain();
+    });
+}
+
+/** Applies priority scheduling before a dedicated worker request is posted. */
+export function createPrioritizedRpcCall(runRaw: RpcCall): RpcCall {
+  const schedule = createRpcScheduler();
+
+  return (method, params, options) =>
+    schedule(options?.priority ?? 'normal', () => runRaw(method, params));
+}

packages/treecrdt-wa-sqlite/src/types.ts

@@ -71,9 +71,11 @@ export type MessagePortProxy = {
   removeEventListener: (type: 'message' | 'messageerror', fn: (ev: any) => void) => void;
 };
 
+export type RpcCallOptions = { priority?: 'foreground' | 'background' };
 export type RpcCall = <M extends RpcMethod>(
   method: M,
   params: RpcParams<M>,
+  options?: RpcCallOptions,
 ) => Promise<RpcResult<M>>;
 export type SharedWorkerFactory = (options?: WorkerOptions & { name?: string }) => SharedWorker;
 export type CrossTabMaterializationScope = {