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

// Three-Phase Commit Protocol Implementation
 
type ParticipantState3PC = 
    | 'WORKING' 
    | 'PREPARED' 
    | 'PRE_COMMITTED'  // New state in 3PC
    | 'COMMITTED' 
    | 'ABORTED';
 
class ThreePhaseCommitCoordinator {
    private transactionLog: TransactionLog;
    private participants: Participant3PC[];
    private timeout: number = 10000;
 
    async executeTransaction(transaction: Transaction): Promise<boolean> {
        const txId = transaction.id;
 
        // ========== PHASE 1: PREPARE ==========
        console.log(`[${txId}] Phase 1: Sending PREPARE`);
        
        const votes = await this.collectVotes(transaction);
        
        if (!votes.allCommit) {
            await this.broadcastAbort(txId);
            return false;
        }
 
        // ========== PHASE 2: PRE-COMMIT ==========
        // This is the key addition in 3PC
        console.log(`[${txId}] Phase 2: Sending PRE_COMMIT`);
        
        await this.transactionLog.write(txId, 'PRE_COMMITTING');
        
        const preCommitAcks = await Promise.all(
            this.participants.map(async (p) => {
                try {
                    await this.withTimeout(p.preCommit(txId), this.timeout);
                    return true;
                } catch (error) {
                    return false;
                }
            })
        );
 
        // If any participant failed to precommit, we can still safely abort
        // because no one has committed yet
        if (!preCommitAcks.every(ack => ack)) {
            await this.broadcastAbort(txId);
            return false;
        }
 
        // ========== PHASE 3: COMMIT ==========
        console.log(`[${txId}] Phase 3: Sending COMMIT`);
        
        await this.transactionLog.write(txId, 'COMMITTING');
        
        await Promise.all(
            this.participants.map(p => 
                this.retryUntilSuccess(() => p.commit(txId))
            )
        );
 
        await this.transactionLog.write(txId, 'COMMITTED');
        return true;
    }
 
    private async collectVotes(tx: Transaction): Promise<{ allCommit: boolean }> {
        const votes = await Promise.all(
            this.participants.map(p => 
                this.withTimeout(p.prepare(tx), this.timeout, 'VOTE_ABORT')
            )
        );
        return { allCommit: votes.every(v => v === 'VOTE_COMMIT') };
    }
}
 
class ThreePhaseCommitParticipant {
    private state: ParticipantState3PC = 'WORKING';
    private participantLog: ParticipantLog;
    private timeoutMs: number = 15000;
 
    async prepare(transaction: Transaction): Promise<VoteResult> {
        // Validate and prepare transaction
        const canCommit = await this.validateAndPrepare(transaction);
        
        if (canCommit) {
            await this.participantLog.write(transaction.id, 'PREPARED');
            this.state = 'PREPARED';
            return 'VOTE_COMMIT';
        } else {
            this.state = 'ABORTED';
            return 'VOTE_ABORT';
        }
    }
 
    async preCommit(txId: string): Promise<void> {
        // Transition to pre-committed state
        await this.participantLog.write(txId, 'PRE_COMMITTED');
        this.state = 'PRE_COMMITTED';
        
        // Start timeout - if no COMMIT received, we can commit autonomously
        this.startCommitTimeout(txId);
    }
 
    async commit(txId: string): Promise<void> {
        await this.participantLog.write(txId, 'COMMITTED');
        this.state = 'COMMITTED';
        // Release locks, apply changes
    }
 
    private startCommitTimeout(txId: string): void {
        setTimeout(async () => {
            if (this.state === 'PRE_COMMITTED') {
                // KEY 3PC INSIGHT: If we're pre-committed and timeout,
                // we KNOW all participants received PRE_COMMIT
                // (otherwise coordinator would have aborted)
                // So we can safely commit autonomously
                console.log(`Timeout: Committing ${txId} autonomously`);
                await this.commit(txId);
            }
        }, this.timeoutMs);
    }
}

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

// 3PC Termination Protocol - How participants recover without coordinator
 
class ThreePhaseTerminationProtocol {
    constructor(
        private localState: ParticipantState3PC,
        private otherParticipants: Participant3PC[]
    ) {}
 
    async electNewCoordinatorAndDecide(txId: string): Promise<'COMMIT' | 'ABORT'> {
        // Collect states from all reachable participants
        const states = await this.collectParticipantStates(txId);
        
        // Apply termination rules
        return this.decide(states);
    }
 
    private decide(states: ParticipantState3PC[]): 'COMMIT' | 'ABORT' {
        // Rule 1: If anyone committed, we must commit
        if (states.includes('COMMITTED')) {
            return 'COMMIT';
        }
 
        // Rule 2: If anyone aborted, we must abort
        if (states.includes('ABORTED')) {
            return 'ABORT';
        }
 
        // Rule 3: If anyone is still WORKING, abort
        // (coordinator would have sent ABORT, not PRE_COMMIT)
        if (states.includes('WORKING')) {
            return 'ABORT';
        }
 
        // Rule 4: If everyone is PREPARED (no one PRE_COMMITTED)
        // Coordinator hadn't sent PRE_COMMIT yet → abort
        if (states.every(s => s === 'PREPARED')) {
            return 'ABORT';
        }
 
        // Rule 5: If at least one is PRE_COMMITTED
        // Coordinator decided to commit → commit
        if (states.includes('PRE_COMMITTED')) {
            return 'COMMIT';
        }
 
        // Should never reach here
        throw new Error('Unexpected state combination');
    }
 
    private async collectParticipantStates(txId: string): Promise<ParticipantState3PC[]> {
        const states: ParticipantState3PC[] = [this.localState];
        
        for (const participant of this.otherParticipants) {
            try {
                const state = await participant.queryState(txId);
                states.push(state);
            } catch (error) {
                // Participant unreachable - skip
                // This is where network partitions cause problems!
            }
        }
        
        return states;
    }
}