diff --git a/package.json b/package.json index fe3010a..bf6737d 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "check": "tsc --noEmit", "self-test": "bun run src/selftest/run.ts", "self-test:postgres-kafka-order": "bun run src/selftest/integration/postgres-kafka-order.ts", + "self-test:postgres-runner-retention-fence": "bun run src/selftest/integration/postgres-runner-retention-fence.ts", "test": "bun run src/selftest/run.ts", "cli": "bun scripts/agentrun-cli.ts" }, diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index facb6ea..ebdf63e 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -79,6 +79,12 @@ export interface CreateRunnerJobInput extends JsonRecord { } export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise { + if (!options.defaults.retention) return await createKubernetesRunnerJobUnderFence(options); + const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace; + return await options.store.withRunnerCreateFence(namespace, async () => await createKubernetesRunnerJobUnderFence(options)); +} + +async function createKubernetesRunnerJobUnderFence(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise { throwIfAborted(options.signal, "runner dispatch attempt"); const commandId = stringField(options.input, "commandId"); const run = await options.store.getRun(options.runId); diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index d04625b..ca415a5 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -5,7 +5,7 @@ import { AgentRunError } from "../common/errors.js"; import { redactJson } from "../common/redaction.js"; import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; -import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; +import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js"; import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js"; import { normalizeRunEventPayload, requireEventType } from "../common/events.js"; @@ -458,12 +458,14 @@ export async function createPostgresAgentRunStore(options: PostgresStoreOptions) export class PostgresAgentRunStore implements AgentRunStore { private readonly pool: Pool; + private readonly runnerCreateFencePool: Pool; private readonly eventOutboxConfig: EventOutboxConfig; private migrationReady = false; private appliedMigrationId: string | null = null; constructor(options: PostgresStoreOptions) { this.pool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-v01", ...(options.poolMax === undefined ? {} : { max: options.poolMax }) }); + this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: 1 }); this.eventOutboxConfig = options.eventOutbox ?? { enabled: false, topic: null, source: null }; } @@ -879,6 +881,50 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( }); } + async withRunnerCreateFence(namespace: string, operation: () => Promise): Promise { + const client = await this.runnerCreateFencePool.connect(); + try { + await client.query("BEGIN"); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-runner-create", namespace]); + const result = await operation(); + await client.query("COMMIT"); + return result; + } catch (error) { + await client.query("ROLLBACK"); + throw error; + } finally { + client.release(); + } + } + + async withRunnerRetentionFence(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise): Promise { + return this.withTransaction(async (client) => { + const run = await this.requireRunForUpdate(client, input.runId); + const commandRows = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [run.id]); + const commands = commandRows.rows.slice(0, 100).map(commandFromRow); + const commandRow = commandRows.rows.find((item) => String(item.id) === input.commandId); + if (!commandRow) throw new AgentRunError("schema-invalid", `command ${input.commandId} does not belong to run ${input.runId}`, { httpStatus: 409 }); + const command = commandFromRow(commandRow); + const eventRows = await client.query("SELECT * FROM agentrun_events WHERE run_id = $1 ORDER BY seq ASC LIMIT 500", [run.id]); + const result = await operation({ run, command, commands, events: eventRows.rows.map(eventFromRow) }); + if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) return result; + + const at = nowIso(); + const updatedCommands = await client.query("UPDATE agentrun_commands SET state = 'failed', updated_at = $2 WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') RETURNING *", [run.id, at]); + await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, input.reason, at); + const updatedRun = await client.query("UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = 'infra-failed', failure_message = $2, claimed_by = NULL, lease_expires_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *", [run.id, input.reason, at]); + await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]); + for (const row of updatedCommands.rows) { + const item = commandFromRow(row); + await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: item.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true }); + } + await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true }); + const next = runFromRow(updatedRun.rows[0]); + await this.touchSessionForRun(client, next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); + return result; + }); + } + async claimRun(runId: string, runnerId: string, leaseMs: number): Promise { return this.withTransaction(async (client) => { const run = await this.requireRunForUpdate(client, runId); @@ -1393,7 +1439,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( } async close(): Promise { - await this.pool.end(); + await Promise.all([this.pool.end(), this.runnerCreateFencePool.end()]); } private async createCancelRequest(client: PoolClient, input: { targetKind: CancelTargetKind; targetId: string; run: RunRecord; command: CommandRecord | null; reason: string; at: string; stage: CancelStage }): Promise { diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index 134e941..6ba47d4 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -166,12 +166,26 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu let deletedAssociatedResourceCount = 0; for (const item of selected) { await revalidateSelectedRunner(input.store, input.options, item); + if (!item.runId || !item.commandId || !item.dbFactsFingerprint) { + throw new AgentRunError("infra-failed", "runner retention selected resource has no fenceable database identity", { + httpStatus: 409, + details: { reason: "runner-retention-db-fence-unavailable", namespace: input.options.namespace, resourceKind: item.resourceKind, name: item.name, valuesPrinted: false }, + }); + } + await input.store.withRunnerRetentionFence({ + runId: item.runId, + commandId: item.commandId, + terminalizeStalePending: item.candidateKind === "stale-pending", + reason: `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`, + }, async (facts) => { + const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events); + if (currentDbFingerprint !== item.dbFactsFingerprint) throw databaseFenceRejected(input.options, item, currentDbFingerprint, facts.run); + await deleteRunnerResourceCas(input.options, item); + }); if (item.resourceKind === "job") { - await deleteRunnerResource(input.options, item); deletedRunnerJobCount++; deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName); } else { - await deleteRunnerResource(input.options, item); deletedRunnerPodCount++; } } @@ -430,16 +444,74 @@ async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRet }); } -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", +function databaseFenceRejected(options: RunnerRetentionOptions, item: RunnerResourceEntry, currentDbFingerprint: string, run: RunRecord): AgentRunError { + return new AgentRunError("infra-failed", "runner retention database facts changed before atomic deletion", { + httpStatus: 409, + details: { + reason: "runner-retention-db-fence-rejected", + changeReason: run.claimedBy || run.status === "claimed" || run.status === "running" ? "run-became-active" : "database-fingerprint-changed", namespace: options.namespace, resourceKind: item.resourceKind, name: item.name, + selectedDbFingerprint: item.dbFactsFingerprint, + currentDbFingerprint, + runStatus: run.status, + claimedBy: run.claimedBy, + leaseExpiresAt: run.leaseExpiresAt, + valuesPrinted: false, + }, + }); +} + +async function deleteRunnerResourceCas(options: RunnerRetentionOptions, item: RunnerResourceEntry): Promise { + if (item.resourceKind === "job") { + for (const pod of [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name))) { + await deleteKubernetesObjectCas(options, item, "pod", pod.name, pod.uid, pod.resourceVersion); + } + const currentJob = await kubectlGetObject(options, "job", item.name); + const expectedIdentity = { + "agentrun.pikastech.local/run-id": item.runId ?? "", + "agentrun.pikastech.local/command-id": item.commandId ?? "", + "agentrun.pikastech.local/runner-id": item.runnerId ?? "", + }; + if (currentJob.metadata?.uid !== item.resourceUid || !resourceIdentityMatches(expectedIdentity, currentJob.metadata?.annotations ?? {})) { + throw new AgentRunError("infra-failed", "runner retention Job identity changed after Pod CAS deletion", { + httpStatus: 409, + details: { reason: "runner-retention-k8s-cas-rejected", changeReason: "job-identity-changed", namespace: options.namespace, resourceKind: "job", name: item.name, expectedUid: item.resourceUid, currentUid: currentJob.metadata?.uid ?? null, valuesPrinted: false }, + }); + } + await deleteKubernetesObjectCas(options, item, "job", item.name, currentJob.metadata?.uid ?? null, currentJob.metadata?.resourceVersion ?? null); + return; + } + await deleteKubernetesObjectCas(options, item, item.resourceKind, item.name, item.resourceUid, item.resourceVersion); +} + +async function deleteKubernetesObjectCas(options: RunnerRetentionOptions, selected: RunnerResourceEntry, resourceKind: "job" | "pod", name: string, uid: string | null, resourceVersion: string | null): Promise { + if (!uid || !resourceVersion) { + throw new AgentRunError("infra-failed", "runner retention cannot bind Kubernetes delete preconditions", { + httpStatus: 409, + details: { reason: "runner-retention-k8s-cas-unavailable", namespace: options.namespace, selectedResourceKind: selected.resourceKind, selectedName: selected.name, resourceKind, name, uidPresent: Boolean(uid), resourceVersionPresent: Boolean(resourceVersion), valuesPrinted: false }, + }); + } + const resourcePath = resourceKind === "job" + ? `/apis/batch/v1/namespaces/${encodeURIComponent(options.namespace)}/jobs/${encodeURIComponent(name)}` + : `/api/v1/namespaces/${encodeURIComponent(options.namespace)}/pods/${encodeURIComponent(name)}`; + const deleteOptions = JSON.stringify({ apiVersion: "v1", kind: "DeleteOptions", propagationPolicy: "Background", preconditions: { uid, resourceVersion } }); + const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", "--raw", resourcePath, "-f", "-"], options.signal, deleteOptions); + if (result.code === 0) return; + const conflict = /\b(?:Conflict|PreconditionFailed|resourceVersion|UID)\b/iu.test(`${result.stderr}\n${result.stdout}`); + throw new AgentRunError("infra-failed", `Kubernetes CAS delete ${resourceKind} failed with code ${result.code}`, { + httpStatus: conflict ? 409 : 502, + details: redactJson({ + reason: "runner-retention-k8s-cas-rejected", + changeReason: conflict ? "object-identity-or-version-changed" : "delete-request-failed", + namespace: options.namespace, + selectedResourceKind: selected.resourceKind, + selectedName: selected.name, + resourceKind, + name, + expectedUid: uid, + expectedResourceVersion: resourceVersion, exitCode: result.code, signal: result.signal, stderr: redactText(result.stderr.slice(-2000)), @@ -460,6 +532,12 @@ async function kubectlGetList(options: RunnerRetentionOptions, resource: string, return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : []; } +async function kubectlGetObject(options: RunnerRetentionOptions, resource: "job" | "pod", name: string): Promise { + const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["get", resource, name, "-n", options.namespace, "-o", "json"], options.signal); + if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource}/${name} failed with code ${result.code}`, { httpStatus: 409, details: redactJson({ reason: "runner-retention-k8s-cas-rejected", changeReason: "resource-missing-before-cas", namespace: options.namespace, resourceKind: resource, name, stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); + return parseJsonObject(result.stdout, `kubectl get ${resource}/${name}`) as unknown as K8sObject; +} + async function kubectlGetPodEvents(options: RunnerRetentionOptions): Promise { const args = ["get", "events", "-n", options.namespace, "--field-selector", "involvedObject.kind=Pod,type=Warning", "-o", "json"]; const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args, options.signal); @@ -480,15 +558,19 @@ async function deleteAssociatedResources(options: RunnerRetentionOptions, jobNam return deleted; } -async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> { +async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal, stdin?: string): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> { throwIfAborted(abortSignal, "kubectl"); - const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] }); + const child = spawn(kubectlCommand, args, { stdio: [stdin === undefined ? "ignore" : "pipe", "pipe", "pipe"] }); let stdout = ""; let stderr = ""; - child.stdout.setEncoding("utf8"); - child.stderr.setEncoding("utf8"); - child.stdout.on("data", (chunk) => { stdout += String(chunk); }); - child.stderr.on("data", (chunk) => { stderr += String(chunk); }); + const stdoutStream = child.stdout; + const stderrStream = child.stderr; + if (!stdoutStream || !stderrStream) throw new AgentRunError("infra-failed", "kubectl output pipes were not created", { httpStatus: 503 }); + stdoutStream.setEncoding("utf8"); + stderrStream.setEncoding("utf8"); + stdoutStream.on("data", (chunk) => { stdout += String(chunk); }); + stderrStream.on("data", (chunk) => { stderr += String(chunk); }); + if (stdin !== undefined) child.stdin?.end(stdin); const result = await waitForChildProcess(child, { ...(abortSignal ? { signal: abortSignal } : {}), label: "kubectl" }); return { ...result, stdout, stderr }; } diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 9186d71..5382f36 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -32,6 +32,20 @@ export interface DurableQueueClaim { attemptCount: number; } +export interface RunnerRetentionFenceFacts { + run: RunRecord; + command: CommandRecord; + commands: CommandRecord[]; + events: RunEvent[]; +} + +export interface RunnerRetentionFenceInput { + runId: string; + commandId: string; + terminalizeStalePending: boolean; + reason: string; +} + export interface AgentRunStore { health(): MaybePromise; createRun(input: CreateRunInput): MaybePromise; @@ -57,6 +71,8 @@ export interface AgentRunStore { getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): MaybePromise; saveRunnerJob(input: SaveRunnerJobInput): MaybePromise; updateRunnerJobResult(runnerJobId: string, patch: JsonRecord): MaybePromise; + withRunnerCreateFence(namespace: string, operation: () => Promise): MaybePromise; + withRunnerRetentionFence(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise): MaybePromise; claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise; heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise; ackCommand(commandId: string): MaybePromise; @@ -179,6 +195,8 @@ export class MemoryAgentRunStore implements AgentRunStore { private readonly kafkaEventOutbox = new Map(); private readonly queueTasks = new Map(); private readonly queueReadCursors = new Map(); + private readonly runnerCreateFenceTails = new Map>(); + private readonly runnerRetentionFences = new Set(); private queueVersion = 0; private sessionVersion = 0; private kafkaOutboxSeq = 0; @@ -417,7 +435,40 @@ export class MemoryAgentRunStore implements AgentRunStore { return next; } + async withRunnerCreateFence(namespace: string, operation: () => Promise): Promise { + const predecessor = this.runnerCreateFenceTails.get(namespace) ?? Promise.resolve(); + let release = (): void => {}; + const current = new Promise((resolve) => { release = resolve; }); + const tail = predecessor.then(async () => current); + this.runnerCreateFenceTails.set(namespace, tail); + await predecessor; + try { + return await operation(); + } finally { + release(); + if (this.runnerCreateFenceTails.get(namespace) === tail) this.runnerCreateFenceTails.delete(namespace); + } + } + + async withRunnerRetentionFence(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise): Promise { + if (this.runnerRetentionFences.has(input.runId)) throw runnerRetentionFenceConflict(input.runId); + const run = this.getRun(input.runId); + const command = this.getCommand(input.commandId); + if (command.runId !== run.id) throw new AgentRunError("schema-invalid", `command ${input.commandId} does not belong to run ${input.runId}`, { httpStatus: 409 }); + this.runnerRetentionFences.add(input.runId); + try { + const commands = this.listCommands(run.id, 0, 100); + const events = this.listEvents(run.id, 0, 500); + const result = await operation({ run, command, commands, events }); + if (input.terminalizeStalePending && !isTerminalRunStatus(run.status)) this.terminalizeRunnerRetention(run, commands, input.reason); + return result; + } finally { + this.runnerRetentionFences.delete(input.runId); + } + } + claimRun(runId: string, runnerId: string, leaseMs: number): RunRecord { + if (this.runnerRetentionFences.has(runId)) throw runnerRetentionFenceConflict(runId); const run = this.getRun(runId); if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); if (run.claimedBy && run.claimedBy !== runnerId && !isLeaseExpired(run.leaseExpiresAt)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 }); @@ -428,6 +479,7 @@ export class MemoryAgentRunStore implements AgentRunStore { } heartbeat(runId: string, runnerId: string, leaseMs: number): RunRecord { + if (this.runnerRetentionFences.has(runId)) throw runnerRetentionFenceConflict(runId); const run = this.getRun(runId); if (isTerminalRunStatus(run.status)) return run; if (run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is not claimed by ${runnerId}`, { httpStatus: 409 }); @@ -800,6 +852,20 @@ export class MemoryAgentRunStore implements AgentRunStore { return next; } + private terminalizeRunnerRetention(run: RunRecord, commands: CommandRecord[], reason: string): void { + const at = nowIso(); + for (const command of commands.filter((item) => !isTerminalCommandState(item.state))) { + this.commands.set(command.id, { ...command, state: "failed", updatedAt: at }); + this.cancelRunnerDispatchIntentForCommand(command.id, reason, at); + const prepared = this.prepareEvent(run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message: reason, retentionFence: true }); + this.commitPreparedEvent(prepared); + } + const next = this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: reason, claimedBy: null, leaseExpiresAt: null }); + this.cancelRunnerDispatchIntentsForRun(run.id, reason, at); + this.commitPreparedEvent(this.prepareEvent(run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message: reason, retentionFence: true })); + this.touchSessionForRun(next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); + } + private requireRunnerDispatchIntent(intentId: string): RunnerDispatchIntentRecord { const intent = this.runnerDispatchIntents.get(intentId); if (!intent) throw new AgentRunError("schema-invalid", `runner dispatch intent ${intentId} was not found`, { httpStatus: 404 }); @@ -1239,6 +1305,13 @@ function staleClaimError(component: string, claim: DurableQueueClaim): AgentRunE return new AgentRunError("runner-lease-conflict", `${component} claim is stale`, { httpStatus: 409, details: { id: claim.id, attemptCount: claim.attemptCount, valuesPrinted: false } }); } +function runnerRetentionFenceConflict(runId: string): AgentRunError { + return new AgentRunError("runner-lease-conflict", `run ${runId} is fenced by runner retention`, { + httpStatus: 409, + details: { reason: "runner-retention-db-fenced", runId, valuesPrinted: false }, + }); +} + function isActiveRunnerDispatchIntent(intent: RunnerDispatchIntentRecord): boolean { return intent.state === "pending" || intent.state === "dispatching" || intent.state === "retry"; } diff --git a/src/selftest/cases/20-runner-k8s-job.ts b/src/selftest/cases/20-runner-k8s-job.ts index 5c89278..79a3a58 100644 --- a/src/selftest/cases/20-runner-k8s-job.ts +++ b/src/selftest/cases/20-runner-k8s-job.ts @@ -299,6 +299,8 @@ function runnerJob(name, createdAt, identity) { metadata: { name, namespace: "agentrun-v02", + uid: "uid-" + name, + resourceVersion: "1", creationTimestamp: createdAt, labels: { "app.kubernetes.io/part-of": "agentrun", "app.kubernetes.io/name": "agentrun-runner", "app.kubernetes.io/component": "runner", "agentrun.pikastech.local/lane": "v0.2", "job-name": name }, annotations: { "agentrun.pikastech.local/run-id": identity.runId, "agentrun.pikastech.local/command-id": identity.commandId, "agentrun.pikastech.local/runner-id": "runner-old", token: null, secret: null }, @@ -306,17 +308,27 @@ function runnerJob(name, createdAt, identity) { status: {}, }; } +function retentionJobs() { + const facts = JSON.parse(process.env.AGENTRUN_SELFTEST_RETENTION_DB_FACTS || "[]"); + return [ + runnerJob("agentrun-v02-runner-old-a", "2026-01-01T00:00:00Z", facts[0]), + runnerJob("agentrun-v02-runner-old-b", "2026-01-01T00:01:00Z", facts[1]), + ]; +} if (args[0] === "get" && args[1] === "jobs") { const deleted = await readDeleted(); const deletedJobs = new Set(deleted.filter((item) => item.resource === "job").map((item) => item.name)); - const facts = JSON.parse(process.env.AGENTRUN_SELFTEST_RETENTION_DB_FACTS || "[]"); - const items = [ - runnerJob("agentrun-v02-runner-old-a", "2026-01-01T00:00:00Z", facts[0]), - runnerJob("agentrun-v02-runner-old-b", "2026-01-01T00:01:00Z", facts[1]), - ].filter((item) => !deletedJobs.has(item.metadata.name)); + const items = retentionJobs().filter((item) => !deletedJobs.has(item.metadata.name)); console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items })); process.exit(0); } +if (args[0] === "get" && args[1] === "job" && args[2]) { + const deleted = await readDeleted(); + const item = retentionJobs().find((candidate) => candidate.metadata.name === args[2]); + if (!item || deleted.some((entry) => entry.resource === "job" && entry.name === args[2])) process.exit(1); + console.log(JSON.stringify(item)); + process.exit(0); +} if (args[0] === "get" && args[1] === "pods") { console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); process.exit(0); @@ -324,8 +336,15 @@ if (args[0] === "get" && args[1] === "pods") { if (args[0] === "get" && args[1] === "events") { console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); process.exit(0); } if (args[0] === "delete") { const deleted = await readDeleted(); - const resource = args[1]; - const name = args[2] && !args[2].startsWith("-") ? args[2] : args.join(" "); + const rawIndex = args.indexOf("--raw"); + const rawPath = rawIndex >= 0 ? args[rawIndex + 1] : null; + const resource = rawPath?.includes("/jobs/") ? "job" : args[1]; + const name = rawPath ? decodeURIComponent(rawPath.split("/").filter(Boolean).at(-1)) : args[2] && !args[2].startsWith("-") ? args[2] : args.join(" "); + if (rawPath) { + const body = JSON.parse(await readStdin()); + const current = retentionJobs().find((candidate) => candidate.metadata.name === name); + if (!current || body?.preconditions?.uid !== current.metadata.uid || body?.preconditions?.resourceVersion !== current.metadata.resourceVersion) process.exit(1); + } deleted.push({ resource, name, args }); await writeDeleted(deleted); console.log(String(resource) + "/" + String(name).replace(/[^a-z0-9-]+/g, "-") + " deleted"); diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index 2ffbf60..87d3097 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/src/selftest/cases/22-runner-retention-stale-pending.ts @@ -4,7 +4,7 @@ import path from "node:path"; import { AgentRunError } from "../../common/errors.js"; import type { CommandRecord, JsonRecord, RunEvent, RunRecord } from "../../common/types.js"; import { enforceRunnerRetentionBeforeCreate, type RunnerRetentionOptions, type RunnerRetentionSummary } from "../../mgr/runner-retention.js"; -import { MemoryAgentRunStore } from "../../mgr/store.js"; +import { MemoryAgentRunStore, type RunnerRetentionFenceFacts, type RunnerRetentionFenceInput } from "../../mgr/store.js"; import type { SelfTestCase } from "../harness.js"; const stalePendingMaxAgeMs = 15 * 60_000; @@ -26,16 +26,21 @@ interface RetentionFixture { events?: JsonRecord[]; eventsAfterFirstRead?: JsonRecord[]; racePods?: JsonRecord[]; + raceJobs?: JsonRecord[]; eventsError?: boolean; trackEventReads?: boolean; churnEventMetadata?: boolean; raceAfterEvents?: boolean; + raceOnCasDelete?: boolean; + bumpJobResourceVersionAfterPodDelete?: boolean; + bumpedJobResourceVersion?: string; deleteErrorResources?: string[]; retainDeletedJobs?: boolean; } class RetentionFactStore extends MemoryAgentRunStore { private readonly facts: RetentionFact[]; + private beforeFence: (() => void) | null = null; constructor(facts: RetentionFact[]) { super(); @@ -64,6 +69,33 @@ class RetentionFactStore extends MemoryAgentRunStore { if (!fact) throw new AgentRunError("schema-invalid", `run ${runId} was not found`, { httpStatus: 404 }); return fact.events.filter((event) => event.seq > afterSeq).slice(0, limit); } + + setBeforeFence(callback: (() => void) | null): void { + this.beforeFence = callback; + } + + claimFact(runId: string, runnerId: string): void { + const fact = this.facts.find((item) => item.run.id === runId); + if (!fact) throw new Error(`missing retention fact ${runId}`); + const at = new Date(Date.now() + 60_000).toISOString(); + fact.run = { ...fact.run, status: "claimed", claimedBy: runnerId, leaseExpiresAt: at, updatedAt: at }; + fact.events.push(runEvent(runId, fact.events.length + 1, at, "backend_status", { phase: "run-claimed", runnerId, commandId: fact.command.id })); + } + + override async withRunnerRetentionFence(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise): Promise { + const beforeFence = this.beforeFence; + this.beforeFence = null; + beforeFence?.(); + const fact = this.facts.find((item) => item.run.id === input.runId && item.command.id === input.commandId); + if (!fact) throw new Error(`missing retention fence facts ${input.runId}/${input.commandId}`); + const result = await operation({ run: fact.run, command: fact.command, commands: this.listCommands(fact.run.id, 0, 100), events: this.listEvents(fact.run.id, 0, 500) }); + if (input.terminalizeStalePending) { + const at = new Date().toISOString(); + fact.command = { ...fact.command, state: "failed", updatedAt: at }; + fact.run = { ...fact.run, status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: input.reason, claimedBy: null, leaseExpiresAt: null, updatedAt: at }; + } + return result; + } } const selfTest: SelfTestCase = async (context) => { @@ -81,8 +113,12 @@ const selfTest: SelfTestCase = async (context) => { await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand }); await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand }); await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertDbClaimAfterRecheckFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertPodCasRaceFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertDeleteFailureFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertMemoryDbFenceRejectsConcurrentClaim(); + await assertRunnerCreateFenceSerializesNamespace(); } finally { restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath); restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath); @@ -107,8 +143,12 @@ const selfTest: SelfTestCase = async (context) => { "event-facts-unavailable-protected", "terminal-ledger-running-pod-protected", "selected-runner-race-fails-closed", - "runner-delete-exit-code-checked", + "db-claim-after-recheck-fails-closed", + "pod-resource-version-cas-fails-closed", + "runner-delete-cas-exit-code-checked", "runner-delete-postcondition-checked", + "memory-db-fence-rejects-concurrent-claim", + "namespace-runner-create-fence-prevents-capacity-overshoot", ], }; }; @@ -124,6 +164,8 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s events: facts.map((fact, index) => failedMountEvent(fact, index)), trackEventReads: true, churnEventMetadata: true, + bumpJobResourceVersionAfterPodDelete: true, + bumpedJobResourceVersion: "2", }; await writeFixture(input, fixture); const store = new RetentionFactStore(facts); @@ -163,6 +205,8 @@ 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"]); + const jobCas = deleted.find((item) => item.resource === "cas-request" && item.targetKind === "job") ?? {}; + assert.equal(jobCas.expectedResourceVersion, "2"); assert.equal(deleted.filter((item) => item.resource === "read" && item.name === "events").length, 4); } @@ -237,6 +281,43 @@ async function assertPendingToRunningRaceFailsClosed(input: { fixturePath: strin assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); } +async function assertDbClaimAfterRecheckFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(94, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { jobs: [runnerJob(fact, 94)], pods: [runnerPod(fact, 94, "pending")], events: [failedMountEvent(fact, 94)] }); + const store = new RetentionFactStore([fact]); + store.setBeforeFence(() => store.claimFact(fact.run.id, "runner-raced-after-recheck")); + + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store, options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 })); + assert.equal(error.details?.reason, "runner-retention-db-fence-rejected"); + assert.equal(error.details?.changeReason, "run-became-active"); + assert.equal(error.details?.claimedBy, "runner-raced-after-recheck"); + const state = await readDeletionState(input.statePath); + assert.equal(deletionOperations(state).length, 0); + assert.equal(state.filter((item) => item.resource === "cas-request").length, 0); +} + +async function assertPodCasRaceFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(95, new Date(Date.now() - 60 * 60_000).toISOString()); + const runningPod = runnerPod(fact, 95, "running"); + (runningPod.metadata as JsonRecord).resourceVersion = "2"; + await writeFixture(input, { + jobs: [runnerJob(fact, 95)], + pods: [runnerPod(fact, 95, "pending")], + events: [failedMountEvent(fact, 95)], + raceOnCasDelete: 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-k8s-cas-rejected"); + assert.equal(error.details?.changeReason, "object-identity-or-version-changed"); + assert.equal(error.details?.resourceKind, "pod"); + assert.equal(error.details?.expectedResourceVersion, "1"); + const state = await readDeletionState(input.statePath); + assert.equal(deletionOperations(state).length, 0); + assert.equal(state.filter((item) => item.resource === "cas-request" && item.targetKind === "pod").length, 1); +} + 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, { @@ -251,10 +332,13 @@ async function assertDeleteFailureFailsClosed(input: { fixturePath: string; stat options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1, })); - assert.equal(error.details?.reason, "runner-retention-delete-failed"); + assert.equal(error.details?.reason, "runner-retention-k8s-cas-rejected"); + assert.equal(error.details?.changeReason, "delete-request-failed"); assert.equal(error.details?.resourceKind, "job"); assert.equal(error.details?.exitCode, 42); - assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); + const state = await readDeletionState(input.statePath); + assert.deepEqual(deletionOperations(state).filter((item) => item.resource === "job"), []); + assert.deepEqual(deletionOperations(state).filter((item) => item.resource === "pod").map((item) => item.name), ["agentrun-v02-runner-stale-92-pod-92"]); } async function assertDeletionPostconditionFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { @@ -279,6 +363,63 @@ async function assertDeletionPostconditionFailsClosed(input: { fixturePath: stri assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-93"]); } +async function assertMemoryDbFenceRejectsConcurrentClaim(): Promise { + const store = new MemoryAgentRunStore(); + const run = store.createRun({ tenantId: "unidesk", projectId: "pikasTech/agentrun", workspaceRef: { kind: "host-path", path: "/tmp/agentrun-retention-memory" }, providerId: "NC01", backendProfile: "codex", executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } }, traceSink: null }); + const command = store.createCommand(run.id, { type: "turn", payload: { prompt: "memory retention fence" } }); + let enterFence = (): void => {}; + let releaseFence = (): void => {}; + const entered = new Promise((resolve) => { enterFence = resolve; }); + const release = new Promise((resolve) => { releaseFence = resolve; }); + const fenced = Promise.resolve(store.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "retention memory self-test" }, async () => { + enterFence(); + await release; + })); + await entered; + assert.throws( + () => store.claimRun(run.id, "runner-during-retention", 60_000), + (error) => error instanceof AgentRunError && error.failureKind === "runner-lease-conflict" && error.details?.reason === "runner-retention-db-fenced", + ); + releaseFence(); + await fenced; + assert.equal(store.getRun(run.id).status, "failed"); + assert.equal(store.getCommand(command.id).state, "failed"); + assert.throws(() => store.claimRun(run.id, "runner-after-retention", 60_000), (error) => error instanceof AgentRunError && error.httpStatus === 409); +} + +async function assertRunnerCreateFenceSerializesNamespace(): Promise { + const store = new MemoryAgentRunStore(); + const maxRunners = 20; + let liveRunnerCount = 19; + let enterFirst = (): void => {}; + let releaseFirst = (): void => {}; + const firstEntered = new Promise((resolve) => { enterFirst = resolve; }); + const firstRelease = new Promise((resolve) => { releaseFirst = resolve; }); + let secondEntered = false; + const first = Promise.resolve(store.withRunnerCreateFence("agentrun-nc01-v02", async () => { + assert.ok(liveRunnerCount + 1 <= maxRunners); + enterFirst(); + await firstRelease; + liveRunnerCount++; + return "first"; + })); + await firstEntered; + const second = Promise.resolve(store.withRunnerCreateFence("agentrun-nc01-v02", async () => { + secondEntered = true; + if (liveRunnerCount + 1 > maxRunners) throw new AgentRunError("infra-failed", "capacity fence rejected second incoming runner", { httpStatus: 409 }); + liveRunnerCount++; + return "second"; + })); + await Promise.resolve(); + assert.equal(secondEntered, false); + releaseFirst(); + const results = await Promise.allSettled([first, second]); + assert.equal(results[0]?.status, "fulfilled"); + assert.equal(results[1]?.status, "rejected"); + assert.equal(secondEntered, true); + assert.equal(liveRunnerCount, maxRunners); +} + async function assertProtected( input: { fixturePath: string; statePath: string; kubectlCommand: string }, facts: RetentionFact[], diff --git a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs index ab36ab5..c589587 100755 --- a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs +++ b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs @@ -29,6 +29,25 @@ const eventReadCount = state.filter((item) => item.resource === "read" && item.n const deletedJobs = fixture.retainDeletedJobs === true ? new Set() : new Set(state.filter((item) => item.resource === "job").map((item) => item.name)); +const deletedPods = new Set(state.filter((item) => item.resource === "pod").map((item) => item.name)); + +function currentJob(name) { + const base = fixture.jobs.find((item) => item?.metadata?.name === name); + if (!base || fixture.bumpJobResourceVersionAfterPodDelete !== true) return base; + const podDeleted = fixture.pods.some((pod) => podJobName(pod) === name && deletedPods.has(pod?.metadata?.name)); + if (!podDeleted) return base; + return { ...base, metadata: { ...(base.metadata ?? {}), resourceVersion: fixture.bumpedJobResourceVersion ?? "2" } }; +} + +if (args[0] === "get" && args[1] === "job" && args[2]) { + const item = currentJob(String(args[2])); + if (!item || deletedJobs.has(item?.metadata?.name)) { + console.error(`Error from server (NotFound): job/${String(args[2])} was not found`); + process.exit(1); + } + console.log(JSON.stringify(item)); + process.exit(0); +} if (args[0] === "get" && args[1] === "jobs") { const items = fixture.jobs.filter((item) => !deletedJobs.has(item?.metadata?.name)); @@ -40,7 +59,7 @@ if (args[0] === "get" && args[1] === "pods") { const sourcePods = fixture.raceAfterEvents === true && eventReadCount > 0 && Array.isArray(fixture.racePods) ? fixture.racePods : fixture.pods; - const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item))); + const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item)) && !deletedPods.has(item?.metadata?.name)); console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items })); process.exit(0); } @@ -70,6 +89,32 @@ if (args[0] === "get" && args[1] === "events") { } if (args[0] === "delete") { + const rawIndex = args.indexOf("--raw"); + if (rawIndex >= 0) { + const resourcePath = String(args[rawIndex + 1] ?? ""); + const resource = resourcePath.includes("/jobs/") ? "job" : resourcePath.includes("/pods/") ? "pod" : "unknown"; + const name = decodeURIComponent(resourcePath.split("/").filter(Boolean).at(-1) ?? "unknown"); + const deleteOptions = await readStdinJson(); + const expected = deleteOptions?.preconditions ?? {}; + const source = fixture.raceOnCasDelete === true && Array.isArray(fixture.racePods) ? fixture.racePods : fixture.pods; + const current = resource === "job" + ? (fixture.raceOnCasDelete === true && Array.isArray(fixture.raceJobs) ? fixture.raceJobs.find((item) => item?.metadata?.name === name) : currentJob(name)) + : source.find((item) => item?.metadata?.name === name); + state.push({ resource: "cas-request", name, targetKind: resource, expectedUid: expected.uid ?? null, expectedResourceVersion: expected.resourceVersion ?? null, valuesPrinted: false }); + await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`); + if (!expected.uid || !expected.resourceVersion || !current || current?.metadata?.uid !== expected.uid || current?.metadata?.resourceVersion !== expected.resourceVersion) { + console.error(`Error from server (Conflict): ${resource}/${name} resourceVersion or UID precondition failed`); + process.exit(1); + } + 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: null, cas: true, valuesPrinted: false }); + await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`); + console.log(JSON.stringify({ kind: "Status", apiVersion: "v1", status: "Success", details: { name, kind: resource } })); + process.exit(0); + } const resource = String(args[1] ?? "unknown"); const directName = args[2] && !String(args[2]).startsWith("-") ? String(args[2]) : null; const selectorIndex = args.indexOf("-l"); @@ -85,5 +130,13 @@ if (args[0] === "delete") { process.exit(0); } +async function readStdinJson() { + try { + return JSON.parse(await Bun.stdin.text()); + } catch { + return null; + } +} + console.error(`unsupported fake retention kubectl args: ${args.join(" ")}`); process.exit(1); diff --git a/src/selftest/integration/postgres-runner-retention-fence.ts b/src/selftest/integration/postgres-runner-retention-fence.ts new file mode 100644 index 0000000..52e2872 --- /dev/null +++ b/src/selftest/integration/postgres-runner-retention-fence.ts @@ -0,0 +1,80 @@ +import assert from "node:assert/strict"; +import { AgentRunError } from "../../common/errors.js"; +import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; + +const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); +if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory"); + +const primary = await createPostgresAgentRunStore({ connectionString }); +const concurrent = await createPostgresAgentRunStore({ connectionString }); + +try { + await assertNamespaceCreateFence(); + await assertRetentionFenceBlocksClaim(); + console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-run-retention-fence-blocks-claim"] })); +} finally { + await Promise.all([primary.close(), concurrent.close()]); +} + +async function assertNamespaceCreateFence(): Promise { + let firstEntered = (): void => {}; + let releaseFirst = (): void => {}; + const entered = new Promise((resolve) => { firstEntered = resolve; }); + const release = new Promise((resolve) => { releaseFirst = resolve; }); + let secondEntered = false; + const first = primary.withRunnerCreateFence("agentrun-postgres-retention-selftest", async () => { + firstEntered(); + await release; + return "first"; + }); + await entered; + const second = concurrent.withRunnerCreateFence("agentrun-postgres-retention-selftest", async () => { + secondEntered = true; + return "second"; + }); + await delay(50); + assert.equal(secondEntered, false); + releaseFirst(); + assert.deepEqual(await Promise.all([first, second]), ["first", "second"]); +} + +async function assertRetentionFenceBlocksClaim(): Promise { + const run = await primary.createRun({ + tenantId: "selftest", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-selftest" }, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } }, + traceSink: null, + }); + const command = await primary.createCommand(run.id, { type: "turn", payload: { prompt: "postgres retention fence self-test" } }); + let fenceEntered = (): void => {}; + let releaseFence = (): void => {}; + const entered = new Promise((resolve) => { fenceEntered = resolve; }); + const release = new Promise((resolve) => { releaseFence = resolve; }); + const fenced = primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention fence self-test" }, async (facts) => { + assert.equal(facts.run.id, run.id); + assert.equal(facts.command.id, command.id); + fenceEntered(); + await release; + }); + await entered; + let claimSettled = false; + const claim = concurrent.claimRun(run.id, "runner-postgres-retention-race", 60_000) + .then(() => null, (error: unknown) => error) + .finally(() => { claimSettled = true; }); + await delay(50); + assert.equal(claimSettled, false); + releaseFence(); + await fenced; + const claimError = await claim; + assert.ok(claimError instanceof AgentRunError); + assert.equal(claimError.httpStatus, 409); + assert.equal((await primary.getRun(run.id)).status, "failed"); + assert.equal((await primary.getCommand(command.id)).state, "failed"); +} + +async function delay(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +}