diff --git a/src/selfhost/pg-dialect.ts b/src/selfhost/pg-dialect.ts index 9bda2c7221..39bff6dbad 100644 --- a/src/selfhost/pg-dialect.ts +++ b/src/selfhost/pg-dialect.ts @@ -1,9 +1,9 @@ // SQLite → Postgres SQL dialect translation for the self-host Postgres backend (#977). gittensory's core and // drizzle-orm/d1 emit SQLite-dialect SQL; this translates the bounded set of SQLite-isms the codebase uses -// (placeholders + a handful of scalar functions + INSERT OR REPLACE/IGNORE) so the SAME queries run on -// Postgres. The timestamp columns are TEXT (ISO strings written by the app), so the datetime/CURRENT_TIMESTAMP -// translations return TEXT in SQLite's format to preserve the existing text-comparison semantics. Validated -// against a real Postgres (all 56 migrations + the runtime query paths). +// (placeholders + a handful of scalar functions + INSERT OR REPLACE/IGNORE + the rowid pseudo-column) so +// the SAME queries run on Postgres. The timestamp columns are TEXT (ISO strings written by the app), so the +// datetime/CURRENT_TIMESTAMP translations return TEXT in SQLite's format to preserve the existing +// text-comparison semantics. Validated against a real Postgres (all 56 migrations + the runtime query paths). // INSERT OR REPLACE needs an explicit conflict target on Postgres; map the (few) tables that use it to their PK. const REPLACE_CONFLICT_KEYS: Record = { @@ -14,14 +14,29 @@ const REPLACE_CONFLICT_KEYS: Record = { orb_signals: ["instance_id", "repo_hash", "pr_hash"], }; -/** Replace `?` placeholders with `$1,$2,…`, skipping any `?` inside single-quoted string literals. */ +/** Replace `?` placeholders with `$1,$2,…`, skipping any `?` inside single-quoted string literals. A `?` + * immediately followed by digits is SQLite's *numbered* placeholder (`?1`, `?2`, …, e.g. retention.ts's + * `retentionWhere()` and repositories.ts's `claimRegateFanoutSlot()`) — its index is reused verbatim as + * `$1`/`$2` rather than folded into the anonymous-placeholder counter below, otherwise `?1` corrupts to + * `$1` + a literal trailing `1` (i.e. `$11`), which Postgres reads as bind parameter 11. Per SQLite's own + * rule, a later anonymous `?` gets "one greater than the largest parameter number already assigned", so a + * numbered placeholder also raises the anonymous counter's floor — otherwise a later `?` could collide with + * an earlier `?N` (e.g. `?1` then `?` must yield `$1`, `$2`, not `$1`, `$1`). */ export function toNumberedPlaceholders(sql: string): string { let out = ""; let n = 0; let inString = false; - for (const ch of sql) { + for (let i = 0; i < sql.length; i++) { + const ch = sql[i] as string; if (ch === "'") inString = !inString; if (ch === "?" && !inString) { + const numbered = /^\d+/.exec(sql.slice(i + 1))?.[0]; + if (numbered) { + n = Math.max(n, Number(numbered)); + out += `$${numbered}`; + i += numbered.length; + continue; + } n += 1; out += `$${n}`; } else { @@ -50,6 +65,18 @@ export function translateFunctions(sql: string): string { ); } +/** Translate SQLite's `rowid` pseudo-column to Postgres's `ctid` system column. Both give a stable, + * per-row identifier for the lifetime of a single statement/snapshot — exactly how the codebase uses + * it: `DELETE ... WHERE rowid IN (SELECT rowid FROM t WHERE ... LIMIT n)` for bounded batched pruning + * (retention.ts) and `ORDER BY rowid` for insertion-order tie-breaking (orb/relay.ts, tests). `ctid` is + * a *physical* row location that can change across `VACUUM FULL` / row rewrites, so this is only safe + * for the codebase's existing usage — internal bookkeeping resolved within one statement — never for + * durable application-facing row identity. Fixes the self-host Postgres dead-letter where the raw + * `rowid` reached Postgres verbatim ("column \"rowid\" does not exist"). */ +export function translateRowid(sql: string): string { + return sql.replace(/\browid\b/gi, "ctid"); +} + /** Translate INSERT OR REPLACE / INSERT OR IGNORE to Postgres ON CONFLICT. */ export function translateInsertOr(sql: string): string { if (/^\s*INSERT\s+OR\s+IGNORE\s+INTO/i.test(sql)) { @@ -87,7 +114,7 @@ export function stripConflictTargetQualifiers(sql: string): string { /** Translate a runtime query (SQLite → Postgres). */ export function translateSql(sql: string): string { - return toNumberedPlaceholders(stripConflictTargetQualifiers(translateFunctions(translateInsertOr(sql)))); + return toNumberedPlaceholders(stripConflictTargetQualifiers(translateRowid(translateFunctions(translateInsertOr(sql))))); } /** Migrations are applied as whole multi-statement files via exec(), so the statement-anchored diff --git a/test/integration/selfhost-pg.test.ts b/test/integration/selfhost-pg.test.ts index 498a05c08c..0589d168c8 100644 --- a/test/integration/selfhost-pg.test.ts +++ b/test/integration/selfhost-pg.test.ts @@ -6,6 +6,8 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import pg from "pg"; import { runSelfHostMigrations } from "../../src/selfhost/migrate"; import { createPgAdapter } from "../../src/selfhost/pg-adapter"; +import { pruneExpiredRecords } from "../../src/db/retention"; +import { processJob } from "../../src/queue/processors"; const URL = process.env.PG_TEST_URL; const suite = URL ? describe : describe.skip; @@ -55,4 +57,31 @@ suite("Postgres backend (#977) — real Postgres", () => { const still = await db.prepare("SELECT COUNT(*) AS n FROM system_flags WHERE key=?").bind("batch_probe").first<{ n: number }>(); expect(still?.n).toBe(1); // the DELETE rolled back }); + + it("prunes rows past the retention window and processJob('prune-retention') does not dead-letter (regression for the live self-host incident: job _selfhost_jobs.id=61132 failed with 'column \"rowid\" does not exist')", async () => { + const db = createPgAdapter(pool); + const env = { DB: db } as unknown as Env; + const oldIso = new Date(Date.now() - 100 * 86_400_000).toISOString(); + const recentIso = new Date(Date.now() - 1 * 86_400_000).toISOString(); + for (const [id, createdAt] of [ + ["pg-old-1", oldIso], + ["pg-old-2", oldIso], + ["pg-recent", recentIso], + ] as const) { + await db + .prepare("INSERT INTO ai_usage_events (id, feature, model, status, estimated_neurons, created_at) VALUES (?, 'f', 'm', 'ok', 1, ?)") + .bind(id, createdAt) + .run(); + } + + const results = await pruneExpiredRecords(env, { policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }] }); + expect(results[0]?.deleted).toBe(2); // the two old rows, bounded-batch deleted via ctid (not rowid) + const remaining = await db.prepare("SELECT COUNT(*) AS n FROM ai_usage_events").first<{ n: number }>(); + expect(remaining?.n).toBe(1); + + // The exact live incident: the job queue's processJob("prune-retention") dispatch must not throw. + await expect(processJob(env, { type: "prune-retention", requestedBy: "schedule" })).resolves.toBeUndefined(); + const audit = await db.prepare("SELECT outcome FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string }>(); + expect(audit?.outcome).toBe("success"); + }); }); diff --git a/test/unit/selfhost-pg-dialect.test.ts b/test/unit/selfhost-pg-dialect.test.ts index 425b1524ac..b5dcd1578c 100644 --- a/test/unit/selfhost-pg-dialect.test.ts +++ b/test/unit/selfhost-pg-dialect.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { stripConflictTargetQualifiers, toNumberedPlaceholders, translateDdl, translateFunctions, translateInsertOr, translateMigrationInserts, translateSql } from "../../src/selfhost/pg-dialect"; +import { stripConflictTargetQualifiers, toNumberedPlaceholders, translateDdl, translateFunctions, translateInsertOr, translateMigrationInserts, translateRowid, translateSql } from "../../src/selfhost/pg-dialect"; describe("pg-dialect (#977 SQLite → Postgres)", () => { it("numbers placeholders, skipping `?` inside string literals", () => { @@ -7,6 +7,21 @@ describe("pg-dialect (#977 SQLite → Postgres)", () => { expect(toNumberedPlaceholders("SELECT '?' AS lit WHERE a=?")).toBe("SELECT '?' AS lit WHERE a=$1"); }); + it("REGRESSION: reuses a SQLite numbered placeholder's own index instead of corrupting it via the anonymous counter", () => { + // Before the fix: `?1` scanned as anonymous `?` (→ $1) followed by a literal `1`, corrupting to `$11` + // — a bind index Postgres has no value for. retentionWhere() (retention.ts) and claimRegateFanoutSlot() + // (repositories.ts) both use this numbered syntax. + expect(toNumberedPlaceholders("created_at < ?1")).toBe("created_at < $1"); + expect(toNumberedPlaceholders("last_regate_fanout_at = ?1 WHERE id = 'singleton' AND (x IS NULL OR x < ?2)")).toBe( + "last_regate_fanout_at = $1 WHERE id = 'singleton' AND (x IS NULL OR x < $2)", + ); + // A later anonymous `?` continues from the highest index already assigned (SQLite's own rule), so it + // must not collide with an earlier numbered placeholder. + expect(toNumberedPlaceholders("a=?1 AND b=?")).toBe("a=$1 AND b=$2"); + // A literal `?1` inside a string is left untouched, same as a bare `?` literal. + expect(toNumberedPlaceholders("SELECT '?1' AS lit WHERE a=?")).toBe("SELECT '?1' AS lit WHERE a=$1"); + }); + it("translates datetime/strftime/CURRENT_TIMESTAMP/json to Postgres (text-returning to match SQLite)", () => { expect(translateFunctions("x > datetime('now', ?)")).toContain("to_char(now() + (?)::interval"); expect(translateFunctions("datetime('now')")).toContain("to_char(now(),"); @@ -80,4 +95,29 @@ describe("pg-dialect (#977 SQLite → Postgres)", () => { expect(out).toContain("values ($1, $2)"); // placeholders numbered expect(out).toContain("set \"status\" = $3"); }); + + it("translates the rowid pseudo-column to Postgres's ctid system column", () => { + expect(translateRowid("SELECT rowid FROM t WHERE a = ?")).toBe("SELECT ctid FROM t WHERE a = ?"); + expect(translateRowid("ORDER BY rowid DESC")).toBe("ORDER BY ctid DESC"); + expect(translateRowid("ORDER BY ROWID ASC")).toBe("ORDER BY ctid ASC"); // case-insensitive + // Only the bare `rowid` token is rewritten — identifiers that merely contain it are left alone. + expect(translateRowid("SELECT row_id, my_rowid_col FROM t")).toBe("SELECT row_id, my_rowid_col FROM t"); + expect(translateRowid("SELECT 1")).toBe("SELECT 1"); // no-op passthrough + }); + + it("REGRESSION (self-host Postgres prune-retention dead-letter): translateSql strips rowid from the exact batched-delete shape retention.ts emits", () => { + // The literal shape src/db/retention.ts's pruneExpiredRecords() builds for its bounded batched delete. + // Before the fix, this reached Postgres verbatim and failed with `column "rowid" does not exist`. + const deleteSql = 'DELETE FROM ai_usage_events WHERE rowid IN (SELECT rowid FROM ai_usage_events WHERE created_at < ?1 LIMIT 1000)'; + const out = translateSql(deleteSql); + expect(out.toLowerCase()).not.toContain("rowid"); + expect(out).toBe("DELETE FROM ai_usage_events WHERE ctid IN (SELECT ctid FROM ai_usage_events WHERE created_at < $1 LIMIT 1000)"); + }); + + it("also fixes the rowid tie-break ORDER BY used by orb/relay.ts enrollment resolution", () => { + const sql = "SELECT relay_mode FROM orb_enrollments WHERE installation_id = ? ORDER BY enrolled_at DESC, rowid DESC"; + const out = translateSql(sql); + expect(out.toLowerCase()).not.toContain("rowid"); + expect(out).toContain("ORDER BY enrolled_at DESC, ctid DESC"); + }); }); diff --git a/test/unit/selfhost-pg-retention.test.ts b/test/unit/selfhost-pg-retention.test.ts new file mode 100644 index 0000000000..96c9742d70 --- /dev/null +++ b/test/unit/selfhost-pg-retention.test.ts @@ -0,0 +1,123 @@ +// Unit tests for data-retention pruning (src/db/retention.ts) against the Postgres backend (#977). Mocks +// pg.Pool so no real DB is needed — real-Postgres integration coverage lives in test/integration/selfhost-pg.test.ts. +// retention.ts itself is unchanged: it still emits SQLite-dialect SQL (rowid, `?1`-style numbered +// placeholders); src/selfhost/pg-dialect.ts's translateSql() is what makes it Postgres-safe, same as every +// other query path on this backend. These tests exercise that translation end-to-end through the real +// pruneExpiredRecords()/processJob() call path, not just the dialect translator in isolation. +import { describe, expect, it, vi } from "vitest"; +import type { Pool } from "pg"; +import { createPgAdapter } from "../../src/selfhost/pg-adapter"; +import { pruneExpiredRecords } from "../../src/db/retention"; +import { processJob, runRetentionPrune } from "../../src/queue/processors"; + +interface MockPgPool { + pool: Pool; + calls: string[]; + remaining: Record; +} + +/** A minimal fake Postgres that also acts as a regression guard: if untranslated SQLite SQL (the `rowid` + * pseudo-column) ever reaches it again, it throws the exact error a real Postgres raised in the live + * self-host incident (dead-lettered job `_selfhost_jobs.id = 61132`, `prune-retention`, 5 attempts). */ +function makeRetentionPgPool(remaining: Record = {}): MockPgPool { + const calls: string[] = []; + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + const q = String(sql); + calls.push(q); + if (/\browid\b/i.test(q)) throw new Error('column "rowid" does not exist'); + + const countMatch = /^SELECT count\(\*\) AS n FROM (\w+) WHERE/i.exec(q); + if (countMatch) { + const table = countMatch[1] as string; + return { rows: [{ n: remaining[table] ?? 0 }], rowCount: 1 }; + } + + const deleteMatch = /^DELETE FROM (\w+) WHERE ctid IN \(SELECT ctid FROM \1 WHERE .*? LIMIT (\d+)\)$/i.exec(q); + if (deleteMatch) { + const table = deleteMatch[1] as string; + const limit = Number(deleteMatch[2]); + const have = remaining[table] ?? 0; + const changes = Math.min(have, limit); + remaining[table] = have - changes; + return { rows: [], rowCount: changes }; + } + + if (/^insert into "?audit_events"?/i.test(q)) return { rows: [], rowCount: 1 }; + + return { rows: [], rowCount: 0 }; + }); + return { pool: { query: fn } as unknown as Pool, calls, remaining }; +} + +function makeEnv(remaining: Record = {}): { env: Env; mock: MockPgPool } { + const mock = makeRetentionPgPool(remaining); + return { env: { DB: createPgAdapter(mock.pool) } as unknown as Env, mock }; +} + +describe("pruneExpiredRecords on the Postgres backend (#977)", () => { + it("dry-run counts eligible rows without issuing any delete", async () => { + const { env, mock } = makeEnv({ ai_usage_events: 5 }); + const results = await pruneExpiredRecords(env, { + dryRun: true, + policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }], + }); + expect(results[0]?.deleted).toBe(5); + expect(mock.calls.some((q) => /^DELETE/i.test(q))).toBe(false); + expect(mock.remaining.ai_usage_events).toBe(5); // untouched + }); + + it("deletes across multiple bounded batches and stops at the per-table cap, same as the SQLite path", async () => { + const { env, mock } = makeEnv({ ai_usage_events: 5 }); + const results = await pruneExpiredRecords(env, { + batchSize: 2, + maxPerTable: 4, + policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }], + }); + expect(results[0]?.deleted).toBe(4); // 2 + 2, then cap reached + expect(mock.remaining.ai_usage_events).toBe(1); // one row left for the next run + const deletes = mock.calls.filter((q) => /^DELETE/i.test(q)); + expect(deletes).toHaveLength(2); + }); + + it("keeps the audit_events durable-event exclusion in the translated Postgres delete", async () => { + const { env, mock } = makeEnv({ audit_events: 1 }); + await pruneExpiredRecords(env, { policy: [{ table: "audit_events", column: "created_at", days: 90 }] }); + const [deleteSql] = mock.calls.filter((q) => /^DELETE/i.test(q)); + expect(deleteSql).toContain("event_type NOT IN ('github_app.pr_public_surface_published')"); + expect(deleteSql?.toLowerCase()).not.toContain("rowid"); + }); + + it("translates the numbered `?1` cutoff placeholder to Postgres's $1, not a corrupted $11", async () => { + const { env, mock } = makeEnv({ ai_usage_events: 0 }); + await pruneExpiredRecords(env, { + dryRun: true, + policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }], + }); + const [countSql] = mock.calls; + expect(countSql).toContain("created_at < $1"); + expect(countSql).not.toMatch(/\$1\d/); // not $11, $12, ... + }); +}); + +describe("runRetentionPrune + processJob on the Postgres backend (#977)", () => { + it("audits a dry-run without deleting", async () => { + const { env, mock } = makeEnv({ ai_usage_events: 2 }); + await runRetentionPrune(env, "test", true); + const [insertSql] = mock.calls.filter((q) => /^insert into "?audit_events"?/i.test(q)); + expect(insertSql).toBeDefined(); + expect(mock.remaining.ai_usage_events).toBe(2); // nothing deleted + }); + + it("REGRESSION (self-host dead-letter, job id 61132): processJob prune-retention no longer throws the rowid column error on Postgres", async () => { + const { env } = makeEnv(); // every table reports 0 eligible rows — exercises the full default policy + await expect(processJob(env, { type: "prune-retention", requestedBy: "schedule" })).resolves.toBeUndefined(); + }); + + it("processJob prune-retention deletes eligible rows and records a success audit event on Postgres", async () => { + const { env, mock } = makeEnv({ ai_usage_events: 3 }); + await processJob(env, { type: "prune-retention", requestedBy: "schedule" }); + expect(mock.remaining.ai_usage_events).toBe(0); + const [insertSql] = mock.calls.filter((q) => /^insert into "?audit_events"?/i.test(q)); + expect(insertSql).toBeDefined(); + }); +});