mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-10 05:12:14 +00:00
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.
This commit is contained in:
1 parent
cfa5fe537b
commit
b4166cdc84
25 files changed
+386
-104
No files matched your search
@@ -145,9 +145,10 @@ describe("background-task settle contract", () => {
|
||||
it("registers what fireAndForget starts", async () => {
|
||||
let release: () => void = () => {};
|
||||
fireAndForget(
|
||||
new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
}),
|
||||
() =>
|
||||
new Promise<void>((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<void>((resolve) =>
|
||||
setTimeout(() => {
|
||||
innerDone = true;
|
||||
resolve();
|
||||
}, 10),
|
||||
),
|
||||
{ action: "guard.background.cascade.inner" },
|
||||
);
|
||||
}),
|
||||
() =>
|
||||
Promise.resolve().then(() => {
|
||||
fireAndForget(
|
||||
() =>
|
||||
new Promise<void>((resolve) =>
|
||||
setTimeout(() => {
|
||||
innerDone = true;
|
||||
resolve();
|
||||
}, 10),
|
||||
),
|
||||
{ action: "guard.background.cascade.inner" },
|
||||
);
|
||||
}),
|
||||
{ action: "guard.background.cascade.outer" },
|
||||
);
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<string, string> = {
|
||||
"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);
|
||||
});
|
||||
});
|
||||
@@ -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" },
|
||||
);
|
||||
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
|
||||
|
||||
Vendored
+13
-16
@@ -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<T = unknown> {
|
||||
@@ -406,23 +407,19 @@ export class ServerCache<T> {
|
||||
// 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({
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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";
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<object>();
|
||||
|
||||
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<T>(start: () => Promise<T>): Promise<T> {
|
||||
const parent = currentEvent();
|
||||
if (!parent) return start();
|
||||
try {
|
||||
return detachedFrom.run(parent, start);
|
||||
} catch (error) {
|
||||
return Promise.reject(error);
|
||||
}
|
||||
}
|
||||
@@ -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 `<surface>.<noun>.<verb>` 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<unknown>,
|
||||
start: () => Promise<unknown>,
|
||||
context: { action: string; meta?: Record<string, unknown> },
|
||||
): 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: {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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" },
|
||||
);
|
||||
|
||||
|
||||
@@ -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") {
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
}),
|
||||
);
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
}),
|
||||
}),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user