Conflict-free Replicated Data Types (CRDTs)

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

// STATE-BASED CRDT: G-Counter (Grow-only Counter)
// Each node tracks its own count; merge takes max of each
 
interface GCounterState {
    [nodeId: string]: number;
}
 
class StateBasedGCounter {
    private nodeId: string;
    private state: GCounterState = {};
    
    constructor(nodeId: string) {
        this.nodeId = nodeId;
        this.state[nodeId] = 0;
    }
    
    increment(): void {
        this.state[this.nodeId] = (this.state[this.nodeId] ?? 0) + 1;
    }
    
    value(): number {
        return Object.values(this.state).reduce((a, b) => a + b, 0);
    }
    
    getState(): GCounterState {
        return { ...this.state };
    }
    
    // Merge: component-wise maximum
    merge(remoteState: GCounterState): void {
        for (const [nodeId, count] of Object.entries(remoteState)) {
            this.state[nodeId] = Math.max(
                this.state[nodeId] ?? 0, 
                count
            );
        }
    }
}
 
// OPERATION-BASED CRDT: G-Counter
// Operations are broadcast; each node applies same ops
 
type CounterOp = { type: 'increment'; nodeId: string; delta: number };
 
class OpBasedGCounter {
    private counts: Map<string, number> = new Map();
    
    // Apply operation (from local or remote)
    apply(op: CounterOp): void {
        if (op.type === 'increment') {
            const current = this.counts.get(op.nodeId) ?? 0;
            this.counts.set(op.nodeId, current + op.delta);
        }
    }
    
    value(): number {
        return Array.from(this.counts.values()).reduce((a, b) => a + b, 0);
    }
    
    // Generate increment operation
    increment(nodeId: string): CounterOp {
        return { type: 'increment', nodeId, delta: 1 };
    }
}
 
// For op-based: operations must be delivered exactly once
// For state-based: delivery can be unreliable (merge handles it)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Redis CRDB counter example
# EU region:
EU> INCRBY pageviews 100
(integer) 100
 
# US region (concurrent):
US> INCRBY pageviews 50
(integer) 50
 
# After sync, both regions see:
> GET pageviews
(integer) 150   # 100 + 50, not LWW!
 
# This is a CRDT counter, not a regular string.
 
# Sets with OR-Set semantics:
EU> SADD cart item1 item2
(integer) 2
 
US> SREM cart item1
(integer) 1
 
# After sync: cart contains {item2}
# The remove is observed and wins over the add

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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

/**
 * Production-grade PN-Counter CRDT
 * Supports increment, decrement, and merging across replicas.
 */
 
interface PNCounterState {
    p: { [nodeId: string]: number };  // Increments per node
    n: { [nodeId: string]: number };  // Decrements per node
}
 
export class PNCounter {
    private nodeId: string;
    private p: Map<string, number> = new Map();
    private n: Map<string, number> = new Map();
    
    constructor(nodeId: string) {
        this.nodeId = nodeId;
        this.p.set(nodeId, 0);
        this.n.set(nodeId, 0);
    }
    
    /**
     * Increment the counter by 1.
     */
    increment(): void {
        this.p.set(this.nodeId, (this.p.get(this.nodeId) ?? 0) + 1);
    }
    
    /**
     * Decrement the counter by 1.
     */
    decrement(): void {
        this.n.set(this.nodeId, (this.n.get(this.nodeId) ?? 0) + 1);
    }
    
    /**
     * Increment by a specific amount.
     */
    incrementBy(amount: number): void {
        if (amount < 0) throw new Error('Use decrement for negative values');
        this.p.set(this.nodeId, (this.p.get(this.nodeId) ?? 0) + amount);
    }
    
    /**
     * Decrement by a specific amount.
     */
    decrementBy(amount: number): void {
        if (amount < 0) throw new Error('Use increment for negative values');
        this.n.set(this.nodeId, (this.n.get(this.nodeId) ?? 0) + amount);
    }
    
    /**
     * Get the current value (sum of increments - sum of decrements).
     */
    value(): number {
        const pSum = Array.from(this.p.values()).reduce((a, b) => a + b, 0);
        const nSum = Array.from(this.n.values()).reduce((a, b) => a + b, 0);
        return pSum - nSum;
    }
    
    /**
     * Get the total number of increment operations (useful for analytics).
     */
    totalIncrements(): number {
        return Array.from(this.p.values()).reduce((a, b) => a + b, 0);
    }
    
    /**
     * Get the raw state for synchronization.
     */
    getState(): PNCounterState {
        return {
            p: Object.fromEntries(this.p),
            n: Object.fromEntries(this.n),
        };
    }
    
    /**
     * Merge another counter's state into this one.
     * This is the join operation—commutative, associative, idempotent.
     */
    merge(remoteState: PNCounterState): void {
        // Merge increments: take max for each node
        for (const [nodeId, count] of Object.entries(remoteState.p)) {
            const local = this.p.get(nodeId) ?? 0;
            this.p.set(nodeId, Math.max(local, count));
        }
        
        // Merge decrements: take max for each node
        for (const [nodeId, count] of Object.entries(remoteState.n)) {
            const local = this.n.get(nodeId) ?? 0;
            this.n.set(nodeId, Math.max(local, count));
        }
    }
    
    /**
     * Compare two states for debugging/testing.
     */
    equals(other: PNCounter): boolean {
        const thisState = this.getState();
        const otherState = other.getState();
        return JSON.stringify(thisState) === JSON.stringify(otherState);
    }
}
 
// Usage demonstration:
function demo() {
    // Two replicas
    const counterEU = new PNCounter('eu-node');
    const counterUS = new PNCounter('us-node');
    
    // Concurrent operations
    counterEU.increment();  // EU: +1
    counterEU.increment();  // EU: +1
    counterUS.increment();  // US: +1
    counterUS.decrementBy(2);  // US: -2
    
    console.log('Before merge:');
    console.log('EU value:', counterEU.value());  // 2
    console.log('US value:', counterUS.value());  // -1
    
    // Sync states
    counterEU.merge(counterUS.getState());
    counterUS.merge(counterEU.getState());
    
    console.log('After merge:');
    console.log('EU value:', counterEU.value());  // 1 (2 + 1 - 2)
    console.log('US value:', counterUS.value());  // 1 (converged!)
    console.log('States equal:', counterEU.equals(counterUS));  // true
}