fix(insights): stop a timed-out warm from evicting its own freshly-warmed caches

The on-demand force-warm bounds the comprehensive briefing with a 45 s
timeout and then warms the per-status + generic-metric caches itself. The
timeout could not cancel the underlying generation, so a comprehensive that
resolved after the timeout reached its own per-status eviction and deleted
the rows the warm passes had just written — leaving the user back on a cold
cache.

Thread an AbortSignal into the comprehensive generation and abort it when the
timeout fires; the generation now bails out before its write + evict once the
signal is aborted, so a late resolve can never undo the warm.

Also align the force-warm gating with the route: the route admits the job on
the per-user assessment surface, so warm the per-status + generic-metric
cards whenever that surface is enabled, even when the global briefing switch
is off. Only the comprehensive briefing stays gated on the briefing switch.

Factor the per-status cache-action shape into a single helper so the key is
built in one place rather than hand-assembled across the generators, the
eviction sweep, and the ingest debounce.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
(cherry picked from commit 75e0b5a9129893915f7a213ec96253df19341d15)
This commit is contained in:
Marc-André Bombeck committed 2026-06-02 17:34:52 +02:00
1 parent 94415e468e
commit b13721cc97
13 files changed
+237 -48

No files matched your search

+2 -1
View File
@@ -45,6 +45,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -112,7 +113,7 @@ export async function generateBloodPressureStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.blood-pressure-status.${locale}`;
const cacheAction = statusCacheAction("blood-pressure", locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
+2 -1
View File
@@ -25,6 +25,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -50,7 +51,7 @@ export async function generateBmiStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.bmi-status.${locale}`;
const cacheAction = statusCacheAction("bmi", locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
+30 -6
View File
@@ -52,7 +52,11 @@ import {
type InsightStatusScope,
} from "@/lib/jobs/insight-status-generate-shared";
import { normalizeLocale } from "@/lib/insights/status-shared";
import { isTimeoutStub } from "@/lib/insights/status-cache";
import {
isTimeoutStub,
statusCacheAction,
statusCacheActionPrefix,
} from "@/lib/insights/status-cache";
import {
metricIdForMeasurementType,
metricStatusScope,
@@ -309,7 +313,7 @@ export async function invalidateStatusInsightsForTypes(
where: {
userId,
OR: staleScopes.map((scope) => ({
action: { startsWith: `insights.${scope}-status.` },
action: { startsWith: statusCacheActionPrefix(scope) },
})),
},
});
@@ -352,13 +356,12 @@ async function findRecentlyWarmedScopes(
): Promise<Set<InsightStatusScope>> {
const cutoff = new Date(Date.now() - INGEST_INVALIDATE_DEBOUNCE_MS);
// Match exactly the cache actions for the candidate scopes in this locale.
const candidateActions = Array.from(
scopes,
(scope) => `insights.${scope}-status.${locale}`,
const candidateActions = Array.from(scopes, (scope) =>
statusCacheAction(scope, locale),
);
const actionToScope = new Map<string, InsightStatusScope>();
for (const scope of scopes) {
actionToScope.set(`insights.${scope}-status.${locale}`, scope);
actionToScope.set(statusCacheAction(scope, locale), scope);
}
const rows = await prisma.auditLog.findMany({
@@ -401,6 +404,17 @@ interface GenerateOptions {
locale: "de" | "en";
/** Skip the 24 h cache short-circuit and force a fresh generation. */
force?: boolean;
/**
* v1.9.0 — abort the generation when the caller has given up on it.
*
* The on-demand `forceWarmUser` path bounds the comprehensive step with a
* timeout and then proceeds to warm the per-status caches itself. Without
* this signal a comprehensive that resolves AFTER the timeout would still
* reach `evictPerStatusInsightCache` and delete the rows the warm passes
* had just written — undoing its own work. The signal lets the caller cut
* the generation off before the eviction so the warmed rows survive.
*/
signal?: AbortSignal;
}
/**
@@ -420,6 +434,7 @@ export async function generateComprehensiveInsight(
): Promise<GenerateOutcome> {
const { locale } = options;
const force = options.force === true;
const signal = options.signal;
const dbUser = await prisma.user.findUnique({
where: { id: userId },
@@ -527,6 +542,15 @@ export async function generateComprehensiveInsight(
return { status: "failed", reason: "invalid-json" };
}
// v1.9.0 — the caller abandoned this generation (its bounded timeout
// fired and it has moved on to warm the per-status caches itself). Bail
// out before the write + evict so a late-resolving comprehensive can never
// delete the rows the warm passes have since written. The provider call
// already cost what it cost; what we must NOT do now is touch the cache.
if (signal?.aborted) {
return { status: "failed", reason: "aborted" };
}
await prisma.user.update({
where: { id: userId },
data: {
+2 -1
View File
@@ -28,6 +28,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -65,7 +66,7 @@ export async function generateGeneralStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.general-status.${locale}`;
const cacheAction = statusCacheAction("general", locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
@@ -30,6 +30,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
isTimeoutStub,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -65,7 +66,7 @@ export async function generateMedicationComplianceStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.medication-compliance-status.${locale}`;
const cacheAction = statusCacheAction("medication-compliance", locale);
const todayKey = toBerlinDayKey(new Date());
// This route carries a richer cached envelope (`summary` +
+2 -1
View File
@@ -55,6 +55,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -124,7 +125,7 @@ export async function generateMetricStatus(args: {
const force = args.force === true;
const readOnly = args.readOnly === true;
const scope = metricStatusScope(meta.id);
const cacheAction = `insights.${scope}-status.${locale}`;
const cacheAction = statusCacheAction(scope, locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
+2 -1
View File
@@ -38,6 +38,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -63,7 +64,7 @@ export async function generateMoodStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.mood-status.${locale}`;
const cacheAction = statusCacheAction("mood", locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
+2 -1
View File
@@ -31,6 +31,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -56,7 +57,7 @@ export async function generatePulseStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.pulse-status.${locale}`;
const cacheAction = statusCacheAction("pulse", locale);
const todayKey = toBerlinDayKey(new Date());
const cached = await readFreshStatusText({
+21 -1
View File
@@ -28,6 +28,26 @@ import { annotate } from "@/lib/logging/context";
* the fallback for the day, and a fresh generation is attempted instead.
*/
/**
* The single source of truth for the per-status cache-action shape.
*
* Every per-metric assessment is persisted as an `auditLog` row whose
* `action` is `insights.<scope>-status.<locale>`. The shape IS the cache
* key, so building it by hand in multiple places risks the same silent
* drift the queryKey factory guards against on the client. `statusCacheAction`
* is the one builder; `statusCacheActionPrefix` is its locale-agnostic
* sibling for the `startsWith` eviction that drops every locale variant of a
* scope at once.
*/
export function statusCacheAction(scope: string, locale: string): string {
return `insights.${scope}-status.${locale}`;
}
/** Locale-agnostic prefix for `startsWith`-eviction of every locale of a scope. */
export function statusCacheActionPrefix(scope: string): string {
return `insights.${scope}-status.`;
}
interface ParsedStatusCache {
dateKey?: string;
locale?: string;
@@ -194,7 +214,7 @@ export async function resolveReadOnlyStatusMiss(args: {
const hasProvider = await hasUsableStatusProvider(args.userId);
if (!hasProvider) return { kind: "no-provider" };
const cacheAction = `insights.${args.metric}-status.${args.locale}`;
const cacheAction = statusCacheAction(args.metric, args.locale);
// v1.8.7 — stale-while-revalidate. Surface the last good (non-stub)
// assessment for this scope so the card renders the previous text
+2 -1
View File
@@ -35,6 +35,7 @@ import { runStatusCompletion } from "@/lib/insights/status-provider";
import {
readFreshStatusText,
resolveReadOnlyStatusMiss,
statusCacheAction,
} from "@/lib/insights/status-cache";
import { returnTimeoutFallback } from "@/lib/insights/timeout-fallback";
import { annotate } from "@/lib/logging/context";
@@ -105,7 +106,7 @@ export async function generateWeightStatusForUser(
const locale = normalizeLocale(options?.locale);
const force = options?.force === true;
const readOnly = options?.readOnly === true;
const cacheAction = `insights.weight-status.${locale}`;
const cacheAction = statusCacheAction("weight", locale);
const todayKey = toBerlinDayKey(new Date());
// Serve today's real assessment when present. The shared reader
+9
View File
@@ -32,6 +32,14 @@ export function withTimeout<T>(
start: () => Promise<T>,
ms: number,
fallback: T,
/**
* v1.9.0 — fired once when the budget expires before the upstream settles.
* `withTimeout` cannot cancel the underlying promise, so a caller that owns
* a cancellation handle (e.g. an `AbortController`) passes it here to cut
* the detached work off — preventing a late-resolving promise from running
* side effects the caller has already moved past.
*/
onTimeout?: () => void,
): Promise<TimeoutEnvelope<T>> {
let timer: ReturnType<typeof setTimeout> | null = null;
let settled = false;
@@ -44,6 +52,7 @@ export function withTimeout<T>(
action: { name: "insights.status.provider_timeout" },
meta: { timeout_ms: ms },
});
onTimeout?.();
resolve({ timedOut: true, errored: false, value: fallback });
}, ms);
@@ -82,7 +82,11 @@ function makePrisma(users: Array<{ id: string; locale: string | null }>) {
beforeEach(() => {
vi.clearAllMocks();
getAssistantFlags.mockResolvedValue({ enabled: true, briefing: true });
getAssistantFlags.mockResolvedValue({
enabled: true,
briefing: true,
insightStatus: true,
});
checkRateLimit.mockResolvedValue({ allowed: true });
});
@@ -395,9 +399,14 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => {
warmGenericMetrics,
});
// Comprehensive forced once, in the active locale only.
// Comprehensive forced once, in the active locale only. The forced path
// also threads an AbortSignal so a timeout can cut the generation off.
expect(generate).toHaveBeenCalledTimes(1);
expect(generate).toHaveBeenCalledWith("u1", { locale: "en", force: true });
expect(generate).toHaveBeenCalledWith("u1", {
locale: "en",
force: true,
signal: expect.any(AbortSignal),
});
// Each specialised generator forced exactly once — the active locale,
// NOT both de+en like the nightly cron.
for (const g of statusGenerators) {
@@ -528,17 +537,116 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => {
}
});
it("short-circuits to a no-op when the briefing surface is disabled globally", async () => {
getAssistantFlags.mockResolvedValue({ enabled: false, briefing: false });
it("short-circuits to a no-op when the whole assistant is disabled globally", async () => {
getAssistantFlags.mockResolvedValue({
enabled: false,
briefing: false,
insightStatus: false,
});
const { prisma } = makePrisma([]);
const generate = vi.fn();
const statusGenerators = Array.from({ length: 7 }, () =>
vi.fn().mockResolvedValue({ hasProvider: true, cached: false }),
);
const result = await forceWarmUser(prisma as never, "u1", "de", {
generate,
statusGenerators,
});
expect(generate).not.toHaveBeenCalled();
for (const g of statusGenerators) {
expect(g).not.toHaveBeenCalled();
}
expect(result.comprehensive).toBe("skipped");
expect(result.assessmentsWarmed).toBe(0);
});
it("warms the per-status + generic caches when `briefing` is off but `insightStatus` is on (L1)", async () => {
// v1.9.0 — the route admits the job on the per-user `insightStatus`
// surface. A global `briefing` kill-switch must not suppress the
// assessment cards the user has enabled; only the comprehensive briefing
// belongs to the `briefing` surface.
getAssistantFlags.mockResolvedValue({
enabled: true,
briefing: false,
insightStatus: true,
});
const { prisma } = makePrisma([]);
const generate = vi.fn();
const statusGenerators = Array.from({ length: 7 }, () =>
vi.fn().mockResolvedValue({ hasProvider: true, cached: false }),
);
const warmGenericMetrics = vi.fn().mockResolvedValue(2);
const result = await forceWarmUser(prisma as never, "u1", "de", {
generate,
statusGenerators,
warmGenericMetrics,
});
// Comprehensive belongs to the `briefing` surface — skipped here.
expect(generate).not.toHaveBeenCalled();
// But the per-status + generic passes warm.
for (const g of statusGenerators) {
expect(g).toHaveBeenCalledTimes(1);
}
expect(warmGenericMetrics).toHaveBeenCalledWith("u1", ["de"]);
expect(result).toMatchObject({
comprehensive: "skipped",
assessmentsWarmed: 7,
metricAssessmentsWarmed: 2,
});
});
it("aborts the comprehensive on timeout so its late resolve cannot evict the warmed rows (M-1)", async () => {
// v1.9.0 race fix: `withTimeout` cannot cancel the detached generation,
// so a comprehensive that resolves AFTER the timeout would reach its own
// `evictPerStatusInsightCache` and delete the rows the warm passes wrote.
// The forced path threads an AbortController; on timeout it must abort,
// and the generation observes `signal.aborted` before its evict.
const { prisma } = makePrisma([]);
let observedSignal: AbortSignal | undefined;
const generate = vi.fn().mockImplementation(
(
_userId: string,
opts: { signal?: AbortSignal },
) =>
new Promise((resolve) => {
observedSignal = opts.signal;
setTimeout(
() => resolve({ status: "generated", providerType: "x" }),
120_000,
);
}),
);
const statusGenerators = Array.from({ length: 7 }, () =>
vi.fn().mockResolvedValue({ hasProvider: true, cached: false }),
);
const warmGenericMetrics = vi.fn().mockResolvedValue(3);
vi.useFakeTimers();
try {
const promise = forceWarmUser(prisma as never, "u1", "de", {
generate,
statusGenerators,
warmGenericMetrics,
});
await vi.advanceTimersByTimeAsync(60_000);
const result = await promise;
// The timeout fired and aborted the still-running generation.
expect(observedSignal?.aborted).toBe(true);
expect(result.comprehensive).toBe("timeout");
// The warm passes ran and are not undone by the abandoned comprehensive.
for (const g of statusGenerators) {
expect(g).toHaveBeenCalledTimes(1);
}
expect(result.assessmentsWarmed).toBe(7);
expect(result.metricAssessmentsWarmed).toBe(3);
} finally {
vi.useRealTimers();
}
});
});
+48 -28
View File
@@ -534,7 +534,7 @@ export async function forceWarmUser(
options: {
generate?: (
userId: string,
opts: { locale: "de" | "en"; force?: boolean },
opts: { locale: "de" | "en"; force?: boolean; signal?: AbortSignal },
) => Promise<GenerateOutcome>;
statusGenerators?: ReadonlyArray<StatusGenerator>;
warmGenericMetrics?: (
@@ -557,10 +557,17 @@ export async function forceWarmUser(
metricAssessmentsWarmed: 0,
};
// Master kill-switch — when the operator disabled the briefing surface
// globally there is nothing to warm, on demand or otherwise.
// v1.9.0 — gate each section on the flag for the surface it actually
// warms, matching the route. The HTTP route admits this job on the per-user
// `insightStatus` surface, so warming the per-status + generic-metric cards
// must run whenever `insightStatus` is enabled. Only the comprehensive
// briefing belongs to the `briefing` surface — gating the whole warm on
// `briefing` (the previous behaviour) let an operator's global `briefing`
// kill-switch silently suppress the assessment cards the user has enabled.
// When the master assistant switch is off both flags resolve false, so the
// whole thing still no-ops.
const flags = await getAssistantFlags();
if (!flags.briefing) return result;
if (!flags.briefing && !flags.insightStatus) return result;
const locales: ReadonlyArray<"de" | "en"> = [locale];
@@ -571,32 +578,45 @@ export async function forceWarmUser(
// comprehensive insight (each generator resolves its own provider chain
// and writes its own cache row). A timeout / error / skipped comprehensive
// now falls through to the warm passes instead of aborting the whole job.
const comprehensive = await withTimeout(
() => generate(userId, { locale, force: true }),
FORCE_WARM_COMPREHENSIVE_TIMEOUT_MS,
null,
);
if (comprehensive.timedOut) {
result.comprehensive = "timeout";
} else if (comprehensive.errored || comprehensive.value === null) {
result.comprehensive = "failed";
} else {
result.comprehensive = comprehensive.value.status;
//
// On a timeout `withTimeout` cannot cancel the detached generation; without
// a cut-off a late-resolving comprehensive would reach its own
// `evictPerStatusInsightCache` and delete the rows the warm passes write
// below. Thread an AbortController so the timeout aborts the generation
// before it can touch the cache.
if (flags.briefing) {
const controller = new AbortController();
const comprehensive = await withTimeout(
() => generate(userId, { locale, force: true, signal: controller.signal }),
FORCE_WARM_COMPREHENSIVE_TIMEOUT_MS,
null,
() => controller.abort(),
);
if (comprehensive.timedOut) {
result.comprehensive = "timeout";
} else if (comprehensive.errored || comprehensive.value === null) {
result.comprehensive = "failed";
} else {
result.comprehensive = comprehensive.value.status;
}
}
// Always run the per-status + generic-metric warm passes, regardless of
// the comprehensive outcome. Each generator independently no-ops to its
// no-key fallback when no provider is configured (no LLM call) and serves
// its fallback without persisting on a per-card timeout, so running them
// after a failed comprehensive is safe and bounded by their own per-card
// budgets. The only outcome that has nothing to warm is a missing provider,
// and the generators detect that themselves at near-zero cost.
result.assessmentsWarmed = await warmPerStatusCaches(
userId,
locales,
statusGenerators,
);
result.metricAssessmentsWarmed = await warmGenericMetrics(userId, locales);
// Run the per-status + generic-metric warm passes whenever the
// `insightStatus` surface is enabled, regardless of the comprehensive
// outcome. Each generator independently no-ops to its no-key fallback when
// no provider is configured (no LLM call) and serves its fallback without
// persisting on a per-card timeout, so running them after a failed
// comprehensive is safe and bounded by their own per-card budgets. The only
// outcome that has nothing to warm is a missing provider, and the
// generators detect that themselves at near-zero cost.
if (flags.insightStatus) {
result.assessmentsWarmed = await warmPerStatusCaches(
userId,
locales,
statusGenerators,
);
result.metricAssessmentsWarmed = await warmGenericMetrics(userId, locales);
}
annotate({
action: { name: "insights.pregenerate.force" },