From 82d89600445472ea892e90eb3e84d45d7ee47526 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 11:07:28 +0200 Subject: [PATCH] fix(jobs): run every admitted import inside its own background event The admission drain ran each full-history import with no wide event in scope, so every external call, warning and annotation the provider walk recorded was dropped and a failing backfill left only its error message. Each admitted run now opens a job.integration_backfill event with its kind, identity, job id and slice, emitted when the run settles. The Health Connect import worker gets the same treatment. The external call list on one event is bounded so a long walk cannot hold thousands of entries: failed and annotated calls keep the larger share, the rest is counted in external_calls_omitted. --- .../integration-backfill-drain-event.test.ts | 117 ++++++++++++++++++ src/lib/jobs/health-connect-import-worker.ts | 14 ++- src/lib/jobs/integration-backfill-drain.ts | 45 ++++++- .../logging/__tests__/event-builder.test.ts | 49 +++++++- src/lib/logging/event-builder.ts | 24 ++++ src/lib/logging/types.ts | 5 + 6 files changed, 251 insertions(+), 3 deletions(-) create mode 100644 src/lib/jobs/__tests__/integration-backfill-drain-event.test.ts 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?: {