Add support for batch sync to workerd

docs/08_capnweb_interface.md

@@ -86,7 +86,10 @@ interface SyncRPC {
   // when it wants to wait for the wire to drain. pushRev /
   // fetchCursor only move when the receiver is acting as a sync
   // peer; otherwise they sit at 0 / { rev: 0, path: null }.
-  watermarks(): Promise<{
+  // `settle` runs the same disk-to-VFS reconciliation as
+  // fetchChanges before reading currentRev. Deferred command sync
+  // uses it to capture a target that includes the command's writes.
+  watermarks(input?: { settle?: boolean }): Promise<{
     currentRev: number;
     pushRev:    number;
     fetchCursor: { rev: number; path: string | null };

packages/computer/src/index.ts

@@ -15,6 +15,7 @@
 // TestBackend stays on the main entry because it's a thin
 // test-only fake with no payload.
 
+export type { SyncBatchBudget, SyncBatchResult } from "@cloudflare/computer-rpc/driver";
 export type {
   ApplyResult,
   DurableObjectStorageLike,
@@ -96,6 +97,7 @@ export {
   withWorkspace,
 } from "./with-workspace.js";
 export {
+  type SyncBatchOptions,
   type SyncRetryIntent,
   type SyncRetryOptions,
   type SyncRetryScheduler,

packages/computer/src/retry.test.ts

@@ -34,6 +34,7 @@ class MemoryRetryScheduler implements SyncRetryScheduler {
 function retryBackend(options: {
   onExec(): void;
   fetchChanges: import("@cloudflare/computer-rpc").SyncRPC["fetchChanges"];
+  watermarks?: import("@cloudflare/computer-rpc").SyncRPC["watermarks"];
   close?: () => Promise<void>;
 }): WorkspaceBackend {
   const sync: import("@cloudflare/computer-rpc").SyncRPC = {
@@ -50,9 +51,13 @@ function retryBackend(options: {
     fetchObjects() {
       return new ReadableStream({ start: (controller) => controller.close() });
     },
-    async watermarks() {
-      return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } };
-    },
+    watermarks:
+      options.watermarks ??
+      (async () => ({
+        currentRev: 0,
+        pushRev: 0,
+        fetchCursor: { rev: 0, path: null },
+      })),
     async pushObjects() {},
   };
   return {
@@ -304,6 +309,61 @@ describe("Workspace durable pending-sync retries", () => {
     expect(scheduler.cleared).toEqual(["sandbox"]);
   });
 
+  it("does not exhaust retries while bounded batches are making progress", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const entries = Array.from(
+      { length: 6 },
+      (_, index): ChangeEntry => ({
+        kind: "delete",
+        rev: 1,
+        path: `/generated/${index}`,
+        mtime: 1,
+      }),
+    );
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges(input) {
+        const remaining = entries.filter((entry) => {
+          if (!input.after || input.after.rev < entry.rev) return true;
+          return input.after.path !== null && entry.path > input.after.path;
+        });
+        return {
+          currentCursor: { rev: 1, path: null },
+          appliedPushCursor: { rev: 0, path: null },
+          stream: new ReadableStream<ChangeEntry>({
+            start(controller) {
+              for (const entry of remaining) controller.enqueue(entry);
+              controller.close();
+            },
+          }),
+        };
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+      retry: { initialDelayMs: 100, maxDelayMs: 1_000, maxAttempts: 3 },
+      now: () => 5_000,
+    });
+    scheduler.intents.set("sandbox", {
+      backend: "sandbox",
+      targetCursor: { rev: 1, path: null },
+      attempt: 1,
+      notBefore: 0,
+    });
+
+    const outcomes: WorkspaceRetryPendingSyncResult[] = [];
+    for (let i = 0; i < 7; i++) {
+      outcomes.push(await ws.retryPendingSync("sandbox", { maxEntries: 1, maxBytes: 1024 }));
+    }
+
+    expect(outcomes.slice(0, -1).every((result) => result.status === "pending")).toBe(true);
+    expect(outcomes.at(-1)).toMatchObject({ status: "complete" });
+    expect(outcomes.some((result) => result.status === "exhausted")).toBe(false);
+    expect(scheduler.intents.size).toBe(0);
+  });
+
   it("coalesces repeated command failures into one pending intent per backend", async () => {
     const scheduler = new MemoryRetryScheduler();
     const backend = retryBackend({
@@ -396,3 +456,160 @@ describe("Workspace durable pending-sync retries", () => {
     expect(closes).toBe(1);
   });
 });
+
+describe("Workspace deferred synchronization", () => {
+  it("schedules before returning a deferred result", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        return {
+          currentCursor: { rev: 0, path: null },
+          appliedPushCursor: { rev: 0, path: null },
+          stream: new ReadableStream<ChangeEntry>({
+            start(controller) {
+              controller.close();
+            },
+          }),
+        };
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+      now: () => 5_000,
+    });
+
+    const handle = await ws.runtime.exec("build", {
+      encoding: "utf8",
+      sync: "defer",
+    });
+    const result = await handle.result();
+
+    expect(result.sync).toMatchObject({
+      status: "pending",
+      backend: "sandbox",
+      targetCursor: { rev: 0, path: null },
+    });
+    expect(scheduler.intents.get("sandbox")).toMatchObject({
+      backend: "sandbox",
+      targetCursor: { rev: 0, path: null },
+    });
+  });
+
+  it("settles the remote filesystem before capturing the deferred target", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const settleInputs: unknown[] = [];
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        throw new Error("not used");
+      },
+      async watermarks(input) {
+        settleInputs.push(input);
+        return {
+          currentRev: input?.settle === true ? 5 : 0,
+          pushRev: 0,
+          fetchCursor: { rev: 0, path: null },
+        };
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+    });
+
+    const handle = await ws.runtime.exec("build", { sync: "defer" });
+    const result = await handle.result();
+
+    expect(settleInputs.at(-1)).toEqual({ settle: true });
+    expect(result.sync).toMatchObject({ targetCursor: { rev: 5, path: null } });
+    expect(scheduler.intents.get("sandbox")).toMatchObject({
+      targetCursor: { rev: 5, path: null },
+    });
+  });
+
+  it("persists an unfenced intent when target capture fails", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        throw new Error("not used");
+      },
+      async watermarks(input) {
+        if (input?.settle === true) throw new Error("settle failed");
+        return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } };
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+    });
+
+    const handle = await ws.runtime.exec("build", { sync: "defer" });
+    const result = await handle.result();
+
+    expect(result.sync).toMatchObject({
+      status: "pending",
+      error: expect.stringContaining("settle failed"),
+    });
+    expect(scheduler.intents.get("sandbox")).toEqual(
+      expect.objectContaining({ backend: "sandbox", attempt: 1 }),
+    );
+    expect(scheduler.intents.get("sandbox")).not.toHaveProperty("targetCursor");
+  });
+
+  it("widens an existing intent when another deferred command finishes", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    let currentRev = 0;
+    const backend = retryBackend({
+      onExec() {
+        currentRev++;
+      },
+      async fetchChanges() {
+        throw new Error("not used");
+      },
+      async watermarks() {
+        return {
+          currentRev,
+          pushRev: 0,
+          fetchCursor: { rev: 0, path: null },
+        };
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+    });
+
+    const first = await ws.runtime.exec("first", { sync: "defer" });
+    await first.result();
+    const second = await ws.runtime.exec("second", { sync: "defer" });
+    const result = await second.result();
+
+    expect(result.sync).toMatchObject({ targetCursor: { rev: 2, path: null } });
+    expect(scheduler.intents.get("sandbox")).toMatchObject({
+      targetCursor: { rev: 2, path: null },
+    });
+  });
+});
+
+it("rejects deferred execution without a retry scheduler", async () => {
+  let execs = 0;
+  const backend = retryBackend({
+    onExec: () => {
+      execs += 1;
+    },
+    async fetchChanges() {
+      throw new Error("not expected");
+    },
+  });
+  const ws = new Workspace({ storage: new SQLiteTestStorage(), backends: [backend] });
+
+  await expect(ws.runtime.exec("build", { sync: "defer" })).rejects.toThrow("retryScheduler");
+  expect(execs).toBe(0);
+});

packages/computer/src/runtime/runtime.ts

@@ -69,6 +69,7 @@ export class WorkspaceRuntime {
       env: options.env,
       stdin: options.stdin,
       timeoutMs: options.timeoutMs,
+      sync: options.sync,
     });
     return wrapModuleHandle(
       runtime,

packages/computer/src/runtime/types.ts

@@ -110,6 +110,7 @@ export interface WorkspaceRuntimeExecOptions<E extends ExecEncoding = undefined>
   env?: Record<string, string>;
   stdin?: Uint8Array | string;
   timeoutMs?: number;
+  sync?: "wait" | "defer";
 }
 
 export interface WorkspaceRuntimeGetOptions<E extends ExecEncoding = undefined> {
@@ -144,6 +145,7 @@ export interface ModuleExecutionInput {
   env?: Record<string, string>;
   stdin?: Uint8Array | string;
   timeoutMs?: number;
+  sync?: "wait" | "defer";
 }
 
 export interface ModuleExecutionEnvelope {