JSONbored · loopover-orb · Jul 28, 2026 · Jul 28, 2026
@@ -0,0 +1,22 @@
+-- #9472/#9473: 0193 gave every RETENTION_POLICY table of its era a leading-column index on its retention
+-- timestamp, so each batched delete's inner SELECT is an index range scan rather than a full table scan.
+-- Two later batches of policy entries never got the same treatment:
+--
+--   * #9415's five (check_summaries, pull_request_files, repo_github_totals_snapshots,
+--     recent_merged_pull_requests, orb_pr_outcomes) shipped with no migration at all, while also moving the
+--     prune to HOURLY. check_summaries (511MB / 180,820 rows) and pull_request_files (179MB / 63,091 rows)
+--     were measured as the two largest tables in the production database, so this was a full scan + sort of
+--     both, every hour -- and on the synchronous SQLite adapter that blocks the event loop.
+--   * #9473's four newly-bounded per-event tables, added alongside this migration.
+--
+-- Paired with the RETENTION_PK_COLUMN entries added in the same change (without which pkColumnFor() falls
+-- back to `rowid` -> `ctid` on Postgres, making the outer `IN` a seq scan regardless of this index).
+CREATE INDEX IF NOT EXISTS idx_check_summaries_updated_at ON check_summaries(updated_at);
+CREATE INDEX IF NOT EXISTS idx_pull_request_files_updated_at ON pull_request_files(updated_at);
+CREATE INDEX IF NOT EXISTS idx_repo_github_totals_snapshots_fetched_at ON repo_github_totals_snapshots(fetched_at);
+CREATE INDEX IF NOT EXISTS idx_recent_merged_pull_requests_updated_at ON recent_merged_pull_requests(updated_at);
+CREATE INDEX IF NOT EXISTS idx_orb_pr_outcomes_occurred_at ON orb_pr_outcomes(occurred_at);
+CREATE INDEX IF NOT EXISTS idx_pull_request_reviews_updated_at ON pull_request_reviews(updated_at);
+CREATE INDEX IF NOT EXISTS idx_predicted_gate_calibration_ledger_created_at ON predicted_gate_calibration_ledger(created_at);
+CREATE INDEX IF NOT EXISTS idx_contributor_gate_history_created_at ON contributor_gate_history(created_at);
+CREATE INDEX IF NOT EXISTS idx_decision_replay_inputs_created_at ON decision_replay_inputs(created_at);
@@ -75,6 +75,22 @@ export const RETENTION_POLICY: readonly RetentionRule[] = [
   { table: "repo_github_totals_snapshots", column: "fetched_at", days: 30 },
   { table: "recent_merged_pull_requests", column: "updated_at", days: 30 },
   { table: "orb_pr_outcomes", column: "occurred_at", days: 90 },
+  // #9473: four more members of the same re-derivable/per-event class #9415 bounded, found by an audit sweep
+  // for tables written per event with NO delete path anywhere in src/. Two have a pruned sibling, which is
+  // what makes the omission clearly unintentional rather than a retention decision:
+  //   - pull_request_reviews is the SIXTH GitHub mirror alongside pull_request_files/check_summaries above,
+  //     synced by the same backfill segment machinery and re-fetched on the next sync.
+  //   - predicted_gate_calibration_ledger is per (login, project, PR, COMMIT); its sibling
+  //     predicted_gate_calls is already pruned at 90d.
+  //   - contributor_gate_history is per (login, project, PR, HEAD_SHA), so every push adds a row, and every
+  //     reader is already windowed (contributor-gate-eval / predicted-gate-agreement both use created_at >= ?)
+  //     -- aged rows are pure dead weight.
+  //   - decision_replay_inputs holds one replay_json blob per decision record, whose parent decision_records
+  //     is pruned at 180d above; without a matching rule these rows outlive the thing they describe.
+  { table: "pull_request_reviews", column: "updated_at", days: 30 },
+  { table: "predicted_gate_calibration_ledger", column: "created_at", days: 90 },
+  { table: "contributor_gate_history", column: "created_at", days: 90 },
+  { table: "decision_replay_inputs", column: "created_at", days: 180 },
 ];
 
 // #9083: a real, single-column, indexable primary key for the ordered-range delete below, keyed by table
@@ -88,7 +104,7 @@ export const RETENTION_POLICY: readonly RetentionRule[] = [
 // inner SELECT is an index range scan, not a scan of its own. A table absent from this map (a composite
 // primary key, or a caller-supplied ad-hoc rule in a test) falls back to rowid/ctid -- still correct, just
 // not scan-optimal, which is acceptable for the lower row counts of the tables that fall back today.
-const RETENTION_PK_COLUMN: Readonly<Record<string, string>> = {
+export const RETENTION_PK_COLUMN: Readonly<Record<string, string>> = {
   audit_events: "id",
   ai_usage_events: "id",
   product_usage_events: "id",
@@ -103,8 +119,44 @@ const RETENTION_PK_COLUMN: Readonly<Record<string, string>> = {
   review_audit: "id",
   decision_records: "id",
   orb_webhook_events: "delivery_id",
+  // #9472: #9415 added the five tables below to RETENTION_POLICY but not here, so pkColumnFor() fell back to
+  // `rowid` -- which pg-dialect rewrites to `ctid`, turning each batched delete's outer `IN` into a full
+  // sequential scan of the whole table, hourly. All five have a single-column `id TEXT PRIMARY KEY`.
+  check_summaries: "id",
+  pull_request_files: "id",
+  repo_github_totals_snapshots: "id",
+  recent_merged_pull_requests: "id",
+  // #9473's additions carry their own single-column primary keys for the same reason.
+  pull_request_reviews: "id",
+  predicted_gate_calibration_ledger: "id",
+  contributor_gate_history: "id",
+  // decision_replay_inputs keys on record_id (decision_records.id), not an `id` column.
+  decision_replay_inputs: "record_id",
 };
 
+/**
+ * Policy tables that legitimately have NO single-column primary key, so {@link RETENTION_PK_COLUMN} cannot
+ * name one and pkColumnFor() falls back to `rowid` for them. Listing them explicitly (rather than letting an
+ * absence mean either "composite PK" or "someone forgot") is what lets the completeness guard in
+ * test/unit/retention.test.ts be strict: every policy table must appear in one of the two, so a new entry
+ * cannot ship unmapped by accident the way #9415's five did.
+ *
+ * NOTE the cost of being here: on the self-host Postgres backend `rowid` is rewritten to `ctid`, so the
+ * batched delete's outer `IN` is a sequential scan of the whole table. The retention-column index from
+ * 0193/0196 still serves the inner SELECT, so the scan is bounded by batch size rather than table size, but a
+ * future table with a genuinely high row count should prefer adding a surrogate `id` over joining this list.
+ */
+export const RETENTION_COMPOSITE_PK_TABLES: ReadonlySet<string> = new Set([
+  // PRIMARY KEY (repository_full_name, pr_number)
+  "orb_pr_outcomes",
+  // PRIMARY KEY (repo_full_name, path, head_sha)
+  "grounding_file_content_cache",
+  // PRIMARY KEY (repo_full_name, pull_number, head_sha[, linked_issue_number])
+  "ai_review_cache",
+  "ai_slop_cache",
+  "linked_issue_satisfaction_cache",
+]);
+
 function pkColumnFor(table: string): string {
   return RETENTION_PK_COLUMN[table] ?? "rowid";
 }
@@ -247,9 +299,17 @@ export const LATEST_ONLY_SIGNAL_SNAPSHOT_TYPES = [
  * for its trend/change reader. This keeps only the latest row per
  * (signal_type, target_key), batched PER signal_type (not one table-wide window-function delete) so
  * each statement stays within D1's per-statement CPU budget -- the same batching split used during
- * the incident's manual remediation. "Latest" is the highest rowid per key: signal_snapshots is
- * populated by a single sequential batch job, so insertion order and generated_at agree, and rowid
- * (unlike generated_at) can never tie.
+ * the incident's manual remediation.
+ *
+ * "Latest" is `(generated_at, id)` DESC per key -- NOT rowid (#9470). rowid was chosen because it "can never
+ * tie" where generated_at can, but on the self-host Postgres backend pg-dialect's translateRowid rewrites every
+ * `rowid` to `ctid`, which is a PHYSICAL heap location, not insertion order: once the age-prune frees pages, a
+ * NEWER row inserted into a reclaimed early page gets a LOWER ctid than an older row on a later page, so
+ * MAX(ctid) selected a STALE row and this delete removed the genuinely newest snapshot. Confirmed on production
+ * (2026-07-27): ~36% of multi-row keys for contributor-evidence-graph had ctid order disagreeing with recency.
+ * translateRowid's own doc says it is only safe for bookkeeping resolved WITHIN one statement and "never for
+ * durable application-facing row identity" -- deciding which row survives a DELETE is exactly that. The id
+ * tiebreak restores the total ordering rowid was providing, without depending on physical layout.
  */
 export async function dedupeSignalSnapshots(
   env: Env,
@@ -266,7 +326,10 @@ export async function dedupeSignalSnapshots(
     .all<{ signal_type: string }>();
 
   for (const { signal_type: signalType } of types.results) {
-    const staleCondition = `signal_type = ?1 AND rowid NOT IN (SELECT MAX(rowid) FROM signal_snapshots WHERE signal_type = ?1 GROUP BY target_key)`;
+    // One index-backed lookup per distinct target_key (signal_snapshots_target_idx covers
+    // (signal_type, target_key, generated_at)), rather than a table-wide window function -- same
+    // per-statement-budget discipline as the batching above.
+    const staleCondition = `signal_type = ?1 AND id NOT IN (SELECT (SELECT newest.id FROM signal_snapshots AS newest WHERE newest.signal_type = ?1 AND newest.target_key = keys.target_key ORDER BY newest.generated_at DESC, newest.id DESC LIMIT 1) FROM (SELECT DISTINCT target_key FROM signal_snapshots WHERE signal_type = ?1) AS keys)`;
 
     if (dryRun) {
       const row = await env.DB.prepare(`SELECT count(*) AS n FROM signal_snapshots WHERE ${staleCondition}`).bind(signalType).first<{ n: number }>();
@@ -276,7 +339,9 @@ export async function dedupeSignalSnapshots(
 
     let deleted = 0;
     for (;;) {
-      const result = await env.DB.prepare(`DELETE FROM signal_snapshots WHERE rowid IN (SELECT rowid FROM signal_snapshots WHERE ${staleCondition} LIMIT ${batchSize})`)
+      // Batch by the real primary key, not rowid -- see the #9470 note above: on Postgres `rowid` becomes
+      // `ctid`, so the outer `IN` degrades to a full seq scan AND carries the same physical-location semantics.
+      const result = await env.DB.prepare(`DELETE FROM signal_snapshots WHERE id IN (SELECT id FROM signal_snapshots WHERE ${staleCondition} LIMIT ${batchSize})`)
         .bind(signalType)
         .run();
       const changes = Number(result.meta?.changes ?? 0);

@@ -1,7 +1,7 @@
 import { describe, expect, it } from "vitest";
 import { createApp } from "../../src/api/routes";
 import { getDb } from "../../src/db/client";
-import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention";
+import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_COMPOSITE_PK_TABLES, RETENTION_PK_COLUMN, RETENTION_POLICY } from "../../src/db/retention";
 import { agentContextSnapshots, aiReviewCache, aiSlopCache, aiUsageEvents, groundingFileContentCache, linkedIssueSatisfactionCache, webhookEvents } from "../../src/db/schema";
 import { processJob, runRetentionPrune } from "../../src/queue/processors";
 import { REPO_FOCUS_MANIFEST_SIGNAL, REPO_PUBLIC_FOCUS_MANIFEST_SIGNAL } from "../../src/signals/focus-manifest-loader";
@@ -561,7 +561,7 @@ describe("dedupeSignalSnapshots", () => {
       DB: {
         prepare: (sql: string) => ({
           bind: (..._binds: unknown[]) =>
-            sql.includes("SELECT DISTINCT")
+            sql.includes("DISTINCT signal_type")
               ? { all: async () => ({ results: [{ signal_type: "repo-culture-profile" }] }) }
               : { first: async () => undefined }, // count query returns no row → `row?.n ?? 0` fallback fires
         }),
@@ -576,7 +576,7 @@ describe("dedupeSignalSnapshots", () => {
       DB: {
         prepare: (sql: string) => ({
           bind: (..._binds: unknown[]) =>
-            sql.includes("SELECT DISTINCT")
+            sql.includes("DISTINCT signal_type")
               ? { all: async () => ({ results: [{ signal_type: "repo-culture-profile" }] }) }
               : { run: async () => ({}) }, // no meta → `result.meta?.changes ?? 0` fallback fires, so changes = 0 < batchSize
         }),
@@ -685,3 +685,106 @@ describe("retention preview route", () => {
     expect(await countSignalSnapshots(env)).toBe(2); // preview is read-only
   });
 });
+
+// #9472 drift guard: #9415 added five tables to RETENTION_POLICY but shipped neither a RETENTION_PK_COLUMN
+// entry nor an index migration for any of them, so every batched delete fell back to `rowid` (-> `ctid` on
+// Postgres) and ran as a full sequential scan plus a full sort -- hourly, on the two largest tables in the
+// database. Nothing caught it, because nothing asserted the three sites stay in step. These tests are that
+// assertion: a future policy entry cannot ship without its PK mapping and its retention-column index.
+describe("RETENTION_POLICY completeness (drift guard, #9472)", () => {
+  it("every policy table is either PK-mapped or explicitly listed as composite-PK (never silently unmapped)", () => {
+    // An absent entry used to mean either "composite PK, rowid fallback is correct" or "someone forgot" --
+    // indistinguishable, which is how #9415's five shipped unmapped. Now it must be a deliberate choice.
+    const unaccounted = RETENTION_POLICY.filter((rule) => !(rule.table in RETENTION_PK_COLUMN) && !RETENTION_COMPOSITE_PK_TABLES.has(rule.table)).map(
+      (rule) => rule.table,
+    );
+    expect(unaccounted).toEqual([]);
+  });
+
+  it("no table claims both a single-column PK and a composite-PK exemption", () => {
+    const both = Object.keys(RETENTION_PK_COLUMN).filter((table) => RETENTION_COMPOSITE_PK_TABLES.has(table));
+    expect(both).toEqual([]);
+  });
+
+  it("every policy table has an index leading with its retention column somewhere in migrations/", async () => {
+    const { readdir, readFile } = await import("node:fs/promises");
+    const { join } = await import("node:path");
+    const dir = join(process.cwd(), "migrations");
+    const files = (await readdir(dir)).filter((name) => name.endsWith(".sql"));
+    const sql = (await Promise.all(files.map((name) => readFile(join(dir, name), "utf8")))).join("\n").toLowerCase();
+    // The index must LEAD with the retention column -- a trailing position cannot serve the ordered range
+    // scan the batched delete performs (that was 0193's whole point).
+    const missing = RETENTION_POLICY.filter((rule) => !sql.includes(`on ${rule.table}(${rule.column})`) && !sql.includes(`on ${rule.table} (${rule.column})`)).map(
+      (rule) => `${rule.table}(${rule.column})`,
+    );
+    expect(missing).toEqual([]);
+  });
+
+  it("every RETENTION_PK_COLUMN entry names a table the policy actually prunes (no dead mappings)", () => {
+    const policyTables = new Set(RETENTION_POLICY.map((rule) => rule.table));
+    const orphaned = Object.keys(RETENTION_PK_COLUMN).filter((table) => !policyTables.has(table));
+    expect(orphaned).toEqual([]);
+  });
+});
+
+// #9470 regression: dedupe kept the row with the highest `rowid`, which pg-dialect rewrites to `ctid` -- a
+// PHYSICAL heap location. Once the age-prune frees pages, a NEWER row inserted into a reclaimed early page
+// gets a LOWER ctid than an older row on a later page, so MAX(ctid) selected a STALE snapshot and this delete
+// removed the genuinely newest one. Confirmed live on production Postgres (2026-07-27): ~36% of multi-row
+// keys for contributor-evidence-graph had ctid order disagreeing with recency, across ~344 keys of
+// dedupe-eligible types, with the dedupe deleting thousands of rows per daily run.
+//
+// SQLite cannot reproduce ctid page reuse, so these tests pin the property that makes the bug impossible
+// either way: the survivor is chosen by (generated_at, id), never by physical/insertion order. The row
+// inserted LAST here is deliberately the OLDEST by generated_at -- under the old MAX(rowid) rule it would
+// have survived and the newest would have been deleted.
+describe("dedupeSignalSnapshots survivor selection (#9470)", () => {
+  it("keeps the newest row by generated_at even when it was inserted FIRST", async () => {
+    const env = createTestEnv();
+    await insertSignalSnapshot(env, "s-newest", "repo-culture-profile", "JSONbored/loopover", "2026-06-13T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-middle", "repo-culture-profile", "JSONbored/loopover", "2026-06-11T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-oldest", "repo-culture-profile", "JSONbored/loopover", "2026-06-10T00:00:00.000Z");
+
+    const results = await dedupeSignalSnapshots(env);
+
+    expect(results).toEqual([{ signalType: "repo-culture-profile", deleted: 2 }]);
+    const survivor = await env.DB.prepare("SELECT id FROM signal_snapshots WHERE signal_type = ?").bind("repo-culture-profile").first<{ id: string }>();
+    expect(survivor?.id).toBe("s-newest");
+  });
+
+  it("breaks a generated_at tie deterministically on id, keeping exactly one row per key", async () => {
+    // generated_at CAN tie (it was the original reason rowid was chosen); the id tiebreak restores the total
+    // ordering without depending on physical layout.
+    const env = createTestEnv();
+    const tied = "2026-06-12T00:00:00.000Z";
+    await insertSignalSnapshot(env, "s-aaa", "repo-culture-profile", "JSONbored/loopover", tied);
+    await insertSignalSnapshot(env, "s-zzz", "repo-culture-profile", "JSONbored/loopover", tied);
+
+    const results = await dedupeSignalSnapshots(env);
+
+    expect(results).toEqual([{ signalType: "repo-culture-profile", deleted: 1 }]);
+    const survivor = await env.DB.prepare("SELECT id FROM signal_snapshots WHERE signal_type = ?").bind("repo-culture-profile").first<{ id: string }>();
+    expect(survivor?.id).toBe("s-zzz"); // highest id wins the tie
+  });
+
+  it("dedupes each target_key independently, keeping the newest row of every key", async () => {
+    const env = createTestEnv();
+    await insertSignalSnapshot(env, "a-new", "repo-culture-profile", "JSONbored/loopover", "2026-06-13T00:00:00.000Z");
+    await insertSignalSnapshot(env, "a-old", "repo-culture-profile", "JSONbored/loopover", "2026-06-01T00:00:00.000Z");
+    await insertSignalSnapshot(env, "b-new", "repo-culture-profile", "JSONbored/metagraphed", "2026-06-13T00:00:00.000Z");
+    await insertSignalSnapshot(env, "b-old", "repo-culture-profile", "JSONbored/metagraphed", "2026-06-01T00:00:00.000Z");
+
+    await dedupeSignalSnapshots(env);
+
+    const rows = await env.DB.prepare("SELECT id FROM signal_snapshots WHERE signal_type = ? ORDER BY id").bind("repo-culture-profile").all<{ id: string }>();
+    expect(rows.results.map((row) => row.id)).toEqual(["a-new", "b-new"]);
+  });
+
+  it("leaves a single-row key untouched (nothing to dedupe)", async () => {
+    const env = createTestEnv();
+    await insertSignalSnapshot(env, "only", "repo-culture-profile", "JSONbored/loopover", "2026-06-13T00:00:00.000Z");
+
+    expect(await dedupeSignalSnapshots(env)).toEqual([{ signalType: "repo-culture-profile", deleted: 0 }]);
+    expect(await countSignalSnapshots(env, "repo-culture-profile")).toBe(1);
+  });
+});