Implementation Approaches

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

class LamportClock {
  private timestamp: number = 0;
  
  // Rule 1: Tick before any local operation
  tick(): number {
    return ++this.timestamp;
  }
  
  // Rule 2: Get timestamp to send with message
  send(): number {
    return this.tick();
  }
  
  // Rule 3: Update clock on message receive
  receive(messageTimestamp: number): number {
    this.timestamp = Math.max(this.timestamp, messageTimestamp) + 1;
    return this.timestamp;
  }
  
  // Current value (read-only)
  current(): number {
    return this.timestamp;
  }
}
 
// Example usage
const clockA = new LamportClock();
const clockB = new LamportClock();
 
// Process A performs local operation
const a1 = clockA.tick();  // a1 = 1
 
// Process A sends message to B
const msgTs = clockA.send();  // msgTs = 2
const b1 = clockB.receive(msgTs);  // b1 = 3
 
// Process B sends reply to A
const replyTs = clockB.send();  // replyTs = 4
const a2 = clockA.receive(replyTs);  // a2 = 5
 
// Causal chain: a1(1) → msg(2) → b1(3) → reply(4) → a2(5)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

class VectorClock {
  private clock: Map<string, number>;
  private processId: string;
  
  constructor(processId: string) {
    this.processId = processId;
    this.clock = new Map();
  }
  
  // Increment own counter before local operation
  tick(): VectorClock {
    const current = this.clock.get(this.processId) || 0;
    this.clock.set(this.processId, current + 1);
    return this.clone();
  }
  
  // Merge with received clock: component-wise max, then tick
  receive(other: VectorClock): void {
    for (const [id, value] of other.clock) {
      const current = this.clock.get(id) || 0;
      this.clock.set(id, Math.max(current, value));
    }
    this.tick();
  }
  
  // Compare two vector clocks
  compare(other: VectorClock): 'before' | 'after' | 'concurrent' | 'equal' {
    let hasLess = false;
    let hasGreater = false;
    
    const allKeys = new Set([...this.clock.keys(), ...other.clock.keys()]);
    
    for (const key of allKeys) {
      const thisVal = this.clock.get(key) || 0;
      const otherVal = other.clock.get(key) || 0;
      
      if (thisVal < otherVal) hasLess = true;
      if (thisVal > otherVal) hasGreater = true;
    }
    
    if (!hasLess && !hasGreater) return 'equal';
    if (hasLess && !hasGreater) return 'before';
    if (!hasLess && hasGreater) return 'after';
    return 'concurrent';  // Both hasLess and hasGreater are true
  }
  
  clone(): VectorClock {
    const cloned = new VectorClock(this.processId);
    for (const [id, value] of this.clock) {
      cloned.clock.set(id, value);
    }
    return cloned;
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

interface VersionedValue<T> {
  value: T;
  version: VersionVector;
}
 
class VersionVector {
  private versions: Map<string, number>;  // replicaId -> version number
  
  constructor() {
    this.versions = new Map();
  }
  
  // Increment version for a specific replica (when that replica writes)
  increment(replicaId: string): void {
    const current = this.versions.get(replicaId) || 0;
    this.versions.set(replicaId, current + 1);
  }
  
  // Merge two version vectors (after reconciliation)
  merge(other: VersionVector): VersionVector {
    const merged = new VersionVector();
    const allReplicas = new Set([
      ...this.versions.keys(),
      ...other.versions.keys()
    ]);
    
    for (const replicaId of allReplicas) {
      merged.versions.set(
        replicaId,
        Math.max(
          this.versions.get(replicaId) || 0,
          other.versions.get(replicaId) || 0
        )
      );
    }
    return merged;
  }
  
  // Determine relationship
  descends(ancestor: VersionVector): boolean {
    // Returns true if this version is a descendant of ancestor
    for (const [replicaId, version] of ancestor.versions) {
      if ((this.versions.get(replicaId) || 0) < version) {
        return false;
      }
    }
    return true;
  }
  
  // Detect conflicts (concurrent updates)
  conflicts(other: VersionVector): boolean {
    const thisDescends = this.descends(other);
    const otherDescends = other.descends(this);
    
    // Conflict if neither descends from the other
    return !thisDescends && !otherDescends;
  }
}

Scenario: Alice and Bob both edit a document offline
 
Initial state at all replicas:
  value = "Hello"
  version = {A: 1, B: 0}  (originated at replica A)
 
Alice's replica (A):
  - Alice edits: "Hello World"
  - Version becomes: {A: 2, B: 0}
 
Bob's replica (B):  
  - Bob edits (from same original): "Hello Everyone"
  - Version becomes: {A: 1, B: 1}
 
When replicas synchronize:
  Alice's version: {A: 2, B: 0}
  Bob's version:   {A: 1, B: 1}
  
  Comparison:
  - Alice's A=2 > Bob's A=1 (Alice is ahead on A)
  - Alice's B=0 < Bob's B=1 (Bob is ahead on B)
  - Neither descends from the other → CONFLICT DETECTED
  
Resolution options:
  1. Last-write-wins (using wall clock, may lose Bob's data)
  2. Merge the changes (application-specific)
  3. Create siblings: keep both, let next client resolve

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

class HybridLogicalClock {
  // l: physical time portion (coarse-grained wall clock)
  // c: logical counter (for events in the same physical time)
  private l: number;
  private c: number;
  
  constructor() {
    this.l = Date.now();  // Physical time in milliseconds
    this.c = 0;
  }
  
  // Get current HLC timestamp
  now(): { l: number; c: number } {
    const pt = Date.now();  // Physical time
    
    if (pt > this.l) {
      // Physical time has advanced - reset counter
      this.l = pt;
      this.c = 0;
    } else {
      // Physical time hasn't advanced - increment counter
      this.c++;
    }
    
    return { l: this.l, c: this.c };
  }
  
  // Receive a timestamp from another node
  receive(remote: { l: number; c: number }): { l: number; c: number } {
    const pt = Date.now();
    
    if (pt > this.l && pt > remote.l) {
      // Physical time is ahead of both - use it
      this.l = pt;
      this.c = 0;
    } else if (this.l === remote.l) {
      // Same physical time as remote - take max counter + 1
      this.c = Math.max(this.c, remote.c) + 1;
    } else if (this.l > remote.l) {
      // We're ahead - just increment our counter
      this.c++;
    } else {
      // Remote is ahead - adopt remote time, increment counter
      this.l = remote.l;
      this.c = remote.c + 1;
    }
    
    return { l: this.l, c: this.c };
  }
  
  // Compare two HLC timestamps
  static compare(a: { l: number; c: number }, b: { l: number; c: number }): number {
    if (a.l !== b.l) return a.l - b.l;
    return a.c - b.c;
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

interface CausalWrite {
  key: string;
  value: any;
  
  // The write's unique identifier
  writeId: string;
  
  // Dependencies: what this write causally depends on
  // Could be version vectors, specific write IDs, or other markers
  dependencies: VersionVector;
  
  // Timestamp for ordering among independent writes
  timestamp: HLC;
}
 
class CausalReplica {
  private data: Map<string, VersionedValue>;
  private pendingWrites: CausalWrite[] = [];  // Writes waiting for deps
  private appliedVersions: VersionVector;     // What we've applied
  
  // Receive a write from replication
  receiveWrite(write: CausalWrite): void {
    if (this.dependenciesSatisfied(write)) {
      this.applyWrite(write);
      this.tryApplyPending();  // Some pending writes might now be ready
    } else {
      // Buffer until dependencies arrive
      this.pendingWrites.push(write);
    }
  }
  
  private dependenciesSatisfied(write: CausalWrite): boolean {
    // Check if we've seen everything this write depends on
    return this.appliedVersions.descends(write.dependencies);
  }
  
  private applyWrite(write: CausalWrite): void {
    // Apply the write to our local state
    this.data.set(write.key, {
      value: write.value,
      version: this.appliedVersions.merge(write.dependencies)
    });
    
    // Update our version vector to reflect this write
    this.appliedVersions.increment(write.writeId);
  }
  
  private tryApplyPending(): void {
    let applied = true;
    while (applied) {
      applied = false;
      for (let i = this.pendingWrites.length - 1; i >= 0; i--) {
        if (this.dependenciesSatisfied(this.pendingWrites[i])) {
          this.applyWrite(this.pendingWrites[i]);
          this.pendingWrites.splice(i, 1);
          applied = true;
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

// Causal broadcast implementation using vector clocks
class CausalBroadcast {
  private nodeId: string;
  private clock: VectorClock;
  private pending: Map<string, PendingMessage[]>;  // Per-sender pending queues
  private delivered: Set<string>;  // Message IDs we've delivered
  
  constructor(nodeId: string) {
    this.nodeId = nodeId;
    this.clock = new VectorClock(nodeId);
    this.pending = new Map();
    this.delivered = new Set();
  }
  
  // Broadcast a message to all nodes
  broadcast(message: any): BroadcastMessage {
    const timestamp = this.clock.tick();
    
    const broadcastMsg: BroadcastMessage = {
      id: `${this.nodeId}-${timestamp.get(this.nodeId)}`,
      sender: this.nodeId,
      payload: message,
      vectorClock: timestamp.clone()
    };
    
    // Send to all other nodes (implementation depends on network layer)
    this.sendToAll(broadcastMsg);
    
    // Deliver locally immediately
    this.deliverLocally(broadcastMsg);
    
    return broadcastMsg;
  }
  
  // Receive a broadcast message from another node
  receive(msg: BroadcastMessage): void {
    if (this.delivered.has(msg.id)) {
      return;  // Already delivered
    }
    
    if (this.canDeliver(msg)) {
      this.deliver(msg);
    } else {
      // Buffer until we can deliver
      this.bufferPending(msg);
    }
  }
  
  private canDeliver(msg: BroadcastMessage): boolean {
    // Can deliver if we've seen all messages this one depends on
    // Sender's component should be exactly one more than what we have
    // All other components should be <= what we have
    
    const senderComponent = msg.vectorClock.get(msg.sender);
    const ourSenderProgress = this.clock.get(msg.sender) || 0;
    
    if (senderComponent !== ourSenderProgress + 1) {
      return false;  // Missing earlier messages from this sender
    }
    
    for (const [nodeId, value] of msg.vectorClock.entries()) {
      if (nodeId !== msg.sender) {
        const ourValue = this.clock.get(nodeId) || 0;
        if (value > ourValue) {
          return false;  // Missing messages from another sender
        }
      }
    }
    
    return true;
  }
  
  private deliver(msg: BroadcastMessage): void {
    // Update our clock
    this.clock.receive(msg.vectorClock);
    this.delivered.add(msg.id);
    
    // Deliver to application
    this.deliverToApplication(msg.payload);
    
    // Check if any pending messages can now be delivered
    this.processPending();
  }
  
  private processPending(): void {
    let delivered = true;
    while (delivered) {
      delivered = false;
      for (const [sender, messages] of this.pending) {
        for (let i = 0; i < messages.length; i++) {
          if (this.canDeliver(messages[i].msg)) {
            this.deliver(messages[i].msg);
            messages.splice(i, 1);
            delivered = true;
            break;
          }
        }
      }
    }
  }
}