From 6fb0d184f1e17352b2d2467c8b1ec38354835ffb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 10 Nov 2026 19:08:36 +0100 Subject: [PATCH] fix(sync): hold a drained cursor a minute back so late commits arrive A page's watermark is its last row's stamp, and a long import stamps its rows before it commits them: a quick write committing in between carried the watermark past them, and the keyset never returned them. A caught-up page now hands out watermarks at most a minute before serverNow in every row domain, as start=now does, so the next pull reads that minute again. The mirror domain keeps its watermark. --- docs/api/openapi.yaml | 4 +- src/app/api/sync/changes/route.ts | 6 ++- src/lib/openapi/routes/sync.ts | 4 +- src/lib/sync/changes/pull.ts | 2 + src/lib/sync/cursor.ts | 28 +++++++++++++ tests/integration/apple-health-mirror.test.ts | 18 +++++++- .../sync-changes-start-now.test.ts | 42 ++++++++++++++++++- 7 files changed, 97 insertions(+), 7 deletions(-) diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 21f8a4772..e4e977459 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -50185,7 +50185,9 @@ components: - "null" hasMore: type: boolean - description: False once the client is caught up as of `serverNow`. + description: False once the client is caught up as of `serverNow`. The cursor of such a page sits at most a minute + before `serverNow` in each row domain, so the next pull reads rows changed in that last minute again (a + write still in flight may commit a row stamped there); upserts are idempotent. cursorExpired: type: boolean description: True when the supplied cursor predates tombstone retention, or was issued before the account's data was diff --git a/src/app/api/sync/changes/route.ts b/src/app/api/sync/changes/route.ts index 6f126f02c..4d6574219 100644 --- a/src/app/api/sync/changes/route.ts +++ b/src/app/api/sync/changes/route.ts @@ -30,7 +30,11 @@ * server `id` (the client mirrors those by server id). * - `hasMore` is true when ANY served domain still has rows past its * page; the next `cursor` advances every domain that returned rows. - * When `hasMore` is false the client is caught up as of `serverNow`. + * When `hasMore` is false the client is caught up as of `serverNow`, + * and that page's cursor is held back to a minute before it in every + * row domain (`holdBackDrainedCursor`): a row stamped in that minute + * whose write commits after the pull still arrives on the next one, at + * the cost of reading that minute twice. * - `cursorExpired: true` when ANY domain watermark predates the * tombstone-retention horizon — the client drops its cursor and does * a clean initial sync (a deletion older than retention may have diff --git a/src/lib/openapi/routes/sync.ts b/src/lib/openapi/routes/sync.ts index 3a6cb02ab..bfc9207a7 100644 --- a/src/lib/openapi/routes/sync.ts +++ b/src/lib/openapi/routes/sync.ts @@ -285,7 +285,9 @@ export const syncChangesResponse = z .describe("Opaque cursor to echo into the next request."), hasMore: z .boolean() - .describe("False once the client is caught up as of `serverNow`."), + .describe( + "False once the client is caught up as of `serverNow`. The cursor of such a page sits at most a minute before `serverNow` in each row domain, so the next pull reads rows changed in that last minute again (a write still in flight may commit a row stamped there); upserts are idempotent.", + ), cursorExpired: z .boolean() .describe( diff --git a/src/lib/sync/changes/pull.ts b/src/lib/sync/changes/pull.ts index 49dc08c7d..4b5ca8be6 100644 --- a/src/lib/sync/changes/pull.ts +++ b/src/lib/sync/changes/pull.ts @@ -14,6 +14,7 @@ import { decodeCursorIssuedAt, encodeCursor, encodeCursorAtNow, + holdBackDrainedCursor, SYNC_DOMAINS, type SyncCursor, type SyncDomain, @@ -232,6 +233,7 @@ export async function pullSyncChanges(input: { cd.hasMore || cy.hasMore || mm.hasMore; + if (!hasMore) holdBackDrainedCursor(nextCursor, serverNow.getTime()); annotate({ action: { name: "sync.changes.pull" }, diff --git a/src/lib/sync/cursor.ts b/src/lib/sync/cursor.ts index ab2fa1694..7cacd33ac 100644 --- a/src/lib/sync/cursor.ts +++ b/src/lib/sync/cursor.ts @@ -129,6 +129,34 @@ export function encodeCursorAtNow(nowMs: number): string { return encodeCursor(cursor, nowMs); } +/** + * Hold a drained walk's watermarks back to `START_NOW_OVERLAP_MS` before + * `nowMs`, in place. The watermark of a page is the stamp of its last row, + * and that stamp says nothing about rows still in flight: a long write (a + * Garmin or Withings chunk holds its transaction for seconds) stamps its rows + * first and commits them last, so a quick write that commits in between can + * carry the watermark past them, and the keyset would never return them. + * Held back a minute, the next pull reads the last minute again and the + * late rows come with it; anything older is passed as before. Only the + * cursor of a walk that is done is held back: one that still has pages keeps + * moving forward, or a minute holding more than a page could never drain. + * + * `measurementMirror` is left alone. Its watermark is a token's mirror + * stamp, written by one short update, and reading its last minute again + * would send every reading of a token whose mirror switch just flipped + * once more on every pull in that minute. + */ +export function holdBackDrainedCursor(cursor: SyncCursor, nowMs: number): void { + const ceiling = nowMs - START_NOW_OVERLAP_MS; + for (const domain of SYNC_DOMAINS) { + if (domain === "measurementMirror") continue; + const wm = cursor[domain]; + if (wm && wm.updatedAtMs > ceiling) { + cursor[domain] = { updatedAtMs: ceiling, id: "0" }; + } + } +} + function parseWatermark(raw: unknown): DomainWatermark | null { if (typeof raw !== "object" || raw === null) return null; const u = (raw as { u?: unknown }).u; diff --git a/tests/integration/apple-health-mirror.test.ts b/tests/integration/apple-health-mirror.test.ts index fb62b59d3..70ce6cdbf 100644 --- a/tests/integration/apple-health-mirror.test.ts +++ b/tests/integration/apple-health-mirror.test.ts @@ -259,7 +259,13 @@ describe("switching the setting", () => { const next = await callRoute<{ changes: { - measurements: { upserts: unknown[] }; + measurements: { + upserts: Array<{ + id: string; + updatedAt: string; + syncVersion: number; + }>; + }; measurementMirror: { upserts: Array<{ id: string; appleHealthMirror: boolean }>; }; @@ -270,7 +276,15 @@ describe("switching the setting", () => { query: { cursor: first.data.cursor }, auth: owner, }); - expect(next.data.changes.measurements.upserts).toEqual([]); + // The reading was written within the last minute, so the drained cursor + // reads it once more; the switch itself changed nothing about it. + expect( + next.data.changes.measurements.upserts.map((r) => ({ + id: r.id, + updatedAt: new Date(r.updatedAt), + syncVersion: r.syncVersion, + })), + ).toEqual([{ id, ...before }]); expect(next.data.changes.measurementMirror.upserts).toEqual([ { id, appleHealthMirror: true }, ]); diff --git a/tests/integration/sync-changes-start-now.test.ts b/tests/integration/sync-changes-start-now.test.ts index f77ee7591..72a63bb5d 100644 --- a/tests/integration/sync-changes-start-now.test.ts +++ b/tests/integration/sync-changes-start-now.test.ts @@ -138,12 +138,50 @@ describe("a cursor that starts now", () => { ); expect(mirrored.every((r) => r.appleHealthMirror === true)).toBe(true); - // And the cursor it hands back is a normal one: draining further is quiet. + // The drained cursor it hands back sits a minute back like the start + // one, so the row written in that minute comes once more and nothing + // else does. The mirror domain is not held back. const after = await pull({ cursor: next.cursor! }); - expect(after.changes.measurements.upserts).toEqual([]); + expect(after.changes.measurements.upserts.map((r) => r.id)).toEqual([ + fresh.id, + ]); expect(after.changes.measurementMirror.upserts).toEqual([]); }); + it("delivers a row stamped before a drained page's last row that commits after it", async () => { + // A long write stamps its rows first and commits them last; a quick + // write that commits in between moves the last row of the page past + // them. + const quick = await makeMeasurement( + { id: userId }, + { + source: "EXTERNAL", + ingestTokenId: scale.tokenId, + measuredAt: new Date(Date.UTC(2026, 10, 4, 7)), + externalId: "quick", + }, + ); + const first = await pull({}); + expect(first.hasMore).toBe(false); + expect(first.changes.measurements.upserts.map((r) => r.id)).toContain( + quick.id, + ); + const late = await makeMeasurement( + { id: userId }, + { + source: "EXTERNAL", + ingestTokenId: scale.tokenId, + measuredAt: new Date(Date.UTC(2026, 10, 4, 8)), + externalId: "late-after-page", + updatedAt: new Date(quick.updatedAt.getTime() - 1_000), + }, + ); + const next = await pull({ cursor: first.cursor! }); + expect(next.changes.measurements.upserts.map((r) => r.id)).toContain( + late.id, + ); + }); + it("delivers a row stamped before the cursor that commits after it", async () => { const start = await pull({ start: "now" }); // A write in flight when the cursor was handed out: its stamp predates