1import * as machines from "@distilled.cloud/fly-io/machines";
2import * as Fly from "@/Fly";
3import * as Output from "@/Output";
4import type { ScratchStack } from "@/Test/Alchemy";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { FileSystem } from "effect/FileSystem";
8import * as Path from "effect/Path";
9import * as Redacted from "effect/Redacted";
10import * as Schedule from "effect/Schedule";
11import * as HttpClient from "effect/http/HttpClient";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import { randomBytes } from "node:crypto";
14import { Ledger, ledgerLayer } from "./bluegreen-worker-ledger.ts";
15import {
16 Worker,
17 workerLayer,
18 type WorkerOptions,
19} from "./bluegreen-worker-managed.ts";
20import {
21 Cache,
22 LedgerSite,
23 WorkerSite,
24 services,
25 type LedgerEvent,
26 type Snapshot,
27 scripts,
28} from "./bluegreen-worker-shared.ts";
29
30const readinessProbe = `
31const report = (value) => console.log(value);
32report(process.env.REDIS_URL ? "redis-url:present" : "redis-url:missing");
33report(process.env.LEDGER_TOKEN ? "ledger-token:present" : "ledger-token:missing");
34report(["1", "2"].includes(process.env.WORKERS) ? "workers:valid" : "workers:missing-or-invalid");
35report(["0", "1"].includes(process.env.RUN_ONLY) ? "run-only:valid" : "run-only:missing-or-invalid");
36try {
37 const value = JSON.parse(process.env.WORKERS);
38 report(typeof value === "number" ? "workers-decoded:number" : "workers-decoded:other");
39} catch { report("workers-decoded:other"); }
40for (const [family, host] of [["ipv4", "127.0.0.1"], ["ipv6", "[::1]"]]) {
41 try {
42 const response = await fetch("http://" + host + ":3000/health", {
43 signal: AbortSignal.timeout(1500),
44 });
45 report(family + (response.status === 200 ? ":ready" : response.status === 503 ? ":starting" : ":other-status"));
46 await response.body?.cancel();
47 } catch { report(family + ":unreachable"); }
48}
49if (process.env.LEDGER_URL) {
50 try {
51 const response = await fetch(process.env.LEDGER_URL, {
52 method: "POST",
53 headers: { authorization: "Bearer " + process.env.LEDGER_TOKEN, "content-type": "application/json" },
54 body: JSON.stringify({ operation: "snapshot", args: [] }),
55 signal: AbortSignal.timeout(1500),
56 });
57 report(response.status === 200 ? "ledger:authenticated" : response.status === 401 ? "ledger:unauthorized" : "ledger:other-status");
58 await response.body?.cancel();
59 } catch { report("ledger:unreachable"); }
60}
61`;
62
63const readinessProbeLabels = new Set([
64 "redis-url:present",
65 "redis-url:missing",
66 "ledger-token:present",
67 "ledger-token:missing",
68 "workers:valid",
69 "workers:missing-or-invalid",
70 "run-only:valid",
71 "run-only:missing-or-invalid",
72 "workers-decoded:number",
73 "workers-decoded:other",
74 "ipv4:ready",
75 "ipv4:starting",
76 "ipv4:other-status",
77 "ipv4:unreachable",
78 "ipv6:ready",
79 "ipv6:starting",
80 "ipv6:other-status",
81 "ipv6:unreachable",
82 "ledger:authenticated",
83 "ledger:unauthorized",
84 "ledger:other-status",
85 "ledger:unreachable",
86]);
87
88const diagnoseMachines = (appName: string, role: "worker" | "ledger") =>
89 Effect.gen(function* () {
90 const live = yield* machines
91 .listMachines({ app_name: appName })
92 .pipe(Effect.timeout("8 seconds"));
93 yield* Effect.logError("Worker readiness machine census", {
94 role,
95 appName,
96 machines: live.map((machine) => ({
97 id: machine.id,
98 state: machine.state,
99 checks: machine.checks?.map((check) => ({
100 name: check.name,
101 status: check.status,
102 })),
103 })),
104 });
105 for (const machine of live.slice(0, 2)) {
106 const machineId = machine.id;
107 if (!machineId || machine.state !== "started") continue;
108 yield* machines
109 .execMachine({
110 app_name: appName,
111 machine_id: machineId,
112 command: ["node", "--input-type=module", "-e", readinessProbe],
113 timeout: 6,
114 })
115 .pipe(
116 Effect.timeout("8 seconds"),
117 Effect.flatMap((response) =>
118 Effect.logError("Worker readiness safe exec observations", {
119 role,
120 machineId,
121 exitCode: response.exit_code,
122 observations: (response.stdout ?? "")
123 .split("\n")
124 .filter((line) => readinessProbeLabels.has(line)),
125 }),
126 ),
127 Effect.catchCause(() =>
128 Effect.logWarning("Worker readiness exec unavailable", {
129 role,
130 machineId,
131 }),
132 ),
133 );
134 }
135 }).pipe(
136 Effect.catchCause(() =>
137 Effect.logWarning("Worker readiness machine census unavailable", {
138 role,
139 appName,
140 }),
141 ),
142 );
143
144export const requireValue = <A>(value: A | undefined, name: string) =>
145 value === undefined
146 ? Effect.fail(new Error(`Missing ${name}`))
147 : Effect.succeed(value);
148
149export const makeScenario = (stack: ScratchStack) =>
150 Effect.gen(function* () {
151 const token = yield* Effect.sync(() =>
152 Array.from(randomBytes(32), (byte) =>
153 byte.toString(16).padStart(2, "0"),
154 ).join(""),
155 );
156 const fs = yield* FileSystem;
157 const path = yield* Path.Path;
158 const raw = yield* fs.readFileString(
159 yield* path.fromFileUrl(
160 new URL("./bluegreen-worker-raw.mjs", import.meta.url),
161 ),
162 );
163 const deploy = (
164 options: WorkerOptions & {
165 raw?: boolean;
166 rawSignal?: "SIGQUIT" | "SIGTERM";
167 gatewayOnly?: boolean;
168 },
169 ) =>
170 stack.deploy(
171 Effect.gen(function* () {
172 const cache = yield* Cache;
173 const ledgerApp = yield* LedgerSite;
174 const ledgerSecret = yield* Fly.Secret("LedgerToken", {
175 app: ledgerApp,
176 name: "LEDGER_TOKEN",
177 value: Redacted.make(token),
178 });
179 yield* Fly.IpAssignment("LedgerIp", {
180 app: ledgerApp,
181 type: "shared_v4",
182 });
183 const ledger = yield* Ledger.pipe(
184 Effect.provide(ledgerLayer(ledgerSecret.digest)),
185 );
186 const ledgerUrl = ledger.url.pipe(
187 Output.mapEffect((url) =>
188 url
189 ? Effect.succeed(url)
190 : Effect.die(new Error("Ledger service URL missing")),
191 ),
192 );
193 const workerApp = yield* WorkerSite;
194 const workerSecret = yield* Fly.Secret("WorkerToken", {
195 app: workerApp,
196 name: "LEDGER_TOKEN",
197 value: Redacted.make(token),
198 });
199 if (!options.runOnly)
200 yield* Fly.IpAssignment("WorkerIp", {
201 app: workerApp,
202 type: "shared_v4",
203 });
204 const worker = options.gatewayOnly
205 ? undefined
206 : options.raw
207 ? yield* Fly.Machine("RawWorker", {
208 app: workerApp,
209 image: "node:26-slim",
210 region: "iad",
211 guest: { cpuKind: "shared", cpus: 1, memoryMb: 256 },
212 init: { exec: ["node", "--input-type=module", "-e", raw] },
213 env: {
214 VERSION: options.version,
215 MODE: options.mode ?? "drain",
216 AFTER_SIGNAL_MS: String(options.afterSignalMs ?? 1000),
217 SHUTDOWN_MS: String(
218 options.timeout === "60 seconds"
219 ? 60000
220 : options.timeout === "10 seconds"
221 ? 10000
222 : 30000,
223 ),
224 LEDGER_URL: ledgerUrl,
225 SECRET_READY: workerSecret.digest.pipe(
226 Output.mapEffect((digest) =>
227 requireValue(digest, "worker secret digest").pipe(
228 Effect.orDie,
229 ),
230 ),
231 ),
232 },
233 deploy: {
234 strategy: "bluegreen",
235 healthTimeout: "45 seconds",
236 },
237 shutdown: {
238 signal: options.rawSignal ?? options.signal,
239 timeout: options.timeout,
240 },
241 services: options.runOnly ? [] : services,
242 checks: options.runOnly
243 ? {
244 ready: {
245 type: "http",
246 port: 3000,
247 path: "/health",
248 interval: "2s",
249 timeout: "1s",
250 },
251 }
252 : undefined,
253 })
254 : yield* Worker.pipe(Effect.provide(workerLayer(options)));
255 return { cache, ledger, ledgerUrl, worker, workerApp, ledgerApp };
256 }),
257 );
258 type Deployment = Effect.Success<ReturnType<typeof deploy>>;
259 let inventory: Deployment | undefined;
260 const trackedDeploy = (options: Parameters<typeof deploy>[0]) =>
261 Effect.gen(function* () {
262 if (inventory === undefined) {
263 inventory = yield* deploy({ ...options, gatewayOnly: true });
264 if (options.gatewayOnly) return inventory;
265 }
266 const value = yield* deploy(options);
267 inventory = value;
268 return value;
269 }).pipe(
270 Effect.tapCause(() => {
271 const observed = inventory;
272 if (!observed) return Effect.void;
273 return Effect.all(
274 [
275 diagnoseMachines(observed.workerApp.appName, "worker"),
276 diagnoseMachines(observed.ledgerApp.appName, "ledger"),
277 snapshot(observed.ledgerUrl).pipe(
278 Effect.flatMap((value) =>
279 Effect.logError(
280 "Worker lifecycle ledger at failure",
281 value.events.slice(-30),
282 ),
283 ),
284 Effect.catchCause(() =>
285 Effect.logWarning(
286 "Worker lifecycle ledger unavailable during failure diagnostics",
287 ),
288 ),
289 ),
290 ],
291 { concurrency: "unbounded", discard: true },
292 );
293 }),
294 );
295 const call = (
296 url: string,
297 operation: keyof typeof scripts,
298 args: string[] = [],
299 requestTimeout: "8 seconds" | "30 seconds" = "8 seconds",
300 ) =>
301 HttpClient.execute(
302 HttpClientRequest.post(url).pipe(
303 HttpClientRequest.setHeader("authorization", `Bearer ${token}`),
304 HttpClientRequest.bodyJsonUnsafe({ operation, args }),
305 ),
306 ).pipe(
307 Effect.flatMap((response) =>
308 response.status === 200
309 ? response.json
310 : Effect.fail(new Error(`ledger HTTP ${response.status}`)),
311 ),
312 Effect.map((body) => (body as { result: unknown }).result),
313 Effect.timeout(requestTimeout),
314 Effect.catchTag("TimeoutError", () =>
315 Effect.fail(
316 new Error(`Ledger ${operation} request exceeded ${requestTimeout}`),
317 ),
318 ),
319 );
320 const snapshot = (url: string) =>
321 // Durable observations have a separate budget from the worker's shutdown grace.
322 call(url, "snapshot", [], "30 seconds").pipe(
323 Effect.map((result) => {
324 const value = JSON.parse(String(result)) as Snapshot;
325 return {
326 ...value,
327 events: (Array.isArray(value.events) ? value.events : []).map(
328 (event) => JSON.parse(event) as LedgerEvent,
329 ),
330 results: Array.isArray(value.results) ? value.results : [],
331 checkpoints: Array.isArray(value.checkpoints)
332 ? value.checkpoints
333 : [],
334 produced: Array.isArray(value.produced) ? value.produced : [],
335 };
336 }),
337 );
338 const wait = (
339 url: string,
340 predicate: (
341 value: Effect.Success<ReturnType<typeof snapshot>>,
342 ) => boolean,
343 ) =>
344 snapshot(url).pipe(
345 Effect.repeat({
346 until: predicate,
347 times: 10,
348 schedule: Schedule.spaced("1 second"),
349 }),
350 Effect.timeout("45 seconds"),
351 Effect.catchTag("TimeoutError", () =>
352 Effect.fail(
353 new Error("Ledger predicate observation exceeded 45 seconds"),
354 ),
355 ),
356 Effect.tap((value) =>
357 Effect.sync(() => expect(predicate(value)).toBe(true)),
358 ),
359 );
360 const settle = (url: string) =>
361 Effect.gen(function* () {
362 yield* call(url, "pause");
363 const ledger = yield* wait(
364 url,
365 (value) =>
366 value.pending === 0 &&
367 value.results.length === value.produced.length * 2,
368 );
369 expect(ledger.entries).toBe(ledger.produced.length);
370 const completed = ledger.results.filter(
371 (_value, index) => index % 2 === 0,
372 );
373 expect([...completed].sort()).toEqual([...ledger.produced].sort());
374 return ledger;
375 });
376 const cleanup = Effect.gen(function* () {
377 yield* stack.destroy();
378 if (!inventory) return;
379 for (const appName of [
380 inventory.workerApp.appName,
381 inventory.ledgerApp.appName,
382 ]) {
383 const gone = yield* machines.getApp({ app_name: appName }).pipe(
384 Effect.as(false),
385 Effect.catchTag("NotFound", () => Effect.succeed(true)),
386 Effect.repeat({
387 until: (value) => value,
388 times: 8,
389 schedule: Schedule.spaced("1 second"),
390 }),
391 );
392 expect(gone).toBe(true);
393 }
394 expect(
395 yield* Fly.findRedisAddOn({
396 id: inventory.cache.redisId,
397 name: inventory.cache.name,
398 }),
399 ).toBeUndefined();
400 }).pipe(Effect.orDie);
401 return { deploy: trackedDeploy, call, snapshot, wait, settle, cleanup };
402 });
403
404export const assertOrder = (
405 events: LedgerEvent[],
406 machine: string,
407 first: string,
408 second: string,
409 worker?: string,
410) => {
411 const selected = events.filter(
412 (event) =>
413 event.machine === machine &&
414 (worker === undefined || event.worker === worker),
415 );
416 expect(
417 selected.findIndex((event) => event.event === first),
418 ).toBeGreaterThanOrEqual(0);
419 expect(selected.findIndex((event) => event.event === second)).toBeGreaterThan(
420 selected.findIndex((event) => event.event === first),
421 );
422};
423export const assertStopped = (events: LedgerEvent[], machine: string) => {
424 const stopped = events.findLastIndex(
425 (event) => event.machine === machine && event.event === "stopped",
426 );
427 expect(stopped).toBeGreaterThanOrEqual(0);
428 expect(
429 events
430 .slice(stopped + 1)
431 .filter(
432 (event) =>
433 event.machine === machine &&
434 ["claimed", "reclaimed", "producer"].includes(event.event),
435 ),
436 ).toEqual([]);
437};
438export const assertReplacement = (
439 appName: string,
440 oldId: string,
441 newId: string,
442) =>
443 Effect.gen(function* () {
444 expect(newId).not.toBe(oldId);
445 const live = (yield* machines.listMachines({ app_name: appName })).filter(
446 (machine) => machine.state !== "destroyed",
447 );
448 expect(live.map((machine) => machine.id)).toEqual([newId]);
449 expect(
450 yield* machines.getMachine({ app_name: appName, machine_id: oldId }).pipe(
451 Effect.map((machine) => machine.state === "destroyed"),
452 Effect.catchTag("NotFound", () => Effect.succeed(true)),
453 ),
454 ).toBe(true);
455 });