fix(google-health): keep long backfills out of reauth parking and settle overlapping sleep segments

A history backfill carried one access token through walks that outlive it.
After an hour every request answered 401, which classifies as a revoked
grant, so the account was parked at error_reauth with a refresh token that
was still valid, and every later run skipped it until a manual reconnect.
Running beside the hourly sync, the backfill also pushed the account over
Google's per-minute quota, and each 429 failed its whole collection.

- Walks take a token source that renews the access token before it expires
  and once more when Google refuses it mid-walk.
- A 429 is waited out (Retry-After, RetryInfo, else a doubling backoff) and
  the request repeated; it is never a reauth.
- error_reauth only short-circuits a run when the token endpoint refused the
  grant (needsReauth). A connection parked by a data request's 401 refreshes
  on its next run and clears itself when that run completes.
- One run per account at a time: syncs, the backfill and the sleep repair
  share an advisory lock. The hourly poll skips a held account, a manual
  sync answers 409 sync_in_progress, the backfill waits a bounded time.
- A backfill attempt records each collection it finishes cleanly
  (backfill_progress) and the next attempt resumes with the rest.
- The incremental watermark is stamped with the cycle's start, not its end.

Overlapping sleep segments that share a stage and an end share the natural
key (type, measuredAt, source, sleepStage). A re-score that moved one row
onto a sibling's slot threw P2002 and failed the sync on every run.

- A batch keeps one reading per natural key, chosen the same way each time
  (longer segment, then smaller externalId).
- A move that still collides is settled by reconcileExternalMeasurement,
  which retires the stale row, instead of failing the sync.
- Updates run before inserts, and the key-format rescue no longer re-keys a
  row this batch is moving.
- The sleep replace-by-window protects every session's fresh rows, so two
  overlapping sessions no longer tombstone and resurrect each other hourly.

Refs #1194, #1195

src/app/api/google-health/callback/route.ts

@@ -154,6 +154,9 @@ export const GET = apiHandler(async (request: NextRequest) => {
         // reconnect CTA and the sync path stops short-circuiting.
         needsReauth: false,
         backfillCompletedAt: null,
+        // A new grant walks history from the start: what an earlier
+        // backfill finished was read under the old one.
+        backfillProgress: Prisma.DbNull,
       },
       create: {
         userId: user.id,

src/app/api/google-health/sync/route.ts

@@ -104,6 +104,15 @@ export const POST = apiHandler(async (request: NextRequest) => {
   }
 
   const result = await syncUserGoogleHealth(user.id, { fullSync });
+  // Another run of this account holds it (the history backfill after a
+  // reconnect, or the hourly poll). Starting a second one beside it is what
+  // pushed accounts over Google's per-minute quota; the card follows the
+  // running one through the status route instead.
+  if (result.busy) {
+    return apiError("A Google Health sync is already running", 409, {
+      errorCode: "sync_in_progress",
+    });
+  }
   // A run that failed AND wrote nothing is an error, as it always was. A run
   // that failed after writing some of its resources is a partial, and answering
   // 502 for it threw away the honest half of the result.

src/lib/google-health/__tests__/rate-limit-and-token.test.ts

@@ -0,0 +1,213 @@
+/**
+ * #1194 — how one Google Health data request answers a 429 and a 401, and
+ * which of two readings sharing a natural key a write keeps (#1195).
+ *
+ * A 429 is Google's per-user, per-minute quota. It is waited out and the
+ * request sent again: the delay Google asks for when it asks (`Retry-After`,
+ * or a `RetryInfo` detail in the body), otherwise a doubling backoff, and a
+ * request still throttled after the last retry fails as the transient it is.
+ * Never as a reauth.
+ *
+ * A 401 with a token source is answered by asking the source once for a
+ * fresh token and repeating the request; the token had expired, not the
+ * grant. A second 401, or a 401 on a fixed token, still fails as a reauth,
+ * because then it is one.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() }));
+vi.mock("@/lib/safe-fetch", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@/lib/safe-fetch")>();
+  return { ...actual, safeFetch: safeFetchMock };
+});
+
+import {
+  GOOGLE_HEALTH_DATA_TYPES,
+  GOOGLE_HEALTH_RATE_LIMIT,
+  fetchDataPoints,
+  googleHealthRetryDelayMs,
+} from "../client";
+import { collapseNaturalKeyTwins } from "../sync-core";
+import { GoogleHealthApiError } from "../response-classifier";
+
+function response(
+  status: number,
+  body: unknown = {},
+  headers: Record<string, string> = {},
+): Response {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { "content-type": "application/json", ...headers },
+  });
+}
+
+const throttled = () =>
+  response(429, {
+    error: { code: 429, status: "RESOURCE_EXHAUSTED", message: "Quota" },
+  });
+
+beforeEach(() => {
+  safeFetchMock.mockReset();
+});
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe("a 429 from a data read", () => {
+  it("waits with a doubling backoff and repeats the request", async () => {
+    vi.useFakeTimers();
+    safeFetchMock
+      .mockResolvedValueOnce(throttled())
+      .mockResolvedValueOnce(throttled())
+      .mockResolvedValueOnce(response(200, { dataPoints: [{ a: 1 }] }));
+
+    const read = fetchDataPoints(
+      GOOGLE_HEALTH_DATA_TYPES.weight,
+      "token",
+      "fetchWeight",
+    );
+    await vi.advanceTimersByTimeAsync(GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs);
+    expect(safeFetchMock).toHaveBeenCalledTimes(2);
+    await vi.advanceTimersByTimeAsync(GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs * 2);
+
+    await expect(read).resolves.toEqual([{ a: 1 }]);
+    expect(safeFetchMock).toHaveBeenCalledTimes(3);
+  });
+
+  it("fails as transient, never as a reauth, once the retries run out", async () => {
+    vi.useFakeTimers();
+    safeFetchMock.mockImplementation(async () => throttled());
+
+    const read = fetchDataPoints(
+      GOOGLE_HEALTH_DATA_TYPES.weight,
+      "token",
+      "fetchWeight",
+    ).catch((err: unknown) => err);
+    await vi.advanceTimersByTimeAsync(10 * 60 * 1000);
+    const err = await read;
+
+    expect(err).toBeInstanceOf(GoogleHealthApiError);
+    expect((err as GoogleHealthApiError).classification).toBe("transient");
+    expect(safeFetchMock).toHaveBeenCalledTimes(
+      GOOGLE_HEALTH_RATE_LIMIT.maxRetries + 1,
+    );
+  });
+
+  it("does not sleep out a delay longer than the cap", async () => {
+    safeFetchMock.mockResolvedValue(
+      response(429, {}, { "retry-after": "3600" }),
+    );
+
+    await expect(
+      fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, "token", "fetchWeight"),
+    ).rejects.toMatchObject({ classification: "transient" });
+    expect(safeFetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("reads the delay Google asks for from the header or the body", () => {
+    const now = Date.parse("2026-10-10T12:00:00.000Z");
+    expect(
+      googleHealthRetryDelayMs(response(429, {}, { "retry-after": "7" }), {}),
+    ).toBe(7000);
+    expect(
+      googleHealthRetryDelayMs(
+        response(429, {}, { "retry-after": "Sat, 10 Oct 2026 12:00:30 GMT" }),
+        {},
+        now,
+      ),
+    ).toBe(30_000);
+    expect(
+      googleHealthRetryDelayMs(response(429), {
+        error: {
+          details: [
+            {
+              "@type": "type.googleapis.com/google.rpc.RetryInfo",
+              retryDelay: "12.5s",
+            },
+          ],
+        },
+      }),
+    ).toBe(12_500);
+    expect(googleHealthRetryDelayMs(response(429), {})).toBeUndefined();
+  });
+});
+
+describe("a 401 from a data read", () => {
+  it("asks a token source once for a fresh token and repeats the request", async () => {
+    safeFetchMock
+      .mockResolvedValueOnce(response(401))
+      .mockResolvedValueOnce(response(200, { dataPoints: [{ a: 1 }] }));
+    const source = vi.fn(async ({ forceRefresh }: { forceRefresh: boolean }) =>
+      forceRefresh ? "fresh" : "stale",
+    );
+
+    await expect(
+      fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, source, "fetchWeight"),
+    ).resolves.toEqual([{ a: 1 }]);
+    expect(source.mock.calls.map(([o]) => o.forceRefresh)).toEqual([
+      false,
+      true,
+    ]);
+    const auth = (n: number) =>
+      new Headers((safeFetchMock.mock.calls[n]![1] as RequestInit).headers).get(
+        "authorization",
+      );
+    expect(auth(0)).toBe("Bearer stale");
+    expect(auth(1)).toBe("Bearer fresh");
+  });
+
+  it("is a reauth when the fresh token is refused too", async () => {
+    safeFetchMock.mockResolvedValue(response(401));
+
+    await expect(
+      fetchDataPoints(
+        GOOGLE_HEALTH_DATA_TYPES.weight,
+        async () => "token",
+        "fetchWeight",
+      ),
+    ).rejects.toMatchObject({ classification: "reauth_required" });
+    expect(safeFetchMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("is a reauth at once on a fixed token", async () => {
+    safeFetchMock.mockResolvedValue(response(401));
+
+    await expect(
+      fetchDataPoints(GOOGLE_HEALTH_DATA_TYPES.weight, "token", "fetchWeight"),
+    ).rejects.toMatchObject({ classification: "reauth_required" });
+    expect(safeFetchMock).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("collapseNaturalKeyTwins (#1195)", () => {
+  const end = new Date("2026-10-09T02:00:00.000Z");
+  const reading = (externalId: string, value: number, stage = "CORE") => ({
+    type: "SLEEP_DURATION",
+    value,
+    unit: "minutes",
+    measuredAt: end,
+    externalId,
+    sleepStage: stage as "CORE" | "DEEP",
+  });
+
+  it("keeps the longer of two segments sharing a stage and an end", () => {
+    const { kept, dropped } = collapseNaturalKeyTwins([
+      reading("s:sleep:00:30", 90),
+      reading("s:sleep:00:00", 120),
+      reading("s:sleep:01:00", 60, "DEEP"),
+    ]);
+    expect(kept.map((r) => r.externalId)).toEqual([
+      "s:sleep:00:00",
+      "s:sleep:01:00",
+    ]);
+    expect(dropped).toBe(1);
+  });
+
+  it("picks the same reading whatever order the pair arrives in", () => {
+    const a = reading("phone:sleep:00:00", 60);
+    const b = reading("watch:sleep:00:00", 60);
+    expect(collapseNaturalKeyTwins([a, b]).kept).toEqual([a]);
+    expect(collapseNaturalKeyTwins([b, a]).kept).toEqual([a]);
+  });
+});

src/lib/google-health/__tests__/sync-activity-failsoft.test.ts

@@ -28,6 +28,10 @@ vi.mock("../sync-core", () => ({
     connection: { id: "c1", googleUserId: "g1" },
   })),
   handleCollectionFetchError: handleErrorMock,
+  googleHealthTokenSource:
+    (_userId: string, info: { accessToken: string }) => async () =>
+      info.accessToken,
+  runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
   upsertGoogleHealthMeasurements: vi.fn(async () => ({
     imported: 0,
     touched: [],

src/lib/google-health/__tests__/sync-failsoft.test.ts

@@ -28,6 +28,7 @@ const {
   recordSyncSuccessMock: vi.fn(async () => {}),
   invalidateUserMeasurementsMock: vi.fn(),
   prismaMock: {
+    integrationStatus: { findUnique: vi.fn(async () => null) },
     googleHealthConnection: {
       findUnique: vi.fn(),
       update: vi.fn(async () => ({})),
@@ -47,6 +48,18 @@ const {
 }));
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
+
+// The account lock is a Postgres advisory lock on a connection of its own;
+// these cases run without a database, so the lock is always free.
+vi.mock("../sync-lock", () => ({
+  withGoogleHealthSyncLock: async (
+    _userId: string,
+    run: () => Promise<unknown>,
+  ) => ({
+    ran: true,
+    result: await run(),
+  }),
+}));
 vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() }));
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn(async () => false),

src/lib/google-health/__tests__/sync-fairness.test.ts

@@ -14,6 +14,7 @@ const { calls, prismaMock, recordSyncSuccess, resources } = vi.hoisted(() => {
   return {
     calls,
     prismaMock: {
+      integrationStatus: { findUnique: vi.fn(async () => null) },
       googleHealthConnection: {
         findUnique: vi.fn(async () => ({ lastSyncedAt: null })),
         update: vi.fn(async () => ({})),
@@ -45,6 +46,18 @@ const { calls, prismaMock, recordSyncSuccess, resources } = vi.hoisted(() => {
 });
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
+
+// The account lock is a Postgres advisory lock on a connection of its own;
+// these cases run without a database, so the lock is always free.
+vi.mock("../sync-lock", () => ({
+  withGoogleHealthSyncLock: async (
+    _userId: string,
+    run: () => Promise<unknown>,
+  ) => ({
+    ran: true,
+    result: await run(),
+  }),
+}));
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn(async () => false),
   recordSyncSuccess,

src/lib/google-health/__tests__/sync-intraday-daily-wide.test.ts

@@ -11,6 +11,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const { prismaMock, metrics } = vi.hoisted(() => ({
   prismaMock: {
+    integrationStatus: { findUnique: vi.fn(async () => null) },
     googleHealthConnection: {
       findUnique: vi.fn(),
       update: vi.fn(async () => ({})),
@@ -20,6 +21,18 @@ const { prismaMock, metrics } = vi.hoisted(() => ({
 }));
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
+
+// The account lock is a Postgres advisory lock on a connection of its own;
+// these cases run without a database, so the lock is always free.
+vi.mock("../sync-lock", () => ({
+  withGoogleHealthSyncLock: async (
+    _userId: string,
+    run: () => Promise<unknown>,
+  ) => ({
+    ran: true,
+    result: await run(),
+  }),
+}));
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn(async () => false),
   recordSyncSuccess: vi.fn(async () => {}),

src/lib/google-health/__tests__/sync-intraday-overlap.test.ts

@@ -21,7 +21,10 @@ vi.mock("../sync-core", async (importOriginal) => {
   const actual = await importOriginal<typeof import("../sync-core")>();
   return {
     ...actual,
-    getValidToken: vi.fn(async () => ({ accessToken: "token" })),
+    getValidToken: vi.fn(async () => ({
+      accessToken: "token",
+      expiresAt: new Date(Date.now() + 60 * 60 * 1000),
+    })),
     handleCollectionFetchError: vi.fn(async () => 0),
     upsertGoogleHealthMeasurements: vi.fn(async () => ({
       imported: 0,

src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts

@@ -28,6 +28,10 @@ vi.mock("../sync-core", () => ({
   getValidToken: vi.fn(async () => ({ accessToken: "token" })),
   handleCollectionFetchError: vi.fn(async () => 0),
   upsertGoogleHealthMeasurements: upsertMock,
+  googleHealthTokenSource:
+    (_userId: string, info: { accessToken: string }) => async () =>
+      info.accessToken,
+  runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
 }));
 
 vi.mock("@/lib/db", () => ({

src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts

@@ -54,6 +54,10 @@ vi.mock("../sync-core", () => ({
     connection: { id: "c1", googleUserId: "g1" },
   })),
   handleCollectionFetchError: handleErrorMock,
+  googleHealthTokenSource:
+    (_userId: string, info: { accessToken: string }) => async () =>
+      info.accessToken,
+  runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
   upsertGoogleHealthMeasurements: vi.fn(async () => ({
     imported: 0,
     touched: [],

src/lib/google-health/__tests__/sync-outcomes.test.ts

@@ -112,6 +112,7 @@ const resourceOutcomes = [
 
 const { prismaMock, runCycle } = vi.hoisted(() => ({
   prismaMock: {
+    integrationStatus: { findUnique: vi.fn(async () => null) },
     googleHealthConnection: {
       findUnique: vi.fn(async () => ({ lastSyncedAt: null })),
       update: vi.fn(async () => ({})),
@@ -121,6 +122,18 @@ const { prismaMock, runCycle } = vi.hoisted(() => ({
 }));
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
+
+// The account lock is a Postgres advisory lock on a connection of its own;
+// these cases run without a database, so the lock is always free.
+vi.mock("../sync-lock", () => ({
+  withGoogleHealthSyncLock: async (
+    _userId: string,
+    run: () => Promise<unknown>,
+  ) => ({
+    ran: true,
+    result: await run(),
+  }),
+}));
 vi.mock("@/lib/integrations/status", () => ({
   isReauthRequired: vi.fn(async () => false),
   recordSyncSuccess: vi.fn(async () => {}),
@@ -134,6 +147,7 @@ vi.mock("@/lib/insights/comprehensive-generate", () => ({
 }));
 vi.mock("../sync-core", () => ({
   GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
+  getValidToken: vi.fn(),
   incrementalStart: vi.fn(() => undefined),
   intradayOverlapMs: vi.fn(() => 2 * 60 * 60 * 1000),
   markSynced: vi.fn(async () => {}),

src/lib/google-health/__tests__/sync-sleep-morning-refresh.test.ts

@@ -36,6 +36,10 @@ vi.mock("../client", () => ({
 vi.mock("../sync-core", () => ({
   getValidToken: vi.fn(async () => ({ accessToken: "tok" })),
   handleCollectionFetchError: vi.fn(() => 0),
+  googleHealthTokenSource:
+    (_userId: string, info: { accessToken: string }) => async () =>
+      info.accessToken,
+  runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
   noteHardFailure: vi.fn(),
   replaceStaleGoogleHealthSleep: vi.fn(async () => 0),
   upsertGoogleHealthMeasurements: upsertMock,

src/lib/google-health/__tests__/sync-workout-insert-identity.test.ts

@@ -38,6 +38,10 @@ vi.mock("../client", () => ({
 vi.mock("../sync-core", () => ({
   getValidToken,
   handleCollectionFetchError: vi.fn(),
+  googleHealthTokenSource:
+    (_userId: string, info: { accessToken: string }) => async () =>
+      info.accessToken,
+  runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
   noteHardFailure: vi.fn(),
 }));
 

src/lib/google-health/client.ts

@@ -106,6 +106,11 @@ export function noteGoogleHealthOutcomeFailure(
   if (tracker && !tracker.reasonCode) tracker.reasonCode = reasonCode;
 }
 
+/** Whether a walk in the current resource scope has ended short so far. */
+export function googleHealthOutcomeTruncated(): boolean {
+  return clientOutcomeStorage.getStore()?.truncated === true;
+}
+
 export async function runWithGoogleHealthClientOutcome<T>(
   fn: () => Promise<T>,
 ): Promise<{ result: T; outcome: GoogleHealthClientOutcome }> {
@@ -729,6 +734,167 @@ export function extractGoogleApiErrorDetail(json: unknown): string | undefined {
   return [head, tail].filter(Boolean).join(" ").slice(0, 500);
 }
 
+// ─── Token source + one data request ───────────────────────────
+
+/**
+ * Where a walk gets its access token. A plain string is used as given. A
+ * function is asked before every request, so a walk that runs for longer than
+ * the token lives (an access token lasts an hour; a full-history heart-rate
+ * walk can take several) picks up the refreshed one instead of carrying an
+ * expired token into a 401. `forceRefresh` asks for a new token even though
+ * the current one has not reached its expiry, after Google refused it.
+ */
+export type GoogleHealthTokenSource = (opts: {
+  forceRefresh: boolean;
+}) => Promise<string>;
+
+export type GoogleHealthAccessToken = string | GoogleHealthTokenSource;
+
+/**
+ * How a data request sits out Google's per-user, per-minute quota (429
+ * `RESOURCE_EXHAUSTED`). The quota refills within the minute, so a short wait
+ * turns a throttled request into a completed one, where failing it used to
+ * cost the whole collection. The waits follow `Retry-After` (or the
+ * `google.rpc.RetryInfo` delay in the body) when Google sends one, and
+ * otherwise double from `baseDelayMs`: 2, 4, 8, 16, 32 s, a minute and a
+ * bit in all, which spans a full quota window. A requested delay above
+ * `maxDelayMs` is not slept out; the request fails as the transient it is
+ * and the next run picks the window up.
+ */
+export const GOOGLE_HEALTH_RATE_LIMIT = {
+  maxRetries: 5,
+  baseDelayMs: 2_000,
+  maxDelayMs: 60_000,
+} as const;
+
+/**
+ * The delay Google asked for on a 429, in ms: the `Retry-After` header in
+ * either of its forms, else a `google.rpc.RetryInfo` detail (`"30s"`) in the
+ * error body. Undefined when it asked for nothing readable.
+ */
+export function googleHealthRetryDelayMs(
+  res: Response,
+  json: unknown,
+  now: number = Date.now(),
+): number | undefined {
+  const header = res.headers.get("retry-after")?.trim();
+  if (header) {
+    if (/^\d+$/.test(header)) return Number.parseInt(header, 10) * 1000;
+    const at = Date.parse(header);
+    if (Number.isFinite(at)) return Math.max(0, at - now);
+  }
+  const details = (json as { error?: { details?: unknown } } | null)?.error
+    ?.details;
+  if (Array.isArray(details)) {
+    for (const d of details) {
+      const delay = (d as { retryDelay?: unknown } | null)?.retryDelay;
+      if (typeof delay === "string") {
+        const m = /^(\d+(?:\.\d+)?)s$/.exec(delay.trim());
+        if (m) return Math.round(Number.parseFloat(m[1]!) * 1000);
+      }
+    }
+  }
+  return undefined;
+}
+
+function waitMs(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+interface GoogleHealthDataResponse<T> {
+  status: number;
+  json: T | null;
+  verdict: ReturnType<typeof classifyGoogleHealthResponse>;
+  apiErrorDetail: string | undefined;
+}
+
+/**
+ * Send one data request and answer with the response it finally got.
+ *
+ * Two answers are handled here rather than handed back:
+ *   - 429: the request waits (see `GOOGLE_HEALTH_RATE_LIMIT`) and is sent
+ *     again, up to `maxRetries` times, unless the caller's stop condition
+ *     says to end the walk.
+ *   - 401 with a token source: the token expired or was revoked between
+ *     the source's last check and this request. The source is asked once for
+ *     a fresh token and the request is repeated. A second 401, or a 401 on a
+ *     fixed string token, is returned like any other answer.
+ *
+ * Every attempt is recorded as its own external call, so the wide event
+ * shows the throttling and the waits rather than only their outcome.
+ */
+async function googleHealthDataRequest<T>(
+  token: GoogleHealthAccessToken,
+  label: string,
+  send: (accessToken: string) => Promise<Response>,
+  isExpectedFallback: (status: number) => boolean = () => false,
+): Promise<GoogleHealthDataResponse<T>> {
+  let rateLimited = 0;
+  let refreshed = false;
+  let forceRefresh = false;
+  for (;;) {
+    const accessToken =
+      typeof token === "string" ? token : await token({ forceRefresh });
+    forceRefresh = false;
+    const started = performance.now();
+    const res = await send(accessToken);
+    const json = (await res.json().catch(() => null)) as T | null;
+    const verdict = classifyGoogleHealthResponse(res.status);
+    const apiErrorDetail =
+      verdict.classification === "success"
+        ? undefined
+        : extractGoogleApiErrorDetail(json);
+
+    const retryAfterMs =
+      res.status === 429 ? googleHealthRetryDelayMs(res, json) : undefined;
+    const backoffMs =
+      res.status === 429
+        ? (retryAfterMs ??
+          GOOGLE_HEALTH_RATE_LIMIT.baseDelayMs * 2 ** rateLimited)
+        : 0;
+    const willWait =
+      res.status === 429 &&
+      rateLimited < GOOGLE_HEALTH_RATE_LIMIT.maxRetries &&
+      backoffMs <= GOOGLE_HEALTH_RATE_LIMIT.maxDelayMs &&
+      !stopRequested();
+    const willRefresh =
+      res.status === 401 && typeof token !== "string" && !refreshed;
+
+    getEvent()?.addExternalCall({
+      service: "google-health",
+      method: label,
+      duration_ms: Math.round(performance.now() - started),
+      status: res.status,
+      ...(isExpectedFallback(res.status)
+        ? { note: "date_filter_rejected_retrying_snake_case" }
+        : willWait
+          ? { note: `rate_limited_retrying_in_ms=${backoffMs}` }
+          : willRefresh
+            ? { note: "token_rejected_retrying_with_fresh_token" }
+            : {
+                error:
+                  verdict.classification === "success"
+                    ? undefined
+                    : apiErrorDetail
+                      ? `${verdict.reason} ${apiErrorDetail}`
+                      : verdict.reason,
+              }),
+    });
+
+    if (willWait) {
+      rateLimited += 1;
+      await waitMs(backoffMs);
+      continue;
+    }
+    if (willRefresh) {
+      refreshed = true;
+      forceRefresh = true;
+      continue;
+    }
+    return { status: res.status, json, verdict, apiErrorDetail };
+  }
+}
+
 /**
  * Walk every `DataPoint` for one data type since the incremental cursor via
  * `dataPoints.list` with `nextPageToken` pagination. The data-type id is
@@ -748,7 +914,7 @@ export function extractGoogleApiErrorDetail(json: unknown): string | undefined {
  */
 export async function fetchDataPoints(
   dataType: GoogleHealthDataType,
-  accessToken: string,
+  accessToken: GoogleHealthAccessToken,
   verb: string,
   query: DataPointQuery = {},
 ): Promise<GoogleHealthDataPoint[]> {
@@ -776,7 +942,7 @@ export async function fetchDataPoints(
  */
 export async function forEachDataPointPage(
   dataType: GoogleHealthDataType,
-  accessToken: string,
+  accessToken: GoogleHealthAccessToken,
   verb: string,
   query: DataPointQuery,
   onPage: (points: GoogleHealthDataPoint[]) => void | Promise<void>,
@@ -812,50 +978,31 @@ export async function forEachDataPointPage(
       if (pageToken) params.set("pageToken", pageToken);
 
       requestCount += 1;
-      const pageStart = performance.now();
-      const res = await safeFetch(
-        `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
-        {
-          method: "GET",
-          headers: { Authorization: `Bearer ${accessToken}` },
-        },
-      );
-
-      const json = (await res
-        .json()
-        .catch(() => null)) as GoogleHealthDataPointPage | null;
-      const verdict = classifyGoogleHealthResponse(res.status);
-      const apiErrorDetail =
-        verdict.classification === "success"
-          ? undefined
-          : extractGoogleApiErrorDetail(json);
       // v1.42 (#1023) — the first filtered daily-summary read answering 400
       // is the expected trigger for the snake_case retry below, not a
       // failure: some accounts reject the documented camelCase prefix on
       // some types, and the very next request answers 200. It was logged as
       // an error next to the 200 that followed it. It is recorded as a note
       // now; any other 400 still carries its error.
-      const expectedFallback =
-        canFallBack &&
-        dateStyle === "camel" &&
-        requestCount === 1 &&
-        res.status === 400;
-      getEvent()?.addExternalCall({
-        service: "google-health",
-        method: `${verb}(page=${pageCount})`,
-        duration_ms: Math.round(performance.now() - pageStart),
-        status: res.status,
-        ...(expectedFallback
-          ? { note: "date_filter_rejected_retrying_snake_case" }
-          : {
-              error:
-                verdict.classification === "success"
-                  ? undefined
-                  : apiErrorDetail
-                    ? `${verdict.reason} ${apiErrorDetail}`
-                    : verdict.reason,
-            }),
-      });
+      const firstRequest = requestCount === 1;
+      const { json, verdict, apiErrorDetail } =
+        await googleHealthDataRequest<GoogleHealthDataPointPage>(
+          accessToken,
+          `${verb}(page=${pageCount})`,
+          (token) =>
+            safeFetch(
+              `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
+              {
+                method: "GET",
+                headers: { Authorization: `Bearer ${token}` },
+              },
+            ),
+          (status) =>
+            canFallBack &&
+            dateStyle === "camel" &&
+            firstRequest &&
+            status === 400,
+        );
       if (verdict.classification !== "success") {
         throw new GoogleHealthApiError({
           verb,
@@ -1078,7 +1225,7 @@ export function buildDailyRollUpBody(
  */
 export async function fetchDailyRollUp(
   dataType: GoogleHealthDataType,
-  accessToken: string,
+  accessToken: GoogleHealthAccessToken,
   verb: string,
   query: RollupQuery = {},
 ): Promise<GoogleHealthRollupPoint[]> {
@@ -1108,38 +1255,23 @@ export async function fetchDailyRollUp(
         const body = buildDailyRollUpBody(chunk, pageToken ?? undefined);
 
         requestCount += 1;
-        const reqStart = performance.now();
-        const res = await safeFetch(
-          `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`,
-          {
-            method: "POST",
-            headers: {
-              Authorization: `Bearer ${accessToken}`,
-              "Content-Type": "application/json",
-            },
-            body: JSON.stringify(body),
-          },
-        );
-        const json = (await res
-          .json()
-          .catch(() => null)) as GoogleHealthRollupPage | null;
-        const verdict = classifyGoogleHealthResponse(res.status);
-        const apiErrorDetail =
-          verdict.classification === "success"
-            ? undefined
-            : extractGoogleApiErrorDetail(json);
-        getEvent()?.addExternalCall({
-          service: "google-health",
-          method: `${verb}(chunk=${chunkIndex},page=${pageCount})`,
-          duration_ms: Math.round(performance.now() - reqStart),
-          status: res.status,
-          error:
-            verdict.classification === "success"
-              ? undefined
-              : apiErrorDetail
-                ? `${verdict.reason} ${apiErrorDetail}`
-                : verdict.reason,
-        });
+        const { json, verdict, apiErrorDetail } =
+          await googleHealthDataRequest<GoogleHealthRollupPage>(
+            accessToken,
+            `${verb}(chunk=${chunkIndex},page=${pageCount})`,
+            (token) =>
+              safeFetch(
+                `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`,
+                {
+                  method: "POST",
+                  headers: {
+                    Authorization: `Bearer ${token}`,
+                    "Content-Type": "application/json",
+                  },
+                  body: JSON.stringify(body),
+                },
+              ),
+          );
         if (verdict.classification !== "success") {
           throw new GoogleHealthApiError({
             verb,

src/lib/google-health/sync-activity.ts

@@ -38,7 +38,9 @@ import {
 } from "./client";
 import {
   getValidToken,
+  googleHealthTokenSource,
   handleCollectionFetchError,
+  runGoogleHealthCollection,
   upsertGoogleHealthMeasurements,
   type GoogleHealthMeasurementUpsert,
   type GoogleHealthResourceSyncOptions,
@@ -96,6 +98,7 @@ export async function syncUserActivity(
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
+  const token = googleHealthTokenSource(userId, tokenInfo);
 
   // The dailyRollUp request range is civil and user-local; resolve the user's
   // stored zone so the range bounds land on the correct civil days rather than
@@ -118,51 +121,51 @@ export async function syncUserActivity(
   const rollupEmptyResponse: string[] = [];
   const rollupDroppedAll: string[] = [];
   for (const resource of ROLLUP_RESOURCES) {
-    let points: Record<string, unknown>[];
-    try {
-      points = await fetchDailyRollUp(
-        resource.dataType,
-        tokenInfo.accessToken,
-        resource.verb,
-        { start, tz },
-      );
-    } catch (err) {
-      imported += await handleCollectionFetchError(resource.verb, userId, err);
-      continue;
-    }
-
-    // The mapper runs INSIDE a per-type catch too (mirrors sync-metrics): a
-    // single malformed point whose `resource.map(point)` throws must not
-    // escape the ROLLUP_RESOURCES loop and abort every sibling type ordered
-    // after it. Route a map throw through the same ledger as a fetch failure —
-    // record it, fail the cycle, and move on to the next type.
-    const readings: GoogleHealthMeasurementUpsert[] = [];
-    try {
-      for (const point of points) {
-        for (const m of resource.map(point, tz)) {
-          readings.push({
-            type: m.type,
-            value: m.value,
-            unit: m.unit,
-            measuredAt: m.measuredAt,
-            externalId: externalIdFor(m),
-          });
-        }
+    imported += await runGoogleHealthCollection(resource.verb, async () => {
+      let points: Record<string, unknown>[];
+      try {
+        points = await fetchDailyRollUp(
+          resource.dataType,
+          token,
+          resource.verb,
+          { start, tz },
+        );
+      } catch (err) {
+        return handleCollectionFetchError(resource.verb, userId, err);
       }
-    } catch (err) {
-      imported += await handleCollectionFetchError(resource.verb, userId, err);
-      continue;
-    }
-    if (points.length === 0) {
-      rollupEmptyResponse.push(resource.verb);
-    } else if (readings.length === 0) {
-      rollupDroppedAll.push(resource.verb);
-    }
-    imported += (
-      await upsertGoogleHealthMeasurements(userId, readings, {
-        deferRollup: opts.deferRollup,
-      })
-    ).imported;
+
+      // The mapper runs INSIDE a per-type catch too (mirrors sync-metrics): a
+      // single malformed point whose `resource.map(point)` throws must not
+      // escape the ROLLUP_RESOURCES loop and abort every sibling type ordered
+      // after it. Route a map throw through the same ledger as a fetch failure —
+      // record it, fail the cycle, and move on to the next type.
+      const readings: GoogleHealthMeasurementUpsert[] = [];
+      try {
+        for (const point of points) {
+          for (const m of resource.map(point, tz)) {
+            readings.push({
+              type: m.type,
+              value: m.value,
+              unit: m.unit,
+              measuredAt: m.measuredAt,
+              externalId: externalIdFor(m),
+            });
+          }
+        }
+      } catch (err) {
+        return handleCollectionFetchError(resource.verb, userId, err);
+      }
+      if (points.length === 0) {
+        rollupEmptyResponse.push(resource.verb);
+      } else if (readings.length === 0) {
+        rollupDroppedAll.push(resource.verb);
+      }
+      return (
+        await upsertGoogleHealthMeasurements(userId, readings, {
+          deferRollup: opts.deferRollup,
+        })
+      ).imported;
+    });
   }
   if (rollupEmptyResponse.length > 0 || rollupDroppedAll.length > 0) {
     annotate({
@@ -174,33 +177,35 @@ export async function syncUserActivity(
   }
 
   // VO2 max — a daily summary (list + `.date` filter), not a rollup type.
-  try {
-    const points = await fetchDataPoints(
-      GOOGLE_HEALTH_DATA_TYPES.vo2Max,
-      tokenInfo.accessToken,
-      "fetchVo2Max",
-      { start },
-    );
-    const readings: GoogleHealthMeasurementUpsert[] = [];
-    for (const point of points) {
-      for (const m of mapVo2Max(point, tz)) {
-        readings.push({
-          type: m.type,
-          value: m.value,
-          unit: m.unit,
-          measuredAt: m.measuredAt,
-          externalId: externalIdFor(m),
-        });
+  imported += await runGoogleHealthCollection("fetchVo2Max", async () => {
+    try {
+      const points = await fetchDataPoints(
+        GOOGLE_HEALTH_DATA_TYPES.vo2Max,
+        token,
+        "fetchVo2Max",
+        { start },
+      );
+      const readings: GoogleHealthMeasurementUpsert[] = [];
+      for (const point of points) {
+        for (const m of mapVo2Max(point, tz)) {
+          readings.push({
+            type: m.type,
+            value: m.value,
+            unit: m.unit,
+            measuredAt: m.measuredAt,
+            externalId: externalIdFor(m),
+          });
+        }
       }
+      return (
+        await upsertGoogleHealthMeasurements(userId, readings, {
+          deferRollup: opts.deferRollup,
+        })
+      ).imported;
+    } catch (err) {
+      return handleCollectionFetchError("fetchVo2Max", userId, err);
     }
-    imported += (
-      await upsertGoogleHealthMeasurements(userId, readings, {
-        deferRollup: opts.deferRollup,
-      })
-    ).imported;
-  } catch (err) {
-    imported += await handleCollectionFetchError("fetchVo2Max", userId, err);
-  }
+  });
 
   // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`).
   annotate({

src/lib/google-health/sync-core.ts

@@ -19,8 +19,13 @@ import { AsyncLocalStorage } from "node:async_hooks";
 import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { encrypt, decrypt } from "@/lib/crypto";
-import { getEvent } from "@/lib/logging/context";
+import { annotate, getEvent } from "@/lib/logging/context";
 import { userDayKey } from "@/lib/tz/format";
+import { isP2002 } from "@/lib/prisma-errors";
+import {
+  reconcileExternalMeasurement,
+  type MeasurementReconciliationVerdict,
+} from "@/lib/measurements/reconcile-external-measurement";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status";
@@ -30,10 +35,12 @@ import {
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
+  googleHealthOutcomeTruncated,
   noteGoogleHealthMapped,
   noteGoogleHealthOutcomeFailure,
   noteGoogleHealthWritten,
   refreshAccessToken,
+  type GoogleHealthTokenSource,
 } from "./client";
 import { getUserGoogleHealthCredentials } from "./credentials";
 import {
@@ -99,6 +106,8 @@ export function intradayOverlapMs(
 
 export interface GoogleHealthTokenInfo {
   accessToken: string;
+  /** When `accessToken` stops being accepted. */
+  expiresAt: Date;
   connection: { id: string; googleUserId: string };
 }
 
@@ -115,6 +124,7 @@ export interface GoogleHealthTokenInfo {
  */
 export async function getValidToken(
   userId: string,
+  opts: { forceRefresh?: boolean } = {},
 ): Promise<GoogleHealthTokenInfo | null> {
   const connection = await prisma.googleHealthConnection.findUnique({
     where: { userId },
@@ -125,8 +135,8 @@ export async function getValidToken(
   const refreshToken = decrypt(connection.refreshToken);
 
   if (
-    connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS <
-    Date.now()
+    opts.forceRefresh === true ||
+    connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < Date.now()
   ) {
     try {
       const creds = await getUserGoogleHealthCredentials(userId);
@@ -170,6 +180,7 @@ export async function getValidToken(
 
       return {
         accessToken: newTokens.access_token,
+        expiresAt,
         connection: {
           id: connection.id,
           googleUserId: connection.googleUserId,
@@ -209,6 +220,7 @@ export async function getValidToken(
 
   return {
     accessToken,
+    expiresAt: connection.tokenExpiresAt,
     connection: {
       id: connection.id,
       googleUserId: connection.googleUserId,
@@ -216,6 +228,51 @@ export async function getValidToken(
   };
 }
 
+/**
+ * Thrown by a token source when no valid token can be had any more. The
+ * failure is already on the ledger and the cycle's verdict by then
+ * (`getValidToken` records it), so the collection that hit it only has to
+ * stop.
+ */
+export class GoogleHealthTokenUnavailableError extends Error {
+  constructor() {
+    super("Google Health access token unavailable");
+    this.name = "GoogleHealthTokenUnavailableError";
+  }
+}
+
+/**
+ * The token source a resource hands to its walks (see
+ * `GoogleHealthTokenSource`). It starts from the token the resource resolved
+ * and renews it whenever it comes within the refresh buffer of expiring, or
+ * when a request reports it refused.
+ *
+ * Before this the resource resolved one token and every walk carried it to
+ * the end. An access token lives an hour, a full-history heart-rate walk runs
+ * for several, and once the hour was up every remaining request answered 401.
+ * A 401 classifies as a revoked grant, so the account was parked at
+ * `error_reauth` over a token that had simply expired, with a refresh token
+ * that was still perfectly good, and stayed parked until it was reconnected
+ * by hand (#1194).
+ */
+export function googleHealthTokenSource(
+  userId: string,
+  initial: GoogleHealthTokenInfo,
+): GoogleHealthTokenSource {
+  let current = initial;
+  return async ({ forceRefresh }) => {
+    if (
+      forceRefresh ||
+      current.expiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < Date.now()
+    ) {
+      const next = await getValidToken(userId, { forceRefresh });
+      if (!next) throw new GoogleHealthTokenUnavailableError();
+      current = next;
+    }
+    return current.accessToken;
+  };
+}
+
 /**
  * True when a caught error is a per-data-class 403 (forbidden). The Google
  * Health Restricted bundles are granted independently in the consent flow, so a
@@ -343,6 +400,106 @@ export async function runWithGoogleHealthSyncCycle<T>(
   };
 }
 
+/**
+ * Collections a full-history backfill already walked to the end, carried from
+ * one attempt of the backfill to the next.
+ *
+ * A backfill that ended early (its job budget ran out, a collection failed)
+ * used to start the next attempt from the first page of the first collection,
+ * so the dense heart-rate walk, thousands of pages, was paid for again on
+ * every retry, and an account whose walk never fit into one attempt never
+ * finished at all. Each collection that completes cleanly is now written to
+ * `backfillProgress`, and the next attempt skips it and goes on with the
+ * collections it did not reach. A collection is the unit because Google
+ * offers no cursor that survives between attempts; its page tokens are not
+ * documented to outlive the walk that received them.
+ */
+interface BackfillResumeTracker {
+  userId: string;
+  done: Set<string>;
+  /** Collections this attempt skipped because an earlier one finished them. */
+  skipped: string[];
+}
+const backfillResumeStorage = new AsyncLocalStorage<BackfillResumeTracker>();
+
+/** Read the `backfillProgress` column into the set of finished collections. */
+export function readGoogleHealthBackfillDone(value: unknown): Set<string> {
+  const done =
+    value && typeof value === "object" && !Array.isArray(value)
+      ? (value as { done?: unknown }).done
+      : undefined;
+  return new Set(
+    Array.isArray(done)
+      ? done.filter(
+          (key): key is string => typeof key === "string" && key.length <= 64,
+        )
+      : [],
+  );
+}
+
+/**
+ * Run a full-history backfill attempt that resumes: collections in `done` are
+ * skipped, and each collection finished cleanly during this attempt is added
+ * to it and persisted at once, so an attempt that dies part-way keeps what it
+ * finished.
+ */
+export async function runWithGoogleHealthBackfillResume<T>(
+  userId: string,
+  done: Set<string>,
+  fn: () => Promise<T>,
+): Promise<{ result: T; skipped: string[] }> {
+  const tracker: BackfillResumeTracker = { userId, done, skipped: [] };
+  const result = await backfillResumeStorage.run(tracker, fn);
+  return { result, skipped: tracker.skipped };
+}
+
+/**
+ * Walk one collection, under the backfill's resume bookkeeping when there is
+ * one in scope. Outside a backfill this is just `fn()`.
+ *
+ * Inside one, a collection already walked to the end by an earlier attempt is
+ * skipped. Otherwise it runs, and it counts as finished only when nothing
+ * went wrong while it ran: no hard failure on the cycle's ledger, no 403
+ * soft-skip, and no walk that ended short (its page ceiling, or the job's
+ * budget asking it to stop). Anything less is walked again next attempt.
+ */
+export async function runGoogleHealthCollection(
+  key: string,
+  fn: () => Promise<number>,
+): Promise<number> {
+  const resume = backfillResumeStorage.getStore();
+  if (!resume) return fn();
+  if (resume.done.has(key)) {
+    resume.skipped.push(key);
+    return 0;
+  }
+  const hardBefore = hardFailStorage.getStore()?.failures.length ?? 0;
+  const softBefore = softSkipStorage.getStore()?.count ?? 0;
+  const truncatedBefore = googleHealthOutcomeTruncated();
+  const imported = await fn();
+  const clean =
+    (hardFailStorage.getStore()?.failures.length ?? 0) === hardBefore &&
+    (softSkipStorage.getStore()?.count ?? 0) === softBefore &&
+    !truncatedBefore &&
+    !googleHealthOutcomeTruncated();
+  if (clean) {
+    resume.done.add(key);
+    await prisma.googleHealthConnection
+      .update({
+        where: { userId: resume.userId },
+        data: { backfillProgress: { done: [...resume.done] } },
+      })
+      .catch((err) =>
+        // Losing the note costs a re-walk of this collection next attempt,
+        // nothing more; it must not fail a collection that did complete.
+        getEvent()?.addWarning(
+          `google-health: backfill progress write failed: ${err}`,
+        ),
+      );
+  }
+  return imported;
+}
+
 /**
  * Single-source the per-resource collection-fetch error handling. A 403 on one
  * data class soft-skips it (warn + return 0) so sibling resources still sync; a
@@ -362,6 +519,14 @@ export async function handleCollectionFetchError(
   userId: string,
   err: unknown,
 ): Promise<number> {
+  if (err instanceof GoogleHealthTokenUnavailableError) {
+    // `getValidToken` already recorded the failure and failed the cycle;
+    // recording it again would count one dead token twice.
+    getEvent()?.addWarning(
+      `google-health ${resource} stopped for ${userId}: no valid access token`,
+    );
+    return 0;
+  }
   if (isCollectionForbidden(err)) {
     getEvent()?.addWarning(
       `google-health ${resource} sync skipped for ${userId}: collection 403 (soft-skip)`,
@@ -445,6 +610,51 @@ export interface GoogleHealthMeasurementUpsert {
   sleepStage?: "IN_BED" | "AWAKE" | "ASLEEP" | "REM" | "CORE" | "DEEP" | null;
 }
 
+/**
+ * Keep one reading per natural key — `(type, measuredAt, sleepStage)`, the
+ * second unique index on measurements — out of a batch.
+ *
+ * Google reports overlapping sleep segments: two sessions for one night, or
+ * two segments of one session with the same stage and the same end. Both
+ * readings carry their own externalId, but only one row can hold their
+ * natural key. Writing both was never possible; what happened instead
+ * depended on the order of the batch. A fresh pair lost one reading to
+ * `skipDuplicates` without a trace, and a re-score that moved one row onto
+ * the other's slot threw P2002 and failed the whole sync (#1195).
+ *
+ * The rule picks the same reading every time Google sends the same pair, so
+ * the row that holds the slot does not change hands from one sync to the
+ * next: the longer one (for a sleep segment, the span that covers the
+ * shorter one ending at the same instant), then the smaller externalId.
+ * Readings that share an externalId as well are the same reading fetched
+ * twice, and collapse the same way.
+ */
+export function collapseNaturalKeyTwins<
+  T extends GoogleHealthMeasurementUpsert,
+>(readings: readonly T[]): { kept: T[]; dropped: number } {
+  const byKey = new Map<string, T>();
+  for (const r of readings) {
+    const key = `${r.type}|${r.measuredAt.getTime()}|${r.sleepStage ?? ""}`;
+    const held = byKey.get(key);
+    if (
+      !held ||
+      r.value > held.value ||
+      (r.value === held.value && r.externalId < held.externalId)
+    ) {
+      byKey.set(key, r);
+    }
+  }
+  return {
+    kept: readings.filter(
+      (r) =>
+        byKey.get(
+          `${r.type}|${r.measuredAt.getTime()}|${r.sleepStage ?? ""}`,
+        ) === r,
+    ),
+    dropped: readings.length - byKey.size,
+  };
+}
+
 /** Chunk size for the batched `createMany` insert of fresh readings. */
 const GOOGLE_HEALTH_CREATE_CHUNK = 500;
 
@@ -454,7 +664,10 @@ export interface GoogleHealthSleepReplaceWindow {
   windowStart: Date | null;
   /** Latest segment end (UTC). */
   windowEnd: Date | null;
-  /** The fresh externalIds for THIS session — never soft-deleted. */
+  /**
+   * The fresh externalIds that must survive — every session's of the fetch,
+   * since sessions can overlap (see `replaceStaleGoogleHealthSleep`).
+   */
   keepIds: string[];
 }
 
@@ -466,11 +679,12 @@ export interface GoogleHealthSleepReplaceWindow {
  * night-total silently double-counted (a 7h35 night read as 10h+). For each
  * just-fetched session this soft-deletes any LIVE `GOOGLE_HEALTH`
  * `SLEEP_DURATION` row whose `measuredAt` falls inside the session's
- * `[windowStart, windowEnd]` but was NOT re-produced by this fetch (`keepIds`
- * are this session's fresh externalIds). Sleep sessions do not overlap in time,
- * so a scan bounded to one session's window only ever touches that session's own
- * rows — a fresh row is protected by `keepIds`, and any leftover (old volatile
- * key, or a segment Google dropped) is cleared. Rows OUTSIDE every returned
+ * `[windowStart, windowEnd]` but was NOT re-produced by this fetch. Sessions
+ * CAN overlap in time (Google reports two sessions over one stretch of a night
+ * from two sources), so the caller passes every fresh externalId of the fetch
+ * in `keepIds`, not only this session's: a window then never clears a row
+ * another session just re-produced, and any leftover (old volatile key, or a
+ * segment Google dropped) is cleared. Rows OUTSIDE every returned
  * window are never touched, so a night Google did not re-report this tick stays
  * intact — no data loss, only stale duplicates and re-score orphans go. This is
  * also the repair path: a full backfill re-fetches history and cleans each night
@@ -554,11 +768,15 @@ export async function upsertGoogleHealthMeasurements(
   // before they can become rows. A refused reading is dropped and tallied on
   // the ambient event, never clamped and never written for a later reader to
   // discover.
-  const readings = dropImplausibleMeasurements(
-    "googleHealth",
-    incoming,
-    (r) => ({ type: r.type, value: r.value }),
+  const { kept: readings, dropped: naturalKeyTwins } = collapseNaturalKeyTwins(
+    dropImplausibleMeasurements("googleHealth", incoming, (r) => ({
+      type: r.type,
+      value: r.value,
+    })),
   );
+  if (naturalKeyTwins > 0) {
+    annotate({ meta: { "googleHealth.naturalKeyTwins": naturalKeyTwins } });
+  }
   if (readings.length === 0) return { imported: 0, touched: [], inserted: [] };
 
   // Probe EVERY existing row (live AND tombstoned) for the batch's externalIds
@@ -712,6 +930,13 @@ export async function upsertGoogleHealthMeasurements(
   // value, the NEW externalId (the migration itself), `deletedAt: null`.
   if (toCreate.length > 0) {
     try {
+      // A row this batch already matched by externalId belongs to that
+      // reading. When the reading moves it (a re-scored segment end), the
+      // slot it leaves is free by the time the inserts run (updates go
+      // first, below), so a fresh reading for that slot is an insert, not a
+      // rescue: re-keying the moving row would hand it to two readings at
+      // once and drop one of them.
+      const claimedIds = new Set([...rowByKey.values()].map((row) => row.id));
       const rescued = new Set<number>();
       const naturalKeyOf = (
         type: string,
@@ -749,7 +974,7 @@ export async function upsertGoogleHealthMeasurements(
           const id = byNaturalKey.get(
             naturalKeyOf(c.type, c.measuredAt, c.sleepStage),
           );
-          if (id) {
+          if (id && !claimedIds.has(id)) {
             rescued.add(i + j);
             toUpdate.push({
               id,
@@ -789,6 +1014,121 @@ export async function upsertGoogleHealthMeasurements(
     measuredAt: Date;
   }> = [];
 
+  // Existing-row overwrites first, inserts second: a re-scored segment that
+  // moves its row frees the slot the row held, and a fresh reading for that
+  // slot can only land once the move has happened.
+  //
+  // Per-row update (differing values) on the matched id, so the re-fetched
+  // daily summary overwrites in place and bumps `syncVersion`.
+  // `deletedAt: null` rides along unconditionally — a no-op on a live row, a
+  // deliberate RESURRECTION on a tombstoned one (Google is the source of truth
+  // for its own rows; see TOMBSTONES RESURRECT above).
+  //
+  // A move can still land on a slot another row holds: Google reports
+  // overlapping sleep segments, and a re-score can put one segment's end on
+  // the end another segment of the same stage already has. The natural-key
+  // index refuses that write (P2002). It is not a failed sync: the row that
+  // holds the slot is a stale copy of the same reading, so the move is handed
+  // to `reconcileExternalMeasurement`, which retires the stale copy and lands
+  // the reading, the same way every other provider settles the two
+  // identities. Only a reconcile that itself fails holds the watermark.
+  const conflicts: typeof toUpdate = [];
+  for (const entry of toUpdate) {
+    const { id, r, reKeyTo, movedFrom } = entry;
+    try {
+      await prisma.measurement.update({
+        where: { id },
+        data: {
+          value: r.value,
+          unit: r.unit,
+          measuredAt: r.measuredAt,
+          sleepStage: r.sleepStage ?? null,
+          deletedAt: null,
+          // Key-format migration: adopt the fresh externalId in place.
+          ...(reKeyTo ? { externalId: reKeyTo } : {}),
+          // Surface the server-side mutation to the iOS LWW reconciler.
+          syncVersion: { increment: 1 },
+        },
+      });
+      touched.push({
+        type: r.type as MeasurementType,
+        measuredAt: r.measuredAt,
+      });
+      // A row that moved (a daily anchor re-cut on the user's local noon)
+      // leaves its old day too; that day's rollup needs the refold.
+      if (movedFrom) {
+        touched.push({
+          type: r.type as MeasurementType,
+          measuredAt: movedFrom,
+        });
+      }
+      imported++;
+    } catch (err) {
+      if (isP2002(err)) {
+        conflicts.push(entry);
+        continue;
+      }
+      getEvent()?.addWarning(
+        `google-health: failed to update measurement: ${err}`,
+      );
+      // Same rule as the create catch: hold the watermark so the overwrite is
+      // retried next tick rather than silently dropped.
+      noteHardFailure("measurements:update");
+    }
+  }
+
+  if (conflicts.length > 0) {
+    let resolved = 0;
+    for (const { r, movedFrom } of conflicts) {
+      const verdict = await prisma
+        .$transaction((tx) =>
+          reconcileExternalMeasurement(tx, {
+            userId,
+            type: r.type as MeasurementType,
+            source: "GOOGLE_HEALTH",
+            value: r.value,
+            unit: r.unit,
+            measuredAt: r.measuredAt,
+            externalId: r.externalId,
+            sleepStage: r.sleepStage ?? null,
+          }),
+        )
+        .catch((err): MeasurementReconciliationVerdict => ({
+          status: "failed",
+          error: { message: String(err) },
+        }));
+      if (verdict.status === "failed") {
+        getEvent()?.addWarning(
+          `google-health: failed to reconcile measurement: ${verdict.error.message}`,
+        );
+        noteHardFailure("measurements:update");
+        continue;
+      }
+      if (verdict.status === "rejected_range") continue;
+      resolved++;
+      imported++;
+      touched.push({
+        type: r.type as MeasurementType,
+        measuredAt: r.measuredAt,
+      });
+      if (movedFrom) {
+        touched.push({
+          type: r.type as MeasurementType,
+          measuredAt: movedFrom,
+        });
+      }
+      for (const dirty of verdict.dirtyIdentities ?? []) {
+        touched.push({ type: dirty.type, measuredAt: dirty.measuredAt });
+      }
+    }
+    annotate({
+      meta: {
+        "googleHealth.naturalKeyConflicts": conflicts.length,
+        "googleHealth.naturalKeyConflictsResolved": resolved,
+      },
+    });
+  }
+
   // Fresh inserts: chunked `createMany` (server-owned rows, field-by-field).
   // `skipDuplicates` guards the partial-unique index in the rare race where a
   // concurrent run inserted the same live key between the probe and the write.
@@ -831,50 +1171,6 @@ export async function upsertGoogleHealthMeasurements(
     }
   }
 
-  // Existing-row overwrites: per-row update (differing values) on the matched
-  // id, so the re-fetched daily summary overwrites in place and bumps
-  // `syncVersion`. `deletedAt: null` rides along unconditionally — a no-op on
-  // a live row, a deliberate RESURRECTION on a tombstoned one (Google is the
-  // source of truth for its own rows; see TOMBSTONES RESURRECT above).
-  for (const { id, r, reKeyTo, movedFrom } of toUpdate) {
-    try {
-      await prisma.measurement.update({
-        where: { id },
-        data: {
-          value: r.value,
-          unit: r.unit,
-          measuredAt: r.measuredAt,
-          sleepStage: r.sleepStage ?? null,
-          deletedAt: null,
-          // Key-format migration: adopt the fresh externalId in place.
-          ...(reKeyTo ? { externalId: reKeyTo } : {}),
-          // Surface the server-side mutation to the iOS LWW reconciler.
-          syncVersion: { increment: 1 },
-        },
-      });
-      touched.push({
-        type: r.type as MeasurementType,
-        measuredAt: r.measuredAt,
-      });
-      // A row that moved (a daily anchor re-cut on the user's local noon)
-      // leaves its old day too; that day's rollup needs the refold.
-      if (movedFrom) {
-        touched.push({
-          type: r.type as MeasurementType,
-          measuredAt: movedFrom,
-        });
-      }
-      imported++;
-    } catch (err) {
-      getEvent()?.addWarning(
-        `google-health: failed to update measurement: ${err}`,
-      );
-      // Same rule as the create catch: hold the watermark so the overwrite is
-      // retried next tick rather than silently dropped.
-      noteHardFailure("measurements:update");
-    }
-  }
-
   // On a `fullSync` backfill the caller defers the rollup fold: it collapses
   // every resource's touched type-days into a SINGLE range-recompute at the end
   // of the cycle. The incremental path keeps the inline per-day hook here.
@@ -919,14 +1215,21 @@ export async function upsertGoogleHealthMeasurements(
 }
 
 /**
- * Stamp `lastSyncedAt = now`. Called ONCE per cycle by `syncUserGoogleHealth`
- * after a non-degenerate run — never per resource, so the watermark can't move
- * mid-cycle and shrink a later resource's fetch window.
+ * Stamp `lastSyncedAt`. Called ONCE per cycle by `syncUserGoogleHealth` after
+ * a non-degenerate run — never per resource, so the watermark can't move
+ * mid-cycle and shrink a later resource's fetch window. The orchestrator
+ * passes the instant the cycle STARTED: a sample written to Google while the
+ * cycle was still walking is newer than what the walk read, and stamping the
+ * end would put the watermark past it. That gap is seconds for an hourly run
+ * and hours for a full-history backfill.
  */
-export async function markSynced(userId: string): Promise<void> {
+export async function markSynced(
+  userId: string,
+  at: Date = new Date(),
+): Promise<void> {
   await prisma.googleHealthConnection.update({
     where: { userId },
-    data: { lastSyncedAt: new Date() },
+    data: { lastSyncedAt: at },
   });
 }
 

src/lib/google-health/sync-lock.ts

@@ -0,0 +1,48 @@
+/**
+ * One Google Health run per account at a time, across processes.
+ *
+ * The hourly poll, the manual trigger, the full-history backfill and the
+ * sleep repair all read the same account's collections with the same grant.
+ * Run side by side they share Google's per-user, per-minute request quota:
+ * a backfill deep in a dense heart-rate walk and an hourly sync starting
+ * beside it pushed the account over the quota, and every request either of
+ * them made in that minute answered 429. They also write the same rows, so
+ * two runs racing one night's sleep segments can each move a row the other
+ * is about to move.
+ *
+ * So every run for one account holds the same advisory lock for as long as
+ * it reads and writes (`withJobLock`, a connection of its own, released when
+ * the run ends or the process dies). A run that finds it taken does no work:
+ * the hourly poll skips the account for this tick, the manual trigger answers
+ * that a sync is already running, and the backfill and the repair wait a
+ * bounded time for a short run to finish before handing their job back to
+ * pg-boss to retry.
+ */
+import { withJobLock, type GuardedRun } from "@/lib/jobs/job-lock";
+
+/** The lock key for one account's Google Health runs. */
+export function googleHealthSyncLockKey(userId: string): string {
+  return `google-health-sync:${userId}`;
+}
+
+/**
+ * How long a backfill or repair waits for an hourly or manual run to finish.
+ * Those runs read a day of history at most and take a minute or two, so ten
+ * minutes covers one that is also sitting out a rate limit.
+ */
+export const GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS = 10 * 60 * 1000;
+
+/**
+ * Run `run` while holding the account's Google Health lock. Resolves
+ * `{ ran: false }` without calling `run` when another run holds it and does
+ * not let go within `waitMs` (zero: give up at once).
+ */
+export function withGoogleHealthSyncLock<T>(
+  userId: string,
+  run: () => Promise<T>,
+  options: { waitMs?: number } = {},
+): Promise<GuardedRun<T>> {
+  return withJobLock(googleHealthSyncLockKey(userId), run, undefined, {
+    waitMs: options.waitMs ?? 0,
+  });
+}

src/lib/google-health/sync-metrics.ts

@@ -30,6 +30,7 @@ import {
   GOOGLE_HEALTH_DENSE_MAX_PAGES,
   type GoogleHealthDataType,
   type GoogleHealthMappedMeasurement,
+  type GoogleHealthTokenSource,
   fetchDataPoints,
   forEachDataPointPage,
   mapBloodGlucose,
@@ -46,7 +47,9 @@ import {
 } from "./client";
 import {
   getValidToken,
+  googleHealthTokenSource,
   handleCollectionFetchError,
+  runGoogleHealthCollection,
   upsertGoogleHealthMeasurements,
   type GoogleHealthMeasurementUpsert,
   type GoogleHealthResourceSyncOptions,
@@ -137,6 +140,7 @@ export async function syncUserMetrics(
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
+  const token = googleHealthTokenSource(userId, tokenInfo);
 
   // The incremental lower bound is the cycle-wide watermark snapshotted once by
   // `syncUserGoogleHealth` — never re-read here, so a sibling resource's stamp
@@ -167,38 +171,9 @@ export async function syncUserMetrics(
   // the failure stay written: every write is keyed, so the retry that the
   // failed verdict causes overwrites them in place.
   for (const resource of METRIC_RESOURCES) {
-    try {
-      await forEachDataPointPage(
-        resource.dataType,
-        tokenInfo.accessToken,
-        resource.verb,
-        {
-          start: resource.intraday ? (opts.intradayStart ?? start) : start,
-          maxPages: resource.maxPages,
-        },
-        async (points) => {
-          const readings: GoogleHealthMeasurementUpsert[] = [];
-          for (const point of points) {
-            for (const m of resource.map(point, tz)) {
-              readings.push({
-                type: m.type,
-                value: m.value,
-                unit: m.unit,
-                measuredAt: m.measuredAt,
-                externalId: m.fieldTag,
-              });
-            }
-          }
-          imported += (
-            await upsertGoogleHealthMeasurements(userId, readings, {
-              deferRollup: opts.deferRollup,
-            })
-          ).imported;
-        },
-      );
-    } catch (err) {
-      imported += await handleCollectionFetchError(resource.verb, userId, err);
-    }
+    imported += await runGoogleHealthCollection(resource.verb, () =>
+      syncMetricCollection(userId, resource, token, tz, opts),
+    );
   }
 
   // Height → User.heightCm, only when the user has no height yet. Never mint a
@@ -208,7 +183,7 @@ export async function syncUserMetrics(
     try {
       heightPoints = await fetchDataPoints(
         GOOGLE_HEALTH_DATA_TYPES.height,
-        tokenInfo.accessToken,
+        token,
         "fetchHeight",
         { start },
       );
@@ -255,3 +230,53 @@ export async function syncUserMetrics(
   // mid-cycle.
   return imported;
 }
+
+/**
+ * Walk one metric collection page by page and write each page as it arrives.
+ * A failure (fetch, map or write) goes through the collection ledger and
+ * stops only this collection; the siblings still run.
+ */
+async function syncMetricCollection(
+  userId: string,
+  resource: MetricResource,
+  token: GoogleHealthTokenSource,
+  tz: string,
+  opts: GoogleHealthResourceSyncOptions,
+): Promise<number> {
+  let imported = 0;
+  try {
+    await forEachDataPointPage(
+      resource.dataType,
+      token,
+      resource.verb,
+      {
+        start: resource.intraday
+          ? (opts.intradayStart ?? opts.start)
+          : opts.start,
+        maxPages: resource.maxPages,
+      },
+      async (points) => {
+        const readings: GoogleHealthMeasurementUpsert[] = [];
+        for (const point of points) {
+          for (const m of resource.map(point, tz)) {
+            readings.push({
+              type: m.type,
+              value: m.value,
+              unit: m.unit,
+              measuredAt: m.measuredAt,
+              externalId: m.fieldTag,
+            });
+          }
+        }
+        imported += (
+          await upsertGoogleHealthMeasurements(userId, readings, {
+            deferRollup: opts.deferRollup,
+          })
+        ).imported;
+      },
+    );
+  } catch (err) {
+    imported += await handleCollectionFetchError(resource.verb, userId, err);
+  }
+  return imported;
+}

src/lib/google-health/sync-sleep.ts

@@ -28,8 +28,10 @@ import {
 } from "./client";
 import {
   getValidToken,
+  googleHealthTokenSource,
   handleCollectionFetchError,
   noteHardFailure,
+  runGoogleHealthCollection,
   replaceStaleGoogleHealthSleep,
   upsertGoogleHealthMeasurements,
   type GoogleHealthMeasurementUpsert,
@@ -43,6 +45,15 @@ import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger"
 export async function syncUserSleep(
   userId: string,
   opts: GoogleHealthResourceSyncOptions = {},
+): Promise<number> {
+  return runGoogleHealthCollection("sleep", () =>
+    syncSleepSessions(userId, opts),
+  );
+}
+
+async function syncSleepSessions(
+  userId: string,
+  opts: GoogleHealthResourceSyncOptions,
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
@@ -60,7 +71,7 @@ export async function syncUserSleep(
   try {
     points = await fetchDataPoints(
       GOOGLE_HEALTH_DATA_TYPES.sleep,
-      tokenInfo.accessToken,
+      googleHealthTokenSource(userId, tokenInfo),
       "fetchSleep",
       { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE },
     );
@@ -116,7 +127,16 @@ export async function syncUserSleep(
     });
   }
 
-  await replaceStaleGoogleHealthSleep(userId, replaceWindows);
+  // Every session's fresh rows are protected in every window, not only in
+  // their own. Google can report two sessions over the same stretch of a
+  // night (two sources, or a nap inside the main sleep), and a window that
+  // protected only its own session's rows tombstoned the other session's rows
+  // on every sync, for the upsert to bring them straight back.
+  const allKeepIds = readings.map((r) => r.externalId);
+  await replaceStaleGoogleHealthSleep(
+    userId,
+    replaceWindows.map((w) => ({ ...w, keepIds: allKeepIds })),
+  );
 
   const { imported, inserted } = await upsertGoogleHealthMeasurements(
     userId,

src/lib/google-health/sync-workout.ts

@@ -25,8 +25,10 @@ import {
 } from "./client";
 import {
   getValidToken,
+  googleHealthTokenSource,
   handleCollectionFetchError,
   noteHardFailure,
+  runGoogleHealthCollection,
   type GoogleHealthResourceSyncOptions,
 } from "./sync-core";
 import { prisma } from "@/lib/db";
@@ -38,6 +40,15 @@ import { resolveUserTimezone } from "@/lib/tz/resolver";
 export async function syncUserWorkout(
   userId: string,
   opts: GoogleHealthResourceSyncOptions = {},
+): Promise<number> {
+  return runGoogleHealthCollection("fetchExercise", () =>
+    syncExerciseSessions(userId, opts),
+  );
+}
+
+async function syncExerciseSessions(
+  userId: string,
+  opts: GoogleHealthResourceSyncOptions,
 ): Promise<number> {
   const tokenInfo = await getValidToken(userId);
   if (!tokenInfo) return 0;
@@ -57,7 +68,7 @@ export async function syncUserWorkout(
   try {
     points = await fetchDataPoints(
       GOOGLE_HEALTH_DATA_TYPES.exercise,
-      tokenInfo.accessToken,
+      googleHealthTokenSource(userId, tokenInfo),
       "fetchExercise",
       { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE, tz },
     );

src/lib/google-health/sync.ts

@@ -1,7 +1,7 @@
 import pLimit from "p-limit";
 import { prisma } from "@/lib/db";
 import { annotate, getEvent } from "@/lib/logging/context";
-import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status";
+import { recordSyncSuccess } from "@/lib/integrations/status";
 import {
   collapseToTypeDayKeys,
   recomputeUserRollups,
@@ -17,8 +17,10 @@ import { syncUserActivity } from "./sync-activity";
 import { syncUserMetrics } from "./sync-metrics";
 import { syncUserSleep } from "./sync-sleep";
 import { syncUserWorkout } from "./sync-workout";
+import { withGoogleHealthSyncLock } from "./sync-lock";
 import {
   GOOGLE_HEALTH_INTEGRATION_KEY,
+  getValidToken,
   incrementalStart,
   intradayOverlapMs,
   markSynced,
@@ -40,6 +42,64 @@ export interface GoogleHealthSyncResult {
   imported: number;
   failed: boolean;
   resources?: GoogleHealthResourceOutcome[];
+  /**
+   * Set when nothing ran because another run for this account holds the
+   * account's sync lock (a backfill, the hourly poll, a manual trigger).
+   * `failed` is true alongside it so a caller that ignores the flag cannot
+   * mistake the skipped run for a clean one.
+   */
+  busy?: true;
+}
+
+/**
+ * Whether an account's Google Health connection is parked for a reason no
+ * retry can fix, so the run is not attempted.
+ *
+ * `parked` (a persistent failure older than a day) is one: it waits for the
+ * operator or the user to resume it. `error_reauth` is one only when the
+ * grant itself was refused: the token endpoint answered `invalid_grant` or
+ * 401, which `getValidToken` marks on the connection as `needsReauth`. The
+ * ledger also reaches `error_reauth` from a 401 on a data request, and that
+ * one says nothing certain about the grant: an access token that expired
+ * part-way through a long walk produced it, with a refresh token that was
+ * still valid, and the account stayed parked until it was reconnected by
+ * hand (#1194). Such a connection is not parked here; the next run refreshes
+ * the token first, a refresh that works clears the state when the run
+ * completes, and a refresh that is refused marks `needsReauth` and parks it
+ * properly.
+ */
+export async function isGoogleHealthParked(userId: string): Promise<boolean> {
+  const state = await ledgerState(userId);
+  if (state === "parked") return true;
+  if (state !== "error_reauth") return false;
+  const connection = await prisma.googleHealthConnection.findUnique({
+    where: { userId },
+    select: { needsReauth: true },
+  });
+  return connection?.needsReauth !== false;
+}
+
+async function ledgerState(userId: string): Promise<string | null> {
+  const row = await prisma.integrationStatus.findUnique({
+    where: {
+      userId_integration: {
+        userId,
+        integration: GOOGLE_HEALTH_INTEGRATION_KEY,
+      },
+    },
+    select: { state: true },
+  });
+  return row?.state ?? null;
+}
+
+/** Options for one `syncUserGoogleHealth` run. */
+export interface GoogleHealthSyncOptions {
+  fullSync?: boolean;
+  /**
+   * How long to wait for another run of this account to let go of the
+   * account's sync lock. Zero (the default) answers `busy` at once.
+   */
+  waitForLockMs?: number;
 }
 
 const RESOURCE_STATUSES = new Set<GoogleHealthResourceStatus>([
@@ -135,15 +195,39 @@ function terminalResourceOutcome(
  */
 export async function syncUserGoogleHealth(
   userId: string,
-  opts: { fullSync?: boolean } = {},
+  opts: GoogleHealthSyncOptions = {},
 ): Promise<GoogleHealthSyncResult> {
-  if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) {
+  if (await isGoogleHealthParked(userId)) {
     getEvent()?.addWarning(
       `google-health sync skipped for ${userId}: parked at error_reauth`,
     );
     return { state: "failed", imported: 0, failed: true, resources: [] };
   }
 
+  const run = await withGoogleHealthSyncLock(
+    userId,
+    () => runGoogleHealthSyncCycle(userId, opts),
+    { waitMs: opts.waitForLockMs ?? 0 },
+  );
+  if (!run.ran) {
+    annotate({ meta: { "googleHealth.sync.busy": true } });
+    return {
+      state: "in_progress",
+      imported: 0,
+      failed: true,
+      resources: [],
+      busy: true,
+    };
+  }
+  return run.result;
+}
+
+/** One sync cycle, run while holding the account's sync lock. */
+async function runGoogleHealthSyncCycle(
+  userId: string,
+  opts: GoogleHealthSyncOptions,
+): Promise<GoogleHealthSyncResult> {
+  const cycleStartedAt = new Date();
   const connection = await prisma.googleHealthConnection.findUnique({
     where: { userId },
     select: { lastSyncedAt: true },
@@ -152,6 +236,19 @@ export async function syncUserGoogleHealth(
     return { state: "failed", imported: 0, failed: true, resources: [] };
   }
 
+  // A connection held at `error_reauth` by a data request's 401 (see
+  // `isGoogleHealthParked`) gets a fresh token before anything else: the
+  // refresh is the one call that can tell an expired access token from a
+  // revoked grant. Refused, it records the reauth and the run stops here.
+  if ((await ledgerState(userId)) === "error_reauth") {
+    const probe = await runWithGoogleHealthSyncCycle(() =>
+      getValidToken(userId, { forceRefresh: true }),
+    );
+    if (!probe.result) {
+      return { state: "failed", imported: 0, failed: true, resources: [] };
+    }
+  }
+
   const progress = await startGoogleHealthSyncProgress(userId);
   const persistProgress = async (
     state: GoogleHealthSyncState,
@@ -287,7 +384,7 @@ export async function syncUserGoogleHealth(
   const failed = anyFailed || allSoftSkipped || truncated;
 
   if (!failed) {
-    await markSynced(userId);
+    await markSynced(userId, cycleStartedAt);
     await recordSyncSuccess(userId, GOOGLE_HEALTH_INTEGRATION_KEY);
   }
 

src/lib/jobs/__tests__/google-health-backfill.test.ts

@@ -7,14 +7,19 @@
  */
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-const { prismaMock, syncUserGoogleHealthMock, isReauthRequiredMock } =
-  vi.hoisted(() => ({
+const { prismaMock, syncUserGoogleHealthMock, isParkedMock } = vi.hoisted(
+  () => ({
     prismaMock: {
-      googleHealthConnection: { findMany: vi.fn(), update: vi.fn() },
+      googleHealthConnection: {
+        findMany: vi.fn(),
+        findUnique: vi.fn(async () => null),
+        update: vi.fn(),
+      },
     },
     syncUserGoogleHealthMock: vi.fn(),
-    isReauthRequiredMock: vi.fn(async () => false),
-  }));
+    isParkedMock: vi.fn(async () => false),
+  }),
+);
 
 vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
 
@@ -33,14 +38,17 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
   return { ...actual, safeFetch: safeFetchMock };
 });
 
-vi.mock("@/lib/integrations/status", () => ({
-  isReauthRequired: isReauthRequiredMock,
-}));
-
 vi.mock("@/lib/google-health/sync-core", () => ({
   GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
+  readGoogleHealthBackfillDone: () => new Set<string>(),
+  runWithGoogleHealthBackfillResume: async (
+    _userId: string,
+    _done: Set<string>,
+    fn: () => Promise<unknown>,
+  ) => ({ result: await fn(), skipped: [] }),
 }));
 vi.mock("@/lib/google-health/sync", () => ({
+  isGoogleHealthParked: isParkedMock,
   syncUserGoogleHealth: (...a: unknown[]) => syncUserGoogleHealthMock(...a),
 }));
 
@@ -53,7 +61,7 @@ import {
 
 beforeEach(() => {
   vi.clearAllMocks();
-  isReauthRequiredMock.mockResolvedValue(false);
+  isParkedMock.mockResolvedValue(false);
   prismaMock.googleHealthConnection.update.mockResolvedValue({});
 });
 
@@ -66,6 +74,7 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
     expect(imported).toBe(42);
     expect(syncUserGoogleHealthMock).toHaveBeenCalledWith("u1", {
       fullSync: true,
+      waitForLockMs: expect.any(Number),
     });
     const updateArg = prismaMock.googleHealthConnection.update.mock
       .calls[0]![0] as {
@@ -74,6 +83,21 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
     };
     expect(updateArg.where).toEqual({ userId: "u1" });
     expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date);
+    // The resume note goes with the stamp: a later backfill starts over.
+    expect(updateArg.data).toHaveProperty("backfillProgress");
+  });
+
+  it("throws without stamping when another sync of the account holds it", async () => {
+    syncUserGoogleHealthMock.mockResolvedValue({
+      imported: 0,
+      failed: true,
+      busy: true,
+    });
+
+    await expect(runGoogleHealthBackfillForUser("u1")).rejects.toThrow(
+      /another sync of the account is running/,
+    );
+    expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
   });
 
   it("a failed verdict THROWS without stamping — pg-boss retries become real", async () => {
@@ -86,7 +110,7 @@ describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
   });
 
   it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => {
-    isReauthRequiredMock.mockResolvedValue(true);
+    isParkedMock.mockResolvedValue(true);
 
     await expect(runGoogleHealthBackfillForUser("u1")).resolves.toEqual({
       imported: 0,

src/lib/jobs/__tests__/google-health-sleep-repair.test.ts

@@ -30,6 +30,20 @@ vi.mock("@/lib/google-health/sync-sleep", () => ({
   syncUserSleep: (...a: unknown[]) => syncUserSleep(...a),
 }));
 
+vi.mock("@/lib/google-health/sync", () => ({
+  isGoogleHealthParked: isReauthRequiredMock,
+}));
+
+// The account lock is a Postgres advisory lock on a connection of its own;
+// these cases run without a database, so the lock is always free.
+vi.mock("@/lib/google-health/sync-lock", () => ({
+  GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS: 0,
+  withGoogleHealthSyncLock: async (
+    _userId: string,
+    run: () => Promise<unknown>,
+  ) => ({ ran: true, result: await run() }),
+}));
+
 vi.mock("@/lib/logging/context", () => ({
   annotate: () => {},
   getEvent: () => null,

src/lib/jobs/google-health-backfill.ts

@@ -11,13 +11,20 @@
  * provisions it and the boot enqueue silently never drains (the v1.4.37
  * dead-queue class).
  */
+import { Prisma } from "@/generated/prisma/client";
 import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
-import { isReauthRequired } from "@/lib/integrations/status";
-import { GOOGLE_HEALTH_INTEGRATION_KEY } from "@/lib/google-health/sync-core";
-import { syncUserGoogleHealth } from "@/lib/google-health/sync";
+import {
+  readGoogleHealthBackfillDone,
+  runWithGoogleHealthBackfillResume,
+} from "@/lib/google-health/sync-core";
+import {
+  isGoogleHealthParked,
+  syncUserGoogleHealth,
+} from "@/lib/google-health/sync";
+import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock";
 import { runWithGoogleHealthStop } from "@/lib/google-health/client";
 
 export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
@@ -53,14 +60,24 @@ export interface GoogleHealthBackfillPayload {
  * heart-rate history runs to thousands of pages, and the walk asks it before
  * every page: once it says stop, the walk ends where it is, the cycle reports
  * itself truncated, and this throws like any other incomplete run, without
- * the stamp. The rows already written stay; the retry or the next boot walks
- * again and the key-stable upserts overwrite them.
+ * the stamp. The rows already written stay.
+ *
+ * Resumable: every collection the walk finishes cleanly is noted in
+ * `backfillProgress`, and the next attempt (the pg-boss retry, or the next
+ * boot's discovery) skips those and walks only what is left
+ * (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp
+ * lands, and on every reconnect.
+ *
+ * Never beside another run of the same account: the sync takes the account's
+ * lock, waiting a bounded time for an hourly or manual run to finish. If the
+ * lock does not come free the attempt throws without doing anything, and the
+ * retry tries again.
  */
 export async function runGoogleHealthBackfillForUser(
   userId: string,
   shouldStop: () => boolean = () => false,
 ): Promise<{ imported: number }> {
-  if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) {
+  if (await isGoogleHealthParked(userId)) {
     annotate({
       action: {
         name: "google_health.backfill.skipped_reauth",
@@ -70,9 +87,32 @@ export async function runGoogleHealthBackfillForUser(
     return { imported: 0 };
   }
 
-  const { imported, failed } = await runWithGoogleHealthStop(shouldStop, () =>
-    syncUserGoogleHealth(userId, { fullSync: true }),
+  const connection = await prisma.googleHealthConnection.findUnique({
+    where: { userId },
+    select: { backfillProgress: true },
+  });
+  const done = readGoogleHealthBackfillDone(connection?.backfillProgress);
+
+  const {
+    result: { imported, failed, busy },
+    skipped,
+  } = await runWithGoogleHealthBackfillResume(userId, done, () =>
+    runWithGoogleHealthStop(shouldStop, () =>
+      syncUserGoogleHealth(userId, {
+        fullSync: true,
+        waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS,
+      }),
+    ),
   );
+  if (skipped.length > 0) {
+    annotate({ meta: { "googleHealth.backfill.resumedPast": skipped } });
+  }
+
+  if (busy) {
+    throw new Error(
+      `google-health backfill for user ${userId} deferred: another sync of the account is running`,
+    );
+  }
 
   if (failed) {
     // Surface through the pg-boss retry path (retryLimit 3, backoff). If the
@@ -85,7 +125,7 @@ export async function runGoogleHealthBackfillForUser(
 
   await prisma.googleHealthConnection.update({
     where: { userId },
-    data: { backfillCompletedAt: new Date() },
+    data: { backfillCompletedAt: new Date(), backfillProgress: Prisma.DbNull },
   });
 
   annotate({

src/lib/jobs/google-health-sleep-repair.ts

@@ -27,13 +27,16 @@ import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission";
-import { isReauthRequired } from "@/lib/integrations/status";
 import {
-  GOOGLE_HEALTH_INTEGRATION_KEY,
   GOOGLE_HEALTH_TOKEN_HARD_FAIL,
   runWithGoogleHealthHardFailLedger,
 } from "@/lib/google-health/sync-core";
 import { syncUserSleep } from "@/lib/google-health/sync-sleep";
+import { isGoogleHealthParked } from "@/lib/google-health/sync";
+import {
+  GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS,
+  withGoogleHealthSyncLock,
+} from "@/lib/google-health/sync-lock";
 
 export const GOOGLE_HEALTH_SLEEP_REPAIR_QUEUE = "google-health-sleep-repair";
 
@@ -71,7 +74,7 @@ export interface GoogleHealthSleepRepairPayload {
 export async function runGoogleHealthSleepRepairForUser(
   userId: string,
 ): Promise<{ imported: number }> {
-  if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) {
+  if (await isGoogleHealthParked(userId)) {
     annotate({
       action: {
         name: "googleHealth.sleepRepair.skipped_reauth",
@@ -81,10 +84,22 @@ export async function runGoogleHealthSleepRepairForUser(
     return { imported: 0 };
   }
 
-  const { result: imported, failures } =
-    await runWithGoogleHealthHardFailLedger(() =>
-      syncUserSleep(userId, { deferRollup: false }),
+  // The repair re-reads the account's sleep, so it takes the account's lock
+  // like every other run (`withGoogleHealthSyncLock`).
+  const run = await withGoogleHealthSyncLock(
+    userId,
+    () =>
+      runWithGoogleHealthHardFailLedger(() =>
+        syncUserSleep(userId, { deferRollup: false }),
+      ),
+    { waitMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS },
+  );
+  if (!run.ran) {
+    throw new Error(
+      `google-health sleep repair for user ${userId} deferred: another sync of the account is running`,
     );
+  }
+  const { result: imported, failures } = run.result;
 
   if (failures.length > 0) {
     if (failures.every((f) => f === GOOGLE_HEALTH_TOKEN_HARD_FAIL)) {

src/lib/jobs/job-lock.ts

@@ -24,15 +24,31 @@ import { caughtAs } from "@/lib/logging/signal";
 /** What a guarded run came to. */
 export type GuardedRun<T> = { ran: true; result: T } | { ran: false };
 
+export interface JobLockOptions {
+  /**
+   * How long to keep asking for a lock another run holds before giving up.
+   * Zero (the default) gives up at once. A caller that would rather wait out
+   * a short run than drop its own work sets this; a run whose holder can take
+   * hours never should.
+   */
+  waitMs?: number;
+  /** The pause between two asks while waiting. */
+  pollMs?: number;
+}
+
+const DEFAULT_LOCK_POLL_MS = 2_000;
+
 /**
  * Run `run` while holding the advisory lock for `key`. Resolves
- * `{ ran: false }` at once, without calling `run`, when another run holds it.
+ * `{ ran: false }` without calling `run` when another run holds it, at once or,
+ * with `waitMs`, once that long has passed without the lock coming free.
  * Rejects exactly as `run` does otherwise.
  */
 export async function withJobLock<T>(
   key: string,
   run: () => Promise<T>,
   connectionString: string | undefined = process.env.DATABASE_URL,
+  options: JobLockOptions = {},
 ): Promise<GuardedRun<T>> {
   const client = new Client({ connectionString, keepAlive: true });
   // An idle client whose server goes away emits `error`; unhandled, that
@@ -41,11 +57,17 @@ export async function withJobLock<T>(
   client.on("error", () => {});
   await client.connect();
   try {
-    const { rows } = await client.query<{ held: boolean }>(
-      "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
-      [key],
-    );
-    if (!rows[0]?.held) return { ran: false };
+    const deadline = Date.now() + Math.max(0, options.waitMs ?? 0);
+    const pollMs = Math.max(1, options.pollMs ?? DEFAULT_LOCK_POLL_MS);
+    for (;;) {
+      const { rows } = await client.query<{ held: boolean }>(
+        "SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS held",
+        [key],
+      );
+      if (rows[0]?.held) break;
+      if (Date.now() + pollMs > deadline) return { ran: false };
+      await new Promise((resolve) => setTimeout(resolve, pollMs));
+    }
     try {
       return { ran: true, result: await run() };
     } finally {

src/lib/jobs/reminder/google-health-sync.ts

@@ -12,9 +12,9 @@ import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import {
   GOOGLE_HEALTH_POLL_CONCURRENCY,
+  isGoogleHealthParked,
   syncUserGoogleHealth,
 } from "@/lib/google-health/sync";
-import { isReauthRequired } from "@/lib/integrations/status";
 import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
 import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup";
 import {
@@ -64,16 +64,19 @@ export async function handleGoogleHealthSync(
       const verdicts = await Promise.all(
         targets.map(({ userId }) =>
           limit(async (): Promise<IntegrationUserVerdict> => {
-            if (await isReauthRequired(userId, "google-health")) {
+            if (await isGoogleHealthParked(userId)) {
               return { status: "parked", imported: 0 };
             }
 
             try {
               const result = await syncUserGoogleHealth(userId);
+              // Another run of this account (its backfill, a manual sync)
+              // holds the account; this tick leaves it to that run.
+              if (result.busy) return { status: "skipped", imported: 0 };
               const parked =
                 result.failed &&
                 result.imported === 0 &&
-                (await isReauthRequired(userId, "google-health"));
+                (await isGoogleHealthParked(userId));
               const status = parked
                 ? "parked"
                 : result.failed

src/lib/openapi/error-codes.ts

@@ -433,6 +433,7 @@ export const ERROR_CODE_CATALOGUE: Readonly<Record<string, readonly string[]>> =
       "private_origin_not_approved",
       "private_origin_not_grantable",
       "rate_limited_self",
+      "sync_in_progress",
       "too_many_rows",
       "unreadable_json",
       "upstream_invalid_json",

src/lib/openapi/routes/integrations.ts

@@ -1955,6 +1955,11 @@ export const integrationPaths: NonNullable<ZodOpenApiObject["paths"]> = {
             "The run failed and wrote nothing. A run that wrote something before failing answers 200 instead.",
           content: { "application/json": { schema: errorEnvelope } },
         },
+        "409": {
+          description:
+            "Another sync of this account is already running (the history backfill after a reconnect, or the hourly poll), so none was started. `meta.errorCode` = `sync_in_progress`. `GET /api/google-health/sync/status` reports the running one.",
+          content: { "application/json": { schema: errorEnvelope } },
+        },
         ...stdResponses,
         "422": {
           description: