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