feat(db): add heartbeat_runs.issue_lock_release_count (BLO-22060, split from #1124)

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<void>> = [];
+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<string> {
+  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,
+  );
+});

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;

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
     }
   ]
 }

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"),