feat(sync): seal clamps the HLC physical candidate to server date + MAX_DRIFT (T13314)

.changeset/t13314-seal-clock-clamp.md

@@ -0,0 +1,17 @@
+---
+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.
+- 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.

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,85 @@ 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, 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));
+      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);
+    // 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 () => {
     const { db, replica } = await pushing();
     write(db, addTask('T1'));

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,56 @@ 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, 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.
+ * @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<typeof ServerClockJson>;
+  try {
+    const r = ServerClockJson.safeParse(JSON.parse(row.value));
+    if (!r.success) return null;
+    parsed = r.data;
+  } catch {
+    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;
+}
+
 /** Offset of the replica id inside an encoded HLC (13 + 1 + 6 + 1). */
 const REPLICA_OFFSET = 21;
 

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';
@@ -136,15 +138,16 @@ const empty = (stream: string, fields: Partial<PushStreamReport>): 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);
 }
 
 /**
@@ -182,14 +185,24 @@ export async function pushStream(
       refusedKind: 'genesis-pending',
     });
   }
-  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;

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,
@@ -1107,6 +1107,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
@@ -1364,7 +1368,7 @@ function sealInTransaction(
           ...(rest as Omit<SealedOp, 'h'>),
           seq,
           last,
-          h: tickClock(db, replica, at(op)?.at_ms ?? now()),
+          h: tickClock(db, replica, wall(at(op)?.at_ms ?? now())),
         };
       },
     );