diff --git a/apps/dispatcher/src/env.ts b/apps/dispatcher/src/env.ts index 7baddfe..341d64a 100644 --- a/apps/dispatcher/src/env.ts +++ b/apps/dispatcher/src/env.ts @@ -308,6 +308,14 @@ export interface Env { */ readonly SANDBOX_PREVIEW_HOSTNAME?: string; + /** + * The Worker-wide sandbox transport for a repo no `sandbox.transport:` + * key pins: `http`, `websocket` or `rpc`, and `rpc` when unset or not a + * transport name. The dispatcher pins the resolved transport onto every + * container (`resolveSandboxTransport`). A var, not a secret. + */ + readonly SANDBOX_TRANSPORT?: string; + /** * The dispatcher's own public origin (e.g. * `https://flare-dispatch..workers.dev`, or a custom domain) — diff --git a/apps/dispatcher/src/workflow.ts b/apps/dispatcher/src/workflow.ts index 6304ae7..1e9318c 100644 --- a/apps/dispatcher/src/workflow.ts +++ b/apps/dispatcher/src/workflow.ts @@ -569,17 +569,11 @@ export class RunWorkflow extends WorkflowEntrypoint { } } - // Per-repo container transport — the canary knob for `rpc`. - // - // The SDK's streaming file APIs exist only on its `rpc` client (`http` is - // the "route-based compatibility" path), which is why the R2 dependency - // cache misses on every run: its restore hands `writeFile` a stream, the SDK - // routes any stream to `writeFileStream`, and that raises off `rpc`. - // `SANDBOX_TRANSPORT=rpc` as a Worker var would fix it for every repo at - // once; this lets one prove it first. - // - // A value the SDK does not recognise is ignored by it with a warning, so a - // typo degrades to the default rather than breaking the run. + // Per-repo container transport pin. It outranks the Worker-wide + // `SANDBOX_TRANSPORT` in either direction, and with neither set containers + // run on `rpc`, the one transport that streams a cache archive in + // (`resolveSandboxTransport`). A value that is not a transport name is + // skipped, so a typo falls through rather than breaking the run. // Container path only — the substrate owns its own transport, so a run // there should not pay a KV read to be told about one it cannot use. const sandboxTransport = @@ -716,10 +710,9 @@ export class RunWorkflow extends WorkflowEntrypoint { ? { aiGatewayAuthToken: this.env.AI_GATEWAY_AUTH_TOKEN } : {}), sandboxPreviewHostname: this.env.SANDBOX_PREVIEW_HOSTNAME, - ...(sandboxTransport === "rpc" || - sandboxTransport === "websocket" || - sandboxTransport === "http" - ? { sandboxTransport } + ...(sandboxTransport != null ? { sandboxTransport } : {}), + ...(this.env.SANDBOX_TRANSPORT !== undefined + ? { sandboxTransportDefault: this.env.SANDBOX_TRANSPORT } : {}), ...(publicOrigin !== undefined ? { publicOrigin } : {}), ...(logsBaseUrl !== undefined ? { logsViewerBase: logsBaseUrl } : {}), diff --git a/infra/Dockerfile.sandbox-demo b/infra/Dockerfile.sandbox-demo index 31096e6..aa57f52 100644 --- a/infra/Dockerfile.sandbox-demo +++ b/infra/Dockerfile.sandbox-demo @@ -128,6 +128,70 @@ RUN if [ "$WITH_RUST" = "true" ]; then \ echo "WITH_RUST=false — no Rust toolchain baked. Lean image."; \ fi +# ── Prebuilt DuckDB shared library (gated by the WITH_RUST build arg) ─────── +# +# specs/adr/0007-cost-efficient-execution.md, "Prebuilt heavy native +# dependencies in the image". A cold `cargo test` of a workspace on the +# `duckdb` crate spends most of its time compiling DuckDB's C++ from source. +# This layer ships the release's `libduckdb.so` and headers, pinned by +# SHA-256, under /opt/duckdb, so a consumer that links DuckDB skips that +# compile. The static archive in the release zip is left out: the layer +# serves dynamic linking only, and the archive doubles the layer's size. +# +# libduckdb-sys (1.10505.x, build.rs) reads DUCKDB_LIB_DIR and +# DUCKDB_INCLUDE_DIR only in its `build_linked` path, compiled under +# `#[cfg(not(feature = "bundled"))]`. A consumer that builds with the +# `bundled` feature compiles from source and ignores both, so exporting them +# changes nothing for it. In linked mode the build script emits +# `rustc-link-search=$DUCKDB_LIB_DIR` and `rustc-link-lib=dylib=duckdb`, which +# links but does not make the library loadable at runtime; the ld.so.conf.d +# entry below puts /opt/duckdb/lib on the loader path for every exec user. +# +# Consumer opt-in, owned by the consumer: +# - build `duckdb` / `libduckdb-sys` without the `bundled` feature in its CI +# run command (release builds may stay bundled); +# - assert the loaded library's version at startup (`SELECT version()` or +# `duckdb_library_version()` against the pinned crate's DuckDB version), +# because a linked library of the wrong version compiles cleanly and +# misbehaves at runtime; +# - move DUCKDB_VERSION and the digest below together with its crate pin: +# libduckdb-sys 1.10505.x pairs with DuckDB v1.5.5. +# +# In a linked Linux build DUCKDB_LIB_DIR wins over every other source, +# `DUCKDB_DOWNLOAD_LIB=1` included: a consumer whose crate pin differs from +# DUCKDB_VERSION links this library, not the one its crate would download, +# unless its run command unsets DUCKDB_LIB_DIR first. +# +# The ENV paths are set in every variant; with WITH_RUST=false the directory +# is absent and nothing in the image reads them. The release asset is +# linux-amd64 only, matching the container platform; any other build +# architecture fails the image build. +ARG DUCKDB_VERSION=v1.5.5 +ENV DUCKDB_LIB_DIR=/opt/duckdb/lib \ + DUCKDB_INCLUDE_DIR=/opt/duckdb/include +RUN if [ "$WITH_RUST" = "true" ]; then \ + set -eux; \ + arch="$(uname -m)"; \ + case "$arch" in \ + x86_64) sha256=1fb8ce388157d84a25abe685a8a2520bf00c00321821968e4bb398fd766e7abb ;; \ + *) echo "libduckdb: unsupported architecture $arch" >&2; exit 1 ;; \ + esac; \ + archive="/tmp/libduckdb-${DUCKDB_VERSION}-linux-amd64.zip"; \ + curl -fsSL -o "$archive" \ + "https://github.com/duckdb/duckdb/releases/download/${DUCKDB_VERSION}/libduckdb-linux-amd64.zip"; \ + echo "${sha256} ${archive}" | sha256sum -c -; \ + mkdir -p "$DUCKDB_LIB_DIR" "$DUCKDB_INCLUDE_DIR"; \ + unzip -q -j "$archive" libduckdb.so -d "$DUCKDB_LIB_DIR"; \ + unzip -q -j "$archive" duckdb.h duckdb.hpp duckdb_extension.h -d "$DUCKDB_INCLUDE_DIR"; \ + rm -f "$archive"; \ + chmod -R a+rX /opt/duckdb; \ + echo "$DUCKDB_LIB_DIR" > /etc/ld.so.conf.d/duckdb.conf; \ + ldconfig; \ + ldconfig -p | grep -q "libduckdb.so "; \ + else \ + echo "WITH_RUST=false — no prebuilt DuckDB baked. Lean image."; \ + fi + # ── demo-agent layer ──────────────────────────────────────────────────────── # # The `product-demo` run's bundled CLI (record start/stop, play, gif, diff --git a/packages/core/src/primitives/install-cached.ts b/packages/core/src/primitives/install-cached.ts index 67013a3..ce4234c 100644 --- a/packages/core/src/primitives/install-cached.ts +++ b/packages/core/src/primitives/install-cached.ts @@ -47,6 +47,15 @@ type ToolSpec = { readonly paths: readonly string[]; /** Base the `paths` are packed from, when the tool's home is not the checkout. */ readonly base?: string; + /** + * Command run after every restore, hit or miss, when `paths` cannot hold + * the whole installed tree. A pnpm workspace links each member's + * dependencies into its own `packages//node_modules`, which the root + * `node_modules` archive does not carry; a hit without a relink leaves + * every member unable to import its dependencies. After a miss the install + * already ran, so the relink finds the tree up to date and returns at once. + */ + readonly relink?: string; }; // Exported so the invariant above is assertable. It is the contract this @@ -60,6 +69,7 @@ export const TOOLS = { // NOT `.pnpm-store`: nothing creates one in a checkout, and `tar czf` exits // 2 on a missing member — listing it disabled the whole pnpm save. paths: ["node_modules"], + relink: "pnpm install --frozen-lockfile --prefer-offline", }, npm: { lockfile: "package-lock.json", install: "npm ci", paths: ["node_modules"] }, cargo: { @@ -117,7 +127,7 @@ export const installCached = (opts: { return; } const spec: ToolSpec = TOOLS[tool]; - const { lockfile, install, paths, base } = spec; + const { lockfile, install, paths, base, relink } = spec; // Hash the lockfile inside the container to key the R2 cache entry. const hash = yield* sandbox.exec({ @@ -144,4 +154,14 @@ export const installCached = (opts: { ...(opts.redactValues !== undefined ? { redactValues: opts.redactValues } : {}), }), }); + + if (relink !== undefined) { + yield* sandbox.exec({ + cwd: opts.dir, + container: opts.container, + command: relink, + ...(opts.env !== undefined ? { env: opts.env } : {}), + ...(opts.redactValues !== undefined ? { redactValues: opts.redactValues } : {}), + }); + } }); diff --git a/packages/core/src/primitives/primitives.test.ts b/packages/core/src/primitives/primitives.test.ts index 000f322..17013a8 100644 --- a/packages/core/src/primitives/primitives.test.ts +++ b/packages/core/src/primitives/primitives.test.ts @@ -7,6 +7,7 @@ // covers `workspace`, `installCached`, `sharded`, `bootApp`, `probeHttp`. import { Effect } from "effect"; +import { Cache } from "../services/cache"; import { sandbox } from "../services/sandbox"; import { describe, expect, it } from "vitest"; import { bootApp } from "./boot-app"; @@ -155,6 +156,42 @@ describe("installCached", () => { expect(cmds.some((c) => c.includes("pnpm install"))).toBe(true); }); + // PROOF (a pnpm hit still relinks the workspace): the archive holds the + // root `node_modules` only, while pnpm links each workspace member's + // dependencies into `packages//node_modules`. A hit that skipped every + // install left those members unable to import anything. + it("relinks a pnpm workspace after a cache hit, without the full install", async () => { + const { layer, handles } = makeCFRuntimeTest({ + sandboxProgram: { + "test -f pnpm-lock.yaml": { exitCode: 0 }, + "sha256sum pnpm-lock.yaml": { exitCode: 0, stdout: "feedface\n" }, + "pnpm install --frozen-lockfile --prefer-offline": { exitCode: 0 }, + }, + }); + const hit = { + restoreOr: () => Effect.void, + save: () => Effect.void, + } as unknown as Cache["Type"]; + + await Effect.runPromise( + installCached({ container: { id: "c-1" }, dir: "/workspace/repo" }).pipe( + Effect.provideService(Cache, hit), + Effect.provide(layer), + ), + ); + + const cmds = handles.sandbox.execs.map((e) => String(e.command)); + expect(cmds).toContain("pnpm install --frozen-lockfile --prefer-offline"); + expect(cmds).not.toContain("pnpm install --frozen-lockfile"); + }); + + it("relinks nothing for a tool whose cached paths hold the whole tree", () => { + expect(TOOLS.pnpm.relink).toBe("pnpm install --frozen-lockfile --prefer-offline"); + expect("relink" in TOOLS.npm).toBe(false); + expect("relink" in TOOLS.cargo).toBe(false); + expect("relink" in TOOLS.uv).toBe(false); + }); + it("no recognized lockfile → no install run", async () => { const { layer, handles } = makeCFRuntimeTest({ sandboxProgram: { diff --git a/packages/runtime-cf/src/cache-r2-live.test.ts b/packages/runtime-cf/src/cache-r2-live.test.ts new file mode 100644 index 0000000..f4d512a --- /dev/null +++ b/packages/runtime-cf/src/cache-r2-live.test.ts @@ -0,0 +1,209 @@ +// Unit coverage for the live `makeCacheR2Live` closures: the transport gate on +// restore, the `unapplied` outcome, and the save's upload-once check. +// +// `@cloudflare/sandbox` is `vi.mock`ed to an in-memory fake box, and the R2 +// bucket is a plain fake, so these run in the Node pool with no container and +// no Miniflare. +// +// Spec: specs/adr/0007-cost-efficient-execution.md § Upload each cache key once. + +import { Effect, Logger, LogLevel } from "effect"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { Cache } from "@fractalboxdev/flare-dispatch-core"; + +type FakeBox = { + execs: string[]; + writes: number; + writeFile: (path: string, body: unknown) => Promise; + exec: (command: string) => Promise<{ exitCode: number; stdout: string; stderr: string }>; + readFileStream: () => Promise>; +}; + +let box: FakeBox; + +vi.mock("@cloudflare/sandbox", () => ({ + getSandbox: () => box, +})); + +const { makeCacheR2Live } = await import("./cache-r2"); + +const makeBox = (writeFile?: () => Promise, untarExit = 0): FakeBox => { + const b: FakeBox = { + execs: [], + writes: 0, + writeFile: async () => { + b.writes += 1; + if (writeFile !== undefined) await writeFile(); + }, + exec: async (command) => { + b.execs.push(command); + if (command.startsWith("tar xzf") && untarExit !== 0) + return { exitCode: untarExit, stdout: "", stderr: "gzip: stdin: unexpected end of file" }; + return { exitCode: 0, stdout: command.startsWith("stat") ? "3\n" : "", stderr: "" }; + }, + readFileStream: async () => + new ReadableStream({ + start: (c) => { + c.enqueue(new Uint8Array([1, 2, 3])); + c.close(); + }, + }), + }; + return b; +}; + +/** A fake bucket holding `stored` keys; records which calls touched it. */ +const makeBucket = (stored: readonly string[]) => { + const keys = new Set(stored); + const calls: string[] = []; + let cancelled = false; + const bucket = { + head: async (key: string) => { + calls.push(`head ${key}`); + return keys.has(key) ? { key } : null; + }, + get: async (key: string) => { + calls.push(`get ${key}`); + if (!keys.has(key)) return null; + return { + body: new ReadableStream({ + cancel: () => { + cancelled = true; + }, + }), + }; + }, + put: async (key: string) => { + calls.push(`put ${key}`); + keys.add(key); + }, + }; + return { bucket: bucket as unknown as R2Bucket, calls, wasCancelled: () => cancelled }; +}; + +const ns = {} as DurableObjectNamespace; +const KEY = "cache/o/r/pnpm-abc.tar.gz"; +const container = { id: "c1" } as const; + +const run = ( + layer: ReturnType, + times = 1, +): Promise<{ onMiss: number; lines: string[] }> => { + const lines: string[] = []; + let onMiss = 0; + const once = Effect.flatMap(Cache, (c) => + c.restoreOr({ + key: "pnpm-abc", + paths: ["node_modules"], + container, + dir: "/workspace/r", + onMiss: () => + Effect.sync(() => { + onMiss += 1; + }), + }), + ); + return Effect.runPromise( + Effect.repeatN(once, times - 1).pipe( + Effect.provide(layer), + Effect.provide( + Logger.replace( + Logger.defaultLogger, + Logger.make(({ message }) => lines.push(String(message))), + ), + ), + Logger.withMinimumLogLevel(LogLevel.Info), + ), + ).then(() => ({ onMiss, lines })); +}; + +beforeEach(() => { + box = makeBox(); +}); + +describe("makeCacheR2Live restore", () => { + it("http transport + stored archive — unapplied without reading the body, no re-upload", async () => { + const { bucket, calls } = makeBucket([KEY]); + const { onMiss } = await run(makeCacheR2Live(bucket, ns, "o/r", "http")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`head ${KEY}`]); + expect(box.writes).toBe(0); + expect(box.execs.some((c) => c.startsWith("tar czf"))).toBe(false); + }); + + it("http transport — logs the transport once per execution", async () => { + const { bucket } = makeBucket([KEY]); + const { onMiss, lines } = await run(makeCacheR2Live(bucket, ns, "o/r", "http"), 3); + expect(onMiss).toBe(3); + expect(lines.filter((l) => l.includes("http sandbox transport"))).toHaveLength(1); + }); + + it("http transport + no archive — miss: onMiss, pack, upload", async () => { + const { bucket, calls } = makeBucket([]); + const { onMiss } = await run(makeCacheR2Live(bucket, ns, "o/r", "http")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`head ${KEY}`, `head ${KEY}`, `put ${KEY}`]); + expect(box.execs.some((c) => c.startsWith("tar czf"))).toBe(true); + }); + + it("rpc transport + stored archive — hit: streamed in and extracted", async () => { + const { bucket, calls } = makeBucket([KEY]); + const { onMiss } = await run(makeCacheR2Live(bucket, ns, "o/r", "rpc")); + expect(onMiss).toBe(0); + expect(calls).toEqual([`get ${KEY}`]); + expect(box.writes).toBe(1); + expect(box.execs.some((c) => c.startsWith("tar xzf"))).toBe(true); + }); + + it("websocket transport + stored archive — unapplied from the head alone", async () => { + const { bucket, calls } = makeBucket([KEY]); + const { onMiss, lines } = await run(makeCacheR2Live(bucket, ns, "o/r", "websocket")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`head ${KEY}`]); + expect(box.writes).toBe(0); + expect(lines.some((l) => l.includes("websocket sandbox transport"))).toBe(true); + }); + + it("rpc transport + no archive — miss: onMiss, pack, upload", async () => { + const { bucket, calls } = makeBucket([]); + const { onMiss } = await run(makeCacheR2Live(bucket, ns, "o/r", "rpc")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`get ${KEY}`, `head ${KEY}`, `put ${KEY}`]); + expect(box.writes).toBe(0); + expect(box.execs.some((c) => c.startsWith("tar czf"))).toBe(true); + }); + + it("rpc transport + a failing tar xzf — unapplied: onMiss runs, no save", async () => { + box = makeBox(undefined, 2); + const { bucket, calls } = makeBucket([KEY]); + const { onMiss, lines } = await run(makeCacheR2Live(bucket, ns, "o/r", "rpc")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`get ${KEY}`]); + expect(box.execs.some((c) => c.startsWith("tar czf"))).toBe(false); + expect(lines.join("\n")).toContain("tar xzf exited 2"); + }); + + it("rpc transport + a failing apply — unapplied: onMiss runs, body cancelled, no save", async () => { + box = makeBox(() => Promise.reject(new Error("writeFileStream requires the rpc transport"))); + const { bucket, calls, wasCancelled } = makeBucket([KEY]); + const { onMiss, lines } = await run(makeCacheR2Live(bucket, ns, "o/r", "rpc")); + expect(onMiss).toBe(1); + expect(calls).toEqual([`get ${KEY}`]); + expect(wasCancelled()).toBe(true); + expect(box.execs.some((c) => c.startsWith("tar czf"))).toBe(false); + expect(lines.join("\n")).toContain("writeFileStream requires the rpc transport"); + }); +}); + +describe("makeCacheR2Live save", () => { + it("a stored key skips tar czf and the upload", async () => { + const { bucket, calls } = makeBucket([KEY]); + await Effect.runPromise( + Effect.flatMap(Cache, (c) => + c.save({ key: "pnpm-abc", paths: ["node_modules"], container, dir: "/workspace/r" }), + ).pipe(Effect.provide(makeCacheR2Live(bucket, ns, "o/r", "rpc"))), + ); + expect(calls).toEqual([`head ${KEY}`]); + expect(box.execs).toEqual([]); + }); +}); diff --git a/packages/runtime-cf/src/cache-r2.test.ts b/packages/runtime-cf/src/cache-r2.test.ts index 7b46ab7..c1c2e52 100644 --- a/packages/runtime-cf/src/cache-r2.test.ts +++ b/packages/runtime-cf/src/cache-r2.test.ts @@ -2,13 +2,15 @@ // // Drives composeRestoreOr with fake `restore` / `save` Effects — no container, // no R2, no Miniflare. Covers the best-effort contract: a hit skips `onMiss`, -// a miss runs `onMiss` then `save`, a restore error degrades to a miss, a -// `save` failure is swallowed, and an `onMiss` failure propagates. +// a miss runs `onMiss` then `save`, `unapplied` runs `onMiss` and skips +// `save`, a restore error degrades to a miss, a `save` failure is swallowed, +// and an `onMiss` failure propagates. Also covers `composeSave`'s +// upload-once check. import { Effect, Exit, Logger, LogLevel } from "effect"; import { describe, expect, it } from "vitest"; import { CacheError } from "@fractalboxdev/flare-dispatch-core"; -import { composeRestoreOr } from "./cache-restore-or"; +import { composeRestoreOr, composeSave } from "./cache-restore-or"; const container = { id: "c1" } as const; const baseOpts = { key: "k", paths: ["node_modules"], container }; @@ -18,7 +20,7 @@ describe("composeRestoreOr", () => { let onMissRan = false; let saveRan = false; const restoreOr = composeRestoreOr( - () => Effect.succeed(true), + () => Effect.succeed("hit" as const), () => Effect.sync(() => { saveRan = true; @@ -41,7 +43,7 @@ describe("composeRestoreOr", () => { let onMissRan = false; let saveRan = false; const restoreOr = composeRestoreOr( - () => Effect.succeed(false), + () => Effect.succeed("miss" as const), () => Effect.sync(() => { saveRan = true; @@ -60,12 +62,52 @@ describe("composeRestoreOr", () => { expect(saveRan).toBe(true); }); - it("a restore error is treated as a miss — onMiss runs", async () => { + // ADR-0007: the key is content-addressed, so the stored archive already + // holds what a second pack and upload would write. + it("unapplied — onMiss runs, save is skipped", async () => { let onMissRan = false; + let saveRan = false; const restoreOr = composeRestoreOr( - () => Effect.fail(new CacheError({ phase: "restore", key: "k", cause: "boom" })), + () => Effect.succeed("unapplied" as const), + () => + Effect.sync(() => { + saveRan = true; + }), + ); + await Effect.runPromise( + restoreOr({ + ...baseOpts, + onMiss: () => + Effect.sync(() => { + onMissRan = true; + }), + }), + ); + expect(onMissRan).toBe(true); + expect(saveRan).toBe(false); + }); + + it("unapplied — an onMiss failure still propagates", async () => { + const restoreOr = composeRestoreOr( + () => Effect.succeed("unapplied" as const), () => Effect.void, ); + const exit = await Effect.runPromiseExit( + restoreOr({ ...baseOpts, onMiss: () => Effect.fail("install failed" as const) }), + ); + expect(Exit.isFailure(exit)).toBe(true); + }); + + it("a restore error is treated as a miss — onMiss runs, then save", async () => { + let onMissRan = false; + let saveRan = false; + const restoreOr = composeRestoreOr( + () => Effect.fail(new CacheError({ phase: "restore", key: "k", cause: "boom" })), + () => + Effect.sync(() => { + saveRan = true; + }), + ); await Effect.runPromise( restoreOr({ ...baseOpts, @@ -76,11 +118,12 @@ describe("composeRestoreOr", () => { }), ); expect(onMissRan).toBe(true); + expect(saveRan).toBe(true); }); it("a save failure is swallowed — the run still succeeds", async () => { const restoreOr = composeRestoreOr( - () => Effect.succeed(false), + () => Effect.succeed("miss" as const), () => Effect.fail(new CacheError({ phase: "save", key: "k", cause: "boom" })), ); const exit = await Effect.runPromiseExit( @@ -92,7 +135,7 @@ describe("composeRestoreOr", () => { it("a swallowed save failure is logged, not discarded silently", async () => { const lines: string[] = []; const restoreOr = composeRestoreOr( - () => Effect.succeed(false), + () => Effect.succeed("miss" as const), () => Effect.fail(new CacheError({ phase: "save", key: "k", cause: "tar czf exited 2" })), ); const exit = await Effect.runPromiseExit( @@ -140,7 +183,7 @@ describe("composeRestoreOr", () => { const restoreOr = composeRestoreOr( (o) => { seen.push(o.base); - return Effect.succeed(false); + return Effect.succeed("miss" as const); }, (o) => { seen.push(o.base); @@ -160,7 +203,7 @@ describe("composeRestoreOr", () => { it("an onMiss failure propagates", async () => { const restoreOr = composeRestoreOr( - () => Effect.succeed(false), + () => Effect.succeed("miss" as const), () => Effect.void, ); const exit = await Effect.runPromiseExit( @@ -172,3 +215,59 @@ describe("composeRestoreOr", () => { expect(Exit.isFailure(exit)).toBe(true); }); }); + +describe("composeSave", () => { + const packer = () => { + const packed: string[] = []; + return { + packed, + pack: (o: { key: string }) => + Effect.sync(() => { + packed.push(o.key); + }), + }; + }; + + it("a stored key skips the pack and the upload", async () => { + const { packed, pack } = packer(); + await Effect.runPromise(composeSave(() => Effect.succeed(true), pack)(baseOpts)); + expect(packed).toEqual([]); + }); + + it("an absent key packs and uploads", async () => { + const { packed, pack } = packer(); + await Effect.runPromise(composeSave(() => Effect.succeed(false), pack)(baseOpts)); + expect(packed).toEqual(["k"]); + }); + + it("a failed existence check packs as before, and is logged", async () => { + const lines: string[] = []; + const { packed, pack } = packer(); + await Effect.runPromise( + composeSave( + () => Effect.fail(new CacheError({ phase: "save", key: "k", cause: "head unreachable" })), + pack, + )(baseOpts).pipe( + Effect.provide( + Logger.replace( + Logger.defaultLogger, + Logger.make(({ message }) => lines.push(String(message))), + ), + ), + Logger.withMinimumLogLevel(LogLevel.Warning), + ), + ); + expect(packed).toEqual(["k"]); + expect(lines.join("\n")).toContain("head unreachable"); + }); + + it("a pack failure propagates to restoreOr's swallow", async () => { + const exit = await Effect.runPromiseExit( + composeSave( + () => Effect.succeed(false), + () => Effect.fail(new CacheError({ phase: "save", key: "k", cause: "tar" })), + )(baseOpts), + ); + expect(Exit.isFailure(exit)).toBe(true); + }); +}); diff --git a/packages/runtime-cf/src/cache-r2.ts b/packages/runtime-cf/src/cache-r2.ts index 658bf62..e9a3372 100644 --- a/packages/runtime-cf/src/cache-r2.ts +++ b/packages/runtime-cf/src/cache-r2.ts @@ -4,10 +4,23 @@ // entry is a single gzipped tar of the requested paths. // // * restoreOr — `bucket.get` the archive; on a hit, stream it into the -// container and `tar xzf` it in place. On a miss — or any restore error — -// run `onMiss`, then best-effort `save`. -// * save — `tar czf` the paths inside the container, stream the archive out, -// and `bucket.put` it. +// container and `tar xzf` it in place. On a miss — or a restore error — +// run `onMiss`, then best-effort `save`. An archive that exists but does +// not apply is `unapplied`: `onMiss` runs and the save is skipped. +// * save — `bucket.head` the key first; an archive already stored skips the +// pack and the upload. Otherwise `tar czf` the paths inside the container, +// stream the archive out, and `bucket.put` it. +// +// --- Transport --------------------------------------------------------------- +// +// Streaming a body into the container (`writeFile` with a stream) exists only +// on the SDK's `rpc` transport; on `http` / `websocket` it throws. Containers +// run on `rpc` unless a repo pin or the Worker-wide var says otherwise +// (sandbox-transport.ts), so a restore streams by default. A restore on +// another transport reads only the archive's `head` and reports `unapplied`, so +// the run pays for neither the download nor a re-upload of an archive R2 +// already holds. The transport is logged once per Layer instance: once per +// execution, again after a Workflow replay rebuilds the runtime. // // --- Cache key isolation (review fix) --------------------------------------- // @@ -28,19 +41,28 @@ // ============================================================================ // // The container side (`tar`, `readFileStream`, `writeFile`) cannot run in -// `vitest-pool-workers` — Miniflare has no container runtime — so the live -// `restore` / `save` closures are verified by typecheck + `wrangler deploy -// --dry-run`. Their orchestration is covered by cache-r2.test.ts via -// `composeRestoreOr` with fakes. +// `vitest-pool-workers` — Miniflare has no container runtime. The live +// `restore` / `save` closures are covered by cache-r2-live.test.ts against a +// mocked `getSandbox` and a fake bucket; their orchestration by +// cache-r2.test.ts via `composeRestoreOr` / `composeSave` with fakes. // -// Spec: specs/03-dsl.md § cache, specs/02-runs.md § cache-pnpm, plan § PR8. +// Spec: specs/03-dsl.md § cache, specs/02-runs.md § cache-pnpm, plan § PR8, +// specs/adr/0007-cost-efficient-execution.md § Upload each cache key once. import { getSandbox, type Sandbox } from "@cloudflare/sandbox"; import { Effect, Layer } from "effect"; import { Cache, CacheError, type CacheService } from "@fractalboxdev/flare-dispatch-core"; import { readContainerFile } from "./container-file-stream"; import { putStream } from "./r2-put-stream"; -import { composeRestoreOr, type RestoreFn, type SaveFn } from "./cache-restore-or"; +import type { SandboxTransport } from "./sandbox-transport"; +import { + composeRestoreOr, + composeSave, + type ExistsFn, + type RestoreFn, + type RestoreOutcome, + type SaveFn, +} from "./cache-restore-or"; /** Per-repo, content-addressed R2 key for a cache entry. */ const archiveKey = (repo: string, key: string): string => `cache/${repo}/${key}.tar.gz`; @@ -55,13 +77,33 @@ const TARBALL_PATH = "/tmp/fd-cache.tar.gz"; * @param bucket the R2 binding (`env.RUNS_STORAGE`). * @param ns the `RUNS_SANDBOX` namespace for the supplied Sandbox subclass. * @param repo the `owner/name` slug — scopes the archive key per repo. + * @param transport the transport this execution's containers run on, as + * `resolveSandboxTransport` resolved it — only `rpc` can stream + * an archive into a container. */ export const makeCacheR2Live = ( bucket: R2Bucket, ns: DurableObjectNamespace, repo: string, + transport: SandboxTransport, ): Layer.Layer => { - const save: SaveFn = (opts) => + const streams = transport === "rpc"; + let transportLogged = false; + const logTransportOnce = Effect.suspend(() => { + if (transportLogged) return Effect.void; + transportLogged = true; + return Effect.logWarning( + `cache: the ${transport} sandbox transport cannot stream an archive into a container — stored archives report unapplied for this execution`, + ); + }); + + const exists: ExistsFn = (key) => + Effect.tryPromise({ + try: async () => (await bucket.head(archiveKey(repo, key))) !== null, + catch: (cause) => new CacheError({ phase: "save", key, cause }), + }); + + const pack: SaveFn = (opts) => Effect.tryPromise({ try: async () => { const box = getSandbox(ns, opts.container.id); @@ -86,23 +128,51 @@ export const makeCacheR2Live = ( }); const restore: RestoreFn = (opts) => - Effect.tryPromise({ - try: async () => { - const archive = await bucket.get(archiveKey(repo, opts.key)); - if (archive === null) return false; - const box = getSandbox(ns, opts.container.id); - await box.writeFile(TARBALL_PATH, archive.body); - const untar = await box.exec(`tar xzf ${TARBALL_PATH}`, { - cwd: opts.base ?? opts.dir, + Effect.gen(function* () { + const objectKey = archiveKey(repo, opts.key); + if (!streams) { + const stored = yield* Effect.tryPromise({ + try: () => bucket.head(objectKey), + catch: (cause) => new CacheError({ phase: "restore", key: opts.key, cause }), }); - if (untar.exitCode !== 0) { - throw new Error(`tar xzf exited ${untar.exitCode}: ${untar.stderr}`); - } - return true; - }, - catch: (cause) => new CacheError({ phase: "restore", key: opts.key, cause }), + if (stored === null) return "miss" as const; + yield* logTransportOnce; + return "unapplied" as const; + } + const archive = yield* Effect.tryPromise({ + try: () => bucket.get(objectKey), + catch: (cause) => new CacheError({ phase: "restore", key: opts.key, cause }), + }); + if (archive === null) return "miss" as const; + return yield* Effect.tryPromise({ + try: async (): Promise => { + const box = getSandbox(ns, opts.container.id); + try { + await box.writeFile(TARBALL_PATH, archive.body); + } catch (cause) { + // An unread body holds the R2 read open; a locked one rejects the + // cancel, which changes nothing. + await archive.body.cancel().catch(() => undefined); + throw cause; + } + const untar = await box.exec(`tar xzf ${TARBALL_PATH}`, { + cwd: opts.base ?? opts.dir, + }); + if (untar.exitCode !== 0) { + throw new Error(`tar xzf exited ${untar.exitCode}: ${untar.stderr}`); + } + return "hit"; + }, + catch: (cause) => new CacheError({ phase: "restore", key: opts.key, cause }), + }).pipe( + Effect.catchAll((e) => + Effect.as(Effect.logWarning(`${e.message} — stored archive unapplied`), "unapplied" as const), + ), + ); }); + const save = composeSave(exists, pack); + const service: CacheService = { restoreOr: composeRestoreOr(restore, save), save, diff --git a/packages/runtime-cf/src/cache-restore-or.ts b/packages/runtime-cf/src/cache-restore-or.ts index b8f402f..6274b87 100644 --- a/packages/runtime-cf/src/cache-restore-or.ts +++ b/packages/runtime-cf/src/cache-restore-or.ts @@ -3,20 +3,32 @@ // Factored out of cache-r2.ts so it imports NO Cloudflare bindings (no // `@cloudflare/sandbox`) and is therefore unit-testable in plain vitest — // cache-r2.ts wires the live, container-backed `restore` / `save`; this file -// owns only the hit/miss/error sequencing. +// owns only the hit/miss/unapplied/error sequencing and the save's +// upload-once check. // -// Spec: specs/03-dsl.md § cache, specs/pm/plan.md § PR8. +// Spec: specs/03-dsl.md § cache, specs/pm/plan.md § PR8, +// specs/adr/0007-cost-efficient-execution.md § Upload each cache key once. -import { Effect } from "effect"; +import { Effect, Match } from "effect"; import { CacheError, type Container } from "@fractalboxdev/flare-dispatch-core"; -/** Restore a cached archive into a container. `true` = hit, `false` = miss. */ +/** + * What a restore did with a key: + * + * - `hit` — the archive exists and is applied in the container; + * - `miss` — no archive is stored under the key; + * - `unapplied` — the archive exists, applying it failed (or the transport + * cannot stream it into the container). + */ +export type RestoreOutcome = "hit" | "miss" | "unapplied"; + +/** Restore a cached archive into a container. */ export type RestoreFn = (opts: { key: string; container: Container; dir?: string; base?: string; -}) => Effect.Effect; +}) => Effect.Effect; /** Pack container paths into an archive and persist it. */ export type SaveFn = (opts: { @@ -27,17 +39,41 @@ export type SaveFn = (opts: { base?: string; }) => Effect.Effect; +/** Whether an archive is already stored under a cache key. */ +export type ExistsFn = (key: string) => Effect.Effect; + +/** + * A save that packs and uploads only when no archive is stored under its key. + * Keys are content-addressed, so a stored archive already holds this content: + * parallel misses on one key upload once instead of once per part. A failed + * existence check packs as before — the check saves work, it guards nothing. + */ +export const composeSave = + (exists: ExistsFn, pack: SaveFn): SaveFn => + (opts) => + exists(opts.key).pipe( + Effect.catchAll((e) => Effect.as(Effect.logWarning(e.message), false)), + Effect.flatMap((stored) => + stored + ? Effect.logInfo(`cache: ${opts.key} is already stored — skipping pack and upload`) + : pack(opts), + ), + ); + /** * The pure `restoreOr` orchestration, factored out of any binding so it is * unit-testable with fake `restore` / `save`. Best-effort by construction: * - * - a restore error is degraded to a miss; - * - on a hit, `onMiss` is skipped and the Effect completes with `void`; - * - on a miss, `onMiss` runs, then `save` runs and a save failure does not - * fail the Effect — the `onMiss` work already succeeded. + * - on a `hit`, `onMiss` is skipped and the Effect completes with `void`; + * - on a `miss`, `onMiss` runs, then `save` runs and a save failure does not + * fail the Effect — the `onMiss` work already succeeded; + * - on `unapplied`, `onMiss` runs and the save is skipped: the stored + * archive already holds this key's content; + * - a restore error is degraded to a miss — whether an archive is stored is + * unknown, and the save's existence check covers the case where it is. * - * Never silent, though: both failures are logged. A discarded save is - * otherwise indistinguishable from a working cache. + * Never silent, though: restore and save failures and a skipped save are + * logged. A discarded save is otherwise indistinguishable from a working cache. * * Returns `void`: on a cache hit there is no `onMiss` value, so `restoreOr` * cannot honestly yield an `A`. Consumers (`installCached`) discard it anyway. @@ -53,15 +89,27 @@ export const composeRestoreOr = onMiss: () => Effect.Effect; }): Effect.Effect => restore(opts).pipe( - Effect.catchAll((e) => Effect.as(Effect.logWarning(e.message), false)), - Effect.flatMap((hit) => - hit - ? Effect.void - : opts.onMiss().pipe( + Effect.catchAll((e) => Effect.as(Effect.logWarning(e.message), "miss" as const)), + Effect.flatMap((outcome) => + Match.value(outcome).pipe( + Match.when("hit", () => Effect.void), + Match.when("miss", () => + opts.onMiss().pipe( Effect.tap(() => save(opts).pipe(Effect.catchAll((e) => Effect.logWarning(e.message))), ), Effect.asVoid, ), + ), + Match.when("unapplied", () => + opts.onMiss().pipe( + Effect.tap(() => + Effect.logInfo(`cache: ${opts.key} is stored but unapplied — skipping save`), + ), + Effect.asVoid, + ), + ), + Match.exhaustive, + ), ), ); diff --git a/packages/runtime-cf/src/runtime.ts b/packages/runtime-cf/src/runtime.ts index 766711d..4a8d5f5 100644 --- a/packages/runtime-cf/src/runtime.ts +++ b/packages/runtime-cf/src/runtime.ts @@ -43,6 +43,7 @@ import { makeOidcLive, type OidcLiveConfig } from "./oidc-live"; import { type ExecutionContext, makeD1ExecutionsLive } from "./executions-d1"; import { makeIOLive } from "./io-live"; import { makeSandboxCloudflareLive } from "./sandbox-cf"; +import { resolveSandboxTransport } from "./sandbox-transport"; import type { CheckCommandOwner, Container } from "@fractalboxdev/flare-dispatch-core"; import { resolveAppCredentials } from "./sandbox-clone-auth"; import { CacheOnFacade, makeSandboxFacadeLive, type SandboxFacadeOptions } from "./sandbox-facade"; @@ -181,12 +182,20 @@ export type CFRuntimeLiveOptions = { */ readonly sandboxPreviewHostname?: string; /** - * Pin this execution's containers to a transport instead of the Worker-wide - * `SANDBOX_TRANSPORT` default — see `makeSandboxCloudflareLive`'s `transport` - * option for why the choice is per-execution rather than per-Worker. - * Container path only; the substrate facade owns its own transport. + * The `CONFIG_KV` `sandbox.transport:` pin, as read. It outranks + * `sandboxTransportDefault` in either direction, so an operator moves one + * repo onto or off `rpc` without a deploy. A value that is not a transport + * name is skipped. Container path only; the substrate facade owns its own + * transport. See `resolveSandboxTransport`. */ - readonly sandboxTransport?: "http" | "websocket" | "rpc"; + readonly sandboxTransport?: string; + /** + * The Worker-wide `SANDBOX_TRANSPORT` var, as read. Applies when no repo pin + * names a transport; when it names none either, containers run on `rpc`. + * The resolved transport is pinned onto every container the execution + * acquires and tells the `cache` Layer whether an archive can stream in. + */ + readonly sandboxTransportDefault?: string; /** * The dispatcher's own public origin (e.g. * `https://..workers.dev`) — prefixed onto the @@ -315,6 +324,13 @@ export const makeCFRuntimeLive = ( ...(opts.checks !== undefined ? { checks: opts.checks } : {}), payloadRepo: opts.execution.repo, }); + // The repo pin, else the Worker-wide var, else `rpc` — pinned onto every + // container this execution acquires, so the cache restores only where the + // container streams. Pure and unit-tested in `sandbox-transport.ts`. + const sandboxTransport = resolveSandboxTransport({ + repoPin: opts.sandboxTransport, + workerDefault: opts.sandboxTransportDefault, + }); // The substrate path and the container path are mutually exclusive by // construction: one execution runs entirely on one of them, so there is never // a call that reaches a container the rest of the runtime does not know about. @@ -335,7 +351,7 @@ export const makeCFRuntimeLive = ( cloneAuth, opts.sandboxPreviewHostname, opts.logsViewerBase, - opts.sandboxTransport, + sandboxTransport, opts.checkCommandOwner, ); const stepRunner = makeStepRunnerCloudflare(opts.executionId); @@ -346,7 +362,12 @@ export const makeCFRuntimeLive = ( const cache = opts.substrate !== undefined ? CacheOnFacade - : makeCacheR2Live(opts.bucket, opts.sandboxNs!, opts.execution.repo); + : makeCacheR2Live( + opts.bucket, + opts.sandboxNs!, + opts.execution.repo, + sandboxTransport, + ); // `Config` is live when the `CONFIG_KV` binding is present; absent, the // dying stub keeps a config-reading run from silently mis-behaving. const config = diff --git a/packages/runtime-cf/src/sandbox-cf.test.ts b/packages/runtime-cf/src/sandbox-cf.test.ts index 45609e2..71d9d9b 100644 --- a/packages/runtime-cf/src/sandbox-cf.test.ts +++ b/packages/runtime-cf/src/sandbox-cf.test.ts @@ -323,15 +323,16 @@ describe("makeSandboxCloudflareLive — container routing", () => { }), ); - it.effect("no transport option leaves the container on the Worker-wide default", () => + it.effect("no transport option leaves the container on the SDK default", () => Effect.gen(function* () { currentBox = makeFakeBox({ proc: null }); transportCalls.length = 0; yield* Effect.flatMap(SandboxTag, (s) => s.acquire({})).pipe( Effect.provide(makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-2")), ); - // Untouched: the default is whatever `SANDBOX_TRANSPORT` says, and a - // Layer that pinned it unasked would silently change every consumer. + // Untouched: the Layer pins only what its caller resolved. The runtime + // always passes one (`resolveSandboxTransport`), so this is the SDK + // default for a caller that opts out. expect(transportCalls).toEqual([]); }), ); diff --git a/packages/runtime-cf/src/sandbox-cf.ts b/packages/runtime-cf/src/sandbox-cf.ts index e52736f..3e6c885 100644 --- a/packages/runtime-cf/src/sandbox-cf.ts +++ b/packages/runtime-cf/src/sandbox-cf.ts @@ -305,28 +305,24 @@ export const makeSandboxCloudflareLive = ( */ logsViewerBase?: string, /** - * Pin every container this execution acquires to a transport, rather than - * taking the Worker-wide `SANDBOX_TRANSPORT` default. + * Pin every container this execution acquires to a transport. The runtime + * always passes one — `resolveSandboxTransport`'s repo pin, else the + * Worker-wide `SANDBOX_TRANSPORT`, else `rpc` — because the SDK leaves an + * unpinned container on `http`. `undefined` leaves the SDK default. * - * WHY THIS EXISTS, and why it is per-execution. The SDK's streaming file APIs - * live only on the `rpc` client — its own comment calls `rpc` the "primary - * container-control client" and `http`/`websocket` the "route-based - * compatibility client". `writeFileStream` is a bare `throw` off `rpc`, which - * is why the R2 dependency cache has missed on every run since it was added: - * the restore hands `writeFile` a `ReadableStream`, the SDK routes any stream - * to `writeFileStream`, and it raises. + * WHY `rpc`. The SDK's streaming file APIs live only on the `rpc` client — + * its own comment calls `rpc` the "primary container-control client" and + * `http`/`websocket` the "route-based compatibility client". `writeFileStream` + * is a bare `throw` off `rpc`, and the R2 cache's restore hands `writeFile` a + * `ReadableStream`, which the SDK routes to `writeFileStream` — so off `rpc` a + * stored archive cannot be applied, and the cache reports it `unapplied` + * (cache-r2.ts § Transport). * - * `SANDBOX_TRANSPORT=rpc` as a Worker var would fix that in one line and - * change the control path for EVERY repo this dispatcher serves at once. - * This threads the same choice per execution instead, so one consumer can - * prove it before the rest follow. That is the whole difference between a - * rollout and a flip. - * - * Note the precedence the SDK applies on cold start: a transport written to - * the DO's storage WINS over the env-derived default. `setTransport` persists, - * so a container pinned here stays pinned for its lifetime regardless of what - * the var says — which is what makes this safe to scope, and what would make - * it sticky if it were ever pointed at the wrong value. + * A container keeps the transport it started with for its whole life. The + * SDK prefers a transport written to the DO's storage over the env-derived + * default on cold start, and `setTransport` persists, so a container pinned + * here stays pinned whatever the var later says. Ids are per execution, so a + * changed pin or var takes effect on the next execution. */ transport?: "http" | "websocket" | "rpc", checkOwner?: (container: Container) => CheckCommandOwner, @@ -527,8 +523,8 @@ export const makeSandboxCloudflareLive = ( const id = opts.key === undefined ? sandboxId : previewSafeSandboxId(`${executionId}:${opts.key}`); // Pin the container's transport before anything runs in it, when the - // caller asked for one. See `transport` in this Layer's options for why - // this is per-container rather than a Worker-wide env var. + // caller passed one. See `transport` in this Layer's options for why + // the pin is per container rather than left to the SDK default. // // ONCE PER CONTAINER, not once per acquire. `acquire` is also how a // caller derives an id without provisioning anything — `ensureWorkspace` @@ -537,8 +533,8 @@ export const makeSandboxCloudflareLive = ( // each of those would wake a Durable Object to re-assert a setting it // already has. // - // Best-effort: a container that stays on the default transport works, - // it just cannot stream a file. Failing acquisition over a transport + // Best-effort: a container that stays on the SDK default transport + // works, it just cannot stream a file. Failing acquisition over a transport // preference would trade a slow cache for a dead run. if (transport !== undefined && !pinned.has(id)) { pinned.add(id); diff --git a/packages/runtime-cf/src/sandbox-transport.test.ts b/packages/runtime-cf/src/sandbox-transport.test.ts new file mode 100644 index 0000000..c9b4eb0 --- /dev/null +++ b/packages/runtime-cf/src/sandbox-transport.test.ts @@ -0,0 +1,49 @@ +// Unit coverage for `resolveSandboxTransport` — the per-repo pin, else the +// Worker-wide var, else `rpc` (see sandbox-transport.ts header). + +import { describe, expect, it } from "vitest"; +import { parseSandboxTransport, resolveSandboxTransport } from "./sandbox-transport"; + +describe("resolveSandboxTransport", () => { + it("defaults to rpc when neither the pin nor the var is set", () => { + expect(resolveSandboxTransport({})).toBe("rpc"); + expect(resolveSandboxTransport({ repoPin: null, workerDefault: undefined })).toBe("rpc"); + }); + + it("takes the Worker-wide var when no repo pin is set", () => { + expect(resolveSandboxTransport({ workerDefault: "http" })).toBe("http"); + expect(resolveSandboxTransport({ repoPin: null, workerDefault: "websocket" })).toBe( + "websocket", + ); + }); + + it("lets the repo pin move a repo back to http over an rpc default", () => { + expect(resolveSandboxTransport({ repoPin: "http" })).toBe("http"); + expect(resolveSandboxTransport({ repoPin: "http", workerDefault: "rpc" })).toBe("http"); + }); + + it("lets the repo pin select rpc over an http var", () => { + expect(resolveSandboxTransport({ repoPin: "rpc", workerDefault: "http" })).toBe("rpc"); + }); + + it("skips an unrecognised pin and falls through to the var", () => { + expect(resolveSandboxTransport({ repoPin: "RPC", workerDefault: "http" })).toBe("http"); + expect(resolveSandboxTransport({ repoPin: "", workerDefault: "websocket" })).toBe("websocket"); + }); + + it("skips an unrecognised var and falls through to rpc", () => { + expect(resolveSandboxTransport({ workerDefault: "grpc" })).toBe("rpc"); + expect(resolveSandboxTransport({ repoPin: "typo", workerDefault: "typo" })).toBe("rpc"); + }); +}); + +describe("parseSandboxTransport", () => { + it("accepts exactly the three transport names", () => { + expect(parseSandboxTransport("rpc")).toBe("rpc"); + expect(parseSandboxTransport("websocket")).toBe("websocket"); + expect(parseSandboxTransport("http")).toBe("http"); + expect(parseSandboxTransport(" rpc")).toBeUndefined(); + expect(parseSandboxTransport(null)).toBeUndefined(); + expect(parseSandboxTransport(undefined)).toBeUndefined(); + }); +}); diff --git a/packages/runtime-cf/src/sandbox-transport.ts b/packages/runtime-cf/src/sandbox-transport.ts new file mode 100644 index 0000000..2fc47be --- /dev/null +++ b/packages/runtime-cf/src/sandbox-transport.ts @@ -0,0 +1,39 @@ +// The sandbox transport an execution's containers run on. +// +// The `@cloudflare/sandbox` SDK speaks to a container over `http` (its own +// default), `websocket` or `rpc`. Only `rpc` streams a file into a container, +// so only `rpc` lets the R2 cache apply a stored archive (cache-r2.ts +// § Transport). The dispatcher therefore defaults to `rpc` and pins the +// effective transport onto every container it acquires (sandbox-cf.ts), rather +// than leaving an unpinned container on the SDK's `http`. +// +// Precedence, highest first: +// +// 1. the per-repo `CONFIG_KV` pin `sandbox.transport:` — in either +// direction, so an operator moves one repo back to `http` without a deploy; +// 2. the Worker-wide `SANDBOX_TRANSPORT` var; +// 3. `rpc`. +// +// A value that is not a transport name is ignored at its level, so a typo in +// the pin falls through to the var, and one in the var falls through to `rpc`. + +export type SandboxTransport = "http" | "websocket" | "rpc"; + +/** The transport a container runs on when neither the pin nor the var names one. */ +export const SANDBOX_TRANSPORT_DEFAULT: SandboxTransport = "rpc"; + +/** `raw` as a transport name, or `undefined` when it is not one. */ +export const parseSandboxTransport = (raw: string | null | undefined): SandboxTransport | undefined => + raw === "rpc" || raw === "websocket" || raw === "http" ? raw : undefined; + +/** + * The effective transport: the per-repo pin, else the Worker-wide var, else + * `rpc`. Unrecognised values at either level are skipped. + */ +export const resolveSandboxTransport = (input: { + readonly repoPin?: string | null | undefined; + readonly workerDefault?: string | null | undefined; +}): SandboxTransport => + parseSandboxTransport(input.repoPin) ?? + parseSandboxTransport(input.workerDefault) ?? + SANDBOX_TRANSPORT_DEFAULT; diff --git a/runs/README.md b/runs/README.md index 7f03895..17b9c69 100644 --- a/runs/README.md +++ b/runs/README.md @@ -503,34 +503,42 @@ flowchart TB ### Container transport (`sandbox.transport:`) +Containers run on the SDK's `rpc` transport. The dispatcher resolves one +transport per execution and pins it onto every container the execution +acquires, highest first: + +1. the `CONFIG_KV` key `sandbox.transport:`; +2. the Worker-wide `SANDBOX_TRANSPORT` var; +3. `rpc`. + +Each level takes `http`, `websocket` or `rpc`. The pin overrides the var in +either direction, so one repo moves back to `http` without a deploy: + ```bash -wrangler kv key put --binding=CONFIG_KV "sandbox.transport:owner/repo" "rpc" +wrangler kv key put --binding=CONFIG_KV "sandbox.transport:owner/repo" "http" ``` -Pins every container that repo's executions acquire to one of `http` (the -default), `websocket`, or `rpc`. **Any other value is dropped by the dispatcher -before it reaches the SDK** — silently, and the run proceeds on the default. A -typo therefore degrades rather than breaks, but it also says nothing: check the -key back if a transport change appears to have had no effect. +Deleting the key returns the repo to the var, or to `rpc`. **A value that is not +a transport name is skipped at its level**, silently: a typo in the pin falls +through to the var, and one in the var falls through to `rpc`. Check the key +back if a transport change appears to have had no effect. The precedence lives +in `packages/runtime-cf/src/sandbox-transport.ts`. -**Why it exists.** The SDK's streaming file APIs live only on its `rpc` client — +**Why `rpc`.** The SDK's streaming file APIs live only on its `rpc` client — its own comment calls `rpc` the "primary container-control client" and `http`/`websocket` the "route-based compatibility client", and `writeFileStream` -is a bare `throw` off `rpc`. That is why the R2 dependency cache misses on every -run: `installCached`'s restore hands `writeFile` a `ReadableStream`, the SDK -routes any stream to `writeFileStream`, it raises, and `composeRestoreOr` -records a miss. - -`SANDBOX_TRANSPORT=rpc` as a Worker var fixes that in one line — and changes the -control path for every repo this dispatcher serves, at once. This key is the -same choice scoped to one consumer, so a change with that blast radius can be -proved before it is taken. +is a bare `throw` off `rpc`. `installCached`'s restore hands `writeFile` a +`ReadableStream`, and the SDK routes any stream to `writeFileStream`, so off +`rpc` a stored archive cannot be applied. The R2 cache receives the resolved +transport: on `rpc` it streams the archive in; on `http` or `websocket` it reads +only the archive's head, reports `unapplied`, runs the install and skips the +save, so the run pays for neither the download nor a second upload. **It is sticky.** `setTransport` persists to the container's Durable Object storage, and the SDK prefers a stored transport over the env-derived default on -cold start. A container pinned here keeps that transport for its lifetime; -since ids are per execution, the pin is re-applied per run and costs one DO call -at acquire. +cold start. A container keeps the transport it started with for its lifetime; +since ids are per execution, a changed pin or var takes effect on the next run, +and the pin costs one DO call per container at acquire. ### A stage does not assume its checkout diff --git a/runs/contextful-gate.ts b/runs/contextful-gate.ts index 4400409..3b62e22 100644 --- a/runs/contextful-gate.ts +++ b/runs/contextful-gate.ts @@ -628,6 +628,10 @@ export const contextfulGate = defineRun({ failOnNonZeroExit: true, timeoutSec: STAGE_TIMEOUT_SEC, admissionMaxQueueAgeSec: STAGE_QUEUE_SEC, + // Stays on: what the stage runs is the consumer's `contextful-ci`, + // and its `.cargo/config.toml` may set `net.offline`, so whether + // the stage fetches its own crates is not decidable here + // (specs/adr/0007-cost-efficient-execution.md § Fan-out parts). install: true, ...(prebuilt === null ? {} diff --git a/wrangler.jsonc b/wrangler.jsonc index 59fe104..c219183 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -189,7 +189,13 @@ // class) bakes the native toolchain + Rust: webhook-mode runs have no // setup step to install one, and consumer Rust workspaces with `*-sys` // crates both need `cc` to link and more memory than standard-2 gives a - // cold `cargo test` of a large dep graph. + // cold `cargo test` of a large dep graph. `WITH_RUST` also bakes a + // sha256-pinned prebuilt DuckDB shared library under /opt/duckdb and + // exports `DUCKDB_LIB_DIR` / `DUCKDB_INCLUDE_DIR` (ADR-0007, "Prebuilt + // heavy native dependencies"). A consumer opts in by building `duckdb` + // without its `bundled` feature in CI, asserts the loaded library's + // version at startup, and moves its crate pin with the image's + // `DUCKDB_VERSION`; a bundled build ignores the variables. "containers": [ { "class_name": "RunSandbox",