From 46cf7ac3439d86b67b8aacbd424707462c4ab95a Mon Sep 17 00:00:00 2001 From: Marcus Pousette Date: Sun, 12 Jul 2026 17:00:03 +0200 Subject: [PATCH] Add race-safe outbound sync controller --- .changeset/clean-outbound-sync.md | 9 + packages/sync-protocol/protocol/src/sync.ts | 131 ++++-- .../protocol/tests/smoke.test.ts | 39 ++ packages/treecrdt-sync/README.md | 28 ++ packages/treecrdt-sync/src/controller.ts | 344 +++++++++++++++ packages/treecrdt-sync/src/index.ts | 4 + packages/treecrdt-sync/src/types.ts | 58 +++ .../treecrdt-sync/tests/controller.test.ts | 409 ++++++++++++++++++ 8 files changed, 984 insertions(+), 38 deletions(-) create mode 100644 .changeset/clean-outbound-sync.md create mode 100644 packages/treecrdt-sync/src/controller.ts create mode 100644 packages/treecrdt-sync/tests/controller.test.ts diff --git a/.changeset/clean-outbound-sync.md b/.changeset/clean-outbound-sync.md new file mode 100644 index 00000000..8535934f --- /dev/null +++ b/.changeset/clean-outbound-sync.md @@ -0,0 +1,9 @@ +--- +'@treecrdt/sync': minor +'@treecrdt/sync-protocol': patch +--- + +Add an outbound sync helper for queued local-op upload to remote peer transports. Standard +TreeCRDT operations are deduped by operation id by default; custom op shapes may provide `opKey`. +Timed-out direct pushes are aborted before later chunks can be sent, and their ops remain queued +for retry. diff --git a/packages/sync-protocol/protocol/src/sync.ts b/packages/sync-protocol/protocol/src/sync.ts index a16f581e..c78ee6e7 100644 --- a/packages/sync-protocol/protocol/src/sync.ts +++ b/packages/sync-protocol/protocol/src/sync.ts @@ -151,6 +151,8 @@ export type SyncOnceOptions = { export type SyncPushOptions = { /** Split a direct push into smaller wire batches to avoid giant frames. */ maxOpsPerBatch?: number; + /** Abort an in-flight direct push before it sends any later chunks. */ + signal?: AbortSignal; /** * Reuse an existing opsBatch stream id for a direct push. * @@ -239,6 +241,41 @@ function waitForAbort(signal: AbortSignal): Promise { ); } +function syncAbortReason(signal: AbortSignal): Error { + if (signal.reason instanceof Error) return signal.reason; + const error = new Error(signal.reason === undefined ? 'sync aborted' : String(signal.reason)); + error.name = 'AbortError'; + return error; +} + +function throwIfSyncAborted(signal?: AbortSignal): void { + if (signal?.aborted) throw syncAbortReason(signal); +} + +function awaitSyncStep(step: T | PromiseLike, signal?: AbortSignal): Promise { + const promise = Promise.resolve(step); + if (!signal) return promise; + if (signal.aborted) return Promise.reject(syncAbortReason(signal)); + + return new Promise((resolve, reject) => { + const onAbort = () => { + signal.removeEventListener('abort', onAbort); + reject(syncAbortReason(signal)); + }; + signal.addEventListener('abort', onAbort, { once: true }); + promise.then( + (value) => { + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error) => { + signal.removeEventListener('abort', onAbort); + reject(error); + }, + ); + }); +} + type ResponderSubscription = { subscriptionId: string; filter: Filter; @@ -523,27 +560,35 @@ export class SyncPeer { private async refreshHelloCapabilities( transport: DuplexTransport>, + signal?: AbortSignal, ): Promise { this.assertTransportActive(transport); if (!this.auth?.helloCapabilities) return; - const [maxLamport, advertisedCapabilities] = await Promise.all([ - this.backend.maxLamport(), - this.auth.helloCapabilities({ docId: this.backend.docId }), - ]); + throwIfSyncAborted(signal); + const [maxLamport, advertisedCapabilities] = await awaitSyncStep( + Promise.all([ + this.backend.maxLamport(), + this.auth.helloCapabilities({ docId: this.backend.docId }), + ]), + signal, + ); const capabilities = copyCapabilities(advertisedCapabilities); this.assertTransportActive(transport); const { exchangeId, ack } = this.beginHelloExchange(transport); try { - await transport.send({ - v: 0, - docId: this.backend.docId, - payload: { - case: 'hello', - value: { exchangeId, capabilities, filters: [], maxLamport }, - }, - }); - await ack.promise; + await awaitSyncStep( + transport.send({ + v: 0, + docId: this.backend.docId, + payload: { + case: 'hello', + value: { exchangeId, capabilities, filters: [], maxLamport }, + }, + }), + signal, + ); + await awaitSyncStep(ack.promise, signal); } finally { deleteTransportOwned(this.pendingHelloExchanges, transport, exchangeId); } @@ -1087,18 +1132,23 @@ export class SyncPeer { ): Promise { if (ops.length === 0) return; - await this.refreshHelloCapabilities(transport); + const signal = opts.signal; + const step = (value: T | PromiseLike) => awaitSyncStep(value, signal); + throwIfSyncAborted(signal); + await this.refreshHelloCapabilities(transport, signal); const peerSnapshot = this.peerCapabilitySnapshot(transport); const peerCapabilities = peerSnapshot.capabilities; let outgoingOps: readonly Op[] = ops; if (this.auth?.filterOutgoingOps) { - const allowed = await this.auth.filterOutgoingOps(outgoingOps, { - docId: this.backend.docId, - purpose: 'reconcile', - filter: { all: {} }, - capabilities: peerCapabilities, - }); + const allowed = await step( + this.auth.filterOutgoingOps(outgoingOps, { + docId: this.backend.docId, + purpose: 'reconcile', + filter: { all: {} }, + capabilities: peerCapabilities, + }), + ); assertOutgoingFilterLength(allowed, outgoingOps.length); assertCurrentCapabilityLease(this.isCurrentPeerCapabilitySnapshot(transport, peerSnapshot)); outgoingOps = outgoingOps.filter((_op, index) => allowed[index] === true); @@ -1110,34 +1160,39 @@ export class SyncPeer { const shouldAttachAuth = peerAdvertisedOpAuth(peerCapabilities); for (let start = 0; start < outgoingOps.length; start += batchSize) { + throwIfSyncAborted(signal); const chunk = outgoingOps.slice(start, start + batchSize); const auth = shouldAttachAuth && this.auth?.signOps - ? await this.auth.signOps(chunk, { - docId: this.backend.docId, - purpose: 'reconcile', - filterId: streamId, - }) + ? await step( + this.auth.signOps(chunk, { + docId: this.backend.docId, + purpose: 'reconcile', + filterId: streamId, + }), + ) : undefined; if (auth && auth.length !== chunk.length) { throw new Error(`signOps returned ${auth.length} entries for ${chunk.length} ops`); } assertCurrentCapabilityLease(this.isCurrentPeerCapabilitySnapshot(transport, peerSnapshot)); - await transport.send({ - v: 0, - docId: this.backend.docId, - payload: { - case: 'opsBatch', - value: { - filterId: streamId, - ops: [...chunk], - ...(auth ? { auth } : {}), - done: start + batchSize >= outgoingOps.length, + await step( + transport.send({ + v: 0, + docId: this.backend.docId, + payload: { + case: 'opsBatch', + value: { + filterId: streamId, + ops: [...chunk], + ...(auth ? { auth } : {}), + done: start + batchSize >= outgoingOps.length, + }, }, - }, - }); - await yieldToMacrotask(); + }), + ); + await step(yieldToMacrotask()); } } diff --git a/packages/sync-protocol/protocol/tests/smoke.test.ts b/packages/sync-protocol/protocol/tests/smoke.test.ts index 29ba2942..7b6b1ba5 100644 --- a/packages/sync-protocol/protocol/tests/smoke.test.ts +++ b/packages/sync-protocol/protocol/tests/smoke.test.ts @@ -1642,6 +1642,45 @@ test('a stale failing HelloAck rejects its real filtered session', async () => { } }); +test('pushOps aborts an in-flight chunk without sending later chunks', async () => { + const docId = 'doc-push-abort'; + const root = '0'.repeat(32); + const backend = new MemoryBackend(docId); + const ops = [1, 2].map((counter) => + makeOp(replicas.a, counter, counter, { + type: 'insert', + parent: root, + node: nodeIdFromInt(counter), + orderKey: orderKeyFromPosition(counter - 1), + }), + ); + const sent: SyncMessage[] = []; + let releaseFirstSend!: () => void; + const firstSend = new Promise((resolve) => { + releaseFirstSend = resolve; + }); + const transport: DuplexTransport> = { + async send(message) { + sent.push(message); + if (sent.length === 1) await firstSend; + }, + onMessage: () => () => {}, + }; + const peer = new SyncPeer(backend, { maxOpsPerBatch: 1 }); + const controller = new AbortController(); + + const push = peer.pushOps(transport, ops, { signal: controller.signal }); + await waitUntil(() => sent.length === 1); + controller.abort(new Error('push timed out')); + + await expect(push).rejects.toThrow('push timed out'); + releaseFirstSend(); + await tick(); + + expect(sent).toHaveLength(1); + expect(sent[0]?.payload.case).toBe('opsBatch'); +}); + test('pushOps refreshes replay capabilities before uploading newly authorized ops', async () => { const docId = 'doc-push-capability-refresh'; const root = '0'.repeat(32); diff --git a/packages/treecrdt-sync/README.md b/packages/treecrdt-sync/README.md index 23a449fc..c03918ef 100644 --- a/packages/treecrdt-sync/README.md +++ b/packages/treecrdt-sync/README.md @@ -9,10 +9,38 @@ High-level **client** library for TreeCRDT sync v0. It combines **`@treecrdt/dis - You want a single dependency to **open a websocket** to a sync server and run reconciliation against a SQLite-backed client store. - You are fine with the built-in discovery + WebSocket + protobuf wiring. +## Multi-peer outbound upload + +Use `createOutboundSync` with a `localPeer` when one `SyncPeer` owns several transports, such as +local-tab mesh peers plus a remote websocket server. The app still manages transport discovery, but +outbound sync owns the committed-local-op hook: it attaches remote upload targets, wakes live +subscriptions on the `localPeer`, and queues exact local-op upload with dedupe and offline retry. + +```ts +import { createOutboundSync } from '@treecrdt/sync'; + +const outbound = createOutboundSync({ + localPeer: peer, + isOnline: () => navigator.onLine, +}); + +const detachRemote = outbound.attachTarget('remote:server', websocketTransport); + +const op = await client.local.payload(replica, node, payload); +outbound.queueOps([op]); // live subscription wakeup + remote websocket upload/retry +``` + +Use `addTarget` only when the transport was already attached to the same `localPeer`. `queueOps` +dedupes standard TreeCRDT `Operation` values by `meta.id`. Pass `opKey` only for a custom op shape +or custom coalescing behavior. `pushTimeoutMs` aborts a stalled direct push and leaves its ops +queued for a later flush. + ## When not to - You only need the protocol types and `SyncPeer` (use **`@treecrdt/sync-protocol`**). - You use a custom transport, no discovery, or an in-memory backend (depend on the protocol and/or **`@treecrdt/discovery`** as needed). +- You want exact low-level control over each `syncOnce`, `startLive`, and direct push call; use + `connectTreecrdtWebSocketSync` directly. ## Repo location diff --git a/packages/treecrdt-sync/src/controller.ts b/packages/treecrdt-sync/src/controller.ts new file mode 100644 index 00000000..6e885bb2 --- /dev/null +++ b/packages/treecrdt-sync/src/controller.ts @@ -0,0 +1,344 @@ +import type { Operation } from '@treecrdt/interface'; +import { bytesToHex } from '@treecrdt/interface/ids'; +import type { SyncMessage } from '@treecrdt/sync-protocol'; +import type { DuplexTransport } from '@treecrdt/sync-protocol/transport'; +import type { OutboundSync, OutboundSyncOptions, OutboundSyncStatus } from './types.js'; + +function outboundSyncStatusSnapshot( + targets: ReadonlyMap>>, + pendingOps: readonly Op[], + running: boolean, + scheduled: boolean, +): OutboundSyncStatus { + return { + targetCount: targets.size, + pendingOps: pendingOps.length, + running, + scheduled, + }; +} + +async function pushWithTimeout( + run: (signal: AbortSignal) => Promise, + controller: AbortController, + ms: number | undefined, + message: string, + parentSignal?: AbortSignal, +): Promise { + if (ms !== undefined && (!Number.isFinite(ms) || ms <= 0)) { + throw new Error(`invalid pushTimeoutMs: ${ms}`); + } + + const forwardAbort = () => controller.abort(parentSignal?.reason); + if (parentSignal?.aborted) forwardAbort(); + else parentSignal?.addEventListener('abort', forwardAbort, { once: true }); + + const timer = + ms === undefined ? undefined : setTimeout(() => controller.abort(new Error(message)), ms); + try { + return await run(controller.signal); + } finally { + if (timer !== undefined) clearTimeout(timer); + parentSignal?.removeEventListener('abort', forwardAbort); + } +} + +function defaultOpKey(op: unknown): string | undefined { + const id = (op as { meta?: { id?: { replica?: unknown; counter?: unknown } } })?.meta?.id; + if (!(id?.replica instanceof Uint8Array)) return undefined; + if (typeof id.counter !== 'number') return undefined; + return `${bytesToHex(id.replica)}:${id.counter}`; +} + +/** + * Queue exact committed local writes for a single {@link SyncPeer} that is attached to one or more + * outbound transports. + * + * `queueOps` also notifies the low-level peer about the local update, so apps can report a local + * write once instead of separately waking live subscriptions and queueing remote upload. + * + * Apps that also use local-tab mesh transports should attach those transports directly to the + * low-level peer, but only register outbound upload targets with this controller. + */ +export function createOutboundSync( + options: OutboundSyncOptions, +): OutboundSync { + const targets = new Map>>(); + const pendingOps: Op[] = []; + const pendingOpKeys = new Set(); + let running = false; + let scheduled = false; + let closed = false; + let targetsRevision = 0; + let activePushController: AbortController | undefined; + let activeFlush: Promise | undefined; + let autoFlushQueued = false; + + const emitStatus = () => { + try { + options.onStatus?.(outboundSyncStatusSnapshot(targets, pendingOps, running, scheduled)); + } catch { + // Observers must not interrupt queue bookkeeping. + } + }; + + const reportError = (targetId: string, error: unknown) => { + try { + options.onError?.({ targetId, error }); + } catch { + // Observers must not turn a handled push failure into an unhandled rejection. + } + }; + + const invalidateTargets = () => { + targetsRevision += 1; + activePushController?.abort(new Error('outbound targets changed')); + }; + + const addPendingOps = (ops: readonly Op[]) => { + for (const op of ops) { + const key = options.opKey?.(op) ?? defaultOpKey(op); + if (key !== undefined) { + if (pendingOpKeys.has(key)) continue; + pendingOpKeys.add(key); + } + pendingOps.push(op); + } + }; + + const restorePendingOps = (ops: readonly Op[]) => { + if (ops.length === 0) return; + const existing = pendingOps.splice(0, pendingOps.length); + pendingOpKeys.clear(); + addPendingOps(ops); + addPendingOps(existing); + }; + + const takePendingOps = () => { + const ops = pendingOps.splice(0, pendingOps.length); + pendingOpKeys.clear(); + return ops; + }; + + const pushTimeoutMs = (targetId: string) => + typeof options.pushTimeoutMs === 'function' + ? options.pushTimeoutMs(targetId) + : options.pushTimeoutMs; + + type AttachedTarget = { + transport: DuplexTransport>; + detach: () => void; + registrations: number; + }; + const targetDetaches = new Map(); + + const detachTarget = (targetId: string, expectedTransport?: DuplexTransport>) => { + const attached = targetDetaches.get(targetId); + if (!attached) return; + if (expectedTransport && attached.transport !== expectedTransport) return; + targetDetaches.delete(targetId); + try { + attached.detach(); + } catch { + // ignore + } + }; + + const retainAttachedTarget = (targetId: string, registration: AttachedTarget) => { + registration.registrations += 1; + let released = false; + return () => { + if (released) return; + released = true; + if (targetDetaches.get(targetId) !== registration) return; + registration.registrations -= 1; + if (registration.registrations === 0) { + removeTarget(targetId, registration.transport); + } + }; + }; + + const addTarget = (targetId: string, transport: DuplexTransport>) => { + if (closed) return; + const previous = targets.get(targetId); + const attached = targetDetaches.get(targetId); + if (attached && attached.transport !== transport) detachTarget(targetId); + targets.set(targetId, transport); + if (previous !== transport) invalidateTargets(); + emitStatus(); + if (pendingOps.length > 0) scheduleFlush(); + }; + + const removeTarget = (targetId: string, expectedTransport?: DuplexTransport>) => { + const transport = targets.get(targetId); + if (expectedTransport && transport !== expectedTransport) return; + if (targets.delete(targetId)) invalidateTargets(); + detachTarget(targetId, expectedTransport); + emitStatus(); + }; + + const clearTargets = () => { + if (targets.size > 0) invalidateTargets(); + targets.clear(); + for (const targetId of Array.from(targetDetaches.keys())) detachTarget(targetId); + emitStatus(); + }; + + const queueAutoFlush = () => { + if (closed || autoFlushQueued) return; + autoFlushQueued = true; + queueMicrotask(() => { + autoFlushQueued = false; + if (closed || activeFlush || !scheduled) return; + void controller.flush().catch(() => {}); + }); + }; + + const scheduleFlush = () => { + if (closed) return; + scheduled = true; + emitStatus(); + if (!activeFlush) queueAutoFlush(); + }; + + const runFlush = async () => { + if (closed) return; + if (!scheduled && pendingOps.length > 0) scheduled = true; + if (!scheduled) { + emitStatus(); + return; + } + + running = true; + emitStatus(); + try { + while (scheduled && !closed) { + scheduled = false; + if (options.isOnline && !options.isOnline()) { + emitStatus(); + return; + } + + const uploadTargets = Array.from(targets.entries()); + const uploadTargetsRevision = targetsRevision; + if (uploadTargets.length === 0) { + emitStatus(); + return; + } + + const ops = takePendingOps(); + if (ops.length === 0) { + emitStatus(); + continue; + } + + let failed = false; + for (const [targetId, transport] of uploadTargets) { + const pushController = new AbortController(); + activePushController = pushController; + try { + const pushOptions = options.pushOptions?.(targetId); + await pushWithTimeout( + (signal) => options.localPeer.pushOps(transport, ops, { ...pushOptions, signal }), + pushController, + pushTimeoutMs(targetId), + `outbound push with ${targetId.slice(0, 8)} timed out`, + pushOptions?.signal, + ); + } catch (error) { + if (!closed && targetsRevision === uploadTargetsRevision) { + failed = true; + reportError(targetId, error); + } + } finally { + if (activePushController === pushController) activePushController = undefined; + } + if (targetsRevision !== uploadTargetsRevision) break; + } + + if (closed) return; + + const targetsChanged = targetsRevision !== uploadTargetsRevision; + if (failed || targetsChanged) { + restorePendingOps(ops); + if (targetsChanged) scheduled = true; + emitStatus(); + if (scheduled) continue; + return; + } + + emitStatus(); + } + } finally { + running = false; + emitStatus(); + } + }; + + const flush = (): Promise => { + if (closed) return Promise.resolve(); + if (activeFlush) return activeFlush; + + const barrier = Promise.resolve().then(async () => { + do { + await runFlush(); + } while (!closed && scheduled); + }); + activeFlush = barrier; + const clear = () => { + if (activeFlush === barrier) activeFlush = undefined; + }; + void barrier.then(clear, clear); + return barrier; + }; + + const controller: OutboundSync = { + get status() { + return outboundSyncStatusSnapshot(targets, pendingOps, running, scheduled); + }, + get pendingOpCount() { + return pendingOps.length; + }, + get targetCount() { + return targets.size; + }, + attachTarget: (targetId, transport) => { + if (closed) return () => {}; + const existing = targetDetaches.get(targetId); + if (existing?.transport === transport) { + return retainAttachedTarget(targetId, existing); + } + + const detach = options.localPeer.attach(transport); + if (existing) detachTarget(targetId); + const registration = { transport, detach, registrations: 0 }; + targetDetaches.set(targetId, registration); + addTarget(targetId, transport); + return retainAttachedTarget(targetId, registration); + }, + addTarget, + removeTarget, + clearTargets, + queueOps: (ops) => { + if (closed || ops.length === 0) return; + try { + void options.localPeer.notifyLocalUpdate(ops).catch(() => {}); + } catch { + // Notification is best-effort; exact outbound upload remains queued below. + } + addPendingOps(ops); + if (pendingOps.length > 0) scheduleFlush(); + }, + flush, + close: () => { + closed = true; + scheduled = false; + pendingOps.splice(0, pendingOps.length); + pendingOpKeys.clear(); + clearTargets(); + }, + }; + + emitStatus(); + return controller; +} diff --git a/packages/treecrdt-sync/src/index.ts b/packages/treecrdt-sync/src/index.ts index 5bf9c0b7..949f79d7 100644 --- a/packages/treecrdt-sync/src/index.ts +++ b/packages/treecrdt-sync/src/index.ts @@ -1,10 +1,14 @@ export { connectTreecrdtWebSocketSync } from './connect.js'; +export { createOutboundSync } from './controller.js'; export { createTreecrdtWebSocketSyncFromTransport } from './create-sync-from-transport.js'; export type { ConnectTreecrdtWebSocketSyncOptions, CreateTreecrdtWebSocketSyncFromTransportOptions, MaterializationEvent, MaterializationListener, + OutboundSync, + OutboundSyncOptions, + OutboundSyncStatus, TreecrdtWebSocketSync, TreecrdtWebSocketSyncClient, } from './types.js'; diff --git a/packages/treecrdt-sync/src/types.ts b/packages/treecrdt-sync/src/types.ts index 9d7e752a..6a215184 100644 --- a/packages/treecrdt-sync/src/types.ts +++ b/packages/treecrdt-sync/src/types.ts @@ -8,10 +8,14 @@ import type { TreecrdtSyncBackendClient } from '@treecrdt/sync-sqlite/backend'; import type { Filter, SyncAuth, + SyncMessage, SyncPeerOptions, SyncOnceOptions, + SyncPeer, + SyncPushOptions, SyncSubscribeOptions, } from '@treecrdt/sync-protocol'; +import type { DuplexTransport } from '@treecrdt/sync-protocol/transport'; import type { DiscoveryRouteCache } from '@treecrdt/discovery'; /** @@ -72,4 +76,58 @@ export type CreateTreecrdtWebSocketSyncFromTransportOptions = { onLiveError?: (error: unknown) => void; }; +export type OutboundSyncStatus = { + targetCount: number; + pendingOps: number; + running: boolean; + scheduled: boolean; +}; + +export type OutboundSyncOptions = { + localPeer: SyncPeer; + /** + * Stable key used to coalesce repeated local write hints before upload. + * + * Defaults to TreeCRDT `Operation.meta.id` when the queued op has the standard operation shape. + * Provide this only for custom op shapes or custom coalescing behavior. + */ + opKey?: (op: Op) => string; + /** + * Allows apps to keep queued work while offline instead of turning transient offline state into + * sync errors. + */ + isOnline?: () => boolean; + pushOptions?: (targetId: string) => SyncPushOptions | undefined; + /** Abort a stalled push after this duration while keeping its ops queued for retry. */ + pushTimeoutMs?: number | ((targetId: string) => number | undefined); + onError?: (ctx: { targetId: string; error: unknown }) => void; + onStatus?: (status: OutboundSyncStatus) => void; +}; + +export type OutboundSync = { + readonly status: OutboundSyncStatus; + readonly pendingOpCount: number; + readonly targetCount: number; + /** + * Attach a remote transport to `localPeer` and register it as an outbound upload target. The + * returned cleanup removes the target and detaches the transport. Repeated calls with the same + * id and transport share one attachment until their last cleanup runs. + */ + attachTarget: (targetId: string, transport: DuplexTransport>) => () => void; + /** + * Register an already-attached transport as an outbound upload target. + */ + addTarget: (targetId: string, transport: DuplexTransport>) => void; + removeTarget: (targetId: string) => void; + clearTargets: () => void; + /** + * Report exact committed local ops. Wakes live subscriptions on `localPeer` and queues the same + * ops for registered outbound upload targets. + */ + queueOps: (ops: readonly Op[]) => void; + /** Flush queued work. Concurrent calls share the active drain barrier. */ + flush: () => Promise; + close: () => void; +}; + export type { MaterializationEvent, MaterializationListener }; diff --git a/packages/treecrdt-sync/tests/controller.test.ts b/packages/treecrdt-sync/tests/controller.test.ts new file mode 100644 index 00000000..ae026b9f --- /dev/null +++ b/packages/treecrdt-sync/tests/controller.test.ts @@ -0,0 +1,409 @@ +import { expect, test, vi } from 'vitest'; +import type { Operation } from '@treecrdt/interface'; +import { makeOp, nodeIdFromInt } from '@treecrdt/benchmark'; +import type { SyncPeer } from '@treecrdt/sync-protocol'; + +import { createOutboundSync } from '../src/controller.js'; +import { ROOT, orderKeyFromPosition } from './test-helpers.js'; + +function replicaFromLabel(label: string): Uint8Array { + const encoded = new TextEncoder().encode(label); + if (encoded.length === 0) throw new Error('replica label must not be empty'); + const out = new Uint8Array(32); + for (let i = 0; i < out.length; i += 1) out[i] = encoded[i % encoded.length]!; + return out; +} + +function deferred() { + let resolve!: () => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +const replicas = { a: replicaFromLabel('a') }; + +function makeInsertOp(counter = 1): Operation { + return makeOp(replicas.a, counter, counter, { + type: 'insert', + parent: ROOT, + node: nodeIdFromInt(counter), + orderKey: orderKeyFromPosition(counter - 1), + }); +} + +function createFakePeer(opts: { failPushes?: number } = {}) { + let failPushes = opts.failPushes ?? 0; + const pushed: Op[][] = []; + const detachTransport = vi.fn(); + const peer = { + attach: vi.fn(() => detachTransport), + notifyLocalUpdate: vi.fn(async () => {}), + pushOps: vi.fn(async (_transport: unknown, ops: readonly Op[]) => { + if (failPushes > 0) { + failPushes -= 1; + throw new Error('direct push failed'); + } + pushed.push([...ops]); + }), + } as unknown as SyncPeer; + return { detachTransport, peer, pushed }; +} + +test('outbound sync queues local ops until an outbound target is available', async () => { + const op = makeInsertOp(); + const { peer, pushed } = createFakePeer(); + const controller = createOutboundSync({ + localPeer: peer, + }); + + controller.queueOps([op, op]); + await controller.flush(); + + expect(peer.notifyLocalUpdate).toHaveBeenCalledWith([op, op]); + expect(controller.pendingOpCount).toBe(1); + expect(pushed).toHaveLength(0); + + controller.addTarget('remote:server', {} as any); + await controller.flush(); + + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([[op]]); +}); + +test('outbound sync can attach and clean up a remote target', async () => { + const op = makeInsertOp(); + const transport = {} as any; + const { detachTransport, peer, pushed } = createFakePeer(); + const controller = createOutboundSync({ + localPeer: peer, + }); + + const detachTarget = controller.attachTarget('remote:server', transport); + + expect(peer.attach).toHaveBeenCalledWith(transport); + expect(controller.targetCount).toBe(1); + + controller.queueOps([op]); + await controller.flush(); + + expect(pushed).toEqual([[op]]); + + detachTarget(); + detachTarget(); + + expect(detachTransport).toHaveBeenCalledTimes(1); + expect(controller.targetCount).toBe(0); +}); + +test('outbound sync shares repeated attachments without stale cleanup leaks', () => { + const oldTransport = { id: 'old' } as any; + const newTransport = { id: 'new' } as any; + const detachOld = vi.fn(); + const detachNew = vi.fn(); + const peer = { + attach: vi.fn((transport: unknown) => (transport === oldTransport ? detachOld : detachNew)), + notifyLocalUpdate: vi.fn(async () => {}), + pushOps: vi.fn(async () => {}), + } as unknown as SyncPeer; + const controller = createOutboundSync({ localPeer: peer }); + + const releaseOldA = controller.attachTarget('remote:server', oldTransport); + const releaseOldB = controller.attachTarget('remote:server', oldTransport); + expect(peer.attach).toHaveBeenCalledTimes(1); + + releaseOldA(); + releaseOldA(); + expect(detachOld).not.toHaveBeenCalled(); + expect(controller.targetCount).toBe(1); + + const releaseNew = controller.attachTarget('remote:server', newTransport); + expect(peer.attach).toHaveBeenCalledTimes(2); + expect(detachOld).toHaveBeenCalledTimes(1); + + releaseOldB(); + expect(detachNew).not.toHaveBeenCalled(); + expect(controller.targetCount).toBe(1); + + releaseNew(); + releaseNew(); + expect(detachNew).toHaveBeenCalledTimes(1); + expect(controller.targetCount).toBe(0); +}); + +test('outbound sync close detaches attached remote targets', () => { + const { detachTransport, peer } = createFakePeer(); + const controller = createOutboundSync({ + localPeer: peer, + }); + + controller.attachTarget('remote:server', {} as any); + controller.close(); + controller.close(); + + expect(detachTransport).toHaveBeenCalledTimes(1); + expect(controller.targetCount).toBe(0); +}); + +test('outbound sync keeps queued ops while offline', async () => { + const op = makeInsertOp(); + let online = false; + const { peer, pushed } = createFakePeer(); + const controller = createOutboundSync({ + localPeer: peer, + isOnline: () => online, + }); + controller.addTarget('remote:server', {} as any); + + controller.queueOps([op]); + await controller.flush(); + + expect(controller.pendingOpCount).toBe(1); + expect(pushed).toEqual([]); + + online = true; + await controller.flush(); + + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([[op]]); +}); + +test('outbound sync keeps failed direct pushes queued', async () => { + const op = makeInsertOp(); + const { peer, pushed } = createFakePeer({ failPushes: 1 }); + const errors: unknown[] = []; + const controller = createOutboundSync({ + localPeer: peer, + onError: ({ error }) => errors.push(error), + }); + controller.addTarget('remote:server', {} as any); + + controller.queueOps([op]); + await controller.flush(); + + expect(controller.pendingOpCount).toBe(1); + expect(errors).toHaveLength(1); + expect(pushed).toHaveLength(0); + + await controller.flush(); + + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([[op]]); +}); + +test('concurrent flush waits for work scheduled during the active run', async () => { + const firstOp = makeInsertOp(1); + const secondOp = makeInsertOp(2); + const deliveries: Operation[][] = []; + const firstGate = deferred(); + const secondGate = deferred(); + const firstStarted = deferred(); + const secondStarted = deferred(); + const peer = { + attach: vi.fn(() => () => {}), + notifyLocalUpdate: vi.fn(async () => {}), + pushOps: vi.fn(async (_transport: unknown, ops: readonly Operation[]) => { + deliveries.push([...ops]); + if (deliveries.length === 1) { + firstStarted.resolve(); + await firstGate.promise; + } else { + secondStarted.resolve(); + await secondGate.promise; + } + }), + } as unknown as SyncPeer; + const controller = createOutboundSync({ localPeer: peer }); + controller.addTarget('remote:server', {} as any); + + controller.queueOps([firstOp]); + const activeFlush = controller.flush(); + await firstStarted.promise; + controller.queueOps([secondOp]); + const concurrentFlush = controller.flush(); + let concurrentResolved = false; + void concurrentFlush.then(() => { + concurrentResolved = true; + }); + + expect(concurrentFlush).toBe(activeFlush); + firstGate.resolve(); + await secondStarted.promise; + expect(concurrentResolved).toBe(false); + + secondGate.resolve(); + await Promise.all([activeFlush, concurrentFlush]); + + expect(deliveries).toEqual([[firstOp], [secondOp]]); + expect(controller.pendingOpCount).toBe(0); +}); + +test('outbound sync aborts a timed-out push before retrying it', async () => { + const op = makeInsertOp(); + const pushed: Operation[][] = []; + const errors: unknown[] = []; + let attempts = 0; + let activePushes = 0; + let maxActivePushes = 0; + let timedOutSignal: AbortSignal | undefined; + + const peer = { + attach: vi.fn(() => () => {}), + notifyLocalUpdate: vi.fn(async () => {}), + pushOps: vi.fn( + (_transport: unknown, ops: readonly Operation[], opts?: { signal?: AbortSignal }) => { + attempts += 1; + activePushes += 1; + maxActivePushes = Math.max(maxActivePushes, activePushes); + + if (attempts === 1) { + timedOutSignal = opts?.signal; + return new Promise((_resolve, reject) => { + const onAbort = () => { + activePushes -= 1; + reject(timedOutSignal?.reason); + }; + if (timedOutSignal?.aborted) onAbort(); + else timedOutSignal?.addEventListener('abort', onAbort, { once: true }); + }); + } + + activePushes -= 1; + pushed.push([...ops]); + return Promise.resolve(); + }, + ), + } as unknown as SyncPeer; + const controller = createOutboundSync({ + localPeer: peer, + pushTimeoutMs: 5, + onError: ({ error }) => errors.push(error), + }); + controller.addTarget('remote:server', {} as any); + + controller.queueOps([op]); + await controller.flush(); + + expect(timedOutSignal?.aborted).toBe(true); + expect(controller.pendingOpCount).toBe(1); + expect(errors).toHaveLength(1); + + await controller.flush(); + + expect(maxActivePushes).toBe(1); + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([[op]]); +}); + +test('outbound sync replays an in-flight batch to a replacement target', async () => { + const op = makeInsertOp(); + const oldTransport = { id: 'old' } as any; + const newTransport = { id: 'new' } as any; + const deliveries: Array<{ transport: unknown; ops: Operation[] }> = []; + let oldPushSignal: AbortSignal | undefined; + const peer = { + attach: vi.fn(() => () => {}), + notifyLocalUpdate: vi.fn(async () => {}), + pushOps: vi.fn( + async (transport: unknown, ops: readonly Operation[], opts?: { signal?: AbortSignal }) => { + deliveries.push({ transport, ops: [...ops] }); + if (transport === oldTransport) { + oldPushSignal = opts?.signal; + await new Promise((_resolve, reject) => + oldPushSignal?.addEventListener('abort', () => reject(oldPushSignal?.reason), { + once: true, + }), + ); + } + }, + ), + } as unknown as SyncPeer; + const controller = createOutboundSync({ localPeer: peer }); + controller.addTarget('remote:server', oldTransport); + + controller.queueOps([op]); + const flushing = controller.flush(); + await Promise.resolve(); + expect(deliveries).toEqual([{ transport: oldTransport, ops: [op] }]); + + controller.addTarget('remote:server', newTransport); + await flushing; + + expect(oldPushSignal?.aborted).toBe(true); + expect(deliveries).toEqual([ + { transport: oldTransport, ops: [op] }, + { transport: newTransport, ops: [op] }, + ]); + expect(controller.pendingOpCount).toBe(0); +}); + +test('outbound sync isolates observer failures and rejected local notifications', async () => { + const op = makeInsertOp(); + const peer = { + attach: vi.fn(() => () => {}), + notifyLocalUpdate: vi.fn(async () => { + throw new Error('local notification failed'); + }), + pushOps: vi.fn(async () => { + throw new Error('direct push failed'); + }), + } as unknown as SyncPeer; + const controller = createOutboundSync({ + localPeer: peer, + onStatus: () => { + throw new Error('status observer failed'); + }, + onError: () => { + throw new Error('error observer failed'); + }, + }); + + controller.addTarget('remote:server', {} as any); + controller.queueOps([op]); + await expect(controller.flush()).resolves.toBeUndefined(); + + expect(controller.pendingOpCount).toBe(1); + expect(controller.status.running).toBe(false); +}); + +test('outbound sync accepts custom op keys for non-TreeCRDT op shapes', async () => { + type CustomOp = { id: string }; + const op: CustomOp = { id: 'local-write-1' }; + const { peer, pushed } = createFakePeer(); + const controller = createOutboundSync({ + localPeer: peer, + opKey: (next) => next.id, + }); + + controller.queueOps([op, { id: op.id }]); + await controller.flush(); + + expect(peer.notifyLocalUpdate).toHaveBeenCalledWith([op, { id: op.id }]); + expect(controller.pendingOpCount).toBe(1); + expect(pushed).toHaveLength(0); + + controller.addTarget('remote:server', {} as any); + await controller.flush(); + + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([[op]]); +}); + +test('outbound sync ignores empty op batches', async () => { + const { peer, pushed } = createFakePeer(); + const statuses: number[] = []; + const controller = createOutboundSync({ + localPeer: peer, + onStatus: (status) => statuses.push(status.pendingOps), + }); + controller.addTarget('remote:server', {} as any); + + controller.queueOps([]); + await controller.flush(); + + expect(peer.notifyLocalUpdate).not.toHaveBeenCalled(); + expect(controller.pendingOpCount).toBe(0); + expect(pushed).toEqual([]); + expect(statuses.at(-1)).toBe(0); +});