diff --git a/prisma/migrations/0418_audit_log_geo_attempt/migration.sql b/prisma/migrations/0418_audit_log_geo_attempt/migration.sql new file mode 100644 index 000000000..cd5cd5d84 --- /dev/null +++ b/prisma/migrations/0418_audit_log_geo_attempt/migration.sql @@ -0,0 +1,30 @@ +-- The geo backfill notes when it last tried a row, and finds its candidates +-- through an index that holds only them. +-- +-- The hourly backfill (`src/lib/jobs/geo-backfill.ts`) re-resolves audit rows +-- whose location or carrier is missing. An address no database will ever +-- place (a private range that reached the log through a proxy) stays missing, +-- so the same few rows were looked up again on every pass, and finding them +-- meant walking thirty days of `audit_logs` through `created_at`, filtering +-- every row that had long been resolved. +-- +-- `geo_attempted_at` records the try; the backfill skips a row it tried less +-- than `GEO_BACKFILL_RETRY_DAYS` ago. NULL for every existing row: each is +-- tried once more, then waits like any other. +-- +-- `audit_logs_geo_backfill_idx` is partial on the rows still missing +-- something, so it holds the backlog and nothing else, and shrinks as rows +-- resolve. Prisma cannot express the predicate, so the index lives here only +-- (noted at the model in `schema.prisma`). A plain build: it takes SHARE on +-- `audit_logs` while it reads the table once, which `migrate deploy` does at +-- boot before the app serves. +-- +-- Reversibility: +-- DROP INDEX IF EXISTS "audit_logs_geo_backfill_idx"; +-- ALTER TABLE "audit_logs" DROP COLUMN IF EXISTS "geo_attempted_at"; + +ALTER TABLE "audit_logs" ADD COLUMN IF NOT EXISTS "geo_attempted_at" TIMESTAMP(3); + +CREATE INDEX IF NOT EXISTS "audit_logs_geo_backfill_idx" + ON "audit_logs" ("created_at") + WHERE "ip_address" IS NOT NULL AND ("location" IS NULL OR "carrier" IS NULL); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index f3337df8c..c802f3fd4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -5415,18 +5415,26 @@ model AuditLog { /// lives. actorUserId String? @map("actor_user_id") - action String // e.g. "auth.login", "measurement.create", "export.csv" - details String? // JSON string, no sensitive data - ipAddress String? @map("ip_address") - location String? // e.g. "Berlin, DE" — resolved from IP at log time + action String // e.g. "auth.login", "measurement.create", "export.csv" + details String? // JSON string, no sensitive data + ipAddress String? @map("ip_address") + location String? // e.g. "Berlin, DE" — resolved from IP at log time // v1.4.27 B3 — resolved at audit-creation time from the bundled // GeoLite2-ASN MMDB. `asn` is the autonomous-system number, // `carrier` is the organisation string (e.g. "Deutsche Telekom AG"). // Both nullable so older rows, private-IP rows, and offline-miss // rows stay valid. - asn Int? // autonomous-system number - carrier String? // autonomous-system organisation - createdAt DateTime @default(now()) @map("created_at") + asn Int? // autonomous-system number + carrier String? // autonomous-system organisation + /// When the geo backfill (`src/lib/jobs/geo-backfill.ts`) last tried this + /// row. A row it could not complete waits `GEO_BACKFILL_RETRY_DAYS` before + /// the next try, so an address no database will ever place (a private + /// range behind a proxy) is not resolved again on every hourly pass. NULL: + /// never tried. The backfill reads its candidates through the partial index + /// `audit_logs_geo_backfill_idx` (migration 0418, raw SQL only: Prisma + /// cannot express its predicate). + geoAttemptedAt DateTime? @map("geo_attempted_at") + createdAt DateTime @default(now()) @map("created_at") user User? @relation(fields: [userId], references: [id], onDelete: SetNull) diff --git a/src/lib/jobs/__tests__/geo-backfill.test.ts b/src/lib/jobs/__tests__/geo-backfill.test.ts index 0bd4bdec3..8a6b4a835 100644 --- a/src/lib/jobs/__tests__/geo-backfill.test.ts +++ b/src/lib/jobs/__tests__/geo-backfill.test.ts @@ -9,6 +9,7 @@ import { GEO_BACKFILL_BATCH_CAP, GEO_BACKFILL_CRON, GEO_BACKFILL_QUEUE, + GEO_BACKFILL_RETRY_DAYS, GEO_BACKFILL_WINDOW_DAYS, runGeoBackfill, } from "../geo-backfill"; @@ -65,6 +66,21 @@ describe("runGeoBackfill", () => { createdAt: { gt: new Date(now.getTime() - GEO_BACKFILL_WINDOW_DAYS * 86_400_000), }, + // A row tried less than the retry wait ago is not picked again. + AND: [ + { + OR: [ + { geoAttemptedAt: null }, + { + geoAttemptedAt: { + lt: new Date( + now.getTime() - GEO_BACKFILL_RETRY_DAYS * 86_400_000, + ), + }, + }, + ], + }, + ], }); expect(args?.take).toBe(GEO_BACKFILL_BATCH_CAP); }); @@ -91,7 +107,7 @@ describe("runGeoBackfill", () => { }); expect(prisma.auditLog.update).toHaveBeenCalledWith({ where: { id: "a1" }, - data: { location: "Berlin, DE" }, + data: { location: "Berlin, DE", geoAttemptedAt: expect.any(Date) }, }); }); @@ -117,6 +133,7 @@ describe("runGeoBackfill", () => { location: "München, DE", asn: 3320, carrier: "Deutsche Telekom AG", + geoAttemptedAt: expect.any(Date), }, }); }); @@ -136,7 +153,11 @@ describe("runGeoBackfill", () => { expect(summary.carrierResolved).toBe(1); expect(prisma.auditLog.update).toHaveBeenCalledWith({ where: { id: "a2b" }, - data: { location: "Bochum, DE", carrier: "Deutsche Telekom AG" }, + data: { + location: "Bochum, DE", + carrier: "Deutsche Telekom AG", + geoAttemptedAt: expect.any(Date), + }, }); }); @@ -158,15 +179,20 @@ describe("runGeoBackfill", () => { }); expect(prisma.auditLog.update).toHaveBeenCalledWith({ where: { id: "a3" }, - data: { carrier: "Vodafone GmbH", asn: 3209 }, + data: { + carrier: "Vodafone GmbH", + asn: 3209, + geoAttemptedAt: expect.any(Date), + }, }); }); - it("skips the update when the resolver misses entirely, counts as still-unresolved", async () => { + it("stamps only the attempt when the resolver misses entirely, counts as still-unresolved", async () => { const prisma = makePrismaMock([{ id: "a4", ipAddress: "192.0.2.1" }]); vi.mocked(lookupIpGeo).mockResolvedValueOnce(EMPTY_GEO); + const now = new Date("2026-05-15T12:00:00Z"); - const summary = await runGeoBackfill(prisma); + const summary = await runGeoBackfill(prisma, now); expect(summary).toEqual({ scanned: 1, @@ -174,7 +200,12 @@ describe("runGeoBackfill", () => { carrierResolved: 0, stillUnresolved: 1, }); - expect(prisma.auditLog.update).not.toHaveBeenCalled(); + // The try is recorded, so the next passes leave the row alone until + // the retry wait has passed. + expect(prisma.auditLog.update).toHaveBeenCalledWith({ + where: { id: "a4" }, + data: { geoAttemptedAt: now }, + }); }); it("handles a mixed batch of hits and misses", async () => { @@ -204,7 +235,7 @@ describe("runGeoBackfill", () => { carrierResolved: 2, stillUnresolved: 1, }); - expect(prisma.auditLog.update).toHaveBeenCalledTimes(2); + expect(prisma.auditLog.update).toHaveBeenCalledTimes(3); }); it("treats a Prisma update throw as still-unresolved without aborting the pass", async () => { diff --git a/src/lib/jobs/geo-backfill.ts b/src/lib/jobs/geo-backfill.ts index f0e7bd12d..919cb8a01 100644 --- a/src/lib/jobs/geo-backfill.ts +++ b/src/lib/jobs/geo-backfill.ts @@ -34,9 +34,16 @@ * * Idempotent: a row that the resolver still cannot match (private IP * sneaking through CF egress, freshly-allocated range the GeoLite2 - * release hasn't picked up yet) stays NULL and the next pass tries - * again. The cap is per-call, not per-row, so the helper can be - * scheduled at any cadence without doubling up. + * release hasn't picked up yet) stays NULL. Every row a pass looks at is + * stamped `geoAttemptedAt`, and a row stamped less than + * `GEO_BACKFILL_RETRY_DAYS` ago is not picked again: an address no + * database will ever place used to be looked up on every hourly pass for + * its whole 30-day window, and finding it walked every audit row of those + * 30 days. The wait still lets a range a later GeoLite2 release learns + * resolve on a later pass. The candidates are read through the partial + * index `audit_logs_geo_backfill_idx` (migration 0418), which holds only + * the rows still missing something. The cap is per-call, not per-row, so + * the helper can be scheduled at any cadence without doubling up. * * No notifications, no event ladder — the maintainer can read the * counts off the returned summary and decide whether to keep @@ -47,6 +54,8 @@ import { lookupIpGeo } from "@/lib/geo"; export const GEO_BACKFILL_BATCH_CAP = 500; export const GEO_BACKFILL_WINDOW_DAYS = 30; +/** How long a row the backfill could not complete waits for its next try. */ +export const GEO_BACKFILL_RETRY_DAYS = 7; /** * v1.4.37 — pg-boss queue name + cron expression for the recurring @@ -74,6 +83,9 @@ export async function runGeoBackfill( const cutoff = new Date( now.getTime() - GEO_BACKFILL_WINDOW_DAYS * 86_400_000, ); + const retryBefore = new Date( + now.getTime() - GEO_BACKFILL_RETRY_DAYS * 86_400_000, + ); const rows = await prisma.auditLog.findMany({ where: { @@ -85,6 +97,14 @@ export async function runGeoBackfill( OR: [{ location: null }, { carrier: null }], ipAddress: { not: null }, createdAt: { gt: cutoff }, + AND: [ + { + OR: [ + { geoAttemptedAt: null }, + { geoAttemptedAt: { lt: retryBefore } }, + ], + }, + ], }, select: { id: true, ipAddress: true }, orderBy: { createdAt: "desc" }, @@ -109,8 +129,14 @@ export async function runGeoBackfill( // ASN MMDB. The online fallback bounds itself with a request timeout. const { location, asn, carrier } = await lookupIpGeo(ip); - const data: { location?: string; asn?: number; carrier?: string | null } = - {}; + // Stamped whatever the outcome: a row that resolved leaves the candidate + // set anyway, and one that did not waits for its next try. + const data: { + location?: string; + asn?: number; + carrier?: string | null; + geoAttemptedAt: Date; + } = { geoAttemptedAt: now }; if (location) { data.location = location; summary.located += 1; @@ -121,10 +147,8 @@ export async function runGeoBackfill( summary.carrierResolved += 1; } - if (Object.keys(data).length === 0) { - summary.stillUnresolved += 1; - continue; - } + const resolved = data.location !== undefined || data.carrier !== undefined; + if (!resolved) summary.stillUnresolved += 1; try { await prisma.auditLog.update({ @@ -135,7 +159,7 @@ export async function runGeoBackfill( // The row may have been deleted by the retention sweeper between // findMany and update. Treat as unresolved; the next pass will // skip it because the row no longer exists. - summary.stillUnresolved += 1; + if (resolved) summary.stillUnresolved += 1; } } diff --git a/tests/integration/geo-backfill-attempt.test.ts b/tests/integration/geo-backfill-attempt.test.ts new file mode 100644 index 000000000..8c71bf735 --- /dev/null +++ b/tests/integration/geo-backfill-attempt.test.ts @@ -0,0 +1,107 @@ +/** + * The geo backfill tries a row once per retry wait, and finds its candidates + * through the partial index that holds only them. + * + * Before migration 0418 an address no database could place was looked up on + * every hourly pass for thirty days, and every pass walked thirty days of + * audit rows to find it. Run against the real table: the second pass must not + * pick the row again, a pass past the wait must, and the candidate read must + * go through `audit_logs_geo_backfill_idx` (read off the index's own scan + * counter, so the statement under test is the one Prisma actually sends). + */ +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/geo", () => ({ + lookupIpGeo: vi.fn(async (ip: string) => + ip.startsWith("10.") + ? { location: null, asn: null, carrier: null } + : { location: "Bochum, DE", asn: 3320, carrier: "Deutsche Telekom AG" }, + ), +})); + +const { runGeoBackfill, GEO_BACKFILL_RETRY_DAYS } = + await import("@/lib/jobs/geo-backfill"); +const { lookupIpGeo } = await import("@/lib/geo"); + +const prisma = getPrismaClient(); +const USER = "geo-backfill-attempt"; +const DAY = 86_400_000; +const NOW = new Date(); + +async function indexScans(): Promise { + const [row] = await prisma.$queryRaw>` + SELECT idx_scan AS n FROM pg_stat_user_indexes + WHERE indexrelname = 'audit_logs_geo_backfill_idx'`; + return Number(row?.n ?? 0); +} + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ data: { id: USER, username: USER } }); + // Enough resolved history that a sequential walk is the costlier plan. + await prisma.$executeRaw` + INSERT INTO audit_logs (id, user_id, action, ip_address, location, asn, carrier, created_at) + SELECT 'resolved-' || g, ${USER}, 'measurement.create', '84.131.0.1', + 'Bochum, DE', 3320, 'Deutsche Telekom AG', + ${NOW}::timestamp - (g * interval '5 seconds') + FROM generate_series(1, 20000) g`; + await prisma.auditLog.createMany({ + data: [ + // Never placeable: a private address. + { + id: "private", + userId: USER, + action: "auth.login", + ipAddress: "10.1.2.3", + createdAt: new Date(NOW.getTime() - 2 * DAY), + }, + // Placeable, missing only its carrier. + { + id: "carrierless", + userId: USER, + action: "auth.login", + ipAddress: "84.131.0.9", + location: "Bochum, DE", + createdAt: new Date(NOW.getTime() - 3 * DAY), + }, + ], + }); + await prisma.$executeRawUnsafe("ANALYZE audit_logs"); +}, 120_000); + +describe("runGeoBackfill", () => { + it("reads its candidates through the partial index", async () => { + const before = await indexScans(); + const summary = await runGeoBackfill(prisma, NOW); + expect(summary.scanned).toBe(2); + expect(summary.carrierResolved).toBe(1); + expect(summary.stillUnresolved).toBe(1); + // Statistics reach the view asynchronously; give them a moment. + let after = await indexScans(); + for (let i = 0; i < 40 && after === before; i += 1) { + await new Promise((r) => setTimeout(r, 250)); + after = await indexScans(); + } + expect(after).toBeGreaterThan(before); + }); + + it("does not try the unplaceable row again inside the wait", async () => { + vi.mocked(lookupIpGeo).mockClear(); + const summary = await runGeoBackfill( + prisma, + new Date(NOW.getTime() + 60 * 60_000), + ); + expect(summary.scanned).toBe(0); + expect(lookupIpGeo).not.toHaveBeenCalled(); + }); + + it("tries it again once the wait has passed", async () => { + const later = new Date(NOW.getTime() + (GEO_BACKFILL_RETRY_DAYS + 1) * DAY); + const summary = await runGeoBackfill(prisma, later); + expect(summary.scanned).toBe(1); + const row = await prisma.auditLog.findUnique({ where: { id: "private" } }); + expect(row?.geoAttemptedAt?.getTime()).toBe(later.getTime()); + }); +});