From d51f42ed648c4711a5e6bc4c41a6d4ff4e6e187b Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 29 Jul 2026 14:23:12 -0700 Subject: [PATCH] test(server): drain heartbeat runs to quiescence in sibling suite teardown (#10464) ## Thinking Path > - Paperclip uses heartbeats to run work. > - Test suites share heartbeat run state during teardown. > - Late heartbeat work can race shared table deletes. > - That race can deadlock or fail foreign key checks. > - The primary suite already uses a drain helper to wait for quiescence. > - This pull request reuses that helper in the sibling suites that share the race. > - The benefit is stable teardown and fewer flake failures. ## Linked Issues or Issue Description No public GitHub issue exists for this change. Refs: #10450 This pull request reuses the quiescence drain from the primary suite. ## What Changed - Added `server/src/__tests__/helpers/drain-heartbeat-runs.ts`. - Reused the shared helper in `low-trust-red-team-routes.test.ts`. - Applied the drain to the eight sibling suites that share the race. - Kept the existing test intent unchanged. ## Verification - `git log --oneline origin/master..origin/test/heartbeat-teardown-quiescence-drain-sweep` - `git diff --stat origin/master...origin/test/heartbeat-teardown-quiescence-drain-sweep` - Existing local test evidence in the handoff shows the primary suite and the guarded suites pass. - The handoff also records a stress loop with no `40P01` or `23503` errors. ## Risks - Low risk. The change touches test teardown only. - The helper waits for active runs to drain. A new real background execution path may need the same guard. ## Model Used OpenAI GPT-5, tool-use and code execution enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used with version and capability details - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either linked existing issues with `Fixes: #` / `Closes #` / `Refs #` or described the issue in-PR following the relevant issue template - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes, or no docs update was required for this test-only change - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...at-accepted-plan-workspace-refresh.test.ts | 23 +++++++--------- .../heartbeat-issue-rewake-throttle.test.ts | 15 +++++++---- ...artbeat-responsible-user-invariant.test.ts | 20 +++++++------- .../heartbeat-retry-scheduling.test.ts | 8 ++++++ ...tbeat-workspace-branch-containment.test.ts | 25 +++++++----------- ...eartbeat-workspace-finalize-branch.test.ts | 19 +++++++------- .../heartbeat-worktree-suppression.test.ts | 9 +++++++ .../__tests__/helpers/drain-heartbeat-runs.ts | 26 +++++++++++++++++++ .../low-trust-red-team-routes.test.ts | 24 +---------------- .../plugin-orchestration-apis.test.ts | 10 +++++++ 10 files changed, 101 insertions(+), 78 deletions(-) create mode 100644 server/src/__tests__/helpers/drain-heartbeat-runs.ts diff --git a/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts b/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts index 5ae4be084a..416577e4db 100644 --- a/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts +++ b/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts @@ -33,6 +33,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { heartbeatService } from "../services/heartbeat.ts"; import { instanceSettingsService } from "../services/instance-settings.ts"; import { issueService } from "../services/issues.ts"; @@ -113,20 +114,14 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { afterEach(async () => { adapterExecute.mockClear(); - 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 >= 5) break; - } else { - idlePolls = 0; - } - await new Promise((resolve) => setTimeout(resolve, 50)); - } + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A wakeup claims a run and dispatches its execution + // fire-and-forget, and that run can dispatch a follow-up wakeup, so a run or + // wakeup can still write heartbeat_runs and issues rows when teardown starts + // and would race the deletes. The shared drain also awaits an in-flight + // wakeup that is still before run registration, which a plain run table + // status poll cannot see. + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); while (tempRoots.length > 0) { const root = tempRoots.pop(); if (root) await rm(root, { recursive: true, force: true }).catch(() => undefined); diff --git a/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts index 2d044de81c..f8bfbb2049 100644 --- a/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts +++ b/server/src/__tests__/heartbeat-issue-rewake-throttle.test.ts @@ -22,6 +22,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.ts"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { runningProcesses } from "../adapters/index.ts"; const mockAdapterExecute = vi.hoisted(() => @@ -69,11 +70,15 @@ describeEmbeddedPostgres("heartbeat issue rewake throttle", () => { afterEach(async () => { runningProcesses.clear(); - for (let attempt = 0; attempt < 100; attempt += 1) { - const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns); - if (!runs.some((run) => run.status === "queued" || run.status === "running")) break; - await new Promise((resolve) => setTimeout(resolve, 50)); - } + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A wakeup claims a run and dispatches its execution + // fire-and-forget, and that run can dispatch a follow-up wakeup, so a run or + // wakeup can still write heartbeat_runs and issues rows when teardown starts + // and would race the deletes (a heartbeat_runs delete deadlocks on the ON + // DELETE SET NULL cascade to issues). The shared drain also awaits an + // in-flight wakeup that is still before run registration, which a plain run + // table status poll cannot see. + await drainHeartbeatRunsToQuiescence(db, heartbeat); // Post-run bookkeeping (run-event records, follow-up wake scheduling) can // still write for a moment after a run reaches a terminal status, so a // single delete sweep can hit a foreign-key violation when a late insert diff --git a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts index 32a1ed3a88..c92977ddb2 100644 --- a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts +++ b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -21,6 +21,7 @@ import { } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.ts"; import { runningProcesses } from "../adapters/index.ts"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => ({ @@ -91,15 +92,14 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { afterEach(async () => { mockAdapterExecute.mockClear(); runningProcesses.clear(); - await new Promise((resolve) => setTimeout(resolve, 500)); - for (let attempt = 0; attempt < 40; attempt += 1) { - const activeRuns = await db - .select() - .from(heartbeatRuns) - .where(inArray(heartbeatRuns.status, ["queued", "running"])); - if (activeRuns.length === 0) break; - await new Promise((resolve) => setTimeout(resolve, 50)); - } + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A wakeup claims a run and dispatches its execution + // fire-and-forget, and that run can dispatch a follow-up wakeup, so a run or + // wakeup can still write heartbeat_runs and issues rows when teardown starts + // and would race the deletes. The shared drain also awaits an in-flight + // wakeup that is still before run registration, which a plain run table + // status poll cannot see. + await drainHeartbeatRunsToQuiescence(db, heartbeat); await db.delete(issueComments); await db.delete(activityLog); await deleteHeartbeatRunsAfterEvents(db); diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 9d8a1cbe0e..bc04d27db2 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -22,6 +22,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; import { BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS, @@ -91,6 +92,13 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { }, 20_000); afterEach(async () => { + // Await every in-flight background heartbeat run to quiescence before the + // cleanup deletes. heartbeat.invoke claims a run and dispatches its + // execution fire-and-forget, and that run can schedule a follow-up retry + // wakeup, so a run or wakeup can still write heartbeat_runs and issues rows + // when teardown starts. The cleanup deletes issues before heartbeat_runs, so + // a late write races the deletes and can deadlock or break a foreign key. + await drainHeartbeatRunsToQuiescence(db, heartbeat); await cleanupRetryFixture(); }); diff --git a/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts b/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts index 8e1b942679..ed899e0dfc 100644 --- a/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts +++ b/server/src/__tests__/heartbeat-workspace-branch-containment.test.ts @@ -35,6 +35,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { heartbeatService } from "../services/heartbeat.ts"; import { instanceSettingsService } from "../services/instance-settings.ts"; import { @@ -170,21 +171,6 @@ async function waitForRunToFinish(heartbeat: Heartbeat, runId: string, timeoutMs return heartbeat.getRun(runId); } -async function waitForHeartbeatIdle(db: Db, timeoutMs = 5_000) { - const deadline = Date.now() + timeoutMs; - let idleSince: number | null = null; - while (Date.now() < deadline) { - const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns); - if (!runs.some((run) => run.status === "queued" || run.status === "running")) { - idleSince ??= Date.now(); - if (Date.now() - idleSince >= 250) return; - } else { - idleSince = null; - } - await new Promise((resolve) => setTimeout(resolve, 50)); - } -} - async function deleteHeartbeatRunsForCleanup(db: Db) { let lastError: unknown = null; for (let attempt = 0; attempt < 5; attempt += 1) { @@ -875,7 +861,14 @@ describeEmbeddedPostgres("heartbeat workspace branch containment", () => { }, 20_000); afterEach(async () => { - await waitForHeartbeatIdle(db); + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. resumeQueuedRuns claims a run and dispatches its execution + // fire-and-forget, and the containment path can dispatch a follow-up + // recovery wakeup, so a run or wakeup can still write heartbeat_runs and + // issues rows when teardown starts. The shared drain also awaits an + // in-flight wakeup that is still before run registration, which a plain run + // table status poll cannot see. + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); adapterExecute.mockReset(); adapterExecute.mockImplementation(async () => ({ exitCode: 0, diff --git a/server/src/__tests__/heartbeat-workspace-finalize-branch.test.ts b/server/src/__tests__/heartbeat-workspace-finalize-branch.test.ts index 419284e140..2cb8a2404a 100644 --- a/server/src/__tests__/heartbeat-workspace-finalize-branch.test.ts +++ b/server/src/__tests__/heartbeat-workspace-finalize-branch.test.ts @@ -34,6 +34,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { heartbeatService } from "../services/heartbeat.ts"; import { instanceSettingsService } from "../services/instance-settings.ts"; @@ -102,15 +103,6 @@ async function waitForRunToFinish(heartbeat: Heartbeat, runId: string, timeoutMs return heartbeat.getRun(runId); } -async function waitForHeartbeatIdle(db: Db, timeoutMs = 5_000) { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns); - if (!runs.some((run) => run.status === "queued" || run.status === "running")) return; - await new Promise((resolve) => setTimeout(resolve, 50)); - } -} - async function waitForRuntimeStateLastRun(db: Db, agentId: string, runId: string, timeoutMs = 5_000) { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { @@ -275,7 +267,14 @@ describeEmbeddedPostgres("heartbeat workspace finalization branch guard", () => }, 20_000); afterEach(async () => { - await waitForHeartbeatIdle(db); + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A wakeup claims a run and dispatches its execution + // fire-and-forget, and finalization success can dispatch a follow-up + // wakeup, so a run or wakeup can still write heartbeat_runs and issues rows + // when teardown starts. The shared drain also awaits an in-flight wakeup + // that is still before run registration, which a plain run table status + // poll cannot see. + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); adapterExecute.mockReset(); adapterExecute.mockImplementation(async () => ({ exitCode: 0, diff --git a/server/src/__tests__/heartbeat-worktree-suppression.test.ts b/server/src/__tests__/heartbeat-worktree-suppression.test.ts index b1998f9584..1e3a6c5f7c 100644 --- a/server/src/__tests__/heartbeat-worktree-suppression.test.ts +++ b/server/src/__tests__/heartbeat-worktree-suppression.test.ts @@ -22,6 +22,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { heartbeatService, resolveHeartbeatSchedulingSuppression } from "../services/heartbeat.ts"; import { instanceSettingsService } from "../services/instance-settings.ts"; @@ -66,6 +67,14 @@ describeEmbeddedPostgres("heartbeat worktree suppression", () => { }, 20_000); afterEach(async () => { + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A live wakeup claims a run and dispatches its execution + // fire-and-forget, so a run or wakeup can still write heartbeat_runs and + // issues rows when teardown starts and would race the deletes. The heartbeat + // service tracks in-flight run and wakeup promises in module state shared + // across service instances, so a fresh instance here drains the runs the + // per-test instances dispatched. + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); await db.delete(issueComments); await db.delete(issueDocuments); await db.delete(documentRevisions); diff --git a/server/src/__tests__/helpers/drain-heartbeat-runs.ts b/server/src/__tests__/helpers/drain-heartbeat-runs.ts new file mode 100644 index 0000000000..132647c20a --- /dev/null +++ b/server/src/__tests__/helpers/drain-heartbeat-runs.ts @@ -0,0 +1,26 @@ +import { createDb, heartbeatRuns } from "@paperclipai/db"; +import type { heartbeatService } from "../../services/heartbeat.js"; + +type Db = ReturnType; +type Heartbeat = ReturnType; + +// Await every background heartbeat run until the run table is quiescent. A route +// dispatches a wakeup fire-and-forget (void heartbeat.wakeup(...) in +// routes/issues.ts). Such a wakeup, or a run it dispatches, can write issues, +// issue_comments, and heartbeat_runs rows during teardown and race the deletes +// in a suite afterEach or afterAll (a heartbeat_runs delete deadlocks on the ON +// DELETE SET NULL cascade to issues; an issue_comments insert breaks the later +// delete of issues). drainActiveRunExecutions() awaits both in-flight wakeup +// promises and in-flight run executions, so it also waits for a wakeup that is +// still before run registration. Re-check the run table after the drain as a +// backstop, and give a late run a macrotask before the next attempt, until no +// run is queued or running. +export async function drainHeartbeatRunsToQuiescence(db: Db, heartbeat: Heartbeat) { + for (let attempt = 0; attempt < 50; attempt += 1) { + await heartbeat.drainActiveRunExecutions(); + const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns); + const hasPending = runs.some((run) => run.status === "queued" || run.status === "running"); + if (!hasPending) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} diff --git a/server/src/__tests__/low-trust-red-team-routes.test.ts b/server/src/__tests__/low-trust-red-team-routes.test.ts index b6028d9bd4..10fe7f96c7 100644 --- a/server/src/__tests__/low-trust-red-team-routes.test.ts +++ b/server/src/__tests__/low-trust-red-team-routes.test.ts @@ -41,6 +41,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { parseWakePayloadFromMessage } from "./helpers/wake-message.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { errorHandler } from "../middleware/index.js"; import { agentRoutes } from "../routes/agents.js"; import { issueRoutes } from "../routes/issues.js"; @@ -79,29 +80,6 @@ function isHeartbeatCleanupFkError(error: unknown) { ); } -// Await every background heartbeat run until the run table is quiescent. A route -// dispatches a wakeup fire-and-forget (void heartbeat.wakeup(...) in -// routes/issues.ts). Such a wakeup, or a run it dispatches, can write issues, -// issue_comments, and heartbeat_runs rows during teardown and race the deletes -// below (a heartbeat_runs delete deadlocks on the ON DELETE SET NULL cascade to -// issues; an issue_comments insert breaks the later delete of issues). -// drainActiveRunExecutions() awaits both in-flight wakeup promises and in-flight -// run executions, so it now also waits for a wakeup that is still before run -// registration. Re-check the run table after the drain as a backstop, and give a -// late run a macrotask before the next attempt, until no run is queued or running. -async function drainHeartbeatRunsToQuiescence( - db: Db, - heartbeat: ReturnType, -) { - for (let attempt = 0; attempt < 50; attempt += 1) { - await heartbeat.drainActiveRunExecutions(); - const runs = await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns); - const hasPending = runs.some((run) => run.status === "queued" || run.status === "running"); - if (!hasPending) return; - await new Promise((resolve) => setTimeout(resolve, 10)); - } -} - async function deleteHeartbeatRunsAndWakeupsAfterActivityLogDrains(db: Db) { for (let attempt = 0; attempt < 10; attempt += 1) { await db.delete(heartbeatRunEvents); diff --git a/server/src/__tests__/plugin-orchestration-apis.test.ts b/server/src/__tests__/plugin-orchestration-apis.test.ts index 1f3f2e0658..e7226267b1 100644 --- a/server/src/__tests__/plugin-orchestration-apis.test.ts +++ b/server/src/__tests__/plugin-orchestration-apis.test.ts @@ -31,6 +31,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { buildHostServices } from "../services/plugin-host-services.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -94,6 +96,14 @@ describeEmbeddedPostgres("plugin orchestration APIs", () => { afterEach(async () => { await Promise.all(tempRoots.map((root) => fs.rm(root, { recursive: true, force: true }))); tempRoots.length = 0; + // Await every in-flight background heartbeat run to quiescence before the + // deletes below. A createComment-triggered wakeup dispatches its run + // fire-and-forget (void heartbeat.wakeup(...)), so a run or wakeup can still + // write heartbeat_runs and issues rows when teardown starts and would race + // the deletes. The heartbeat service tracks in-flight run and wakeup + // promises in module state shared across service instances, so a fresh + // instance here drains the runs the per-test host services dispatched. + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); await db.delete(costEvents); await deleteHeartbeatRunsWithDependents(); await db.delete(agentWakeupRequests);