diff --git a/migrations/0196_retention_column_indexes_round_two.sql b/migrations/0196_retention_column_indexes_round_two.sql new file mode 100644 index 0000000000..176dfd574d --- /dev/null +++ b/migrations/0196_retention_column_indexes_round_two.sql @@ -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); diff --git a/src/db/retention.ts b/src/db/retention.ts index 1e6bec7db2..0ac1dc98dc 100644 --- a/src/db/retention.ts +++ b/src/db/retention.ts @@ -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> = { +export const RETENTION_PK_COLUMN: Readonly> = { audit_events: "id", ai_usage_events: "id", product_usage_events: "id", @@ -103,8 +119,44 @@ const RETENTION_PK_COLUMN: Readonly> = { 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 = 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); diff --git a/test/unit/retention.test.ts b/test/unit/retention.test.ts index bfa8400913..98b6fca21a 100644 --- a/test/unit/retention.test.ts +++ b/test/unit/retention.test.ts @@ -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); + }); +});