From abde16126705f7ec915aff68a53cb41cbe0e12d5 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:25:31 +0000 Subject: [PATCH 1/2] dofs: stop advancing pushRev locally on upstream apply The loopback-suppression optimization in applyChanges advanced the local pushRev to currentRev after every upstream apply, on the theory that the apply's own rev bumps would otherwise get re-pushed. That theory was correct in isolation but the implementation was unsound: it moved our pushRev past entries the remote did not know we had shipped, while the remote's fetchRev (echoed back as appliedPushRev on every fetchChanges response) stayed where the last real push had put it. The cross-side invariant check in pullOnce then trips on the very next pull and the post-drain pull in the exec bracket swallows the error, leaving every subsequent container-side write invisible to the host until something reconciles the watermarks. Drop the local advance. The next pushOnce ships the apply's rev bumps, the receiver's alreadyApplied() check drops them as no-ops, and the container's fetchRev catches up to our pushRev in the same round trip. One extra push per upstream apply, bounded by the batch's coalesce output. The cross-side invariant stays intact. The two F1 tests on this behavior already covered the unsafe case (do not strand unpushed locals); they still pass. The two tests that pinned the optimization (one in apply.test.ts, one in wire.test.ts) are updated to assert the new contract. --- packages/dofs/src/sync/apply.test.ts | 49 ++++++++++-------- packages/dofs/src/sync/apply.ts | 76 ++++++++++------------------ packages/rpc/tests/wire.test.ts | 14 ++--- 3 files changed, 65 insertions(+), 74 deletions(-) diff --git a/packages/dofs/src/sync/apply.test.ts b/packages/dofs/src/sync/apply.test.ts index b326e0be..ed60c8c0 100644 --- a/packages/dofs/src/sync/apply.test.ts +++ b/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); }); }); }); diff --git a/packages/dofs/src/sync/apply.ts b/packages/dofs/src/sync/apply.ts index 3b02b202..35aa76d6 100644 --- a/packages/dofs/src/sync/apply.ts +++ b/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, options: ApplyOptions = {}, ): Promise { - // 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, 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 }; } diff --git a/packages/rpc/tests/wire.test.ts b/packages/rpc/tests/wire.test.ts index 6db78517..58e564c7 100644 --- a/packages/rpc/tests/wire.test.ts +++ b/packages/rpc/tests/wire.test.ts @@ -508,7 +508,7 @@ describe("push semantics — external vs sync peer", () => { } }); - it("push with senderRev>0 (sync peer) advances pushRev to silence loopback", async () => { + it("push with senderRev>0 (sync peer) leaves pushRev for the next push, advances fetchRev to senderRev", async () => { harness = await startHarness(); const { currentRev, readWatermark, writeWatermark } = await import("@cloudflare/dofs"); const client = createSyncClient({ url: harness.url }); @@ -546,11 +546,13 @@ describe("push semantics — external vs sync peer", () => { }), }); - // Loopback suppression kicks in: pushRev was advanced - // to currentRev so the receiver doesn't push these - // entries back to the peer. - const cur = currentRev(harness.db); - expect(readWatermark(harness.db, "pushRev")).toBe(cur); + // pushRev stays where we seeded it: the receiver does + // not advance pushRev locally on upstream apply (that + // was unsound — it desynced our pushRev from the wire- + // visible fetchRev and broke the cross-side invariant + // on the next pull). The next pushOnce ships the + // apply's rev bumps and alreadyApplied() drops them. + expect(readWatermark(harness.db, "pushRev")).toBe(1); // fetchRev was advanced to senderRev. expect(readWatermark(harness.db, "fetchRev")).toBe(42); } finally { From 162f15e2e8d44aade4e6d534ba9970eb70e3b44b Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Mon, 15 Jun 2026 14:33:14 +0000 Subject: [PATCH 2/2] 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. --- packages/rpc/src/server.ts | 11 +- packages/rpc/src/sync-driver.test.ts | 149 ++++++++++++++++++++++----- packages/rpc/src/sync-driver.ts | 98 ++++++++++++++++-- 3 files changed, 216 insertions(+), 42 deletions(-) diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index d7f12ac8..b77484f5 100644 --- a/packages/rpc/src/server.ts +++ b/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 diff --git a/packages/rpc/src/sync-driver.test.ts b/packages/rpc/src/sync-driver.test.ts index 472bc55f..4fc701a7 100644 --- a/packages/rpc/src/sync-driver.test.ts +++ b/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 }) => { @@ -427,36 +447,80 @@ 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 }; }; } @@ -464,7 +528,9 @@ describe("sync driver — cross-side invariant", () => { }, }) 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 { diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index 355d5f29..45943679 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -98,8 +98,27 @@ export async function pullOnce( remote: SyncRPC, backend?: string, ): Promise { + // 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 { // 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; }