From f1145c0e9163868dfa4bf302dc5d1f28b8f5f8cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 2 Jun 2026 16:21:53 +0200 Subject: [PATCH] fix(insights): debounce ingest-driven status invalidation to coalesce the regeneration storm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A constantly-syncing client drips measurement batches every few minutes. Every batch deleted the dirtied per-metric assessment cache rows and re-enqueued a regeneration. The per-(user,scope,locale) enqueue singleton is only 120 s, so it could not coalesce across batches that arrive minutes apart, and each delete dropped the row a fresh enqueue would otherwise have found warm — so a metric card regenerated several times an hour and its assessment felt regenerated on every visit. Before evicting and re-enqueuing, check whether the scope's cached assessment was already (re)generated within a 30 min debounce window. A scope that is fresh is left intact and not re-enqueued; a genuinely stale or missing scope still refreshes immediately, so correctness holds. A timeout stub never counts as fresh, so a scope that recently stalled still retries. One indexed read per user answers it for every candidate scope at once. Co-Authored-By: Claude Opus 4.8 (1M context) (cherry picked from commit 024baea664490ef0494add78c87492f01e61c6b9) --- .../invalidate-status-insights.test.ts | 71 ++++++- src/lib/insights/comprehensive-generate.ts | 197 ++++++++++++++---- 2 files changed, 222 insertions(+), 46 deletions(-) diff --git a/src/lib/insights/__tests__/invalidate-status-insights.test.ts b/src/lib/insights/__tests__/invalidate-status-insights.test.ts index c6c38c9e8..8ba5550e5 100644 --- a/src/lib/insights/__tests__/invalidate-status-insights.test.ts +++ b/src/lib/insights/__tests__/invalidate-status-insights.test.ts @@ -11,12 +11,16 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; const deleteMany = vi.fn(); +const auditFindMany = vi.fn(); const enqueueStatusGeneration = vi.fn(); const userFindUnique = vi.fn(); vi.mock("@/lib/db", () => ({ prisma: { - auditLog: { deleteMany: (...a: unknown[]) => deleteMany(...a) }, + auditLog: { + deleteMany: (...a: unknown[]) => deleteMany(...a), + findMany: (...a: unknown[]) => auditFindMany(...a), + }, user: { findUnique: (...a: unknown[]) => userFindUnique(...a) }, }, })); @@ -30,6 +34,9 @@ import { invalidateStatusInsightsForTypes } from "../comprehensive-generate"; beforeEach(() => { vi.clearAllMocks(); deleteMany.mockResolvedValue({ count: 0 }); + // Default: no recently-warmed cache rows, so every dirtied scope is stale + // and refreshes — the pre-debounce contract the bulk of these tests pin. + auditFindMany.mockResolvedValue([]); enqueueStatusGeneration.mockResolvedValue(undefined); userFindUnique.mockResolvedValue({ locale: "de" }); }); @@ -176,4 +183,66 @@ describe("invalidateStatusInsightsForTypes", () => { enqueuedScopes().some((s) => s.startsWith("metric:")), ).toBe(false); }); + + describe("ingest-invalidation debounce (v1.9.0)", () => { + /** Build a recent (within-window) real assessment cache row. */ + function freshRow(scope: string, locale = "de") { + return { + action: `insights.${scope}-status.${locale}`, + details: JSON.stringify({ text: "fresh assessment", model: "gpt" }), + }; + } + + it("skips a scope whose assessment was warmed within the window", async () => { + // `general` was regenerated minutes ago; a fresh WEIGHT sample must not + // delete or re-enqueue it. weight + bmi are still stale, so they refresh. + auditFindMany.mockResolvedValue([freshRow("general")]); + await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]); + expect(deletedScopes()).toEqual(["bmi", "weight"]); + expect(enqueuedScopes()).toEqual(["bmi", "weight"]); + }); + + it("is a complete no-op (no delete, no enqueue) when every scope is fresh", async () => { + auditFindMany.mockResolvedValue([ + freshRow("weight"), + freshRow("bmi"), + freshRow("general"), + ]); + await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]); + expect(deleteMany).not.toHaveBeenCalled(); + expect(enqueueStatusGeneration).not.toHaveBeenCalled(); + }); + + it("treats a recent timeout stub as not-fresh and still refreshes the scope", async () => { + // A stub carries no real assessment, so a scope that recently stalled + // must retry rather than be debounced into staying cold. + auditFindMany.mockResolvedValue([ + { + action: "insights.general-status.de", + details: JSON.stringify({ model: "timeout-stub", text: "stub" }), + }, + ]); + await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]); + expect(deletedScopes()).toEqual(["bmi", "general", "weight"]); + expect(enqueuedScopes()).toEqual(["bmi", "general", "weight"]); + }); + + it("scopes the freshness probe to the user's resolved locale and the window cutoff", async () => { + userFindUnique.mockResolvedValue({ locale: "en" }); + await invalidateStatusInsightsForTypes("u1", ["WEIGHT"]); + const where = auditFindMany.mock.calls[0][0].where; + expect(where.userId).toBe("u1"); + // Only the en cache actions for the dirtied scopes are probed. + expect(new Set(where.action.in)).toEqual( + new Set([ + "insights.weight-status.en", + "insights.bmi-status.en", + "insights.general-status.en", + ]), + ); + // A recency floor is applied (createdAt >= cutoff in the past). + expect(where.createdAt.gte).toBeInstanceOf(Date); + expect(where.createdAt.gte.getTime()).toBeLessThan(Date.now()); + }); + }); }); diff --git a/src/lib/insights/comprehensive-generate.ts b/src/lib/insights/comprehensive-generate.ts index aaa54229b..7b6705532 100644 --- a/src/lib/insights/comprehensive-generate.ts +++ b/src/lib/insights/comprehensive-generate.ts @@ -47,15 +47,42 @@ import { runRawCompletionWithFallback, } from "@/lib/ai/provider-runner"; import { invalidateUserInsights } from "@/lib/cache/invalidate"; -import { enqueueStatusGeneration } from "@/lib/jobs/insight-status-generate-shared"; +import { + enqueueStatusGeneration, + type InsightStatusScope, +} from "@/lib/jobs/insight-status-generate-shared"; import { normalizeLocale } from "@/lib/insights/status-shared"; +import { isTimeoutStub } from "@/lib/insights/status-cache"; import { metricIdForMeasurementType, metricStatusScope, } from "@/lib/insights/metric-status-registry"; +import { annotate } from "@/lib/logging/context"; const CACHE_TTL_MS = 24 * 60 * 60 * 1000; +/** + * Debounce window for ingest-driven status invalidation. A scope whose + * cached assessment was (re)generated within this window is left intact on + * a fresh measurement ingest — neither evicted nor re-enqueued. + * + * A constantly-syncing client (Apple Health drips batches every few + * minutes) used to delete + re-enqueue every dirtied scope on every batch, + * and because each delete dropped the cache row the per-(user,scope,locale) + * enqueue singleton (120 s) could not coalesce across batches that arrive + * minutes apart — so STEPS / general / a metric card regenerated several + * times an hour and the assessment felt "regenerated on every visit". This + * window coalesces the storm: a genuinely stale scope (no fresh assessment) + * still refreshes immediately, but a scope refreshed inside the window is + * skipped, so a fresh assessment survives the day's sync drip and a burst + * of batches costs at most one regeneration per scope per window. + * + * 30 min is comfortably longer than a typical sync cadence yet far shorter + * than the 24 h assessment TTL, so a card still tracks new data within the + * same session without thrashing the provider. + */ +const INGEST_INVALIDATE_DEBOUNCE_MS = 30 * 60 * 1000; + const MAX_DOWNGRADE_TOKENS: ReadonlyArray = [ "anthropometrics", "medications", @@ -221,52 +248,26 @@ export async function invalidateStatusInsightsForTypes( userId: string, types: Iterable, ): Promise { - const scopes = new Set(); - // v1.8.7.1 — a fresh measurement of a type that backs a generic - // HealthKit assessment card (the ~30 registry metrics) also dirties that - // card's `metric:` scope. Collect those alongside the seven - // specialised scopes so the generic tier is invalidated + re-warmed on - // ingest with the same debounced enqueue, rather than lagging until the - // nightly cron. Only registered, data-bearing types map here; the seven - // specialised metrics and any unregistered type resolve to null and are - // skipped, so the constant sync cannot fan out to unwanted scopes. - const metricScopes = new Set<`metric:${string}`>(); + // One ordered set of every scope the batch dirties. The seven specialised + // scopes are bare slugs (`weight`, `general`, …); the generic HealthKit + // cards (v1.8.7.1) carry a `metric:` prefix. Both share the cache-key + // shape `insights.-status.`, so a single set covers the + // eviction + the debounce filter + the enqueue uniformly. Only registered, + // data-bearing types contribute a generic scope; the seven specialised + // metrics and any unregistered type resolve to null and are skipped, so + // the constant sync cannot fan out to unwanted scopes. + const scopes = new Set(); for (const type of types) { for (const scope of statusScopesForMeasurementType(type)) { scopes.add(scope); } const metricId = metricIdForMeasurementType(type); if (metricId) { - metricScopes.add(metricStatusScope(metricId)); + scopes.add(metricStatusScope(metricId)); } } - if (scopes.size === 0 && metricScopes.size === 0) return; + if (scopes.size === 0) return; - await prisma.auditLog.deleteMany({ - where: { - userId, - OR: [ - ...Array.from(scopes, (scope) => ({ - action: { startsWith: `insights.${scope}-status.` }, - })), - ...Array.from(metricScopes, (scope) => ({ - action: { startsWith: `insights.${scope}-status.` }, - })), - ], - }, - }); - - // v1.8.7 — regenerate-on-invalidate. Deleting the today-row alone left - // the card to re-warm only on the user's next category open (a miss → a - // worker round-trip while they wait) or the nightly cron. Instead, - // proactively enqueue a debounced regenerate for each dirtied scope so - // the cache is re-warmed in the background. The enqueue is coalesced per - // `(user, metric, locale)` via the queue's `singletonKey` (120 s window), - // so the constant Apple-Health sync cannot fan out a storm of jobs — a - // burst of samples collapses into one regenerate per scope. - // The stale-while-revalidate read keeps the previous assessment visible - // until the fresh one lands, so the user never waits on a card. - // // v1.8.7 — regenerate only the user's resolved locale, matching the // read-path (every `*-status` GET serves `normalizeLocale(user.locale)`). // Warming both locales doubled provider spend on every sync, half of it @@ -277,16 +278,122 @@ export async function invalidateStatusInsightsForTypes( select: { locale: true }, }); const locale = normalizeLocale(localeRow?.locale); + + // v1.9.0 — debounce the ingest-invalidation storm. A constantly-syncing + // client drips batches every few minutes; deleting + re-enqueuing every + // dirtied scope on every batch regenerated the same card several times an + // hour (the per-(user,scope,locale) enqueue singleton is only 120 s, so it + // could not coalesce across batches arriving minutes apart, and each + // delete dropped the row a fresh enqueue would otherwise have found warm). + // Skip any scope whose cached assessment was (re)generated within the + // debounce window — leave its row intact and do NOT re-enqueue. A + // genuinely stale or missing scope still refreshes immediately, so + // correctness holds; only the redundant churn is removed. + const freshScopes = await findRecentlyWarmedScopes(userId, locale, scopes); + const staleScopes = Array.from(scopes).filter( + (scope) => !freshScopes.has(scope), + ); + if (staleScopes.length === 0) { + annotate({ + action: { name: "insights.status.invalidate.debounced" }, + meta: { skipped: scopes.size, refreshed: 0 }, + }); + return; + } + + // Drop the cached per-metric assessment rows that are actually stale, so + // the next mount / nightly warm pass regenerates them against the new + // data. The `-status.` substring guard keeps the sweep off the + // comprehensive cache and any unrelated `insights.*` audit row. + await prisma.auditLog.deleteMany({ + where: { + userId, + OR: staleScopes.map((scope) => ({ + action: { startsWith: `insights.${scope}-status.` }, + })), + }, + }); + + // v1.8.7 — regenerate-on-invalidate. Deleting the today-row alone left + // the card to re-warm only on the user's next category open (a miss → a + // worker round-trip while they wait) or the nightly cron. Instead, + // proactively enqueue a debounced regenerate for each dirtied scope so + // the cache is re-warmed in the background. The enqueue is coalesced per + // `(user, metric, locale)` via the queue's `singletonKey` (120 s window); + // the debounce above is the second, wider coalescing layer that survives + // across the sync drip. The stale-while-revalidate read keeps the + // previous assessment visible until the fresh one lands. + for (const scope of staleScopes) { + void enqueueStatusGeneration({ userId, metric: scope, locale }); + } + + annotate({ + action: { name: "insights.status.invalidate.debounced" }, + meta: { skipped: freshScopes.size, refreshed: staleScopes.length }, + }); +} + +/** + * v1.9.0 — return the subset of `scopes` whose cached assessment for + * `locale` was generated within `INGEST_INVALIDATE_DEBOUNCE_MS` and is a + * real (non-stub) assessment. Those scopes are skipped by the ingest + * invalidator so a fresh assessment survives the sync drip. + * + * One indexed read per user (a single `findMany` over this user's recent + * status-cache rows, newest-first) answers it for every candidate scope at + * once — cheaper than a per-scope probe and bounded by `take`. A timeout + * stub never counts as fresh (it carries no real assessment), so a scope + * that recently stalled still gets a retry enqueued. + */ +async function findRecentlyWarmedScopes( + userId: string, + locale: "de" | "en", + scopes: ReadonlySet, +): Promise> { + 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 actionToScope = new Map(); for (const scope of scopes) { - void enqueueStatusGeneration({ userId, metric: scope, locale }); + actionToScope.set(`insights.${scope}-status.${locale}`, scope); } - // v1.8.7.1 — re-warm the matching generic metric scopes the same way, so - // a freshly-synced HealthKit metric's assessment card refreshes in the - // background instead of lagging until the nightly warm pass. Same - // single-locale, singleton-debounced enqueue as the seven scopes above. - for (const scope of metricScopes) { - void enqueueStatusGeneration({ userId, metric: scope, locale }); + + const rows = await prisma.auditLog.findMany({ + where: { + userId, + action: { in: candidateActions }, + createdAt: { gte: cutoff }, + }, + orderBy: { createdAt: "desc" }, + select: { action: true, details: true }, + }); + + const fresh = new Set(); + for (const row of rows) { + const scope = actionToScope.get(row.action); + if (!scope || fresh.has(scope)) continue; + if (!row.details) continue; + try { + const parsed = JSON.parse(row.details) as { + model?: string; + timeout?: boolean; + text?: string; + }; + // A stub is not a real assessment — let a stalled scope retry. + if (isTimeoutStub(parsed)) continue; + if (typeof parsed.text !== "string" || parsed.text.trim().length === 0) { + continue; + } + fresh.add(scope); + } catch { + // Malformed payload — treat as not-fresh so the scope refreshes. + continue; + } } + return fresh; } interface GenerateOptions {