mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
fix(sync): expire delta cursors issued before a restore or a data wipe
A restore writes rows back with their original updatedAt, behind every paired client's watermarks, and removes rows without leaving tombstones. A client that kept its cursor would miss both on its next incremental pull. The same holds after the person deletes all their data. `users.sync_reset_at` (migration 0363) records when the record was last replaced: the restore stamps it as its last write and again right after the commit, the per-account and the admin wipe stamp it inside their transactions. Cursors now carry the time they were issued, and /api/sync/changes answers its existing `cursorExpired` to a cursor issued before the reset, or one without an issue time on an account that has one. The comparison is on the issue time, because after a restore a freshly drained cursor still carries old watermarks.
This commit is contained in:
1 parent
419ff04765
commit
eda87fd4a7
14 files changed
+375
-13
No files matched your search
@@ -113,7 +113,11 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
// Every account row survives; the personal data carried on its own
|
||||
// columns does not. Same classification contract as the model list.
|
||||
await tx.user.updateMany({ data: USER_RESET });
|
||||
// Paired clients drop their delta cursors and start over, instead of
|
||||
// catching up from a position whose rows are gone (`syncResetAt`).
|
||||
await tx.user.updateMany({
|
||||
data: { ...USER_RESET, syncResetAt: new Date() },
|
||||
});
|
||||
|
||||
return perModel;
|
||||
},
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
wipeDelegateKey,
|
||||
WIPE_MODELS,
|
||||
} from "@/lib/data-wipe/wipe-plan";
|
||||
import { stampSyncReset } from "@/lib/sync/reset";
|
||||
|
||||
export const dynamic = "force-dynamic";
|
||||
|
||||
@@ -104,6 +105,9 @@ export const DELETE = apiHandler(async (request: NextRequest) => {
|
||||
// The account row survives; the personal data carried on its own columns
|
||||
// does not. Same classification contract as the model list.
|
||||
await tx.user.update({ where: { id: userId }, data: USER_RESET });
|
||||
// Paired clients drop their delta cursors and start over, instead of
|
||||
// catching up from a position whose rows are gone.
|
||||
await stampSyncReset(tx, userId);
|
||||
|
||||
// The nightly off-host copies of the record are part of "all your
|
||||
// data". The request commits with the wipe; the purge job deletes
|
||||
|
||||
@@ -34,7 +34,9 @@
|
||||
* - `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
|
||||
* been pruned, so an incremental delta could silently miss it).
|
||||
* been pruned, so an incremental delta could silently miss it) — or
|
||||
* when the cursor was issued before the account's `syncResetAt` (a
|
||||
* backup restore or a delete-all-data since, see the schema note).
|
||||
* - `syncVersion` is echoed per row so the client can keep its mirror's
|
||||
* version monotonic.
|
||||
* - The `cycleDays` / `cycles` blocks are MODULE-GATED: an account with
|
||||
@@ -64,6 +66,7 @@ import { contextForWire, type MoodContextWire } from "@/lib/mood/context";
|
||||
import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
|
||||
import {
|
||||
decodeCursor,
|
||||
decodeCursorIssuedAt,
|
||||
encodeCursor,
|
||||
SYNC_DOMAINS,
|
||||
type DomainWatermark,
|
||||
@@ -271,14 +274,39 @@ export const GET = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
// A stale cursor (ANY domain watermark older than retention) forces a
|
||||
// clean re-init: a deletion older than retention may already be pruned.
|
||||
const expired = SYNC_DOMAINS.some((domain) => {
|
||||
const pastRetention = SYNC_DOMAINS.some((domain) => {
|
||||
const wm = cursor[domain];
|
||||
return wm !== undefined && wm.updatedAtMs < retentionHorizon.getTime();
|
||||
});
|
||||
// A cursor issued before the account's record was last replaced (a backup
|
||||
// restore, or the person deleting all their data) cannot be caught up
|
||||
// either: the restore wrote rows back with their original `updatedAt`,
|
||||
// behind the cursor's watermarks, and removed rows without tombstones. The
|
||||
// client re-initialises instead. Compared by the time the cursor was
|
||||
// issued, not by its watermarks, which a restore leaves meaningless; a
|
||||
// cursor from before the issue time was recorded counts as issued before
|
||||
// any reset.
|
||||
let beforeReset = false;
|
||||
if (parsed.data.cursor) {
|
||||
const account = await prisma.user.findUnique({
|
||||
where: { id: user.id },
|
||||
select: { syncResetAt: true },
|
||||
});
|
||||
if (account?.syncResetAt) {
|
||||
const issuedAt = decodeCursorIssuedAt(parsed.data.cursor);
|
||||
beforeReset =
|
||||
issuedAt === null || issuedAt < account.syncResetAt.getTime();
|
||||
}
|
||||
}
|
||||
const expired = pastRetention || beforeReset;
|
||||
if (expired) {
|
||||
annotate({
|
||||
action: { name: "sync.changes.pull" },
|
||||
meta: { cursor_expired: true, returned: 0 },
|
||||
meta: {
|
||||
cursor_expired: true,
|
||||
cursor_expired_reason: beforeReset ? "sync_reset" : "retention",
|
||||
returned: 0,
|
||||
},
|
||||
});
|
||||
return apiSuccess({
|
||||
serverNow: serverNow.toISOString(),
|
||||
@@ -552,7 +580,7 @@ export const GET = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
return apiSuccess({
|
||||
serverNow: serverNow.toISOString(),
|
||||
cursor: encodeCursor(nextCursor),
|
||||
cursor: encodeCursor(nextCursor, serverNow.getTime()),
|
||||
hasMore,
|
||||
cursorExpired: false,
|
||||
changes: {
|
||||
|
||||
@@ -469,6 +469,8 @@ export const USER_KEPT_FIELDS: Readonly<Record<string, string>> = {
|
||||
"the version of the medical disclaimer the account acknowledged",
|
||||
documentQuotaBytes:
|
||||
"a per-account storage quota set by the operator, not by the person",
|
||||
syncResetAt:
|
||||
"sync bookkeeping, not record content: the wipe sets it to its own time rather than clearing it, so a paired client's delta cursor from before the wipe is refused and the client starts over",
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@@ -759,7 +759,10 @@ describe("backup round trip — export, wire schema, restore", () => {
|
||||
});
|
||||
|
||||
const userUpdate = written.find(
|
||||
(w) => w.model === "user" && w.op === "update",
|
||||
(w) =>
|
||||
w.model === "user" &&
|
||||
w.op === "update" &&
|
||||
"insightsCachedText" in w.data,
|
||||
);
|
||||
expect(
|
||||
userUpdate,
|
||||
@@ -798,8 +801,13 @@ describe("backup round trip — export, wire schema, restore", () => {
|
||||
// whether the restore over-invalidates on a genuine no-op.
|
||||
const { written: writtenUnchanged } = await roundTrip(liveSeed, payload);
|
||||
|
||||
// The restore also stamps the account's sync reset; only an update
|
||||
// that touches the cached briefing is what this asks about.
|
||||
const userUpdate = writtenUnchanged.find(
|
||||
(w) => w.model === "user" && w.op === "update",
|
||||
(w) =>
|
||||
w.model === "user" &&
|
||||
w.op === "update" &&
|
||||
"insightsCachedText" in w.data,
|
||||
);
|
||||
expect(
|
||||
userUpdate,
|
||||
|
||||
@@ -103,6 +103,7 @@ import { restoreEcgData } from "@/lib/export/ecg-backup";
|
||||
import { restoredMedicationCreatedAt } from "@/lib/export/medication-created-at";
|
||||
import { invalidateUserData } from "@/lib/cache/invalidate";
|
||||
import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client";
|
||||
import { stampSyncReset } from "@/lib/sync/reset";
|
||||
import {
|
||||
classifyRestoreFailure,
|
||||
RESTORE_FAILURE_CAUSE_MESSAGES,
|
||||
@@ -2412,6 +2413,11 @@ export async function restoreBackup(
|
||||
foreignBefore,
|
||||
await countForeignReferences(tx, ownerId, tenantEdges),
|
||||
);
|
||||
// Paired clients' delta cursors are no longer valid (see
|
||||
// `User.syncResetAt`). Stamped as the last write, so it commits with
|
||||
// the data, and stamped again right after the commit below to cover
|
||||
// a cursor issued while this transaction was still running.
|
||||
await stampSyncReset(tx, ownerId);
|
||||
progress.sectionsDone = RESTORE_SECTION_STEPS.length;
|
||||
report("sections", true);
|
||||
return {
|
||||
@@ -2479,6 +2485,22 @@ export async function restoreBackup(
|
||||
);
|
||||
}
|
||||
|
||||
// A pull that started before the commit read the pre-restore rows and may
|
||||
// have been issued a cursor after the stamp inside the transaction. The
|
||||
// second stamp expires that one too; it can only expire more cursors, so a
|
||||
// failure here costs a client at most one extra incremental page.
|
||||
try {
|
||||
await stampSyncReset(prisma, ownerId);
|
||||
} catch (err) {
|
||||
annotate({
|
||||
meta: {
|
||||
restore_sync_reset_failed: true,
|
||||
restore_sync_reset_error:
|
||||
err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const { cleared, skipped, expiredTombstonesSkipped } = outcome;
|
||||
const measurementsRestored =
|
||||
streamed.measurementCount - expiredTombstonesSkipped;
|
||||
|
||||
@@ -257,7 +257,7 @@ export const syncChangesResponse = z
|
||||
cursorExpired: z
|
||||
.boolean()
|
||||
.describe(
|
||||
"True when the supplied cursor predates tombstone retention — drop the cursor and do a clean initial sync.",
|
||||
"True when the supplied cursor predates tombstone retention, or was issued before the account's data was last restored from a backup or deleted — drop the cursor and do a clean initial sync.",
|
||||
),
|
||||
changes: z.object({
|
||||
measurements: z.object({
|
||||
|
||||
@@ -3,7 +3,12 @@
|
||||
*/
|
||||
import { describe, it, expect } from "vitest";
|
||||
|
||||
import { encodeCursor, decodeCursor, type SyncCursor } from "../cursor";
|
||||
import {
|
||||
encodeCursor,
|
||||
decodeCursor,
|
||||
decodeCursorIssuedAt,
|
||||
type SyncCursor,
|
||||
} from "../cursor";
|
||||
|
||||
describe("sync cursor codec", () => {
|
||||
it("round-trips a per-domain keyset map through encode/decode", () => {
|
||||
@@ -112,3 +117,20 @@ describe("sync cursor codec", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("cursor issue time (sync reset)", () => {
|
||||
const cursor: SyncCursor = {
|
||||
measurements: { updatedAtMs: 1_000, id: "clx0m" },
|
||||
};
|
||||
|
||||
it("round-trips the issue time beside the watermarks", () => {
|
||||
const token = encodeCursor(cursor, 1_790_000_000_000);
|
||||
expect(decodeCursorIssuedAt(token)).toBe(1_790_000_000_000);
|
||||
expect(decodeCursor(token)).toEqual(cursor);
|
||||
});
|
||||
|
||||
it("reads no issue time from a token that carries none", () => {
|
||||
expect(decodeCursorIssuedAt(encodeCursor(cursor))).toBeNull();
|
||||
expect(decodeCursorIssuedAt("not a token")).toBeNull();
|
||||
});
|
||||
});
|
||||
+31
-3
@@ -63,14 +63,27 @@ const CURSOR_VERSION = 2;
|
||||
* `cycleDays`/`cycles` domains; they decode as absent (fresh scan). */
|
||||
const SUPPORTED_CURSOR_VERSIONS: readonly number[] = [1, 2];
|
||||
|
||||
/** Encode a per-domain keyset map into an opaque token. */
|
||||
export function encodeCursor(cursor: SyncCursor): string {
|
||||
/**
|
||||
* Encode a per-domain keyset map into an opaque token.
|
||||
*
|
||||
* `issuedAtMs` is when the server handed the token out. The watermarks cannot
|
||||
* say that: a restore writes rows back with their original `updatedAt`, so a
|
||||
* cursor drained entirely after a restore can still carry watermarks older
|
||||
* than it. The issue time is what `/api/sync/changes` compares against the
|
||||
* account's `syncResetAt`. Additive inside the v2 envelope: a token without
|
||||
* it decodes as before and counts as issued before any reset.
|
||||
*/
|
||||
export function encodeCursor(cursor: SyncCursor, issuedAtMs?: number): string {
|
||||
const d: Record<string, { u: number; i: string }> = {};
|
||||
for (const domain of SYNC_DOMAINS) {
|
||||
const wm = cursor[domain];
|
||||
if (wm) d[domain] = { u: wm.updatedAtMs, i: wm.id };
|
||||
}
|
||||
const json = JSON.stringify({ v: CURSOR_VERSION, d });
|
||||
const json = JSON.stringify(
|
||||
issuedAtMs === undefined
|
||||
? { v: CURSOR_VERSION, d }
|
||||
: { v: CURSOR_VERSION, t: issuedAtMs, d },
|
||||
);
|
||||
return Buffer.from(json, "utf-8").toString("base64url");
|
||||
}
|
||||
|
||||
@@ -117,3 +130,18 @@ export function decodeCursor(token: string): SyncCursor | null {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* When a token was issued (see {@link encodeCursor}), or null for a token
|
||||
* that does not say: one from before the field existed, or one that does not
|
||||
* parse.
|
||||
*/
|
||||
export function decodeCursorIssuedAt(token: string): number | null {
|
||||
try {
|
||||
const json = Buffer.from(token, "base64url").toString("utf-8");
|
||||
const t = (JSON.parse(json) as { t?: unknown } | null)?.t;
|
||||
return typeof t === "number" && Number.isFinite(t) ? t : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
/**
|
||||
* Expire every delta cursor an account's paired clients hold.
|
||||
*
|
||||
* Called when the account's record is replaced wholesale: a backup restore
|
||||
* (inside its transaction and again right after the commit) and the
|
||||
* person's delete-all-data. `/api/sync/changes` answers `cursorExpired` to
|
||||
* any cursor issued before the stamp, and the client re-initialises. See
|
||||
* `User.syncResetAt` in the schema for why an incremental catch-up from such
|
||||
* a cursor would miss rows.
|
||||
*/
|
||||
import type { Prisma, PrismaClient } from "@/generated/prisma/client";
|
||||
|
||||
export async function stampSyncReset(
|
||||
db: Pick<PrismaClient | Prisma.TransactionClient, "user">,
|
||||
userId: string,
|
||||
at: Date = new Date(),
|
||||
): Promise<void> {
|
||||
await db.user.update({ where: { id: userId }, data: { syncResetAt: at } });
|
||||
}
|
||||
Reference in new issue
Block a user