From 1237e573e1a30b7e7174e6a4c0e90248fb317e72 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 23:59:53 -0700 Subject: [PATCH 1/3] Load QuickJS lazily on first artifact render --- apps/cloud/src/quickjs.ts | 39 +++++++++++++++++++++++++++++++++++---- 1 file changed, 35 insertions(+), 4 deletions(-) diff --git a/apps/cloud/src/quickjs.ts b/apps/cloud/src/quickjs.ts index 9d84484ed..24ddee8b8 100644 --- a/apps/cloud/src/quickjs.ts +++ b/apps/cloud/src/quickjs.ts @@ -10,6 +10,7 @@ import baseVariant from "@jitl/quickjs-wasmfile-release-sync"; import wasmModule from "./quickjs-engine.wasm"; import { setQuickJSModule } from "@executor-js/runtime-quickjs"; +import { SpanStatusCode, trace } from "@opentelemetry/api"; // --------------------------------------------------------------------------- // QuickJS-on-Workers WASM loading. @@ -20,16 +21,46 @@ import { setQuickJSModule } from "@executor-js/runtime-quickjs"; // fetched/compiled at runtime. `newVariant(base, { wasmModule })` hands it the // statically-imported, pre-compiled module, and `setQuickJSModule` makes every // `makeQuickJsExecutor()` reuse it. Preloaded once per isolate. +// +// Callers trigger this lazily, from the artifact smoke-render path only — see +// `session-durable-object.ts`. Instantiation is ~1.4s of CPU, and most +// sessions never render an artifact, so it must not sit on every session +// init. The client span records exactly when/where that cost lands. // --------------------------------------------------------------------------- let preloaded: Promise | null = null; +const tracer = trace.getTracer("executor-cloud-quickjs"); + export const preloadQuickJs = (): Promise => { if (!preloaded) { - const variant = newVariant(baseVariant, { wasmModule }); - preloaded = newQuickJSWASMModuleFromVariant(variant).then((mod) => { - setQuickJSModule(mod); - }); + // oxlint-disable-next-line executor/no-promise-catch -- boundary: this module has no Effect context; reset the memoized promise on failure so a retried artifact render can trigger a fresh instantiation instead of caching a permanent rejection + preloaded = tracer + .startActiveSpan("quickjs.preload", async (span) => { + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: promise-native WASM instantiation; recording exception detail on the span before rethrowing + try { + const variant = newVariant(baseVariant, { wasmModule }); + const mod = await newQuickJSWASMModuleFromVariant(variant); + setQuickJSModule(mod); + } catch (err) { + // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: normalizing an untyped instantiation failure only for the OTel span record; the original error is rethrown below unchanged + const cause = err instanceof Error ? err : String(err); + span.recordException(cause); + // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: same normalization as the recordException line above + const message = typeof cause === "string" ? cause : cause.message; + span.setStatus({ code: SpanStatusCode.ERROR, message }); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: rethrow after recording so the caller (and the memoization reset below) still observes the failure + throw err; + } finally { + span.end(); + } + }) + // oxlint-disable-next-line executor/no-promise-catch -- boundary: this module has no Effect context; reset the memoized promise on failure so a retried artifact render can trigger a fresh instantiation instead of caching a permanent rejection + .catch((err: unknown) => { + preloaded = null; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: rethrow so the caller's awaited promise still rejects with the original error + throw err; + }); } return preloaded; }; From a024e1ada1062c1557309c838a1ac888faeb9a62 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 23:59:58 -0700 Subject: [PATCH 2/3] Cap concurrent session builds per isolate --- .../src/mcp/session-build-semaphore.test.ts | 120 ++++++++++++++++++ apps/cloud/src/mcp/session-build-semaphore.ts | 66 ++++++++++ apps/cloud/src/mcp/session-durable-object.ts | 87 ++++++++----- 3 files changed, 244 insertions(+), 29 deletions(-) create mode 100644 apps/cloud/src/mcp/session-build-semaphore.test.ts create mode 100644 apps/cloud/src/mcp/session-build-semaphore.ts diff --git a/apps/cloud/src/mcp/session-build-semaphore.test.ts b/apps/cloud/src/mcp/session-build-semaphore.test.ts new file mode 100644 index 000000000..ac34f3db7 --- /dev/null +++ b/apps/cloud/src/mcp/session-build-semaphore.test.ts @@ -0,0 +1,120 @@ +import { describe, expect, it, beforeEach } from "@effect/vitest"; + +import { + acquireBuildSlot, + releaseBuildSlot, + resetBuildSlotsForTest, + currentActiveBuildsForTest, + currentQueueLengthForTest, +} from "./session-build-semaphore"; + +describe("session-build-semaphore", () => { + beforeEach(() => { + resetBuildSlotsForTest(); + }); + + it("grants up to the cap immediately, with no wait", async () => { + const waits = await Promise.all([ + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + ]); + expect(waits).toEqual([0, 0, 0, 0]); + expect(currentActiveBuildsForTest()).toBe(4); + expect(currentQueueLengthForTest()).toBe(0); + }); + + it("queues a build past the cap until a slot is released", async () => { + await Promise.all([ + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + ]); + expect(currentActiveBuildsForTest()).toBe(4); + + let fifthResolved = false; + const fifth = acquireBuildSlot().then((waitedMs) => { + fifthResolved = true; + return waitedMs; + }); + + expect(currentQueueLengthForTest()).toBe(1); + // Nothing releases the pending fifth build without an explicit release — + // it must not resolve on its own. + await Promise.resolve(); + await Promise.resolve(); + expect(fifthResolved).toBe(false); + + releaseBuildSlot(); + const waitedMs = await fifth; + expect(fifthResolved).toBe(true); + expect(waitedMs).toBeGreaterThanOrEqual(0); + // The freed slot went straight to the waiter — total active stays at cap. + expect(currentActiveBuildsForTest()).toBe(4); + expect(currentQueueLengthForTest()).toBe(0); + }); + + it("releases queued waiters in FIFO order", async () => { + await Promise.all([ + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + ]); + + const order: number[] = []; + const second = acquireBuildSlot().then(() => order.push(2)); + const third = acquireBuildSlot().then(() => order.push(3)); + const fourth = acquireBuildSlot().then(() => order.push(4)); + expect(currentQueueLengthForTest()).toBe(3); + + releaseBuildSlot(); + await second; + releaseBuildSlot(); + await third; + releaseBuildSlot(); + await fourth; + + expect(order).toEqual([2, 3, 4]); + }); + + it("never deadlocks: releasing a slot always makes forward progress for the next waiter", async () => { + await Promise.all([ + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + acquireBuildSlot(), + ]); + + // 6 more builds arrive while all 4 slots are held — all 6 queue. + const queued = Array.from({ length: 6 }, () => acquireBuildSlot()); + expect(currentQueueLengthForTest()).toBe(6); + + // The 4 in-flight builds finish one at a time; each release must hand its + // slot straight to the next queued waiter rather than sitting idle. + for (let i = 0; i < 4; i++) releaseBuildSlot(); + await Promise.all(queued.slice(0, 4)); + expect(currentActiveBuildsForTest()).toBe(4); + expect(currentQueueLengthForTest()).toBe(2); + + // Those 4 finish too, freeing the last 2 queued waiters. + for (let i = 0; i < 4; i++) releaseBuildSlot(); + await Promise.all(queued.slice(4)); + expect(currentActiveBuildsForTest()).toBe(2); + expect(currentQueueLengthForTest()).toBe(0); + + // And the last 2 finish, draining the semaphore completely. + releaseBuildSlot(); + releaseBuildSlot(); + expect(currentActiveBuildsForTest()).toBe(0); + expect(currentQueueLengthForTest()).toBe(0); + }); + + it("does not go negative when released more times than acquired", () => { + releaseBuildSlot(); + releaseBuildSlot(); + expect(currentActiveBuildsForTest()).toBe(0); + }); +}); diff --git a/apps/cloud/src/mcp/session-build-semaphore.ts b/apps/cloud/src/mcp/session-build-semaphore.ts new file mode 100644 index 000000000..c04e4ea9d --- /dev/null +++ b/apps/cloud/src/mcp/session-build-semaphore.ts @@ -0,0 +1,66 @@ +/** + * Module-scope (== per-isolate, same reasoning as + * `packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts`) semaphore + * bounding concurrent COLD `buildMcpServer` builds. + * + * A burst of new sessions can land on one isolate at once. Each cold build + * runs execution-stack setup and MCP server construction, which is real CPU + * — concentrating several of those builds at the same instant is what drags + * the isolate's request p95 during a wave. Capping how many builds run at + * once smooths that burst into a short FIFO queue instead of letting every + * arrival pay full concurrent cost. + * + * Deliberately dependency-free: a tiny promise-chain queue, not a library. + */ + +const MAX_CONCURRENT_BUILDS = 4; + +let activeBuilds = 0; +const waitQueue: Array<() => void> = []; + +/** + * Reserve a build slot, queueing FIFO when the cap is already held. Resolves + * with the number of milliseconds spent waiting for a slot — 0 when one was + * immediately free, which is the common case outside a burst. + * + * Always resolves, never rejects: there is nothing to fail here, only to + * wait for. + */ +export const acquireBuildSlot = (): Promise => { + const requestedAt = Date.now(); + if (activeBuilds < MAX_CONCURRENT_BUILDS) { + activeBuilds += 1; + return Promise.resolve(0); + } + return new Promise((resolve) => { + waitQueue.push(() => { + activeBuilds += 1; + resolve(Date.now() - requestedAt); + }); + }); +}; + +/** + * Release a build slot. Must be called exactly once per slot a caller + * actually acquired (i.e. `acquireBuildSlot` resolved) — callers release from + * a `finally`/`ensuring` so a build that throws still frees its slot and + * never deadlocks the queue behind it. + * + * Wakes the next FIFO waiter, if any, handing it the freed slot directly + * rather than making it race a fresh `acquireBuildSlot` caller. + */ +export const releaseBuildSlot = (): void => { + activeBuilds = Math.max(0, activeBuilds - 1); + const next = waitQueue.shift(); + if (next) next(); +}; + +/** Test-only: isolate-scoped module state outlives a single test case. */ +export const resetBuildSlotsForTest = (): void => { + activeBuilds = 0; + waitQueue.length = 0; +}; + +export const currentActiveBuildsForTest = (): number => activeBuilds; + +export const currentQueueLengthForTest = (): number => waitQueue.length; diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index 5ff5ee2f0..9263c4dfc 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -44,6 +44,7 @@ import { } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; +import { acquireBuildSlot, releaseBuildSlot } from "./session-build-semaphore"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its // engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is @@ -162,6 +163,20 @@ const loadAppShellHtml = makeAssetsShellHtmlLoader({ import("virtual:executor-mcp-apps-shell-dev-html").then((mod) => mod.devShellHtml), }); +// QuickJS-WASM must be loaded before the smoke render asks for a sandbox: the +// default variant cannot fetch its own `.wasm` on Workers. `../quickjs` is +// imported dynamically here, not at module scope, so a session that never +// calls create_artifact/edit_artifact never pays for it — see the comment on +// the dynamic import block in `buildMcpServer` for why that matters on a cold +// isolate. `preloadQuickJs()` itself is memoized per isolate (and resets on +// failure), so concurrent artifact calls, and repeat calls after the first, +// are all free past the first successful load. +const smokeRenderArtifactAfterQuickJsPreload: typeof smokeRenderArtifact = async (code) => { + const { preloadQuickJs } = await import("../quickjs"); + await preloadQuickJs(); + return smokeRenderArtifact(code); +}; + // --------------------------------------------------------------------------- // Durable Object // --------------------------------------------------------------------------- @@ -235,39 +250,42 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { const self = this; - return Effect.gen(function* () { + let buildSlotAcquired = false; + const build = Effect.gen(function* () { + // A burst of cold sessions landing on one isolate at once used to pay + // full concurrent build cost; this bounds it to `MAX_CONCURRENT_BUILDS` + // at a time and queues the rest FIFO. See `session-build-semaphore.ts`. + const buildQueueMs = yield* Effect.promise(() => acquireBuildSlot()); + buildSlotAcquired = true; + if (buildQueueMs > 0) { + yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_ms": buildQueueMs }); + } + // Imported here rather than at module scope. Cloudflare requires a // Durable Object class to be exported from the Worker entry, so every // static import this module makes is evaluated by *every* cold isolate — // including the ones that only render a page or forward a passthrough - // proxy and never open an MCP session. These three roots pull the whole - // code-execution stack (sucrase, ajv, QuickJS-WASM): measured at 1.9 MB - // of the Worker's startup closure, for code only a real session runs. + // proxy and never open an MCP session. These two roots pull the whole + // code-execution stack (sucrase, ajv): measured at 1.9 MB of the + // Worker's startup closure, for code only a real session runs. // `apps/cloud/scripts/start-closure.mjs` reports that number and will - // show it moving back if these become static again. - const [{ preloadQuickJs }, { makeExecutionStack }, { CloudMeteredExecutionStackLayer }] = - yield* Effect.promise( - () => - Promise.all([ - import("../quickjs"), - import("../engine/execution-stack"), - import("../engine/execution-stack-metered"), - ]) as Promise< - [ - typeof import("../quickjs"), - typeof import("../engine/execution-stack"), - typeof import("../engine/execution-stack-metered"), - ] - >, - ); + // show it moving back if these become static again. QuickJS-WASM is a + // separate dynamic import off the artifact smoke-render path only (see + // `smokeRenderArtifactAfterQuickJsPreload` above) — it is never needed + // during init, so it no longer lives in this Promise.all at all. + const [{ makeExecutionStack }, { CloudMeteredExecutionStackLayer }] = yield* Effect.promise( + () => + Promise.all([ + import("../engine/execution-stack"), + import("../engine/execution-stack-metered"), + ]) as Promise< + [ + typeof import("../engine/execution-stack"), + typeof import("../engine/execution-stack-metered"), + ] + >, + ); - // QuickJS-WASM must be loaded before anything asks for a sandbox: the - // default variant cannot fetch its own `.wasm` on Workers. Cloud runs - // user `execute` code on the dynamic-worker runtime, but the artifact - // smoke render is a QuickJS sandbox on every host — without this it fails - // open on each create and the check silently does nothing. - // Idempotent per isolate. - yield* Effect.promise(() => preloadQuickJs()); const { executor, engine } = yield* makeExecutionStack( sessionMeta.userId, sessionMeta.organizationId, @@ -307,7 +325,7 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase self.persistAppsEnabled(appsEnabled), loadAppShellHtml, - smokeRenderArtifact, + smokeRenderArtifact: smokeRenderArtifactAfterQuickJsPreload, artifactUrl: artifactUrlFor( env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", sessionMeta.organizationSlug, @@ -333,8 +351,19 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { + if (buildSlotAcquired) releaseBuildSlot(); + }), + ), Effect.provide(makeSessionServices(dbHandle)), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: runtime-build failures surface as the base's tapCause/cleanup defect Effect.orDie, From 8a08453ec94e1c33f017a346714d8c0a7e3ae1b9 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sat, 29 Aug 2026 00:12:25 -0700 Subject: [PATCH 3/3] Fix build-semaphore slot leak on interruption, add queue timeout A queued waiter interrupted mid-wait (client disconnect during a burst) used to leave its slot permanently unowned: the underlying promise couldn't be cancelled, so a later release would still hand it the slot with no fiber left to free it. acquireBuildSlot now returns a handle with an idempotent cancel() that dequeues an unstarted waiter or hands back an already-granted slot, wired through Effect.promise's abort signal plus an ensuring finalizer that both call the same cancel(). Also add a 10s max queue wait: a waiter that isn't granted a slot in time proceeds without one and is flagged on the span, so a stalled build degrades queued admission back to old unbounded-concurrent behavior instead of stalling everything behind it forever. --- .../src/mcp/session-build-semaphore.test.ts | 205 +++++++++++++++--- apps/cloud/src/mcp/session-build-semaphore.ts | 149 +++++++++++-- apps/cloud/src/mcp/session-durable-object.ts | 46 +++- 3 files changed, 345 insertions(+), 55 deletions(-) diff --git a/apps/cloud/src/mcp/session-build-semaphore.test.ts b/apps/cloud/src/mcp/session-build-semaphore.test.ts index ac34f3db7..3d4ad7634 100644 --- a/apps/cloud/src/mcp/session-build-semaphore.test.ts +++ b/apps/cloud/src/mcp/session-build-semaphore.test.ts @@ -14,30 +14,35 @@ describe("session-build-semaphore", () => { }); it("grants up to the cap immediately, with no wait", async () => { - const waits = await Promise.all([ - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), + const results = await Promise.all([ + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + ]); + expect(results).toEqual([ + { acquired: true, waitMs: 0, timedOut: false }, + { acquired: true, waitMs: 0, timedOut: false }, + { acquired: true, waitMs: 0, timedOut: false }, + { acquired: true, waitMs: 0, timedOut: false }, ]); - expect(waits).toEqual([0, 0, 0, 0]); expect(currentActiveBuildsForTest()).toBe(4); expect(currentQueueLengthForTest()).toBe(0); }); it("queues a build past the cap until a slot is released", async () => { await Promise.all([ - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, ]); expect(currentActiveBuildsForTest()).toBe(4); let fifthResolved = false; - const fifth = acquireBuildSlot().then((waitedMs) => { + const fifth = acquireBuildSlot().promise.then((result) => { fifthResolved = true; - return waitedMs; + return result; }); expect(currentQueueLengthForTest()).toBe(1); @@ -48,9 +53,11 @@ describe("session-build-semaphore", () => { expect(fifthResolved).toBe(false); releaseBuildSlot(); - const waitedMs = await fifth; + const result = await fifth; expect(fifthResolved).toBe(true); - expect(waitedMs).toBeGreaterThanOrEqual(0); + expect(result.acquired).toBe(true); + expect(result.timedOut).toBe(false); + expect(result.waitMs).toBeGreaterThanOrEqual(0); // The freed slot went straight to the waiter — total active stays at cap. expect(currentActiveBuildsForTest()).toBe(4); expect(currentQueueLengthForTest()).toBe(0); @@ -58,16 +65,16 @@ describe("session-build-semaphore", () => { it("releases queued waiters in FIFO order", async () => { await Promise.all([ - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, ]); const order: number[] = []; - const second = acquireBuildSlot().then(() => order.push(2)); - const third = acquireBuildSlot().then(() => order.push(3)); - const fourth = acquireBuildSlot().then(() => order.push(4)); + const second = acquireBuildSlot().promise.then(() => order.push(2)); + const third = acquireBuildSlot().promise.then(() => order.push(3)); + const fourth = acquireBuildSlot().promise.then(() => order.push(4)); expect(currentQueueLengthForTest()).toBe(3); releaseBuildSlot(); @@ -82,14 +89,14 @@ describe("session-build-semaphore", () => { it("never deadlocks: releasing a slot always makes forward progress for the next waiter", async () => { await Promise.all([ - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), - acquireBuildSlot(), + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, ]); // 6 more builds arrive while all 4 slots are held — all 6 queue. - const queued = Array.from({ length: 6 }, () => acquireBuildSlot()); + const queued = Array.from({ length: 6 }, () => acquireBuildSlot().promise); expect(currentQueueLengthForTest()).toBe(6); // The 4 in-flight builds finish one at a time; each release must hand its @@ -117,4 +124,150 @@ describe("session-build-semaphore", () => { releaseBuildSlot(); expect(currentActiveBuildsForTest()).toBe(0); }); + + it("cancelling a queued waiter dequeues it, and a subsequent release goes to the next waiter", async () => { + await Promise.all([ + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + ]); + expect(currentActiveBuildsForTest()).toBe(4); + + const cancelled = acquireBuildSlot(); + let cancelledSettled = false; + void cancelled.promise.then(() => { + cancelledSettled = true; + }); + + const next = acquireBuildSlot(); + let nextResult: Awaited | undefined; + void next.promise.then((r) => { + nextResult = r; + }); + + expect(currentQueueLengthForTest()).toBe(2); + + // Interrupted before it was ever granted a slot (e.g. client disconnect + // while queued) — this must not hand it a slot later, and must not block + // the waiter behind it. + cancelled.cancel(); + expect(currentQueueLengthForTest()).toBe(1); + + releaseBuildSlot(); + await next.promise; + + // The cancelled waiter's promise never resolves and never gets a slot; + // the freed slot went straight to the next waiter instead. + await Promise.resolve(); + await Promise.resolve(); + expect(cancelledSettled).toBe(false); + expect(nextResult).toEqual({ acquired: true, waitMs: expect.any(Number), timedOut: false }); + expect(currentActiveBuildsForTest()).toBe(4); + expect(currentQueueLengthForTest()).toBe(0); + }); + + it("cancel-after-grant releases the slot instead of leaking it", async () => { + await Promise.all([ + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + ]); + expect(currentActiveBuildsForTest()).toBe(4); + + // Queues, then gets granted a slot by the release below. + const grantedThenAbandoned = acquireBuildSlot(); + releaseBuildSlot(); + const result = await grantedThenAbandoned.promise; + expect(result.acquired).toBe(true); + // Still at the cap: the freed slot was handed straight to this waiter. + expect(currentActiveBuildsForTest()).toBe(4); + + // The caller never gets to consume the grant (e.g. its fiber was + // interrupted in the same tick the grant happened) and cancels instead — + // this must give the slot back rather than leaking it forever. + grantedThenAbandoned.cancel(); + expect(currentActiveBuildsForTest()).toBe(3); + + // Idempotent: a second cancel must not release it again. + grantedThenAbandoned.cancel(); + expect(currentActiveBuildsForTest()).toBe(3); + + // And the freed slot is usable by a fresh acquire. + const fresh = await acquireBuildSlot().promise; + expect(fresh).toEqual({ acquired: true, waitMs: 0, timedOut: false }); + expect(currentActiveBuildsForTest()).toBe(4); + }); + + it("cancelling an immediately-granted (never-queued) slot releases it exactly once", async () => { + const handle = acquireBuildSlot(); + await handle.promise; + expect(currentActiveBuildsForTest()).toBe(1); + + handle.cancel(); + expect(currentActiveBuildsForTest()).toBe(0); + + // Idempotent. + handle.cancel(); + expect(currentActiveBuildsForTest()).toBe(0); + }); + + it("proceeds without a slot when the queue wait exceeds the timeout, and does not count it as active", async () => { + await Promise.all([ + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + ]); + expect(currentActiveBuildsForTest()).toBe(4); + + const timedOutHandle = acquireBuildSlot(10); + const result = await timedOutHandle.promise; + + expect(result).toEqual({ acquired: false, waitMs: expect.any(Number), timedOut: true }); + expect(result.waitMs).toBeGreaterThanOrEqual(10); + // A timed-out waiter never counted against the cap. + expect(currentActiveBuildsForTest()).toBe(4); + expect(currentQueueLengthForTest()).toBe(0); + + // A timed-out waiter never held a slot, so cancelling it must not + // release one that was never granted. + timedOutHandle.cancel(); + expect(currentActiveBuildsForTest()).toBe(4); + + // The 4 originally-held slots are still releasable normally. + releaseBuildSlot(); + releaseBuildSlot(); + releaseBuildSlot(); + releaseBuildSlot(); + expect(currentActiveBuildsForTest()).toBe(0); + }); + + it("a waiter that times out is removed from the queue and does not block waiters behind it", async () => { + await Promise.all([ + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + acquireBuildSlot().promise, + ]); + + const timesOut = acquireBuildSlot(10); + const staysQueued = acquireBuildSlot(); + expect(currentQueueLengthForTest()).toBe(2); + + const timedOutResult = await timesOut.promise; + expect(timedOutResult.timedOut).toBe(true); + // Only the timed-out waiter left the queue; the other is still waiting. + expect(currentQueueLengthForTest()).toBe(1); + + releaseBuildSlot(); + const staysQueuedResult = await staysQueued.promise; + expect(staysQueuedResult).toEqual({ + acquired: true, + waitMs: expect.any(Number), + timedOut: false, + }); + expect(currentQueueLengthForTest()).toBe(0); + }); }); diff --git a/apps/cloud/src/mcp/session-build-semaphore.ts b/apps/cloud/src/mcp/session-build-semaphore.ts index c04e4ea9d..4e778deb2 100644 --- a/apps/cloud/src/mcp/session-build-semaphore.ts +++ b/apps/cloud/src/mcp/session-build-semaphore.ts @@ -15,36 +15,151 @@ const MAX_CONCURRENT_BUILDS = 4; +/** + * Max time a build waits in the FIFO queue for a slot before proceeding + * without one. Admission control exists only to smooth bursts into a short + * queue — it must never turn a *slow* build into a *stuck* one. If the + * isolate is wedged behind a hung dependency (e.g. a stalled DB during an + * incident) and a slot never frees up, every waiter behind it would otherwise + * block forever. Past this timeout a waiter gives up on the queue and builds + * uncapped instead, same as if the semaphore did not exist, and reports + * `timedOut: true` so it is visible in telemetry rather than silently + * degrading. + */ +export const MAX_QUEUE_WAIT_MS = 10_000; + let activeBuilds = 0; -const waitQueue: Array<() => void> = []; + +/** Lifecycle of one queued waiter. See `acquireBuildSlot` for the full state machine. */ +type WaiterState = "queued" | "granted" | "timed-out" | "cancelled"; + +interface Waiter { + state: WaiterState; + readonly grant: () => void; +} + +const waitQueue: Waiter[] = []; + +export interface BuildSlotResult { + /** False only when the wait timed out; the caller must NOT call `releaseBuildSlot` for it. */ + readonly acquired: boolean; + /** Milliseconds spent waiting for a slot (0 when one was immediately free). */ + readonly waitMs: number; + /** True when the wait hit `MAX_QUEUE_WAIT_MS` and proceeded without a slot. */ + readonly timedOut: boolean; +} + +export interface BuildSlotHandle { + /** Resolves exactly once, with the outcome of this acquisition. Never rejects. */ + readonly promise: Promise; + /** + * Cancel this acquisition. Safe to call at any point in its lifecycle, + * including redundantly or after `promise` has already settled — it is + * idempotent and only ever frees a slot once: + * - Still queued: dequeues the waiter. Nothing was ever granted, so there + * is nothing to release. + * - Already granted a slot — this races `releaseBuildSlot` handing the + * slot to this waiter (that happens synchronously inside + * `releaseBuildSlot`) against this `cancel` — releases that slot + * immediately so it is not held with no owner, and wakes the next FIFO + * waiter. + * - Already timed out or already cancelled: no-op. + * + * Callers should call this from a `finally`/`ensuring` unconditionally + * (success, failure, or interruption alike) rather than trying to track + * separately whether a slot was actually granted — idempotency here makes + * that bookkeeping unnecessary and closes the acquire/release race that + * bookkeeping was prone to. + */ + readonly cancel: () => void; +} /** - * Reserve a build slot, queueing FIFO when the cap is already held. Resolves - * with the number of milliseconds spent waiting for a slot — 0 when one was - * immediately free, which is the common case outside a burst. + * Reserve a build slot, queueing FIFO when the cap is already held, up to + * `maxQueueWaitMs` (defaults to `MAX_QUEUE_WAIT_MS`; overridable for tests). * - * Always resolves, never rejects: there is nothing to fail here, only to - * wait for. + * Returns a handle rather than a bare promise so a caller that stops waiting + * — most notably an `Effect` fiber interrupted mid-queue on a client + * disconnect — can cancel cleanly instead of leaking the slot a later + * `releaseBuildSlot` would otherwise hand to a waiter nobody is listening to + * anymore. */ -export const acquireBuildSlot = (): Promise => { +export const acquireBuildSlot = (maxQueueWaitMs: number = MAX_QUEUE_WAIT_MS): BuildSlotHandle => { const requestedAt = Date.now(); + if (activeBuilds < MAX_CONCURRENT_BUILDS) { activeBuilds += 1; - return Promise.resolve(0); + let released = false; + return { + promise: Promise.resolve({ acquired: true, waitMs: 0, timedOut: false }), + cancel: () => { + if (released) return; + released = true; + releaseBuildSlot(); + }, + }; } - return new Promise((resolve) => { - waitQueue.push(() => { - activeBuilds += 1; - resolve(Date.now() - requestedAt); - }); + + let resolvePromise!: (value: BuildSlotResult) => void; + const promise = new Promise((resolve) => { + resolvePromise = resolve; }); + + const waiter: Waiter = { + state: "queued", + grant: () => { + // Guaranteed "queued" here: `releaseBuildSlot` only reaches a waiter by + // shifting it out of `waitQueue`, and both other transitions + // (timeout, cancel) remove the waiter from the queue in the same tick + // they change its state, so a granted waiter can't have been anything + // else. + waiter.state = "granted"; + activeBuilds += 1; + resolvePromise({ acquired: true, waitMs: Date.now() - requestedAt, timedOut: false }); + }, + }; + waitQueue.push(waiter); + + const timer = setTimeout(() => { + if (waiter.state !== "queued") return; + waiter.state = "timed-out"; + const idx = waitQueue.indexOf(waiter); + if (idx !== -1) waitQueue.splice(idx, 1); + resolvePromise({ acquired: false, waitMs: Date.now() - requestedAt, timedOut: true }); + }, maxQueueWaitMs); + + return { + promise, + cancel: () => { + clearTimeout(timer); + if (waiter.state === "queued") { + waiter.state = "cancelled"; + const idx = waitQueue.indexOf(waiter); + if (idx !== -1) waitQueue.splice(idx, 1); + return; + } + if (waiter.state === "granted") { + // Grant already happened (raced with a concurrent `releaseBuildSlot` + // shifting this waiter off the queue) before this cancel ran, and + // the caller never got to consume the slot — hand it back now + // instead of leaking it. + waiter.state = "cancelled"; + releaseBuildSlot(); + return; + } + // "timed-out": never held a slot. "cancelled": a prior call already + // handled this. Either way, nothing to do — idempotent no-op. + }, + }; }; /** * Release a build slot. Must be called exactly once per slot a caller - * actually acquired (i.e. `acquireBuildSlot` resolved) — callers release from - * a `finally`/`ensuring` so a build that throws still frees its slot and - * never deadlocks the queue behind it. + * actually acquired (i.e. `acquireBuildSlot`'s `promise` resolved with + * `acquired: true`) — callers release from a `finally`/`ensuring` (typically + * via the returned handle's `cancel`, which is idempotent) so a build that + * throws, or is interrupted, still frees its slot and never deadlocks the + * queue behind it. * * Wakes the next FIFO waiter, if any, handing it the freed slot directly * rather than making it race a fresh `acquireBuildSlot` caller. @@ -52,7 +167,7 @@ export const acquireBuildSlot = (): Promise => { export const releaseBuildSlot = (): void => { activeBuilds = Math.max(0, activeBuilds - 1); const next = waitQueue.shift(); - if (next) next(); + if (next) next.grant(); }; /** Test-only: isolate-scoped module state outlives a single test case. */ diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index 9263c4dfc..6b9963eb8 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -44,7 +44,7 @@ import { } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; -import { acquireBuildSlot, releaseBuildSlot } from "./session-build-semaphore"; +import { acquireBuildSlot, type BuildSlotHandle } from "./session-build-semaphore"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its // engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is @@ -250,15 +250,35 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { const self = this; - let buildSlotAcquired = false; + // Set synchronously, inside the `Effect.promise` executor below, before + // any `await` — so there is no window where interruption can land after + // the handle exists but before this is assigned. `cancel` is idempotent + // (see `session-build-semaphore.ts`), so it is safe to invoke from both + // the abort listener below AND the `Effect.ensuring` finalizer without + // risking a double release. + let buildSlot: BuildSlotHandle | undefined; const build = Effect.gen(function* () { // A burst of cold sessions landing on one isolate at once used to pay // full concurrent build cost; this bounds it to `MAX_CONCURRENT_BUILDS` - // at a time and queues the rest FIFO. See `session-build-semaphore.ts`. - const buildQueueMs = yield* Effect.promise(() => acquireBuildSlot()); - buildSlotAcquired = true; - if (buildQueueMs > 0) { - yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_ms": buildQueueMs }); + // at a time and queues the rest FIFO, degrading to old concurrent + // behavior (see `timedOut` below) rather than stalling if the queue + // itself gets stuck. See `session-build-semaphore.ts`. + // + // `Effect.promise` hands its executor an `AbortSignal` tied to this + // fiber's own interruption — firing it (rather than merely walking + // away) is what lets a waiter still sitting in the semaphore's queue + // be dequeued immediately on a client disconnect, instead of getting + // granted a slot later with no one left to release it. + const slot = yield* Effect.promise((signal) => { + const handle = acquireBuildSlot(); + buildSlot = handle; + signal.addEventListener("abort", () => handle.cancel(), { once: true }); + return handle.promise; + }); + if (slot.timedOut) { + yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_timeout": true }); + } else if (slot.waitMs > 0) { + yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_ms": slot.waitMs }); } // Imported here rather than at module scope. Cloudflare requires a @@ -355,13 +375,15 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase { - if (buildSlotAcquired) releaseBuildSlot(); + buildSlot?.cancel(); }), ), Effect.provide(makeSessionServices(dbHandle)),