From dbde7db91617c55d0679517f643be9c53818e39c Mon Sep 17 00:00:00 2001 From: root Date: Sat, 11 Jul 2026 01:01:46 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=85=B3=E9=97=AD=20runner=20retention?= =?UTF-8?q?=20=E5=88=A0=E9=99=A4=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/runner-retention.ts | 238 ++++++++++++++++-- .../22-runner-retention-stale-pending.ts | 101 +++++++- .../fake-kubectl-retention-stale-pending.mjs | 31 ++- 3 files changed, 340 insertions(+), 30 deletions(-) diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index c44e220..134e941 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -50,6 +50,8 @@ type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending"; interface RunnerPodObservation extends JsonRecord { name: string; + uid: string | null; + resourceVersion: string | null; phase: string | null; createdAt: string | null; ageMs: number | null; @@ -57,6 +59,7 @@ interface RunnerPodObservation extends JsonRecord { podStartTime: string | null; ready: boolean; runnerContainerState: "waiting" | "running" | "terminated" | "unknown"; + runnerContainerLastState: "running" | "terminated" | "none"; runnerContainerStarted: boolean; runnerContainerReady: boolean; runnerContainerRestartCount: number; @@ -72,6 +75,8 @@ interface K8sObject { metadata?: { name?: string; namespace?: string; + uid?: string; + resourceVersion?: string; creationTimestamp?: string; labels?: Record; annotations?: Record; @@ -93,6 +98,8 @@ interface RunnerResourceEntry { resourceKind: "job" | "pod"; name: string; jobName: string; + resourceUid: string | null; + resourceVersion: string | null; runId: string | null; commandId: string | null; runnerId: string | null; @@ -105,6 +112,7 @@ interface RunnerResourceEntry { candidateKind: CandidateKind; classificationReason: string; lastActiveAtMs: number; + dbFactsFingerprint: string | null; } interface Snapshot { @@ -157,16 +165,37 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu let deletedRunnerPodCount = 0; let deletedAssociatedResourceCount = 0; for (const item of selected) { + await revalidateSelectedRunner(input.store, input.options, item); if (item.resourceKind === "job") { - await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "job", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal); + await deleteRunnerResource(input.options, item); deletedRunnerJobCount++; deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName); } else { - await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "pod", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal); + await deleteRunnerResource(input.options, item); deletedRunnerPodCount++; } } const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before; + if (after.liveRunnerCount > targetBeforeCreate) { + throw new AgentRunError("infra-failed", "runner retention did not make enough room for a new runner", { + httpStatus: 409, + details: { + reason: "runner-retention-postcondition-failed", + namespace: input.options.namespace, + maxRunners: input.options.maxRunners, + incomingRunnerCount, + targetBeforeCreate, + liveRunnerCountBefore: before.liveRunnerCount, + liveRunnerCountAfter: after.liveRunnerCount, + selectedRunnerCount: selected.length, + deletedRunnerJobCount, + deletedRunnerPodCount, + deletedAssociatedResourceCount, + selected: selected.map(resourceSummary), + valuesPrinted: false, + }, + }); + } return { enabled: true, namespace: input.options.namespace, @@ -287,6 +316,8 @@ async function jobEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe resourceKind: "job", name, jobName: name, + resourceUid: item.metadata?.uid ?? null, + resourceVersion: item.metadata?.resourceVersion ?? null, createdAtMs: objectCreatedAtMs(item), terminal: jobTerminal(item), podPhase: null, @@ -314,6 +345,8 @@ async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe resourceKind: "pod", name, jobName, + resourceUid: item.metadata?.uid ?? null, + resourceVersion: item.metadata?.resourceVersion ?? null, createdAtMs: objectCreatedAtMs(item), terminal: podTerminal(item), podPhase: stringPath(item, ["status", "phase"]), @@ -322,7 +355,7 @@ async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe }; } -type ActivityFields = Pick; +type ActivityFields = Pick; async function activityFor(store: AgentRunStore, input: { runId: string | null; @@ -335,41 +368,85 @@ async function activityFor(store: AgentRunStore, input: { podObservations: RunnerPodObservation[]; podIdentityMatch: boolean; }): Promise { - if (!input.runId || !input.commandId || !input.runnerId) return protectedActivity(input, "identity-facts-unavailable", input.createdAtMs); - if (!input.podIdentityMatch) return protectedActivity(input, "pod-identity-mismatch", input.createdAtMs); + if (!input.runId || !input.commandId || !input.runnerId) return protectedActivity(input, "identity-facts-unavailable", input.createdAtMs, null); + if (!input.podIdentityMatch) return protectedActivity(input, "pod-identity-mismatch", input.createdAtMs, null); let run: RunRecord; let command: CommandRecord; try { [run, command] = await Promise.all([store.getRun(input.runId), store.getCommand(input.commandId)]); } catch { - return protectedActivity(input, "db-facts-unavailable", input.createdAtMs); + return protectedActivity(input, "db-facts-unavailable", input.createdAtMs, null); } const lastActiveAtMs = Math.max(input.createdAtMs, isoMs(run.updatedAt), isoMs(command.updatedAt), isoMs(run.leaseExpiresAt)); - if (command.runId !== run.id) return protectedActivity(input, "db-identity-mismatch", lastActiveAtMs); - const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs)); - if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs); - const activePodReason = activePodProtectionReason(input.podObservations); - if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs); - if (isTerminalRunStatus(run.status)) return candidateActivity(input, "terminal", "terminal-run", lastActiveAtMs); - if (isTerminalCommandState(command.state)) return candidateActivity(input, input.terminal ? "terminal" : "idle", input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource", lastActiveAtMs); - if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs); - - const podBlockReason = stalePendingPodBlockReason(input.podObservations); - if (podBlockReason) return protectedActivity(input, podBlockReason, lastActiveAtMs); - + if (command.runId !== run.id) return protectedActivity(input, "db-identity-mismatch", lastActiveAtMs, null); let commands: CommandRecord[]; let events: import("../common/types.js").RunEvent[]; try { [commands, events] = await Promise.all([store.listCommands(run.id, 0, 100), store.listEvents(run.id, 0, 500)]); } catch { - return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs); + return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs, null); } + const dbFactsFingerprint = databaseFactsFingerprint(run, command, commands, events); + const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs)); + if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs, dbFactsFingerprint); + const activePodReason = activePodProtectionReason(input.podObservations); + if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs, dbFactsFingerprint); + if (isTerminalRunStatus(run.status)) return candidateActivity(input, "terminal", "terminal-run", lastActiveAtMs, dbFactsFingerprint); + if (isTerminalCommandState(command.state)) return candidateActivity(input, input.terminal ? "terminal" : "idle", input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource", lastActiveAtMs, dbFactsFingerprint); + if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs, dbFactsFingerprint); + + const podBlockReason = stalePendingPodBlockReason(input.podObservations); + if (podBlockReason) return protectedActivity(input, podBlockReason, lastActiveAtMs, dbFactsFingerprint); if (!hasNoBusinessProgress(run, command, commands, events, input.stalePendingMaxAgeMs)) { - return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs); + return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs, dbFactsFingerprint); } - return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs); + return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint); +} + +async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, selected: RunnerResourceEntry): Promise { + const snapshot = await runnerSnapshot(store, options); + const resources = selected.resourceKind === "job" ? snapshot.jobs : snapshot.pods; + const current = resources.find((item) => item.name === selected.name) ?? null; + const selectedFingerprint = runnerSafetyFingerprint(selected); + const currentFingerprint = current ? runnerSafetyFingerprint(current) : null; + if (current && !current.protectedActive && currentFingerprint === selectedFingerprint) return; + + throw new AgentRunError("infra-failed", "runner retention selected resource changed before deletion", { + httpStatus: 409, + details: { + reason: "runner-retention-selected-runner-changed", + changeReason: current === null ? "resource-missing" : current.protectedActive ? "protected-active" : "safety-fingerprint-changed", + namespace: options.namespace, + resourceKind: selected.resourceKind, + name: selected.name, + selectedFingerprint, + currentFingerprint, + selected: resourceSummary(selected), + current: current ? resourceSummary(current) : null, + valuesPrinted: false, + }, + }); +} + +async function deleteRunnerResource(options: RunnerRetentionOptions, item: RunnerResourceEntry): Promise { + const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", item.resourceKind, item.name, "-n", options.namespace], options.signal); + if (result.code === 0) return; + throw new AgentRunError("infra-failed", `kubectl delete ${item.resourceKind} failed with code ${result.code}`, { + httpStatus: 502, + details: redactJson({ + reason: "runner-retention-delete-failed", + namespace: options.namespace, + resourceKind: item.resourceKind, + name: item.name, + exitCode: result.code, + signal: result.signal, + stderr: redactText(result.stderr.slice(-2000)), + stdout: redactText(result.stdout.slice(-1000)), + valuesPrinted: false, + }), + }); } async function kubectlGetList(options: RunnerRetentionOptions, resource: string, selector: string): Promise { @@ -430,6 +507,8 @@ function resourceSummary(item: RunnerResourceEntry): JsonRecord { resourceKind: item.resourceKind, name: item.name, jobName: item.jobName, + resourceUid: item.resourceUid, + resourceVersion: item.resourceVersion, runId: item.runId, commandId: item.commandId, runnerId: item.runnerId, @@ -510,10 +589,13 @@ function observePod(item: K8sObject, stalePendingMaxAgeMs: number, events: K8sOb ]; const runnerContainerRestartCount = typeof runner?.restartCount === "number" ? runner.restartCount : 0; const runnerLastState = jsonRecord(runner?.lastState); + const runnerContainerLastState: RunnerPodObservation["runnerContainerLastState"] = runnerLastState?.running ? "running" : runnerLastState?.terminated ? "terminated" : "none"; const runnerContainerStarted = runner?.started === true || runnerContainerRestartCount > 0 || runnerContainerState === "running" || runnerContainerState === "terminated" || Boolean(runnerLastState?.running) || Boolean(runnerLastState?.terminated); const runnerContainerReady = runner?.ready === true; return { name: objectName(item) ?? "unknown", + uid: item.metadata?.uid ?? null, + resourceVersion: item.metadata?.resourceVersion ?? null, phase, createdAt: createdAtMs > 0 ? new Date(createdAtMs).toISOString() : null, ageMs, @@ -521,6 +603,7 @@ function observePod(item: K8sObject, stalePendingMaxAgeMs: number, events: K8sOb podStartTime: boundedRedactedText(item.status?.startTime), ready, runnerContainerState, + runnerContainerLastState, runnerContainerStarted, runnerContainerReady, runnerContainerRestartCount, @@ -591,12 +674,117 @@ function hasNoBusinessProgress(run: RunRecord, command: CommandRecord, commands: return events.every((event) => event.payload.commandId === undefined || event.payload.commandId === command.id); } -function protectedActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, reason: string, lastActiveAtMs: number): ActivityFields { - return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, lastActiveAtMs }; +function protectedActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string | null): ActivityFields { + return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, lastActiveAtMs, dbFactsFingerprint }; } -function candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number): ActivityFields { - return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, lastActiveAtMs }; +function candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string): ActivityFields { + return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, lastActiveAtMs, dbFactsFingerprint }; +} + +function databaseFactsFingerprint(run: RunRecord, command: CommandRecord, commands: CommandRecord[], events: import("../common/types.js").RunEvent[]): string { + const commandFacts = (item: CommandRecord): JsonRecord => ({ + id: item.id, + runId: item.runId, + seq: item.seq, + type: item.type, + state: item.state, + payloadHash: item.payloadHash, + cancelEpoch: item.cancelEpoch, + cancelRequestId: item.cancelRequestId, + cancelRequestedAt: item.cancelRequestedAt, + cancelReason: item.cancelReason, + createdAt: item.createdAt, + updatedAt: item.updatedAt, + acknowledgedAt: item.acknowledgedAt, + dispatchIntentHash: stableHash(item.dispatchIntent ?? null), + valuesPrinted: false, + }); + return stableHash({ + run: { + id: run.id, + status: run.status, + terminalStatus: run.terminalStatus, + failureKind: run.failureKind, + failureMessageHash: run.failureMessage ? stableHash(run.failureMessage) : null, + cancelEpoch: run.cancelEpoch, + cancelRequestId: run.cancelRequestId, + cancelRequestedAt: run.cancelRequestedAt, + cancelReason: run.cancelReason, + createdAt: run.createdAt, + updatedAt: run.updatedAt, + claimedBy: run.claimedBy, + leaseExpiresAt: run.leaseExpiresAt, + valuesPrinted: false, + }, + command: commandFacts(command), + commands: [...commands].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map(commandFacts), + events: [...events].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map((event) => ({ + id: event.id, + runId: event.runId, + seq: event.seq, + type: event.type, + createdAt: event.createdAt, + payloadHash: stableHash(event.payload), + valuesPrinted: false, + })), + valuesPrinted: false, + }); +} + +function runnerSafetyFingerprint(item: RunnerResourceEntry): string { + return stableHash({ + schema: "agentrun-runner-retention-safety-v1", + resourceKind: item.resourceKind, + name: item.name, + jobName: item.jobName, + resourceUid: item.resourceUid, + resourceVersion: item.resourceVersion, + runId: item.runId, + commandId: item.commandId, + runnerId: item.runnerId, + createdAtMs: item.createdAtMs, + terminal: item.terminal, + podPhase: item.podPhase, + protectedActive: item.protectedActive, + protectedReason: item.protectedReason, + candidateKind: item.candidateKind, + classificationReason: item.classificationReason, + lastActiveAtMs: item.lastActiveAtMs, + dbFactsFingerprint: item.dbFactsFingerprint, + podObservations: [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name)).map(podSafetyFacts), + valuesPrinted: false, + }); +} + +function podSafetyFacts(item: RunnerPodObservation): JsonRecord { + const waiting = item.waiting.map((entry) => ({ + source: typeof entry.source === "string" ? entry.source : null, + name: typeof entry.name === "string" ? entry.name : null, + reason: typeof entry.reason === "string" ? entry.reason : null, + deterministic: entry.deterministic === true, + valuesPrinted: false, + })).sort((left, right) => JSON.stringify(left).localeCompare(JSON.stringify(right))); + return { + name: item.name, + uid: item.uid, + resourceVersion: item.resourceVersion, + phase: item.phase, + createdAt: item.createdAt, + stale: item.stale, + podStartTime: item.podStartTime, + ready: item.ready, + runnerContainerState: item.runnerContainerState, + runnerContainerLastState: item.runnerContainerLastState, + runnerContainerStarted: item.runnerContainerStarted, + runnerContainerReady: item.runnerContainerReady, + runnerContainerRestartCount: item.runnerContainerRestartCount, + waiting, + waitingEvidenceSources: [...item.waitingEvidenceSources].sort(), + eventFactsAvailable: item.eventFactsAvailable, + factsAvailable: item.factsAvailable, + valuesPrinted: false, + }; } function boundedResourceSummaries(items: RunnerResourceEntry[]): { items: JsonRecord[]; omittedCount: number } { diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index fd7dbe0..2ffbf60 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/src/selftest/cases/22-runner-retention-stale-pending.ts @@ -20,6 +20,20 @@ interface RetentionFact { events: RunEvent[]; } +interface RetentionFixture { + jobs: JsonRecord[]; + pods: JsonRecord[]; + events?: JsonRecord[]; + eventsAfterFirstRead?: JsonRecord[]; + racePods?: JsonRecord[]; + eventsError?: boolean; + trackEventReads?: boolean; + churnEventMetadata?: boolean; + raceAfterEvents?: boolean; + deleteErrorResources?: string[]; + retainDeletedJobs?: boolean; +} + class RetentionFactStore extends MemoryAgentRunStore { private readonly facts: RetentionFact[]; @@ -66,6 +80,9 @@ const selfTest: SelfTestCase = async (context) => { await assertTwentyStalePendingSelectsOnlyOldest({ fixturePath, statePath, kubectlCommand }); await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand }); await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand }); + await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertDeleteFailureFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand }); } finally { restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath); restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath); @@ -89,6 +106,9 @@ const selfTest: SelfTestCase = async (context) => { "pod-identity-mismatch-protected", "event-facts-unavailable-protected", "terminal-ledger-running-pod-protected", + "selected-runner-race-fails-closed", + "runner-delete-exit-code-checked", + "runner-delete-postcondition-checked", ], }; }; @@ -102,6 +122,8 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s jobs: facts.map((fact, index) => runnerJob(fact, index)), pods: facts.map((fact, index) => runnerPod(fact, index, "pending")), events: facts.map((fact, index) => failedMountEvent(fact, index)), + trackEventReads: true, + churnEventMetadata: true, }; await writeFixture(input, fixture); const store = new RetentionFactStore(facts); @@ -111,7 +133,7 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s assert.equal(withinLimit.liveRunnerCountBefore, 20); assert.equal(withinLimit.inactiveCandidateCount, 20); assert.equal(withinLimit.selectedRunnerCount, 0); - assert.equal((await readDeletionState(input.statePath)).length, 0); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); const cleanup = await enforceRunnerRetentionBeforeCreate({ store, options, incomingRunnerCount: 1 }); assert.equal(cleanup.liveRunnerCountBefore, 20); @@ -141,6 +163,7 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s assert.equal(serialized.includes("REDACTED"), true); const deleted = await readDeletionState(input.statePath); assert.deepEqual(deleted.filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-00"]); + assert.equal(deleted.filter((item) => item.resource === "read" && item.name === "events").length, 4); } async function assertProtectedMatrix(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { @@ -190,6 +213,72 @@ async function assertEventOnlyFailedMountCandidate(input: { fixturePath: string; assert.equal((await readDeletionState(input.statePath)).length, 0); } +async function assertPendingToRunningRaceFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(91, new Date(Date.now() - 60 * 60_000).toISOString()); + const runningPod = runnerPod(fact, 91, "running"); + (runningPod.metadata as JsonRecord).resourceVersion = "2"; + await writeFixture(input, { + jobs: [runnerJob(fact, 91)], + pods: [runnerPod(fact, 91, "pending")], + events: [failedMountEvent(fact, 91)], + raceAfterEvents: true, + racePods: [runningPod], + }); + + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ + store: new RetentionFactStore([fact]), + options: retentionOptions(input.kubectlCommand, 1), + incomingRunnerCount: 1, + })); + assert.equal(error.details?.reason, "runner-retention-selected-runner-changed"); + assert.equal(error.details?.changeReason, "protected-active"); + const current = (error.details?.current ?? {}) as JsonRecord; + assert.equal(current.protectedReason, "runner-pod-started-running-or-ready"); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); +} + +async function assertDeleteFailureFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(92, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 92)], + pods: [runnerPod(fact, 92, "pending")], + events: [failedMountEvent(fact, 92)], + deleteErrorResources: ["job"], + }); + + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ + store: new RetentionFactStore([fact]), + options: retentionOptions(input.kubectlCommand, 1), + incomingRunnerCount: 1, + })); + assert.equal(error.details?.reason, "runner-retention-delete-failed"); + assert.equal(error.details?.resourceKind, "job"); + assert.equal(error.details?.exitCode, 42); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); +} + +async function assertDeletionPostconditionFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(93, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 93)], + pods: [runnerPod(fact, 93, "pending")], + events: [failedMountEvent(fact, 93)], + retainDeletedJobs: true, + }); + + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ + store: new RetentionFactStore([fact]), + options: retentionOptions(input.kubectlCommand, 1), + incomingRunnerCount: 1, + })); + assert.equal(error.details?.reason, "runner-retention-postcondition-failed"); + assert.equal(error.details?.targetBeforeCreate, 0); + assert.equal(error.details?.liveRunnerCountBefore, 1); + assert.equal(error.details?.liveRunnerCountAfter, 1); + assert.equal(error.details?.deletedRunnerJobCount, 1); + assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-93"]); +} + async function assertProtected( input: { fixturePath: string; statePath: string; kubectlCommand: string }, facts: RetentionFact[], @@ -287,6 +376,8 @@ function runnerJob(fact: RetentionFact, index: number): JsonRecord { metadata: { name, namespace: "agentrun-nc01-v02", + uid: `job-uid-${suffix}`, + resourceVersion: "1", creationTimestamp: fact.run.createdAt, labels: { ...labels, "job-name": name }, annotations: identityAnnotations(fact), @@ -319,6 +410,8 @@ function runnerPod(fact: RetentionFact, index: number, state: "pending" | "runni metadata: { name, namespace: "agentrun-nc01-v02", + uid: `pod-uid-${suffix}`, + resourceVersion: "1", creationTimestamp: fact.run.createdAt, labels: { ...labels, "job-name": jobName }, annotations: identityAnnotations(fact), @@ -372,7 +465,7 @@ function retentionOptions(kubectlCommand: string, maxRunners: number): RunnerRet }; } -async function writeFixture(input: { fixturePath: string; statePath: string }, fixture: { jobs: JsonRecord[]; pods: JsonRecord[]; events?: JsonRecord[]; eventsError?: boolean }): Promise { +async function writeFixture(input: { fixturePath: string; statePath: string }, fixture: RetentionFixture): Promise { await writeFile(input.fixturePath, `${JSON.stringify(fixture, null, 2)}\n`, "utf8"); await writeFile(input.statePath, "[]\n", "utf8"); } @@ -381,6 +474,10 @@ async function readDeletionState(statePath: string): Promise { return JSON.parse(await readFile(statePath, "utf8")) as JsonRecord[]; } +function deletionOperations(state: JsonRecord[]): JsonRecord[] { + return state.filter((item) => item.resource === "job" || item.resource === "pod" || item.resource === "secret" || item.resource === "pvc"); +} + async function retentionError(run: () => Promise): Promise { try { await run(); diff --git a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs index a199708..ab36ab5 100755 --- a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs +++ b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs @@ -25,7 +25,10 @@ function podJobName(item) { const fixture = await readJson(fixturePath, { jobs: [], pods: [] }); const state = await readJson(statePath, []); -const deletedJobs = new Set(state.filter((item) => item.resource === "job").map((item) => item.name)); +const eventReadCount = state.filter((item) => item.resource === "read" && item.name === "events").length; +const deletedJobs = fixture.retainDeletedJobs === true + ? new Set() + : new Set(state.filter((item) => item.resource === "job").map((item) => item.name)); if (args[0] === "get" && args[1] === "jobs") { const items = fixture.jobs.filter((item) => !deletedJobs.has(item?.metadata?.name)); @@ -34,7 +37,10 @@ if (args[0] === "get" && args[1] === "jobs") { } if (args[0] === "get" && args[1] === "pods") { - const items = fixture.pods.filter((item) => !deletedJobs.has(podJobName(item))); + const sourcePods = fixture.raceAfterEvents === true && eventReadCount > 0 && Array.isArray(fixture.racePods) + ? fixture.racePods + : fixture.pods; + const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item))); console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items })); process.exit(0); } @@ -44,7 +50,22 @@ if (args[0] === "get" && args[1] === "events") { console.error("events access denied by self-test fixture"); process.exit(1); } - console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: fixture.events ?? [] })); + const sourceItems = eventReadCount > 0 && Array.isArray(fixture.eventsAfterFirstRead) + ? fixture.eventsAfterFirstRead + : fixture.events ?? []; + const items = fixture.churnEventMetadata === true + ? sourceItems.map((item) => ({ + ...item, + count: Number(item?.count ?? 0) + eventReadCount, + lastTimestamp: new Date(Date.parse(item?.lastTimestamp ?? "2026-01-01T00:00:00.000Z") + eventReadCount * 1000).toISOString(), + message: `${String(item?.message ?? "warning")} occurrence=${eventReadCount + 1}`, + })) + : sourceItems; + if (fixture.trackEventReads === true || fixture.raceAfterEvents === true || Array.isArray(fixture.eventsAfterFirstRead)) { + state.push({ resource: "read", name: "events", sequence: eventReadCount + 1, valuesPrinted: false }); + await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`); + } + console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items })); process.exit(0); } @@ -54,6 +75,10 @@ if (args[0] === "delete") { const selectorIndex = args.indexOf("-l"); const selector = selectorIndex >= 0 ? String(args[selectorIndex + 1] ?? "") : null; const name = directName ?? selector ?? "selected"; + if (Array.isArray(fixture.deleteErrorResources) && fixture.deleteErrorResources.includes(resource)) { + console.error(`self-test forced ${resource} deletion failure`); + process.exit(42); + } state.push({ resource, name, selector, valuesPrinted: false }); await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`); console.log(`${resource}/${name.replace(/[^a-z0-9.-]+/giu, "-")} deleted`);