From 59c97a38f52ddde2e063577000937ff243da6f7c Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:44:02 -0700 Subject: [PATCH 1/2] feat(sync): seal clamps the HLC physical candidate to server date + MAX_DRIFT (T13314) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Spec §1.3 "Local clock ahead": sealing clamps the physical candidate to min(at_ms, serverDate + MAX_DRIFT) when a server date from the last 24 h is known. Captures stamped while the clock ran ahead used to seal with future HLCs once push resumed. - push records sync.server_clock {offsetMs, atMs} whenever it has a server offset (setPushMeta replaces markClockAhead; gate 28 count unchanged). - clock-store sealWallCeiling(db, now): now + offset + MAX_DRIFT, or null when the observation is missing, unreadable or older than 24 h. - the sealer computes the ceiling once per seal transaction and ticks each op at min(at_ms or now, ceiling). Tests: an hour-ahead capture after a fresh server date seals at or below serverNow + MAX_DRIFT; with no observation or a stale one it seals at its at_ms. Mutations killed: no clamp, no TTL, no record, wrong margin. Co-Authored-By: Claude Opus 5.5 --- .changeset/t13314-seal-clock-clamp.md | 15 ++++ .../src/store/sync/__tests__/push.test.ts | 70 +++++++++++++++++++ packages/core/src/store/sync/clock-store.ts | 45 ++++++++++++ packages/core/src/store/sync/push.ts | 35 +++++++--- packages/core/src/store/sync/sealer.ts | 8 ++- 5 files changed, 160 insertions(+), 13 deletions(-) create mode 100644 .changeset/t13314-seal-clock-clamp.md diff --git a/.changeset/t13314-seal-clock-clamp.md b/.changeset/t13314-seal-clock-clamp.md new file mode 100644 index 0000000000..48c6f14b33 --- /dev/null +++ b/.changeset/t13314-seal-clock-clamp.md @@ -0,0 +1,15 @@ +--- +id: t13314-seal-clock-clamp +tasks: [T13314] +kind: feat +summary: "Sealing clamps each op's clock to the server's date plus MAX_DRIFT, so writes stamped while the clock ran ahead never carry future HLCs" +--- + +A device whose clock ran ahead of the server's pauses push (journal spec §1.3), but its writes are still captured with the +clock's ahead time. Once the clock was put right and push resumed, those captures used to seal with future HLCs. A future HLC +wins every last-writer-wins race on every device. + +- Each push now records the server's clock offset and when it was observed (`sync.server_clock` in `_sync_meta`). +- The sealer stamps each op no later than the server's estimated date plus `MAX_DRIFT` (5 minutes) when that observation is + under 24 hours old. Without a recent observation, sealing is unclamped, as before. +- HLCs stay monotonic. The clamp only bounds the physical candidate, and the counter still orders ops. diff --git a/packages/core/src/store/sync/__tests__/push.test.ts b/packages/core/src/store/sync/__tests__/push.test.ts index 10fe2eff50..2f2237632b 100644 --- a/packages/core/src/store/sync/__tests__/push.test.ts +++ b/packages/core/src/store/sync/__tests__/push.test.ts @@ -23,11 +23,14 @@ import { ROW_IDENTITY_RECIPE_KEY, } from '../../row-identity.js'; import { finishCaptureFrame, openCaptureFrame, setCaptureEnabled } from '../capture.js'; +import { SERVER_CLOCK_KEY, SERVER_CLOCK_TTL_MS } from '../clock-store.js'; import { setSyncFlag } from '../flags.js'; import { completeGenesis, cutGenesis } from '../genesis.js'; +import { MAX_DRIFT_MS, parseHlc } from '../hlc.js'; import { CLOCK_AHEAD_KEY, pushStream, type SegmentUploader } from '../push.js'; import { ensureProjectReplica, storeHwm } from '../replica.js'; import { ReplicaRegistry } from '../replica-registry.js'; +import { sealPending } from '../sealer.js'; import type { PersistedSegment } from '../segments.js'; const SYNC_SCHEMA = resolve(import.meta.dirname, '../../../../migrations/sync-journal'); @@ -210,6 +213,73 @@ describe('pushStream (S4-2)', () => { expect(n(db, 'SELECT count(*) AS n FROM _sync_meta WHERE key = ?', CLOCK_AHEAD_KEY)).toBe(0); }); + it("captures stamped while the clock ran ahead seal no later than the server's date plus MAX_DRIFT (T13314, §1.3)", async () => { + const { db, replica } = await pushing(); + write(db, addTask('T1')); + // Stamped an hour ahead; the clock has since been put right. + db.exec('UPDATE _sync_capture SET at_ms = at_ms + 3600000'); + const server = fakeServer(); + const r = await pushStream(db, opts(replica, server, { serverOffsetMs: 0 })); + expect(r).toMatchObject({ clockAhead: false, pushed: 1 }); + const observed = JSON.parse( + ( + db.prepare('SELECT value FROM _sync_meta WHERE key = ?').get(SERVER_CLOCK_KEY) as { + value: string; + } + ).value, + ) as { offsetMs: number; atMs: number }; + expect(observed.offsetMs).toBe(0); + const hlcs = db.prepare('SELECT hlc FROM _sync_op').all() as Array<{ hlc: string }>; + expect(hlcs.length).toBeGreaterThan(0); + for (const { hlc } of hlcs) { + expect(parseHlc(hlc).phys).toBeLessThanOrEqual(clock + MAX_DRIFT_MS); + } + }); + + it('a server date older than 24 h, or none, leaves sealing unclamped (T13314)', async () => { + const { db, replica } = await pushing(); + const stampedAhead = (id: string) => { + write(db, addTask(id)); + db.exec("UPDATE _sync_capture SET at_ms = at_ms + 3600000 WHERE state = 'live'"); + return Number( + ( + db.prepare("SELECT max(at_ms) AS n FROM _sync_capture WHERE state = 'live'").get() as { + n: number; + } + ).n, + ); + }; + const maxPhys = () => + Math.max( + ...(db.prepare('SELECT hlc FROM _sync_op').all() as Array<{ hlc: string }>).map( + (o) => parseHlc(o.hlc).phys, + ), + ); + const seal = () => + sealPending(db, { + scope: 'project', + replica, + now: () => ++clock, + env: {}, + allowUnreleased: true, + }); + // No server date known. + const first = stampedAhead('T1'); + seal(); + expect(maxPhys()).toBe(first); + // A stale one. + db.prepare( + 'INSERT INTO _sync_meta (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value', + ).run( + SERVER_CLOCK_KEY, + JSON.stringify({ offsetMs: 0, atMs: clock - SERVER_CLOCK_TTL_MS - 60_000 }), + new Date(clock).toISOString(), + ); + const second = stampedAhead('T2'); + seal(); + expect(maxPhys()).toBe(second); + }); + it('a server ahead of the store (restored or copied) is refused before anything is sealed or sent', async () => { const { db, replica } = await pushing(); write(db, addTask('T1')); diff --git a/packages/core/src/store/sync/clock-store.ts b/packages/core/src/store/sync/clock-store.ts index b8e9e4759f..027e0e2c32 100644 --- a/packages/core/src/store/sync/clock-store.ts +++ b/packages/core/src/store/sync/clock-store.ts @@ -11,6 +11,7 @@ */ import type { DatabaseSync } from 'node:sqlite'; +import { z } from 'zod'; import { encodeHlc, genesisHlc, @@ -35,6 +36,50 @@ export const CLOCK_HEAL_SOURCES: ReadonlyArray<{ table: string; column: string } { table: '_sync_row_meta', column: 'hlc' }, ]; +/** + * `_sync_meta` key: the server's clock as push last observed it (§1.3), JSON + * `{offsetMs, atMs}` — `offsetMs` is the server's date minus this device's + * clock, `atMs` this device's clock when it was observed. + */ +export const SERVER_CLOCK_KEY = 'sync.server_clock'; + +/** How long an observed server date bounds sealing (§1.3: "a server date from the last 24 h"). */ +export const SERVER_CLOCK_TTL_MS = 24 * 60 * 60 * 1000; + +const ServerClockJson = z.object({ offsetMs: z.number(), atMs: z.number() }).strict(); + +/** + * The highest physical candidate a seal may use now (journal spec §1.3, local + * clock ahead): the server's date, estimated from the last observed offset, + * plus `MAX_DRIFT`. Null when no server date from the last 24 h is known (or + * the record is unreadable): sealing is then unclamped. Read-only. + * + * @param db - The store. + * @param nowMs - This device's clock. + * @param maxDriftMs - The bound; defaults to {@link MAX_DRIFT_MS}. + */ +export function sealWallCeiling( + db: DatabaseSync, + nowMs: number, + maxDriftMs: number = MAX_DRIFT_MS, +): number | null { + if (!hasTable(db, '_sync_meta')) return null; + const row = db.prepare('SELECT value FROM _sync_meta WHERE key = ?').get(SERVER_CLOCK_KEY) as + | { value: string } + | undefined; + if (row === undefined) return null; + let parsed: z.infer; + try { + const r = ServerClockJson.safeParse(JSON.parse(row.value)); + if (!r.success) return null; + parsed = r.data; + } catch { + return null; + } + if (Math.abs(nowMs - parsed.atMs) > SERVER_CLOCK_TTL_MS) return null; + return nowMs + parsed.offsetMs + maxDriftMs; +} + /** Offset of the replica id inside an encoded HLC (13 + 1 + 6 + 1). */ const REPLICA_OFFSET = 21; diff --git a/packages/core/src/store/sync/push.ts b/packages/core/src/store/sync/push.ts index 3db8df7a9a..477077ed39 100644 --- a/packages/core/src/store/sync/push.ts +++ b/packages/core/src/store/sync/push.ts @@ -6,10 +6,11 @@ * 1. the preconditions: `sync.push` on, the stream cut and its genesis * checkpoint stored (`genesis_pending` clear), so no segment ever * precedes the checkpoint it extends; - * 2. the local-clock check: when this device's clock runs more than - * `MAX_DRIFT` ahead of the server's date, push pauses (persisted as - * `sync.clock_ahead` for `cleo doctor`), so future-stamped HLCs do not - * spread; local writes are never refused; + * 2. the local-clock check: the server's date is recorded for the sealer's + * clamp (`sync.server_clock`, read by `sealWallCeiling`); when this + * device's clock runs more than `MAX_DRIFT` ahead of it, push pauses + * (persisted as `sync.clock_ahead` for `cleo doctor`), so future-stamped + * HLCs do not spread; local writes are never refused; * 3. the rollback check: when the server already holds a later segment from * this replica than the store has persisted, the store is behind (a * restore or a copy) and must rebind (§1.5, §2.9); nothing is sent; @@ -29,6 +30,7 @@ import type { DatabaseSync } from 'node:sqlite'; import type { TableScope } from '@cleocode/contracts'; +import { SERVER_CLOCK_KEY } from './clock-store.js'; import { isSyncFlagOn } from './flags.js'; import { genesisCutOf, genesisPending } from './genesis.js'; import { storeHwm } from './replica.js'; @@ -117,15 +119,16 @@ const empty = (stream: string, fields: Partial): PushStreamRep }); /** Set or clear {@link CLOCK_AHEAD_KEY}. */ -function markClockAhead(db: DatabaseSync, value: string | null, atIso: string): void { +/** Set (or, with null, clear) one of push's `_sync_meta` keys. */ +function setPushMeta(db: DatabaseSync, key: string, value: string | null, atIso: string): void { if (value === null) { - db.prepare('DELETE FROM _sync_meta WHERE key = ?').run(CLOCK_AHEAD_KEY); + db.prepare('DELETE FROM _sync_meta WHERE key = ?').run(key); return; } db.prepare( 'INSERT INTO _sync_meta (key, value, updated_at) VALUES (?, ?, ?) ' + 'ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at', - ).run(CLOCK_AHEAD_KEY, value, atIso); + ).run(key, value, atIso); } /** @@ -157,14 +160,24 @@ export async function pushStream( refused: `${o.stream}'s genesis checkpoint is not stored yet: run \`cleo sync enable push\` again`, }); } - const atIso = new Date(now()).toISOString(); - // §1.3: a clock running ahead of the server's pauses push, never writes. + const nowMs = now(); + const atIso = new Date(nowMs).toISOString(); + // §1.3: the observed server date bounds what the sealer stamps + // (sealWallCeiling), and a clock running ahead of it pauses push, never writes. + if (o.serverOffsetMs !== null) { + setPushMeta( + db, + SERVER_CLOCK_KEY, + JSON.stringify({ offsetMs: o.serverOffsetMs, atMs: nowMs }), + atIso, + ); + } const drift = o.serverOffsetMs === null ? 0 : -o.serverOffsetMs; if (drift > (o.maxDriftMs ?? MAX_DRIFT_MS)) { - markClockAhead(db, JSON.stringify({ driftMs: drift }), atIso); + setPushMeta(db, CLOCK_AHEAD_KEY, JSON.stringify({ driftMs: drift }), atIso); return empty(o.stream, { clockAhead: true }); } - markClockAhead(db, null, atIso); + setPushMeta(db, CLOCK_AHEAD_KEY, null, atIso); // §1.5 / §2.9: the server is authoritative. A later segment there than this // store ever persisted means the store was restored or copied behind it. const local = storeHwm(db, o.replica)[o.stream] ?? null; diff --git a/packages/core/src/store/sync/sealer.ts b/packages/core/src/store/sync/sealer.ts index 50b48193da..768740293a 100644 --- a/packages/core/src/store/sync/sealer.ts +++ b/packages/core/src/store/sync/sealer.ts @@ -60,7 +60,7 @@ import { SECRET_MARKER, syncSetTables, } from './capture.js'; -import { tickClock, withImmediateTransaction } from './clock-store.js'; +import { sealWallCeiling, tickClock, withImmediateTransaction } from './clock-store.js'; import { actorOpOf, clearFieldLeaves, @@ -1079,6 +1079,10 @@ function sealInTransaction( const started = performance.now(); const maxMs = opts.maxMs ?? 200; const replica = opts.replica; + // §1.3: a capture stamped while this clock ran ahead seals no later than + // the server's date plus MAX_DRIFT, once a recent server date is known. + const ceiling = sealWallCeiling(db, now()); + const wall = (ms: number): number => (ceiling === null ? ms : Math.min(ms, ceiling)); // 1. Read a batch. A frame cut by the budget is completed only when it is // the batch's first group (it must seal whole); otherwise it is left for @@ -1336,7 +1340,7 @@ function sealInTransaction( ...(rest as Omit), seq, last, - h: tickClock(db, replica, at(op)?.at_ms ?? now()), + h: tickClock(db, replica, wall(at(op)?.at_ms ?? now())), }; }, ); From 1e0e9be094c653a17361bcfcf6db05aea68b4685 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:59:42 -0700 Subject: [PATCH 2/2] fix(sync): a clock set back since the server date was observed drops the stale offset (T13314 review LOW) When nowMs < atMs the clock was corrected after the observation; using the stale negative offset would stamp seals outside push up to |offset| in the past for a day. sealWallCeiling now returns null then. Test: an observation dated after now leaves sealing unclamped; removing the guard fails it. Co-Authored-By: Claude Opus 5.5 --- .changeset/t13314-seal-clock-clamp.md | 2 ++ .../core/src/store/sync/__tests__/push.test.ts | 14 +++++++++++++- packages/core/src/store/sync/clock-store.ts | 12 +++++++++--- 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/.changeset/t13314-seal-clock-clamp.md b/.changeset/t13314-seal-clock-clamp.md index 48c6f14b33..1f97e95024 100644 --- a/.changeset/t13314-seal-clock-clamp.md +++ b/.changeset/t13314-seal-clock-clamp.md @@ -12,4 +12,6 @@ wins every last-writer-wins race on every device. - Each push now records the server's clock offset and when it was observed (`sync.server_clock` in `_sync_meta`). - The sealer stamps each op no later than the server's estimated date plus `MAX_DRIFT` (5 minutes) when that observation is under 24 hours old. Without a recent observation, sealing is unclamped, as before. +- If this device's clock is now earlier than when the offset was observed, the clock was set back since. That is the + correction of a clock that ran ahead, so the old offset is dropped rather than stamping seals in the past. - HLCs stay monotonic. The clamp only bounds the physical candidate, and the counter still orders ops. diff --git a/packages/core/src/store/sync/__tests__/push.test.ts b/packages/core/src/store/sync/__tests__/push.test.ts index 2f2237632b..e84971b223 100644 --- a/packages/core/src/store/sync/__tests__/push.test.ts +++ b/packages/core/src/store/sync/__tests__/push.test.ts @@ -236,7 +236,7 @@ describe('pushStream (S4-2)', () => { } }); - it('a server date older than 24 h, or none, leaves sealing unclamped (T13314)', async () => { + it('a server date older than 24 h, one from before a clock set back, or none, leaves sealing unclamped (T13314)', async () => { const { db, replica } = await pushing(); const stampedAhead = (id: string) => { write(db, addTask(id)); @@ -278,6 +278,18 @@ describe('pushStream (S4-2)', () => { const second = stampedAhead('T2'); seal(); expect(maxPhys()).toBe(second); + // One observed after now: this clock was set back since (a correction), so the + // stale negative offset must not stamp seals in the past. + db.prepare( + 'INSERT INTO _sync_meta (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value', + ).run( + SERVER_CLOCK_KEY, + JSON.stringify({ offsetMs: -3_600_000, atMs: clock + 3_600_000 }), + new Date(clock).toISOString(), + ); + const third = stampedAhead('T3'); + seal(); + expect(maxPhys()).toBe(third); }); it('a server ahead of the store (restored or copied) is refused before anything is sealed or sent', async () => { diff --git a/packages/core/src/store/sync/clock-store.ts b/packages/core/src/store/sync/clock-store.ts index 027e0e2c32..9953988a37 100644 --- a/packages/core/src/store/sync/clock-store.ts +++ b/packages/core/src/store/sync/clock-store.ts @@ -51,8 +51,10 @@ const ServerClockJson = z.object({ offsetMs: z.number(), atMs: z.number() }).str /** * The highest physical candidate a seal may use now (journal spec §1.3, local * clock ahead): the server's date, estimated from the last observed offset, - * plus `MAX_DRIFT`. Null when no server date from the last 24 h is known (or - * the record is unreadable): sealing is then unclamped. Read-only. + * plus `MAX_DRIFT`. Null when no server date from the last 24 h is known, the + * record is unreadable, or this clock was set back since the observation (a + * corrected clock: the offset no longer holds): sealing is then unclamped. + * Read-only. * * @param db - The store. * @param nowMs - This device's clock. @@ -76,7 +78,11 @@ export function sealWallCeiling( } catch { return null; } - if (Math.abs(nowMs - parsed.atMs) > SERVER_CLOCK_TTL_MS) return null; + // A clock now earlier than when the offset was observed was set back since + // (the correction of a clock that ran ahead): the offset is stale, and + // using it would stamp seals in the past. Wait for the next server date. + if (nowMs < parsed.atMs) return null; + if (nowMs - parsed.atMs > SERVER_CLOCK_TTL_MS) return null; return nowMs + parsed.offsetMs + maxDriftMs; }