rpc: fix sync-driver build and same-rev pull divergence

The fetch watermark became a rev/path cursor, but reconcileWatermarks
still read remoteWatermarks.fetchRev, a field that no longer exists on
the watermarks() result. The reference broke the type check, failed
the rpc build, and cascaded into every workspace that depends on the
built package. Read the rev off the cursor instead.

The same cursor conversion also widened the pullOnce recovery
predicate. The inline reset path exists to absorb a remote that forgot
what we pushed, which shows up as the remote's applied push cursor rev
regressing below our local pushRev. Comparing the full cursor instead
of its rev folded a second, unrelated case into the recoverable set: a
same-rev partial cursor, where the remote applied part of the rev we
pushed and then tore down mid-apply. That is a genuine cross-side
invariant violation. Resetting and replaying cannot mend lost state
inside a rev the remote claimed to hold, so the predicate is rev-only
again and the assertion surfaces the partial-apply case.

Point the envelope-disposal test at a same-rev partial cursor so it
exercises a real invariant trip rather than the recoverable rev
regression, which the inline reset now swallows by design.

Author: aron-cf

packages/rpc/src/sync-driver.test.ts

@@ -1,4 +1,5 @@
 import {
+  type ChangeCursor,
   type ChangeEntry,
   currentRev,
   Database,
@@ -58,7 +59,7 @@ async function drainStream<T>(stream: ReadableStream<T>): Promise<T[]> {
 // applyChanges to throw on a missing object mid-stream).
 function trackFetchDisposal(
   rpc: SyncRPC,
-  opts: { lowerCursor?: boolean; failHasObjects?: boolean } = {},
+  opts: { appliedPushCursor?: ChangeCursor; failHasObjects?: boolean } = {},
 ): { rpc: SyncRPC; disposeCount: () => number } {
   let disposeCount = 0;
   const wrapped = new Proxy(rpc as object, {
@@ -68,7 +69,9 @@ function trackFetchDisposal(
           const real = await Reflect.get(target, prop, receiver).call(target, ...args);
           return {
             ...real,
-            ...(opts.lowerCursor ? { appliedPushCursor: { rev: 0, path: null } } : {}),
+            ...(opts.appliedPushCursor !== undefined
+              ? { appliedPushCursor: opts.appliedPushCursor }
+              : {}),
             [Symbol.dispose]() {
               disposeCount += 1;
             },
@@ -460,13 +463,17 @@ describe("sync driver — pullOnce envelope disposal", () => {
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       // Local claims to have pushed rev 42; the wrapper echoes back a
-      // rev-0 cursor, so assertAppliedPushCursor throws before the
-      // stream is ever read.
+      // same-rev partial cursor (rev 42 but stalled mid-apply at a
+      // path). That is a non-recoverable divergence — the inline
+      // reset path only catches a rev regression — so
+      // assertAppliedPushCursor throws before the stream is read.
       writeWatermark(local, "pushRev", 42);
       const providerR = new SQLiteWorkspaceProvider(remote.db, { now: () => 1 });
       providerR.writeFileSync("/seed.txt", "x");
 
-      const tracked = trackFetchDisposal(remote.rpc, { lowerCursor: true });
+      const tracked = trackFetchDisposal(remote.rpc, {
+        appliedPushCursor: { rev: 42, path: "/partial.txt" },
+      });
       await expect(pullOnce(local, tracked.rpc)).rejects.toThrow(/cross-side invariant violated/i);
       expect(tracked.disposeCount()).toBe(1);
     } finally {

packages/rpc/src/sync-driver.ts

@@ -107,10 +107,10 @@ async function pullOnceImpl(
   try {
     const { currentCursor, appliedPushCursor } = fetchResult;
     // Cross-side watermark divergence. Two shapes are recoverable:
-    //   * appliedPushCursor < {rev: localPushRev, path: null}: the
-    //     remote forgot what we pushed (typically a process-lifetime
-    //     wsd restart while the WebSocket survived, so
-    //     reconcileWatermarks on connect never re-ran).
+    //   * appliedPushCursor.rev < localPushRev: the remote forgot
+    //     what we pushed (typically a process-lifetime wsd restart
+    //     while the WebSocket survived, so reconcileWatermarks on
+    //     connect never re-ran).
     //   * currentCursor < after: the remote's log is shorter than we
     //     remember — same root cause, different symptom.
     // Both are the inline equivalent of reconcileWatermarks: reset
@@ -119,10 +119,16 @@ async function pullOnceImpl(
     // re-ships everything incrementally and the receiver's
     // alreadyApplied() check absorbs the redundant work.
     //
+    // The divergence test is rev-only on purpose. A same-rev partial
+    // appliedPushCursor (the remote applied part of the rev we
+    // pushed, then tore down mid-apply) is not a recoverable race —
+    // it means the receiver lost state inside a rev it told us it
+    // had. Resetting and replaying cannot mend that, so we let the
+    // assertion below surface it instead of looping.
+    //
     // A second divergence after a reset is a real protocol break:
     // surface it via the assertion below rather than loop.
-    const pushDiverged =
-      compareChangeCursors(appliedPushCursor, { rev: localPushRev, path: null }) < 0;
+    const pushDiverged = appliedPushCursor.rev < localPushRev;
     const fetchDiverged = compareChangeCursors(currentCursor, after) < 0;
     if (!retried && (pushDiverged || fetchDiverged)) {
       // Cancel the stream before disposing the envelope. For a real
@@ -410,20 +416,20 @@ export async function reconcileWatermarks(
     fetchRevReset = true;
   }
 
-  // The remote's fetchRev is the largest senderRev it has applied
-  // from us — every push handler advances fetchRev to the incoming
-  // senderRev, and fetchChanges echoes that value back as
-  // appliedPushRev. If it's below our local pushRev, the remote has
-  // not seen what we claimed to ship; reset our pushRev so the next
-  // pushOnce re-baselines from rev 0.
+  // The remote's fetch cursor rev is the largest senderRev it has
+  // applied from us — every push handler advances the fetch cursor to
+  // the incoming senderRev, and fetchChanges echoes that cursor back
+  // as appliedPushCursor. If its rev is below our local pushRev, the
+  // remote has not seen what we claimed to ship; reset our pushRev so
+  // the next pushOnce re-baselines from rev 0.
   //
   // We deliberately do NOT compare against remoteWatermarks.pushRev:
   // that field is the remote's own *outbound* push progress and
   // stays at 0 in topologies where the remote never initiates a push
   // (e.g. the container side of a DO↔container backend), which would
   // make every reconcile spuriously reset pushRev and force a full
   // re-push on every reconnect.
-  if (remoteWatermarks.fetchRev < localPushRev) {
+  if (remoteWatermarks.fetchCursor.rev < localPushRev) {
     writeWatermark(db, "pushRev", 0, backend);
     pushRevReset = true;
   }