diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index ca415a5..6ae8b93 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -906,8 +906,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( 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 facts = { run, command, commands, events: eventRows.rows.map(eventFromRow) }; + if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) return await operation(facts); 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]); @@ -921,7 +921,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( 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; + return await operation(facts); }); } diff --git a/src/mgr/runner-reconciler.ts b/src/mgr/runner-reconciler.ts index 1d6721e..5e19c51 100644 --- a/src/mgr/runner-reconciler.ts +++ b/src/mgr/runner-reconciler.ts @@ -6,18 +6,24 @@ import { emitAgentRunOtelSpan } from "../common/otel-trace.js"; import { nowIso, stableHash } from "../common/validation.js"; import type { AgentRunStore } from "./store.js"; import { isTerminalCommandState, isTerminalRunStatus } from "./store.js"; +import { reconcileStalePendingRunners, type RunnerRetentionOptions } from "./runner-retention.js"; export interface RunnerReconcilerOptions { store: AgentRunStore; namespace?: string; kubectlCommand?: string; limit?: number; + intervalMs?: number; + nextReconcileAt?: string | null; + retention?: RunnerRetentionOptions; } export interface RunnerReconcilerLoopOptions extends RunnerReconcilerOptions { batchSize: number; intervalMs: number; onError?: (error: unknown) => void; + onResult?: (result: JsonRecord) => void; + onSchedule?: (nextReconcileAt: string | null) => void; } interface K8sObject { @@ -68,7 +74,9 @@ interface TerminalOutboxSearch { } export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): Promise { + const startedAt = nowIso(); const limit = clampLimit(input.limit ?? 20); + const stalePendingRetention = await reconcileStalePendingWithinRunnerLoop(input, limit); const jobs = await input.store.listRunnerJobsForReconciliation(limit); const items: JsonRecord[] = []; let observeFailedCount = 0; @@ -97,19 +105,76 @@ export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): P }); } + const reconciledAt = nowIso(); + const nextReconcileAt = input.nextReconcileAt !== undefined + ? input.nextReconcileAt + : input.intervalMs && input.intervalMs > 0 ? new Date(Date.parse(reconciledAt) + input.intervalMs).toISOString() : null; return { action: "runner-job-reconcile", - reconciledAt: nowIso(), + startedAt, + reconciledAt, + nextReconcileAt, limit, scannedCount: jobs.length, updatedCount: jobs.length, observeFailedCount, runClosureCount, + stalePendingRetention, items, valuesPrinted: false, }; } +async function reconcileStalePendingWithinRunnerLoop(input: RunnerReconcilerOptions, limit: number): Promise { + if (!input.retention) { + return { + enabled: false, + action: "runner-stale-pending-reconcile", + namespace: input.namespace ?? null, + reason: "runner-retention-disabled", + valuesPrinted: false, + }; + } + const options: RunnerRetentionOptions = { + ...input.retention, + ...(input.namespace ? { namespace: input.namespace } : {}), + ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}), + }; + try { + return await reconcileStalePendingRunners({ store: input.store, options, limit }); + } catch (error) { + return { + enabled: true, + action: "runner-stale-pending-reconcile", + namespace: options.namespace, + stalePendingMaxAgeMs: options.stalePendingMaxAgeMs, + state: "failed-closed", + reason: "runner-stale-pending-observation-failed", + error: reconcilerErrorSummary(error), + valuesPrinted: false, + }; + } +} + +function reconcilerErrorSummary(error: unknown): JsonRecord { + if (error instanceof AgentRunError) { + return redactJson({ + failureKind: error.failureKind, + httpStatus: error.httpStatus, + message: redactText(error.message).slice(0, 300), + details: error.details, + valuesPrinted: false, + }) as JsonRecord; + } + return { + failureKind: "infra-failed", + httpStatus: 500, + message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 300), + details: null, + valuesPrinted: false, + }; +} + async function recoverTerminalOutboxIfNeeded(store: AgentRunStore, job: RunnerJobRecord, observation: JsonRecord, input: { namespace?: string; kubectlCommand?: string }): Promise { const phase = stringValue(observation.observedRunnerPhase); if (phase !== "k8s:succeeded" && phase !== "k8s:failed") return null; @@ -273,23 +338,37 @@ export function startRunnerJobReconciler(input: RunnerReconcilerLoopOptions): () const intervalMs = Math.max(1_000, input.intervalMs); let stopped = false; let running = false; + let timer: NodeJS.Timeout | null = null; const tick = async (): Promise => { if (stopped || running) return; running = true; + input.onSchedule?.(null); + let result: JsonRecord | null = null; try { - await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) }); + result = await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}), ...(input.retention ? { retention: input.retention } : {}) }); } catch (error) { input.onError?.(error); } finally { running = false; } + if (stopped) { + input.onSchedule?.(null); + return; + } + const nextReconcileAt = new Date(Date.now() + intervalMs).toISOString(); + input.onSchedule?.(nextReconcileAt); + if (result) { + result.nextReconcileAt = nextReconcileAt; + input.onResult?.(result); + } + timer = setTimeout(() => { void tick(); }, intervalMs); + timer.unref?.(); }; - const timer = setInterval(() => { void tick(); }, intervalMs); - timer.unref?.(); void tick(); return () => { stopped = true; - clearInterval(timer); + if (timer) clearTimeout(timer); + input.onSchedule?.(null); }; } diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index 6ba47d4..dd5017e 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -2,7 +2,7 @@ import { spawn } from "node:child_process"; import { AgentRunError } from "../common/errors.js"; import { redactJson, redactText } from "../common/redaction.js"; import type { CommandRecord, JsonRecord, JsonValue, RunRecord } from "../common/types.js"; -import { stableHash } from "../common/validation.js"; +import { nowIso, stableHash } from "../common/validation.js"; import { isTerminalCommandState, isTerminalRunStatus } from "./store.js"; import type { AgentRunStore } from "./store.js"; import { throwIfAborted, waitForChildProcess } from "./abortable-child.js"; @@ -46,6 +46,30 @@ export interface RunnerRetentionSummary extends JsonRecord { valuesPrinted: false; } +export interface RunnerStalePendingReconcileSummary extends JsonRecord { + enabled: true; + action: "runner-stale-pending-reconcile"; + observedAt: string; + namespace: string; + limit: number; + maxRunners: number; + stalePendingMaxAgeMs: number; + observedRunnerCount: number; + liveRunnerCountBefore: number; + liveRunnerCountAfter: number | null; + stalePendingCandidateCount: number; + protectedActiveRunnerCount: number; + selectedRunnerCount: number; + terminalizedRunCount: number; + terminalizedCommandCount: number; + deletedRunnerJobCount: number; + deletedRunnerPodCount: number; + deletedAssociatedResourceCount: number; + failedRunnerCount: number; + reason: string; + valuesPrinted: false; +} + type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending"; interface RunnerPodObservation extends JsonRecord { @@ -165,29 +189,10 @@ 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.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") { - deletedRunnerJobCount++; - deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName); - } else { - deletedRunnerPodCount++; - } + const retired = await retireSelectedRunner(input.store, input.options, item, `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`); + deletedRunnerJobCount += retired.deletedRunnerJobCount; + deletedRunnerPodCount += retired.deletedRunnerPodCount; + if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName); } const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before; if (after.liveRunnerCount > targetBeforeCreate) { @@ -239,6 +244,100 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu } satisfies RunnerRetentionSummary; } +export async function reconcileStalePendingRunners(input: { store: AgentRunStore; options: RunnerRetentionOptions; limit: number }): Promise { + const limit = clampReconcileLimit(input.limit); + return await input.store.withRunnerCreateFence(input.options.namespace, async () => { + const observedAt = nowIso(); + const before = await runnerSnapshot(input.store, input.options); + const resources = [...before.jobs, ...before.pods]; + const stalePendingCandidates = resources + .filter((item) => !item.protectedActive && item.candidateKind === "stale-pending") + .sort(compareCandidates); + const protectedActiveRunners = resources.filter((item) => item.protectedActive); + const selected = stalePendingCandidates.slice(0, limit); + const outcomes: JsonRecord[] = []; + let terminalizedRunCount = 0; + let terminalizedCommandCount = 0; + let deletedRunnerJobCount = 0; + let deletedRunnerPodCount = 0; + let deletedAssociatedResourceCount = 0; + let failedRunnerCount = 0; + + for (const item of selected) { + let retired: RetiredRunnerResource | null = null; + try { + retired = await retireSelectedRunner(input.store, input.options, item, `runner reconciler retired stale-pending Kubernetes ${item.resourceKind}/${item.name}`); + terminalizedRunCount += item.candidateKind === "stale-pending" ? 1 : 0; + terminalizedCommandCount += item.candidateKind === "stale-pending" ? 1 : 0; + deletedRunnerJobCount += retired.deletedRunnerJobCount; + deletedRunnerPodCount += retired.deletedRunnerPodCount; + const deletedAssociatedForRunner = item.resourceKind === "job" ? await deleteAssociatedResources(input.options, item.jobName) : 0; + deletedAssociatedResourceCount += deletedAssociatedForRunner; + outcomes.push({ ...resourceIdentitySummary(item), state: "retired", terminalized: item.candidateKind === "stale-pending", ...retired, deletedAssociatedResourceCount: deletedAssociatedForRunner, valuesPrinted: false }); + } catch (error) { + failedRunnerCount++; + outcomes.push({ ...resourceIdentitySummary(item), state: retired ? "retired-associated-cleanup-failed" : "failed-closed", coreRetired: retired !== null, ...(retired ?? {}), error: retentionErrorSummary(error), valuesPrinted: false }); + } + } + + let after: Snapshot | null = before; + let postObservationError: JsonRecord | null = null; + if (selected.length > 0) { + try { + after = await runnerSnapshot(input.store, input.options); + } catch (error) { + after = null; + postObservationError = retentionErrorSummary(error); + } + } + const candidateSummaries = boundedResourceSummaries(stalePendingCandidates); + const protectedSummaries = boundedResourceSummaries(protectedActiveRunners); + const selectedSummaries = boundedResourceSummaries(selected); + const boundedOutcomes = boundedJsonRecords(outcomes); + return { + enabled: true, + action: "runner-stale-pending-reconcile", + observedAt, + namespace: input.options.namespace, + limit, + maxRunners: input.options.maxRunners, + stalePendingMaxAgeMs: input.options.stalePendingMaxAgeMs, + observedRunnerCount: resources.length, + liveRunnerCountBefore: before.liveRunnerCount, + liveRunnerCountAfter: after?.liveRunnerCount ?? null, + runnerJobCountBefore: before.runnerJobCount, + nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount, + orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount, + inactiveCandidateCount: resources.filter((item) => !item.protectedActive).length, + stalePendingCandidateCount: stalePendingCandidates.length, + stalePendingCandidates: candidateSummaries.items, + stalePendingCandidateOmittedCount: candidateSummaries.omittedCount, + protectedActiveRunnerCount: protectedActiveRunners.length, + protectedActiveRunners: protectedSummaries.items, + protectedActiveRunnerOmittedCount: protectedSummaries.omittedCount, + protectedReasonCounts: reasonCounts(protectedActiveRunners.map((item) => item.protectedReason ?? "unknown")), + selectedRunnerCount: selected.length, + selectedRunners: selectedSummaries.items, + selectedRunnerOmittedCount: selectedSummaries.omittedCount, + terminalizedRunCount, + terminalizedCommandCount, + deletedRunnerJobCount, + deletedRunnerPodCount, + deletedAssociatedResourceCount, + failedRunnerCount, + outcomes: boundedOutcomes.items, + outcomeOmittedCount: boundedOutcomes.omittedCount, + state: postObservationError || failedRunnerCount > 0 ? "partial" : "completed", + postObservationState: postObservationError ? "failed" : "completed", + postObservationError, + activeRunRisk: protectedActiveRunners.length > 0, + reason: postObservationError ? "stale-pending-post-observation-failed" : selected.length === 0 ? "no-stale-pending-candidate" : failedRunnerCount > 0 ? "stale-pending-reconcile-partial" : "stale-pending-reconciled", + cleanupPolicy: cleanupPolicySummary(), + valuesPrinted: false, + } satisfies RunnerStalePendingReconcileSummary; + }); +} + async function runnerSnapshot(store: AgentRunStore, options: RunnerRetentionOptions): Promise { const observed = await Promise.allSettled([ kubectlGetList(options, "jobs", labelSelector(options.matchLabels)), @@ -419,6 +518,35 @@ async function activityFor(store: AgentRunStore, input: { return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint); } +interface RetiredRunnerResource extends JsonRecord { + deletedRunnerJobCount: number; + deletedRunnerPodCount: number; +} + +async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, item: RunnerResourceEntry, reason: string): Promise { + await revalidateSelectedRunner(store, 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: options.namespace, resourceKind: item.resourceKind, name: item.name, valuesPrinted: false }, + }); + } + await store.withRunnerRetentionFence({ + runId: item.runId, + commandId: item.commandId, + terminalizeStalePending: item.candidateKind === "stale-pending", + reason, + }, async (facts) => { + const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events); + if (currentDbFingerprint !== item.dbFactsFingerprint) throw databaseFenceRejected(options, item, currentDbFingerprint, facts.run); + await deleteRunnerResourceCas(options, item); + }); + if (item.resourceKind === "job") { + return { deletedRunnerJobCount: 1, deletedRunnerPodCount: 0 }; + } + return { deletedRunnerJobCount: 0, deletedRunnerPodCount: 1 }; +} + 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; @@ -599,13 +727,22 @@ function resourceSummary(item: RunnerResourceEntry): JsonRecord { classificationReason: item.classificationReason, terminal: item.terminal, podPhase: item.podPhase, - podObservations: item.podObservations.slice(0, 5), + podObservations: item.podObservations.slice(0, 5).map(podObservationSummary), podObservationOmittedCount: Math.max(0, item.podObservations.length - 5), lastActiveAt: new Date(item.lastActiveAtMs).toISOString(), valuesPrinted: false, }) as JsonRecord; } +function podObservationSummary(item: RunnerPodObservation): JsonRecord { + return { + ...item, + waiting: item.waiting.slice(0, 5), + waitingOmittedCount: Math.max(0, item.waiting.length - 5), + valuesPrinted: false, + }; +} + function cleanupPolicySummary(): JsonRecord { return { mode: "db-ledger-and-k8s-observation", @@ -874,6 +1011,54 @@ function boundedResourceSummaries(items: RunnerResourceEntry[]): { items: JsonRe return { items: items.slice(0, limit).map(resourceSummary), omittedCount: Math.max(0, items.length - limit) }; } +function boundedJsonRecords(items: JsonRecord[]): { items: JsonRecord[]; omittedCount: number } { + const limit = 20; + return { items: items.slice(0, limit), omittedCount: Math.max(0, items.length - limit) }; +} + +function resourceIdentitySummary(item: RunnerResourceEntry): JsonRecord { + return { + resourceKind: item.resourceKind, + name: item.name, + jobName: item.jobName, + runId: item.runId, + commandId: item.commandId, + runnerId: item.runnerId, + candidateKind: item.candidateKind, + classificationReason: item.classificationReason, + valuesPrinted: false, + }; +} + +function retentionErrorSummary(error: unknown): JsonRecord { + if (error instanceof AgentRunError) { + return redactJson({ + failureKind: error.failureKind, + httpStatus: error.httpStatus, + message: redactText(error.message).slice(0, 300), + details: error.details, + valuesPrinted: false, + }) as JsonRecord; + } + return { + failureKind: "infra-failed", + httpStatus: 500, + message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 300), + details: null, + valuesPrinted: false, + }; +} + +function reasonCounts(reasons: string[]): JsonRecord { + const counts: Record = {}; + for (const reason of reasons) counts[reason] = (counts[reason] ?? 0) + 1; + return Object.fromEntries(Object.entries(counts).sort(([left], [right]) => left.localeCompare(right))) as JsonRecord; +} + +function clampReconcileLimit(limit: number): number { + return Math.max(1, Math.min(Math.floor(limit), 500)); +} + function jsonRecord(value: unknown): JsonRecord | null { return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : null; } diff --git a/src/mgr/server.ts b/src/mgr/server.ts index 2fc39ef..0cc1265 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -115,13 +115,42 @@ function runnerRetentionOptionsForRequest(defaults: RunnerRetentionOptions | und }; } -function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runnerReconcilerOptions"] | undefined, runnerJobDefaults: ManagerServerOptions["runnerJobDefaults"] | undefined): RunnerReconcilerRuntimeOptions { +function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runnerReconcilerOptions"] | undefined, runnerJobDefaults: RunnerJobDefaults): RunnerReconcilerRuntimeOptions { const enabled = defaults?.enabled ?? optionalBooleanEnv("AGENTRUN_MANAGER_RECONCILER_ENABLED", process.env.AGENTRUN_MANAGER_RECONCILER_ENABLED, false); const namespace = defaults?.namespace ?? runnerJobDefaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01"; const intervalMs = defaults?.intervalMs ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS", process.env.AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS) ?? 30_000; const batchSize = defaults?.batchSize ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE", process.env.AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE) ?? 20; const kubectlCommand = defaults?.kubectlCommand ?? runnerJobDefaults?.kubectlCommand; - return { enabled, namespace, intervalMs, batchSize, ...(kubectlCommand ? { kubectlCommand } : {}) }; + return { enabled, namespace, intervalMs, batchSize, nextReconcileAt: null, ...(kubectlCommand ? { kubectlCommand } : {}), ...(runnerJobDefaults.retention ? { retention: runnerJobDefaults.retention } : {}) }; +} + +function runnerReconcilerLogSummary(result: JsonRecord): JsonRecord { + const retention = typeof result.stalePendingRetention === "object" && result.stalePendingRetention !== null && !Array.isArray(result.stalePendingRetention) + ? result.stalePendingRetention as JsonRecord + : {}; + return { + code: "runner-reconciler-completed", + reconciledAt: result.reconciledAt ?? null, + nextReconcileAt: result.nextReconcileAt ?? null, + scannedCount: result.scannedCount ?? 0, + observeFailedCount: result.observeFailedCount ?? 0, + runClosureCount: result.runClosureCount ?? 0, + stalePendingRetention: { + enabled: retention.enabled === true, + state: retention.state ?? "completed", + reason: retention.reason ?? null, + observedRunnerCount: retention.observedRunnerCount ?? 0, + stalePendingCandidateCount: retention.stalePendingCandidateCount ?? 0, + protectedActiveRunnerCount: retention.protectedActiveRunnerCount ?? 0, + protectedReasonCounts: retention.protectedReasonCounts ?? {}, + selectedRunnerCount: retention.selectedRunnerCount ?? 0, + terminalizedRunCount: retention.terminalizedRunCount ?? 0, + deletedRunnerJobCount: retention.deletedRunnerJobCount ?? 0, + failedRunnerCount: retention.failedRunnerCount ?? 0, + valuesPrinted: false, + }, + valuesPrinted: false, + }; } function runnerDispatcherOptionsForRuntime(defaults: ManagerServerOptions["runnerDispatcherOptions"] | undefined): RunnerDispatcherOptions { @@ -201,7 +230,9 @@ interface RunnerReconcilerRuntimeOptions { namespace: string; intervalMs: number; batchSize: number; + nextReconcileAt: string | null; kubectlCommand?: string; + retention?: RunnerRetentionOptions; } export interface StartedManagerServer { @@ -217,15 +248,25 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr const sessionPvcDefaults = options.sessionPvcOptions; const providerProfileDefaults = options.providerProfileOptions; const toolCredentialDefaults = options.toolCredentialOptions; - const runnerReconcilerOptions = runnerReconcilerOptionsForRuntime(options.runnerReconcilerOptions, runnerJobDefaults); + const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit); + const runnerReconcilerOptions = runnerReconcilerOptionsForRuntime(options.runnerReconcilerOptions, resolvedRunnerJobDefaults); const runnerDispatcherOptions = runnerDispatcherOptionsForRuntime(options.runnerDispatcherOptions); const kafkaOutboxRelayOptions = kafkaOutboxRelayOptionsForRuntime(options.kafkaOutboxRelayOptions); - const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit); assertRunnerRuntimeConfigured(runnerJobDefaults); const aipodSpecDir = options.aipodSpecDir ?? process.env.AGENTRUN_AIPOD_SPEC_DIR; const auth = options.auth ?? managerServerAuthConfigFromEnv(); const authSummary = managerAuthSummary(auth); - const stopRunnerReconciler = runnerReconcilerOptions.enabled ? startRunnerJobReconciler({ store, namespace: runnerReconcilerOptions.namespace, intervalMs: runnerReconcilerOptions.intervalMs, batchSize: runnerReconcilerOptions.batchSize, ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}), onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; + const stopRunnerReconciler = runnerReconcilerOptions.enabled ? startRunnerJobReconciler({ + store, + namespace: runnerReconcilerOptions.namespace, + intervalMs: runnerReconcilerOptions.intervalMs, + batchSize: runnerReconcilerOptions.batchSize, + ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}), + ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}), + onSchedule: (nextReconcileAt) => { runnerReconcilerOptions.nextReconcileAt = nextReconcileAt; }, + onResult: (result) => console.info(JSON.stringify(runnerReconcilerLogSummary(result))), + onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })), + }) : null; const stopRunnerDispatcher = runnerDispatcherOptions.enabled ? startRunnerDispatcher({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions, onError: (error) => console.warn(JSON.stringify({ code: "runner-dispatcher-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; const stopKafkaOutboxRelay = kafkaOutboxRelayOptions.enabled ? startKafkaEventOutboxRelay({ store, options: kafkaOutboxRelayOptions, onError: (error) => console.warn(JSON.stringify({ code: "kafka-outbox-relay-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null; const server = createServer(async (req, res) => { @@ -869,7 +910,14 @@ async function route({ method, url, body, store, sourceCommit, authSummary, runn const record = body === null ? {} : asRecord(body, "runnerReconciler"); const limit = Math.max(1, Math.min(numberField(record, "limit", runnerReconcilerOptions?.batchSize ?? 20), 500)); const namespace = optionalString(record.namespace) ?? runnerReconcilerOptions?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01"; - return await reconcileRunnerJobsOnce({ store, namespace, limit, ...(runnerReconcilerOptions?.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}) }) as JsonValue; + return await reconcileRunnerJobsOnce({ + store, + namespace, + limit, + nextReconcileAt: runnerReconcilerOptions.nextReconcileAt, + ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}), + ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}), + }) as JsonValue; } if (method === "POST" && path === "/api/v1/reconciler/runner-dispatch") { return await dispatchRunnerIntentsOnce({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions }) as JsonValue; diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index 87d3097..36341e4 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/src/selftest/cases/22-runner-retention-stale-pending.ts @@ -3,7 +3,10 @@ import { chmod, readFile, writeFile } from "node:fs/promises"; import path from "node:path"; import { AgentRunError } from "../../common/errors.js"; import type { CommandRecord, JsonRecord, RunEvent, RunRecord } from "../../common/types.js"; +import { ManagerClient } from "../../mgr/client.js"; +import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "../../mgr/runner-reconciler.js"; import { enforceRunnerRetentionBeforeCreate, type RunnerRetentionOptions, type RunnerRetentionSummary } from "../../mgr/runner-retention.js"; +import { startManagerServer } from "../../mgr/server.js"; import { MemoryAgentRunStore, type RunnerRetentionFenceFacts, type RunnerRetentionFenceInput } from "../../mgr/store.js"; import type { SelfTestCase } from "../harness.js"; @@ -41,6 +44,7 @@ interface RetentionFixture { class RetentionFactStore extends MemoryAgentRunStore { private readonly facts: RetentionFact[]; private beforeFence: (() => void) | null = null; + terminalizationCount = 0; constructor(facts: RetentionFact[]) { super(); @@ -90,6 +94,7 @@ class RetentionFactStore extends MemoryAgentRunStore { 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) { + this.terminalizationCount++; 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 }; @@ -119,6 +124,12 @@ const selfTest: SelfTestCase = async (context) => { await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertMemoryDbFenceRejectsConcurrentClaim(); await assertRunnerCreateFenceSerializesNamespace(); + await assertPeriodicReconcilerCleansAtCapacity({ fixturePath, statePath, kubectlCommand }); + await assertPeriodicReconcilerProtectsUnsafeRunners({ fixturePath, statePath, kubectlCommand }); + await assertPeriodicReconcilerCasConflictFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertConcurrentPeriodicReconcileIsSerialized({ fixturePath, statePath, kubectlCommand }); + await assertControlledReconcileApiExposesRetention({ fixturePath, statePath, kubectlCommand }); + await assertAutomaticReconcilerReportsNextSchedule(); } finally { restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath); restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath); @@ -149,6 +160,12 @@ const selfTest: SelfTestCase = async (context) => { "runner-delete-postcondition-checked", "memory-db-fence-rejects-concurrent-claim", "namespace-runner-create-fence-prevents-capacity-overshoot", + "periodic-reconciler-cleans-stale-pending-at-capacity", + "periodic-reconciler-protects-unsafe-runners", + "periodic-reconciler-cas-conflict-fails-closed", + "periodic-reconciler-namespace-fence-prevents-duplicate-terminalization", + "controlled-reconcile-api-exposes-bounded-retention-result", + "automatic-reconciler-reports-next-schedule", ], }; }; @@ -420,6 +437,226 @@ async function assertRunnerCreateFenceSerializesNamespace(): Promise { assert.equal(liveRunnerCount, maxRunners); } +async function assertPeriodicReconcilerCleansAtCapacity(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const now = Date.now(); + const facts = Array.from({ length: 20 }, (_, index) => pendingFact(index + 300, new Date(now - (60 - index) * 60_000).toISOString())); + await writeFixture(input, { + jobs: facts.map((fact, index) => runnerJob(fact, index + 300)), + pods: facts.map((fact, index) => runnerPod(fact, index + 300, "pending")), + events: facts.map((fact, index) => failedMountEvent(fact, index + 300)), + bumpJobResourceVersionAfterPodDelete: true, + bumpedJobResourceVersion: "2", + }); + const store = new RetentionFactStore(facts); + const summary = await reconcileRunnerJobsOnce({ + store, + namespace: "agentrun-nc01-v02", + kubectlCommand: input.kubectlCommand, + retention: retentionOptions(input.kubectlCommand, 20), + limit: 3, + intervalMs: 30_000, + }); + const retention = summary.stalePendingRetention as JsonRecord; + assert.equal(retention.liveRunnerCountBefore, 20); + assert.equal(retention.liveRunnerCountAfter, 17); + assert.equal(retention.stalePendingCandidateCount, 20); + assert.equal(retention.selectedRunnerCount, 3); + assert.equal(retention.terminalizedRunCount, 3); + assert.equal(retention.terminalizedCommandCount, 3); + assert.equal(retention.deletedRunnerJobCount, 3); + assert.equal(retention.failedRunnerCount, 0); + assert.equal(retention.reason, "stale-pending-reconciled"); + assert.equal(store.terminalizationCount, 3); + for (const fact of facts.slice(0, 3)) { + assert.equal(store.getRun(fact.run.id).status, "failed"); + assert.equal(store.getCommand(fact.command.id).state, "failed"); + } + assert.equal(store.getRun((facts[3] as RetentionFact).run.id).status, "pending"); + const reconciledAt = Date.parse(String(summary.reconciledAt)); + const nextReconcileAt = Date.parse(String(summary.nextReconcileAt)); + assert.equal(nextReconcileAt - reconciledAt, 30_000); + assert.deepEqual( + deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), + ["agentrun-v02-runner-stale-300", "agentrun-v02-runner-stale-301", "agentrun-v02-runner-stale-302"], + ); +} + +async function assertPeriodicReconcilerProtectsUnsafeRunners(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const oldAt = new Date(Date.now() - 60 * 60_000).toISOString(); + const freshAt = new Date(Date.now() - 60_000).toISOString(); + const fresh = pendingFact(320, freshAt); + const running = pendingFact(321, oldAt); + const ready = pendingFact(322, oldAt); + const claimed = pendingFact(323, oldAt, "claimed"); + const toolExecuting = pendingFact(324, oldAt, "tool-executing"); + const unavailable = pendingFact(325, oldAt); + const allFacts = [fresh, running, ready, claimed, toolExecuting, unavailable]; + await writeFixture(input, { + jobs: allFacts.map((fact, index) => runnerJob(fact, index + 320)), + pods: [ + runnerPod(fresh, 320, "pending"), + runnerPod(running, 321, "running"), + runnerPod(ready, 322, "ready"), + runnerPod(claimed, 323, "pending"), + runnerPod(toolExecuting, 324, "failed-config"), + runnerPod(unavailable, 325, "pending"), + ], + events: allFacts.map((fact, index) => failedMountEvent(fact, index + 320)), + }); + const summary = await reconcileRunnerJobsOnce({ + store: new RetentionFactStore([fresh, running, ready, claimed, toolExecuting]), + namespace: "agentrun-nc01-v02", + kubectlCommand: input.kubectlCommand, + retention: retentionOptions(input.kubectlCommand, 20), + limit: 20, + }); + const retention = summary.stalePendingRetention as JsonRecord; + assert.equal(retention.stalePendingCandidateCount, 0); + assert.equal(retention.selectedRunnerCount, 0); + assert.equal(retention.protectedActiveRunnerCount, 6); + const reasonCounts = retention.protectedReasonCounts as JsonRecord; + assert.equal(reasonCounts["stale-pending-window-active"], 1); + assert.equal(reasonCounts["runner-pod-started-running-or-ready"], 2); + assert.equal(reasonCounts["fresh-active-run"], 1); + assert.equal(reasonCounts["business-progress-observed-or-unverifiable"], 1); + assert.equal(reasonCounts["db-facts-unavailable"], 1); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); +} + +async function assertPeriodicReconcilerCasConflictFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(330, new Date(Date.now() - 60 * 60_000).toISOString()); + const racedPod = runnerPod(fact, 330, "running"); + (racedPod.metadata as JsonRecord).resourceVersion = "2"; + await writeFixture(input, { + jobs: [runnerJob(fact, 330)], + pods: [runnerPod(fact, 330, "pending")], + events: [failedMountEvent(fact, 330)], + raceOnCasDelete: true, + racePods: [racedPod], + }); + const store = new RetentionFactStore([fact]); + const summary = await reconcileRunnerJobsOnce({ + store, + namespace: "agentrun-nc01-v02", + kubectlCommand: input.kubectlCommand, + retention: retentionOptions(input.kubectlCommand, 20), + limit: 1, + }); + const retention = summary.stalePendingRetention as JsonRecord; + assert.equal(retention.selectedRunnerCount, 1); + assert.equal(retention.failedRunnerCount, 1); + assert.equal(retention.terminalizedRunCount, 0); + assert.equal(retention.deletedRunnerJobCount, 0); + assert.equal(store.terminalizationCount, 0); + assert.equal(store.getRun(fact.run.id).status, "pending"); + assert.equal(store.getCommand(fact.command.id).state, "pending"); + const outcome = ((retention.outcomes as JsonRecord[])[0] ?? {}) as JsonRecord; + assert.equal(outcome.state, "failed-closed"); + const error = (outcome.error ?? {}) as JsonRecord; + const details = (error.details ?? {}) as JsonRecord; + assert.equal(details.reason, "runner-retention-k8s-cas-rejected"); + assert.equal(details.changeReason, "object-identity-or-version-changed"); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0); +} + +async function assertConcurrentPeriodicReconcileIsSerialized(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(340, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 340)], + pods: [runnerPod(fact, 340, "pending")], + events: [failedMountEvent(fact, 340)], + bumpJobResourceVersionAfterPodDelete: true, + bumpedJobResourceVersion: "2", + }); + const store = new RetentionFactStore([fact]); + const options = { + store, + namespace: "agentrun-nc01-v02", + kubectlCommand: input.kubectlCommand, + retention: retentionOptions(input.kubectlCommand, 20), + limit: 1, + }; + const [first, second] = await Promise.all([reconcileRunnerJobsOnce(options), reconcileRunnerJobsOnce(options)]); + const summaries = [first.stalePendingRetention as JsonRecord, second.stalePendingRetention as JsonRecord]; + assert.equal(summaries.reduce((total, item) => total + Number(item.selectedRunnerCount ?? 0), 0), 1); + assert.equal(summaries.reduce((total, item) => total + Number(item.terminalizedRunCount ?? 0), 0), 1); + assert.equal(store.terminalizationCount, 1); + assert.equal(store.getRun(fact.run.id).status, "failed"); + assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-340"]); +} + +async function assertControlledReconcileApiExposesRetention(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(350, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 350)], + pods: [runnerPod(fact, 350, "pending")], + events: [failedMountEvent(fact, 350)], + bumpJobResourceVersionAfterPodDelete: true, + bumpedJobResourceVersion: "2", + }); + const store = new RetentionFactStore([fact]); + const retention = retentionOptions(input.kubectlCommand, 20); + const server = await startManagerServer({ + port: 0, + host: "127.0.0.1", + sourceCommit: "self-test-282", + store, + runnerJobDefaults: { + namespace: "agentrun-nc01-v02", + managerUrl: "http://agentrun-mgr.agentrun-nc01-v02.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-v02-runner-api-key", key: "api-key" }, + image: "registry.example.invalid/agentrun@sha256:1111111111111111111111111111111111111111111111111111111111111111", + serviceAccountName: "agentrun-v02-runner", + jobNamePrefix: "agentrun-v02-runner", + lane: "v0.2", + kubectlCommand: input.kubectlCommand, + retention, + }, + runnerReconcilerOptions: { + enabled: false, + namespace: "agentrun-nc01-v02", + kubectlCommand: input.kubectlCommand, + intervalMs: 45_000, + batchSize: 1, + }, + }); + try { + const summary = await new ManagerClient(server.baseUrl).post("/api/v1/reconciler/runner-jobs", { limit: 1, namespace: "agentrun-nc01-v02" }) as JsonRecord; + const cleanup = summary.stalePendingRetention as JsonRecord; + assert.equal(cleanup.enabled, true); + assert.equal(cleanup.selectedRunnerCount, 1); + assert.equal(cleanup.terminalizedRunCount, 1); + assert.equal(cleanup.deletedRunnerJobCount, 1); + assert.equal(cleanup.failedRunnerCount, 0); + assert.equal(cleanup.valuesPrinted, false); + assert.equal(summary.nextReconcileAt, null); + assert.equal(JSON.stringify(summary).includes("abc123-secret-value"), false); + } finally { + await new Promise((resolve) => server.server.close(() => resolve())); + } +} + +async function assertAutomaticReconcilerReportsNextSchedule(): Promise { + let scheduledAt: string | null = null; + let resolveResult = (_result: JsonRecord): void => {}; + const resultPromise = new Promise((resolve) => { resolveResult = resolve; }); + const stop = startRunnerJobReconciler({ + store: new MemoryAgentRunStore(), + batchSize: 1, + intervalMs: 1_000, + onSchedule: (nextReconcileAt) => { scheduledAt = nextReconcileAt; }, + onResult: resolveResult, + }); + try { + const result = await resultPromise; + assert.equal(typeof result.nextReconcileAt, "string"); + assert.equal(result.nextReconcileAt, scheduledAt); + assert.ok(Date.parse(String(result.nextReconcileAt)) > Date.parse(String(result.reconciledAt))); + } finally { + stop(); + } +} + async function assertProtected( input: { fixturePath: string; statePath: string; kubectlCommand: string }, facts: RetentionFact[], diff --git a/src/selftest/integration/postgres-runner-retention-fence.ts b/src/selftest/integration/postgres-runner-retention-fence.ts index 52e2872..88915ca 100644 --- a/src/selftest/integration/postgres-runner-retention-fence.ts +++ b/src/selftest/integration/postgres-runner-retention-fence.ts @@ -11,7 +11,9 @@ 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"] })); + await assertRetentionTerminalizationIsIdempotent(); + await assertRetentionTerminalizationRollsBackOnCasFailure(); + console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization"] })); } finally { await Promise.all([primary.close(), concurrent.close()]); } @@ -75,6 +77,55 @@ async function assertRetentionFenceBlocksClaim(): Promise { assert.equal((await primary.getCommand(command.id)).state, "failed"); } +async function assertRetentionTerminalizationIsIdempotent(): Promise { + const run = await primary.createRun({ + tenantId: "selftest", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-idempotent" }, + 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 idempotence self-test" } }); + const fenceInput = { runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention idempotence self-test" }; + await primary.withRunnerRetentionFence(fenceInput, async (facts) => { + assert.equal(facts.run.status, "pending"); + }); + await primary.withRunnerRetentionFence(fenceInput, async (facts) => { + assert.equal(facts.run.status, "failed"); + assert.equal(facts.command.state, "failed"); + }); + const events = await primary.listEvents(run.id, 0, 500); + assert.equal(events.filter((event) => event.type === "backend_status" && event.payload.retentionFence === true).length, 1); + assert.equal(events.filter((event) => event.type === "terminal_status" && event.payload.retentionFence === true).length, 1); + assert.equal((await primary.getRun(run.id)).status, "failed"); + assert.equal((await primary.getCommand(command.id)).state, "failed"); +} + +async function assertRetentionTerminalizationRollsBackOnCasFailure(): Promise { + const run = await primary.createRun({ + tenantId: "selftest", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-rollback" }, + 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 rollback self-test" } }); + await assert.rejects( + primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention rollback self-test" }, async () => { + throw new AgentRunError("infra-failed", "simulated Kubernetes CAS conflict", { httpStatus: 409, details: { reason: "runner-retention-k8s-cas-rejected", valuesPrinted: false } }); + }), + (error) => error instanceof AgentRunError && error.details?.reason === "runner-retention-k8s-cas-rejected", + ); + assert.equal((await primary.getRun(run.id)).status, "pending"); + assert.equal((await primary.getCommand(command.id)).state, "pending"); + const events = await primary.listEvents(run.id, 0, 500); + assert.equal(events.filter((event) => event.payload.retentionFence === true).length, 0); +} + async function delay(ms: number): Promise { await new Promise((resolve) => setTimeout(resolve, ms)); }