Convergence Guarantees

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

interface AntiEntropyProcess {
    // Run periodically (e.g., every 10 seconds)
    async synchronize(localNode: Node, remoteNode: Node): Promise<void> {
        // 1. Get a digest of local data (not the full data)
        const localDigest = await this.computeDigest(localNode);
        const remoteDigest = await remoteNode.getDigest();
        
        // 2. Compare digests efficiently (e.g., using Merkle trees)
        const differences = this.findDifferences(localDigest, remoteDigest);
        
        if (differences.length === 0) {
            return; // Already synchronized
        }
        
        // 3. For each difference, exchange actual data
        for (const key of differences) {
            const localValue = await localNode.get(key);
            const remoteValue = await remoteNode.get(key);
            
            // 4. Resolve conflicts using versioning/timestamps
            const resolved = this.resolveConflict(localValue, remoteValue);
            
            // 5. Update both nodes with resolved value
            if (resolved !== localValue) {
                await localNode.put(key, resolved);
            }
            if (resolved !== remoteValue) {
                await remoteNode.put(key, resolved);
            }
        }
    }
    
    // Conflict resolution strategies
    resolveConflict(v1: Value, v2: Value): Value {
        // Option 1: Last-Write-Wins (LWW)
        return v1.timestamp > v2.timestamp ? v1 : v2;
        
        // Option 2: Vector clock comparison
        // return this.mergeByVectorClock(v1, v2);
        
        // Option 3: Return both, let application decide
        // return { conflict: true, versions: [v1, v2] };
    }
}

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

class GossipProtocol {
    private nodeId: string;
    private state: Map<string, VersionedValue>;
    private peers: string[];
    private gossipInterval: number = 1000; // 1 second
    
    async startGossiping(): Promise<void> {
        setInterval(async () => {
            await this.gossipRound();
        }, this.gossipInterval);
    }
    
    private async gossipRound(): Promise<void> {
        // 1. Select random peer (fanout = 1 for simplicity)
        const peer = this.selectRandomPeer();
        if (!peer) return;
        
        // 2. Send our state digest
        const ourDigest = this.computeDigest();
        
        // 3. Exchange with peer
        const peerDigest = await this.send(peer, {
            type: 'GOSSIP_SYN',
            digest: ourDigest
        });
        
        // 4. Identify data we need from peer
        const needFromPeer = this.findNewerInRemote(ourDigest, peerDigest);
        
        // 5. Identify data peer needs from us
        const peerNeeds = this.findNewerInLocal(ourDigest, peerDigest);
        
        // 6. Exchange actual data (SYN-ACK)
        const peerData = await this.send(peer, {
            type: 'GOSSIP_SYN_ACK',
            request: needFromPeer,
            data: this.getValues(peerNeeds)
        });
        
        // 7. Update local state with peer's data
        this.mergeState(peerData);
        
        // 8. Send final ACK with any remaining data
        await this.send(peer, {
            type: 'GOSSIP_ACK',
            data: this.getValues(peerData.stillNeeded)
        });
    }
    
    private selectRandomPeer(): string | null {
        if (this.peers.length === 0) return null;
        const index = Math.floor(Math.random() * this.peers.length);
        return this.peers[index];
    }
    
    private computeDigest(): Map<string, number> {
        // Return key -> version map (not full data)
        const digest = new Map<string, number>();
        for (const [key, value] of this.state) {
            digest.set(key, value.version);
        }
        return digest;
    }
}

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

interface MerkleNode {
    hash: string;
    left?: MerkleNode;
    right?: MerkleNode;
    keyRange?: { min: string; max: string }; // For leaf nodes
}
 
class MerkleTree {
    private root: MerkleNode;
    private data: Map<string, string>; // key -> value
    
    constructor(data: Map<string, string>) {
        this.data = data;
        this.root = this.build(Array.from(data.keys()).sort());
    }
    
    private build(keys: string[]): MerkleNode {
        if (keys.length === 0) {
            return { hash: this.hash('empty') };
        }
        
        if (keys.length <= 10) { // Leaf node: small range of keys
            const content = keys
                .map(k => `${k}:${this.data.get(k)}`)
                .join('|');
            return {
                hash: this.hash(content),
                keyRange: { min: keys[0], max: keys[keys.length - 1] }
            };
        }
        
        // Internal node: split and recurse
        const mid = Math.floor(keys.length / 2);
        const left = this.build(keys.slice(0, mid));
        const right = this.build(keys.slice(mid));
        
        return {
            hash: this.hash(left.hash + right.hash),
            left,
            right
        };
    }
    
    // Find differences between two Merkle trees
    static findDifferences(local: MerkleNode, remote: MerkleNode): string[] {
        if (local.hash === remote.hash) {
            return []; // No differences in this subtree
        }
        
        // Leaf nodes: return the key range for detailed comparison
        if (local.keyRange) {
            return [local.keyRange.min + '-' + local.keyRange.max];
        }
        
        // Recurse into children
        const diffs: string[] = [];
        
        if (local.left && remote.left) {
            diffs.push(...this.findDifferences(local.left, remote.left));
        }
        if (local.right && remote.right) {
            diffs.push(...this.findDifferences(local.right, remote.right));
        }
        
        return diffs;
    }
    
    private hash(content: string): string {
        // In practice: use SHA-256 or similar
        return crypto.createHash('sha256').update(content).digest('hex');
    }
}
 
// Usage in anti-entropy:
// 1. Each replica maintains a Merkle tree of its data
// 2. During sync, compare root hashes
// 3. If different, traverse to find differing ranges
// 4. Exchange only the differing data

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

class ReadRepairCoordinator {
    private replicas: ReplicaNode[];
    private consistencyLevel: number; // How many replicas to read from
    
    async read(key: string): Promise<Value> {
        // 1. Send read requests to all replicas
        const responses = await Promise.allSettled(
            this.replicas.map(replica => 
                replica.read(key).then(value => ({ replica, value }))
            )
        );
        
        // 2. Collect successful responses
        const values = responses
            .filter(r => r.status === 'fulfilled')
            .map(r => (r as PromiseFulfilledResult<any>).value);
        
        // 3. Check if we have enough responses
        if (values.length < this.consistencyLevel) {
            throw new Error('Insufficient replicas available');
        }
        
        // 4. Find the most recent value (e.g., highest timestamp)
        const mostRecent = values.reduce((latest, current) =>
            current.value.timestamp > latest.value.timestamp 
                ? current 
                : latest
        );
        
        // 5. Identify replicas with stale data
        const staleReplicas = values.filter(
            v => v.value.timestamp < mostRecent.value.timestamp
        );
        
        // 6. Perform read repair asynchronously (don't block client)
        this.repairAsync(key, mostRecent.value, staleReplicas);
        
        // 7. Return most recent value to client
        return mostRecent.value;
    }
    
    private async repairAsync(
        key: string, 
        correctValue: Value, 
        staleReplicas: { replica: ReplicaNode; value: Value }[]
    ): Promise<void> {
        // Fire-and-forget repair requests
        for (const { replica } of staleReplicas) {
            replica.write(key, correctValue).catch(err => {
                // Log error but don't fail - background anti-entropy will catch it
                console.warn(`Read repair failed for ${replica.id}: ${err}`);
            });
        }
    }
}

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

interface Hint {
    targetNode: string;
    key: string;
    value: Value;
    timestamp: number;
    expiresAt: number; // Hints shouldn't be kept forever
}
 
class HintedHandoffManager {
    private hintStore: Map<string, Hint[]> = new Map(); // targetNode -> hints
    private hintTTL: number = 3 * 60 * 60 * 1000; // 3 hours default
    
    async writeWithHints(
        key: string, 
        value: Value, 
        replicas: string[]
    ): Promise<WriteResult> {
        const results = await Promise.allSettled(
            replicas.map(replica => this.writeToReplica(replica, key, value))
        );
        
        // Process results and create hints for failures
        for (let i = 0; i < results.length; i++) {
            if (results[i].status === 'rejected') {
                const failedReplica = replicas[i];
                this.storeHint({
                    targetNode: failedReplica,
                    key,
                    value,
                    timestamp: Date.now(),
                    expiresAt: Date.now() + this.hintTTL
                });
                console.log(`Stored hint for ${failedReplica}: ${key}`);
            }
        }
        
        // Check if we achieved required consistency level
        const successCount = results.filter(r => r.status === 'fulfilled').length;
        return { success: successCount >= this.writeConsistencyLevel };
    }
    
    private storeHint(hint: Hint): void {
        if (!this.hintStore.has(hint.targetNode)) {
            this.hintStore.set(hint.targetNode, []);
        }
        this.hintStore.get(hint.targetNode)!.push(hint);
    }
    
    // Called when gossip indicates a node has recovered
    async onNodeRecovered(nodeId: string): Promise<void> {
        const hints = this.hintStore.get(nodeId) || [];
        if (hints.length === 0) return;
        
        console.log(`Replaying ${hints.length} hints to ${nodeId}`);
        
        for (const hint of hints) {
            // Check if hint has expired
            if (Date.now() > hint.expiresAt) {
                console.log(`Hint expired for ${hint.key}, skipping`);
                continue;
            }
            
            try {
                await this.writeToReplica(hint.targetNode, hint.key, hint.value);
                console.log(`Successfully replayed hint for ${hint.key}`);
            } catch (error) {
                console.error(`Failed to replay hint for ${hint.key}: ${error}`);
                // Keep the hint for next attempt
                continue;
            }
        }
        
        // Clear delivered hints
        this.hintStore.delete(nodeId);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Write arrives
    │
    ├── Direct write to available replicas (immediate)
    │
    ├── Hinted Handoff for unavailable replicas
    │   └── Replayed when replica recovers (seconds-hours)
    │
    └── If inconsistency persists:
        │
        ├── Read Repair (on reads)
        │   └── Compares all replicas, fixes stale ones
        │
        └── Anti-Entropy Repair (background)
            ├── Full repair: Compares entire dataset using Merkle trees
            └── Incremental repair: Only repairs recently written data
 
Each layer catches failures that slip through earlier layers.
Together, they guarantee eventual convergence.