diff --git a/src/lib/jobs/__tests__/integration-backfill-drain-event.test.ts b/src/lib/jobs/__tests__/integration-backfill-drain-event.test.ts new file mode 100644 index 000000000..e6b8421af --- /dev/null +++ b/src/lib/jobs/__tests__/integration-backfill-drain-event.test.ts @@ -0,0 +1,117 @@ +/** + * #1220 — every admitted full-history import runs inside its own background + * event, so what the provider walk records on the ambient event (each + * request with its status, every warning, every annotation) reaches the log. + * Before, the drain ran the import with no event in scope, every one of those + * calls was a no-op, and a failing backfill left nothing but its error + * message behind. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +import type { WideEvent } from "@/lib/logging/types"; + +const { emitted, runGoogleHealthBackfillMock } = vi.hoisted(() => ({ + emitted: [] as WideEvent[], + runGoogleHealthBackfillMock: vi.fn(), +})); + +vi.mock("@/lib/logging/transports", () => ({ + emitIfSampled: (event: WideEvent) => emitted.push(event), +})); +vi.mock("@/lib/jobs/integration-backfill-lock", () => ({ + withIntegrationBackfillLock: async (_id: string, run: () => unknown) => ({ + ran: true, + result: await run(), + }), +})); +vi.mock("@/lib/jobs/google-health-backfill", () => ({ + runGoogleHealthBackfillForUser: runGoogleHealthBackfillMock, +})); +vi.mock("@/lib/jobs/reminder/shared", () => ({ workerLog: () => {} })); + +import { drainIntegrationBackfillAdmission } from "../integration-backfill-drain"; +import type { IntegrationBackfillAdmissionPayload } from "../integration-backfill-admission"; +import { annotate, getEvent } from "@/lib/logging/context"; + +function job( + data: IntegrationBackfillAdmissionPayload, +): Job { + return { + id: "job-1", + data, + expireInSeconds: 3600, + } as unknown as Job; +} + +const payload: IntegrationBackfillAdmissionPayload = { + kind: "google-health-backfill", + data: { userId: "u1", enqueuedAt: "2026-10-14T11:00:00Z", slice: 2 }, +}; + +beforeEach(() => { + emitted.length = 0; + runGoogleHealthBackfillMock.mockReset(); +}); + +describe("drainIntegrationBackfillAdmission — one event per admitted run", () => { + it("emits what the import recorded, with the job it belongs to", async () => { + runGoogleHealthBackfillMock.mockImplementation(async () => { + getEvent()?.addExternalCall({ + service: "google-health", + method: "fetchHeartRate(page=0)", + duration_ms: 120, + status: 200, + }); + annotate({ meta: { "googleHealth.backfill.slice": { pages: 1 } } }); + return { imported: 1000, continued: true }; + }); + + await drainIntegrationBackfillAdmission([job(payload)]); + + expect(emitted).toHaveLength(1); + const [event] = emitted; + expect(event!.background).toMatchObject({ + task_name: "job.integration_backfill", + result: { kind: "google-health-backfill", imported: 1000, continued: 1 }, + }); + expect(event!.meta).toMatchObject({ + kind: "google-health-backfill", + identity: "google-health-backfill|u1", + job_id: "job-1", + slice: 2, + "googleHealth.backfill.slice": { pages: 1 }, + }); + expect(event!.external_calls).toEqual([ + expect.objectContaining({ + method: "fetchHeartRate(page=0)", + status: 200, + }), + ]); + }); + + it("emits the calls of a run that throws, with its error", async () => { + runGoogleHealthBackfillMock.mockImplementation(async () => { + getEvent()?.addExternalCall({ + service: "google-health", + method: "fetchHeartRate(page=812)", + duration_ms: 15000, + status: 503, + error: "transient HTTP 503", + }); + throw new Error("google-health backfill incomplete for user u1"); + }); + + await expect( + drainIntegrationBackfillAdmission([job(payload)]), + ).rejects.toThrow(/incomplete/); + + expect(emitted).toHaveLength(1); + expect(emitted[0]!.level).toBe("error"); + expect(emitted[0]!.error?.message).toMatch(/incomplete/); + expect(emitted[0]!.external_calls?.[0]).toMatchObject({ + method: "fetchHeartRate(page=812)", + status: 503, + }); + }); +}); diff --git a/src/lib/jobs/health-connect-import-worker.ts b/src/lib/jobs/health-connect-import-worker.ts index 074caf22e..58ba0c6db 100644 --- a/src/lib/jobs/health-connect-import-worker.ts +++ b/src/lib/jobs/health-connect-import-worker.ts @@ -32,6 +32,7 @@ import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalida import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; import { trackBackgroundTask } from "@/lib/logging/background-tasks"; import { annotate } from "@/lib/logging/context"; +import { withBackgroundEvent } from "@/lib/logging/background"; import { recomputeUserRollups, ROLLUP_FOLD_WINDOW_MS, @@ -102,7 +103,18 @@ export async function handleHealthConnectImport( ): Promise { let failure: JobOutcome | null = null; for (const job of jobs) { - const outcome = await runHealthConnectImport(job.data); + // Its own event per import, so the `annotate` calls below and anything + // the import records on the ambient event reach the log instead of + // vanishing for want of one. + const outcome = await withBackgroundEvent( + "job.health_connect_import", + async (event) => { + event.addMeta("import_job_id", job.data.importJobId); + const result = await runHealthConnectImport(job.data); + if (!result.ok) event.setError(result.cause ?? result.reason); + return result; + }, + ); if (!outcome.ok && failure === null) failure = outcome; } return failure ?? jobDone({ jobs: jobs.length }); diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts index 68d7f7174..df6ad556e 100644 --- a/src/lib/jobs/integration-backfill-drain.ts +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -143,6 +143,49 @@ async function runAdmitted( } } +/** The task name of the event each admitted run writes. */ +export const INTEGRATION_BACKFILL_RUN_TASK = "job.integration_backfill"; + +/** + * Run one admitted job inside its own background event. + * + * The runners write what they do through the ambient event: every request a + * provider walk makes (`addExternalCall`, with its status, duration and the + * page it asked for), every warning, every `annotate`. Without an event in + * scope each of those is a no-op, so a full-history import that failed left + * nothing behind but the job's own error message: not the collection it was + * on, not the status the provider answered, not how far it got (#1220). The + * hourly syncs always ran inside one; the imports did not. Every kind gets + * one here, emitted when the run settles, thrown error included. + */ +async function runObserved( + job: Job, + identity: string, + shouldStop: () => boolean, + totals: RunCounts, +): Promise { + await withBackgroundEvent(INTEGRATION_BACKFILL_RUN_TASK, async (event) => { + event.addMeta("kind", job.data.kind); + event.addMeta("identity", identity); + event.addMeta("job_id", job.id); + event.addMeta("slice", job.data.data.slice ?? 0); + const before = { ...totals }; + await runAdmitted(job.data, shouldStop, totals); + event.setBackground({ + task_name: INTEGRATION_BACKFILL_RUN_TASK, + result: { + kind: job.data.kind, + imported: totals.imported - before.imported, + continued: totals.continued - before.continued, + removed: totals.removed - before.removed, + deleted: totals.deleted - before.deleted, + markers: totals.markers - before.markers, + linked: totals.linked - before.linked, + }, + }); + }); +} + /** * Drain a batch of admission jobs. There is no per-job error isolation: a * runner that throws fails the job and the admission retry policy applies. So @@ -166,7 +209,7 @@ export async function drainIntegrationBackfillAdmission( for (const job of jobs) { const identity = integrationBackfillIdentity(job.data); const run = await withIntegrationBackfillLock(identity, () => - runAdmitted(job.data, shouldStop, totals), + runObserved(job, identity, shouldStop, totals), ); if (!run.ran) { alreadyRunning += 1; diff --git a/src/lib/logging/__tests__/event-builder.test.ts b/src/lib/logging/__tests__/event-builder.test.ts index 2ff65ce65..c8900f8f9 100644 --- a/src/lib/logging/__tests__/event-builder.test.ts +++ b/src/lib/logging/__tests__/event-builder.test.ts @@ -1,5 +1,9 @@ import { describe, expect, it } from "vitest"; -import { WideEventBuilder } from "../event-builder"; +import { + EXTERNAL_CALLS_MAX, + EXTERNAL_CALLS_PLAIN_MAX, + WideEventBuilder, +} from "../event-builder"; describe("WideEventBuilder.setHttp", () => { it("stores plain http metadata as-is", () => { @@ -140,3 +144,46 @@ describe("WideEventBuilder.capStatusLevel", () => { expect(evt.toJSON().level).toBe("warn"); }); }); + +describe("WideEventBuilder — bounded external calls", () => { + it("lists failures beyond the share of plain successes, and counts the rest", () => { + const evt = new WideEventBuilder("background"); + for (let i = 0; i < 500; i++) { + evt.addExternalCall({ + service: "google-health", + method: `page=${i}`, + duration_ms: 1, + status: 200, + }); + } + evt.addExternalCall({ + service: "google-health", + method: "page=500", + duration_ms: 1, + status: 503, + error: "transient HTTP 503", + }); + const event = evt.toJSON(); + expect(event.external_calls).toHaveLength(EXTERNAL_CALLS_PLAIN_MAX + 1); + expect(event.external_calls!.at(-1)).toMatchObject({ status: 503 }); + expect(event.external_calls_omitted).toBe(500 - EXTERNAL_CALLS_PLAIN_MAX); + + for (let i = 0; i < 1000; i++) { + evt.addExternalCall({ + service: "google-health", + method: `retry=${i}`, + duration_ms: 1, + status: 429, + note: "rate_limited_retrying_in_ms=2000", + }); + } + const after = evt.toJSON(); + expect(after.external_calls).toHaveLength(EXTERNAL_CALLS_MAX); + expect(after.external_calls_omitted).toBe( + 500 - + EXTERNAL_CALLS_PLAIN_MAX + + 1000 - + (EXTERNAL_CALLS_MAX - EXTERNAL_CALLS_PLAIN_MAX - 1), + ); + }); +}); diff --git a/src/lib/logging/event-builder.ts b/src/lib/logging/event-builder.ts index 09afd9819..2429e9d59 100644 --- a/src/lib/logging/event-builder.ts +++ b/src/lib/logging/event-builder.ts @@ -35,6 +35,12 @@ function redactDeep(value: T, depth = 0): T { return value; } +/** The most external calls one event lists. */ +export const EXTERNAL_CALLS_MAX = 200; + +/** The most calls without an error or a note one event lists. */ +export const EXTERNAL_CALLS_PLAIN_MAX = 60; + /** * Baut ein Wide Event Schritt fuer Schritt auf. * Wird am Anfang eines Requests/Tasks erstellt und am Ende emittiert. @@ -44,6 +50,7 @@ export class WideEventBuilder { private startTime: number; private highestLevel: LogLevel = "info"; private statusLevelCeiling: LogLevel = "error"; + private plainExternalCalls = 0; constructor(kind: EventKind = "http") { this.startTime = performance.now(); @@ -208,6 +215,23 @@ export class WideEventBuilder { addExternalCall(call: NonNullable[0]): this { if (!this.event.external_calls) this.event.external_calls = []; + // Bounded. A long background run (a full-history backfill walks hundreds + // of pages in one event) would otherwise hold every call in memory and + // write them all as one log line. Calls that failed or carry a note are + // what a reader looks for, so they keep the larger share; plain successes + // stop being listed sooner. Whatever is left out is counted, never lost + // silently. + const calls = this.event.external_calls; + const notable = call.error !== undefined || call.note !== undefined; + if ( + calls.length >= EXTERNAL_CALLS_MAX || + (!notable && this.plainExternalCalls >= EXTERNAL_CALLS_PLAIN_MAX) + ) { + this.event.external_calls_omitted = + (this.event.external_calls_omitted ?? 0) + 1; + return this; + } + if (!notable) this.plainExternalCalls += 1; // An outbound failure carries whatever the remote said, and for several // integrations that is the request URL — which for Telegram embeds the bot // token. `setError` and `setHttp` scrub on the way in; this entry point did diff --git a/src/lib/logging/types.ts b/src/lib/logging/types.ts index 086357230..a126d5681 100644 --- a/src/lib/logging/types.ts +++ b/src/lib/logging/types.ts @@ -123,6 +123,11 @@ export interface WideEvent { */ note?: string; }>; + /** + * How many external calls the event did not list because it had already + * listed its share (`EXTERNAL_CALLS_MAX`, `EXTERNAL_CALLS_PLAIN_MAX`). + */ + external_calls_omitted?: number; // Hintergrundaufgabe (nur bei kind === "background") background?: {