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/index.ts b/packages/computer/src/index.ts index 725a0ea1..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, @@ -96,6 +97,7 @@ export { withWorkspace, } from "./with-workspace.js"; export { + type SyncBatchOptions, type SyncRetryIntent, type SyncRetryOptions, type SyncRetryScheduler, diff --git a/packages/computer/src/retry.test.ts b/packages/computer/src/retry.test.ts index 26575272..81043c5c 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({ @@ -396,3 +456,160 @@ describe("Workspace durable pending-sync retries", () => { expect(closes).toBe(1); }); }); + +describe("Workspace deferred synchronization", () => { + it("schedules before returning a deferred result", async () => { + const scheduler = new MemoryRetryScheduler(); + const backend = retryBackend({ + onExec() {}, + async fetchChanges() { + return { + currentCursor: { rev: 0, path: null }, + appliedPushCursor: { rev: 0, path: null }, + stream: new ReadableStream({ + start(controller) { + controller.close(); + }, + }), + }; + }, + }); + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + now: () => 5_000, + }); + + const handle = await ws.runtime.exec("build", { + encoding: "utf8", + sync: "defer", + }); + const result = await handle.result(); + + expect(result.sync).toMatchObject({ + status: "pending", + backend: "sandbox", + targetCursor: { rev: 0, path: null }, + }); + expect(scheduler.intents.get("sandbox")).toMatchObject({ + backend: "sandbox", + targetCursor: { rev: 0, path: null }, + }); + }); + + it("settles the remote filesystem before capturing the deferred target", async () => { + const scheduler = new MemoryRetryScheduler(); + const settleInputs: unknown[] = []; + const backend = retryBackend({ + onExec() {}, + async fetchChanges() { + throw new Error("not used"); + }, + async watermarks(input) { + settleInputs.push(input); + return { + currentRev: input?.settle === true ? 5 : 0, + pushRev: 0, + fetchCursor: { rev: 0, path: null }, + }; + }, + }); + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + }); + + const handle = await ws.runtime.exec("build", { sync: "defer" }); + const result = await handle.result(); + + expect(settleInputs.at(-1)).toEqual({ settle: true }); + expect(result.sync).toMatchObject({ targetCursor: { rev: 5, path: null } }); + expect(scheduler.intents.get("sandbox")).toMatchObject({ + targetCursor: { rev: 5, path: null }, + }); + }); + + it("persists an unfenced intent when target capture fails", async () => { + const scheduler = new MemoryRetryScheduler(); + const backend = retryBackend({ + onExec() {}, + async fetchChanges() { + throw new Error("not used"); + }, + async watermarks(input) { + if (input?.settle === true) throw new Error("settle failed"); + return { currentRev: 0, pushRev: 0, fetchCursor: { rev: 0, path: null } }; + }, + }); + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + }); + + const handle = await ws.runtime.exec("build", { sync: "defer" }); + const result = await handle.result(); + + expect(result.sync).toMatchObject({ + status: "pending", + error: expect.stringContaining("settle failed"), + }); + expect(scheduler.intents.get("sandbox")).toEqual( + expect.objectContaining({ backend: "sandbox", attempt: 1 }), + ); + expect(scheduler.intents.get("sandbox")).not.toHaveProperty("targetCursor"); + }); + + it("widens an existing intent when another deferred command finishes", async () => { + const scheduler = new MemoryRetryScheduler(); + let currentRev = 0; + const backend = retryBackend({ + onExec() { + currentRev++; + }, + async fetchChanges() { + throw new Error("not used"); + }, + async watermarks() { + return { + currentRev, + pushRev: 0, + fetchCursor: { rev: 0, path: null }, + }; + }, + }); + const ws = new Workspace({ + storage: new SQLiteTestStorage(), + backends: [backend], + retryScheduler: scheduler, + }); + + const first = await ws.runtime.exec("first", { sync: "defer" }); + await first.result(); + const second = await ws.runtime.exec("second", { sync: "defer" }); + const result = await second.result(); + + expect(result.sync).toMatchObject({ targetCursor: { rev: 2, path: null } }); + expect(scheduler.intents.get("sandbox")).toMatchObject({ + targetCursor: { rev: 2, path: null }, + }); + }); +}); + +it("rejects deferred execution without a retry scheduler", async () => { + let execs = 0; + const backend = retryBackend({ + onExec: () => { + execs += 1; + }, + async fetchChanges() { + throw new Error("not expected"); + }, + }); + const ws = new Workspace({ storage: new SQLiteTestStorage(), backends: [backend] }); + + await expect(ws.runtime.exec("build", { sync: "defer" })).rejects.toThrow("retryScheduler"); + expect(execs).toBe(0); +}); 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..c92a9f64 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?: "wait" | "defer"; } export interface WorkspaceRuntimeGetOptions { @@ -144,6 +145,7 @@ export interface ModuleExecutionInput { env?: Record; stdin?: Uint8Array | string; timeoutMs?: number; + sync?: "wait" | "defer"; } export interface ModuleExecutionEnvelope { diff --git a/packages/computer/src/shell.test.ts b/packages/computer/src/shell.test.ts index 07928c19..bccfc945 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: "defer", + }); + 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 = { @@ -309,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({ @@ -496,3 +527,79 @@ 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); + }); + + 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: "defer", + }); + + 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 1747862e..6d0fd3ad 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?: "wait" | "defer"; } export interface GetExecOptions { @@ -96,6 +105,12 @@ 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; + }>; + assertDeferredReady?(): void | Promise; } type ShellExecInput = Parameters[0]; @@ -140,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 === "defer") await this.#sync.assertDeferredReady?.(); const input: ShellExecInput = { source, id: options.id, @@ -163,12 +179,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 === "defer" + ? 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 }, }; } @@ -255,6 +274,9 @@ export function withPostPull( try { reader.releaseLock(); } catch {} + try { + await sync.onPullPending?.(error, runtimeId); + } catch {} resolveOutcome({ applied: 0, skipped: [], @@ -268,6 +290,9 @@ export function withPostPull( await reader.cancel(reason); } finally { reader.releaseLock(); + try { + await sync.onPullPending?.(reason, runtimeId); + } catch {} resolveOutcome({ applied: 0, skipped: [], @@ -281,6 +306,94 @@ 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 () => { + let metadata: Awaited>> = {}; + let error: unknown; + 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 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) { + 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 settleUnfenced(error); + controller.error(error); + } + }, + 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 { + while (!(await reader.read()).done) {} + reader.releaseLock(); + await settle(); + } catch (error) { + try { + reader.releaseLock(); + } catch {} + await settleUnfenced(error); + } + }, + }, + { 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.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..818942e4 100644 --- a/packages/computer/src/workspace.ts +++ b/packages/computer/src/workspace.ts @@ -10,9 +10,19 @@ // 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, + compareChangeCursors, Database, type DurableObjectStorageLike, initializeSchema, @@ -53,6 +63,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; @@ -72,6 +83,11 @@ export interface SyncRetryScheduler { clear(backend: string): Promise; } +export interface SyncBatchOptions extends SyncBatchBudget { + mode: "batch"; + targetCursor?: ChangeCursor; +} + export interface SyncRetryOptions { initialDelayMs?: number; maxDelayMs?: number; @@ -87,7 +103,9 @@ export type WorkspaceRetryPendingSyncResult = runtimeId?: string; attempt: number; notBefore: number; - error: string; + cursor?: ChangeCursor; + targetCursor?: ChangeCursor; + error?: string; } | { status: "exhausted"; @@ -101,6 +119,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 @@ -594,33 +616,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, + ); + }, + ), + ); } /** @@ -631,7 +711,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"); @@ -652,7 +735,25 @@ 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") { + // 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", + ...next, + cursor: result.cursor, + targetCursor: result.targetCursor, + }; + } await scheduler.clear(resolvedId); return { status: "complete", @@ -683,13 +784,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, @@ -715,26 +856,85 @@ 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; + 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) { + 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; + } } - await scheduler.schedule(this.#retryIntent(resolvedId, 1, runtimeId)); + + await scheduler.schedule(this.#retryIntent(resolvedId, 1, intentRuntimeId, targetCursor)); + if (captureError !== undefined) throw captureError; + return { + backend: resolvedId, + ...(intentRuntimeId === undefined ? {} : { runtimeId: intentRuntimeId }), + ...(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, @@ -926,6 +1126,7 @@ export class Workspace { timeoutMs: input.timeoutMs, env: input.env, stdin: input.stdin, + sync: input.sync, }); this.#rememberExecutionRuntime(id, envelope.id, envelope.runtimeId); return { @@ -1161,7 +1362,15 @@ 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), + assertDeferredReady: () => { + if (this.#retryScheduler === undefined) { + throw new Error("Workspace requires a retryScheduler for deferred synchronization"); + } + }, }, this.#observer, dispatch, @@ -1259,6 +1468,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, 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..4fc180b4 100644 --- a/packages/dofs/src/sync/watermarks.ts +++ b/packages/dofs/src/sync/watermarks.ts @@ -102,6 +102,35 @@ 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, + ); + 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( + 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, diff --git a/packages/rpc/src/interface.ts b/packages/rpc/src/interface.ts index e12c3293..967986bf 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; @@ -62,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 c9c46dcb..83eacda8 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), @@ -193,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 5fb6b8a7..2f1fea3a 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 @@ -468,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", () => { @@ -1282,3 +1320,145 @@ 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(); + } + }); + + 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 fba350a5..56e556e2 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"); @@ -71,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); } @@ -278,89 +305,402 @@ 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); } } -// 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. -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); +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 minimumCursor(a: ChangeCursor, b: ChangeCursor): ChangeCursor { + return compareChangeCursors(a, b) <= 0 ? a : b; +} + +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); + 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; + 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, + ); + try { + assertAppliedPushCursor(fetchResult.appliedPushCursor, pushCursor); + } catch (error) { + maybeDispose(fetchResult); + throw error; + } + if (compareChangeCursors(after, targetCursor) >= 0) { + maybeDispose(fetchResult); + 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 = await 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 = await 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[] = []; + let exhausted = true; + for await (const entry of coalesceChanges(db, cursor, { through: targetCursor })) { + candidates.push(entry); + if (candidates.length >= options.budget.maxEntries) { + exhausted = false; + break; } } - 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)); + 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(); + }, + }); + const response = await remote.push({ + senderRev: targetCursor.rev, + senderCursor: targetCursor, + changes, + }); + assertAppliedPushCursor(response.appliedPushCursor, targetCursor); + writePushCursor(db, targetCursor, backend); + return { + status: "complete", + entries: 0, + bytes: 0, + applied: 0, + skipped: [], + cursor: targetCursor, + targetCursor, + }; } - const missing = wantedHashes.filter((h) => !remoteHas.has(hex(h))); - if (missing.length > 0) { + 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 h of missing) { + for (const chunk of transferable) { const row = db.one<{ bytes: Uint8Array }>( "SELECT bytes FROM vfs_blob_bytes WHERE hash = ?", - h, + chunk.hash, ); - if (row === undefined) { - throw new Error(`pushOnce: missing local blob ${hex(h)}`); - } - yield { hash: h, bytes: row.bytes }; + 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 bytesStream = new ReadableStream<{ hash: Uint8Array; bytes: Uint8Array }>({ + 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(bytesStream); + 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 acknowledgedCursor = exhausted ? targetCursor : lastCursor; const entryStream = new ReadableStream({ start(controller) { - for (const e of entries) controller.enqueue(e); + for (const entry of selected) controller.enqueue(entry); 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; + const response = await remote.push({ + senderRev: targetCursor.rev, + senderCursor: acknowledgedCursor, + changes: entryStream, + }); + assertAppliedPushCursor(response.appliedPushCursor, acknowledgedCursor); + writePushCursor(db, acknowledgedCursor, backend); + return { + status: compareChangeCursors(acknowledgedCursor, targetCursor) >= 0 ? "complete" : "pending", + entries: selected.length, + bytes, + applied: response.applied ?? selected.length, + skipped: [], + cursor: acknowledgedCursor, + 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. +export async function pushOnce(db: Database, remote: SyncRPC, backend?: string): Promise { + 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 }, + }); + targetCursor = result.targetCursor; + pushed += result.entries; + if (result.status === "complete") return pushed; + } } // One full tick: pull, then push. The order matters \u2014 pulling @@ -400,7 +740,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; @@ -426,8 +766,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; }