diff --git a/packages/db/src/heartbeat-runs-issue-lock-release-count-migration.test.ts b/packages/db/src/heartbeat-runs-issue-lock-release-count-migration.test.ts new file mode 100644 index 000000000000..3b150cb19fa0 --- /dev/null +++ b/packages/db/src/heartbeat-runs-issue-lock-release-count-migration.test.ts @@ -0,0 +1,151 @@ +/** + * BLO-22060: heartbeat_runs.issue_lock_release_count is the durable mark that + * makes a stale-lock release stick. The sweep clears the issue row's lock + * fields, so the issue row cannot carry the count; the run must. + * + * Two properties of the migration are load-bearing and asserted here: + * - the column is NOT NULL with a server default of 0, so every insert path + * gets a usable count without having to name the column; + * - pre-existing rows backfill to 0 rather than to the bound, so a run that + * is legitimately mid-flight at deploy time is never stranded. + */ +import { createHash } from "node:crypto"; +import fs from "node:fs"; +import { afterEach, describe, expect, it } from "vitest"; +import postgres from "postgres"; +import { applyPendingMigrations, inspectMigrations } from "./client.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./test-embedded-postgres.js"; + +// Resolved by suffix rather than hardcoded. This migration has already been +// renumbered four times by rebases (0213 -> 0214 -> 0216 -> 0219); each time, a +// stale literal here would fail the suite with an ENOENT that reads as +// unrelated to the rename. +const MIGRATION_SUFFIX = "_heartbeat_runs_issue_lock_release_count.sql"; +const COLUMN_NAME = "issue_lock_release_count"; + +const cleanups: Array<() => Promise> = []; +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +function resolveMigrationFile(): string { + const migrationsDir = new URL("./migrations/", import.meta.url); + const matches = fs + .readdirSync(migrationsDir) + .filter((file) => file.endsWith(MIGRATION_SUFFIX)) + .sort(); + if (matches.length !== 1) { + throw new Error( + `expected exactly one migration ending in ${MIGRATION_SUFFIX}, found ${matches.length}: ${matches.join(", ")}`, + ); + } + return matches[0]!; +} + +const MIGRATION_FILE = resolveMigrationFile(); + +async function migrationHash(): Promise { + const content = await fs.promises.readFile( + new URL(`./migrations/${MIGRATION_FILE}`, import.meta.url), + "utf8", + ); + return createHash("sha256").update(content).digest("hex"); +} + +afterEach(async () => { + while (cleanups.length > 0) await cleanups.pop()?.(); +}, 60_000); + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres issue lock release count migration tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("heartbeat-run issue lock release count migration", () => { + it( + "adds a NOT NULL default-0 counter and backfills pre-existing runs to 0", + async () => { + const database = await startEmbeddedPostgresTestDatabase( + "paperclip-heartbeat-issue-lock-release-count-", + ); + cleanups.push(database.cleanup); + const sql = postgres(database.connectionString, { max: 1, onnotice: () => {} }); + cleanups.push(async () => sql.end()); + + // Rewind to the pre-migration shape, then seed a run that predates it. + // session_replication_role=replica skips the FK checks so the row needs + // no company/agent fixtures. + await sql.unsafe(` + ALTER TABLE heartbeat_runs DROP COLUMN ${COLUMN_NAME}; + SET session_replication_role = replica; + INSERT INTO heartbeat_runs (id, company_id, agent_id, status, context_snapshot) + VALUES ( + '33333333-3333-4333-8333-333333333333', + '11111111-1111-4111-8111-111111111111', + '22222222-2222-4222-8222-222222222222', + 'scheduled_retry', + '{}'::jsonb + ); + SET session_replication_role = origin; + `); + await sql` + DELETE FROM "drizzle"."__drizzle_migrations" + WHERE "hash" = ${await migrationHash()} + `; + + expect(await inspectMigrations(database.connectionString)).toMatchObject({ + status: "needsMigrations", + pendingMigrations: [MIGRATION_FILE], + }); + + await applyPendingMigrations(database.connectionString); + expect((await inspectMigrations(database.connectionString)).status).toBe("upToDate"); + + const columns = await sql< + { data_type: string; is_nullable: string; column_default: string | null }[] + >` + SELECT "data_type", "is_nullable", "column_default" + FROM "information_schema"."columns" + WHERE "table_schema" = 'public' + AND "table_name" = 'heartbeat_runs' + AND "column_name" = ${COLUMN_NAME} + `; + expect(columns).toEqual([ + { data_type: "integer", is_nullable: "NO", column_default: "0" }, + ]); + + // The pre-existing park backfills to 0, not to the bound: it keeps a full + // adoption budget rather than being stranded by the deploy. + const seeded = await sql<{ issue_lock_release_count: number }[]>` + SELECT "issue_lock_release_count" + FROM "heartbeat_runs" + WHERE "id" = '33333333-3333-4333-8333-333333333333' + `; + expect(seeded).toEqual([{ issue_lock_release_count: 0 }]); + + // A fresh insert that never names the column still gets a usable count. + await sql.unsafe(` + SET session_replication_role = replica; + INSERT INTO heartbeat_runs (id, company_id, agent_id, status, context_snapshot) + VALUES ( + '44444444-4444-4444-8444-444444444444', + '11111111-1111-4111-8111-111111111111', + '22222222-2222-4222-8222-222222222222', + 'queued', + '{}'::jsonb + ); + SET session_replication_role = origin; + `); + const inserted = await sql<{ issue_lock_release_count: number }[]>` + SELECT "issue_lock_release_count" + FROM "heartbeat_runs" + WHERE "id" = '44444444-4444-4444-8444-444444444444' + `; + expect(inserted).toEqual([{ issue_lock_release_count: 0 }]); + }, + 60_000, + ); +}); diff --git a/packages/db/src/migrations/0219_heartbeat_runs_issue_lock_release_count.sql b/packages/db/src/migrations/0219_heartbeat_runs_issue_lock_release_count.sql new file mode 100644 index 000000000000..88ce6eda6996 --- /dev/null +++ b/packages/db/src/migrations/0219_heartbeat_runs_issue_lock_release_count.sql @@ -0,0 +1,37 @@ +-- BLO-22060: make a stale-lock release durable so the 6h cap cannot be renewed +-- by the same dead holder. +-- +-- sweepStaleIssueLocks (server/src/services/recovery/service.ts) bounds a +-- pre-claim issue lock at STALE_PRE_CLAIM_ISSUE_LOCK_MS measured from +-- issues.execution_locked_at, and on expiry clears execution_run_id / +-- execution_locked_at *without* cancelling the run — a `scheduled_retry` park +-- must survive the release so it can still fire when its deadline arrives. +-- +-- That post-sweep state (execution_run_id null, run still parked) is exactly +-- the precondition for re-adoption in enqueueWakeup's legacy-run fallback: +-- the fallback selects any run for the issue whose status is queued/running/ +-- scheduled_retry, has no exclusion for the run that just lost the lock, and +-- cancelStaleScheduledRetry declines to cancel a park belonging to the issue's +-- own assignee. It then re-stamps execution_locked_at = now(). The clock +-- restarts, so a capacity park deadlined days out could re-acquire a fresh 6h +-- lock on every wake, indefinitely. +-- +-- The release therefore has to leave a mark on the run itself; the issue row is +-- the wrong place because the sweep nulls it. This counter is that mark. +-- Adoption declines once it reaches MAX_SWEPT_ISSUE_LOCK_RELEASES, which bounds +-- total lock time attributable to one run at a single window regardless of wake +-- volume. It is incremented for every holder the sweep releases, not only +-- parks: `queued` (BLO-18995) and silent-`running` (BLO-19941) locks are +-- released by the same path and are selectable by the same fallback, and a +-- released park that is later promoted arrives there as `queued` still carrying +-- this count. +-- +-- Backfilled to 0: pre-existing rows have no recorded release, so they keep +-- exactly one more adoption. That is the conservative direction — it cannot +-- strand a run that is legitimately mid-flight at deploy time, and any run that +-- is genuinely wedged gets bounded on its next sweep release. +-- +-- ADD COLUMN with a non-volatile DEFAULT is a catalog-only rewrite on +-- PostgreSQL 11+, so this does not scan heartbeat_runs (~1.8 GB) and holds +-- ACCESS EXCLUSIVE only for the catalog update. +ALTER TABLE "heartbeat_runs" ADD COLUMN "issue_lock_release_count" integer DEFAULT 0 NOT NULL; diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index b4b2f4b18d41..14b5fc3c8830 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -1478,6 +1478,13 @@ "when": 1786682096165, "tag": "0218_plugin_install_dir", "breakpoints": true + }, + { + "idx": 219, + "version": "7", + "when": 1786682097165, + "tag": "0219_heartbeat_runs_issue_lock_release_count", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 6a1fa76d965d..559cef91c170 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -44,6 +44,21 @@ export const heartbeatRuns = pgTable( onDelete: "set null", }), processLossRetryCount: integer("process_loss_retry_count").notNull().default(0), + // BLO-22060: how many times sweepStaleIssueLocks has released an issue + // execution lock held by this run. The sweep deliberately does not cancel + // the run, so the run survives the release — a `scheduled_retry` park still + // has to fire at its deadline — and was previously free to be re-adopted as + // the issue's executionRunId by enqueueWakeup's legacy-run fallback, + // re-stamping executionLockedAt and resetting the 6h staleness clock on + // every wake. This counter is what makes that release durable: adoption + // declines once it reaches MAX_SWEPT_ISSUE_LOCK_RELEASES, so total lock + // time attributable to one run is bounded no matter how many wakes arrive. + // + // Counted for every holder the sweep releases — `queued` (BLO-18995), + // silent-`running` (BLO-19941) and `scheduled_retry` (BLO-21309) — because + // the fallback can select all three, and a released park that is later + // promoted arrives there as `queued` still carrying this count. + issueLockReleaseCount: integer("issue_lock_release_count").notNull().default(0), scheduledRetryAt: timestamp("scheduled_retry_at", { withTimezone: true }), scheduledRetryAttempt: integer("scheduled_retry_attempt").notNull().default(0), scheduledRetryReason: text("scheduled_retry_reason"),