workspace: Add durable sync retries

Let Durable Object hosts persist one pending pull intent per backend and wake Workspace through retryPendingSync. Failed attempts use bounded exponential backoff while exhausted intents remain available for inspection.\n\nCover scheduling, coalescing, cursor resume, convergence, exhaustion, and RPC cleanup with an in-memory scheduler.

Author: mattzcarey

packages/workspace/README.md

@@ -239,6 +239,98 @@ A workspace with two backends that both write into
 [`docs/05_shell_interface.md`](../../docs/05_shell_interface.md)
 for the caveat.
 
+## Durable pending-sync retries
+
+A command can finish after changing backend files while its post-command pull
+fails. The command result reports `sync.status: "pending"`. If you configure a
+`SyncRetryScheduler`, Workspace also writes one durable retry intent for that
+backend. The library does not use an in-memory timer and cannot own the host
+Durable Object's alarm.
+
+The scheduler contract contains only values that a host can persist:
+
+```ts
+import {
+  type SyncRetryIntent,
+  type SyncRetryScheduler,
+  Workspace,
+} from "@cloudflare/workspace";
+
+const RETRY_PREFIX = "workspace:sync-retry:";
+
+class DurableObjectRetryScheduler implements SyncRetryScheduler {
+  constructor(private readonly state: DurableObjectState) {}
+
+  async get(backend: string): Promise<SyncRetryIntent | undefined> {
+    return this.state.storage.get(`${RETRY_PREFIX}${backend}`);
+  }
+
+  async schedule(intent: SyncRetryIntent): Promise<void> {
+    // schedule replaces the backend's existing intent, so repeated
+    // failures coalesce instead of creating an alarm queue.
+    await this.state.storage.put(`${RETRY_PREFIX}${intent.backend}`, intent);
+
+    const intents = await this.state.storage.list<SyncRetryIntent>({
+      prefix: RETRY_PREFIX,
+    });
+    const next = Math.min(...[...intents.values()].map((item) => item.notBefore));
+    await this.state.storage.setAlarm(next);
+  }
+
+  async clear(backend: string): Promise<void> {
+    await this.state.storage.delete(`${RETRY_PREFIX}${backend}`);
+  }
+}
+```
+
+Pass the scheduler to `Workspace` and invoke `retryPendingSync` from the host's
+alarm or another durable scheduler:
+
+```ts
+export class WorkspaceHost extends DurableObject<Env> {
+  readonly scheduler = new DurableObjectRetryScheduler(this.ctx);
+  readonly workspace = new Workspace({
+    storage: this.ctx.storage,
+    backends: [/* ... */],
+    retryScheduler: this.scheduler,
+    retry: {
+      initialDelayMs: 1_000,
+      maxDelayMs: 60_000,
+      maxAttempts: 5,
+    },
+  });
+
+  async alarm(): Promise<void> {
+    const intents = await this.ctx.storage.list<SyncRetryIntent>({
+      prefix: RETRY_PREFIX,
+    });
+    const now = Date.now();
+    for (const intent of intents.values()) {
+      if (intent.notBefore <= now) {
+        await this.workspace.retryPendingSync(intent.backend);
+      }
+    }
+
+    const remaining = await this.ctx.storage.list<SyncRetryIntent>({
+      prefix: RETRY_PREFIX,
+    });
+    if (remaining.size > 0) {
+      await this.ctx.storage.setAlarm(
+        Math.min(...[...remaining.values()].map((item) => item.notBefore)),
+      );
+    }
+  }
+}
+```
+
+`retryPendingSync(backend?)` enters the same per-backend FIFO as commands,
+`push`, and `pull`. It resumes `pullOnce` from the cursor already persisted in
+SQLite. Success clears the intent. Failure replaces it with the next bounded
+exponential-backoff attempt. Once `maxAttempts` fails, the final intent stays
+in storage and the method returns `status: "exhausted"`; the host can inspect,
+alert on, or explicitly clear it. Calling the method with no pending intent
+returns `status: "idle"`.
+
 ## Worker-side consumption
 
 ```ts

packages/workspace/src/index.ts

@@ -68,4 +68,11 @@ export {
   WorkspaceShellStub,
   WorkspaceStub,
 } from "./stub.js";
-export { Workspace, type WorkspaceOptions } from "./workspace.js";
+export {
+  type SyncRetryIntent,
+  type SyncRetryOptions,
+  type SyncRetryScheduler,
+  Workspace,
+  type WorkspaceOptions,
+  type WorkspaceRetryPendingSyncResult,
+} from "./workspace.js";

packages/workspace/src/retry.test.ts

@@ -0,0 +1,284 @@
+import type { ChangeEntry } from "@cloudflare/dofs";
+import { SQLiteTestStorage } from "@cloudflare/dofs/testing";
+import { describe, expect, it } from "vitest";
+
+import type { BackendHandle, WorkspaceBackend } from "./backend.js";
+import type {
+  SyncRetryIntent,
+  SyncRetryScheduler,
+  WorkspaceRetryPendingSyncResult,
+} from "./workspace.js";
+import { Workspace } from "./workspace.js";
+
+class MemoryRetryScheduler implements SyncRetryScheduler {
+  readonly intents = new Map<string, SyncRetryIntent>();
+  readonly scheduled: SyncRetryIntent[] = [];
+  readonly cleared: string[] = [];
+
+  async get(backend: string): Promise<SyncRetryIntent | undefined> {
+    return this.intents.get(backend);
+  }
+
+  async schedule(intent: SyncRetryIntent): Promise<void> {
+    this.intents.set(intent.backend, intent);
+    this.scheduled.push(intent);
+  }
+
+  async clear(backend: string): Promise<void> {
+    this.intents.delete(backend);
+    this.cleared.push(backend);
+  }
+}
+
+function retryBackend(options: {
+  onExec(): void;
+  fetchChanges: import("@cloudflare/workspace-rpc").SyncRPC["fetchChanges"];
+  close?: () => Promise<void>;
+}): WorkspaceBackend {
+  const sync: import("@cloudflare/workspace-rpc").SyncRPC = {
+    async push(input) {
+      return { rev: 0, appliedPushCursor: { rev: input.senderRev, path: null } };
+    },
+    fetchChanges: options.fetchChanges,
+    async readEntry() {
+      return null;
+    },
+    async hasObjects(hashes) {
+      return hashes;
+    },
+    fetchObjects() {
+      return new ReadableStream({ start: (controller) => controller.close() });
+    },
+    async watermarks() {
+      return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } };
+    },
+    async pushObjects() {},
+  };
+  return {
+    id: "sandbox",
+    type: "fake",
+    async connect(): Promise<BackendHandle> {
+      return {
+        rpc: {
+          sync,
+          shell: {
+            async exec() {
+              options.onExec();
+              return {
+                id: "command-1",
+                events: new ReadableStream({
+                  start(controller) {
+                    controller.enqueue({ id: "command-1", seq: 1, name: "exit", value: 0 });
+                    controller.close();
+                  },
+                }),
+              };
+            },
+            async getExec() {
+              throw new Error("not used");
+            },
+            async killExec() {},
+            async disposeExec() {},
+          },
+        },
+        close: options.close ?? (async () => {}),
+      };
+    },
+  };
+}
+
+async function runCommand(ws: Workspace): Promise<WorkspaceRetryPendingSyncResult | undefined> {
+  const handle = await ws.shell.exec("build", { encoding: "utf8" });
+  const result = await handle.result();
+  expect(result.sync.status).toBe("pending");
+  return undefined;
+}
+
+describe("Workspace durable pending-sync retries", () => {
+  it("schedules the exact durable retry intent after a post-command pull failure", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    let execs = 0;
+    const backend = retryBackend({
+      onExec: () => execs++,
+      async fetchChanges() {
+        throw new Error("backend unavailable");
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+      retry: { initialDelayMs: 2_000, maxDelayMs: 30_000, maxAttempts: 4 },
+      now: () => 10_000,
+    });
+
+    await runCommand(ws);
+
+    expect(execs).toBe(1);
+    expect(scheduler.scheduled).toEqual([{ backend: "sandbox", attempt: 1, notBefore: 12_000 }]);
+  });
+
+  it("resumes a partial batch from the persisted cursor and converges without rerunning the command", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const after: Array<{ rev: number; path: string | null } | undefined> = [];
+    let fetches = 0;
+    let execs = 0;
+    const entries = Array.from(
+      { length: 257 },
+      (_, index): ChangeEntry => ({
+        kind: "delete",
+        rev: 1,
+        path: `/generated/${index.toString().padStart(3, "0")}`,
+        mtime: 1,
+      }),
+    );
+    const backend = retryBackend({
+      onExec: () => execs++,
+      async fetchChanges(input) {
+        after.push(input.after);
+        fetches++;
+        const remaining = entries.filter((entry) => {
+          if (!input.after || input.after.rev < entry.rev) return true;
+          return (
+            input.after.rev === entry.rev &&
+            input.after.path !== null &&
+            entry.path > input.after.path
+          );
+        });
+        return {
+          currentCursor: { rev: 1, path: null },
+          appliedPushCursor: { rev: 0, path: null },
+          stream:
+            fetches === 1
+              ? new ReadableStream<ChangeEntry>({
+                  pull(controller) {
+                    const entry = remaining.shift();
+                    if (entry !== undefined) {
+                      controller.enqueue(entry);
+                      return;
+                    }
+                    controller.error(new Error("lost after first batch"));
+                  },
+                })
+              : 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,
+    });
+
+    await runCommand(ws);
+    const retried = await ws.retryPendingSync("sandbox");
+
+    expect(retried).toMatchObject({ status: "complete", applied: 1 });
+    expect(execs).toBe(1);
+    expect(after).toEqual([
+      { rev: 0, path: null },
+      { rev: 1, path: "/generated/255" },
+    ]);
+    expect(scheduler.intents.size).toBe(0);
+    expect(scheduler.cleared).toEqual(["sandbox"]);
+  });
+
+  it("coalesces repeated command failures into one pending intent per backend", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        throw new Error("still unavailable");
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+      now: () => 1_000,
+    });
+
+    await Promise.all([runCommand(ws), runCommand(ws)]);
+
+    expect(scheduler.scheduled).toHaveLength(1);
+    expect(scheduler.intents.size).toBe(1);
+  });
+
+  it("reschedules with bounded exponential backoff and leaves exhaustion visible", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        throw new Error("still unavailable");
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+      retry: { initialDelayMs: 100, maxDelayMs: 150, maxAttempts: 3 },
+      now: () => 1_000,
+    });
+
+    await runCommand(ws);
+    expect(await ws.retryPendingSync()).toMatchObject({ status: "pending", attempt: 2 });
+    expect(await ws.retryPendingSync()).toMatchObject({ status: "pending", attempt: 3 });
+    expect(await ws.retryPendingSync()).toMatchObject({ status: "exhausted", attempt: 3 });
+
+    expect(scheduler.scheduled).toEqual([
+      { backend: "sandbox", attempt: 1, notBefore: 1_100 },
+      { backend: "sandbox", attempt: 2, notBefore: 1_150 },
+      { backend: "sandbox", attempt: 3, notBefore: 1_150 },
+    ]);
+    expect(scheduler.intents.get("sandbox")).toEqual({
+      backend: "sandbox",
+      attempt: 3,
+      notBefore: 1_150,
+    });
+  });
+
+  it("disposes a failed retry envelope and closes its RPC handle", async () => {
+    const scheduler = new MemoryRetryScheduler();
+    let closes = 0;
+    let disposals = 0;
+    const backend = retryBackend({
+      onExec() {},
+      async fetchChanges() {
+        return {
+          currentCursor: { rev: 1, path: null },
+          appliedPushCursor: { rev: 0, path: null },
+          stream: new ReadableStream<ChangeEntry>({
+            start(controller) {
+              controller.error(new Error("cancel retry stream"));
+            },
+          }),
+          [Symbol.dispose]() {
+            disposals++;
+          },
+        };
+      },
+      close: async () => {
+        closes++;
+      },
+    });
+    const ws = new Workspace({
+      storage: new SQLiteTestStorage(),
+      backends: [backend],
+      retryScheduler: scheduler,
+    });
+    scheduler.intents.set("sandbox", { backend: "sandbox", attempt: 1, notBefore: 0 });
+
+    await expect(ws.retryPendingSync()).resolves.toMatchObject({ status: "pending" });
+    await ws.close();
+
+    expect(disposals).toBe(1);
+    expect(closes).toBe(1);
+  });
+});