fix(jobs): persist bounded success outcomes

This commit is contained in:
Marc-André Bombeck committed 2026-07-29 15:50:42 +02:00
1 parent 7acc711d9e
commit 8db21ab718
3 files changed
+135 -12

No files matched your search

+45 -6
View File
@@ -28,21 +28,60 @@ beforeEach(() => {
});
describe("runJob", () => {
it("resolves and reports nothing when the handler says it did the work", async () => {
it("returns a bounded useful result and reports nothing on success", async () => {
const wrapped = runJob("rate-limit-cleanup", async () =>
jobDone({ deleted: 12 }),
jobDone({ outcome: "useful", deleted: 12 }),
);
await expect(wrapped(noJobs)).resolves.toBeUndefined();
await expect(wrapped(noJobs)).resolves.toEqual({
ok: true,
did: { outcome: "useful", deleted: 12 },
});
expect(reportWorkerError).not.toHaveBeenCalled();
});
it("treats a run with nothing to do as a success", async () => {
it("returns a durable clean-zero result when there was nothing to do", async () => {
const wrapped = runJob("mood-reminder-cleanup", async () =>
jobDone({ deleted: 0 }),
jobDone({ outcome: "clean_zero", deleted: 0 }),
);
await expect(wrapped(noJobs)).resolves.toBeUndefined();
await expect(wrapped(noJobs)).resolves.toEqual({
ok: true,
did: { outcome: "clean_zero", deleted: 0 },
});
expect(reportWorkerError).not.toHaveBeenCalled();
});
it("preserves an honest bounded partial outcome", async () => {
const wrapped = runJob("google-health-sync", async () =>
jobDone({
outcome: "partial",
users_complete: 2,
users_partial: 1,
users_failed: 1,
}),
);
await expect(wrapped(noJobs)).resolves.toEqual({
ok: true,
did: {
outcome: "partial",
users_complete: 2,
users_partial: 1,
users_failed: 1,
},
});
expect(reportWorkerError).not.toHaveBeenCalled();
});
it("refuses to persist credentials or unbounded facts on success", async () => {
const wrapped = runJob("google-health-sync", async () =>
jobDone({
refresh_token: "private-token",
} as unknown as Record<string, string>),
);
await expect(wrapped(noJobs)).rejects.toThrow(/fact key is not allowed/i);
expect(reportWorkerError).not.toHaveBeenCalled();
});
+13 -6
View File
@@ -20,7 +20,13 @@
*/
import type { Job, JobWithMetadata } from "pg-boss";
import type { JobFact, JobFacts, JobOutcome } from "./job-outcome";
import {
serializeJobOutcome,
type JobFact,
type JobFacts,
type JobOutcome,
type SerializedJobOutcome,
} from "./job-outcome";
import { reportWorkerError } from "./report-worker-error";
/** A converted handler: it says what it did instead of just finishing. */
@@ -61,8 +67,9 @@ function factMeta(did: JobFacts | undefined): Record<string, JobFact> {
}
/**
* Wrap a handler for `boss.work`. Resolves on `ok: true`; reports and
* rethrows on `ok: false`.
* Wrap a handler for `boss.work`. Resolves with the bounded, redacted
* persistence representation on `ok: true`; reports and rethrows on
* `ok: false`.
*
* A handler that throws is left alone — the throw already fails the job, and
* re-wrapping it would only bury the original stack. `runJob` exists for the
@@ -71,10 +78,10 @@ function factMeta(did: JobFacts | undefined): Record<string, JobFact> {
export function runJob<J>(
queue: string,
handler: (jobs: J[]) => Promise<JobOutcome>,
): (jobs: J[]) => Promise<void> {
return async (jobs: J[]): Promise<void> => {
): (jobs: J[]) => Promise<SerializedJobOutcome> {
return async (jobs: J[]): Promise<SerializedJobOutcome> => {
const outcome = await handler(jobs);
if (outcome.ok) return;
if (outcome.ok) return serializeJobOutcome(outcome);
const failure = new JobFailure(queue, outcome.reason, outcome.cause);
const causeMessage =
@@ -0,0 +1,77 @@
/**
* Durable success-output contract against pg-boss's own schema and worker.
*
* A unit test can prove `runJob` returns a value, but only pg-boss can prove
* that the value reaches the completed job row's `output` JSONB column. This
* test deliberately registers the production wrapper as the real worker
* callback and reads the completed row back through pg-boss.
*/
import { PgBoss, type Job } from "pg-boss";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { jobDone } from "@/lib/jobs/job-outcome";
import { runJob } from "@/lib/jobs/run-job";
const QUEUE = `job-success-outcome-${Date.now()}-${Math.floor(
Math.random() * 1e6,
)}`;
let boss: PgBoss;
async function waitForCompletedOutput(
jobId: string,
): Promise<Record<string, unknown>> {
const deadline = Date.now() + 15_000;
while (Date.now() < deadline) {
const job = await boss.getJobById(QUEUE, jobId);
if (job?.state === "completed") {
return (job.output ?? {}) as Record<string, unknown>;
}
await new Promise((resolve) => setTimeout(resolve, 50));
}
throw new Error("pg-boss job did not complete before the test deadline");
}
beforeAll(async () => {
boss = new PgBoss({
connectionString: process.env.DATABASE_URL,
supervise: false,
schedule: false,
});
await boss.start();
await boss.createQueue(QUEUE);
}, 120_000);
afterAll(async () => {
await boss?.stop({ graceful: false, close: true });
});
describe("runJob durable pg-boss success output", () => {
it("persists only the bounded serialized outcome on completion", async () => {
await boss.work<object, Record<string, unknown>>(
QUEUE,
{ pollingIntervalSeconds: 0.5 },
runJob(QUEUE, async (_jobs: Job<object>[]) =>
jobDone({
provider: "google_health",
outcome: "useful",
users_complete: 1,
measurements_imported: 4,
}),
),
);
const jobId = await boss.send(QUEUE, {});
if (!jobId) throw new Error("queue rejected the integration-test job");
await expect(waitForCompletedOutput(jobId)).resolves.toEqual({
ok: true,
did: {
provider: "google_health",
outcome: "useful",
users_complete: 1,
measurements_imported: 4,
},
});
});
});