perf(jobs): stop the geo backfill from retrying the same rows hourly

An audit row whose address no database can place stayed without a
location, so every hourly pass looked it up again for its whole 30-day
window, and finding it walked every audit row of those 30 days through
the created_at index.

Migration 0418 adds audit_logs.geo_attempted_at and a partial index on
the rows still missing a location or carrier. Every row a pass looks at
is stamped, and a row tried less than seven days ago is not picked
again, which still lets a range a later GeoLite2 release learns resolve.
The candidate read goes through the partial index, which holds only the
backlog.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 20:26:30 +02:00
1 parent 1cfae83578
commit c6c51be280
5 files changed
+224 -24

No files matched your search

@@ -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);
+15 -7
View File
@@ -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)
+38 -7
View File
@@ -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 () => {
+34 -10
View File
@@ -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;
}
}
@@ -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<number> {
const [row] = await prisma.$queryRaw<Array<{ n: bigint }>>`
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());
});
});