From 6a122d3b643c14b01520998bfa2058e007fef47f Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:19:50 +0000 Subject: [PATCH 01/14] dofs: Add path-aware push cursors --- packages/dofs/src/index.ts | 2 ++ packages/dofs/src/schema/core.ts | 2 +- packages/dofs/src/schema/index.ts | 6 +++++ packages/dofs/src/schema/migrations.ts | 19 ++++++++++++++++ packages/dofs/src/schema/sync.ts | 7 ++++++ packages/dofs/src/sync/watermarks.test.ts | 13 +++++++++++ packages/dofs/src/sync/watermarks.ts | 27 +++++++++++++++++++++++ 7 files changed, 75 insertions(+), 1 deletion(-) diff --git a/packages/dofs/src/index.ts b/packages/dofs/src/index.ts index e5638a45..468ec22a 100644 --- a/packages/dofs/src/index.ts +++ b/packages/dofs/src/index.ts @@ -56,8 +56,10 @@ export { compareChangeCursors, currentRev, readFetchCursor, + readPushCursor, readWatermark, writeFetchCursor, + writePushCursor, writeWatermark, } from "./sync/watermarks.js"; export type { ExecutedStatement } from "./testing-recording.js"; diff --git a/packages/dofs/src/schema/core.ts b/packages/dofs/src/schema/core.ts index bc77dd62..88b7daa5 100644 --- a/packages/dofs/src/schema/core.ts +++ b/packages/dofs/src/schema/core.ts @@ -13,7 +13,7 @@ // dirents leaf so the (parent, name) resolve read is covering // (no separate index needed). See `schema/migrations.ts` for the // migration list; `sync.ts` carries the fresh-install DDL. -export const SCHEMA_VERSION = 5; +export const SCHEMA_VERSION = 6; export const ROOT_INODE = 1; export const CORE_STATEMENTS = [ diff --git a/packages/dofs/src/schema/index.ts b/packages/dofs/src/schema/index.ts index e0147e96..dcf85803 100644 --- a/packages/dofs/src/schema/index.ts +++ b/packages/dofs/src/schema/index.ts @@ -69,6 +69,12 @@ export function initializeSchema(db: Database, now: () => number): void { "fetch", null, ); + db.run( + "INSERT OR IGNORE INTO _vfs_push_cursor (k, backend, rev, path) VALUES (?, 'default', ?, ?)", + "push", + 0, + null, + ); db.run( `INSERT OR IGNORE INTO vfs_nodes diff --git a/packages/dofs/src/schema/migrations.ts b/packages/dofs/src/schema/migrations.ts index d2425ada..31b6f374 100644 --- a/packages/dofs/src/schema/migrations.ts +++ b/packages/dofs/src/schema/migrations.ts @@ -142,11 +142,30 @@ function v4_to_v5_without_rowid(db: Database): void { db.run(`CREATE INDEX vfs_chunks_by_hash ON vfs_chunks(hash)`); } +function v5_to_v6_push_cursor(db: Database): void { + db.run( + `CREATE TABLE IF NOT EXISTS _vfs_push_cursor ( + k TEXT NOT NULL CHECK(k = 'push'), + backend TEXT NOT NULL DEFAULT 'default', + rev INTEGER NOT NULL DEFAULT 0, + path TEXT, + PRIMARY KEY (k, backend) + )`, + ); + db.run( + `INSERT OR IGNORE INTO _vfs_push_cursor (k, backend, rev, path) + SELECT 'push', backend, v, NULL + FROM _vfs_watermark + WHERE k = 'pushRev'`, + ); +} + export const MIGRATIONS: readonly Migration[] = [ { from: 1, to: 2, migrator: v1_to_v2_add_mounts_mode }, { from: 2, to: 3, migrator: v2_to_v3_add_size_column }, { from: 3, to: 4, migrator: v3_to_v4_watermark_backend_column }, { from: 4, to: 5, migrator: v4_to_v5_without_rowid }, + { from: 5, to: 6, migrator: v5_to_v6_push_cursor }, ] as const; // Apply every migration whose `from` matches the current version, diff --git a/packages/dofs/src/schema/sync.ts b/packages/dofs/src/schema/sync.ts index fdbac101..78f70787 100644 --- a/packages/dofs/src/schema/sync.ts +++ b/packages/dofs/src/schema/sync.ts @@ -45,6 +45,13 @@ export const SYNC_STATEMENTS = [ path TEXT, PRIMARY KEY (k, backend) )`, + `CREATE TABLE IF NOT EXISTS _vfs_push_cursor ( + k TEXT NOT NULL CHECK(k = 'push'), + backend TEXT NOT NULL DEFAULT 'default', + rev INTEGER NOT NULL DEFAULT 0, + path TEXT, + PRIMARY KEY (k, backend) + )`, // The `mode` column was added at schema v2; `schema/migrations.ts` // owns the ALTER for existing databases. Keep the CHECK // constraint here aligned with the migration's CHECK so fresh diff --git a/packages/dofs/src/sync/watermarks.test.ts b/packages/dofs/src/sync/watermarks.test.ts index 7b9213b0..ef3122c4 100644 --- a/packages/dofs/src/sync/watermarks.test.ts +++ b/packages/dofs/src/sync/watermarks.test.ts @@ -6,8 +6,10 @@ import { compareChangeCursors, currentRev, readFetchCursor, + readPushCursor, readWatermark, writeFetchCursor, + writePushCursor, writeWatermark, } from "./watermarks.js"; @@ -35,6 +37,17 @@ describe("watermarks", () => { }); }); + it("persists a path-aware push cursor per backend", async () => { + await withDB(async (db) => { + expect(readPushCursor(db)).toEqual({ rev: 0, path: null }); + writePushCursor(db, { rev: 12, path: "/dir/file.txt" }, "container"); + expect(readPushCursor(db, "container")).toEqual({ rev: 12, path: "/dir/file.txt" }); + expect(readPushCursor(db, "worker")).toEqual({ rev: 0, path: null }); + writePushCursor(db, { rev: 13, path: null }, "container"); + expect(readPushCursor(db, "container")).toEqual({ rev: 13, path: null }); + }); + }); + it("does not persist an intermediate full-rev cursor when a partial cursor write fails", async () => { await withDB(async (db) => { writeFetchCursor(db, { rev: 12, path: null }); diff --git a/packages/dofs/src/sync/watermarks.ts b/packages/dofs/src/sync/watermarks.ts index 084c3c68..87084bb1 100644 --- a/packages/dofs/src/sync/watermarks.ts +++ b/packages/dofs/src/sync/watermarks.ts @@ -102,6 +102,33 @@ export function readFetchCursor(db: Database, backend: string = DEFAULT_BACKEND_ return { rev, path: path ?? null }; } +export function readPushCursor(db: Database, backend: string = DEFAULT_BACKEND_ID): ChangeCursor { + const row = db.one<{ rev: number; path: string | null }>( + "SELECT rev, path FROM _vfs_push_cursor WHERE k = ? AND backend = ?", + "push", + backend, + ); + return row === undefined ? { rev: 0, path: null } : { rev: row.rev, path: row.path }; +} + +export function writePushCursor( + db: Database, + cursor: ChangeCursor, + backend: string = DEFAULT_BACKEND_ID, +): void { + db.transactionSync(() => { + db.run( + "INSERT INTO _vfs_push_cursor (k, backend, rev, path) VALUES (?, ?, ?, ?) " + + "ON CONFLICT(k, backend) DO UPDATE SET rev = excluded.rev, path = excluded.path", + "push", + backend, + cursor.rev, + cursor.path, + ); + writeWatermarkValue(db, "pushRev", cursor.rev, backend); + }); +} + export function writeFetchCursor( db: Database, cursor: ChangeCursor, From ff6704460dc00ccb9c8fe43ec1cbab666a3ff5c7 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:25:55 +0000 Subject: [PATCH 02/14] rpc: Add bounded sync batches --- packages/rpc/src/interface.ts | 13 +- packages/rpc/src/server.ts | 28 ++- packages/rpc/src/sync-driver.test.ts | 126 +++++++++- packages/rpc/src/sync-driver.ts | 347 ++++++++++++++++++++++++++- 4 files changed, 498 insertions(+), 16 deletions(-) diff --git a/packages/rpc/src/interface.ts b/packages/rpc/src/interface.ts index e12c3293..5cf87698 100644 --- a/packages/rpc/src/interface.ts +++ b/packages/rpc/src/interface.ts @@ -28,8 +28,13 @@ export interface SyncRPC { // advances its fetch cursor to this completed rev after the apply // settles, and echoes that cursor back as `appliedPushCursor` so // the sender can assert applied covers pushed on every response. - push(input: { senderRev: number; changes: ReadableStream }): Promise<{ + push(input: { + senderRev: number; + senderCursor?: ChangeCursor; + changes: ReadableStream; + }): Promise<{ rev: number; + applied?: number; appliedPushCursor: ChangeCursor; }>; @@ -45,7 +50,11 @@ export interface SyncRPC { // mirroring the same check on push. // // Per-file entries carry (hash, size) chunk lists; no bytes inline. - fetchChanges(input: { after?: ChangeCursor; ignore?: string[] }): Promise<{ + fetchChanges(input: { + after?: ChangeCursor; + through?: ChangeCursor; + ignore?: string[]; + }): Promise<{ currentCursor: ChangeCursor; appliedPushCursor: ChangeCursor; stream: ReadableStream; diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index c9c46dcb..638fedaa 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -101,8 +101,9 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { async push(input: { senderRev: number; + senderCursor?: ChangeCursor; changes: ReadableStream; - }): Promise<{ rev: number; appliedPushCursor: ChangeCursor }> { + }): Promise<{ rev: number; applied?: number; appliedPushCursor: ChangeCursor }> { const entries: ChangeEntry[] = []; const reader = input.changes.getReader(); try { @@ -128,7 +129,8 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { // local writes: bump rev through the normal apply path, // leave pushRev untouched so the outbound sync loop // ships them upstream on the next tick. - const isPeer = input.senderRev > 0; + const senderCursor = input.senderCursor ?? { rev: input.senderRev, path: null }; + const isPeer = senderCursor.rev > 0; // Wrap the whole batch in a single transactionSync so a // mid-stream failure (e.g. a missing chunk in applyChangesSync's // assembly step) rolls back every prior entry. Without this @@ -138,11 +140,8 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { applyChangesSync(this.db, entries, new Map(), { source: isPeer ? "upstream" : "local", }); - if (isPeer) { - const nextCursor = { rev: input.senderRev, path: null }; - if (compareChangeCursors(nextCursor, readFetchCursor(this.db)) > 0) { - writeFetchCursor(this.db, nextCursor); - } + if (isPeer && compareChangeCursors(senderCursor, readFetchCursor(this.db)) > 0) { + writeFetchCursor(this.db, senderCursor); } }); if (this.options.afterApply !== undefined && entries.length > 0) { @@ -157,11 +156,16 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { } return { rev: currentRev(this.db), - appliedPushCursor: { rev: input.senderRev, path: null }, + applied: entries.length, + appliedPushCursor: isPeer ? senderCursor : { rev: 0, path: null }, }; } - async fetchChanges(input: { after?: ChangeCursor; ignore?: string[] }): Promise<{ + async fetchChanges(input: { + after?: ChangeCursor; + through?: ChangeCursor; + ignore?: string[]; + }): Promise<{ currentCursor: ChangeCursor; appliedPushCursor: ChangeCursor; stream: ReadableStream; @@ -179,7 +183,11 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { const after = input.after ?? { rev: 0, path: null }; const ignore = input.ignore ?? this.options.ignore; const snapshotRev = currentRev(this.db); - const currentCursor = { rev: snapshotRev, path: null }; + const snapshotCursor = { rev: snapshotRev, path: null }; + const currentCursor = + input.through !== undefined && compareChangeCursors(input.through, snapshotCursor) < 0 + ? input.through + : snapshotCursor; return { currentCursor, appliedPushCursor: readFetchCursor(this.db), diff --git a/packages/rpc/src/sync-driver.test.ts b/packages/rpc/src/sync-driver.test.ts index 5fb6b8a7..bfaec329 100644 --- a/packages/rpc/src/sync-driver.test.ts +++ b/packages/rpc/src/sync-driver.test.ts @@ -16,7 +16,14 @@ import { describe, expect, it } from "vitest"; import type { SyncRPC } from "./interface.js"; import { createSyncServer } from "./server.js"; -import { pullOnce, pushOnce, reconcileWatermarks, tick } from "./sync-driver.js"; +import { + pullBatch, + pullOnce, + pushBatch, + pushOnce, + reconcileWatermarks, + tick, +} from "./sync-driver.js"; // Two peers wired up as direct in-process SyncRPC stubs. No // WebSocket; we already have the real-wire convergence test in @@ -1282,3 +1289,120 @@ async function sha256(bytes: Uint8Array): Promise { hash.update(bytes); return new Uint8Array(hash.digest()); } + +describe("bounded synchronization", () => { + it("pulls one entry per batch and resumes at the captured target", async () => { + const upstream = makePeer(); + const downstream = makePeer(); + try { + const provider = new SQLiteWorkspaceProvider(upstream.db, { now: () => 1 }); + await provider.writeFile("/one.txt", "one"); + await provider.writeFile("/two.txt", "two"); + + const first = await pullBatch(downstream.db, upstream.rpc, { + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(first.status).toBe("pending"); + expect(first.entries).toBe(1); + + const second = await pullBatch(downstream.db, upstream.rpc, { + targetCursor: first.targetCursor, + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(second.status).toBe("pending"); + const third = await pullBatch(downstream.db, upstream.rpc, { + targetCursor: first.targetCursor, + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(third.status).toBe("complete"); + expect(second.entries + third.entries).toBe(1); + expect(fileEntries(downstream.db)).toEqual(["one.txt", "two.txt"]); + } finally { + upstream.close(); + downstream.close(); + } + }); + + it("stages a large file across pull batches without redownloading chunks", async () => { + const upstream = makePeer(); + const downstream = makePeer(); + try { + const provider = new SQLiteWorkspaceProvider(upstream.db, { now: () => 1 }); + const large = new Uint8Array(3 * 512 * 1024); + large.fill(1, 0, 512 * 1024); + large.fill(2, 512 * 1024, 2 * 512 * 1024); + large.fill(3, 2 * 512 * 1024); + await provider.writeFile("/large.bin", large); + let fetches = 0; + const rpc = new Proxy(upstream.rpc as object, { + get(target, property, receiver) { + if (property === "fetchObjects") { + return (hashes: Uint8Array[]) => { + fetches += hashes.length; + return Reflect.get(target, property, receiver).call(target, hashes); + }; + } + return Reflect.get(target, property, receiver); + }, + }) as SyncRPC; + + const first = await pullBatch(downstream.db, rpc, { + budget: { maxEntries: 8, maxBytes: 512 * 1024 }, + }); + expect(first.status).toBe("pending"); + expect(first.entries).toBe(0); + expect(first.bytes).toBe(512 * 1024); + + const second = await pullBatch(downstream.db, rpc, { + targetCursor: first.targetCursor, + budget: { maxEntries: 8, maxBytes: 512 * 1024 }, + }); + expect(second.status).toBe("pending"); + expect(second.bytes).toBe(512 * 1024); + + const third = await pullBatch(downstream.db, rpc, { + targetCursor: first.targetCursor, + budget: { maxEntries: 8, maxBytes: 512 * 1024 }, + }); + expect(third.status).toBe("complete"); + expect(fetches).toBe(3); + } finally { + upstream.close(); + downstream.close(); + } + }); + + it("pushes one bounded unit and resumes through a revision", async () => { + const upstream = makePeer(); + const downstream = makePeer(); + try { + const provider = new SQLiteWorkspaceProvider(upstream.db, { now: () => 1 }); + await provider.writeFile("/one.txt", "one"); + await provider.writeFile("/two.txt", "two"); + + const first = await pushBatch(upstream.db, downstream.rpc, { + backend: "container", + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(first.status).toBe("pending"); + expect(first.entries).toBe(1); + + const second = await pushBatch(upstream.db, downstream.rpc, { + backend: "container", + targetCursor: first.targetCursor, + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(second.status).toBe("pending"); + const third = await pushBatch(upstream.db, downstream.rpc, { + backend: "container", + targetCursor: first.targetCursor, + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + expect(third.status).toBe("complete"); + expect(fileEntries(downstream.db)).toEqual(["one.txt", "two.txt"]); + } finally { + upstream.close(); + downstream.close(); + } + }); +}); diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index fba350a5..e54247ee 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -21,15 +21,45 @@ import { type Database, hasObjects, readFetchCursor, + readPushCursor, readWatermark, type SkippedEntry, stageBlob, writeFetchCursor, + writePushCursor, writeWatermark, } from "@cloudflare/dofs"; import type { SyncRPC } from "./interface.js"; +export interface SyncBatchBudget { + maxEntries: number; + maxBytes: number; + maxWallTimeMs?: number; +} + +export interface SyncBatchResult { + status: "complete" | "pending"; + entries: number; + bytes: number; + applied: number; + skipped: SkippedEntry[]; + cursor: ChangeCursor; + targetCursor: ChangeCursor; +} + +export interface PullBatchOptions { + backend?: string; + targetCursor?: ChangeCursor; + budget: SyncBatchBudget; +} + +export interface PushBatchOptions { + backend?: string; + targetCursor?: ChangeCursor; + budget: SyncBatchBudget; +} + function hex(bytes: Uint8Array): string { let s = ""; for (let i = 0; i < bytes.byteLength; i++) s += bytes[i].toString(16).padStart(2, "0"); @@ -278,14 +308,325 @@ function cursorComplete(after: ChangeCursor, current: ChangeCursor): boolean { } function writeFetchCursorIfAhead(db: Database, cursor: ChangeCursor, backend?: string): void { - // Overlapping pulls can complete out of order, so checkpoint writes - // compare against the latest persisted cursor instead of the value - // observed when this pull started. if (compareChangeCursors(cursor, readFetchCursor(db, backend)) > 0) { writeFetchCursor(db, cursor, backend); } } +function validateBudget(budget: SyncBatchBudget): void { + if (!Number.isSafeInteger(budget.maxEntries) || budget.maxEntries <= 0) { + throw new Error("Sync batch maxEntries must be a positive safe integer"); + } + if (!Number.isSafeInteger(budget.maxBytes) || budget.maxBytes <= 0) { + throw new Error("Sync batch maxBytes must be a positive safe integer"); + } + if ( + budget.maxWallTimeMs !== undefined && + (!Number.isFinite(budget.maxWallTimeMs) || budget.maxWallTimeMs <= 0) + ) { + throw new Error("Sync batch maxWallTimeMs must be positive when provided"); + } +} + +function entryCursor(entry: ChangeEntry): ChangeCursor { + return { rev: entry.rev, path: entry.path }; +} + +function entryHashes(entry: ChangeEntry): { hash: Uint8Array; size: number }[] { + return entry.kind === "file" ? entry.chunks : []; +} + +function entryContentBytes(entry: ChangeEntry): number { + return entryHashes(entry).reduce((total, chunk) => total + chunk.size, 0); +} + +function minimumCursor(a: ChangeCursor, b: ChangeCursor): ChangeCursor { + return compareChangeCursors(a, b) <= 0 ? a : b; +} + +export async function pullBatch( + db: Database, + remote: SyncRPC, + options: PullBatchOptions, +): Promise { + validateBudget(options.budget); + const backend = options.backend; + const after = readFetchCursor(db, backend); + const fetchResult = await remote.fetchChanges({ after, through: options.targetCursor }); + const targetCursor = minimumCursor( + options.targetCursor ?? fetchResult.currentCursor, + fetchResult.currentCursor, + ); + const pushCursor = readPushCursor(db, backend); + assertAppliedPushCursor(fetchResult.appliedPushCursor, pushCursor); + if (compareChangeCursors(after, targetCursor) >= 0) { + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor: after, + targetCursor, + }; + } + + const reader = fetchResult.stream.getReader(); + let streamDone = false; + let cursor = after; + let entries = 0; + let bytes = 0; + let applied = 0; + const skipped: SkippedEntry[] = []; + const started = Date.now(); + try { + while (entries < options.budget.maxEntries) { + if ( + options.budget.maxWallTimeMs !== undefined && + Date.now() - started >= options.budget.maxWallTimeMs + ) { + break; + } + const next = await reader.read(); + if (next.done) { + streamDone = true; + break; + } + const entry = next.value; + const chunks = entryHashes(entry); + const hashes = chunks.map((chunk) => chunk.hash); + const localHave = new Set(hasObjects(db, hashes).map(hex)); + const remoteHave = new Set( + (hashes.length === 0 ? [] : await remote.hasObjects(hashes)).map(hex), + ); + const missingRemote = chunks.filter((chunk) => !remoteHave.has(hex(chunk.hash))); + if (missingRemote.length > 0) { + throw new Error(`pullBatch: remote is missing object ${hex(missingRemote[0].hash)}`); + } + const missingLocal = chunks.filter((chunk) => !localHave.has(hex(chunk.hash))); + const transferable: { hash: Uint8Array; size: number }[] = []; + let availableBytes = options.budget.maxBytes - bytes; + for (const chunk of missingLocal) { + if (chunk.size <= availableBytes || transferable.length === 0) { + transferable.push(chunk); + availableBytes -= chunk.size; + } else { + break; + } + } + if (transferable.length < missingLocal.length) { + if (transferable.length > 0) { + const objectStream = remote.fetchObjects(transferable.map((chunk) => chunk.hash)); + const objectReader = objectStream.getReader(); + try { + while (true) { + const object = await objectReader.read(); + if (object.done) break; + stageBlob(db, object.value.hash, object.value.bytes, Date.now()); + bytes += object.value.bytes.byteLength; + } + } finally { + objectReader.releaseLock(); + } + } + return { + status: "pending", + entries, + bytes, + applied, + skipped, + cursor, + targetCursor, + }; + } + if (transferable.length > 0) { + const objectStream = remote.fetchObjects(transferable.map((chunk) => chunk.hash)); + const objectReader = objectStream.getReader(); + try { + while (true) { + const object = await objectReader.read(); + if (object.done) break; + stageBlob(db, object.value.hash, object.value.bytes, Date.now()); + bytes += object.value.bytes.byteLength; + } + } finally { + objectReader.releaseLock(); + } + } + const result = await applyChanges(db, [entry], new Map(), { + source: "upstream", + backend, + }); + const nextCursor = entryCursor(entry); + writeFetchCursorIfAhead(db, nextCursor, backend); + cursor = nextCursor; + entries += 1; + applied += result.applied; + skipped.push(...result.skipped); + } + if (streamDone) { + writeFetchCursorIfAhead(db, targetCursor, backend); + cursor = targetCursor; + } + return { + status: compareChangeCursors(cursor, targetCursor) >= 0 ? "complete" : "pending", + entries, + bytes, + applied, + skipped, + cursor, + targetCursor, + }; + } finally { + if (!streamDone) await reader.cancel().catch(() => {}); + reader.releaseLock(); + maybeDispose(fetchResult); + } +} + +export async function pushBatch( + db: Database, + remote: SyncRPC, + options: PushBatchOptions, +): Promise { + validateBudget(options.budget); + const backend = options.backend; + const cursor = readPushCursor(db, backend); + const targetCursor = minimumCursor(options.targetCursor ?? { rev: currentRev(db), path: null }, { + rev: currentRev(db), + path: null, + }); + if (compareChangeCursors(cursor, targetCursor) >= 0) { + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor, + targetCursor, + }; + } + + const candidates: ChangeEntry[] = []; + for await (const entry of coalesceChanges(db, cursor, { through: targetCursor })) { + candidates.push(entry); + if (candidates.length >= options.budget.maxEntries) break; + } + if (candidates.length === 0) { + writePushCursor(db, targetCursor, backend); + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor: targetCursor, + targetCursor, + }; + } + + const wanted: { hash: Uint8Array; size: number }[] = []; + const seen = new Set(); + for (const entry of candidates) { + for (const chunk of entryHashes(entry)) { + const key = hex(chunk.hash); + if (!seen.has(key)) { + seen.add(key); + wanted.push(chunk); + } + } + } + const have = new Set( + (wanted.length === 0 ? [] : await remote.hasObjects(wanted.map((c) => c.hash))).map(hex), + ); + const missing = wanted.filter((chunk) => !have.has(hex(chunk.hash))); + const transferable: { hash: Uint8Array; size: number }[] = []; + let availableBytes = options.budget.maxBytes; + for (const chunk of missing) { + if (chunk.size <= availableBytes || transferable.length === 0) { + transferable.push(chunk); + availableBytes -= chunk.size; + } else { + break; + } + } + let bytes = 0; + if (transferable.length > 0) { + const local = (function* () { + for (const chunk of transferable) { + const row = db.one<{ bytes: Uint8Array }>( + "SELECT bytes FROM vfs_blob_bytes WHERE hash = ?", + chunk.hash, + ); + if (row === undefined) throw new Error(`pushBatch: missing local blob ${hex(chunk.hash)}`); + bytes += row.bytes.byteLength; + yield { hash: chunk.hash, bytes: row.bytes }; + } + })(); + const objectStream = new ReadableStream<{ hash: Uint8Array; bytes: Uint8Array }>({ + pull(controller) { + const next = local.next(); + if (next.done) controller.close(); + else controller.enqueue(next.value); + }, + }); + await remote.pushObjects(objectStream); + } + if (transferable.length < missing.length) { + return { + status: "pending", + entries: 0, + bytes, + applied: 0, + skipped: [], + cursor, + targetCursor, + }; + } + + const selected = candidates.filter((entry) => { + const entryMissing = entryHashes(entry).filter((chunk) => !have.has(hex(chunk.hash))); + return entryMissing.every((chunk) => + transferable.some((item) => hex(item.hash) === hex(chunk.hash)), + ); + }); + if (selected.length === 0) { + return { + status: "pending", + entries: 0, + bytes, + applied: 0, + skipped: [], + cursor, + targetCursor, + }; + } + const lastCursor = entryCursor(selected[selected.length - 1]); + const entryStream = new ReadableStream({ + start(controller) { + for (const entry of selected) controller.enqueue(entry); + controller.close(); + }, + }); + const response = await remote.push({ + senderRev: targetCursor.rev, + senderCursor: lastCursor, + changes: entryStream, + }); + assertAppliedPushCursor(response.appliedPushCursor, lastCursor); + writePushCursor(db, lastCursor, backend); + return { + status: compareChangeCursors(lastCursor, targetCursor) >= 0 ? "complete" : "pending", + entries: selected.length, + bytes, + applied: response.applied ?? selected.length, + skipped: [], + cursor: lastCursor, + targetCursor, + }; +} + // Push every entry the local store has produced since the last // successful push. The wire shape mirrors pullOnce in reverse: // stage bytes the remote lacks, then push the entry stream. From 96ce3e23493c40032d6eb4a011e26067eb021546 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:28:59 +0000 Subject: [PATCH 03/14] computer: Add bounded sync options --- packages/computer/src/index.ts | 1 + packages/computer/src/workspace.test.ts | 18 +++- packages/computer/src/workspace.ts | 105 +++++++++++++++++++++--- 3 files changed, 113 insertions(+), 11 deletions(-) diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index 725a0ea1..d93c8843 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -96,6 +96,7 @@ export { withWorkspace, } from "./with-workspace.js"; export { + type SyncBatchOptions, type SyncRetryIntent, type SyncRetryOptions, type SyncRetryScheduler, diff --git a/packages/computer/src/workspace.test.ts b/packages/computer/src/workspace.test.ts index 3b89a672..9bccce16 100644 --- a/packages/computer/src/workspace.test.ts +++ b/packages/computer/src/workspace.test.ts @@ -76,7 +76,10 @@ function fakeRpc(): import("@cloudflare/computer-rpc").SyncRPC { } finally { reader.releaseLock(); } - return { rev: 0, appliedPushCursor: { rev: input.senderRev, path: null } }; + return { + rev: 0, + appliedPushCursor: input.senderCursor ?? { rev: input.senderRev, path: null }, + }; }, async fetchChanges() { return { @@ -1164,6 +1167,19 @@ describe("Workspace.pull return shape", () => { const result = await ws.pull(); expect(result).toEqual({ applied: 0, skipped: [] }); }); + + it("uses batch options on pull without changing the default overload", async () => { + const ws = new Workspace({ storage: makeStorage(), backends: [makeBackend("fake")] }); + await ws.ready(); + const result = await ws.pull("fake", { + mode: "batch", + maxEntries: 1, + maxBytes: 1024, + }); + expect(result.status).toBe("complete"); + expect(result.entries).toBe(0); + expect(result.targetCursor).toEqual({ rev: 0, path: null }); + }); }); describe("Workspace mutation serialization", () => { diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index b0a66948..24410f47 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -10,9 +10,18 @@ // routed through Workspace.runtime.exec. import type { ShellRPC } from "@cloudflare/computer-rpc"; -import { pullOnce, pushOnce, reconcileWatermarks } from "@cloudflare/computer-rpc/driver"; +import { + pullBatch, + pullOnce, + pushBatch, + pushOnce, + reconcileWatermarks, + type SyncBatchBudget, + type SyncBatchResult, +} from "@cloudflare/computer-rpc/driver"; import { type ApplyResult, + type ChangeCursor, Database, type DurableObjectStorageLike, initializeSchema, @@ -72,6 +81,11 @@ export interface SyncRetryScheduler { clear(backend: string): Promise; } +export interface SyncBatchOptions extends SyncBatchBudget { + mode: "batch"; + targetCursor?: ChangeCursor; +} + export interface SyncRetryOptions { initialDelayMs?: number; maxDelayMs?: number; @@ -594,33 +608,91 @@ export class Workspace { // Both methods emit a `workspace.sync.push` / `workspace.sync.pull` // span on the configured observer, tagged with the resolved // backend id and the entry count. - push(id?: string): Promise { + push(id?: string): Promise; + push(options: SyncBatchOptions): Promise; + push(id: string | undefined, options: SyncBatchOptions): Promise; + push( + idOrOptions?: string | SyncBatchOptions, + options?: SyncBatchOptions, + ): Promise { + const id = + typeof idOrOptions === "string" || idOrOptions === undefined ? idOrOptions : undefined; + const batch = typeof idOrOptions === "object" ? idOrOptions : options; return this.#serialize(id, (resolvedId) => withSpan( this.#observer, "workspace.sync.push", { "workspace.sync.backend": resolvedId }, async () => { + if (batch !== undefined) { + if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) { + return emptyBatchResult(batch.targetCursor); + } + return this.#runWithReconnect(resolvedId, "pushBatch", async (handle) => { + if (handle.sync === "none") return emptyBatchResult(batch.targetCursor); + return pushBatch(this.#db, handle.rpc.sync, { + backend: resolvedId, + targetCursor: batch.targetCursor, + budget: batch, + }); + }); + } if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) return 0; return this.#runWithReconnect(resolvedId, "push", async (handle) => { - // A backend that reuses the host store as its sole - // source of truth has nothing to ship and no remote to - // ship to. Short-circuit so the shell exec bracket can - // keep calling push() unconditionally without paying - // for it. if (handle.sync === "none") return 0; return pushOnce(this.#db, handle.rpc.sync, resolvedId); }); }, (span, outcome) => { - if (outcome.ok) span.setAttribute("workspace.sync.pushed", outcome.value); + if (!outcome.ok) return; + span.setAttribute( + "workspace.sync.pushed", + typeof outcome.value === "number" ? outcome.value : outcome.value.entries, + ); }, ), ); } - pull(id?: string): Promise { - return this.#serialize(id, (resolvedId) => this.#pullResolved(resolvedId)); + pull(id?: string): Promise; + pull(options: SyncBatchOptions): Promise; + pull(id: string | undefined, options: SyncBatchOptions): Promise; + pull( + idOrOptions?: string | SyncBatchOptions, + options?: SyncBatchOptions, + ): Promise { + const id = + typeof idOrOptions === "string" || idOrOptions === undefined ? idOrOptions : undefined; + const batch = typeof idOrOptions === "object" ? idOrOptions : options; + if (batch === undefined) + return this.#serialize(id, (resolvedId) => this.#pullResolved(resolvedId)); + return this.#serialize(id, (resolvedId) => + withSpan( + this.#observer, + "workspace.sync.pull", + { "workspace.sync.backend": resolvedId }, + async () => { + if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) { + return emptyBatchResult(batch.targetCursor); + } + return this.#runWithReconnect(resolvedId, "pullBatch", async (handle) => { + if (handle.sync === "none") return emptyBatchResult(batch.targetCursor); + return pullBatch(this.#db, handle.rpc.sync, { + backend: resolvedId, + targetCursor: batch.targetCursor, + budget: batch, + }); + }); + }, + (span, outcome) => { + if (!outcome.ok) return; + span.setAttribute( + "workspace.sync.applied", + typeof outcome.value === "object" ? outcome.value.applied : outcome.value, + ); + }, + ), + ); } /** @@ -1259,6 +1331,19 @@ export class Workspace { } } +function emptyBatchResult(targetCursor?: ChangeCursor): SyncBatchResult { + const target = targetCursor ?? { rev: 0, path: null }; + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor: target, + targetCursor: target, + }; +} + function assertExecutionRuntime( executionId: string, expectedRuntimeId: string | undefined, From 3c76355322c125f025cd89101ba97f75e339843c Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:36:05 +0000 Subject: [PATCH 04/14] computer: Add deferred command synchronization --- packages/computer/src/retry.test.ts | 42 +++++++ packages/computer/src/runtime/runtime.ts | 1 + packages/computer/src/runtime/types.ts | 2 + packages/computer/src/shell.test.ts | 31 +++++ packages/computer/src/shell.ts | 91 ++++++++++++++- packages/computer/src/workspace.ts | 142 +++++++++++++++++++++-- 6 files changed, 292 insertions(+), 17 deletions(-) diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 26575272..513dcaf2 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -396,3 +396,45 @@ 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({ + 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: "deferred", + }); + 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 }, + }); + }); +}); diff --git a/packages/computer/src/runtime/runtime.ts b/packages/computer/src/runtime/runtime.ts index ccae0577..ec43d5df 100644 --- a/packages/computer/src/runtime/runtime.ts +++ b/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, diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index a909f494..edf9ccec 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -110,6 +110,7 @@ export interface WorkspaceRuntimeExecOptions env?: Record; stdin?: Uint8Array | string; timeoutMs?: number; + sync?: "inline" | "deferred"; } export interface WorkspaceRuntimeGetOptions { @@ -144,6 +145,7 @@ export interface ModuleExecutionInput { env?: Record; stdin?: Uint8Array | string; timeoutMs?: number; + sync?: "inline" | "deferred"; } export interface ModuleExecutionEnvelope { diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index 07928c19..60bf8fe1 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -299,6 +299,37 @@ describe("CommandExecutor.exec — envelope events", () => { }); describe("CommandExecutor.exec — push/pull bracket", () => { + it("defers post-command synchronization until a durable intent is scheduled", async () => { + const f = fakeRpc({ events: [exit(1, 0)] }); + const order: string[] = []; + const sync: Sync = { + async push() { + order.push("push"); + return 0; + }, + async pull() { + order.push("pull"); + return applied(1); + }, + async onPostExecPending() { + order.push("schedule"); + return { backend: "container", runtimeId: "runtime-1" }; + }, + }; + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop", { + sync: "deferred", + }); + const { outcome } = await drain(execution); + expect(order).toEqual(["push", "schedule"]); + expect(outcome).toMatchObject({ + sync: { + status: "pending", + backend: "container", + runtimeId: "runtime-1", + }, + }); + }); + it("reports pushed up front and the pull outcome after drain", async () => { const f = fakeRpc({ events: [stdout(1, "hi"), exit(2, 0)] }); const sync: Sync = { diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index 1747862e..161296b1 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -21,7 +21,7 @@ // whatever landed between reattach and drain. import type { ExecEvent, ShellRPC } from "@cloudflare/computer-rpc"; -import type { ApplyResult, SkippedEntry } from "@cloudflare/dofs"; +import type { ApplyResult, ChangeCursor, SkippedEntry } from "@cloudflare/dofs"; import { noopObserver, safeErrorMessage, type WorkspaceObserver, withSpan } from "./observe.js"; import { assertNotTemplate } from "./sh.js"; @@ -39,7 +39,15 @@ export type WorkspaceExecEvent = export type ExecSyncResult = | { status: "complete"; applied: number; skipped: SkippedEntry[] } - | { status: "pending"; applied: number; skipped: SkippedEntry[]; error: string }; + | { + status: "pending"; + applied: number; + skipped: SkippedEntry[]; + error?: string; + backend?: string; + runtimeId?: string; + targetCursor?: ChangeCursor; + }; export type KillSignal = "SIGTERM" | "SIGKILL" | "SIGINT" | "SIGHUP"; @@ -73,6 +81,7 @@ export interface ExecOptions { // Standard input fed to the command. Bytes, or a string encoded // as UTF-8. stdin?: Uint8Array | string; + sync?: "inline" | "deferred"; } export interface GetExecOptions { @@ -96,6 +105,11 @@ export interface Sync { push(): Promise; pull(runtimeId?: string): Promise; onPullPending?(error: unknown, runtimeId?: string): Promise; + onPostExecPending?(runtimeId?: string): Promise<{ + backend?: string; + runtimeId?: string; + targetCursor?: ChangeCursor; + }>; } type ShellExecInput = Parameters[0]; @@ -163,12 +177,15 @@ export class CommandExecutor { // call — because the inner stream is handed off to the caller // and the envelope can't be bound with `using` here. const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); - const { stream, outcome } = withPostPull(drained, this.#sync, envelope.runtimeId); + const wrapped = + options.sync === "deferred" + ? withDeferredPostPull(drained, this.#sync, envelope.runtimeId) + : withPostPull(drained, this.#sync, envelope.runtimeId); return { id: envelope.id, runtimeId: envelope.runtimeId, - events: stream, - sync: { pushed, outcome }, + events: wrapped.stream, + sync: { pushed, outcome: wrapped.outcome }, }; } @@ -281,6 +298,70 @@ export function withPostPull( return { stream, outcome }; } +export function withDeferredPostPull( + source: ReadableStream, + sync: Sync, + runtimeId?: string, +): { stream: ReadableStream; outcome: Promise } { + const reader = source.getReader(); + let resolveOutcome!: (outcome: PostPullOutcome) => void; + const outcome = new Promise((resolve) => { + resolveOutcome = resolve; + }); + const settle = async (reason?: unknown) => { + let metadata: Awaited>> = {}; + let error: unknown = reason; + try { + metadata = (await sync.onPostExecPending?.(runtimeId)) ?? {}; + } catch (caught) { + error = caught; + } + resolveOutcome({ + applied: 0, + skipped: [], + sync: { + status: "pending", + applied: 0, + skipped: [], + ...(error === undefined ? {} : { error: safeErrorMessage(error) }), + ...metadata, + }, + }); + }; + const stream = new ReadableStream( + { + async pull(controller) { + try { + const next = await reader.read(); + if (!next.done) { + controller.enqueue(next.value); + return; + } + reader.releaseLock(); + await settle(); + controller.close(); + } catch (error) { + try { + reader.releaseLock(); + } catch {} + await settle(error); + controller.error(error); + } + }, + async cancel(reason) { + try { + await reader.cancel(reason); + } finally { + reader.releaseLock(); + await settle(reason); + } + }, + }, + { highWaterMark: 0 }, + ); + return { stream, outcome }; +} + async function runPostPull(sync: Sync, runtimeId?: string): Promise { try { const result = await sync.pull(runtimeId); diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index 24410f47..26ef40e0 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -62,6 +62,7 @@ import { export interface SyncRetryIntent { backend: string; + targetCursor?: ChangeCursor; // Container process whose post-command changes are pending. Durable // retries must not report success against an empty replacement. runtimeId?: string; @@ -101,7 +102,9 @@ export type WorkspaceRetryPendingSyncResult = runtimeId?: string; attempt: number; notBefore: number; - error: string; + cursor?: ChangeCursor; + targetCursor?: ChangeCursor; + error?: string; } | { status: "exhausted"; @@ -115,6 +118,10 @@ export type WorkspaceRetryPendingSyncResult = const DEFAULT_RETRY_INITIAL_DELAY_MS = 1_000; const DEFAULT_RETRY_MAX_DELAY_MS = 60_000; const DEFAULT_RETRY_MAX_ATTEMPTS = 5; +const DEFAULT_SYNC_BATCH_BUDGET: SyncBatchBudget = { + maxEntries: 64, + maxBytes: 4 * 1024 * 1024, +}; // When a backend RPC fails with a transport error, how much replay // the operation tolerates. "always" suits idempotent calls; a @@ -703,7 +710,10 @@ export class Workspace { * intent. A failed pull advances bounded exponential backoff; the * last failed attempt remains stored and is reported as exhausted. */ - retryPendingSync(id?: string): Promise { + retryPendingSync( + id?: string, + budget: SyncBatchBudget = DEFAULT_SYNC_BATCH_BUDGET, + ): Promise { return this.#serialize(id, async (resolvedId) => { if (resolvedId === undefined) { throw new Error("Workspace has no backend configured for pending sync retry"); @@ -724,7 +734,36 @@ export class Workspace { }; } try { - const result = await this.#pullResolved(resolvedId, intent.runtimeId); + const result = await this.#pullBatchResolved( + resolvedId, + intent.runtimeId, + budget, + intent.targetCursor, + ); + if (result.status === "pending") { + if (intent.attempt >= this.#retryMaxAttempts) { + return { + status: "exhausted", + backend: resolvedId, + ...(intent.runtimeId === undefined ? {} : { runtimeId: intent.runtimeId }), + attempt: intent.attempt, + error: "pending sync retry attempts exhausted", + }; + } + const next = this.#retryIntent( + resolvedId, + intent.attempt + 1, + intent.runtimeId, + result.targetCursor, + ); + await scheduler.schedule(next); + return { + status: "pending", + ...next, + cursor: result.cursor, + targetCursor: result.targetCursor, + }; + } await scheduler.clear(resolvedId); return { status: "complete", @@ -755,13 +794,53 @@ export class Workspace { error: message, }; } - const next = this.#retryIntent(resolvedId, intent.attempt + 1, intent.runtimeId); + const next = this.#retryIntent( + resolvedId, + intent.attempt + 1, + intent.runtimeId, + intent.targetCursor, + ); await scheduler.schedule(next); return { status: "pending", ...next, error: message }; } }); } + #pullBatchResolved( + resolvedId: string | undefined, + expectedRuntimeId: string | undefined, + budget: SyncBatchBudget, + targetCursor?: ChangeCursor, + ): Promise { + return withSpan( + this.#observer, + "workspace.sync.pull.batch", + { "workspace.sync.backend": resolvedId }, + async () => { + if (resolvedId === undefined || this.#moduleBackendsById.has(resolvedId)) { + return emptyBatchResult(targetCursor); + } + return this.#runWithReconnect(resolvedId, "pullBatch", async (handle) => { + if (expectedRuntimeId !== undefined) { + assertExecutionRuntime("post-command sync", expectedRuntimeId, handle.runtimeId); + } + if (handle.sync === "none") return emptyBatchResult(targetCursor); + return pullBatch(this.#db, handle.rpc.sync, { + backend: resolvedId, + targetCursor, + budget, + }); + }); + }, + (span, outcome) => { + if (!outcome.ok) return; + span.setAttribute("workspace.sync.entries", outcome.value.entries); + span.setAttribute("workspace.sync.bytes", outcome.value.bytes); + span.setAttribute("workspace.sync.applied", outcome.value.applied); + }, + ); + } + #pullResolved(resolvedId: string | undefined, expectedRuntimeId?: string): Promise { return withSpan( this.#observer, @@ -787,26 +866,61 @@ export class Workspace { ); } - async #schedulePendingSync(id: string, runtimeId?: string): Promise { + async #schedulePendingSync( + id: string, + runtimeId?: string, + captureTarget = false, + ): Promise<{ backend?: string; runtimeId?: string; targetCursor?: ChangeCursor }> { const scheduler = this.#retryScheduler; - if (scheduler === undefined) return; - await this.#serialize(id, async (resolvedId) => { - if (resolvedId === undefined) return; + if (scheduler === undefined) { + if (captureTarget) { + throw new Error("Workspace requires a retryScheduler for deferred synchronization"); + } + return {}; + } + return this.#serialize(id, async (resolvedId) => { + if (resolvedId === undefined) return {}; const existing = await scheduler.get(resolvedId); if (existing !== undefined && (runtimeId === undefined || existing.runtimeId === runtimeId)) { - return; + return { + backend: resolvedId, + ...(existing.runtimeId === undefined ? {} : { runtimeId: existing.runtimeId }), + ...(existing.targetCursor === undefined ? {} : { targetCursor: existing.targetCursor }), + }; } - await scheduler.schedule(this.#retryIntent(resolvedId, 1, runtimeId)); + let targetCursor: ChangeCursor | undefined; + if (captureTarget) { + const handle = await this.#handleFor(resolvedId); + if (runtimeId !== undefined) { + assertExecutionRuntime("post-command sync", runtimeId, handle.runtimeId); + } + targetCursor = + handle.sync === "none" + ? { rev: 0, path: null } + : { rev: (await handle.rpc.sync.watermarks()).currentRev, path: null }; + } + await scheduler.schedule(this.#retryIntent(resolvedId, 1, runtimeId, targetCursor)); + return { + backend: resolvedId, + ...(runtimeId === undefined ? {} : { runtimeId }), + ...(targetCursor === undefined ? {} : { targetCursor }), + }; }); } - #retryIntent(backend: string, attempt: number, runtimeId?: string): SyncRetryIntent { + #retryIntent( + backend: string, + attempt: number, + runtimeId?: string, + targetCursor?: ChangeCursor, + ): SyncRetryIntent { const delay = Math.min( this.#retryMaxDelayMs, this.#retryInitialDelayMs * 2 ** Math.max(0, attempt - 1), ); return { backend, + ...(targetCursor === undefined ? {} : { targetCursor }), ...(runtimeId === undefined ? {} : { runtimeId }), attempt, notBefore: this.#now() + delay, @@ -998,6 +1112,7 @@ export class Workspace { timeoutMs: input.timeoutMs, env: input.env, stdin: input.stdin, + sync: input.sync, }); this.#rememberExecutionRuntime(id, envelope.id, envelope.runtimeId); return { @@ -1233,7 +1348,10 @@ export class Workspace { { push: () => this.push(id), pull: (runtimeId) => this.#pullForExec(id, runtimeId), - onPullPending: (_error, runtimeId) => this.#schedulePendingSync(id, runtimeId), + onPullPending: async (_error, runtimeId) => { + await this.#schedulePendingSync(id, runtimeId); + }, + onPostExecPending: (runtimeId) => this.#schedulePendingSync(id, runtimeId, true), }, this.#observer, dispatch, From 5a9f988637beda1440fad9d068027f9aab2fa4d9 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:39:14 +0000 Subject: [PATCH 05/14] dofs: Preserve legacy push watermarks --- packages/dofs/src/sync/watermarks.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/dofs/src/sync/watermarks.ts b/packages/dofs/src/sync/watermarks.ts index 87084bb1..4fc180b4 100644 --- a/packages/dofs/src/sync/watermarks.ts +++ b/packages/dofs/src/sync/watermarks.ts @@ -108,7 +108,9 @@ export function readPushCursor(db: Database, backend: string = DEFAULT_BACKEND_I "push", backend, ); - return row === undefined ? { rev: 0, path: null } : { rev: row.rev, path: row.path }; + const watermark = readWatermark(db, "pushRev", backend); + if (row === undefined || watermark > row.rev) return { rev: watermark, path: null }; + return { rev: row.rev, path: row.path }; } export function writePushCursor( From 31412275a2a99d9982e22636e498314be376deeb Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:39:14 +0000 Subject: [PATCH 06/14] rpc: Resume compatibility sync helpers --- packages/rpc/src/sync-driver.ts | 131 ++++++++++++-------------------- 1 file changed, 50 insertions(+), 81 deletions(-) diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index e54247ee..0560f745 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -101,9 +101,6 @@ export async function pullOnce( remote: SyncRPC, backend?: string, ): Promise { - // Delegate to the inner implementation with retried=false. See - // pullOnceImpl for the fetchChanges round trip, invariant check, - // reset-and-retry path, and batched apply loop. return pullOnceImpl(db, remote, backend, false); } @@ -336,30 +333,50 @@ function entryHashes(entry: ChangeEntry): { hash: Uint8Array; size: number }[] { return entry.kind === "file" ? entry.chunks : []; } -function entryContentBytes(entry: ChangeEntry): number { - return entryHashes(entry).reduce((total, chunk) => total + chunk.size, 0); -} - function minimumCursor(a: ChangeCursor, b: ChangeCursor): ChangeCursor { return compareChangeCursors(a, b) <= 0 ? a : b; } -export async function pullBatch( +export function pullBatch( db: Database, remote: SyncRPC, options: PullBatchOptions, ): Promise { validateBudget(options.budget); + return pullBatchImpl(db, remote, options, false); +} + +async function pullBatchImpl( + db: Database, + remote: SyncRPC, + options: PullBatchOptions, + retried: boolean, +): Promise { const backend = options.backend; const after = readFetchCursor(db, backend); const fetchResult = await remote.fetchChanges({ after, through: options.targetCursor }); + const pushCursor = readPushCursor(db, backend); + const pushDiverged = fetchResult.appliedPushCursor.rev < pushCursor.rev; + const fetchDiverged = compareChangeCursors(fetchResult.currentCursor, after) < 0; + if (!retried && (pushDiverged || fetchDiverged)) { + await fetchResult.stream.cancel().catch(() => {}); + maybeDispose(fetchResult); + if (pushDiverged) writePushCursor(db, { rev: 0, path: null }, backend); + if (fetchDiverged) writeFetchCursor(db, { rev: 0, path: null }, backend); + return pullBatchImpl(db, remote, options, true); + } const targetCursor = minimumCursor( options.targetCursor ?? fetchResult.currentCursor, fetchResult.currentCursor, ); - const pushCursor = readPushCursor(db, backend); - assertAppliedPushCursor(fetchResult.appliedPushCursor, pushCursor); + try { + assertAppliedPushCursor(fetchResult.appliedPushCursor, pushCursor); + } catch (error) { + maybeDispose(fetchResult); + throw error; + } if (compareChangeCursors(after, targetCursor) >= 0) { + maybeDispose(fetchResult); return { status: "complete", entries: 0, @@ -416,7 +433,7 @@ export async function pullBatch( } if (transferable.length < missingLocal.length) { if (transferable.length > 0) { - const objectStream = remote.fetchObjects(transferable.map((chunk) => chunk.hash)); + const objectStream = await remote.fetchObjects(transferable.map((chunk) => chunk.hash)); const objectReader = objectStream.getReader(); try { while (true) { @@ -440,7 +457,7 @@ export async function pullBatch( }; } if (transferable.length > 0) { - const objectStream = remote.fetchObjects(transferable.map((chunk) => chunk.hash)); + const objectStream = await remote.fetchObjects(transferable.map((chunk) => chunk.hash)); const objectReader = objectStream.getReader(); try { while (true) { @@ -514,6 +531,17 @@ export async function pushBatch( if (candidates.length >= options.budget.maxEntries) break; } if (candidates.length === 0) { + const changes = new ReadableStream({ + start(controller) { + controller.close(); + }, + }); + const response = await remote.push({ + senderRev: targetCursor.rev, + senderCursor: targetCursor, + changes, + }); + assertAppliedPushCursor(response.appliedPushCursor, targetCursor); writePushCursor(db, targetCursor, backend); return { status: "complete", @@ -631,77 +659,18 @@ export async function pushBatch( // successful push. The wire shape mirrors pullOnce in reverse: // stage bytes the remote lacks, then push the entry stream. export async function pushOnce(db: Database, remote: SyncRPC, backend?: string): Promise { - const sincePush = readWatermark(db, "pushRev", backend); - const localRev = currentRev(db); - if (localRev <= sincePush) return 0; - - const entries: ChangeEntry[] = []; - const wantedHashes: Uint8Array[] = []; - const seenHash = new Set(); - for await (const e of coalesceChanges(db, { rev: sincePush, path: null })) { - entries.push(e); - if (e.kind === "file") { - for (const c of e.chunks) { - const k = hex(c.hash); - if (!seenHash.has(k)) { - seenHash.add(k); - wantedHashes.push(c.hash); - } - } - } - } - if (entries.length === 0) return 0; - - // Probe the remote for the chunks it already holds; ship the - // complement. - const remoteHas = new Set(); - if (wantedHashes.length > 0) { - const have = await remote.hasObjects(wantedHashes); - for (const h of have) remoteHas.add(hex(h)); - } - const missing = wantedHashes.filter((h) => !remoteHas.has(hex(h))); - - if (missing.length > 0) { - const local = (function* () { - for (const h of missing) { - const row = db.one<{ bytes: Uint8Array }>( - "SELECT bytes FROM vfs_blob_bytes WHERE hash = ?", - h, - ); - if (row === undefined) { - throw new Error(`pushOnce: missing local blob ${hex(h)}`); - } - yield { hash: h, bytes: row.bytes }; - } - })(); - const bytesStream = new ReadableStream<{ hash: Uint8Array; bytes: Uint8Array }>({ - pull(controller) { - const next = local.next(); - if (next.done) controller.close(); - else controller.enqueue(next.value); - }, + let targetCursor: ChangeCursor | undefined; + let pushed = 0; + while (true) { + const result = await pushBatch(db, remote, { + backend, + targetCursor, + budget: { maxEntries: PULL_BATCH_SIZE, maxBytes: 4 * 1024 * 1024 }, }); - await remote.pushObjects(bytesStream); + targetCursor = result.targetCursor; + pushed += result.entries; + if (result.status === "complete") return pushed; } - - const entryStream = new ReadableStream({ - start(controller) { - for (const e of entries) controller.enqueue(e); - controller.close(); - }, - }); - const response = await remote.push({ senderRev: localRev, changes: entryStream }); - - // Cross-side invariant: the receiver must echo back a cursor that - // covers the rev we just claimed to push. A drift means the apply - // path lost data, or a stale receiver is serving an old snapshot. - // Tear down loudly rather than corrupt watermarks. - assertAppliedPushCursor(response.appliedPushCursor, { rev: localRev, path: null }); - - // Local pushRev advances to the rev we observed at the start of - // this round. Anything written after that gets caught next tick. - writeWatermark(db, "pushRev", localRev, backend); - return entries.length; } // One full tick: pull, then push. The order matters \u2014 pulling From 26c5386664a4ff57b0322ca0e460e36f3ed2ed76 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:39:53 +0000 Subject: [PATCH 07/14] computer: Schedule sync on stream cancellation --- packages/computer/src/shell.test.ts | 22 ++++++++++++++++++++++ packages/computer/src/shell.ts | 6 ++++++ 2 files changed, 28 insertions(+) diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index 60bf8fe1..bfe1a11a 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -527,3 +527,25 @@ describe("CommandExecutor.get — reattach", () => { expect((outcome as { applied: number }).applied).toBe(2); }); }); + +describe("CommandExecutor cancellation", () => { + it("schedules synchronization when the event stream is cancelled", async () => { + const f = fakeRpc({ events: [stdout(1, "output"), exit(2, 0)] }); + let scheduled = 0; + const sync: Sync = { + async push() { + return 0; + }, + async pull() { + return applied(0); + }, + async onPullPending() { + scheduled += 1; + }, + }; + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + await execution.events.cancel("consumer stopped"); + await execution.sync.outcome; + expect(scheduled).toBe(1); + }); +}); diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index 161296b1..0b79d444 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -272,6 +272,9 @@ export function withPostPull( try { reader.releaseLock(); } catch {} + try { + await sync.onPullPending?.(error, runtimeId); + } catch {} resolveOutcome({ applied: 0, skipped: [], @@ -285,6 +288,9 @@ export function withPostPull( await reader.cancel(reason); } finally { reader.releaseLock(); + try { + await sync.onPullPending?.(reason, runtimeId); + } catch {} resolveOutcome({ applied: 0, skipped: [], From 2af816ae6c95a32aa3852f88e8b0cbc18f09662b Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:41:10 +0000 Subject: [PATCH 08/14] rpc: Reset path-aware cursors on reconcile --- packages/rpc/src/sync-driver.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index 0560f745..9d25fce1 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -710,7 +710,7 @@ export async function reconcileWatermarks( ): Promise<{ fetchRevReset: boolean; pushRevReset: boolean }> { const remoteWatermarks = await remote.watermarks(); const localFetchCursor = readFetchCursor(db, backend); - const localPushRev = readWatermark(db, "pushRev", backend); + const localPushCursor = readPushCursor(db, backend); let fetchRevReset = false; let pushRevReset = false; @@ -736,8 +736,8 @@ export async function reconcileWatermarks( // (e.g. the container side of a DO↔container backend), which would // make every reconcile spuriously reset pushRev and force a full // re-push on every reconnect. - if (remoteWatermarks.fetchCursor.rev < localPushRev) { - writeWatermark(db, "pushRev", 0, backend); + if (compareChangeCursors(remoteWatermarks.fetchCursor, localPushCursor) < 0) { + writePushCursor(db, { rev: 0, path: null }, backend); pushRevReset = true; } From 77ba2db2c367ce90f9b7141381132624211f37fb Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:41:10 +0000 Subject: [PATCH 09/14] computer: Export sync batch types --- packages/computer/src/index.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index d93c8843..15baf169 100644 --- a/packages/computer/src/index.ts +++ b/packages/computer/src/index.ts @@ -22,6 +22,7 @@ export type { SQLiteWorkspaceProviderOptions, } from "@cloudflare/dofs"; export { SQLiteWorkspaceProvider } from "@cloudflare/dofs"; +export type { SyncBatchBudget, SyncBatchResult } from "@cloudflare/computer-rpc/driver"; export type { BackendHandle, WorkspaceBackend } from "./backend.js"; export { TestBackend, type TestBackendOptions } from "./backends/test.js"; export { From f553dae9dffa225d66bec1e28925bfd3e303f208 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:43:21 +0000 Subject: [PATCH 10/14] computer, rpc: Fence deferred synchronization --- packages/computer/src/retry.test.ts | 18 ++++++++++++++++++ packages/computer/src/shell.ts | 2 ++ packages/computer/src/workspace.ts | 5 +++++ packages/rpc/src/sync-driver.ts | 11 +++++++++++ 4 files changed, 36 insertions(+) diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 513dcaf2..07d60f7a 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -438,3 +438,21 @@ describe("Workspace deferred synchronization", () => { }); }); }); + +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: "deferred" })).rejects.toThrow( + "retryScheduler", + ); + expect(execs).toBe(0); +}); diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index 0b79d444..359db2b0 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -110,6 +110,7 @@ export interface Sync { runtimeId?: string; targetCursor?: ChangeCursor; }>; + assertDeferredReady?(): void | Promise; } type ShellExecInput = Parameters[0]; @@ -154,6 +155,7 @@ export class CommandExecutor { // with stale or incomplete workspace contents is not safe. async exec(source: string, options: ExecOptions = {}): Promise { assertNotTemplate(source); + if (options.sync === "deferred") await this.#sync.assertDeferredReady?.(); const input: ShellExecInput = { source, id: options.id, diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index 26ef40e0..e50f1c70 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -1352,6 +1352,11 @@ export class Workspace { await this.#schedulePendingSync(id, runtimeId); }, onPostExecPending: (runtimeId) => this.#schedulePendingSync(id, runtimeId, true), + assertDeferredReady: () => { + if (this.#retryScheduler === undefined) { + throw new Error("Workspace requires a retryScheduler for deferred synchronization"); + } + }, }, this.#observer, dispatch, diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index 9d25fce1..4d3b83b9 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -531,6 +531,17 @@ export async function pushBatch( if (candidates.length >= options.budget.maxEntries) break; } if (candidates.length === 0) { + if (cursor.rev === 0 && cursor.path === null) { + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor, + targetCursor, + }; + } const changes = new ReadableStream({ start(controller) { controller.close(); From f0470870a2010be212c83521b4460576eb455ff5 Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:44:32 +0000 Subject: [PATCH 11/14] computer: Format sync exports --- packages/computer/src/index.ts | 2 +- packages/computer/src/retry.test.ts | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/packages/computer/src/index.ts b/packages/computer/src/index.ts index 15baf169..70f2adfd 100644 --- a/packages/computer/src/index.ts +++ b/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, @@ -22,7 +23,6 @@ export type { SQLiteWorkspaceProviderOptions, } from "@cloudflare/dofs"; export { SQLiteWorkspaceProvider } from "@cloudflare/dofs"; -export type { SyncBatchBudget, SyncBatchResult } from "@cloudflare/computer-rpc/driver"; export type { BackendHandle, WorkspaceBackend } from "./backend.js"; export { TestBackend, type TestBackendOptions } from "./backends/test.js"; export { diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 07d60f7a..5c8ce4bb 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -451,8 +451,6 @@ it("rejects deferred execution without a retry scheduler", async () => { }); const ws = new Workspace({ storage: new SQLiteTestStorage(), backends: [backend] }); - await expect(ws.runtime.exec("build", { sync: "deferred" })).rejects.toThrow( - "retryScheduler", - ); + await expect(ws.runtime.exec("build", { sync: "deferred" })).rejects.toThrow("retryScheduler"); expect(execs).toBe(0); }); From c989b398ad5440379e4d148710bf1ab353f5fb0f Mon Sep 17 00:00:00 2001 From: OpenAI Date: Thu, 20 Aug 2026 20:46:36 +0000 Subject: [PATCH 12/14] rpc: Acknowledge complete push targets --- packages/rpc/src/sync-driver.ts | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index 4d3b83b9..e9b0b557 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -526,9 +526,13 @@ export async function pushBatch( } const candidates: ChangeEntry[] = []; + let exhausted = true; for await (const entry of coalesceChanges(db, cursor, { through: targetCursor })) { candidates.push(entry); - if (candidates.length >= options.budget.maxEntries) break; + if (candidates.length >= options.budget.maxEntries) { + exhausted = false; + break; + } } if (candidates.length === 0) { if (cursor.rev === 0 && cursor.path === null) { @@ -642,6 +646,7 @@ export async function pushBatch( }; } const lastCursor = entryCursor(selected[selected.length - 1]); + const acknowledgedCursor = exhausted ? targetCursor : lastCursor; const entryStream = new ReadableStream({ start(controller) { for (const entry of selected) controller.enqueue(entry); @@ -650,18 +655,18 @@ export async function pushBatch( }); const response = await remote.push({ senderRev: targetCursor.rev, - senderCursor: lastCursor, + senderCursor: acknowledgedCursor, changes: entryStream, }); - assertAppliedPushCursor(response.appliedPushCursor, lastCursor); - writePushCursor(db, lastCursor, backend); + assertAppliedPushCursor(response.appliedPushCursor, acknowledgedCursor); + writePushCursor(db, acknowledgedCursor, backend); return { - status: compareChangeCursors(lastCursor, targetCursor) >= 0 ? "complete" : "pending", + status: compareChangeCursors(acknowledgedCursor, targetCursor) >= 0 ? "complete" : "pending", entries: selected.length, bytes, applied: response.applied ?? selected.length, skipped: [], - cursor: lastCursor, + cursor: acknowledgedCursor, targetCursor, }; } From bc54994150a99bce72527fedc52db928b58e918f Mon Sep 17 00:00:00 2001 From: Pi Agent Date: Fri, 21 Aug 2026 14:09:24 +0000 Subject: [PATCH 13/14] computer, rpc: Fence deferred sync retries Settle the remote filesystem before capturing a deferred sync target and merge later command targets into an existing durable intent. Treat bounded batch completion as progress rather than a failed retry. Keep canceled command streams draining until the command finishes, and leave already-satisfied pull cursors in place. --- docs/08_capnweb_interface.md | 5 +- packages/computer/src/retry.test.ts | 165 ++++++++++++++++++++++++++- packages/computer/src/shell.test.ts | 54 +++++++++ packages/computer/src/shell.ts | 38 ++++-- packages/computer/src/workspace.ts | 64 +++++++---- packages/rpc/src/interface.ts | 10 +- packages/rpc/src/server.ts | 13 ++- packages/rpc/src/sync-driver.test.ts | 56 +++++++++ packages/rpc/src/sync-driver.ts | 14 +++ 9 files changed, 380 insertions(+), 39 deletions(-) diff --git a/docs/08_capnweb_interface.md b/docs/08_capnweb_interface.md index 72672373..956ebd8a 100644 --- a/docs/08_capnweb_interface.md +++ b/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 }; diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 5c8ce4bb..7a100c26 100644 --- a/packages/computer/src/retry.test.ts +++ b/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; }): 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({ + 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({ @@ -437,6 +497,105 @@ describe("Workspace deferred synchronization", () => { 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: "deferred" }); + 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: "deferred" }); + 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: "deferred" }); + await first.result(); + const second = await ws.runtime.exec("second", { sync: "deferred" }); + 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 () => { diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index bfe1a11a..ff8a7943 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -548,4 +548,58 @@ describe("CommandExecutor cancellation", () => { await execution.sync.outcome; expect(scheduled).toBe(1); }); + + it("waits for a deferred command to finish before capturing its target after cancellation", async () => { + let source!: ReadableStreamDefaultController; + let sourceCancelled = false; + const shell: ShellRPC = { + async exec() { + return { + id: "still-running", + events: new ReadableStream({ + start(controller) { + source = controller; + }, + cancel() { + sourceCancelled = true; + }, + }), + }; + }, + async getExec() { + throw new Error("unused"); + }, + async killExec() {}, + async disposeExec() {}, + }; + let scheduled = 0; + const sync: Sync = { + async push() { + return 0; + }, + async pull() { + return applied(0); + }, + async onPostExecPending() { + scheduled++; + return { targetCursor: { rev: 2, path: null } }; + }, + }; + const execution = await new CommandExecutor(shell, sync).exec("build", { + sync: "deferred", + }); + + const cancelling = execution.events.cancel("consumer stopped"); + await Promise.resolve(); + expect(sourceCancelled).toBe(false); + expect(scheduled).toBe(0); + + source.enqueue(exit(1, 0)); + source.close(); + await cancelling; + await expect(execution.sync.outcome).resolves.toMatchObject({ + sync: { status: "pending", targetCursor: { rev: 2, path: null } }, + }); + expect(scheduled).toBe(1); + }); }); diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index 359db2b0..2b918e38 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -316,9 +316,9 @@ export function withDeferredPostPull( const outcome = new Promise((resolve) => { resolveOutcome = resolve; }); - const settle = async (reason?: unknown) => { + const settle = async () => { let metadata: Awaited>> = {}; - let error: unknown = reason; + let error: unknown; try { metadata = (await sync.onPostExecPending?.(runtimeId)) ?? {}; } catch (caught) { @@ -336,6 +336,21 @@ export function withDeferredPostPull( }, }); }; + const settleUnfenced = async (error: unknown) => { + try { + await sync.onPullPending?.(error, runtimeId); + } catch {} + resolveOutcome({ + applied: 0, + skipped: [], + sync: { + status: "pending", + applied: 0, + skipped: [], + error: safeErrorMessage(error), + }, + }); + }; const stream = new ReadableStream( { async pull(controller) { @@ -352,16 +367,25 @@ export function withDeferredPostPull( try { reader.releaseLock(); } catch {} - await settle(error); + await settleUnfenced(error); controller.error(error); } }, - async cancel(reason) { + async cancel() { + // Cancelling an event subscription does not stop the command. + // Keep draining so backpressure cannot stall it, then capture + // the target only after its stream closes. The cancellation + // promise is the durability boundary: it resolves only after + // the retry intent has been stored. try { - await reader.cancel(reason); - } finally { + while (!(await reader.read()).done) {} reader.releaseLock(); - await settle(reason); + await settle(); + } catch (error) { + try { + reader.releaseLock(); + } catch {} + await settleUnfenced(error); } }, }, diff --git a/packages/computer/src/workspace.ts b/packages/computer/src/workspace.ts index e50f1c70..818942e4 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -22,6 +22,7 @@ import { import { type ApplyResult, type ChangeCursor, + compareChangeCursors, Database, type DurableObjectStorageLike, initializeSchema, @@ -741,21 +742,10 @@ export class Workspace { intent.targetCursor, ); if (result.status === "pending") { - if (intent.attempt >= this.#retryMaxAttempts) { - return { - status: "exhausted", - backend: resolvedId, - ...(intent.runtimeId === undefined ? {} : { runtimeId: intent.runtimeId }), - attempt: intent.attempt, - error: "pending sync retry attempts exhausted", - }; - } - const next = this.#retryIntent( - resolvedId, - intent.attempt + 1, - intent.runtimeId, - result.targetCursor, - ); + // Hitting a batch budget is successful progress, not a failed + // retry. Reset the consecutive-failure count so large trees + // can drain through any number of bounded alarm turns. + const next = this.#retryIntent(resolvedId, 1, intent.runtimeId, result.targetCursor); await scheduler.schedule(next); return { status: "pending", @@ -881,28 +871,52 @@ export class Workspace { return this.#serialize(id, async (resolvedId) => { if (resolvedId === undefined) return {}; const existing = await scheduler.get(resolvedId); - if (existing !== undefined && (runtimeId === undefined || existing.runtimeId === runtimeId)) { + const sameRuntime = + existing !== undefined && (runtimeId === undefined || existing.runtimeId === runtimeId); + if (!captureTarget && sameRuntime) { return { backend: resolvedId, ...(existing.runtimeId === undefined ? {} : { runtimeId: existing.runtimeId }), ...(existing.targetCursor === undefined ? {} : { targetCursor: existing.targetCursor }), }; } + + const intentRuntimeId = runtimeId ?? (sameRuntime ? existing.runtimeId : undefined); let targetCursor: ChangeCursor | undefined; + let captureError: unknown; if (captureTarget) { - const handle = await this.#handleFor(resolvedId); - if (runtimeId !== undefined) { - assertExecutionRuntime("post-command sync", runtimeId, handle.runtimeId); + try { + const handle = await this.#handleFor(resolvedId); + if (runtimeId !== undefined) { + assertExecutionRuntime("post-command sync", runtimeId, handle.runtimeId); + } + targetCursor = + handle.sync === "none" + ? { rev: 0, path: null } + : { + rev: (await handle.rpc.sync.watermarks({ settle: true })).currentRev, + path: null, + }; + if ( + sameRuntime && + existing.targetCursor !== undefined && + compareChangeCursors(existing.targetCursor, targetCursor) > 0 + ) { + targetCursor = existing.targetCursor; + } + } catch (error) { + // Preserve a durable, unfenced retry when the settle/capture + // step fails. Its first pull will capture a fresh target. + captureError = error; + targetCursor = undefined; } - targetCursor = - handle.sync === "none" - ? { rev: 0, path: null } - : { rev: (await handle.rpc.sync.watermarks()).currentRev, path: null }; } - await scheduler.schedule(this.#retryIntent(resolvedId, 1, runtimeId, targetCursor)); + + await scheduler.schedule(this.#retryIntent(resolvedId, 1, intentRuntimeId, targetCursor)); + if (captureError !== undefined) throw captureError; return { backend: resolvedId, - ...(runtimeId === undefined ? {} : { runtimeId }), + ...(intentRuntimeId === undefined ? {} : { runtimeId: intentRuntimeId }), ...(targetCursor === undefined ? {} : { targetCursor }), }; }); diff --git a/packages/rpc/src/interface.ts b/packages/rpc/src/interface.ts index 5cf87698..967986bf 100644 --- a/packages/rpc/src/interface.ts +++ b/packages/rpc/src/interface.ts @@ -71,8 +71,14 @@ export interface SyncRPC { // progress. // // pushRev / fetchCursor only move when the receiver is acting as - // a sync peer. Otherwise they sit at 0. - watermarks(): Promise<{ currentRev: number; pushRev: number; fetchCursor: ChangeCursor }>; + // a sync peer. Otherwise they sit at 0. Pass `settle: true` to run + // the receiver's pre-fetch reconciliation before currentRev is read; + // deferred command sync uses that as its durable target fence. + watermarks(input?: { settle?: boolean }): Promise<{ + currentRev: number; + pushRev: number; + fetchCursor: ChangeCursor; + }>; // Materialise the receiver's view of a single path as a // ChangeEntry. Returns null when the path doesn't exist and diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index 638fedaa..83eacda8 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -201,7 +201,18 @@ class SyncRPCServer extends RpcTarget implements SyncRPC { return materialiseChange(this.db, path); } - async watermarks(): Promise<{ currentRev: number; pushRev: number; fetchCursor: ChangeCursor }> { + async watermarks(input: { settle?: boolean } = {}): Promise<{ + currentRev: number; + pushRev: number; + fetchCursor: ChangeCursor; + }> { + // Deferred command synchronization needs a cursor that includes + // writes still waiting in the userspace shim. Unlike the ordinary + // diagnostic read, a settled read propagates hook failures so the + // caller can persist an unfenced retry instead of a stale target. + if (input.settle === true && this.options.beforeFetch !== undefined) { + await this.options.beforeFetch(); + } return { currentRev: currentRev(this.db), pushRev: readWatermark(this.db, "pushRev"), diff --git a/packages/rpc/src/sync-driver.test.ts b/packages/rpc/src/sync-driver.test.ts index bfaec329..2f1fea3a 100644 --- a/packages/rpc/src/sync-driver.test.ts +++ b/packages/rpc/src/sync-driver.test.ts @@ -475,6 +475,37 @@ describe("SyncRPC server — beforeFetch hook", () => { b.close(); } }); + + it("settles before returning a deferred synchronization target", async () => { + const b = makeReceiverWithSpy(); + try { + const provider = new SQLiteWorkspaceProvider(b.db, { now: () => 2 }); + b.setBeforeFetch(() => { + provider.writeFileSync("/late.txt", "settled"); + }); + + const watermarks = await b.rpc.watermarks({ settle: true }); + + expect(b.calls).toBe(1); + expect(watermarks.currentRev).toBe(currentRev(b.db)); + expect(watermarks.currentRev).toBeGreaterThan(1); + } finally { + b.close(); + } + }); + + it("surfaces a failed settle instead of returning a stale target", async () => { + const b = makeReceiverWithSpy(); + try { + b.setBeforeFetch(() => { + throw new Error("settle failed"); + }); + + await expect(b.rpc.watermarks({ settle: true })).rejects.toThrow("settle failed"); + } finally { + b.close(); + } + }); }); describe("SyncRPC server — fetchChanges snapshots", () => { @@ -1405,4 +1436,29 @@ describe("bounded synchronization", () => { downstream.close(); } }); + + it("leaves an advanced fetch cursor alone when an old target is already satisfied", async () => { + const upstream = makePeer(); + const downstream = makePeer(); + try { + const provider = new SQLiteWorkspaceProvider(upstream.db, { now: () => 1 }); + await provider.writeFile("/one.txt", "one"); + const oldTarget = { rev: currentRev(upstream.db), path: null }; + await provider.writeFile("/two.txt", "two"); + await pullOnce(downstream.db, upstream.rpc); + const advanced = readFetchCursor(downstream.db); + + const result = await pullBatch(downstream.db, upstream.rpc, { + targetCursor: oldTarget, + budget: { maxEntries: 1, maxBytes: 1024 }, + }); + + expect(result.status).toBe("complete"); + expect(result.cursor).toEqual(advanced); + expect(readFetchCursor(downstream.db)).toEqual(advanced); + } finally { + upstream.close(); + downstream.close(); + } + }); }); diff --git a/packages/rpc/src/sync-driver.ts b/packages/rpc/src/sync-driver.ts index e9b0b557..56e556e2 100644 --- a/packages/rpc/src/sync-driver.ts +++ b/packages/rpc/src/sync-driver.ts @@ -354,6 +354,20 @@ async function pullBatchImpl( ): Promise { const backend = options.backend; const after = readFetchCursor(db, backend); + if ( + options.targetCursor !== undefined && + compareChangeCursors(after, options.targetCursor) >= 0 + ) { + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor: after, + targetCursor: options.targetCursor, + }; + } const fetchResult = await remote.fetchChanges({ after, through: options.targetCursor }); const pushCursor = readPushCursor(db, backend); const pushDiverged = fetchResult.appliedPushCursor.rev < pushCursor.rev; From 8f7123514fc76298dc3c151587bcfe0311f530c9 Mon Sep 17 00:00:00 2001 From: aron <263346377+aron-cf@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:02:05 +0000 Subject: [PATCH 14/14] computer: Rename exec sync modes Use the action-oriented `wait` and `defer` values to describe whether execution waits for post-command synchronization or records it for later. --- packages/computer/src/retry.test.ts | 12 ++++++------ packages/computer/src/runtime/types.ts | 4 ++-- packages/computer/src/shell.test.ts | 6 +++--- packages/computer/src/shell.ts | 6 +++--- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 7a100c26..81043c5c 100644 --- a/packages/computer/src/retry.test.ts +++ b/packages/computer/src/retry.test.ts @@ -483,7 +483,7 @@ describe("Workspace deferred synchronization", () => { const handle = await ws.runtime.exec("build", { encoding: "utf8", - sync: "deferred", + sync: "defer", }); const result = await handle.result(); @@ -521,7 +521,7 @@ describe("Workspace deferred synchronization", () => { retryScheduler: scheduler, }); - const handle = await ws.runtime.exec("build", { sync: "deferred" }); + const handle = await ws.runtime.exec("build", { sync: "defer" }); const result = await handle.result(); expect(settleInputs.at(-1)).toEqual({ settle: true }); @@ -549,7 +549,7 @@ describe("Workspace deferred synchronization", () => { retryScheduler: scheduler, }); - const handle = await ws.runtime.exec("build", { sync: "deferred" }); + const handle = await ws.runtime.exec("build", { sync: "defer" }); const result = await handle.result(); expect(result.sync).toMatchObject({ @@ -586,9 +586,9 @@ describe("Workspace deferred synchronization", () => { retryScheduler: scheduler, }); - const first = await ws.runtime.exec("first", { sync: "deferred" }); + const first = await ws.runtime.exec("first", { sync: "defer" }); await first.result(); - const second = await ws.runtime.exec("second", { sync: "deferred" }); + const second = await ws.runtime.exec("second", { sync: "defer" }); const result = await second.result(); expect(result.sync).toMatchObject({ targetCursor: { rev: 2, path: null } }); @@ -610,6 +610,6 @@ it("rejects deferred execution without a retry scheduler", async () => { }); const ws = new Workspace({ storage: new SQLiteTestStorage(), backends: [backend] }); - await expect(ws.runtime.exec("build", { sync: "deferred" })).rejects.toThrow("retryScheduler"); + await expect(ws.runtime.exec("build", { sync: "defer" })).rejects.toThrow("retryScheduler"); expect(execs).toBe(0); }); diff --git a/packages/computer/src/runtime/types.ts b/packages/computer/src/runtime/types.ts index edf9ccec..c92a9f64 100644 --- a/packages/computer/src/runtime/types.ts +++ b/packages/computer/src/runtime/types.ts @@ -110,7 +110,7 @@ export interface WorkspaceRuntimeExecOptions env?: Record; stdin?: Uint8Array | string; timeoutMs?: number; - sync?: "inline" | "deferred"; + sync?: "wait" | "defer"; } export interface WorkspaceRuntimeGetOptions { @@ -145,7 +145,7 @@ export interface ModuleExecutionInput { env?: Record; stdin?: Uint8Array | string; timeoutMs?: number; - sync?: "inline" | "deferred"; + sync?: "wait" | "defer"; } export interface ModuleExecutionEnvelope { diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index ff8a7943..bccfc945 100644 --- a/packages/computer/src/shell.test.ts +++ b/packages/computer/src/shell.test.ts @@ -317,7 +317,7 @@ describe("CommandExecutor.exec — push/pull bracket", () => { }, }; const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop", { - sync: "deferred", + sync: "defer", }); const { outcome } = await drain(execution); expect(order).toEqual(["push", "schedule"]); @@ -340,7 +340,7 @@ describe("CommandExecutor.exec — push/pull bracket", () => { return applied(7); }, }; - const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop"); + const execution = await new CommandExecutor(f.rpc.shell, sync).exec("noop", { sync: "wait" }); expect(execution.sync.pushed).toBe(5); const { outcome } = await drain(execution); expect(outcome).toEqual({ @@ -586,7 +586,7 @@ describe("CommandExecutor cancellation", () => { }, }; const execution = await new CommandExecutor(shell, sync).exec("build", { - sync: "deferred", + sync: "defer", }); const cancelling = execution.events.cancel("consumer stopped"); diff --git a/packages/computer/src/shell.ts b/packages/computer/src/shell.ts index 2b918e38..6d0fd3ad 100644 --- a/packages/computer/src/shell.ts +++ b/packages/computer/src/shell.ts @@ -81,7 +81,7 @@ export interface ExecOptions { // Standard input fed to the command. Bytes, or a string encoded // as UTF-8. stdin?: Uint8Array | string; - sync?: "inline" | "deferred"; + sync?: "wait" | "defer"; } export interface GetExecOptions { @@ -155,7 +155,7 @@ export class CommandExecutor { // with stale or incomplete workspace contents is not safe. async exec(source: string, options: ExecOptions = {}): Promise { assertNotTemplate(source); - if (options.sync === "deferred") await this.#sync.assertDeferredReady?.(); + if (options.sync === "defer") await this.#sync.assertDeferredReady?.(); const input: ShellExecInput = { source, id: options.id, @@ -180,7 +180,7 @@ export class CommandExecutor { // and the envelope can't be bound with `using` here. const drained = disposeOnDone(envelope.events, () => maybeDispose(envelope)); const wrapped = - options.sync === "deferred" + options.sync === "defer" ? withDeferredPostPull(drained, this.#sync, envelope.runtimeId) : withPostPull(drained, this.#sync, envelope.runtimeId); return {