rpc,dofs: Track fetch cursors by rev and path

Large directory renames can produce thousands of entries at one
revision. A scalar fetch watermark can only resume at rev boundaries,
so a crash in the middle of one of those streams forces the next pull
to replay the whole rev.

Store fetch progress as a rev/path cursor and checkpoint committed
batches inside a rev. fetchChanges now advertises a snapshot cursor
and streams only entries at or before that cursor, which keeps retry
behavior deterministic while materialized entries read current data.

This changes the RPC fetch shape from scalar revs to cursors. The
durable object and wsd are deployed as a matched pair, so the protocol
is updated in lockstep rather than negotiated across mixed versions.

Author: ndisidore

docs/02_sync_protocol.md

@@ -54,10 +54,10 @@ A typical `exec()` round-trip:
    [06. Mount Interface](./06_mount_interface.md).
 3. **Exec.** The command runs. FUSE writes are captured by the
    in-container VFS as they happen, each stamped with a fresh revision.
-4. **Fetch.** The DO calls `fetchChanges({ sinceRev: fetchRev })`. The
-   container streams `ChangeEntry` records — one per touched path,
-   per-file entries carrying `chunks: (hash, size)[]`. No bytes
-   inline.
+4. **Fetch.** The DO calls `fetchChanges({ after: fetchCursor })`.
+   The container streams `ChangeEntry` records after that `(rev, path)`
+   cursor — one per touched path, per-file entries carrying
+   `chunks: (hash, size)[]`. No bytes inline.
 5. **Diff.** The DO reads up to `PULL_BATCH_SIZE` (256) entries from the
    stream, unions the chunk hashes referenced by that batch, probes
    its own `vfs_blobs` for which it already has, and calls
@@ -69,13 +69,13 @@ A typical `exec()` round-trip:
    `transactionSync` inside `writeFile`/`mkdir`/`rm`/`symlink` is the
    real durability boundary. The driver then loops back to step 5 for
    the next batch.
-   `fetchRev` is advanced **per committed batch** to the max `rev`
-   any entry in that batch carried. `coalesceChanges` emits entries
-   in ascending rev order so this checkpoint is safe — everything
-   below `batchMaxRev` has been applied. A crash mid-pull resumes
-   from the last per-batch advance, so re-fetched work is bounded
-   by `PULL_BATCH_SIZE` (256) entries, not the whole stream. The
-   receiver's `alreadyApplied` check inside `applyChanges` still
+   The fetch cursor is advanced **per committed batch** to the last
+   streamed entry's `(rev, path)`. `coalesceChanges` emits entries in
+   ascending `rev`, then ascending `path`, so this checkpoint is safe
+   even when one rev contains more than one batch. A crash mid-pull
+   resumes from the last per-batch advance, so re-fetched work is
+   bounded by `PULL_BATCH_SIZE` (256) entries, not the whole stream.
+   The receiver's `alreadyApplied` check inside `applyChanges` still
    drops already-applied entries on the floor so re-apply is
    idempotent and cheap.
 
@@ -105,6 +105,56 @@ applies the upstream entry. This is last-writer-wins conflict handling:
 it converges the tree, but local-only children under the conflicting
 path are discarded without separate tombstones.
 
+## Alternatives considered
+
+Representing a rename as a full-subtree restamp produces one wire entry
+per subtree item at a single revision. Two cheaper encodings were
+considered and rejected.
+
+### A rename opcode
+
+A dedicated `rename` entry carrying `{ fromPath, toPath, inode }` would
+collapse a directory move to one wire row. It was rejected because it is
+an operation, while the rest of the protocol is state-based:
+`materialiseChange` resolves each entry to the path's current state at
+fetch time, the receiver reconciles against its own live state, and
+`alreadyApplied` makes re-apply idempotent without ordered replay.
+
+An opcode breaks that model in three ways. It is relative to the
+receiver's prior state: `relink(from -> to)` is meaningless to a peer
+that never held `from`, so a cold-start peer pulling from rev 0 has
+nothing to relink. The pull path is deliberately receiver-history
+agnostic: the producer answers "changes after cursor X" by
+materialising current state and knows nothing about what a given
+receiver has seen, so it cannot decide when an opcode is safe to emit.
+And making the opcode idempotent against final state requires
+re-deriving the same state reconciliation the opcode was meant to avoid,
+while still not solving cold start. The per-subtree cost is the price of
+keeping one state-based representation that bootstraps, converges, and
+replays under a single rule.
+
+### Chunking a rename across revisions
+
+A scalar fetch watermark can only resume at revision boundaries, so a
+large rename at one rev forces a crash to replay the whole rev. One way
+to bound that without a path cursor is to split a single rename across
+many revisions, so a scalar watermark resumes at a chunk boundary.
+
+This was rejected because it weakens an invariant the protocol relies
+on: `rev` is bumped atomically once per mutation (see
+[03. Filesystem Schema](./03_filesystem_schema.md)), so every `rev`
+value names one committed, point-in-time snapshot of the tree.
+`currentCursor` is built on that meaning, and the same meaning keeps
+room for snapshot reads at an arbitrary `rev`. Chunking would mint
+intermediate revisions that never committed as a whole, leaving most
+`rev` values describing tree states that never existed.
+
+The `(rev, path)` fetch cursor avoids that. `path` is an orthogonal
+second coordinate: "within committed snapshot `rev`, consumed up to
+`path`." A rename still stamps exactly one revision across its subtree,
+while a crash can resume mid-rev. Resumability is bought without
+weakening what a revision means.
+
 ### Chunking
 
 Files are split at a fixed `CHUNK_SIZE` (512 KiB). Chunk boundaries are
@@ -127,10 +177,10 @@ one name.
 | Watermark | Owner | Meaning |
 | --- | --- | --- |
 | `pushRev` | DO | Last DO-side `rev` successfully pushed to the container. |
-| `fetchRev` | DO | Last container-side `rev` the DO has fetched. |
+| `fetchCursor` | DO | Last container-side cursor the DO has fetched; `path = null` means the whole rev is complete. |
 | `currentRev` | DO | Latest `rev` stamped on a DO-side mutation. |
 | `currentRev` | Container | Latest `rev` stamped on a container-side mutation. |
-| `appliedPushRev` | Container | Largest DO `rev` the container has fully applied. Echoed on every **push** response. |
+| `appliedPushCursor` | Container | DO-side cursor the container has applied. Echoed on every **push** and **fetchChanges** response. |
 
 The DO watermarks live in the `_vfs_watermark` table so they survive DO
 restarts. The container's watermarks live in the same `Database`
@@ -144,15 +194,14 @@ fresh receiver).
 ### Cross-side invariant
 
 After every successful `push` **and** every `fetchChanges`, the
-response carries the receiver's current `appliedPushRev` (the
-largest `senderRev` it has fully applied). The DO asserts
-`appliedPushRev >= pushRev` before continuing. The two sides never
-share a single clock, but echoing the largest applied rev makes the
-"receiver is caught up with our pushes" invariant inspectable on
-the wire instead of load-bearing in-process state. A regression in
-the post-apply `pushRev` advancement path (see step 1 above) trips
-the assertion on the next push or pull rather than corrupting data
-silently.
+response carries the receiver's current `appliedPushCursor`. The DO
+asserts that cursor covers its local `{ rev: pushRev, path: null }`
+before continuing. The two sides never share a single clock, but
+echoing the applied cursor makes the "receiver is caught up with our
+pushes" invariant inspectable on the wire instead of load-bearing
+in-process state. A regression in the post-apply cursor advancement
+path trips the assertion on the next push or pull rather than
+corrupting data silently.
 
 ## Wire shape
 
@@ -161,10 +210,14 @@ records, both probe with `hasObjects`, both transfer bytes by hash.
 Naming follows git's vocabulary — the DO *pushes* entries and
 objects to the container, and *fetches* entries and objects back.
 
+The DO and `wsd` are deployed as a matched pair. The protocol has no
+version negotiation, so changes to request or response shapes are hard
+wire breaks and require lockstep rollout.
+
 | RPC | Direction | Returns | Notes |
 | --- | --- | --- | --- |
-| `push({ senderRev, changes })` | DO → container | `{ rev, appliedPushRev }` | Streams a coalesced batch of `ChangeEntry` via the `changes` `ReadableStream`. The sender then calls `hasObjects` on the referenced hashes and follows up with `pushObjects` for the missing subset. See the `senderRev` branches below. |
-| `fetchChanges({ sinceRev?, ignore? })` | container → DO | `Promise<{ currentRev, appliedPushRev, stream: ReadableStream<ChangeEntry> }>` | Streams one entry per touched path. For files, `chunks: (hash, size)[]` (no bytes inline); for dirs, metadata; for deletes, a tombstone. `currentRev` is the receiver's rev at stream open; the puller advances `fetchRev` no further than this. `appliedPushRev` carries the cross-side invariant check on the pull path. |
+| `push({ senderRev, changes })` | DO → container | `{ rev, appliedPushCursor }` | Streams a coalesced batch of `ChangeEntry` via the `changes` `ReadableStream`. The sender then calls `hasObjects` on the referenced hashes and follows up with `pushObjects` for the missing subset. See the `senderRev` branches below. |
+| `fetchChanges({ after?, ignore? })` | container → DO | `Promise<{ currentCursor, appliedPushCursor, stream: ReadableStream<ChangeEntry> }>` | Streams one entry per touched path after `after`, ordered by `rev` then `path`. For files, `chunks: (hash, size)[]` (no bytes inline); for dirs, metadata; for deletes, a tombstone. `currentCursor` is `{ rev: currentRev, path: null }` at stream open; the puller writes it after a clean drain. `appliedPushCursor` carries the cross-side invariant check on the pull path. |
 | `hasObjects(hashes[])` | sender probes receiver | `Uint8Array[]` | Returns the subset of the input the receiver already holds. The git `have` line, batched. |
 | `fetchObjects(hashes[])` | container → DO | `ReadableStream<{ hash, bytes }>` | Streams chunk bytes by hash. The git `want`/pack response on the fetch path. |
 | `pushObjects(objects)` | DO → container | `void` | Streams chunk bytes by hash. The push-direction mirror of `fetchObjects`. |
@@ -177,7 +230,8 @@ load-test rationale):
 
 - **`senderRev > 0` — sync peer.** A DO calling its container counterpart
   (or vice versa). The receiver applies the batch as `upstream`,
-  advances its own `fetchRev` to `senderRev`, and on the *sender's*
+  advances its own fetch cursor to `{ rev: senderRev, path: null }`,
+  and on the *sender's*
   side `pushRev` is advanced past the rev just shipped (gated on no
   interleaved local writes — see step 1 above).
 - **`senderRev === 0` — external writer / fresh receiver.** Used by
@@ -197,7 +251,7 @@ edited file) shows up exactly once on the wire. See
 - **Container restart mid-exec.** The DO's connection detects the
   closed WebSocket and self-destructs. The next call transparently
   rebuilds against the still-running `wsd` (or restarts it if needed).
-  `pushRev` and `fetchRev` mean the catch-up is incremental, modulo
+  `pushRev` and the fetch cursor mean the catch-up is incremental, modulo
   whatever the container's deployment chose for its DB lifetime.
 - **Container crash mid-apply.** `push` is atomic from the DO's
   perspective on the receiver: the server wraps the whole batch in a
@@ -209,11 +263,12 @@ edited file) shows up exactly once on the wire. See
   applied so far; the receiver never sees a partial push. The pull
   path keeps the per-mutation model because the streaming batches
   can't hold a synchronous transaction across network I/O.
-- **DO restart mid-pull.** `fetchRev` advances per committed batch
-  to the max `rev` the batch carried, so a restart mid-pull resumes
-  from the last per-batch checkpoint. Wasted work is bounded by
-  `PULL_BATCH_SIZE` entries (256), not the whole stream. End state is
-  correct either way — apply is idempotent.
+- **DO restart mid-pull.** The fetch cursor advances per committed
+  batch to the last entry's `(rev, path)`, so a restart mid-pull
+  resumes from the last per-batch checkpoint, including within a
+  single large rev. Wasted work is bounded by `PULL_BATCH_SIZE`
+  entries (256), not the whole stream. End state is correct either
+  way — apply is idempotent.
 - **DO restart.** Watermarks are persisted, so the new DO instance
   picks up where the old one left off. The container keeps `wsd`
   alive across the gap.

docs/03_filesystem_schema.md

@@ -70,10 +70,9 @@ positional read primitive can read it directly instead of running
 `SUM(size) FROM vfs_chunks` on every call. Every write path stamps
 it alongside `mode`/`mtime`/`rev`.
 
-The `vfs_nodes_by_rev` index supports `coalesceChanges`'s
-`WHERE rev > sinceRev` scan over live inodes, which the sync protocol
-calls once per pull to enumerate everything modified since the last
-fetch watermark.
+The `vfs_nodes_by_rev` index supports `coalesceChanges`'s cursor scan
+over live inodes, which the sync protocol calls once per pull to
+enumerate everything modified after the last fetch cursor.
 
 There is no `ignored` column: ignored paths are entirely invisible to
 the DO-side filesystem API (see
@@ -218,6 +217,20 @@ Stores `pushRev` and `fetchRev` (see
 [02. Sync Protocol](./02_sync_protocol.md#watermarks)). Survives DO
 restarts so reconnects resume cleanly.
 
+### `_vfs_fetch_cursor` — fetch path tie-breaker
+
+```sql
+CREATE TABLE _vfs_fetch_cursor (
+  k    TEXT PRIMARY KEY CHECK(k = 'fetch'),
+  path TEXT
+);
+```
+
+Stores the path component for the fetch cursor. The numeric rev remains
+in `_vfs_watermark.fetchRev`; `path = NULL` means the whole rev has
+been applied. When `path` is not null, only revs before
+`_vfs_watermark.fetchRev` are fully complete.
+
 ### `_vfs_mounts` — mount index state
 
 ```sql

docs/06_mount_interface.md

@@ -284,7 +284,7 @@ This is deliberate, not an oversight:
 - R2 and GitHub have no monotonic rev clock. Treating them as peers
   would force per-tick polling and a diff against a remembered
   snapshot.
-- The protocol's invariants — `appliedPushRev`, watermark
+- The protocol's invariants — `appliedPushCursor`, watermark
   reconciliation, tombstones — assume one peer. They don't generalize
   to N peers without a real CRDT / LWW story.
 - The "container always wins" conflict policy is a deliberate

docs/08_capnweb_interface.md

@@ -58,34 +58,39 @@ interface SyncRPC {
   // not inline: the DO sends ChangeEntry records with chunk hashes,
   // the receiver calls back via hasObjects / pushObjects for the
   // missing subset. Returns the receiver's new rev plus the
-  // appliedPushRev it stamped for this batch.
+  // appliedPushCursor it stamped for this batch.
   push(input: {
     senderRev: number;
     changes:   ReadableStream<ChangeEntry>;
-  }): Promise<{ rev: number; appliedPushRev: number }>;
+  }): Promise<{
+    rev:               number;
+    appliedPushCursor: { rev: number; path: string | null };
+  }>;
 
-  // Container ← DO. Stream every ChangeEntry with rev > sinceRev,
-  // alongside the receiver's currentRev (cursor the puller advances
-  // fetchRev to) and appliedPushRev (cross-side invariant check on
-  // the pull path, mirroring the push response). Per-file entries
-  // carry (hash, size) chunk lists; no bytes inline.
+  // Container ← DO. Stream every ChangeEntry after `after`, ordered
+  // by rev then path. A cursor with path=null means the whole rev is
+  // complete; a cursor with path set resumes after that path inside
+  // the same rev. `currentCursor` is the receiver's currentRev at
+  // stream open with path=null, and `appliedPushCursor` is the
+  // receiver's cursor for sender changes it has applied. Per-file
+  // entries carry (hash, size) chunk lists; no bytes inline.
   fetchChanges(input: {
-    sinceRev?: number;
-    ignore?:   string[];
+    after?:  { rev: number; path: string | null };
+    ignore?: string[];
   }): Promise<{
-    currentRev:     number;
-    appliedPushRev: number;
+    currentCursor:  { rev: number; path: string | null };
+    appliedPushCursor: { rev: number; path: string | null };
     stream:         ReadableStream<ChangeEntry>;
   }>;
 
   // Diagnostic surface for soak tests, dashboards, and the agent
   // when it wants to wait for the wire to drain. pushRev /
-  // fetchRev only move when the receiver is acting as a sync
-  // peer; otherwise they sit at 0.
+  // fetchCursor only move when the receiver is acting as a sync
+  // peer; otherwise they sit at 0 / { rev: 0, path: null }.
   watermarks(): Promise<{
     currentRev: number;
     pushRev:    number;
-    fetchRev:   number;
+    fetchCursor: { rev: number; path: string | null };
   }>;
 
   // Materialise a single path as a ChangeEntry without driving
@@ -116,26 +121,32 @@ interface SyncRPC {
 }
 ```
 
+The durable object and `wsd` are deployed as a matched pair. This
+interface has no version negotiation, so request and response shape
+changes are hard wire breaks and require lockstep rollout.
+
 `ChangeEntry` is defined in `packages/dofs/src/sync/changes.ts`. Schema
 column references match [03. Filesystem Schema](./03_filesystem_schema.md).
 
 #### Rev-0 baseline (no separate snapshot)
 
 There is no dedicated `snapshot()` RPC. A fresh DO with no watermark
-calls `fetchChanges({ sinceRev: 0 })`, which streams every live entry
-plus any tombstones the receiver has retained. Treating the baseline as
-a degenerate fetch keeps the wire shape minimal: the same pull path
-covers both cold-start replication and incremental catch-up.
+calls `fetchChanges({ after: { rev: 0, path: null } })`, which streams
+every live entry plus any tombstones the receiver has retained.
+Treating the baseline as a degenerate fetch keeps the wire shape
+minimal: the same pull path covers both cold-start replication and
+incremental catch-up.
 
 #### Push semantics: peer vs external
 
 `push` distinguishes two callers via `senderRev`:
 
 - **`senderRev > 0` — sync peer.** The sender is replicating its own
-  log forward. The receiver advances `fetchRev` to `senderRev` once
-  the batch settles, echoes it back as `appliedPushRev`, and uses
-  that value to silence the loopback (the next `fetchChanges` from
-  the peer won't replay these entries back at it).
+  log forward. The receiver advances its fetch cursor to
+  `{ rev: senderRev, path: null }` once the batch settles, echoes it
+  back as `appliedPushCursor`, and uses that value to silence the
+  loopback (the next `fetchChanges` from the peer won't replay these
+  entries back at it).
 - **`senderRev === 0` — external orchestrator.** The sender doesn't
   have a rev space of its own (an agent, a CI script, a one-shot
   writer). The receiver applies the batch as ordinary local writes,
@@ -199,19 +210,19 @@ ends:
   streaming `ChangeEntry` records, the container calls `hasObjects` on
   the chunk hashes referenced, the DO follows up with `pushObjects`
   (itself a stream) for the missing subset, the container applies the
-  batch and returns `{ rev, appliedPushRev }`.
-- **Fetch (container → DO).** The DO calls `fetchChanges({ sinceRev })`,
-  which returns `{ currentRev, appliedPushRev, stream }` in one round-
-  trip. `currentRev` is the target watermark; `appliedPushRev`
+  batch and returns `{ rev, appliedPushCursor }`.
+- **Fetch (container → DO).** The DO calls `fetchChanges({ after })`,
+  which returns `{ currentCursor, appliedPushCursor, stream }` in one round-
+  trip. `currentCursor` is the target cursor; `appliedPushCursor`
   carries the cross-side invariant (the puller asserts it covers
-  the local `pushRev` before draining). The DO then streams
+  `{ rev: pushRev, path: null }` before draining). The DO then streams
   `ChangeEntry` records, accumulates chunk hashes, calls `hasObjects`
   on itself (cheap, local) to find what it already has, then calls
   `fetchObjects` for the rest.
 
 | Aspect | Value |
 | --- | --- |
-| Round-trips per fetch | 1 streaming `fetchChanges` (carries `currentRev` + `appliedPushRev` + entry stream) + 1 `hasObjects` per batch + 1 streaming `fetchObjects` per batch (only if any hashes are missing) |
+| Round-trips per fetch | 1 streaming `fetchChanges` (carries `currentCursor` + `appliedPushCursor` + entry stream) + 1 `hasObjects` per batch + 1 streaming `fetchObjects` per batch (only if any hashes are missing) |
 | Round-trips per push | 1 streaming `push` (carries `senderRev`) + 1 `hasObjects` (server-driven) + 1 streaming `pushObjects` (only if any hashes are missing) |
 | Bytes inline in `ChangeEntry` | None — entries carry chunk hashes only |
 | Object transfer shape | `ReadableStream<{ hash, bytes }>` in both directions |