wsd, rpc: fix shim reverse-direction race via beforeFetch hook

Under FUSE_MOUNT=shim, a write inside an exec'd process didn't
make it back to the DO unless something later triggered a fresh
push/pull cycle. The shim's disk\u2192VFS path is a 250ms poll; the
worker's post-exec pullOnce races against that poll, and on a
fast exec the pull beats the poll \u2014 fetchChanges streams the
empty pre-write state and the file stays stranded in wsd's VFS
forever (or until the next exec drives another bracket).

The push direction already had a symmetric escape hatch:
afterApply on the SyncRPC fires after a peer batch commits and
the shim flushes VFS\u2192disk before the push returns. Add the
matching beforeFetch hook in the rpc server. wsd wires it to
shim.reconcileNow(), a new public method on ShimMount that runs
the same disk\u2192VFS reconcile the poll loop runs, serialised
through the same internal mutex.

With the hook in place, fetchChanges awaits the shim's reconcile
before computing the change set, so a Workspace.pull issued
right after shell.exec sees every file the spawned process
wrote. Verified end-to-end against examples/wsd-container under
wrangler dev (FUSE_MOUNT=auto resolves to shim with no /dev/fuse):
the smoke script's step 4 now passes with SETTLE_SECONDS=0, and
the previously-stranded single-exec scenario reads back the
written file immediately.

beforeFetch is documented to fire on every fetch, including ones
that would otherwise stream zero entries \u2014 the hook is what
produces the entries in the first place. Errors are caught and
logged so a wedged shim can't take down the wire. Test coverage
mirrors the afterApply set: fires once per pull, surfaces hook-
materialised writes, swallows thrown hooks.

The example script drops its post-exec sleep now that the
contract is synchronous.

Author: aron-cf

examples/wsd-container/script/run

@@ -18,7 +18,6 @@ BASE_URL="${BASE_URL%/}"
 NAME="${NAME:-demo}"
 PATH_VIA_API="api-wrote.txt"
 PATH_VIA_EXEC="exec-wrote.txt"
-SETTLE_SECONDS="${SETTLE_SECONDS:-3}"
 
 step() { printf '\n=== %s ===\n' "$*"; }
 fail() { printf '\nFAIL: %s\n' "$*" >&2; exit 1; }
@@ -41,9 +40,9 @@ curl -fsS -X POST "${BASE_URL}/c/${NAME}/exec" \
   -H 'content-type: application/json' \
   -d "{\"command\":\"echo hello-from-exec > /workspace/${PATH_VIA_EXEC} && cat /workspace/${PATH_VIA_EXEC}\",\"encoding\":\"utf8\"}"
 printf '\n'
-# The shim's reverse-direction reconciler polls on a short tick.
-# Sleep a beat so the GET below sees the new bytes the first time.
-sleep "$SETTLE_SECONDS"
+# No settle wait needed: wsd's `beforeFetch` hook runs the shim's
+# disk→VFS reconcile synchronously inside `fetchChanges`, so the
+# GET below is guaranteed to see whatever the exec just wrote.
 
 step "4. read via API: GET /c/${NAME}/file/workspace/${PATH_VIA_EXEC}"
 read_via_api=$(curl -fsS "${BASE_URL}/c/${NAME}/file/workspace/${PATH_VIA_EXEC}")

packages/rpc/src/server.ts

@@ -59,13 +59,28 @@ export interface ServerOptions {
    * push failure.
    */
   afterApply?: () => void | Promise<void>;
+  /**
+   * Optional hook fired inside the SyncRPC `fetchChanges` handler,
+   * right before the receiver computes the change set the puller
+   * will see. Resolved before any entries stream. Used by wsd to
+   * settle the userspace shim's disk→VFS reconcile so a
+   * `Workspace.pull()` issued right after `shell.exec` returns the
+   * files the exec'd process wrote, without waiting on the shim's
+   * periodic poll.
+   *
+   * Fires on every fetch, including ones that would otherwise
+   * stream zero entries — the hook is what produces the entries in
+   * the first place. Errors are caught and logged; a hook failure
+   * must not fail the fetch.
+   */
+  beforeFetch?: () => void | Promise<void>;
 }
 
 class SyncRPCServer extends RpcTarget implements SyncRPC {
   constructor(
     private readonly db: Database,
     private readonly options: Required<Pick<ServerOptions, "ignore">> &
-      Pick<ServerOptions, "afterApply">,
+      Pick<ServerOptions, "afterApply" | "beforeFetch">,
   ) {
     super();
     trackStub(this);
@@ -134,6 +149,16 @@ class SyncRPCServer extends RpcTarget implements SyncRPC {
     appliedPushRev: number;
     stream: ReadableStream<ChangeEntry>;
   }> {
+    if (this.options.beforeFetch !== undefined) {
+      try {
+        await this.options.beforeFetch();
+      } catch (err) {
+        // Settle hook failures must not surface as fetch failures —
+        // we still want to stream whatever's already in the store.
+        // Log so the operator notices a wedged shim, then carry on.
+        console.warn("[SyncRPCServer] beforeFetch hook failed:", err);
+      }
+    }
     const sinceRev = input.sinceRev ?? 0;
     const ignore =
       input.ignore ?? (this.options.ignore.length > 0 ? this.options.ignore : DEFAULT_IGNORE);
@@ -255,7 +280,11 @@ class WorkspaceRPCServer extends RpcTarget implements WorkspaceRPC {
 // back the object to mount on each connection via
 // acceptWebSocketSession().
 export function createSyncServer(db: Database, options: ServerOptions = {}): SyncRPC {
-  return new SyncRPCServer(db, { ignore: options.ignore ?? [], afterApply: options.afterApply });
+  return new SyncRPCServer(db, {
+    ignore: options.ignore ?? [],
+    afterApply: options.afterApply,
+    beforeFetch: options.beforeFetch,
+  });
 }
 
 // Construct a ShellRPC bound to a Runner. wsd holds the only

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

@@ -234,6 +234,107 @@ describe("SyncRPC server — afterApply hook", () => {
   });
 });
 
+describe("SyncRPC server — beforeFetch hook", () => {
+  // Symmetric to the afterApply spy above. beforeFetch runs on the
+  // receiver right before fetchChanges streams entries, giving the
+  // host a chance to settle any out-of-band writes (e.g. wsd's shim
+  // pulling disk changes into the VFS) into the store the fetch is
+  // about to read.
+  function makeReceiverWithSpy(): {
+    db: Database;
+    rpc: SyncRPC;
+    close: () => void;
+    calls: number;
+    setBeforeFetch: (fn: () => void | Promise<void>) => void;
+  } {
+    const storage = new SQLiteTestStorage();
+    const db = new Database(storage);
+    initializeSchema(db, () => 1000);
+    let hook: () => void | Promise<void> = () => {};
+    let calls = 0;
+    const rpc = createSyncServer(db, {
+      beforeFetch: async () => {
+        calls += 1;
+        await hook();
+      },
+    });
+    return {
+      db,
+      rpc,
+      close: () => storage.close(),
+      get calls() {
+        return calls;
+      },
+      setBeforeFetch: (fn) => {
+        hook = fn;
+      },
+    };
+  }
+
+  it("fires once per fetchChanges and is awaited before entries stream", async () => {
+    const a = makePeer();
+    const b = makeReceiverWithSpy();
+    try {
+      // Stage a write in the hook itself — simulates the shim's
+      // disk→VFS reconcile picking up a file that wasn't in the VFS
+      // when pullOnce started.
+      const providerB = new SQLiteWorkspaceProvider(b.db, { now: () => 2 });
+      b.setBeforeFetch(() => {
+        providerB.writeFileSync("/late.txt", "materialised by hook");
+      });
+
+      const result = await pullOnce(a.db, b.rpc);
+      expect(b.calls).toBe(1);
+      // The pull must surface the entry the hook wrote — that's the
+      // whole point of beforeFetch existing.
+      expect(result.applied).toBeGreaterThan(0);
+      expect(fileEntries(a.db)).toContain("late.txt");
+    } finally {
+      a.close();
+      b.close();
+    }
+  });
+
+  it("fires even when there are no changes to stream", async () => {
+    // Pulling against an empty receiver still has to call the hook,
+    // because the hook is what produces "any changes" in the first
+    // place — conditional firing would defeat the contract.
+    const a = makePeer();
+    const b = makeReceiverWithSpy();
+    try {
+      const result = await pullOnce(a.db, b.rpc);
+      expect(b.calls).toBe(1);
+      expect(result.applied).toBe(0);
+    } finally {
+      a.close();
+      b.close();
+    }
+  });
+
+  it("a thrown hook does not fail the fetch", async () => {
+    const a = makePeer();
+    const b = makeReceiverWithSpy();
+    try {
+      const providerB = new SQLiteWorkspaceProvider(b.db, { now: () => 2 });
+      providerB.writeFileSync("/already-there.txt", "x");
+
+      b.setBeforeFetch(() => {
+        throw new Error("reconcile blew up");
+      });
+
+      // The fetch must still succeed and return the pre-existing
+      // entry. The receiver logs and swallows hook errors.
+      const result = await pullOnce(a.db, b.rpc);
+      expect(b.calls).toBe(1);
+      expect(result.applied).toBe(1);
+      expect(fileEntries(a.db)).toContain("already-there.txt");
+    } finally {
+      a.close();
+      b.close();
+    }
+  });
+});
+
 describe("sync driver — bidirectional convergence", () => {
   it("two peers writing in parallel converge after a few ticks", async () => {
     const a = makePeer();

packages/wsd/src/cli/wsd.ts

@@ -419,9 +419,9 @@ async function main(): Promise<void> {
 
   let fuse: FuseMount | undefined;
   // When running on the userspace shim, capture the typed handle
-  // so we can wire `flush()` into the SyncRPC `push` afterApply
-  // hook below. A real FUSE mount serves reads straight from the
-  // VFS, so it doesn't need an explicit settle.
+  // so we can wire `flush()` and `reconcileNow()` into the SyncRPC
+  // afterApply / beforeFetch hooks below. A real FUSE mount serves
+  // reads straight from the VFS, so it doesn't need either settle.
   let shim: ShimMount | undefined;
   if (backend.kind !== "none") {
     // The VFS stores everything under `mountPoint` so capnweb pulls,
@@ -467,7 +467,19 @@ async function main(): Promise<void> {
     // exec()/read against the host fs after a push sees the new
     // files. Real FUSE doesn't need this — the kernel-FUSE driver
     // serves reads from the VFS directly.
-    ...(shim ? { afterApply: () => shim.flush() } : {}),
+    // Symmetric shim settles:
+    //   - afterApply (push side): wait for the VFS→disk flush so
+    //     a subsequent `shell.exec` sees the just-pushed files.
+    //   - beforeFetch (pull side): wait for the disk→VFS reconcile
+    //     so a `Workspace.pull()` issued right after `shell.exec`
+    //     observes files the exec'd process just wrote, without
+    //     waiting on the next periodic poll tick.
+    ...(shim
+      ? {
+          afterApply: () => shim.flush(),
+          beforeFetch: () => shim.reconcileNow(),
+        }
+      : {}),
   });
   const http = createHTTPServer(info, rpc);
 

packages/wsd/src/shim/shim.test.ts

@@ -172,6 +172,54 @@ test("shim.flush() resolves on an unmounted shim without throwing", async (_ctx)
   await shim.flush();
 });
 
+test("shim.reconcileNow() settles disk writes into the VFS before resolving", async (_ctx) => {
+  // Mirror of the flush() test above, in the reverse direction.
+  // A very slow poll guarantees the periodic reconcile can't be
+  // serving the assertion; if reconcileNow() works the file is in
+  // the VFS as soon as the call returns.
+  const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "wsd-shim-reconcile-"));
+  const { vfs } = await createNodeVirtualFileSystem();
+  const shim = await mountShim({ vfs, mountPoint, pollIntervalMs: 60_000 });
+  onTestFinished(async () => {
+    await shim.unmount();
+    await fs.rm(mountPoint, { recursive: true, force: true });
+  });
+
+  await fs.mkdir(path.join(mountPoint, "proj"), { recursive: true });
+  await fs.writeFile(path.join(mountPoint, "proj", "x.txt"), "from disk");
+  await fs.writeFile(path.join(mountPoint, "proj", "y.txt"), "also from disk");
+
+  // No periodic reconcile has fired yet — the VFS is empty until
+  // reconcileNow() walks the disk.
+  expect(vfs.existsSync(`${mountPoint}/proj/x.txt`)).toBe(false);
+
+  await shim.reconcileNow();
+
+  expect(vfs.readFileSync(`${mountPoint}/proj/x.txt`).toString()).toBe("from disk");
+  expect(vfs.readFileSync(`${mountPoint}/proj/y.txt`).toString()).toBe("also from disk");
+});
+
+test("shim.reconcileNow() is idempotent and cheap on a clean tree", async (_ctx) => {
+  const { vfs, mountPoint, shim } = await setup();
+  await fs.writeFile(path.join(mountPoint, "stable.txt"), "steady");
+  await shim.reconcileNow();
+  const rev1 = vfs.statSync(`${mountPoint}/stable.txt`).mtime.getTime();
+  await shim.reconcileNow();
+  const rev2 = vfs.statSync(`${mountPoint}/stable.txt`).mtime.getTime();
+  expect(rev2).toBe(rev1, "reconcileNow on an unchanged tree should not bump VFS mtime");
+});
+
+test("shim.reconcileNow() resolves on an unmounted shim without throwing", async (_ctx) => {
+  const mountPoint = await fs.mkdtemp(path.join(os.tmpdir(), "wsd-shim-reconcile-unmount-"));
+  const { vfs } = await createNodeVirtualFileSystem();
+  const shim = await mountShim({ vfs, mountPoint, pollIntervalMs: TICK_MS });
+  onTestFinished(async () => {
+    await fs.rm(mountPoint, { recursive: true, force: true });
+  });
+  await shim.unmount();
+  await shim.reconcileNow();
+});
+
 test("shim drops VFS writes outside the mount point", async (_ctx) => {
   // Pin the cross-namespace contract that backed the original bug:
   // a write into the VFS at `${mountPoint}/foo` lands on disk at

packages/wsd/src/shim/shim.ts

@@ -50,6 +50,19 @@ export interface ShimMount {
    * so flushing twice in a row is cheap.
    */
   flush(): Promise<void>;
+  /**
+   * Block until the VFS reflects the on-disk tree's current state.
+   * Called by the SyncRPC `fetchChanges` handler right before it
+   * computes the change set the puller will see, so a
+   * `Workspace.pull()` issued after `shell.exec` returns observes
+   * files the exec'd process wrote without waiting on the next
+   * periodic poll tick.
+   *
+   * Runs the same disk→VFS reconcile the polling loop runs,
+   * serialised through the same internal mutex so a request-time
+   * call can't race with the tick. Idempotent on a clean tree.
+   */
+  reconcileNow(): Promise<void>;
 }
 
 export interface MountShimOptions {
@@ -155,9 +168,13 @@ export async function mountShim(options: MountShimOptions): Promise<ShimMount> {
 
   // Disk -> VFS via periodic reconcile. Walks the mount point,
   // diffs against the shadow, applies changes to the VFS.
+  // Shared by the periodic poll below and the on-demand
+  // reconcileNow() hook; both go through `run` so they serialise
+  // against the VFS watcher loop.
+  const reconcile = (): Promise<void> => run(() => reconcileDiskToVfs(vfs, mountPoint, shadow));
   const pollTimer = setInterval(() => {
     if (stopped) return;
-    void run(() => reconcileDiskToVfs(vfs, mountPoint, shadow)).catch((error) => {
+    void reconcile().catch((error) => {
       console.error("[shim] disk reconcile failed:", error);
     });
   }, pollIntervalMs);
@@ -190,6 +207,10 @@ export async function mountShim(options: MountShimOptions): Promise<ShimMount> {
         await flushVfsToDisk(vfs, mountPoint, shadow);
       });
     },
+    async reconcileNow(): Promise<void> {
+      if (stopped) return;
+      await reconcile();
+    },
   };
 }