1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Fiber from "effect/Fiber";
7import * as Scope from "effect/Scope";
8import type { Server } from "node:http";
9
10/** Internal HTTP and dependency lifetimes, separate from managed run fibers. */
11export class ManagedHttpShutdown extends Context.Service<
12 ManagedHttpShutdown,
13 {
14 readonly scope: Scope.Closeable;
15 readonly dependencies: Scope.Closeable;
16 readonly servers: Set<Server>;
17 readonly drainTimeoutMs: number;
18 readonly isStopping: () => boolean;
19 readonly observeRequest: (fiber: Fiber.Fiber<unknown, unknown>) => void;
20 readonly runnerFinished: (
21 exit: Exit.Exit<unknown, unknown>,
22 last: boolean,
23 ) => void;
24 }
25>()("Alchemy.Runtime.ManagedHttpShutdown") {}
26
27/**
28 * Coordinate shutdown of an Alchemy-owned Fly Service process, including HTTP
29 * requests, background runners, and their shared dependencies.
30 *
31 * The Fly bootstrap enables this when `ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS` is set.
32 * The Service provider sets it for managed Services using blue/green or an
33 * explicit shutdown policy. Raw images and external servers own their shutdown.
34 *
35 * This is a process-entrypoint helper, not a request timeout or a general-purpose
36 * Effect combinator. Running it installs process-wide SIGTERM and SIGINT
37 * listeners. They coexist with other listeners, but cannot prevent another
38 * listener from exiting the process. Importing this module installs no listeners.
39 *
40 * Shutdown stops HTTP acceptance and interrupts background runners concurrently.
41 * Existing responses and request finalizers drain before shared dependencies
42 * close. Deadlines start on a signal, runner failure, completion of all runner
43 * bodies, or program completion, not after `timeoutMs` of normal execution.
44 *
45 * At 80% of the budget, remaining HTTP connections are forcibly closed. At 90%,
46 * a native timer calls `process.exit(1)`, even if Effect finalizers are still
47 * running. This reserves time before Fly's own termination deadline; it does
48 * not guarantee cleanup completes. These timers require a responsive event loop.
49 * Repeated signals do not extend the budget. Listeners and timers are removed
50 * when the wrapper's scope closes, unless the process exits first.
51 *
52 * @param program - The whole managed process program, not an individual handler.
53 * @param timeoutMs - Fly's configured shutdown budget, from 1 to 300000 ms.
54 * @returns Whether shutdown was requested before the program completed.
55 * Cleanup failures and exceeded drain deadlines fail the Effect; the caller
56 * owns the normal process exit after successful cleanup.
57 * @internal
58 */
59export const withManagedHttpShutdown = (
60 program: Effect.Effect<unknown, unknown>,
61 timeoutMs: number,
62) =>
63 Effect.gen(function* () {
64 if (
65 !Number.isSafeInteger(timeoutMs) ||
66 timeoutMs <= 0 ||
67 timeoutMs > 300_000
68 ) {
69 return yield* Effect.fail(
70 new Error(
71 "ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS must be a positive integer at most 300000",
72 ),
73 );
74 }
75
76 const signal = yield* Deferred.make<void>();
77 const scope = yield* Scope.make("parallel");
78 const dependencies = yield* Scope.make();
79 const servers = new Set<Server>();
80 let stopping = false;
81 let drained = false;
82 let timedOut = false;
83 const runnerFailures: Cause.Cause<unknown>[] = [];
84 const requestFailures: Cause.Cause<unknown>[] = [];
85 const requests = new Set<Fiber.Fiber<unknown, unknown>>();
86 let drainDeadline: ReturnType<typeof setTimeout> | undefined;
87 let deadline: ReturnType<typeof setTimeout> | undefined;
88 // Leave time for request/instance finalizers, then for Fly's own stop deadline.
89 const drainTimeoutMs = Math.floor(timeoutMs * 0.8);
90 const exitTimeoutMs = Math.min(Math.floor(timeoutMs * 0.9), 2 ** 31 - 1);
91
92 // Native timers remain independent of uninterruptible Effect finalizers.
93 const beginShutdown = () => {
94 if (stopping) return;
95 stopping = true;
96 drainDeadline = setTimeout(() => {
97 if (drained) return;
98 timedOut = true;
99 console.error(
100 "Managed HTTP drain deadline exceeded; closing connections.",
101 );
102 for (const server of servers) server.closeAllConnections();
103 }, drainTimeoutMs);
104 deadline = setTimeout(() => {
105 console.error("Managed HTTP shutdown deadline exceeded; exiting.");
106 process.exit(1);
107 }, exitTimeoutMs);
108 };
109
110 yield* Effect.acquireRelease(
111 Effect.sync(() => {
112 const onSignal = () => {
113 beginShutdown();
114 Deferred.doneUnsafe(signal, Effect.void);
115 };
116 process.on("SIGTERM", onSignal);
117 process.on("SIGINT", onSignal);
118 return onSignal;
119 }),
120 (onSignal) =>
121 Effect.sync(() => {
122 process.off("SIGTERM", onSignal);
123 process.off("SIGINT", onSignal);
124 if (drainDeadline !== undefined) clearTimeout(drainDeadline);
125 if (deadline !== undefined) clearTimeout(deadline);
126 }),
127 );
128 yield* Effect.addFinalizer((exit) => Scope.close(dependencies, exit));
129 yield* Effect.addFinalizer((exit) => Scope.close(scope, exit));
130
131 const fiber = yield* program.pipe(
132 Effect.provideService(ManagedHttpShutdown, {
133 scope,
134 dependencies,
135 servers,
136 drainTimeoutMs,
137 isStopping: () => stopping,
138 observeRequest: (request) => {
139 if (requests.has(request)) return;
140 requests.add(request);
141 request.addObserver((exit) => {
142 requests.delete(request);
143 if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause)) {
144 requestFailures.push(exit.cause);
145 console.error(
146 "Managed HTTP request cleanup failed",
147 Cause.pretty(exit.cause),
148 );
149 }
150 });
151 },
152 runnerFinished: (exit, last) => {
153 if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause)) {
154 runnerFailures.push(exit.cause);
155 console.error(
156 "Managed process cleanup failed",
157 Cause.pretty(exit.cause),
158 );
159 beginShutdown();
160 Deferred.doneUnsafe(signal, Effect.void);
161 } else if (last) {
162 beginShutdown();
163 }
164 },
165 }),
166 Effect.forkChild,
167 );
168 const interrupted = yield* Effect.raceFirst(
169 Deferred.await(signal).pipe(Effect.as(true)),
170 Fiber.await(fiber).pipe(Effect.as(false)),
171 );
172 yield* Effect.sync(beginShutdown);
173 // Neither HTTP drain nor one worker's finalizers may delay another's stop.
174 const [httpExit, programExit] = yield* Effect.all(
175 [
176 Scope.close(scope, Exit.void).pipe(Effect.exit),
177 Effect.gen(function* () {
178 if (interrupted) yield* Fiber.interrupt(fiber);
179 return yield* Fiber.await(fiber);
180 }),
181 ],
182 { concurrency: "unbounded" },
183 );
184 // Socket closure is not completion of the request fiber's finalizers.
185 yield* Fiber.awaitAll(requests);
186 drained = true;
187 const dependencyExit = yield* Scope.close(dependencies, programExit).pipe(
188 Effect.exit,
189 );
190 const failures = runnerFailures.concat(
191 requestFailures,
192 [httpExit, programExit, dependencyExit].flatMap((exit) =>
193 Exit.isFailure(exit) &&
194 !(interrupted && Cause.hasInterruptsOnly(exit.cause))
195 ? [exit.cause]
196 : [],
197 ),
198 );
199 for (const cause of failures) {
200 yield* Effect.logError("Managed process cleanup failed", cause);
201 }
202 if (failures.length > 0) return yield* Effect.failCause(failures[0]!);
203 if (timedOut) {
204 return yield* Effect.fail(
205 new Error("Managed process drain deadline exceeded"),
206 );
207 }
208 return interrupted;
209 }).pipe(Effect.scoped);