1import { findAvailablePort, nodeLoaderArgs } from "@/Util/Node.ts";
2import { PlatformServices } from "@/Util/PlatformServices.ts";
3import { describe, expect, it } from "alchemy-test";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
8import * as Path from "effect/Path";
9import * as Result from "effect/Result";
10import * as Schedule from "effect/Schedule";
11import * as Stream from "effect/Stream";
12import * as FetchHttpClient from "effect/http/FetchHttpClient";
13import * as HttpClient from "effect/http/HttpClient";
14import * as ChildProcess from "effect/process/ChildProcess";
15import { createHash } from "node:crypto";
16
17const services = Layer.mergeAll(PlatformServices, FetchHttpClient.layer);
18
19const spawnFixture = (
20 env: Record<string, string | undefined> = {},
21 waitForReady = true,
22) =>
23 Effect.gen(function* () {
24 const path = yield* Path.Path;
25 const fixture = yield* path.fromFileUrl(
26 new URL("./fixtures/managed-http-shutdown.ts", import.meta.url),
27 );
28 const port = yield* findAvailablePort();
29 const args = yield* Effect.sync(() => nodeLoaderArgs(fixture));
30 const handle = yield* ChildProcess.make("node", [...args, fixture], {
31 env: {
32 ALCHEMY_STACK_NAME: "managed-http-shutdown",
33 ALCHEMY_STAGE: "test",
34 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "3000",
35 PORT: String(port),
36 ...env,
37 },
38 extendEnv: true,
39 stdin: "ignore",
40 stdout: "pipe",
41 stderr: "pipe",
42 killSignal: "SIGKILL",
43 });
44 const output: string[] = [];
45 const ready = yield* Deferred.make<void>();
46 yield* Stream.merge(handle.stdout, handle.stderr).pipe(
47 Stream.decodeText,
48 Stream.runForEach((chunk) =>
49 Effect.sync(() => {
50 output.push(chunk);
51 const marker = env.RUN_ONLY === "1" ? "worker1 ready" : "http ready";
52 if (output.join("").includes(marker))
53 Deferred.doneUnsafe(ready, Effect.void);
54 }),
55 ),
56 Effect.forkScoped,
57 );
58 if (waitForReady) {
59 yield* Deferred.await(ready).pipe(
60 Effect.timeout("15 seconds"),
61 Effect.tapError(() => Effect.logError(output.join(""))),
62 );
63 }
64 const client = yield* HttpClient.HttpClient;
65 return {
66 handle,
67 output: () => output.join(""),
68 signal: (signal: "SIGTERM" | "SIGINT") =>
69 Effect.sync(() => process.kill(handle.pid, signal)),
70 get: (route: string) =>
71 client.get(`http://127.0.0.1:${port}${route}`, {
72 headers: { connection: "close" },
73 }),
74 };
75 });
76
77const waitForOutput = (fixture: { output: () => string }, text: string) =>
78 Effect.sync(fixture.output).pipe(
79 Effect.repeat({
80 schedule: Schedule.spaced("200 millis"),
81 until: (output) => output.includes(text),
82 times: 10,
83 }),
84 Effect.tap((output) => Effect.sync(() => expect(output).toContain(text))),
85 );
86
87// Measure the child's shutdown budget separately from signal delivery and reaping.
88const shutdownElapsed = (fixture: { output: () => string }) =>
89 waitForOutput(fixture, "shutdown elapsed ms:").pipe(
90 Effect.map((output) => {
91 const match = /shutdown elapsed ms: ([\d.]+)/.exec(output);
92 expect(match).not.toBeNull();
93 return Number(match![1]);
94 }),
95 );
96
97const refusesNewRequests = (
98 fixture: Effect.Success<ReturnType<typeof spawnFixture>>,
99) =>
100 fixture.get("/health").pipe(
101 Effect.timeout("200 millis"),
102 Effect.result,
103 Effect.repeat({
104 schedule: Schedule.spaced("25 millis"),
105 until: Result.isFailure,
106 times: 8,
107 }),
108 Effect.tap((result) =>
109 Effect.sync(() => expect(Result.isFailure(result)).toBe(true)),
110 ),
111 );
112
113const assertFinalizerOrder = (output: string, finished: string) => {
114 expect(output.indexOf(finished)).toBeGreaterThanOrEqual(0);
115 expect(output.indexOf("request finalized")).toBeGreaterThan(
116 output.indexOf(finished),
117 );
118 expect(output.indexOf("instance finalizing")).toBeGreaterThan(
119 output.indexOf("request finalized"),
120 );
121 expect(output).toContain("instance finalized");
122};
123
124const before = (output: string, first: string, second: string) => {
125 expect(output.indexOf(first)).toBeGreaterThanOrEqual(0);
126 expect(output.indexOf(second)).toBeGreaterThan(output.indexOf(first));
127};
128
129const assertWorkerOrder = (output: string, name: string) => {
130 before(output, `${name} stop started`, `${name} stopped`);
131 before(output, `${name} stopped`, `${name} drain started`);
132 before(output, `${name} drain started`, `${name} drained`);
133 before(output, `${name} drained`, `${name} work scope closed`);
134 before(output, `${name} work scope closed`, `${name} client released`);
135 const afterStop = output.slice(output.indexOf(`${name} stopped`));
136 expect(afterStop).not.toContain(`${name} claim`);
137 expect(output).not.toContain("already closed");
138};
139
140describe.sequential("managed Fly HTTP shutdown", { tags: ["local"] }, () => {
141 it.live(
142 "R03 SIGTERM stops acceptance and drains a slow response before instance finalizers",
143 () =>
144 Effect.gen(function* () {
145 const fixture = yield* spawnFixture();
146 const response = yield* fixture.get("/slow").pipe(
147 Effect.flatMap((response) => response.text),
148 Effect.forkChild,
149 );
150 yield* waitForOutput(fixture, "request started");
151 yield* fixture.signal("SIGTERM");
152 yield* refusesNewRequests(fixture);
153 yield* fixture.signal("SIGINT");
154 expect(yield* Fiber.join(response)).toBe("completed");
155 expect(yield* fixture.handle.exitCode).toBe(0);
156 yield* waitForOutput(fixture, "instance finalized");
157 assertFinalizerOrder(fixture.output(), "response ready");
158 }).pipe(Effect.scoped, Effect.provide(services)),
159 { timeout: 25_000 },
160 );
161
162 it.live(
163 "R04 SIGINT drains the entire streaming body before closing scopes",
164 () =>
165 Effect.gen(function* () {
166 const fixture = yield* spawnFixture();
167 const first = yield* Deferred.make<void>();
168 const chunks: string[] = [];
169 const response = yield* fixture.get("/stream").pipe(
170 Effect.flatMap((response) =>
171 response.stream.pipe(
172 Stream.decodeText,
173 Stream.runForEach((chunk) =>
174 Effect.sync(() => {
175 chunks.push(chunk);
176 Deferred.doneUnsafe(first, Effect.void);
177 }),
178 ),
179 ),
180 ),
181 Effect.forkChild,
182 );
183 yield* Deferred.await(first);
184 yield* fixture.signal("SIGINT");
185 yield* refusesNewRequests(fixture);
186 yield* Fiber.join(response);
187 expect(chunks.join("")).toBe("first\nsecond\nlast\n");
188 expect(yield* fixture.handle.exitCode).toBe(0);
189 yield* waitForOutput(fixture, "instance finalized");
190 assertFinalizerOrder(fixture.output(), "chunk last");
191 }).pipe(Effect.scoped, Effect.provide(services)),
192 { timeout: 25_000 },
193 );
194
195 it.live(
196 "R05 closes a stalled stream at the drain deadline and finishes instance cleanup",
197 () =>
198 Effect.gen(function* () {
199 const fixture = yield* spawnFixture({
200 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "2000",
201 RECORD_SHUTDOWN_TIMING: "1",
202 });
203 const first = yield* Deferred.make<void>();
204 const response = yield* fixture.get("/stream-hang").pipe(
205 Effect.flatMap((response) =>
206 response.stream.pipe(
207 Stream.runForEach(() => Deferred.succeed(first, undefined)),
208 ),
209 ),
210 Effect.result,
211 Effect.forkChild,
212 );
213 yield* Deferred.await(first);
214 yield* fixture.signal("SIGTERM");
215 expect(
216 yield* fixture.handle.exitCode.pipe(Effect.timeout("5 seconds")),
217 ).toBe(1);
218 expect(Result.isFailure(yield* Fiber.join(response))).toBe(true);
219 const elapsed = yield* shutdownElapsed(fixture);
220 expect(elapsed).toBeGreaterThanOrEqual(1500);
221 expect(elapsed).toBeLessThan(2000);
222 yield* waitForOutput(fixture, "instance finalized");
223 assertFinalizerOrder(fixture.output(), "chunk last");
224 expect(fixture.output()).toContain("drain deadline exceeded");
225 expect(fixture.output()).not.toContain("shutdown deadline exceeded");
226 }).pipe(Effect.scoped, Effect.provide(services)),
227 { timeout: 25_000 },
228 );
229
230 it.live(
231 "R05 bounds a stuck request and instance finalizer below the Fly stop deadline",
232 () =>
233 Effect.gen(function* () {
234 const fixture = yield* spawnFixture({
235 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "2000",
236 HANG_FINALIZER: "1",
237 RECORD_SHUTDOWN_TIMING: "1",
238 });
239 const response = yield* fixture
240 .get("/hang")
241 .pipe(Effect.result, Effect.forkChild);
242 yield* waitForOutput(fixture, "request started");
243 yield* fixture.signal("SIGTERM");
244 yield* refusesNewRequests(fixture);
245 expect(
246 yield* fixture.handle.exitCode.pipe(Effect.timeout("5 seconds")),
247 ).toBe(1);
248 const elapsed = yield* shutdownElapsed(fixture);
249 expect(elapsed).toBeGreaterThanOrEqual(1500);
250 expect(elapsed).toBeLessThan(2000);
251 expect(Result.isFailure(yield* Fiber.join(response))).toBe(true);
252 yield* waitForOutput(fixture, "shutdown deadline exceeded");
253 expect(fixture.output()).toContain("request finalized");
254 expect(fixture.output()).toContain("instance finalizing");
255 expect(fixture.output()).not.toContain("instance finalized");
256 }).pipe(Effect.scoped, Effect.provide(services)),
257 { timeout: 25_000 },
258 );
259
260 it.live(
261 "R03/R07 run-only jobs survive SIGTERM until the worker drains and releases its client",
262 () =>
263 Effect.gen(function* () {
264 const fixture = yield* spawnFixture({ RUN_ONLY: "1", WORKERS: "1" });
265 yield* fixture.signal("SIGTERM");
266 yield* waitForOutput(fixture, "worker1 stop started");
267 yield* fixture.signal("SIGTERM");
268 yield* fixture.signal("SIGINT");
269 expect(yield* fixture.handle.exitCode).toBe(0);
270 yield* waitForOutput(fixture, "instance finalized");
271 const output = fixture.output();
272 assertWorkerOrder(output, "worker1");
273 before(output, "worker1 stopped", "worker1 job 1 completed");
274 before(output, "worker1 job 1 completed", "worker1 drained");
275 before(output, "worker1 client released", "instance finalizing");
276 expect(output.match(/worker1 stop started/g)?.length).toBe(1);
277 }).pipe(Effect.scoped, Effect.provide(services)),
278 { timeout: 25_000 },
279 );
280
281 it.live(
282 "R03/R07 mixed HTTP and jobs begin stopping concurrently and retain shared dependencies",
283 () =>
284 Effect.gen(function* () {
285 const fixture = yield* spawnFixture({
286 WORKERS: "2",
287 NESTED_WORKER: "1",
288 });
289 const response = yield* fixture.get("/slow").pipe(
290 Effect.flatMap((response) => response.text),
291 Effect.forkChild,
292 );
293 yield* waitForOutput(fixture, "request started");
294 yield* fixture.signal("SIGINT");
295 yield* refusesNewRequests(fixture);
296 expect(yield* Fiber.join(response)).toBe("completed");
297 expect(yield* fixture.handle.exitCode).toBe(0);
298 yield* waitForOutput(fixture, "instance finalized");
299 const output = fixture.output();
300 for (const name of ["worker1", "worker2"]) {
301 assertWorkerOrder(output, name);
302 before(output, `${name} stopped`, "response ready");
303 before(output, `${name} client released`, "instance finalizing");
304 }
305 assertFinalizerOrder(output, "response ready");
306 }).pipe(Effect.scoped, Effect.provide(services)),
307 { timeout: 25_000 },
308 );
309
310 it.live(
311 "R04 SIGINT preserves a slow stream's length and checksum while jobs drain",
312 () =>
313 Effect.gen(function* () {
314 const fixture = yield* spawnFixture({ WORKERS: "1" });
315 const first = yield* Deferred.make<void>();
316 const chunks: string[] = [];
317 const response = yield* fixture.get("/stream-large").pipe(
318 Effect.flatMap((response) =>
319 response.stream.pipe(
320 Stream.decodeText,
321 Stream.runForEach((chunk) =>
322 Effect.sync(() => {
323 chunks.push(chunk);
324 Deferred.doneUnsafe(first, Effect.void);
325 }),
326 ),
327 ),
328 ),
329 Effect.forkChild,
330 );
331 yield* Deferred.await(first);
332 yield* fixture.signal("SIGINT");
333 yield* refusesNewRequests(fixture);
334 yield* Fiber.join(response);
335 const actual = chunks.join("");
336 const expected = Array.from({ length: 16 }, (_, index) =>
337 String(index).padStart(2, "0").repeat(8192),
338 ).join("");
339 expect(actual.length).toBe(262144);
340 const hashes = yield* Effect.sync(() =>
341 [actual, expected].map((body) =>
342 createHash("sha256").update(body).digest("hex"),
343 ),
344 );
345 expect(hashes[0]).toBe(hashes[1]);
346 expect(yield* fixture.handle.exitCode).toBe(0);
347 yield* waitForOutput(fixture, "instance finalized");
348 assertWorkerOrder(fixture.output(), "worker1");
349 assertFinalizerOrder(fixture.output(), "chunk last");
350 }).pipe(Effect.scoped, Effect.provide(services)),
351 { timeout: 25_000 },
352 );
353
354 it.live(
355 "R05 delayed acquisition stop is a barrier without delaying another worker",
356 () =>
357 Effect.gen(function* () {
358 const fixture = yield* spawnFixture({
359 RUN_ONLY: "1",
360 WORKERS: "2",
361 WORKER_MODE: "stop-delay",
362 });
363 yield* fixture.signal("SIGTERM");
364 expect(yield* fixture.handle.exitCode).toBe(0);
365 yield* waitForOutput(fixture, "instance finalized");
366 const output = fixture.output();
367 assertWorkerOrder(output, "worker1");
368 assertWorkerOrder(output, "worker2");
369 before(output, "worker2 stopped", "worker1 stopped");
370 before(output, "worker1 claim 2", "worker1 stopped");
371 const claims = [...output.matchAll(/worker1 claim (\d+)/g)].map(
372 (match) => match[1],
373 );
374 for (const id of claims) {
375 before(output, `worker1 job ${id} completed`, "worker1 drained");
376 }
377 }).pipe(Effect.scoped, Effect.provide(services)),
378 { timeout: 25_000 },
379 );
380
381 it.live(
382 "R05 a failed stop checkpoints without a success drain and cannot suppress HTTP or another worker",
383 () =>
384 Effect.gen(function* () {
385 const fixture = yield* spawnFixture({
386 WORKERS: "2",
387 WORKER_MODE: "stop-fail",
388 });
389 const response = yield* fixture.get("/slow").pipe(
390 Effect.flatMap((response) => response.text),
391 Effect.forkChild,
392 );
393 yield* waitForOutput(fixture, "request started");
394 yield* fixture.signal("SIGTERM");
395 yield* refusesNewRequests(fixture);
396 expect(yield* Fiber.join(response)).toBe("completed");
397 expect(yield* fixture.handle.exitCode).toBe(1);
398 yield* waitForOutput(fixture, "instance finalized");
399 const output = fixture.output();
400 expect(output).toContain("worker1 checkpoint");
401 expect(output).not.toContain("worker1 drain started");
402 expect(output).not.toContain("worker1 stopped");
403 before(output, "worker1 checkpoint", "worker1 work scope closed");
404 before(output, "worker1 work scope closed", "worker1 client released");
405 before(output, "worker1 client released", "instance finalizing");
406 assertWorkerOrder(output, "worker2");
407 before(output, "worker2 client released", "instance finalizing");
408 assertFinalizerOrder(output, "response ready");
409 expect(output).toContain("Managed process cleanup failed");
410 }).pipe(Effect.scoped, Effect.provide(services)),
411 { timeout: 25_000 },
412 );
413
414 for (const mode of ["stop-hang", "job-hang"]) {
415 it.live(
416 `R05 ${mode} cannot suppress another worker or extend the absolute hard deadline`,
417 () =>
418 Effect.gen(function* () {
419 const fixture = yield* spawnFixture({
420 WORKERS: "2",
421 WORKER_MODE: mode,
422 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "2000",
423 });
424 const response = yield* fixture.get("/slow").pipe(
425 Effect.flatMap((response) => response.text),
426 Effect.forkChild,
427 );
428 yield* waitForOutput(fixture, "request started");
429 const started = yield* Effect.sync(() => performance.now());
430 yield* fixture.signal("SIGTERM");
431 yield* refusesNewRequests(fixture);
432 expect(yield* Fiber.join(response)).toBe("completed");
433 yield* fixture.signal("SIGINT");
434 expect(yield* fixture.handle.exitCode).toBe(1);
435 const elapsed = yield* Effect.sync(() => performance.now() - started);
436 expect(elapsed).toBeGreaterThanOrEqual(1700);
437 expect(elapsed).toBeLessThan(2200);
438 yield* waitForOutput(fixture, "shutdown deadline exceeded");
439 const output = fixture.output();
440 assertWorkerOrder(output, "worker2");
441 expect(output).toContain("request finalized");
442 expect(output).not.toContain("instance finalizing");
443 expect(output).not.toContain("worker1 client released");
444 expect(output.match(/worker1 stop started/g)?.length).toBe(1);
445 if (mode === "stop-hang")
446 expect(output).not.toContain("worker1 drain started");
447 }).pipe(Effect.scoped, Effect.provide(services)),
448 { timeout: 25_000 },
449 );
450 }
451
452 for (const mode of ["normal", "error"]) {
453 it.live(
454 `R03 ${mode} run completion cleans jobs and shared dependencies without a signal`,
455 () =>
456 Effect.gen(function* () {
457 const fixture = yield* spawnFixture({
458 RUN_ONLY: "1",
459 WORKERS: "1",
460 WORKER_MODE: mode,
461 });
462 expect(yield* fixture.handle.exitCode).toBe(
463 mode === "normal" ? 0 : 1,
464 );
465 yield* waitForOutput(fixture, "instance finalized");
466 assertWorkerOrder(fixture.output(), "worker1");
467 before(
468 fixture.output(),
469 "worker1 client released",
470 "instance finalizing",
471 );
472 if (mode === "error")
473 expect(fixture.output()).toContain("worker failed");
474 }).pipe(Effect.scoped, Effect.provide(services)),
475 { timeout: 25_000 },
476 );
477 }
478
479 it.live(
480 "R03 a completed background runner does not stop a still-running HTTP service",
481 () =>
482 Effect.gen(function* () {
483 const fixture = yield* spawnFixture({
484 WORKERS: "1",
485 WORKER_MODE: "normal",
486 });
487 yield* waitForOutput(fixture, "worker1 client released");
488 yield* Effect.sleep("2 seconds");
489 expect((yield* fixture.get("/health")).status).toBe(200);
490 yield* fixture.signal("SIGTERM");
491 expect(yield* fixture.handle.exitCode).toBe(0);
492 yield* waitForOutput(fixture, "instance finalized");
493 assertWorkerOrder(fixture.output(), "worker1");
494 expect(fixture.output()).not.toContain("deadline exceeded");
495 }).pipe(Effect.scoped, Effect.provide(services)),
496 { timeout: 25_000 },
497 );
498
499 it.live(
500 "R03 a runner error initiates sibling cleanup while an HTTP response is still in flight",
501 () =>
502 Effect.gen(function* () {
503 const fixture = yield* spawnFixture({
504 WORKERS: "2",
505 WORKER_MODE: "error-delay",
506 });
507 const response = yield* fixture.get("/slow").pipe(
508 Effect.flatMap((response) => response.text),
509 Effect.forkChild,
510 );
511 yield* waitForOutput(fixture, "worker1 stop started");
512 yield* refusesNewRequests(fixture);
513 expect(yield* Fiber.join(response)).toBe("completed");
514 expect(yield* fixture.handle.exitCode).toBe(1);
515 yield* waitForOutput(fixture, "instance finalized");
516 const output = fixture.output();
517 for (const name of ["worker1", "worker2"]) {
518 assertWorkerOrder(output, name);
519 before(output, `${name} stopped`, "response ready");
520 before(output, `${name} client released`, "instance finalizing");
521 }
522 assertFinalizerOrder(output, "response ready");
523 expect(output).toContain("worker failed");
524 }).pipe(Effect.scoped, Effect.provide(services)),
525 { timeout: 25_000 },
526 );
527
528 for (const route of ["/slow", "/stream"]) {
529 it.live(
530 `review R2 ${route} request cleanup defects survive SIGTERM and independent cleanup`,
531 () =>
532 Effect.gen(function* () {
533 const fixture = yield* spawnFixture({
534 WORKERS: "2",
535 FAIL_REQUEST_FINALIZER: "1",
536 });
537 const response = yield* fixture.get(route).pipe(
538 Effect.flatMap((response) => response.text),
539 Effect.forkChild,
540 );
541 yield* waitForOutput(fixture, "request started");
542 yield* fixture.signal("SIGTERM");
543 expect(yield* Fiber.join(response)).toBe(
544 route === "/slow" ? "completed" : "first\nsecond\nlast\n",
545 );
546 expect(yield* fixture.handle.exitCode).toBe(1);
547 yield* waitForOutput(fixture, "instance finalized");
548 const output = fixture.output();
549 expect(output).toContain("Managed HTTP request cleanup failed");
550 expect(output).toContain("request cleanup defect");
551 before(output, "request dependency released", "instance finalizing");
552 assertWorkerOrder(output, "worker1");
553 assertWorkerOrder(output, "worker2");
554 before(output, "worker2 client released", "instance finalizing");
555 }).pipe(Effect.scoped, Effect.provide(services)),
556 { timeout: 25_000 },
557 );
558 }
559
560 it.live(
561 "review R2 ordinary handler defects do not become managed cleanup failures",
562 () =>
563 Effect.gen(function* () {
564 const fixture = yield* spawnFixture();
565 expect((yield* fixture.get("/handler-error")).status).toBe(500);
566 yield* waitForOutput(fixture, "request dependency released");
567 yield* fixture.signal("SIGTERM");
568 expect(yield* fixture.handle.exitCode).toBe(0);
569 expect(fixture.output()).toContain("ordinary handler defect");
570 expect(fixture.output()).not.toContain(
571 "Managed HTTP request cleanup failed",
572 );
573 }).pipe(Effect.scoped, Effect.provide(services)),
574 { timeout: 25_000 },
575 );
576
577 it.live(
578 "review R2 expected client abort does not become a cleanup failure",
579 () =>
580 Effect.gen(function* () {
581 const fixture = yield* spawnFixture();
582 const response = yield* fixture.get("/hang").pipe(Effect.forkChild);
583 yield* waitForOutput(fixture, "request started");
584 yield* Fiber.interrupt(response);
585 yield* waitForOutput(fixture, "request dependency released");
586 yield* fixture.signal("SIGTERM");
587 expect(yield* fixture.handle.exitCode).toBe(0);
588 expect(fixture.output()).not.toContain(
589 "Managed HTTP request cleanup failed",
590 );
591 }).pipe(Effect.scoped, Effect.provide(services)),
592 { timeout: 25_000 },
593 );
594
595 for (const exit of ["normal", "error"]) {
596 it.live(
597 `review R1 opaque nested ${exit} cleanup is bounded from an actual signal, not its hidden body exit`,
598 () =>
599 Effect.gen(function* () {
600 const fixture = yield* spawnFixture({
601 WORKERS: "2",
602 NESTED_WORKER: "1",
603 WORKER_MODE: "stop-hang",
604 RUN_EXIT: exit,
605 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "2000",
606 });
607 yield* waitForOutput(fixture, "worker1 stop started");
608 yield* Effect.sleep("2200 millis");
609 expect((yield* fixture.get("/health")).status).toBe(200);
610 expect(fixture.output()).not.toContain("shutdown deadline exceeded");
611 expect(fixture.output()).not.toContain("worker2 stop started");
612 const signaled = yield* Effect.sync(() => performance.now());
613 yield* fixture.signal("SIGTERM");
614 yield* refusesNewRequests(fixture);
615 expect(yield* fixture.handle.exitCode).toBe(1);
616 const elapsed = yield* Effect.sync(
617 () => performance.now() - signaled,
618 );
619 expect(elapsed).toBeGreaterThanOrEqual(1700);
620 expect(elapsed).toBeLessThan(2500);
621 yield* waitForOutput(fixture, "shutdown deadline exceeded");
622 assertWorkerOrder(fixture.output(), "worker2");
623 expect(fixture.output()).not.toContain("instance finalizing");
624 }).pipe(Effect.scoped, Effect.provide(services)),
625 { timeout: 25_000 },
626 );
627 }
628
629 for (const signal of ["SIGTERM", "SIGINT"] as const) {
630 it.live(
631 `review R1 recommended scoped never worker drains independently on ${signal}`,
632 () =>
633 Effect.gen(function* () {
634 const fixture = yield* spawnFixture({
635 RUN_ONLY: "1",
636 WORKERS: "1",
637 NESTED_WORKER: "1",
638 });
639 yield* fixture.signal(signal);
640 expect(yield* fixture.handle.exitCode).toBe(0);
641 yield* waitForOutput(fixture, "instance finalized");
642 assertWorkerOrder(fixture.output(), "worker1");
643 before(
644 fixture.output(),
645 "worker1 job 1 completed",
646 "worker1 client released",
647 );
648 before(
649 fixture.output(),
650 "worker1 client released",
651 "instance finalizing",
652 );
653 }).pipe(Effect.scoped, Effect.provide(services)),
654 { timeout: 25_000 },
655 );
656 }
657
658 for (const exit of ["normal", "error"]) {
659 it.live(
660 `R05 ${exit} completion starts a deadline before runtime-owned scope cleanup`,
661 () =>
662 Effect.gen(function* () {
663 const fixture = yield* spawnFixture({
664 RUN_ONLY: "1",
665 WORKERS: "1",
666 WORKER_MODE: "stop-hang",
667 RUN_EXIT: exit,
668 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "2000",
669 });
670 expect(yield* fixture.handle.exitCode).toBe(1);
671 yield* waitForOutput(fixture, "shutdown deadline exceeded");
672 expect(fixture.output()).toContain("worker1 stop started");
673 expect(fixture.output()).not.toContain("instance finalizing");
674 }).pipe(Effect.scoped, Effect.provide(services)),
675 { timeout: 25_000 },
676 );
677 }
678
679 it.live(
680 "R05 an initially empty job set cannot bypass a delayed acquisition stop",
681 () =>
682 Effect.gen(function* () {
683 const fixture = yield* spawnFixture({
684 RUN_ONLY: "1",
685 WORKERS: "1",
686 WORKER_MODE: "stop-delay-empty",
687 });
688 expect(fixture.output()).not.toContain("worker1 claim");
689 yield* fixture.signal("SIGTERM");
690 expect(yield* fixture.handle.exitCode).toBe(0);
691 yield* waitForOutput(fixture, "instance finalized");
692 const output = fixture.output();
693 before(output, "worker1 stop started", "worker1 claim 1");
694 assertWorkerOrder(output, "worker1");
695 for (const match of output.matchAll(/worker1 claim (\d+)/g)) {
696 before(
697 output,
698 `worker1 job ${match[1]} completed`,
699 "worker1 drained",
700 );
701 }
702 }).pipe(Effect.scoped, Effect.provide(services)),
703 { timeout: 25_000 },
704 );
705
706 it.live(
707 "R05 dependency finalizer failure retains a nonzero outcome after successful drains",
708 () =>
709 Effect.gen(function* () {
710 const fixture = yield* spawnFixture({
711 RUN_ONLY: "1",
712 WORKERS: "2",
713 FAIL_FINALIZER: "1",
714 });
715 yield* fixture.signal("SIGTERM");
716 expect(yield* fixture.handle.exitCode).toBe(1);
717 yield* waitForOutput(fixture, "dependency finalizer failed");
718 assertWorkerOrder(fixture.output(), "worker1");
719 assertWorkerOrder(fixture.output(), "worker2");
720 before(
721 fixture.output(),
722 "worker2 client released",
723 "instance finalizing",
724 );
725 }).pipe(Effect.scoped, Effect.provide(services)),
726 { timeout: 25_000 },
727 );
728
729 it.live(
730 "R06 accepts the 300s parser boundary without waiting for that deadline",
731 () =>
732 Effect.gen(function* () {
733 const fixture = yield* spawnFixture({
734 RUN_ONLY: "1",
735 WORKERS: "1",
736 WORKER_MODE: "normal",
737 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: "300000",
738 });
739 expect(yield* fixture.handle.exitCode).toBe(0);
740 yield* waitForOutput(fixture, "instance finalized");
741 }).pipe(Effect.scoped, Effect.provide(services)),
742 { timeout: 25_000 },
743 );
744
745 it.live(
746 "R06 run-only workers without opt-in retain native signal termination",
747 () =>
748 Effect.gen(function* () {
749 const fixture = yield* spawnFixture({
750 RUN_ONLY: "1",
751 WORKERS: "1",
752 ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: undefined,
753 });
754 yield* fixture.signal("SIGINT");
755 expect(
756 Result.isFailure(yield* fixture.handle.exitCode.pipe(Effect.result)),
757 ).toBe(true);
758 expect(fixture.output()).not.toContain("worker1 stop started");
759 expect(fixture.output()).not.toContain("instance finalizing");
760 }).pipe(Effect.scoped, Effect.provide(services)),
761 { timeout: 25_000 },
762 );
763
764 for (const [name, env] of [
765 [
766 "Fly without the shutdown env",
767 { ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: undefined },
768 ],
769 ["unrelated hosts even with the Fly env", { UNMANAGED_HOST: "1" }],
770 ] as const) {
771 it.live(
772 `R06 preserves default signal behavior for ${name}`,
773 () =>
774 Effect.gen(function* () {
775 const fixture = yield* spawnFixture(env);
776 const response = yield* fixture
777 .get("/slow")
778 .pipe(Effect.result, Effect.forkChild);
779 yield* waitForOutput(fixture, "request started");
780 yield* fixture.signal("SIGTERM");
781 const exit = yield* fixture.handle.exitCode.pipe(Effect.result);
782 expect(Result.isFailure(exit)).toBe(true);
783 expect(Result.isFailure(yield* Fiber.join(response))).toBe(true);
784 expect(fixture.output()).not.toContain("response ready");
785 expect(fixture.output()).not.toContain("instance finalizing");
786 }).pipe(Effect.scoped, Effect.provide(services)),
787 { timeout: 25_000 },
788 );
789 }
790
791 for (const timeout of [
792 "0",
793 "-1",
794 "1.5",
795 "invalid",
796 "300001",
797 "9007199254740992",
798 ]) {
799 it.live(
800 `R06 rejects invalid shutdown timeout ${timeout}`,
801 () =>
802 Effect.gen(function* () {
803 const fixture = yield* spawnFixture(
804 { ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS: timeout },
805 false,
806 );
807 expect(yield* fixture.handle.exitCode).toBe(1);
808 yield* waitForOutput(fixture, "must be a positive integer");
809 expect(fixture.output()).not.toContain("ready");
810 }).pipe(Effect.scoped, Effect.provide(services)),
811 { timeout: 25_000 },
812 );
813 }
814});