| 1 | import * as Duration from "effect/Duration"; |
| 2 | import * as Effect from "effect/Effect"; |
| 3 | import * as Layer from "effect/Layer"; |
| 4 | import * as Schedule from "effect/Schedule"; |
| 5 | import * as Semaphore from "effect/Semaphore"; |
| 6 | import * as HttpClientRequest from "effect/http/HttpClientRequest"; |
| 7 | import * as HttpClientResponse from "effect/http/HttpClientResponse"; |
| 8 | import * as HttpServerRequest from "effect/http/HttpServerRequest"; |
| 9 | import * as HttpServerResponse from "effect/http/HttpServerResponse"; |
| 10 | import { ALCHEMY_PHASE } from "../../Phase.ts"; |
| 11 | import { makeFetchRpcStub } from "../../Rpc.ts"; |
| 12 | import { type Fetcher } from "../Fetcher.ts"; |
| 13 | import { DurableObjectState } from "../Workers/DurableObjectState.ts"; |
| 14 | import { |
| 15 | ContainerCrashedError, |
| 16 | ContainerError, |
| 17 | ContainerRateLimitedError, |
| 18 | ContainerTag, |
| 19 | NoContainerInstanceError, |
| 20 | type Container, |
| 21 | type ContainerStartupOptions, |
| 22 | } from "./Container.ts"; |
| 23 | |
| 24 | /** |
| 25 | * Per-container-instance start coordination (start mutex + confirmed-ready |
| 26 | * port cache). |
| 27 | * |
| 28 | * `startContainer` can be invoked more than once for the same physical |
| 29 | * container: the {@link layer} helper calls it once per Durable Object, but a |
| 30 | * caller may also use `startContainer` directly from several places. If the |
| 31 | * mutex/ready-cache lived in each `startContainer` closure, those calls would |
| 32 | * not coordinate — two of them could both observe `running === false` and both |
| 33 | * call `container.start()`, with the second throwing "already running" (see |
| 34 | * cloudflare/containers#173). |
| 35 | * |
| 36 | * Keying off the Durable Object's `DurableObjectState` (stable for the life of |
| 37 | * a DO instance, and a distinct object per instance) makes the coordination |
| 38 | * per-container-instance regardless of how many `startContainer` calls share |
| 39 | * it. The map is module-scoped, but Workers isolates are per-isolate and each |
| 40 | * DO instance has its own `DurableObjectState`, so instances never collide; the |
| 41 | * `WeakMap` lets entries be collected with the instance. We deliberately do NOT |
| 42 | * key on the container's logical id — that is shared across every |
| 43 | * `getByName(...)` instance of the same class, which must each start |
| 44 | * independently. |
| 45 | */ |
| 46 | const startCoordination = new WeakMap< |
| 47 | object, |
| 48 | { |
| 49 | readonly startMutex: ReturnType<typeof Semaphore.makeUnsafe>; |
| 50 | readonly readyPorts: Set<number>; |
| 51 | } |
| 52 | >(); |
| 53 | |
| 54 | const getStartCoordination = (key: object) => { |
| 55 | let coordination = startCoordination.get(key); |
| 56 | if (!coordination) { |
| 57 | // `makeUnsafe` (rather than `Effect.makeSemaphore`) because this runs in a |
| 58 | // synchronous cache-miss branch; the semaphore is a plain mutable primitive |
| 59 | // shared across effects, created once per DO instance. |
| 60 | coordination = { |
| 61 | startMutex: Semaphore.makeUnsafe(1), |
| 62 | readyPorts: new Set<number>(), |
| 63 | }; |
| 64 | startCoordination.set(key, coordination); |
| 65 | } |
| 66 | return coordination; |
| 67 | }; |
| 68 | |
| 69 | export const layer = <Image extends Container.Decl.Any>( |
| 70 | container: Image, |
| 71 | options?: ContainerStartupOptions, |
| 72 | ) => { |
| 73 | const id = (container as any)["~alchemy/Id"] as string; |
| 74 | // Provide the *started* instance under the same tag `yield* MyContainer` |
| 75 | // resolves (keyed by logical id) so the two compose. |
| 76 | return Layer.effect( |
| 77 | ContainerTag(id), |
| 78 | startContainer(container, options), |
| 79 | ) as Layer.Layer<InstanceType<Image>>; |
| 80 | }; |
| 81 | |
| 82 | /** |
| 83 | * Runs the Container in a Durable Object and monitors it, providing a durable fetch and RPC interface to it. |
| 84 | */ |
| 85 | export const startContainer = Effect.fn(function* < |
| 86 | Image extends Container.Decl.Any, |
| 87 | >(containerEff: Image, options?: ContainerStartupOptions) { |
| 88 | const bindEff = (containerEff as any)["~alchemy/Container/Binding"] as |
| 89 | | Effect.Effect<Effect.Effect<Container>, never, any> |
| 90 | | undefined; |
| 91 | const bound = yield* ( |
| 92 | bindEff ?? (containerEff as any as Effect.Effect<any, never, never>) |
| 93 | ); |
| 94 | const container: Container = Effect.isEffect(bound) |
| 95 | ? yield* bound as Effect.Effect<Container> |
| 96 | : (bound as Container); |
| 97 | |
| 98 | // Every constant below is taken directly from Cloudflare's own |
| 99 | // `@cloudflare/containers` runtime (`dist/lib/container.js`) rather than |
| 100 | // guessed, so our readiness behaviour matches `startAndWaitForPorts`: |
| 101 | // |
| 102 | // INSTANCE_POLL_INTERVAL_MS = 300 → fixed poll interval |
| 103 | // PING_TIMEOUT_MS = 5_000 → per-probe cap (addTimeoutSignal) |
| 104 | // TIMEOUT_TO_GET_CONTAINER_MS = 8_000 → phase 1: find+start an instance |
| 105 | // TIMEOUT_TO_GET_PORTS_MS = 20_000 → phase 2: wait for the port to listen |
| 106 | // |
| 107 | // Native derives its retry COUNTS as ceil(timeout / pollInterval); we keep |
| 108 | // the two phases separate (as native does) on a 300ms cadence. |
| 109 | const READINESS_POLL_INTERVAL = Duration.millis(300); // INSTANCE_POLL_INTERVAL_MS |
| 110 | const READINESS_PROBE_TIMEOUT = Duration.seconds(5); // PING_TIMEOUT_MS |
| 111 | const TIMEOUT_TO_GET_CONTAINER_MS = 8_000; |
| 112 | const TIMEOUT_TO_GET_PORTS_MS = 20_000; |
| 113 | const GET_CONTAINER_RETRIES = Math.ceil(TIMEOUT_TO_GET_CONTAINER_MS / 300); // 27 |
| 114 | const PORT_READY_RETRIES = Math.ceil(TIMEOUT_TO_GET_PORTS_MS / 300); // 67 |
| 115 | // When rate limited, back off hard before retrying — hammering `start()` |
| 116 | // only prolongs the limit (native returns 429 immediately and does not loop). |
| 117 | const RATE_LIMIT_BACKOFF = Duration.seconds(2); |
| 118 | const RATE_LIMIT_RETRIES = 5; |
| 119 | // Light retry for the real request only — covers the transient |
| 120 | // "Network connection lost" window when a container instance is recycled. |
| 121 | const REQUEST_RETRIES = 3; |
| 122 | |
| 123 | // Error classification, mirroring native's string matchers in |
| 124 | // `@cloudflare/containers` (`isErrorOfType` against these exact phrases). |
| 125 | const errString = (e: unknown) => |
| 126 | (e instanceof Error ? e.message : String(e)).toLowerCase(); |
| 127 | const isNoInstanceError = (e: unknown) => |
| 128 | errString(e).includes( |
| 129 | "there is no container instance that can be provided to this durable object", |
| 130 | ); |
| 131 | const isRateLimitedError = (e: unknown) => |
| 132 | errString(e).includes("you are requesting too many containers per second"); |
| 133 | |
| 134 | type ReadinessError = |
| 135 | | ContainerError |
| 136 | | NoContainerInstanceError |
| 137 | | ContainerRateLimitedError |
| 138 | | ContainerCrashedError; |
| 139 | |
| 140 | // Map a raw defect to a typed readiness error (native classifies the same |
| 141 | // message phrases via `isErrorOfType`). |
| 142 | const classifyDefect = ( |
| 143 | defect: unknown, |
| 144 | portNumber: number, |
| 145 | ): ContainerError | NoContainerInstanceError | ContainerRateLimitedError => |
| 146 | isRateLimitedError(defect) |
| 147 | ? new ContainerRateLimitedError({ |
| 148 | message: "Rate limited starting container", |
| 149 | cause: defect, |
| 150 | }) |
| 151 | : isNoInstanceError(defect) |
| 152 | ? new NoContainerInstanceError({ |
| 153 | message: "No container instance available", |
| 154 | cause: defect, |
| 155 | }) |
| 156 | : new ContainerError({ |
| 157 | message: `Container not ready on port ${portNumber}: ${defect}`, |
| 158 | cause: defect, |
| 159 | }); |
| 160 | |
| 161 | // Coalesce concurrent starts and share the confirmed-ready port cache across |
| 162 | // every caller for THIS Durable Object instance (see `startCoordination`). |
| 163 | // Native uses a `startInFlight` promise so two racing callers never both |
| 164 | // invoke `container.start()` ("already running" — cloudflare/containers#173); |
| 165 | // the 1-permit mutex serialises the start path so the second entrant sees |
| 166 | // `running === true` and skips. `readyPorts` lets steady-state requests skip |
| 167 | // the readiness probe entirely (like native's `healthy` state). |
| 168 | // |
| 169 | // Key on `DurableObjectState` (stable + unique per DO instance). Requiring it |
| 170 | // is correct: a container only ever runs inside a Durable Object, so this is |
| 171 | // an honest dependency rather than a silent fallback. |
| 172 | const doState = yield* DurableObjectState; |
| 173 | const coordinationKey: object = |
| 174 | (doState.container as object | undefined) ?? doState; |
| 175 | const { startMutex, readyPorts } = getStartCoordination(coordinationKey); |
| 176 | |
| 177 | const launchMonitor = Effect.forkDetach( |
| 178 | container.monitor().pipe( |
| 179 | Effect.flatMap(() => Effect.logInfo("Container monitor exited")), |
| 180 | Effect.catchTag("ContainerError", (error) => |
| 181 | Effect.logError(`Container monitor error: ${error.message}`), |
| 182 | ), |
| 183 | // When the container exits (stopped, crashed, or slept), its ports are no |
| 184 | // longer listening — drop the confirmed-ready marks so the next request |
| 185 | // re-probes the freshly (re)started instance instead of fetching a dead |
| 186 | // port. This is the analog of native's monitor clearing the `healthy` |
| 187 | // state on exit. |
| 188 | Effect.ensuring(Effect.sync(() => readyPorts.clear())), |
| 189 | ), |
| 190 | ); |
| 191 | |
| 192 | // The actual (re)start, serialised through the mutex so racing callers never |
| 193 | // both invoke `container.start()` ("already running" — see |
| 194 | // cloudflare/containers#173). Re-checks `running` under the lock. |
| 195 | const startOnce = Semaphore.withPermits( |
| 196 | startMutex, |
| 197 | 1, |
| 198 | )( |
| 199 | Effect.gen(function* () { |
| 200 | if (yield* container.running) return; |
| 201 | yield* Effect.logDebug("Container not running, starting..."); |
| 202 | // A (re)start means nothing is listening yet — clear any stale ready |
| 203 | // marks from a previous run so the next request re-probes this instance. |
| 204 | yield* Effect.sync(() => readyPorts.clear()); |
| 205 | yield* container.start(options).pipe( |
| 206 | Effect.catchDefect((defect) => Effect.fail(classifyDefect(defect, -1))), |
| 207 | // A rate limit is transient — back off well clear of the per-second |
| 208 | // window and try again a few times before surfacing it. |
| 209 | Effect.retry({ |
| 210 | while: ( |
| 211 | e: |
| 212 | | ContainerError |
| 213 | | NoContainerInstanceError |
| 214 | | ContainerRateLimitedError, |
| 215 | ) => e._tag === "ContainerRateLimitedError", |
| 216 | schedule: Schedule.spaced(RATE_LIMIT_BACKOFF), |
| 217 | times: RATE_LIMIT_RETRIES, |
| 218 | }), |
| 219 | ); |
| 220 | yield* launchMonitor; |
| 221 | }), |
| 222 | ); |
| 223 | |
| 224 | // Ensure the container is running. Fast path: when it's already running (the |
| 225 | // steady state) this is a single cheap `running` read and avoids contending |
| 226 | // on the start mutex. Only an actual (re)start acquires the lock — which is |
| 227 | // also what makes auto-restart work: a stopped/crashed container has |
| 228 | // `running === false`, so the next call transparently restarts it. |
| 229 | const ensureRunning = Effect.gen(function* () { |
| 230 | if (yield* container.running) return; |
| 231 | yield* startOnce; |
| 232 | }); |
| 233 | |
| 234 | // A single readiness probe: any response at all (even a non-2xx) proves the |
| 235 | // port is accepting connections. Each probe is hard-capped so a hung connect |
| 236 | // to a not-yet-bound port cannot stall the request. If the instance has gone |
| 237 | // away (`running === false`) after we started it, surface a crash rather than |
| 238 | // polling out the whole budget (native's `if (!this.container.running)`). |
| 239 | const probePort = (portNumber: number) => |
| 240 | container.getTcpPort(portNumber).pipe( |
| 241 | Effect.andThen((port: Fetcher) => |
| 242 | port.fetch( |
| 243 | HttpClientRequest.get("http://containerstarthealthcheck") as any, |
| 244 | ), |
| 245 | ), |
| 246 | Effect.timeout(READINESS_PROBE_TIMEOUT), |
| 247 | Effect.catchDefect((defect: unknown) => |
| 248 | container.running.pipe( |
| 249 | Effect.andThen((running) => |
| 250 | Effect.fail<ReadinessError>( |
| 251 | !running && |
| 252 | !isRateLimitedError(defect) && |
| 253 | !isNoInstanceError(defect) |
| 254 | ? new ContainerCrashedError({ |
| 255 | message: `Container exited while waiting for port ${portNumber}`, |
| 256 | cause: defect, |
| 257 | }) |
| 258 | : classifyDefect(defect, portNumber), |
| 259 | ), |
| 260 | ), |
| 261 | ), |
| 262 | ), |
| 263 | // A probe that hangs past the per-attempt cap, or fails the HTTP fetch |
| 264 | // before the port is up, is just "not ready yet" — retry it like any |
| 265 | // other transient readiness failure. |
| 266 | Effect.catchTag(["TimeoutError", "HttpClientError"], (cause: unknown) => |
| 267 | container.running.pipe( |
| 268 | Effect.andThen((running) => |
| 269 | Effect.fail<ReadinessError>( |
| 270 | !running |
| 271 | ? new ContainerCrashedError({ |
| 272 | message: `Container exited while waiting for port ${portNumber}`, |
| 273 | cause, |
| 274 | }) |
| 275 | : new ContainerError({ |
| 276 | message: `Container probe not ready on port ${portNumber}`, |
| 277 | cause, |
| 278 | }), |
| 279 | ), |
| 280 | ), |
| 281 | ), |
| 282 | ), |
| 283 | ); |
| 284 | |
| 285 | // Retry ONLY failures that can clear by waiting on the SAME instance: |
| 286 | // - `ContainerError` — the port isn't listening yet (the app is |
| 287 | // still booting; the container process is up) |
| 288 | // - `NoContainerInstanceError` — an instance is still being allocated |
| 289 | // Fail fast on the rest: |
| 290 | // - `ContainerCrashedError` — the container PROCESS has exited |
| 291 | // (`running === false`). `running` reflects |
| 292 | // the process, not the app, so this is never |
| 293 | // "still starting" — it's fatal for this |
| 294 | // instance. Spinning the budget on a |
| 295 | // crash-looping container just wastes ~20-60s |
| 296 | // per request; surface it in ~one poll. |
| 297 | // - `ContainerRateLimitedError`— back off in `ensureRunning`, don't spin. |
| 298 | // This matches `@cloudflare/containers`, whose `waitForPort` / |
| 299 | // `doStartContainer` throw immediately on `!running`. |
| 300 | const isTransientReadiness = (e: ReadinessError) => |
| 301 | e._tag === "ContainerError" || e._tag === "NoContainerInstanceError"; |
| 302 | |
| 303 | // Phase 2 (native `waitForPort`): poll the port until it accepts connections. |
| 304 | // Crash detection is poll-based (`probePort` surfaces `ContainerCrashedError` |
| 305 | // on an observed `!running`, which `isTransientReadiness` does NOT retry). |
| 306 | // |
| 307 | // NB: we deliberately do NOT race `container.monitor()` for faster crash |
| 308 | // detection. The monitor wrapper settles immediately for a not-yet-allocated |
| 309 | // instance (and loses the exit code that native uses to tell "queued" from |
| 310 | // "exited"), so racing it mis-reports healthy, still-provisioning containers |
| 311 | // as crashed under concurrent startup — it dropped a 100-instance cold start |
| 312 | // to <10% success. Polling is correct under load; the price is that crash |
| 313 | // detection is bounded by the probe cadence rather than instant. |
| 314 | const waitForPort = (portNumber: number) => |
| 315 | readyPorts.has(portNumber) |
| 316 | ? Effect.void |
| 317 | : probePort(portNumber).pipe( |
| 318 | Effect.tapError((err) => |
| 319 | Effect.logDebug(`Container not ready (will retry): ${err}`), |
| 320 | ), |
| 321 | Effect.retry({ |
| 322 | while: isTransientReadiness, |
| 323 | schedule: Schedule.spaced(READINESS_POLL_INTERVAL), |
| 324 | times: PORT_READY_RETRIES, |
| 325 | }), |
| 326 | Effect.andThen(() => |
| 327 | Effect.sync(() => { |
| 328 | readyPorts.add(portNumber); |
| 329 | }), |
| 330 | ), |
| 331 | ); |
| 332 | |
| 333 | // Phase 1 (native `doStartContainer`) + phase 2: ensure the instance is |
| 334 | // started (bounded, rate-limit aware), then wait for the port. `ensureRunning` |
| 335 | // is ALWAYS run (cheap when already running) rather than short-circuiting on |
| 336 | // `readyPorts` — otherwise a container that has since stopped/crashed would |
| 337 | // never be restarted. The port probe (`waitForPort`) is what `readyPorts` |
| 338 | // skips in steady state. |
| 339 | const ensureReady = (portNumber: number) => |
| 340 | ensureRunning.pipe( |
| 341 | Effect.retry({ |
| 342 | while: ( |
| 343 | e: |
| 344 | | ContainerError |
| 345 | | NoContainerInstanceError |
| 346 | | ContainerRateLimitedError, |
| 347 | ) => e._tag === "NoContainerInstanceError", |
| 348 | schedule: Schedule.spaced(READINESS_POLL_INTERVAL), |
| 349 | times: GET_CONTAINER_RETRIES, |
| 350 | }), |
| 351 | Effect.andThen(() => waitForPort(portNumber)), |
| 352 | ); |
| 353 | |
| 354 | const getTcpPort = (portNumber: number) => |
| 355 | Effect.succeed({ |
| 356 | fetch: (( |
| 357 | request: |
| 358 | | HttpClientRequest.HttpClientRequest |
| 359 | | HttpServerRequest.HttpServerRequest, |
| 360 | ) => |
| 361 | // Block until the port is ready (bounded), THEN forward the real |
| 362 | // request exactly once. Readiness is a cheap healthcheck, so the user |
| 363 | // request is never replayed against a cold container. |
| 364 | ensureReady(portNumber).pipe( |
| 365 | Effect.andThen(() => container.getTcpPort(portNumber)), |
| 366 | Effect.andThen((port: Fetcher) => port.fetch(request as any)), |
| 367 | Effect.catchDefect((defect: unknown) => |
| 368 | Effect.fail( |
| 369 | new ContainerError({ |
| 370 | message: `Container fetch failed on port ${portNumber}: ${defect}`, |
| 371 | cause: defect, |
| 372 | }), |
| 373 | ), |
| 374 | ), |
| 375 | // Only retry a generic transient blip: a not-yet-ready port |
| 376 | // (`ContainerError`) or a transport hiccup on the real request such |
| 377 | // as "Network connection lost" (`HttpClientError`). A crash, rate |
| 378 | // limit, or exhausted no-instance budget is surfaced immediately so a |
| 379 | // crash-looping container fails fast instead of re-starting 3× more. |
| 380 | Effect.retry({ |
| 381 | while: (e) => |
| 382 | e._tag === "ContainerError" || e._tag === "HttpClientError", |
| 383 | schedule: Schedule.spaced(READINESS_POLL_INTERVAL), |
| 384 | times: REQUEST_RETRIES, |
| 385 | }), |
| 386 | )) as { |
| 387 | ( |
| 388 | request: HttpClientRequest.HttpClientRequest, |
| 389 | ): Effect.Effect<HttpClientResponse.HttpClientResponse>; |
| 390 | ( |
| 391 | request: HttpServerRequest.HttpServerRequest, |
| 392 | ): Effect.Effect<HttpServerResponse.HttpServerResponse>; |
| 393 | }, |
| 394 | }); |
| 395 | |
| 396 | const phase = yield* ALCHEMY_PHASE; |
| 397 | |
| 398 | // eagerly start the container when in runtime, no-op during planning |
| 399 | if (phase === "runtime") { |
| 400 | // erase the RuntimeContext color (we are applying it eagerly as an optimization only during runtime) |
| 401 | yield* ensureRunning as Effect.Effect<void>; |
| 402 | } |
| 403 | |
| 404 | // The container exposes its non-`fetch` shape methods (declared on the |
| 405 | // Container class) over the plain-`fetch` RPC protocol served by the |
| 406 | // container runtime (`serveRpc`). Wrap the instance so that any method that |
| 407 | // isn't one of the built-in handle methods (`running`/`start`/`getTcpPort`/…) |
| 408 | // is dispatched as `POST http://container/__rpc__/{name}` over the |
| 409 | // container's port-3000 server. |
| 410 | const base = { |
| 411 | ...container, |
| 412 | getTcpPort, |
| 413 | fetch: getTcpPort(3000), |
| 414 | }; |
| 415 | return makeFetchRpcStub<Container.Instance<InstanceType<Image>>>({ |
| 416 | fetch: (request) => |
| 417 | getTcpPort(3000).pipe(Effect.flatMap((port) => port.fetch(request))), |
| 418 | baseUrl: "http://container", |
| 419 | base: base as Record<string, unknown>, |
| 420 | }); |
| 421 | }); |