fix(jobs): retire the six per-metric status crons behind the nightly batch

The 02:00 batch builds every status card and falls each metric it cannot
cover back to that card's own path, so the 02:05 to 02:30 crons only
re-read cards the batch had just written. Their schedules and queues are
removed on an instance that still has them (`retireQueues`), and the
fresh-switch integration test drives the batch loop instead.
This commit is contained in:
Marc-André Bombeck committed 2026-10-20 03:34:51 +02:00
1 parent 8eb4ef3d56
commit 0770ae3590
11 files changed
+187 -384

No files matched your search

@@ -41,10 +41,6 @@ const AI_JOBS: Record<string, JobGate> = {
gate: "aiCapabilityForJob",
keys: ["statusText"],
},
"lib/jobs/reminder/insights-handlers.ts": {
gate: "aiCapabilityForJob",
keys: ["statusText"],
},
"lib/jobs/insight-status-generate.ts": {
gate: "aiCapabilityForJob",
keys: ["statusText"],
@@ -175,44 +175,53 @@ describe("findStatusCronCandidates — gates", () => {
});
});
describe("mood-status queue registration (dead-queue guard)", () => {
// v1.18.1 — the nightly status-ladder wiring moved out of the 2143-LOC
// reminder-worker boot file into the status registrar. The dead-queue guard
// follows the wiring there.
describe("the nightly status batch is the one status cron", () => {
const workerSrc = fs.readFileSync(
path.resolve(__dirname, "../reminder/register-status.ts"),
"utf8",
);
const RETIRED = [
"insights-blood-pressure-status",
"insights-weight-status",
"insights-pulse-status",
"insights-bmi-status",
"insights-mood-status",
"insights-medication-compliance-status",
];
it("registers the queue in the allQueues createQueue loop", () => {
it("registers and schedules the 02:00 batch with the nightly pass options", () => {
const match = workerSrc.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/);
expect(match).not.toBeNull();
expect(match![1]).toMatch(/\bMOOD_STATUS_QUEUE\b/);
});
it("schedules the 02:30 cron with the nightly pass options (retries and expiry)", () => {
expect(match![1]).toMatch(/\bGENERAL_STATUS_QUEUE\b/);
expect(workerSrc).toMatch(
/\[\s*MOOD_STATUS_QUEUE\s*,\s*MOOD_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/,
/\[\s*GENERAL_STATUS_QUEUE\s*,\s*GENERAL_STATUS_CRON\s*,\s*nightlyInsightPassOptions\s*\]/,
);
expect(workerSrc).toMatch(/MOOD_STATUS_CRON\s*=\s*"30 2 \* \* \*"/);
expect(workerSrc).toMatch(/GENERAL_STATUS_CRON\s*=\s*"0 2 \* \* \*"/);
});
it("registers a createAndWork handler for the queue", () => {
expect(workerSrc).toMatch(/createAndWork[\s\S]{0,120}MOOD_STATUS_QUEUE/);
it("retires the six per-metric crons on an instance that still has them", () => {
const retired = workerSrc.match(
/const RETIRED_STATUS_QUEUES\s*=\s*\[([\s\S]*?)\]/,
);
expect(retired).not.toBeNull();
for (const name of RETIRED) expect(retired![1]).toContain(`"${name}"`);
expect(workerSrc).toMatch(
/await retireQueues\(boss,\s*RETIRED_STATUS_QUEUES\)/,
);
const live = workerSrc.replace(retired![0], "");
for (const name of RETIRED) expect(live).not.toContain(`"${name}"`);
});
it("drives all seven status crons through the shared discovery", () => {
it("drives the batch through the shared discovery", () => {
const statusSrc = fs.readFileSync(
path.resolve(__dirname, "../reminder/insights-handlers.ts"),
"utf8",
);
expect(statusSrc).toMatch(/findStatusCronCandidates\(prisma\)/);
// The old iterate-every-user discovery must be gone from the status
// handlers (the WHOOP/data-backup cohort reads keep their own scans).
const statusBlock = statusSrc.slice(
statusSrc.indexOf("async function runStatusCronGenerate"),
statusSrc.indexOf("async function runStatusBatchCron"),
statusSrc.indexOf("async function handleInsightPregenerateJob"),
);
expect(statusBlock).toMatch(/findStatusCronCandidates\(prisma\)/);
expect(statusBlock).not.toMatch(/prisma\.user\.findMany/);
});
});
+2 -2
View File
@@ -749,7 +749,7 @@ export async function runInsightPregenerate(
// (`insights:${userId}`), so this never starves a user's manual
// regenerate quota. The gate bounds the comprehensive cost only: a
// blocked user still gets the refill-only status warm below, because
// the 02:xx status crons already skipped every pregenerate candidate
// the 02:00 status batch already skipped every pregenerate candidate
// on the assumption that THIS pass warms their cards — exiting early
// here left those cards cold until the first on-visit generation.
const budget = briefing.available
@@ -906,7 +906,7 @@ export async function runInsightPregenerate(
// generator's content-hash gate turns a same-data regeneration into a
// timestamp refresh with no provider call. The warm runs on every
// comprehensive outcome (budget-blocked, failed, timed-out,
// consent-skipped included) because the 02:xx status crons skip every
// consent-skipped included) because the 02:00 status batch skips every
// pregenerate candidate on the assumption that THIS pass covers them.
//
// Nothing to warm when the status notes themselves are unavailable.
+11 -20
View File
@@ -72,15 +72,6 @@ export type QueueRuntime =
const short = (why: string): QueueRuntime => ({ runtime: "short", why });
const nightlyStatusPass = (fn: string): QueueRuntime => ({
runtime: "long",
expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS,
expiryVia: ["nightlyInsightPassOptions"],
stop: { file: "lib/jobs/reminder/insights-handlers.ts", fn },
exclusive: "lockedPass",
why: "Nightly status pass: one provider call per candidate account, serially, over the whole instance.",
});
const cohortSync = (provider: string): QueueRuntime =>
short(
`${provider} sync: a recent window per connection (incremental since the last sync); a webhook job is one connection.`,
@@ -140,17 +131,17 @@ export const QUEUE_RUNTIME: Readonly<Record<string, QueueRuntime>> = {
"strava-backfill": forwarder,
// ── Status and insights ────────────────────────────────────────────
"insights-general-status": nightlyStatusPass("handleGeneralStatusGenerate"),
"insights-blood-pressure-status": nightlyStatusPass(
"handleBloodPressureStatusGenerate",
),
"insights-weight-status": nightlyStatusPass("handleWeightStatusGenerate"),
"insights-pulse-status": nightlyStatusPass("handlePulseStatusGenerate"),
"insights-bmi-status": nightlyStatusPass("handleBmiStatusGenerate"),
"insights-mood-status": nightlyStatusPass("handleMoodStatusGenerate"),
"insights-medication-compliance-status": nightlyStatusPass(
"handleMedicationComplianceStatusGenerate",
),
"insights-general-status": {
runtime: "long",
expireInSeconds: NIGHTLY_INSIGHT_PASS_EXPIRE_SECONDS,
expiryVia: ["nightlyInsightPassOptions"],
stop: {
file: "lib/jobs/reminder/insights-handlers.ts",
fn: "handleGeneralStatusGenerate",
},
exclusive: "lockedPass",
why: "Nightly status batch: one provider call per candidate account, serially, over the whole instance.",
},
"insight-pregenerate": {
runtime: "long",
expireInSeconds: INSIGHT_PREGENERATE_EXPIRE_SECONDS,
@@ -0,0 +1,45 @@
/**
* The registrar's boot-time decisions that reach pg-boss on every instance.
*/
import { describe, expect, it, vi } from "vitest";
vi.mock("@/lib/db", () => ({
prisma: { $executeRaw: vi.fn().mockResolvedValue(0) },
}));
vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));
import { retireQueues } from "../registrar-shared";
function fakeBoss() {
return {
createQueue: vi.fn().mockResolvedValue(undefined),
updateQueue: vi.fn().mockResolvedValue(undefined),
schedule: vi.fn().mockResolvedValue(undefined),
work: vi.fn().mockResolvedValue("worker-id"),
unschedule: vi.fn().mockResolvedValue(undefined),
deleteQueue: vi.fn().mockResolvedValue(undefined),
};
}
describe("retireQueues", () => {
it("removes the schedule before the queue, and one failure does not stop the rest", async () => {
const boss = fakeBoss();
const order: string[] = [];
boss.unschedule.mockImplementation(async (name: string) => {
order.push(`unschedule:${name}`);
});
boss.deleteQueue.mockImplementation(async (name: string) => {
order.push(`delete:${name}`);
if (name === "old-a") throw new Error("locked");
});
await expect(
retireQueues(boss as never, ["old-a", "old-b"]),
).resolves.toBeUndefined();
expect(order).toEqual([
"unschedule:old-a",
"delete:old-a",
"unschedule:old-b",
"delete:old-b",
]);
});
});
+19 -202
View File
@@ -1,5 +1,6 @@
/**
* Nightly insight status crons (general / blood pressure / weight / pulse / BMI / mood / medication compliance) plus the insight pregenerate and per-user status-generate handlers.
* The nightly status batch plus the insight pregenerate and per-user
* status-generate handlers.
*
* Extracted from reminder-worker.ts, which owns the queue names, cron
* schedules, and boss.work registrations.
@@ -23,155 +24,32 @@ import {
import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome";
import { jobBudget } from "@/lib/jobs/job-budget";
import { withBackgroundEvent } from "@/lib/logging/background";
import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status";
import { generateWeightStatusForUser } from "@/lib/insights/weight-status";
import { generatePulseStatusForUser } from "@/lib/insights/pulse-status";
import { generateBmiStatusForUser } from "@/lib/insights/bmi-status";
import { generateMoodStatusForUser } from "@/lib/insights/mood-status";
import { generateMedicationComplianceStatusForUser } from "@/lib/insights/medication-compliance-status";
import { generateStatusBatchForUser } from "@/lib/insights/status-batch";
import { findStatusCronCandidates } from "@/lib/jobs/status-cron-candidates";
import { annotate } from "@/lib/logging/context";
import { aiCapabilityForJob } from "@/lib/ai/capabilities/gate";
import { getWorkerPrisma } from "./shared";
export interface GeneralStatusPayload {
triggeredAt: string;
}
export interface BloodPressureStatusPayload {
triggeredAt: string;
}
export interface WeightStatusPayload {
triggeredAt: string;
}
export interface PulseStatusPayload {
triggeredAt: string;
}
export interface BmiStatusPayload {
triggeredAt: string;
}
export interface MoodStatusPayload {
triggeredAt?: string;
}
export interface MedicationComplianceStatusPayload {
triggeredAt: string;
}
/**
* Shared driver for the nightly 02:xx per-metric status crons. User
* discovery is centralised in `findStatusCronCandidates`, which applies
* the operator's `insightStatus` switch, the person's AI analysis switch,
* and the pregenerate-candidate skip (users with a stale comprehensive
* cache belong to the 04:30 pre-generate pass, which re-warms every
* per-status note anyway — see `status-cron-candidates.ts` for the full
* division of nightly labour). Each user's `statusText` capability is then
* resolved before a generator runs.
* The generators normalise `locale` themselves (de stays de, everything
* else gets English prose).
* The nightly status batch. The 02:00 cron runs `generateStatusBatchForUser`
* for every candidate (`findStatusCronCandidates`): it builds every
* per-metric snapshot once, issues ONE provider call for the metrics still
* needing the model, and fans the answer into the per-metric cache rows. A
* metric the batch omits, or every metric when the call fails, falls back to
* that card's own path inside the batch, so the pass covers every card on its
* own. Each user's `statusText` capability is resolved inside the batch before
* a snapshot is built.
*
* The pass is what succeeds or fails. One user whose generation threw is that
* user's cold card, reported as the `failed` count; failing the job would
* re-run the whole cohort over it.
*
* The pass walks every candidate with a provider call each, so on a large
* instance it can outlast its job. `shouldStop` is the job's time budget: the
* pass stops between users once it is spent and reports `stopped_early`. The
* users it did not reach keep yesterday's note until the next night or their
* next visit; nothing is retried beside a pass that is still running.
* The pass is what succeeds or fails. One user whose batch threw is that
* user's cold cards, reported as the `failed` count; failing the job would
* re-run the whole cohort over it. `shouldStop` is the job's time budget: the
* pass stops between users once it is spent and reports `stopped_early`, and
* the users it did not reach are warmed by their next visit or night.
*/
export async function runStatusCronGenerate(
taskName: string,
generate: (
userId: string,
options: { locale: string | null; force: boolean },
) => Promise<unknown>,
shouldStop: () => boolean = () => false,
): Promise<JobOutcome> {
return withBackgroundEvent(taskName, async (evt) => {
const prisma = getWorkerPrisma();
try {
const users = await findStatusCronCandidates(prisma);
if (users.length === 0) {
return jobDone({ total: 0, generated: 0, failed: 0 });
}
let generated = 0;
let failed = 0;
let stoppedEarly = false;
for (const user of users) {
if (shouldStop()) {
stoppedEarly = true;
break;
}
// The capability before the generator builds a snapshot. The batch
// entry below checks for itself.
const capability = await aiCapabilityForJob(user.id, "statusText");
if (!capability.available) {
annotate({
action: { name: "insights.status.cron.skipped" },
meta: { task: taskName, reason: capability.reason },
});
continue;
}
try {
await generate(user.id, {
locale: await resolveJobLocale(user.locale),
force: false,
});
generated++;
} catch (error) {
failed++;
evt.addWarning(
`${taskName} generation failed for user ${user.id}: ${error}`,
);
}
}
evt.setBackground({
task_name: taskName,
result: { generated, failed, total: users.length, stoppedEarly },
});
return jobDone({
total: users.length,
generated,
failed,
stopped_early: stoppedEarly,
});
} catch (err) {
evt.setError(err);
recordError();
await reportWorkerError(taskName, err);
throw err;
}
});
}
/**
* v1.18.11 (P2) — nightly status batch. The 02:00 anchor cron runs
* `generateStatusBatchForUser` for every candidate: it builds all seven
* per-metric snapshots once and issues ONE provider call for the metrics
* still needing the LLM (seed-pinned + grounded inside `runStatusCompletion`,
* unchanged), fanning the response into the SAME per-metric cache rows the
* standalone generators wrote.
*
* The six later per-metric crons (02:05–02:30) stay registered and keep their
* own drivers: a card the batch wrote today resolves inside their `prepare`
* step as a calendar-day cache hit (`served`, no provider call, no snapshot
* rebuild), so they cost a cache read and cover the edge cases the batch
* can't — a user discovered after 02:00, a metric the batch omitted, or a
* batch-call failure (which falls each metric back to its single-card path
* inside the batch itself). Net effect: the nightly ladder pays one call per
* user instead of seven, with no queue/registry churn.
*/
async function runStatusBatchCron(
export async function runStatusBatchCron(
taskName: string,
shouldStop: () => boolean,
): Promise<JobOutcome> {
@@ -189,8 +67,8 @@ async function runStatusBatchCron(
let failed = 0;
let stoppedEarly = false;
for (const user of users) {
// Same budget as the per-metric passes: stop between users, and
// leave the rest to the later crons and the next night.
// Stop between users; the rest are warmed by a visit or the next
// night.
if (shouldStop()) {
stoppedEarly = true;
break;
@@ -233,8 +111,7 @@ async function runStatusBatchCron(
},
});
// A user whose batch call threw already fell back to the single-card
// path inside the batch, and the six later per-metric crons cover what
// is still cold. The batch pass ran, so it reports the count.
// path inside the batch. The pass ran, so it reports the count.
return jobDone({
total: users.length,
generated,
@@ -257,66 +134,6 @@ export function handleGeneralStatusGenerate(
return runStatusBatchCron("job.insights.batch", jobBudget(jobs));
}
export function handleBloodPressureStatusGenerate(
jobs: Job<BloodPressureStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.blood_pressure",
generateBloodPressureStatusForUser,
jobBudget(jobs),
);
}
export function handleWeightStatusGenerate(
jobs: Job<WeightStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.weight",
generateWeightStatusForUser,
jobBudget(jobs),
);
}
export function handlePulseStatusGenerate(
jobs: Job<PulseStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.pulse",
generatePulseStatusForUser,
jobBudget(jobs),
);
}
export function handleBmiStatusGenerate(
jobs: Job<BmiStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.bmi",
generateBmiStatusForUser,
jobBudget(jobs),
);
}
export function handleMoodStatusGenerate(
jobs: Job<MoodStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.mood",
generateMoodStatusForUser,
jobBudget(jobs),
);
}
export function handleMedicationComplianceStatusGenerate(
jobs: Job<MedicationComplianceStatusPayload>[],
): Promise<JobOutcome> {
return runStatusCronGenerate(
"job.insights.medication_compliance",
generateMedicationComplianceStatusForUser,
jobBudget(jobs),
);
}
export async function handleInsightPregenerateJob(
jobs: Job<InsightPregeneratePayload>[],
): Promise<JobOutcome> {
+20 -113
View File
@@ -1,8 +1,8 @@
/**
* Status / insight / computed-score queue registrar.
*
* Owns the nightly per-metric insight-status ladder (general / BP / weight /
* pulse / BMI / mood / medication-compliance), the comprehensive-insight
* Owns the nightly status batch (every per-metric status card, one provider
* call per account), the comprehensive-insight
* pre-generation, the on-demand per-metric status generation, the nightly
* computed scores (recovery / stress / strain), the proactive Coach nudge, the
* Coach memory refresh, the period-narrative warm, and the medication
@@ -112,24 +112,13 @@ import {
createAndWork,
insightRetryOptions,
nightlyInsightPassOptions,
retireQueues,
type QueuePolicyTable,
type ScheduleEntry,
} from "./registrar-shared";
import {
GeneralStatusPayload,
BloodPressureStatusPayload,
WeightStatusPayload,
PulseStatusPayload,
BmiStatusPayload,
MoodStatusPayload,
MedicationComplianceStatusPayload,
handleGeneralStatusGenerate,
handleBloodPressureStatusGenerate,
handleWeightStatusGenerate,
handlePulseStatusGenerate,
handleBmiStatusGenerate,
handleMoodStatusGenerate,
handleMedicationComplianceStatusGenerate,
handleInsightPregenerateJob,
handleInsightStatusGenerate,
} from "./insights-handlers";
@@ -138,45 +127,25 @@ const GENERAL_STATUS_QUEUE = "insights-general-status";
const GENERAL_STATUS_CRON = "0 2 * * *"; // daily at 02:00
const BLOOD_PRESSURE_STATUS_QUEUE = "insights-blood-pressure-status";
const BLOOD_PRESSURE_STATUS_CRON = "5 2 * * *"; // daily at 02:05
const WEIGHT_STATUS_QUEUE = "insights-weight-status";
const WEIGHT_STATUS_CRON = "10 2 * * *"; // daily at 02:10
const PULSE_STATUS_QUEUE = "insights-pulse-status";
const PULSE_STATUS_CRON = "15 2 * * *"; // daily at 02:15
const BMI_STATUS_QUEUE = "insights-bmi-status";
const BMI_STATUS_CRON = "20 2 * * *"; // daily at 02:20
// v1.15.20 — mood joins the nightly per-metric status ladder. Same gate +
// discovery as the six older crons (see status-cron-candidates.ts); 02:30
// continues the 5-minute stagger after BMI (02:20) and compliance (02:25).
const MOOD_STATUS_QUEUE = "insights-mood-status";
const MOOD_STATUS_CRON = "30 2 * * *"; // daily at 02:30
const MEDICATION_COMPLIANCE_STATUS_QUEUE =
"insights-medication-compliance-status";
const MEDICATION_COMPLIANCE_STATUS_CRON = "25 2 * * *"; // daily at 02:25
/**
* The six per-metric status crons that followed the 02:00 batch (02:05 to
* 02:30). The batch builds every card and falls each metric it could not
* cover back to that card's own path, so they only re-read cards the batch
* had just written; a user who joins later is warmed by the visit or the next
* night. Their schedules and queues are removed on an instance that still has
* them.
*/
const RETIRED_STATUS_QUEUES = [
"insights-blood-pressure-status",
"insights-weight-status",
"insights-pulse-status",
"insights-bmi-status",
"insights-mood-status",
"insights-medication-compliance-status",
] as const;
const allQueues = [
GENERAL_STATUS_QUEUE,
BLOOD_PRESSURE_STATUS_QUEUE,
WEIGHT_STATUS_QUEUE,
PULSE_STATUS_QUEUE,
BMI_STATUS_QUEUE,
// v1.15.20 — mood joins the nightly status ladder. The queue MUST be
// registered here or pg-boss never provisions it and the 02:30 schedule
// silently no-ops (the v1.4.37 dead-queue class).
MOOD_STATUS_QUEUE,
MEDICATION_COMPLIANCE_STATUS_QUEUE,
// v1.7.0 — nightly comprehensive-insight pre-generation so the daily
// briefing is warm before the user opens /insights or the dashboard
// snapshot. Without this entry the 04:30 schedule silently no-ops.
@@ -238,21 +207,6 @@ const allQueues = [
const schedules: ScheduleEntry[] = [
[GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, nightlyInsightPassOptions],
[
BLOOD_PRESSURE_STATUS_QUEUE,
BLOOD_PRESSURE_STATUS_CRON,
nightlyInsightPassOptions,
],
[WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, nightlyInsightPassOptions],
[PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, nightlyInsightPassOptions],
[BMI_STATUS_QUEUE, BMI_STATUS_CRON, nightlyInsightPassOptions],
// v1.15.20 — mood status nightly, continuing the 02:xx ladder.
[MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, nightlyInsightPassOptions],
[
MEDICATION_COMPLIANCE_STATUS_QUEUE,
MEDICATION_COMPLIANCE_STATUS_CRON,
nightlyInsightPassOptions,
],
// v1.7.0 — nightly 04:30 Europe/Berlin comprehensive-insight
// pre-generation. Budget-gated per user inside the handler.
[
@@ -383,6 +337,7 @@ export async function registerStatusQueues(
boss: PgBoss,
): Promise<readonly string[]> {
await createAndSchedule(boss, allQueues, schedules, queuePolicies);
await retireQueues(boss, RETIRED_STATUS_QUEUES);
await createAndWork<GeneralStatusPayload>(
boss,
@@ -394,54 +349,6 @@ export async function registerStatusQueues(
handleGeneralStatusGenerate,
),
);
await createAndWork<BloodPressureStatusPayload>(
boss,
BLOOD_PRESSURE_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(
BLOOD_PRESSURE_STATUS_QUEUE,
() => WHOLE_PASS,
handleBloodPressureStatusGenerate,
),
);
await createAndWork<WeightStatusPayload>(
boss,
WEIGHT_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(
WEIGHT_STATUS_QUEUE,
() => WHOLE_PASS,
handleWeightStatusGenerate,
),
);
await createAndWork<PulseStatusPayload>(
boss,
PULSE_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(PULSE_STATUS_QUEUE, () => WHOLE_PASS, handlePulseStatusGenerate),
);
await createAndWork<BmiStatusPayload>(
boss,
BMI_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(BMI_STATUS_QUEUE, () => WHOLE_PASS, handleBmiStatusGenerate),
);
await createAndWork<MoodStatusPayload>(
boss,
MOOD_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(MOOD_STATUS_QUEUE, () => WHOLE_PASS, handleMoodStatusGenerate),
);
await createAndWork<MedicationComplianceStatusPayload>(
boss,
MEDICATION_COMPLIANCE_STATUS_QUEUE,
{ localConcurrency: 1 },
lockedPass(
MEDICATION_COMPLIANCE_STATUS_QUEUE,
() => WHOLE_PASS,
handleMedicationComplianceStatusGenerate,
),
);
// v1.7.0 — nightly comprehensive-insight pre-generation.
//
// v1.16.8 — localConcurrency 2 (was 1). The queue carries BOTH the
+21 -1
View File
@@ -169,7 +169,7 @@ export const insightRetryOptions = {
} as const;
/**
* Send options for a nightly AI pass over every account (the status crons).
* Send options for a nightly AI pass over every account (the status batch).
*
* The retries of `insightRetryOptions`, plus a two-hour expiry. Each pass
* makes a provider call per account, and on a large instance that outlasted
@@ -333,6 +333,26 @@ export async function createAndSchedule(
}
}
/**
* Remove queues the tree no longer declares: their schedule first, so the
* timekeeper stops minting jobs nobody works, then the queue with whatever it
* still holds. A queue that is already gone is a no-op, so every boot after
* the first does nothing. Never fails a boot.
*/
export async function retireQueues(
boss: PgBoss,
queues: readonly string[],
): Promise<void> {
for (const name of queues) {
try {
await boss.unschedule(name);
await boss.deleteQueue(name);
} catch (err) {
workerLog("error", `[queue-retire] failed to retire ${name}`, err);
}
}
}
/**
* Bind a handler to a queue. The one legal way to reach `boss.work` in this
* tree — the `healthlog/job-handler-outcome` lint rule refuses a bare
+9 -9
View File
@@ -1,7 +1,7 @@
/**
* Shared user discovery for the nightly 02:xx per-metric status crons
* (general / blood-pressure / weight / pulse / bmi / medication-
* compliance / mood).
* User discovery for the nightly 02:00 status batch, which writes every
* per-metric status card (general / blood-pressure / weight / pulse / bmi /
* medication-compliance / mood).
*
* Before this module each cron iterated EVERY user row: accounts that
* disabled the coach surface (`disableCoach`) and operator-disabled
@@ -12,18 +12,18 @@
* Division of nightly labour (documented here once, referenced by the
* handlers):
*
* - 02:xx status crons → users the 04:30 `insight-pregenerate` pass will
* - 02:00 status batch → users the 04:30 `insight-pregenerate` pass will
* not reach: a comprehensive cache fresher than `PREGENERATE_STALE_MS`,
* so that pass skips them, while their per-status notes still age out
* daily and only these crons re-fill them.
* - 04:30 insight-pregenerate → the rest: users with a comprehensive cache
* older than `PREGENERATE_STALE_MS`. That pass warms every per-status
* note whenever `statusText` is available for the user, whatever became
* of the briefing, so a 02:xx generation for them would be redone two
* of the briefing, so a 02:00 generation for them would be redone two
* hours later.
*
* Cheap gates, in SQL and in order (the per-user `statusText` capability in
* each handler, and the chokepoint behind it, are what decide):
* Cheap gates, in SQL and in order (the per-user `statusText` capability
* inside the batch, and the chokepoint behind it, are what decide):
* 1. The operator's `insightStatus` switch (master applied). Off, the whole
* pass is empty.
* 2. The person's AI analysis switch (the `insights` module). Someone who
@@ -32,7 +32,7 @@
* else.
* 3. The pregenerate-candidate skip described above, only while the
* `briefing` switch is on; when it is off the 04:30 pass no-ops, so the
* 02:xx crons keep covering everyone.
* 02:00 batch keeps covering everyone.
*/
import type { PrismaClient } from "@/generated/prisma/client";
import { getAssistantFlags } from "@/lib/feature-flags";
@@ -59,7 +59,7 @@ export async function findStatusCronCandidates(
if (users.length === 0) return [];
// When the briefing switch is off the 04:30 pregenerate pass no-ops, so no
// user has a comprehensive claim and the 02:xx crons keep all.
// user has a comprehensive claim and the 02:00 batch keeps all.
if (!flags.briefing) {
return users.map((u) => ({ id: u.id, locale: u.locale }));
}
-6
View File
@@ -50,12 +50,6 @@ const subsystemSurface = {
// register-status.ts
"insights-general-status": { audience: "system" },
"insights-blood-pressure-status": { audience: "system" },
"insights-weight-status": { audience: "system" },
"insights-pulse-status": { audience: "system" },
"insights-bmi-status": { audience: "system" },
"insights-mood-status": { audience: "system" },
"insights-medication-compliance-status": { audience: "system" },
"insight-pregenerate": { audience: "system" },
"insight-status-generate": { audience: "system" },
"morning-digest-refresh": { audience: "system" },
@@ -4,16 +4,17 @@
* Reads inside one request are memoised on the request's wide event, so a
* page that resolves the AI capabilities five times reads the settings row
* once. A background job opens one wide event for its whole run, and the
* status crons loop over every candidate inside it: memoising there meant the
* status batch loops over every candidate inside it: memoising there meant the
* switches, the modules and the consent kinds were read once at the start of
* the run and served to every later user, and the wire re-check read the same
* cached answer. Turning the Assistant off at 02:05 did not stop the 02:00
* pass.
*
* Driven here over the real loop, the real capability loader and Postgres:
* the first user's generation turns the switch off, and the remaining users
* must not reach a generator, while the wire re-check inside the first user's
* own run must see the change too.
* Driven here over the real loop, the real capability loader and Postgres.
* The batch is stubbed past its first step, the `statusText` capability it
* resolves before building anything: the first user's run turns the switch
* off, the remaining users must not get past that step, and the wire re-check
* inside the first user's own run must see the change too.
*/
import { beforeEach, describe, expect, it, vi } from "vitest";
@@ -27,7 +28,30 @@ vi.mock("@/lib/jobs/status-cron-candidates", () => ({
),
}));
import { runStatusCronGenerate } from "@/lib/jobs/reminder/insights-handlers";
const batch = vi.hoisted(() => ({
run: async (_userId: string): Promise<void> => {},
}));
vi.mock("@/lib/insights/status-batch", async () => {
const { aiCapabilityForJob } = await import("@/lib/ai/capabilities/gate");
return {
generateStatusBatchForUser: vi.fn(async (userId: string) => {
const result = {
served: 0,
batched: 0,
fellBack: 0,
batchCallMade: false,
};
if (!(await aiCapabilityForJob(userId, "statusText")).available) {
return result;
}
await batch.run(userId);
return { ...result, batched: 1, batchCallMade: true };
}),
};
});
import { runStatusBatchCron } from "@/lib/jobs/reminder/insights-handlers";
import { aiEgressRefusal } from "@/lib/ai/capabilities/egress";
let counter = 0;
@@ -74,7 +98,7 @@ describe("a switch turned off mid-run", () => {
const reached: string[] = [];
const wire: Array<string | null> = [];
const generate = vi.fn(async (userId: string) => {
batch.run = async (userId: string) => {
reached.push(userId);
// The wire re-check before the switch moves: open.
wire.push(
@@ -87,9 +111,9 @@ describe("a switch turned off mid-run", () => {
(await aiEgressRefusal("statusText", userId, ["anthropic"]))?.reason ??
null,
);
});
};
await runStatusCronGenerate("job.test.fresh_switch", generate);
await runStatusBatchCron("job.test.fresh_switch", () => false);
expect(reached).toEqual([candidates.ids[0]]);
expect(wire[0]).toBeNull();