diff --git a/packages/stack/README.md b/packages/stack/README.md index 50dd10f464..6f412c8944 100644 --- a/packages/stack/README.md +++ b/packages/stack/README.md @@ -198,7 +198,7 @@ service and its projected status. ### Readiness ```typescript -await stack.ready(); // Inherit the stack's finite two-minute default +await stack.ready(); // Inherit the stack's finite three-minute default await stack.ready({ mode: "finite", timeoutMs: 30_000 }); await stack.ready({ mode: "infinite" }); // Explicit debugging override await stack.serviceReady("postgres"); @@ -209,7 +209,10 @@ In eager mode, `start()` blocks until every enabled service is ready. In lazy mo for direct listeners and services activated so far. Unrequested lazy services report `Dormant`. Calling `serviceReady()` for a dormant lazy service fails immediately; activate it through the proxy or call `startService()` first. Foreground -and detached stacks use the same readiness rules. +and detached stacks use the same readiness rules. The configured policy also applies to service +start, restart, activation, and reload operations; a call-specific option overrides it. A finite +deadline fails with `STACK_READINESS_TIMEOUT` and disposes the local runtime, so the handle cannot +be used to relaunch processes afterward. ### Status @@ -308,15 +311,16 @@ try { } ``` -| Code | Description | -| ------------------- | -------------------------------------------- | -| `SERVICE_NOT_FOUND` | Referenced a service that doesn't exist | -| `SERVICE_NOT_READY` | Service failed to become healthy | -| `BUILD_ERROR` | Failed to build the service dependency graph | -| `BINARY_NOT_FOUND` | No binary available for the current platform | -| `DOWNLOAD_ERROR` | Binary download failed | -| `PORT_CONFLICT` | Requested port is already in use | -| `PORT_ALLOCATION` | Failed to allocate a free port | +| Code | Description | +| ------------------------- | -------------------------------------------- | +| `SERVICE_NOT_FOUND` | Referenced a service that doesn't exist | +| `SERVICE_NOT_READY` | Service failed to become healthy | +| `STACK_READINESS_TIMEOUT` | Stack readiness exceeded its finite deadline | +| `BUILD_ERROR` | Failed to build the service dependency graph | +| `BINARY_NOT_FOUND` | No binary available for the current platform | +| `DOWNLOAD_ERROR` | Binary download failed | +| `PORT_CONFLICT` | Requested port is already in use | +| `PORT_ALLOCATION` | Failed to allocate a free port | ## Examples diff --git a/packages/stack/docs/architecture.md b/packages/stack/docs/architecture.md index 336964d1d2..a6ba09c9d2 100644 --- a/packages/stack/docs/architecture.md +++ b/packages/stack/docs/architecture.md @@ -48,11 +48,13 @@ functions options, and per-service configuration. `false` disables an optional s 4. applies per-service defaults and current `DEFAULT_VERSIONS`; 5. records auto-managed paths for scoped cleanup. -Readiness policy is part of the resolved configuration. The package default is a finite two-minute +Readiness policy is part of the resolved configuration. The package default is a finite three-minute deadline; callers can choose a different finite deadline or explicit infinite waiting. Per-call `ReadyOptions` take precedence over the stack policy, while `inherit` delegates to the stack -policy. The Promise `ready()` / `serviceReady()` Adapter applies that resolver today. The Effect -lifecycle Interface remains unbounded until deadline enforcement is centralized there. +policy. The local Implementation applies this resolver to startup, service activation, restart, +reload, and explicit readiness waits. A finite deadline fails with `StackReadinessError` and runs +the same scoped cleanup used by disposal. Promise and remote Adapters pass `ReadyOptions` through +to that Implementation instead of layering a second timeout rule around it. The current zero-config stack enables PostgreSQL, PostgREST, Auth, and Edge Runtime. Realtime, Storage, imgproxy, Mailpit, Postgres Meta, Studio, Analytics, Vector, and Supavisor are enabled only @@ -119,7 +121,7 @@ Vector without Analytics, and Studio without Postgres Meta. ## Lifecycle ownership -The current local Implementation is `StackLifecycleCoordinator`. It owns one scoped lifecycle: +The local Implementation is `LocalStack`. Its scoped layer owns one lifecycle: - preparation and its single-flight deferred; - graph construction and the process-compose runtime; @@ -129,19 +131,21 @@ The current local Implementation is `StackLifecycleCoordinator`. It owns one sco - exact cleanup targets and metadata persistence; - disposal of processes, Docker resources, ports, and auto-managed paths. -`Stack.ts` currently provides a thin public Effect Interface over that coordinator. `ApiProxy` -uses the narrower `StackServiceActivator` Interface so an incoming request can activate a lazy -backend without gaining unrelated lifecycle operations. +`Stack.ts` contains only the public Effect Interface and transport schemas. `LocalStack` constructs +the state once and publishes both `Stack` and the narrower `StackServiceActivator` Interface from +the same scoped layer. `ApiProxy` therefore activates a lazy backend without gaining unrelated +lifecycle operations or requiring a second pass-through lifecycle tag. -Before the orchestrator exists, the coordinator publishes synthetic `Pending` and `Downloading` +Before the orchestrator exists, `LocalStack` publishes synthetic `Pending` and `Downloading` states. After construction, it subscribes to raw process-compose state and publishes only public projected states. `StackServiceState` adds `Downloading`, `Initializing`, and `Dormant` to the raw process statuses. `start()` prepares artifacts, creates the runtime once, starts the appropriate services, and waits for their generic process-compose readiness. `stop()` preserves explicit per-service stop intent; -`dispose()` additionally closes the scoped runtime and executes cleanup. Generic process-compose -readiness remains intentionally policy-free and unbounded. +`dispose()` additionally closes the scoped runtime and executes cleanup. Stack readiness policy is +enforced around generic process-compose waits, which remain intentionally policy-free and +unbounded. Structural `Equal.equals` comparison suppresses duplicate projected state emissions. ## Eager and lazy activation @@ -237,8 +241,9 @@ Detached mode adds: - `StateManager`: atomically persists and discovers durable metadata and live state. The management transport includes health, status, status stream, start/stop, readiness, -per-service lifecycle, logs/history, and Edge Runtime reload routes. It is local Unix-socket -transport, not the public Supabase API proxy. +per-service lifecycle, logs/history, and Edge Runtime reload routes. Readiness waits use validated +`ReadyOptions` JSON bodies and preserve `StackReadinessError` across the transport. It is local +Unix-socket transport, not the public Supabase API proxy. See [detach mode](./detach-mode.md) for paths, process startup, and compiled executable dispatch. diff --git a/packages/stack/docs/detach-mode.md b/packages/stack/docs/detach-mode.md index 812c8e861a..9fae1ee7b7 100644 --- a/packages/stack/docs/detach-mode.md +++ b/packages/stack/docs/detach-mode.md @@ -114,14 +114,16 @@ metadata and service data remain. `DaemonServer` exposes the local `Stack` Interface on the Unix socket. Current routes include: - `/health`, `/status`, and `/status/stream`; -- `/start`, `/stop`, and `/ready`; +- `/start`, `/stop`, and `POST /ready`; - per-service start, stop, restart, and readiness; - merged and per-service live logs plus buffered history; - functions and Edge Runtime reload. -State and log streams use SSE. Ordinary responses and typed failures use validated JSON shapes. -`RemoteStack` decodes that transport back into the same Effect `Stack` Interface used in -foreground mode, including `ServiceNotFoundError`, `ServiceReadyError`, and `StackBuildError`. +State and log streams use SSE. Readiness routes use `POST` with a validated readiness-policy body; +omitting an override sends the explicit `inherit` representation. Ordinary responses and typed +failures use validated JSON shapes. `RemoteStack` decodes that transport back into the same Effect +`Stack` Interface used in foreground mode, including `ServiceNotFoundError`, `ServiceReadyError`, +`StackBuildError`, and `StackReadinessError`. The management socket is not the public local API endpoint. `ApiProxy` still owns the configured HTTP API port inside the daemon process. @@ -141,6 +143,14 @@ package. In particular: On normal `/stop`, the daemon gracefully stops the stack, signals HTTP shutdown after the response has had time to flush, disposes both managed runtimes, and removes live state/runtime paths. +A readiness deadline is terminal for that local runtime: the stack disposes its scoped resources, +the daemon returns the typed timeout response, and then the daemon shuts down. This prevents later +requests from relaunching processes after cleanup has already run. The boundary is deliberately +fail-closed across the whole daemon, rather than isolated to the service that timed out: once +processes and port leases are being released, the management and proxy servers cannot safely keep +advertising a usable runtime. This terminal path does not drain unrelated in-flight requests to +otherwise healthy services; callers must reconnect after starting a fresh daemon. + If the daemon has died, CLI stop/status detects a stale PID. Stop can use cleanup targets persisted in `stack.json` to force-remove known Docker containers before removing the stale state. This crash-recovery metadata is deliberately separate from user-facing `/status` connection data. diff --git a/packages/stack/src/ApiProxy.ts b/packages/stack/src/ApiProxy.ts index 632e467b09..9a3f98041c 100644 --- a/packages/stack/src/ApiProxy.ts +++ b/packages/stack/src/ApiProxy.ts @@ -1,4 +1,4 @@ -import { Effect, Layer, Option, Context, Duration, Schedule, Result } from "effect"; +import { Deferred, Effect, Layer, Option, Context, Schedule, Result } from "effect"; import { Headers, HttpBody, @@ -9,12 +9,11 @@ import { HttpServerRequest, HttpServerResponse, } from "effect/unstable/http"; -import { activationTimeoutSecondsForService, StackServiceActivator } from "./ServiceActivation.ts"; -import type { ServiceName } from "./versions.ts"; +import { StackServiceActivator } from "./ServiceActivation.ts"; +import type { ServiceName } from "./ServiceName.ts"; export interface ProxyConfig { readonly listenPort: number; - readonly activationTimeout?: Duration.Input; readonly gotruePort: number; readonly postgrestPort: number; readonly postgrestAdminPort: number; @@ -132,19 +131,17 @@ function makeProxyHandler( client: HttpClient.HttpClient, config: ProxyConfig, activator: StackServiceActivator["Service"], + signalTerminalFailure: Effect.Effect, opts: ProxyHandlerOptions, ) { return (req: HttpServerRequest.HttpServerRequest) => Effect.gen(function* () { - const activation = yield* activator - .activate(opts.service) - .pipe( - Effect.timeout( - config.activationTimeout ?? - Duration.seconds(activationTimeoutSecondsForService(opts.service)), - ), - Effect.result, - ); + const activation = yield* activator.activate(opts.service).pipe( + Effect.tapError((error) => + error._tag === "StackReadinessError" ? signalTerminalFailure : Effect.void, + ), + Effect.result, + ); if (Result.isFailure(activation)) { return HttpServerResponse.text("Service unavailable", { status: 503, @@ -225,6 +222,8 @@ export class ApiProxy extends Context.Service< ApiProxy, { readonly address: HttpServer.Address; + /** Completes when terminal lazy-activation failure requires daemon teardown. */ + readonly awaitTerminalFailure: Effect.Effect; } >()("local/ApiProxy") { static layer = ( @@ -239,13 +238,15 @@ export class ApiProxy extends Context.Service< const server = yield* HttpServer.HttpServer; const client = yield* HttpClient.HttpClient; const activator = yield* StackServiceActivator; + const terminalFailure = yield* Deferred.make(); + const signalTerminalFailure = Deferred.succeed(terminalFailure, void 0).pipe(Effect.asVoid); const routes = [ HttpRouter.route("*", "/health", HttpServerResponse.text("OK", { status: 200 })), HttpRouter.route( "*", "/.well-known/oauth-authorization-server", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "auth", backendPort: config.gotruePort, backendPath: "/.well-known/oauth-authorization-server", @@ -254,7 +255,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/auth/v1/verify", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "auth", backendPort: config.gotruePort, stripPrefix: "/auth/v1", @@ -263,7 +264,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/auth/v1/callback", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "auth", backendPort: config.gotruePort, stripPrefix: "/auth/v1", @@ -272,7 +273,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/auth/v1/authorize", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "auth", backendPort: config.gotruePort, stripPrefix: "/auth/v1", @@ -281,7 +282,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/auth/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "auth", backendPort: config.gotruePort, stripPrefix: "/auth/v1", @@ -291,7 +292,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/rest/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "postgrest", backendPort: config.postgrestPort, stripPrefix: "/rest/v1", @@ -301,7 +302,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/rest-admin/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "postgrest", backendPort: config.postgrestAdminPort, stripPrefix: "/rest-admin/v1", @@ -310,7 +311,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/graphql/v1", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "postgrest", backendPort: config.postgrestPort, backendPath: "/rpc/graphql", @@ -321,7 +322,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/functions/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "edge-runtime", backendPort: config.edgeRuntimePort, stripPrefix: "/functions/v1", @@ -333,7 +334,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/realtime/v1/api/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "realtime", backendPort: config.realtimePort, stripPrefix: "/realtime/v1", @@ -343,7 +344,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/realtime/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "realtime", backendPort: config.realtimePort, stripPrefix: "/realtime/v1", @@ -352,7 +353,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/storage/v1/s3/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "storage", backendPort: config.storagePort, stripPrefix: "/storage/v1", @@ -361,7 +362,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/storage/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "storage", backendPort: config.storagePort, stripPrefix: "/storage/v1", @@ -371,7 +372,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/pg/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "pgmeta", backendPort: config.pgmetaPort, stripPrefix: "/pg", @@ -380,7 +381,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/analytics/v1/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "analytics", backendPort: config.analyticsPort, stripPrefix: "/analytics/v1", @@ -389,7 +390,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/pooler/v2/*", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "pooler", backendPort: config.poolerPort, stripPrefix: "/pooler", @@ -398,7 +399,7 @@ export class ApiProxy extends Context.Service< HttpRouter.route( "*", "/mcp", - makeProxyHandler(client, config, activator, { + makeProxyHandler(client, config, activator, signalTerminalFailure, { service: "studio", backendPort: config.studioPort, backendPath: "/api/mcp", @@ -423,6 +424,7 @@ export class ApiProxy extends Context.Service< return { address: server.address, + awaitTerminalFailure: Deferred.await(terminalFailure), }; }), ); diff --git a/packages/stack/src/ApiProxy.unit.test.ts b/packages/stack/src/ApiProxy.unit.test.ts index 6e97bd0a6c..2d13427352 100644 --- a/packages/stack/src/ApiProxy.unit.test.ts +++ b/packages/stack/src/ApiProxy.unit.test.ts @@ -5,7 +5,7 @@ import { Effect, Layer, ManagedRuntime } from "effect"; import { FetchHttpClient } from "effect/unstable/http"; import { afterAll, beforeAll, describe, expect, test } from "vitest"; import { ApiProxy, type ProxyConfig } from "./ApiProxy.ts"; -import { StackNotRunningError } from "./errors.ts"; +import { StackNotRunningError, StackReadinessError } from "./errors.ts"; import { StackServiceActivator } from "./ServiceActivation.ts"; import type { ServiceName } from "./versions.ts"; @@ -120,7 +120,11 @@ function buildProxyLayer( async function startProxy( config: ProxyConfig, activatorLayer?: Layer.Layer, -): Promise<{ url: string; dispose: () => Promise }> { +): Promise<{ + url: string; + dispose: () => Promise; + awaitTerminalFailure: () => Promise; +}> { const proxyRuntime = ManagedRuntime.make(buildProxyLayer(config, activatorLayer)); const proxy = await proxyRuntime.runPromise(ApiProxy); const addr = proxy.address; @@ -129,7 +133,11 @@ async function startProxy( const host = addr.hostname === "0.0.0.0" ? "127.0.0.1" : addr.hostname; url = `http://${host}:${addr.port}`; } - return { url, dispose: () => proxyRuntime.dispose() }; + return { + url, + dispose: () => proxyRuntime.dispose(), + awaitTerminalFailure: () => proxyRuntime.runPromise(proxy.awaitTerminalFailure), + }; } describe("ApiProxy", () => { @@ -246,20 +254,22 @@ describe("ApiProxy", () => { } }); - test("returns 503 when service activation does not complete before the request deadline", async () => { + test("signals daemon teardown after a terminal activation failure", async () => { const activatorLayer = Layer.succeed(StackServiceActivator, { - activate: () => Effect.never, + activate: () => + Effect.fail( + new StackReadinessError({ + target: "postgrest", + timeoutMs: 30_000, + detail: "PostgREST did not become ready", + }), + ), }); - const proxy = await startProxy( - { ...configForPort(echoServer.port), activationTimeout: "10 millis" }, - activatorLayer, - ); + const proxy = await startProxy(configForPort(echoServer.port), activatorLayer); try { - const res = await fetch(`${proxy.url}/rest/v1/users`, { - signal: AbortSignal.timeout(1_000), - }); + const res = await fetch(`${proxy.url}/rest/v1/users`); expect(res.status).toBe(503); - expect(res.headers.get("retry-after")).toBe("1"); + await expect(proxy.awaitTerminalFailure()).resolves.toBeUndefined(); } finally { await proxy.dispose(); } diff --git a/packages/stack/src/DaemonProtocol.ts b/packages/stack/src/DaemonProtocol.ts index 7bfdf9fb94..88262482cb 100644 --- a/packages/stack/src/DaemonProtocol.ts +++ b/packages/stack/src/DaemonProtocol.ts @@ -4,6 +4,7 @@ import { StackStateSchema } from "./StateManager.ts"; const DaemonErrorCodeSchema = Schema.Literals([ "SERVICE_NOT_FOUND", "SERVICE_NOT_READY", + "STACK_READINESS_TIMEOUT", "STACK_BUILD_ERROR", ]); @@ -12,6 +13,7 @@ export const DaemonErrorResponseSchema = Schema.Struct({ error: Schema.String, service: Schema.optionalKey(Schema.String), exitCode: Schema.optionalKey(Schema.Number), + timeoutMs: Schema.optionalKey(Schema.Number), }); export type DaemonErrorResponse = typeof DaemonErrorResponseSchema.Type; diff --git a/packages/stack/src/DaemonServer.integration.test.ts b/packages/stack/src/DaemonServer.integration.test.ts index e16ec4ac7f..1cb2227f68 100644 --- a/packages/stack/src/DaemonServer.integration.test.ts +++ b/packages/stack/src/DaemonServer.integration.test.ts @@ -4,6 +4,7 @@ import { Effect, Layer, ManagedRuntime, Stream } from "effect"; import * as http from "node:http"; import { afterAll, beforeAll, describe, expect, test } from "vitest"; import { DaemonServer } from "./DaemonServer.ts"; +import { StackReadinessError } from "./errors.ts"; import { Stack, type StackInfo } from "./Stack.ts"; import { StackServiceState } from "./StackServiceState.ts"; @@ -53,13 +54,22 @@ const MOCK_LOGS: ReadonlyArray = [ // Mock Stack // --------------------------------------------------------------------------- -function mockStack() { +function mockStack(options: { readonly startTimeoutMs?: number } = {}) { let stopped = false; const serviceCalls: string[] = []; const layer = Layer.succeed(Stack, { getInfo: () => Effect.succeed(MOCK_INFO), - start: () => Effect.void, + start: () => + options.startTimeoutMs === undefined + ? Effect.void + : Effect.fail( + new StackReadinessError({ + target: "stack", + timeoutMs: options.startTimeoutMs, + detail: `Timed out waiting for stack readiness after ${options.startTimeoutMs}ms`, + }), + ), stop: () => Effect.sync(() => { stopped = true; @@ -321,6 +331,29 @@ describe("DaemonServer", () => { expect(mock.serviceCalls).toContain("restart:postgres"); }); + test("POST readiness routes validate the shared override representation", async () => { + const stackReady = await fetch(`${url}/ready`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ mode: "inherit" }), + }); + expect(stackReady.status).toBe(200); + + const serviceReady = await fetch(`${url}/services/postgres/ready`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ mode: "finite", timeoutMs: 100 }), + }); + expect(serviceReady.status).toBe(200); + + const malformed = await fetch(`${url}/ready`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ mode: "finite", timeoutMs: 0 }), + }); + expect(malformed.status).not.toBe(200); + }); + test("POST /edge-runtime/reload returns 200", async () => { const res = await fetch(`${url}/edge-runtime/reload`, { method: "POST", @@ -358,6 +391,26 @@ describe("DaemonServer", () => { expect(body.error).toContain("unknown"); }); + test("a startup readiness timeout returns the typed failure and shuts down the daemon", async () => { + const freshRuntime = ManagedRuntime.make(buildDaemonLayer(mockStack({ startTimeoutMs: 75 }))); + try { + const daemon = await freshRuntime.runPromise(DaemonServer); + const shutdownPromise = freshRuntime.runPromise(daemon.awaitShutdown); + const response = await fetch(`${getUrl(daemon.address)}/start`, { method: "POST" }); + + expect(response.status).toBe(500); + expect(await response.json()).toEqual({ + code: "STACK_READINESS_TIMEOUT", + error: "Timed out waiting for stack readiness after 75ms", + service: "stack", + timeoutMs: 75, + }); + await shutdownPromise; + } finally { + await freshRuntime.dispose(); + } + }); + // ------------------------------------------------------------------------- // Stop (tested last since it modifies daemon state) // ------------------------------------------------------------------------- diff --git a/packages/stack/src/DaemonServer.ts b/packages/stack/src/DaemonServer.ts index 638029c885..4900fcba8c 100644 --- a/packages/stack/src/DaemonServer.ts +++ b/packages/stack/src/DaemonServer.ts @@ -9,6 +9,7 @@ import { import * as Sse from "effect/unstable/encoding/Sse"; import type { DaemonErrorResponse } from "./DaemonProtocol.ts"; import { EdgeRuntimeReloadConfigSchema, Stack } from "./Stack.ts"; +import { ReadyOptionsSchema } from "./StackConfig.ts"; // --------------------------------------------------------------------------- // Service @@ -50,6 +51,29 @@ export class DaemonServer extends Context.Service< ); const buildErrorResponse = (detail: string) => errorResponse({ code: "STACK_BUILD_ERROR", error: detail }, 500); + const readinessTimeoutResponse = (target: string, timeoutMs: number, detail: string) => + errorResponse( + { + code: "STACK_READINESS_TIMEOUT", + error: detail, + service: target, + timeoutMs, + }, + 500, + ); + const beginShutdown = beforeShutdown.pipe( + Effect.ensuring( + // The HTTP module has no response-flushed hook. Delay the process + // shutdown signal long enough for the final JSON response to leave + // the socket. + Deferred.succeed(shutdownDeferred, void 0).pipe( + Effect.delay("25 millis"), + Effect.forkDetach, + ), + ), + ); + const terminalReadinessResponse = (target: string, timeoutMs: number, detail: string) => + beginShutdown.pipe(Effect.as(readinessTimeoutResponse(target, timeoutMs, detail))); // Helper: wrap an Effect Stream as a text/event-stream response const sseResponse = ( @@ -113,20 +137,29 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), HttpRouter.route( - "GET", + "POST", "/ready", - stack.waitAllReady().pipe( - Effect.as(HttpServerResponse.jsonUnsafe({ ok: true })), + Effect.gen(function* () { + const opts = yield* HttpServerRequest.schemaBodyJson(ReadyOptionsSchema); + yield* stack.waitAllReady(opts); + return HttpServerResponse.jsonUnsafe({ ok: true }); + }).pipe( Effect.catchTag("ServiceReadyError", (e) => Effect.succeed(notReadyResponse(e.name, e.reason, e.exitCode)), ), Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), @@ -136,18 +169,7 @@ export class DaemonServer extends Context.Service< "/stop", Effect.gen(function* () { yield* stack.stop(); - yield* beforeShutdown.pipe( - Effect.ensuring( - // The HTTP module has no response-flushed hook. Delay the - // process shutdown signal long enough for this small JSON - // response to leave the socket; stopDaemon also tolerates a - // dropped response and confirms termination by polling PID. - Deferred.succeed(shutdownDeferred, void 0).pipe( - Effect.delay("25 millis"), - Effect.forkDetach, - ), - ), - ); + yield* beginShutdown; return HttpServerResponse.jsonUnsafe({ ok: true }); }), ), @@ -219,15 +241,19 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), HttpRouter.route( - "GET", + "POST", "/services/:name/ready", Effect.gen(function* () { const routeParams = yield* HttpRouter.params; - yield* stack.waitReady(routeParams.name!); + const opts = yield* HttpServerRequest.schemaBodyJson(ReadyOptionsSchema); + yield* stack.waitReady(routeParams.name!, opts); return HttpServerResponse.jsonUnsafe({ ok: true }); }).pipe( Effect.catchTag("ServiceNotFoundError", (e) => @@ -239,6 +265,9 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), @@ -276,6 +305,9 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), @@ -299,6 +331,9 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), @@ -319,6 +354,9 @@ export class DaemonServer extends Context.Service< Effect.catchTag("StackBuildError", (e) => Effect.succeed(buildErrorResponse(e.detail)), ), + Effect.catchTag("StackReadinessError", (e) => + terminalReadinessResponse(e.target, e.timeoutMs, e.detail), + ), ), ), ]; diff --git a/packages/stack/src/LocalStack.ts b/packages/stack/src/LocalStack.ts new file mode 100644 index 0000000000..363a7ac48a --- /dev/null +++ b/packages/stack/src/LocalStack.ts @@ -0,0 +1,894 @@ +import { LogBuffer, Orchestrator } from "@supabase/process-compose"; +import { ServiceNotFoundError } from "@supabase/process-compose"; +import type { ResolvedGraph, ServiceReadyError } from "@supabase/process-compose"; +import { + Context, + Deferred, + Duration, + Effect, + Equal, + FileSystem, + Layer, + Path, + Ref, + Semaphore, + Stream, + SubscriptionRef, +} from "effect"; +import { ChildProcessSpawner } from "effect/unstable/process"; +import type { CleanupTargets } from "./CleanupTargets.ts"; +import { cleanupLocalStackResources } from "./cleanup.ts"; +import { StackBuildError, StackNotRunningError, StackReadinessError } from "./errors.ts"; +import { configureFunctionsRuntime, type FunctionsConfig } from "./functions.ts"; +import { detectPlatform, dockerHostAddress } from "./Platform.ts"; +import type { PortLease } from "./PortAllocator.ts"; +import { + activationTargetsForService, + eagerServices, + lifecycleTargetsForService, + StackServiceActivator, +} from "./ServiceActivation.ts"; +import { portFieldsForService } from "./ServicePorts.ts"; +import { StackMetadataPersistence } from "./StackMetadataPersistence.ts"; +import { StackPreparation } from "./StackPreparation.ts"; +import type { PreparedStackArtifacts } from "./StackPreparation.ts"; +import { + enabledServicesForConfig, + StackBuilder, + validateResolvedConfig, + versionsForConfig, +} from "./StackBuilder.ts"; +import { resolveReadinessPolicy } from "./StackConfig.ts"; +import type { ReadinessPolicy, ReadyOptions, ResolvedStackConfig } from "./StackConfig.ts"; +import { projectStackStates, type StackServiceProjectionCatalog } from "./StackStateProjection.ts"; +import { StackServiceState } from "./StackServiceState.ts"; +import { Stack } from "./Stack.ts"; +import type { EdgeRuntimeReloadConfig, StackInfo } from "./Stack.ts"; +import { SERVICE_NAMES, type ServiceName } from "./versions.ts"; + +type LifecyclePhase = + | "idle" + | "preparing" + | "prepared" + | "starting" + | "running" + | "stopping" + | "stopped" + | "disposed"; + +type StackService = typeof Stack.Service; + +interface RuntimeState { + readonly orchestrator: Orchestrator["Service"]; + readonly graph: ResolvedGraph; + readonly serviceProjection: StackServiceProjectionCatalog; + readonly cleanupTargets: CleanupTargets; +} + +const initialPublicStates = (config: ResolvedStackConfig): ReadonlyArray => + enabledServicesForConfig(config).map( + (name) => + new StackServiceState({ + name, + status: "Pending", + pid: null, + exitCode: null, + restartCount: 0, + startedAt: null, + error: null, + }), + ); + +const stackInfoFor = (config: ResolvedStackConfig): StackInfo => { + const apiUrl = `http://127.0.0.1:${config.apiPort}`; + return { + url: apiUrl, + dbUrl: `postgresql://postgres:postgres@127.0.0.1:${config.dbPort}/postgres`, + publishableKey: config.publishableKey, + secretKey: config.secretKey, + anonJwt: config.anonJwt, + serviceRoleJwt: config.serviceRoleJwt, + serviceEndpoints: { + ...(config.auth === false ? {} : { auth: `${apiUrl}/auth/v1` }), + ...(config.postgrest === false ? {} : { postgrest: `${apiUrl}/rest/v1` }), + ...(config.edgeRuntime === false + ? {} + : { + functions: `${apiUrl}/functions/v1`, + edge_runtime: `${apiUrl}/functions/v1`, + }), + ...(config.realtime === false ? {} : { realtime: `${apiUrl}/realtime/v1` }), + ...(config.storage === false + ? {} + : { + storage: `${apiUrl}/storage/v1`, + storage_s3: `${apiUrl}/storage/v1/s3`, + }), + ...(config.imgproxy === false || config.startupMode === "lazy" + ? {} + : { imgproxy: `http://127.0.0.1:${config.imgproxy.port}` }), + ...(config.mailpit === false + ? {} + : { + mailpit: `http://127.0.0.1:${config.mailpit.port}`, + mailpit_smtp: `smtp://127.0.0.1:${config.mailpit.smtpPort}`, + mailpit_pop3: `pop3://127.0.0.1:${config.mailpit.pop3Port}`, + }), + ...(config.pgmeta === false ? {} : { pgmeta: `${apiUrl}/pg` }), + ...(config.studio === false ? {} : { studio: `http://127.0.0.1:${config.studio.port}` }), + ...(config.analytics === false ? {} : { analytics: `${apiUrl}/analytics/v1` }), + ...(config.pooler === false + ? {} + : { + pooler: `postgresql://postgres:postgres@127.0.0.1:${config.pooler.port}/postgres`, + pooler_admin: `http://127.0.0.1:${config.pooler.apiPort}`, + }), + }, + }; +}; + +const changedStatesBetween = ( + previous: ReadonlyArray | undefined, + current: ReadonlyArray, +): ReadonlyArray => { + if (previous === undefined) { + return current; + } + + const previousByName = new Map(previous.map((state) => [state.name, state] as const)); + return current.filter((state) => !Equal.equals(previousByName.get(state.name), state)); +}; + +/** + * The private in-process Stack implementation. Its scoped construction owns + * lifecycle state once and publishes both public seams from that same state. + */ +export const localStackLayer = ( + config: ResolvedStackConfig, + portLease: PortLease, +): Layer.Layer< + Stack | StackServiceActivator, + StackBuildError, + | StackBuilder + | StackPreparation + | ChildProcessSpawner.ChildProcessSpawner + | StackMetadataPersistence + | FileSystem.FileSystem + | Path.Path +> => + Layer.effectContext( + Effect.gen(function* () { + const builder = yield* StackBuilder; + const preparation = yield* StackPreparation; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const metadataPersistence = yield* StackMetadataPersistence; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const scope = yield* Effect.scope; + + const info = stackInfoFor(config); + const enabledServices = enabledServicesForConfig(config); + const stateRef = yield* SubscriptionRef.make(initialPublicStates(config)); + const phaseRef = yield* Ref.make("idle"); + const lifecycleLock = Semaphore.makeUnsafe(1); + const projectionLock = Semaphore.makeUnsafe(1); + + const logBufferServices = yield* Layer.buildWithScope(LogBuffer.layer, scope); + const logBuffer = Context.get(logBufferServices, LogBuffer); + + const updateState = (nextState: StackServiceState) => + SubscriptionRef.update(stateRef, (current) => { + const previous = current.find((entry) => entry.name === nextState.name); + if (Equal.equals(previous, nextState)) { + return current; + } + return current.some((entry) => entry.name === nextState.name) + ? current.map((entry) => (entry.name === nextState.name ? nextState : entry)) + : [...current, nextState]; + }); + + const syncProjectedStates = ( + orchestrator: Orchestrator["Service"], + serviceProjection: StackServiceProjectionCatalog, + ) => + Effect.gen(function* () { + const rawStates = yield* orchestrator.getAllStates(); + yield* Effect.forEach(projectStackStates(rawStates, serviceProjection), updateState, { + discard: true, + }); + }).pipe(projectionLock.withPermit); + + const requireKnownService = (name: string) => + Effect.gen(function* () { + const currentStates = SubscriptionRef.getUnsafe(stateRef); + const match = currentStates.find((state) => state.name === name); + if (match === undefined) { + return yield* Effect.fail(new ServiceNotFoundError({ name })); + } + return match; + }); + const requireKnownServiceName = ( + name: string, + ): Effect.Effect => + Effect.gen(function* () { + yield* requireKnownService(name); + const service = SERVICE_NAMES.find((candidate) => candidate === name); + if (service === undefined) { + return yield* Effect.fail(new ServiceNotFoundError({ name })); + } + return service; + }); + + let preparedArtifacts: PreparedStackArtifacts | undefined; + let prepareDeferred: Deferred.Deferred | undefined; + let runtimeState: RuntimeState | undefined; + let runtimeDeferred: Deferred.Deferred | undefined; + + const ensurePrepared = Effect.suspend(() => { + if (preparedArtifacts !== undefined) { + return Effect.succeed(preparedArtifacts); + } + if (prepareDeferred !== undefined) { + return Deferred.await(prepareDeferred); + } + + const deferred = Deferred.makeUnsafe(); + prepareDeferred = deferred; + + const effect = Effect.gen(function* () { + yield* validateResolvedConfig(config); + yield* Ref.set(phaseRef, "preparing"); + + let prepared: PreparedStackArtifacts | undefined; + yield* preparation + .prepareEvents({ + mode: config.mode, + services: enabledServicesForConfig(config), + versions: versionsForConfig(config), + }) + .pipe( + Stream.mapError( + (cause) => + new StackBuildError({ + detail: "Failed to prepare stack assets", + cause, + }), + ), + ) + .pipe( + Stream.runForEach((event) => { + switch (event._tag) { + case "ServiceDownloadStarted": + return updateState( + new StackServiceState({ + name: event.service, + status: "Downloading", + pid: null, + exitCode: null, + restartCount: 0, + startedAt: null, + error: null, + }), + ); + case "ServiceDownloadFinished": + return updateState( + new StackServiceState({ + name: event.service, + status: "Pending", + pid: null, + exitCode: null, + restartCount: 0, + startedAt: null, + error: null, + }), + ); + case "PreparationCompleted": + return Effect.sync(() => { + prepared = event.artifacts; + }); + } + }), + ); + + if (prepared === undefined) { + return yield* Effect.fail( + new StackBuildError({ + detail: "Stack preparation completed without prepared artifacts", + }), + ); + } + + yield* Ref.set(phaseRef, "prepared"); + return prepared; + }).pipe( + Effect.tap((value) => + Effect.sync(() => { + preparedArtifacts = value; + }), + ), + Effect.onError(() => Ref.set(phaseRef, "idle")), + Effect.ensuring( + Effect.sync(() => { + prepareDeferred = undefined; + }), + ), + ); + + return Effect.gen(function* () { + yield* Effect.forkIn(effect.pipe(Deferred.into(deferred)), scope); + return yield* Deferred.await(deferred); + }); + }); + + const ensureRuntime = Effect.suspend(() => { + if (runtimeState !== undefined) { + return Effect.succeed(runtimeState); + } + if (runtimeDeferred !== undefined) { + return Deferred.await(runtimeDeferred); + } + + const deferred = Deferred.makeUnsafe(); + runtimeDeferred = deferred; + + const effect = Effect.gen(function* () { + const prepared = yield* ensurePrepared; + const { graph, serviceProjection, cleanupTargets } = yield* builder.build( + config, + prepared, + ); + + yield* metadataPersistence.persistCleanupTargets(cleanupTargets).pipe( + Effect.mapError( + (cause) => + new StackBuildError({ + detail: "Failed to persist stack cleanup metadata", + cause, + }), + ), + ); + + const orchLayer = Orchestrator.layer(graph).pipe( + Layer.provide(Layer.succeed(LogBuffer, logBuffer)), + Layer.provide(Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner)), + ); + const orchServices = yield* Layer.buildWithScope(orchLayer, scope); + const orchestrator = Context.get(orchServices, Orchestrator); + + yield* syncProjectedStates(orchestrator, serviceProjection); + yield* orchestrator.allStateChanges().pipe( + Stream.runForEach(() => syncProjectedStates(orchestrator, serviceProjection)), + Effect.ignore, + Effect.forkIn(scope), + ); + + return { + orchestrator, + graph, + serviceProjection, + cleanupTargets, + } satisfies RuntimeState; + }).pipe( + Effect.tap((value) => + Effect.sync(() => { + runtimeState = value; + }), + ), + Effect.ensuring( + Effect.sync(() => { + runtimeDeferred = undefined; + }), + ), + ); + + return Effect.gen(function* () { + yield* Effect.forkIn(effect.pipe(Deferred.into(deferred)), scope); + return yield* Deferred.await(deferred); + }); + }); + + let disposed = false; + let disposing = false; + const runtimeHost = Effect.gen(function* () { + const prepared = yield* ensurePrepared; + const platform = yield* detectPlatform; + const edgeRuntimeResolution = prepared.resolutions["edge-runtime"]; + return { + hostname: + edgeRuntimeResolution?.type === "docker" ? dockerHostAddress(platform.os) : "127.0.0.1", + }; + }); + const providePlatform = ( + effect: Effect.Effect, + ): Effect.Effect => + effect.pipe( + Effect.provideService(FileSystem.FileSystem, fs), + Effect.provideService(Path.Path, path), + ); + const configureFunctions = ( + nextConfig: ResolvedStackConfig, + ): Effect.Effect => + Effect.gen(function* () { + yield* providePlatform(configureFunctionsRuntime(nextConfig, yield* runtimeHost)); + }).pipe( + Effect.mapError( + (cause) => + new StackBuildError({ + detail: "Failed to configure Edge Functions", + cause, + }), + ), + ); + const configWithFunctionOptions = (opts?: FunctionsConfig): ResolvedStackConfig => { + if (opts === undefined) { + return config; + } + const base = config.functions === false ? { noVerifyJwt: false } : config.functions; + return { + ...config, + functions: { + envFile: opts.envFile ?? base.envFile, + noVerifyJwt: opts.noVerifyJwt ?? base.noVerifyJwt, + }, + }; + }; + const configWithEdgeRuntimeOptions = ( + opts: EdgeRuntimeReloadConfig, + ): Effect.Effect => + Effect.gen(function* () { + if (config.edgeRuntime === false || opts.edgeRuntime.enabled === false) { + return yield* Effect.fail(new ServiceNotFoundError({ name: "edge-runtime" })); + } + + const base = configWithFunctionOptions(opts.functions); + return { + ...base, + edgeRuntime: { + ...config.edgeRuntime, + enabled: opts.edgeRuntime.enabled ?? config.edgeRuntime.enabled, + inspectorPort: opts.edgeRuntime.inspectorPort ?? config.edgeRuntime.inspectorPort, + policy: opts.edgeRuntime.policy ?? config.edgeRuntime.policy, + env: opts.edgeRuntime.env ?? config.edgeRuntime.env, + }, + }; + }); + const publicAllStateChanges = () => + SubscriptionRef.changes(stateRef).pipe( + Stream.mapAccum< + ReadonlyArray | undefined, + ReadonlyArray, + StackServiceState + >( + () => undefined, + (previous, current) => [current, changedStatesBetween(previous, current)], + ), + ); + const withLifecycleLock = lifecycleLock.withPermit; + const syncRuntimeProjectedStates = (runtime: RuntimeState) => + syncProjectedStates(runtime.orchestrator, runtime.serviceProjection); + const serviceStartOptions = { + beforeStart: (name: string) => portLease.reserve(portFieldsForService(name)), + beforeSpawn: (name: string) => portLease.release(portFieldsForService(name)), + }; + const knownServiceError = (service: string, cause: ServiceNotFoundError) => + new StackBuildError({ + detail: `Prepared graph does not contain enabled service ${service}`, + cause, + }); + const beginStartTargets = ( + root: ServiceName, + allowExplicitlyStopped: ReadonlySet, + ) => + Effect.gen(function* () { + const runtime = yield* ensureRuntime; + const targets = activationTargetsForService(enabledServices, root); + const targetClosure = new Set( + targets.flatMap((target) => + runtime.graph.startOrderFor(target).map((definition) => definition.name), + ), + ); + + for (const dependency of targetClosure) { + const state = yield* runtime.orchestrator + .getState(dependency) + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError(dependency, cause)), + ), + ); + const publicDependency = SERVICE_NAMES.find((candidate) => candidate === dependency); + if ( + state.desired === "stopped" && + publicDependency !== undefined && + !allowExplicitlyStopped.has(publicDependency) + ) { + return yield* Effect.fail( + new StackBuildError({ + detail: `Cannot activate ${root} because dependency ${dependency} was explicitly stopped`, + }), + ); + } + } + + for (const target of targets) { + yield* runtime.orchestrator + .startService(target, serviceStartOptions) + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError(target, cause)), + ), + ); + } + return { runtime, targets }; + }); + const waitForTargets = ({ + runtime, + targets, + }: { + readonly runtime: RuntimeState; + readonly targets: ReadonlyArray; + }) => + Effect.gen(function* () { + yield* Effect.forEach( + targets, + (target) => + runtime.orchestrator + .waitReady(target) + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError(target, cause)), + ), + ), + { concurrency: "unbounded", discard: true }, + ); + yield* syncRuntimeProjectedStates(runtime); + }); + const inspectStartedTargets = (root: ServiceName) => + Effect.gen(function* () { + const runtime = yield* ensureRuntime; + const targets = activationTargetsForService(enabledServices, root); + const states = yield* Effect.forEach(targets, (target) => + runtime.orchestrator + .getState(target) + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError(target, cause)), + ), + ), + ); + if (states.some((state) => state.desired !== "running")) { + return undefined; + } + return { + runtime, + targets, + ready: states.every( + (state) => + state.status === "Healthy" || (state.status === "Stopped" && state.exitCode === 0), + ), + }; + }); + const requireRunningPhase = Effect.gen(function* () { + const phase = yield* Ref.get(phaseRef); + if (phase !== "running") { + return yield* Effect.fail(new StackNotRunningError({ phase })); + } + }); + const requireMutable = (operation: string) => + Effect.suspend(() => + disposed || disposing + ? Effect.fail( + new StackBuildError({ + detail: `Cannot ${operation} after stack disposal has begun`, + }), + ) + : Effect.void, + ); + const disposeOnce = () => + Effect.suspend(() => { + disposing = true; + return Effect.gen(function* () { + if (disposed) { + return; + } + disposed = true; + yield* Ref.set(phaseRef, "stopping"); + yield* cleanupLocalStackResources({ + stop: () => + runtimeState === undefined ? Effect.void : runtimeState.orchestrator.stop(), + cleanupTargets: runtimeState?.cleanupTargets ?? { dockerContainerNames: [] }, + config, + }).pipe( + Effect.ensuring(portLease.releaseAll), + Effect.ensuring(Ref.set(phaseRef, "disposed")), + ); + }).pipe(withLifecycleLock); + }).pipe(Effect.uninterruptible); + + const withReadinessPolicy = ( + effect: Effect.Effect, + target: string, + readyOptions?: ReadyOptions, + ): Effect.Effect => { + const policy: ReadinessPolicy = resolveReadinessPolicy({ + readyOptions, + stackPolicy: config.readiness, + }); + if (policy.mode === "infinite") { + return effect; + } + return effect.pipe( + Effect.timeoutOrElse({ + duration: Duration.millis(policy.timeoutMs), + orElse: () => + Effect.fail( + new StackReadinessError({ + target, + timeoutMs: policy.timeoutMs, + detail: `Timed out waiting for ${target} readiness after ${policy.timeoutMs}ms`, + }), + ), + }), + ); + }; + const cleanupOnReadinessFailure = ( + effect: Effect.Effect, + ): Effect.Effect => + effect.pipe( + Effect.catchTag("StackReadinessError", (error) => + disposeOnce().pipe(Effect.andThen(Effect.fail(error))), + ), + ); + + yield* Effect.addFinalizer(disposeOnce); + + const activateService = (name: ServiceName) => + Effect.gen(function* () { + yield* requireRunningPhase; + const service = yield* requireKnownServiceName(name); + const existing = yield* inspectStartedTargets(service); + if (existing?.ready === true) { + // Close the race with a concurrent stack stop before taking + // the lock-free healthy-request fast path. + yield* requireRunningPhase; + return; + } + if (existing !== undefined) { + yield* waitForTargets(existing).pipe((effect) => withReadinessPolicy(effect, name)); + return; + } + const started = yield* Effect.gen(function* () { + yield* requireRunningPhase; + const concurrentlyStarted = yield* inspectStartedTargets(service); + if (concurrentlyStarted !== undefined) return concurrentlyStarted; + return yield* beginStartTargets(service, new Set()); + }).pipe(withLifecycleLock); + yield* waitForTargets(started).pipe((effect) => withReadinessPolicy(effect, name)); + }).pipe(cleanupOnReadinessFailure); + + const stack = { + getInfo: () => Effect.succeed(info), + start: () => + Effect.gen(function* () { + yield* requireMutable("start"); + yield* Ref.set(phaseRef, "starting"); + const runtime = yield* ensureRuntime; + yield* configureFunctions(config); + + if (config.startupMode === "lazy") { + const readiness: Array> = []; + if ( + runtime.graph.startOrder.some((definition) => definition.name === "postgres-init") + ) { + yield* runtime.orchestrator + .startService("postgres-init", serviceStartOptions) + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError("postgres-init", cause)), + ), + ); + readiness.push( + runtime.orchestrator + .waitReady("postgres-init") + .pipe( + Effect.catchTag("ServiceNotFoundError", (cause) => + Effect.fail(knownServiceError("postgres-init", cause)), + ), + ), + ); + } + for (const service of eagerServices(enabledServices)) { + const started = yield* beginStartTargets( + service, + new Set(lifecycleTargetsForService(enabledServices, service)), + ); + readiness.push(waitForTargets(started)); + } + yield* Effect.all(readiness, { concurrency: "unbounded", discard: true }).pipe( + (effect) => withReadinessPolicy(effect, "stack"), + ); + } else { + yield* runtime.orchestrator.start(serviceStartOptions); + yield* runtime.orchestrator + .waitAllReady() + .pipe((effect) => withReadinessPolicy(effect, "stack")); + yield* syncRuntimeProjectedStates(runtime); + } + yield* Ref.set(phaseRef, "running"); + }).pipe( + Effect.onError(() => Ref.set(phaseRef, "stopped")), + withLifecycleLock, + cleanupOnReadinessFailure, + ), + stop: () => + Effect.gen(function* () { + if (disposed) { + return; + } + if (runtimeState === undefined) { + yield* Ref.set(phaseRef, "stopped"); + return; + } + yield* Ref.set(phaseRef, "stopping"); + yield* runtimeState.orchestrator.stop(); + yield* Ref.set(phaseRef, "stopped"); + }).pipe(withLifecycleLock), + dispose: disposeOnce, + startService: (name) => + Effect.gen(function* () { + const started = yield* Effect.gen(function* () { + yield* requireMutable(`start service ${name}`); + const service = yield* requireKnownServiceName(name); + return yield* beginStartTargets( + service, + new Set(lifecycleTargetsForService(enabledServices, service)), + ); + }).pipe(withLifecycleLock); + yield* waitForTargets(started).pipe((effect) => withReadinessPolicy(effect, name)); + }).pipe(cleanupOnReadinessFailure), + stopService: (name) => + Effect.gen(function* () { + yield* requireMutable(`stop service ${name}`); + const service = yield* requireKnownServiceName(name); + const runtime = yield* ensureRuntime; + for (const target of lifecycleTargetsForService( + enabledServices, + service, + ).toReversed()) { + yield* runtime.orchestrator.stopService(target); + } + }).pipe(withLifecycleLock), + restartService: (name) => + Effect.gen(function* () { + const started = yield* Effect.gen(function* () { + yield* requireMutable(`restart service ${name}`); + const service = yield* requireKnownServiceName(name); + const runtime = yield* ensureRuntime; + yield* runtime.orchestrator.restartService(service, serviceStartOptions); + return { runtime, targets: [service] }; + }).pipe(withLifecycleLock); + yield* waitForTargets(started).pipe((effect) => withReadinessPolicy(effect, name)); + }).pipe(cleanupOnReadinessFailure), + reloadFunctions: (opts) => + Effect.gen(function* () { + const started = yield* Effect.gen(function* () { + yield* requireMutable("reload functions"); + yield* requireKnownService("edge-runtime"); + yield* configureFunctions(configWithFunctionOptions(opts)); + const runtime = yield* ensureRuntime; + const state = yield* runtime.orchestrator.getState("edge-runtime"); + if (state.desired !== "running") { + return yield* beginStartTargets("edge-runtime", new Set(["edge-runtime"])); + } + yield* runtime.orchestrator.restartService("edge-runtime", serviceStartOptions); + return { runtime, targets: ["edge-runtime"] as const }; + }).pipe(withLifecycleLock); + yield* waitForTargets(started).pipe((effect) => + withReadinessPolicy(effect, "edge-runtime"), + ); + }).pipe(cleanupOnReadinessFailure), + reloadEdgeRuntime: (opts) => + Effect.gen(function* () { + const started = yield* Effect.gen(function* () { + yield* requireMutable("reload Edge Runtime"); + yield* requireKnownService("edge-runtime"); + const nextConfig = yield* configWithEdgeRuntimeOptions(opts); + const prepared = yield* ensurePrepared; + const runtime = yield* ensureRuntime; + const buildResult = yield* builder.build(nextConfig, prepared); + const edgeRuntimeDef = buildResult.graph.startOrder.find( + (def) => def.name === "edge-runtime", + ); + + if (edgeRuntimeDef === undefined) { + return yield* Effect.fail(new ServiceNotFoundError({ name: "edge-runtime" })); + } + + yield* configureFunctions(nextConfig); + yield* runtime.orchestrator + .updateServiceDefinition("edge-runtime", edgeRuntimeDef) + .pipe( + Effect.mapError( + (cause) => + new StackBuildError({ + detail: "Failed to update edge-runtime service definition", + cause, + }), + ), + ); + const state = yield* runtime.orchestrator.getState("edge-runtime"); + if (state.desired !== "running") { + return yield* beginStartTargets("edge-runtime", new Set(["edge-runtime"])); + } + yield* runtime.orchestrator.restartService("edge-runtime", serviceStartOptions); + return { runtime, targets: ["edge-runtime"] as const }; + }).pipe(withLifecycleLock); + yield* waitForTargets(started).pipe((effect) => + withReadinessPolicy(effect, "edge-runtime"), + ); + }).pipe(cleanupOnReadinessFailure), + getState: (name) => + Effect.gen(function* () { + const currentStates = SubscriptionRef.getUnsafe(stateRef); + const match = currentStates.find((state) => state.name === name); + if (match === undefined) { + return yield* Effect.fail(new ServiceNotFoundError({ name })); + } + return match; + }), + getAllStates: () => Effect.sync(() => SubscriptionRef.getUnsafe(stateRef)), + stateChanges: (name) => + Effect.gen(function* () { + yield* requireKnownService(name); + return Stream.filter(publicAllStateChanges(), (state) => state.name === name); + }), + allStateChanges: publicAllStateChanges, + waitReady: (name, opts) => + Effect.gen(function* () { + const phase = yield* Ref.get(phaseRef); + if (phase !== "running") { + return yield* Effect.fail( + new StackBuildError({ + detail: `Cannot wait for service ${name} while the stack is ${phase}`, + }), + ); + } + yield* requireKnownServiceName(name); + const runtime = yield* ensureRuntime; + yield* runtime.orchestrator + .waitReady(name) + .pipe((effect) => withReadinessPolicy(effect, name, opts)); + yield* syncRuntimeProjectedStates(runtime); + }).pipe(cleanupOnReadinessFailure), + waitAllReady: (opts) => + Effect.gen(function* () { + const phase = yield* Ref.get(phaseRef); + if (phase !== "running") { + return yield* Effect.fail( + new StackBuildError({ + detail: `Cannot wait for stack readiness while the stack is ${phase}`, + }), + ); + } + const runtime = yield* ensureRuntime; + yield* runtime.orchestrator + .waitAllReady() + .pipe((effect) => withReadinessPolicy(effect, "stack", opts)); + yield* syncRuntimeProjectedStates(runtime); + }).pipe(cleanupOnReadinessFailure), + subscribeLogs: (name) => logBuffer.subscribe(name), + subscribeAllLogs: (services) => + services === undefined || services.length === 0 + ? logBuffer.subscribeAll() + : logBuffer + .subscribeAll() + .pipe(Stream.filter((entry) => services.includes(entry.service))), + logHistory: (name, limit) => logBuffer.history(name, limit), + logHistoryAll: (limit, services) => logBuffer.historyAll(limit, services), + } satisfies StackService; + + return Context.make(Stack, stack).pipe( + Context.add(StackServiceActivator, { activate: activateService }), + ); + }), + ); diff --git a/packages/stack/src/RemoteStack.integration.test.ts b/packages/stack/src/RemoteStack.integration.test.ts index dc520bab1e..43930a1a2a 100644 --- a/packages/stack/src/RemoteStack.integration.test.ts +++ b/packages/stack/src/RemoteStack.integration.test.ts @@ -4,9 +4,10 @@ import { Effect, Fiber, Layer, ManagedRuntime, Stream } from "effect"; import * as http from "node:http"; import { afterAll, beforeAll, describe, expect, test } from "vitest"; import { DaemonServer } from "./DaemonServer.ts"; -import { StackBuildError } from "./errors.ts"; +import { StackBuildError, StackReadinessError } from "./errors.ts"; import { RemoteStack } from "./RemoteStack.ts"; import { Stack, type StackInfo } from "./Stack.ts"; +import type { ReadyOptions } from "./StackConfig.ts"; import { StackServiceState } from "./StackServiceState.ts"; import { UnixHttpClient, UnixHttpClientError } from "./UnixHttpClient.ts"; @@ -75,11 +76,13 @@ function mockStack( readonly startServiceBuildError?: string; readonly startServiceReadyError?: string; readonly waitReadyBuildError?: string; + readonly waitReadyTimeoutMs?: number; readonly restartServiceReadyError?: string; } = {}, ) { let stopped = false; const serviceCalls: string[] = []; + const readinessCalls: Array<{ readonly target: string; readonly options?: ReadyOptions }> = []; const layer = Layer.succeed(Stack, { getInfo: () => Effect.succeed(MOCK_INFO), @@ -146,18 +149,29 @@ function mockStack( : Effect.fail(new ServiceNotFoundError({ name })); }, allStateChanges: () => Stream.fromIterable(MOCK_STATES), - waitReady: (name: string) => { + waitReady: (name: string, readyOptions?: ReadyOptions) => { const match = MOCK_STATES.find((s) => s.name === name); if (match === undefined) return Effect.fail(new ServiceNotFoundError({ name })); if (options.waitReadyBuildError !== undefined) { return Effect.fail(new StackBuildError({ detail: options.waitReadyBuildError })); } + if (options.waitReadyTimeoutMs !== undefined) { + return Effect.fail( + new StackReadinessError({ + target: name, + timeoutMs: options.waitReadyTimeoutMs, + detail: `Timed out waiting for ${name}`, + }), + ); + } return Effect.sync(() => { + readinessCalls.push({ target: name, options: readyOptions }); serviceCalls.push(`ready:${name}`); }); }, - waitAllReady: () => + waitAllReady: (readyOptions?: ReadyOptions) => Effect.sync(() => { + readinessCalls.push({ target: "stack", options: readyOptions }); serviceCalls.push("ready:all"); }), subscribeLogs: (name: string) => @@ -185,6 +199,7 @@ function mockStack( return stopped; }, serviceCalls, + readinessCalls, }; } @@ -288,9 +303,15 @@ describe("RemoteStack integration", () => { expect(exit._tag).toBe("Failure"); }); - test("waitReady delegates to the daemon coordinator", async () => { - await clientRuntime.runPromise(Effect.flatMap(Stack, (stack) => stack.waitReady("auth"))); + test("waitReady passes one validated finite override through the daemon", async () => { + await clientRuntime.runPromise( + Effect.flatMap(Stack, (stack) => stack.waitReady("auth", { mode: "finite", timeoutMs: 250 })), + ); expect(mock.serviceCalls).toContain("ready:auth"); + expect(mock.readinessCalls).toContainEqual({ + target: "auth", + options: { mode: "finite", timeoutMs: 250 }, + }); }); test("waitReady rejects dot path segments locally", async () => { @@ -301,9 +322,38 @@ describe("RemoteStack integration", () => { expect(mock.serviceCalls).not.toContain("ready:all"); }); - test("waitAllReady delegates to the daemon coordinator", async () => { + test("waitAllReady sends explicit inherit semantics to the daemon", async () => { await clientRuntime.runPromise(Effect.flatMap(Stack, (stack) => stack.waitAllReady())); expect(mock.serviceCalls).toContain("ready:all"); + expect(mock.readinessCalls).toContainEqual({ + target: "stack", + options: { mode: "inherit" }, + }); + }); + + test("preserves StackReadinessError across the daemon transport", async () => { + const failingMock = mockStack({ waitReadyTimeoutMs: 75 }); + const failingServer = ManagedRuntime.make(buildServerLayer(failingMock)); + let failingClient: ManagedRuntime.ManagedRuntime | undefined; + try { + const daemon = await failingServer.runPromise(DaemonServer); + const addr = daemon.address; + if (addr._tag !== "TcpAddress") throw new Error("Expected TcpAddress"); + const host = addr.hostname === "0.0.0.0" ? "127.0.0.1" : addr.hostname; + failingClient = ManagedRuntime.make(buildClientLayer(`http://${host}:${addr.port}`)); + + const error = await failingClient.runPromise( + Effect.flatMap(Stack, (stack) => stack.waitReady("auth")).pipe(Effect.flip), + ); + expect(error._tag).toBe("StackReadinessError"); + if (error._tag === "StackReadinessError") { + expect(error.target).toBe("auth"); + expect(error.timeoutMs).toBe(75); + } + } finally { + await failingClient?.dispose(); + await failingServer.dispose(); + } }); test("interrupting waitReady aborts the daemon request", async () => { diff --git a/packages/stack/src/RemoteStack.ts b/packages/stack/src/RemoteStack.ts index df7388559f..dd0f8ca6a4 100644 --- a/packages/stack/src/RemoteStack.ts +++ b/packages/stack/src/RemoteStack.ts @@ -3,8 +3,9 @@ import { Effect, Layer, Schema, Stream } from "effect"; import * as Sse from "effect/unstable/encoding/Sse"; import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"; import { DaemonErrorResponseSchema } from "./DaemonProtocol.ts"; -import { StackBuildError } from "./errors.ts"; +import { StackBuildError, StackReadinessError } from "./errors.ts"; import { Stack, StackInfoSchema } from "./Stack.ts"; +import { inheritReadyOptions } from "./StackConfig.ts"; import { StackServiceState, StackServiceStatusSchema } from "./StackServiceState.ts"; import { UnixHttpClient, UnixHttpClientError } from "./UnixHttpClient.ts"; import { SERVICE_NAMES } from "./versions.ts"; @@ -105,7 +106,10 @@ function withAbortSignal( const failDaemonResponse = ( response: HttpClientResponse.HttpClientResponse, fallbackName: string, -): Effect.Effect => +): Effect.Effect< + never, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError +> => Effect.gen(function* () { const body = yield* HttpClientResponse.schemaBodyJson(DaemonErrorResponseSchema)(response).pipe( Effect.orDie, @@ -121,13 +125,22 @@ const failDaemonResponse = ( }); case "STACK_BUILD_ERROR": return yield* new StackBuildError({ detail: body.error }); + case "STACK_READINESS_TIMEOUT": + return yield* new StackReadinessError({ + target: body.service ?? fallbackName, + timeoutMs: body.timeoutMs ?? 0, + detail: body.error, + }); } }); const expectDaemonOk = ( response: HttpClientResponse.HttpClientResponse, fallbackName: string, -): Effect.Effect => +): Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError +> => response.status >= 200 && response.status < 300 ? Effect.void : failDaemonResponse(response, fallbackName); @@ -301,6 +314,7 @@ export const RemoteStack = { }); yield* expectDaemonOk(response, name).pipe( Effect.catchTag("ServiceReadyError", (error) => Effect.die(error)), + Effect.catchTag("StackReadinessError", (error) => Effect.die(error)), ); }), ), @@ -392,7 +406,7 @@ export const RemoteStack = { }), ), - waitReady: (name: string) => + waitReady: (name, opts) => withUnixHttpClient( withAbortSignal((signal) => Effect.gen(function* () { @@ -400,18 +414,28 @@ export const RemoteStack = { const response = yield* unixResponse( socketPath, `/services/${servicePath}/ready`, - { signal }, + { + method: "POST", + signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify(opts ?? inheritReadyOptions), + }, ); yield* expectDaemonOk(response, name); }), ), ), - waitAllReady: () => + waitAllReady: (opts) => withUnixHttpClient( withAbortSignal((signal) => Effect.gen(function* () { - const response = yield* unixResponse(socketPath, "/ready", { signal }); + const response = yield* unixResponse(socketPath, "/ready", { + method: "POST", + signal, + headers: { "content-type": "application/json" }, + body: JSON.stringify(opts ?? inheritReadyOptions), + }); yield* expectDaemonOk(response, "stack").pipe( Effect.catchTag("ServiceNotFoundError", (error) => Effect.die(error)), ); diff --git a/packages/stack/src/ServiceActivation.ts b/packages/stack/src/ServiceActivation.ts index ae5c05189b..27f055eae2 100644 --- a/packages/stack/src/ServiceActivation.ts +++ b/packages/stack/src/ServiceActivation.ts @@ -1,7 +1,7 @@ import { ServiceNotFoundError } from "@supabase/process-compose"; import type { ServiceReadyError } from "@supabase/process-compose"; import { Context, Effect, Layer } from "effect"; -import { StackBuildError, StackNotRunningError } from "./errors.ts"; +import { StackBuildError, StackNotRunningError, StackReadinessError } from "./errors.ts"; import { stackServiceStartupBudgetSeconds } from "./services/health-budgets.ts"; import { SERVICE_NAMES, serviceMetadata } from "./ServiceCatalog.ts"; import type { ServiceName } from "./ServiceName.ts"; @@ -74,7 +74,11 @@ export class StackServiceActivator extends Context.Service< service: ServiceName, ) => Effect.Effect< void, - ServiceNotFoundError | ServiceReadyError | StackBuildError | StackNotRunningError + | ServiceNotFoundError + | ServiceReadyError + | StackBuildError + | StackNotRunningError + | StackReadinessError >; } >()("stack/StackServiceActivator") { diff --git a/packages/stack/src/Stack.ts b/packages/stack/src/Stack.ts index dbe2ce70e0..6d07355597 100644 --- a/packages/stack/src/Stack.ts +++ b/packages/stack/src/Stack.ts @@ -1,10 +1,9 @@ import { ServiceNotFoundError } from "@supabase/process-compose"; import type { LogEntry, ServiceReadyError } from "@supabase/process-compose"; -import { Effect, Layer, Schema, Context, Stream } from "effect"; -import { StackBuildError } from "./errors.ts"; +import { Context, Effect, Schema, Stream } from "effect"; +import { StackBuildError, StackReadinessError } from "./errors.ts"; import type { FunctionsConfig } from "./functions.ts"; -import { StackLifecycleCoordinator } from "./StackLifecycleCoordinator.ts"; -import type { EdgeRuntimeConfig, ResolvedStackConfig } from "./StackConfig.ts"; +import type { EdgeRuntimeConfig, ReadyOptions } from "./StackConfig.ts"; import { StackServiceState } from "./StackServiceState.ts"; export interface StackInfo { @@ -49,30 +48,43 @@ export interface EdgeRuntimeReloadConfig { readonly functions?: FunctionsConfig; } -type StackService = typeof Stack.Service; - export class Stack extends Context.Service< Stack, { readonly getInfo: () => Effect.Effect; - readonly start: () => Effect.Effect; + readonly start: () => Effect.Effect< + void, + ServiceReadyError | StackBuildError | StackReadinessError + >; readonly stop: () => Effect.Effect; readonly dispose: () => Effect.Effect; readonly startService: ( name: string, - ) => Effect.Effect; + ) => Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError + >; readonly stopService: ( name: string, ) => Effect.Effect; readonly restartService: ( name: string, - ) => Effect.Effect; + ) => Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError + >; readonly reloadFunctions: ( opts?: FunctionsConfig, - ) => Effect.Effect; + ) => Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError + >; readonly reloadEdgeRuntime: ( opts: EdgeRuntimeReloadConfig, - ) => Effect.Effect; + ) => Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError + >; readonly getState: (name: string) => Effect.Effect; readonly getAllStates: () => Effect.Effect>; readonly stateChanges: ( @@ -81,8 +93,14 @@ export class Stack extends Context.Service< readonly allStateChanges: () => Stream.Stream; readonly waitReady: ( name: string, - ) => Effect.Effect; - readonly waitAllReady: () => Effect.Effect; + opts?: ReadyOptions, + ) => Effect.Effect< + void, + ServiceNotFoundError | ServiceReadyError | StackBuildError | StackReadinessError + >; + readonly waitAllReady: ( + opts?: ReadyOptions, + ) => Effect.Effect; readonly subscribeLogs: (name: string) => Stream.Stream; readonly subscribeAllLogs: (services?: ReadonlyArray) => Stream.Stream; readonly logHistory: (name: string, limit?: number) => Effect.Effect>; @@ -91,35 +109,4 @@ export class Stack extends Context.Service< services?: ReadonlyArray, ) => Effect.Effect>; } ->()("stack/Stack") { - static layer = ( - _config: ResolvedStackConfig, - ): Layer.Layer => - Layer.effect( - this, - Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - return { - getInfo: coordinator.getInfo, - start: coordinator.start, - stop: coordinator.stop, - dispose: coordinator.dispose, - startService: coordinator.startService, - stopService: coordinator.stopService, - restartService: coordinator.restartService, - reloadFunctions: coordinator.reloadFunctions, - reloadEdgeRuntime: coordinator.reloadEdgeRuntime, - getState: coordinator.getState, - getAllStates: coordinator.getAllStates, - stateChanges: coordinator.stateChanges, - allStateChanges: coordinator.allStateChanges, - waitReady: coordinator.waitReady, - waitAllReady: coordinator.waitAllReady, - subscribeLogs: coordinator.subscribeLogs, - subscribeAllLogs: coordinator.subscribeAllLogs, - logHistory: coordinator.logHistory, - logHistoryAll: coordinator.logHistoryAll, - } satisfies StackService; - }), - ); -} +>()("stack/Stack") {} diff --git a/packages/stack/src/Stack.unit.test.ts b/packages/stack/src/Stack.unit.test.ts index 189a7f9e11..c52918dedb 100644 --- a/packages/stack/src/Stack.unit.test.ts +++ b/packages/stack/src/Stack.unit.test.ts @@ -6,8 +6,9 @@ import { mockChildProcessSpawner } from "../../process-compose/tests/helpers/moc import { mockBinaryResolver } from "../tests/helpers/mocks.ts"; import { defaultPublishableKey, defaultSecretKey, generateJwt } from "./JwtGenerator.ts"; import type { AllocatedPorts, PortField, PortLease } from "./PortAllocator.ts"; +import { StackServiceActivator } from "./ServiceActivation.ts"; import { Stack } from "./Stack.ts"; -import { StackLifecycleCoordinator } from "./StackLifecycleCoordinator.ts"; +import { localStackLayer } from "./LocalStack.ts"; import { StackMetadataPersistence } from "./StackMetadataPersistence.ts"; import { StackPreparation } from "./StackPreparation.ts"; import { StackBuilder } from "./StackBuilder.ts"; @@ -115,7 +116,7 @@ function setupLayer( ) { const resolver = mockBinaryResolver(); const stackPreparationLayer = StackPreparation.layer.pipe(Layer.provide(resolver.layer)); - const coordinatorLayer = StackLifecycleCoordinator.layer(config, portLease).pipe( + const layer = localStackLayer(config, portLease).pipe( Layer.provide(StackBuilder.layer), Layer.provide(stackPreparationLayer), Layer.provide(StackMetadataPersistence.noop), @@ -123,9 +124,7 @@ function setupLayer( Layer.provide(BunServices.layer), ); - const layer = Stack.layer(config).pipe(Layer.provide(coordinatorLayer)); - - return { coordinatorLayer, layer, resolver, spawner }; + return { layer, resolver, spawner }; } describe("Stack", () => { @@ -306,16 +305,12 @@ describe("Stack", () => { }); const spawner = mockChildProcessSpawner(); const stackPreparationLayer = StackPreparation.layer.pipe(Layer.provide(resolver.layer)); - const coordinatorLayer = StackLifecycleCoordinator.layer( - defaultConfig, - noopPortLease(defaultConfig.ports), - ).pipe( + const layer = localStackLayer(defaultConfig, noopPortLease(defaultConfig.ports)).pipe( Layer.provide(StackBuilder.layer), Layer.provide(stackPreparationLayer), Layer.provide(StackMetadataPersistence.noop), ); - const layer = Stack.layer(defaultConfig).pipe( - Layer.provide(coordinatorLayer), + const providedLayer = layer.pipe( Layer.provide(spawner.layer), Layer.provide(BunServices.layer), ); @@ -334,7 +329,38 @@ describe("Stack", () => { yield* Fiber.interrupt(startFiber); expect(states.map((state) => state.status)).toContain("Downloading"); - }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(providedLayer)); + }); + + it.live("starts the readiness deadline after artifact preparation", () => { + const resolver = mockBinaryResolver({ + downloadedServices: ["postgres"], + downloadDelayMs: 1_000, + }); + const spawner = mockChildProcessSpawner(); + const config = { + ...defaultConfig, + postgrest: false, + auth: false, + readiness: { mode: "finite", timeoutMs: 250 }, + } satisfies ResolvedStackConfig; + const stackPreparationLayer = StackPreparation.layer.pipe(Layer.provide(resolver.layer)); + const layer = localStackLayer(config, noopPortLease(config.ports)).pipe( + Layer.provide(StackBuilder.layer), + Layer.provide(stackPreparationLayer), + Layer.provide(StackMetadataPersistence.noop), + Layer.provide(spawner.layer), + Layer.provide(BunServices.layer), + ); + + return Effect.gen(function* () { + const stack = yield* Stack; + const startedAt = Date.now(); + const exit = yield* stack.start().pipe(Effect.exit); + + expect(Date.now() - startedAt).toBeGreaterThanOrEqual(900); + expect(Exit.isSuccess(exit)).toBe(true); + }).pipe(Effect.provide(layer), Effect.scoped, Effect.timeout("5 seconds")); }); it.effect("getState fails for internal helper services", () => { @@ -391,16 +417,12 @@ describe("Stack", () => { const resolver = mockBinaryResolver({ failServices: ["postgres", "postgrest", "auth"] }); const spawner = mockChildProcessSpawner({ exitCode: 1 }); const stackPreparationLayer = StackPreparation.layer.pipe(Layer.provide(resolver.layer)); - const coordinatorLayer = StackLifecycleCoordinator.layer( - defaultConfig, - noopPortLease(defaultConfig.ports), - ).pipe( + const layer = localStackLayer(defaultConfig, noopPortLease(defaultConfig.ports)).pipe( Layer.provide(StackBuilder.layer), Layer.provide(stackPreparationLayer), Layer.provide(StackMetadataPersistence.noop), ); - const layer = Stack.layer(defaultConfig).pipe( - Layer.provide(coordinatorLayer), + const providedLayer = layer.pipe( Layer.provide(spawner.layer), Layer.provide(BunServices.layer), ); @@ -413,7 +435,7 @@ describe("Stack", () => { // No container was ever started: only prepare-phase docker commands ran. const startedContainers = spawner.spawned.filter((record) => record.args[0] === "run"); expect(startedContainers).toEqual([]); - }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(providedLayer)); }); it.live("lazy startup starts direct services without starting HTTP backends", () => { @@ -455,21 +477,22 @@ describe("Stack", () => { version: DEFAULT_VERSIONS.imgproxy, }, }; - const { coordinatorLayer } = setupLayer(config); + const { layer } = setupLayer(config); return Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - yield* coordinator.stopService("imgproxy"); + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + yield* stack.stopService("imgproxy"); - const error = yield* coordinator.activateService("storage").pipe(Effect.flip); + const error = yield* activator.activate("storage").pipe(Effect.flip); expect(error._tag).toBe("StackBuildError"); if (error._tag === "StackBuildError") { expect(error.detail).toContain("imgproxy was explicitly stopped"); } - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer), Effect.timeout("5 seconds")); + yield* stack.stop(); + }).pipe(Effect.provide(layer), Effect.timeout("5 seconds")); }); it.live("lazy readiness includes an activation that is still starting", () => @@ -484,34 +507,35 @@ describe("Stack", () => { : Effect.void, }); const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config, noopPortLease(config.ports), spawner); + const { layer } = setupLayer(config, noopPortLease(config.ports), spawner); yield* Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - expect((yield* coordinator.getState("auth")).status).toBe("Dormant"); - const activeStateFiber = yield* coordinator.allStateChanges().pipe( + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + expect((yield* stack.getState("auth")).status).toBe("Dormant"); + const activeStateFiber = yield* stack.allStateChanges().pipe( Stream.filter((state) => state.name === "auth" && state.status !== "Dormant"), Stream.runHead, Effect.forkChild({ startImmediately: true }), ); - const activationFiber = yield* coordinator - .activateService("auth") + const activationFiber = yield* activator + .activate("auth") .pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.await(spawnStarted); expect((yield* Fiber.join(activeStateFiber))._tag).toBe("Some"); - expect((yield* coordinator.getState("auth")).status).not.toBe("Dormant"); + expect((yield* stack.getState("auth")).status).not.toBe("Dormant"); - const readyFiber = yield* coordinator + const readyFiber = yield* stack .waitAllReady() .pipe(Effect.forkChild({ startImmediately: true })); yield* Effect.yieldNow; expect(readyFiber.pollUnsafe()).toBeUndefined(); - yield* coordinator.stop().pipe(Effect.timeout("1 second")); + yield* stack.stop().pipe(Effect.timeout("1 second")); yield* Fiber.interrupt(readyFiber); yield* Fiber.interrupt(activationFiber); - }).pipe(Effect.provide(coordinatorLayer)); + }).pipe(Effect.provide(layer)); }).pipe(Effect.scoped, Effect.timeout("5 seconds")), ); @@ -529,25 +553,26 @@ describe("Stack", () => { : Effect.void, }); const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config, noopPortLease(config.ports), spawner); + const { layer } = setupLayer(config, noopPortLease(config.ports), spawner); yield* Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - const manualStart = yield* coordinator + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + const manualStart = yield* stack .startService("auth") .pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.await(authSpawnStarted); const activationCompleted = yield* Effect.race( - coordinator.activateService("postgres").pipe(Effect.as(true)), + activator.activate("postgres").pipe(Effect.as(true)), Effect.sleep("200 millis").pipe(Effect.as(false)), ); yield* Fiber.interrupt(manualStart); expect(activationCompleted).toBe(true); - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer)); + yield* stack.stop(); + }).pipe(Effect.provide(layer)); }).pipe(Effect.scoped, Effect.timeout("5 seconds")), ); @@ -582,13 +607,11 @@ describe("Stack", () => { : Effect.void, releaseAll: Effect.void, }; - const { coordinatorLayer } = setupLayer(config, lease); + const { layer } = setupLayer(config, lease); yield* Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - const starting = yield* coordinator - .start() - .pipe(Effect.forkChild({ startImmediately: true })); + const stack = yield* Stack; + const starting = yield* stack.start().pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.await(postgresReleaseStarted); const mailpitBeganConcurrently = yield* Effect.race( @@ -599,8 +622,8 @@ describe("Stack", () => { yield* Fiber.interrupt(starting); expect(mailpitBeganConcurrently).toBe(true); - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer)); + yield* stack.stop(); + }).pipe(Effect.provide(layer)); }).pipe(Effect.scoped, Effect.timeout("5 seconds")), ); @@ -619,17 +642,18 @@ describe("Stack", () => { : Effect.void, }); const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config, noopPortLease(config.ports), spawner); + const { layer } = setupLayer(config, noopPortLease(config.ports), spawner); yield* Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - const activationFiber = yield* coordinator - .activateService("auth") + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + const activationFiber = yield* activator + .activate("auth") .pipe(Effect.forkChild({ startImmediately: true })); yield* Deferred.await(spawnStarted); - const disposeFiber = yield* coordinator + const disposeFiber = yield* stack .dispose() .pipe(Effect.forkChild({ startImmediately: true })); yield* Fiber.join(disposeFiber); @@ -639,9 +663,108 @@ describe("Stack", () => { expect(spawner.spawned.some((record) => record.command.endsWith("/auth"))).toBe(false); - const error = yield* coordinator.activateService("auth").pipe(Effect.flip); + const error = yield* activator.activate("auth").pipe(Effect.flip); expect(error._tag).toBe("StackNotRunningError"); - }).pipe(Effect.provide(coordinatorLayer)); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.scoped, Effect.timeout("5 seconds")), + ); + + it.live("uses the stack readiness deadline for explicit lazy activation and cleans up", () => + Effect.gen(function* () { + const spawner = mockChildProcessSpawner(); + let releasedAll = false; + const config = { + ...defaultConfig, + startupMode: "lazy", + readiness: { mode: "finite", timeoutMs: 500 }, + } satisfies ResolvedStackConfig; + const lease: PortLease = { + ...noopPortLease(config.ports), + releaseAll: Effect.sync(() => { + releasedAll = true; + }), + }; + const { layer } = setupLayer(config, lease, spawner); + + yield* Effect.gen(function* () { + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + + const error = yield* activator.activate("auth").pipe(Effect.flip); + + expect(error._tag).toBe("StackReadinessError"); + if (error._tag === "StackReadinessError") { + expect(error.target).toBe("auth"); + expect(error.timeoutMs).toBe(500); + } + expect(releasedAll).toBe(true); + const spawnCountAfterDisposal = spawner.spawned.length; + expect((yield* activator.activate("postgres").pipe(Effect.flip))._tag).toBe( + "StackNotRunningError", + ); + for (const operation of [ + stack.start(), + stack.startService("postgres"), + stack.stopService("postgres"), + stack.restartService("postgres"), + stack.reloadFunctions(), + stack.reloadEdgeRuntime({ edgeRuntime: {} }), + ]) { + expect((yield* operation.pipe(Effect.flip))._tag).toBe("StackBuildError"); + } + yield* stack.stop(); + expect(spawner.spawned).toHaveLength(spawnCountAfterDisposal); + }).pipe(Effect.provide(layer)); + }).pipe(Effect.scoped, Effect.timeout("5 seconds")), + ); + + it.live("allows a finite wait override against an infinite stack policy", () => + Effect.gen(function* () { + const spawnStarted = yield* Deferred.make(); + const spawner = mockChildProcessSpawner({ + beforeSpawn: (record) => + record.args.some((arg) => + Buffer.from(arg, "base64url").toString().includes('"command":"/cache/auth/'), + ) + ? Deferred.succeed(spawnStarted, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + }); + let releasedAll = false; + const config = { + ...defaultConfig, + startupMode: "lazy", + readiness: { mode: "infinite" }, + } satisfies ResolvedStackConfig; + const lease: PortLease = { + ...noopPortLease(config.ports), + releaseAll: Effect.sync(() => { + releasedAll = true; + }), + }; + const { layer } = setupLayer(config, lease, spawner); + + yield* Effect.gen(function* () { + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + const activation = yield* activator + .activate("auth") + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(spawnStarted); + + const error = yield* stack + .waitAllReady({ mode: "finite", timeoutMs: 25 }) + .pipe(Effect.flip); + + expect(error._tag).toBe("StackReadinessError"); + if (error._tag === "StackReadinessError") { + expect(error.target).toBe("stack"); + expect(error.timeoutMs).toBe(25); + } + expect(releasedAll).toBe(true); + yield* Fiber.interrupt(activation); + }).pipe(Effect.provide(layer)); }).pipe(Effect.scoped, Effect.timeout("5 seconds")), ); @@ -664,31 +787,32 @@ describe("Stack", () => { if (authConfig === false) { throw new Error("Expected auth to be enabled in the default test config"); } - const { coordinatorLayer, spawner } = setupLayer({ + const { layer, spawner } = setupLayer({ ...defaultConfig, startupMode: "lazy", ports: { ...defaultPorts, authPort }, auth: { ...authConfig, port: authPort }, }); yield* Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; + const stack = yield* Stack; + const activator = yield* StackServiceActivator; const isAuthStart = (record: { readonly args: ReadonlyArray }) => record.args.some((arg) => Buffer.from(arg, "base64url").toString().includes('"command":"/cache/auth/'), ); - yield* coordinator.start(); - yield* coordinator.activateService("auth"); + yield* stack.start(); + yield* activator.activate("auth"); const initialAuthStarts = spawner.spawned.filter(isAuthStart).length; expect(initialAuthStarts).toBeGreaterThan(0); - yield* coordinator.stopService("postgres"); - yield* coordinator.restartService("postgres"); - yield* coordinator.waitAllReady(); + yield* stack.stopService("postgres"); + yield* stack.restartService("postgres"); + yield* stack.waitAllReady(); expect(spawner.spawned.filter(isAuthStart)).toHaveLength(initialAuthStarts); - expect((yield* coordinator.getState("auth")).status).toBe("Stopped"); - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer)); + expect((yield* stack.getState("auth")).status).toBe("Stopped"); + yield* stack.stop(); + }).pipe(Effect.provide(layer)); }).pipe(Effect.scoped, Effect.timeout("5 seconds")); }); @@ -710,57 +834,53 @@ describe("Stack", () => { it.live("keeps unactivated services dormant after a stop and start cycle", () => { const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config); + const { layer } = setupLayer(config); return Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - expect((yield* coordinator.getState("auth")).status).toBe("Dormant"); + const stack = yield* Stack; + yield* stack.start(); + expect((yield* stack.getState("auth")).status).toBe("Dormant"); - yield* coordinator.stop(); - yield* coordinator.start(); + yield* stack.stop(); + yield* stack.start(); - expect((yield* coordinator.getState("auth")).status).toBe("Dormant"); - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer), Effect.timeout("5 seconds")); + expect((yield* stack.getState("auth")).status).toBe("Dormant"); + yield* stack.stop(); + }).pipe(Effect.provide(layer), Effect.timeout("5 seconds")); }); it.live("rejects a cached activation after the stack has stopped", () => { const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config); + const { layer } = setupLayer(config); return Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - yield* coordinator.stop(); + const stack = yield* Stack; + const activator = yield* StackServiceActivator; + yield* stack.start(); + yield* stack.stop(); - const error = yield* coordinator.activateService("postgres").pipe(Effect.flip); + const error = yield* activator.activate("postgres").pipe(Effect.flip); expect(error._tag).toBe("StackNotRunningError"); - }).pipe(Effect.provide(coordinatorLayer), Effect.timeout("5 seconds")); + }).pipe(Effect.provide(layer), Effect.timeout("5 seconds")); }); it.live("preserves an explicitly stopped service across a stack restart", () => { const config = { ...defaultConfig, startupMode: "lazy" } satisfies ResolvedStackConfig; - const { coordinatorLayer } = setupLayer(config); + const { layer } = setupLayer(config); return Effect.gen(function* () { - const coordinator = yield* StackLifecycleCoordinator; - yield* coordinator.start(); - const stoppedState = yield* coordinator.allStateChanges().pipe( - Stream.filter((state) => state.name === "auth" && state.status === "Stopped"), - Stream.runHead, - Effect.forkChild({ startImmediately: true }), - ); - yield* coordinator.stopService("auth"); - expect((yield* Fiber.join(stoppedState))._tag).toBe("Some"); - expect((yield* coordinator.getState("auth")).status).toBe("Stopped"); + const stack = yield* Stack; + yield* stack.start(); + yield* stack.stopService("auth"); + yield* Effect.sleep("20 millis"); + expect((yield* stack.getState("auth")).status).toBe("Stopped"); - yield* coordinator.stop(); - yield* coordinator.start(); + yield* stack.stop(); + yield* stack.start(); - expect((yield* coordinator.getState("auth")).status).toBe("Stopped"); - yield* coordinator.stop(); - }).pipe(Effect.provide(coordinatorLayer), Effect.timeout("5 seconds")); + expect((yield* stack.getState("auth")).status).toBe("Stopped"); + yield* stack.stop(); + }).pipe(Effect.provide(layer), Effect.timeout("5 seconds")); }); it.live("releases only the ports in a lazy service dependency closure", () => { diff --git a/packages/stack/src/StackBuilder.ts b/packages/stack/src/StackBuilder.ts index 17f5e75ed4..9cc12215d9 100644 --- a/packages/stack/src/StackBuilder.ts +++ b/packages/stack/src/StackBuilder.ts @@ -53,21 +53,19 @@ const dockerOnlyServices = SERVICE_NAMES.filter( (service) => serviceMetadata(service).runtimeSupport === "docker-only", ); +// Serial health-check paths used by dependency waits; keep each path aligned +// with the corresponding service's transitive dependencies. +const postgresStartupPath: ReadonlyArray = ["postgres"]; +const storageStartupPath: ReadonlyArray = ["postgres", "storage"]; +const analyticsStartupPath: ReadonlyArray = ["postgres", "analytics"]; + +const postgresDependencyTimeoutSeconds = dependencyTimeoutSecondsForServices(postgresStartupPath); + const dependsOnPostgres = (hasPostgresInit: boolean): ReadonlyArray => hasPostgresInit ? [{ service: "postgres-init", condition: "completed" }] : [{ service: "postgres", condition: "healthy" }]; -const POSTGRES_DEPENDENCY_TIMEOUT_SECONDS = dependencyTimeoutSecondsForServices(["postgres"]); -const STORAGE_DEPENDENCY_TIMEOUT_SECONDS = dependencyTimeoutSecondsForServices([ - "postgres", - "storage", -]); -const ANALYTICS_DEPENDENCY_TIMEOUT_SECONDS = dependencyTimeoutSecondsForServices([ - "postgres", - "analytics", -]); - const publicServiceProjection = ( defs: ReadonlyArray, hasPostgresInit: boolean, @@ -251,15 +249,17 @@ export class StackBuilder extends Context.Service< ); const hasPostgresInit = postgresResolution.type === "binary"; const postgresDeps = dependsOnPostgres(hasPostgresInit); + const postgresInitCompletionBudgetSeconds = hasPostgresInit + ? POSTGRES_INIT_COMPLETION_BUDGET_SECONDS + : 0; const postgresConsumerDependencyTimeoutSeconds = - POSTGRES_DEPENDENCY_TIMEOUT_SECONDS + - (hasPostgresInit ? POSTGRES_INIT_COMPLETION_BUDGET_SECONDS : 0); + postgresDependencyTimeoutSeconds + postgresInitCompletionBudgetSeconds; const storageDependencyTimeoutSeconds = - STORAGE_DEPENDENCY_TIMEOUT_SECONDS + - (hasPostgresInit ? POSTGRES_INIT_COMPLETION_BUDGET_SECONDS : 0); + dependencyTimeoutSecondsForServices(storageStartupPath) + + postgresInitCompletionBudgetSeconds; const analyticsDependencyTimeoutSeconds = - ANALYTICS_DEPENDENCY_TIMEOUT_SECONDS + - (hasPostgresInit ? POSTGRES_INIT_COMPLETION_BUDGET_SECONDS : 0); + dependencyTimeoutSecondsForServices(analyticsStartupPath) + + postgresInitCompletionBudgetSeconds; const jwtJwks = generateJwks(config.jwtSecret); const defs: Array = [ @@ -293,7 +293,7 @@ export class StackBuilder extends Context.Service< dbPort: config.dbPort, autoExposeNewTables: config.postgres.autoExposeNewTables, }), - dependencyTimeoutSeconds: POSTGRES_DEPENDENCY_TIMEOUT_SECONDS, + dependencyTimeoutSeconds: postgresDependencyTimeoutSeconds, enabled: true, }); } diff --git a/packages/stack/src/StackConfig.ts b/packages/stack/src/StackConfig.ts index cbf6552fc8..77b23e0bec 100644 --- a/packages/stack/src/StackConfig.ts +++ b/packages/stack/src/StackConfig.ts @@ -1,3 +1,4 @@ +import { Schema } from "effect"; import type { FunctionsConfig, ResolvedFunctionsConfig } from "./functions.ts"; import type { AllocatedPorts } from "./PortAllocator.ts"; @@ -10,10 +11,26 @@ export type ReadinessPolicy = export type ReadyOptions = { readonly mode: "inherit" } | ReadinessPolicy; -/** Standalone stacks wait at most two minutes unless a caller or launch Adapter chooses otherwise. */ +const ReadinessPolicySchema = Schema.Union([ + Schema.Struct({ + mode: Schema.Literal("finite"), + timeoutMs: Schema.Int.check(Schema.isGreaterThan(0)), + }), + Schema.Struct({ mode: Schema.Literal("infinite") }), +]); + +/** The single wire representation accepted by Effect, Promise, and daemon Adapters. */ +export const ReadyOptionsSchema = Schema.Union([ + Schema.Struct({ mode: Schema.Literal("inherit") }), + ReadinessPolicySchema, +]); + +export const inheritReadyOptions: ReadyOptions = { mode: "inherit" }; + +/** Standalone stacks wait at most three minutes unless a caller or launch Adapter chooses otherwise. */ export const DEFAULT_STACK_READINESS_POLICY: ReadinessPolicy = { mode: "finite", - timeoutMs: 120_000, + timeoutMs: 180_000, }; /** Resolve readiness with per-call policy taking precedence over stack policy and package default. */ diff --git a/packages/stack/src/StackConfig.unit.test.ts b/packages/stack/src/StackConfig.unit.test.ts index 9c3d3b0a84..21978dcd9e 100644 --- a/packages/stack/src/StackConfig.unit.test.ts +++ b/packages/stack/src/StackConfig.unit.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from "vitest"; +import { Schema } from "effect"; import { DEFAULT_STACK_READINESS_POLICY, + ReadyOptionsSchema, resolveReadinessPolicy, type ReadinessPolicy, } from "./StackConfig.ts"; @@ -16,9 +18,9 @@ describe("resolveReadinessPolicy", () => { expect( resolveReadinessPolicy({ readyOptions: { mode: "inherit" }, - stackPolicy: finite(120_000), + stackPolicy: finite(180_000), }), - ).toEqual(finite(120_000)); + ).toEqual(finite(180_000)); expect( resolveReadinessPolicy({ readyOptions: { mode: "inherit" }, @@ -31,13 +33,13 @@ describe("resolveReadinessPolicy", () => { expect( resolveReadinessPolicy({ readyOptions: finite(5_000), - stackPolicy: finite(120_000), + stackPolicy: finite(180_000), }), ).toEqual(finite(5_000)); expect( resolveReadinessPolicy({ readyOptions: finite(300_000), - stackPolicy: finite(120_000), + stackPolicy: finite(180_000), }), ).toEqual(finite(300_000)); }); @@ -46,7 +48,7 @@ describe("resolveReadinessPolicy", () => { expect( resolveReadinessPolicy({ readyOptions: { mode: "infinite" }, - stackPolicy: finite(120_000), + stackPolicy: finite(180_000), }), ).toEqual({ mode: "infinite" }); expect( @@ -57,3 +59,22 @@ describe("resolveReadinessPolicy", () => { ).toEqual(finite(30_000)); }); }); + +describe("ReadyOptionsSchema", () => { + const decode = Schema.decodeUnknownSync(ReadyOptionsSchema); + + it("accepts the three readiness override modes", () => { + expect(decode({ mode: "inherit" })).toEqual({ mode: "inherit" }); + expect(decode({ mode: "infinite" })).toEqual({ mode: "infinite" }); + expect(decode({ mode: "finite", timeoutMs: 25 })).toEqual({ + mode: "finite", + timeoutMs: 25, + }); + }); + + it("rejects malformed and non-positive finite deadlines", () => { + expect(() => decode({ mode: "finite", timeoutMs: 0 })).toThrow(); + expect(() => decode({ mode: "finite", timeoutMs: -1 })).toThrow(); + expect(() => decode({ mode: "forever" })).toThrow(); + }); +}); diff --git a/packages/stack/src/StackLifecycleCoordinator.ts b/packages/stack/src/StackLifecycleCoordinator.ts deleted file mode 100644 index 5127288e21..0000000000 --- a/packages/stack/src/StackLifecycleCoordinator.ts +++ /dev/null @@ -1,872 +0,0 @@ -import { LogBuffer, Orchestrator } from "@supabase/process-compose"; -import { ServiceNotFoundError } from "@supabase/process-compose"; -import type { LogEntry, ResolvedGraph, ServiceReadyError } from "@supabase/process-compose"; -import { - Deferred, - Effect, - FileSystem, - Layer, - Path, - Ref, - Semaphore, - Context, - Stream, - SubscriptionRef, -} from "effect"; -import { ChildProcessSpawner } from "effect/unstable/process"; -import type { CleanupTargets } from "./CleanupTargets.ts"; -import { cleanupLocalStackResources } from "./cleanup.ts"; -import { StackBuildError, StackNotRunningError } from "./errors.ts"; -import { configureFunctionsRuntime, type FunctionsConfig } from "./functions.ts"; -import { detectPlatform, dockerHostAddress } from "./Platform.ts"; -import type { PortLease } from "./PortAllocator.ts"; -import { - activationTargetsForService, - eagerServices, - lifecycleTargetsForService, -} from "./ServiceActivation.ts"; -import { portFieldsForService } from "./ServicePorts.ts"; -import { StackMetadataPersistence } from "./StackMetadataPersistence.ts"; -import { StackPreparation } from "./StackPreparation.ts"; -import type { PreparedStackArtifacts } from "./StackPreparation.ts"; -import { - enabledServicesForConfig, - StackBuilder, - validateResolvedConfig, - versionsForConfig, -} from "./StackBuilder.ts"; -import type { ResolvedStackConfig } from "./StackConfig.ts"; -import { projectStackStates, type StackServiceProjectionCatalog } from "./StackStateProjection.ts"; -import { StackServiceState } from "./StackServiceState.ts"; -import type { EdgeRuntimeReloadConfig, StackInfo } from "./Stack.ts"; -import { SERVICE_NAMES } from "./ServiceCatalog.ts"; -import type { ServiceName } from "./ServiceName.ts"; - -type LifecyclePhase = - | "idle" - | "preparing" - | "prepared" - | "starting" - | "running" - | "stopping" - | "stopped"; - -interface RuntimeState { - readonly orchestrator: Orchestrator["Service"]; - readonly graph: ResolvedGraph; - readonly serviceProjection: StackServiceProjectionCatalog; - readonly cleanupTargets: CleanupTargets; -} - -const sameState = (a: StackServiceState | undefined, b: StackServiceState): boolean => - a?.name === b.name && - a.status === b.status && - a.pid === b.pid && - a.exitCode === b.exitCode && - a.restartCount === b.restartCount && - a.startedAt === b.startedAt && - a.error === b.error; - -const initialPublicStates = (config: ResolvedStackConfig): ReadonlyArray => - enabledServicesForConfig(config).map( - (name) => - new StackServiceState({ - name, - status: "Pending", - pid: null, - exitCode: null, - restartCount: 0, - startedAt: null, - error: null, - }), - ); - -const stackInfoFor = (config: ResolvedStackConfig): StackInfo => { - const apiUrl = `http://127.0.0.1:${config.apiPort}`; - return { - url: apiUrl, - dbUrl: `postgresql://postgres:postgres@127.0.0.1:${config.dbPort}/postgres`, - publishableKey: config.publishableKey, - secretKey: config.secretKey, - anonJwt: config.anonJwt, - serviceRoleJwt: config.serviceRoleJwt, - serviceEndpoints: { - ...(config.auth === false ? {} : { auth: `${apiUrl}/auth/v1` }), - ...(config.postgrest === false ? {} : { postgrest: `${apiUrl}/rest/v1` }), - ...(config.edgeRuntime === false - ? {} - : { - functions: `${apiUrl}/functions/v1`, - edge_runtime: `${apiUrl}/functions/v1`, - }), - ...(config.realtime === false ? {} : { realtime: `${apiUrl}/realtime/v1` }), - ...(config.storage === false - ? {} - : { - storage: `${apiUrl}/storage/v1`, - storage_s3: `${apiUrl}/storage/v1/s3`, - }), - ...(config.imgproxy === false || config.startupMode === "lazy" - ? {} - : { imgproxy: `http://127.0.0.1:${config.imgproxy.port}` }), - ...(config.mailpit === false - ? {} - : { - mailpit: `http://127.0.0.1:${config.mailpit.port}`, - mailpit_smtp: `smtp://127.0.0.1:${config.mailpit.smtpPort}`, - mailpit_pop3: `pop3://127.0.0.1:${config.mailpit.pop3Port}`, - }), - ...(config.pgmeta === false ? {} : { pgmeta: `${apiUrl}/pg` }), - ...(config.studio === false ? {} : { studio: `http://127.0.0.1:${config.studio.port}` }), - ...(config.analytics === false ? {} : { analytics: `${apiUrl}/analytics/v1` }), - ...(config.pooler === false - ? {} - : { - pooler: `postgresql://postgres:postgres@127.0.0.1:${config.pooler.port}/postgres`, - pooler_admin: `http://127.0.0.1:${config.pooler.apiPort}`, - }), - }, - }; -}; - -const changedStatesBetween = ( - previous: ReadonlyArray | undefined, - current: ReadonlyArray, -): ReadonlyArray => { - if (previous === undefined) { - return current; - } - - const previousByName = new Map(previous.map((state) => [state.name, state] as const)); - return current.filter((state) => !sameState(previousByName.get(state.name), state)); -}; - -export class StackLifecycleCoordinator extends Context.Service< - StackLifecycleCoordinator, - { - readonly getInfo: () => Effect.Effect; - readonly getCleanupTargets: () => Effect.Effect; - readonly start: () => Effect.Effect; - readonly stop: () => Effect.Effect; - readonly dispose: () => Effect.Effect; - readonly startService: ( - name: string, - ) => Effect.Effect; - readonly activateService: ( - name: ServiceName, - ) => Effect.Effect< - void, - ServiceNotFoundError | ServiceReadyError | StackBuildError | StackNotRunningError - >; - readonly stopService: ( - name: string, - ) => Effect.Effect; - readonly restartService: ( - name: string, - ) => Effect.Effect; - readonly reloadFunctions: ( - opts?: FunctionsConfig, - ) => Effect.Effect; - readonly reloadEdgeRuntime: ( - opts: EdgeRuntimeReloadConfig, - ) => Effect.Effect; - readonly getState: (name: string) => Effect.Effect; - readonly getAllStates: () => Effect.Effect>; - readonly stateChanges: ( - name: string, - ) => Effect.Effect, ServiceNotFoundError>; - readonly allStateChanges: () => Stream.Stream; - readonly waitReady: ( - name: string, - ) => Effect.Effect; - readonly waitAllReady: () => Effect.Effect; - readonly subscribeLogs: (name: string) => Stream.Stream; - readonly subscribeAllLogs: (services?: ReadonlyArray) => Stream.Stream; - readonly logHistory: (name: string, limit?: number) => Effect.Effect>; - readonly logHistoryAll: ( - limit?: number, - services?: ReadonlyArray, - ) => Effect.Effect>; - } ->()("stack/StackLifecycleCoordinator") { - static layer = ( - config: ResolvedStackConfig, - portLease: PortLease, - ): Layer.Layer< - StackLifecycleCoordinator, - StackBuildError, - | StackBuilder - | StackPreparation - | ChildProcessSpawner.ChildProcessSpawner - | StackMetadataPersistence - | FileSystem.FileSystem - | Path.Path - > => - Layer.effect( - this, - Effect.gen(function* () { - const builder = yield* StackBuilder; - const preparation = yield* StackPreparation; - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const metadataPersistence = yield* StackMetadataPersistence; - const fs = yield* FileSystem.FileSystem; - const path = yield* Path.Path; - const scope = yield* Effect.scope; - - const info = stackInfoFor(config); - const enabledServices = enabledServicesForConfig(config); - const stateRef = yield* SubscriptionRef.make(initialPublicStates(config)); - const phaseRef = yield* Ref.make("idle"); - const lifecycleLock = Semaphore.makeUnsafe(1); - const projectionLock = Semaphore.makeUnsafe(1); - - const logBufferServices = yield* Layer.buildWithScope(LogBuffer.layer, scope); - const logBuffer = Context.get(logBufferServices, LogBuffer); - - const updateState = (nextState: StackServiceState) => - SubscriptionRef.update(stateRef, (current) => { - const previous = current.find((entry) => entry.name === nextState.name); - if (sameState(previous, nextState)) { - return current; - } - return current.some((entry) => entry.name === nextState.name) - ? current.map((entry) => (entry.name === nextState.name ? nextState : entry)) - : [...current, nextState]; - }); - - const syncProjectedStates = ( - orchestrator: Orchestrator["Service"], - serviceProjection: StackServiceProjectionCatalog, - ) => - Effect.gen(function* () { - const rawStates = yield* orchestrator.getAllStates(); - yield* Effect.forEach(projectStackStates(rawStates, serviceProjection), updateState, { - discard: true, - }); - }).pipe(projectionLock.withPermit); - - const requireKnownService = (name: string) => - Effect.gen(function* () { - const currentStates = SubscriptionRef.getUnsafe(stateRef); - const match = currentStates.find((state) => state.name === name); - if (match === undefined) { - return yield* Effect.fail(new ServiceNotFoundError({ name })); - } - return match; - }); - const requireKnownServiceName = ( - name: string, - ): Effect.Effect => - Effect.gen(function* () { - yield* requireKnownService(name); - const service = SERVICE_NAMES.find((candidate) => candidate === name); - if (service === undefined) { - return yield* Effect.fail(new ServiceNotFoundError({ name })); - } - return service; - }); - - let preparedArtifacts: PreparedStackArtifacts | undefined; - let prepareDeferred: Deferred.Deferred | undefined; - let runtimeState: RuntimeState | undefined; - let runtimeDeferred: Deferred.Deferred | undefined; - - const ensurePrepared = Effect.suspend(() => { - if (preparedArtifacts !== undefined) { - return Effect.succeed(preparedArtifacts); - } - if (prepareDeferred !== undefined) { - return Deferred.await(prepareDeferred); - } - - const deferred = Deferred.makeUnsafe(); - prepareDeferred = deferred; - - const effect = Effect.gen(function* () { - yield* validateResolvedConfig(config); - yield* Ref.set(phaseRef, "preparing"); - - let prepared: PreparedStackArtifacts | undefined; - yield* preparation - .prepareEvents({ - mode: config.mode, - services: enabledServicesForConfig(config), - versions: versionsForConfig(config), - }) - .pipe( - Stream.mapError( - (cause) => - new StackBuildError({ - detail: "Failed to prepare stack assets", - cause, - }), - ), - ) - .pipe( - Stream.runForEach((event) => { - switch (event._tag) { - case "ServiceDownloadStarted": - return updateState( - new StackServiceState({ - name: event.service, - status: "Downloading", - pid: null, - exitCode: null, - restartCount: 0, - startedAt: null, - error: null, - }), - ); - case "ServiceDownloadFinished": - return updateState( - new StackServiceState({ - name: event.service, - status: "Pending", - pid: null, - exitCode: null, - restartCount: 0, - startedAt: null, - error: null, - }), - ); - case "PreparationCompleted": - return Effect.sync(() => { - prepared = event.artifacts; - }); - } - }), - ); - - if (prepared === undefined) { - return yield* Effect.fail( - new StackBuildError({ - detail: "Stack preparation completed without prepared artifacts", - }), - ); - } - - yield* Ref.set(phaseRef, "prepared"); - return prepared; - }).pipe( - Effect.tap((value) => - Effect.sync(() => { - preparedArtifacts = value; - }), - ), - Effect.onError(() => Ref.set(phaseRef, "idle")), - Effect.ensuring( - Effect.sync(() => { - prepareDeferred = undefined; - }), - ), - ); - - return Effect.gen(function* () { - yield* Effect.forkIn(effect.pipe(Deferred.into(deferred)), scope); - return yield* Deferred.await(deferred); - }); - }); - - const ensureRuntime = Effect.suspend(() => { - if (runtimeState !== undefined) { - return Effect.succeed(runtimeState); - } - if (runtimeDeferred !== undefined) { - return Deferred.await(runtimeDeferred); - } - - const deferred = Deferred.makeUnsafe(); - runtimeDeferred = deferred; - - const effect = Effect.gen(function* () { - const prepared = yield* ensurePrepared; - const { graph, serviceProjection, cleanupTargets } = yield* builder.build( - config, - prepared, - ); - - yield* metadataPersistence.persistCleanupTargets(cleanupTargets).pipe( - Effect.mapError( - (cause) => - new StackBuildError({ - detail: "Failed to persist stack cleanup metadata", - cause, - }), - ), - ); - - const orchLayer = Orchestrator.layer(graph).pipe( - Layer.provide(Layer.succeed(LogBuffer, logBuffer)), - Layer.provide(Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner)), - ); - const orchServices = yield* Layer.buildWithScope(orchLayer, scope); - const orchestrator = Context.get(orchServices, Orchestrator); - - yield* syncProjectedStates(orchestrator, serviceProjection); - yield* orchestrator.allStateChanges().pipe( - Stream.runForEach(() => syncProjectedStates(orchestrator, serviceProjection)), - Effect.ignore, - Effect.forkIn(scope), - ); - - return { - orchestrator, - graph, - serviceProjection, - cleanupTargets, - } satisfies RuntimeState; - }).pipe( - Effect.tap((value) => - Effect.sync(() => { - runtimeState = value; - }), - ), - Effect.ensuring( - Effect.sync(() => { - runtimeDeferred = undefined; - }), - ), - ); - - return Effect.gen(function* () { - yield* Effect.forkIn(effect.pipe(Deferred.into(deferred)), scope); - return yield* Deferred.await(deferred); - }); - }); - - let disposed = false; - const runtimeHost = Effect.gen(function* () { - const prepared = yield* ensurePrepared; - const platform = yield* detectPlatform; - const edgeRuntimeResolution = prepared.resolutions["edge-runtime"]; - return { - hostname: - edgeRuntimeResolution?.type === "docker" - ? dockerHostAddress(platform.os) - : "127.0.0.1", - }; - }); - const providePlatform = ( - effect: Effect.Effect, - ): Effect.Effect => - effect.pipe( - Effect.provideService(FileSystem.FileSystem, fs), - Effect.provideService(Path.Path, path), - ); - const configureFunctions = ( - nextConfig: ResolvedStackConfig, - ): Effect.Effect => - Effect.gen(function* () { - yield* providePlatform(configureFunctionsRuntime(nextConfig, yield* runtimeHost)); - }).pipe( - Effect.mapError( - (cause) => - new StackBuildError({ - detail: "Failed to configure Edge Functions", - cause, - }), - ), - ); - const configWithFunctionOptions = (opts?: FunctionsConfig): ResolvedStackConfig => { - if (opts === undefined) { - return config; - } - const base = config.functions === false ? { noVerifyJwt: false } : config.functions; - return { - ...config, - functions: { - envFile: opts.envFile ?? base.envFile, - noVerifyJwt: opts.noVerifyJwt ?? base.noVerifyJwt, - }, - }; - }; - const configWithEdgeRuntimeOptions = ( - opts: EdgeRuntimeReloadConfig, - ): Effect.Effect => - Effect.gen(function* () { - if (config.edgeRuntime === false || opts.edgeRuntime.enabled === false) { - return yield* Effect.fail(new ServiceNotFoundError({ name: "edge-runtime" })); - } - - const base = configWithFunctionOptions(opts.functions); - return { - ...base, - edgeRuntime: { - ...config.edgeRuntime, - enabled: opts.edgeRuntime.enabled ?? config.edgeRuntime.enabled, - inspectorPort: opts.edgeRuntime.inspectorPort ?? config.edgeRuntime.inspectorPort, - policy: opts.edgeRuntime.policy ?? config.edgeRuntime.policy, - env: opts.edgeRuntime.env ?? config.edgeRuntime.env, - }, - }; - }); - const publicAllStateChanges = () => - SubscriptionRef.changes(stateRef).pipe( - Stream.mapAccum< - ReadonlyArray | undefined, - ReadonlyArray, - StackServiceState - >( - () => undefined, - (previous, current) => [current, changedStatesBetween(previous, current)], - ), - ); - const withLifecycleLock = lifecycleLock.withPermit; - const syncRuntimeProjectedStates = (runtime: RuntimeState) => - syncProjectedStates(runtime.orchestrator, runtime.serviceProjection); - const serviceStartOptions = { - beforeStart: (name: string) => portLease.reserve(portFieldsForService(name)), - beforeSpawn: (name: string) => portLease.release(portFieldsForService(name)), - }; - const knownServiceError = (service: string, cause: ServiceNotFoundError) => - new StackBuildError({ - detail: `Prepared graph does not contain enabled service ${service}`, - cause, - }); - const beginStartTargets = ( - root: ServiceName, - allowExplicitlyStopped: ReadonlySet, - ) => - Effect.gen(function* () { - const runtime = yield* ensureRuntime; - const targets = activationTargetsForService(enabledServices, root); - const targetClosure = new Set( - targets.flatMap((target) => - runtime.graph.startOrderFor(target).map((definition) => definition.name), - ), - ); - - for (const dependency of targetClosure) { - const state = yield* runtime.orchestrator - .getState(dependency) - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError(dependency, cause)), - ), - ); - const publicDependency = SERVICE_NAMES.find((candidate) => candidate === dependency); - if ( - state.desired === "stopped" && - publicDependency !== undefined && - !allowExplicitlyStopped.has(publicDependency) - ) { - return yield* Effect.fail( - new StackBuildError({ - detail: `Cannot activate ${root} because dependency ${dependency} was explicitly stopped`, - }), - ); - } - } - - for (const target of targets) { - yield* runtime.orchestrator - .startService(target, serviceStartOptions) - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError(target, cause)), - ), - ); - } - return { runtime, targets }; - }); - const waitForTargets = ({ - runtime, - targets, - }: { - readonly runtime: RuntimeState; - readonly targets: ReadonlyArray; - }) => - Effect.gen(function* () { - yield* Effect.forEach( - targets, - (target) => - runtime.orchestrator - .waitReady(target) - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError(target, cause)), - ), - ), - { concurrency: "unbounded", discard: true }, - ); - yield* syncRuntimeProjectedStates(runtime); - }); - const inspectStartedTargets = (root: ServiceName) => - Effect.gen(function* () { - const runtime = yield* ensureRuntime; - const targets = activationTargetsForService(enabledServices, root); - const states = yield* Effect.forEach(targets, (target) => - runtime.orchestrator - .getState(target) - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError(target, cause)), - ), - ), - ); - if (states.some((state) => state.desired !== "running")) { - return undefined; - } - return { - runtime, - targets, - ready: states.every( - (state) => - state.status === "Healthy" || - (state.status === "Stopped" && state.exitCode === 0), - ), - }; - }); - const requireRunningPhase = Effect.gen(function* () { - const phase = yield* Ref.get(phaseRef); - if (phase !== "running") { - return yield* Effect.fail(new StackNotRunningError({ phase })); - } - }); - const disposeOnce = () => - Effect.gen(function* () { - if (disposed) { - return; - } - disposed = true; - yield* Ref.set(phaseRef, "stopping"); - yield* cleanupLocalStackResources({ - stop: () => - runtimeState === undefined ? Effect.void : runtimeState.orchestrator.stop(), - cleanupTargets: runtimeState?.cleanupTargets ?? { dockerContainerNames: [] }, - config, - }).pipe( - Effect.ensuring(portLease.releaseAll), - Effect.ensuring(Ref.set(phaseRef, "stopped")), - ); - }).pipe(withLifecycleLock); - - yield* Effect.addFinalizer(disposeOnce); - - return { - getInfo: () => Effect.succeed(info), - getCleanupTargets: () => - Effect.succeed(runtimeState?.cleanupTargets ?? { dockerContainerNames: [] }), - start: () => - Effect.gen(function* () { - yield* Ref.set(phaseRef, "starting"); - const runtime = yield* ensureRuntime; - yield* configureFunctions(config); - - if (config.startupMode === "lazy") { - const readiness: Array> = - []; - if ( - runtime.graph.startOrder.some((definition) => definition.name === "postgres-init") - ) { - yield* runtime.orchestrator - .startService("postgres-init", serviceStartOptions) - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError("postgres-init", cause)), - ), - ); - readiness.push( - runtime.orchestrator - .waitReady("postgres-init") - .pipe( - Effect.catchTag("ServiceNotFoundError", (cause) => - Effect.fail(knownServiceError("postgres-init", cause)), - ), - ), - ); - } - for (const service of eagerServices(enabledServices)) { - const started = yield* beginStartTargets( - service, - new Set(lifecycleTargetsForService(enabledServices, service)), - ); - readiness.push(waitForTargets(started)); - } - yield* Effect.all(readiness, { concurrency: "unbounded", discard: true }); - } else { - yield* runtime.orchestrator.start(serviceStartOptions); - yield* runtime.orchestrator.waitAllReady(); - yield* syncRuntimeProjectedStates(runtime); - } - yield* Ref.set(phaseRef, "running"); - }).pipe( - Effect.onError(() => Ref.set(phaseRef, "stopped")), - withLifecycleLock, - ), - stop: () => - Effect.gen(function* () { - if (runtimeState === undefined) { - yield* Ref.set(phaseRef, "stopped"); - return; - } - yield* Ref.set(phaseRef, "stopping"); - yield* runtimeState.orchestrator.stop(); - yield* Ref.set(phaseRef, "stopped"); - }).pipe(withLifecycleLock), - dispose: disposeOnce, - startService: (name) => - Effect.gen(function* () { - const started = yield* Effect.gen(function* () { - const service = yield* requireKnownServiceName(name); - return yield* beginStartTargets( - service, - new Set(lifecycleTargetsForService(enabledServices, service)), - ); - }).pipe(withLifecycleLock); - yield* waitForTargets(started); - }), - activateService: (name) => - Effect.gen(function* () { - yield* requireRunningPhase; - const service = yield* requireKnownServiceName(name); - const existing = yield* inspectStartedTargets(service); - if (existing?.ready === true) { - // Close the race with a concurrent stack stop before taking - // the lock-free healthy-request fast path. - yield* requireRunningPhase; - return; - } - if (existing !== undefined) { - yield* waitForTargets(existing); - return; - } - const started = yield* Effect.gen(function* () { - yield* requireRunningPhase; - const concurrentlyStarted = yield* inspectStartedTargets(service); - if (concurrentlyStarted !== undefined) return concurrentlyStarted; - return yield* beginStartTargets(service, new Set()); - }).pipe(withLifecycleLock); - yield* waitForTargets(started); - }), - stopService: (name) => - Effect.gen(function* () { - const service = yield* requireKnownServiceName(name); - const runtime = yield* ensureRuntime; - for (const target of lifecycleTargetsForService( - enabledServices, - service, - ).toReversed()) { - yield* runtime.orchestrator.stopService(target); - } - }).pipe(withLifecycleLock), - restartService: (name) => - Effect.gen(function* () { - const started = yield* Effect.gen(function* () { - const service = yield* requireKnownServiceName(name); - const runtime = yield* ensureRuntime; - yield* runtime.orchestrator.restartService(service, serviceStartOptions); - return { runtime, targets: [service] }; - }).pipe(withLifecycleLock); - yield* waitForTargets(started); - }), - reloadFunctions: (opts) => - Effect.gen(function* () { - const started = yield* Effect.gen(function* () { - yield* requireKnownService("edge-runtime"); - yield* configureFunctions(configWithFunctionOptions(opts)); - const runtime = yield* ensureRuntime; - const state = yield* runtime.orchestrator.getState("edge-runtime"); - if (state.desired !== "running") { - return yield* beginStartTargets("edge-runtime", new Set(["edge-runtime"])); - } - yield* runtime.orchestrator.restartService("edge-runtime", serviceStartOptions); - return { runtime, targets: ["edge-runtime"] as const }; - }).pipe(withLifecycleLock); - yield* waitForTargets(started); - }), - reloadEdgeRuntime: (opts) => - Effect.gen(function* () { - const started = yield* Effect.gen(function* () { - yield* requireKnownService("edge-runtime"); - const nextConfig = yield* configWithEdgeRuntimeOptions(opts); - const prepared = yield* ensurePrepared; - const runtime = yield* ensureRuntime; - const buildResult = yield* builder.build(nextConfig, prepared); - const edgeRuntimeDef = buildResult.graph.startOrder.find( - (def) => def.name === "edge-runtime", - ); - - if (edgeRuntimeDef === undefined) { - return yield* Effect.fail(new ServiceNotFoundError({ name: "edge-runtime" })); - } - - yield* configureFunctions(nextConfig); - yield* runtime.orchestrator - .updateServiceDefinition("edge-runtime", edgeRuntimeDef) - .pipe( - Effect.mapError( - (cause) => - new StackBuildError({ - detail: "Failed to update edge-runtime service definition", - cause, - }), - ), - ); - const state = yield* runtime.orchestrator.getState("edge-runtime"); - if (state.desired !== "running") { - return yield* beginStartTargets("edge-runtime", new Set(["edge-runtime"])); - } - yield* runtime.orchestrator.restartService("edge-runtime", serviceStartOptions); - return { runtime, targets: ["edge-runtime"] as const }; - }).pipe(withLifecycleLock); - yield* waitForTargets(started); - }), - getState: (name) => - Effect.gen(function* () { - const currentStates = SubscriptionRef.getUnsafe(stateRef); - const match = currentStates.find((state) => state.name === name); - if (match === undefined) { - return yield* Effect.fail(new ServiceNotFoundError({ name })); - } - return match; - }), - getAllStates: () => Effect.sync(() => SubscriptionRef.getUnsafe(stateRef)), - stateChanges: (name) => - Effect.gen(function* () { - yield* requireKnownService(name); - return Stream.filter(publicAllStateChanges(), (state) => state.name === name); - }), - allStateChanges: publicAllStateChanges, - waitReady: (name) => - Effect.gen(function* () { - const phase = yield* Ref.get(phaseRef); - if (phase !== "running") { - return yield* Effect.fail( - new StackBuildError({ - detail: `Cannot wait for service ${name} while the stack is ${phase}`, - }), - ); - } - yield* requireKnownServiceName(name); - const runtime = yield* ensureRuntime; - yield* runtime.orchestrator.waitReady(name); - yield* syncRuntimeProjectedStates(runtime); - }), - waitAllReady: () => - Effect.gen(function* () { - const phase = yield* Ref.get(phaseRef); - if (phase !== "running") { - return yield* Effect.fail( - new StackBuildError({ - detail: `Cannot wait for stack readiness while the stack is ${phase}`, - }), - ); - } - const runtime = yield* ensureRuntime; - yield* runtime.orchestrator.waitAllReady(); - yield* syncRuntimeProjectedStates(runtime); - }), - subscribeLogs: (name) => logBuffer.subscribe(name), - subscribeAllLogs: (services) => - services === undefined || services.length === 0 - ? logBuffer.subscribeAll() - : logBuffer - .subscribeAll() - .pipe(Stream.filter((entry) => services.includes(entry.service))), - logHistory: (name, limit) => logBuffer.history(name, limit), - logHistoryAll: (limit, services) => logBuffer.historyAll(limit, services), - }; - }), - ); -} diff --git a/packages/stack/src/createStack.ts b/packages/stack/src/createStack.ts index 9f0559a4fc..6d06b9666f 100644 --- a/packages/stack/src/createStack.ts +++ b/packages/stack/src/createStack.ts @@ -1,10 +1,11 @@ import type { LogEntry } from "@supabase/process-compose"; -import { Duration, Effect, type Layer, ManagedRuntime, Stream } from "effect"; +import { Context, Effect, type Layer, ManagedRuntime, Stream } from "effect"; import { FileSystem, Path } from "effect"; import { HttpServer } from "effect/unstable/http"; import { ChildProcessSpawner } from "effect/unstable/process"; import { cleanupAutoManagedPaths, dockerForceRemove } from "./cleanup.ts"; import type { CleanupTargets } from "./CleanupTargets.ts"; +import { ApiProxy } from "./ApiProxy.ts"; import { toStackError } from "./errors.ts"; import type { FunctionsConfig } from "./functions.ts"; import { daemonLayer, foregroundLayer, type DaemonStartError } from "./layers.ts"; @@ -13,7 +14,6 @@ import { allocatedPortFieldsForConfig } from "./ServicePorts.ts"; import { Stack } from "./Stack.ts"; import type { EdgeRuntimeReloadConfig } from "./Stack.ts"; import type { ReadyOptions, ResolvedStackConfig, StackConfig } from "./StackConfig.ts"; -import { resolveReadinessPolicy } from "./StackConfig.ts"; import { resolveConfig } from "./StackConfigResolver.ts"; import type { StackServiceState } from "./StackServiceState.ts"; import { InvalidStackStateError, StackAlreadyRunningError } from "./StateManager.ts"; @@ -34,6 +34,22 @@ export interface PlatformFactoryOptions { export type PlatformFactory = (options: PlatformFactoryOptions) => PlatformLayer; +/** @internal Converts foreground operation failures and closes the runtime after terminal timeouts. */ +export async function runForegroundOperation( + operation: Promise, + dispose: () => Promise, +): Promise { + try { + return await operation; + } catch (error: unknown) { + const stackError = toStackError(error); + if (stackError.code === "STACK_READINESS_TIMEOUT") { + await dispose(); + } + throw stackError; + } +} + export interface StackHandle extends AsyncDisposable { readonly url: string; readonly dbUrl: string; @@ -139,21 +155,24 @@ export async function createStack( try { const services = await runtime.context(); - const localStack = await runtime.runPromise( - Effect.gen(function* () { - return yield* Stack; - }), - ); + const localStack = Context.get(services, Stack); + const apiProxy = Context.get(services, ApiProxy); const info = await runtime.runPromise(localStack.getInfo()); + let disposal: Promise | undefined; + const gracefulDispose = () => { + disposal ??= runtime.dispose().catch(() => {}); + return disposal; + }; const run = (effect: Effect.Effect) => - runtime.runPromise(effect).catch((error: unknown) => { - throw toStackError(error); - }); + runForegroundOperation(runtime.runPromise(effect), gracefulDispose); - const gracefulDispose = async () => { - await runtime.dispose().catch(() => {}); - }; + // A terminal lazy-activation timeout disposes LocalStack. Close the + // foreground runtime as well so the public API port cannot outlive it. + void runtime + .runPromise(apiProxy.awaitTerminalFailure) + .then(gracefulDispose) + .catch(() => {}); const stack: StackHandle = { url: info.url, @@ -168,28 +187,8 @@ export async function createStack( restartService: (name) => run(localStack.restartService(name)), reloadFunctions: (opts) => run(localStack.reloadFunctions(opts)), reloadEdgeRuntime: (opts) => run(localStack.reloadEdgeRuntime(opts)), - ready: (opts) => { - const policy = resolveReadinessPolicy({ - readyOptions: opts, - stackPolicy: resolved.readiness, - }); - const effect = - policy.mode === "finite" - ? localStack.waitAllReady().pipe(Effect.timeout(Duration.millis(policy.timeoutMs))) - : localStack.waitAllReady(); - return run(effect); - }, - serviceReady: (name, opts) => { - const policy = resolveReadinessPolicy({ - readyOptions: opts, - stackPolicy: resolved.readiness, - }); - const effect = - policy.mode === "finite" - ? localStack.waitReady(name).pipe(Effect.timeout(Duration.millis(policy.timeoutMs))) - : localStack.waitReady(name); - return run(effect); - }, + ready: (opts) => run(localStack.waitAllReady(opts)), + serviceReady: (name, opts) => run(localStack.waitReady(name, opts)), getStatus: () => run(localStack.getAllStates()), getServiceStatus: (name) => run(localStack.getState(name)), statusChanges: () => Stream.toAsyncIterableWith(localStack.allStateChanges(), services), diff --git a/packages/stack/src/createStack.unit.test.ts b/packages/stack/src/createStack.unit.test.ts index 747b16686b..90516c1884 100644 --- a/packages/stack/src/createStack.unit.test.ts +++ b/packages/stack/src/createStack.unit.test.ts @@ -2,7 +2,8 @@ import { describe, expect, it } from "vitest"; import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { StackHandle } from "./createStack.ts"; +import { runForegroundOperation, type StackHandle } from "./createStack.ts"; +import { StackReadinessError } from "./errors.ts"; import type { AllocatedPorts } from "./PortAllocator.ts"; import { DEFAULT_MANAGED_STACK_NAME, projectKeyForProjectDir } from "./paths.ts"; import { stackMetadata } from "./StackMetadata.ts"; @@ -66,6 +67,37 @@ function writeStackMetadata( ); } +describe("foreground operation lifecycle", () => { + it("disposes the foreground runtime after a direct readiness timeout", async () => { + let disposeCount = 0; + const operation = Promise.reject( + new StackReadinessError({ + target: "stack", + timeoutMs: 10, + detail: "Timed out waiting for stack readiness", + }), + ); + + await expect( + runForegroundOperation(operation, async () => { + disposeCount += 1; + }), + ).rejects.toMatchObject({ code: "STACK_READINESS_TIMEOUT" }); + expect(disposeCount).toBe(1); + }); + + it("keeps the foreground runtime open after a non-terminal operation failure", async () => { + let disposeCount = 0; + + await expect( + runForegroundOperation(Promise.reject(new Error("failed")), async () => { + disposeCount += 1; + }), + ).rejects.toMatchObject({ code: "UNKNOWN" }); + expect(disposeCount).toBe(0); + }); +}); + describe("createStack types", () => { it("StackHandle interface has expected shape", () => { const check = (_stack: StackHandle) => { @@ -261,7 +293,7 @@ describe("resolveConfig startup mode", () => { describe("resolveConfig readiness policy", () => { it("uses a finite package default", async () => { const config = await resolveConfig(); - expect(config.readiness).toEqual({ mode: "finite", timeoutMs: 120_000 }); + expect(config.readiness).toEqual({ mode: "finite", timeoutMs: 180_000 }); }); it("preserves an explicit infinite policy", async () => { diff --git a/packages/stack/src/daemon.ts b/packages/stack/src/daemon.ts index 30361ecb92..de7671d72e 100644 --- a/packages/stack/src/daemon.ts +++ b/packages/stack/src/daemon.ts @@ -1,5 +1,6 @@ import { Effect, Layer, ManagedRuntime } from "effect"; import { HttpServer } from "effect/unstable/http"; +import { ApiProxy } from "./ApiProxy.ts"; import type { PlatformFactory } from "./createStack.ts"; import { DaemonServer } from "./DaemonServer.ts"; import { PORT_FIELDS, reservePorts, type PortLease } from "./PortAllocator.ts"; @@ -46,7 +47,7 @@ export async function runDaemon( const msg = await waitForMessage(); const { socketPath } = msg; - let appRuntime: ManagedRuntime.ManagedRuntime | undefined; + let appRuntime: ManagedRuntime.ManagedRuntime | undefined; let daemonRuntime: ManagedRuntime.ManagedRuntime | undefined; let portLease: PortLease | undefined; @@ -73,12 +74,14 @@ export async function runDaemon( // Build the app layer (Stack + ApiProxy) const appLayer = foregroundDaemonLayer(config, platformFactory, portLease); - appRuntime = ManagedRuntime.make(appLayer); + const localAppRuntime = ManagedRuntime.make(appLayer); + appRuntime = localAppRuntime; // Build the stack (services are started later via POST /start) - const localStack = await appRuntime.runPromise(Stack); - const info = await appRuntime.runPromise(localStack.getInfo()); - const localStateManager = await appRuntime.runPromise(StateManager); + const localStack = await localAppRuntime.runPromise(Stack); + const apiProxy = await localAppRuntime.runPromise(ApiProxy); + const info = await localAppRuntime.runPromise(localStack.getInfo()); + const localStateManager = await localAppRuntime.runPromise(StateManager); // Build daemon management server on Unix socket const daemonLayer = DaemonServer.layer.pipe( @@ -115,7 +118,15 @@ export async function runDaemon( process.disconnect?.(); const daemon = await daemonRuntime.runPromise(DaemonServer); - await Promise.race([daemonRuntime.runPromise(daemon.awaitShutdown), waitForSignal()]); + // A terminal activation deadline has already disposed the stack's scoped processes and port + // leases. Treat that as a whole-runtime failure: keeping the management or proxy servers alive + // would expose partially disposed state. This path deliberately does not drain unrelated + // in-flight proxy requests; callers should reconnect after starting a fresh daemon. + await Promise.race([ + daemonRuntime.runPromise(daemon.awaitShutdown), + localAppRuntime.runPromise(apiProxy.awaitTerminalFailure), + waitForSignal(), + ]); await shutdownDaemon({ appRuntime, daemonRuntime }); process.exit(0); } catch (err) { @@ -159,7 +170,7 @@ function waitForSignal(): Promise<"SIGINT" | "SIGTERM"> { } async function shutdownDaemon(opts: { - readonly appRuntime?: ManagedRuntime.ManagedRuntime; + readonly appRuntime?: ManagedRuntime.ManagedRuntime; readonly daemonRuntime?: ManagedRuntime.ManagedRuntime; }): Promise { await opts.daemonRuntime?.dispose().catch(() => {}); diff --git a/packages/stack/src/effect.ts b/packages/stack/src/effect.ts index 215e6b6892..e757aaeb58 100644 --- a/packages/stack/src/effect.ts +++ b/packages/stack/src/effect.ts @@ -13,6 +13,7 @@ export { PortConflictError, StackBuildError, StackError, + StackReadinessError, toStackError, } from "./errors.ts"; diff --git a/packages/stack/src/errors.ts b/packages/stack/src/errors.ts index 1da68cdd63..937d3330db 100644 --- a/packages/stack/src/errors.ts +++ b/packages/stack/src/errors.ts @@ -31,6 +31,12 @@ export class StackNotRunningError extends Data.TaggedError("StackNotRunningError readonly phase: string; }> {} +export class StackReadinessError extends Data.TaggedError("StackReadinessError")<{ + readonly target: string; + readonly timeoutMs: number; + readonly detail: string; +}> {} + export class PortConflictError extends Data.TaggedError("PortConflictError")<{ readonly port: number; readonly service: string; @@ -71,6 +77,12 @@ export function toStackError(err: unknown): StackError { message: taggedMessage, cause: err, }); + case "StackReadinessError": + return new StackError({ + code: "STACK_READINESS_TIMEOUT", + message: taggedMessage, + cause: err, + }); case "BinaryNotFoundError": return new StackError({ code: "BINARY_NOT_FOUND", diff --git a/packages/stack/src/layers.ts b/packages/stack/src/layers.ts index 30fb1c3eb4..c0afe22dcb 100644 --- a/packages/stack/src/layers.ts +++ b/packages/stack/src/layers.ts @@ -11,9 +11,8 @@ import type { DaemonMessage, DaemonStartMessage } from "./daemon.ts"; import { DaemonMessageSchema } from "./DaemonProtocol.ts"; import type { PortLease } from "./PortAllocator.ts"; import { RemoteStack } from "./RemoteStack.ts"; -import { StackServiceActivator } from "./ServiceActivation.ts"; import { Stack } from "./Stack.ts"; -import { StackLifecycleCoordinator } from "./StackLifecycleCoordinator.ts"; +import { localStackLayer } from "./LocalStack.ts"; import { StackMetadataPersistence } from "./StackMetadataPersistence.ts"; import { StackPreparation } from "./StackPreparation.ts"; import { @@ -46,7 +45,7 @@ export const foregroundLayer = ( config: ResolvedStackConfig, platformFactory: PlatformFactory, portLease: PortLease, -): Layer.Layer => { +): Layer.Layer => { const platform = platformFactory({ apiPort: config.apiPort, releaseApiPort: portLease.release(["apiPort"]), @@ -56,18 +55,11 @@ export const foregroundLayer = ( Layer.provide(FetchHttpClient.layer), ); const stackPreparationLayer = StackPreparation.layer.pipe(Layer.provide(binaryResolverLayer)); - const coordinatorLayer = StackLifecycleCoordinator.layer(config, portLease).pipe( + const stackLayer = localStackLayer(config, portLease).pipe( Layer.provide(StackBuilder.layer), Layer.provide(stackPreparationLayer), Layer.provide(StackMetadataPersistence.noop), ); - const stackLayer = Stack.layer(config); - const serviceActivatorLayer = Layer.effect( - StackServiceActivator, - Effect.map(StackLifecycleCoordinator, (coordinator) => ({ - activate: coordinator.activateService, - })), - ); const proxyConfig: ProxyConfig = { listenPort: config.apiPort, @@ -88,14 +80,10 @@ export const foregroundLayer = ( }; const apiProxyLayer = ApiProxy.layer(proxyConfig).pipe( Layer.provide(FetchHttpClient.layer), - Layer.provide(serviceActivatorLayer), + Layer.provide(stackLayer), ); - return Layer.mergeAll(stackLayer, apiProxyLayer).pipe( - Layer.provide(coordinatorLayer), - Layer.provide(platform), - Layer.orDie, - ); + return Layer.mergeAll(stackLayer, apiProxyLayer).pipe(Layer.provide(platform), Layer.orDie); }; // --------------------------------------------------------------------------- @@ -114,7 +102,7 @@ export const foregroundDaemonLayer = ( config: ResolvedDaemonConfig, platformFactory: PlatformFactory, portLease: PortLease, -): Layer.Layer => { +): Layer.Layer => { const platform = platformFactory({ apiPort: config.apiPort, releaseApiPort: portLease.release(["apiPort"]), @@ -140,16 +128,6 @@ export const foregroundDaemonLayer = ( anonJwt: config.anonJwt, serviceRoleJwt: config.serviceRoleJwt, }; - const serviceActivatorLayer = Layer.effect( - StackServiceActivator, - Effect.map(StackLifecycleCoordinator, (coordinator) => ({ - activate: coordinator.activateService, - })), - ); - const apiProxyLayer = ApiProxy.layer(proxyConfig).pipe( - Layer.provide(FetchHttpClient.layer), - Layer.provide(serviceActivatorLayer), - ); const stateManagerLayer = StateManager.make( singleStackStateManagerPaths(config.stackRoot, config.runtimeRoot, config.name), ); @@ -157,15 +135,17 @@ export const foregroundDaemonLayer = ( const metadataPersistenceLayer = StackMetadataPersistence.fromStateManager(config.name).pipe( Layer.provide(stateManagerLayer), ); - const coordinatorLayer = StackLifecycleCoordinator.layer(config, portLease).pipe( + const stackLayer = localStackLayer(config, portLease).pipe( Layer.provide(StackBuilder.layer), Layer.provide(stackPreparationLayer), Layer.provide(metadataPersistenceLayer), ); - const stackLayer = Stack.layer(config); + const apiProxyLayer = ApiProxy.layer(proxyConfig).pipe( + Layer.provide(FetchHttpClient.layer), + Layer.provide(stackLayer), + ); return Layer.mergeAll(stackLayer, apiProxyLayer, stateManagerLayer).pipe( - Layer.provide(coordinatorLayer), Layer.provide(platform), Layer.orDie, ); diff --git a/packages/stack/src/services/health-budgets.ts b/packages/stack/src/services/health-budgets.ts index 7c623bfb6d..3cf965441f 100644 --- a/packages/stack/src/services/health-budgets.ts +++ b/packages/stack/src/services/health-budgets.ts @@ -1,5 +1,5 @@ import { defaults, type HealthCheckConfig } from "@supabase/process-compose"; -import type { ServiceName } from "../versions.ts"; +import type { ServiceName } from "../ServiceName.ts"; type HealthBudget = Required< Pick< diff --git a/packages/stack/tests/createStack-docker.e2e.test.ts b/packages/stack/tests/createStack-docker.e2e.test.ts index f957492e72..8f81ad820e 100644 --- a/packages/stack/tests/createStack-docker.e2e.test.ts +++ b/packages/stack/tests/createStack-docker.e2e.test.ts @@ -101,11 +101,8 @@ dockerDescribe("createStack e2e (docker mode)", () => { async () => { const functionsRes = await fetch(`${stack.url}/functions/v1/test`); await stack.serviceReady("edge-runtime"); - - const [runningImages, states] = await Promise.all([ - Promise.resolve(execSync("docker ps --format '{{.Image}}'").toString()), - stack.getStatus(), - ]); + const runningImages = execSync("docker ps --format '{{.Image}}'").toString(); + const states = await stack.getStatus(); expect(runningImages).toContain("supabase/edge-runtime"); expect(states).toEqual(