diff --git a/src/api/routes.ts b/src/api/routes.ts index 1344a4bfda..0c89aa4c6f 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -107,7 +107,7 @@ import { getGlobalAgentFrozenState, setGlobalAgentFrozen, } from "../db/repositories"; -import { pruneExpiredRecords, RETENTION_POLICY } from "../db/retention"; +import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_POLICY } from "../db/retention"; import { backfillOpenPullRequestDetails, backfillRegisteredRepositories, @@ -3808,11 +3808,19 @@ export function createApp() { return c.json(await getRepositoryAiKeyStatus(c.env, fullName)); }); - // Read-only retention preview: counts the rows the daily prune cron would delete, per table. Does NOT - // delete anything (dry-run); the actual prune runs on the schedule via the prune-retention job. + // Read-only retention preview: counts the rows the daily prune cron would delete, per table, plus the + // duplicate signal_snapshots rows the dedup pass would remove. Does NOT delete anything (dry-run); the + // actual prune + dedup runs on the schedule via the prune-retention job. app.get("/v1/internal/retention/preview", async (c) => { const results = await pruneExpiredRecords(c.env, { dryRun: true }); - return c.json({ policy: RETENTION_POLICY, eligible: results, totalEligible: results.reduce((sum, r) => sum + r.deleted, 0) }); + const dedupeResults = await dedupeSignalSnapshots(c.env, { dryRun: true }); + return c.json({ + policy: RETENTION_POLICY, + eligible: results, + totalEligible: results.reduce((sum, r) => sum + r.deleted, 0), + signalSnapshotDuplicates: dedupeResults, + totalSignalSnapshotDuplicates: dedupeResults.reduce((sum, r) => sum + r.deleted, 0), + }); }); app.post("/v1/internal/repos/:owner/:repo/ai-key", async (c) => { diff --git a/src/db/retention.ts b/src/db/retention.ts index 92fcefaaab..459f8dc7b3 100644 --- a/src/db/retention.ts +++ b/src/db/retention.ts @@ -88,3 +88,50 @@ export async function pruneExpiredRecords( return results; } + +export type SignalSnapshotDedupeResult = { signalType: string; deleted: number }; + +/** + * signal_snapshots has no dedup: `generate-signal-snapshots` inserts a NEW row per (signal_type, + * target_key) on every run rather than replacing the prior one, so within RETENTION_POLICY's 90-day + * age window a key can accumulate hundreds of superseded snapshots (#3810 -- 342,243 rows for 2,183 + * distinct keys contributed to hitting D1's size cap). 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. + */ +export async function dedupeSignalSnapshots( + env: Env, + options: { dryRun?: boolean; batchSize?: number; maxPerType?: number } = {}, +): Promise { + const dryRun = options.dryRun ?? false; + const batchSize = options.batchSize ?? BATCH_SIZE; + const maxPerType = options.maxPerType ?? MAX_DELETED_PER_TABLE; + const results: SignalSnapshotDedupeResult[] = []; + + const types = await env.DB.prepare("SELECT DISTINCT signal_type FROM signal_snapshots").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)`; + + if (dryRun) { + const row = await env.DB.prepare(`SELECT count(*) AS n FROM signal_snapshots WHERE ${staleCondition}`).bind(signalType).first<{ n: number }>(); + results.push({ signalType, deleted: Number(row?.n ?? 0) }); + continue; + } + + 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})`) + .bind(signalType) + .run(); + const changes = Number(result.meta?.changes ?? 0); + deleted += changes; + if (changes < batchSize || deleted >= maxPerType) break; + } + results.push({ signalType, deleted }); + } + + return results; +} diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 2a091dd091..fd55f724da 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -83,7 +83,7 @@ import { upsertPullRequestFromGitHub, upsertRepositoryFromGitHub, } from "../db/repositories"; -import { pruneExpiredRecords } from "../db/retention"; +import { dedupeSignalSnapshots, pruneExpiredRecords } from "../db/retention"; import { effectiveIssueCapForAccountAge, isBelowAccountAgeThreshold, @@ -858,8 +858,10 @@ function refreshLiveMergeState( } /** - * Run (or dry-run) the data-retention prune across the configured log/snapshot tables and audit the - * outcome. The per-table windows live in RETENTION_POLICY; only append-only/superseded tables are pruned. + * Run (or dry-run) the data-retention prune across the configured log/snapshot tables, plus the + * signal_snapshots dedup pass (#3810 -- signal_snapshots has no natural dedup, so within its own + * retention window a key can still accumulate many superseded rows), and audit the combined outcome. + * The per-table windows live in RETENTION_POLICY; only append-only/superseded tables are pruned. */ export async function runRetentionPrune( env: Env, @@ -867,18 +869,22 @@ export async function runRetentionPrune( dryRun: boolean, ): Promise { const results = await pruneExpiredRecords(env, { dryRun }); + const dedupeResults = await dedupeSignalSnapshots(env, { dryRun }); const totalDeleted = results.reduce((sum, result) => sum + result.deleted, 0); + const totalDeduped = dedupeResults.reduce((sum, result) => sum + result.deleted, 0); await recordAuditEvent(env, { eventType: "retention.prune", actor: requestedBy, outcome: dryRun ? "completed" : "success", detail: dryRun - ? `dry-run: ${totalDeleted} row(s) eligible` - : `pruned ${totalDeleted} row(s)`, + ? `dry-run: ${totalDeleted} row(s) eligible, ${totalDeduped} duplicate signal_snapshots row(s) eligible` + : `pruned ${totalDeleted} row(s), deduped ${totalDeduped} signal_snapshots row(s)`, metadata: { dryRun, totalDeleted, perTable: Object.fromEntries(results.map((r) => [r.table, r.deleted])), + totalDeduped, + perSignalType: Object.fromEntries(dedupeResults.map((r) => [r.signalType, r.deleted])), }, }); } diff --git a/test/unit/retention.test.ts b/test/unit/retention.test.ts index a2acd2b133..187930c9ae 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 { pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention"; +import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention"; import { aiUsageEvents, webhookEvents } from "../../src/db/schema"; import { processJob, runRetentionPrune } from "../../src/queue/processors"; import { createTestEnv } from "../helpers/d1"; @@ -26,6 +26,21 @@ async function seed(env: Env) { const countWebhook = async (env: Env) => (await env.DB.prepare("SELECT count(*) AS n FROM webhook_events").first<{ n: number }>())?.n ?? 0; +async function insertSignalSnapshot(env: Env, id: string, signalType: string, targetKey: string, generatedAt: string) { + await env.DB.prepare( + "INSERT INTO signal_snapshots (id, signal_type, target_key, repo_full_name, payload_json, generated_at) VALUES (?,?,?,?,?,?)", + ) + .bind(id, signalType, targetKey, "JSONbored/gittensory", "{}", generatedAt) + .run(); +} + +const countSignalSnapshots = async (env: Env, signalType?: string) => + ( + await env.DB.prepare(signalType ? "SELECT count(*) AS n FROM signal_snapshots WHERE signal_type = ?" : "SELECT count(*) AS n FROM signal_snapshots") + .bind(...(signalType ? [signalType] : [])) + .first<{ n: number }>() + )?.n ?? 0; + describe("pruneExpiredRecords", () => { it("dry-run reports eligible rows per table without deleting anything", async () => { const env = createTestEnv(); @@ -95,6 +110,83 @@ describe("pruneExpiredRecords", () => { }); }); +describe("dedupeSignalSnapshots", () => { + it("returns no results when the table is empty", async () => { + const env = createTestEnv(); + const results = await dedupeSignalSnapshots(env); + expect(results).toEqual([]); + }); + + it("dry-run counts duplicates per signal_type without deleting anything", async () => { + const env = createTestEnv(); + await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-3", "repo_culture", "other/repo", "2026-06-01T00:00:00.000Z"); // distinct key, not a duplicate + const results = await dedupeSignalSnapshots(env, { dryRun: true }); + expect(results).toEqual([{ signalType: "repo_culture", deleted: 1 }]); + expect(await countSignalSnapshots(env)).toBe(3); // nothing actually deleted + }); + + it("keeps only the highest-rowid row per (signal_type, target_key) and leaves other signal_types untouched", async () => { + const env = createTestEnv(); + await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-3", "repo_culture", "JSONbored/gittensory", "2026-06-03T00:00:00.000Z"); // latest, kept + await insertSignalSnapshot(env, "s-4", "burden_forecast", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); // sole row, kept + + const results = await dedupeSignalSnapshots(env); + expect(results.find((r) => r.signalType === "repo_culture")?.deleted).toBe(2); + expect(results.find((r) => r.signalType === "burden_forecast")?.deleted).toBe(0); + expect(await countSignalSnapshots(env, "repo_culture")).toBe(1); + expect(await countSignalSnapshots(env, "burden_forecast")).toBe(1); + const remaining = await env.DB.prepare("SELECT id FROM signal_snapshots WHERE signal_type = ?").bind("repo_culture").first<{ id: string }>(); + expect(remaining?.id).toBe("s-3"); + }); + + it("deletes across multiple batches per signal_type and stops at the per-type cap", async () => { + const env = createTestEnv(); + for (let i = 0; i < 6; i++) { + await insertSignalSnapshot(env, `s-${i}`, "repo_culture", "JSONbored/gittensory", `2026-06-0${i + 1}T00:00:00.000Z`); + } + // The 6th insert (highest generated_at, inserted last so it also has the highest rowid) is kept, leaving 5 + // duplicates; batchSize 2 forces multiple full (changes === batchSize) delete iterations before maxPerType 4 + // is reached, so the loop continues past its first batch instead of stopping there. + const results = await dedupeSignalSnapshots(env, { batchSize: 2, maxPerType: 4 }); + expect(results).toEqual([{ signalType: "repo_culture", deleted: 4 }]); // 2 + 2, then cap reached + expect(await countSignalSnapshots(env, "repo_culture")).toBe(2); // 1 kept + 1 duplicate left for the next run + }); + + it("dry-run falls back to 0 when the count query returns no row (defensive ?? 0 arm)", async () => { + const noRowEnv = { + DB: { + prepare: (sql: string) => ({ + all: async () => ({ results: [{ signal_type: "repo_culture" }] }), // the DISTINCT signal_type query + bind: (..._binds: unknown[]) => ({ + first: async () => undefined, // count query returns no row → `row?.n ?? 0` fallback fires + }), + }), + }, + } as unknown as Env; + const results = await dedupeSignalSnapshots(noRowEnv, { dryRun: true }); + expect(results).toEqual([{ signalType: "repo_culture", deleted: 0 }]); + }); + + it("falls back to 0 changes when a delete run() result lacks meta (defensive ?? 0 arm)", async () => { + const noMetaEnv = { + DB: { + prepare: (sql: string) => ({ + all: async () => ({ results: [{ signal_type: "repo_culture" }] }), // the DISTINCT signal_type query + bind: (..._binds: unknown[]) => ({ + run: async () => ({}), // no meta → `result.meta?.changes ?? 0` fallback fires, so changes = 0 < batchSize + }), + }), + }, + } as unknown as Env; + const results = await dedupeSignalSnapshots(noMetaEnv); + expect(results).toEqual([{ signalType: "repo_culture", deleted: 0 }]); + }); +}); + describe("runRetentionPrune + processJob", () => { it("audits a dry-run without deleting", async () => { const env = createTestEnv(); @@ -106,13 +198,17 @@ describe("runRetentionPrune + processJob", () => { expect(audit?.detail).toMatch(/dry-run/); }); - it("processJob prune-retention deletes and audits", async () => { + it("processJob prune-retention deletes, dedupes signal_snapshots, and audits both", async () => { const env = createTestEnv(); await seed(env); + await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z"); await processJob(env, { type: "prune-retention", requestedBy: "schedule" }); expect(await countWebhook(env)).toBe(3); - const audit = await env.DB.prepare("SELECT outcome FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string }>(); + expect(await countSignalSnapshots(env, "repo_culture")).toBe(1); + const audit = await env.DB.prepare("SELECT outcome, detail FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string; detail: string }>(); expect(audit?.outcome).toBe("success"); + expect(audit?.detail).toMatch(/deduped 1 signal_snapshots row/); }); }); @@ -121,11 +217,21 @@ describe("retention preview route", () => { const app = createApp(); const env = createTestEnv(); await seed(env); + await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); + await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z"); const res = await app.request("/v1/internal/retention/preview", { headers: { authorization: `Bearer ${env.INTERNAL_JOB_TOKEN}` } }, env); expect(res.status).toBe(200); - const body = (await res.json()) as { totalEligible: number; eligible: Array<{ table: string; deleted: number }> }; + const body = (await res.json()) as { + totalEligible: number; + eligible: Array<{ table: string; deleted: number }>; + totalSignalSnapshotDuplicates: number; + signalSnapshotDuplicates: Array<{ signalType: string; deleted: number }>; + }; expect(body.totalEligible).toBeGreaterThanOrEqual(1); expect(body.eligible.find((r) => r.table === "webhook_events")).toBeUndefined(); + expect(body.totalSignalSnapshotDuplicates).toBe(1); + expect(body.signalSnapshotDuplicates).toEqual([{ signalType: "repo_culture", deleted: 1 }]); expect(await countWebhook(env)).toBe(3); // preview is read-only + expect(await countSignalSnapshots(env)).toBe(2); // preview is read-only }); });