Distributed Transactions

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

// Transaction Record maintained by the Coordinator
interface TransactionRecord {
    // Unique identifier for the distributed transaction
    transactionId: string;
    
    // Current state in the coordinator's state machine
    state: 'INITIAL' | 'WAIT' | 'COMMIT' | 'ABORT' | 'END';
    
    // List of all participants in this transaction
    participants: ParticipantInfo[];
    
    // Timestamp when commit process began
    prepareStartTime: number;
    
    // Timeout value for vote collection
    voteTimeout: number;
    
    // Received votes from participants
    votes: Map<string, 'VOTE_COMMIT' | 'VOTE_ABORT'>;
    
    // Acknowledgments received from participants
    acknowledgments: Set<string>;
    
    // The global decision once determined
    decision?: 'COMMIT' | 'ABORT';
}
 
// Information about each participant
interface ParticipantInfo {
    // Unique identifier for the participant
    participantId: string;
    
    // Network address for communication
    endpoint: string;
    
    // Whether we've received their vote
    hasVoted: boolean;
    
    // The vote received (if any)
    vote?: 'VOTE_COMMIT' | 'VOTE_ABORT';
    
    // Whether they've acknowledged the decision
    hasAcknowledged: boolean;
    
    // Number of retry attempts for this participant
    retryCount: number;
}
 
// The Coordinator's persistent transaction table
class CoordinatorTransactionManager {
    // In-memory transaction table (also persisted to log)
    private activeTransactions: Map<string, TransactionRecord>;
    
    // Pending timer handles for timeouts
    private timeoutHandles: Map<string, Timer>;
    
    // Durable log for recovery
    private durableLog: DurableLog;
}

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

class Coordinator {
    /**
     * Initiate the prepare phase for a distributed transaction
     */
    async initiateCommit(txId: string): Promise<'COMMIT' | 'ABORT'> {
        const txRecord = this.activeTransactions.get(txId);
        if (!txRecord || txRecord.state !== 'INITIAL') {
            throw new Error(`Invalid transaction state for commit: ${txId}`);
        }
        
        // Step 1: Force-write prepare record to durable log
        await this.durableLog.forceWrite({
            type: 'PREPARE',
            transactionId: txId,
            participants: txRecord.participants.map(p => p.participantId),
            timestamp: Date.now()
        });
        
        // Step 2: Transition to WAIT state
        txRecord.state = 'WAIT';
        txRecord.prepareStartTime = Date.now();
        
        // Step 3: Start timeout timer
        this.startVoteTimeout(txId);
        
        // Step 4: Send PREPARE to all participants (parallel)
        const preparePromises = txRecord.participants.map(p =>
            this.sendPrepare(p, txId)
        );
        
        // Don't await these - we'll collect votes as they arrive
        // Fire and forget the PREPARE messages
        
        // Step 5: Wait for votes (with timeout)
        const decision = await this.collectVotes(txId);
        
        // Step 6: Execute commit phase
        return this.executeDecision(txId, decision);
    }
    
    /**
     * Collect votes from all participants
     */
    private async collectVotes(txId: string): Promise<'COMMIT' | 'ABORT'> {
        const txRecord = this.activeTransactions.get(txId)!;
        
        return new Promise((resolve) => {
            const checkComplete = () => {
                // Check if all votes received
                if (txRecord.votes.size === txRecord.participants.length) {
                    // All votes in - check if unanimous COMMIT
                    const allCommit = Array.from(txRecord.votes.values())
                        .every(v => v === 'VOTE_COMMIT');
                    
                    this.clearVoteTimeout(txId);
                    resolve(allCommit ? 'COMMIT' : 'ABORT');
                    return true;
                }
                
                // Check for any ABORT vote (early termination)
                if (Array.from(txRecord.votes.values()).includes('VOTE_ABORT')) {
                    this.clearVoteTimeout(txId);
                    resolve('ABORT');
                    return true;
                }
                
                return false;
            };
            
            // Register vote handler
            this.voteHandlers.set(txId, (participantId, vote) => {
                txRecord.votes.set(participantId, vote);
                txRecord.participants.find(p => p.participantId === participantId)!.hasVoted = true;
                checkComplete();
            });
            
            // Register timeout handler
            this.timeoutHandlers.set(txId, () => {
                // Timeout - not all votes received, must abort
                resolve('ABORT');
            });
        });
    }
}

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

class Coordinator {
    /**
     * Execute the commit or abort decision
     */
    async executeDecision(txId: string, decision: 'COMMIT' | 'ABORT'): Promise<'COMMIT' | 'ABORT'> {
        const txRecord = this.activeTransactions.get(txId)!;
        
        // CRITICAL: Force-write decision to log BEFORE sending to participants
        await this.durableLog.forceWrite({
            type: decision,
            transactionId: txId,
            timestamp: Date.now()
        });
        
        // Update in-memory state
        txRecord.state = decision;
        txRecord.decision = decision;
        
        // Determine which participants need notification
        // For COMMIT: all participants
        // For ABORT: only those that voted COMMIT (others already aborted)
        const participantsToNotify = decision === 'COMMIT'
            ? txRecord.participants
            : txRecord.participants.filter(p => p.vote === 'VOTE_COMMIT');
        
        // Send decision to all relevant participants
        for (const participant of participantsToNotify) {
            this.sendDecision(participant, txId, decision);
        }
        
        // Start acknowledgment collection phase
        this.startAckPhase(txId, participantsToNotify);
        
        // Wait for all acknowledgments (with retry loop)
        await this.collectAcknowledgments(txId, participantsToNotify);
        
        // All acknowledged - log END and clean up
        await this.completeTransaction(txId);
        
        return decision;
    }
    
    /**
     * Collect acknowledgments with retry for reliability
     */
    private async collectAcknowledgments(
        txId: string, 
        participants: ParticipantInfo[]
    ): Promise<void> {
        const txRecord = this.activeTransactions.get(txId)!;
        
        while (true) {
            // Check if all have acknowledged
            const allAcked = participants.every(p => 
                txRecord.acknowledgments.has(p.participantId)
            );
            
            if (allAcked) {
                return; // All done
            }
            
            // Find who hasn't acknowledged
            const unackedParticipants = participants.filter(p =>
                !txRecord.acknowledgments.has(p.participantId)
            );
            
            // Retry sending decision to unacknowledged participants
            for (const participant of unackedParticipants) {
                participant.retryCount++;
                
                if (participant.retryCount > MAX_RETRIES) {
                    // Participant seems permanently failed
                    // Log this and consider them acknowledged
                    // (when they recover, they'll query us)
                    this.log.warn(`Participant ${participant.participantId} not responding`);
                    txRecord.acknowledgments.add(participant.participantId);
                    continue;
                }
                
                // Retry sending decision
                this.sendDecision(participant, txId, txRecord.decision!);
            }
            
            // Wait before next retry round
            await this.sleep(RETRY_INTERVAL * Math.min(participant.retryCount, 5));
        }
    }
    
    /**
     * Complete the transaction - log END and clean up
     */
    private async completeTransaction(txId: string): Promise<void> {
        // Log transaction end
        await this.durableLog.forceWrite({
            type: 'END',
            transactionId: txId,
            timestamp: Date.now()
        });
        
        // Clean up in-memory structures
        this.activeTransactions.delete(txId);
        this.timeoutHandles.delete(txId);
        this.voteHandlers.delete(txId);
    }
}

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

class TimeoutManager {
    private timers: Map<string, NodeJS.Timeout> = new Map();
    private deadlines: Map<string, number> = new Map(); // Monotonic time
    
    /**
     * Start a timeout for vote collection
     */
    startVoteTimeout(txId: string, timeoutMs: number, onTimeout: () => void): void {
        // Record deadline using monotonic time
        const deadline = this.getMonotonicTime() + timeoutMs;
        this.deadlines.set(txId, deadline);
        
        // Schedule callback
        const handle = setTimeout(() => {
            this.timers.delete(txId);
            this.deadlines.delete(txId);
            onTimeout();
        }, timeoutMs);
        
        this.timers.set(txId, handle);
    }
    
    /**
     * Cancel a pending timeout
     */
    cancelTimeout(txId: string): void {
        const handle = this.timers.get(txId);
        if (handle) {
            clearTimeout(handle);
            this.timers.delete(txId);
            this.deadlines.delete(txId);
        }
    }
    
    /**
     * On recovery, recalculate remaining timeout
     */
    resumeTimeout(txId: string, originalDeadline: number, onTimeout: () => void): void {
        const now = this.getMonotonicTime();
        const remaining = Math.max(0, originalDeadline - now);
        
        if (remaining === 0) {
            // Already expired
            onTimeout();
        } else {
            // Resume with remaining time
            this.startVoteTimeout(txId, remaining, onTimeout);
        }
    }
    
    /**
     * Get monotonic time (not affected by wall-clock adjustments)
     */
    private getMonotonicTime(): number {
        // In Node.js: process.hrtime()
        // In browser: performance.now()
        return performance.now();
    }
}

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

class CoordinatorRecovery {
    /**
     * Recover coordinator state after crash
     */
    async recover(): Promise<void> {
        console.log('Starting coordinator recovery...');
        
        // Step 1: Scan log and categorize transactions
        const transactions = await this.classifyTransactions();
        
        // Step 2: Handle each category
        for (const [txId, state] of transactions) {
            await this.recoverTransaction(txId, state);
        }
        
        console.log('Coordinator recovery complete');
    }
    
    /**
     * Scan log and classify transaction states
     */
    private async classifyTransactions(): Promise<Map<string, TransactionState>> {
        const transactions = new Map<string, TransactionState>();
        
        // Read log from checkpoint
        const logRecords = await this.durableLog.readFromCheckpoint();
        
        for (const record of logRecords) {
            switch (record.type) {
                case 'PREPARE':
                    transactions.set(record.transactionId, {
                        phase: 'WAIT',
                        participants: record.participants,
                        decision: undefined
                    });
                    break;
                    
                case 'COMMIT':
                    const commitTx = transactions.get(record.transactionId);
                    if (commitTx) {
                        commitTx.decision = 'COMMIT';
                    }
                    break;
                    
                case 'ABORT':
                    const abortTx = transactions.get(record.transactionId);
                    if (abortTx) {
                        abortTx.decision = 'ABORT';
                    }
                    break;
                    
                case 'END':
                    // Transaction complete - remove from recovery list
                    transactions.delete(record.transactionId);
                    break;
            }
        }
        
        return transactions;
    }
    
    /**
     * Recover a single transaction
     */
    private async recoverTransaction(txId: string, state: TransactionState): Promise<void> {
        if (state.decision === 'COMMIT') {
            // Decision was COMMIT - need to complete it
            console.log(`Recovering COMMIT for ${txId}`);
            await this.completeCommit(txId, state.participants);
            
        } else if (state.decision === 'ABORT') {
            // Decision was ABORT - need to notify voting COMMIT participants
            console.log(`Recovering ABORT for ${txId}`);
            await this.completeAbort(txId, state.participants);
            
        } else {
            // No decision - crashed during vote collection
            // Safe choice: ABORT
            console.log(`No decision found for ${txId} - aborting`);
            await this.durableLog.forceWrite({ type: 'ABORT', transactionId: txId });
            await this.completeAbort(txId, state.participants);
        }
    }
    
    /**
     * Handle participant query about transaction outcome
     */
    handleDecisionQuery(txId: string): 'COMMIT' | 'ABORT' {
        // Look up in recovered transactions
        const state = this.activeTransactions.get(txId);
        
        if (state?.decision === 'COMMIT') {
            return 'COMMIT';
        }
        
        // No record or explicit ABORT - return ABORT
        // (Presumed Abort)
        return 'ABORT';
    }
}

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

class GroupCommitLog {
    private pendingRecords: LogRecord[] = [];
    private waiters: Map<string, () => void> = new Map();
    private flushTimer: NodeJS.Timeout | null = null;
    
    private readonly MAX_BATCH_SIZE = 100;
    private readonly MAX_WAIT_MS = 10;
    
    /**
     * Write a log record with group commit optimization
     */
    async writeWithGroupCommit(record: LogRecord): Promise<void> {
        return new Promise((resolve) => {
            // Add record to pending batch
            this.pendingRecords.push(record);
            this.waiters.set(record.transactionId, resolve);
            
            // Check if batch is full
            if (this.pendingRecords.length >= this.MAX_BATCH_SIZE) {
                this.flushBatch();
                return;
            }
            
            // Start timer if not already running
            if (!this.flushTimer) {
                this.flushTimer = setTimeout(
                    () => this.flushBatch(), 
                    this.MAX_WAIT_MS
                );
            }
        });
    }
    
    /**
     * Flush pending records to disk
     */
    private async flushBatch(): Promise<void> {
        if (this.flushTimer) {
            clearTimeout(this.flushTimer);
            this.flushTimer = null;
        }
        
        if (this.pendingRecords.length === 0) return;
        
        // Capture current batch
        const batch = this.pendingRecords;
        const batchWaiters = new Map(this.waiters);
        
        this.pendingRecords = [];
        this.waiters = new Map();
        
        // Write and sync
        await this.writeToDisk(batch);
        await this.syncToDisk();
        
        // Notify all waiters
        for (const resolve of batchWaiters.values()) {
            resolve();
        }
    }
}