1import type {
2 FlyMachineConfig,
3 FlyMachineMount,
4 FlyMachineService,
5 FlyMachineServiceCheck,
6 FlyStopConfig,
7 ImageRef as FlyImageRef,
8 Machine as FlyMachine,
9 Volume as FlyVolume,
10} from "@distilled.cloud/fly-io/machines";
11import * as machines from "@distilled.cloud/fly-io/machines";
12import * as Retry from "@distilled.cloud/fly-io/Retry";
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Schedule from "effect/Schedule";
16import * as Result from "effect/Result";
17import { deepEqual } from "../Diff.ts";
18import {
19 validateDeployment,
20 type DeploymentPolicy,
21 type MachineCheck,
22} from "./Deployment.ts";
23import { reconcileBlueGreen, setRouting } from "./bluegreen.ts";
24import {
25 classifyDeploymentState,
26 validProtocol2Generation,
27} from "./DeploymentState.ts";
28import { usingMachineLeases, type MachineLeases } from "./leases.ts";
29import { regionOfReplica } from "./Region.ts";
30import { listOwnedApps } from "./App.ts";
31import type {
32 MachineGuest,
33 MachineImageRef,
34 MachineService,
35 MachineServiceCheck,
36} from "./Machine.ts";
37import {
38 alchemyMetadataKeys,
39 createMachineMetadata,
40 isAlchemyOwnedMetadata,
41 sanitizeFlyAppName,
42 type FlyAlchemyType,
43} from "./Metadata.ts";
44import type { DiskSpec, MountedDisk } from "./MountVolume.ts";
45import {
46 deleteVolume,
47 ensureVolumeGroup,
48 getVolumeById,
49 volumeGroupName,
50} from "./Volume.ts";
51
52const WAIT_TIMEOUT_SECONDS = 8;
53const waitBackoff = Schedule.exponential("500 millis");
54const SERVICE_CHECK_NAME_PREFIX = "servicecheck-";
55
56export class ReplicaNotCreated extends Data.TaggedError(
57 "Fly.ReplicaNotCreated",
58)<{
59 name: string;
60 appName: string;
61}> {}
62
63export class ReplicaChecksNotPassing extends Data.TaggedError(
64 "Fly.ReplicaChecksNotPassing",
65)<{
66 appName: string;
67 machineId: string;
68 checks: ReadonlyArray<{
69 name: string | undefined;
70 status: string | undefined;
71 output: string | undefined;
72 }>;
73}> {
74 get message() {
75 const checks = this.checks.map(
76 (check) =>
77 `${check.name ?? "unnamed"}: ${check.status ?? "unknown"}${check.output ? ` (${check.output})` : ""}`,
78 );
79 return `Service checks did not pass for ${this.appName}/${this.machineId}: ${checks.join("; ") || "no service check results"}`;
80 }
81}
82
83export interface Replica {
84 machineId: string;
85 name: string;
86 baseName?: string;
87 region: string;
88 state: string;
89 instanceId: string | undefined;
90 privateIp: string | undefined;
91 imageRef: MachineImageRef | undefined;
92 guest: MachineGuest | undefined;
93 mounts: MountedDisk[];
94}
95
96export interface ReplicaSet {
97 rolloutPending?: boolean;
98 appName: string;
99 machineId: string;
100 machineIds: string[];
101 name: string;
102 baseName?: string;
103 /** Region of replica 0. */
104 region: string;
105 /** Regions the replicas run in, in replica order. */
106 regions: string[];
107 state: string;
108 instanceId: string | undefined;
109 privateIp: string | undefined;
110 imageRef: MachineImageRef | undefined;
111 guest: MachineGuest | undefined;
112 /** Observed proxy services published by replica 0. */
113 services: FlyMachineService[] | undefined;
114 count: number;
115 mounts: MountedDisk[];
116 replicas: Replica[];
117}
118
119const compactRecord = (
120 record: Record<string, string | undefined> | null | undefined,
121): Record<string, string> =>
122 Object.fromEntries(
123 Object.entries(record ?? {}).flatMap(([key, value]) =>
124 value === undefined ? [] : [[key, value]],
125 ),
126 );
127
128export const gone = (machine: FlyMachine | undefined) =>
129 machine === undefined || machine.state === "destroyed";
130
131export const getMachineById = (appName: string, machineId: string) =>
132 machines.getMachine({ app_name: appName, machine_id: machineId }).pipe(
133 Effect.map((machine) => (gone(machine) ? undefined : machine)),
134 Effect.catchTag("NotFound", () => Effect.succeed(undefined)),
135 );
136
137export const listMachinesByApp = (appName: string) =>
138 machines.listMachines({ app_name: appName }).pipe(
139 Effect.map((machines) => machines.filter((machine) => !gone(machine))),
140 Effect.catchTag("NotFound", () => Effect.succeed([])),
141 );
142
143export const resolveCount = (count: number | undefined) =>
144 Math.max(1, Math.floor(count ?? 1));
145
146export const replicaMachineName = (
147 base: string,
148 index: number,
149 count: number,
150) => {
151 if (count <= 1 && index === 0) return base;
152 const suffix = `-${index}`;
153 const room = 30 - suffix.length;
154 const clipped = base.slice(0, Math.max(1, room)).replace(/-+$/g, "");
155 return sanitizeFlyAppName(`${clipped}${suffix}`);
156};
157
158export const replicaIndexOf = (machine: FlyMachine): number => {
159 const raw = compactRecord(machine.config?.metadata)[
160 alchemyMetadataKeys.replica
161 ];
162 const parsed = raw === undefined ? 0 : Number(raw);
163 return Number.isFinite(parsed) ? parsed : 0;
164};
165
166export const alchemyIdOf = (machine: FlyMachine): string | undefined => {
167 const id = compactRecord(machine.config?.metadata)[alchemyMetadataKeys.id];
168 return id !== undefined && id.length > 0 ? id : undefined;
169};
170
171export const isOwnedType = (machine: FlyMachine, type: FlyAlchemyType) => {
172 const metadata = compactRecord(machine.config?.metadata);
173 return (
174 isAlchemyOwnedMetadata(metadata) &&
175 metadata[alchemyMetadataKeys.type] === type
176 );
177};
178
179export const toImageRef = (
180 ref: FlyImageRef | undefined,
181): MachineImageRef | undefined => {
182 if (ref === undefined) return undefined;
183 const imageRef: MachineImageRef = {
184 registry: ref.registry,
185 repository: ref.repository,
186 tag: ref.tag,
187 digest: ref.digest,
188 };
189 return imageRef.registry === undefined &&
190 imageRef.repository === undefined &&
191 imageRef.tag === undefined &&
192 imageRef.digest === undefined
193 ? undefined
194 : imageRef;
195};
196
197export const toGuestAttrs = (
198 guest:
199 | {
200 cpu_kind?: string;
201 cpus?: number;
202 memory_mb?: number;
203 gpu_kind?: string;
204 gpus?: number;
205 }
206 | undefined,
207): MachineGuest | undefined => {
208 if (guest === undefined) return undefined;
209 return {
210 cpuKind: guest.cpu_kind,
211 cpus: guest.cpus,
212 memoryMb: guest.memory_mb,
213 gpuKind: guest.gpu_kind,
214 gpus: guest.gpus,
215 };
216};
217
218export const toFlyServiceCheck = (
219 check: MachineServiceCheck,
220): FlyMachineServiceCheck => ({
221 type: check.type,
222 port: check.port,
223 interval: check.interval,
224 timeout: check.timeout,
225 grace_period: check.gracePeriod,
226 method: check.method,
227 path: check.path,
228 protocol: check.protocol,
229 headers: check.headers?.map((header) => ({
230 name: header.name,
231 values: header.values,
232 })),
233 tls_server_name: check.tlsServerName,
234 tls_skip_verify: check.tlsSkipVerify,
235});
236
237export const toFlyService = (service: MachineService): FlyMachineService => ({
238 protocol: service.protocol,
239 internal_port: service.internalPort,
240 autostart: service.autostart,
241 autostop:
242 typeof service.autostop === "boolean"
243 ? service.autostop
244 ? "stop"
245 : "off"
246 : service.autostop,
247 min_machines_running: service.minMachinesRunning,
248 ports: service.ports?.map((port) => ({
249 port: port.port,
250 handlers: port.handlers,
251 force_https: port.forceHttps,
252 start_port: port.startPort,
253 end_port: port.endPort,
254 })),
255 checks: service.checks?.map((check) => ({
256 ...toFlyServiceCheck(check),
257 // Fly clamps service-check intervals; named Machine checks retain longer intervals.
258 interval:
259 (durationNanoseconds(check.interval) ?? 0n) > 60_000_000_000n
260 ? "60s"
261 : check.interval,
262 })),
263});
264
265export const autostopMode = (value: string | boolean | undefined) =>
266 value === true
267 ? "stop"
268 : value === false || value === undefined
269 ? "off"
270 : value;
271
272const normalizedCheckDuration = (value: string | undefined) => {
273 const nanos = durationNanoseconds(value);
274 return nanos === undefined ? value : `${nanos}ns`;
275};
276
277const normalizedCheck = (
278 check: machines.FlyMachineCheck | FlyMachineServiceCheck | undefined,
279) =>
280 check === undefined
281 ? undefined
282 : {
283 ...check,
284 grace_period: normalizedCheckDuration(check.grace_period),
285 interval: normalizedCheckDuration(check.interval),
286 timeout: normalizedCheckDuration(check.timeout),
287 };
288
289export const sameChecks = (
290 observed: FlyMachineConfig["checks"],
291 desired: FlyMachineConfig["checks"],
292) => {
293 const normalize = (checks: FlyMachineConfig["checks"]) =>
294 Object.fromEntries(
295 Object.entries(checks ?? {}).map(([name, check]) => [
296 name,
297 normalizedCheck(check),
298 ]),
299 );
300 return deepEqual(normalize(observed), normalize(desired), {
301 stripNullish: true,
302 });
303};
304
305export const normalizedServices = (services: FlyMachineService[] | undefined) =>
306 (services ?? []).map((service) => ({
307 ...service,
308 autostop: autostopMode(service.autostop),
309 checks: service.checks?.map(normalizedCheck),
310 }));
311
312export const sameServices = (
313 observed: FlyMachineService[] | undefined,
314 desired: FlyMachineService[] | undefined,
315) =>
316 deepEqual(normalizedServices(observed), normalizedServices(desired), {
317 stripNullish: true,
318 });
319
320export const hasPublishedService = (
321 services: FlyMachineService[] | undefined,
322) =>
323 (services ?? []).some((service) =>
324 (service.ports ?? []).some(
325 (port) => port.port !== undefined || port.start_port !== undefined,
326 ),
327 );
328
329export const waitStarted = (appName: string, machineId: string) =>
330 machines
331 .waitMachine({
332 app_name: appName,
333 machine_id: machineId,
334 state: "started",
335 timeout: WAIT_TIMEOUT_SECONDS,
336 })
337 .pipe(
338 Retry.none,
339 Effect.retry({
340 times: 6,
341 schedule: waitBackoff,
342 while: (e) =>
343 e._tag === "GatewayTimeout" || e._tag === "MachineWaitTimeout",
344 }),
345 Effect.timeout("50 seconds"),
346 );
347
348export const configuredCheckNames = (config: FlyMachineConfig | undefined) => [
349 ...Object.keys(config?.checks ?? {}),
350 ...(config?.services ?? []).flatMap((service) =>
351 (service.checks ?? []).map(
352 (check, checkIndex) =>
353 `${SERVICE_CHECK_NAME_PREFIX}${String(checkIndex).padStart(2, "0")}-${check.type ?? "tcp"}-${service.internal_port}`,
354 ),
355 ),
356];
357
358export const checksPassing = (
359 machine: FlyMachine,
360 config: FlyMachineConfig | undefined,
361) => {
362 const expected = configuredCheckNames(config);
363 const checks = machine.checks ?? [];
364 // Fly may also report compatibility mirrors; they do not replace service reports.
365 return (
366 machine.state === "started" &&
367 new Set(expected).size === expected.length &&
368 checks.every(
369 (check) =>
370 check.name !== undefined &&
371 check.status === "passing" &&
372 (expected.includes(check.name) ||
373 expected.some(
374 (name) =>
375 name.startsWith(SERVICE_CHECK_NAME_PREFIX) &&
376 check.name ===
377 name.replace(
378 SERVICE_CHECK_NAME_PREFIX,
379 "bg_deployments_compat-",
380 ),
381 )),
382 ) &&
383 new Set(checks.map((check) => check.name)).size === checks.length &&
384 expected.every((name) => {
385 const reports = checks.filter((check) => check.name === name);
386 return reports.length === 1 && reports[0]?.status === "passing";
387 })
388 );
389};
390
391const TRANSIENT_GET_TAGS = [
392 "TooManyRequests",
393 "InternalServerError",
394 "BadGateway",
395 "ServiceUnavailable",
396 "GatewayTimeout",
397] as const;
398
399/**
400 * After the Machine is started, wait until Fly reports every service
401 * check as passing. No configured checks is a no-op. Empty live
402 * `servicecheck-*` results keep polling — they are not success.
403 * `warning` / `unknown` during grace keep polling.
404 */
405export const waitHealthy = Effect.fn(function* (
406 appName: string,
407 machine: FlyMachine,
408 healthTimeoutMs = 60_000,
409 config: FlyMachineConfig | undefined = machine.config,
410) {
411 const machineId = machine.id;
412 const named = Object.keys(config?.checks ?? {});
413 const expected = (config?.services ?? []).flatMap(
414 (service) => service.checks ?? [],
415 ).length;
416 if (machineId === undefined || (expected === 0 && named.length === 0))
417 return machine;
418
419 let observed = machine;
420 const notPassing = () =>
421 new ReplicaChecksNotPassing({
422 appName,
423 machineId,
424 checks: (observed.checks ?? []).map((check) => ({
425 name: check.name,
426 status: check.status,
427 output: undefined,
428 })),
429 });
430 const passing = yield* getMachineById(appName, machineId).pipe(
431 Retry.none,
432 Effect.map((current) => {
433 if (current === undefined) return false;
434 observed = current;
435 return (
436 current.instance_id !== undefined &&
437 current.instance_id === machine.instance_id &&
438 checksPassing(current, config)
439 );
440 }),
441 Effect.catchTag(TRANSIENT_GET_TAGS, () => Effect.succeed(false)),
442 Effect.catchTag("HttpClientError", (error) =>
443 error.reason._tag === "TransportError"
444 ? Effect.succeed(false)
445 : Effect.fail(error),
446 ),
447 Effect.repeat({
448 schedule: Schedule.spaced(healthTimeoutMs / 10),
449 until: (passing) => passing,
450 times: 10,
451 }),
452 Effect.timeoutOrElse({
453 duration: healthTimeoutMs,
454 orElse: () => Effect.fail(notPassing()),
455 }),
456 );
457 if (!passing) return yield* notPassing();
458 return observed;
459});
460
461export const waitDestroyed = (appName: string, machineId: string) =>
462 machines
463 .waitMachine({
464 app_name: appName,
465 machine_id: machineId,
466 state: "destroyed",
467 timeout: WAIT_TIMEOUT_SECONDS,
468 })
469 .pipe(
470 Retry.none,
471 Effect.as(undefined),
472 Effect.catchTag("NotFound", () => Effect.void),
473 Effect.retry({
474 times: 6,
475 schedule: waitBackoff,
476 while: (e) =>
477 e._tag === "GatewayTimeout" || e._tag === "MachineWaitTimeout",
478 }),
479 Effect.timeout("50 seconds"),
480 );
481
482export const ensureStarted = (
483 appName: string,
484 machine: FlyMachine,
485 skipLaunch: boolean,
486 healthTimeoutMs = 60_000,
487 expectedConfig?: FlyMachineConfig,
488 existingLeases?: MachineLeases,
489) =>
490 usingMachineLeases(appName, existingLeases, (leases) =>
491 ensureLeasedStarted(
492 appName,
493 machine,
494 skipLaunch,
495 healthTimeoutMs,
496 expectedConfig,
497 leases,
498 ),
499 );
500
501const ensureLeasedStarted = Effect.fn(function* (
502 appName: string,
503 machine: FlyMachine,
504 skipLaunch: boolean,
505 healthTimeoutMs: number,
506 expectedConfig: FlyMachineConfig | undefined,
507 leases: MachineLeases,
508) {
509 const machineId = machine.id;
510 if (machineId === undefined || skipLaunch) return machine;
511 yield* leases.acquire([machineId]);
512 const started = yield* Effect.gen(function* () {
513 // Create/update responses can lag Fly's automatic launch.
514 const current = yield* machines.getMachine({
515 app_name: appName,
516 machine_id: machineId,
517 });
518 if (!sameOwnership(current, machine)) {
519 return yield* new ReplicaOwnershipChanged({ appName, machineId });
520 }
521 yield* Effect.logDebug("Fly machine startup", {
522 appName,
523 machineId,
524 state: current.state,
525 });
526 if (
527 current.state === "stopped" ||
528 current.state === "suspended" ||
529 current.state === "failed"
530 ) {
531 yield* leases.mutate(machineId, (lease_nonce) =>
532 machines.startMachine({
533 app_name: appName,
534 machine_id: machineId,
535 lease_nonce,
536 }),
537 );
538 }
539 // Re-observe state between waits instead of retrying the wait in the SDK.
540 yield* machines
541 .waitMachine({
542 app_name: appName,
543 machine_id: machineId,
544 state: "started",
545 instance_id: current.instance_id,
546 timeout: WAIT_TIMEOUT_SECONDS,
547 })
548 .pipe(Retry.none);
549 return yield* machines.getMachine({
550 app_name: appName,
551 machine_id: machineId,
552 });
553 }).pipe(
554 Effect.retry({
555 times: 6,
556 schedule: waitBackoff,
557 while: (error) =>
558 error._tag === "MachineStartFromCreatedState" ||
559 error._tag === "MachineReplacing" ||
560 error._tag === "MachineWaitTimeout" ||
561 error._tag === "Conflict" ||
562 error._tag === "GatewayTimeout",
563 }),
564 Effect.timeout("180 seconds"),
565 );
566 return yield* waitHealthy(
567 appName,
568 started,
569 healthTimeoutMs,
570 expectedConfig ?? machine.config,
571 );
572});
573
574export const deleteMachine = (
575 appName: string,
576 machineId: string,
577 existingLeases?: MachineLeases,
578) =>
579 usingMachineLeases(appName, existingLeases, (leases) =>
580 Effect.gen(function* () {
581 if (appName.length === 0 || machineId.length === 0) return;
582 yield* leases.acquire([machineId]);
583 yield* leases.remove(machineId, (lease_nonce) =>
584 machines
585 .deleteMachine({
586 app_name: appName,
587 machine_id: machineId,
588 force: true,
589 lease_nonce,
590 })
591 .pipe(Effect.timeout("30 seconds")),
592 );
593 yield* waitDestroyed(appName, machineId);
594 }).pipe(Effect.catchTag("NotFound", () => leases.forget(machineId))),
595 );
596
597const durationNanoseconds = (value: string | undefined) => {
598 if (value === undefined || value === "") return undefined;
599 const negative = value.startsWith("-");
600 const duration = /^[+-]/.test(value) ? value.slice(1) : value;
601 if (duration === "0") return 0n;
602 const limit = negative ? 1n << 63n : (1n << 63n) - 1n;
603 const units: Record<string, bigint> = {
604 ns: 1n,
605 us: 1_000n,
606 µs: 1_000n,
607 μs: 1_000n,
608 ms: 1_000_000n,
609 s: 1_000_000_000n,
610 m: 60_000_000_000n,
611 h: 3_600_000_000_000n,
612 };
613 let total = 0n;
614 let consumed = 0;
615 for (const part of duration.matchAll(
616 /(\d+(?:\.\d*)?|\.\d+)(ns|us|µs|μs|ms|s|m|h)/gy,
617 )) {
618 const [integer = "", fraction = ""] = part[1]!.split(".");
619 const whole = integer.replace(/^0+/, "") || "0";
620 if (whole.length > 19) return undefined;
621 const unit = units[part[2]!]!;
622 let fractionalNanos = 0n;
623 // Truncate each component to nanoseconds without unbounded BigInt operands.
624 for (let index = fraction.length - 1; index >= 0; index--) {
625 fractionalNanos =
626 (BigInt(fraction[index]!) * unit + fractionalNanos) / 10n;
627 }
628 total += BigInt(whole) * unit + fractionalNanos;
629 if (total > limit) return undefined;
630 consumed += part[0].length;
631 }
632 if (consumed === 0 || consumed !== duration.length) return undefined;
633 return negative ? -total : total;
634};
635
636const stopTimeoutMillis = (timeout: string | undefined) => {
637 if (timeout === "") return 0;
638 // Shutdown policies retain their unsigned millisecond-or-larger syntax.
639 if (
640 timeout === undefined ||
641 !/^(?:\d+(?:\.\d+)?(?:ms|s|m|h))+$/.test(timeout)
642 ) {
643 return undefined;
644 }
645 const nanos = durationNanoseconds(timeout);
646 return nanos === undefined ? undefined : Number(nanos) / 1_000_000;
647};
648
649export const sameStopConfig = (
650 observed: FlyStopConfig | undefined,
651 desired: FlyStopConfig | undefined,
652) =>
653 observed?.signal === desired?.signal &&
654 stopTimeoutMillis(observed?.timeout) === stopTimeoutMillis(desired?.timeout);
655
656export class ShutdownPolicyMismatch extends Data.TaggedError(
657 "Fly.ShutdownPolicyMismatch",
658)<{ machineId: string; message: string }> {}
659
660export const predecessorShutdown = Effect.fn(function* (machine: FlyMachine) {
661 const persisted = machine.config?.stop_config;
662 const injected = machine.config?.env?.ALCHEMY_FLY_SHUTDOWN_TIMEOUT_MS;
663 const timeoutMs = stopTimeoutMillis(persisted?.timeout);
664 if (injected !== undefined) {
665 const managedTimeout = Number(injected);
666 const signal = persisted?.signal ?? "SIGTERM";
667 if (
668 !/^\d+$/.test(injected) ||
669 !Number.isSafeInteger(managedTimeout) ||
670 managedTimeout <= 0 ||
671 managedTimeout > 300_000 ||
672 (signal !== "SIGTERM" && signal !== "SIGINT") ||
673 (persisted?.timeout !== undefined && timeoutMs !== managedTimeout)
674 ) {
675 return yield* new ShutdownPolicyMismatch({
676 machineId: machine.id ?? "",
677 message:
678 "The predecessor's managed shutdown environment and stop_config disagree; reconcile its shutdown policy before migration.",
679 });
680 }
681 return {
682 signal,
683 timeout: persisted?.timeout ?? `${managedTimeout}ms`,
684 timeoutMs: managedTimeout,
685 };
686 }
687 if (persisted?.timeout !== undefined && (!timeoutMs || timeoutMs > 300_000)) {
688 return yield* new ShutdownPolicyMismatch({
689 machineId: machine.id ?? "",
690 message:
691 "The predecessor has an invalid stop_config timeout; refusing to shorten its shutdown window.",
692 });
693 }
694 // Unspecified raw-image overrides must remain unspecified for Fly's defaults.
695 return {
696 signal: persisted?.signal,
697 timeout: persisted?.timeout,
698 timeoutMs: timeoutMs ?? 300_000,
699 };
700});
701
702export class ReplicaOwnershipChanged extends Data.TaggedError(
703 "Fly.ReplicaOwnershipChanged",
704)<{
705 appName: string;
706 machineId: string;
707}> {}
708
709export class ReplicaRetirementIncomplete extends Data.TaggedError(
710 "Fly.ReplicaRetirementIncomplete",
711)<{
712 appName: string;
713 residuals: Array<{ machineId: string; stage: string }>;
714}> {}
715
716const retirementStep =
717 (appName: string, machineId: string, stage: string) =>
718 <A, E extends { readonly _tag: string }, R>(effect: Effect.Effect<A, E, R>) =>
719 effect.pipe(
720 Effect.mapError((error) =>
721 error._tag === "NotFound"
722 ? error
723 : new ReplicaRetirementIncomplete({
724 appName,
725 residuals: [{ machineId, stage: `${stage}: ${error._tag}` }],
726 }),
727 ),
728 );
729
730const sameOwnership = (observed: FlyMachine, snapshot: FlyMachine) =>
731 observed.id === snapshot.id &&
732 [
733 alchemyMetadataKeys.stack,
734 alchemyMetadataKeys.stage,
735 alchemyMetadataKeys.id,
736 alchemyMetadataKeys.type,
737 alchemyMetadataKeys.instance,
738 alchemyMetadataKeys.fqn,
739 alchemyMetadataKeys.generation,
740 ].every(
741 (key) =>
742 observed.config?.metadata?.[key] === snapshot.config?.metadata?.[key],
743 );
744
745export const leaseSnapshot = Effect.fn(function* (
746 appName: string,
747 snapshot: FlyMachine[],
748 leases: MachineLeases,
749) {
750 const ordered = [...snapshot].sort((a, b) =>
751 (a.id ?? "").localeCompare(b.id ?? ""),
752 );
753 yield* Effect.forEach(
754 ordered,
755 (machine) =>
756 Effect.gen(function* () {
757 if (!machine.id || machine.host_status === "unreachable") return;
758 yield* leases
759 .acquire([machine.id])
760 .pipe(Effect.catchTag("NotFound", () => Effect.void));
761 }),
762 { concurrency: 1 },
763 ).pipe(Effect.timeout("30 seconds"));
764 const observed: FlyMachine[] = [];
765 for (const machine of ordered) {
766 if (!machine.id) continue;
767 if (machine.host_status === "unreachable") {
768 observed.push(machine);
769 continue;
770 }
771 const current = yield* getMachineById(appName, machine.id);
772 if (!current) {
773 yield* leases.forget(machine.id);
774 continue;
775 }
776 if (!sameOwnership(current, machine))
777 return yield* new ReplicaOwnershipChanged({
778 appName,
779 machineId: machine.id,
780 });
781 yield* leases.checkTarget(machine.id);
782 observed.push(current);
783 }
784 yield* leases.check;
785 return observed;
786});
787
788export const retireMachines = (
789 appName: string,
790 snapshot: FlyMachine[],
791 replacementReady = false,
792 existingLeases?: MachineLeases,
793) =>
794 usingMachineLeases(appName, existingLeases, (leases) =>
795 retireLeasedMachines(appName, snapshot, replacementReady, leases),
796 );
797
798const retireLeasedMachines = Effect.fn(function* (
799 appName: string,
800 snapshot: FlyMachine[],
801 replacementReady: boolean,
802 leases: MachineLeases,
803) {
804 const targets = yield* leaseSnapshot(appName, snapshot, leases);
805 const results = yield* Effect.forEach(
806 targets,
807 (machine) =>
808 Effect.gen(function* () {
809 yield* leases.check;
810 const result = yield* Effect.gen(function* () {
811 if (!machine.id) return;
812 if (replacementReady && machine.host_status === "unreachable") {
813 const metadata = machine.config?.metadata;
814 if (
815 !machine.config ||
816 machine.incomplete_config ||
817 (machine.config.mounts?.length ?? 0) > 0 ||
818 !metadata?.[alchemyMetadataKeys.instance] ||
819 !metadata[alchemyMetadataKeys.fqn]
820 ) {
821 return yield* new ReplicaOwnershipChanged({
822 appName,
823 machineId: machine.id,
824 });
825 }
826 yield* Effect.logWarning(
827 "Force-retiring owned stateless Fly Machine on unreachable host; work was not proven drained",
828 { appName, machineId: machine.id },
829 );
830 const machineId = machine.id;
831 const heldNonce = yield* leases.nonceIfHeld(machineId);
832 yield* Effect.gen(function* () {
833 const remove = (lease_nonce?: string) =>
834 machines
835 .deleteMachine({
836 app_name: appName,
837 machine_id: machineId,
838 force: true,
839 lease_nonce,
840 })
841 .pipe(Retry.none, Effect.timeout("30 seconds"));
842 if (heldNonce !== undefined)
843 yield* leases.remove(machineId, remove);
844 else yield* leases.guard(remove());
845 }).pipe(
846 Effect.catchTag("NotFound", () => Effect.void),
847 retirementStep(appName, machineId, "unreachable force-delete"),
848 );
849 yield* leases.forget(machine.id);
850 return yield* waitDestroyed(appName, machine.id);
851 }
852 yield* retireMachine(appName, machine.id, machine, leases);
853 }).pipe(Effect.result);
854 yield* leases.check;
855 return Result.isFailure(result)
856 ? result.failure._tag === "Fly.ReplicaRetirementIncomplete"
857 ? result.failure.residuals
858 : [{ machineId: machine.id ?? "", stage: result.failure._tag }]
859 : [];
860 }),
861 { concurrency: 4 },
862 );
863 const residuals = results.flat();
864 if (residuals.length)
865 return yield* new ReplicaRetirementIncomplete({ appName, residuals });
866});
867
868export const retireMachine = (
869 appName: string,
870 machineId: string,
871 snapshot?: FlyMachine,
872 existingLeases?: MachineLeases,
873) =>
874 usingMachineLeases(appName, existingLeases, (leases) =>
875 retireLeasedMachine(appName, machineId, snapshot, leases).pipe(
876 Effect.tap(() => leases.forget(machineId)),
877 ),
878 );
879
880const retireLeasedMachine = Effect.fn(
881 function* (
882 appName: string,
883 machineId: string,
884 snapshot: FlyMachine | undefined,
885 leases: MachineLeases,
886 ) {
887 yield* leases.acquire([machineId]);
888 const current = yield* getMachineById(appName, machineId);
889 if (current === undefined) return;
890 if (snapshot && !sameOwnership(current, snapshot)) {
891 return yield* new ReplicaOwnershipChanged({ appName, machineId });
892 }
893 const stop = yield* predecessorShutdown(current);
894 if (
895 current.cordoned === true &&
896 current.config?.metadata?.[alchemyMetadataKeys.phase] === "candidate"
897 )
898 return yield* deleteMachine(appName, machineId, leases);
899 if (current.state !== "stopped" && current.state !== "created") {
900 yield* leases
901 .mutate(
902 machineId,
903 (lease_nonce) =>
904 machines
905 .cordonMachine({
906 app_name: appName,
907 machine_id: machineId,
908 lease_nonce,
909 })
910 .pipe(Effect.timeout("30 seconds")),
911 { idempotent: true },
912 )
913 .pipe(retirementStep(appName, machineId, "cordon"));
914 if (hasPublishedService(current.config?.services))
915 yield* Effect.sleep("10 seconds");
916 yield* leases
917 .mutate(
918 machineId,
919 (lease_nonce) =>
920 machines
921 .stopMachine({
922 app_name: appName,
923 machine_id: machineId,
924 lease_nonce,
925 signal: stop.signal,
926 timeout: stop.timeout,
927 })
928 .pipe(Effect.timeout(stop.timeoutMs + 15_000)),
929 { timeoutMs: stop.timeoutMs + 120_000 },
930 )
931 .pipe(retirementStep(appName, machineId, "stop"));
932 yield* machines
933 .waitMachine({
934 app_name: appName,
935 machine_id: machineId,
936 state: "stopped",
937 instance_id: current.instance_id,
938 timeout: WAIT_TIMEOUT_SECONDS,
939 })
940 .pipe(
941 Retry.none,
942 Effect.retry({
943 times: 10,
944 schedule: Schedule.spaced(Math.max(500, stop.timeoutMs / 10)),
945 while: (error) =>
946 error._tag === "MachineWaitTimeout" ||
947 error._tag === "GatewayTimeout",
948 }),
949 Effect.timeout(stop.timeoutMs + 15_000),
950 retirementStep(appName, machineId, "wait-stopped"),
951 );
952 }
953 yield* leases
954 .remove(machineId, (lease_nonce) =>
955 machines
956 .deleteMachine({
957 app_name: appName,
958 machine_id: machineId,
959 force: false,
960 lease_nonce,
961 })
962 .pipe(Effect.timeout("30 seconds")),
963 )
964 .pipe(
965 Effect.catchTag("NotFound", () => Effect.void),
966 retirementStep(appName, machineId, "delete"),
967 );
968 yield* waitDestroyed(appName, machineId).pipe(
969 retirementStep(appName, machineId, "verify-absent"),
970 );
971 },
972 Effect.catchTag("NotFound", () => Effect.void),
973);
974
975const mountedDisksOf = (
976 machine: FlyMachine,
977 volumesById: Map<string, FlyVolume>,
978): MountedDisk[] =>
979 (machine.config?.mounts ?? []).flatMap((mount) => {
980 const volumeId = mount.volume;
981 const path = mount.path;
982 if (volumeId === undefined || path === undefined) return [];
983 const volume = volumesById.get(volumeId);
984 return [
985 {
986 path,
987 volumeId,
988 sizeGb: volume?.size_gb ?? 0,
989 name: volume?.name ?? "",
990 },
991 ];
992 });
993
994export const toReplica = (
995 machine: FlyMachine,
996 volumesById: Map<string, FlyVolume>,
997): Replica => ({
998 machineId: machine.id ?? "",
999 name: machine.name ?? "",
1000 region: machine.region ?? "",
1001 state: machine.state ?? "",
1002 instanceId: machine.instance_id,
1003 privateIp: machine.private_ip,
1004 imageRef: toImageRef(machine.image_ref),
1005 guest: toGuestAttrs(machine.config?.guest),
1006 mounts: mountedDisksOf(machine, volumesById),
1007});
1008
1009export const toReplicaSet = (
1010 replicas: Replica[],
1011 appName: string,
1012 baseName: string,
1013 services?: FlyMachineService[],
1014): ReplicaSet => {
1015 const primary = replicas[0];
1016 return {
1017 appName,
1018 machineId: primary?.machineId ?? "",
1019 machineIds: replicas.map((replica) => replica.machineId),
1020 name: primary?.name ?? baseName,
1021 baseName,
1022 region: primary?.region ?? "",
1023 regions: [
1024 ...new Set(
1025 replicas
1026 .map((replica) => replica.region)
1027 .filter((region) => region.length > 0),
1028 ),
1029 ],
1030 state: primary?.state ?? "",
1031 instanceId: primary?.instanceId,
1032 privateIp: primary?.privateIp,
1033 imageRef: primary?.imageRef,
1034 guest: primary?.guest,
1035 services,
1036 count: replicas.length,
1037 mounts: primary?.mounts ?? [],
1038 replicas,
1039 };
1040};
1041
1042export const ownedReplicas = (
1043 listed: FlyMachine[],
1044 input: {
1045 metadata: Record<string, string>;
1046 resourceInstanceId: string;
1047 fqn: string;
1048 baseName?: string;
1049 machineIds?: readonly string[];
1050 },
1051) =>
1052 listed.filter((machine) => {
1053 const metadata = machine.config?.metadata ?? {};
1054 if (
1055 !Object.entries(input.metadata).every(
1056 ([key, value]) => metadata[key] === value,
1057 )
1058 )
1059 return false;
1060 if (metadata[alchemyMetadataKeys.instance] !== undefined) {
1061 return (
1062 metadata[alchemyMetadataKeys.instance] === input.resourceInstanceId &&
1063 metadata[alchemyMetadataKeys.fqn] === input.fqn
1064 );
1065 }
1066 return (
1067 (machine.id !== undefined && input.machineIds?.includes(machine.id)) ||
1068 (input.baseName !== undefined &&
1069 (machine.name === input.baseName ||
1070 machine.name ===
1071 replicaMachineName(input.baseName, replicaIndexOf(machine), 2)))
1072 );
1073 });
1074
1075export const listReplicas = Effect.fn(function* (input: {
1076 appName: string;
1077 id: string;
1078 type: FlyAlchemyType;
1079}) {
1080 const machines = yield* listMachinesByApp(input.appName);
1081 return machines
1082 .filter(
1083 (machine) =>
1084 isOwnedType(machine, input.type) && alchemyIdOf(machine) === input.id,
1085 )
1086 .sort((left, right) => replicaIndexOf(left) - replicaIndexOf(right));
1087});
1088
1089export const listReplicaSets = Effect.fn(function* (type: FlyAlchemyType) {
1090 const apps = yield* listOwnedApps();
1091 const groups = yield* Effect.forEach(
1092 apps,
1093 (app) =>
1094 listMachinesByApp(app.appName).pipe(
1095 Effect.map((machines) => {
1096 const owned = machines.filter((machine) =>
1097 isOwnedType(machine, type),
1098 );
1099 const byId = new Map<string, FlyMachine[]>();
1100 for (const machine of owned) {
1101 const id = alchemyIdOf(machine);
1102 if (id === undefined) continue;
1103 const metadata = machine.config?.metadata;
1104 const key = JSON.stringify([
1105 metadata?.[alchemyMetadataKeys.stack],
1106 metadata?.[alchemyMetadataKeys.stage],
1107 metadata?.[alchemyMetadataKeys.fqn] ?? id,
1108 metadata?.[alchemyMetadataKeys.instance],
1109 metadata?.[alchemyMetadataKeys.generation],
1110 ]);
1111 const group = byId.get(key) ?? [];
1112 group.push(machine);
1113 byId.set(key, group);
1114 }
1115 return [...byId.values()].map((group) => {
1116 const sorted = [...group].sort(
1117 (left, right) => replicaIndexOf(left) - replicaIndexOf(right),
1118 );
1119 const replicas = sorted.map((machine) =>
1120 toReplica(machine, new Map()),
1121 );
1122 return toReplicaSet(
1123 replicas,
1124 app.appName,
1125 replicas[0]?.name ?? "",
1126 sorted[0]?.config?.services,
1127 );
1128 });
1129 }),
1130 ),
1131 { concurrency: 8 },
1132 );
1133 return groups.flat();
1134});
1135
1136const pickVolume = (
1137 group: FlyVolume[],
1138 used: Set<string>,
1139 preferId: string | undefined,
1140): FlyVolume | undefined => {
1141 if (preferId !== undefined && !used.has(preferId)) {
1142 const preferred = group.find((volume) => volume.id === preferId);
1143 if (preferred !== undefined) return preferred;
1144 }
1145 return group.find(
1146 (volume) => volume.id !== undefined && !used.has(volume.id),
1147 );
1148};
1149
1150export interface ReconcileReplicasInput {
1151 fqn: string;
1152 resourceInstanceId: string;
1153 policy: DeploymentPolicy;
1154 checks?: Record<string, MachineCheck>;
1155 id: string;
1156 type: FlyAlchemyType;
1157 appName: string;
1158 baseName: string;
1159 /**
1160 * Regions to run in. `count` is the total replica count; replica `i`
1161 * runs in `regions[i % regions.length]`.
1162 */
1163 regions: readonly string[];
1164 count: number;
1165 disks: DiskSpec[];
1166 skipLaunch?: boolean;
1167 minSecretsVersion?: number;
1168 outputMachineIds?: readonly string[];
1169 preferVolumeIds?: ReadonlyArray<ReadonlyArray<string>>;
1170 configDrifted: (
1171 machine: FlyMachine,
1172 desired: {
1173 mounts: FlyMachineMount[];
1174 metadata: Record<string, string>;
1175 },
1176 ) => boolean;
1177 buildConfig: (replica: {
1178 index: number;
1179 mounts: FlyMachineMount[];
1180 metadata: Record<string, string>;
1181 }) => FlyMachineConfig;
1182}
1183
1184export const reconcileReplicas = (input: ReconcileReplicasInput) =>
1185 usingMachineLeases(input.appName, undefined, (leases) =>
1186 reconcileLeasedReplicas(input, leases),
1187 );
1188
1189const reconcileLeasedReplicas = Effect.fn(function* (
1190 input: ReconcileReplicasInput,
1191 leases: MachineLeases,
1192) {
1193 const ownership = yield* createMachineMetadata(input.id, input.type);
1194 const alchemy = {
1195 ...ownership,
1196 [alchemyMetadataKeys.instance]: input.resourceInstanceId,
1197 [alchemyMetadataKeys.fqn]: input.fqn,
1198 [alchemyMetadataKeys.baseName]: input.baseName,
1199 };
1200 const buildConfig: ReconcileReplicasInput["buildConfig"] = (replica) => ({
1201 ...input.buildConfig(replica),
1202 checks:
1203 input.checks === undefined
1204 ? undefined
1205 : Object.fromEntries(
1206 Object.entries(input.checks).map(([name, check]) => [
1207 name,
1208 toFlyServiceCheck(check),
1209 ]),
1210 ),
1211 stop_config:
1212 input.policy.shutdown === undefined
1213 ? undefined
1214 : {
1215 signal: input.policy.shutdown.signal,
1216 timeout: input.policy.shutdown.timeout,
1217 },
1218 });
1219 const config = buildConfig({ index: 0, mounts: [], metadata: alchemy });
1220 yield* validateDeployment(
1221 input.policy,
1222 config,
1223 input.disks.length > 0,
1224 input.skipLaunch,
1225 );
1226 if (input.policy.bluegreen)
1227 return yield* reconcileBlueGreen(
1228 { ...input, buildConfig },
1229 ownership,
1230 alchemy,
1231 leases,
1232 );
1233 const desiredNames = new Set(
1234 Array.from({ length: input.count }, (_, index) =>
1235 replicaMachineName(input.baseName, index, input.count),
1236 ),
1237 );
1238 const listed = yield* listMachinesByApp(input.appName);
1239 const owned = yield* leaseSnapshot(
1240 input.appName,
1241 ownedReplicas(listed, {
1242 ...input,
1243 metadata: ownership,
1244 machineIds: input.outputMachineIds,
1245 }),
1246 leases,
1247 );
1248 const byIndex = new Map<number, FlyMachine>();
1249 const preferIds = new Set(
1250 (input.outputMachineIds ?? []).filter((id) => id.length > 0),
1251 );
1252 for (const machine of owned) {
1253 const id = machine.id;
1254 if (id !== undefined && preferIds.has(id)) {
1255 byIndex.set(replicaIndexOf(machine), machine);
1256 }
1257 }
1258 for (const machine of owned) {
1259 const name = machine.name;
1260 if (
1261 name === undefined ||
1262 (!desiredNames.has(name) &&
1263 machine.config?.metadata?.[alchemyMetadataKeys.instance] !==
1264 input.resourceInstanceId)
1265 )
1266 continue;
1267 const index = replicaIndexOf(machine);
1268 if (!byIndex.has(index)) byIndex.set(index, machine);
1269 }
1270
1271 const scaleDown = [...byIndex].filter(([index]) => index >= input.count);
1272 yield* retireMachines(
1273 input.appName,
1274 scaleDown.map(([, machine]) => machine),
1275 false,
1276 leases,
1277 );
1278 for (const [index] of scaleDown) byIndex.delete(index);
1279
1280 // Volumes are regional: one group per disk and region, sized to the
1281 // replicas that run there.
1282 const replicasIn = (region: string) =>
1283 Array.from({ length: input.count }, (_, index) => index).filter(
1284 (index) => regionOfReplica(input.regions, index) === region,
1285 ).length;
1286 const groups: Array<{
1287 disk: DiskSpec;
1288 name: string;
1289 byRegion: Map<string, { volumes: FlyVolume[]; extras: FlyVolume[] }>;
1290 }> = [];
1291 for (const [diskIndex, disk] of input.disks.entries()) {
1292 const name = yield* volumeGroupName(input.id, disk);
1293 const preferIds = (input.preferVolumeIds ?? [])
1294 .map((replica) => replica[diskIndex])
1295 .filter((id): id is string => id !== undefined && id.length > 0);
1296 const byRegion = new Map<
1297 string,
1298 { volumes: FlyVolume[]; extras: FlyVolume[] }
1299 >();
1300 for (const region of input.regions) {
1301 byRegion.set(
1302 region,
1303 yield* ensureVolumeGroup({
1304 appName: input.appName,
1305 name,
1306 region,
1307 count: replicasIn(region),
1308 disk,
1309 preferIds,
1310 }),
1311 );
1312 }
1313 groups.push({ disk, name, byRegion });
1314 }
1315
1316 const usedVolumeIds = new Set<string>();
1317 const live: FlyMachine[] = [];
1318 for (let index = 0; index < input.count; index++) {
1319 const name = replicaMachineName(input.baseName, index, input.count);
1320 const region = regionOfReplica(input.regions, index);
1321 const metadata = {
1322 ...alchemy,
1323 [alchemyMetadataKeys.replica]: String(index),
1324 ...(input.minSecretsVersion === undefined
1325 ? {}
1326 : {
1327 [alchemyMetadataKeys.secretsVersion]: String(
1328 input.minSecretsVersion,
1329 ),
1330 }),
1331 };
1332 const prefer = input.preferVolumeIds?.[index] ?? [];
1333 const mounts: FlyMachineMount[] = [];
1334 for (const [diskIndex, group] of groups.entries()) {
1335 const volume = pickVolume(
1336 group.byRegion.get(region)?.volumes ?? [],
1337 usedVolumeIds,
1338 prefer[diskIndex],
1339 );
1340 const volumeId = volume?.id;
1341 if (volumeId === undefined) {
1342 return yield* new ReplicaNotCreated({
1343 name,
1344 appName: input.appName,
1345 });
1346 }
1347 usedVolumeIds.add(volumeId);
1348 mounts.push({ volume: volumeId, path: group.disk.path });
1349 }
1350 const config = buildConfig({ index, mounts, metadata });
1351 let current = byIndex.get(index);
1352 // Fly cannot move a Machine between regions. Machine names are unique
1353 // per App, so retire the old replica before creating it in its region.
1354 if (current !== undefined && current.region !== region) {
1355 yield* retireMachines(input.appName, [current], false, leases);
1356 byIndex.delete(index);
1357 current = undefined;
1358 }
1359 if (current === undefined) {
1360 const created = yield* machines
1361 .createMachine({
1362 app_name: input.appName,
1363 name,
1364 region,
1365 config,
1366 skip_launch: input.skipLaunch === true ? true : undefined,
1367 min_secrets_version: input.minSecretsVersion,
1368 })
1369 .pipe(Effect.catchTag("Conflict", () => Effect.succeed(undefined)));
1370 current =
1371 created ??
1372 (yield* listMachinesByApp(input.appName).pipe(
1373 Effect.map((machines) =>
1374 ownedReplicas(machines, {
1375 ...input,
1376 metadata: ownership,
1377 machineIds: input.outputMachineIds,
1378 }).find((machine) => machine.name === name),
1379 ),
1380 ));
1381 if (current === undefined || current.id === undefined) {
1382 return yield* new ReplicaNotCreated({
1383 name,
1384 appName: input.appName,
1385 });
1386 }
1387 const observed = (yield* leaseSnapshot(
1388 input.appName,
1389 [current],
1390 leases,
1391 ))[0];
1392 if (!observed || input.configDrifted(observed, { mounts, metadata })) {
1393 return yield* new ReplicaNotCreated({ name, appName: input.appName });
1394 }
1395 current = observed;
1396 } else if (
1397 !sameChecks(current.config?.checks, config.checks) ||
1398 !sameStopConfig(current.config?.stop_config, config.stop_config) ||
1399 input.configDrifted(current, {
1400 mounts,
1401 metadata,
1402 })
1403 ) {
1404 const machineId = current.id!;
1405 const currentVersion = current.instance_id;
1406 const previous = current;
1407 const expected = { ...previous, config };
1408 yield* leases.mutate(machineId, (lease_nonce) =>
1409 machines
1410 .updateMachine({
1411 app_name: input.appName,
1412 machine_id: machineId,
1413 current_version: currentVersion,
1414 config,
1415 skip_launch: input.skipLaunch === true ? true : undefined,
1416 min_secrets_version: input.minSecretsVersion,
1417 lease_nonce,
1418 })
1419 .pipe(Effect.timeout("30 seconds")),
1420 );
1421 // Rolling opt-out removes generation metadata; observe that transition before startup.
1422 current = yield* leases.guard(
1423 Effect.gen(function* () {
1424 const observed = yield* machines
1425 .getMachine({
1426 app_name: input.appName,
1427 machine_id: machineId,
1428 })
1429 .pipe(Retry.none);
1430 if (
1431 !sameOwnership(observed, expected) &&
1432 !sameOwnership(observed, previous)
1433 )
1434 return yield* new ReplicaOwnershipChanged({
1435 appName: input.appName,
1436 machineId,
1437 });
1438 if (
1439 !sameOwnership(observed, expected) ||
1440 input.configDrifted(observed, { mounts, metadata }) ||
1441 !sameChecks(observed.config?.checks, config.checks) ||
1442 !sameStopConfig(observed.config?.stop_config, config.stop_config)
1443 )
1444 return yield* new ReplicaNotCreated({
1445 appName: input.appName,
1446 name,
1447 });
1448 return observed;
1449 }).pipe(
1450 Effect.retry({
1451 times: 8,
1452 schedule: waitBackoff,
1453 while: (error) =>
1454 error._tag === "Fly.ReplicaNotCreated" ||
1455 TRANSIENT_GET_TAGS.some((tag) => tag === error._tag),
1456 }),
1457 Effect.timeout("60 seconds"),
1458 ),
1459 );
1460 }
1461
1462 current = yield* ensureStarted(
1463 input.appName,
1464 current,
1465 input.skipLaunch === true,
1466 input.policy.healthTimeoutMs,
1467 config,
1468 leases,
1469 );
1470 if (current.cordoned === true && !input.skipLaunch) {
1471 yield* setRouting(input.appName, current.id!, false, leases);
1472 if (hasPublishedService(config.services))
1473 yield* Effect.sleep("10 seconds");
1474 current = yield* waitHealthy(
1475 input.appName,
1476 current,
1477 input.policy.healthTimeoutMs,
1478 config,
1479 );
1480 }
1481 live.push(current);
1482 }
1483
1484 const liveIds = new Set(live.map((machine) => machine.id));
1485 yield* retireMachines(
1486 input.appName,
1487 owned.filter((machine) => !liveIds.has(machine.id)),
1488 false,
1489 leases,
1490 );
1491
1492 for (const group of groups) {
1493 for (const extra of [...group.byRegion.values()].flatMap(
1494 (placed) => placed.extras,
1495 )) {
1496 const volumeId = extra.id;
1497 if (volumeId === undefined || usedVolumeIds.has(volumeId)) continue;
1498 yield* deleteVolume(input.appName, volumeId);
1499 }
1500 }
1501
1502 const volumesById = new Map<string, FlyVolume>();
1503 for (const group of groups) {
1504 for (const volume of [...group.byRegion.values()].flatMap(
1505 (placed) => placed.volumes,
1506 )) {
1507 if (volume.id !== undefined) volumesById.set(volume.id, volume);
1508 }
1509 }
1510 const fresh = yield* Effect.forEach(
1511 live,
1512 (machine) =>
1513 machine.id === undefined
1514 ? Effect.succeed(machine)
1515 : getMachineById(input.appName, machine.id).pipe(
1516 Effect.map((next) => next ?? machine),
1517 ),
1518 { concurrency: 4 },
1519 );
1520 const replicas = fresh.map((machine) => toReplica(machine, volumesById));
1521 return toReplicaSet(
1522 replicas,
1523 input.appName,
1524 input.baseName,
1525 fresh[0]?.config?.services,
1526 );
1527});
1528
1529export const deleteReplicaSet = Effect.fn(function* (input: {
1530 appName: string;
1531 id: string;
1532 type: FlyAlchemyType;
1533 fqn: string;
1534 resourceInstanceId: string;
1535 force?: boolean;
1536 machineIds: readonly string[];
1537 volumeIds: readonly string[];
1538}) {
1539 return yield* usingMachineLeases(input.appName, undefined, (leases) =>
1540 Effect.gen(function* () {
1541 const owned = yield* leaseSnapshot(
1542 input.appName,
1543 ownedReplicas(yield* listMachinesByApp(input.appName), {
1544 ...input,
1545 metadata: yield* createMachineMetadata(input.id, input.type),
1546 }),
1547 leases,
1548 );
1549 if (input.force) {
1550 yield* Effect.forEach(
1551 owned,
1552 (machine) =>
1553 Effect.gen(function* () {
1554 if (!machine.id) return;
1555 const current = yield* getMachineById(input.appName, machine.id);
1556 if (!current) return;
1557 if (!sameOwnership(current, machine))
1558 return yield* new ReplicaOwnershipChanged({
1559 appName: input.appName,
1560 machineId: machine.id,
1561 });
1562 yield* deleteMachine(input.appName, machine.id, leases);
1563 }),
1564 { concurrency: 4 },
1565 );
1566 } else yield* retireMachines(input.appName, owned, false, leases);
1567 yield* Effect.forEach(
1568 [...new Set(input.volumeIds)].filter((id) => id.length > 0),
1569 (volumeId) => deleteVolume(input.appName, volumeId),
1570 { concurrency: 4 },
1571 );
1572 }),
1573 );
1574});
1575
1576export const volumeIdsOf = (set: {
1577 mounts?: readonly MountedDisk[];
1578 replicas?: readonly Replica[];
1579}): string[] => {
1580 const ids = new Set<string>();
1581 for (const mount of set.mounts ?? []) ids.add(mount.volumeId);
1582 for (const replica of set.replicas ?? []) {
1583 for (const mount of replica.mounts) ids.add(mount.volumeId);
1584 }
1585 return [...ids];
1586};
1587
1588export const groupGenerations = (machines: FlyMachine[]) => {
1589 const groups = new Map<string | undefined, FlyMachine[]>();
1590 for (const machine of machines) {
1591 const generation =
1592 machine.config?.metadata?.[alchemyMetadataKeys.generation];
1593 const group = groups.get(generation) ?? [];
1594 group.push(machine);
1595 groups.set(generation, group);
1596 }
1597 return groups;
1598};
1599
1600export const observeReplicaSet = Effect.fn(function* (input: {
1601 appName?: string;
1602 fqn: string;
1603 resourceInstanceId: string;
1604 id: string;
1605 type: FlyAlchemyType;
1606 machineIds?: readonly string[];
1607 baseName?: string;
1608}) {
1609 if (input.appName === undefined) return undefined;
1610 const owned = ownedReplicas(yield* listMachinesByApp(input.appName), {
1611 ...input,
1612 metadata: yield* createMachineMetadata(input.id, input.type),
1613 });
1614 const groups = groupGenerations(owned);
1615 const committed = [...groups.entries()]
1616 .filter(([generation, group]) => {
1617 if (generation === undefined) return false;
1618 const count = Number(
1619 group[0]?.config?.metadata?.[alchemyMetadataKeys.count],
1620 );
1621 if (
1622 group.some(
1623 (machine) => classifyDeploymentState(machine).protocol === "invalid",
1624 )
1625 )
1626 return false;
1627 const protocol2 = group.some(
1628 (machine) =>
1629 machine.config?.metadata?.[alchemyMetadataKeys.protocol] === "2",
1630 );
1631 if (protocol2 && !validProtocol2Generation(group)) return false;
1632 return (
1633 Number.isSafeInteger(count) &&
1634 count > 0 &&
1635 group.length === count &&
1636 new Set(group.map(replicaIndexOf)).size === count &&
1637 group.every((machine) => {
1638 const metadata = machine.config?.metadata;
1639 const index = replicaIndexOf(machine);
1640 return (
1641 index >= 0 &&
1642 index < count &&
1643 Number(metadata?.[alchemyMetadataKeys.count]) === count &&
1644 metadata?.[alchemyMetadataKeys.phase] === "active" &&
1645 machine.cordoned === false &&
1646 (metadata[alchemyMetadataKeys.protocol] === undefined ||
1647 ((metadata[alchemyMetadataKeys.protocol] === "1" ||
1648 metadata[alchemyMetadataKeys.protocol] === "2") &&
1649 metadata[alchemyMetadataKeys.restored] === "true" &&
1650 (metadata[alchemyMetadataKeys.role] === "idle" ||
1651 (machine.instance_id !== undefined &&
1652 metadata[alchemyMetadataKeys.checkedInstance] ===
1653 machine.instance_id))))
1654 );
1655 })
1656 );
1657 })
1658 .sort(
1659 ([, a], [, b]) =>
1660 Number(b[0]?.config?.metadata?.[alchemyMetadataKeys.sequence] ?? 0) -
1661 Number(a[0]?.config?.metadata?.[alchemyMetadataKeys.sequence] ?? 0),
1662 );
1663 const legacy = (groups.get(undefined) ?? []).filter((machine) => {
1664 const { protocol } = classifyDeploymentState(machine);
1665 return protocol === "legacy" || protocol === "1";
1666 });
1667 const listed = (committed[0]?.[1] ?? legacy).sort(
1668 (a, b) => replicaIndexOf(a) - replicaIndexOf(b),
1669 );
1670 const rolloutPending = owned.some((machine) => !listed.includes(machine));
1671 if (listed.length > 0) {
1672 const volumesById = new Map<string, FlyVolume>();
1673 for (const machine of listed) {
1674 for (const mount of machine.config?.mounts ?? []) {
1675 const volumeId = mount.volume;
1676 if (volumeId === undefined || volumesById.has(volumeId)) continue;
1677 const volume = yield* getVolumeById(input.appName, volumeId);
1678 if (volume !== undefined) volumesById.set(volumeId, volume);
1679 }
1680 }
1681 const replicas = listed.map((machine) => toReplica(machine, volumesById));
1682 return {
1683 ...toReplicaSet(
1684 replicas,
1685 input.appName,
1686 listed[0]?.config?.metadata?.[alchemyMetadataKeys.baseName] ??
1687 input.baseName ??
1688 replicas[0]?.name ??
1689 "",
1690 listed[0]?.config?.services,
1691 ),
1692 rolloutPending,
1693 };
1694 }
1695 if (owned.length)
1696 return {
1697 ...toReplicaSet([], input.appName, input.baseName ?? ""),
1698 region: owned[0]?.region ?? "",
1699 regions: [
1700 ...new Set(
1701 owned
1702 .map((machine) => machine.region ?? "")
1703 .filter((region) => region.length > 0),
1704 ),
1705 ],
1706 rolloutPending: true,
1707 };
1708 return undefined;
1709});