Bloom Clocks, Hybrid Logical Clocks, and Causality Tracking in Distributed Systems

/*
 * Three relationships between events in a distributed system:
 *
 * 1. HAPPENS-BEFORE (→): A causally precedes B
 *    - A and B are on the same node, A occurs before B
 *    - A is a send event, B is the corresponding receive
 *    - Transitive: A → B and B → C implies A → C
 *
 * 2. CONCURRENT (||): Neither caused the other
 *    - A and B are on different nodes
 *    - No causal chain connects them
 *    - They may have happened at the same wall-clock time or not
 *
 * 3. SAME EVENT: A = B
 *
 * Example:
 *
 *   Node 1: ──A──────B──────D──────
 *               \          ↗
 *                \       /
 *   Node 2: ──────C────E──────F────
 *
 *   A → B (same node, A before B)
 *   A → C (A sent message received by C)
 *   C → E (same node)
 *   E → D (E sent message received by D)
 *   A → D (transitive: A→B, but also A→C→E→D)
 *   B || F (no causal chain: concurrent)
 *   B || C (no causal chain: concurrent)
 *
 * Why physical timestamps fail:
 *
 *   Node 1 clock: 10:00:00.000  Event A
 *   Node 2 clock: 10:00:00.005  Event C
 *   
 *   Does A → C? NOT necessarily.
 *   Node 2's clock could be 5ms ahead.
 *   A might have happened AFTER C in real time.
 *   NTP synchronization: ±10ms typical, ±100ms during correction
 */

/*
 * Lamport timestamp: single integer counter per node
 *
 * Rules:
 * 1. Before any event: counter++
 * 2. When sending message: attach counter value
 * 3. When receiving message: counter = max(local, received) + 1
 *
 * Property: if A → B then L(A) < L(B)
 * Limitation: if L(A) < L(B), A might NOT have caused B
 *             (could be concurrent)
 */

class LamportClock {
    constructor() {
        this.time = 0;
    }
    
    // Local event:
    tick() {
        this.time += 1;
        return this.time;
    }
    
    // Send event:
    send() {
        this.time += 1;
        return { timestamp: this.time, /* + message payload */ };
    }
    
    // Receive event:
    receive(messageTimestamp) {
        this.time = Math.max(this.time, messageTimestamp) + 1;
        return this.time;
    }
}

// Example trace:
// Node1: tick()=1, send()=2              → msg(ts=2) → Node2
// Node2: tick()=1, receive(2)=3, tick()=4
// Node1: tick()=3
//
// Node1 events: 1, 2, 3
// Node2 events: 1, 3, 4
//
// L(Node1.event2) = 2, L(Node2.event1) = 1
// 2 > 1, but Node1.event2 did NOT cause Node2.event1
// Lamport timestamps can't detect concurrency

/*
 * Lamport timestamps:
 *   ✓ Detect "happens-before" (if A→B then L(A)<L(B))
 *   ✗ Cannot detect concurrency (L(A)<L(B) doesn't mean A→B)
 *   ✓ O(1) space (single integer)
 *   ✓ O(1) comparison time
 *   Use: Total ordering where concurrency detection isn't needed
 *   Used by: Paxos, some distributed locks
 */

/*
 * Vector clock: one counter per node
 * Can detect BOTH causality AND concurrency
 *
 * Rules:
 * 1. Before event at node i: VC[i]++
 * 2. Send: attach entire vector
 * 3. Receive: VC[j] = max(VC_local[j], VC_received[j]) for all j, then VC[i]++
 *
 * Comparison:
 *   VC_A ≤ VC_B iff ∀i: VC_A[i] ≤ VC_B[i]  (A happened-before B)
 *   VC_A || VC_B iff ∃i: VC_A[i] > VC_B[i] AND ∃j: VC_B[j] > VC_A[j]
 *                  (concurrent — neither dominates)
 */

class VectorClock {
    constructor(nodeId, numNodes) {
        this.nodeId = nodeId;
        this.clock = new Array(numNodes).fill(0);
    }
    
    tick() {
        this.clock[this.nodeId]++;
        return [...this.clock];
    }
    
    send() {
        this.clock[this.nodeId]++;
        return [...this.clock]; // Copy to avoid mutation
    }
    
    receive(remoteClock) {
        for (let i = 0; i < this.clock.length; i++) {
            this.clock[i] = Math.max(this.clock[i], remoteClock[i]);
        }
        this.clock[this.nodeId]++;
        return [...this.clock];
    }
    
    static compare(vcA, vcB) {
        let aLessOrEqual = true;
        let bLessOrEqual = true;
        
        for (let i = 0; i < vcA.length; i++) {
            if (vcA[i] > vcB[i]) bLessOrEqual = false;
            if (vcB[i] > vcA[i]) aLessOrEqual = false;
        }
        
        if (aLessOrEqual && bLessOrEqual) return 'equal';
        if (aLessOrEqual) return 'before';       // A → B
        if (bLessOrEqual) return 'after';         // B → A
        return 'concurrent';                       // A || B
    }
}

/*
 * Example:
 *   3 nodes: N0, N1, N2
 *
 *   N0: tick() → [1,0,0]
 *   N0: send() → [2,0,0]  ─── msg ──▶ N1
 *   N1: tick() → [0,1,0]
 *   N1: receive([2,0,0]) → [2,2,0]
 *   N2: tick() → [0,0,1]
 *
 *   compare([2,0,0], [2,2,0]) → 'before'     (N0.send → N1.receive)
 *   compare([0,1,0], [0,0,1]) → 'concurrent' (N1.tick || N2.tick)
 *
 * Problem: O(N) space per event where N = number of nodes
 *   - 1000 nodes → 1000-element vector per event
 *   - 1M nodes → impractical
 *   - Dynamic membership (nodes join/leave) requires resizing
 */

/*
 * HLC: combines physical time + logical counter
 *
 * Format: (physical_time, logical_counter, node_id)
 *   - physical_time: wall clock (milliseconds)
 *   - logical_counter: breaks ties when physical time is equal
 *   - node_id: breaks ties when both are equal
 *
 * Properties:
 *   - If A → B then HLC(A) < HLC(B) (causality preserved)
 *   - Physical component stays close to real time (bounded drift)
 *   - O(1) space (not O(N) like vector clocks)
 *   - CANNOT detect concurrency (same limitation as Lamport)
 *   - But provides useful real-world timestamps
 *
 * Used by: CockroachDB, YugabyteDB, TiDB, MongoDB (logical session)
 */

class HybridLogicalClock {
    constructor(nodeId) {
        this.nodeId = nodeId;
        this.l = 0; // Logical physical time (max observed)
        this.c = 0; // Logical counter
    }
    
    now() {
        return Date.now(); // Physical wall clock
    }
    
    // Local or send event:
    tick() {
        const pt = this.now();
        
        if (pt > this.l) {
            // Physical clock advanced — reset counter:
            this.l = pt;
            this.c = 0;
        } else {
            // Physical clock hasn't advanced — increment counter:
            this.c += 1;
        }
        
        return { l: this.l, c: this.c, node: this.nodeId };
    }
    
    // Receive event:
    receive(remote) {
        const pt = this.now();
        
        if (pt > this.l && pt > remote.l) {
            // Local physical clock is ahead of both:
            this.l = pt;
            this.c = 0;
        } else if (this.l === remote.l) {
            // Same logical time — advance counter:
            this.c = Math.max(this.c, remote.c) + 1;
        } else if (remote.l > this.l) {
            // Remote is ahead — adopt remote time:
            this.l = remote.l;
            this.c = remote.c + 1;
        } else {
            // Local is ahead of remote:
            this.c += 1;
        }
        
        return { l: this.l, c: this.c, node: this.nodeId };
    }
    
    static compare(a, b) {
        if (a.l !== b.l) return a.l - b.l;
        if (a.c !== b.c) return a.c - b.c;
        return a.node < b.node ? -1 : a.node > b.node ? 1 : 0;
    }
}

/*
 * Key insight: HLC.l tracks the maximum physical time seen:
 *
 * Node A (clock at 100): tick() → {l:100, c:0}
 * Node B (clock at 90, slow): receive({l:100, c:0}) → {l:100, c:1}
 *   B's logical time jumps to 100 (adopts A's ahead clock)
 *
 * Node B (clock advances to 101): tick() → {l:101, c:0}
 *   Physical clock caught up → counter resets
 *
 * Bounded drift guarantee:
 *   |HLC.l - physical_time| ≤ max_clock_drift_in_cluster
 *   If NTP keeps clocks within 10ms, HLC.l is within 10ms of real time
 *
 * This is why CockroachDB can use HLC timestamps as MVCC versions
 * AND as real-world timestamps for time-travel queries.
 */

/*
 * CockroachDB uses HLC timestamps for MVCC and transactions:
 *
 * Every transaction gets an HLC timestamp:
 *   BEGIN → HLC timestamp = (physical_time, logical, node_id)
 *
 * Read:
 *   Read at timestamp T → return latest version where version.ts ≤ T
 *   If another transaction wrote at T' where T' > read_ts:
 *     → read timestamp is pushed forward (read refresh)
 *     → or transaction restarts with new timestamp
 *
 * Write:
 *   Write at timestamp T → store value with version T
 *   If another transaction already read at T' where T' ≥ T:
 *     → write intent is pushed to T' + 1
 *     → or write-write conflict → one transaction restarts
 *
 * Uncertainty interval:
 *   Because clocks aren't perfectly synchronized:
 *   A transaction at time T considers all writes in [T, T + max_offset]
 *   as potentially concurrent (uncertainty window)
 *
 *   max_offset = configured maximum clock skew (default: 500ms)
 *
 *   If a value is found in the uncertainty interval:
 *     → Transaction restarts with a higher timestamp
 *     → Guarantees linearizability despite clock skew
 *
 * Timeline:
 *   Node A (clock=100): BEGIN txn at HLC(100,0)
 *   Node A reads key K: version at HLC(95,0) → OK, 95 < 100
 *   Node A reads key L: version at HLC(100,3)
 *     100 is within uncertainty [100, 100+500ms]
 *     → Restart txn at HLC(100,4) to get consistent snapshot
 *
 * This is how CockroachDB achieves serializable isolation
 * across multiple nodes without a single timestamp oracle.
 */

/*
 * Problem: Vector clocks are O(N) where N = number of nodes
 * Bloom Clocks: compress vector clock into a fixed-size Bloom filter
 *
 * Instead of: [node0: 5, node1: 3, node2: 7, ..., nodeN: 2]  — O(N)
 * Use: Bloom filter of (nodeId, counter) pairs               — O(1) space
 *
 * Trade-off: false positives possible
 *   - May incorrectly report "happens-before" (false causal ordering)
 *   - Will NEVER miss a true "happens-before" (no false concurrency)
 *   - One-sided error: safe for detecting conflicts (conservative)
 */

class BloomClock {
    constructor(nodeId, size = 1024, numHashes = 3) {
        this.nodeId = nodeId;
        this.bits = new Uint8Array(size);  // Fixed size regardless of N
        this.size = size;
        this.numHashes = numHashes;
        this.localCounter = 0;
    }
    
    // Hash function for Bloom filter:
    hash(nodeId, counter, seed) {
        // Simplified — use MurmurHash or xxHash in production:
        let h = seed;
        const str = `${nodeId}:${counter}`;
        for (let i = 0; i < str.length; i++) {
            h = (h * 31 + str.charCodeAt(i)) & 0x7FFFFFFF;
        }
        return h % this.size;
    }
    
    // Add (nodeId, counter) to the Bloom clock:
    add(nodeId, counter) {
        for (let i = 0; i < this.numHashes; i++) {
            const idx = this.hash(nodeId, counter, i);
            this.bits[idx] = 1;
        }
    }
    
    // Check if (nodeId, counter) is in the Bloom clock:
    contains(nodeId, counter) {
        for (let i = 0; i < this.numHashes; i++) {
            const idx = this.hash(nodeId, counter, i);
            if (this.bits[idx] === 0) return false;
        }
        return true; // Possibly present (false positive possible)
    }
    
    // Tick (local event):
    tick() {
        this.localCounter++;
        this.add(this.nodeId, this.localCounter);
    }
    
    // Merge (receive remote Bloom clock):
    merge(remoteBits) {
        for (let i = 0; i < this.size; i++) {
            this.bits[i] = this.bits[i] | remoteBits[i]; // Bitwise OR
        }
        this.localCounter++;
        this.add(this.nodeId, this.localCounter);
    }
    
    // Compare (check if this happened-before other):
    happenedBefore(other) {
        // this ≤ other iff all bits set in this are also set in other
        for (let i = 0; i < this.size; i++) {
            if (this.bits[i] === 1 && other.bits[i] === 0) {
                return false; // This has info that other doesn't
            }
        }
        return true;
    }
}

/*
 * Bloom Clock comparison:
 *
 * BC_A ⊆ BC_B → A might have happened before B
 *   (could be false positive due to Bloom filter)
 * BC_A ⊄ BC_B AND BC_B ⊄ BC_A → definitely concurrent
 *   (no false negatives for concurrency detection)
 *
 * Error characteristics:
 *   - False "happens-before": possible (Bloom filter FP)
 *   - False "concurrent": impossible (Bloom filter has no FN)
 *   - For conflict detection: conservative (may flag non-conflicts)
 *   - FP rate depends on filter size and number of events
 *
 * Space: O(m) where m = Bloom filter size (fixed, e.g., 1KB)
 * vs Vector Clock: O(N) where N = number of nodes (grows with system)
 *
 * Use when: N is very large (thousands of nodes) and occasional
 * false conflict detection is acceptable
 */

/*
 * Dotted Version Vectors (DVV): optimized vector clocks for databases
 *
 * Problem with basic vector clocks:
 *   Each client update creates a new entry in the vector
 *   1000 clients → 1000-element vector per key
 *
 * Solution: DVV tracks versions per SERVER node, not per client
 *   - Only N entries where N = number of replica nodes (typically 3-5)
 *   - Clients provide context (causal history) with writes
 *
 * Used by: Riak (DVVSets)
 */

class DottedVersionVector {
    constructor() {
        // Base vector: { nodeId: maxContiguousCounter }
        this.base = {};
        // Dot: the specific (node, counter) for this event
        this.dot = null;
    }
    
    // Create a new version on a node:
    static create(nodeId, base) {
        const dvv = new DottedVersionVector();
        dvv.base = { ...base };
        const nextCounter = (base[nodeId] || 0) + 1;
        dvv.base[nodeId] = nextCounter;
        dvv.dot = { node: nodeId, counter: nextCounter };
        return dvv;
    }
    
    // Check if this version dominates (happened after) another:
    dominates(other) {
        for (const [node, counter] of Object.entries(other.base)) {
            if ((this.base[node] || 0) < counter) return false;
        }
        if (other.dot) {
            const baseCounter = this.base[other.dot.node] || 0;
            if (baseCounter < other.dot.counter) return false;
        }
        return true;
    }
    
    // Merge multiple versions (for conflict resolution):
    static merge(versions) {
        const merged = new DottedVersionVector();
        for (const v of versions) {
            for (const [node, counter] of Object.entries(v.base)) {
                merged.base[node] = Math.max(merged.base[node] || 0, counter);
            }
            if (v.dot) {
                merged.base[v.dot.node] = Math.max(
                    merged.base[v.dot.node] || 0,
                    v.dot.counter
                );
            }
        }
        return merged;
    }
}

/*
 * DVV vs Vector Clock:
 *
 * Vector Clock: O(C) where C = number of clients (unbounded)
 * DVV: O(N) where N = number of server replicas (fixed, small)
 *
 * DVV separates:
 *   - Causal context (base vector): what the client has seen
 *   - Event identity (dot): which replica created this version
 *
 * This enables sibling detection without tracking every client.
 */

/*
 * Interval Tree Clock: dynamic membership vector clock
 *
 * Problem: Vector clocks require knowing N (number of nodes) upfront
 * ITC: nodes can fork (split) and join (merge) dynamically
 *
 * Each node owns an "ID" (interval in [0,1]) and an "event" (counter tree)
 *
 * Fork: split ID interval in half
 *   Node A owns [0,1] → fork → A owns [0,0.5], B owns [0.5,1]
 *
 * Join: merge intervals back
 *   A [0,0.5] + B [0.5,1] → C [0,1]
 *
 * Event: counter associated with owned interval
 *   Only the owner of an interval can increment its counter
 */

class ITCStamp {
    constructor(id, event) {
        this.id = id;       // Ownership interval (tree structure)
        this.event = event; // Counter (tree structure)
    }
    
    // Fork: create two stamps from one
    static fork(stamp) {
        const [id1, id2] = ITCStamp.splitId(stamp.id);
        return [
            new ITCStamp(id1, stamp.event), // Clone event to both
            new ITCStamp(id2, stamp.event),
        ];
    }
    
    // Join: merge two stamps into one
    static join(s1, s2) {
        const mergedId = ITCStamp.mergeId(s1.id, s2.id);
        const mergedEvent = ITCStamp.mergeEvent(s1.event, s2.event);
        return new ITCStamp(mergedId, mergedEvent);
    }
    
    // Event: increment counter (only on owned interval)
    tick() {
        // Increment the counter on the interval this stamp owns
        this.event = ITCStamp.incrementEvent(this.event, this.id);
        return this;
    }
    
    // Compare: causality check
    static leq(e1, e2) {
        // e1 ≤ e2 iff all counters in e1 ≤ corresponding counters in e2
        return ITCStamp.eventLeq(e1.event, e2.event);
    }
    
    // Simplified ID splitting:
    static splitId(id) {
        if (typeof id === 'number') {
            return [{ l: id, r: 0 }, { l: 0, r: id }];
        }
        // Tree-based splitting for nested intervals
        return [id.l || 0, id.r || 0];
    }
    
    static mergeId(id1, id2) {
        // Union of intervals
        return { l: id1, r: id2 };
    }
    
    static mergeEvent(e1, e2) {
        // Point-wise max
        if (typeof e1 === 'number' && typeof e2 === 'number') {
            return Math.max(e1, e2);
        }
        return { base: Math.max(e1.base || 0, e2.base || 0) };
    }
    
    static incrementEvent(event, id) {
        return (typeof event === 'number') ? event + 1 : { ...event, base: (event.base || 0) + 1 };
    }
    
    static eventLeq(e1, e2) {
        if (typeof e1 === 'number' && typeof e2 === 'number') {
            return e1 <= e2;
        }
        return (e1.base || 0) <= (e2.base || 0);
    }
}

/*
 * ITC advantages:
 *   - Dynamic: nodes can join/leave without global coordination
 *   - No fixed N: interval subdivides as needed
 *   - Fork/join is local (no broadcast)
 *   - Detects causality and concurrency (like vector clocks)
 *
 * ITC disadvantages:
 *   - More complex implementation than vector clocks
 *   - Deep nesting possible with many forks
 *   - Less widely implemented in production systems
 *
 * Used in: distributed garbage collection, dynamic P2P systems
 */

/*
 * Complete comparison of causality tracking mechanisms:
 *
 * ┌─────────────────┬────────┬──────────┬───────────┬───────────────┐
 * │ Mechanism        │ Space  │ Causality│ Concurrcy │ Dynamic Nodes │
 * ├─────────────────┼────────┼──────────┼───────────┼───────────────┤
 * │ Lamport Clock    │ O(1)   │ Partial  │ No        │ Yes           │
 * │ Vector Clock     │ O(N)   │ Full     │ Yes       │ No (fixed N)  │
 * │ HLC              │ O(1)   │ Partial  │ No        │ Yes           │
 * │ Dotted V. Vec.   │ O(N)*  │ Full     │ Yes       │ Yes (servers) │
 * │ Bloom Clock      │ O(m)†  │ Approx   │ Yes‡      │ Yes           │
 * │ Interval Tree    │ O(k)§  │ Full     │ Yes       │ Yes (fork/join│
 * │ Version Vector   │ O(N)   │ Full     │ Yes       │ No            │
 * └─────────────────┴────────┴──────────┴───────────┴───────────────┘
 *
 * * N = number of server replicas (small, fixed)
 * † m = Bloom filter size (fixed, configurable)
 * ‡ Conservative: may report false causality, never false concurrency
 * § k = depth of interval tree (grows with forks)
 *
 * When to use each:
 *
 * Lamport Clock:
 *   → Total ordering needed, concurrency detection not needed
 *   → Example: log ordering, distributed lock sequencing
 *
 * Vector Clock:
 *   → Small, fixed number of nodes (< 100)
 *   → Need exact causality AND concurrency detection
 *   → Example: Dynamo-style databases, optimistic replication
 *
 * HLC:
 *   → Need real-world timestamps + causal ordering
 *   → Don't need concurrency detection
 *   → Example: MVCC databases, time-travel queries
 *   → Used by: CockroachDB, YugabyteDB
 *
 * Bloom Clock:
 *   → Very large number of nodes (thousands+)
 *   → Approximate concurrency detection is sufficient
 *   → Example: large-scale P2P, IoT sensor networks
 *
 * Dotted Version Vector:
 *   → Database with many clients but few replicas
 *   → Need sibling detection per key
 *   → Example: Riak, distributed KV stores
 *
 * Interval Tree Clock:
 *   → Highly dynamic membership (nodes join/leave frequently)
 *   → P2P systems, mobile networks
 */

// Using HLC for event ordering in a microservices system:

class EventStore {
    constructor(nodeId) {
        this.hlc = new HybridLogicalClock(nodeId);
        this.events = [];
    }
    
    // Publish an event with HLC timestamp:
    async publish(eventType, payload, causalParent = null) {
        let timestamp;
        
        if (causalParent) {
            // This event was caused by receiving another event:
            timestamp = this.hlc.receive(causalParent.timestamp);
        } else {
            // Independent event:
            timestamp = this.hlc.tick();
        }
        
        const event = {
            id: generateUUID(),
            type: eventType,
            payload,
            timestamp,
            // HLC timestamp is both:
            // 1. Causally consistent (if A→B, ts(A) < ts(B))
            // 2. Close to real time (useful for queries)
        };
        
        await this.store(event);
        await this.broadcast(event);
        return event;
    }
    
    // Query events in causal order:
    async getEvents(after, before) {
        // HLC timestamps can be compared like regular timestamps
        // but preserve causal ordering:
        return this.events
            .filter(e => 
                HybridLogicalClock.compare(e.timestamp, after) > 0 &&
                HybridLogicalClock.compare(e.timestamp, before) < 0
            )
            .sort((a, b) => HybridLogicalClock.compare(a.timestamp, b.timestamp));
    }
    
    // Detect potential conflicts:
    async detectConflicts(key, localVersion, remoteVersion) {
        // For systems that need conflict detection, use version vectors:
        const comparison = VectorClock.compare(
            localVersion.vectorClock,
            remoteVersion.vectorClock
        );
        
        switch (comparison) {
            case 'before':
                return { conflict: false, winner: 'remote' };
            case 'after':
                return { conflict: false, winner: 'local' };
            case 'equal':
                return { conflict: false, winner: 'same' };
            case 'concurrent':
                return {
                    conflict: true,
                    local: localVersion,
                    remote: remoteVersion,
                    // Application must resolve this conflict
                };
        }
    }
}

/*
 * Production patterns:
 *
 * 1. Event sourcing with HLC:
 *    → Events stored with HLC timestamps
 *    → Replay in causal order across services
 *    → Time-travel queries using HLC physical component
 *
 * 2. CRDT with vector clocks:
 *    → Each CRDT operation tagged with vector clock
 *    → Merge operations use vector comparison for ordering
 *    → Concurrent operations resolved by CRDT merge function
 *
 * 3. Optimistic replication with DVV:
 *    → Each key-value pair has a DVV
 *    → Write conflicts detected by DVV comparison
 *    → Siblings stored until application resolves conflict
 */