diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index 354a2809cb..92b85ab14c 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -93,21 +93,15 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { afterEach(async () => { vi.clearAllMocks(); runningProcesses.clear(); - let idlePolls = 0; - for (let attempt = 0; attempt < 100; attempt += 1) { - const runs = await db - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns); - const hasActiveRun = runs.some((run) => run.status === "queued" || run.status === "running"); - if (!hasActiveRun) { - idlePolls += 1; - if (idlePolls >= 3) break; - } else { - idlePolls = 0; - } - await new Promise((resolve) => setTimeout(resolve, 50)); - } - await new Promise((resolve) => setTimeout(resolve, 50)); + // reconcileIssueGraphLiveness heals dependency wakes by enqueuing an + // on-demand wake, which dispatches a heartbeat run fire-and-forget (see + // startNextQueuedRunForAgent → executeRun in the heartbeat service). That + // background run keeps writing rows (workspace_operations, heartbeat_run_events) + // after the awaited call resolves. Deterministically await those in-flight + // executions before clearing tables — otherwise an escaping heartbeat_run_events + // insert can land between the events delete and the heartbeat_runs delete and + // trip the run_events → runs foreign key. + await heartbeatService(db).drainActiveRunExecutions(); await db.delete(activityLog); await db.delete(heartbeatRunEvents); await db.delete(costEvents); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6fbe84b699..9276a9ee81 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -554,6 +554,13 @@ const SESSIONED_LOCAL_ADAPTERS = new Set([ // Routes and the scheduler construct separate heartbeatService instances, but // they must agree on in-process adapter executions when reaping stale runs. const activeRunExecutions = new Set(); +// Background heartbeat executions are dispatched fire-and-forget (see +// startNextQueuedRunForAgent), so the promise that resolves once a run's DB +// writes are fully flushed is otherwise unobservable. Track those promises here +// — shared across service instances like activeRunExecutions above — so callers +// that must guarantee no run write is still in flight (graceful shutdown, and +// tests tearing down a shared database) can await drainActiveRunExecutions(). +const activeRunExecutionPromises = new Set>(); const INLINE_BASE64_IMAGE_DATA_RE = /("type":"image","source":\{"type":"base64","data":")([A-Za-z0-9+/=]{1024,})(")/g; type RuntimeConfigSecretResolver = Pick< @@ -11748,14 +11755,34 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (claimedRuns.length === 0) return []; for (const claimedRun of claimedRuns) { - void executeRun(claimedRun.id).catch((err) => { + const execution = executeRun(claimedRun.id).catch((err) => { logger.error({ err, runId: claimedRun.id }, "queued heartbeat execution failed"); }); + // Register the in-flight execution so drainActiveRunExecutions() can await + // it. executeRun resolves only after its finally block finishes flushing + // run rows/events, so awaiting this promise guarantees the run's writes + // have landed before a caller (e.g. a test's afterEach) mutates the DB. + activeRunExecutionPromises.add(execution); + void execution.finally(() => { + activeRunExecutionPromises.delete(execution); + }); } return claimedRuns; }); } + // Await every background heartbeat execution that is currently in flight. A + // draining run can, in its finally block, promote and dispatch the next queued + // run for the same agent — that follow-up execution is registered in the set + // before the parent promise settles, so we loop until the set is empty rather + // than snapshotting once. Callers use this to guarantee no run is still + // writing rows/events (graceful shutdown, deterministic test teardown). + async function drainActiveRunExecutions() { + while (activeRunExecutionPromises.size > 0) { + await Promise.all([...activeRunExecutionPromises]); + } + } + async function executeRun(runId: string) { if ((await getSchedulingSuppression()).suppressed) return; @@ -16912,6 +16939,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // gate on suppression should prefer this over the env-only resolver. resolveSchedulingSuppression: getSchedulingSuppression, drainRunningRunsForShutdown, + drainActiveRunExecutions, promoteDueScheduledRetries, retryScheduledRetryNow,