From 87554c09f31c49d835b1802ac2d4e57f39181c12 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 03:34:46 +0200 Subject: [PATCH] fix(google-health): key a waiting retry on when it is due A slice that found nothing left but a parked collection queued the next slice a day out under its slice number. A boot during that day ran a fresh first slice, met the same wait and queued its own retry under a different slice number, so two chains walked the same history. The waiting retry now goes under a key named after the instant it is due, read from the stored note, so every slice that meets the same wait names the same job and the queue keeps one. A retry already due stays an ordinary next slice. --- .../__tests__/google-health-backfill.test.ts | 41 ++++++++++ src/lib/jobs/google-health-backfill.ts | 82 ++++++++++++++----- 2 files changed, 103 insertions(+), 20 deletions(-) diff --git a/src/lib/jobs/__tests__/google-health-backfill.test.ts b/src/lib/jobs/__tests__/google-health-backfill.test.ts index f89980edc..3fd42cdc0 100644 --- a/src/lib/jobs/__tests__/google-health-backfill.test.ts +++ b/src/lib/jobs/__tests__/google-health-backfill.test.ts @@ -129,6 +129,7 @@ vi.mock("@/lib/google-health/sync", () => ({ import { GOOGLE_HEALTH_BACKFILL_MAX_SLICE, GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS, + googleHealthBackfillRetryKey, runGoogleHealthBackfillForUser, } from "../google-health-backfill"; import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission"; @@ -400,6 +401,46 @@ describe("runGoogleHealthBackfillForUser — slices", () => { }); }); + it("keys a waiting retry on when it is due, so a boot's slice cannot start a second chain", async () => { + resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter( + (k) => k !== "fetchHeartRate", + ); + resumeState.parks = ["fetchHeartRate"]; + resumeState.retryDueAt = Date.now() + 20 * 3_600_000; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false }); + + // The slice that parked the collection, then a boot's first slice meeting + // the same wait from the same note. + await runGoogleHealthBackfillForUser("u1", undefined, 7); + await runGoogleHealthBackfillForUser("u1", undefined, 0); + + const keys = bossSend.mock.calls.map( + (call) => (call[2] as { singletonKey: string }).singletonKey, + ); + expect(keys).toEqual([ + googleHealthBackfillRetryKey("u1", resumeState.retryDueAt), + googleHealthBackfillRetryKey("u1", resumeState.retryDueAt), + ]); + expect(keys[0]).toBe( + `google-health-backfill|u1|retry-${new Date(resumeState.retryDueAt).toISOString()}`, + ); + }); + + it("sends a retry that is already due as an ordinary next slice", async () => { + resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter( + (k) => k !== "fetchHeartRate", + ); + resumeState.parks = ["fetchHeartRate"]; + resumeState.retryDueAt = Date.now() - 1_000; + syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false }); + + await runGoogleHealthBackfillForUser("u1", undefined, 7); + expect(bossSend.mock.calls[0]![2]).toMatchObject({ + singletonKey: "google-health-backfill|u1|slice-8", + startAfter: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + }); + }); + it("does not stamp over a parked collection when the retry cannot be queued", async () => { resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter( (k) => k !== "fetchHeartRate", diff --git a/src/lib/jobs/google-health-backfill.ts b/src/lib/jobs/google-health-backfill.ts index a23a6b016..82b30ed07 100644 --- a/src/lib/jobs/google-health-backfill.ts +++ b/src/lib/jobs/google-health-backfill.ts @@ -17,8 +17,12 @@ import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { enqueueIntegrationBackfillAdmission, + INTEGRATION_BACKFILL_ADMISSION_QUEUE, INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, + integrationBackfillAdmissionSendOptions, + integrationBackfillIdentity, integrationBackfillSourceOptions, + type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; import { runWithGoogleHealthBackfillResume } from "@/lib/google-health/sync-core"; import { @@ -101,6 +105,23 @@ export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE = ) + 2; +/** + * The singleton key of a slice that waits for a parked collection's retry: + * the import's identity and the instant the retry is due. The instant comes + * from the stored note, so every slice that reads the same note derives the + * same key. + */ +export function googleHealthBackfillRetryKey( + userId: string, + dueAt: number, +): string { + const identity = integrationBackfillIdentity({ + kind: "google-health-backfill", + data: { userId, enqueuedAt: "" }, + }); + return `${identity}|retry-${new Date(dueAt).toISOString()}`; +} + /** Test seam: a slice's budget, when not the production one. */ export interface GoogleHealthBackfillSliceOptions { sliceMs?: number; @@ -280,23 +301,36 @@ export async function runGoogleHealthBackfillForUser( ); } - const sendNextSlice = async (delaySeconds: number): Promise => { + /** + * Queue the next slice. A slice that waits for a parked collection's retry + * goes under a key named after when the retry is due, instead of its slice + * number: every slice that meets the same wait, a boot's included, names the + * same job, and the queue keeps one. Under the slice number a boot during + * the day-long wait started a second chain beside the waiting one. + */ + const sendNextSlice = async ( + delaySeconds: number, + retryDueAt?: number, + ): Promise => { if (slice + 1 > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) return false; const boss = getGlobalBoss(); - const sent = boss - ? await enqueueIntegrationBackfillAdmission( - boss, - { - kind: "google-health-backfill", - data: { - userId, - enqueuedAt: new Date().toISOString(), - slice: slice + 1, - }, - }, - delaySeconds, - ) - : null; + if (!boss) return false; + const payload: IntegrationBackfillAdmissionPayload = { + kind: "google-health-backfill", + data: { + userId, + enqueuedAt: new Date().toISOString(), + slice: slice + 1, + }, + }; + const sent = + retryDueAt === undefined + ? await enqueueIntegrationBackfillAdmission(boss, payload, delaySeconds) + : await boss.send(INTEGRATION_BACKFILL_ADMISSION_QUEUE, payload, { + ...integrationBackfillAdmissionSendOptions(payload), + singletonKey: googleHealthBackfillRetryKey(userId, retryDueAt), + startAfter: delaySeconds, + }); return Boolean(sent); }; @@ -310,11 +344,18 @@ export async function runGoogleHealthBackfillForUser( (key) => done.has(key) || pending.keys.includes(key), ); if (pending && (onlyRetriesLeft || !failed)) { + const waitSeconds = Math.ceil((pending.dueAt - Date.now()) / 1000); const delaySeconds = Math.max( INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS, - Math.ceil((pending.dueAt - Date.now()) / 1000), + waitSeconds, ); - if (await sendNextSlice(delaySeconds)) { + // A retry already due is an ordinary next slice; only a wait is keyed on + // when it ends. + const delayed = + waitSeconds > INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS; + if ( + await sendNextSlice(delaySeconds, delayed ? pending.dueAt : undefined) + ) { annotate({ action: { name: "google_health.backfill.retry_scheduled", @@ -323,9 +364,10 @@ export async function runGoogleHealthBackfillForUser( }); return { imported, continued: true }; } - // Not queued (most often one is already waiting under the same slice - // key, sent by an earlier boot's slice). Nothing is lost: the marker is - // not stamped, so the next boot offers the backfill again. + // Not queued: most often the same retry is already waiting, sent by the + // slice that parked the collection, and this is a boot's slice meeting + // the same wait. Nothing is lost either way: the marker is not stamped, + // so the next boot offers the backfill again. annotate({ action: { name: "google_health.backfill.retry_not_queued",