diff --git a/apps/dispatcher/src/workflow.ts b/apps/dispatcher/src/workflow.ts index 474a0af..66585e1 100644 --- a/apps/dispatcher/src/workflow.ts +++ b/apps/dispatcher/src/workflow.ts @@ -444,6 +444,24 @@ 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. + // 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 = + substrateFacade === undefined + ? await this.env.CONFIG_KV?.get(`sandbox.transport:${payload.github.repo}`) + : undefined; + const runtime = makeCFRuntimeLive({ db, bucket: this.env.RUNS_STORAGE, @@ -542,6 +560,11 @@ 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 } + : {}), ...(publicOrigin !== undefined ? { publicOrigin } : {}), ...(logsBaseUrl !== undefined ? { logsViewerBase: logsBaseUrl } : {}), // Wire the live OIDC signing Layer when both the JWK + issuer URL are diff --git a/packages/runtime-cf/src/runtime.ts b/packages/runtime-cf/src/runtime.ts index eea9c3e..500dda2 100644 --- a/packages/runtime-cf/src/runtime.ts +++ b/packages/runtime-cf/src/runtime.ts @@ -172,6 +172,13 @@ export type CFRuntimeLiveOptions = { * Non-browser runs never touch the surface. */ 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. + */ + readonly sandboxTransport?: "http" | "websocket" | "rpc"; /** * The dispatcher's own public origin (e.g. * `https://..workers.dev`) — prefixed onto the @@ -315,6 +322,7 @@ export const makeCFRuntimeLive = (opts: CFRuntimeLiveOptions): Layer.Layer { + transportCalls.push(t); + }), _getLogs: getLogs, _waitForPort: waitForPort, }; }; +/** Every `setTransport` the Layer issued, in order. */ +const transportCalls: string[] = []; + // A stand-in for the SDK's `SessionTerminatedError` (thrown when a command's // shell exits). `vi.hoisted` so it exists before the hoisted `vi.mock` factory // runs and so the tests can construct instances the Layer will `instanceof`. @@ -314,6 +323,63 @@ describe("makeSandboxCloudflareLive — container routing", () => { }), ); + it.effect("no transport option leaves the container on the Worker-wide 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. + expect(transportCalls).toEqual([]); + }), + ); + + it.effect("a transport option pins EVERY container the execution acquires", () => + Effect.gen(function* () { + currentBox = makeFakeBox({ proc: null }); + transportCalls.length = 0; + yield* Effect.flatMap(SandboxTag, (s) => + Effect.all([s.acquire({}), s.acquire({ key: "features" })]), + ).pipe( + Effect.provide( + makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-3", undefined, undefined, undefined, "rpc"), + ), + ); + // Per CONTAINER, not per Layer: a keyed container is a different DO and + // carries its own transport, so pinning the first would leave the stage + // containers on the compatibility client — which is where the streaming + // file APIs do not exist. + expect(transportCalls).toEqual(["rpc", "rpc"]); + }), + ); + + it.effect("re-acquiring a container does not re-pin it", () => + Effect.gen(function* () { + currentBox = makeFakeBox({ proc: null }); + transportCalls.length = 0; + yield* Effect.flatMap(SandboxTag, (s) => + // Three acquires, two containers. `acquire` is also how a caller derives + // an id without provisioning — `ensureWorkspace` on a rebuild, and the + // stage reaper naming the container it is about to destroy. Each of + // those would otherwise wake a Durable Object to re-assert a setting it + // already has. + Effect.all([ + s.acquire({}), + s.acquire({}), + s.acquire({ key: "features" }), + s.acquire({ key: "features" }), + ]), + ).pipe( + Effect.provide( + makeSandboxCloudflareLive(ns, makeBucket().bucket, "route-4", undefined, undefined, undefined, "rpc"), + ), + ); + expect(transportCalls).toEqual(["rpc", "rpc"]); + }), + ); + it.effect("exec routes by the handle it is given, not by the execution", () => Effect.gen(function* () { currentBox = makeFakeBox({ proc: null }); diff --git a/packages/runtime-cf/src/sandbox-cf.ts b/packages/runtime-cf/src/sandbox-cf.ts index ae6e378..a255608 100644 --- a/packages/runtime-cf/src/sandbox-cf.ts +++ b/packages/runtime-cf/src/sandbox-cf.ts @@ -308,6 +308,31 @@ export const makeSandboxCloudflareLive = ( * historical message. Built by the dispatcher (it owns the token secret). */ logsViewerBase?: string, + /** + * Pin every container this execution acquires to a transport, rather than + * taking the Worker-wide `SANDBOX_TRANSPORT` 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. + * + * `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. + */ + transport?: "http" | "websocket" | "rpc", ): Layer.Layer => { // The Durable Object / sandbox id. `getSandbox` routes the DO by this id AND // the SDK embeds it in the `exposePort` preview URL's DNS label, which must @@ -337,6 +362,9 @@ export const makeSandboxCloudflareLive = ( // in line with them. const boxFor = (container?: Container): Sandbox => getSandbox(ns, container?.id ?? sandboxId); + /** Container ids whose transport this Layer has already pinned — see `acquire`. */ + const pinned = new Set(); + // `exec` log keys are unique within a run: the first exec is `exec.ndjson` // (the name the plan's acceptance pins), subsequent execs `exec-2.ndjson`, … @@ -501,12 +529,37 @@ export const makeSandboxCloudflareLive = ( // dispatcher's end-of-run teardown knows the execution's own id and cannot // know what a run named. acquire: (opts) => - Effect.succeed({ - id: + Effect.gen(function* () { + const id = opts.key === undefined ? sandboxId - : previewSafeSandboxId(`${executionId}:${opts.key}`), - } satisfies Container), + : 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. + // + // ONCE PER CONTAINER, not once per acquire. `acquire` is also how a + // caller derives an id without provisioning anything — `ensureWorkspace` + // re-acquires on a rebuild, and `offload-test`'s stage reaper acquires + // purely to name the container it is about to destroy. Without this memo + // 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 + // preference would trade a slow cache for a dead run. + if (transport !== undefined && !pinned.has(id)) { + pinned.add(id); + yield* Effect.promise(async () => { + try { + await getSandbox(ns, id).setTransport(transport); + } catch (cause) { + console.warn(`setTransport(${transport}) failed for ${id} — staying on the default: ${String(cause)}`); + } + }); + } + return { id } satisfies Container; + }), // Best-effort and idempotent — destroying a container that was never // provisioned, or is already gone, is a success. A teardown failure must diff --git a/runs/README.md b/runs/README.md index 07b935e..f94ae5a 100644 --- a/runs/README.md +++ b/runs/README.md @@ -263,6 +263,37 @@ Staged mode is webhook-only: a dispatch that passes `command` skips the config read and stays single-exec. Stages run inside one workflow instance posting one check-run — sequentially by default, concurrently when the next rung says so. +### Container transport (`sandbox.transport:`) + +```bash +wrangler kv key put --binding=CONFIG_KV "sandbox.transport:owner/repo" "rpc" +``` + +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. + +**Why it exists.** 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. + +**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. + ### A stage does not assume its checkout Container disk is ephemeral, and a staged run spanning forty minutes of durable