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.

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",

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<boolean> => {
+  /**
+   * 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<boolean> => {
     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",