Re-work sync engine

.changeset/restartable-sync-engine.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/computer": minor
+---
+
+`Workspace.pull()` and `Workspace.push()` are now async iterables that commit one block of changes per iteration. This allows a caller to implement a deferred post-exec sync across several invocations. See [./docs/02_sync_protocol.md](./docs/02_sync_protocol.md) for details.

docs/02_sync_protocol.md

@@ -222,6 +222,148 @@ 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.
 
+## Restartable synchronization
+
+A large sync cannot assume it finishes in one invocation. A Durable
+Object can be evicted between two steps, an alarm has a hard cutoff,
+and a JavaScript generator survives neither. `Workspace.pull()`
+and `Workspace.push()` therefore treat the iterator as
+disposable and SQLite as the durability surface.
+
+One `next()` performs one complete block: plan it, transfer it, apply
+it, persist the cursor, yield. No RPC stream stays open across a yield,
+because the caller may never return. Recreating the iterable resumes
+from durable state, so these two lines are equivalent to iterating
+twice on one iterator:
+
+```ts
+await workspace.pull()[Symbol.asyncIterator]().next();
+// Durable Object may be evicted here.
+await workspace.pull()[Symbol.asyncIterator]().next();
+```
+
+Durable state lives in two places. `_vfs_sync_operations` holds one row
+per `(backend, direction)` carrying the fixed target, the generation
+that fences superseded executions, and the block sizing profile. The
+committed progress cursor stays in the watermark tables above, because
+filesystem application and cursor advancement have to commit together.
+The operation row records where a sync is going; the watermark records
+how far it got.
+
+Callers pass no cursor, target, or budget. Block sizing is internal and
+shrinks automatically when a block is detected as interrupted, so a
+caller cannot ask for a block too large to complete.
+
+### The library never owns an alarm
+
+`Workspace` does not call `setAlarm` or `deleteAlarm`. Scheduling
+belongs to the application, which knows what else its alarm is for.
+A strict one-step handler gives each block a fresh CPU allowance:
+
+```ts
+async alarm() {
+  // Leave a durable watchdog before entering a block. A hard reset
+  // cannot erase this future invocation.
+  await this.ctx.storage.setAlarm(Date.now() + 30_000);
+
+  const iterator = this.workspace.pull()[Symbol.asyncIterator]();
+  const { value, done } = await iterator.next();
+
+  if (done || value.complete) {
+    await this.ctx.storage.deleteAlarm();
+  } else {
+    await this.ctx.storage.setAlarm(Date.now() + 1_000);
+  }
+}
+```
+
+Completion is exposed on the yielded value rather than only through a
+following `done: true`, so a one-step handler can decide whether to
+reschedule without paying for a second `next()`.
+
+An application that would rather drive several blocks per invocation
+owns that wall-time budget too:
+
+```ts
+async alarm() {
+  const stopAt = Date.now() + 12 * 60_000;
+  await this.ctx.storage.setAlarm(Date.now() + 30_000);
+
+  for await (const progress of this.workspace.pull()) {
+    if (progress.complete) return;
+    if (Date.now() >= stopAt) {
+      await this.ctx.storage.setAlarm(Date.now() + 1_000);
+      return;
+    }
+  }
+}
+```
+
+Breaking out of either loop stops local driving without failing the
+operation. The last yielded cursor stays durable and a later iterable
+resumes from it. The corollary is that the library cannot guarantee
+convergence if the application never calls it again — that is the
+deliberate price of leaving scheduling to the implementor.
+
+### Entry mode and pack mode
+
+Small windows ship one `ChangeEntry` at a time, with a `hasObjects`
+probe and a `fetchObjects` pull for the bytes the receiver lacks. That
+is cheap for a handful of changes and hopeless for tens of thousands of
+them.
+
+Above either threshold the operation switches to a **change pack**: one
+gzip stream carrying the entries interleaved with the unique objects
+they reference, so a block needs no object round trips at all.
+
+| Threshold | Value |
+| --- | --- |
+| Entries in the cursor window | 20,000 |
+| Estimated payload bytes | 100 MiB |
+
+Selection happens once per operation and is persisted on the operation
+row. The mode cannot change mid-operation: a block's encoding must not
+depend on when it was requested, or a replayed block would not match
+the original and the receiver could not absorb it as a no-op. Since the
+target is fixed at creation, the window cannot grow underneath the
+estimate.
+
+Pack framing is length-prefixed records inside the gzip stream, closed
+by a footer carrying the start cursor, block cursor, target, generation,
+counts, and a digest. A truncated or tampered pack is rejected as a
+protocol error before anything applies, leaving the cursor untouched —
+advancing over entries that never arrived would lose them silently.
+
+`fetchChangePack` and `applyChangePack` are optional on `SyncRPC`. A
+peer that predates them keeps working in entry mode, so the two sides
+can be deployed in either order.
+
+### Replay and idempotency
+
+Because the cursor advances only after a block applies, any block
+interrupted before its acknowledgment is re-applied. Replay safety is
+per entry kind rather than a blanket property:
+
+| Entry | Replay behavior |
+| --- | --- |
+| `file` | `alreadyApplied` compares the manifest hash; identical content is a no-op. |
+| `dir` | Compares mode; a match is a no-op. |
+| `symlink` | Compares mode and target; a match is a no-op. |
+| `delete` | Compares the tombstone's rev against the live inode's. A path recreated **above** the tombstone rev is newer information, so the delete is dropped. |
+
+The delete case is why replay cannot be assumed idempotent for free. A
+tombstone describes a path as of the rev it was stamped with; replaying
+it against a newer local recreation would destroy data the tombstone
+never described. The recreation is itself a change the next push ships
+upstream, so dropping the stale delete loses no work.
+
+Entries the receiver refuses — most often a container-side write under
+a read-only mount — are reported in `SyncProgress.skipped` **and**
+written to `_vfs_sync_skips`. The cursor advances past them, because an
+entry that can never apply would otherwise stall the operation forever.
+Recording them durably keeps the drop auditable rather than visible
+only to whoever happened to read that one progress value.
+
 ## Wire shape
 
 The wire is symmetric: push and fetch both move `ChangeEntry`

docs/05_runtime_interface.md

@@ -64,7 +64,7 @@ interface WorkspaceRuntimeResult {
 }
 ```
 
-Command backends leave `value` unset. `worker-javascript` uses `value` for the module's structured return value and reports a zero-entry completed sync. A command can complete while its post-command pull fails; in that case `sync.status` is `"pending"`, and a configured `SyncRetryScheduler` can durably retry the pull without rerunning the command.
+Command backends leave `value` unset. `worker-javascript` uses `value` for the module's structured return value and reports a zero-entry completed sync. A command can complete while its post-command pull fails; in that case `sync.status` is `"pending"`, and the next `pull()` resumes the operation from its durable watermark without rerunning the command.
 
 ## Backend routing
 
@@ -98,7 +98,7 @@ push → spawn → events/result → pull
 
 A backend with `sync: "none"`, such as `worker-shell`, shares the host store and reports zero push/pull counts. A Container has its own VFS and synchronizes changes before and after command execution. Fully draining either `result()` or the event stream completes the post-command pull before the stream closes.
 
-The pre-command push is a safety gate, not a best-effort optimization. The push and spawn use the same backend handle. If that handle fails before dispatch, the reconnect retry repeats both steps on the replacement container so the command cannot skip its push. If the push still fails, `exec()` rejects before the spawn request is sent; `pushed: 0` means a successful push found no entries, not that synchronization failed. A failed post-command pull does not change the completed command result. The pull is fenced to the runtime UUID that ran the command, so reconnecting to an empty replacement cannot report a clean zero-entry sync. It reports `sync.status: "pending"` and persists that UUID with the retry intent. `retryPendingSync()` resumes against the original runtime when possible; if that runtime was replaced, it clears the unrecoverable intent and reports `status: "lost"` so a later command can schedule sync for the live runtime.
+The pre-command push is a safety gate, not a best-effort optimization. The push and spawn use the same backend handle. If that handle fails before dispatch, the reconnect retry repeats both steps on the replacement container so the command cannot skip its push. If the push still fails, `exec()` rejects before the spawn request is sent; `pushed: 0` means a successful push found no entries, not that synchronization failed. A failed post-command pull does not change the completed command result. The pull is fenced to the runtime UUID that ran the command, so reconnecting to an empty replacement cannot report a clean zero-entry sync. It reports `sync.status: "pending"` and persists that UUID with the sync operation. A later `pull()` resumes against the original runtime when possible; if that runtime was replaced, the unrecoverable operation is cleared so a later command can synchronize the live runtime.
 
 Container connection failures also get one backend-internal reconnect attempt. Sync calls are safe to repeat. `getExec`, `killExec`, and `disposeExec` are retried only when the new connection reaches the same computerd runtime; a replacement container returns `EEXEC_LOST` instead of applying an old execution id to its new process table. `shell.exec` is different: the backend retries it only when connection setup failed or a locally disposed stub proves that no request was sent. If the transport fails after computerd may have accepted the spawn, the error states that the command may have started and the backend does not replay it. A failure while reading the event stream also invalidates the connection without rerunning the command.
 

packages/computer/README.md

@@ -476,17 +476,18 @@ When assigning a workspace to a Think agent's `workspace`, pass
 `useThink: true` so Think's compatibility methods are added alongside
 `workspace.fs` and `workspace.runtime`.
 
-### Durable pending-sync retries
+### Resuming an incomplete sync
 
 A command can change backend files and then have its post-command pull
-fail; the result exposes `sync: { status: "pending", ... }`. Configure a
-`SyncRetryScheduler` on `Workspace` to persist one coalesced retry per
-backend, then call `workspace.retryPendingSync(backend)` from your DO's
-alarm. Retries use bounded exponential backoff and return `"exhausted"`
-after the configured maximum. A container replacement returns `"lost"`
-and clears the unrecoverable intent so new work is not blocked. The library
-does not own your DO's alarm. See `SyncRetryScheduler`, `SyncRetryIntent`,
-and `SyncRetryOptions` in the package exports.
+fail; the result exposes `sync: { status: "pending", ... }`. Nothing
+further is required to recover it. The sync operation and its watermark
+hold the progress durably, so the next `pull()` resumes from where the
+failed one stopped and drains the rest of the fixed target.
+
+A deferred exec fixes that target when the command finishes, which pins
+the command's changes rather than capturing a newer target that could
+have raced ahead. Attempt counts, backoff, and exhaustion are not part
+of the API: a caller that wants to stop trying stops iterating.
 
 ### Observability
 

packages/computer/src/deferred-sync.test.ts

@@ -0,0 +1,156 @@
+import { createSyncServer } from "@cloudflare/computer-rpc/server";
+import {
+  Database,
+  initializeSchema,
+  readOperation,
+  SQLiteWorkspaceProvider,
+} from "@cloudflare/dofs";
+import { SQLiteTestStorage } from "@cloudflare/dofs/testing";
+import { describe, expect, it } from "vitest";
+
+import type { BackendHandle, WorkspaceBackend } from "./backend.js";
+import { Workspace } from "./workspace.js";
+
+// Deferred synchronization used to need a host-supplied retry
+// scheduler: the host persisted an intent, set its own alarm, and
+// called back into retryPendingSync. The durable operation row makes
+// that indirection unnecessary. A deferred exec captures the pull
+// target, and any later caller resumes it by iterating pull().
+
+function peerBackend(id: string): {
+  backend: WorkspaceBackend;
+  remote: Database;
+  close: () => void;
+} {
+  const storage = new SQLiteTestStorage();
+  const remote = new Database(storage);
+  initializeSchema(remote, () => 1000);
+  const sync = createSyncServer(remote);
+  const execId = () => `${id}-exec`;
+  const shell: import("@cloudflare/computer-rpc").ShellRPC = {
+    async exec(input) {
+      const eid = input.id ?? execId();
+      return {
+        id: eid,
+        events: new ReadableStream<import("@cloudflare/computer-rpc").ExecEvent>({
+          start(controller) {
+            controller.enqueue({ id: eid, seq: 1, name: "exit", code: 0 });
+            controller.close();
+          },
+        }),
+      };
+    },
+    getExec: () => Promise.reject(new Error("not used")),
+    killExec: () => Promise.reject(new Error("not used")),
+    disposeExec: () => Promise.reject(new Error("not used")),
+  };
+  const backend: WorkspaceBackend = {
+    id,
+    type: "fake",
+    async connect(): Promise<BackendHandle> {
+      return { rpc: { sync, shell }, close: async () => {} };
+    },
+  };
+  return { backend, remote, close: () => storage.close() };
+}
+
+describe("deferred synchronization without a host scheduler", () => {
+  it("leaves a resumable pull operation after a deferred exec", async () => {
+    const peer = peerBackend("fake");
+    try {
+      const provider = new SQLiteWorkspaceProvider(peer.remote);
+      provider.writeFileSync("/produced.txt", "by the command");
+
+      const ws = new Workspace({
+        storage: new SQLiteTestStorage(),
+        backends: [peer.backend],
+      });
+      await ws.ready();
+
+      // A deferred exec returns without draining the pull.
+      const handle = await ws.runtime.exec("true", { sync: "defer" });
+      await handle.result();
+
+      // The application resumes on its own schedule. No retry
+      // scheduler, no host-persisted intent.
+      for await (const progress of ws.pull()) {
+        if (progress.complete) break;
+      }
+
+      expect(await ws.fs.readFile("/produced.txt", "utf8")).toBe("by the command");
+    } finally {
+      peer.close();
+    }
+  });
+
+  it("does not require a retryScheduler to defer", async () => {
+    const peer = peerBackend("fake");
+    try {
+      const ws = new Workspace({
+        storage: new SQLiteTestStorage(),
+        backends: [peer.backend],
+      });
+      await ws.ready();
+
+      // Previously this threw without a configured retryScheduler.
+      const handle = await ws.runtime.exec("true", { sync: "defer" });
+      await expect(handle.result()).resolves.toBeDefined();
+    } finally {
+      peer.close();
+    }
+  });
+
+  it("reports a deferred pull as pending rather than complete", async () => {
+    const peer = peerBackend("fake");
+    try {
+      const provider = new SQLiteWorkspaceProvider(peer.remote);
+      provider.writeFileSync("/produced.txt", "by the command");
+
+      const ws = new Workspace({
+        storage: new SQLiteTestStorage(),
+        backends: [peer.backend],
+      });
+      await ws.ready();
+
+      const handle = await ws.runtime.exec("true", { sync: "defer" });
+      const execution = await handle.result();
+
+      expect(execution.sync.status).toBe("pending");
+      // Nothing applied yet: that is the point of deferring.
+      expect(execution.sync.applied).toBe(0);
+    } finally {
+      peer.close();
+    }
+  });
+
+  it("converges a deferred pull driven one block at a time", async () => {
+    const peer = peerBackend("fake");
+    try {
+      const provider = new SQLiteWorkspaceProvider(peer.remote);
+      for (let i = 0; i < 5; i++) {
+        provider.writeFileSync(`/f${i}.txt`, `content-${i}`);
+      }
+
+      const ws = new Workspace({
+        storage: new SQLiteTestStorage(),
+        backends: [peer.backend],
+      });
+      await ws.ready();
+      const handle = await ws.runtime.exec("true", { sync: "defer" });
+      await handle.result();
+
+      // One block per "alarm", each from a fresh iterable.
+      for (let i = 0; i < 20; i++) {
+        const iterator = ws.pull()[Symbol.asyncIterator]();
+        const { value, done } = await iterator.next();
+        if (done || value.complete) break;
+      }
+
+      expect(await ws.fs.readFile("/f4.txt", "utf8")).toBe("content-4");
+      // Completed operations leave no row behind.
+      expect(readOperation(ws.db, "fake", "pull")).toBeUndefined();
+    } finally {
+      peer.close();
+    }
+  });
+});

packages/computer/src/index.ts

@@ -15,7 +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 { SyncProgress } from "@cloudflare/computer-rpc/driver";
 export type {
   ApplyResult,
   DurableObjectStorageLike,
@@ -44,6 +44,17 @@ export type {
   MountFactory,
   MountWriteAPI,
 } from "./mounts/types.js";
+export {
+  createSyncLogger,
+  DEFAULT_CPU_LIMIT_MS,
+  type SyncBlockTelemetry,
+  type SyncLogger,
+  type SyncLoggerOptions,
+  type SyncOperationTelemetry,
+  syncBlockLog,
+  syncOperationLog,
+  type TelemetryRecord,
+} from "./observe/sync-telemetry.js";
 export {
   noopObserver,
   type WorkspaceAttributes,
@@ -97,13 +108,8 @@ export {
   withWorkspace,
 } from "./with-workspace.js";
 export {
-  type SyncBatchOptions,
-  type SyncRetryIntent,
-  type SyncRetryOptions,
-  type SyncRetryScheduler,
   type ThinkWorkspaceCompatibility,
   Workspace,
   type WorkspaceGitFactory,
   type WorkspaceOptions,
-  type WorkspaceRetryPendingSyncResult,
 } from "./workspace.js";