Async Replication Patterns

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

// Primary-Replica Async Replication Implementation
 
interface ReplicationEvent {
    sequenceNumber: number;
    timestamp: Date;
    operation: 'INSERT' | 'UPDATE' | 'DELETE';
    table: string;
    data: Record<string, any>;
}
 
class PrimaryNode {
    private sequenceNumber: number = 0;
    private writeAheadLog: ReplicationEvent[] = [];
    private replicas: ReplicaNode[] = [];
    private replicationQueue: ReplicationEvent[] = [];
 
    async write(operation: WriteOperation): Promise<WriteResult> {
        // 1. Execute write locally
        const result = await this.executeLocally(operation);
        
        // 2. Append to WAL (durably before acknowledging)
        const event = this.createReplicationEvent(operation);
        await this.appendToWAL(event);
        
        // 3. Acknowledge to client IMMEDIATELY (async replication)
        // Replication happens in background
        
        // 4. Queue for async replication
        this.replicationQueue.push(event);
        this.triggerAsyncReplication();
        
        return result; // Client doesn't wait for replication
    }
 
    private async triggerAsyncReplication(): Promise<void> {
        // Non-blocking: replicate in background
        setImmediate(async () => {
            while (this.replicationQueue.length > 0) {
                const event = this.replicationQueue.shift()!;
                
                // Fan out to all replicas (fire and forget)
                await Promise.allSettled(
                    this.replicas.map(replica => 
                        replica.receiveReplicationEvent(event)
                    )
                );
            }
        });
    }
 
    getReplicationLag(): Map<string, number> {
        const lags = new Map<string, number>();
        
        for (const replica of this.replicas) {
            const lastApplied = replica.getLastAppliedSequence();
            const lag = this.sequenceNumber - lastApplied;
            lags.set(replica.id, lag);
        }
        
        return lags;
    }
}
 
class ReplicaNode {
    id: string;
    private lastAppliedSequence: number = 0;
    private replicationBuffer: ReplicationEvent[] = [];
 
    async receiveReplicationEvent(event: ReplicationEvent): Promise<void> {
        // Buffer events that arrive out of order
        this.replicationBuffer.push(event);
        this.replicationBuffer.sort((a, b) => a.sequenceNumber - b.sequenceNumber);
        
        // Apply events in order
        while (this.replicationBuffer.length > 0) {
            const next = this.replicationBuffer[0];
            
            if (next.sequenceNumber === this.lastAppliedSequence + 1) {
                await this.applyEvent(next);
                this.lastAppliedSequence = next.sequenceNumber;
                this.replicationBuffer.shift();
            } else if (next.sequenceNumber <= this.lastAppliedSequence) {
                // Duplicate - discard
                this.replicationBuffer.shift();
            } else {
                // Gap - wait for missing events
                break;
            }
        }
    }
 
    getLastAppliedSequence(): number {
        return this.lastAppliedSequence;
    }
}

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

// Semi-Synchronous Replication Implementation
 
type SyncMode = 'ASYNC' | 'SEMI_SYNC_ONE' | 'SEMI_SYNC_MAJORITY' | 'FULLY_SYNC';
 
class SemiSyncPrimary {
    private replicas: ReplicaNode[] = [];
    private syncMode: SyncMode = 'SEMI_SYNC_ONE';
    private syncTimeout: number = 500; // ms
 
    async write(operation: WriteOperation): Promise<WriteResult> {
        // Execute locally and log
        const result = await this.executeAndLog(operation);
        const event = this.getLastEvent();
 
        // Determine how many acks we need
        const requiredAcks = this.calculateRequiredAcks();
 
        if (requiredAcks === 0) {
            // Async mode - fire and forget
            this.replicateAsync(event);
            return result;
        }
 
        // Wait for required acknowledgments (with timeout)
        try {
            await this.waitForAcks(event, requiredAcks, this.syncTimeout);
        } catch (error) {
            // Timeout - decide what to do
            if (this.shouldFallbackToAsync()) {
                console.warn('Semi-sync timeout - falling back to async');
                // Write is durable on primary, proceed anyway
            } else {
                throw new ReplicationTimeoutError('Failed to replicate');
            }
        }
 
        return result;
    }
 
    private async waitForAcks(
        event: ReplicationEvent,
        required: number,
        timeoutMs: number
    ): Promise<void> {
        return new Promise((resolve, reject) => {
            let ackCount = 0;
            const timeout = setTimeout(() => {
                reject(new Error('Replication timeout'));
            }, timeoutMs);
 
            // Send to all replicas in parallel
            this.replicas.forEach(async (replica) => {
                try {
                    await replica.receiveAndAck(event);
                    ackCount++;
                    
                    if (ackCount >= required) {
                        clearTimeout(timeout);
                        resolve();
                    }
                } catch (error) {
                    // Replica failed - others might succeed
                }
            });
        });
    }
 
    private calculateRequiredAcks(): number {
        switch (this.syncMode) {
            case 'ASYNC': return 0;
            case 'SEMI_SYNC_ONE': return 1;
            case 'SEMI_SYNC_MAJORITY': return Math.floor(this.replicas.length / 2) + 1;
            case 'FULLY_SYNC': return this.replicas.length;
        }
    }
}

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

// Multi-Primary Replication with Conflict Detection
 
interface VectorClock {
    [nodeId: string]: number;
}
 
interface VersionedRecord {
    data: Record<string, any>;
    vectorClock: VectorClock;
    lastModifiedBy: string;
}
 
class MultiPrimaryNode {
    id: string;
    private vectorClock: VectorClock = {};
    private store: Map<string, VersionedRecord> = new Map();
    private pendingConflicts: Map<string, VersionedRecord[]> = new Map();
 
    async write(key: string, data: Record<string, any>): Promise<WriteResult> {
        // Increment our position in the vector clock
        this.vectorClock[this.id] = (this.vectorClock[this.id] || 0) + 1;
 
        const record: VersionedRecord = {
            data,
            vectorClock: { ...this.vectorClock },
            lastModifiedBy: this.id
        };
 
        this.store.set(key, record);
        
        // Replicate to other primaries
        await this.replicateToPeers(key, record);
        
        return { success: true, version: record.vectorClock };
    }
 
    async receiveReplication(key: string, incomingRecord: VersionedRecord): Promise<void> {
        const existing = this.store.get(key);
 
        if (!existing) {
            // No conflict - just store
            this.store.set(key, incomingRecord);
            this.mergeVectorClock(incomingRecord.vectorClock);
            return;
        }
 
        const relationship = this.compareVectorClocks(
            existing.vectorClock,
            incomingRecord.vectorClock
        );
 
        switch (relationship) {
            case 'BEFORE':
                // Incoming is newer - replace
                this.store.set(key, incomingRecord);
                this.mergeVectorClock(incomingRecord.vectorClock);
                break;
 
            case 'AFTER':
                // Existing is newer - ignore incoming
                break;
 
            case 'CONCURRENT':
                // CONFLICT! Both modified concurrently
                await this.handleConflict(key, existing, incomingRecord);
                break;
        }
    }
 
    private compareVectorClocks(
        a: VectorClock, 
        b: VectorClock
    ): 'BEFORE' | 'AFTER' | 'CONCURRENT' {
        let aBeforeB = false;
        let bBeforeA = false;
 
        const allNodes = new Set([...Object.keys(a), ...Object.keys(b)]);
 
        for (const node of allNodes) {
            const aVal = a[node] || 0;
            const bVal = b[node] || 0;
 
            if (aVal < bVal) aBeforeB = true;
            if (bVal < aVal) bBeforeA = true;
        }
 
        if (aBeforeB && !bBeforeA) return 'BEFORE';
        if (bBeforeA && !aBeforeB) return 'AFTER';
        return 'CONCURRENT'; // Both have changes the other doesn't
    }
 
    private async handleConflict(
        key: string,
        local: VersionedRecord,
        incoming: VersionedRecord
    ): Promise<void> {
        // Strategy 1: Last-Write-Wins (simple but loses data)
        // Strategy 2: Store both versions, let application resolve
        // Strategy 3: Automatic merge (if data structure supports it)
 
        // Example: Store both for later resolution
        this.pendingConflicts.set(key, [local, incoming]);
        
        console.warn(`Conflict detected for key ${key}`);
    }
}