From b4166cdc84356d81f9d50cad69489262651f90ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 23 Oct 2026 16:32:10 +0200 Subject: [PATCH] fix(logging): keep detached work from renaming the request's wide event Work a request starts and does not wait for runs inside the request's context, so an annotate({ action }) from it landed on the request's event whenever it ran before the event was emitted. The measurement tail's status re-warm logged a hand-entered reading as insights.status.invalidate.skipped on some requests and as measurement.create on others. runDetached starts such work so that its action is dropped for the event it was detached from; meta, warnings and the auth it reads stay as they were, and an event the work opens for itself keeps its own action. fireAndForget now takes a function and starts it detached, and every trackBackgroundTask handle on a request path is built with runDetached, which a structural guard holds. --- .../background-task-settle-guard.test.ts | 31 ++++--- .../baselines/source-file-lines.json | 4 +- src/__tests__/detached-work-guard.test.ts | 93 +++++++++++++++++++ src/app/api/consent/ai/route.ts | 9 +- src/app/api/labs/ocr/commit/route.ts | 2 +- src/app/api/labs/route.ts | 2 +- src/lib/cache/server-cache.ts | 29 +++--- src/lib/fitbit/jobs/sync-handlers.ts | 2 +- src/lib/jobs/boss-ready.ts | 19 ++-- src/lib/jobs/import-tail.ts | 9 +- src/lib/jobs/reminder/google-health-sync.ts | 4 +- src/lib/jobs/reminder/poll-cohort.ts | 2 +- src/lib/jobs/reminder/whoop-sync.ts | 2 +- src/lib/jobs/reminder/withings-sync.ts | 6 +- src/lib/logging/__tests__/detached.test.ts | 82 ++++++++++++++++ .../logging/__tests__/fire-and-forget.test.ts | 10 +- src/lib/logging/context.ts | 10 +- src/lib/logging/detached.ts | 47 ++++++++++ src/lib/logging/fire-and-forget.ts | 16 ++-- src/lib/measurements/batch-ingest/report.ts | 2 +- .../measurements/create/create-readings.ts | 25 ++--- .../measurements/create/create-sleep-stage.ts | 2 +- .../after-measurement-mutation.test.ts | 34 +++++++ src/lib/rollups/after-measurement-mutation.ts | 9 +- src/lib/rollups/mood-rollups.ts | 39 ++++---- 25 files changed, 386 insertions(+), 104 deletions(-) create mode 100644 src/__tests__/detached-work-guard.test.ts create mode 100644 src/lib/logging/__tests__/detached.test.ts create mode 100644 src/lib/logging/detached.ts diff --git a/src/__tests__/background-task-settle-guard.test.ts b/src/__tests__/background-task-settle-guard.test.ts index 322c7dd6f..e6b564813 100644 --- a/src/__tests__/background-task-settle-guard.test.ts +++ b/src/__tests__/background-task-settle-guard.test.ts @@ -145,9 +145,10 @@ describe("background-task settle contract", () => { it("registers what fireAndForget starts", async () => { let release: () => void = () => {}; fireAndForget( - new Promise((resolve) => { - release = resolve; - }), + () => + new Promise((resolve) => { + release = resolve; + }), { action: "guard.background.registered" }, ); expect(pendingBackgroundTaskCount()).toBe(1); @@ -161,17 +162,19 @@ describe("background-task settle contract", () => { let innerDone = false; fireAndForget( - Promise.resolve().then(() => { - fireAndForget( - new Promise((resolve) => - setTimeout(() => { - innerDone = true; - resolve(); - }, 10), - ), - { action: "guard.background.cascade.inner" }, - ); - }), + () => + Promise.resolve().then(() => { + fireAndForget( + () => + new Promise((resolve) => + setTimeout(() => { + innerDone = true; + resolve(); + }, 10), + ), + { action: "guard.background.cascade.inner" }, + ); + }), { action: "guard.background.cascade.outer" }, ); diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index 0989373ab..822bd738b 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/src/__tests__/baselines/source-file-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", "count": 101, - "total": 116900, + "total": 116897, "entries": { "app/medications/page-client.tsx": 839, "app/page-client.tsx": 2324, @@ -48,7 +48,7 @@ "lib/analytics/sleep-night.ts": 1345, "lib/analytics/summaries-slice.ts": 1248, "lib/auth/session.ts": 811, - "lib/cache/server-cache.ts": 851, + "lib/cache/server-cache.ts": 848, "lib/cycle/prediction.ts": 883, "lib/daily/digest.ts": 1317, "lib/daily/load-digest.ts": 1000, diff --git a/src/__tests__/detached-work-guard.test.ts b/src/__tests__/detached-work-guard.test.ts new file mode 100644 index 000000000..8f3992048 --- /dev/null +++ b/src/__tests__/detached-work-guard.test.ts @@ -0,0 +1,93 @@ +/** + * Structural guard: work a request starts and does not wait for is started + * through `runDetached`, so it cannot rename the request's wide event. + * + * Detached work runs inside the request's context. Before `runDetached`, an + * `annotate({ action })` from it landed on the request's event whenever it + * ran before the event was emitted: the measurement tail's status re-warm + * logged a hand-entered reading as `insights.status.invalidate.skipped` on + * some requests and as `measurement.create` on others. + * + * Two shapes start detached work. `fireAndForget` takes a function and + * starts it detached itself (its signature makes a running promise a type + * error). `trackBackgroundTask` registers a handle, so every call outside + * the logging module has to build that handle with `runDetached`, unless the + * file is listed below with the reason no request can be its caller. + */ +import { describe, expect, it } from "vitest"; + +import { readSourceFolder, stripComments } from "./helpers/source-files"; + +/** Files whose detached work never starts inside a request, and why. */ +const NO_REQUEST_CALLER: Record = { + "lib/db-session.ts": + "the pool's connect hook pins the session time zone; it annotates nothing", + "lib/pwa/query-persister.ts": + "runs in the browser, where there is no wide event (and no async_hooks)", + "lib/jobs/reminder/climb-backfill-at-boot.ts": + "started once at worker boot, outside any request", +}; + +/** The argument text of every `trackBackgroundTask(...)` call in `code`. */ +export function trackedHandles(code: string): string[] { + const out: string[] = []; + for (const m of code.matchAll(/\btrackBackgroundTask\s*\(/g)) { + let i = m.index + m[0].length; + let depth = 1; + while (depth > 0 && i < code.length) { + if (code[i] === "(") depth += 1; + else if (code[i] === ")") depth -= 1; + i += 1; + } + out.push(code.slice(m.index + m[0].length, i - 1)); + } + return out; +} + +const startsDetached = (handle: string) => /^\s*runDetached\s*\(/.test(handle); + +describe("detached work cannot rename the request's event", () => { + const files = readSourceFolder("src", { + floor: 3000, + include: (rel) => + !rel.startsWith("generated/") && + !rel.startsWith("lib/logging/") && + !rel.includes("__tests__/") && + !/\.test\.tsx?$/.test(rel), + }); + const calls = files.flatMap(([rel, text]) => + trackedHandles(stripComments(text)).map((handle) => ({ rel, handle })), + ); + + it("finds the detached starts it polices", () => { + expect(calls.length).toBeGreaterThanOrEqual(8); + expect(calls.map((c) => c.rel)).toContain( + "lib/rollups/after-measurement-mutation.ts", + ); + }); + + it("starts every request-path handle with runDetached", () => { + const offenders = calls + .filter((c) => !(c.rel in NO_REQUEST_CALLER)) + .filter((c) => !startsDetached(c.handle)) + .map((c) => `${c.rel}: ${c.handle.replace(/\s+/g, " ").slice(0, 80)}`); + expect(offenders).toEqual([]); + }); + + it("lists no file that stopped starting detached work", () => { + const callers = new Set(calls.map((c) => c.rel)); + expect( + Object.keys(NO_REQUEST_CALLER).filter((rel) => !callers.has(rel)), + ).toEqual([]); + }); + + it("the matcher reads a nested handle and tells the two shapes apart", () => { + const [plain, detached] = trackedHandles( + "trackBackgroundTask(work(a, b(c)).catch(() => {}));\n" + + "trackBackgroundTask(runDetached(() => work()).catch(() => {}));", + ); + expect(plain).toBe("work(a, b(c)).catch(() => {})"); + expect(startsDetached(plain)).toBe(false); + expect(startsDetached(detached)).toBe(true); + }); +}); diff --git a/src/app/api/consent/ai/route.ts b/src/app/api/consent/ai/route.ts index 86421e8ca..13d49882f 100644 --- a/src/app/api/consent/ai/route.ts +++ b/src/app/api/consent/ai/route.ts @@ -74,10 +74,11 @@ export const POST = apiHandler(async (request: NextRequest) => { // is rebuilt from a backup. Fire-and-forget; audit failures must // never bubble into a user-facing 500. fireAndForget( - auditLog("consent.ai.grant", { - userId: user.id, - details: { kind, receiptId: receipt.id }, - }), + () => + auditLog("consent.ai.grant", { + userId: user.id, + details: { kind, receiptId: receipt.id }, + }), { action: "consent.ai.audit" }, ); diff --git a/src/app/api/labs/ocr/commit/route.ts b/src/app/api/labs/ocr/commit/route.ts index 9db6f0e75..cade37169 100644 --- a/src/app/api/labs/ocr/commit/route.ts +++ b/src/app/api/labs/ocr/commit/route.ts @@ -402,7 +402,7 @@ async function commitOcrRows(request: NextRequest) { // A lab panel just landed — resolve any "annual blood panel" reminders now // rather than waiting on the cron. Fire-and-forget. if (inserted.length > 0) { - fireAndForget(enqueueReminderSatisfy(user.id), { + fireAndForget(() => enqueueReminderSatisfy(user.id), { action: "reminder.satisfy.enqueue", }); diff --git a/src/app/api/labs/route.ts b/src/app/api/labs/route.ts index 4732acab9..65373f80c 100644 --- a/src/app/api/labs/route.ts +++ b/src/app/api/labs/route.ts @@ -373,7 +373,7 @@ async function postLabResult(request: NextRequest) { // v1.18.1 — eventful Lab↔Vorsorge satisfaction. A lab panel just // landed; resolve the user's free-text "annual blood panel" reminders // now rather than waiting on the 15-min cron. Fire-and-forget. - fireAndForget(enqueueReminderSatisfy(user.id), { + fireAndForget(() => enqueueReminderSatisfy(user.id), { action: "reminder.satisfy.enqueue", }); diff --git a/src/lib/cache/server-cache.ts b/src/lib/cache/server-cache.ts index 48a2dd842..265d1215f 100644 --- a/src/lib/cache/server-cache.ts +++ b/src/lib/cache/server-cache.ts @@ -36,6 +36,7 @@ */ import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { runDetached } from "@/lib/logging/detached"; import type { WorkoutsProjection } from "@/lib/workouts/list-read"; export interface ServerCacheOptions { @@ -406,23 +407,19 @@ export class ServerCache { // single-flight slot as `wrap`, so an invalidation that lands // while it is in flight discards its (pre-write) result instead // of re-caching it as fresh. - const refresh = this.startBuild(key, builder, ttlMsOverride).promise; - // Detach: the foreground caller does not await the rebuild. - // Handle the rejection on the detached handle so an - // unhandled-rejection warning never fires — but never silently: - // a refresh that keeps failing means the bucket serves an - // ever-older stale value until the SWR window finally lapses, - // so the failure must reach the logs. There is no request - // context here (the caller already returned), so a structured - // console.error stands in for the wide-event annotation; the - // key is logged as its djb2 hash, matching the cache-outcome - // meta convention, so the userId segment never reaches stdout. - // Registered so the test suite can await it: the error below is a - // console write on a detached handle, and one that lands while a - // vitest worker is closing its RPC channel fails the whole run. - // No-op outside test. + // Detached (`runDetached`): the caller does not await the rebuild, + // and the rebuild cannot rename the caller's event. Its rejection is + // handled, but never silently: a refresh that keeps failing serves + // an ever-older value until the SWR window lapses. It may outlive + // the event, so a structured console.error carries it, the key as + // its djb2 hash (the cache-outcome meta convention) so the userId + // segment never reaches stdout. Registered so the test suite can + // await it: a console write landing while a vitest worker closes + // its RPC channel fails the whole run. No-op outside test. trackBackgroundTask( - refresh.catch((err) => { + runDetached( + () => this.startBuild(key, builder, ttlMsOverride).promise, + ).catch((err) => { console.error( `cache.${this.opts.name ?? "unnamed"}.refresh_failed`, JSON.stringify({ diff --git a/src/lib/fitbit/jobs/sync-handlers.ts b/src/lib/fitbit/jobs/sync-handlers.ts index 3b9ea163d..148c89428 100644 --- a/src/lib/fitbit/jobs/sync-handlers.ts +++ b/src/lib/fitbit/jobs/sync-handlers.ts @@ -79,7 +79,7 @@ export async function handleFitbitSync( // reminders eventfully. Fire-and-forget; the cron is the net. onUserSynced: (userId, imported) => { if (imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } diff --git a/src/lib/jobs/boss-ready.ts b/src/lib/jobs/boss-ready.ts index 1388fcfff..0c06af8f2 100644 --- a/src/lib/jobs/boss-ready.ts +++ b/src/lib/jobs/boss-ready.ts @@ -23,6 +23,7 @@ import type { PgBoss } from "pg-boss"; import { startupPhase } from "@/lib/boot/startup-state"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { runDetached } from "@/lib/logging/detached"; /** The global slot a producer (re)start is parked in while it runs. */ export const BOSS_START_KEY = "__healthlog_pgboss_start__" as const; @@ -80,14 +81,16 @@ export async function enqueueWhenBossReady( } if (!bossStartPending()) return "skipped"; trackBackgroundTask( - whenBossReady(label).then(async (boss) => { - if (!boss) return; - try { - await send(boss); - } catch (err) { - console.warn(`[pg-boss] deferred ${label} enqueue failed:`, err); - } - }), + runDetached(() => + whenBossReady(label).then(async (boss) => { + if (!boss) return; + try { + await send(boss); + } catch (err) { + console.warn(`[pg-boss] deferred ${label} enqueue failed:`, err); + } + }), + ), ); return "deferred"; } diff --git a/src/lib/jobs/import-tail.ts b/src/lib/jobs/import-tail.ts index a63dbf71d..0df38f7ca 100644 --- a/src/lib/jobs/import-tail.ts +++ b/src/lib/jobs/import-tail.ts @@ -25,6 +25,7 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { runDetached } from "@/lib/logging/detached"; import { recomputeUserRollups, ROLLUP_FOLD_WINDOW_MS, @@ -68,9 +69,11 @@ export async function afterBulkImport( if (types.length > 0) { // Registered so the test suite can await the detached handle. trackBackgroundTask( - invalidateStatusInsightsForTypes(userId, types, { - source: "manual", - }).catch((err) => { + runDetached(() => + invalidateStatusInsightsForTypes(userId, types, { + source: "manual", + }), + ).catch((err) => { console.warn( `[${label}] status-insight invalidate failed for user ${userId}`, err, diff --git a/src/lib/jobs/reminder/google-health-sync.ts b/src/lib/jobs/reminder/google-health-sync.ts index 0e49f4f64..44e59eadf 100644 --- a/src/lib/jobs/reminder/google-health-sync.ts +++ b/src/lib/jobs/reminder/google-health-sync.ts @@ -105,7 +105,7 @@ export async function handleGoogleHealthSync( // next hour's, which is not a failure to retry. if (result.failed && shouldStop()) { if (result.imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } @@ -128,7 +128,7 @@ export async function handleGoogleHealthSync( : "complete"; if (result.imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } diff --git a/src/lib/jobs/reminder/poll-cohort.ts b/src/lib/jobs/reminder/poll-cohort.ts index 7bd205f59..25f62ebde 100644 --- a/src/lib/jobs/reminder/poll-cohort.ts +++ b/src/lib/jobs/reminder/poll-cohort.ts @@ -225,7 +225,7 @@ export function makePollCohortHandler({ // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge // reminders eventfully. Fire-and-forget; the cron is the net. if (imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index 429463e18..bf22df904 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -131,7 +131,7 @@ export async function runWhoopResourceSync( // sync tail the webhook / per-resource cron paths pass through — // `syncUserWhoop` covers the manual + backfill paths itself. invalidateUserMeasurements(userId); - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } diff --git a/src/lib/jobs/reminder/withings-sync.ts b/src/lib/jobs/reminder/withings-sync.ts index af4447739..ca5de77a6 100644 --- a/src/lib/jobs/reminder/withings-sync.ts +++ b/src/lib/jobs/reminder/withings-sync.ts @@ -124,7 +124,7 @@ export async function handleWithingsFallbackSync( // v1.18.1 — a fresh reading landed; resolve this user's Vorsorge // reminders eventfully. Fire-and-forget; the cron is the net. if (result.imported > 0) { - fireAndForget(enqueueReminderSatisfy(connection.userId), { + fireAndForget(() => enqueueReminderSatisfy(connection.userId), { action: "reminder.satisfy.enqueue", }); } @@ -304,7 +304,7 @@ export async function handleWithingsActivitySync( imported, }); if (imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } @@ -377,7 +377,7 @@ export async function handleWithingsSleepSync( imported, }); if (imported > 0) { - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); } diff --git a/src/lib/logging/__tests__/detached.test.ts b/src/lib/logging/__tests__/detached.test.ts new file mode 100644 index 000000000..0166e3dd4 --- /dev/null +++ b/src/lib/logging/__tests__/detached.test.ts @@ -0,0 +1,82 @@ +/** + * Detached work keeps its hands off the action of the event it was started + * from, and keeps everything else it may do there. + */ +import { describe, expect, it } from "vitest"; + +import { withBackgroundEvent } from "../background"; +import { settleBackgroundTasks } from "../background-tasks"; +import { annotate, eventStorage, getEvent } from "../context"; +import { runDetached } from "../detached"; +import { WideEventBuilder } from "../event-builder"; +import { fireAndForget } from "../fire-and-forget"; + +describe("runDetached", () => { + it("drops the detached work's action for the parent event and keeps its meta", async () => { + const event = new WideEventBuilder(); + await eventStorage.run(event, async () => { + annotate({ action: { name: "measurement.create" } }); + await runDetached(async () => { + annotate({ + action: { name: "insights.status.invalidate.skipped" }, + meta: { reason: "no_provider" }, + }); + // The request's own event is still the one it reads. + expect(getEvent()).toBe(event); + }); + expect(event.toJSON().action?.name).toBe("measurement.create"); + // Back in the request, the request names its action as before. + annotate({ action: { name: "measurement.create.done" } }); + }); + expect(event.toJSON().action?.name).toBe("measurement.create.done"); + expect(event.toJSON().meta?.reason).toBe("no_provider"); + }); + + it("leaves an event the detached work opens for itself its own action", async () => { + const event = new WideEventBuilder(); + let inner: string | undefined; + await eventStorage.run(event, async () => { + annotate({ action: { name: "measurement.create" } }); + await runDetached(() => + withBackgroundEvent("test.inner", async (evt) => { + annotate({ action: { name: "test.inner.ran" } }); + inner = evt.toJSON().action?.name; + }), + ); + }); + expect(inner).toBe("test.inner.ran"); + expect(event.toJSON().action?.name).toBe("measurement.create"); + }); + + it("starts the work synchronously and turns a synchronous throw into a rejection", async () => { + let started = false; + const event = new WideEventBuilder(); + await eventStorage.run(event, async () => { + const pending = runDetached(async () => { + started = true; + }); + expect(started).toBe(true); + await pending; + await expect( + runDetached(() => { + throw new Error("sync"); + }), + ).rejects.toThrow("sync"); + }); + }); + + it("detaches what fireAndForget starts", async () => { + const event = new WideEventBuilder(); + await eventStorage.run(event, async () => { + annotate({ action: { name: "labs.create" } }); + fireAndForget( + async () => { + annotate({ action: { name: "reminder.satisfy.enqueue" } }); + }, + { action: "reminder.satisfy.enqueue" }, + ); + await settleBackgroundTasks(); + }); + expect(event.toJSON().action?.name).toBe("labs.create"); + }); +}); diff --git a/src/lib/logging/__tests__/fire-and-forget.test.ts b/src/lib/logging/__tests__/fire-and-forget.test.ts index a0fbb3e23..608151d68 100644 --- a/src/lib/logging/__tests__/fire-and-forget.test.ts +++ b/src/lib/logging/__tests__/fire-and-forget.test.ts @@ -12,7 +12,7 @@ afterEach(() => { describe("fireAndForget", () => { it("does not throw and does not reject when the promise resolves", async () => { const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); - fireAndForget(Promise.resolve("ok"), { action: "test.noop.run" }); + fireAndForget(() => Promise.resolve("ok"), { action: "test.noop.run" }); await flush(); expect(warn).not.toHaveBeenCalled(); }); @@ -22,7 +22,7 @@ describe("fireAndForget", () => { const builder = new WideEventBuilder(); await eventStorage.run(builder, async () => { - fireAndForget(Promise.reject(new Error("boom")), { + fireAndForget(() => Promise.reject(new Error("boom")), { action: "reminder.satisfy.enqueue", meta: { userScope: "u_1" }, }); @@ -39,7 +39,7 @@ describe("fireAndForget", () => { it("warns with the action label only — never the error payload", async () => { const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); - fireAndForget(Promise.reject(new Error("secret-token-xyz")), { + fireAndForget(() => Promise.reject(new Error("secret-token-xyz")), { action: "integration.reauth.mark", }); await flush(); @@ -52,7 +52,7 @@ describe("fireAndForget", () => { it("is a safe no-op outside a request context", async () => { vi.spyOn(console, "warn").mockImplementation(() => {}); expect(() => { - fireAndForget(Promise.reject(new Error("no context")), { + fireAndForget(() => Promise.reject(new Error("no context")), { action: "test.orphan.run", }); }).not.toThrow(); @@ -63,7 +63,7 @@ describe("fireAndForget", () => { vi.spyOn(console, "warn").mockImplementation(() => {}); const builder = new WideEventBuilder(); await eventStorage.run(builder, async () => { - fireAndForget(Promise.reject("plain-string"), { + fireAndForget(() => Promise.reject("plain-string"), { action: "test.reject.string", }); await flush(); diff --git a/src/lib/logging/context.ts b/src/lib/logging/context.ts index a7e51266b..82ea9fd4f 100644 --- a/src/lib/logging/context.ts +++ b/src/lib/logging/context.ts @@ -1,6 +1,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import type { WideEvent } from "./types"; import { WideEventBuilder } from "./event-builder"; +import { detachedParent, registerDetachedEventSource } from "./detached"; import { registerSourcePriorityParseObserver } from "@/lib/validations/source-priority"; import { registerRecurrenceParseErrorObserver } from "@/lib/medications/scheduling/recurrence"; import { @@ -16,6 +17,10 @@ export function getEvent(): WideEventBuilder | null { return eventStorage.getStore() ?? null; } +// `runDetached` (./detached) reads the current event through this getter; +// registered here so the two modules do not import each other. +registerDetachedEventSource(getEvent); + /** * Convenience: Felder im aktuellen Event annotieren. * Safe to call anywhere — no-op wenn kein Kontext aktiv. @@ -26,7 +31,10 @@ export function annotate(fields: { }): void { const evt = getEvent(); if (!evt) return; - if (fields.action) evt.setAction(fields.action); + // Detached work never renames the event it was started from. + if (fields.action && detachedParent() !== evt) { + evt.setAction(fields.action); + } if (fields.meta) { for (const [key, value] of Object.entries(fields.meta)) { evt.addMeta(key, value); diff --git a/src/lib/logging/detached.ts b/src/lib/logging/detached.ts new file mode 100644 index 000000000..7b241d6bb --- /dev/null +++ b/src/lib/logging/detached.ts @@ -0,0 +1,47 @@ +/** + * Start work the caller does not wait for (a status re-warm, a reminder + * enqueue, a stale-cache rebuild) so it cannot rename the event it was + * started from. + * + * Detached work still runs inside the request's context: it reads the + * request's auth for its audit rows and adds meta, warnings and external + * calls to the request's event while that event is open. What it must not do + * is set the event's action. Its `annotate({ action })` used to land on the + * request's event whenever it ran before the event was emitted, so the same + * write was logged as `measurement.create` on one request and as the + * re-warm's `insights.status.invalidate.skipped` on the next. Under + * `runDetached` an action from the detached work is dropped for the parent + * event (`annotate` in ./context checks `detachedParent`); an event the work + * opens for itself (`withBackgroundEvent`) is its own and keeps its action. + * + * Kept apart from ./context, which most unit tests replace wholesale: a test + * that mocks the context still gets a working `runDetached`, which then + * simply starts the work. + */ +import { AsyncLocalStorage } from "node:async_hooks"; + +/** The event a piece of detached work was started from. */ +const detachedFrom = new AsyncLocalStorage(); + +let currentEvent: () => object | null = () => null; + +/** Called once by ./context with its event getter. */ +export function registerDetachedEventSource(source: () => object | null): void { + currentEvent = source; +} + +/** The event the running work was detached from, if any. */ +export function detachedParent(): object | undefined { + return detachedFrom.getStore(); +} + +/** Start `start` detached from the current event; it runs synchronously. */ +export function runDetached(start: () => Promise): Promise { + const parent = currentEvent(); + if (!parent) return start(); + try { + return detachedFrom.run(parent, start); + } catch (error) { + return Promise.reject(error); + } +} diff --git a/src/lib/logging/fire-and-forget.ts b/src/lib/logging/fire-and-forget.ts index a6f06895c..6a45b2535 100644 --- a/src/lib/logging/fire-and-forget.ts +++ b/src/lib/logging/fire-and-forget.ts @@ -1,4 +1,5 @@ import { annotate } from "./context"; +import { runDetached } from "./detached"; import { trackBackgroundTask } from "./background-tasks"; /** @@ -22,23 +23,26 @@ import { trackBackgroundTask } from "./background-tasks"; * Only the caller-supplied action label — never the error text — reaches * the raw console sink, keeping it redaction-safe by construction. * - * The promise stays fire-and-forget: the caller does not await it and its - * resolution is discarded exactly as before. Under test the settled handle is - * registered with `trackBackgroundTask` so the suite setup can await it — see + * The work stays fire-and-forget: the caller does not await it and its + * resolution is discarded exactly as before. It is started through + * `runDetached`, so nothing it annotates can rename the caller's event; that + * is why it is passed as a function rather than as a promise already + * running. Under test the settled handle is registered with + * `trackBackgroundTask` so the suite setup can await it — see * `./background-tasks` for why a late `console.warn` is otherwise a teardown * hazard. * - * @param promise The background work. May already be running. + * @param start Starts the background work. * @param context.action A stable `..` label (never user * input) identifying which background task failed. * @param context.meta Optional extra fields to attach to the breadcrumb; they * ride the annotate/redact path with the error message. */ export function fireAndForget( - promise: Promise, + start: () => Promise, context: { action: string; meta?: Record }, ): void { - const settled = Promise.resolve(promise).catch((error: unknown) => { + const settled = runDetached(start).catch((error: unknown) => { const message = error instanceof Error ? error.message : String(error); annotate({ meta: { diff --git a/src/lib/measurements/batch-ingest/report.ts b/src/lib/measurements/batch-ingest/report.ts index 8cce1851e..d47ebe43c 100644 --- a/src/lib/measurements/batch-ingest/report.ts +++ b/src/lib/measurements/batch-ingest/report.ts @@ -201,7 +201,7 @@ export async function enqueueRecordJobs(args: { // Eventful Vorsorge satisfaction. A matching reading just landed; resolve // the user's reminders now rather than waiting on the 15-min cron. // Fire-and-forget — the cron reconciles on enqueue miss. - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); try { diff --git a/src/lib/measurements/create/create-readings.ts b/src/lib/measurements/create/create-readings.ts index 0ab7ea82d..c89a2fc5f 100644 --- a/src/lib/measurements/create/create-readings.ts +++ b/src/lib/measurements/create/create-readings.ts @@ -208,7 +208,7 @@ async function afterReadingsCreated( // Eventful Vorsorge satisfaction. Resolve the user's reminders against the // just-landed readings now. Fire-and-forget. - fireAndForget(enqueueReminderSatisfy(user.id), { + fireAndForget(() => enqueueReminderSatisfy(user.id), { action: "reminder.satisfy.enqueue", }); @@ -232,17 +232,18 @@ async function afterReadingsCreated( // symptom-coupled emergency copy. Fire-and-forget — a notification failure // never fails the write. fireAndForget( - runSafetyFloorCheck({ - userId: user.id, - written: rows.map((r) => ({ - type: r.type, - value: r.value, - measuredAt: r.measuredAt, - glucoseContext: r.glucoseContext, - })), - symptomsPresent: options.symptomsPresent, - timezone: user.timezone ?? undefined, - }), + () => + runSafetyFloorCheck({ + userId: user.id, + written: rows.map((r) => ({ + type: r.type, + value: r.value, + measuredAt: r.measuredAt, + glucoseContext: r.glucoseContext, + })), + symptomsPresent: options.symptomsPresent, + timezone: user.timezone ?? undefined, + }), { action: "measurement.safety_floor.check" }, ); diff --git a/src/lib/measurements/create/create-sleep-stage.ts b/src/lib/measurements/create/create-sleep-stage.ts index f42b235d9..79a7da6f8 100644 --- a/src/lib/measurements/create/create-sleep-stage.ts +++ b/src/lib/measurements/create/create-sleep-stage.ts @@ -147,7 +147,7 @@ export async function createNamedSleepStage( }, }); invalidateUserMeasurements(userId, { evict: true }); - fireAndForget(enqueueReminderSatisfy(userId), { + fireAndForget(() => enqueueReminderSatisfy(userId), { action: "reminder.satisfy.enqueue", }); if (status === "inserted") { diff --git a/src/lib/rollups/__tests__/after-measurement-mutation.test.ts b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts index 17fd3ca91..5d330e882 100644 --- a/src/lib/rollups/__tests__/after-measurement-mutation.test.ts +++ b/src/lib/rollups/__tests__/after-measurement-mutation.test.ts @@ -41,6 +41,9 @@ vi.mock("@/lib/insights/status-invalidation", () => ({ })); import { afterMeasurementMutation } from "../after-measurement-mutation"; +import { annotate, eventStorage } from "@/lib/logging/context"; +import { WideEventBuilder } from "@/lib/logging/event-builder"; +import { settleBackgroundTasks } from "@/lib/logging/background-tasks"; import type { MeasurementType } from "@/generated/prisma/client"; beforeEach(() => { @@ -163,4 +166,35 @@ describe("afterMeasurementMutation", () => { warn.mockRestore(); } }); + + it("keeps the request's action name while the detached re-warm annotates its own", async () => { + // The re-warm is detached from the request but starts inside its + // context. It annotates an action of its own when it skips; that must + // never rename the write the request is reporting. + mocks.invalidate.mockImplementation(async () => { + annotate({ + action: { name: "insights.status.invalidate.skipped" }, + meta: { reason: "no_provider" }, + }); + }); + const event = new WideEventBuilder(); + await eventStorage.run(event, async () => { + annotate({ action: { name: "measurement.create" } }); + await afterMeasurementMutation( + "user-1", + [ + { + type: "WEIGHT" as MeasurementType, + measuredAt: day("2026-05-10T08:00:00Z"), + }, + ], + { source: "manual" }, + ); + await settleBackgroundTasks(); + }); + expect(mocks.invalidate).toHaveBeenCalledTimes(1); + expect(event.toJSON().action?.name).toBe("measurement.create"); + // What the detached work says still reaches the request's event. + expect(event.toJSON().meta?.reason).toBe("no_provider"); + }); }); diff --git a/src/lib/rollups/after-measurement-mutation.ts b/src/lib/rollups/after-measurement-mutation.ts index 6cd1f834e..1dfaa8955 100644 --- a/src/lib/rollups/after-measurement-mutation.ts +++ b/src/lib/rollups/after-measurement-mutation.ts @@ -37,6 +37,7 @@ import { type StatusIngestSource, } from "@/lib/insights/status-invalidation"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; +import { runDetached } from "@/lib/logging/detached"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { earliestMeasuredAt } from "@/lib/cache/closed-block"; import type { MeasurementType } from "@/generated/prisma/client"; @@ -83,9 +84,11 @@ export async function afterMeasurementMutation( // write on a detached handle, and one that lands while a vitest worker is // closing its RPC channel fails the whole run. No-op outside test. trackBackgroundTask( - invalidateStatusInsightsForTypes(userId, types, { - source: options.source, - }).catch((err) => { + runDetached(() => + invalidateStatusInsightsForTypes(userId, types, { + source: options.source, + }), + ).catch((err) => { console.warn(`[${label}] status-insight invalidate failed`, err); }), ); diff --git a/src/lib/rollups/mood-rollups.ts b/src/lib/rollups/mood-rollups.ts index 4dfe341e7..cf249bb05 100644 --- a/src/lib/rollups/mood-rollups.ts +++ b/src/lib/rollups/mood-rollups.ts @@ -49,6 +49,7 @@ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { enqueueWhenBossReady } from "@/lib/jobs/boss-ready"; import { annotate } from "@/lib/logging/context"; +import { runDetached } from "@/lib/logging/detached"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { @@ -318,25 +319,27 @@ export async function recomputeMoodBucketsForEntry( // write on a detached handle, and one that lands while a vitest worker is // closing its RPC channel fails the whole run. No-op outside test. trackBackgroundTask( - Promise.all( - ASYNC_GRANULARITIES.map((granularity) => { - const { from, to } = bucketSpan(dateLabel, granularity); - return enqueueMoodRollupRecompute({ - userId, - granularity, - from, - to, - }).catch((err: unknown) => { - const message = err instanceof Error ? err.message : String(err); - annotate({ - meta: { - mood_rollup_enqueue_failed: true, - mood_rollup_enqueue_error: message, - }, + runDetached(() => + Promise.all( + ASYNC_GRANULARITIES.map((granularity) => { + const { from, to } = bucketSpan(dateLabel, granularity); + return enqueueMoodRollupRecompute({ + userId, + granularity, + from, + to, + }).catch((err: unknown) => { + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + mood_rollup_enqueue_failed: true, + mood_rollup_enqueue_error: message, + }, + }); + console.error("[mood-rollups] async enqueue failed:", message); }); - console.error("[mood-rollups] async enqueue failed:", message); - }); - }), + }), + ), ), ); }