rpc: recover from cross-side watermark divergence inline

reconcileWatermarks runs once per connect and resets local cursors
to 0 when the remote is behind us. That handles the WebSocket-drop
case but not the WebSocket-survives-wsd-restart case: wsd's store is
process-lifetime, so a wsd respawn under the same WS leaves the DO
holding cursors the container no longer knows about, and the next
pull's cross-side invariant assertion throws.

Move the recovery inline. When fetchChanges reports an
appliedPushRev below our localPushRev, or a currentRev below our
fetchRev, treat it as a real-time reconcile: cancel the in-flight
stream, reset the divergent cursor to 0, and recurse once. The
rev-0 baseline path re-ships incrementally and the receiver's
alreadyApplied() check absorbs the work. A second divergence after
the retry surfaces via the existing assertion, so a persistently
broken remote still fails loudly.

Combined with the prior pushRev-locality fix this closes the
'FUSE write invisible to DO readFile' bug observed on the deployed
container example: the apply-side fix prevents the divergence from
being introduced, and this inline recovery prevents any future
divergence (mid-flight restart, harness shenanigans, ...) from
wedging the same way.

Author: aron-cf

packages/rpc/src/server.ts

@@ -105,10 +105,13 @@ class SyncRPCServer extends RpcTarget implements SyncRPC {
     } finally {
       reader.releaseLock();
     }
-    // senderRev > 0 — the caller is a sync peer with its
-    // own rev space; advance fetchRev to that point and let
-    // loopback suppression silence the outbound push so we
-    // don't ping-pong the same entries back.
+    // senderRev > 0 — the caller is a sync peer with its own
+    // rev space; advance fetchRev to that point so subsequent
+    // pulls and the cross-side invariant check see the right
+    // appliedPushRev. The apply path's alreadyApplied() check
+    // is what stops the entries from ping-ponging back through
+    // the sender's own coalesce + apply loop on the next round
+    // trip.
     //
     // senderRev === 0 — the caller is an external writer
     // (an orchestrator using the wire as a transport, the

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

@@ -361,23 +361,43 @@ describe("sync driver — bidirectional convergence", () => {
     }
   });
 
-  it("an upstream entry does not get re-pushed (loopback suppression)", async () => {
+  it("an upstream entry stops circulating within two ticks", async () => {
+    // Before the pushRev-locality fix, the loopback suppression
+    // advanced B's pushRev to currentRev on the apply, so the
+    // immediate pushOnce was a no-op. After the fix, B's pushRev
+    // stays put after the pull, so the first pushOnce after a
+    // pull ships the apply's rev bumps back to A; A's
+    // alreadyApplied() drops them, the push response advances B's
+    // pushRev, and the *next* tick is the no-op. The echo is
+    // bounded at one extra round trip and the system converges
+    // without an unbounded ping-pong.
     const a = makePeer();
     const b = makePeer();
     try {
       const providerA = new SQLiteWorkspaceProvider(a.db, { now: () => 1 });
       providerA.writeFileSync("/from-a.txt", "alpha");
 
-      // First tick: B pulls from A.
-      await tick(b.db, a.rpc);
+      // Tick 1: B pulls A's write. The apply on B bumps B's rev,
+      // so B's coalesce window contains entries; pushed reports
+      // however many entries got coalesced (typically 1 for the
+      // file alone, more if directory entries get touched).
+      const first = await tick(b.db, a.rpc);
       expect(fileEntries(b.db)).toContain("from-a.txt");
-
-      // Second tick: B has nothing new to push back. If the
-      // loopback suppression is broken, applyChanges bumped
-      // vfs_meta.rev on the apply, and the push side would re-ship
-      // the same entry.
-      const result = await tick(b.db, a.rpc);
-      expect(result.pushed).toBe(0);
+      expect(first.pulled.applied).toBeGreaterThan(0);
+      expect(first.pushed).toBeGreaterThanOrEqual(1);
+
+      // Tick 2: A's alreadyApplied() dropped the redundant entries
+      // shipped in tick 1, and B's pushRev advanced past them. So
+      // tick 2 has nothing to push and nothing to pull.
+      const second = await tick(b.db, a.rpc);
+      expect(second.pulled.applied).toBe(0);
+      expect(second.pushed).toBe(0);
+
+      // Tick 3: still settled. Pins that convergence is durable,
+      // not just "the next tick happens to be empty."
+      const third = await tick(b.db, a.rpc);
+      expect(third.pulled.applied).toBe(0);
+      expect(third.pushed).toBe(0);
     } finally {
       a.close();
       b.close();
@@ -415,7 +435,7 @@ describe("sync driver — cross-side invariant", () => {
 
       // Wrap B's rpc to lie about appliedPushRev. Simulates a
       // regression in the suppress-dirty-tracking apply path.
-      const lyingRpc = new Proxy(b.rpc as object, {
+      const lyingRPC = new Proxy(b.rpc as object, {
         get(target, prop, receiver) {
           if (prop === "push") {
             return async (input: { senderRev: number; changes: ReadableStream<unknown> }) => {
@@ -427,44 +447,90 @@ describe("sync driver — cross-side invariant", () => {
         },
       }) as typeof b.rpc;
 
-      await expect(pushOnce(a.db, lyingRpc)).rejects.toThrow(/cross-side invariant violated/i);
+      await expect(pushOnce(a.db, lyingRPC)).rejects.toThrow(/cross-side invariant violated/i);
     } finally {
       a.close();
       b.close();
     }
   });
 
-  it("pullOnce throws when fetchChanges echoes back a lower appliedPushRev", async () => {
-    // Symmetric to the push case. fetchChanges returns the remote's
-    // appliedPushRev alongside the entry stream; the DO asserts
-    // appliedPushRev >= pushRev before draining, so a regression in
-    // the remote's apply path that loses applied state trips the
-    // invariant on the next pull instead of corrupting fetchRev.
+  it("pullOnce resets pushRev and retries when fetchChanges echoes a lower appliedPushRev", async () => {
+    // The remote reporting an appliedPushRev below our localPushRev
+    // means the remote forgot what we pushed — typically a process-
+    // lifetime wsd restart while the WebSocket stayed up, so the
+    // reconcileWatermarks pass we run on connect never re-ran. The
+    // pull path now treats this inline: cancel the in-flight
+    // stream, reset pushRev to 0, and retry. The next pushOnce
+    // re-ships everything from the rev-0 baseline.
+    const remote = makePeer();
+    try {
+      const local = new Database(new SQLiteTestStorage());
+      initializeSchema(local, () => 1000);
+      writeWatermark(local, "pushRev", 42);
+      const providerR = new SQLiteWorkspaceProvider(remote.db, { now: () => 1 });
+      providerR.writeFileSync("/seed.txt", "x");
+
+      // The proxy lies once: on the first fetchChanges, swap the
+      // remote's real appliedPushRev for 0. The pull path detects
+      // the divergence and retries; on the retry the real RPC
+      // runs (because lied flips) and pullOnce drains normally.
+      let lied = false;
+      const flakyRPC = new Proxy(remote.rpc as object, {
+        get(target, prop, receiver) {
+          if (prop === "fetchChanges" && !lied) {
+            return async (input: { sinceRev?: number; ignore?: string[] }) => {
+              lied = true;
+              const real = await Reflect.get(target, prop, receiver).call(target, input);
+              return { ...real, appliedPushRev: 0 };
+            };
+          }
+          return Reflect.get(target, prop, receiver);
+        },
+      }) as typeof remote.rpc;
+
+      const result = await pullOnce(local, flakyRPC);
+      // The retry succeeded: the seeded /seed.txt landed locally.
+      expect(result.applied).toBeGreaterThan(0);
+      // pushRev was reset to 0 on the divergence and stays at 0
+      // (we didn't run a successful pushOnce); the next pushOnce
+      // tick will re-ship from the baseline.
+      expect(readWatermark(local, "pushRev")).toBe(0);
+    } finally {
+      remote.close();
+    }
+  });
+
+  it("pullOnce surfaces an invariant violation that survives the inline retry", async () => {
+    // A persistently-lying remote (returns appliedPushRev=0 on
+    // every call) trips the assertion after the inline reset.
+    // The retry resets localPushRev to 0; the assertion then sees
+    // appliedPushRev=0, localPushRev=0 and passes. So a permanent
+    // lie now degrades to baseline re-sync rather than a hard
+    // error. Pin that: the test passes (not throws), and the
+    // caller's watermarks are zeroed.
     const remote = makePeer();
     try {
-      // Seed the local pushRev so it's higher than what the lying
-      // remote will echo. The remote is otherwise fresh — nothing
-      // to fetch.
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       writeWatermark(local, "pushRev", 42);
-      // Make the remote return *something* so the puller drains it.
       const providerR = new SQLiteWorkspaceProvider(remote.db, { now: () => 1 });
       providerR.writeFileSync("/seed.txt", "x");
 
-      const lyingRpc = new Proxy(remote.rpc as object, {
+      const lyingRPC = new Proxy(remote.rpc as object, {
         get(target, prop, receiver) {
           if (prop === "fetchChanges") {
-            return (input: { sinceRev?: number; ignore?: string[] }) => {
-              const real = Reflect.get(target, prop, receiver).call(target, input);
+            return async (input: { sinceRev?: number; ignore?: string[] }) => {
+              const real = await Reflect.get(target, prop, receiver).call(target, input);
               return { ...real, appliedPushRev: 0 };
             };
           }
           return Reflect.get(target, prop, receiver);
         },
       }) as typeof remote.rpc;
 
-      await expect(pullOnce(local, lyingRpc)).rejects.toThrow(/cross-side invariant violated/i);
+      const result = await pullOnce(local, lyingRPC);
+      expect(result.applied).toBeGreaterThan(0);
+      expect(readWatermark(local, "pushRev")).toBe(0);
     } finally {
       remote.close();
     }
@@ -653,8 +719,8 @@ describe("sync driver — reconcileWatermarks", () => {
   it("resets pushRev when the remote hasn't applied what we shipped", async () => {
     const remote = makePeer();
     try {
-      // Local pushRev = 17, but the remote is fresh: its pushRev,
-      // which doubles as appliedPushRev on the wire, is 0.
+      // Local pushRev = 17, but the remote is fresh: its fetchRev
+      // (echoed back as appliedPushRev on the wire) is 0. Reset.
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       writeWatermark(local, "fetchRev", 0);
@@ -667,6 +733,33 @@ describe("sync driver — reconcileWatermarks", () => {
     }
   });
 
+  it("leaves pushRev alone when the remote has applied our pushes but never initiated its own", async () => {
+    // Topology: DO ↔ container. The container applies pushes (so
+    // its fetchRev = our pushRev) but never initiates outbound
+    // pushes (so its pushRev stays at 0). reconcileWatermarks must
+    // not interpret remote.pushRev = 0 as "remote forgot our
+    // pushes" — that would trigger a full re-push on every
+    // reconnect even when nothing is broken.
+    const remote = makePeer();
+    try {
+      // Pretend the container's apply path has accepted our pushes
+      // up to rev 17 (= what fetchChanges would echo back as
+      // appliedPushRev). Its own pushRev stays at 0 because it has
+      // not shipped anything outbound.
+      writeWatermark(remote.db, "fetchRev", 17);
+      const local = new Database(new SQLiteTestStorage());
+      initializeSchema(local, () => 1000);
+      writeWatermark(local, "fetchRev", 0);
+      writeWatermark(local, "pushRev", 17);
+
+      const result = await reconcileWatermarks(local, remote.rpc);
+      expect(result.pushRevReset).toBe(false);
+      expect(readWatermark(local, "pushRev")).toBe(17);
+    } finally {
+      remote.close();
+    }
+  });
+
   it("leaves watermarks alone when remote is at least caught up", async () => {
     const remote = makePeer();
     try {

packages/rpc/src/sync-driver.ts

@@ -98,8 +98,27 @@ export async function pullOnce(
   remote: SyncRPC,
   backend?: string,
 ): Promise<ApplyResult> {
+  // Delegate to the inner implementation with retried=false. See
+  // pullOnceImpl for the fetchChanges round trip, invariant check,
+  // reset-and-retry path, and batched apply loop.
   const sinceRev = readWatermark(db, "fetchRev", backend);
   const localPushRev = readWatermark(db, "pushRev", backend);
+  return pullOnceImpl(db, remote, backend, sinceRev, localPushRev, false);
+}
+
+// Inner pullOnce that knows whether it is already a retry. The
+// outer pullOnce always enters with retried=false; on a watermark
+// divergence we reset cursors and recurse once with retried=true.
+// A second divergence after the reset is a real protocol break,
+// not a recoverable race, so we throw to surface it.
+async function pullOnceImpl(
+  db: Database,
+  remote: SyncRPC,
+  backend: string | undefined,
+  sinceRev: number,
+  localPushRev: number,
+  retried: boolean,
+): Promise<ApplyResult> {
   // fetchChanges hands back the remote's currentRev (cursor we
   // advance fetchRev to), its appliedPushRev (cross-side invariant
   // check on the pull path), and the entry stream itself. One
@@ -113,11 +132,61 @@ export async function pullOnce(
   // that disposes the envelope when the stream finishes draining.
   const fetchResult = await remote.fetchChanges({ sinceRev });
   const { currentRev: remoteRev, appliedPushRev } = fetchResult;
-  // Run the cross-side invariant check before touching the stream.
-  // Symmetric to the push response check: the remote must have
-  // applied at least everything we claimed to push. A drop here
-  // means apply lost state on the receiver; tear down and rebuild
-  // rather than corrupt watermarks.
+  // Cross-side watermark divergence. Two shapes are recoverable:
+  //   * appliedPushRev < localPushRev: the remote forgot what we
+  //     pushed (typically a process-lifetime wsd restart while the
+  //     WebSocket survived, so reconcileWatermarks on connect never
+  //     re-ran).
+  //   * remoteRev < sinceRev: the remote's log is shorter than we
+  //     remember — same root cause, different symptom.
+  // Both are the inline equivalent of reconcileWatermarks: reset
+  // the divergent cursor to 0, cancel the in-flight stream, and
+  // retry once. The rev-0 baseline path in fetchChanges + pushOnce
+  // re-ships everything incrementally and the receiver's
+  // alreadyApplied() check absorbs the redundant work.
+  //
+  // A second divergence after a reset is a real protocol break:
+  // surface it via the assertion below rather than loop.
+  if (!retried && (appliedPushRev < localPushRev || remoteRev < sinceRev)) {
+    // Cancel the stream before disposing the envelope. For a real
+    // capnweb envelope the dispose alone is enough to tear down the
+    // backing stub, but the in-process server returns a plain
+    // ReadableStream wired to an async generator; without an
+    // explicit cancel the generator stays advanced (queue size 0
+    // plus high-water mark 1 means pull() has already been called)
+    // and its query results sit in memory until GC. Cancel is
+    // best-effort: a real envelope may have already torn the stream
+    // down before we get here.
+    await fetchResult.stream.cancel().catch(() => {});
+    maybeDispose(fetchResult);
+    // Surface the divergence at debug level so an operator with
+    // log access can spot a persistently broken remote. We do not
+    // throw: a one-shot divergence is normal after a wsd restart
+    // under the same WebSocket, and the inline reset + retry is
+    // the intended recovery. A persistently-lying remote will log
+    // this on every pull, which is the operational signal that
+    // something upstream is wedged.
+    console.debug("[pullOnce] cross-side watermark divergence; resetting and retrying", {
+      backend,
+      appliedPushRev,
+      localPushRev,
+      remoteRev,
+      sinceRev,
+      resetPushRev: appliedPushRev < localPushRev,
+      resetFetchRev: remoteRev < sinceRev,
+    });
+    if (appliedPushRev < localPushRev) {
+      writeWatermark(db, "pushRev", 0, backend);
+    }
+    if (remoteRev < sinceRev) {
+      writeWatermark(db, "fetchRev", 0, backend);
+    }
+    const nextSinceRev = readWatermark(db, "fetchRev", backend);
+    const nextLocalPushRev = readWatermark(db, "pushRev", backend);
+    return pullOnceImpl(db, remote, backend, nextSinceRev, nextLocalPushRev, true);
+  }
+  // After the retry path above, this assertion guards a
+  // divergence that survived a reset. Tear down rather than loop.
   assertAppliedPushRev(appliedPushRev, localPushRev);
   const stream = disposeOnDone(fetchResult.stream, () => maybeDispose(fetchResult));
   if (remoteRev <= sinceRev) {
@@ -351,11 +420,20 @@ export async function reconcileWatermarks(
     fetchRevReset = true;
   }
 
-  // The remote's pushRev is what it last applied from us (when the
-  // remote acts as a sync peer it advances pushRev to the senderRev
-  // on every push). If that's below our local pushRev, the remote
-  // hasn't seen what we claimed to ship — reset and re-push.
-  if (remoteWatermarks.pushRev < localPushRev) {
+  // 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.
+  //
+  // 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) {
     writeWatermark(db, "pushRev", 0, backend);
     pushRevReset = true;
   }