diff --git a/.changeset/shared-worker-priority-scheduler.md b/.changeset/shared-worker-priority-scheduler.md new file mode 100644 index 00000000..be3bbff4 --- /dev/null +++ b/.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. diff --git a/packages/sync-protocol/material/sqlite/src/backend.ts b/packages/sync-protocol/material/sqlite/src/backend.ts index e0c25ebc..e5211b18 100644 --- a/packages/sync-protocol/material/sqlite/src/backend.ts +++ b/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; get: (opRefs: OpRef[]) => Promise; - appendMany: (ops: Operation[]) => Promise; + appendMany: (ops: Operation[], opts?: WriteOptions) => Promise; }; }; @@ -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 diff --git a/packages/treecrdt-sync/tests/in-memory-sync.test.ts b/packages/treecrdt-sync/tests/in-memory-sync.test.ts index 94322300..7829b54f 100644 --- a/packages/treecrdt-sync/tests/in-memory-sync.test.ts +++ b/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 = []; 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 () => { diff --git a/packages/treecrdt-ts/src/engine.ts b/packages/treecrdt-ts/src/engine.ts index c6e8a58e..c4cfc881 100644 --- a/packages/treecrdt-ts/src/engine.ts +++ b/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 = { diff --git a/packages/treecrdt-wa-sqlite/src/client.ts b/packages/treecrdt-wa-sqlite/src/client.ts index 57ab3e45..3e01819c 100644 --- a/packages/treecrdt-wa-sqlite/src/client.ts +++ b/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, @@ -324,15 +325,8 @@ async function createWorkerClient(opts: { >(); let terminalError: Error | null = null; let closed = false; - let callQueue: Promise = Promise.resolve(); const closedError = new Error(CLIENT_CLOSED_ERROR); - const settleQueue = (promise: Promise): Promise => - promise.then( - () => undefined, - () => undefined, - ); - const callRaw = (method: M, params: RpcParams): Promise> => { if (closed) return Promise.reject(closedError); const id = nextId++; @@ -342,11 +336,7 @@ async function createWorkerClient(opts: { worker.postMessage({ id, method, params } satisfies RpcRequest); }); }; - const call = (method: M, params: RpcParams): Promise> => { - const run = callQueue.then(() => callRaw(method, params)); - callQueue = settleQueue(run); - return run; - }; + const call = createPrioritizedRpcCall(callRaw); const onMessage = (ev: MessageEvent) => { const data = ev.data; @@ -744,7 +734,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 +765,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 +797,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,26 +819,30 @@ 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; } @@ -853,7 +850,11 @@ function makeTreecrdtClientFromCall(opts: { 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 +929,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, diff --git a/packages/treecrdt-wa-sqlite/src/rpc-scheduler.ts b/packages/treecrdt-wa-sqlite/src/rpc-scheduler.ts new file mode 100644 index 00000000..ceaeed42 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/src/rpc-scheduler.ts @@ -0,0 +1,70 @@ +import type { RpcCall, RpcCallOptions } from './types.js'; + +export type RpcSchedulePriority = NonNullable | 'normal'; + +type ScheduledJob = { + priority: RpcSchedulePriority; + run: () => Promise; + 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 (priority: RpcSchedulePriority, run: () => Promise): Promise => + new Promise((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)); +} diff --git a/packages/treecrdt-wa-sqlite/src/types.ts b/packages/treecrdt-wa-sqlite/src/types.ts index deffe270..539e955e 100644 --- a/packages/treecrdt-wa-sqlite/src/types.ts +++ b/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 = ( method: M, params: RpcParams, + options?: RpcCallOptions, ) => Promise>; export type SharedWorkerFactory = (options?: WorkerOptions & { name?: string }) => SharedWorker; export type CrossTabMaterializationScope = { diff --git a/packages/treecrdt-wa-sqlite/tests/rpc-scheduler.test.ts b/packages/treecrdt-wa-sqlite/tests/rpc-scheduler.test.ts new file mode 100644 index 00000000..6f5bc370 --- /dev/null +++ b/packages/treecrdt-wa-sqlite/tests/rpc-scheduler.test.ts @@ -0,0 +1,128 @@ +import { createMaterializationDispatcher } from '@treecrdt/interface/engine'; +import { expect, test } from 'vitest'; +import { makeTreecrdtClientFromCall } from '../src/client.js'; +import type { RpcMethod } from '../src/rpc.js'; +import { + createPrioritizedRpcCall, + createRpcScheduler, + type RpcSchedulePriority, +} from '../src/rpc-scheduler.js'; +import type { ClientMaterializationDispatcher, RpcCall } from '../src/types.js'; + +function deferred() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +async function queuedOrder( + jobs: Array<[RpcSchedulePriority, string]>, + initialPriority: RpcSchedulePriority = 'background', +): Promise { + const schedule = createRpcScheduler(); + const gate = deferred(); + const order: string[] = []; + const running = schedule(initialPriority, async () => { + order.push('initial'); + await gate.promise; + }); + await Promise.resolve(); + const queued = jobs.map(([priority, label]) => + schedule(priority, async () => { + order.push(label); + }), + ); + gate.resolve(); + await Promise.all([running, ...queued]); + return order; +} + +test('foreground work can run between background jobs', async () => { + const order = await queuedOrder([ + ['background', 'background'], + ['foreground', 'foreground'], + ]); + expect(order).toEqual(['initial', 'foreground', 'background']); +}); + +test('normal work is an ordering barrier for later foreground work', async () => { + const order = await queuedOrder([ + ['normal', 'normal'], + ['background', 'background'], + ['foreground', 'foreground'], + ]); + expect(order).toEqual(['initial', 'normal', 'foreground', 'background']); +}); + +test('foreground bursts do not starve queued background work', async () => { + const order = await queuedOrder( + [ + ['background', 'background'], + ...Array.from( + { length: 9 }, + (_, index) => ['foreground', `foreground-${index}`] as [RpcSchedulePriority, string], + ), + ], + 'normal', + ); + expect(order.indexOf('background')).toBe(9); +}); + +test('engine read classification composes with background scheduling', async () => { + const gate = deferred(); + const order: RpcMethod[] = []; + let appendCalls = 0; + const outcome = { headSeq: 0, changes: [] }; + const runRaw = (async (method: RpcMethod) => { + order.push(method); + if (method === 'appendMany') { + appendCalls += 1; + if (appendCalls === 1) await gate.promise; + return outcome; + } + if (method === 'treeNodeCount') return 0; + throw new Error(`unexpected test method: ${method}`); + }) as RpcCall; + const dispatcher = createMaterializationDispatcher(); + const materialized = { + ...dispatcher, + enableCrossTab: () => undefined, + emitIncomingEvent: dispatcher.emitEvent, + close: () => undefined, + } satisfies ClientMaterializationDispatcher; + const client = makeTreecrdtClientFromCall({ + mode: 'worker', + runtime: 'dedicated-worker', + storage: 'memory', + docId: 'rpc-priority-test', + call: createPrioritizedRpcCall(runRaw), + materialized, + close: async () => undefined, + drop: async () => undefined, + }); + + const firstBackground = client.ops.appendMany([], { priority: 'background' }); + await Promise.resolve(); + const secondBackground = client.ops.appendMany([], { priority: 'background' }); + const foregroundRead = client.tree.nodeCount(); + gate.resolve(); + + await Promise.all([firstBackground, secondBackground, foregroundRead]); + expect(order).toEqual(['appendMany', 'treeNodeCount', 'appendMany']); +}); + +test.each(['foreground', 'normal', 'background'] satisfies RpcSchedulePriority[])( + 'a rejected %s job does not stall the scheduler', + async (priority) => { + const schedule = createRpcScheduler(); + const rejected = schedule(priority, async () => { + throw new Error('expected failure'); + }); + const next = schedule('normal', async () => 'completed'); + + await expect(rejected).rejects.toThrow('expected failure'); + await expect(next).resolves.toBe('completed'); + }, +);