Hybrid Logical Clocks & TrueTime - Commit Wait, Uncertainty Intervals & External Consistency

# Hybrid Logical Clock (Kulkarni et al., 2014): timestamp = (l, c).
#  l tracks the largest physical time seen; c is a small counter that breaks ties
#  when physical time has not moved past l. Causality is preserved like Lamport,
#  but l stays within the clock-skew bound of real time, so it is human readable.
from dataclasses import dataclass

MAX_OFFSET_MS = 500  # like CockroachDB's default --max-offset (example guard value)

@dataclass
class HLC:
    name: str
    skew_ms: int        # this node's physical clock error (example values)
    l: int = 0
    c: int = 0

    def pt(self, true_ms: int) -> int:
        return true_ms + self.skew_ms

    def now(self, true_ms: int):
        """Local event or send."""
        p = self.pt(true_ms)
        if p > self.l:
            self.l, self.c = p, 0
        else:
            self.c += 1
        return (self.l, self.c)

    def recv(self, true_ms: int, m):
        ml, mc = m
        p = self.pt(true_ms)
        if ml - p > MAX_OFFSET_MS:   # sender is implausibly far in the future: refuse
            raise ValueError(f"{self.name}: remote ts {ml} is {ml-p} ms ahead of local clock")
        new_l = max(self.l, ml, p)
        if new_l == self.l == ml:   self.c = max(self.c, mc) + 1
        elif new_l == self.l:       self.c += 1
        elif new_l == ml:           self.c = mc + 1
        else:                       self.c = 0
        self.l = new_l
        return (self.l, self.c)

A = HLC("A", skew_ms=0)
B = HLC("B", skew_ms=-40)   # 40 ms slow
C = HLC("C", skew_ms=+80)   # 80 ms fast

trace = []
m = C.now(1000);             trace.append(("C send", 1000, m))
r = A.recv(1005, m);         trace.append(("A recv", 1005, r))
m2 = A.now(1006);            trace.append(("A send", 1006, m2))
r2 = B.recv(1010, m2);       trace.append(("B recv", 1010, r2))
r3 = B.now(1011);            trace.append(("B local", 1011, r3))
r4 = B.now(1200);            trace.append(("B local", 1200, r4))

print(f"{'event':8} {'true':>5} {'phys':>5}  hlc(l,c)")
for ev, t, ts in trace:
    node = {"A": A, "B": B, "C": C}[ev[0]]
    print(f"{ev:8} {t:>5} {node.pt(t):>5}  {ts}")
print("causal order preserved:", trace[0][2] < trace[1][2] < trace[2][2] < trace[3][2] < trace[4][2])
print("max (l - true time) observed:", max(ts[0] - t for _, t, ts in trace), "ms (bounded by worst skew)")

# A node whose clock jumped 10 s ahead must not drag the whole cluster's HLC with it.
bad = HLC("D", skew_ms=10_000)
try:
    A.recv(1300, bad.now(1300))
except ValueError as e:
    print("guard:", e)

event     true  phys  hlc(l,c)
C send    1000  1080  (1080, 0)
A recv    1005  1005  (1080, 1)
A send    1006  1006  (1080, 2)
B recv    1010   970  (1080, 3)
B local   1011   971  (1080, 4)
B local   1200  1160  (1160, 0)
causal order preserved: True
max (l - true time) observed: 80 ms (bounded by worst skew)
guard: A: remote ts 11300 is 10000 ms ahead of local clock

// Part 1: Spanner-style commit wait with a TrueTime-like interval clock (simulated).
// TT.now() returns [earliest, latest]; true time is guaranteed to be inside.
// Part 2: CockroachDB-style read uncertainty restart with a max clock offset.
// All epsilon / offset values are EXAMPLE values.

class TrueTimeSim {
  constructor(public trueMs: number, private epsMs: number) {}
  now(): { earliest: number; latest: number } {
    return { earliest: this.trueMs - this.epsMs, latest: this.trueMs + this.epsMs };
  }
  sleep(ms: number): void { this.trueMs += ms; }
}

function commit(tt: TrueTimeSim): { ts: number; waitedMs: number } {
  const s = tt.now().latest;          // pick commit ts no earlier than any possible "now"
  let waited = 0;
  while (tt.now().earliest <= s) {    // commit wait: until s is definitely in the past
    tt.sleep(0.5); waited += 0.5;
  }
  return { ts: s, waitedMs: waited }; // only now release locks / acknowledge the client
}

console.log("Part 1: commit wait ~ 2 x epsilon");
for (const eps of [1, 4, 7]) {
  const tt = new TrueTimeSim(10_000, eps);
  const t1 = commit(tt);
  // T2 starts strictly after T1 was acknowledged, possibly on another machine.
  const t2 = commit(tt);
  console.log(`  eps=${eps}ms  T1.ts=${t1.ts} waited ${t1.waitedMs}ms  T2.ts=${t2.ts}  T1<T2: ${t1.ts < t2.ts}`);
}

console.log("\nPart 2: uncertainty interval restart (max offset 500 ms)");
const MAX_OFFSET = 500;
// Versions of key k written by other nodes, stamped with THEIR clocks.
const versions = [{ ts: 9_700, v: "old" }, { ts: 10_300, v: "new" }];
function read(readTs: number): string {
  let ts = readTs;
  for (;;) {
    const uncertaintyLimit = ts + MAX_OFFSET;
    const visible = versions.filter((x) => x.ts <= ts).sort((a, b) => b.ts - a.ts)[0];
    const uncertain = versions.find((x) => x.ts > ts && x.ts <= uncertaintyLimit);
    if (!uncertain) return `read@${ts} -> ${visible?.v ?? "nothing"}`;
    // The 'future' write may really have happened before us (its writer's clock ran ahead),
    // so we cannot ignore it: bump our read timestamp past it and retry.
    console.log(`  read@${ts}: value at ${uncertain.ts} is within uncertainty window (${ts}, ${uncertaintyLimit}] -> restart`);
    ts = uncertain.ts;
  }
}
console.log(" ", read(10_000));
console.log(" ", read(11_000));

Part 1: commit wait ~ 2 x epsilon
  eps=1ms  T1.ts=10001 waited 2.5ms  T2.ts=10003.5  T1<T2: true
  eps=4ms  T1.ts=10004 waited 8.5ms  T2.ts=10012.5  T1<T2: true
  eps=7ms  T1.ts=10007 waited 14.5ms  T2.ts=10021.5  T1<T2: true

Part 2: uncertainty interval restart (max offset 500 ms)
  read@10000: value at 10300 is within uncertainty window (10000, 10500] -> restart
  read@10300 -> new
  read@11000 -> new