1import type * as machines from "@distilled.cloud/fly-io/machines";
2import type { ScratchStack } from "@/Test/Alchemy";
3import { scratchStack } from "@/Test/Core";
4import { expect } from "alchemy-test";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as HttpClient from "effect/http/HttpClient";
8import {
9 throughProxy,
10 transportProxy,
11 type TransportEvent,
12} from "./transport.ts";
13
14const metadataKeys = [
15 "alchemy.stack",
16 "alchemy.stage",
17 "alchemy.id",
18 "alchemy.type",
19 "alchemy.replica",
20 "alchemy.instance",
21 "alchemy.fqn",
22 "alchemy.base-name",
23 "alchemy.generation",
24 "alchemy.workload",
25 "alchemy.sequence",
26 "alchemy.count",
27 "alchemy.image",
28 "alchemy.phase",
29 "alchemy.deployment-protocol",
30 "alchemy.readiness-roles",
31 "alchemy.readiness-role",
32 "alchemy.predecessors",
33 "alchemy.idle-policy-restored",
34 "alchemy.checked-instance",
35 "alchemy.min-secrets-version",
36];
37
38const record = (value: unknown): Record<string, unknown> =>
39 typeof value === "object" && value !== null && !Array.isArray(value)
40 ? (value as Record<string, unknown>)
41 : {};
42const string = (value: unknown) =>
43 typeof value === "string" ? value : undefined;
44const metadataOf = (value: unknown) => {
45 const metadata = record(value);
46 return Object.fromEntries(
47 metadataKeys.flatMap((key) =>
48 typeof metadata[key] === "string" ? [[key, metadata[key] as string]] : [],
49 ),
50 );
51};
52const servicesOf = (value: unknown) =>
53 Array.isArray(value)
54 ? value.map((item) => {
55 const service = record(item);
56 return {
57 protocol: string(service.protocol),
58 port:
59 typeof service.internal_port === "number"
60 ? service.internal_port
61 : undefined,
62 autostop:
63 service.autostop === true
64 ? "stop"
65 : service.autostop === "stop" || service.autostop === "suspend"
66 ? service.autostop
67 : "off",
68 autostart:
69 typeof service.autostart === "boolean"
70 ? service.autostart
71 : undefined,
72 floor:
73 typeof service.min_machines_running === "number"
74 ? service.min_machines_running
75 : undefined,
76 ...(Array.isArray(service.checks)
77 ? {
78 checks: service.checks.map((value) => {
79 const check = record(value);
80 return {
81 type: string(check.type),
82 port:
83 typeof check.port === "number" ? check.port : undefined,
84 interval: string(check.interval),
85 timeout: string(check.timeout),
86 gracePeriod: string(check.grace_period),
87 method: string(check.method),
88 protocol: string(check.protocol),
89 };
90 }),
91 }
92 : {}),
93 };
94 })
95 : [];
96
97export interface ReadinessEvent extends TransportEvent {
98 metadata?: Record<string, string>;
99 metadataOnly?: boolean;
100 skipLaunch?: boolean;
101 services?: ReturnType<typeof servicesOf>;
102 serviceConfig?: string;
103}
104
105/** Only these two excluded suites install this observer; bodies and headers are never retained. */
106export const readinessProxy = () =>
107 transportProxy().pipe(
108 Effect.map((proxy) => ({ ...proxy, readiness: [] as ReadinessEvent[] })),
109 Effect.tap((proxy) =>
110 Effect.addFinalizer(() =>
111 Effect.forEach(
112 Array.from(
113 { length: Math.ceil(proxy.readiness.length / 50) },
114 (_, index) => index,
115 ),
116 (index) =>
117 Effect.logInfo("Idle/cadence readiness journal", {
118 offset: index * 50,
119 events: proxy.readiness.slice(index * 50, (index + 1) * 50),
120 }),
121 { discard: true },
122 ),
123 ),
124 ),
125 );
126
127type ReadinessProxy = Effect.Success<ReturnType<typeof readinessProxy>>;
128
129export const readinessActor = (
130 parent: ScratchStack,
131 title: string,
132 file: string,
133 proxy: ReadinessProxy,
134 beforeRequest?: (event: ReadinessEvent) => Effect.Effect<void>,
135) =>
136 Effect.sync(() => {
137 const observer = Layer.effect(
138 HttpClient.HttpClient,
139 Effect.gen(function* () {
140 const client = yield* HttpClient.HttpClient;
141 return HttpClient.transform(client, (response, request) =>
142 Effect.gen(function* () {
143 const event = yield* Effect.sync(() => {
144 const path = new URL(request.url).pathname;
145 if (!/\/apps\/[^/]+\/machines(?:\/|$)/.test(path))
146 return undefined;
147 const event: ReadinessEvent = {
148 sequence: proxy.readiness.length,
149 stage: "request",
150 method: request.method,
151 path,
152 machineId: path.match(/\/machines\/([^/]+)/)?.[1],
153 };
154 if (request.body._tag === "Uint8Array") {
155 try {
156 const body = record(
157 JSON.parse(new TextDecoder().decode(request.body.body)),
158 );
159 const config = record(body.config);
160 const metadata = record(body.metadata ?? config.metadata);
161 event.metadata = metadataOf(metadata);
162 event.phase = event.metadata["alchemy.phase"];
163 event.skipLaunch =
164 typeof body.skip_launch === "boolean"
165 ? body.skip_launch
166 : undefined;
167 event.metadataOnly =
168 Object.keys(body).length === 1 &&
169 "metadata" in body &&
170 Object.keys(metadata).every(
171 (key) =>
172 metadataKeys.includes(key) &&
173 typeof metadata[key] === "string",
174 );
175 } catch {
176 // An unreadable body cannot qualify for the bookkeeping exemption.
177 }
178 }
179 proxy.readiness.push(event);
180 return event;
181 });
182 if (event && beforeRequest) yield* beforeRequest(event);
183 const result = yield* response;
184 if (event) {
185 const value =
186 result.status >= 200 &&
187 result.status < 300 &&
188 /\/machines(?:\/[^/]+)?$/.test(event.path)
189 ? yield* result.json.pipe(
190 Effect.catch(() => Effect.succeed(undefined)),
191 )
192 : undefined;
193 yield* Effect.sync(() => {
194 const machine = record(value);
195 const config = record(machine.config);
196 const observed: ReadinessEvent = {
197 ...event,
198 // Receipt here is after the unmodified proxy forwarded the response.
199 stage: "forwarded",
200 status: result.status,
201 };
202 if (typeof machine.id === "string") {
203 observed.machineId = machine.id;
204 observed.instanceId = string(machine.instance_id);
205 observed.state = string(machine.state);
206 observed.cordoned =
207 typeof machine.cordoned === "boolean"
208 ? machine.cordoned
209 : undefined;
210 observed.digest = string(record(machine.image_ref).digest);
211 observed.metadata = metadataOf(config.metadata);
212 observed.phase = observed.metadata["alchemy.phase"];
213 observed.services = servicesOf(config.services);
214 observed.serviceConfig = JSON.stringify(observed.services);
215 observed.checks = Array.isArray(machine.checks)
216 ? machine.checks.map((item) => {
217 const check = record(item);
218 return {
219 name: string(check.name),
220 status: string(check.status),
221 };
222 })
223 : [];
224 }
225 proxy.readiness.push(observed);
226 });
227 }
228 return result;
229 }),
230 );
231 }),
232 ).pipe(Layer.provideMerge(throughProxy(() => proxy.url)));
233 const actor = scratchStack(
234 { providers: observer, stage: parent.stage },
235 title,
236 file,
237 );
238 expect(actor.name).toBe(parent.name);
239 expect(actor.stage).toBe(parent.stage);
240 expect(actor.state).not.toBe(parent.state);
241 return actor;
242 });
243
244export const retires = (event: TransportEvent, ids: readonly string[]) =>
245 event.stage === "request" &&
246 ids.includes(event.machineId!) &&
247 (event.path.endsWith("/cordon") ||
248 event.path.endsWith("/stop") ||
249 event.path.endsWith("/suspend") ||
250 (event.path.endsWith("/metadata") && event.phase === "retiring") ||
251 (event.method === "DELETE" && /\/machines\/[^/]+$/.test(event.path)));
252
253const identity = (metadata: Record<string, string> | undefined) =>
254 Object.fromEntries(
255 Object.entries(metadata ?? {}).filter(
256 ([key]) => key !== "alchemy.phase" && key !== "alchemy.checked-instance",
257 ),
258 );
259
260export const readinessChecksPassing = (
261 checks: ReadinessEvent["checks"],
262 expected: readonly string[],
263) => {
264 if (
265 !checks ||
266 expected.length === 0 ||
267 new Set(expected).size !== expected.length
268 )
269 return false;
270 const allowed = new Set([
271 ...expected,
272 ...expected
273 .filter((name) => name.startsWith("servicecheck-"))
274 .map(
275 (name) => `bg_deployments_compat-${name.slice("servicecheck-".length)}`,
276 ),
277 ]);
278 const reported = new Set(checks.map((check) => check.name));
279 return (
280 reported.size === checks.length &&
281 expected.every((name) => reported.has(name)) &&
282 checks.every(
283 (check) =>
284 check.name !== undefined &&
285 allowed.has(check.name) &&
286 check.status === "passing",
287 )
288 );
289};
290
291const sameMetadata = (left: ReadinessEvent, right: ReadinessEvent) =>
292 left.method === "PUT" &&
293 right.method === "PUT" &&
294 left.path.endsWith("/metadata") &&
295 left.path === right.path &&
296 left.metadataOnly === true &&
297 right.metadataOnly === true &&
298 left.metadata !== undefined &&
299 right.metadata !== undefined &&
300 JSON.stringify(Object.entries(left.metadata).sort()) ===
301 JSON.stringify(Object.entries(right.metadata).sort());
302
303/** Fresh GET receipt proves observation order, not the remote check sample's timestamp. */
304export const assertReadinessCommit = (
305 events: readonly ReadinessEvent[],
306 priorIds: readonly string[],
307 candidates: machines.Machine[],
308 runningSlots: readonly number[],
309 checkNames: readonly string[],
310 allowIdle: boolean,
311) => {
312 const retirement =
313 priorIds.length > 0
314 ? events.findIndex((event) => retires(event, priorIds))
315 : events.length;
316 expect(retirement).toBeGreaterThan(0);
317 const ordered = [...candidates].sort(
318 (a, b) =>
319 Number(a.config?.metadata?.["alchemy.replica"]) -
320 Number(b.config?.metadata?.["alchemy.replica"]),
321 );
322 const ids = ordered.map((machine) => machine.id!);
323 const creates = events
324 .filter(
325 (event) =>
326 event.stage === "forwarded" &&
327 event.method === "POST" &&
328 event.path.endsWith("/machines") &&
329 ids.includes(event.machineId!),
330 )
331 .map((event) => event.sequence);
332 const mutations = events.filter(
333 (event) =>
334 event.stage === "request" &&
335 event.method !== "GET" &&
336 !event.path.endsWith("/lease") &&
337 (ids.includes(event.machineId!) || creates.includes(event.sequence)),
338 );
339 const isCommit = (event: ReadinessEvent) =>
340 event.method === "PUT" &&
341 event.path.endsWith("/metadata") &&
342 event.phase === "active" &&
343 event.metadataOnly === true;
344 const isValidating = (event: ReadinessEvent) =>
345 event.method === "PUT" &&
346 event.path.endsWith("/metadata") &&
347 event.phase === "validating" &&
348 event.metadataOnly === true;
349 const successful = (event: ReadinessEvent) =>
350 event.status! >= 200 && event.status! < 300;
351 const operations: {
352 request: ReadinessEvent;
353 machineId: string;
354 completed: number;
355 }[] = [];
356 for (const request of mutations) {
357 const receipts = events.filter(
358 (event) =>
359 event.sequence === request.sequence &&
360 event.stage === "forwarded" &&
361 event.method === request.method &&
362 event.path === request.path,
363 );
364 expect(receipts).toHaveLength(1);
365 const receipt = receipts[0]!;
366 const completed = events.indexOf(receipt);
367 const requested = events.indexOf(request);
368 expect(completed).toBeGreaterThan(requested);
369 expect(completed).toBeLessThan(retirement);
370 const machineId = request.machineId ?? receipt.machineId!;
371 if (request.path.endsWith("/start") && receipt.status === 412) {
372 const candidate = ordered.find((machine) => machine.id === machineId)!;
373 const commit = events.findIndex(
374 (event) => isCommit(event) && event.machineId === machineId,
375 );
376 expect(
377 events.some(
378 (event, index) =>
379 index > completed &&
380 index < commit &&
381 event.stage === "forwarded" &&
382 event.method === "GET" &&
383 successful(event) &&
384 event.machineId === machineId &&
385 event.state === "started" &&
386 event.instanceId === candidate.instance_id,
387 ),
388 ).toBe(true);
389 continue;
390 }
391 const previous = operations.findLast(
392 (operation) => operation.machineId === machineId,
393 );
394 // Only settled, retryable metadata failures may join an equivalent retry.
395 if (
396 previous &&
397 !successful(events[previous.completed]!) &&
398 sameMetadata(previous.request, request)
399 ) {
400 expect([429, 500, 502, 503, 504]).toContain(
401 events[previous.completed]!.status,
402 );
403 expect(previous.completed).toBeLessThan(requested);
404 previous.completed = completed;
405 } else {
406 operations.push({ request, machineId, completed });
407 }
408 }
409 for (const operation of operations) {
410 expect(successful(events[operation.completed]!)).toBe(true);
411 }
412 const commits = operations.filter((operation) => isCommit(operation.request));
413 expect(commits).toHaveLength(ordered.length);
414 expect(commits.map((operation) => operation.machineId).sort()).toEqual(
415 [...ids].sort(),
416 );
417 expect(commits.at(-1)?.machineId).toBe(ordered[0]!.id);
418 const firstCommit = events.indexOf(commits[0]!.request);
419 expect(firstCommit).toBeGreaterThan(0);
420 const invalidating = operations.filter(
421 (operation) => !isCommit(operation.request),
422 );
423 expect(invalidating.length).toBeGreaterThan(0);
424 const reset = Math.max(
425 ...invalidating.map((operation) => operation.completed),
426 );
427 expect(reset).toBeLessThan(firstCommit);
428 const zeroCommit = events.indexOf(commits.at(-1)!.request);
429 for (const operation of commits.slice(0, -1)) {
430 expect(operation.completed).toBeLessThan(zeroCommit);
431 }
432
433 for (const [slot, machine] of ordered.entries()) {
434 const id = machine.id!;
435 const run = runningSlots.includes(slot);
436 const commit = commits.find(
437 (operation) => operation.machineId === id,
438 )!.request;
439 const finalMetadata = metadataOf(machine.config?.metadata);
440 const finalServices = servicesOf(machine.config?.services);
441 if (allowIdle) {
442 expect(finalServices.length).toBeGreaterThan(0);
443 expect(finalServices.every((service) => service.autostop !== "off")).toBe(
444 true,
445 );
446 }
447 expect(finalMetadata["alchemy.replica"]).toBe(String(slot));
448 expect(finalMetadata["alchemy.readiness-role"]).toBe(run ? "run" : "idle");
449 expect(finalMetadata["alchemy.readiness-roles"]).toBe(
450 ordered
451 .map((_, index) => (runningSlots.includes(index) ? "run" : "idle"))
452 .join(","),
453 );
454 expect(finalMetadata["alchemy.phase"]).toBe("active");
455 expect(finalMetadata["alchemy.idle-policy-restored"]).toBe("true");
456 expect(commit.metadata).toEqual(finalMetadata);
457 expect(machine.cordoned).toBe(false);
458 for (const key of [
459 "alchemy.stack",
460 "alchemy.stage",
461 "alchemy.id",
462 "alchemy.type",
463 "alchemy.instance",
464 "alchemy.fqn",
465 "alchemy.generation",
466 "alchemy.workload",
467 "alchemy.image",
468 ]) {
469 expect(finalMetadata[key]).toBeDefined();
470 }
471 const observations = events.flatMap((event, index) => {
472 if (
473 index >= firstCommit ||
474 event.stage !== "forwarded" ||
475 event.method !== "GET" ||
476 !event.path.endsWith(`/machines/${id}`) ||
477 event.machineId !== id ||
478 event.status !== 200
479 )
480 return [];
481 const requested = events.findIndex(
482 (request) =>
483 request.stage === "request" && request.sequence === event.sequence,
484 );
485 return requested >= 0 && requested < index
486 ? [{ event, index, requested }]
487 : [];
488 });
489 const pending = observations.findLast(({ requested }) => requested > reset);
490 expect(pending).toBeDefined();
491 if (!pending) continue;
492 expect(pending.event.phase).toBe("validating");
493 expect(pending.event.metadata?.["alchemy.idle-policy-restored"]).toBe(
494 "true",
495 );
496 expect(pending.event.instanceId).toBe(machine.instance_id);
497 let proof = pending;
498 if (run && pending.event.state !== "started") {
499 expect(allowIdle).toBe(true);
500 expect(["stopped", "suspended"]).toContain(pending.event.state);
501 const targetReset = Math.max(
502 -1,
503 ...invalidating
504 .filter(
505 (operation) =>
506 operation.machineId === id && !isValidating(operation.request),
507 )
508 .map((operation) => operation.completed),
509 );
510 const restored = observations.findLast(
511 ({ event, requested, index }) =>
512 requested > targetReset &&
513 index < pending.index &&
514 ["promoting", "validating"].includes(event.phase ?? "") &&
515 event.metadata?.["alchemy.idle-policy-restored"] === "true" &&
516 event.instanceId === pending.event.instanceId &&
517 event.state === "started" &&
518 readinessChecksPassing(event.checks, checkNames),
519 );
520 expect(restored).toBeDefined();
521 if (!restored) continue;
522 proof = restored;
523 } else if (run) {
524 expect(readinessChecksPassing(proof.event.checks, checkNames)).toBe(true);
525 } else {
526 expect(["created", "stopped", "suspended"]).toContain(
527 pending.event.state,
528 );
529 }
530 expect(proof.event.instanceId).toBeDefined();
531 expect(proof.event.digest).toBe(machine.image_ref?.digest);
532 expect(proof.event.services).toEqual(finalServices);
533 expect(proof.event.cordoned).toBe(false);
534 expect(identity(proof.event.metadata)).toEqual(identity(finalMetadata));
535 expect(commit.metadata).toEqual({
536 ...pending.event.metadata,
537 "alchemy.phase": "active",
538 ...(run ? { "alchemy.checked-instance": proof.event.instanceId } : {}),
539 });
540 if (run) {
541 expect(commit.metadata?.["alchemy.checked-instance"]).toBe(
542 proof.event.instanceId,
543 );
544 expect(
545 allowIdle ? ["started", "stopped", "suspended"] : ["started"],
546 ).toContain(machine.state);
547 } else {
548 expect(["created", "stopped", "suspended"]).toContain(machine.state);
549 for (const created of events.filter(
550 (event) =>
551 event.stage === "forwarded" &&
552 event.method === "POST" &&
553 event.path.endsWith("/machines") &&
554 event.machineId === id,
555 )) {
556 expect(created.skipLaunch).toBe(true);
557 }
558 expect(
559 events.some(
560 (event) =>
561 event.machineId === id &&
562 ((event.stage === "request" && event.path.endsWith("/start")) ||
563 (event.stage === "forwarded" && event.state === "started")),
564 ),
565 ).toBe(false);
566 }
567 // Idle proof survives only the unchanged pending stamp and final commit.
568 for (const operation of invalidating.filter(
569 (operation) =>
570 operation.machineId === id && operation.completed > proof.index,
571 )) {
572 expect(isValidating(operation.request)).toBe(true);
573 expect(events.indexOf(operation.request)).toBeGreaterThan(proof.index);
574 expect(operation.request.metadata).toEqual({
575 ...proof.event.metadata,
576 "alchemy.phase": "validating",
577 });
578 }
579 for (const later of events
580 .slice(proof.index + 1)
581 .filter(
582 (event) =>
583 event.stage === "forwarded" &&
584 event.machineId === id &&
585 event.status! >= 200 &&
586 event.status! < 300 &&
587 event.path.endsWith(`/machines/${id}`),
588 )) {
589 expect(later.instanceId).toBe(proof.event.instanceId);
590 expect(later.digest).toBe(proof.event.digest);
591 expect(later.services).toEqual(finalServices);
592 expect(identity(later.metadata)).toEqual(identity(finalMetadata));
593 expect(later.metadata?.["alchemy.idle-policy-restored"]).toBe("true");
594 expect(["promoting", "validating", "active"]).toContain(later.phase);
595 expect(later.cordoned).toBe(false);
596 if (run) {
597 expect(
598 allowIdle ? ["started", "stopped", "suspended"] : ["started"],
599 ).toContain(later.state);
600 if (later.state === "started" && events.indexOf(later) < firstCommit) {
601 expect(readinessChecksPassing(later.checks, checkNames)).toBe(true);
602 }
603 if (later.phase === "active") {
604 expect(later.metadata?.["alchemy.checked-instance"]).toBe(
605 proof.event.instanceId,
606 );
607 }
608 }
609 }
610 }
611};