1import * as machines from "@distilled.cloud/fly-io/machines";
2import * as Retry from "@distilled.cloud/fly-io/Retry";
3import type { ScratchStack } from "@/Test/Alchemy";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Path from "effect/Path";
8import * as Schedule from "effect/Schedule";
9import * as Schema from "effect/Schema";
10import {
11 Witness as ProcessDeathWitness,
12 census,
13 evidencePaths,
14 identity,
15 machine,
16 nowSeconds,
17 persistedRow,
18} from "./process-death.ts";
19import type { TransportEvent } from "./transport.ts";
20
21export const signalOverlapFile = "test/Fly/BlueGreen.test.ts";
22export const cases = [
23 { name: "sigint-create", signal: "SIGINT", phase: "create" },
24 { name: "sigint-promotion", signal: "SIGINT", phase: "promotion" },
25 { name: "sigint-overlap", signal: "SIGINT", phase: "overlap" },
26 { name: "sigint-retirement", signal: "SIGINT", phase: "retirement" },
27 { name: "sigkill-overlap", signal: "SIGKILL", phase: "overlap" },
28] as const;
29export type SignalCase = (typeof cases)[number];
30
31const Boundary = Schema.Struct({
32 sequence: Schema.Number,
33 method: Schema.String,
34 path: Schema.String,
35 status: Schema.Number,
36 machineId: Schema.String,
37});
38
39export const Witness = Schema.Struct({
40 ...ProcessDeathWitness.fields,
41 signal: Schema.Literals(["SIGINT", "SIGKILL"]),
42 phase: Schema.Literals(["create", "promotion", "overlap", "retirement"]),
43 case: Schema.String,
44 barrier: Boundary,
45 // The subsequent routing read proves the provider consumed the uncordon response.
46 returnedUncordon: Schema.optional(Boundary),
47});
48export type Witness = typeof Witness.Type;
49
50const LeaseObservation = Schema.Struct({
51 machineId: Schema.String,
52 present: Schema.Boolean,
53 expiresAt: Schema.Number,
54 at: Schema.Number,
55});
56
57export const Finalized = Schema.Struct({
58 pid: Schema.Number,
59 case: Schema.String,
60 witnessRecordedAt: Schema.Number,
61 interruptedOnly: Schema.Boolean,
62 at: Schema.Number,
63 releases: Schema.Array(Boundary),
64 leases: Schema.Array(LeaseObservation),
65});
66
67export const RunnerInterrupted = Schema.Struct({
68 pid: Schema.Number,
69 case: Schema.String,
70 signal: Schema.Literal("SIGINT"),
71 witnessRecordedAt: Schema.Number,
72 at: Schema.Number,
73 deployFinalized: Schema.Boolean,
74});
75
76export const pathsFor = (stack: ScratchStack) =>
77 Effect.gen(function* () {
78 const paths = yield* evidencePaths(stack);
79 const path = yield* Path.Path;
80 return {
81 ...paths,
82 runnerInterrupted: path.join(paths.directory, "runner-interrupted.json"),
83 blocked: path.join(paths.directory, "interruption-limit.json"),
84 };
85 });
86
87export const readEvidence = <S extends Schema.Top>(file: string, schema: S) =>
88 Effect.gen(function* () {
89 const fs = yield* FileSystem.FileSystem;
90 return yield* fs.readFileString(file).pipe(
91 Effect.flatMap(
92 Schema.decodeUnknownEffect(Schema.fromJsonString(schema), {
93 onExcessProperty: "error",
94 }),
95 ),
96 Effect.mapError(
97 () =>
98 new Error("Missing or invalid signal evidence; preserve the attempt"),
99 ),
100 );
101 });
102
103export const assertSingleRunner = Effect.gen(function* () {
104 const args = yield* Effect.sync(() => process.argv.slice(2));
105 const focused =
106 args[0] === "--exclude" && args[1] === "test/Railway"
107 ? args.slice(2)
108 : args;
109 expect(focused).toEqual([
110 signalOverlapFile,
111 "-t",
112 " > process signals and overlap > ",
113 "--profile",
114 "testing",
115 "--retry",
116 "0",
117 "--concurrency",
118 "1",
119 "--sequential",
120 "--timeout",
121 "750000",
122 ]);
123 expect(process.env.FLY_SIGNAL_OVERLAP_ONLY_RUNNER).toBe("1");
124 expect(process.env.FLY_PROCESS_DEATH_CASE).toBeUndefined();
125 expect(process.env.FLY_PROCESS_DEATH_MODE).toBeUndefined();
126 expect(process.env.ALCHEMY_PROFILE).toBe("testing");
127 expect(process.env.ALCHEMY_TEST_DEV).toBe("0");
128 expect(process.env.ALCHEMY_DEV).toBe("0");
129 expect(process.env.ALCHEMY_TEST_STAGE).toBe("test_fly_signal_overlap");
130 expect(process.env.FAST).toBeUndefined();
131 expect(process.versions.bun).toBeDefined();
132 const fs = yield* FileSystem.FileSystem;
133 for (const file of [
134 signalOverlapFile,
135 "test/Fly/fixtures/signal-overlap.ts",
136 "test/Fly/fixtures/process-death.ts",
137 "test/Fly/fixtures/transport.ts",
138 ]) {
139 const source = yield* fs.readFileString(file);
140 expect(source).not.toMatch(
141 /(?:from\s+|import\s*(?:\(\s*)?|require\s*\()\s*["'](?:node:)?child_process["']/,
142 );
143 expect(source).not.toMatch(
144 /\b(?:Bun|Deno)\s*\.\s*(?:spawn|spawnSync|Command)\s*\(/,
145 );
146 expect(source).not.toMatch(
147 /from\s+["'][^"']*(?:ChildProcess|CommandExecutor)[^"']*["']/,
148 );
149 }
150});
151
152export const successful = (event: TransportEvent) =>
153 event.status !== undefined && event.status >= 200 && event.status < 300;
154
155export const boundary = (event: TransportEvent) => {
156 expect(successful(event)).toBe(true);
157 expect(event.machineId).toBeDefined();
158 return {
159 sequence: event.sequence,
160 method: event.method,
161 path: event.path,
162 status: event.status!,
163 machineId: event.machineId!,
164 };
165};
166
167export const firstReturnedUncordon = (
168 events: readonly TransportEvent[],
169 appName: string,
170) =>
171 events.find(
172 (event) =>
173 event.stage === "forwarded" &&
174 successful(event) &&
175 event.method === "POST" &&
176 event.path.startsWith(`/v1/apps/${appName}/machines/`) &&
177 event.path.endsWith("/uncordon"),
178 );
179
180export const matches = (
181 phase: SignalCase["phase"],
182 appName: string,
183 predecessorId: string,
184 events: readonly TransportEvent[],
185 event: TransportEvent,
186) => {
187 const base = `/v1/apps/${appName}/machines`;
188 if (phase === "create") return event.method === "POST" && event.path === base;
189 if (phase === "promotion")
190 return (
191 event.method === "POST" &&
192 event.path.startsWith(`${base}/`) &&
193 event.path.endsWith("/uncordon")
194 );
195 if (phase === "retirement")
196 return (
197 event.method === "DELETE" && event.path === `${base}/${predecessorId}`
198 );
199 const returned = firstReturnedUncordon(events, appName);
200 return (
201 returned !== undefined &&
202 returned.machineId !== predecessorId &&
203 event.sequence > returned.sequence &&
204 event.method === "GET" &&
205 event.path === `${base}/${returned.machineId}`
206 );
207};
208
209export const assertBoundary = (witness: Witness) =>
210 Effect.sync(() => {
211 const { barrier, returnedUncordon } = witness;
212 expect(barrier.sequence).toBeGreaterThan(0);
213 expect(barrier.status).toBeGreaterThanOrEqual(200);
214 expect(barrier.status).toBeLessThan(300);
215 expect(barrier.machineId).toBe(
216 witness.phase === "retirement"
217 ? witness.predecessor.id
218 : witness.candidate.id,
219 );
220 if (witness.phase === "overlap") {
221 expect(returnedUncordon).toBeDefined();
222 expect(returnedUncordon!.method).toBe("POST");
223 expect(returnedUncordon!.path).toBe(
224 `/v1/apps/${witness.appName}/machines/${witness.candidate.id}/uncordon`,
225 );
226 expect(returnedUncordon!.machineId).toBe(witness.candidate.id);
227 expect(returnedUncordon!.status).toBeGreaterThanOrEqual(200);
228 expect(returnedUncordon!.status).toBeLessThan(300);
229 expect(returnedUncordon!.sequence).toBeGreaterThan(0);
230 expect(barrier.sequence).toBeGreaterThan(returnedUncordon!.sequence);
231 } else expect(returnedUncordon).toBeUndefined();
232 expect(
233 matches(
234 witness.phase,
235 witness.appName,
236 witness.predecessor.id,
237 returnedUncordon ? [{ ...returnedUncordon, stage: "forwarded" }] : [],
238 { ...barrier, stage: "held" },
239 ),
240 ).toBe(true);
241 });
242
243export const assertInventory = (stack: ScratchStack, witness: Witness) =>
244 Effect.gen(function* () {
245 const live = yield* census(witness.appName);
246 expect(live.map((value) => value.id).sort()).toEqual(
247 (witness.phase === "retirement"
248 ? [witness.candidate.id]
249 : [witness.predecessor.id, witness.candidate.id]
250 ).sort(),
251 );
252 const candidate = yield* machine(witness.appName, witness.candidate.id);
253 expect(yield* identity(candidate, stack)).toEqual(witness.candidate);
254 expect(candidate.config?.env?.VERSION).toBe("two");
255 expect(candidate.cordoned).toBe(witness.phase === "create");
256 expect(witness.candidate.phase).toBe(
257 witness.phase === "create"
258 ? "candidate"
259 : witness.phase === "retirement"
260 ? "active"
261 : "promoting",
262 );
263 if (witness.phase !== "retirement") {
264 const predecessor = yield* machine(
265 witness.appName,
266 witness.predecessor.id,
267 );
268 expect(yield* identity(predecessor, stack)).toEqual(witness.predecessor);
269 expect(predecessor.cordoned).toBe(false);
270 expect(predecessor.state).toBe("started");
271 expect(predecessor.config?.env?.VERSION).toBe("one");
272 }
273 });
274
275// Only presence and timestamps escape the SDK; nonce and transport headers never do.
276export const observeLease = (appName: string, machineId: string) =>
277 machines.getMachineLease({ app_name: appName, machine_id: machineId }).pipe(
278 Retry.none,
279 Effect.map((value) => ({
280 present: !!value.data?.nonce,
281 expiresAt: value.data?.expires_at ?? 0,
282 })),
283 Effect.catchTag("NotFound", () =>
284 Effect.succeed({ present: false, expiresAt: 0 }),
285 ),
286 Effect.timeout("10 seconds"),
287 Effect.mapError(
288 (error) => new Error(`Signal lease observation failed (${error._tag})`),
289 ),
290 Effect.flatMap((value) =>
291 nowSeconds.pipe(Effect.map((at) => ({ machineId, ...value, at }))),
292 ),
293 );
294
295export const observeLeases = (witness: Witness) =>
296 Effect.forEach(
297 witness.leases,
298 (held) => observeLease(witness.appName, held.machineId),
299 { concurrency: 2 },
300 );
301
302export const observeExpiry = (witness: Witness) =>
303 Effect.gen(function* () {
304 expect(witness.signal).toBe("SIGKILL");
305 expect(witness.leases.length).toBeGreaterThan(0);
306 const initial = yield* observeLeases(witness);
307 for (const [index, value] of initial.entries()) {
308 expect(value.present).toBe(true);
309 expect(value.expiresAt).toBeGreaterThan(value.at);
310 expect(value.expiresAt).toBeGreaterThanOrEqual(
311 witness.leases[index]!.expiresAt,
312 );
313 expect(value.expiresAt).toBeLessThanOrEqual(witness.recordedAt + 130);
314 }
315 const samples: (typeof LeaseObservation.Type)[] = [];
316 const expired = yield* Effect.forEach(
317 initial,
318 (held) =>
319 Effect.gen(function* () {
320 const value = yield* observeLease(witness.appName, held.machineId);
321 yield* Effect.sync(() => samples.push(value));
322 if (value.present) expect(value.expiresAt).toBe(held.expiresAt);
323 else expect(value.at).toBeGreaterThanOrEqual(held.expiresAt);
324 return !value.present;
325 }),
326 { concurrency: 2 },
327 ).pipe(
328 Effect.repeat({
329 schedule: Schedule.spaced("3 seconds"),
330 times: 65,
331 until: (all) => all.every(Boolean),
332 }),
333 Effect.timeout("230 seconds"),
334 );
335 expect(expired.every(Boolean)).toBe(true);
336 for (const held of initial) {
337 const surviving = yield* machine(witness.appName, held.machineId);
338 expect(surviving.state).not.toBe("destroyed");
339 }
340 return { initial, samples };
341 });
342
343export const assertReleased = (
344 witness: Witness,
345 finalized: typeof Finalized.Type,
346) =>
347 Effect.gen(function* () {
348 expect(witness.signal).toBe("SIGINT");
349 expect(witness.leases.length).toBeGreaterThan(0);
350 expect(finalized.pid).toBe(witness.pid);
351 expect(finalized.case).toBe(witness.case);
352 expect(finalized.witnessRecordedAt).toBe(witness.recordedAt);
353 expect(finalized.interruptedOnly).toBe(true);
354 expect(finalized.at).toBeGreaterThanOrEqual(witness.recordedAt);
355 expect(finalized.leases.map((value) => value.machineId).sort()).toEqual(
356 witness.leases.map((value) => value.machineId).sort(),
357 );
358 for (const held of witness.leases) {
359 const observed = finalized.leases.find(
360 (value) => value.machineId === held.machineId,
361 )!;
362 expect(observed.present).toBe(false);
363 expect(observed.at).toBeGreaterThanOrEqual(witness.recordedAt);
364 expect(observed.at).toBeLessThan(held.expiresAt);
365 expect(finalized.at).toBeLessThan(held.expiresAt);
366 const release = finalized.releases.find(
367 (event) =>
368 event.machineId === held.machineId &&
369 event.status >= 200 &&
370 event.status < 300,
371 );
372 expect(release).toBeDefined();
373 expect(release!.method).toBe("DELETE");
374 expect(release!.path).toBe(
375 `/v1/apps/${witness.appName}/machines/${held.machineId}/lease`,
376 );
377 expect(release!.sequence).toBeGreaterThan(witness.barrier.sequence);
378 // A deleted Machine cannot masquerade as a released lease.
379 const surviving = yield* machine(witness.appName, held.machineId);
380 expect(surviving.state).not.toBe("destroyed");
381 }
382 const current = yield* observeLeases(witness);
383 expect(current.every((value) => !value.present)).toBe(true);
384 return { finalized, current };
385 });
386
387export const assertConverged = (
388 stack: ScratchStack,
389 witness: Witness,
390 ids: string[],
391) =>
392 Effect.gen(function* () {
393 expect(ids).toEqual([witness.candidate.id]);
394 const live = yield* census(witness.appName);
395 expect(live.map((value) => value.id)).toEqual(ids);
396 const current = yield* machine(witness.appName, witness.candidate.id).pipe(
397 Effect.repeat({
398 schedule: Schedule.spaced("2 seconds"),
399 times: 20,
400 until: (value) =>
401 value.state === "started" &&
402 value.cordoned === false &&
403 value.checks?.some(
404 (check) => check.name === "ready" && check.status === "passing",
405 ) === true,
406 }),
407 Effect.timeout("60 seconds"),
408 );
409 expect(yield* identity(current, stack)).toEqual({
410 ...witness.candidate,
411 phase: "active",
412 });
413 expect(current.state).toBe("started");
414 expect(current.cordoned).toBe(false);
415 expect(current.config?.env?.VERSION).toBe("two");
416 const ready =
417 current.checks?.filter((check) => check.name === "ready") ?? [];
418 expect(ready.length).toBeGreaterThan(0);
419 expect(ready.every((check) => check.status === "passing")).toBe(true);
420 expect(current.instance_id).toBeDefined();
421 expect(current.config?.metadata?.["alchemy.checked-instance"]).toBe(
422 current.instance_id,
423 );
424 expect(current.config?.metadata?.["alchemy.predecessors"]).toBe(
425 witness.predecessor.id,
426 );
427 expect(
428 current.config?.metadata?.["alchemy.image"]?.endsWith(
429 `@${witness.candidate.digest}`,
430 ),
431 ).toBe(true);
432 const row = yield* persistedRow(stack, witness.row.fqn);
433 expect(row.instanceId).toBe(witness.row.instanceId);
434 expect(["created", "updated"]).toContain(row.status);
435 });