Yield between inbound sync apply batches

Author: marcus-pousette

.changeset/inbound-sync-batch-yield.md

@@ -0,0 +1,5 @@
+---
+'@treecrdt/sync-protocol': patch
+---
+
+Yield between queued inbound sync apply batches so UI work can run between remote opsBatch applies.

packages/sync-protocol/protocol/src/sync.ts

@@ -1675,7 +1675,10 @@ export class SyncPeer<Op> {
       .catch(() => {
         // A prior batch failure should not permanently poison the queue.
       })
-      .then(() => this.onOpsBatch(transport, batch));
+      .then(async () => {
+        await this.onOpsBatch(transport, batch);
+        if (batch.ops.length > 0 && !batch.done) await yieldToMacrotask();
+      });
     this.opsBatchQueues.set(batch.filterId, current);
     try {
       await current;

packages/sync-protocol/protocol/tests/smoke.test.ts

@@ -30,6 +30,16 @@ async function tick(): Promise<void> {
   await new Promise<void>((resolve) => setTimeout(resolve, 0));
 }
 
+function deferredPromise<T = void>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
 function orderKeyFromPosition(position: number): Uint8Array {
   if (!Number.isInteger(position) || position < 0) throw new Error(`invalid position: ${position}`);
   const n = position + 1;
@@ -443,6 +453,70 @@ test('syncOnce waits for responder to apply uploaded ops before resolving', asyn
   expect(b.hasOp(replicaHex.a, 3)).toBe(true);
 });
 
+test('incoming opsBatch queue yields to macrotasks between applies', async () => {
+  const docId = 'doc-sync-inbound-apply-yield';
+  const root = '0'.repeat(32);
+  const firstApplyCanFinish = deferredPromise();
+  let applyCalls = 0;
+  let macrotaskRan = false;
+  let secondApplyStartedAfterMacrotask = false;
+
+  class ProbeBackend extends MemoryBackend {
+    override async applyOps(ops: Operation[]): Promise<void> {
+      applyCalls += 1;
+      if (applyCalls === 1) {
+        await firstApplyCanFinish.promise;
+      } else if (applyCalls === 2) {
+        secondApplyStartedAfterMacrotask = macrotaskRan;
+      }
+      await super.applyOps(ops);
+    }
+  }
+
+  const a = new MemoryBackend(docId);
+  const b = new ProbeBackend(docId);
+  const ops = [1, 2, 3].map((counter, index) =>
+    makeOp(replicas.a, counter, counter, {
+      type: 'insert',
+      parent: root,
+      node: nodeIdFromInt(counter),
+      orderKey: orderKeyFromPosition(index),
+    }),
+  );
+
+  const [wa, wb] = createMacrotaskDuplex<Uint8Array>();
+  const ta = wrapDuplexTransportWithCodec(wa, treecrdtSyncV0ProtobufCodec);
+  const tb = wrapDuplexTransportWithCodec(wb, treecrdtSyncV0ProtobufCodec);
+  const pa = new SyncPeer(a);
+  const pb = new SyncPeer(b);
+  pa.attach(ta);
+  pb.attach(tb);
+
+  const pushDone = pa.pushOps(ta, ops, {
+    filterId: 'apply-yield-probe',
+    maxOpsPerBatch: 1,
+  });
+
+  await waitUntil(() => applyCalls === 1, {
+    message: 'expected first inbound apply to start',
+  });
+  await pushDone;
+
+  setImmediate(() => {
+    macrotaskRan = true;
+  });
+  firstApplyCanFinish.resolve();
+
+  await waitUntil(() => applyCalls >= 2, {
+    message: 'expected second inbound apply to start',
+  });
+  expect(secondApplyStartedAfterMacrotask).toBe(true);
+
+  await waitUntil(() => b.hasOp(replicaHex.a, 3), {
+    message: 'expected all pushed ops to apply',
+  });
+});
+
 test('pushOps uploads direct ops without reconcile roundtrips', async () => {
   const docId = 'doc-push-direct';
   const root = '0'.repeat(32);

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

@@ -1,14 +1,23 @@
 import { createTreecrdtClient, type TreecrdtClient } from '@treecrdt/wa-sqlite';
 import { makeOp, nodeIdFromInt, type BenchmarkResult } from '@treecrdt/benchmark';
 import type { Operation } from '@treecrdt/interface';
+import { bytesToHex } from '@treecrdt/interface/ids';
+import { SyncPeer, type SyncBackend } from '@treecrdt/sync-protocol';
+import { treecrdtSyncV0ProtobufCodec } from '@treecrdt/sync-protocol/protobuf';
+import {
+  createInMemoryDuplex,
+  wrapDuplexTransportWithCodec,
+} from '@treecrdt/sync-protocol/transport';
 import { replicaFromLabel } from './op-helpers.js';
 
 type RuntimeChoice = 'direct' | 'dedicated-worker' | 'shared-worker';
 type StorageChoice = 'memory' | 'opfs';
+type RemoteIngestChoice = 'append-many' | 'sync-peer';
 
 type RuntimeMixedWriteBenchOptions = {
   runtime: RuntimeChoice;
   storage?: StorageChoice;
+  remoteIngest?: RemoteIngestChoice;
   docId?: string;
   filename?: string;
   prefillOps?: number;
@@ -25,6 +34,7 @@ type RuntimeMixedWriteBenchResult = BenchmarkResult & {
   extra: {
     runtime: RuntimeChoice;
     storage: StorageChoice;
+    remoteIngest: RemoteIngestChoice;
     prefillOps: number;
     remoteOps: number;
     remoteBatchSize: number;
@@ -46,6 +56,7 @@ type RuntimeMixedWriteBenchResult = BenchmarkResult & {
     readP50Ms: number;
     readP95Ms: number;
     readMaxMs: number;
+    interBatchReadMs: number | null;
     localWriteIntervalMs: number;
     readIntervalMs: number;
     yieldBetweenRemoteBatchesMs: number;
@@ -96,6 +107,16 @@ function summarizeDurations(durationsMs: number[]) {
   };
 }
 
+function deferredPromise<T = void>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  let reject!: (reason?: unknown) => void;
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+  return { promise, resolve, reject };
+}
+
 function makeInsertOps(opts: {
   replica: Uint8Array;
   count: number;
@@ -122,6 +143,7 @@ export async function runRuntimeMixedWriteBench(
 ): Promise<RuntimeMixedWriteBenchResult> {
   const runtime = opts.runtime;
   const storage = opts.storage ?? 'opfs';
+  const remoteIngest = opts.remoteIngest ?? 'sync-peer';
   const prefillOps = opts.prefillOps ?? 0;
   const remoteOps = opts.remoteOps ?? 2_000;
   const remoteBatchSize = opts.remoteBatchSize ?? 500;
@@ -164,10 +186,12 @@ export async function runRuntimeMixedWriteBench(
     const remoteBatchDurationsMs: number[] = [];
     const localWriteDurationsMs: number[] = [];
     const readDurationsMs: number[] = [];
+    let interBatchReadMs: number | null = null;
+    let interBatchReadDone: Promise<void> | null = null;
     let finalChildCount = 0;
     const start = performance.now();
 
-    const runRemoteIngest = async () => {
+    const runAppendManyRemoteIngest = async () => {
       for (let batchIndex = 0; batchIndex < remoteBatchCount; batchIndex += 1) {
         const batchStartCounter = batchIndex * remoteBatchSize + 1;
         const remaining = remoteOps - batchIndex * remoteBatchSize;
@@ -195,6 +219,90 @@ export async function runRuntimeMixedWriteBench(
       }
     };
 
+    const runSyncPeerRemoteIngest = async () => {
+      if (remoteOps === 0) return;
+
+      const remoteApplied = deferredPromise();
+      let appliedRemoteOps = 0;
+      const receiverBackend: SyncBackend<Operation> = {
+        docId,
+        maxLamport: async () => BigInt(await client.meta.headLamport()),
+        listOpRefs: async (filter) => {
+          if ('all' in filter) return client.opRefs.all();
+          return client.opRefs.children(bytesToHex(filter.children.parent));
+        },
+        getOpsByOpRefs: async (opRefs) => client.ops.get(opRefs),
+        applyOps: async (ops) => {
+          if (ops.length === 0) return;
+
+          const batchStart = performance.now();
+          try {
+            await client.ops.appendMany(ops);
+          } catch (error) {
+            remoteApplied.reject(error);
+            throw error;
+          }
+
+          remoteBatchDurationsMs.push(performance.now() - batchStart);
+          const nextAppliedRemoteOps = appliedRemoteOps + ops.length;
+          if (appliedRemoteOps === 0 && nextAppliedRemoteOps < remoteOps) {
+            interBatchReadDone = new Promise<void>((resolve, reject) => {
+              setTimeout(() => {
+                const readStart = performance.now();
+                client.tree.children(rootNode()).then(() => {
+                  interBatchReadMs = performance.now() - readStart;
+                  resolve();
+                }, reject);
+              }, 0);
+            });
+          }
+          appliedRemoteOps = nextAppliedRemoteOps;
+          if (appliedRemoteOps >= remoteOps) remoteApplied.resolve();
+        },
+      };
+      const senderBackend: SyncBackend<Operation> = {
+        docId,
+        maxLamport: async () => 0n,
+        listOpRefs: async () => [],
+        getOpsByOpRefs: async () => [],
+        applyOps: async () => {},
+      };
+      const [wireA, wireB] = createInMemoryDuplex<Uint8Array>();
+      const transportA = wrapDuplexTransportWithCodec(wireA, treecrdtSyncV0ProtobufCodec);
+      const transportB = wrapDuplexTransportWithCodec(wireB, treecrdtSyncV0ProtobufCodec);
+      const senderPeer = new SyncPeer(senderBackend, { maxOpsPerBatch: remoteBatchSize });
+      const receiverPeer = new SyncPeer(receiverBackend, { maxOpsPerBatch: remoteBatchSize });
+      const detachSender = senderPeer.attach(transportA);
+      const detachReceiver = receiverPeer.attach(transportB);
+
+      try {
+        const ops = makeInsertOps({
+          replica: remoteReplica,
+          count: remoteOps,
+          startCounter: 1,
+          startLamport: prefillOps + 1,
+          startNodeInt: 100_000,
+          startOrderOffset: 100_000,
+        });
+        await senderPeer.pushOps(transportA, ops, {
+          filterId: `runtime-sync-peer-${crypto.randomUUID()}`,
+          maxOpsPerBatch: remoteBatchSize,
+        });
+        await remoteApplied.promise;
+        if (interBatchReadDone) await interBatchReadDone;
+      } catch (error) {
+        throw new Error(
+          `runtime mixed benchmark remote sync ingest failed: ${errorMessage(error)}`,
+        );
+      } finally {
+        detachSender();
+        detachReceiver();
+      }
+    };
+
+    const runRemoteIngest =
+      remoteIngest === 'sync-peer' ? runSyncPeerRemoteIngest : runAppendManyRemoteIngest;
+
     const runLocalWrites = async () => {
       for (let i = 0; i < localWrites; i += 1) {
         if (i > 0 && localWriteIntervalMs > 0) await sleep(localWriteIntervalMs);
@@ -242,13 +350,14 @@ export async function runRuntimeMixedWriteBench(
     const readSummary = summarizeDurations(readDurationsMs);
 
     return {
-      name: `runtime-mixed-sync-ingest-local-writes-${storage}-${runtime}-prefill-${prefillOps}`,
+      name: `runtime-mixed-${remoteIngest}-ingest-local-writes-${storage}-${runtime}-prefill-${prefillOps}`,
       totalOps,
       durationMs,
       opsPerSec: durationMs > 0 ? (totalOps / durationMs) * 1000 : Infinity,
       extra: {
         runtime,
         storage,
+        remoteIngest,
         prefillOps,
         remoteOps,
         remoteBatchSize,
@@ -270,6 +379,7 @@ export async function runRuntimeMixedWriteBench(
         readP50Ms: readSummary.p50Ms,
         readP95Ms: readSummary.p95Ms,
         readMaxMs: readSummary.maxMs,
+        interBatchReadMs,
         localWriteIntervalMs,
         readIntervalMs,
         yieldBetweenRemoteBatchesMs,

packages/treecrdt-wa-sqlite/e2e/tests/bench-runtime.spec.ts

@@ -5,6 +5,7 @@ import type { BenchmarkResult } from '@treecrdt/benchmark';
 
 type RuntimeChoice = 'direct' | 'dedicated-worker' | 'shared-worker';
 type StorageChoice = 'memory' | 'opfs';
+type RemoteIngestChoice = 'append-many' | 'sync-peer';
 type RuntimeScenario = {
   id: string;
   runtime: RuntimeChoice;
@@ -20,6 +21,7 @@ const defaultScenarios: RuntimeScenario[] = [
 
 const remoteOps = Number(process.env.TREECRDT_RUNTIME_BENCH_REMOTE_OPS ?? 2_000);
 const remoteBatchSize = Number(process.env.TREECRDT_RUNTIME_BENCH_REMOTE_BATCH_SIZE ?? 500);
+const remoteIngest = envRemoteIngest('TREECRDT_RUNTIME_BENCH_REMOTE_INGEST', 'sync-peer');
 const localWrites = Number(process.env.TREECRDT_RUNTIME_BENCH_LOCAL_WRITES ?? 20);
 const readSamples = Number(process.env.TREECRDT_RUNTIME_BENCH_READ_SAMPLES ?? 20);
 const readIntervalMs = Number(process.env.TREECRDT_RUNTIME_BENCH_READ_INTERVAL_MS ?? 0);
@@ -40,6 +42,12 @@ function envNumberList(name: string, fallback: number[]): number[] {
   return values.length > 0 ? values : fallback;
 }
 
+function envRemoteIngest(name: string, fallback: RemoteIngestChoice): RemoteIngestChoice {
+  const raw = process.env[name];
+  if (raw === 'append-many' || raw === 'sync-peer') return raw;
+  return fallback;
+}
+
 function envScenarioList(name: string, fallback: RuntimeScenario[]): RuntimeScenario[] {
   const raw = process.env[name];
   if (!raw) return fallback;
@@ -83,6 +91,7 @@ test('wa-sqlite runtime/storage mixed sync-ingest/local-write benchmarks', async
           storage: scenario.storage,
           docId: `runtime-mixed-${scenario.id}-${prefillOps}-${suffix}`,
           filename: `/runtime-mixed-${scenario.id}-${prefillOps}-${suffix}.db`,
+          remoteIngest,
           prefillOps,
           remoteOps,
           remoteBatchSize,
@@ -97,6 +106,7 @@ test('wa-sqlite runtime/storage mixed sync-ingest/local-write benchmarks', async
       expect(result.totalOps).toBe(remoteOps + localWrites);
       expect(result.extra.runtime).toBe(scenario.runtime);
       expect(result.extra.storage).toBe(scenario.storage);
+      expect(result.extra.remoteIngest).toBe(remoteIngest);
       expect(result.extra.finalChildCount).toBe(prefillOps + remoteOps + localWrites);
       expect(result.extra.remoteBatchDurationsMs.length).toBe(
         Math.ceil(remoteOps / remoteBatchSize),