| 1 | import { regionOfReplica } from "./Region.ts"; |
| 2 | import * as machines from "@distilled.cloud/fly-io/machines"; |
| 3 | import type { Machine } from "@distilled.cloud/fly-io/machines"; |
| 4 | import * as Retry from "@distilled.cloud/fly-io/Retry"; |
| 5 | import * as Data from "effect/Data"; |
| 6 | import * as Effect from "effect/Effect"; |
| 7 | import * as Schedule from "effect/Schedule"; |
| 8 | import { deepEqual } from "../Diff.ts"; |
| 9 | import { sha256Object } from "../Util/sha256.ts"; |
| 10 | import { canonicalContainers } from "./MachineContainers.ts"; |
| 11 | import { alchemyMetadataKeys as keys } from "./Metadata.ts"; |
| 12 | import { |
| 13 | applyImageSet, |
| 14 | encodeImageSet, |
| 15 | pinsFromConfig, |
| 16 | sameImageSet, |
| 17 | validObservedImageSet, |
| 18 | } from "./DeploymentImages.ts"; |
| 19 | import { |
| 20 | classifyDeploymentState, |
| 21 | validProtocol2Generation, |
| 22 | } from "./DeploymentState.ts"; |
| 23 | import { usingMachineLeases, type MachineLeases } from "./leases.ts"; |
| 24 | import { |
| 25 | autostopMode, |
| 26 | checksPassing, |
| 27 | deleteMachine, |
| 28 | ensureStarted, |
| 29 | getMachineById, |
| 30 | groupGenerations, |
| 31 | hasPublishedService, |
| 32 | listMachinesByApp, |
| 33 | leaseSnapshot, |
| 34 | ownedReplicas, |
| 35 | ReplicaNotCreated, |
| 36 | replicaIndexOf, |
| 37 | retireMachines, |
| 38 | sameChecks, |
| 39 | sameServices, |
| 40 | sameStopConfig, |
| 41 | toReplica, |
| 42 | toReplicaSet, |
| 43 | waitHealthy, |
| 44 | type ReconcileReplicasInput, |
| 45 | } from "./replicas.ts"; |
| 46 | |
| 47 | const metadataOf = (machine: Machine) => |
| 48 | Object.fromEntries( |
| 49 | Object.entries(machine.config?.metadata ?? {}).flatMap(([key, value]) => |
| 50 | value === undefined ? [] : [[key, value]], |
| 51 | ), |
| 52 | ); |
| 53 | |
| 54 | const setMetadata = Effect.fn(function* ( |
| 55 | appName: string, |
| 56 | machine: Machine, |
| 57 | metadata: Record<string, string>, |
| 58 | leases: MachineLeases, |
| 59 | ) { |
| 60 | yield* leases.checkTarget(machine.id!); |
| 61 | // Metadata is not nonce-fenced by Fly; it is recovery evidence, not a lock. |
| 62 | yield* leases.mutate( |
| 63 | machine.id!, |
| 64 | () => |
| 65 | machines |
| 66 | .patchMachineMetadata({ |
| 67 | app_name: appName, |
| 68 | machine_id: machine.id!, |
| 69 | metadata, |
| 70 | }) |
| 71 | .pipe(Effect.timeout("30 seconds")), |
| 72 | { idempotent: true }, |
| 73 | ); |
| 74 | yield* Effect.sync(() => { |
| 75 | machine.config = { ...machine.config, metadata }; |
| 76 | }); |
| 77 | }); |
| 78 | |
| 79 | const phase = Effect.fn(function* ( |
| 80 | appName: string, |
| 81 | machine: Machine, |
| 82 | value: string, |
| 83 | leases: MachineLeases, |
| 84 | ) { |
| 85 | if (machine.config?.metadata?.[keys.phase] !== value) { |
| 86 | yield* setMetadata( |
| 87 | appName, |
| 88 | machine, |
| 89 | { |
| 90 | ...metadataOf(machine), |
| 91 | [keys.phase]: value, |
| 92 | }, |
| 93 | leases, |
| 94 | ); |
| 95 | } |
| 96 | }); |
| 97 | |
| 98 | export const setRouting = ( |
| 99 | appName: string, |
| 100 | machineId: string, |
| 101 | cordoned: boolean, |
| 102 | existingLeases?: MachineLeases, |
| 103 | ) => |
| 104 | usingMachineLeases(appName, existingLeases, (leases) => |
| 105 | Effect.gen(function* () { |
| 106 | yield* leases.acquire([machineId]); |
| 107 | yield* leases.mutate( |
| 108 | machineId, |
| 109 | (lease_nonce) => |
| 110 | Effect.gen(function* () { |
| 111 | const request = { |
| 112 | app_name: appName, |
| 113 | machine_id: machineId, |
| 114 | lease_nonce, |
| 115 | }; |
| 116 | if (cordoned) yield* machines.cordonMachine(request); |
| 117 | else yield* machines.uncordonMachine(request); |
| 118 | }).pipe(Effect.timeout("30 seconds")), |
| 119 | { idempotent: true }, |
| 120 | ); |
| 121 | yield* getMachineById(appName, machineId).pipe( |
| 122 | Effect.filterOrFail( |
| 123 | (machine) => machine !== undefined && machine.cordoned === cordoned, |
| 124 | () => new ReplicaNotCreated({ appName, name: machineId }), |
| 125 | ), |
| 126 | Effect.retry({ |
| 127 | times: 8, |
| 128 | schedule: Schedule.spaced("500 millis"), |
| 129 | while: (error) => error._tag === "Fly.ReplicaNotCreated", |
| 130 | }), |
| 131 | ); |
| 132 | }), |
| 133 | ); |
| 134 | |
| 135 | const pinnedImage = (machine: Machine) => { |
| 136 | const image = machine.image_ref; |
| 137 | if (!image?.digest || !image.repository) return undefined; |
| 138 | return `${image.registry ? `${image.registry}/` : ""}${image.repository}@${image.digest}`; |
| 139 | }; |
| 140 | |
| 141 | export class DeploymentRecoveryAmbiguous extends Data.TaggedError( |
| 142 | "Fly.DeploymentRecoveryAmbiguous", |
| 143 | )<{ |
| 144 | appName: string; |
| 145 | message: string; |
| 146 | }> {} |
| 147 | |
| 148 | export const readinessRoles = ( |
| 149 | config: machines.FlyMachineConfig, |
| 150 | count: number, |
| 151 | predecessors: Machine[], |
| 152 | ): Array<"run" | "idle"> => { |
| 153 | const services = config.services ?? []; |
| 154 | if ( |
| 155 | !services.length || |
| 156 | services.some((service) => autostopMode(service.autostop) === "off") |
| 157 | ) { |
| 158 | return Array.from({ length: count }, () => "run"); |
| 159 | } |
| 160 | const roles: Array<"run" | "idle"> = Array.from( |
| 161 | { length: count }, |
| 162 | (_, index) => |
| 163 | predecessors.some( |
| 164 | (machine) => |
| 165 | replicaIndexOf(machine) === index && machine.state === "started", |
| 166 | ) |
| 167 | ? "run" |
| 168 | : "idle", |
| 169 | ); |
| 170 | const floor = Math.min( |
| 171 | count, |
| 172 | Math.max( |
| 173 | 1, |
| 174 | ...services.map((service) => service.min_machines_running ?? 0), |
| 175 | ), |
| 176 | ); |
| 177 | for ( |
| 178 | let index = 0; |
| 179 | roles.filter((role) => role === "run").length < floor; |
| 180 | index++ |
| 181 | ) |
| 182 | roles[index] = "run"; |
| 183 | return roles; |
| 184 | }; |
| 185 | |
| 186 | const idle = (machine: Machine) => |
| 187 | machine.state === "stopped" || |
| 188 | machine.state === "suspended" || |
| 189 | machine.state === "created"; |
| 190 | |
| 191 | /** Cloud metadata records preparation and promotion; replica zero commits a complete set. */ |
| 192 | export const reconcileBlueGreen = Effect.fn(function* ( |
| 193 | input: ReconcileReplicasInput, |
| 194 | ownership: Record<string, string>, |
| 195 | metadata: Record<string, string>, |
| 196 | leases: MachineLeases, |
| 197 | ) { |
| 198 | const config = input.buildConfig({ index: 0, mounts: [], metadata }); |
| 199 | const containerPins = |
| 200 | config.containers === undefined ? undefined : pinsFromConfig(config); |
| 201 | const containerMode = config.containers !== undefined; |
| 202 | const ambiguous = (message: string) => |
| 203 | new DeploymentRecoveryAmbiguous({ appName: input.appName, message }); |
| 204 | if (containerMode && containerPins === undefined) |
| 205 | return yield* ambiguous( |
| 206 | "Blue/green requires a complete immutable container image set.", |
| 207 | ); |
| 208 | const workload = yield* sha256Object({ |
| 209 | config: containerMode |
| 210 | ? { |
| 211 | ...config, |
| 212 | containers: canonicalContainers(config.containers), |
| 213 | } |
| 214 | : config, |
| 215 | count: input.count, |
| 216 | minSecretsVersion: input.minSecretsVersion, |
| 217 | // A single region hashes as before, so existing generations stay current. |
| 218 | region: input.regions.length === 1 ? input.regions[0] : [...input.regions], |
| 219 | }); |
| 220 | const observe = listMachinesByApp(input.appName).pipe( |
| 221 | Effect.flatMap((listed) => { |
| 222 | const owned = ownedReplicas(listed, { |
| 223 | ...input, |
| 224 | metadata: ownership, |
| 225 | machineIds: input.outputMachineIds, |
| 226 | }); |
| 227 | const changed = listed.find( |
| 228 | (machine) => |
| 229 | machine.id && |
| 230 | input.outputMachineIds?.includes(machine.id) && |
| 231 | !owned.includes(machine), |
| 232 | ); |
| 233 | return changed |
| 234 | ? Effect.fail( |
| 235 | ambiguous( |
| 236 | `Cached Machine ${changed.id} no longer has the expected ownership.`, |
| 237 | ), |
| 238 | ) |
| 239 | : Effect.succeed(owned); |
| 240 | }), |
| 241 | ); |
| 242 | const snapshot = yield* observe; |
| 243 | const owned = yield* leaseSnapshot(input.appName, snapshot, leases); |
| 244 | // A vanished snapshot member can have a successor outside this inventory. |
| 245 | if ( |
| 246 | snapshot.some( |
| 247 | (machine) => !owned.some((current) => current.id === machine.id), |
| 248 | ) |
| 249 | ) { |
| 250 | return yield* ambiguous( |
| 251 | "Owned Machine membership changed while acquiring leases; retry with a fresh deployment snapshot.", |
| 252 | ); |
| 253 | } |
| 254 | for (const machine of owned) { |
| 255 | const state = classifyDeploymentState(machine); |
| 256 | if (state.protocol === "invalid") |
| 257 | return yield* ambiguous(`Machine ${machine.id}: ${state.reason}`); |
| 258 | } |
| 259 | |
| 260 | const idleAllowed = |
| 261 | (config.services?.length ?? 0) > 0 && |
| 262 | config.services!.every( |
| 263 | (service) => autostopMode(service.autostop) !== "off", |
| 264 | ); |
| 265 | const preparationServices = config.services?.map((service) => ({ |
| 266 | ...service, |
| 267 | autostop: "off", |
| 268 | })); |
| 269 | const mismatchOf = (machine: Machine): string | undefined => { |
| 270 | const observed = metadataOf(machine); |
| 271 | if (containerMode) { |
| 272 | if ( |
| 273 | observed[keys.protocol] !== "2" || |
| 274 | !validObservedImageSet(machine) || |
| 275 | !sameImageSet(pinsFromConfig(machine.config), containerPins) |
| 276 | ) |
| 277 | return "container image set"; |
| 278 | } else { |
| 279 | const pinned = observed[keys.image]; |
| 280 | if (pinned === undefined && observed[keys.phase] !== "candidate") |
| 281 | return "metadata.image"; |
| 282 | if ( |
| 283 | !pinnedImage(machine) || |
| 284 | (pinned !== undefined && pinned !== pinnedImage(machine)) |
| 285 | ) |
| 286 | return "image_ref"; |
| 287 | } |
| 288 | const secretsVersion = Number(observed[keys.secretsVersion] ?? -1); |
| 289 | if ( |
| 290 | input.minSecretsVersion !== undefined && |
| 291 | (!Number.isSafeInteger(secretsVersion) || |
| 292 | secretsVersion < input.minSecretsVersion) |
| 293 | ) |
| 294 | return "metadata.secretsVersion"; |
| 295 | const temporary = |
| 296 | observed[keys.role] === "run" && |
| 297 | observed[keys.restored] === "false" && |
| 298 | sameServices(machine.config?.services, preparationServices); |
| 299 | if (observed[keys.workload] !== workload) return "metadata.workload"; |
| 300 | if ( |
| 301 | input.configDrifted( |
| 302 | { |
| 303 | ...machine, |
| 304 | config: { |
| 305 | ...machine.config, |
| 306 | image: config.image, |
| 307 | services: temporary ? config.services : machine.config?.services, |
| 308 | }, |
| 309 | }, |
| 310 | { |
| 311 | mounts: [], |
| 312 | metadata: { |
| 313 | ...Object.fromEntries( |
| 314 | Object.entries(observed).filter(([key]) => |
| 315 | key.startsWith("alchemy."), |
| 316 | ), |
| 317 | ), |
| 318 | ...metadata, |
| 319 | }, |
| 320 | }, |
| 321 | ) |
| 322 | ) |
| 323 | return "config"; |
| 324 | if (!sameChecks(machine.config?.checks, config.checks)) |
| 325 | return "config.checks"; |
| 326 | if (!sameStopConfig(machine.config?.stop_config, config.stop_config)) |
| 327 | return "config.stop_config"; |
| 328 | return undefined; |
| 329 | }; |
| 330 | const matches = (machine: Machine) => mismatchOf(machine) === undefined; |
| 331 | const groups = groupGenerations(owned); |
| 332 | const sequences = new Set<string>(); |
| 333 | for (const [generation, group] of groups) { |
| 334 | if (generation === undefined) continue; |
| 335 | const sequence = group[0]?.config?.metadata?.[keys.sequence]; |
| 336 | if ( |
| 337 | !sequence || |
| 338 | !Number.isSafeInteger(Number(sequence)) || |
| 339 | Number(sequence) < 1 || |
| 340 | sequences.has(sequence) || |
| 341 | group.some( |
| 342 | (machine) => machine.config?.metadata?.[keys.sequence] !== sequence, |
| 343 | ) || |
| 344 | (group.some( |
| 345 | (machine) => machine.config?.metadata?.[keys.protocol] === "2", |
| 346 | ) && |
| 347 | !validProtocol2Generation(group)) |
| 348 | ) { |
| 349 | return yield* ambiguous( |
| 350 | "Owned generations have ambiguous sequence/lineage metadata; preserving all capacity.", |
| 351 | ); |
| 352 | } |
| 353 | sequences.add(sequence); |
| 354 | } |
| 355 | const reusable = [...groups.entries()] |
| 356 | .filter( |
| 357 | ([generation, group]) => generation !== undefined && group.every(matches), |
| 358 | ) |
| 359 | .sort( |
| 360 | ([, a], [, b]) => |
| 361 | Number(b[0]?.config?.metadata?.[keys.sequence] ?? 0) - |
| 362 | Number(a[0]?.config?.metadata?.[keys.sequence] ?? 0), |
| 363 | )[0]; |
| 364 | const generation = |
| 365 | reusable?.[0] ?? |
| 366 | (yield* sha256Object({ |
| 367 | workload, |
| 368 | predecessors: owned.map((machine) => machine.id).sort(), |
| 369 | })).slice(0, 20); |
| 370 | const desired = reusable?.[1] ?? []; |
| 371 | if ( |
| 372 | new Set(desired.map(replicaIndexOf)).size !== desired.length || |
| 373 | desired.some((machine) => replicaIndexOf(machine) >= input.count) |
| 374 | ) { |
| 375 | return yield* ambiguous( |
| 376 | "The recovered generation has duplicate or out-of-range replica indices.", |
| 377 | ); |
| 378 | } |
| 379 | const sequence = |
| 380 | desired[0]?.config?.metadata?.[keys.sequence] ?? |
| 381 | String( |
| 382 | 1 + |
| 383 | Math.max( |
| 384 | 0, |
| 385 | ...owned.map((machine) => |
| 386 | Number(machine.config?.metadata?.[keys.sequence] ?? 0), |
| 387 | ), |
| 388 | ), |
| 389 | ); |
| 390 | const predecessors = owned.filter( |
| 391 | (machine) => machine.config?.metadata?.[keys.generation] !== generation, |
| 392 | ); |
| 393 | const recordedRoles = desired[0]?.config?.metadata?.[keys.roles]; |
| 394 | const roles = |
| 395 | recordedRoles?.split(",") ?? |
| 396 | readinessRoles(config, input.count, predecessors); |
| 397 | if ( |
| 398 | roles.length !== input.count || |
| 399 | roles.some((role) => role !== "run" && role !== "idle") || |
| 400 | !roles.includes("run") || |
| 401 | desired.some( |
| 402 | (machine) => |
| 403 | machine.config?.metadata?.[keys.roles] !== recordedRoles || |
| 404 | ((machine.config?.metadata?.[keys.protocol] === "1" || |
| 405 | machine.config?.metadata?.[keys.protocol] === "2") && |
| 406 | machine.config.metadata[keys.role] !== |
| 407 | roles[replicaIndexOf(machine)]), |
| 408 | ) |
| 409 | ) { |
| 410 | return yield* ambiguous( |
| 411 | "The recovered generation has inconsistent readiness roles.", |
| 412 | ); |
| 413 | } |
| 414 | const complete = |
| 415 | desired.length === input.count && |
| 416 | desired.every( |
| 417 | (machine) => |
| 418 | machine.config?.metadata?.[keys.phase] === "active" && |
| 419 | machine.config.metadata[keys.restored] === "true" && |
| 420 | (roles[replicaIndexOf(machine)] === "idle" || |
| 421 | (machine.instance_id !== undefined && |
| 422 | machine.config.metadata[keys.checkedInstance] === |
| 423 | machine.instance_id)) && |
| 424 | machine.cordoned === false, |
| 425 | ); |
| 426 | if ( |
| 427 | complete && |
| 428 | !predecessors.length && |
| 429 | desired.every( |
| 430 | (machine) => |
| 431 | machine.state === "started" || (idleAllowed && idle(machine)), |
| 432 | ) |
| 433 | ) { |
| 434 | return toReplicaSet( |
| 435 | [...desired] |
| 436 | .sort((a, b) => replicaIndexOf(a) - replicaIndexOf(b)) |
| 437 | .map((machine) => toReplica(machine, new Map())), |
| 438 | input.appName, |
| 439 | input.baseName, |
| 440 | config.services, |
| 441 | ); |
| 442 | } |
| 443 | const candidates: Machine[] = []; |
| 444 | const snapshotIds = new Set(owned.map((machine) => machine.id)); |
| 445 | let image = desired[0] ? pinnedImage(desired[0]) : undefined; |
| 446 | const order = Array.from({ length: input.count }, (_, index) => index).sort( |
| 447 | (a, b) => Number(roles[b] === "run") - Number(roles[a] === "run") || a - b, |
| 448 | ); |
| 449 | // Creation-time predecessor IDs are diagnostic lineage, not deletion authority. |
| 450 | const predecessorIds = predecessors |
| 451 | .flatMap((machine) => (machine.id ? [machine.id] : [])) |
| 452 | .sort() |
| 453 | .join(","); |
| 454 | yield* Effect.gen(function* () { |
| 455 | for (const index of order) { |
| 456 | const suffix = `-${generation}-${index}`; |
| 457 | const name = `${input.baseName.slice(0, 30 - suffix.length).replace(/-+$/g, "")}${suffix}`; |
| 458 | let current = desired.find( |
| 459 | (machine) => replicaIndexOf(machine) === index, |
| 460 | ); |
| 461 | if (current === undefined) { |
| 462 | const run = roles[index] === "run"; |
| 463 | const candidateConfig = input.buildConfig({ |
| 464 | index, |
| 465 | mounts: [], |
| 466 | metadata: { |
| 467 | ...metadata, |
| 468 | [keys.replica]: String(index), |
| 469 | [keys.generation]: generation, |
| 470 | [keys.workload]: workload, |
| 471 | [keys.phase]: "candidate", |
| 472 | [keys.sequence]: sequence, |
| 473 | [keys.count]: String(input.count), |
| 474 | [keys.protocol]: containerMode ? "2" : "1", |
| 475 | [keys.roles]: roles.join(","), |
| 476 | [keys.role]: roles[index]!, |
| 477 | [keys.predecessors]: predecessorIds, |
| 478 | [keys.restored]: String( |
| 479 | !run || sameServices(config.services, preparationServices), |
| 480 | ), |
| 481 | ...(input.minSecretsVersion === undefined |
| 482 | ? {} |
| 483 | : { [keys.secretsVersion]: String(input.minSecretsVersion) }), |
| 484 | ...(containerPins |
| 485 | ? { [keys.containerImageSet]: encodeImageSet(containerPins)! } |
| 486 | : image |
| 487 | ? { [keys.image]: image } |
| 488 | : {}), |
| 489 | }, |
| 490 | }); |
| 491 | const pinnedConfig = containerPins |
| 492 | ? applyImageSet(candidateConfig, containerPins) |
| 493 | : candidateConfig; |
| 494 | if (pinnedConfig === undefined) |
| 495 | return yield* ambiguous("Incomplete candidate image set."); |
| 496 | const readback = observe.pipe( |
| 497 | Effect.map((listed) => |
| 498 | listed.find( |
| 499 | (machine) => |
| 500 | machine.name === name && |
| 501 | machine.config?.metadata?.[keys.generation] === generation && |
| 502 | replicaIndexOf(machine) === index && |
| 503 | matches(machine), |
| 504 | ), |
| 505 | ), |
| 506 | Effect.filterOrFail( |
| 507 | (machine) => machine !== undefined, |
| 508 | () => new ReplicaNotCreated({ appName: input.appName, name }), |
| 509 | ), |
| 510 | Effect.retry({ |
| 511 | times: 8, |
| 512 | schedule: Schedule.spaced("500 millis"), |
| 513 | while: (error) => error._tag === "Fly.ReplicaNotCreated", |
| 514 | }), |
| 515 | ); |
| 516 | current = yield* machines |
| 517 | .createMachine({ |
| 518 | app_name: input.appName, |
| 519 | name, |
| 520 | region: regionOfReplica(input.regions, index), |
| 521 | config: { |
| 522 | ...pinnedConfig, |
| 523 | image: containerPins ? undefined : (image ?? config.image), |
| 524 | services: run ? preparationServices : config.services, |
| 525 | }, |
| 526 | skip_launch: !run, |
| 527 | skip_service_registration: true, |
| 528 | min_secrets_version: input.minSecretsVersion, |
| 529 | }) |
| 530 | .pipe( |
| 531 | Retry.none, |
| 532 | Effect.timeout("30 seconds"), |
| 533 | Effect.catchTag( |
| 534 | ["Conflict", "GatewayTimeout", "TimeoutError"], |
| 535 | () => readback, |
| 536 | ), |
| 537 | Effect.catchTag("HttpClientError", (error) => |
| 538 | error.reason._tag === "TransportError" |
| 539 | ? readback |
| 540 | : Effect.fail(error), |
| 541 | ), |
| 542 | ); |
| 543 | } |
| 544 | if (!current?.id) |
| 545 | return yield* new ReplicaNotCreated({ appName: input.appName, name }); |
| 546 | const leased = (yield* leaseSnapshot( |
| 547 | input.appName, |
| 548 | [current], |
| 549 | leases, |
| 550 | ))[0]; |
| 551 | if (!leased?.id) |
| 552 | return yield* ambiguous( |
| 553 | `Candidate ${current.id} disappeared before its lease was acquired.`, |
| 554 | ); |
| 555 | const mismatch = mismatchOf(leased); |
| 556 | if (mismatch !== undefined) |
| 557 | return yield* ambiguous( |
| 558 | `Candidate ${current.id} changed before its lease was acquired. Mismatch: ${mismatch}.`, |
| 559 | ); |
| 560 | current = leased; |
| 561 | snapshotIds.add(current.id); |
| 562 | if (containerPins) { |
| 563 | if ( |
| 564 | !validObservedImageSet(current) || |
| 565 | !sameImageSet(pinsFromConfig(current.config), containerPins) |
| 566 | ) |
| 567 | return yield* ambiguous( |
| 568 | `Container image pin mismatch on ${current.id}.`, |
| 569 | ); |
| 570 | } else { |
| 571 | const resolvedImage = pinnedImage(current); |
| 572 | if (!resolvedImage || (image !== undefined && image !== resolvedImage)) |
| 573 | return yield* ambiguous(`Image pin mismatch on ${current.id}.`); |
| 574 | image = resolvedImage; |
| 575 | if (current.config?.metadata?.[keys.image] !== image) |
| 576 | yield* setMetadata( |
| 577 | input.appName, |
| 578 | current, |
| 579 | { |
| 580 | ...metadataOf(current), |
| 581 | [keys.image]: image, |
| 582 | }, |
| 583 | leases, |
| 584 | ); |
| 585 | } |
| 586 | const checked = |
| 587 | current.config?.metadata?.[keys.checkedInstance] === |
| 588 | current.instance_id && current.instance_id !== undefined; |
| 589 | if ( |
| 590 | roles[index] === "run" && |
| 591 | !(idleAllowed && checked && idle(current)) |
| 592 | ) { |
| 593 | current = yield* ensureStarted( |
| 594 | input.appName, |
| 595 | current, |
| 596 | false, |
| 597 | input.policy.healthTimeoutMs, |
| 598 | current.config, |
| 599 | leases, |
| 600 | ); |
| 601 | } else if (!idle(current)) { |
| 602 | current = yield* waitHealthy( |
| 603 | input.appName, |
| 604 | current, |
| 605 | input.policy.healthTimeoutMs, |
| 606 | current.config, |
| 607 | ); |
| 608 | } |
| 609 | candidates[index] = current; |
| 610 | } |
| 611 | }).pipe( |
| 612 | Effect.catch((error) => |
| 613 | Effect.gen(function* () { |
| 614 | const found = (yield* observe).filter( |
| 615 | (machine) => |
| 616 | snapshotIds.has(machine.id) && |
| 617 | machine.config?.metadata?.[keys.generation] === generation, |
| 618 | ); |
| 619 | // A possibly promoted candidate is capacity, not rollback garbage. |
| 620 | if ( |
| 621 | found.every( |
| 622 | (machine) => |
| 623 | (!containerMode || matches(machine)) && |
| 624 | machine.config?.metadata?.[keys.phase] === "candidate" && |
| 625 | machine.cordoned === true, |
| 626 | ) |
| 627 | ) { |
| 628 | yield* Effect.forEach( |
| 629 | found, |
| 630 | (machine) => |
| 631 | deleteMachine(input.appName, machine.id!, leases).pipe( |
| 632 | Effect.catch((cleanup) => |
| 633 | Effect.logWarning("Fly candidate cleanup remains pending", { |
| 634 | appName: input.appName, |
| 635 | machineId: machine.id, |
| 636 | error: cleanup._tag, |
| 637 | }), |
| 638 | ), |
| 639 | ), |
| 640 | { concurrency: 4 }, |
| 641 | ); |
| 642 | } |
| 643 | return yield* Effect.fail(error); |
| 644 | }), |
| 645 | ), |
| 646 | ); |
| 647 | |
| 648 | const needsPromotion = candidates.some( |
| 649 | (machine) => |
| 650 | machine.cordoned !== false || |
| 651 | machine.config?.metadata?.[keys.phase] !== "active", |
| 652 | ); |
| 653 | if (needsPromotion) { |
| 654 | for (const machine of candidates) |
| 655 | yield* phase(input.appName, machine, "promoting", leases); |
| 656 | // Promotion metadata resets reports; validate the whole set before routing any member. |
| 657 | for (const [index, machine] of candidates.entries()) { |
| 658 | const current = yield* getMachineById(input.appName, machine.id!); |
| 659 | if (!current || !matches(current)) |
| 660 | return yield* ambiguous( |
| 661 | `Candidate ${machine.id} changed before promotion.`, |
| 662 | ); |
| 663 | if (current.state === "started") { |
| 664 | candidates[index] = yield* waitHealthy( |
| 665 | input.appName, |
| 666 | current, |
| 667 | input.policy.healthTimeoutMs, |
| 668 | current.config, |
| 669 | ); |
| 670 | } else if ( |
| 671 | !idleAllowed || |
| 672 | !idle(current) || |
| 673 | (roles[index] === "run" && |
| 674 | (current.instance_id === undefined || |
| 675 | current.config?.metadata?.[keys.checkedInstance] !== |
| 676 | current.instance_id)) |
| 677 | ) { |
| 678 | return yield* ambiguous( |
| 679 | `Candidate ${machine.id} has no readiness proof before promotion.`, |
| 680 | ); |
| 681 | } else candidates[index] = current; |
| 682 | } |
| 683 | for (const machine of candidates) |
| 684 | if (machine.cordoned !== false) |
| 685 | yield* setRouting(input.appName, machine.id!, false, leases); |
| 686 | if (hasPublishedService(config.services)) yield* Effect.sleep("10 seconds"); |
| 687 | } |
| 688 | for (const [index, machine] of candidates.entries()) { |
| 689 | let current = machine; |
| 690 | if (current.config?.metadata?.[keys.restored] === "false") { |
| 691 | // Restoring services creates a new instance; its predecessor's checks cannot prove readiness. |
| 692 | const machineId = current.id!; |
| 693 | const currentVersion = current.instance_id; |
| 694 | const restoreConfig = input.buildConfig({ |
| 695 | index, |
| 696 | mounts: [], |
| 697 | metadata: { ...metadataOf(current), [keys.restored]: "true" }, |
| 698 | }); |
| 699 | const restored = containerPins |
| 700 | ? applyImageSet(restoreConfig, containerPins) |
| 701 | : { ...restoreConfig, image }; |
| 702 | if (restored === undefined) |
| 703 | return yield* ambiguous( |
| 704 | "Incomplete image set during service restoration.", |
| 705 | ); |
| 706 | current = yield* leases.mutate(machineId, (lease_nonce) => |
| 707 | machines |
| 708 | .updateMachine({ |
| 709 | app_name: input.appName, |
| 710 | machine_id: machineId, |
| 711 | lease_nonce, |
| 712 | current_version: currentVersion, |
| 713 | config: restored, |
| 714 | min_secrets_version: input.minSecretsVersion, |
| 715 | }) |
| 716 | .pipe(Effect.timeout("30 seconds")), |
| 717 | ); |
| 718 | current = yield* ensureStarted( |
| 719 | input.appName, |
| 720 | current, |
| 721 | false, |
| 722 | input.policy.healthTimeoutMs, |
| 723 | config, |
| 724 | leases, |
| 725 | ); |
| 726 | } else { |
| 727 | current = (yield* getMachineById(input.appName, current.id!)) ?? current; |
| 728 | if (current.instance_id !== machine.instance_id) { |
| 729 | current = yield* ensureStarted( |
| 730 | input.appName, |
| 731 | current, |
| 732 | false, |
| 733 | input.policy.healthTimeoutMs, |
| 734 | config, |
| 735 | leases, |
| 736 | ); |
| 737 | } |
| 738 | if ( |
| 739 | current.state !== "started" && |
| 740 | roles[index] === "run" && |
| 741 | current.config?.metadata?.[keys.checkedInstance] !== current.instance_id |
| 742 | ) { |
| 743 | current = yield* ensureStarted( |
| 744 | input.appName, |
| 745 | current, |
| 746 | false, |
| 747 | input.policy.healthTimeoutMs, |
| 748 | config, |
| 749 | leases, |
| 750 | ); |
| 751 | } |
| 752 | } |
| 753 | candidates[index] = current; |
| 754 | } |
| 755 | // Reset reports while still pending; an interrupted validation is not a commit. |
| 756 | for (const machine of candidates) |
| 757 | yield* phase(input.appName, machine, "validating", leases); |
| 758 | for (const [index, machine] of candidates.entries()) { |
| 759 | const current = yield* getMachineById(input.appName, machine.id!); |
| 760 | if ( |
| 761 | !current || |
| 762 | !matches(current) || |
| 763 | current.config?.metadata?.[keys.restored] !== "true" |
| 764 | ) |
| 765 | return yield* ambiguous( |
| 766 | `Candidate ${machine.id} changed before commitment.`, |
| 767 | ); |
| 768 | if (current.state === "started") |
| 769 | candidates[index] = yield* waitHealthy( |
| 770 | input.appName, |
| 771 | current, |
| 772 | input.policy.healthTimeoutMs, |
| 773 | config, |
| 774 | ); |
| 775 | else if ( |
| 776 | !idleAllowed || |
| 777 | !idle(current) || |
| 778 | (roles[index] === "run" && |
| 779 | (current.instance_id === undefined || |
| 780 | (current.config?.metadata?.[keys.checkedInstance] !== |
| 781 | current.instance_id && |
| 782 | !( |
| 783 | machine.instance_id === current.instance_id && |
| 784 | checksPassing(machine, config) |
| 785 | )))) |
| 786 | ) { |
| 787 | return yield* ambiguous( |
| 788 | `Candidate ${machine.id} has no readiness proof for its current instance.`, |
| 789 | ); |
| 790 | } else candidates[index] = current; |
| 791 | } |
| 792 | // All final checks precede the commit stamps, with replica zero last. |
| 793 | for (const machine of [...candidates].reverse()) { |
| 794 | const committed = { |
| 795 | ...metadataOf(machine), |
| 796 | [keys.phase]: "active", |
| 797 | ...(machine.state === "started" || |
| 798 | roles[replicaIndexOf(machine)] === "run" |
| 799 | ? { [keys.checkedInstance]: machine.instance_id! } |
| 800 | : {}), |
| 801 | }; |
| 802 | if (!deepEqual(metadataOf(machine), committed)) |
| 803 | yield* setMetadata(input.appName, machine, committed, leases); |
| 804 | } |
| 805 | // Legacy ownership is required; predecessor phase stamps are only advisory. |
| 806 | for (const machine of predecessors) { |
| 807 | if (machine.config?.metadata?.[keys.protocol] === "2") { |
| 808 | const observed = yield* getMachineById(input.appName, machine.id!); |
| 809 | if ( |
| 810 | !observed || |
| 811 | observed.config?.metadata?.[keys.protocol] !== "2" || |
| 812 | !validObservedImageSet(observed) || |
| 813 | !sameImageSet( |
| 814 | pinsFromConfig(observed.config), |
| 815 | pinsFromConfig(machine.config), |
| 816 | ) |
| 817 | ) |
| 818 | return yield* ambiguous( |
| 819 | `Predecessor ${machine.id} image identity changed before retirement.`, |
| 820 | ); |
| 821 | } |
| 822 | if (machine.config?.metadata?.[keys.instance] === undefined) |
| 823 | yield* setMetadata( |
| 824 | input.appName, |
| 825 | machine, |
| 826 | { |
| 827 | ...metadataOf(machine), |
| 828 | ...metadata, |
| 829 | }, |
| 830 | leases, |
| 831 | ); |
| 832 | if (machine.host_status !== "unreachable") |
| 833 | yield* phase(input.appName, machine, "retiring", leases).pipe( |
| 834 | Effect.catch((error) => |
| 835 | Effect.logWarning("Fly predecessor phase metadata was not updated", { |
| 836 | appName: input.appName, |
| 837 | machineId: machine.id, |
| 838 | error: error._tag, |
| 839 | }), |
| 840 | ), |
| 841 | ); |
| 842 | } |
| 843 | yield* retireMachines(input.appName, predecessors, true, leases); |
| 844 | return toReplicaSet( |
| 845 | candidates.map((machine) => toReplica(machine, new Map())), |
| 846 | input.appName, |
| 847 | input.baseName, |
| 848 | config.services, |
| 849 | ); |
| 850 | }); |