Multi-Leader Replication

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

interface PhysicalTimestampedValue<T> {
  value: T;
  timestamp: number;        // milliseconds since epoch
  sourceNodeId: string;     // for tie-breaking
}
 
class PhysicalLWWStore<T> {
  private data = new Map<string, PhysicalTimestampedValue<T>>();
  
  constructor(private readonly nodeId: string) {}
  
  // Local write: generate timestamp from wall clock
  write(key: string, value: T): PhysicalTimestampedValue<T> {
    const entry: PhysicalTimestampedValue<T> = {
      value,
      timestamp: Date.now(),     // Wall-clock time
      sourceNodeId: this.nodeId
    };
    
    return this.applyWrite(key, entry);
  }
  
  // Replicated write: apply incoming entry if it wins
  applyReplicatedWrite(key: string, incoming: PhysicalTimestampedValue<T>): boolean {
    const current = this.data.get(key);
    
    if (!current) {
      this.data.set(key, incoming);
      return true; // Applied
    }
    
    if (this.isNewer(incoming, current)) {
      this.data.set(key, incoming);
      return true; // Applied
    }
    
    return false; // Rejected (current is newer)
  }
  
  private isNewer(
    a: PhysicalTimestampedValue<T>, 
    b: PhysicalTimestampedValue<T>
  ): boolean {
    if (a.timestamp !== b.timestamp) {
      return a.timestamp > b.timestamp;
    }
    // Tie-breaker: lexicographic node ID comparison
    return a.sourceNodeId > b.sourceNodeId;
  }
  
  private applyWrite(key: string, entry: PhysicalTimestampedValue<T>): PhysicalTimestampedValue<T> {
    const current = this.data.get(key);
    
    if (!current || this.isNewer(entry, current)) {
      this.data.set(key, entry);
      return entry;
    }
    
    // Rare: our own write lost to existing entry (shouldn't happen in normal operation)
    return current;
  }
}

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

/**
 * Hybrid Logical Clock (HLC) combines physical time with logical counter.
 * Properties:
 * - Maintains causality: if A → B, then HLC(A) < HLC(B)
 * - Stays close to physical time
 * - Monotonically increasing within a node
 */
interface HybridTimestamp {
  // Physical component: wall-clock time, used as primary ordering
  physical: number;
  // Logical component: breaks ties when physical times are equal
  logical: number;
  // Source node ID for cross-node tie-breaking
  nodeId: string;
}
 
class HybridLogicalClock {
  private lastPhysical: number = 0;
  private logical: number = 0;
  
  constructor(private readonly nodeId: string) {}
  
  // Generate timestamp for a local event
  now(): HybridTimestamp {
    const wallClock = Date.now();
    
    if (wallClock > this.lastPhysical) {
      // Wall clock has advanced: use it, reset logical
      this.lastPhysical = wallClock;
      this.logical = 0;
    } else {
      // Wall clock hasn't advanced (or went backwards!): increment logical
      this.logical++;
    }
    
    return {
      physical: this.lastPhysical,
      logical: this.logical,
      nodeId: this.nodeId
    };
  }
  
  // Update clock based on received timestamp (maintains causality)
  receive(remote: HybridTimestamp): HybridTimestamp {
    const wallClock = Date.now();
    
    if (wallClock > this.lastPhysical && wallClock > remote.physical) {
      // Wall clock is ahead of both: use it
      this.lastPhysical = wallClock;
      this.logical = 0;
    } else if (remote.physical > this.lastPhysical) {
      // Remote is ahead: adopt remote's physical, increment logical
      this.lastPhysical = remote.physical;
      this.logical = remote.logical + 1;
    } else if (this.lastPhysical > remote.physical) {
      // Local is ahead: keep local physical, increment logical
      this.logical++;
    } else {
      // Physical times equal: take max logical + 1
      this.logical = Math.max(this.logical, remote.logical) + 1;
    }
    
    return {
      physical: this.lastPhysical,
      logical: this.logical,
      nodeId: this.nodeId
    };
  }
}
 
// HLC Comparison for LWW
function compareHLC(a: HybridTimestamp, b: HybridTimestamp): number {
  if (a.physical !== b.physical) return a.physical - b.physical;
  if (a.logical !== b.logical) return a.logical - b.logical;
  return a.nodeId.localeCompare(b.nodeId);
}

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

-- Cassandra cell-level LWW example
 
-- Each column update carries a timestamp
-- These can come from different coordinators at different times
 
-- Node 1 writes at T=1000
INSERT INTO users (user_id, name, email) 
VALUES ('alice', 'Alice Smith', 'alice@old.com')
USING TIMESTAMP 1000000000000;
 
-- Node 2 writes at T=1200 (200μs later, different column)
INSERT INTO users (user_id, name, email)
VALUES ('alice', 'Alice Williams', 'alice@new.com')
USING TIMESTAMP 1000000000200;
 
-- Result after replication:
-- user_id: 'alice'
-- name: 'Alice Williams' (T=1200 wins)
-- email: 'alice@new.com' (T=1200 wins)
 
-- If only email was updated at T=1200:
UPDATE users SET email = 'alice@new.com'
WHERE user_id = 'alice'
USING TIMESTAMP 1000000000200;
 
-- Result:
-- name: 'Alice Smith' (T=1000, never updated)
-- email: 'alice@new.com' (T=1200 wins for this cell)

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

interface WriteRecord<T> {
  value: T;
  timestamp: number;
  nodeId: string;
  writeId: string;  // UUID generated per write
}
 
// Strategy 1: Node ID ordering (simple but biased)
function tieBreakByNodeId<T>(a: WriteRecord<T>, b: WriteRecord<T>): WriteRecord<T> {
  return a.nodeId > b.nodeId ? a : b;
}
 
// Strategy 2: Write ID ordering (no bias, requires UUIDs)
function tieBreakByWriteId<T>(a: WriteRecord<T>, b: WriteRecord<T>): WriteRecord<T> {
  return a.writeId > b.writeId ? a : b;
}
 
// Strategy 3: Value hash (content-dependent, deterministic)
function tieBreakByValueHash<T>(a: WriteRecord<T>, b: WriteRecord<T>): WriteRecord<T> {
  const hashA = computeHash(JSON.stringify(a.value));
  const hashB = computeHash(JSON.stringify(b.value));
  return hashA > hashB ? a : b;
}
 
// Strategy 4: Composite (combine multiple criteria)
function tieBreakComposite<T>(a: WriteRecord<T>, b: WriteRecord<T>): WriteRecord<T> {
  // First: compare timestamps (already equal, but future-proofing)
  if (a.timestamp !== b.timestamp) {
    return a.timestamp > b.timestamp ? a : b;
  }
  
  // Second: prefer certain node types (e.g., primary datacenter)
  const priorityA = getNodePriority(a.nodeId);
  const priorityB = getNodePriority(b.nodeId);
  if (priorityA !== priorityB) {
    return priorityA > priorityB ? a : b;
  }
  
  // Third: fall back to write ID
  return a.writeId > b.writeId ? a : b;
}
 
function getNodePriority(nodeId: string): number {
  // Example: primary datacenter has higher priority
  if (nodeId.startsWith('primary-')) return 100;
  if (nodeId.startsWith('secondary-')) return 50;
  return 10;
}
 
function computeHash(input: string): string {
  // Use consistent hash function (e.g., SHA-256)
  // Simplified for illustration
  return input.split('').reduce((acc, char) => {
    return ((acc << 5) - acc) + char.charCodeAt(0);
  }, 0).toString(16);
}

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

interface ConflictEvent<T> {
  key: string;
  winner: WriteRecord<T>;
  loser: WriteRecord<T>;
  timestampDelta: number;    // How close were they?
  detectedAt: Date;
  resolvedBy: 'lww' | 'tie-breaker';
}
 
class LWWWithLogging<T> {
  private conflictLog: ConflictEvent<T>[] = [];
  
  resolve(key: string, local: WriteRecord<T>, incoming: WriteRecord<T>): WriteRecord<T> {
    if (local.timestamp === incoming.timestamp) {
      // Tie: log and use tie-breaker
      const winner = tieBreakByWriteId(local, incoming);
      const loser = winner === local ? incoming : local;
      
      this.logConflict(key, winner, loser, 'tie-breaker');
      return winner;
    }
    
    const winner = incoming.timestamp > local.timestamp ? incoming : local;
    const loser = winner === local ? incoming : local;
    
    // Only log if timestamps were close (potential clock skew issue)
    const delta = Math.abs(incoming.timestamp - local.timestamp);
    if (delta < 1000) { // Within 1 second
      this.logConflict(key, winner, loser, 'lww');
    }
    
    return winner;
  }
  
  private logConflict(
    key: string, 
    winner: WriteRecord<T>, 
    loser: WriteRecord<T>,
    resolvedBy: 'lww' | 'tie-breaker'
  ) {
    this.conflictLog.push({
      key,
      winner,
      loser,
      timestampDelta: Math.abs(winner.timestamp - loser.timestamp),
      detectedAt: new Date(),
      resolvedBy
    });
    
    // Alert if conflict rate is high
    this.checkConflictRate();
  }
  
  private checkConflictRate() {
    const recentConflicts = this.conflictLog.filter(
      c => c.detectedAt.getTime() > Date.now() - 60000 // Last minute
    );
    
    if (recentConflicts.length > 100) {
      console.warn('High conflict rate detected:', recentConflicts.length, 'in last minute');
      // Trigger alert, investigation
    }
  }
}