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.
This commit is contained in:
Marc-André Bombeck committed 2026-11-10 19:08:36 +01:00
1 parent 928ade5109
commit 6fb0d184f1
7 files changed
+97 -7

No files matched your search

+16 -2
View File
@@ -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 },
]);
@@ -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