fix(integrations): serialize refresh and cursor writes

src/lib/integrations/oauth-refresh.ts

@@ -1,3 +1,5 @@
+import type { Prisma } from "@/generated/prisma/client";
+
 /**
  * Shared OAuth rotating-refresh-token persistence helper.
  *
@@ -58,3 +60,32 @@ export async function persistRotatedToken(
   // rather than parking the connection at reauth.
   return opts.readPeerAccessToken();
 }
+
+export type SerializedRefreshProvider = "whoop" | "withings";
+
+/**
+ * Provider refreshes hold a database transaction open around one bounded
+ * external request. `maxWait` bounds pool acquisition and `timeout` bounds the
+ * advisory-lock wait, provider call, and durable token replacement together.
+ */
+export const PROVIDER_REFRESH_TRANSACTION_OPTIONS = {
+  maxWait: 10_000,
+  timeout: 60_000,
+} as const;
+
+/**
+ * Serialize one provider/user refresh attempt for the transaction lifetime.
+ * Callers must re-read token state after this returns: a prior contender may
+ * have committed a fresh token while this transaction waited for the lock.
+ */
+export async function acquireProviderTokenRefreshLock(
+  tx: Pick<Prisma.TransactionClient, "$queryRaw">,
+  provider: SerializedRefreshProvider,
+  userId: string,
+): Promise<void> {
+  const key = `provider-token-refresh:${provider}:${userId}`;
+  await tx.$queryRaw`
+    SELECT 1 AS locked
+    FROM pg_advisory_xact_lock(hashtextextended(${key}, 0))
+  `;
+}

src/lib/whoop/__tests__/sync.test.ts

@@ -31,6 +31,7 @@ const {
       findMany: vi.fn(),
     },
     $executeRaw: vi.fn(),
+    $queryRaw: vi.fn(),
     $transaction: vi.fn(),
   },
   refreshAccessTokenMock: vi.fn(),
@@ -113,8 +114,8 @@ import { syncUserRecovery } from "../sync-recovery";
 
 beforeEach(() => {
   vi.clearAllMocks();
-  prismaMock.$transaction.mockImplementation(async (run: (tx: unknown) => unknown) =>
-    run({}),
+  prismaMock.$transaction.mockImplementation(
+    async (run: (tx: unknown) => unknown) => run(prismaMock),
   );
   reconcileMock.mockImplementation(
     async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
@@ -150,7 +151,7 @@ describe("getValidToken — rotating refresh", () => {
       refresh_token: "new-refresh",
       expires_in: 3600,
     });
-    prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 1 });
+    prismaMock.whoopConnection.update.mockResolvedValue({});
 
     const result = await getValidToken("user1");
 
@@ -159,15 +160,15 @@ describe("getValidToken — rotating refresh", () => {
       clientId: "cid",
       clientSecret: "csecret",
     });
-    const updateArg = prismaMock.whoopConnection.updateMany.mock.calls[0]![0];
-    // CAS guard: scoped to the connection AND the exact stored ciphertext spent.
+    const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0];
     expect(updateArg.where.id).toBe("conn1");
-    expect(updateArg.where.refreshToken).toBe("enc(old-refresh)");
     expect(updateArg.data.accessToken).toBe("enc(new-access)");
     expect(updateArg.data.refreshToken).toBe("enc(new-refresh)");
+    expect(prismaMock.$queryRaw).toHaveBeenCalledTimes(1);
   });
 
-  it("reuses the peer's rotated token on a lost CAS race (no spurious reauth)", async () => {
+
+  it("re-reads token state under an advisory lock and reuses a peer winner", async () => {
     prismaMock.whoopConnection.findUnique
       .mockResolvedValueOnce({
         id: "conn1",
@@ -176,20 +177,26 @@ describe("getValidToken — rotating refresh", () => {
         refreshToken: "enc(old-refresh)",
         tokenExpiresAt: new Date(Date.now() - 1000),
       })
-      // Re-read after the lost race surfaces the peer's freshly rotated access.
-      .mockResolvedValueOnce({ accessToken: "enc(peer-access)" });
-    refreshAccessTokenMock.mockResolvedValue({
-      access_token: "new-access",
-      refresh_token: "new-refresh",
-      expires_in: 3600,
-    });
-    // A concurrent sync already rotated the token → zero rows match the guard.
-    prismaMock.whoopConnection.updateMany.mockResolvedValue({ count: 0 });
+      .mockResolvedValueOnce({
+        id: "conn1",
+        whoopUserId: "42",
+        accessToken: "enc(winner-access)",
+        refreshToken: "enc(winner-refresh)",
+        tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+      });
 
     const result = await getValidToken("user1");
 
-    expect(result?.accessToken).toBe("peer-access");
-    expect(recordSyncFailure).not.toHaveBeenCalled();
+    expect(result?.accessToken).toBe("winner-access");
+    expect(prismaMock.$queryRaw).toHaveBeenCalledTimes(1);
+    expect(refreshAccessTokenMock).not.toHaveBeenCalled();
+    expect(prismaMock.$transaction).toHaveBeenCalledWith(
+      expect.any(Function),
+      expect.objectContaining({
+        maxWait: expect.any(Number),
+        timeout: expect.any(Number),
+      }),
+    );
   });
 
   it("returns the stored token without refresh when not near expiry", async () => {
@@ -418,6 +425,13 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {
       source: "WHOOP",
       externalId: readings[1].externalId,
     });
+    expect(prismaMock.$transaction).toHaveBeenCalledWith(
+      expect.any(Function),
+      expect.objectContaining({
+        maxWait: expect.any(Number),
+        timeout: expect.any(Number),
+      }),
+    );
   });
 
   it("does not pre-probe and still emits a genuine insert", async () => {
@@ -459,6 +473,99 @@ describe("upsertWhoopMeasurements — exact insertion results", () => {
   });
 });
 
+describe("syncUserRecovery — durable cursor ordering", () => {
+  const recovery = {
+    cycle_id: 1,
+    sleep_id: "sleep-uuid",
+    user_id: 42,
+    created_at: "2026-06-01T06:00:00.000Z",
+    updated_at: "2026-06-01T07:00:00.000Z",
+    score_state: "SCORED",
+    score: {
+      user_calibrating: false,
+      recovery_score: 66,
+      resting_heart_rate: 52,
+      hrv_rmssd_milli: 48.7,
+      spo2_percentage: 97,
+      skin_temp_celsius: 33.4,
+    },
+  };
+
+  beforeEach(() => {
+    prismaMock.whoopConnection.findUnique.mockResolvedValue({
+      id: "conn1",
+      whoopUserId: "42",
+      accessToken: "enc(live-access)",
+      refreshToken: "enc(live-refresh)",
+      tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+      lastSyncedAt: new Date("2026-06-01T00:00:00Z"),
+      resourceCursors: {
+        recovery: "2026-06-01T00:00:00.000Z",
+      },
+    });
+    fetchRecoveriesMock.mockResolvedValue([recovery]);
+  });
+
+  it("keeps the cursor unchanged when any represented measurement write fails", async () => {
+    reconcileMock
+      .mockResolvedValueOnce({
+        status: "inserted",
+        row: {
+          id: "inserted:first",
+          type: "RECOVERY_SCORE",
+          measuredAt: new Date(recovery.updated_at),
+          externalId: "sleep-uuid:recovery_score",
+        },
+      })
+      .mockResolvedValueOnce({
+        status: "failed",
+        reason: "db_error",
+        error: new Error("injected write failure"),
+      });
+
+    await expect(syncUserRecovery("user1")).rejects.toThrow();
+
+    expect(prismaMock.$executeRaw).not.toHaveBeenCalled();
+  });
+
+  it("advances after success and replay updates identities without new arrivals", async () => {
+    expect(await syncUserRecovery("user1")).toBe(5);
+    expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1);
+
+    vi.clearAllMocks();
+    prismaMock.$transaction.mockImplementation(
+      async (run: (tx: unknown) => unknown) => run(prismaMock),
+    );
+    prismaMock.whoopConnection.findUnique.mockResolvedValue({
+      id: "conn1",
+      whoopUserId: "42",
+      accessToken: "enc(live-access)",
+      refreshToken: "enc(live-refresh)",
+      tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+      lastSyncedAt: new Date("2026-06-01T00:00:00Z"),
+      resourceCursors: {
+        recovery: "2026-06-01T00:00:00.000Z",
+      },
+    });
+    fetchRecoveriesMock.mockResolvedValue([recovery]);
+    reconcileMock.mockImplementation(
+      async (_tx: unknown, input: { type: string; measuredAt: Date; externalId: string }) => ({
+        status: "updated",
+        row: {
+          id: `existing:${input.externalId}`,
+          type: input.type,
+          measuredAt: input.measuredAt,
+          externalId: input.externalId,
+        },
+      }),
+    );
+
+    expect(await syncUserRecovery("user1")).toBe(5);
+    expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1);
+    expect(emitArrivalMock).toHaveBeenCalledWith("user1", [], "whoop");
+  });
+});
+
 describe("isCollectionForbidden — tier degradation gate", () => {
   it("is true only for a WhoopApiError carrying HTTP 403", () => {
     const forbidden = new WhoopApiError({

src/lib/whoop/sync.ts

@@ -6,9 +6,9 @@
  *   - `getValidToken` decrypts the stored pair, refreshes at
  *     `tokenExpiresAt - 5 min`, and persists BOTH rotated tokens (WHOOP
  *     invalidates the prior access AND refresh token on every refresh — the
- *     same discipline Withings uses for its rotating refresh token). The persist
- *     is a compare-and-swap on the stored ciphertext so two overlapping syncs
- *     cannot park the connection at reauth (see `persistRotatedToken`).
+ *     same discipline Withings uses for its rotating refresh token). A
+ *     provider/user advisory lock serializes refresh, and token state is
+ *     re-read inside the transaction before the one-time token is spent.
  *   - Each per-resource sync (`sync-recovery` / `sync-sleep` / `sync-cycle` /
  *     `sync-workout`) upserts into `Measurement` / `Workout` keyed on
  *     `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP
@@ -44,7 +44,10 @@ import {
   emitInsertedMeasurementArrivals,
   type InsertedMeasurementArrivalRow,
 } from "@/lib/arrivals/measurement-emit";
-import { persistRotatedToken } from "@/lib/integrations/oauth-refresh";
+import {
+  acquireProviderTokenRefreshLock,
+  PROVIDER_REFRESH_TRANSACTION_OPTIONS,
+} from "@/lib/integrations/oauth-refresh";
 import {
   MeasurementReconciliationError,
   reconcileExternalMeasurement,
@@ -202,91 +205,86 @@ export async function getValidToken(
   });
   if (!connection) return null;
 
-  const accessToken = decrypt(connection.accessToken);
-  const refreshToken = decrypt(connection.refreshToken);
-
   if (
-    connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS <
+    connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >=
     Date.now()
   ) {
-    try {
-      const creds = await getUserWhoopCredentials(userId);
-      if (!creds) {
-        getEvent()?.addWarning(
-          `No WHOOP credentials found for user ${userId} during token refresh`,
-        );
-        await recordSyncFailure({
-          userId,
-          integration: "whoop",
-          kind: "reauth_required",
-          message: "WHOOP credentials missing — token refresh skipped",
-          errorCode: "credentials_missing",
-        });
-        return null;
-      }
-
-      const newTokens = await refreshAccessToken(refreshToken, creds);
-      const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
-
-      // WHOOP rotates the refresh token on every refresh. Persist BOTH the new
-      // access token AND the new refresh token with a compare-and-swap on the
-      // stored ciphertext: a concurrent sync that already rotated the token is
-      // not clobbered, and THIS caller — if it lost the race — reuses the peer's
-      // rotated access token instead of dropping the connection to reauth.
-      const persistedAccessToken = await persistRotatedToken(
-        newTokens.access_token,
-        {
-          conditionalUpdate: async () => {
-            const { count } = await prisma.whoopConnection.updateMany({
-              // `connection.refreshToken` is the exact ciphertext we read +
-              // spent; matching it is the CAS guard (a re-encrypt would differ).
-              where: {
-                id: connection.id,
-                refreshToken: connection.refreshToken,
-              },
-              data: {
-                accessToken: encrypt(newTokens.access_token),
-                refreshToken: encrypt(newTokens.refresh_token),
-                tokenExpiresAt: expiresAt,
-              },
-            });
-            return count;
-          },
-          readPeerAccessToken: async () => {
-            const fresh = await prisma.whoopConnection.findUnique({
-              where: { id: connection.id },
-              select: { accessToken: true },
-            });
-            return fresh ? decrypt(fresh.accessToken) : null;
-          },
-        },
-      );
-
-      if (!persistedAccessToken) return null;
-
-      return {
-        accessToken: persistedAccessToken,
-        connection: {
-          id: connection.id,
-          whoopUserId: connection.whoopUserId,
-        },
-      };
-    } catch (err) {
-      getEvent()?.addWarning(
-        `WHOOP token refresh failed for user ${userId}: ${err}`,
-      );
-      await recordWhoopSyncFailure(userId, err);
-      return null;
-    }
+    return {
+      accessToken: decrypt(connection.accessToken),
+      connection: {
+        id: connection.id,
+        whoopUserId: connection.whoopUserId,
+      },
+    };
   }
 
-  return {
-    accessToken,
-    connection: {
-      id: connection.id,
-      whoopUserId: connection.whoopUserId,
-    },
-  };
+  const creds = await getUserWhoopCredentials(userId);
+  if (!creds) {
+    getEvent()?.addWarning(
+      `No WHOOP credentials found for user ${userId} during token refresh`,
+    );
+    await recordSyncFailure({
+      userId,
+      integration: "whoop",
+      kind: "reauth_required",
+      message: "WHOOP credentials missing — token refresh skipped",
+      errorCode: "credentials_missing",
+    });
+    return null;
+  }
+
+  try {
+    return await prisma.$transaction(async (tx) => {
+      await acquireProviderTokenRefreshLock(tx, "whoop", userId);
+
+      const current = await tx.whoopConnection.findUnique({
+        where: { userId },
+      });
+      if (!current) return null;
+
+      if (
+        current.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >=
+        Date.now()
+      ) {
+        return {
+          accessToken: decrypt(current.accessToken),
+          connection: {
+            id: current.id,
+            whoopUserId: current.whoopUserId,
+          },
+        };
+      }
+
+      const newTokens = await refreshAccessToken(
+        decrypt(current.refreshToken),
+        creds,
+      );
+      const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
+
+      await tx.whoopConnection.update({
+        where: { id: current.id },
+        data: {
+          accessToken: encrypt(newTokens.access_token),
+          refreshToken: encrypt(newTokens.refresh_token),
+          tokenExpiresAt: expiresAt,
+        },
+      });
+
+      return {
+        accessToken: newTokens.access_token,
+        connection: {
+          id: current.id,
+          whoopUserId: current.whoopUserId,
+        },
+      };
+    }, PROVIDER_REFRESH_TRANSACTION_OPTIONS);
+  } catch (err) {
+    getEvent()?.addWarning(
+      `WHOOP token refresh failed for user ${userId}: ${err}`,
+    );
+    await recordWhoopSyncFailure(userId, err);
+    return null;
+  }
 }
 
 /**
@@ -350,30 +348,33 @@ export async function upsertWhoopMeasurements(
     InsertedMeasurementArrivalRow & { externalId: string | null }
   > = [];
 
-  const verdicts = await prisma.$transaction(async (tx) => {
-    const outcomes = [];
-    for (const r of readings) {
-      const verdict = await reconcileExternalMeasurement(
-        tx,
-        {
-          userId,
-          type: r.type as MeasurementType,
-          source: "WHOOP",
-          value: r.value,
-          unit: r.unit,
-          measuredAt: r.measuredAt,
-          externalId: r.externalId,
-          sleepStage: r.sleepStage ?? null,
-        },
-        { exactExternalMatch: "update" },
-      );
-      if (verdict.status === "failed") {
-        throw new MeasurementReconciliationError(verdict);
+  const verdicts = await prisma.$transaction(
+    async (tx) => {
+      const outcomes = [];
+      for (const r of readings) {
+        const verdict = await reconcileExternalMeasurement(
+          tx,
+          {
+            userId,
+            type: r.type as MeasurementType,
+            source: "WHOOP",
+            value: r.value,
+            unit: r.unit,
+            measuredAt: r.measuredAt,
+            externalId: r.externalId,
+            sleepStage: r.sleepStage ?? null,
+          },
+          { exactExternalMatch: "update" },
+        );
+        if (verdict.status === "failed") {
+          throw new MeasurementReconciliationError(verdict);
+        }
+        outcomes.push(verdict);
       }
-      outcomes.push(verdict);
-    }
-    return outcomes;
-  }, { timeout: 60_000 });
+      return outcomes;
+    },
+    { maxWait: 10_000, timeout: 60_000 },
+  );
 
   for (let index = 0; index < readings.length; index++) {
     const reading = readings[index]!;

src/lib/withings/__tests__/sync-measure-watermark.test.ts

@@ -24,6 +24,8 @@ vi.mock("@/lib/db", () => ({
       findUnique: vi.fn(),
       update: vi.fn(),
     },
+    $queryRaw: vi.fn(),
+    $transaction: vi.fn(),
   },
 }));
 
@@ -38,6 +40,13 @@ vi.mock("../client", () => ({
   subscribeWebhook: vi.fn(),
 }));
 
+vi.mock("../credentials", () => ({
+  getUserWithingsCredentials: vi.fn(async () => ({
+    clientId: "cid",
+    clientSecret: "secret",
+  })),
+}));
+
 vi.mock("@/lib/integrations/status", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/integrations/status")
@@ -77,13 +86,13 @@ vi.mock("@/lib/logging/context", () => ({
 }));
 
 import { prisma } from "@/lib/db";
-import { fetchMeasurements } from "../client";
+import { fetchMeasurements, refreshAccessToken } from "../client";
 import {
   recordSyncFailure,
   recordSyncSuccess,
 } from "@/lib/integrations/status";
 
-import { syncUserMeasurements } from "../sync";
+import { getValidToken, syncUserMeasurements } from "../sync";
 
 const P2002 = Object.assign(new Error("Unique constraint failed"), {
   code: "P2002",
@@ -91,6 +100,9 @@ const P2002 = Object.assign(new Error("Unique constraint failed"), {
 
 beforeEach(() => {
   vi.clearAllMocks();
+  vi.mocked(prisma.$transaction).mockImplementation(async (run) =>
+    run(prisma as never),
+  );
   // A live, non-expired connection so `getValidToken` returns the token
   // without touching the refresh path.
   vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({
@@ -107,6 +119,39 @@ beforeEach(() => {
   vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
 });
 
+describe("getValidToken — serialized rotating refresh", () => {
+  it("re-reads token state under an advisory lock and reuses a peer winner", async () => {
+    vi.mocked(prisma.withingsConnection.findUnique)
+      .mockResolvedValueOnce({
+        id: "conn-1",
+        withingsUserId: "wu-1",
+        accessToken: "old-access",
+        refreshToken: "old-refresh",
+        tokenExpiresAt: new Date(Date.now() - 1000),
+      } as never)
+      .mockResolvedValueOnce({
+        id: "conn-1",
+        withingsUserId: "wu-1",
+        accessToken: "winner-access",
+        refreshToken: "winner-refresh",
+        tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+      } as never);
+
+    const result = await getValidToken("user-1");
+
+    expect(result?.accessToken).toBe("winner-access");
+    expect(prisma.$queryRaw).toHaveBeenCalledTimes(1);
+    expect(refreshAccessToken).not.toHaveBeenCalled();
+    expect(prisma.$transaction).toHaveBeenCalledWith(
+      expect.any(Function),
+      expect.objectContaining({
+        maxWait: expect.any(Number),
+        timeout: expect.any(Number),
+      }),
+    );
+  });
+});
+
 describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () => {
   it("holds lastSyncedAt and records a failure when a row hard-fails", async () => {
     vi.mocked(fetchMeasurements).mockResolvedValue([

src/lib/withings/sync.ts

@@ -32,6 +32,10 @@ import {
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import { invalidateUserDashboardSnapshot } from "@/lib/cache/invalidate";
+import {
+  acquireProviderTokenRefreshLock,
+  PROVIDER_REFRESH_TRANSACTION_OPTIONS,
+} from "@/lib/integrations/oauth-refresh";
 
 /**
  * Build the callback URL handed to Withings at `Notify.subscribe` time.
@@ -74,35 +78,58 @@ export async function getValidToken(userId: string): Promise<{
 
   if (!connection) return null;
 
-  const accessToken = decrypt(connection.accessToken);
-  const refreshToken = decrypt(connection.refreshToken);
+  if (connection.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) {
+    return {
+      accessToken: decrypt(connection.accessToken),
+      connection: {
+        id: connection.id,
+        withingsUserId: connection.withingsUserId,
+      },
+    };
+  }
 
-  // Check if token is expired (with 5 min buffer)
-  if (connection.tokenExpiresAt.getTime() - 5 * 60 * 1000 < Date.now()) {
-    try {
-      const creds = await getUserWithingsCredentials(userId);
-      if (!creds) {
-        getEvent()?.addWarning(
-          `No credentials found for user ${userId} during token refresh`,
-        );
-        // Without credentials we cannot refresh OR re-authenticate — the
-        // user has to re-enter the client_id/secret. Mark as reauth so
-        // scheduled syncs back off.
-        await recordSyncFailure({
-          userId,
-          integration: "withings",
-          kind: "reauth_required",
-          message: "Withings credentials missing — token refresh skipped",
-          errorCode: "credentials_missing",
-        });
-        return null;
+  const creds = await getUserWithingsCredentials(userId);
+  if (!creds) {
+    getEvent()?.addWarning(
+      `No credentials found for user ${userId} during token refresh`,
+    );
+    await recordSyncFailure({
+      userId,
+      integration: "withings",
+      kind: "reauth_required",
+      message: "Withings credentials missing — token refresh skipped",
+      errorCode: "credentials_missing",
+    });
+    return null;
+  }
+
+  try {
+    return await prisma.$transaction(async (tx) => {
+      await acquireProviderTokenRefreshLock(tx, "withings", userId);
+
+      const current = await tx.withingsConnection.findUnique({
+        where: { userId },
+      });
+      if (!current) return null;
+
+      if (current.tokenExpiresAt.getTime() - 5 * 60 * 1000 >= Date.now()) {
+        return {
+          accessToken: decrypt(current.accessToken),
+          connection: {
+            id: current.id,
+            withingsUserId: current.withingsUserId,
+          },
+        };
       }
 
-      const newTokens = await refreshAccessToken(refreshToken, creds);
-
+      const newTokens = await refreshAccessToken(
+        decrypt(current.refreshToken),
+        creds,
+      );
       const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000);
-      await prisma.withingsConnection.update({
-        where: { id: connection.id },
+
+      await tx.withingsConnection.update({
+        where: { id: current.id },
         data: {
           accessToken: encrypt(newTokens.access_token),
           refreshToken: encrypt(newTokens.refresh_token),
@@ -113,24 +140,16 @@ export async function getValidToken(userId: string): Promise<{
       return {
         accessToken: newTokens.access_token,
         connection: {
-          id: connection.id,
-          withingsUserId: connection.withingsUserId,
+          id: current.id,
+          withingsUserId: current.withingsUserId,
         },
       };
-    } catch (err) {
-      getEvent()?.addWarning(`Token refresh failed for user ${userId}: ${err}`);
-      await recordWithingsSyncFailure(userId, err);
-      return null;
-    }
+    }, PROVIDER_REFRESH_TRANSACTION_OPTIONS);
+  } catch (err) {
+    getEvent()?.addWarning(`Token refresh failed for user ${userId}: ${err}`);
+    await recordWithingsSyncFailure(userId, err);
+    return null;
   }
-
-  return {
-    accessToken,
-    connection: {
-      id: connection.id,
-      withingsUserId: connection.withingsUserId,
-    },
-  };
 }
 
 /**

tests/integration/provider-token-refresh-serialization.test.ts

@@ -0,0 +1,220 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type * as WhoopClientModule from "@/lib/whoop/client";
+import type * as WithingsClientModule from "@/lib/withings/client";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+process.env.ENCRYPTION_KEY =
+  "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+delete process.env.ENCRYPTION_KEYS;
+delete process.env.ENCRYPTION_ACTIVE_KEY_ID;
+
+const { whoopRefresh, withingsRefresh, whoopRecoveries } = vi.hoisted(() => ({
+  whoopRefresh: vi.fn(),
+  withingsRefresh: vi.fn(),
+  whoopRecoveries: vi.fn(),
+}));
+
+vi.mock("@/lib/whoop/client", async (importOriginal) => ({
+  ...(await importOriginal<typeof WhoopClientModule>()),
+  refreshAccessToken: whoopRefresh,
+  fetchRecoveries: whoopRecoveries,
+}));
+
+vi.mock("@/lib/withings/client", async (importOriginal) => ({
+  ...(await importOriginal<typeof WithingsClientModule>()),
+  refreshAccessToken: withingsRefresh,
+}));
+
+import { encrypt, decrypt } from "@/lib/crypto";
+import { getValidToken as getValidWhoopToken } from "@/lib/whoop/sync";
+import { syncUserRecovery } from "@/lib/whoop/sync-recovery";
+import { getValidToken as getValidWithingsToken } from "@/lib/withings/sync";
+
+const CONTENDERS = 75;
+const EXPIRED_AT = new Date("2026-01-01T00:00:00.000Z");
+
+async function seedProviderUser(provider: "whoop" | "withings") {
+  const prisma = getPrismaClient();
+  const userId = `refresh-${provider}`;
+  await prisma.user.create({
+    data: {
+      id: userId,
+      username: `refresh-${provider}`,
+      email: `refresh-${provider}@example.test`,
+      ...(provider === "whoop"
+        ? {
+            whoopClientIdEncrypted: encrypt("whoop-client"),
+            whoopClientSecretEncrypted: encrypt("whoop-secret"),
+          }
+        : {
+            withingsClientIdEncrypted: encrypt("withings-client"),
+            withingsClientSecretEncrypted: encrypt("withings-secret"),
+          }),
+    },
+  });
+
+  if (provider === "whoop") {
+    await prisma.whoopConnection.create({
+      data: {
+        userId,
+        whoopUserId: "whoop-user",
+        accessToken: encrypt("stale-access"),
+        refreshToken: encrypt("stale-refresh"),
+        tokenExpiresAt: EXPIRED_AT,
+      },
+    });
+  } else {
+    await prisma.withingsConnection.create({
+      data: {
+        userId,
+        withingsUserId: "withings-user",
+        accessToken: encrypt("stale-access"),
+        refreshToken: encrypt("stale-refresh"),
+        tokenExpiresAt: EXPIRED_AT,
+      },
+    });
+  }
+
+  return userId;
+}
+
+async function runContenders(
+  provider: "whoop" | "withings",
+  userId: string,
+) {
+  const getToken =
+    provider === "whoop" ? getValidWhoopToken : getValidWithingsToken;
+  return Promise.all(
+    Array.from({ length: CONTENDERS }, () => getToken(userId)),
+  );
+}
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  vi.clearAllMocks();
+});
+
+describe.each([
+  ["whoop", whoopRefresh],
+  ["withings", withingsRefresh],
+] as const)("%s rotating token refresh serialization (real Postgres)", (provider, refresh) => {
+  it("lets exactly one contender refresh a stale token and shares the winner", async () => {
+    const userId = await seedProviderUser(provider);
+    refresh.mockImplementation(async (refreshToken: string) => {
+      expect(refreshToken).toBe("stale-refresh");
+      return {
+        access_token: "winner-access",
+        refresh_token: "winner-refresh",
+        expires_in: 3600,
+      };
+    });
+
+    const results = await runContenders(provider, userId);
+
+    expect(refresh).toHaveBeenCalledTimes(1);
+    expect(results).toHaveLength(CONTENDERS);
+    expect(results.every((result) => result?.accessToken === "winner-access")).toBe(
+      true,
+    );
+
+    const prisma = getPrismaClient();
+    const row =
+      provider === "whoop"
+        ? await prisma.whoopConnection.findUniqueOrThrow({ where: { userId } })
+        : await prisma.withingsConnection.findUniqueOrThrow({ where: { userId } });
+    expect(decrypt(row.accessToken)).toBe("winner-access");
+    expect(decrypt(row.refreshToken)).toBe("winner-refresh");
+  });
+
+  it("rolls back a failed winner so a later contender can retry", async () => {
+    const userId = await seedProviderUser(provider);
+    refresh
+      .mockRejectedValueOnce(new Error("injected refresh failure"))
+      .mockResolvedValue({
+        access_token: "retry-access",
+        refresh_token: "retry-refresh",
+        expires_in: 3600,
+      });
+
+    const results = await runContenders(provider, userId);
+
+    expect(refresh).toHaveBeenCalledTimes(2);
+    expect(results.filter((result) => result === null)).toHaveLength(1);
+    expect(
+      results.filter((result) => result?.accessToken === "retry-access"),
+    ).toHaveLength(CONTENDERS - 1);
+
+    const prisma = getPrismaClient();
+    const row =
+      provider === "whoop"
+        ? await prisma.whoopConnection.findUniqueOrThrow({ where: { userId } })
+        : await prisma.withingsConnection.findUniqueOrThrow({ where: { userId } });
+    expect(decrypt(row.accessToken)).toBe("retry-access");
+    expect(decrypt(row.refreshToken)).toBe("retry-refresh");
+  });
+});
+
+describe("WHOOP recovery cursor durability (real Postgres)", () => {
+  const oldCursor = "2026-06-01T00:00:00.000Z";
+  const recovery = {
+    cycle_id: 1,
+    sleep_id: "sleep-real",
+    user_id: 42,
+    created_at: "2026-06-01T06:00:00.000Z",
+    updated_at: "2026-06-01T07:00:00.000Z",
+    score_state: "SCORED",
+    score: {
+      user_calibrating: false,
+      recovery_score: 66,
+      resting_heart_rate: 52,
+      hrv_rmssd_milli: 48.7,
+      spo2_percentage: 97,
+      skin_temp_celsius: 33.4,
+    },
+  };
+
+  it("holds the cursor on a durable write failure, then advances idempotently", async () => {
+    const userId = await seedProviderUser("whoop");
+    const prisma = getPrismaClient();
+    await prisma.whoopConnection.update({
+      where: { userId },
+      data: {
+        tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+        lastSyncedAt: new Date(oldCursor),
+        resourceCursors: { recovery: oldCursor },
+      },
+    });
+    whoopRecoveries.mockResolvedValue([recovery]);
+
+    await prisma.$executeRaw`
+      ALTER TABLE "measurements"
+      ADD CONSTRAINT "test_whoop_recovery_write_failure"
+      CHECK ("external_id" <> 'sleep-real:hrv_rmssd')
+    `;
+
+    try {
+      await expect(syncUserRecovery(userId)).rejects.toThrow();
+      const failed = await prisma.whoopConnection.findUniqueOrThrow({
+        where: { userId },
+      });
+      expect(failed.resourceCursors).toEqual({ recovery: oldCursor });
+      expect(await prisma.measurement.count({ where: { userId } })).toBe(0);
+    } finally {
+      await prisma.$executeRaw`
+        ALTER TABLE "measurements"
+        DROP CONSTRAINT IF EXISTS "test_whoop_recovery_write_failure"
+      `;
+    }
+
+    expect(await syncUserRecovery(userId)).toBe(5);
+    const succeeded = await prisma.whoopConnection.findUniqueOrThrow({
+      where: { userId },
+    });
+    expect(succeeded.resourceCursors).not.toEqual({ recovery: oldCursor });
+    expect(await prisma.measurement.count({ where: { userId } })).toBe(5);
+
+    expect(await syncUserRecovery(userId)).toBe(5);
+    expect(await prisma.measurement.count({ where: { userId } })).toBe(5);
+  });
+});