fix(jobs): reconcile durable fact inventory

This commit is contained in:
Marc-André Bombeck committed 2026-07-29 15:59:27 +02:00
1 parent 75aa850eae
commit fb02cf89b1
9 files changed
+308 -32

No files matched your search

@@ -0,0 +1,140 @@
/**
* Production success-fact inventory.
*
* `runJob` now serializes every successful handler result before pg-boss
* completes the row. A newly introduced fact key that is not deliberately
* reviewed would therefore fail the job at runtime. This source-level guard
* keeps the fixed vocabulary and every production `jobDone({...})` call in
* lockstep while the serializer's negative tests keep credentials, URLs,
* identifiers, free text, nested payloads, and health values outside it.
*/
import { readdirSync, readFileSync } from "node:fs";
import { join } from "node:path";
import ts from "typescript";
import { describe, expect, it } from "vitest";
import { JOB_FACT_ALLOWLIST } from "@/lib/jobs/job-outcome";
const JOBS_ROOT = join(process.cwd(), "src/lib/jobs");
function productionJobFiles(directory = JOBS_ROOT): string[] {
return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => {
const path = join(directory, entry.name);
if (entry.isDirectory()) {
return entry.name === "__tests__" ? [] : productionJobFiles(path);
}
return entry.name.endsWith(".ts") ? [path] : [];
});
}
function objectFactKeys(
object: ts.ObjectLiteralExpression,
source: ts.SourceFile,
): string[] {
return object.properties.map((property) => {
if (ts.isSpreadAssignment(property) || !property.name) {
throw new Error(
`jobDone facts must have explicit keys at ${source.fileName}:${
source.getLineAndCharacterOfPosition(property.getStart()).line + 1
}`,
);
}
return property.name.getText(source).replace(/^["']|["']$/g, "");
});
}
function declaredObject(
source: ts.SourceFile,
name: string,
): ts.ObjectLiteralExpression | null {
let found: ts.ObjectLiteralExpression | null = null;
const visit = (node: ts.Node): void => {
if (
ts.isVariableDeclaration(node) &&
ts.isIdentifier(node.name) &&
node.name.text === name &&
node.initializer &&
ts.isObjectLiteralExpression(node.initializer)
) {
found = node.initializer;
}
ts.forEachChild(node, visit);
};
visit(source);
return found;
}
function successFactKeys(file: string): string[] {
const source = ts.createSourceFile(
file,
readFileSync(file, "utf8"),
ts.ScriptTarget.Latest,
true,
);
const keys: string[] = [];
const visit = (node: ts.Node): void => {
if (
ts.isCallExpression(node) &&
ts.isIdentifier(node.expression) &&
node.expression.text === "jobDone" &&
node.arguments.length > 0
) {
const facts = node.arguments[0]!;
if (ts.isObjectLiteralExpression(facts)) {
keys.push(...objectFactKeys(facts, source));
} else if (ts.isIdentifier(facts)) {
const object = declaredObject(source, facts.text);
if (!object) {
throw new Error(
`jobDone facts must resolve to a local object at ${file}:${
source.getLineAndCharacterOfPosition(facts.getStart()).line + 1
}`,
);
}
keys.push(...objectFactKeys(object, source));
} else {
throw new Error(
`jobDone facts must be an auditable object at ${file}:${
source.getLineAndCharacterOfPosition(facts.getStart()).line + 1
}`,
);
}
}
ts.forEachChild(node, visit);
};
visit(source);
return keys;
}
describe("durable production job-outcome fact inventory", () => {
it("allows every explicit production success-fact key", () => {
const unknown = productionJobFiles()
.flatMap((file) =>
successFactKeys(file).map((key) => ({
file: file.replace(`${process.cwd()}/`, ""),
key,
})),
)
.filter(({ key }) => !JOB_FACT_ALLOWLIST.has(key));
expect(unknown).toEqual([]);
});
it("keeps location, credential, capability, and health identifiers out", () => {
for (const forbidden of [
"rotate_active_key_id",
"offhost_backup_endpoint",
"offhost_backup_bucket",
"restore_drill_object_key",
"tls_pin_monitor_host",
"intake_auto_skip_cutoff",
"user_id",
"access_token",
"refresh_token",
"health_value",
]) {
expect(JOB_FACT_ALLOWLIST.has(forbidden), forbidden).toBe(false);
}
});
});
-1
View File
@@ -83,7 +83,6 @@ export async function handleEncryptionKeyRotate(
// deployment no longer configures. The pass still ran, so they ride out
// as a count rather than failing (and retrying) the whole corpus.
return jobDone({
rotate_active_key_id: result.activeKeyId,
rotate_scanned: result.totalScanned,
rotate_rotated: result.totalRotated,
rotate_errors: result.totalErrors,
+1 -1
View File
@@ -169,7 +169,7 @@ export async function handleEnvironmentFetch(
"info",
`[environment-fetch] user=${userId} module disabled — skipping`,
);
return jobDone({ skipped: "environment module disabled" });
return jobDone({ skipped: "environment_module_disabled" });
}
const { startDate, endDate } = resolveRange(payload);
+161 -20
View File
@@ -42,32 +42,173 @@ export type JobFacts = Readonly<Record<string, JobFact>>;
* free-form error, or health value to a durable job result.
*/
export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = new Set([
"provider",
"access_tokens_deleted",
"assessments_warmed",
"auto_resolved",
"backed",
"buckets_collapsed",
"buckets_recomputed",
"budget_blocked",
"cached",
"candidates_scanned",
"connect_tickets_deleted",
"connections_deleted",
"considered",
"created",
"daily_rows_retired",
"days_consolidated",
"days_rebuilt",
"days_recomputed",
"days_skipped",
"days_skipped_no_tombstones",
"days_stored",
"deleted",
"dense_days_consolidated",
"dense_rows_soft_deleted",
"dense_skipped",
"derived_resting_rows_upserted",
"discovery_enqueued",
"discovery_failed",
"discovery_skipped",
"dispatched",
"dispatched_period_confirm",
"dispatched_period_soon",
"document_purge_deleted",
"documents",
"documents_indexed",
"dose_changes_migrated",
"downstream_failed",
"drained",
"duration_ms",
"errored",
"failed",
"failures",
"feedback_buckets",
"feedback_total_rows",
"feedback_window_days",
"fetches",
"finalised",
"finalized",
"forced",
"generated",
"geo_backfill_carrier_resolved",
"geo_backfill_located",
"geo_backfill_scanned",
"geo_backfill_skipped",
"geo_backfill_still_unresolved",
"host_metric_pruned",
"hourly_rows_upserted",
"imported",
"in_slot",
"in_window",
"insufficient",
"intake_auto_skip_count",
"intakes_pruned",
"inventory_expired_count",
"inventory_items_migrated",
"jobs",
"legacy_rows_soft_deleted",
"linked",
"malformed",
"manual_mints_removed",
"markers",
"mean_days_consolidated",
"mean_rows_soft_deleted",
"measurements_imported",
"measurements_migrated",
"measurements_pruned",
"medications",
"medications_evaluated",
"metric_assessments_warmed",
"missed_doses_minted",
"mood_entries_migrated",
"mood_pruned",
"notifications_dispatched",
"notifications_failed",
"notified",
"offhost_backup_configured",
"offhost_backup_failed",
"offhost_backup_total_users",
"offhost_backup_uploaded",
"outcome",
"passes_failed",
"per_sample_rows_deleted",
"per_sample_rows_soft_deleted",
"persisted",
"plan_reviews_minted",
"pr_detection_inserted",
"pr_detection_jobs",
"pr_detection_ties",
"pr_detection_users",
"pr_detection_users_failed",
"processed",
"provider",
"queued",
"rearmed",
"records_read",
"refreshed",
"reminders_due",
"removed",
"restore_drill_age_days",
"restore_drill_ciphertext_bytes",
"restore_drill_intake_events",
"restore_drill_measurements",
"restore_drill_medications",
"restore_drill_mood_entries",
"restore_drill_plaintext_bytes",
"restore_drill_stale",
"retried",
"reviewed",
"rollup_failed",
"rotate_errors",
"rotate_rotated",
"rotate_scanned",
"rows_deleted",
"rows_normalised",
"rows_reinserted",
"rows_resurrected",
"rows_soft_deleted",
"rows_upserted",
"satisfied",
"scheduled",
"sent",
"served",
"side_effects_migrated",
"single_user",
"skipped",
"skipped_already_dispatched",
"skipped_already_logged",
"skipped_already_notified",
"skipped_incomplete",
"skipped_no_channel",
"skipped_not_due",
"skipped_outside_window",
"slots_collapsed",
"states_deleted",
"stored",
"subscription_repair_failed",
"summaries_enqueued",
"suppressed_client_managed",
"suppressed_discreet",
"thumbnails_enqueued",
"tls_pin_monitor_known_count",
"tls_pin_monitor_outcome",
"total",
"users_synced",
"unchanged",
"updated",
"users",
"users_clean_zero",
"users_complete",
"users_partial",
"users_enqueued",
"users_failed",
"users_parked",
"users_skipped",
"users_useful",
"users_clean_zero",
"users_partial",
"users_retryable",
"downstream_failed",
"measurements_imported",
"deleted",
"generated",
"jobs",
"skipped",
"processed",
"created",
"updated",
"failed",
"retried",
"queued",
"sent",
"subscription_repair_failed",
"users_scanned",
"users_skipped",
"users_synced",
"users_useful",
"wedge_skipped",
]);
export const MAX_JOB_FACTS = 32;
-2
View File
@@ -52,8 +52,6 @@ export async function handleOffhostBackup(
offhost_backup_uploaded: report.uploaded,
offhost_backup_failed: report.failed,
offhost_backup_total_users: report.totalUsers,
offhost_backup_endpoint: report.config.endpoint,
offhost_backup_bucket: report.config.bucket,
});
} catch (err) {
// Not configured ⇒ skip silently with a warning, not an error: most
@@ -64,7 +64,6 @@ export async function handleIntakeAutoSkip(
evt.addMeta("intake_auto_skip_cutoff", result.cutoff.toISOString());
return jobDone({
intake_auto_skip_count: result.skippedCount,
intake_auto_skip_cutoff: result.cutoff.toISOString(),
});
} catch (err) {
evt.addWarning(
+3 -3
View File
@@ -149,11 +149,11 @@ export async function handleTlsPinMonitor(
evt.addMeta("tls_pin_monitor_outcome", summary.outcome);
evt.addMeta("tls_pin_monitor_host", summary.host);
evt.addMeta("tls_pin_monitor_known_count", summary.knownPinCount);
// No configured target leaves `host` null; the empty string keeps the
// fact shape scalar, and `outcome: "skipped"` already names that case.
// The configured host remains in the scoped operator event above. It is
// deliberately excluded from the durable pg-boss result, which carries
// only the stable outcome code and non-identifying pin count.
return jobDone({
tls_pin_monitor_outcome: summary.outcome,
tls_pin_monitor_host: summary.host ?? "",
tls_pin_monitor_known_count: summary.knownPinCount,
});
} catch (err) {
+1 -2
View File
@@ -189,7 +189,6 @@ export async function handleRestoreDrill(
// above already reached the operator. Failing the job here would retry a
// reading that cannot change until the next upload lands.
return jobDone({
restore_drill_object_key: report.objectKey,
restore_drill_age_days: report.ageDays,
restore_drill_ciphertext_bytes: report.ciphertextBytes,
restore_drill_plaintext_bytes: report.plaintextBytes,
@@ -204,7 +203,7 @@ export async function handleRestoreDrill(
evt.addWarning(`restore-drill skipped: ${err.message}`);
// Most self-hosters never set the S3 vars. Nothing to verify is not a
// failed verification, and a monthly failed job would be noise.
return jobDone({ skipped: "offhost backup not configured" });
return jobDone({ skipped: "offhost_backup_not_configured" });
}
evt.addWarning(`restore-drill failed: ${err}`);
await reportWorkerError(RESTORE_DRILL_QUEUE, err);
@@ -6,7 +6,7 @@
* 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 { PgBoss } from "pg-boss";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { jobDone } from "@/lib/jobs/job-outcome";
@@ -51,7 +51,7 @@ describe("runJob durable pg-boss success output", () => {
await boss.work<object, Record<string, unknown>>(
QUEUE,
{ pollingIntervalSeconds: 0.5 },
runJob(QUEUE, async (_jobs: Job<object>[]) =>
runJob(QUEUE, async () =>
jobDone({
provider: "google_health",
outcome: "useful",