feat(sync): add a race-safe outbound controller

.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.

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<void> {
   );
 }
 
+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<T>(step: T | PromiseLike<T>, signal?: AbortSignal): Promise<T> {
+  const promise = Promise.resolve(step);
+  if (!signal) return promise;
+  if (signal.aborted) return Promise.reject(syncAbortReason(signal));
+
+  return new Promise<T>((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<Op> = {
   subscriptionId: string;
   filter: Filter;
@@ -523,27 +560,35 @@ export class SyncPeer<Op> {
 
   private async refreshHelloCapabilities(
     transport: DuplexTransport<SyncMessage<Op>>,
+    signal?: AbortSignal,
   ): Promise<void> {
     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<Op> {
   ): Promise<void> {
     if (ops.length === 0) return;
 
-    await this.refreshHelloCapabilities(transport);
+    const signal = opts.signal;
+    const step = <T>(value: T | PromiseLike<T>) => 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<Op> {
     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());
     }
   }
 

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<Operation>[] = [];
+  let releaseFirstSend!: () => void;
+  const firstSend = new Promise<void>((resolve) => {
+    releaseFirstSend = resolve;
+  });
+  const transport: DuplexTransport<SyncMessage<Operation>> = {
+    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);

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