diff --git a/docs/integrations/google-health.md b/docs/integrations/google-health.md index 0b187be94..554f1ed1f 100644 --- a/docs/integrations/google-health.md +++ b/docs/integrations/google-health.md @@ -245,8 +245,10 @@ reading: upper-bound filter Google refused. - `stopReason`: why it ended early. `time_budget` and `page_budget` are the normal end of a slice, `rate_limited` means Google would not take more - requests for now, `job_budget` that the job itself ran out of time. - `null` means the slice read everything it started. + requests for now, `job_budget` that the job itself ran out of time, + `lane_wanted` that the slice made room for another heavy job (a nightly + run, a backup, an upload) that was waiting; the next slice starts two + minutes later. `null` means the slice read everything it started. - `collectionsDone` of `collectionsTotal`: data types read to the end so far. - `collections`: one entry per data type the slice touched, with its diff --git a/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts index ecaba31c6..4054883b7 100644 --- a/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts +++ b/src/lib/google-health/__tests__/walk-resume-and-retry.test.ts @@ -152,6 +152,22 @@ describe("a resumable walk", () => { expect(budget.stopReason).toBe("rate_limited"); }); + it("gives way to a waiting heavy job after its first page, keeping its place", async () => { + safeFetchMock.mockImplementation(async (url: string) => + page(Number(new URL(url).searchParams.get("pageToken") ?? 0), true), + ); + const budget = googleHealthSliceBudget(60_000, 500, Date.now, () => true); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk({ resumable: true }), + budget, + ); + // Never before a page is on disk: a slice that ended empty would fail. + expect(safeFetchMock).toHaveBeenCalledTimes(1); + expect(outcome.truncated).toBe(true); + expect(budget.stopReason).toBe("lane_wanted"); + }); + it("still fails on a rate limit outside a slice", async () => { safeFetchMock.mockResolvedValue( page(0, true, 429, { "retry-after": "3600" }), @@ -178,6 +194,20 @@ describe("a walk that is not resumable", () => { expect(outcome.truncated).toBe(false); }); + it("does not give way part-way, since it would start over", async () => { + safeFetchMock.mockImplementation(async (url: string) => { + const n = Number(new URL(url).searchParams.get("pageToken") ?? 0); + return page(n, n < 3); + }); + const { outcome } = await runWithGoogleHealthStop( + () => false, + () => walk(), + googleHealthSliceBudget(60_000, 500, Date.now, () => true), + ); + expect(safeFetchMock).toHaveBeenCalledTimes(4); + expect(outcome.truncated).toBe(false); + }); + it("still stops on the job's own budget", async () => { safeFetchMock.mockImplementation(async () => page(0, true)); let asked = 0; diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index d12006fc1..a0e3a7aa4 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -155,7 +155,7 @@ interface GoogleHealthWalkLimits { /** Why a backfill slice stopped walking. */ export type GoogleHealthSliceStopReason = - "time_budget" | "page_budget" | "rate_limited" | "job_budget"; + "time_budget" | "page_budget" | "rate_limited" | "job_budget" | "lane_wanted"; /** * One backfill slice's share of the work: a span of wall-clock time and a @@ -184,6 +184,14 @@ export interface GoogleHealthSliceBudget { skippedPages: number; stopReason: GoogleHealthSliceStopReason | null; now: () => number; + /** + * Whether another heavy job waits for the lane this slice holds + * (`laneWanted` in `jobs/heavy-lane.ts`). Asked where the slice's own + * limits are, so giving way keeps every walk's place, and only once the + * slice has read a page, so the slice it ends has something on disk and is + * continued rather than failed. + */ + yieldWanted: () => boolean; } /** A fresh slice budget of `ms` milliseconds and `maxPages` pages. */ @@ -191,6 +199,7 @@ export function googleHealthSliceBudget( ms: number, maxPages: number, now: () => number = Date.now, + yieldWanted: () => boolean = () => false, ): GoogleHealthSliceBudget { return { ms, @@ -200,6 +209,7 @@ export function googleHealthSliceBudget( skippedPages: 0, stopReason: null, now, + yieldWanted, }; } @@ -212,6 +222,8 @@ function sliceExhausted(slice: GoogleHealthSliceBudget): boolean { if (slice.now() >= slice.deadline) slice.stopReason = "time_budget"; else if (slice.pages - slice.skippedPages >= slice.maxPages) { slice.stopReason = "page_budget"; + } else if (slice.pages > 0 && slice.yieldWanted()) { + slice.stopReason = "lane_wanted"; } return slice.stopReason !== null; } diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index 3fd42cdc0..8a8d33ba1 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -133,6 +133,10 @@ import { runGoogleHealthBackfillForUser, } from "../google-health-backfill"; import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission"; +import { + LANE_DEFER_SECONDS, + LANE_YIELD_DELAY_SECONDS, +} from "@/lib/jobs/heavy-lane"; import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; import { GOOGLE_HEALTH_DATA_TYPES, @@ -472,6 +476,35 @@ describe("runGoogleHealthBackfillForUser — slices", () => { }); }); + it("queues the next slice past the waiting job's return after giving way", async () => { + resumeState.advances = ["fetchHeartRate"]; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true }); + const client = await import("@/lib/google-health/client"); + const spy = vi + .spyOn(client, "googleHealthSliceBudget") + .mockImplementation(() => ({ + ms: 300_000, + deadline: null, + maxPages: 500, + pages: 3, + skippedPages: 0, + stopReason: "lane_wanted", + now: Date.now, + yieldWanted: () => true, + })); + try { + await expect(runGoogleHealthBackfillForUser("u1")).resolves.toMatchObject( + { continued: true }, + ); + } finally { + spy.mockRestore(); + } + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + startAfter: LANE_YIELD_DELAY_SECONDS, + }); + expect(LANE_YIELD_DELAY_SECONDS).toBeGreaterThan(LANE_DEFER_SECONDS); + }); + it("waits longer before the next slice when this one ended on a rate limit", async () => { resumeState.advances = ["fetchHeartRate"]; syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true }); @@ -488,6 +521,7 @@ describe("runGoogleHealthBackfillForUser — slices", () => { skippedPages: 0, stopReason: "rate_limited", now: Date.now, + yieldWanted: () => false, })); try { await runGoogleHealthBackfillForUser("u1"); diff --git a/src/lib/jobs/__tests__/heavy-lane.test.ts b/src/lib/jobs/__tests__/heavy-lane.test.ts index 0b97297ed..6b1bed63b 100644 --- a/src/lib/jobs/__tests__/heavy-lane.test.ts +++ b/src/lib/jobs/__tests__/heavy-lane.test.ts @@ -8,8 +8,10 @@ import type { Job } from "pg-boss"; vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/logging/signal", () => ({ emitSignal: vi.fn() })); -import { heavyLane } from "../heavy-lane"; +import { emitSignal } from "@/lib/logging/signal"; +import { heavyLane, LANE_WAIT_WARN_MS, laneWanted } from "../heavy-lane"; import { jobDone } from "../job-outcome"; type TestJob = Job<{ force?: boolean }>; @@ -36,7 +38,7 @@ function deferrals() { const back: Array<{ queue: string; ids: string[] }> = []; const defer = vi.fn(async (queue: string, jobs: readonly TestJob[]) => { back.push({ queue, ids: jobs.map((j) => j.id) }); - return jobs.length; + return { deferred: jobs.length, oldestCreatedOn: null }; }); return { back, defer }; } @@ -160,3 +162,107 @@ describe("heavyLane", () => { ).toThrow(/not in HEAVY_QUEUES/); }); }); + +describe("laneWanted", () => { + it("tells the holder, and only the holder, that another queue is waiting", async () => { + const seen: Record = { holder: [], beside: [] }; + let release!: () => void; + const released = new Promise((r) => (release = r)); + const { defer } = deferrals(); + expect(laneWanted()).toBe(false); + const held = heavyLane( + "integration-backfill-admission", + async () => { + seen.holder.push(laneWanted()); + await released; + seen.holder.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + )([job("slice")]); + await tick(); + // A job of the same queue going back is not a reason to give way. + await heavyLane( + "integration-backfill-admission", + vi.fn(), + undefined, + defer, + )([job("next-slice")]); + expect(laneWanted()).toBe(false); + await heavyLane( + "insights-general-status", + vi.fn(), + undefined, + defer, + )([job("nightly")]); + // A light job of a heavy queue running beside the lane is never asked, even now. + await heavyLane( + "insight-pregenerate", + async () => { + seen.beside.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + () => false, + defer, + )([job("warm")]); + release(); + await held; + expect(seen.holder).toEqual([false, true]); + expect(seen.beside).toEqual([false]); + // The next holder starts with a clean slate. + await heavyLane( + "data-backup", + async () => { + seen.holder.push(laneWanted()); + return jobDone({ jobs: 1 }); + }, + undefined, + defer, + )([job("weekly")]); + expect(seen.holder).toEqual([false, true, false]); + }); +}); + +describe("a long wait for the lane", () => { + it("is reported at warn once per holder when a job has waited past the limit", async () => { + vi.mocked(emitSignal).mockClear(); + const now = Date.parse("2026-10-20T06:00:00Z"); + let release!: () => void; + const released = new Promise((r) => (release = r)); + const queuedAt = (ms: number) => async () => ({ + deferred: 1, + oldestCreatedOn: new Date(now - ms), + }); + const held = heavyLane("integration-backfill-admission", async () => { + await released; + return jobDone({ jobs: 1 }); + })([job("import")]); + await tick(); + const wait = (ms: number) => + heavyLane( + "insights-general-status", + vi.fn(), + undefined, + queuedAt(ms), + () => now, + )([job("nightly")]); + + await wait(LANE_WAIT_WARN_MS - 60_000); + expect(emitSignal).not.toHaveBeenCalled(); + await wait(LANE_WAIT_WARN_MS + 60_000); + await wait(LANE_WAIT_WARN_MS + 120_000); + expect(emitSignal).toHaveBeenCalledTimes(1); + expect(emitSignal).toHaveBeenCalledWith({ + action: "job.heavy_lane.waiting_long", + level: "warn", + meta: { + queue: "insights-general-status", + holder: "integration-backfill-admission", + waited_minutes: 121, + }, + }); + release(); + await held; + }); +}); diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index 82b30ed07..d070872de 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -38,9 +38,11 @@ import { import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock"; import { googleHealthSliceBudget, + type GoogleHealthSliceStopReason, runWithGoogleHealthStop, } from "@/lib/google-health/client"; import { reportJobProgress } from "@/lib/jobs/job-observer"; +import { LANE_YIELD_DELAY_SECONDS, laneWanted } from "@/lib/jobs/heavy-lane"; import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys"; export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill"; @@ -122,6 +124,21 @@ export function googleHealthBackfillRetryKey( return `${identity}|retry-${new Date(dueAt).toISOString()}`; } +/** + * The pause before the next slice, by why this one stopped: longer after a + * rate limit, and past the waiting job's return after giving way to it, so + * that job takes the lane before this import does again. + */ +export function nextSliceDelaySeconds( + stopReason: GoogleHealthSliceStopReason | null, +): number { + if (stopReason === "rate_limited") { + return GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS; + } + if (stopReason === "lane_wanted") return LANE_YIELD_DELAY_SECONDS; + return INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS; +} + /** Test seam: a slice's budget, when not the production one. */ export interface GoogleHealthBackfillSliceOptions { sliceMs?: number; @@ -240,6 +257,8 @@ export async function runGoogleHealthBackfillForUser( const budget = googleHealthSliceBudget( sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS, sliceOptions.slicePages ?? GOOGLE_HEALTH_BACKFILL_SLICE_PAGES, + Date.now, + laneWanted, ); const startedAt = Date.now(); const { @@ -387,9 +406,7 @@ export async function runGoogleHealthBackfillForUser( const progressed = finished > 0 || advanced.size > 0 || newlyParked > 0; if (progressed) { const sent = await sendNextSlice( - budget.stopReason === "rate_limited" - ? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS - : INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + nextSliceDelaySeconds(budget.stopReason), ); if (sent) { annotate({ diff --git a/src/lib/jobs/heavy-lane.ts b/src/lib/jobs/heavy-lane.ts index 0f924880d..1437e99b8 100644 --- a/src/lib/jobs/heavy-lane.ts +++ b/src/lib/jobs/heavy-lane.ts @@ -23,12 +23,24 @@ * longer than its own expiry was failed, stopped counting as in use, and had * its staged file swept from under it. A deferred job keeps its id, its * retries and its staged upload, and its expiry starts when it runs. + * + * A deferred job also leaves word that the lane is wanted (`laneWanted`). A + * holder that runs in slices asks it at the boundaries where it can already + * stop and keep its place, ends its slice there, and queues the rest a little + * later than the job that waited (`LANE_YIELD_DELAY_SECONDS`), so a long + * import makes room for the nightly runs and for an upload instead of holding + * them back for hours. A holder that cannot stop part-way runs to its end; a + * job that has stood in line for `LANE_WAIT_WARN_MS` is reported at `warn` + * (`job.heavy_lane.waiting_long`), once per holder, so a lane that is held + * for hours shows in the alerts instead of only in `info` lines. */ +import { AsyncLocalStorage } from "node:async_hooks"; import type { Job } from "pg-boss"; import { prisma } from "@/lib/db"; import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; import { annotate } from "@/lib/logging/context"; +import { emitSignal } from "@/lib/logging/signal"; export interface HeavyQueue { /** What makes the job heavy. */ @@ -73,21 +85,39 @@ export const HEAVY_QUEUES: Readonly> = { /** How far ahead a deferred job is put back in the queue. */ export const LANE_DEFER_SECONDS = 60; +/** + * How far ahead a holder that gave way queues the rest of its work: past the + * deferred job's own return, so the job it gave way to takes the lane first. + */ +export const LANE_YIELD_DELAY_SECONDS = 2 * LANE_DEFER_SECONDS; + +/** How long a job may stand in line before the wait is reported at `warn`. */ +export const LANE_WAIT_WARN_MS = 2 * 60 * 60 * 1000; + +/** What one trip back to the queue did. */ +export interface LaneDeferral { + /** Rows that went back. */ + deferred: number; + /** When the oldest of them was first queued, if any went back. */ + oldestCreatedOn: Date | null; +} + /** * Put `jobs` of `queue` back in the queue to run `seconds` from now. The same * reset pg-boss's own `restore()` makes, with `start_after` moved in the same * statement: a restore followed by a separate update leaves a moment in which * another fetch could claim the row, and the completion of this handler would * then settle that newer claim, since a restored job's next claim keeps its - * attempt number. Answers how many rows went back. + * attempt number. Answers how many rows went back, and since when the oldest + * of them has been waiting. */ export async function deferLaneJobs( queue: string, jobs: readonly Job[], seconds: number = LANE_DEFER_SECONDS, -): Promise { +): Promise { const ids = jobs.map((job) => job.id); - return prisma.$executeRaw` + const rows = await prisma.$queryRaw>` UPDATE pgboss.job SET state = 'created', started_on = NULL, @@ -97,42 +127,102 @@ export async function deferLaneJobs( WHERE name = ${queue} AND id = ANY(${ids}::uuid[]) AND state = 'active' + RETURNING created_on `; + const oldest = rows.reduce( + (min, row) => (min === null || row.created_on < min ? row.created_on : min), + null, + ); + return { deferred: rows.length, oldestCreatedOn: oldest }; } -let busy = false; +/** The job holding the lane, and what has happened while it held it. */ +interface LaneHold { + queue: string; + /** The first other queue that found the lane taken during this hold. */ + wantedBy: string | null; + /** Queues whose long wait has been reported during this hold. */ + warned: Set; +} + +let hold: LaneHold | null = null; +const holdScope = new AsyncLocalStorage(); + +/** + * Whether the caller holds the lane and a job of another queue has been sent + * back while it did. False outside the lane, so a job that runs beside the + * lane (a light job of a heavy queue, any other queue) never gives way. + */ +export function laneWanted(): boolean { + const current = holdScope.getStore(); + return current !== undefined && current === hold && current.wantedBy !== null; +} + +function reportLongWait( + current: LaneHold, + queue: string, + deferral: LaneDeferral, + now: number, +): void { + if (!deferral.oldestCreatedOn || current.warned.has(queue)) return; + const waitedMs = now - deferral.oldestCreatedOn.getTime(); + if (waitedMs < LANE_WAIT_WARN_MS) return; + current.warned.add(queue); + emitSignal({ + action: "job.heavy_lane.waiting_long", + level: "warn", + meta: { + queue, + holder: current.queue, + waited_minutes: Math.floor(waitedMs / 60_000), + }, + }); +} /** * Wrap `handler` so the jobs of `queue` run in the lane. `inLane` picks the * jobs that are heavy where a queue carries light ones too (a single * account's warm beside the nightly walk); a batch with none of them runs at - * once. `defer` is the way back to the queue; tests replace it. + * once. `defer` is the way back to the queue; tests replace it, and `now`. */ export function heavyLane>( queue: string, handler: (jobs: J[]) => Promise, inLane: (job: J) => boolean = () => true, - defer: (queue: string, jobs: readonly J[]) => Promise = deferLaneJobs, + defer: ( + queue: string, + jobs: readonly J[], + ) => Promise = deferLaneJobs, + now: () => number = Date.now, ): (jobs: J[]) => Promise { if (!HEAVY_QUEUES[queue]) { throw new Error(`heavyLane: ${queue} is not in HEAVY_QUEUES`); } return async (jobs) => { if (!jobs.some(inLane)) return handler(jobs); - if (busy) { - const deferred = await defer(queue, jobs); + const current = hold; + if (current) { + const deferral = await defer(queue, jobs); + if (queue !== current.queue) current.wantedBy ??= queue; annotate({ action: { name: "job.heavy_lane.deferred" }, - meta: { queue, jobs: jobs.length, deferred }, + meta: { + queue, + holder: current.queue, + jobs: jobs.length, + deferred: deferral.deferred, + }, }); - return jobDone({ lane_deferred: deferred }); + reportLongWait(current, queue, deferral, now()); + return jobDone({ lane_deferred: deferral.deferred }); } - busy = true; + const entered: LaneHold = { queue, wantedBy: null, warned: new Set() }; + hold = entered; annotate({ action: { name: "job.heavy_lane.entered" }, meta: { queue } }); try { - return await handler(jobs); + return await holdScope.run(entered, () => handler(jobs)); } finally { - busy = false; + hold = null; } }; } diff --git a/tests/integration/heavy-lane-defer.test.ts b/tests/integration/heavy-lane-defer.test.ts index bbd0f0485..dfbb613dd 100644 --- a/tests/integration/heavy-lane-defer.test.ts +++ b/tests/integration/heavy-lane-defer.test.ts @@ -59,7 +59,13 @@ describe("deferLaneJobs", () => { expect(fetched?.id).toBe(id); expect((await row(id!))?.state).toBe("active"); - expect(await deferLaneJobs(WAITER_QUEUE, [fetched!], 30)).toBe(1); + const deferral = await deferLaneJobs(WAITER_QUEUE, [fetched!], 30); + expect(deferral.deferred).toBe(1); + // When it was first queued, so a long wait can be told from a short one. + expect(deferral.oldestCreatedOn).toBeInstanceOf(Date); + expect(Date.now() - deferral.oldestCreatedOn!.getTime()).toBeLessThan( + 60_000, + ); expect(await row(id!)).toMatchObject({ state: "created", retry_count: 0, @@ -77,7 +83,10 @@ describe("deferLaneJobs", () => { it("leaves a job alone that is no longer active", async () => { const id = await boss.send(WAITER_QUEUE, { probe: 2 }); const job = { id, name: WAITER_QUEUE, data: {} } as unknown as Job; - expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toBe(0); + expect(await deferLaneJobs(WAITER_QUEUE, [job], 30)).toEqual({ + deferred: 0, + oldestCreatedOn: null, + }); expect((await row(id!))?.future).toBe(false); await boss.deleteJob(WAITER_QUEUE, id!); });