// Phase E / Phase F — the job-migration saga, split out of automanager.ts // as a pure move (norsk-ctl docs/internal/design/job-migration.md §3). // Three strategies, one saga: the target is PLACED through the same // `placeAndProvision` every job takes (injected — the source node excluded, // an explicit target pinned), started with a StartJob carrying the // migration fields, and driven by the job's own history events keyed on // the INSTANCE KEY — plus the four handshake events for `handover`. The // AutoManager owns the event loop that feeds it and delegates its public // migration surface here; the job's binding (source, then target at // `done`) is the shared `JobBindings`. import * as uuid from "uuid"; import type * as ManagerPB from "@norskvideo/norsk-api/lib/manager_pb"; import type * as CommonPB from "@norskvideo/norsk-api/lib/shared/common_pb"; import type * as ManagerSdk from "@norskvideo/norsk-manager-sdk"; import type { JobId, JobWithHistory, NodeId, NorskManager, Role } from "@norskvideo/norsk-manager-sdk"; import type { AutoSettings, MigrationApplySummary, MigrationId, MigrationPhase, MigrationState, MigrationStrategy, PlacementOutcome, } from "./automanager"; import type { CancellableTimer, Clock } from "./clock"; import { debuglog } from "./debuglog"; import { JOB_SOCKET_CAPABILITY } from "./inventory"; import type { JobBindings } from "./jobBinding"; import type { CapabilityRequirement } from "./types"; // Migration phase-deadline defaults. Tuned to the design's expectation // that target launch + workflow init typically completes well under // two minutes, source-stop is workflow-driven and quick, and apply // is a handful of wire hops. A container stop is bounded by docker's // own stop timeout; a prepared container starts in seconds. const DEFAULT_MIGRATION_PROPOSE_DEADLINE_MS = 30_000; const DEFAULT_MIGRATION_LAUNCHING_DEADLINE_MS = 120_000; const DEFAULT_MIGRATION_TARGET_READY_DEADLINE_MS = 120_000; const DEFAULT_MIGRATION_SOURCE_STOP_DEADLINE_MS = 30_000; const DEFAULT_MIGRATION_APPLYING_DEADLINE_MS = 10_000; const DEFAULT_MIGRATION_STOPPING_SOURCE_DEADLINE_MS = 30_000; const DEFAULT_MIGRATION_STARTING_TARGET_DEADLINE_MS = 60_000; /** * The instance KEY a migration target wears: the source's slot plus the * first eight hex of the migration id, e.g. `primary\@3f9c1e2a`. Unique * per migration, so the source and the target of one migration — and * the targets of successive migrations of the same job — are distinct * `ActiveJobKey`s for the Manager's JobNodeMap. The job itself is still * told the SLOT (`primary` / `backup`) in `NORSK_RUNTIME_ROLE`. * @public */ export function migrationTargetRole(slot: Role, migrationId: MigrationId): Role { return `${slot}@${migrationId.slice(0, 8)}`; } /** Decode the apply summary a target sent; `undefined` for none or for * bytes that are not one. Never throws: this rides the event stream. */ export function readApplySummary(payload: Uint8Array | undefined): MigrationApplySummary | undefined { if (!payload || payload.byteLength === 0) return undefined; try { const parsed: unknown = JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(payload)); if (typeof parsed !== "object" || parsed === null) return undefined; const o = parsed as { applied?: unknown; failed?: unknown; skewMs?: unknown }; const applied = Array.isArray(o.applied) ? o.applied.filter((x): x is string => typeof x === "string") : []; const failed = Array.isArray(o.failed) ? o.failed .filter((f): f is { id: string; reason?: unknown } => typeof f === "object" && f !== null && typeof (f as { id?: unknown }).id === "string") .map((f) => ({ id: f.id, reason: typeof f.reason === "string" ? f.reason : "" })) : []; const skewMs = typeof o.skewMs === "number" && Number.isFinite(o.skewMs) ? o.skewMs : 0; return { applied, failed, skewMs }; } catch { return undefined; } } /** * The source's answer to the propose (`NodeControllerMigrationGetState`). * `refused` and `abandoned` both end the migration before any target is * placed; they differ only in who said so — the job, or us giving up. */ type ProposeAnswer = | { kind: "accepted"; payload: Uint8Array } | { kind: "refused"; reason: string } | { kind: "abandoned"; reason: string }; /** * What `placeAndProvision` is asked to do differently for a migration * target: bind as the migration claim's target rather than as the job's * own node, dispatch StartJob with the migration fields, pin `targetNode` * when the operator named one, and never consider `excludeNodes` (the * source's node, plus whatever recovery would skip). Everything else — * pools, requirements, resilience, hot spares, provision-a-node — is the * same path every job takes. */ export interface MigrationPlacement { migrationId: MigrationId; targetRole: Role; runtimeRole: Role; deferStart: boolean; targetNode?: NodeId; excludeNodes: Set; /** Capabilities the target node must advertise on top of the job's * own — a handover needs the worker's job socket, which a worker * running with it disabled does not advertise. */ extraCapabilities?: CapabilityRequirement[]; /** Phase G — this move may take an exclusively reserved hot spare. The * operator's own decision, per migration, not a policy read off the * job. */ overrideReservation?: boolean; } /** What the saga needs from the AutoManager, handed in explicitly. */ export interface MigrationSagaDeps { settings: AutoSettings; clock: Clock; norsk: Pick< NorskManager, | "startJob" | "stopJob" | "startPreparedJob" | "abortMigration" | "notifyMigrationGetState" | "notifyMigrationCutoverReady" | "notifyMigrationCutoverApply" >; bindings: JobBindings; jobOf: (jobId: JobId) => JobWithHistory | undefined; /** The one placement path every job takes; a migration target goes * through it with the `migration` argument set. */ placeAndProvision: ( jobId: JobId, tags: { [key: string]: string }, recovery: { excludeNodes: Set } | undefined, migration?: MigrationPlacement ) => Promise; /** The nodes ordinary recovery would skip right now. */ currentExcludeSet: () => Set; anyOtherNodeAdvertises: (name: string, except: NodeId) => boolean; /** A target's StartJob was refused (the placement metrics count it). */ onDispatchFailed: () => void; /** A migration reached `done` or `aborted` (drains watch for theirs). */ onMigrationEnded: (state: MigrationState) => void; } export class MigrationSaga { // Every migration this instance has run, done / aborted ones included // (ops tooling reads them until `forgetMigration`). "No concurrent // migrations per Job": the IN-FLIGHT one is the job's binding's // migration claim (`bindings.migrationOf`), which also carries the // target's binding and parked placement. private readonly migrationStates: Map = new Map(); // Deadlines — a per-migration timer fires if the current phase hasn't // transitioned by the configured window. Re-armed on every phase // transition; cleared on done/aborted/close(). private readonly migrationPhaseTimers: Map = new Map(); // Phase F. `proposing` is asynchronous — the source's answer arrives as // a MigrationStateEvent on the event stream — but `migrateJob` still // resolves only once a target is placed, so it parks here on a settler // the event handler (or the phase deadline, or any abort) calls. private readonly migrationProposals: Map void> = new Map(); constructor(private readonly deps: MigrationSagaDeps) {} /** Every in-flight migration's deadline timer, cancelled (close). */ cancelTimers(): void { for (const t of this.migrationPhaseTimers.values()) t.cancel(); this.migrationPhaseTimers.clear(); } /** * Migrate a running job to another node. Placement is the same path * every other job takes: `targetNodeId`, when given, is a hard pin * (that node or nowhere, requirements still checked); otherwise the * engine chooses, with the job's current node excluded. The target is * started under a key unique to this migration * (`migrationTargetRole`) while being told the source's slot. * * Resolves once the target has been placed and its StartJob * dispatched (phase `launching`); the rest is event-driven — watch * `onMigrationPhase` / `onMigrationCompleted` / `onMigrationAborted`. * Rejects (and records the migration as aborted) if placement finds * nowhere for the target or the dispatch fails; throws without * recording anything if the job is unknown, not bound to a node, not * running, already migrating, or `targetNodeId` is its current node. * * `sourceRole` defaults to the job's current instance key (its slot, * or the key a previous migration left it under); pass it explicitly * only when addressing one replica of a multi-replica bundle whose * instances share a jobId (they do not, today). * * @public */ async migrateJob( sourceJobId: JobId, opts: { targetNodeId?: NodeId; strategy?: MigrationStrategy; sourceRole?: Role; /** Phase G — allow the target onto an EXCLUSIVELY reserved hot * spare, one set aside for other templates. Steve's case: two * spares held for `xyz`, and an `abc` that has to move now. */ overrideReservation?: boolean; } = {} ): Promise { if (this.deps.bindings.migrationOf(sourceJobId) !== undefined) { throw new Error( `migrateJob: a migration is already in flight for jobId ${sourceJobId}` ); } const jh = this.deps.jobOf(sourceJobId); if (!jh) throw new Error(`migrateJob: unknown jobId ${sourceJobId}`); const req = jh.job.autoPlacement?.requirements; const slot = this.slotForJob(sourceJobId); if (!req || slot === undefined) { throw new Error( `migrateJob: job ${sourceJobId} is not an AutoManager-placed job ` + `(no placement requirements / bundle tags) — only placed jobs migrate` ); } const sourceNodeId = this.deps.bindings.nodeOf(sourceJobId); if (sourceNodeId === undefined) { throw new Error( `migrateJob: job ${sourceJobId} is not bound to a node — nothing to migrate` ); } if (opts.targetNodeId !== undefined && opts.targetNodeId === sourceNodeId) { throw new Error( `migrateJob: target node ${opts.targetNodeId} is where job ${sourceJobId} already runs` ); } const sourceRole = opts.sourceRole ?? this.roleForJob(sourceJobId); // Only a RUNNING instance migrates: the saga's stoppingSource step // waits on the source's JobStopped, which an instance that never // ran cannot produce, and a source still waiting for its node would // never be dispatched (bindProvisionedJob defers to the migration). const newestForSource = newestHistoryEntryForRole(jh, sourceRole); if ( newestForSource?.event !== "running" || newestForSource.nodeMetadata.nodeId !== sourceNodeId ) { throw new Error( `migrateJob: instance ${sourceJobId}/${sourceRole} is not running on ${sourceNodeId} ` + `(newest history: ${newestForSource?.event ?? "none"}) — only a running instance migrates` ); } const strategy = opts.strategy ?? "stopThenStart"; const migrationId = uuid.v4(); const targetRole = migrationTargetRole(slot, migrationId); const now = this.deps.clock.now(); // Phase F: a handover proposes FIRST. Only the handover strategy has // a job socket to ask down, and only it hands state over; the other // two still begin at `placing`. const first: MigrationPhase = strategy === "handover" ? "proposing" : "placing"; const state: MigrationState = { migrationId, sourceJobId, sourceRole, sourceNodeId, strategy, targetRole, startedAt: now, phase: first, phaseHistory: [{ phase: first, at: now }], pastPointOfNoReturn: false, }; this.migrationStates.set(migrationId, state); this.deps.bindings.claimMigration(sourceJobId, migrationId); this.deps.settings.onMigrationPhase?.(state); debuglog("Migration started", { migrationId, sourceJobId, sourceRole, sourceNodeId, strategy, targetRole, phase: first, requestedTarget: opts.targetNodeId, }); if (state.phase === "proposing") { // Arm the propose deadline by hand: advanceMigration is what // normally arms one, and the migration was BORN in this phase. this.armMigrationPhaseDeadline(state); const answer = await this.propose(state); if (answer.kind !== "accepted") { // The source is told to unlock even when it was the one that // refused — the abort is what clears its lock either way, and it // costs one message on a path that has already given up. const reason = answer.kind === "refused" ? `source refused the handover: ${answer.reason}` : answer.reason; // A refusal reached here only if the migration is still in flight; // performAbort is a no-op on one already marked aborted. await this.performAbort(state, reason, true); throw new Error(`migrateJob: migration ${migrationId} aborted: ${reason}`); } state.initialState = answer.payload; debuglog("Migration: source accepted the handover — placing target", { migrationId, stateBytes: answer.payload.length, }); this.advanceMigration(state, "placing"); } // The source node is never a candidate; the nodes recovery would // skip (unhealthy, recently failed) are not either. const excludeNodes = this.deps.currentExcludeSet(); excludeNodes.add(sourceNodeId); const outcome = await this.deps.placeAndProvision(sourceJobId, jh.job.tags ?? {}, undefined, { migrationId, targetRole, runtimeRole: slot, deferStart: strategy === "stopThenStart", ...(opts.targetNodeId !== undefined ? { targetNode: opts.targetNodeId } : {}), excludeNodes, ...(opts.overrideReservation ? { overrideReservation: true } : {}), ...(strategy === "handover" ? { extraCapabilities: [{ name: JOB_SOCKET_CAPABILITY, count: 1 }] } : {}), }); // A dispatch failure inside placeAndProvision aborts the migration // itself; a placement failure leaves that to us. if (state.phase === "aborted") { throw new Error(`migrateJob: migration ${migrationId} aborted: ${state.abortReason ?? outcome.kind}`); } if (outcome.kind !== "bound") { // Placement's own reason is generic (`noCapacityInAnyPool`); when a // handover had no node to go to because nothing else advertises the job // socket, say so — that is the operator's fix (`--job-socket` on a // worker), not capacity. const why = strategy === "handover" && !this.deps.anyOtherNodeAdvertises(JOB_SOCKET_CAPABILITY, sourceNodeId) ? `no other node advertises ${JOB_SOCKET_CAPABILITY}, which the handover strategy needs (the worker's job socket); ` : ""; // A pinned target gets its own verdict: `noCapacityInAnyPool` is what // placement says when its one candidate was filtered out, and the // operator needs to know WHY that node (fix-list 17). const pinned = opts.targetNodeId !== undefined && outcome.targetReason !== undefined ? `; the named target ${opts.targetNodeId} was rejected: ${outcome.targetReason}` : ""; const reason = `placement failed: ${why}${outcome.reason}${pinned}`; // A handover's source has ANSWERED the propose by now, and answering // locked it: the worker refuses every later migration of that instance // until an abort clears the lock ("instance is already in migration … // (source, stateReported)" — observed on norsk-e2e 2026-09-14 after a // placement failure on a mistyped target id left the source locked // for good). So the abort goes on the wire, exactly as it does for a // refusal; there is no target to stop yet, and stopThenStart's source // was never asked anything, so nothing is sent for it. await this.performAbort(state, reason, this.migrationTouchedWire(state)); throw new Error(`migrateJob: migration ${migrationId} aborted: ${reason}`); } state.targetNodeId = outcome.nodeId; this.advanceMigration(state, "launching"); return migrationId; } /** * Abort an in-flight migration cleanly: MigrationAbort to both keys, * StopJob for the target, its reservation released, the source * untouched. No-op (with a debug log) for an unknown migration id or * one already done/aborted. * * Rejects once the migration is past the point of no return — the * source has already been told to stop, so there is nothing to roll * back to; the saga will complete into the target, or fall into * ordinary recovery if the target is lost. * * @public */ async abortMigration( migrationId: MigrationId, reason: string ): Promise { const state = this.migrationStates.get(migrationId); if (!state) { debuglog("abortMigration: unknown migrationId", { migrationId }); return; } if (state.phase === "done" || state.phase === "aborted") return; if (state.pastPointOfNoReturn) { throw new Error( `abortMigration: migration ${migrationId} is past the point of no return ` + `(phase ${state.phase}): the source has been told to stop, so the migration ` + `can only fail forward — it will complete into the target, or fall into ` + `ordinary recovery if the target is lost` ); } await this.performAbort(state, reason, this.migrationTouchedWire(state)); } /** All known migrations (active + done + aborted). @public */ migrations(): MigrationState[] { return Array.from(this.migrationStates.values()); } /** Look up a single migration by id. @public */ migrationById(migrationId: MigrationId): MigrationState | undefined { return this.migrationStates.get(migrationId); } /** Migration active for the given source jobId, if any. @public */ migrationForJob(jobId: JobId): MigrationState | undefined { const id = this.deps.bindings.migrationOf(jobId)?.migrationId; return id ? this.migrationStates.get(id) : undefined; } /** * Forget a TERMINAL migration — `migrations()` keeps every migration * this AutoManager has ever run, which is an unbounded list an operator * has no other way to shorten. * * Refuses an in-flight one: dropping the record would leave the saga * running with nothing watching it. Abort it first (and, past the point * of no return, wait — it fails forward to `done`). The refusal is a * legitimate state, not a fault; the caller turns it into a 409, exactly * as it does `abortMigration`'s point-of-no-return refusal. * * Returns false for an id that is not (or is no longer) known, so a * double-clear is a no-op rather than an error. * * @public */ forgetMigration(migrationId: MigrationId): boolean { const state = this.migrationStates.get(migrationId); if (!state) { debuglog("forgetMigration: unknown migrationId", { migrationId }); return false; } if (state.phase !== "done" && state.phase !== "aborted") { throw new Error( `forgetMigration: migration ${migrationId} is still in flight (phase ${state.phase}): ` + `abort it first — a migration that is still running cannot be forgotten` ); } this.migrationStates.delete(migrationId); // markMigrationDone / markMigrationAborted already release the claim, // but never leave one pointing at a state that no longer exists: // migrationForJob would then answer undefined while migrateJob still // refused the job as "already migrating". this.deps.bindings.releaseMigration(state.sourceJobId, migrationId); this.migrationPhaseTimers.get(migrationId)?.cancel(); this.migrationPhaseTimers.delete(migrationId); this.migrationProposals.delete(migrationId); return true; } /** * Forget every terminal migration, leaving the in-flight ones alone. * Returns the ids forgotten. @public */ forgetTerminalMigrations(): MigrationId[] { const terminal = Array.from(this.migrationStates.values()) .filter((s) => s.phase === "done" || s.phase === "aborted") .map((s) => s.migrationId); for (const id of terminal) this.forgetMigration(id); return terminal; } inFlightMigrations(): MigrationState[] { const out: MigrationState[] = []; for (const id of this.deps.bindings.inFlightMigrationIds()) { const s = this.migrationStates.get(id); if (s) out.push(s); } return out; } /** * The job's runtime SLOT ("primary" / "backup"), from its replicaIndex * tag. What the job is told in NORSK_RUNTIME_ROLE, and the instance * key of a job that has never migrated. Undefined for un-managed jobs. */ slotForJob(jobId: JobId): Role | undefined { const replicaIndexStr = this.deps.jobOf(jobId)?.job.tags?.["replicaIndex"]; if (replicaIndexStr === undefined) return undefined; const replicaIndex = parseInt(replicaIndexStr, 10); if (!Number.isFinite(replicaIndex)) return undefined; return roleFromReplicaIndex(replicaIndex); } /** * The instance KEY of the job's current instance: the key its last * completed migration left it under, else its slot. This is the role * every RPC that addresses the RUNNING instance must use. */ roleForJob(jobId: JobId): Role { return this.deps.bindings.roleOf(jobId) ?? this.slotForJob(jobId) ?? roleFromReplicaIndex(0); } private advanceMigration(state: MigrationState, phase: MigrationPhase): void { if (state.phase === phase) return; const now = this.deps.clock.now(); state.phase = phase; state.phaseHistory.push({ phase, at: now }); this.deps.settings.onMigrationPhase?.(state); debuglog("Migration phase", { migrationId: state.migrationId, phase }); this.armMigrationPhaseDeadline(state); } /** * Re-arm the per-phase timer for the migration. Expiry before the * point of no return aborts; `stoppingSource` expiry proceeds as if * the source had reported stopped (it was told to stop — waiting * longer helps nobody); `startingTarget` expiry fails forward. */ private armMigrationPhaseDeadline(state: MigrationState): void { // Always cancel any prior timer; terminal phases don't re-arm. this.migrationPhaseTimers.get(state.migrationId)?.cancel(); this.migrationPhaseTimers.delete(state.migrationId); if (state.phase === "done" || state.phase === "aborted") return; const deadlineMs = this.cfgMigrationPhaseDeadline(state.phase); if (deadlineMs === undefined) return; const id = state.migrationId; const phase = state.phase; const timer = this.deps.clock.setTimeout(() => { this.migrationPhaseTimers.delete(id); const current = this.migrationStates.get(id); // Only fire if the migration is still in the phase we armed for. if (!current || current.phase !== phase) return; const reason = `phase timeout: ${phase}`; debuglog("Migration phase deadline fired", { migrationId: id, phase, deadlineMs }); switch (phase) { case "stoppingSource": this.deps.settings.onError?.( new Error( `migration ${id}: no JobStopped from source ${current.sourceJobId}/${current.sourceRole} ` + `within ${deadlineMs}ms of StopJob — proceeding as if stopped` ) ); void this.onSourceStopped(current).catch((err) => this.deps.settings.onError?.(err)); break; case "startingTarget": void this.failForward(current, reason).catch((err) => this.deps.settings.onError?.(err)); break; default: void this.performAbort(current, reason, this.migrationTouchedWire(current)).catch((err) => this.deps.settings.onError?.(err) ); } }, deadlineMs); this.migrationPhaseTimers.set(state.migrationId, timer); } private cfgMigrationPhaseDeadline(phase: MigrationPhase): number | undefined { const m = this.deps.settings.migration; switch (phase) { case "proposing": return m?.proposeDeadlineMs ?? DEFAULT_MIGRATION_PROPOSE_DEADLINE_MS; case "launching": return m?.launchingDeadlineMs ?? DEFAULT_MIGRATION_LAUNCHING_DEADLINE_MS; // `settingStateOnTarget` IS the wait for MigrationTargetReady (the // fold), so it shares that deadline rather than inventing one. case "settingStateOnTarget": case "awaitingTargetReady": return m?.awaitingTargetReadyDeadlineMs ?? DEFAULT_MIGRATION_TARGET_READY_DEADLINE_MS; case "awaitingSourceStop": return m?.awaitingSourceStopDeadlineMs ?? DEFAULT_MIGRATION_SOURCE_STOP_DEADLINE_MS; case "applying": return m?.applyingDeadlineMs ?? DEFAULT_MIGRATION_APPLYING_DEADLINE_MS; case "stoppingSource": return m?.stoppingSourceDeadlineMs ?? DEFAULT_MIGRATION_STOPPING_SOURCE_DEADLINE_MS; case "startingTarget": return m?.startingTargetDeadlineMs ?? DEFAULT_MIGRATION_STARTING_TARGET_DEADLINE_MS; case "placing": // synchronous — placement returns before the phase can be observed case "done": case "aborted": return undefined; } } private markMigrationDone(state: MigrationState): void { this.advanceMigration(state, "done"); // The claim went with completeMigration's rebind; release it here too // for a `done` reached any other way. this.deps.bindings.releaseMigration(state.sourceJobId, state.migrationId); this.deps.settings.onMigrationCompleted?.(state.migrationId); this.deps.onMigrationEnded(state); } markMigrationAborted(state: MigrationState, reason: string): void { state.abortReason = reason; // A migration abandoned while `proposing` — a phase deadline, a // worker-reported abort, an operator abort — must release the parked // migrateJob, or its caller waits forever for a target nobody will // place. this.settleProposal(state.migrationId, { kind: "abandoned", reason }); this.deps.bindings.releaseMigration(state.sourceJobId, state.migrationId); this.advanceMigration(state, "aborted"); this.deps.settings.onMigrationAborted?.(state.migrationId, reason); this.deps.onMigrationEnded(state); } /** * Has anything about this migration been put on the wire yet — i.e. is * there anybody to tell when it is abandoned? For a HANDOVER the answer * is yes from the very first phase: `proposing` asks the source for its * state, and a source that answered is LOCKED on it until a cutover or * an abort. The abort sent in `proposing` / `placing` / `launching` is * therefore the source's unlock, and skipping it strands a live * instance holding state nobody will ever collect. For the other two * strategies nothing reaches a worker until placement has dispatched * the target's StartJob, so `placing` really is silent. */ private migrationTouchedWire(state: MigrationState): boolean { return state.strategy === "handover" || state.phase !== "placing"; } /** * Phase F, the propose. Ask the SOURCE what it would hand over and park * until it answers, the phase deadline gives up, or the migration is * aborted from anywhere else. Nothing has been placed at this point and * nothing will be unless the answer is `accepted`. */ private async propose(state: MigrationState): Promise { const answer = new Promise((resolve) => { this.migrationProposals.set(state.migrationId, resolve); }); try { await this.deps.norsk.notifyMigrationGetState( { jobId: state.sourceJobId, role: state.sourceRole }, state.migrationId ); } catch (err) { const msg = err instanceof Error ? err.message : String(err); this.deps.settings.onError?.( new Error( `migration ${state.migrationId}: notifyMigrationGetState for source ` + `${state.sourceJobId}/${state.sourceRole} failed: ${msg}` ) ); this.settleProposal(state.migrationId, { kind: "abandoned", reason: `notifyMigrationGetState failed: ${msg}`, }); } return answer; } /** Hand the parked `migrateJob` its answer. Idempotent: the first * settler wins, so a refusal racing the phase deadline is harmless. */ private settleProposal(id: MigrationId, answer: ProposeAnswer): void { const resolve = this.migrationProposals.get(id); if (!resolve) return; this.migrationProposals.delete(id); resolve(answer); } /** MigrationAbort to both keys, via Manager. Best effort. */ private async dispatchMigrationAborts( state: MigrationState, reason: string ): Promise { await this.deps.norsk .abortMigration(state.sourceJobId, state.sourceRole, state.targetRole, reason, state.migrationId) .catch((err) => this.deps.settings.onError?.(err)); } /** * The clean rollback (pre point-of-no-return only): abort to both * keys and StopJob for the target when anything was put on the wire * (`notifyWire`), the target's reservation released, the migration * marked aborted. The source is never touched — it was never told to * stop. Local state changes first, so a caller iterating bindings * sees them gone even if the RPCs are slow. */ private async performAbort( state: MigrationState, reason: string, notifyWire: boolean ): Promise { if (state.phase === "done" || state.phase === "aborted") return; const { sourceJobId: jobId, targetRole, targetNodeId } = state; this.markMigrationAborted(state, reason); if (!notifyWire) return; await this.dispatchMigrationAborts(state, reason); // The target is never left running after an abort (it wears a key // nobody will address again). Best effort: a target that never got // dispatched, or already died, answers with an error we swallow. if (targetNodeId !== undefined) { await this.deps.norsk .stopJob(jobId, targetRole, targetNodeId) .catch((err) => this.deps.settings.onError?.(err)); } } /** * StartJob for the migration target on `nodeId`, carrying the * migration fields. On failure the migration is treated as having * lost its target (abort before the point of no return; the only * caller after it is `startPreparedJob`, which has its own path). */ async dispatchMigrationTarget(state: MigrationState, nodeId: NodeId): Promise { try { await this.deps.norsk.startJob(state.sourceJobId, state.targetRole, nodeId, { runtimeRole: this.slotForJob(state.sourceJobId) ?? state.sourceRole, deferStart: state.strategy === "stopThenStart", migrationId: state.migrationId, // Phase F: the source's state rides in on the target's StartJob, // so the target has it at boot and the worker can deliver it as // `initial-state` right after hello-ack — no extra round trip. ...(state.initialState !== undefined ? { migrationInitialState: state.initialState } : {}), }); return true; } catch (err) { const msg = err instanceof Error ? err.message : String(err); this.deps.onDispatchFailed(); this.deps.settings.onError?.( new Error( `migration ${state.migrationId}: startJob dispatch failed for target ` + `${state.sourceJobId}/${state.targetRole} on node ${nodeId}: ${msg}` ) ); await this.handleTargetLost(state, `target startJob dispatch failed: ${msg}`); return false; } } /** The target's containers exist and are running (or, for * stopThenStart, have been told to start): what "fail forward into * the target" needs to be true before it is preferable to recovery. */ private targetIsRunning(state: MigrationState): boolean { if (state.strategy === "stopThenStart") return state.phase === "startingTarget"; return ( state.phase === "settingStateOnTarget" || state.phase === "awaitingTargetReady" || state.phase === "awaitingSourceStop" || state.phase === "applying" || state.phase === "stoppingSource" ); } /** The target is gone (failed, rejected, its node dying, its provision * abandoned): a clean abort while the source is untouched; past the * point of no return, fail forward into ordinary recovery. */ async handleTargetLost(state: MigrationState, reason: string): Promise { if (state.phase === "done" || state.phase === "aborted") return; if (state.pastPointOfNoReturn) await this.failForward(state, reason); else await this.performAbort(state, reason, this.migrationTouchedWire(state)); } /** The source is gone (failed, its node dying) mid-migration. If the * target is already running, complete into it — the job carries on * there, which is exactly what a migration is for. Otherwise abort; * the caller then runs the ordinary recovery re-place for the job * (the job's binding still names the dead source, so recovery sees * it as needing a node). */ async handleSourceLost(state: MigrationState, reason: string): Promise { if (state.phase === "done" || state.phase === "aborted") return; if (this.targetIsRunning(state)) { debuglog("Migration: source lost with target running — completing into the target", { migrationId: state.migrationId, reason, }); this.completeMigration(state); return; } await this.performAbort(state, reason, this.migrationTouchedWire(state)); } /** * StopJob(source, migrationId) — the point of no return. If the RPC * itself fails there is no way back either (the Manager may or may * not have acted on it), so proceed as if the source had stopped: * for stopThenStart that starts the prepared target regardless. */ private async beginStoppingSource(state: MigrationState): Promise { state.pastPointOfNoReturn = true; this.advanceMigration(state, "stoppingSource"); try { await this.deps.norsk.stopJob(state.sourceJobId, state.sourceRole, state.sourceNodeId, { migrationId: state.migrationId, }); } catch (err) { const msg = err instanceof Error ? err.message : String(err); this.deps.settings.onError?.( new Error( `migration ${state.migrationId}: stopJob(source ${state.sourceJobId}/${state.sourceRole} ` + `on ${state.sourceNodeId}) failed past the point of no return: ${msg} — proceeding as if stopped` ) ); await this.onSourceStopped(state); } } /** The source's JobStopped landed (or its deadline stood in for it). */ private async onSourceStopped(state: MigrationState): Promise { if (state.phase !== "stoppingSource") return; if (state.strategy !== "stopThenStart") { this.completeMigration(state); return; } this.advanceMigration(state, "startingTarget"); const targetNodeId = state.targetNodeId; if (targetNodeId === undefined) { await this.failForward(state, "no target node recorded at startingTarget"); return; } try { await this.deps.norsk.startPreparedJob(state.sourceJobId, state.targetRole, targetNodeId); } catch (err) { const msg = err instanceof Error ? err.message : String(err); this.deps.settings.onError?.( new Error( `migration ${state.migrationId}: startPreparedJob for ${state.sourceJobId}/${state.targetRole} ` + `on ${targetNodeId} failed: ${msg}` ) ); await this.failForward(state, `startPreparedJob failed: ${msg}`); } } /** * `done`: the job now lives on the target. Its binding rebinds to it, * the target's separate binding folds away, the instance key becomes * the one the running instance wears from here on, and the placement * of record becomes the target's own decision (`JobBindings. * completeMigration`). The source's reservation was released by the * rebind (the derived reservation reads the bindings); its JobNodeMap * entry went with its JobStopped. */ private completeMigration(state: MigrationState): void { if (state.phase === "done" || state.phase === "aborted") return; const jobId = state.sourceJobId; this.deps.bindings.completeMigration(jobId, state.migrationId, state.targetRole); debuglog("Migration done", { migrationId: state.migrationId, jobId, from: state.sourceNodeId, to: state.targetNodeId, activeRole: state.targetRole, }); this.markMigrationDone(state); } /** * Past the point of no return and the target is lost: the source is * gone or going, so the job has nowhere to run. Mark the migration * aborted, drop both bindings, and hand the job to ordinary recovery * — the same re-place a node death would trigger, ceiling and policy * included — under its slot key again (a fresh instance is not the * migrated one). */ private async failForward(state: MigrationState, reason: string): Promise { if (state.phase === "done" || state.phase === "aborted") return; const jobId = state.sourceJobId; debuglog("Migration failing forward into ordinary recovery", { migrationId: state.migrationId, jobId, phase: state.phase, reason, }); const { targetRole, targetNodeId } = state; this.markMigrationAborted(state, reason); // Whatever remains of the target must not linger under a key nobody // will address again. Best effort. if (targetNodeId !== undefined) { void this.deps.norsk .stopJob(jobId, targetRole, targetNodeId) .catch((err) => debuglog("failForward: target stopJob failed (ignored)", { err: String(err) })); } this.deps.bindings.unbind(jobId); const jh = this.deps.jobOf(jobId); const req = jh?.job.autoPlacement?.requirements; if (!jh || !req || jh.job.terminalState) return; if (jh.job.state !== "pre" && jh.job.state !== "active") return; await this.deps.placeAndProvision(jobId, jh.job.tags ?? {}, { excludeNodes: this.deps.currentExcludeSet(), }); } /** * The saga's event intake for job history: every transition that is * not a handshake message arrives as the newest entry of a * `jobUpdated`, keyed by the entry's role — the instance key — which * is what tells the target's `running` from the source's. Returns * true when the entry was about a migrating instance and has been * fully handled, so `handleJobChanged` must NOT also run ordinary * failure recovery for it (a target failure is the saga's to handle; * a source failure that completed into the target needs no recovery). * Returns false to let the ordinary path run — including after a * source failure that aborted the migration, which is then recovered * like any other runtime failure. */ async advanceMigrationOnJobHistory(jh: JobWithHistory): Promise { const newest = jh.history[0]; if (!newest) return false; const state = this.migrationForJob(jh.job.jobId); if (!state) return false; if (!("role" in newest)) return false; // created / updated / completed / rejected carry no key const role = newest.role; const isTarget = role === state.targetRole; const isSource = role === state.sourceRole; if (!isTarget && !isSource) return false; const id = state.migrationId; switch (newest.event) { case "prepared": if (isTarget && state.phase === "launching") { if (state.strategy !== "stopThenStart") { debuglog("Migration: target reported prepared but strategy did not defer start", { migrationId: id, strategy: state.strategy, }); return true; } await this.beginStoppingSource(state); } return true; case "running": if (!isTarget) return false; if (state.phase === "launching") { switch (state.strategy) { case "overlap": await this.beginStoppingSource(state); break; case "handover": // Phase F. The source's state went out on this target's // StartJob, so by the time it is running the worker has // delivered it as `initial-state`. We sit here until the // target reports ready, which is also the ack that the // state was applied. this.advanceMigration(state, "settingStateOnTarget"); break; case "stopThenStart": // Expected `prepared`; the worker started it anyway. The // two instances now overlap — proceed as overlap would, // rather than wait for a `prepared` that will never come. debuglog("Migration: stopThenStart target reported running without prepared — proceeding as overlap", { migrationId: id, }); await this.beginStoppingSource(state); break; } } else if (state.phase === "startingTarget") { this.completeMigration(state); } return true; case "stopped": if (isSource && state.phase === "stoppingSource") { await this.onSourceStopped(state); return true; } return false; case "stopping": { if (newest.reason !== "jobFailed") return false; const detail = newest.additionalReason ?? "worker reported failure"; if (isTarget) { await this.handleTargetLost(state, `target failed: ${detail}`); return true; } // Source failed mid-migration. if (this.targetIsRunning(state)) { debuglog("Migration: source failed with target running — completing into the target", { migrationId: id, }); this.completeMigration(state); return true; } await this.performAbort(state, `source failed: ${detail}`, this.migrationTouchedWire(state)); return false; // ordinary recovery re-places the job } default: return false; } } // ---- handshake events (handover strategy) ------------------------ /** Look up the in-flight migration a handshake event belongs to, * dropping events for unknown jobs and for a migrationId other than * the one in flight (a predecessor's stragglers). */ private migrationForHandshake( what: string, jobKey: CommonPB.ActiveJobKey | undefined, migrationId: string ): MigrationState | undefined { const jobId = jobKey?.jobId?.jobId; if (!jobId) return undefined; const state = this.migrationForJob(jobId); if (!state) { debuglog(`${what} for a job with no migration in flight — dropped`, { jobId, migrationId }); return undefined; } if (migrationId !== state.migrationId) { debuglog(`${what} carries a stale migrationId — dropped`, { jobId, eventMigrationId: migrationId, inFlight: state.migrationId, }); return undefined; } return state; } /** * Phase F — the source's answer to the propose. Either its migratable * state (possibly empty: a job with nothing to hand over still * accepts) or a refusal, which ends the migration with no target ever * placed. The refusal is read off the ENVELOPE, so the coordinator * never has to open the opaque blob to know it was turned down. */ handleMigrationState(raw: ManagerPB.MigrationStateEvent): void { const state = this.migrationForHandshake("MigrationState", raw.jobKey, raw.migrationId); if (!state) return; if (state.phase !== "proposing") { debuglog("MigrationState in unexpected phase", { migrationId: state.migrationId, phase: state.phase, }); return; } if (raw.refused.length > 0) { this.settleProposal(state.migrationId, { kind: "refused", reason: raw.refused }); return; } this.settleProposal(state.migrationId, { kind: "accepted", payload: raw.payload }); } async handleMigrationTargetReady( raw: ManagerPB.MigrationTargetReadyEvent ): Promise { const state = this.migrationForHandshake("MigrationTargetReady", raw.jobKey, raw.migrationId); if (!state) return; if (state.strategy !== "handover") { debuglog("MigrationTargetReady under a non-handover strategy — ignored", { migrationId: state.migrationId, strategy: state.strategy, }); return; } // `launching` is accepted too: the target's ready can race its own // JobRunning history entry through the Manager. if ( state.phase !== "launching" && state.phase !== "settingStateOnTarget" && state.phase !== "awaitingTargetReady" ) { debuglog("MigrationTargetReady in unexpected phase", { migrationId: state.migrationId, phase: state.phase, }); return; } state.targetPayload = raw.payload; // A worker that reports MigrationInitialStateDelivered has already moved // this to awaitingTargetReady; one that predates the event has not, and // for it `ready` is still the ack that the state was applied. Recorded // either way, so the phase bar never skips a step that really completed. this.advanceMigration(state, "awaitingTargetReady"); this.advanceMigration(state, "awaitingSourceStop"); try { await this.deps.norsk.notifyMigrationCutoverReady( { jobId: state.sourceJobId, role: state.sourceRole }, state.migrationId, raw.payload ); } catch (err) { this.deps.settings.onError?.(err); await this.performAbort(state, `notifyMigrationCutoverReady failed: ${String(err)}`, true); } } async handleMigrationSourceStoppedOutput( raw: ManagerPB.MigrationSourceStoppedOutputEvent ): Promise { const state = this.migrationForHandshake( "MigrationSourceStoppedOutput", raw.jobKey, raw.migrationId ); if (!state) return; if (state.phase !== "awaitingSourceStop") { debuglog("MigrationSourceStoppedOutput in unexpected phase", { migrationId: state.migrationId, phase: state.phase, }); return; } state.sourcePayload = raw.payload; this.advanceMigration(state, "applying"); try { await this.deps.norsk.notifyMigrationCutoverApply( { jobId: state.sourceJobId, role: state.targetRole }, state.migrationId, raw.payload ); } catch (err) { this.deps.settings.onError?.(err); // Still before the point of no return: the abort to the source // means "resume output". await this.performAbort(state, `notifyMigrationCutoverApply failed: ${String(err)}`, true); } } /** The target has applied the handover and begun output — only now is * the source stopped. (Marking done on the apply RPC returning was * hole 4: a target that failed to apply left no source to go back to.) */ async handleMigrationTargetApplied( raw: ManagerPB.MigrationTargetAppliedEvent ): Promise { const state = this.migrationForHandshake("MigrationTargetApplied", raw.jobKey, raw.migrationId); if (!state) return; if (state.phase !== "applying") { debuglog("MigrationTargetApplied in unexpected phase", { migrationId: state.migrationId, phase: state.phase, }); return; } const summary = readApplySummary(raw.payload); if (summary) { state.applySummary = summary; if (summary.failed.length > 0) { // Past the point of no return: the source stops regardless. Say so // where an operator will look, since `done` alone would not. this.deps.settings.onError?.( new Error( `migration ${state.migrationId}: the target applied INCOMPLETELY — applied [${summary.applied.join(", ")}], failed ${summary.failed .map((f) => `${f.id} (${f.reason})`) .join(", ")}; skew ${summary.skewMs} ms` ) ); } } await this.beginStoppingSource(state); } /** Task 90, fix-list 27: the worker has handed the source's state to the * target. `settingStateOnTarget` ends here; what follows is the target's * own ready round. Idempotent against a ready that raced it. */ handleMigrationInitialStateDelivered( raw: ManagerPB.MigrationInitialStateDeliveredEvent ): void { const state = this.migrationForHandshake("MigrationInitialStateDelivered", raw.jobKey, raw.migrationId); if (!state) return; if (state.strategy !== "handover") return; // `launching` too: the delivery can race the target's JobRunning through // the Manager, exactly as `ready` can. if (state.phase !== "launching" && state.phase !== "settingStateOnTarget") { debuglog("MigrationInitialStateDelivered in unexpected phase", { migrationId: state.migrationId, phase: state.phase, }); return; } if (state.phase === "launching") this.advanceMigration(state, "settingStateOnTarget"); this.advanceMigration(state, "awaitingTargetReady"); } async handleMigrationAborted(raw: ManagerPB.MigrationAbortedEvent): Promise { const state = this.migrationForHandshake("MigrationAborted", raw.jobKey, raw.migrationId); if (!state) return; const role = raw.jobKey?.role; debuglog("Migration aborted by worker", { migrationId: state.migrationId, role, reason: raw.reason, }); if (!state.pastPointOfNoReturn) { await this.performAbort(state, raw.reason, true); return; } if (role === state.targetRole) { await this.failForward(state, `target aborted: ${raw.reason}`); return; } // The source backing out after it has been told to stop changes // nothing: it is stopping regardless. debuglog("Migration: source-side abort past the point of no return — ignored", { migrationId: state.migrationId, reason: raw.reason, }); } /** * End every in-flight migration for `reason`: the clean rollback * (abort on the wire, target stopped) before the point of no return, * marked lost after it. Returns the wire work still settling, so the * caller can hold the channel open for it. Local state is final on * return either way. */ abortInFlightMigrations(reason: string): Promise[] { const settling: Promise[] = []; for (const state of this.inFlightMigrations()) { if (state.pastPointOfNoReturn) { this.deps.settings.onError?.( new Error( `migration ${state.migrationId} for job ${state.sourceJobId} was lost to a close past ` + `the point of no return (phase ${state.phase}); the source is stopping or stopped, so the ` + `target is left to carry the job and the next instance rebuilds its placement from history` ) ); this.markMigrationAborted(state, reason); } else { settling.push( this.performAbort(state, reason, this.migrationTouchedWire(state)).catch((err) => this.deps.settings.onError?.(err) ) ); } } return settling; } } export function roleFromReplicaIndex(replicaIndex: number): Role { // Preserve the existing primary/backup convention for replicas 0 and 1 // so consumers that grep on role keep working. Higher replica counts // get a numeric form. if (replicaIndex === 0) return "primary"; if (replicaIndex === 1) return "backup"; return `replica-${replicaIndex}`; } /** * The newest history entry about a given instance key (running, * provisioned, prepared, stopping, stopped all carry the key; created / * updated / completed / rejected do not and are skipped). Undefined when * the key has no history — it never existed, or the history has been * trimmed. */ export function newestHistoryEntryForRole( jh: JobWithHistory, role: Role ): | ManagerSdk.JobHistoryJobRunning | ManagerSdk.JobHistoryJobProvisioned | ManagerSdk.JobHistoryJobPrepared | ManagerSdk.JobHistoryJobStopping | ManagerSdk.JobHistoryJobStopped | undefined { for (const e of jh.history) { if ("role" in e && e.role === role) return e; } return undefined; }