Fix issue with sync watermark divergence

packages/dofs/src/sync/apply.test.ts

@@ -123,7 +123,7 @@ describe("applyChanges", () => {
 });
 
 describe("applyChanges loopback suppression", () => {
-  it("advances pushRev to currentRev when source=upstream", async () => {
+  it("does not advance pushRev locally on upstream apply", async () => {
     await withDB(async (db) => {
       // Pre-existing local state: a write the container already
       // pushed. pushRev sits at currentRev.
@@ -133,10 +133,13 @@ describe("applyChanges loopback suppression", () => {
       const beforePushRev = readWatermark(db, "pushRev");
       expect(beforePushRev).toBeGreaterThan(0);
 
-      // Apply an entry as if it came from upstream. The local rev
-      // counter bumps (writeFile bumps rev), but the source flag
-      // makes the helper advance pushRev to match — the bump
-      // looks like it was already pushed.
+      // Apply an entry as if it came from upstream. The apply's
+      // writeFile bumps the local rev counter, but pushRev must
+      // *not* advance with it — advancing locally would move our
+      // pushRev past entries the remote does not know we have
+      // shipped, breaking the cross-side invariant on the next
+      // pull. The next pushOnce re-ships these rev bumps and the
+      // receiver's alreadyApplied() check drops them.
       await applyChanges(
         db,
         [
@@ -156,11 +159,8 @@ describe("applyChanges loopback suppression", () => {
 
       const afterCurrent = currentRev(db);
       const afterPushRev = readWatermark(db, "pushRev");
-      // Apply bumped currentRev (the writeFile inside).
       expect(afterCurrent).toBeGreaterThan(beforePushRev);
-      // pushRev caught up so the next coalesceChanges(db, pushRev)
-      // sees nothing.
-      expect(afterPushRev).toBe(afterCurrent);
+      expect(afterPushRev).toBe(beforePushRev);
     });
   });
 
@@ -186,15 +186,18 @@ describe("applyChanges loopback suppression", () => {
     });
   });
 
-  it("upstream entries do not get re-pushed on the next coalesce", async () => {
+  it("upstream entries surface on the next coalesce and rely on receiver-side alreadyApplied", async () => {
     await withDB(async (db) => {
       const { coalesceChanges } = await import("./coalesce.js");
       const { currentRev, readWatermark, writeWatermark } = await import("./watermarks.js");
       // Seed pushRev at the current point.
       writeWatermark(db, "pushRev", currentRev(db));
 
-      // Upstream sends a file. After apply, pushRev should equal
-      // currentRev, so coalesceChanges(db, pushRev) is empty.
+      // Upstream sends a file. After apply, pushRev stays where it
+      // was (the local advance was unsound — see the test above).
+      // The next coalesceChanges(db, pushRev) re-emits the entry;
+      // the receiver's alreadyApplied() check drops it. One extra
+      // round trip per apply, watermarks stay in lockstep.
       await applyChanges(
         db,
         [
@@ -214,7 +217,7 @@ describe("applyChanges loopback suppression", () => {
       const cursor = readWatermark(db, "pushRev");
       const drained = [];
       for await (const e of coalesceChanges(db, cursor)) drained.push(e);
-      expect(drained).toEqual([]);
+      expect(drained.map((e) => e.path)).toContain("/upstream.txt");
     });
   });
 });
@@ -271,12 +274,18 @@ describe("applyChanges loopback suppression — F1", () => {
     });
   });
 
-  it("still advances pushRev when caller had no unpushed locals", async () => {
+  it("leaves pushRev alone even when the caller had no unpushed locals", async () => {
     await withDB(async (db) => {
       const { currentRev, readWatermark, writeWatermark } = await import("./watermarks.js");
-      // pushRev already caught up to currentRev: caller has
-      // no pending local writes.
-      writeWatermark(db, "pushRev", currentRev(db));
+      // pushRev already caught up to currentRev: caller has no
+      // pending local writes. The old apply path advanced pushRev
+      // here as an optimization; we no longer do that because it
+      // desynced our pushRev from the remote's fetchRev (echoed
+      // back as appliedPushRev on the wire). The next pushOnce
+      // re-ships the apply's rev bump and the receiver's
+      // alreadyApplied() check drops it.
+      const before = currentRev(db);
+      writeWatermark(db, "pushRev", before);
       await applyChanges(
         db,
         [
@@ -293,9 +302,9 @@ describe("applyChanges loopback suppression — F1", () => {
         new Map(),
         { source: "upstream" },
       );
-      // Loopback suppression still works in the safe case:
-      // pushRev advances to cover the apply's own rev bump.
-      expect(readWatermark(db, "pushRev")).toBe(currentRev(db));
+      const after = currentRev(db);
+      expect(after).toBeGreaterThan(before);
+      expect(readWatermark(db, "pushRev")).toBe(before);
     });
   });
 });

packages/dofs/src/sync/apply.ts

@@ -7,7 +7,7 @@ import { writeFile, writeFileSync } from "../fs/writeFile.js";
 import type { Database } from "../storage.js";
 import type { ChangeEntry } from "./changes.js";
 import { computeManifestHash } from "./manifests.js";
-import { currentRev, readWatermark, writeWatermark } from "./watermarks.js";
+import { readWatermark, writeWatermark } from "./watermarks.js";
 
 // One container-side change that landed under a read-only mount and
 // was therefore skipped rather than applied. Callers (the workspace
@@ -53,14 +53,15 @@ export interface ApplyOptions {
   // cursor. Never regresses the watermark.
   advanceFetchRev?: number;
   // Where the entries came from. 'local' (default) treats the apply
-  // path like any other mutation: writeFile/mkdir/etc bump vfs_meta.rev
-  // and the push loop later ships those new revs upstream. 'upstream'
-  // means the entries came from a remote push or fetch; the apply
-  // still bumps rev (so readers see fresh data) but we advance pushRev
-  // to match, so the push loop knows everything in this range is
-  // already on the wire. Without this flag, applying an upstream
-  // entry would generate a push-back on the next tick and the two
-  // sides would ping-pong forever.
+  // path like any other mutation: writeFile/mkdir/etc bump
+  // vfs_meta.rev and the push loop later ships those new revs
+  // upstream. 'upstream' is informational: the apply still bumps
+  // rev so readers see fresh data, and the next pushOnce ships
+  // those rev bumps back to the sender. Loop convergence is the
+  // receiver's job — the apply path on the original sender uses
+  // alreadyApplied() to drop the redundant entries without bumping
+  // rev further, bounding the echo at one extra round trip per
+  // upstream apply.
   source?: "local" | "upstream";
   // Backend id whose watermark row this apply should touch. The
   // DO hosts independent sync cursors per backend; threading the
@@ -98,10 +99,6 @@ export async function applyChanges(
   objects: Map<string, Uint8Array>,
   options: ApplyOptions = {},
 ): Promise<ApplyResult> {
-  // Snapshot rev before we touch anything. Used by the loopback-
-  // suppression at the bottom to decide whether it's safe to
-  // advance pushRev past the entries this apply produced.
-  const revBeforeApply = currentRev(db);
   const maxBytes = options.maxBytesPerBatch ?? DEFAULT_MAX_BYTES;
   const maxPaths = options.maxPathsPerBatch ?? DEFAULT_MAX_PATHS;
 
@@ -207,34 +204,21 @@ export async function applyChanges(
     }
   }
 
-  // Loopback suppression: when this apply pass reflects entries
-  // from upstream, the writeFile/mkdir/symlink/rm calls inside
-  // bumped vfs_meta.rev. Without this advance, the next push tick
-  // would see those rev bumps as fresh local changes and push them
-  // back to upstream, which would apply them and bump again, and
-  // so on.
+  // Loopback suppression used to advance pushRev locally after an
+  // upstream apply so the next push tick wouldn't re-ship the rev
+  // bumps the apply produced. That optimization is unsound: it
+  // moves the *local* pushRev past entries the remote does not
+  // know we have shipped, while the remote's fetchRev (echoed back
+  // as appliedPushRev on every fetchChanges) stays where it was.
+  // The cross-side invariant check in pullOnce then trips on the
+  // very next pull and the post-drain pullOnce in the exec bracket
+  // swallows the error, leaving every subsequent container-side
+  // write invisible to the host until something reconciles.
   //
-  // Subtle: we can only advance pushRev when it already covered
-  // every rev that existed *before* this apply. If the caller had
-  // unpushed local writes sitting between (existing, revBeforeApply],
-  // advancing pushRev past them would strand them — the next
-  // pushOnce would skip them as already-shipped. That was F1: a
-  // pull whose entries were all idempotent-skipped still bumped
-  // pushRev up to currentRev, masking local writes that hadn't
-  // shipped yet.
-  //
-  // In the unsafe case we leave pushRev alone. The next pushOnce
-  // drains both the unpushed locals and the apply's own bumps;
-  // the receiver's alreadyApplied() check suppresses the latter.
-  // One redundant round-trip per apply, bounded.
-  if (options.source === "upstream") {
-    const revAfter = currentRev(db);
-    const existing = readWatermark(db, "pushRev", options.backend);
-    if (existing >= revBeforeApply && revAfter > existing) {
-      writeWatermark(db, "pushRev", revAfter, options.backend);
-    }
-  }
-
+  // The bounded "redundant round-trip" the old comment promised is
+  // still bounded, and the receiver's alreadyApplied() check still
+  // suppresses the entries on the next pushOnce. We just pay one
+  // extra push per upstream apply to keep the two sides in lockstep.
   return { applied, skipped };
 }
 
@@ -252,7 +236,6 @@ export function applyChangesSync(
   objects: Map<string, Uint8Array>,
   options: ApplyOptions = {},
 ): ApplyResult {
-  const revBeforeApply = currentRev(db);
   const maxBytes = options.maxBytesPerBatch ?? DEFAULT_MAX_BYTES;
   const maxPaths = options.maxPathsPerBatch ?? DEFAULT_MAX_PATHS;
 
@@ -342,13 +325,10 @@ export function applyChangesSync(
     }
   }
 
-  if (options.source === "upstream") {
-    const revAfter = currentRev(db);
-    const existing = readWatermark(db, "pushRev", options.backend);
-    if (existing >= revBeforeApply && revAfter > existing) {
-      writeWatermark(db, "pushRev", revAfter, options.backend);
-    }
-  }
+  // See applyChanges() for why pushRev no longer advances locally
+  // on upstream applies. The receiver's alreadyApplied() check
+  // suppresses the redundant entries on the next pushOnce; one
+  // extra push per apply keeps the cross-side invariant intact.
 
   return { applied, skipped };
 }

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