Unified · Split
| @@ -154,6 +154,60 @@ describe("HttpRouter.toWebHandler request scoping", () => { · 153 unmodified lines | |||
| 154 | }); | 154 | }); |
| 155 | }); | 155 | }); |
| 156 | 156 | ||
| 157 | // --------------------------------------------------------------------------- | ||
| 158 | // WHEN the per-request resource dies, relative to the response bytes. | ||
| 159 | // | ||
| 160 | // `requestScopedMiddleware` wraps the ROUTE HANDLER in `Effect.scoped`, and the | ||
| 161 | // response is written later still, by `HttpEffect.toHandled`'s callback. So the | ||
| 162 | // postgres socket is already closed by the time the client has anything. That | ||
| 163 | // is why cloud's HTTP plane leaves `ExecutorConfig.deferToolSync` unset (see | ||
| 164 | // `engine/execution-stack.ts`): there is no window on this plane that is both | ||
| 165 | // after the response and before the socket closes, so `waitUntil` would only | ||
| 166 | // ever hand a dead pool to the deferred batch. | ||
| 167 | // | ||
| 168 | // Pinned rather than assumed. If the request lifecycle is ever restructured so | ||
| 169 | // the scope outlives the written response, this test fails and the deferral | ||
| 170 | // becomes available — which is a thing to be told, not to rediscover. | ||
| 171 | // --------------------------------------------------------------------------- | ||
| 172 | |||
| 173 | describe("per-request resource lifetime vs the written response", () => { | ||
| 174 | it("releases the request-scoped resource before the response resolves", async () => { | ||
| 175 | const order: string[] = []; | ||
| 176 | const Tracked = Layer.effectDiscard( | ||
| 177 | Effect.acquireRelease( | ||
| 178 | Effect.sync(() => { | ||
| 179 | order.push("acquire"); | ||
| 180 | }), | ||
| 181 | () => | ||
| 182 | Effect.sync(() => { | ||
| 183 | order.push("release"); | ||
| 184 | }), | ||
| 185 | ), | ||
| 186 | ); | ||
| 187 | const Route = HttpRouter.add( | ||
| 188 | "GET", | ||
| 189 | "/", | ||
| 190 | Effect.sync(() => { | ||
| 191 | order.push("handler"); | ||
| 192 | return HttpServerResponse.jsonUnsafe({ ok: true }); | ||
| 193 | }), | ||
| 194 | ); | ||
| 195 | const handler = HttpRouter.toWebHandler( | ||
| 196 | Route.pipe( | ||
| 197 | Layer.provide(requestScopedMiddleware(Tracked).layer), | ||
| 198 | Layer.provideMerge(HttpServer.layerServices), | ||
| 199 | ), | ||
| 200 | { disableLogger: true }, | ||
| 201 | ).handler; | ||
| 202 | |||
| 203 | const response = await handler(new Request("http://test.local/")); | ||
| 204 | order.push("response"); | ||
| 205 | |||
| 206 | expect(response.status).toBe(200); | ||
| 207 | expect(order).toEqual(["acquire", "handler", "release", "response"]); | ||
| 208 | }); | ||
| 209 | }); | ||
| 210 | |||
| 157 | // --------------------------------------------------------------------------- | 211 | // --------------------------------------------------------------------------- |
| 158 | // Regression test against the prod handler factory. If anyone reverts | 212 | // Regression test against the prod handler factory. If anyone reverts |
| 159 | // `makeApiLive` back to wiring `RequestScopedServicesLive` via | 213 | // `makeApiLive` back to wiring `RequestScopedServicesLive` via |
| @@ -27,7 +27,7 @@ import { · 26 unmodified lines | |||
| 27 | 27 | ||
| 28 | import { AutumnService } from "../extensions/billing/service"; | 28 | import { AutumnService } from "../extensions/billing/service"; |
| 29 | import type { DbService } from "../db/db"; | 29 | import type { DbService } from "../db/db"; |
| 30 | import { CloudExecutionSeamsLayer } from "../engine/execution-stack"; | 30 | import { CloudExecutionSeamsLayer, makeCloudExecutionSeamsLayer } from "../engine/execution-stack"; |
| 31 | import { makeExecutionLimitGate } from "./execution-gate"; | 31 | import { makeExecutionLimitGate } from "./execution-gate"; |
| 32 | import { makeCloudExecutionRateLimiter } from "./execution-rate-limit"; | 32 | import { makeCloudExecutionRateLimiter } from "./execution-rate-limit"; |
| 33 | import { withExecutionUsageTracking } from "./execution-usage"; | 33 | import { withExecutionUsageTracking } from "./execution-usage"; |
| @@ -75,3 +75,17 @@ export const CloudMeteredExecutionStackLayer: Layer.Layer< · 41 unmodified lines | |||
| 75 | never, | 75 | never, |
| 76 | AutumnService | DbService | 76 | AutumnService | DbService |
| 77 | > = Layer.merge(CloudExecutionSeamsLayer, CloudMeteringEngineDecorator); | 77 | > = Layer.merge(CloudExecutionSeamsLayer, CloudMeteringEngineDecorator); |
| 78 | |||
| 79 | /** | ||
| 80 | * The same stack over a caller-supplied `HostConfig`. The MCP session DO builds | ||
| 81 | * one so it can defer tool-catalog refreshes to its own `ctx.waitUntil` — its | ||
| 82 | * database handle outlives any single request, where the HTTP plane's does not. | ||
| 83 | * Everything else is identical, decorator included. | ||
| 84 | */ | ||
| 85 | export const makeCloudMeteredExecutionStackLayer = ( | ||
| 86 | hostConfig: Layer.Layer<HostConfig>, | ||
| 87 | ): Layer.Layer< | ||
| 88 | DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider | EngineDecorator, | ||
| 89 | never, | ||
| 90 | AutumnService | DbService | ||
| 91 | > => Layer.merge(makeCloudExecutionSeamsLayer(hostConfig), CloudMeteringEngineDecorator); | ||
| @@ -41,6 +41,7 @@ import { · 40 unmodified lines | |||
| 41 | HostConfig, | 41 | HostConfig, |
| 42 | PluginsProvider, | 42 | PluginsProvider, |
| 43 | collectTables, | 43 | collectTables, |
| 44 | type HostConfigShape, | ||
| 44 | } from "@executor-js/api/server"; | 45 | } from "@executor-js/api/server"; |
| 45 | import { makeDynamicWorkerExecutor } from "@executor-js/runtime-dynamic-worker"; | 46 | import { makeDynamicWorkerExecutor } from "@executor-js/runtime-dynamic-worker"; |
| 46 | import type { AnyPlugin } from "@executor-js/sdk"; | 47 | import type { AnyPlugin } from "@executor-js/sdk"; |
| @@ -88,18 +89,69 @@ export const CloudPluginsProvider: Layer.Layer<PluginsProvider> = Layer.succeed( · 41 unmodified lines | |||
| 88 | */ | 89 | */ |
| 89 | export const CLOUD_MOUNT_PREFIX = "/api" as const; | 90 | export const CLOUD_MOUNT_PREFIX = "/api" as const; |
| 90 | 91 | ||
| 91 | export const CloudHostConfig: Layer.Layer<HostConfig> = Layer.sync(HostConfig, () => ({ | 92 | /** |
| 92 | // SSRF / private-network egress guard. Config-driven, NOT a test flag: | 93 | * Cloud's host config, parameterized by the ONE seam its two execution planes |
| 93 | // production leaves `ALLOW_LOCAL_NETWORK` unset so the guard stays ON (`false`); | 94 | * genuinely disagree about: where a deferred tool-catalog refresh runs. |
| 94 | // the e2e dev-server env opts in with `"true"` so in-scenario fixture | 95 | * |
| 95 | // servers on localhost are reachable. See `hosted-http-client.ts`. | 96 | * The MCP session Durable Object holds its database handle for the life of the |
| 96 | allowLocalNetwork: env.ALLOW_LOCAL_NETWORK === "true", | 97 | * session, so it can hand the batch straight to `ctx.waitUntil` and forget it. |
| 97 | webBaseUrl: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", | 98 | * The HTTP plane's handle dies with its request and gets no seam at all (see |
| 98 | oauthCallbackPath: `${CLOUD_MOUNT_PREFIX}/oauth/callback`, | 99 | * {@link CloudHostConfig}). Nothing else about the config differs, which is why |
| 99 | // WorkOS Vault is cloud's credential storage implementation detail, not a | 100 | * this is a parameter rather than a second config module. |
| 100 | // user-selectable provider surface. | 101 | */ |
| 101 | exposeCredentialProviders: false, | 102 | export const makeCloudHostConfig = ( |
| 102 | })); | 103 | deferToolSync: HostConfigShape["deferToolSync"], |
| 104 | ): Layer.Layer<HostConfig> => | ||
| 105 | Layer.sync(HostConfig, () => ({ | ||
| 106 | // SSRF / private-network egress guard. Config-driven, NOT a test flag: | ||
| 107 | // production leaves `ALLOW_LOCAL_NETWORK` unset so the guard stays ON (`false`); | ||
| 108 | // the e2e dev-server env opts in with `"true"` so in-scenario fixture | ||
| 109 | // servers on localhost are reachable. See `hosted-http-client.ts`. | ||
| 110 | allowLocalNetwork: env.ALLOW_LOCAL_NETWORK === "true", | ||
| 111 | webBaseUrl: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", | ||
| 112 | oauthCallbackPath: `${CLOUD_MOUNT_PREFIX}/oauth/callback`, | ||
| 113 | // WorkOS Vault is cloud's credential storage implementation detail, not a | ||
| 114 | // user-selectable provider surface. | ||
| 115 | exposeCredentialProviders: false, | ||
| 116 | deferToolSync, | ||
| 117 | })); | ||
| 118 | |||
| 119 | /** | ||
| 120 | * The HTTP plane's config, and the one seam it CANNOT fill: `deferToolSync` is | ||
| 121 | * absent, so a tools read over `/api/*` still refreshes an expired catalog | ||
| 122 | * inline. | ||
| 123 | * | ||
| 124 | * Not an omission. The postgres pool a request's executor closes over is | ||
| 125 | * acquired and released inside `requestScopedMiddleware`'s `Effect.scoped`, | ||
| 126 | * which wraps the ROUTE HANDLER — so the pool is torn down while the handler's | ||
| 127 | * `HttpServerResponse` is still a value, strictly before `HttpEffect.toHandled` | ||
| 128 | * writes it. There is therefore no window on this plane that is both after the | ||
| 129 | * response and before the socket closes: work scheduled with `waitUntil` finds | ||
| 130 | * a dead pool, and work drained inside the scope is simply inline work with | ||
| 131 | * extra steps. `api.request-scope.node.test.ts` pins that ordering, so a future | ||
| 132 | * change to the request lifecycle will say so rather than making this comment | ||
| 133 | * quietly wrong. | ||
| 134 | * | ||
| 135 | * Holding the pool open past the response instead — deferring the scope close | ||
| 136 | * into `waitUntil` — is the one thing that would make the window exist, and it | ||
| 137 | * is the thing `closePostgres` was written to prevent. That finalizer awaits | ||
| 138 | * the teardown precisely so live-plus-closing sockets stay bounded by what is | ||
| 139 | * in flight; unbounded, the backlog is the sustained-load cascade recorded in | ||
| 140 | * `db/db.ts`. Trading a socket leak for a background refresh is not a trade | ||
| 141 | * this plane should make, and it would make it on EVERY request, not just the | ||
| 142 | * ones that deferred. | ||
| 143 | * | ||
| 144 | * The alternative — giving the batch a short-lived pool of its own via | ||
| 145 | * `makeDbLayer()` — does not fit either: the task the sdk hands over is already | ||
| 146 | * bound to the executor that built it, so re-targeting it means rebuilding a | ||
| 147 | * whole second executor in the background. That is a background scheduler, a | ||
| 148 | * different feature with a different failure surface, not a wiring detail. | ||
| 149 | * | ||
| 150 | * Cloud's high-volume tools read is `tools/list` over MCP, and that plane DOES | ||
| 151 | * defer: the session Durable Object holds its database handle for the life of | ||
| 152 | * the session (see `mcp/session-durable-object.ts`). | ||
| 153 | */ | ||
| 154 | export const CloudHostConfig: Layer.Layer<HostConfig> = makeCloudHostConfig(undefined); | ||
| 103 | 155 | ||
| 104 | export const CloudCodeExecutorProvider: Layer.Layer<CodeExecutorProvider> = Layer.sync( | 156 | export const CloudCodeExecutorProvider: Layer.Layer<CodeExecutorProvider> = Layer.sync( |
| 105 | CodeExecutorProvider, | 157 | CodeExecutorProvider, |
| @@ -114,13 +166,16 @@ export const CloudCodeExecutorProvider: Layer.Layer<CodeExecutorProvider> = Laye · 8 unmodified lines | |||
| 114 | * exported so that overlay builds over the SAME four seams. There is no neutral | 166 | * exported so that overlay builds over the SAME four seams. There is no neutral |
| 115 | * no-op-decorator variant anymore: every cloud execution meters. | 167 | * no-op-decorator variant anymore: every cloud execution meters. |
| 116 | */ | 168 | */ |
| 169 | export const makeCloudExecutionSeamsLayer = ( | ||
| 170 | hostConfig: Layer.Layer<HostConfig>, | ||
| 171 | ): Layer.Layer< | ||
| 172 | DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider, | ||
| 173 | never, | ||
| 174 | DbService | ||
| 175 | > => Layer.mergeAll(CloudDbProvider, CloudPluginsProvider, hostConfig, CloudCodeExecutorProvider); | ||
| 176 | |||
| 117 | export const CloudExecutionSeamsLayer: Layer.Layer< | 177 | export const CloudExecutionSeamsLayer: Layer.Layer< |
| 118 | DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider, | 178 | DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider, |
| 119 | never, | 179 | never, |
| 120 | DbService | 180 | DbService |
| 121 | > = Layer.mergeAll( | 181 | > = makeCloudExecutionSeamsLayer(CloudHostConfig); |
| 122 | CloudDbProvider, | ||
| 123 | CloudPluginsProvider, | ||
| 124 | CloudHostConfig, | ||
| 125 | CloudCodeExecutorProvider, | ||
| 126 | ); | ||
| @@ -46,7 +46,7 @@ import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; · 45 unmodified lines | |||
| 46 | import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; | 46 | import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; |
| 47 | 47 | ||
| 48 | // The DO meters executions just like the HTTP `/api/*` plane: it builds its | 48 | // The DO meters executions just like the HTTP `/api/*` plane: it builds its |
| 49 | // engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is | 49 | // engine with the metered execution stack, so every MCP execution is |
| 50 | // tracked to Autumn (the MCP server is the primary execution surface, so leaving | 50 | // tracked to Autumn (the MCP server is the primary execution surface, so leaving |
| 51 | // it unmetered silently dropped the bulk of real usage). The billing service | 51 | // it unmetered silently dropped the bulk of real usage). The billing service |
| 52 | // (`AutumnService.Default`) is provided LOCALLY to the metered stack below, so | 52 | // (`AutumnService.Default`) is provided LOCALLY to the metered stack below, so |
| @@ -66,9 +66,9 @@ import { · 13 unmodified lines | |||
| 66 | type DrizzleDb, | 66 | type DrizzleDb, |
| 67 | type DbServiceShape, | 67 | type DbServiceShape, |
| 68 | } from "../db/db"; | 68 | } from "../db/db"; |
| 69 | import { makeExecutionStack } from "../engine/execution-stack"; | 69 | import { makeCloudHostConfig, makeExecutionStack } from "../engine/execution-stack"; |
| 70 | import { preloadQuickJs } from "../quickjs"; | 70 | import { preloadQuickJs } from "../quickjs"; |
| 71 | import { CloudMeteredExecutionStackLayer } from "../engine/execution-stack-metered"; | 71 | import { makeCloudMeteredExecutionStackLayer } from "../engine/execution-stack-metered"; |
| 72 | import { AutumnService } from "../extensions/billing/service"; | 72 | import { AutumnService } from "../extensions/billing/service"; |
| 73 | import { DoTelemetryLive, flushTracerProvider } from "../observability/telemetry"; | 73 | import { DoTelemetryLive, flushTracerProvider } from "../observability/telemetry"; |
| 74 | import { | 74 | import { |
| @@ -236,6 +236,46 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase<Env, CloudSessionD · 161 unmodified lines | |||
| 236 | ); | 236 | ); |
| 237 | } | 237 | } |
| 238 | 238 | ||
| 239 | /** | ||
| 240 | * `ExecutorConfig.deferToolSync` for the MCP plane: a tools read's | ||
| 241 | * speculative catalog refresh runs after the read has answered, on this | ||
| 242 | * session's own database handle. | ||
| 243 | * | ||
| 244 | * Detached rather than drained before some closing scope: the handle this | ||
| 245 | * session's executor closes over is opened once in `openSessionDb` and lives | ||
| 246 | * as long as the DO does, so there is nothing to drain before. `ctx.waitUntil` | ||
| 247 | * only holds the invocation open long enough for the batch to finish, exactly | ||
| 248 | * as the pending-approval leases above do. The task is total by contract, so | ||
| 249 | * `runPromise` cannot reject. | ||
| 250 | * | ||
| 251 | * `DoTelemetryLive` IS re-provided here, and this is the one place that is | ||
| 252 | * correct — see the prohibition on `makeSessionServices`. A bare `runPromise` | ||
| 253 | * carries no `Tracer` at all, so every `executor.tools.sync` span the batch | ||
| 254 | * emits would be silently dropped, and the background refresh would be the | ||
| 255 | * one part of this feature invisible in production. Providing it costs the | ||
| 256 | * link to the read's trace, which is the point: this work is deliberately not | ||
| 257 | * part of that request, and hanging it under a span that closed first would | ||
| 258 | * inflate exactly the tools-read latency the deferral exists to remove. The | ||
| 259 | * batch gets its own root trace instead. Flushed at the end because the read's | ||
| 260 | * own `flushTelemetry` has long since run by the time this finishes. | ||
| 261 | */ | ||
| 262 | private readonly deferToolSync = (task: Effect.Effect<void>): Effect.Effect<void> => | ||
| 263 | Effect.sync(() => { | ||
| 264 | this.ctx.waitUntil( | ||
| 265 | Effect.runPromise( | ||
| 266 | task.pipe( | ||
| 267 | Effect.provide(DoTelemetryLive), | ||
| 268 | Effect.andThen( | ||
| 269 | Effect.tryPromise({ | ||
| 270 | try: () => flushTracerProvider(), | ||
| 271 | catch: () => undefined, | ||
| 272 | }).pipe(Effect.ignore), | ||
| 273 | ), | ||
| 274 | ), | ||
| 275 | ), | ||
| 276 | ); | ||
| 277 | }); | ||
| 278 | |||
| 239 | protected override buildMcpServer( | 279 | protected override buildMcpServer( |
| 240 | sessionMeta: SessionMeta, | 280 | sessionMeta: SessionMeta, |
| 241 | dbHandle: CloudSessionDbHandle, | 281 | dbHandle: CloudSessionDbHandle, |
| @@ -261,7 +301,11 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase<Env, CloudSessionD · 19 unmodified lines | |||
| 261 | // outer `makeSessionServices`. When `AUTUMN_SECRET_KEY` is unset the | 301 | // outer `makeSessionServices`. When `AUTUMN_SECRET_KEY` is unset the |
| 262 | // billing service degrades to a no-op tracker, so this stays inert in | 302 | // billing service degrades to a no-op tracker, so this stays inert in |
| 263 | // cloud dev/preview environments that run without a billing backend. | 303 | // cloud dev/preview environments that run without a billing backend. |
| 264 | Effect.provide(CloudMeteredExecutionStackLayer.pipe(Layer.provide(AutumnService.Default))), | 304 | Effect.provide( |
| 305 | makeCloudMeteredExecutionStackLayer(makeCloudHostConfig(self.deferToolSync)).pipe( | ||
| 306 | Layer.provide(AutumnService.Default), | ||
| 307 | ), | ||
| 308 | ), | ||
| 265 | Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), | 309 | Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), |
| 266 | ); | 310 | ); |
| 267 | // Build the description here so `executor.connections.list()` stays under | 311 | // Build the description here so `executor.connections.list()` stays under |