diff --git a/docs/adr/0017-simplified-managed-stack-architecture.md b/docs/adr/0017-simplified-managed-stack-architecture.md index c2655382f7..7940140c3b 100644 --- a/docs/adr/0017-simplified-managed-stack-architecture.md +++ b/docs/adr/0017-simplified-managed-stack-architecture.md @@ -177,6 +177,17 @@ background task for callers that want full lazy preparation. Eager capability activation remains independent of this setting. That runtime session owns the shared preparation fibers. +Lazy traffic capabilities may retire independently after their configured idle timeout (see the +package [traffic stopping documentation](../../packages/stack/README.md)). The traffic policy is +per capability: supported lazy capabilities default to 60 seconds, `false` disables retirement, +eager capabilities never retire, and unsupported capabilities remain opted out. Activity covers +in-flight HTTP requests and streams plus open WebSocket or TCP connections; idle HTTP keep-alive +sockets do not count. The Supervisor serializes workload retirement with activation and lifecycle +operations, and fences new requests for the retiring capability. Unrelated ingress continues; +dependencies stay running while their dependants are active, and Studio with `pg-meta` retires +together. Retained data and listeners let the next request wake the capability and restart its +workloads. + In both modes, lazy activation prepares the requested dependency closure with bounded concurrency before starting its workloads. Explicit `stack.prepare(...)` remains available as a cache-only warmup for callers that @@ -268,7 +279,7 @@ Tests follow consumed boundaries: - stack integration covers identity, sticky ports, durable lifecycle, ownership, stale-owner recovery, and interrupted cleanup; - supervisor integration covers detached ownership, RPC, stop, and - destroy; and + destroy; dedicated idle-stop integration covers retirement and wake-up; and - one shared stack-package E2E journey runs in native and Docker modes, starts with PostgreSQL alone, activates every other service through realistic traffic, verifies cross-service behavior, then exercises diff --git a/packages/stack/README.md b/packages/stack/README.md index e4e0f98345..f725472583 100644 --- a/packages/stack/README.md +++ b/packages/stack/README.md @@ -24,6 +24,29 @@ capability is lazy. Starting the stack therefore launches only PostgreSQL by default; capabilities configured as eager join its startup dependency closure. The remaining lazy capabilities activate through the stack's listeners on demand for the current running session. + +Lazy REST, Auth, Realtime, Studio, and pooler capabilities stop after 60 seconds without traffic +by default. Traffic means an active request or stream; idle HTTP keep-alive sockets do not keep a +service running, while open WebSocket or TCP connections do. Configure a different positive +timeout, or disable traffic stopping for a capability, with `idleTimeoutSeconds`: + +```ts +await stack.start({ + config: { + capabilities: { + rest: { idleTimeoutSeconds: 120 }, + realtime: { idleTimeoutSeconds: false }, + }, + }, +}); +``` + +Eager capabilities never auto-stop. PostgreSQL, Storage, Functions, Mail, and Analytics do not opt +into traffic stopping yet. Studio and its `pg-meta` companion are stopped and started together. +Dependency protection keeps required dependencies available while a capability is running. +Stopping preserves listeners and data, and the next request wakes the lazy capability and restarts +its workloads. + Native workloads have a two-minute readiness budget to allow cold starts to load shared libraries; container workloads retain a 30-second budget, and PostgreSQL uses its configured `health_timeout`. Each readiness probe returns immediately when its endpoint becomes healthy. diff --git a/packages/stack/src/gateway/ActivityTracker.ts b/packages/stack/src/gateway/ActivityTracker.ts new file mode 100644 index 0000000000..fb5f9fccc6 --- /dev/null +++ b/packages/stack/src/gateway/ActivityTracker.ts @@ -0,0 +1,28 @@ +import { Effect } from "effect"; +import type { CapabilityName } from "../public/Capability.ts"; + +/** Tracks gateway work that must keep a lazy capability running. */ +export interface GatewayActivity { + readonly track: ( + capability: CapabilityName, + effect: Effect.Effect, + ) => Effect.Effect; +} + +export interface GatewayActivityCallbacks { + readonly begin: (capability: CapabilityName) => Effect.Effect; + readonly end: (capability: CapabilityName) => Effect.Effect; +} + +/** Adapts gateway lifetimes to the Supervisor-owned traffic controller. */ +export const makeGatewayActivity = ( + callbacks: GatewayActivityCallbacks, +): Effect.Effect => + Effect.succeed({ + track: (capability, effect) => + Effect.acquireUseRelease( + callbacks.begin(capability), + () => effect, + () => callbacks.end(capability), + ), + } satisfies GatewayActivity); diff --git a/packages/stack/src/gateway/Gateway.ts b/packages/stack/src/gateway/Gateway.ts index c7679b6317..3932d35db8 100644 --- a/packages/stack/src/gateway/Gateway.ts +++ b/packages/stack/src/gateway/Gateway.ts @@ -4,6 +4,7 @@ import type { CapabilityName } from "../public/Capability.ts"; import type { PortField } from "../public/Status.ts"; import { makeHttpGateway, type HttpGateway, type HttpGatewayOptions } from "./HttpGateway.ts"; import { makeTcpGateway, type TcpGateway, type TcpGatewayOptions } from "./TcpGateway.ts"; +import type { GatewayActivity } from "./ActivityTracker.ts"; /** A private backend endpoint returned only after activation has completed. */ export interface BackendEndpoint { @@ -112,6 +113,8 @@ export interface StackGatewayOptions { readonly activate: ( capability: CapabilityName, ) => Effect.Effect; + /** Optional traffic lease shared by the supervisor's idle-stop controller. */ + readonly activity?: GatewayActivity; } /** Compose the protocol gateways under one Supervisor-owned lifecycle scope. */ @@ -135,7 +138,11 @@ export const makeGateway = ( }); } const acquired = yield* Effect.exit( - makeHttpGateway({ ...entry.options, activate: options.activate }), + makeHttpGateway({ + ...entry.options, + activate: options.activate, + activity: options.activity, + }), ); if (Exit.isFailure(acquired)) { yield* closeValues(http.values()); @@ -151,7 +158,11 @@ export const makeGateway = ( }); } const acquired = yield* Effect.exit( - makeTcpGateway({ ...entry.options, activate: options.activate }), + makeTcpGateway({ + ...entry.options, + activate: options.activate, + activity: options.activity, + }), ); if (Exit.isFailure(acquired)) { yield* closeValues([...http.values(), ...tcp.values()]); diff --git a/packages/stack/src/gateway/HttpGateway.ts b/packages/stack/src/gateway/HttpGateway.ts index c3fffeb5c0..97eb0c91fb 100644 --- a/packages/stack/src/gateway/HttpGateway.ts +++ b/packages/stack/src/gateway/HttpGateway.ts @@ -23,6 +23,7 @@ import type { PreparedGatewayRoute, } from "./Gateway.ts"; import { GatewayRouteNotFoundError, isGatewayProxyRoute } from "./Gateway.ts"; +import type { GatewayActivity } from "./ActivityTracker.ts"; import type { HostListener, HostListenerHttpEvent, @@ -48,6 +49,7 @@ export interface HttpGatewayOptions { ) => Effect.Effect; readonly cors?: Readonly>; readonly healthPaths?: ReadonlyArray; + readonly activity?: GatewayActivity; } export interface HttpGateway { @@ -231,9 +233,9 @@ const proxy = ( const onIncomingError = (cause: Error) => { if (settled) return; settled = true; - cleanup(); outgoing?.destroy(); incoming?.destroy(); + cleanup(); resume(Effect.fail(new GatewayBackendError({ cause }))); }; const onIncomingAborted = () => onIncomingError(new Error("backend response aborted")); @@ -250,7 +252,6 @@ const proxy = ( onIncomingError(new Error("response closed")); }; const cleanup = () => { - if (outgoing !== undefined) outgoing.off("error", onOutgoingError); request.off("aborted", onRequestAborted); response.off("close", onResponseClose); if (incoming !== undefined) { @@ -295,9 +296,10 @@ const proxy = ( response.once("close", onResponseClose); request.pipe(outgoing); return Effect.sync(() => { - cleanup(); + settled = true; outgoing.destroy(); incoming?.destroy(); + cleanup(); }); }); @@ -381,9 +383,10 @@ const handleRequest = ( proxy(request, response, backend, options, path, headers), ), ); + const tracked = options.activity?.track(route.capability, activation) ?? activation; // Node invokes this handler outside Effect; use the owner-scoped FiberSet // runtime so cancellation of the gateway interrupts in-flight activation. - const fiber = runFork(activation); + const fiber = runFork(tracked); cancelOnRequestClose(request, response, fiber); fiber.addObserver((exit) => { if (!Exit.isFailure(exit) || response.writableEnded || response.destroyed) return; @@ -515,6 +518,7 @@ const handleUpgrade = ( }), ), ); + const tracked = options.activity?.track(route.capability, activation) ?? activation; const onSocketClose = () => fiber.interruptUnsafe(); const onSocketEnd = () => fiber.interruptUnsafe(); const onSocketError = () => fiber.interruptUnsafe(); @@ -523,7 +527,7 @@ const handleUpgrade = ( socket.off("end", onSocketEnd); request.off("aborted", onRequestAborted); }; - const fiber = runFork(activation); + const fiber = runFork(tracked); socket.once("close", onSocketClose); socket.once("end", onSocketEnd); socket.once("error", onSocketError); diff --git a/packages/stack/src/gateway/TcpGateway.ts b/packages/stack/src/gateway/TcpGateway.ts index aa674d897f..0280af78b0 100644 --- a/packages/stack/src/gateway/TcpGateway.ts +++ b/packages/stack/src/gateway/TcpGateway.ts @@ -11,6 +11,7 @@ import type { GatewayRouteRequest, } from "./Gateway.ts"; import type { HostListener } from "../supervisor/HostListener.ts"; +import type { GatewayActivity } from "./ActivityTracker.ts"; export interface TcpGatewayOptions { readonly address?: string; @@ -25,6 +26,7 @@ export interface TcpGatewayOptions { request: GatewayRouteRequest, activation: ActivationResult, ) => Effect.Effect; + readonly activity?: GatewayActivity; } export interface TcpGateway { @@ -140,10 +142,11 @@ const handleConnection = ( return tunnel(source, backend); }), ); + const tracked = options.activity?.track(route.capability, operation) ?? operation; source.once("error", onPreActivationError); // Node invokes this handler outside Effect; use the owner-scoped FiberSet // runtime for the exact accepted connection's lifecycle. - const fiber = runFork(operation); + const fiber = runFork(tracked); const onSourceClose = () => fiber.interruptUnsafe(); source.once("close", onSourceClose); fiber.addObserver(() => { diff --git a/packages/stack/src/gateway/activity.integration.test.ts b/packages/stack/src/gateway/activity.integration.test.ts new file mode 100644 index 0000000000..ac609f0514 --- /dev/null +++ b/packages/stack/src/gateway/activity.integration.test.ts @@ -0,0 +1,337 @@ +import { NodeServices } from "@effect/platform-node"; +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Option, Queue, Ref } from "effect"; +// oxlint-disable-next-line effecttsgo/node-builtin-import -- The integration fixture needs a real Node HTTP client. +import { Agent, createServer as createHttpServer, request as httpRequest } from "node:http"; +import { connect as tcpConnect, createServer as createTcpServer, type Socket } from "node:net"; +import { makeGatewayActivity } from "./ActivityTracker.ts"; +import { makeHttpGateway } from "./HttpGateway.ts"; +import { makeTcpGateway } from "./TcpGateway.ts"; + +const run = (effect: Effect.Effect) => + Effect.scoped(effect).pipe(Effect.provide(NodeServices.layer)); + +const route = { + capability: "rest" as const, + match: (request: { path: string }) => request.path === "/data", +}; + +const tcpRoute = { + capability: "rest" as const, + match: () => true, +}; + +const activityFixture = Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const active = yield* Ref.make(0); + const activity = yield* makeGatewayActivity({ + begin: (capability) => + Ref.update(active, (value) => value + 1).pipe( + Effect.andThen(Queue.offer(events, `begin:${capability}`)), + ), + end: (capability) => + Ref.update(active, (value) => value - 1).pipe( + Effect.andThen(Queue.offer(events, `end:${capability}`)), + ), + }); + return { active, activity, events }; +}); + +const nextEvent = (events: Queue.Queue) => Queue.take(events); + +describe("gateway traffic activity", () => { + it.live("tracks HTTP requests through streamed response completion", () => + run( + Effect.gen(function* () { + const { activity, active, events } = yield* activityFixture; + let finishResponse: (() => void) | undefined; + const backend = createHttpServer((_request, response) => { + response.write("part"); + finishResponse = () => response.end("done"); + }); + yield* Effect.callback((resume) => { + backend.once("error", (error) => resume(Effect.fail(error))); + backend.listen(0, "127.0.0.1", () => resume(Effect.void)); + return Effect.sync(() => backend.close()); + }); + yield* Effect.addFinalizer(() => + Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }), + ); + const address = backend.address(); + if (typeof address !== "object" || address === null) + return yield* Effect.die("backend unavailable"); + const gateway = yield* makeHttpGateway({ + address: "127.0.0.1", + port: 0, + routes: [route], + activity, + activate: () => + Effect.succeed({ + capability: "rest" as const, + endpoint: { host: "127.0.0.1", port: address.port }, + }), + }); + const response = yield* Effect.callback< + { first: string; done: Effect.Effect }, + Error + >((resume) => { + const request = httpRequest( + { host: "127.0.0.1", port: gateway.port, path: "/data" }, + (incoming) => { + incoming.once("data", (chunk) => + resume( + Effect.succeed({ + first: String(chunk), + done: Effect.callback((done) => { + incoming.once("end", () => done(Effect.void)); + return Effect.sync(() => incoming.destroy()); + }), + }), + ), + ); + }, + ); + request.once("error", (error) => resume(Effect.fail(error))); + request.end(); + return Effect.sync(() => request.destroy()); + }); + expect(yield* nextEvent(events)).toBe("begin:rest"); + expect(yield* Ref.get(active)).toBe(1); + expect(response.first).toBe("part"); + finishResponse?.(); + yield* response.done; + expect(yield* nextEvent(events)).toBe("end:rest"); + expect(yield* Ref.get(active)).toBe(0); + yield* gateway.close; + yield* Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }); + }), + ), + ); + + it.live("does not track OPTIONS or local health responses", () => + run( + Effect.gen(function* () { + const { activity, active, events } = yield* activityFixture; + const gateway = yield* makeHttpGateway({ + address: "127.0.0.1", + port: 0, + healthPaths: ["/health"], + routes: [route], + activity, + activate: () => Effect.die("should not activate"), + }); + const request = (method: string, path: string) => + Effect.callback((resume) => { + const client = httpRequest( + { host: "127.0.0.1", port: gateway.port, path, method }, + (response) => { + response.resume(); + response.once("end", () => resume(Effect.succeed(response.statusCode ?? 0))); + }, + ); + client.once("error", (error) => resume(Effect.fail(error))); + client.end(); + return Effect.sync(() => client.destroy()); + }); + expect(yield* request("OPTIONS", "/data")).toBe(204); + expect(yield* request("GET", "/health")).toBe(200); + expect(Option.isNone(yield* Queue.poll(events))).toBe(true); + expect(yield* Ref.get(active)).toBe(0); + yield* gateway.close; + }), + ), + ); + + it.live("holds silent TCP activity until the client connection closes", () => + run( + Effect.gen(function* () { + const { activity, active, events } = yield* activityFixture; + const backend = createTcpServer(() => undefined); + yield* Effect.callback((resume) => { + backend.once("error", (error) => resume(Effect.fail(error))); + backend.listen(0, "127.0.0.1", () => resume(Effect.void)); + return Effect.sync(() => backend.close()); + }); + yield* Effect.addFinalizer(() => + Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }), + ); + const address = backend.address(); + if (typeof address !== "object" || address === null) + return yield* Effect.die("backend unavailable"); + const gateway = yield* makeTcpGateway({ + address: "127.0.0.1", + port: 0, + routes: [tcpRoute], + activity, + activate: () => + Effect.succeed({ + capability: "rest" as const, + endpoint: { host: "127.0.0.1", port: address.port }, + }), + }); + const client = yield* Effect.callback((resume) => { + const socket = tcpConnect(gateway.port, "127.0.0.1", () => + resume(Effect.succeed(socket)), + ); + socket.once("error", (error) => resume(Effect.fail(error))); + return Effect.sync(() => socket.destroy()); + }); + yield* Effect.addFinalizer(() => Effect.sync(() => client.destroy())); + expect(yield* nextEvent(events)).toBe("begin:rest"); + expect(yield* Ref.get(active)).toBe(1); + expect(Option.isNone(yield* Queue.poll(events))).toBe(true); + client.destroy(); + expect(yield* nextEvent(events)).toBe("end:rest"); + expect(yield* Ref.get(active)).toBe(0); + yield* gateway.close; + yield* Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }); + }), + ), + ); + + it.live("holds a silent WebSocket upgrade until its socket closes", () => + run( + Effect.gen(function* () { + const { activity, active, events } = yield* activityFixture; + const backend = createHttpServer(); + backend.on("upgrade", (_request, socket) => + socket.write( + "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n", + ), + ); + yield* Effect.callback((resume) => { + backend.once("error", (error) => resume(Effect.fail(error))); + backend.listen(0, "127.0.0.1", () => resume(Effect.void)); + return Effect.sync(() => backend.close()); + }); + yield* Effect.addFinalizer(() => + Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }), + ); + const address = backend.address(); + if (typeof address !== "object" || address === null) + return yield* Effect.die("backend unavailable"); + const gateway = yield* makeHttpGateway({ + address: "127.0.0.1", + port: 0, + routes: [{ capability: "rest" as const, match: () => true }], + activity, + activate: () => + Effect.succeed({ + capability: "rest" as const, + endpoint: { host: "127.0.0.1", port: address.port }, + }), + }); + const client = yield* Effect.callback((resume) => { + const socket = tcpConnect(gateway.port, "127.0.0.1", () => { + socket.write( + "GET /socket HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n", + ); + resume(Effect.succeed(socket)); + }); + socket.once("error", (error) => resume(Effect.fail(error))); + return Effect.sync(() => socket.destroy()); + }); + yield* Effect.addFinalizer(() => Effect.sync(() => client.destroy())); + expect(yield* nextEvent(events)).toBe("begin:rest"); + expect(yield* Ref.get(active)).toBe(1); + expect(Option.isNone(yield* Queue.poll(events))).toBe(true); + client.destroy(); + expect(yield* nextEvent(events)).toBe("end:rest"); + expect(yield* Ref.get(active)).toBe(0); + yield* gateway.close; + yield* Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }); + }), + ), + ); + + it.live("releases activity when an HTTP client aborts and ignores keep-alive sockets", () => + run( + Effect.gen(function* () { + const { activity, active, events } = yield* activityFixture; + let requestCount = 0; + let responseToFinish: (() => void) | undefined; + let requestObserved = false; + let requestObservedResume: (() => void) | undefined; + const backend = createHttpServer((_request, response) => { + requestCount += 1; + requestObserved = true; + requestObservedResume?.(); + if (requestCount > 1) response.end("done"); + else responseToFinish = () => response.end("done"); + }); + yield* Effect.callback((resume) => { + backend.once("error", (error) => resume(Effect.fail(error))); + backend.listen(0, "127.0.0.1", () => resume(Effect.void)); + return Effect.sync(() => backend.close()); + }); + yield* Effect.addFinalizer(() => + Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }), + ); + const address = backend.address(); + if (typeof address !== "object" || address === null) + return yield* Effect.die("backend unavailable"); + const gateway = yield* makeHttpGateway({ + address: "127.0.0.1", + port: 0, + routes: [route], + activity, + activate: () => + Effect.succeed({ + capability: "rest" as const, + endpoint: { host: "127.0.0.1", port: address.port }, + }), + }); + const client = httpRequest({ host: "127.0.0.1", port: gateway.port, path: "/data" }); + client.on("error", () => undefined); + yield* Effect.addFinalizer(() => Effect.sync(() => client.destroy())); + client.end(); + expect(yield* nextEvent(events)).toBe("begin:rest"); + if (!requestObserved) + yield* Effect.callback((resume) => { + requestObservedResume = () => resume(Effect.void); + return Effect.sync(() => (requestObservedResume = undefined)); + }); + client.destroy(); + expect(yield* nextEvent(events)).toBe("end:rest"); + expect(yield* Ref.get(active)).toBe(0); + responseToFinish?.(); + const agent = new Agent({ keepAlive: true }); + yield* Effect.addFinalizer(() => Effect.sync(() => agent.destroy())); + yield* Effect.callback((resume) => { + const keepAliveRequest = httpRequest( + { host: "127.0.0.1", port: gateway.port, path: "/data", agent }, + (response) => { + response.resume(); + response.once("end", () => resume(Effect.void)); + }, + ); + keepAliveRequest.on("error", (error) => resume(Effect.fail(error))); + keepAliveRequest.end(); + return Effect.sync(() => keepAliveRequest.destroy()); + }); + expect(yield* nextEvent(events)).toBe("begin:rest"); + expect(yield* nextEvent(events)).toBe("end:rest"); + expect(yield* Ref.get(active)).toBe(0); + agent.destroy(); + yield* gateway.close; + yield* Effect.callback((resume) => { + backend.close(() => resume(Effect.void)); + }); + }), + ), + ); +}); diff --git a/packages/stack/src/model/CapabilityModule.ts b/packages/stack/src/model/CapabilityModule.ts index 4b79fce0e2..c9d04e49bf 100644 --- a/packages/stack/src/model/CapabilityModule.ts +++ b/packages/stack/src/model/CapabilityModule.ts @@ -49,6 +49,8 @@ export interface CapabilityModule { readonly defaultSettings: Settings; readonly defaultEnabled: boolean; readonly defaultActivation: ActivationMode; + /** Default traffic idle timeout; false means this capability is not auto-stopped. */ + readonly defaultIdleTimeoutSeconds?: number | false; readonly dependencies: ReadonlyArray; readonly defaultVersion: string; /** Release selectors are the only source of workload versions and artifacts. */ diff --git a/packages/stack/src/model/Compiler.ts b/packages/stack/src/model/Compiler.ts index 9e2dcdd87a..cfcf2b2d8b 100644 --- a/packages/stack/src/model/Compiler.ts +++ b/packages/stack/src/model/Compiler.ts @@ -2,7 +2,7 @@ import { Duration, Effect, Path, Redacted, Schema } from "effect"; import { InvalidStackConfigError, StackVersionUnsupportedError } from "../public/Errors.ts"; import { StackConfigSchema, type StackConfig, type PreparationMode } from "../public/Config.ts"; import type { JwtSigning } from "../public/Config.ts"; -import type { CapabilityName } from "../public/Capability.ts"; +import { CAPABILITY_NAMES, type CapabilityName } from "../public/Capability.ts"; import type { PortField } from "../public/Status.ts"; import type { StackRuntime } from "../public/Runtime.ts"; import { @@ -476,14 +476,28 @@ const releaseFor = ( const enabledSettings = ( name: CapabilityName, raw: unknown, -): { enabled: boolean; activation: "eager" | "lazy"; settings: unknown; raw: unknown } => { +): { + enabled: boolean; + activation: "eager" | "lazy"; + idleTimeoutSeconds: number | false; + settings: unknown; + raw: unknown; +} => { + const defaultIdleTimeout = CAPABILITY_MODULES[name].defaultIdleTimeoutSeconds ?? false; if (name === "database") - return { enabled: true, activation: "eager", settings: extract(raw, "settings") ?? {}, raw }; + return { + enabled: true, + activation: "eager", + idleTimeoutSeconds: false, + settings: extract(raw, "settings") ?? {}, + raw, + }; if (raw === undefined || raw === null) { const module = CAPABILITY_MODULES[name]; return { enabled: module.defaultEnabled, activation: module.defaultActivation, + idleTimeoutSeconds: module.defaultActivation === "lazy" ? defaultIdleTimeout : false, settings: module.defaultSettings, raw: {}, }; @@ -492,21 +506,60 @@ const enabledSettings = ( return { enabled: false, activation: CAPABILITY_MODULES[name].defaultActivation, + idleTimeoutSeconds: false, settings: CAPABILITY_MODULES[name].defaultSettings, raw, }; const activation = extract(raw, "activation"); + const idleTimeoutSeconds = extract(raw, "idleTimeoutSeconds"); + const selectedActivation = + activation === "eager" || activation === "lazy" + ? activation + : CAPABILITY_MODULES[name].defaultActivation; return { enabled: true, - activation: - activation === "eager" || activation === "lazy" - ? activation - : CAPABILITY_MODULES[name].defaultActivation, + activation: selectedActivation, + idleTimeoutSeconds: + selectedActivation === "eager" + ? false + : idleTimeoutSeconds === false + ? false + : defaultIdleTimeout === false + ? false + : typeof idleTimeoutSeconds === "number" && + Number.isFinite(idleTimeoutSeconds) && + idleTimeoutSeconds > 0 + ? idleTimeoutSeconds + : defaultIdleTimeout, settings: extract(raw, "settings") ?? {}, raw, }; }; +const validateIdleTimeouts = ( + config: StackConfig, +): Effect.Effect => { + const capabilities = config.capabilities; + if (capabilities === undefined) return Effect.void; + for (const name of CAPABILITY_NAMES) { + const raw = capabilities[name]; + if (!isRecord(raw)) continue; + const timeout = "idleTimeoutSeconds" in raw ? raw.idleTimeoutSeconds : undefined; + if ( + typeof timeout === "number" && + (CAPABILITY_MODULES[name].defaultIdleTimeoutSeconds === undefined || + CAPABILITY_MODULES[name].defaultIdleTimeoutSeconds === false) + ) + return Effect.fail( + new InvalidStackConfigError({ + message: `Invalid ${name} idleTimeoutSeconds: capability does not support idle stopping`, + setting: `capabilities.${name}.idleTimeoutSeconds`, + }), + ); + } + return Effect.void; +}; + const materializeCapability = ( module: CapabilityModule, raw: unknown, @@ -535,6 +588,7 @@ const materializeCapability = ( return { enabled: selected.enabled, activation: selected.activation, + idleTimeoutSeconds: selected.idleTimeoutSeconds, version, settings: completeSettings, }; @@ -566,6 +620,7 @@ export const compileStack = ( const path = yield* Path.Path; yield* validateFunctionKeys(input.config ?? {}); const config = yield* decodeConfig(input.config ?? {}); + yield* validateIdleTimeouts(config); yield* validateDatabaseHealthTimeout(config); yield* validatePoolerKeys(config); yield* validateStorageFileSizes(config); diff --git a/packages/stack/src/model/ExecutionPlan.ts b/packages/stack/src/model/ExecutionPlan.ts index d868114440..4c94ffc051 100644 --- a/packages/stack/src/model/ExecutionPlan.ts +++ b/packages/stack/src/model/ExecutionPlan.ts @@ -77,6 +77,7 @@ export interface ExecutionPlan { export interface MaterializedCapability { readonly enabled: boolean; readonly activation: "eager" | "lazy"; + readonly idleTimeoutSeconds: number | false; readonly version: string; readonly settings: MaterializedSettings; } diff --git a/packages/stack/src/model/capabilities/auth.ts b/packages/stack/src/model/capabilities/auth.ts index 1b0758a6d5..567f399203 100644 --- a/packages/stack/src/model/capabilities/auth.ts +++ b/packages/stack/src/model/capabilities/auth.ts @@ -395,6 +395,7 @@ export const AuthModule: CapabilityModule = { }, defaultEnabled: true, defaultActivation: "lazy", + defaultIdleTimeoutSeconds: 60, defaultVersion: version, dependencies: ["database"], releases: { diff --git a/packages/stack/src/model/capabilities/pooler.ts b/packages/stack/src/model/capabilities/pooler.ts index 7bef548b4d..77668aec38 100644 --- a/packages/stack/src/model/capabilities/pooler.ts +++ b/packages/stack/src/model/capabilities/pooler.ts @@ -27,6 +27,7 @@ export const PoolerModule: CapabilityModule = { }, defaultEnabled: true, defaultActivation: "lazy", + defaultIdleTimeoutSeconds: 60, defaultVersion: version, dependencies: ["database"], releases: { diff --git a/packages/stack/src/model/capabilities/realtime.ts b/packages/stack/src/model/capabilities/realtime.ts index 142151f78b..29903a6458 100644 --- a/packages/stack/src/model/capabilities/realtime.ts +++ b/packages/stack/src/model/capabilities/realtime.ts @@ -25,6 +25,7 @@ export const RealtimeModule: CapabilityModule = { }, defaultEnabled: true, defaultActivation: "lazy", + defaultIdleTimeoutSeconds: 60, defaultVersion: version, dependencies: ["database"], releases: { diff --git a/packages/stack/src/model/capabilities/rest.ts b/packages/stack/src/model/capabilities/rest.ts index c368e2b567..f15f4c3f03 100644 --- a/packages/stack/src/model/capabilities/rest.ts +++ b/packages/stack/src/model/capabilities/rest.ts @@ -23,6 +23,7 @@ export const RestModule: CapabilityModule = { }, defaultEnabled: true, defaultActivation: "lazy", + defaultIdleTimeoutSeconds: 60, defaultVersion: version, dependencies: ["database"], releases: { diff --git a/packages/stack/src/model/capabilities/studio.ts b/packages/stack/src/model/capabilities/studio.ts index bbc3abce39..d0f55802c0 100644 --- a/packages/stack/src/model/capabilities/studio.ts +++ b/packages/stack/src/model/capabilities/studio.ts @@ -16,6 +16,7 @@ export const StudioModule: CapabilityModule = { defaultSettings: { api_url: "", openai_api_key: undefined }, defaultEnabled: true, defaultActivation: "lazy", + defaultIdleTimeoutSeconds: 60, defaultVersion: version, dependencies: ["rest", "analytics"], releases: { diff --git a/packages/stack/src/model/idle-config.integration.test.ts b/packages/stack/src/model/idle-config.integration.test.ts new file mode 100644 index 0000000000..2ec057019a --- /dev/null +++ b/packages/stack/src/model/idle-config.integration.test.ts @@ -0,0 +1,97 @@ +import { NodeServices } from "@effect/platform-node"; +import { describe, expect, it } from "@effect/vitest"; +import { Cause, Effect, Exit, Option } from "effect"; +import { InvalidStackConfigError } from "../public/Errors.ts"; +import { compileStack } from "./Compiler.ts"; + +const compile = (config: Parameters[0]["config"]) => + compileStack({ projectRoot: "/tmp/supabase-project", runtime: { kind: "native" }, config }).pipe( + Effect.provide(NodeServices.layer), + ); + +const failureOf = (exit: Exit.Exit): E | undefined => + Exit.isFailure(exit) ? Option.getOrUndefined(Cause.findErrorOption(exit.cause)) : undefined; + +describe("traffic idle timeout configuration", () => { + it.live("defaults supported lazy capabilities to a 60 second timeout", () => + Effect.gen(function* () { + const result = yield* compile({}); + for (const name of ["rest", "auth", "realtime", "studio", "pooler"] as const) { + expect(result.definition.capabilities[name]).toMatchObject({ + activation: "lazy", + idleTimeoutSeconds: 60, + }); + } + }), + ); + + it.live("allows disabling idle stopping explicitly", () => + Effect.gen(function* () { + const result = yield* compile({ + capabilities: { + rest: { idleTimeoutSeconds: false }, + auth: { idleTimeoutSeconds: false }, + realtime: { idleTimeoutSeconds: false }, + studio: { idleTimeoutSeconds: false }, + pooler: { idleTimeoutSeconds: false }, + }, + }); + for (const name of ["rest", "auth", "realtime", "studio", "pooler"] as const) + expect(result.definition.capabilities[name].idleTimeoutSeconds).toBe(false); + }), + ); + + it.live("persists a positive finite timeout override", () => + Effect.gen(function* () { + const result = yield* compile({ + capabilities: { rest: { idleTimeoutSeconds: 15.5 } }, + }); + expect(result.definition.capabilities.rest.idleTimeoutSeconds).toBe(15.5); + }), + ); + + it.live("does not idle-stop eager capabilities", () => + Effect.gen(function* () { + const result = yield* compile({ + capabilities: { + rest: { activation: "eager", idleTimeoutSeconds: 15 }, + studio: { activation: "eager" }, + }, + }); + expect(result.definition.capabilities.rest).toMatchObject({ + activation: "eager", + idleTimeoutSeconds: false, + }); + expect(result.definition.capabilities.studio).toMatchObject({ + activation: "eager", + idleTimeoutSeconds: false, + }); + }), + ); + + it.live("keeps unsupported services opted out of idle stopping", () => + Effect.gen(function* () { + const defaults = yield* compile({}); + for (const name of ["database", "storage", "functions", "mail", "analytics"] as const) + expect(defaults.definition.capabilities[name].idleTimeoutSeconds).toBe(false); + + for (const name of ["database", "storage", "functions", "mail", "analytics"] as const) { + const result = yield* compile({ + capabilities: { [name]: { idleTimeoutSeconds: 30 } }, + } as never).pipe(Effect.exit); + expect(failureOf(result)).toBeInstanceOf(InvalidStackConfigError); + } + }), + ); + + it.live("rejects zero, negative, and non-finite timeouts", () => + Effect.gen(function* () { + for (const value of [0, -1, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY, Number.NaN]) { + const result = yield* compile({ + capabilities: { rest: { idleTimeoutSeconds: value } }, + } as never).pipe(Effect.exit); + expect(failureOf(result)).toBeInstanceOf(InvalidStackConfigError); + } + }), + ); +}); diff --git a/packages/stack/src/public/Config.ts b/packages/stack/src/public/Config.ts index a155839112..679daec576 100644 --- a/packages/stack/src/public/Config.ts +++ b/packages/stack/src/public/Config.ts @@ -39,6 +39,9 @@ const optionalCapability = (settings: S) => Schema.Struct({ enabled: Schema.optionalKey(Schema.Literal(true)), activation: Schema.optionalKey(ActivationModeSchema), + idleTimeoutSeconds: Schema.optionalKey( + Schema.Union([Schema.Finite.check(Schema.isGreaterThan(0)), Schema.Literal(false)]), + ), version: Schema.optionalKey(Schema.String), settings: Schema.optionalKey(settings), }), diff --git a/packages/stack/src/public/whole-stack.e2e.test.ts b/packages/stack/src/public/whole-stack.e2e.test.ts index 2e35a88892..92d7402ebd 100644 --- a/packages/stack/src/public/whole-stack.e2e.test.ts +++ b/packages/stack/src/public/whole-stack.e2e.test.ts @@ -356,6 +356,47 @@ const expectEndpointsRefused = async (endpoints: ReadonlyArray): } }; +const waitForLogEntry = async ( + stack: Pick, + iterator: AsyncIterator, + predicate: (entry: StackLogEntry) => boolean, +): Promise => { + const observation = (async (): Promise => { + while (true) { + const next = await iterator.next(); + if (next.done) throw new Error("Stack log stream ended before idle stop"); + if (predicate(next.value)) return next.value; + } + })(); + let timer: ReturnType | undefined; + try { + return await Promise.race([ + observation, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error("Timed out waiting for idle stop log")), + REQUEST_TIMEOUT_MS, + ); + }), + ]); + } catch (cause) { + let diagnostics = "unavailable"; + try { + const [status, logs] = await Promise.all([stack.status(), stack.logs({ tail: 20 })]); + const rest = status.capabilities.find(({ name }) => name === "rest"); + const recent = logs.entries + .map((entry) => `${entry.source}/${entry.stream}: ${entry.message}`) + .join("\n"); + diagnostics = `lifecycle=${status.lifecycle}; rest=${rest?.state ?? "unavailable"}; recent logs:\n${recent}`; + } catch { + // Preserve the stream failure when diagnostics are unavailable. + } + throw new Error(`Idle stop observation failed: ${diagnostics}`, { cause }); + } finally { + if (timer !== undefined) clearTimeout(timer); + } +}; + const expectRuntimeInputsAbsent = async ( stack: Pick, ): Promise => { @@ -1017,6 +1058,15 @@ const runWholeStackScenario = async (mode: (typeof RUNTIME_CASES)[number]): Prom await using stack: TestStack = await createTestStack({ name: `stack-e2e-${identity}`, runtime: mode.runtime, + config: { + capabilities: { + rest: { idleTimeoutSeconds: false }, + auth: { idleTimeoutSeconds: false }, + realtime: { idleTimeoutSeconds: false }, + studio: { idleTimeoutSeconds: false }, + pooler: { idleTimeoutSeconds: false }, + }, + }, setupProject: async (root) => { projectRoot = root; const directory = join(root, "supabase", "functions", functionSlug); @@ -1390,6 +1440,109 @@ describe("managed Supabase stack whole-stack E2E", () => { }, ); + test( + `stops and wakes idle REST in ${mode.name} mode`, + { timeout: E2E_TIMEOUT_MS }, + async () => { + const identity = crypto.randomUUID().replaceAll("-", "").slice(0, 16).toLowerCase(); + const table = `idle_${identity}`; + const marker = `idle-${identity}`; + await using stack: TestStack = await createTestStack({ + name: `stack-idle-rest-${identity}`, + runtime: mode.runtime, + config: { + capabilities: { + database: {}, + rest: { activation: "lazy", idleTimeoutSeconds: 5 }, + // API gateway credentials are materialized by auth even while its workload stays lazy. + auth: {}, + realtime: { enabled: false }, + storage: { enabled: false }, + functions: { enabled: false }, + studio: { enabled: false }, + mail: { enabled: false }, + analytics: { enabled: false }, + pooler: { enabled: false }, + }, + }, + }); + const initial = await stack.status(); + expect(initial.lifecycle).toBe("running"); + const api = endpoint(initial, "api"); + const initialSupervisorPid = await supervisorPid(stack.id); + await expectOwnedWorkloads(mode, stack.id, ["database:database"]); + + const credentials = await stack.credentials(); + await databaseQuery( + credentials.database.url, + `CREATE TABLE public."${table}" (id integer PRIMARY KEY, payload text NOT NULL)`, + ); + await databaseQuery( + credentials.database.url, + `GRANT SELECT ON public."${table}" TO anon, authenticated, service_role`, + ); + expect( + await databaseQuery( + credentials.database.url, + `INSERT INTO public."${table}" (id, payload) VALUES (1, $1) RETURNING id, payload`, + [marker], + ), + ).toEqual([{ id: 1, payload: marker }]); + + const restPath = `/rest/v1/${table}?select=id,payload`; + const beforeLogs = await stack.logs(); + const logIterator = stack.followLogs({ cursor: beforeLogs.cursor })[Symbol.asyncIterator](); + const idleStopLog = waitForLogEntry( + stack, + logIterator, + (entry) => + entry.source === "supervisor" && + entry.stream === "internal" && + entry.message === "Stopped rest after inactivity", + ); + const observedIdleStopLog = idleStopLog.then( + (entry) => ({ ok: true as const, entry }), + (cause: unknown) => ({ ok: false as const, cause }), + ); + try { + const firstRows = await jsonValue( + await request(api.url, restPath, { + headers: { ...apiHeaders(credentials), Accept: "application/json" }, + }), + ); + expect(firstRows).toEqual([{ id: 1, payload: marker }]); + const observed = await observedIdleStopLog; + if (!observed.ok) throw observed.cause; + } finally { + await logIterator.return?.(); + } + + await expectOwnedWorkloads(mode, stack.id, ["database:database"]); + const stoppedRest = await stack.status(); + expect(capabilityState(stoppedRest, "rest")).toBe("dormant"); + expect(endpoint(stoppedRest, "api").port).toBe(api.port); + expect(await supervisorPid(stack.id)).toBe(initialSupervisorPid); + expect( + await databaseQuery( + credentials.database.url, + `SELECT payload FROM public."${table}" WHERE id = 1`, + ), + ).toEqual([{ payload: marker }]); + + const secondRows = await jsonValue( + await request(api.url, restPath, { + headers: { ...apiHeaders(credentials), Accept: "application/json" }, + }), + ); + expect(secondRows).toEqual([{ id: 1, payload: marker }]); + await expectOwnedWorkloads(mode, stack.id, ["database:database", "rest:rest"]); + const restartedRest = await stack.status(); + expect(capabilityState(restartedRest, "rest")).toBe("ready"); + expect(endpoint(restartedRest, "api").port).toBe(api.port); + expect(await supervisorPid(stack.id)).toBe(initialSupervisorPid); + }, + ); + test(`supports the complete user flow in ${mode.name} mode`, { timeout: E2E_TIMEOUT_MS }, () => runWholeStackScenario(mode), ); diff --git a/packages/stack/src/state/StackState.ts b/packages/stack/src/state/StackState.ts index b74b1839bf..cc4d146aa7 100644 --- a/packages/stack/src/state/StackState.ts +++ b/packages/stack/src/state/StackState.ts @@ -118,7 +118,13 @@ const hasCompleteDefaults = (value: unknown, defaults: unknown): boolean => { return true; }; -const capabilityKeys = ["enabled", "activation", "version", "settings"] as const; +const capabilityKeys = [ + "enabled", + "activation", + "idleTimeoutSeconds", + "version", + "settings", +] as const; const invalid = (message: string): Effect.Effect => Effect.fail(new SchemaIssue.InvalidValue({ message })); @@ -230,6 +236,13 @@ const isDefinitionShape = (input: unknown): input is StackDefinition => { if (!hasExactKeys(capability, capabilityKeys)) return false; if (typeof capability.enabled !== "boolean") return false; if (capability.activation !== "eager" && capability.activation !== "lazy") return false; + if ( + capability.idleTimeoutSeconds !== false && + (typeof capability.idleTimeoutSeconds !== "number" || + !Number.isFinite(capability.idleTimeoutSeconds) || + capability.idleTimeoutSeconds <= 0) + ) + return false; if (typeof capability.version !== "string" || capability.version.length === 0) return false; } const listeners = input.listeners; diff --git a/packages/stack/src/supervisor/Ingress.ts b/packages/stack/src/supervisor/Ingress.ts index 3712688809..ffbe89cdad 100644 --- a/packages/stack/src/supervisor/Ingress.ts +++ b/packages/stack/src/supervisor/Ingress.ts @@ -8,6 +8,7 @@ import type { HttpGatewayListenerOptions, StackGateway, } from "../gateway/Gateway.ts"; +import type { GatewayActivity } from "../gateway/ActivityTracker.ts"; import { GatewayActivationError, PortUnavailableError, @@ -58,6 +59,7 @@ export interface SupervisorIngress { activate: ( capability: import("../public/Capability.ts").CapabilityName, ) => Effect.Effect, + activity?: GatewayActivity, ) => Effect.Effect; /** Close gateway, accepted sockets, and exact listeners; safe to call repeatedly. */ readonly close: Effect.Effect; @@ -261,6 +263,7 @@ export const makeSupervisorIngress = ( activate: ( capability: import("../public/Capability.ts").CapabilityName, ) => Effect.Effect, + activity?: GatewayActivity, ): Effect.Effect => lock.withPermit( Effect.gen(function* () { @@ -404,6 +407,7 @@ export const makeSupervisorIngress = ( : new GatewayActivationError({ message: error.message, cause: error }), ), ), + activity, }).pipe(Effect.provideService(Scope.Scope, entry.scope)), ); if (Exit.isFailure(gatewayResult)) { diff --git a/packages/stack/src/supervisor/SessionLauncher.ts b/packages/stack/src/supervisor/SessionLauncher.ts index 8f996d9309..6b8bb778b2 100644 --- a/packages/stack/src/supervisor/SessionLauncher.ts +++ b/packages/stack/src/supervisor/SessionLauncher.ts @@ -17,6 +17,9 @@ export interface SessionLauncher { readonly launch: (plan: ExecutionPlan) => Effect.Effect; /** Stops and removes every workload started in this session in reverse order. */ readonly stop: Effect.Effect; + readonly stopCapabilities: ( + capabilities: ReadonlySet, + ) => Effect.Effect; /** Whether the most recent launch/rollback cleanup completed exactly. */ readonly cleanupProven: Effect.Effect; /** Clears the session after stack-wide runtime cleanup has completed. */ @@ -190,9 +193,18 @@ export const makeSessionLauncher = (options: { }); const stop = Effect.suspend(() => Ref.get(session).pipe(Effect.flatMap(cleanup))); + const stopCapabilities = ( + capabilities: ReadonlySet, + ) => + Ref.get(session).pipe( + Effect.flatMap((entries) => + cleanup(entries.filter(({ workload }) => capabilities.has(workload.capability))), + ), + ); return { launch, stop, + stopCapabilities, cleanupProven: Ref.get(cleanupProven), clear: Ref.set(session, []), } satisfies SessionLauncher; diff --git a/packages/stack/src/supervisor/Supervisor.ts b/packages/stack/src/supervisor/Supervisor.ts index fc653786af..fbb2ff70cc 100644 --- a/packages/stack/src/supervisor/Supervisor.ts +++ b/packages/stack/src/supervisor/Supervisor.ts @@ -3,6 +3,7 @@ import { Context, Crypto, Deferred, + Duration, Effect, Exit, FileSystem, @@ -23,7 +24,7 @@ import { eagerCapabilities, type ExecutionPlan, } from "../model/ExecutionPlan.ts"; -import type { CapabilityName } from "../public/Capability.ts"; +import { CAPABILITY_NAMES, type CapabilityName } from "../public/Capability.ts"; import type { StackConfig } from "../public/Config.ts"; import { GatewayActivationError, @@ -71,6 +72,7 @@ import { } from "../state/SecretStore.ts"; import type { ActivationResult } from "../gateway/Gateway.ts"; +import { makeGatewayActivity } from "../gateway/ActivityTracker.ts"; interface SupervisorLaunchAttempt { /** Rolls back only workloads and ingress acquired by this launch. */ @@ -187,6 +189,8 @@ export const makeSupervisor = ( driver: runtime.driver, }); const active = yield* Ref.make>(new Set()); + const activeRoots = yield* Ref.make>(new Set()); + const currentPlan = yield* Ref.make(undefined); const phase = yield* Ref.make("stopped"); // A failed cleanup leaves the owner in `stopping` so callers can retry an exact cleanup. // This marker is set by the backend cleanup boundary and read only by start failure handling. @@ -211,12 +215,33 @@ export const makeSupervisor = ( } | { readonly _tag: "ready"; readonly result: ActivationResult } >(); - const initializeActivation = (plan: ExecutionPlan) => Ref.set(active, eagerCapabilities(plan)); + const initializeActivation = (input: LifecycleInput) => + Effect.gen(function* () { + const { plan } = input; + const roots = new Set( + CAPABILITY_NAMES.filter( + (name) => + input.definition.capabilities[name].enabled && plan.activation[name] === "eager", + ), + ); + yield* Ref.set(activeRoots, roots); + yield* Ref.set(active, eagerCapabilities(plan)); + yield* Ref.set(currentPlan, plan); + yield* Ref.set( + idleTimeouts, + new Map( + CAPABILITY_NAMES.map((name) => [ + name, + input.definition.capabilities[name].idleTimeoutSeconds, + ]), + ), + ); + }); const resetForSession = (input: LifecycleInput) => Effect.gen(function* () { activationOwned.clear(); yield* launcher.clear; - yield* initializeActivation(input.plan); + yield* initializeActivation(input); }); const observe = () => runtime.driver.observe(options.stackId).pipe(Effect.mapError(mapRuntimeError)); @@ -281,6 +306,253 @@ export const makeSupervisor = ( result: LifecycleResult; }>; const lifecycleActive = yield* Ref.make(undefined); + const traffic = yield* Ref.make>(new Map()); + const idleTimeouts = yield* Ref.make>(new Map()); + const idleGenerations = yield* Ref.make>(new Map()); + const idleTimers = yield* Ref.make< + ReadonlyMap< + CapabilityName, + { readonly token: symbol; readonly fiber: Fiber.Fiber } + > + >(new Map()); + + const idleTimeout = ( + timeouts: ReadonlyMap, + plan: ExecutionPlan, + capability: CapabilityName, + ): number | false => { + if (plan.activation[capability] !== "lazy") return false; + return timeouts.get(capability) ?? false; + }; + + const canRetire = ( + plan: ExecutionPlan, + roots: ReadonlySet, + capability: CapabilityName, + ): boolean => + ![...roots].some( + (root) => root !== capability && dependencyClosure(plan, [root]).has(capability), + ); + + const appendIdleLog = (message: string): Effect.Effect => + options.runtime.logStore + .append({ + source: "supervisor", + stream: "internal", + message, + }) + .pipe( + Effect.catchTag("LogStoreError", (error) => Effect.logWarning(message, error)), + Effect.asVoid, + ); + + function retireIdle( + capability: CapabilityName, + generation: number, + ): Effect.Effect { + return execution.withPermit( + Effect.gen(function* () { + const fenced = yield* admission.withPermit( + Effect.gen(function* () { + const currentGeneration = (yield* Ref.get(idleGenerations)).get(capability) ?? 0; + const count = (yield* Ref.get(traffic)).get(capability) ?? 0; + const plan = yield* Ref.get(currentPlan); + const roots = yield* Ref.get(activeRoots); + const currentPhase = yield* Ref.get(phase); + if ( + generation !== currentGeneration || + count !== 0 || + plan === undefined || + currentPhase !== "running" || + (yield* Ref.get(lifecycleActive)) !== undefined || + !(yield* Ref.get(active)).has(capability) || + !canRetire(plan, roots, capability) + ) + return false; + + // Fence the route before stopping its workloads. Requests admitted after this + // point queue behind execution and create a fresh activation when cleanup ends. + yield* Ref.update(active, (current) => { + const next = new Set(current); + next.delete(capability); + return next; + }); + yield* Ref.update(activeRoots, (current) => { + if (!current.has(capability)) return current; + const next = new Set(current); + next.delete(capability); + return next; + }); + activationOwned.delete(capability); + return true; + }), + ); + if (!fenced) return; + + const stopped = yield* launcher + .stopCapabilities(new Set([capability])) + .pipe(Effect.mapError(mapCleanupError), Effect.exit); + if (Exit.isFailure(stopped)) { + yield* admission.withPermit( + Ref.set(cleanupProven, false).pipe(Effect.andThen(Ref.set(phase, "stopping"))), + ); + const logged = yield* appendIdleLog( + `Failed to stop ${capability} after inactivity: ${Cause.pretty(stopped.cause)}`, + ).pipe(Effect.exit); + if (Exit.isFailure(logged)) + return yield* Effect.failCause(Cause.combine(stopped.cause, logged.cause)); + if (Cause.hasInterrupts(stopped.cause) || Cause.hasDies(stopped.cause)) + return yield* Effect.failCause(stopped.cause); + return; + } + + yield* admission.withPermit(reevaluateIdleTimersInAdmission()); + yield* appendIdleLog(`Stopped ${capability} after inactivity`); + }), + ); + } + + function armIdleTimerInAdmission(capability: CapabilityName): Effect.Effect { + return Effect.gen(function* () { + const plan = yield* Ref.get(currentPlan); + if (plan === undefined) return; + const timeout = idleTimeout(yield* Ref.get(idleTimeouts), plan, capability); + if (timeout === false) return; + const current = yield* Ref.get(active); + if (!current.has(capability) || !canRetire(plan, yield* Ref.get(activeRoots), capability)) + return; + const count = (yield* Ref.get(traffic)).get(capability) ?? 0; + if (count !== 0 || (yield* Ref.get(idleTimers)).has(capability)) return; + const generation = (yield* Ref.get(idleGenerations)).get(capability) ?? 0; + const token = Symbol(); + yield* Effect.uninterruptibleMask(() => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const fiber = yield* Effect.forkIn( + Deferred.await(started).pipe( + Effect.andThen( + Effect.sleep(Duration.seconds(timeout)).pipe( + Effect.ensuring( + Ref.update(idleTimers, (timers) => { + const entry = timers.get(capability); + if (entry?.token !== token) return timers; + const next = new Map(timers); + next.delete(capability); + return next; + }), + ), + Effect.andThen(retireIdle(capability, generation)), + ), + ), + ), + supervisorScope, + { startImmediately: true }, + ); + yield* Ref.update(idleTimers, (timers) => { + const next = new Map(timers); + next.set(capability, { token, fiber }); + return next; + }); + yield* Deferred.succeed(started, undefined); + }), + ); + }); + } + + function armIdleTimer(capability: CapabilityName): Effect.Effect { + return admission.withPermit(armIdleTimerInAdmission(capability)); + } + + function reevaluateIdleTimersInAdmission(): Effect.Effect { + return Ref.get(active).pipe( + Effect.flatMap((capabilities) => + Effect.forEach(capabilities, armIdleTimerInAdmission, { + concurrency: "unbounded", + discard: true, + }), + ), + ); + } + + function reevaluateIdleTimers(): Effect.Effect { + return Ref.get(active).pipe( + Effect.flatMap((capabilities) => + Effect.forEach(capabilities, armIdleTimer, { concurrency: "unbounded", discard: true }), + ), + ); + } + + const cancelIdleTimers: Effect.Effect = Effect.gen(function* () { + const timers = yield* admission.withPermit( + Effect.gen(function* () { + const timers = yield* Ref.modify( + idleTimers, + (current) => [Array.from(current.values()), new Map()] as const, + ); + yield* Ref.update(idleGenerations, (current) => { + const next = new Map(current); + for (const capability of CAPABILITY_NAMES) + next.set(capability, (next.get(capability) ?? 0) + 1); + return next; + }); + return timers; + }), + ); + yield* Effect.forEach(timers, ({ fiber }) => Fiber.interrupt(fiber), { + concurrency: "unbounded", + discard: true, + }); + }); + + const beginTraffic = (capability: CapabilityName): Effect.Effect => + Effect.gen(function* () { + const stale = yield* admission.withPermit( + Effect.gen(function* () { + const entry = (yield* Ref.get(idleTimers)).get(capability); + yield* Ref.update(idleTimers, (current) => { + if (entry === undefined) return current; + const next = new Map(current); + next.delete(capability); + return next; + }); + yield* Ref.update(traffic, (current) => { + const next = new Map(current); + next.set(capability, (next.get(capability) ?? 0) + 1); + return next; + }); + yield* Ref.update(idleGenerations, (current) => { + const next = new Map(current); + next.set(capability, (next.get(capability) ?? 0) + 1); + return next; + }); + return entry?.fiber; + }), + ); + if (stale !== undefined) yield* Fiber.interrupt(stale); + }); + const endTraffic = (capability: CapabilityName): Effect.Effect => + Effect.gen(function* () { + const shouldArm = yield* admission.withPermit( + Effect.gen(function* () { + const count = (yield* Ref.get(traffic)).get(capability) ?? 0; + yield* Ref.update(traffic, (current) => { + const next = new Map(current); + if (count <= 1) next.delete(capability); + else next.set(capability, count - 1); + return next; + }); + yield* Ref.update(idleGenerations, (current) => { + const next = new Map(current); + next.set(capability, (next.get(capability) ?? 0) + 1); + return next; + }); + return count <= 1; + }), + ); + if (shouldArm) yield* armIdleTimer(capability); + }); + + const activity = yield* makeGatewayActivity({ begin: beginTraffic, end: endTraffic }); const ingressActivate = ( capability: CapabilityName, ): Effect.Effect => @@ -371,10 +643,15 @@ export const makeSupervisor = ( ), ); const owner = Effect.gen(function* () { - const result = yield* execution.withPermit(effect).pipe(Effect.exit); + const result = yield* cancelIdleTimers.pipe( + Effect.andThen(execution.withPermit(effect)), + Effect.exit, + ); // Release the admission slot before waking waiters so a completed operation // cannot make the next lifecycle request look like a conflict. yield* release; + if (Exit.isFailure(result) && (yield* Ref.get(phase)) === "running") + yield* reevaluateIdleTimers(); yield* Deferred.succeed(deferred, result); }).pipe(Effect.ensuring(release)); yield* FiberSet.run(ownedFibers, owner, { startImmediately: true }); @@ -441,7 +718,7 @@ export const makeSupervisor = ( if (cause.reasons.length > 0) return yield* Effect.failCause(cause); }); const opened = yield* runtime.ingress - .open(input, reservation, ingressActivate) + .open(input, reservation, ingressActivate, activity) .pipe(Effect.exit); if (Exit.isFailure(opened)) { const rolledBack = yield* rollback.pipe(Effect.exit); @@ -484,8 +761,12 @@ export const makeSupervisor = ( } if (!destroy) { yield* Ref.set(active, new Set()); + yield* Ref.set(activeRoots, new Set()); + yield* Ref.set(traffic, new Map()); } else { yield* launcher.clear; + yield* Ref.set(activeRoots, new Set()); + yield* Ref.set(traffic, new Map()); } }); const backend: LifecycleBackend = { @@ -519,6 +800,8 @@ export const makeSupervisor = ( ), ); const previousActive = yield* Ref.get(active); + const previousRoots = yield* Ref.get(activeRoots); + const nextRoots = new Set([...previousRoots, capability]); const next = new Set([...previousActive, ...dependencyClosure(plan, [capability])]); const input: LifecycleInput = { stackId: options.stackId, @@ -535,6 +818,7 @@ export const makeSupervisor = ( const activated = yield* runtime.activate(capability, input).pipe(Effect.exit); if (Exit.isFailure(activated)) { yield* Ref.set(active, previousActive); + yield* Ref.set(activeRoots, previousRoots); const rolledBack = yield* launched.rollback.pipe(Effect.exit); if (Exit.isFailure(rolledBack)) { yield* Ref.set(phase, "stopping"); @@ -544,6 +828,9 @@ export const makeSupervisor = ( return yield* Effect.failCause(activated.cause); } const endpoint = activated.value; + yield* Ref.set(activeRoots, nextRoots); + yield* Ref.set(currentPlan, plan); + yield* reevaluateIdleTimers(); return { capability, endpoint }; }); @@ -669,6 +956,7 @@ export const makeSupervisor = ( return yield* Effect.failCause(started.cause); } yield* Ref.set(phase, "running"); + yield* reevaluateIdleTimers(); yield* startBackgroundPreparation(started.value); }); const start = (startOptions?: { readonly config?: StackConfig }) => diff --git a/packages/stack/src/supervisor/supervisor.integration.test.ts b/packages/stack/src/supervisor/supervisor.integration.test.ts index 49921c7d8e..db7c151953 100644 --- a/packages/stack/src/supervisor/supervisor.integration.test.ts +++ b/packages/stack/src/supervisor/supervisor.integration.test.ts @@ -14,6 +14,7 @@ import { Ref, Scope, } from "effect"; +import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/unstable/http"; import { Rpc } from "effect/unstable/rpc"; import { RequestId } from "effect/unstable/rpc/RpcMessage"; @@ -46,6 +47,7 @@ import { resolveStackPaths } from "../state/Paths.ts"; import { StackRpcGroup, type StackRpcError } from "../control/StackRpc.ts"; import { makeSupervisor, type Supervisor, type SupervisorRuntime } from "./Supervisor.ts"; import type { SupervisorIngress } from "./Ingress.ts"; +import type { GatewayActivity } from "../gateway/ActivityTracker.ts"; const identity = { projectRoot: "/tmp/supabase-supervisor", @@ -79,6 +81,12 @@ const makeFixture = ( fixtureOptions: { readonly ingress?: SupervisorIngress; readonly timeline?: Ref.Ref>; + readonly logRecords?: Ref.Ref>; + readonly logWritten?: Deferred.Deferred; + readonly logWrittenFor?: string; + readonly logWrittenAdditional?: Deferred.Deferred; + readonly logWrittenAdditionalFor?: string; + readonly logQueue?: Queue.Queue; readonly runtime?: StackRuntime; readonly startGate?: Deferred.Deferred; readonly startStarted?: Deferred.Deferred; @@ -95,7 +103,10 @@ const makeFixture = ( readonly stopGate?: Deferred.Deferred; readonly stopStarted?: Deferred.Deferred; readonly workloadStopFailFirst?: Ref.Ref; + readonly workloadStopGate?: Deferred.Deferred; + readonly workloadStopStarted?: Deferred.Deferred; readonly workloadRemoveFailFirst?: Ref.Ref; + readonly workloadRemoveDieFirst?: Ref.Ref; readonly stopFailFirst?: Ref.Ref; readonly destroyGate?: Deferred.Deferred; readonly destroyStarted?: Deferred.Deferred; @@ -259,6 +270,10 @@ const makeFixture = ( }), stop: (key) => Effect.gen(function* () { + if (fixtureOptions.workloadStopStarted !== undefined) + yield* Deferred.succeed(fixtureOptions.workloadStopStarted, undefined); + if (fixtureOptions.workloadStopGate !== undefined) + yield* Deferred.await(fixtureOptions.workloadStopGate); if (fixtureOptions.workloadStopFailFirst !== undefined) { const fail = yield* Ref.get(fixtureOptions.workloadStopFailFirst); if (fail) { @@ -292,6 +307,13 @@ const makeFixture = ( }); } } + if (fixtureOptions.workloadRemoveDieFirst !== undefined) { + const fail = yield* Ref.get(fixtureOptions.workloadRemoveDieFirst); + if (fail) { + yield* Ref.set(fixtureOptions.workloadRemoveDieFirst, false); + return yield* Effect.die("injected workload remove defect"); + } + } yield* Ref.update(resources, (current) => current.filter((entry) => entry.workloadId !== key.workloadId), ); @@ -423,7 +445,39 @@ const makeFixture = ( }, logStore: { path: "memory://logs", - append: () => Effect.succeed(entry), + append: (record) => + (fixtureOptions.logRecords === undefined + ? Effect.void + : Ref.update(fixtureOptions.logRecords, (current) => [...current, record.message]) + ).pipe( + Effect.andThen( + fixtureOptions.logQueue === undefined + ? Effect.void + : Queue.offer(fixtureOptions.logQueue, record.message), + ), + Effect.andThen( + fixtureOptions.logWritten === undefined || + (fixtureOptions.logWrittenFor !== undefined && + !record.message.includes(fixtureOptions.logWrittenFor)) + ? Effect.void + : Deferred.succeed(fixtureOptions.logWritten, undefined), + ), + Effect.andThen( + fixtureOptions.logWrittenAdditional === undefined || + (fixtureOptions.logWrittenAdditionalFor !== undefined && + !record.message.includes(fixtureOptions.logWrittenAdditionalFor)) + ? Effect.void + : Deferred.succeed(fixtureOptions.logWrittenAdditional, undefined), + ), + Effect.andThen( + Effect.succeed({ + ...entry, + source: record.source, + stream: record.stream, + message: record.message, + }), + ), + ), read: (options) => options?.cursor?.opaque === "not-a-cursor" ? Effect.fail(new InvalidLogCursorError({ message: "Log cursor is invalid" })) @@ -546,6 +600,562 @@ describe("Supervisor composition", () => { ), ); + it.live("retires lazy traffic after its lease ends and reactivates on demand", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const timeline = yield* Ref.make>([]); + const activationStarted = yield* Deferred.make(); + const logWritten = yield* Deferred.make(); + const fixture = yield* makeFixture({ + timeline, + activationStarted, + logWritten, + logWrittenFor: "Stopped rest after inactivity", + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + const release = yield* Deferred.make(); + const request = yield* Effect.forkChild( + tracker.track( + "rest", + fixture.supervisor.activate("rest").pipe(Effect.andThen(Deferred.await(release))), + ), + ); + yield* Deferred.await(activationStarted); + yield* TestClock.adjust("1 second"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(request); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(logWritten); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + expect(yield* Ref.get(timeline)).toContain("stop:rest:rest"); + yield* fixture.supervisor.activate("rest"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("keeps a lazy dependency pinned until its dependent retires", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const timeline = yield* Ref.make>([]); + const activationStarted = yield* Deferred.make(); + const logWritten = yield* Deferred.make(); + const restLogWritten = yield* Deferred.make(); + const fixture = yield* makeFixture({ + timeline, + activationStarted, + logWritten, + logWrittenFor: "Stopped studio after inactivity", + logWrittenAdditional: restLogWritten, + logWrittenAdditionalFor: "Stopped rest after inactivity", + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { + capabilities: { + rest: { activation: "lazy", idleTimeoutSeconds: 1 }, + studio: { activation: "lazy", idleTimeoutSeconds: 1 }, + }, + }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + const release = yield* Deferred.make(); + const request = yield* Effect.forkChild( + tracker.track( + "studio", + fixture.supervisor.activate("studio").pipe(Effect.andThen(Deferred.await(release))), + ), + ); + yield* Deferred.await(activationStarted); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(request); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(logWritten); + const afterStudio = yield* fixture.supervisor.status; + expect(afterStudio.capabilities.find(({ name }) => name === "studio")?.state).toBe( + "dormant", + ); + expect(afterStudio.capabilities.find(({ name }) => name === "rest")?.state).toBe("ready"); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(restLogWritten); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + expect(yield* Ref.get(timeline)).toEqual( + expect.arrayContaining(["stop:studio:pgmeta", "stop:rest:rest"]), + ); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("accepts very small idle timeout values", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const logWritten = yield* Deferred.make(); + const fixture = yield* makeFixture({ + logWritten, + logWrittenFor: "Stopped rest after inactivity", + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1e-7 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 millis"); + yield* Deferred.await(logWritten); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("rearms a zero-rounded idle timeout after its first retirement", () => + run( + Effect.gen(function* () { + const logs = yield* Queue.unbounded(); + const fixture = yield* makeFixture({ + logQueue: logs, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1e-12 } } }, + }); + yield* fixture.supervisor.activate("rest"); + yield* TestClock.adjust("1 millis"); + expect(yield* Queue.take(logs)).toContain("Stopped rest after inactivity"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + + yield* fixture.supervisor.activate("rest"); + yield* TestClock.adjust("1 millis"); + expect(yield* Queue.take(logs)).toContain("Stopped rest after inactivity"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("resets the idle deadline when a second request arrives", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const activationStarted = yield* Deferred.make(); + const logWritten = yield* Deferred.make(); + const fixture = yield* makeFixture({ + activationStarted, + logWritten, + logWrittenFor: "Stopped rest after inactivity", + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + const first = yield* Effect.forkChild( + tracker.track("rest", fixture.supervisor.activate("rest")), + ); + yield* Deferred.await(activationStarted); + yield* Fiber.join(first); + yield* TestClock.adjust("500 millis"); + yield* tracker.track("rest", Effect.void); + yield* TestClock.adjust("500 millis"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(logWritten); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("queues traffic arriving during idle cleanup for a fresh activation", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const activationCalls = yield* Ref.make(0); + const stopStarted = yield* Deferred.make(); + const stopGate = yield* Deferred.make(); + const fixture = yield* makeFixture({ + activationCalls, + workloadStopStarted: stopStarted, + workloadStopGate: stopGate, + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(stopStarted); + + const requestStarted = yield* Deferred.make(); + const request = yield* Effect.forkChild( + Deferred.succeed(requestStarted, undefined).pipe( + Effect.andThen(tracker.track("rest", fixture.supervisor.activate("rest"))), + ), + ); + yield* Deferred.await(requestStarted); + expect(yield* Ref.get(activationCalls)).toBe(1); + + yield* Deferred.succeed(stopGate, undefined); + yield* Fiber.join(request); + expect(yield* Ref.get(activationCalls)).toBe(2); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("keeps unrelated ready traffic flowing during idle cleanup", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const stopStarted = yield* Deferred.make(); + const stopGate = yield* Deferred.make(); + const authCompleted = yield* Deferred.make(); + const fixture = yield* makeFixture({ + workloadStopStarted: stopStarted, + workloadStopGate: stopGate, + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + const body = Effect.gen(function* () { + yield* fixture.supervisor.start({ + config: { + capabilities: { + rest: { activation: "lazy", idleTimeoutSeconds: 1 }, + auth: { activation: "eager" }, + }, + }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("auth", fixture.supervisor.activate("auth")); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(stopStarted); + + const authRequest = yield* Effect.forkChild( + tracker.track( + "auth", + fixture.supervisor + .activate("auth") + .pipe(Effect.andThen(Deferred.succeed(authCompleted, undefined))), + ), + { startImmediately: true }, + ); + yield* Deferred.await(authCompleted); + yield* Deferred.succeed(stopGate, undefined); + yield* Fiber.join(authRequest); + }); + yield* body.pipe(Effect.ensuring(Deferred.succeed(stopGate, undefined))); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("records the original cause when idle cleanup fails", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const logRecords = yield* Ref.make>([]); + const logWritten = yield* Deferred.make(); + const workloadStopFailFirst = yield* Ref.make(true); + const workloadStopStarted = yield* Deferred.make(); + const fixture = yield* makeFixture({ + logRecords, + logWritten, + workloadStopFailFirst, + workloadStopStarted, + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(workloadStopStarted); + yield* Deferred.await(logWritten); + const messages = yield* Ref.get(logRecords); + expect(messages).toEqual( + expect.arrayContaining([ + expect.stringContaining("Failed to stop rest after inactivity"), + expect.stringContaining("injected workload stop failure"), + ]), + ); + expect((yield* fixture.supervisor.status).lifecycle).toBe("stopping"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("fences the session after an idle cleanup defect", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const logRecords = yield* Ref.make>([]); + const logWritten = yield* Deferred.make(); + const workloadRemoveDieFirst = yield* Ref.make(true); + const fixture = yield* makeFixture({ + logRecords, + logWritten, + workloadRemoveDieFirst, + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(logWritten); + + const messages = yield* Ref.get(logRecords); + expect(messages).toEqual( + expect.arrayContaining([ + expect.stringContaining("Failed to stop rest after inactivity"), + expect.stringContaining("injected workload remove defect"), + ]), + ); + expect((yield* fixture.supervisor.status).lifecycle).toBe("stopping"); + const activation = yield* fixture.supervisor.activate("rest").pipe(Effect.exit); + expect(errorOf(activation)).toBeInstanceOf(StackLifecycleConflictError); + + expect((yield* fixture.supervisor.maintenanceHandlers.stop).ok).toBe(true); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const restartedTracker = yield* Ref.get(activity); + if (restartedTracker === undefined) + return yield* Effect.die("restarted gateway activity was not installed"); + yield* restartedTracker.track("rest", fixture.supervisor.activate("rest")); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("cancels idle timers when a stack session stops", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const fixture = yield* makeFixture({ + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const firstTracker = yield* Ref.get(activity); + if (firstTracker === undefined) + return yield* Effect.die("first gateway activity was not installed"); + yield* firstTracker.track("rest", fixture.supervisor.activate("rest")); + yield* fixture.supervisor.maintenanceHandlers.stop; + + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 10 } } }, + }); + const secondTracker = yield* Ref.get(activity); + if (secondTracker === undefined) + return yield* Effect.die("second gateway activity was not installed"); + yield* secondTracker.track("rest", fixture.supervisor.activate("rest")); + yield* TestClock.adjust("1 second"); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("ready"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + + it.live("restores idle deadlines after a rejected running start", () => + run( + Effect.gen(function* () { + const activity = yield* Ref.make(undefined); + const logWritten = yield* Deferred.make(); + const fixture = yield* makeFixture({ + logWritten, + logWrittenFor: "Stopped rest after inactivity", + ingress: { + acquire: () => + Effect.succeed({ + assignments: {}, + privateAssignments: [], + hostListeners: [], + fresh: false, + ownershipToken: Symbol(), + }), + open: (_input, _reservation, _activate, tracker) => + tracker === undefined ? Effect.void : Ref.set(activity, tracker), + close: Effect.void, + }, + }); + yield* fixture.supervisor.start({ + config: { capabilities: { rest: { activation: "lazy", idleTimeoutSeconds: 1 } } }, + }); + const tracker = yield* Ref.get(activity); + if (tracker === undefined) return yield* Effect.die("gateway activity was not installed"); + yield* tracker.track("rest", fixture.supervisor.activate("rest")); + const rejected = yield* fixture.supervisor + .start({ config: { capabilities: { rest: { settings: { schemas: ["private"] } } } } }) + .pipe(Effect.exit); + expect(errorOf(rejected)).toBeInstanceOf(StackMustBeStoppedError); + yield* TestClock.adjust("1 second"); + yield* Deferred.await(logWritten); + expect( + (yield* fixture.supervisor.status).capabilities.find(({ name }) => name === "rest") + ?.state, + ).toBe("dormant"); + }).pipe(Effect.provide(TestClock.layer())), + ), + ); + it.live("persists stopped after startup ingress failure", () => run( Effect.gen(function* () {