dofs: Make fetch progress cursor-only

Fetch progress is now a rev/path cursor, but the exported watermark
helper still accepted scalar fetchRev writes. That left two public
write paths for one logical cursor.

Restrict the scalar watermark API to pushRev, move fetch progress
callers to readFetchCursor and writeFetchCursor, and normalize
equal-rev partial cursors when a peer push proves the full rev was
applied.

Author: ndisidore

docs/11_lifecycle.md

@@ -335,10 +335,12 @@ Two things have to change for capnweb + hibernation to work:
      wake as a fresh session. The peer must retry any in-flight RPC.
      This is the same semantics as a transport reset, which the
      protocol already handles via the rev cursors.
-   - **Sync streams: nothing to store.** `pushRev` and the durable
-     fetch cursor are already written to SQLite alongside the data
-     they describe. On wake, the next `pushOnce` / `pullOnce` reads
-     them from durable storage and resumes. No attachment write is
+   - **Sync streams: nothing to store.** `pushRev` is written to
+     SQLite with the pushed data it describes. The durable fetch
+     cursor is written after each committed pull batch, not in the
+     same transaction as the data apply. On wake, the next `pushOnce`
+     / `pullOnce` reads the durable counters and resumes; any overlap
+     is dropped by the idempotent apply path. No attachment write is
      required.
    - **Exec streams: store `{ [id]: seq }` per in-flight exec.**
      The `WorkspaceShell` driver inside the DO is the only place

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

@@ -15,38 +15,22 @@ describe("watermarks", () => {
   it("readWatermark returns 0 for a fresh DB", async () => {
     await withDB(async (db) => {
       expect(readWatermark(db, "pushRev")).toBe(0);
-      expect(readWatermark(db, "fetchRev")).toBe(0);
     });
   });
 
   it("writeWatermark persists across reads", async () => {
     await withDB(async (db) => {
       writeWatermark(db, "pushRev", 42);
       expect(readWatermark(db, "pushRev")).toBe(42);
-      expect(readWatermark(db, "fetchRev")).toBe(0);
-      writeWatermark(db, "fetchRev", 7);
-      expect(readWatermark(db, "fetchRev")).toBe(7);
     });
   });
 
   it("persists the fetch cursor rev and path separately", async () => {
     await withDB(async (db) => {
       expect(readFetchCursor(db)).toEqual({ rev: 0, path: null });
       writeFetchCursor(db, { rev: 12, path: "/dir/file.txt" });
-      expect(readWatermark(db, "fetchRev")).toBe(12);
       expect(readFetchCursor(db)).toEqual({ rev: 12, path: "/dir/file.txt" });
       writeFetchCursor(db, { rev: 13, path: null });
-      expect(readWatermark(db, "fetchRev")).toBe(13);
-      expect(readFetchCursor(db)).toEqual({ rev: 13, path: null });
-    });
-  });
-
-  it("clears a stale fetch cursor path when writing fetchRev directly", async () => {
-    await withDB(async (db) => {
-      writeFetchCursor(db, { rev: 12, path: "/z.txt" });
-
-      writeWatermark(db, "fetchRev", 13);
-
       expect(readFetchCursor(db)).toEqual({ rev: 13, path: null });
     });
   });
@@ -70,23 +54,6 @@ describe("watermarks", () => {
     });
   });
 
-  it("does not persist an intermediate cursor when a direct fetchRev write fails", async () => {
-    await withDB(async (db) => {
-      writeFetchCursor(db, { rev: 12, path: "/old.txt" });
-
-      const originalRun = db.run.bind(db);
-      db.run = ((query: string, ...bindings: unknown[]) => {
-        if (query.includes("_vfs_fetch_cursor")) {
-          throw new Error("forced cursor path clear failure");
-        }
-        return originalRun(query, ...bindings);
-      }) as typeof db.run;
-
-      expect(() => writeWatermark(db, "fetchRev", 13)).toThrow("forced cursor path clear failure");
-      expect(readFetchCursor(db)).toEqual({ rev: 12, path: "/old.txt" });
-    });
-  });
-
   it("returns a fresh start cursor for a zero fetchRev", async () => {
     await withDB(async (db) => {
       const cursor = readFetchCursor(db);
@@ -136,9 +103,8 @@ describe("watermarks", () => {
   });
 
   it("rejects unknown watermark keys at the type level via the helper signature", () => {
-    // Compile-time only: writeWatermark only accepts the union
-    // "pushRev" | "fetchRev". This test is a placeholder that
-    // documents the contract; the type system catches misuse.
+    // Compile-time only: writeWatermark only accepts "pushRev".
+    // Fetch progress must go through readFetchCursor/writeFetchCursor.
     expect(true).toBe(true);
   });
 

packages/dofs/src/sync/watermarks.ts

@@ -5,20 +5,24 @@ import type { Database } from "../storage.js";
 // own sync cursors. The container's appliedPushRev lives in-memory on
 // the container side; we don't store it here.
 //
-// pushRev   — last DO-side rev successfully pushed to the backend.
-// fetchRev  — last backend-side rev the DO has fetched and applied.
+// pushRev — last DO-side rev successfully pushed to the backend.
 //
-// initializeSchema() seeds both at 0 in _vfs_watermark for the
+// initializeSchema() seeds pushRev at 0 in _vfs_watermark for the
 // default backend. The schema table is the durability surface;
 // readers and writers always go through this module so the SQL
 // stays in one place.
 //
 // `backend` defaults to DEFAULT_BACKEND_ID so older callers that
 // only ran one backend (or ran the package against a schema before
-// per-backend keying landed) keep working unchanged. The v2 → v3
+// per-backend keying landed) keep working unchanged. The v3 → v4
 // schema migration backfills the column on existing rows with the
 // same default.
-export type WatermarkKey = "pushRev" | "fetchRev";
+//
+// Fetch progress is a `{ rev, path }` cursor. Its rev component is
+// still stored in _vfs_watermark for schema compatibility, but callers
+// must use readFetchCursor() / writeFetchCursor() so rev and path stay
+// consistent.
+export type WatermarkKey = "pushRev";
 
 export const DEFAULT_BACKEND_ID = "default";
 
@@ -39,9 +43,19 @@ export function readWatermark(
   );
 }
 
+function readFetchRev(db: Database, backend: string = DEFAULT_BACKEND_ID): number {
+  return (
+    db.scalar<number>(
+      "SELECT v FROM _vfs_watermark WHERE k = ? AND backend = ?",
+      "fetchRev",
+      backend,
+    ) ?? 0
+  );
+}
+
 function writeWatermarkValue(
   db: Database,
-  key: WatermarkKey,
+  key: WatermarkKey | "fetchRev",
   value: number,
   backend: string = DEFAULT_BACKEND_ID,
 ): void {
@@ -74,19 +88,11 @@ export function writeWatermark(
   value: number,
   backend: string = DEFAULT_BACKEND_ID,
 ): void {
-  if (key !== "fetchRev") {
-    writeWatermarkValue(db, key, value, backend);
-    return;
-  }
-
-  db.transactionSync(() => {
-    writeWatermarkValue(db, key, value, backend);
-    writeFetchCursorPath(db, null, backend);
-  });
+  writeWatermarkValue(db, key, value, backend);
 }
 
 export function readFetchCursor(db: Database, backend: string = DEFAULT_BACKEND_ID): ChangeCursor {
-  const rev = readWatermark(db, "fetchRev", backend);
+  const rev = readFetchRev(db, backend);
   if (rev === 0) return { rev: 0, path: null };
   const path = db.scalar<string | null>(
     "SELECT path FROM _vfs_fetch_cursor WHERE k = ? AND backend = ?",

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

@@ -16,6 +16,7 @@ import {
   currentRev,
   Database,
   initializeSchema,
+  readFetchCursor,
   readWatermark,
   SQLiteWorkspaceProvider,
 } from "@cloudflare/dofs";
@@ -84,7 +85,7 @@ describe("sync driver — push throughput", () => {
         for (let i = 0; i < bytes.byteLength; i += 4096) {
           bytes[i] = (i * 31) & 0xff;
         }
-        provider.writeFileSync("/big.bin", bytes);
+        provider.writeFileSync("/big.bin", Buffer.from(bytes));
         await pushOnce(a.db, b.rpc);
       } finally {
         a.close();
@@ -105,7 +106,7 @@ describe("sync driver — push throughput", () => {
         for (let i = 0; i < bytes.byteLength; i += 4096) {
           bytes[i] = (i * 31) & 0xff;
         }
-        provider.writeFileSync("/big.bin", bytes);
+        provider.writeFileSync("/big.bin", Buffer.from(bytes));
         await pushOnce(a.db, b.rpc);
       } finally {
         a.close();
@@ -199,7 +200,7 @@ describe("sync driver — bidirectional convergence", () => {
         // time. Reading watermarks here adds noise we can
         // tolerate vs. running a no-op closure for the
         // baseline. Sanity assert outside the iteration body:
-        if (readWatermark(b.db, "fetchRev") <= 0) throw new Error("pull didn't advance");
+        if (readFetchCursor(b.db).rev <= 0) throw new Error("pull didn't advance");
         if (currentRev(b.db) <= 0) throw new Error("apply didn't bump rev");
       } finally {
         a.close();

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

@@ -665,7 +665,7 @@ describe("sync driver — streaming pullOnce", () => {
         get(target, prop, receiver) {
           if (prop === "hasObjects") {
             return async (hashes: Uint8Array[]) => {
-              sampledRevs.push(readWatermark(b.db, "fetchRev"));
+              sampledRevs.push(readFetchCursor(b.db).rev);
               return Reflect.get(target, prop, receiver).call(target, hashes);
             };
           }
@@ -684,7 +684,7 @@ describe("sync driver — streaming pullOnce", () => {
         expect(sampledRevs[i]).toBeGreaterThan(sampledRevs[i - 1]);
       }
       // End state still matches the remote.
-      expect(readWatermark(b.db, "fetchRev")).toBe(remoteFinalRev);
+      expect(readFetchCursor(b.db).rev).toBe(remoteFinalRev);
     } finally {
       a.close();
       b.close();
@@ -798,6 +798,33 @@ describe("sync driver — streaming pullOnce", () => {
       b.close();
     }
   });
+
+  it("lets a peer push supersede a partial pull cursor", async () => {
+    const a = makePeer();
+    const b = makePeer();
+    try {
+      const providerA = new SQLiteWorkspaceProvider(a.db, { now: () => 1 });
+      const providerB = new SQLiteWorkspaceProvider(b.db, { now: () => 1 });
+      providerA.writeFileSync("/before-stale-path.txt", "pushed");
+      const pushedRev = currentRev(a.db);
+
+      writeFetchCursor(b.db, { rev: pushedRev, path: "/zzzz" });
+
+      expect(await pushOnce(a.db, b.rpc)).toBeGreaterThan(0);
+      expect(readFetchCursor(b.db)).toEqual({ rev: pushedRev, path: null });
+      expect(providerB.readFileSync("/before-stale-path.txt", "utf8")).toBe("pushed");
+
+      providerA.writeFileSync("/after-push.txt", "pulled later");
+      const pulled = await pullOnce(b.db, a.rpc);
+
+      expect(pulled.applied).toBe(1);
+      expect(providerB.readFileSync("/after-push.txt", "utf8")).toBe("pulled later");
+      expect(readFetchCursor(b.db)).toEqual({ rev: currentRev(a.db), path: null });
+    } finally {
+      a.close();
+      b.close();
+    }
+  });
 });
 
 describe("sync driver — push atomicity", () => {
@@ -875,11 +902,11 @@ describe("sync driver — reconcileWatermarks", () => {
       // (currentRev = 1 from initializeSchema seeding the root).
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
-      writeWatermark(local, "fetchRev", 42);
+      writeFetchCursor(local, { rev: 42, path: null });
       writeWatermark(local, "pushRev", 0);
 
       await reconcileWatermarks(local, remote.rpc);
-      expect(readWatermark(local, "fetchRev")).toBe(0);
+      expect(readFetchCursor(local)).toEqual({ rev: 0, path: null });
       expect(readWatermark(local, "pushRev")).toBe(0);
     } finally {
       remote.close();
@@ -893,7 +920,7 @@ describe("sync driver — reconcileWatermarks", () => {
       // which is echoed as appliedPushCursor on the wire, is 0/null.
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
-      writeWatermark(local, "fetchRev", 0);
+      writeFetchCursor(local, { rev: 0, path: null });
       writeWatermark(local, "pushRev", 17);
 
       await reconcileWatermarks(local, remote.rpc);
@@ -913,11 +940,11 @@ describe("sync driver — reconcileWatermarks", () => {
       const local = new Database(new SQLiteTestStorage());
       initializeSchema(local, () => 1000);
       const remoteCurrent = currentRev(remote.db);
-      writeWatermark(local, "fetchRev", remoteCurrent);
+      writeFetchCursor(local, { rev: remoteCurrent, path: null });
       writeWatermark(local, "pushRev", 0);
 
       await reconcileWatermarks(local, remote.rpc);
-      expect(readWatermark(local, "fetchRev")).toBe(remoteCurrent);
+      expect(readFetchCursor(local)).toEqual({ rev: remoteCurrent, path: null });
       expect(readWatermark(local, "pushRev")).toBe(0);
     } finally {
       remote.close();

packages/rpc/tests/wire.test.ts

@@ -500,7 +500,7 @@ describe("push semantics — external vs sync peer", () => {
       expect(readWatermark(harness.db, "pushRev")).toBe(0);
       // The fetch cursor was NOT advanced either — the sender has
       // no rev space.
-      expect(readWatermark(harness.db, "fetchRev")).toBe(0);
+      expect(readFetchCursor(harness.db)).toEqual({ rev: 0, path: null });
       // The entry is in the coalesce stream.
       const drained: { path: string }[] = [];
       for await (const e of coalesceChanges(harness.db, { rev: 0, path: null })) drained.push(e);
@@ -512,7 +512,9 @@ describe("push semantics — external vs sync peer", () => {
 
   it("push with senderRev>0 (sync peer) advances pushRev to silence loopback", async () => {
     harness = await startHarness();
-    const { currentRev, readWatermark, writeWatermark } = await import("@cloudflare/dofs");
+    const { currentRev, readFetchCursor, readWatermark, writeWatermark } = await import(
+      "@cloudflare/dofs"
+    );
     const client = createSyncClient({ url: harness.url });
     try {
       // Seed pushRev at the current point so the F1 guard

packages/workspace/src/workspace.test.ts

@@ -517,14 +517,17 @@ describe("Workspace backend selection", () => {
     };
     const storage = makeStorage();
     const ws = new Workspace({ storage, backends: [makeBackend("only", sync)] });
-    const { writeWatermark, readWatermark } = await import("@cloudflare/dofs");
+    // Pre-seed local watermarks for the "only" backend.
+    const { readFetchCursor, readWatermark, writeFetchCursor, writeWatermark } = await import(
+      "@cloudflare/dofs"
+    );
     writeWatermark(ws.db, "pushRev", 17, "only");
-    writeWatermark(ws.db, "fetchRev", 42, "only");
+    writeFetchCursor(ws.db, { rev: 42, path: null }, "only");
     // ready() alone no longer dials; ready(id) forces the connect.
     await ws.ready("only");
     expect(watermarksCalls).toBe(1);
     expect(readWatermark(ws.db, "pushRev", "only")).toBe(0);
-    expect(readWatermark(ws.db, "fetchRev", "only")).toBe(0);
+    expect(readFetchCursor(ws.db, "only")).toEqual({ rev: 0, path: null });
   });
 
   it("skips push/pull when the backend declares sync: 'none'", async () => {