fix(jobs): name the stage and condition when the tombstone purge fails

A failed run threw the bare Prisma error, which the support report showed
as "Error". The reason now carries the stage and the database condition
(statement_timeout, lock_timeout, pool_timeout, ...), and the report's
error class reads the SQLSTATE out of a Prisma message.

Refs #1239
This commit is contained in:
Marc-André Bombeck committed 2026-10-18 13:10:21 +02:00
1 parent 1d7077286f
commit 947551cd8e
3 files changed
+192 -20

No files matched your search

@@ -0,0 +1,61 @@
/**
* A purge run that hits a database failure says what it was doing and why.
*
* Before #1239 the run threw the bare Prisma error, whose message carries the
* condition only as a SQLSTATE, and the support report showed seven failed
* runs as "Error". The reason now names the stage and the condition.
*/
import { describe, expect, it, vi } from "vitest";
vi.mock("@/lib/db", () => ({ prisma: {} }));
vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss: () => null }));
vi.mock("@/lib/jobs/job-run-report", () => ({ reportJobRun: vi.fn() }));
import { handleCompactionTombstonePurge } from "../compaction-tombstone-purge";
const STATEMENT_TIMEOUT =
"\nInvalid `prisma.measurement.findMany()` invocation:\n\n\nDatabase error. Code: `57014`. Message: `canceling statement due to statement timeout`";
describe("compaction tombstone purge failures", () => {
it("names the stage and the condition when the scan times out", async () => {
const prisma = {
measurement: {
findMany: vi.fn().mockRejectedValue(new Error(STATEMENT_TIMEOUT)),
},
};
const outcome = await handleCompactionTombstonePurge(
[{ id: "job-1", data: {} } as never],
prisma as never,
);
expect(outcome.ok).toBe(false);
if (!outcome.ok) {
expect(outcome.reason).toBe(
"stopped while scanning tombstones (statement_timeout)",
);
expect((outcome.cause as Error).message).toBe(
"stopped while scanning tombstones (statement_timeout)",
);
expect(((outcome.cause as Error).cause as Error).message).toBe(
STATEMENT_TIMEOUT,
);
}
});
it("falls back to the error's class when no condition is named", async () => {
const prisma = {
measurement: {
findMany: vi.fn().mockRejectedValue(new TypeError("boom")),
},
};
const outcome = await handleCompactionTombstonePurge(
[{ id: "job-2", data: {} } as never],
prisma as never,
);
expect(outcome.ok).toBe(false);
if (!outcome.ok) {
expect(outcome.reason).toBe(
"stopped while scanning tombstones (TypeError)",
);
}
});
});
+65 -20
View File
@@ -55,7 +55,8 @@ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
import { holdAccountFoldLock } from "@/lib/export/restore-lock";
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
import { jobBudget } from "@/lib/jobs/job-budget";
import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
import { failureCode } from "@/lib/jobs/failure-condition";
import { reportJobRun, type JobRunCandidate } from "@/lib/jobs/job-run-report";
import { purgeTombstonesByAccount } from "@/lib/jobs/measurement-tombstone-cleanup";
import { PURGE_BATCH_SIZE, PURGE_MAX_BATCHES } from "@/lib/jobs/purge-batch";
@@ -112,6 +113,35 @@ export interface PurgeCompactionTombstonesOptions {
shouldStop?: () => boolean;
}
/**
* A run that stopped on a database failure, named by what it was doing and
* the condition (`failure-condition.ts`). Before #1239 the run threw the bare
* Prisma error, and the support report read seven failed runs as "Error".
*/
class CompactionPurgeStoppedError extends Error {
constructor(
readonly stage: "scanning tombstones" | "deleting a batch",
cause: unknown,
) {
super(`stopped while ${stage} (${failureCode(cause)})`, { cause });
this.name = "CompactionPurgeStoppedError";
}
}
/** Run `work`, naming `stage` on anything it throws. */
async function during<T>(
stage: CompactionPurgeStoppedError["stage"],
work: () => Promise<T>,
): Promise<T> {
try {
return await work();
} catch (err) {
throw err instanceof CompactionPurgeStoppedError
? err
: new CompactionPurgeStoppedError(stage, err);
}
}
type ScannedRow = {
id: string;
userId: string;
@@ -245,27 +275,28 @@ export async function purgeCompactionTombstones(
ready = ready.filter((row) => !skip.has(row.userId));
while (ready.length < take && !exhausted) {
if (options.shouldStop?.()) break;
await classifyNextPage(skip);
await during("scanning tombstones", () => classifyNextPage(skip));
}
return ready.splice(0, take);
},
deleteIds: async (tx, ids, userId) => {
// A re-fold reads an account's live samples and its leftovers under the
// fold lock, so it never sees the leftovers half gone; it reads the
// repair marker under the same lock and stays off them once the repair
// has run (`fold-constituents.ts`).
await holdAccountFoldLock(tx, userId);
// Re-checked at delete time: a row a restore wrote back, or one a
// person resurrected since the scan, is not a tombstone any more.
const result = await tx.measurement.deleteMany({
where: {
id: { in: ids },
deletedAt: { not: null },
source: "APPLE_HEALTH",
},
});
return result.count;
},
deleteIds: (tx, ids, userId) =>
during("deleting a batch", async () => {
// A re-fold reads an account's live samples and its leftovers under the
// fold lock, so it never sees the leftovers half gone; it reads the
// repair marker under the same lock and stays off them once the repair
// has run (`fold-constituents.ts`).
await holdAccountFoldLock(tx, userId);
// Re-checked at delete time: a row a restore wrote back, or one a
// person resurrected since the scan, is not a tombstone any more.
const result = await tx.measurement.deleteMany({
where: {
id: { in: ids },
deletedAt: { not: null },
source: "APPLE_HEALTH",
},
});
return result.count;
}),
});
return {
@@ -301,7 +332,21 @@ export async function handleCompactionTombstonePurge(
);
return withBackgroundEvent("job.compaction_tombstone_purge", async (evt) => {
const startedAt = Date.now();
const result = await purgeCompactionTombstones(prisma, { shouldStop });
let result: CompactionTombstonePurgeOutcome;
try {
result = await purgeCompactionTombstones(prisma, { shouldStop });
} catch (err) {
// The reason names the stage and the condition, which is what the
// admin's failing-jobs card and the support report show; the error
// itself rides as the cause, for the log.
return jobFailed(
err instanceof CompactionPurgeStoppedError
? err.message
: `stopped (${failureCode(err)})`,
err,
{ continuation },
);
}
const candidates: JobRunCandidate[] = [
...result.deferredUserIds.map((userId): JobRunCandidate => ({
key: accountKey(userId),
+66
View File
@@ -0,0 +1,66 @@
/**
* The database condition behind a failure, as a stable code.
*
* Prisma puts the Postgres SQLSTATE in the message ("Database error. Code:
* `57014`. Message: `canceling statement due to statement timeout`") and not
* in a form the support report's error class recognised, so a statement that
* ran past the 60-second `statement_timeout` read as the bare class "Error"
* (#1239). The code is the part an operator can act on, and it carries no
* account data, so it is what the report and a job's failure reason show.
*/
/** SQLSTATEs worth a name; any other one is shown as `sqlstate_<code>`. */
const SQLSTATE_CONDITIONS: Readonly<Record<string, string>> = {
"57014": "statement_timeout",
"55P03": "lock_timeout",
"40P01": "deadlock",
"40001": "serialization_failure",
"25P03": "idle_in_transaction_timeout",
"53100": "disk_full",
"53200": "out_of_memory",
"53300": "too_many_connections",
"57P01": "admin_shutdown",
};
/** Conditions named by the message text alone, in the order they are tried. */
const TEXT_CONDITIONS: ReadonlyArray<readonly [RegExp, string]> = [
[/statement[ _]timeout/i, "statement_timeout"],
[/lock[ _]timeout/i, "lock_timeout"],
[/deadlock/i, "deadlock"],
[/transaction already closed|expired transaction/i, "transaction_closed"],
// `pg`'s own words when no pooled connection came free in time, and
// Prisma's for the same.
[
/timeout exceeded when trying to connect|new connection from the connection pool/i,
"pool_timeout",
],
[
/connection terminated|connection_lost|server closed the connection/i,
"connection_lost",
],
];
/** The condition a failure message names, or null when it names none. */
export function failureCondition(text: string): string | null {
const sqlstate = /\bCode: `([0-9A-Z]{5})`/.exec(text)?.[1];
if (sqlstate) {
return SQLSTATE_CONDITIONS[sqlstate] ?? `sqlstate_${sqlstate}`;
}
for (const [pattern, condition] of TEXT_CONDITIONS) {
if (pattern.test(text)) return condition;
}
return null;
}
/**
* A thrown value as a short, account-free code for a job's failure reason:
* its database condition, else its class name.
*/
export function failureCode(error: unknown): string {
if (error instanceof Error) {
return failureCondition(error.message) ?? error.name;
}
return typeof error === "string"
? (failureCondition(error) ?? "error")
: "error";
}