Merge pull request #496 from MBombeck/release/v1.28.39

release/v1.28.39 — sync data-loss hardening

CHANGELOG.md

@@ -2,6 +2,24 @@
 
 ## [Unreleased]
 
+## [1.28.39] — 2026-07-16 — Sync data-loss hardening
+
+Four fixes to sync paths that could lose or corrupt data in edge cases:
+
+- The nightly pass that folds raw cumulative samples (steps, energy, distance)
+  into daily totals no longer re-counts a sample you had deleted — it was the
+  one such pass missing the deleted-row filter its siblings all carry.
+- That same pass is now safe when two runs overlap (a deploy-timing overlap, or
+  a manual admin trigger racing the nightly job): an advisory lock plus an
+  in-transaction re-read make the fold idempotent, so an overlap can no longer
+  permanently double a day's total.
+- A re-scored Withings sleep segment (its end time shifts) now updates the
+  existing row in place instead of colliding with it and wedging that night —
+  the same re-key rescue the other providers already carry.
+- Withings measurement sync now holds its position when a reading fails to
+  write, instead of advancing past it and stranding the reading; the next sync
+  retries it.
+
 ## [1.28.38] — 2026-07-16 — Documents polish, waist reminders, bulk sharing
 
 The document vault gets a round of overdue polish. The list cards drop a

docs/api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: 3.1.0
 info:
   title: HealthLog API
-  version: 1.28.38
+  version: 1.28.39
   description: >-
     Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest.
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "healthlog",
-  "version": "1.28.38",
+  "version": "1.28.39",
   "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.",
   "license": "PolyForm-Noncommercial-1.0.0",
   "homepage": "https://healthlog.dev",

public/sw.js

@@ -36,7 +36,7 @@ try {
 // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild.
 const CACHE_VERSION =
   (typeof self !== "undefined" && self.__APP_VERSION__) ||
-  /* @sw-version-fallback */ "v1.28.38";
+  /* @sw-version-fallback */ "v1.28.39";
 const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`;
 const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`;
 // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*`

src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts

@@ -390,6 +390,15 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
     const update = vi.fn().mockResolvedValue({ id: "stats-row" });
     const create = vi.fn().mockResolvedValue({ id: "stats-row-new" });
     const deleteMany = vi.fn().mockResolvedValue({ count: 2 });
+    // v1.28.39 F2 — the in-tx re-read of the still-live per-sample contributors
+    // (only fires when an index-A `stats:` total already exists). Returns the
+    // two late rows so `liveIncrement` sums to the same 500 the pre-scan did on
+    // a clean single run.
+    const findMany = vi
+      .fn()
+      .mockResolvedValue([{ value: 300 }, { value: 200 }]);
+    // v1.28.39 F2 — the per-bucket advisory lock (`pg_advisory_xact_lock`).
+    const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]);
 
     // `findFirst` serves three roles inside `writeDay`:
     //   1. index-A lookup (filters on `externalId`) → the existing collapsed
@@ -412,7 +421,10 @@ describe("drainPerSampleCumulative — late-sync fold into existing total", () =
       },
     );
 
-    const tx = { measurement: { update, create, deleteMany, findFirst } };
+    const tx = {
+      $queryRaw,
+      measurement: { update, create, deleteMany, findFirst, findMany },
+    };
 
     return {
       prisma: {
@@ -530,8 +542,13 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", (
         return { unit: "count" };
       },
     );
+    const findMany = vi.fn().mockResolvedValue([]);
+    const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]);
 
-    const tx = { measurement: { update, create, deleteMany, findFirst } };
+    const tx = {
+      $queryRaw,
+      measurement: { update, create, deleteMany, findFirst, findMany },
+    };
     const prisma = {
       user: { findMany: findManyUser },
       measurement: { findMany: findManyMeasurement },
@@ -585,8 +602,13 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", (
         return { unit: "count" };
       },
     );
+    const findMany = vi.fn().mockResolvedValue([]);
+    const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]);
 
-    const tx = { measurement: { update, create, deleteMany, findFirst } };
+    const tx = {
+      $queryRaw,
+      measurement: { update, create, deleteMany, findFirst, findMany },
+    };
     const prisma = {
       user: { findMany: findManyUser },
       measurement: { findMany: findManyMeasurement },
@@ -612,3 +634,122 @@ describe("drainPerSampleCumulative — colliding day does not abort the walk", (
     expect(summary.totals.bucketsCollapsed).toBe(1);
   });
 });
+
+// v1.28.39 F1 — the cumulative drain was the ONLY one of the four consolidation
+// drains whose scan `where` omitted `deletedAt: null`, so a soft-deleted
+// non-`stats:` APPLE_HEALTH sample was re-summed into the day total AND
+// hard-deleted (resurrecting a user-deleted grace-window sample and erasing the
+// legacy-steps audit tombstones into a double-counted second `stats:` total).
+// The scan must pin live rows only, matching the three siblings.
+describe("drainPerSampleCumulative — scan excludes soft-deleted rows (F1)", () => {
+  function buildPrismaMock() {
+    const findManyUser = vi.fn();
+    const findManyMeasurement = vi.fn().mockResolvedValue([]);
+    return {
+      user: { findMany: findManyUser },
+      measurement: { findMany: findManyMeasurement },
+      $transaction: vi.fn(async (cb: (tx: unknown) => Promise<unknown>) =>
+        cb({}),
+      ),
+    } as unknown as PrismaClient & {
+      user: { findMany: ReturnType<typeof vi.fn> };
+      measurement: { findMany: ReturnType<typeof vi.fn> };
+    };
+  }
+
+  it("pins the per-sample scan to deletedAt: null so tombstones are never re-summed or hard-deleted", async () => {
+    const prismaMock = buildPrismaMock();
+    prismaMock.user.findMany.mockResolvedValue([
+      { id: "user-1", timezone: "Europe/Berlin" },
+    ]);
+
+    await drainPerSampleCumulative(prismaMock, { log: () => {} });
+
+    const call = prismaMock.measurement.findMany.mock.calls[0]?.[0];
+    expect(call).toBeDefined();
+    expect(call.where.source).toBe("APPLE_HEALTH");
+    // The fix: the scan is live-rows-only. Pre-fix this key was absent and a
+    // soft-deleted Apple sample fell into the sum + hard delete.
+    expect(call.where.deletedAt).toBeNull();
+  });
+});
+
+// v1.28.39 F2 — the additive late-fold merge (`eidRow.value + <increment>`)
+// double-counted under two overlapping runs of the same (user, type, day): both
+// scan the same live per-sample rows, the winner mints+hard-deletes, and the
+// loser re-added its now-stale scan sum onto the winner's committed total → a
+// permanent 2× that never self-healed. The per-bucket advisory lock serialises
+// the folds and the in-tx re-read makes the serialised-second fold contribute
+// zero.
+describe("drainPerSampleCumulative — concurrent fold does not double-count (F2)", () => {
+  it("re-reads the live contributors in-tx so a second overlapping drain adds zero", async () => {
+    const findManyUser = vi
+      .fn()
+      .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]);
+
+    // This run's SCAN still sees the two per-sample rows (it scanned before the
+    // sibling committed) summing to 500 in `reducedValue`.
+    const findManyMeasurement = vi.fn(
+      async (args: { where: { type: string } }) => {
+        if (args.where.type === "ACTIVITY_STEPS") {
+          return [
+            {
+              id: "s-1",
+              type: "ACTIVITY_STEPS",
+              value: 300,
+              measuredAt: new Date("2026-05-16T20:00:00.000Z"),
+              externalId: "hk-uuid-1",
+            },
+            {
+              id: "s-2",
+              type: "ACTIVITY_STEPS",
+              value: 200,
+              measuredAt: new Date("2026-05-16T21:00:00.000Z"),
+              externalId: "hk-uuid-2",
+            },
+          ];
+        }
+        return [];
+      },
+    );
+
+    const update = vi.fn().mockResolvedValue({ id: "stats-row" });
+    const create = vi.fn().mockResolvedValue({ id: "stats-row-new" });
+    const deleteMany = vi.fn().mockResolvedValue({ count: 0 });
+    // The sibling run already collapsed the day to 9000 (index-A row exists)…
+    const findFirst = vi.fn(
+      async (args: { where: Record<string, unknown> }) => {
+        if ("externalId" in args.where) return { id: "stats-row", value: 9000 };
+        if ("measuredAt" in args.where) return null;
+        return { unit: "count" };
+      },
+    );
+    // …AND already HARD-DELETED s-1 / s-2, so the in-tx re-read of the live
+    // contributors returns nothing → increment 0, not the stale 500.
+    const findMany = vi.fn().mockResolvedValue([]);
+    const $queryRaw = vi.fn().mockResolvedValue([{ locked: 1 }]);
+
+    const tx = {
+      $queryRaw,
+      measurement: { update, create, deleteMany, findFirst, findMany },
+    };
+    const prisma = {
+      user: { findMany: findManyUser },
+      measurement: { findMany: findManyMeasurement },
+      $transaction: vi.fn(async (cb: (t: typeof tx) => Promise<unknown>) =>
+        cb(tx),
+      ),
+    } as unknown as PrismaClient;
+
+    await drainPerSampleCumulative(prisma, { log: () => {} });
+
+    // The advisory lock fired for the bucket.
+    expect($queryRaw).toHaveBeenCalledTimes(1);
+    // The row is adopted in place and left at 9000 — NOT 9500. Pre-fix the
+    // merge added the stale `reducedValue` (500) onto the committed total.
+    expect(create).not.toHaveBeenCalled();
+    expect(update).toHaveBeenCalledTimes(1);
+    const updateArg = update.mock.calls[0]?.[0] as { data: { value: number } };
+    expect(updateArg.data.value).toBe(9000);
+  });
+});

src/lib/measurements/drain-per-sample-cumulative.ts

@@ -215,14 +215,22 @@ export async function drainPerSampleCumulative(
     statsPrefix: DAILY_STATS_PREFIX,
     reduce: sumBucketValues,
     // v1.4.37 W7c — exclude rows inside the grace window so the nightly
-    // scheduled drain never collapses today's still-in-flight watch
-    // syncs. No `deletedAt` filter and no NOT-stats predicate here: the
-    // cumulative drain historically scans every row and relies on the
-    // in-memory bucketer to skip already-collapsed `stats:` rows.
+    // scheduled drain never collapses today's still-in-flight watch syncs.
+    // `deletedAt: null` pins the scan to LIVE rows — matching the three
+    // sibling drains (daily-mean :234, dense-intraday :464, legacy-steps
+    // :287). Without it a soft-deleted non-`stats:` APPLE_HEALTH sample was
+    // (a) re-summed into the day total and (b) hard-deleted below, which both
+    // resurrects a user-deleted sample inside the grace window AND erases the
+    // legacy-steps audit tombstones (double-counting them into a second
+    // `stats:` total). The NOT-`stats:` predicate stays OFF here: the
+    // cumulative drain relies on the in-memory bucketer to skip
+    // already-collapsed `stats:` rows, and the live `stats:` total (never
+    // tombstoned) is still scanned so the late-fold path can see it.
     buildScanWhere: ({ userId, type, cutoffAt }) => ({
       userId,
       source: "APPLE_HEALTH",
       type,
+      deletedAt: null,
       ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}),
     }),
     writeDay: async ({
@@ -253,6 +261,25 @@ export async function drainPerSampleCumulative(
       const foldOnce = async (): Promise<number> => {
         let removed = 0;
         await pc.$transaction(async (tx) => {
+          // Single-flight the per-(user, type, day) fold. Two overlapping
+          // drains — a rolling-deploy nightly overlap, or an operator's manual
+          // `POST /api/admin/drain-per-sample-cumulative` racing the cron —
+          // each scan the same live per-sample rows before either commits.
+          // Without a guard the second run re-adds its already-drained sum onto
+          // the first run's committed `stats:` total → a permanent 2× day total
+          // that never self-heals. The transaction-scoped advisory lock
+          // (released on commit/rollback, keyed on the bucket's canonical
+          // externalId) serialises the two folds so their read+write never
+          // interleave; the live re-read below then makes the serialised-second
+          // fold contribute exactly zero. Mirrors the per-user advisory-lock
+          // guard on the documents quota gate (`api/documents/inbound`).
+          // (`pg_advisory_xact_lock` returns void, which the client cannot
+          // deserialize as a column — selecting FROM it yields a plain int row.)
+          await tx.$queryRaw`
+            SELECT 1 AS locked
+            FROM pg_advisory_xact_lock(hashtextextended('drain-cumulative:' || ${userId} || ':' || ${externalId}, 0))
+          `;
+
           // Index-A row: already carries the target `stats:` externalId. When
           // present, this is the existing collapsed daily total.
           const eidRow = await tx.measurement.findFirst({
@@ -286,8 +313,27 @@ export async function drainPerSampleCumulative(
           // per-sample row that fell on local noon, itself part of this
           // bucket), its value is already inside `reducedValue`, so the merged
           // value is just `reducedValue`. A fresh day mints `reducedValue`.
+          //
+          // Concurrency guard (paired with the advisory lock above): a sibling
+          // fold that committed just before this one may have already merged
+          // some or all of these per-sample rows into `eidRow` and hard-deleted
+          // them. Trusting the pre-scan `reducedValue` for the increment would
+          // then re-add a sibling's already-folded work → the permanent 2×.
+          // Re-read the still-present contributors inside the locked
+          // transaction: whatever survives is the genuine un-folded increment
+          // (all of `reducedValue` on a clean single run; zero for a
+          // serialised-second run whose rows the winner already drained).
+          const liveIncrement =
+            eidRow !== null
+              ? (
+                  await tx.measurement.findMany({
+                    where: { id: { in: sourceRowIds } },
+                    select: { value: true },
+                  })
+                ).reduce((sum, r) => sum + r.value, 0)
+              : 0;
           const mergedValue =
-            eidRow !== null ? eidRow.value + reducedValue : reducedValue;
+            eidRow !== null ? eidRow.value + liveIncrement : reducedValue;
 
           let canonicalRowId: string;
           if (adoptTarget) {

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

@@ -0,0 +1,179 @@
+/**
+ * v1.28.39 F4 — Withings measure-sync watermark-hold contract.
+ *
+ * The documented status-tracking contract (see `syncUserMeasurements`) says a
+ * downstream measurement-upsert failure must record a sync failure rather than
+ * silently advancing the watermark. Pre-fix, `lastSyncedAt` + `recordSyncSuccess`
+ * were stamped UNCONDITIONALLY after a warn-only per-row loop, so a transient
+ * per-row write failure stranded that reading forever behind the 10-minute
+ * overlap window. These tests pin the fix: a hard row failure HOLDS the
+ * watermark (no stamp, no success, records a failure) so the next tick retries;
+ * a benign P2002 idempotent-write collision does NOT hold it; a clean run stamps
+ * as before.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/db", () => ({
+  prisma: {
+    measurement: {
+      findFirst: vi.fn(),
+      create: vi.fn(),
+      update: vi.fn(),
+    },
+    withingsConnection: {
+      findUnique: vi.fn(),
+      update: vi.fn(),
+    },
+  },
+}));
+
+vi.mock("@/lib/crypto", () => ({
+  decrypt: vi.fn((v: string) => v),
+  encrypt: vi.fn((v: string) => v),
+}));
+
+vi.mock("../client", () => ({
+  fetchMeasurements: vi.fn(),
+  refreshAccessToken: vi.fn(),
+  subscribeWebhook: vi.fn(),
+}));
+
+vi.mock("@/lib/integrations/status", async () => {
+  const actual = await vi.importActual<
+    typeof import("@/lib/integrations/status")
+  >("@/lib/integrations/status");
+  return {
+    ...actual,
+    isReauthRequired: vi.fn().mockResolvedValue(false),
+    recordSyncFailure: vi.fn().mockResolvedValue(undefined),
+    recordSyncSuccess: vi.fn().mockResolvedValue(undefined),
+  };
+});
+
+vi.mock("@/lib/rollups/measurement-rollups", async () => {
+  const actual = await vi.importActual<
+    typeof import("@/lib/rollups/measurement-rollups")
+  >("@/lib/rollups/measurement-rollups");
+  return {
+    ...actual,
+    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+  };
+});
+
+vi.mock("@/lib/insights/comprehensive-generate", () => ({
+  invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined),
+}));
+
+vi.mock("@/lib/cache/invalidate", () => ({
+  invalidateUserDashboardSnapshot: vi.fn(),
+}));
+
+vi.mock("@/lib/logging/context", () => ({
+  getEvent: vi.fn(() => ({
+    addExternalCall: vi.fn(),
+    addWarning: vi.fn(),
+  })),
+  annotate: vi.fn(),
+}));
+
+import { prisma } from "@/lib/db";
+import { fetchMeasurements } from "../client";
+import {
+  recordSyncFailure,
+  recordSyncSuccess,
+} from "@/lib/integrations/status";
+
+import { syncUserMeasurements } from "../sync";
+
+const P2002 = Object.assign(new Error("Unique constraint failed"), {
+  code: "P2002",
+});
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  // A live, non-expired connection so `getValidToken` returns the token
+  // without touching the refresh path.
+  vi.mocked(prisma.withingsConnection.findUnique).mockResolvedValue({
+    id: "conn-1",
+    withingsUserId: "wu-1",
+    accessToken: "enc-access",
+    refreshToken: "enc-refresh",
+    tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000),
+    lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000),
+  } as never);
+  vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never);
+  vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null);
+  vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+  vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
+});
+
+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([
+      {
+        type: "WEIGHT",
+        value: 80,
+        measuredAt: new Date("2026-05-16T08:00:00Z"),
+      },
+      {
+        type: "PULSE",
+        value: 60,
+        measuredAt: new Date("2026-05-16T08:05:00Z"),
+      },
+    ] as never);
+    // The first write hard-fails (a transient DB error), the second succeeds.
+    vi.mocked(prisma.measurement.create)
+      .mockRejectedValueOnce(new Error("connection reset by peer"))
+      .mockResolvedValue({} as never);
+
+    const imported = await syncUserMeasurements("user-1");
+
+    // Only the second row persisted.
+    expect(imported).toBe(1);
+    // The watermark is HELD — no stamp — so the next tick refetches the window.
+    expect(prisma.withingsConnection.update).not.toHaveBeenCalled();
+    expect(recordSyncSuccess).not.toHaveBeenCalled();
+    // The failure is recorded so the streak + admin threshold reflect it.
+    expect(recordSyncFailure).toHaveBeenCalledTimes(1);
+    expect(recordSyncFailure).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: "user-1", integration: "withings" }),
+    );
+  });
+
+  it("does NOT hold the watermark for a benign P2002 idempotent-write collision", async () => {
+    vi.mocked(fetchMeasurements).mockResolvedValue([
+      {
+        type: "WEIGHT",
+        value: 80,
+        measuredAt: new Date("2026-05-16T08:00:00Z"),
+      },
+    ] as never);
+    vi.mocked(prisma.measurement.create).mockRejectedValueOnce(P2002);
+
+    const imported = await syncUserMeasurements("user-1");
+
+    // P2002 means the row is already present — nothing lost — so the cycle is
+    // still healthy: stamp the watermark + success, record no failure.
+    expect(imported).toBe(0);
+    expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1);
+    expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
+    expect(recordSyncFailure).not.toHaveBeenCalled();
+  });
+
+  it("stamps the watermark + success on a clean run", async () => {
+    vi.mocked(fetchMeasurements).mockResolvedValue([
+      {
+        type: "WEIGHT",
+        value: 80,
+        measuredAt: new Date("2026-05-16T08:00:00Z"),
+      },
+    ] as never);
+
+    const imported = await syncUserMeasurements("user-1");
+
+    expect(imported).toBe(1);
+    expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1);
+    expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings");
+    expect(recordSyncFailure).not.toHaveBeenCalled();
+  });
+});

src/lib/withings/__tests__/sync-sleep.test.ts

@@ -322,6 +322,46 @@ describe("syncUserSleep — segment writes + idempotency", () => {
     });
   });
 
+  it("re-keys a re-scored segment in place via the natural key instead of colliding (F3, 0055 wedge)", async () => {
+    // Withings re-scores a night: the segment's END (measuredAt + the 0055
+    // natural key) stays fixed while its START (externalId) shifts, same stage.
+    // The externalId probe MISSES (new START ⇒ new id), but the natural key
+    // `(userId, type, measuredAt, source, sleepStage)` is still occupied by the
+    // prior row. A blind create would P2002 on it (swallowed → row lost, then
+    // the sweep tombstones the old row → the night wedges forever). The
+    // natural-key rescue must UPDATE the surviving row in place, re-keying it.
+    installFetchMock([
+      { startdate: 1715000500, enddate: 1715003600, state: 2, id: 42 },
+    ]);
+    vi.mocked(prisma.measurement.findFirst).mockImplementation((async (args: {
+      where: Record<string, unknown>;
+    }) => {
+      const where = args.where;
+      // externalId probe (START shifted ⇒ new id) → miss.
+      if ("externalId" in where) return null;
+      // natural-key probe (END + stage unchanged) → the surviving prior row.
+      if ("measuredAt" in where) return { id: "old-row" };
+      return null;
+    }) as never);
+    vi.mocked(prisma.measurement.update).mockResolvedValue({} as never);
+    vi.mocked(prisma.measurement.create).mockResolvedValue({} as never);
+
+    const imported = await syncUserSleep("user-1");
+    expect(imported).toBe(1);
+
+    // No collision: the survivor is re-keyed in place, never re-created.
+    expect(prisma.measurement.create).not.toHaveBeenCalled();
+    expect(prisma.measurement.update).toHaveBeenCalledWith({
+      where: { id: "old-row" },
+      data: {
+        value: expect.any(Number),
+        // Re-keyed onto the shifted START-based externalId and resurrected.
+        externalId: "withings:sleep:user-1:42:1715000500",
+        deletedAt: null,
+      },
+    });
+  });
+
   it("stamps every row with an externalId keyed on session id + segment START", async () => {
     installFetchMock([
       { startdate: 1715000000, enddate: 1715003600, state: 2, id: 42 },

src/lib/withings/sync-sleep.ts

@@ -471,18 +471,56 @@ export async function syncUserSleep(
           },
         });
       } else {
-        await prisma.measurement.create({
-          data: {
+        // Natural-key rescue — mirrors the measure path's findFirst in
+        // `sync.ts`. When Withings re-scores a night it can shift a segment's
+        // START (its externalId) while the END (`measuredAt`) and the stage
+        // stay put: the externalId probe above then MISSES, but the 0055
+        // natural key `(userId, type, measuredAt, source, sleepStage)`
+        // (NULLS NOT DISTINCT) is already occupied by the prior row. A blind
+        // `create` P2002s on that key; the collision is swallowed by the catch
+        // below and the sweep then tombstones the old row — so the re-keyed
+        // segment is lost and, because a tombstoned row still holds the natural
+        // key, every later sync re-collides: the night wedges permanently (the
+        // "0055 erased re-keyed sleep" class). Probe the natural key
+        // (tombstone-inclusive — no `deletedAt` filter) and re-key that row in
+        // place, so a re-scored segment updates rather than colliding.
+        const naturalKeyRow = await prisma.measurement.findFirst({
+          where: {
             userId,
             type: SLEEP_TYPE,
-            value: minutes,
-            unit: getUnitForType(SLEEP_TYPE),
-            measuredAt,
             source: "WITHINGS",
+            measuredAt,
             sleepStage: stage,
-            externalId,
           },
+          select: { id: true },
         });
+        if (naturalKeyRow) {
+          await prisma.measurement.update({
+            where: { id: naturalKeyRow.id },
+            // Re-key the surviving row onto the new START-based externalId and
+            // resurrect it if the sweep had tombstoned it. `measuredAt` and
+            // `sleepStage` already match by construction (they ARE the natural
+            // key we matched on), so only value + externalId + deletedAt move.
+            data: {
+              value: minutes,
+              externalId,
+              deletedAt: null,
+            },
+          });
+        } else {
+          await prisma.measurement.create({
+            data: {
+              userId,
+              type: SLEEP_TYPE,
+              value: minutes,
+              unit: getUnitForType(SLEEP_TYPE),
+              measuredAt,
+              source: "WITHINGS",
+              sleepStage: stage,
+              externalId,
+            },
+          });
+        }
       }
       touched.push({ type: SLEEP_TYPE, measuredAt });
       imported++;

src/lib/withings/sync.ts

@@ -12,7 +12,8 @@ import {
   subscribeWebhook,
 } from "./client";
 import { getUserWithingsCredentials } from "./credentials";
-import { getEvent } from "@/lib/logging/context";
+import { annotate, getEvent } from "@/lib/logging/context";
+import { isP2002 } from "@/lib/prisma-errors";
 import {
   isReauthRequired,
   recordSyncFailure,
@@ -213,6 +214,16 @@ export async function syncUserMeasurements(
   // the user's sync.
   const touched: Array<{ type: MeasurementType; measuredAt: Date }> = [];
 
+  // v1.28.39 — hold-watermark-on-hard-failure (mirrors google-health /
+  // fitbit's `hardFailStorage` verdict). A per-row write that HARD-fails
+  // (anything but a benign P2002 idempotent-write collision) must not let
+  // `lastSyncedAt` advance past the unpersisted reading — the 10-min overlap
+  // window would otherwise never re-cover it and the reading is stranded
+  // forever. Track any hard failure so the watermark stamp + `recordSyncSuccess`
+  // below are held and the failure is recorded for the next tick to retry.
+  let anyRowFailed = false;
+  let firstRowError: unknown = null;
+
   for (const m of measures) {
     const measType = m.type as MeasurementType;
     try {
@@ -258,7 +269,22 @@ export async function syncUserMeasurements(
       touched.push({ type: measType, measuredAt: m.measuredAt });
       imported++;
     } catch (err) {
-      getEvent()?.addWarning(`Failed to upsert measure: ${err}`);
+      if (isP2002(err)) {
+        // Benign: the composite unique index serialised a concurrent insert of
+        // this exact reading — the row is present, nothing is lost. Warn and
+        // continue WITHOUT holding the watermark (this is the expected
+        // idempotent-write race the findFirst+create models around).
+        getEvent()?.addWarning(
+          `Withings measure upsert hit an idempotent P2002 (row already present): ${err}`,
+        );
+      } else {
+        // A genuine write failure strands this reading. Flag the cycle so the
+        // watermark is held and the next scheduled tick refetches the overlap
+        // window and retries.
+        anyRowFailed = true;
+        if (firstRowError === null) firstRowError = err;
+        getEvent()?.addWarning(`Failed to upsert measure: ${err}`);
+      }
     }
   }
 
@@ -300,6 +326,29 @@ export async function syncUserMeasurements(
     );
   }
 
+  // v1.28.39 — HOLD the watermark on any hard row failure. Implements the
+  // documented status-tracking contract above ("downstream measurement-upsert
+  // that fails ALL items → recordSyncFailure"), extended to a partial hard
+  // failure: a single hard-failed row must not advance `lastSyncedAt` past the
+  // reading it failed to persist, or the 10-min overlap window loses it. Record
+  // the failure (classified transient by default → the connection stays live
+  // and the next tick retries from the un-advanced watermark) and return
+  // WITHOUT stamping the watermark or success. The rows that DID write kept
+  // their rollup recompute above.
+  if (anyRowFailed) {
+    annotate({
+      action: {
+        name: "withings.sync.watermark_held",
+        details: { imported },
+      },
+    });
+    await recordWithingsSyncFailure(
+      userId,
+      firstRowError ?? new Error("Withings measure row write failed"),
+    );
+    return imported;
+  }
+
   // Update last synced timestamp
   await prisma.withingsConnection.update({
     where: { userId },