Participant Role

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

interface ParticipantTransactionContext {
    // Global transaction identifier from coordinator
    globalTxId: string;
    
    // Local transaction identifier
    localTxId: string;
    
    // Coordinator information for recovery
    coordinator: {
        id: string;
        endpoint: string;
    };
    
    // Current state in participant state machine
    state: 'ACTIVE' | 'PREPARED' | 'COMMITTED' | 'ABORTED';
    
    // Locks held by this transaction
    heldLocks: Set<LockHandle>;
    
    // Read set for validation
    readSet: Map<DataItemId, Version>;
    
    // Write set with old/new values
    writeSet: Map<DataItemId, { oldValue: any; newValue: any }>;
    
    // Local log position for undo/redo
    logSequenceNumber: number;
}
 
class ParticipantTransactionManager {
    private activeTransactions: Map<string, ParticipantTransactionContext>;
    
    /**
     * Execute a local operation within a distributed transaction
     */
    async executeOperation(
        globalTxId: string,
        operation: Operation
    ): Promise<OperationResult> {
        const ctx = this.activeTransactions.get(globalTxId);
        
        if (!ctx || ctx.state !== 'ACTIVE') {
            throw new Error(`Transaction ${globalTxId} not active`);
        }
        
        // Acquire necessary locks
        const locks = await this.lockManager.acquireLocks(
            operation.requiredLocks,
            globalTxId
        );
        ctx.heldLocks = new Set([...ctx.heldLocks, ...locks]);
        
        if (operation.type === 'READ') {
            // Track read for validation
            const item = await this.storage.read(operation.itemId);
            ctx.readSet.set(operation.itemId, item.version);
            return { data: item.value };
            
        } else if (operation.type === 'WRITE') {
            // Log old value for undo
            const oldItem = await this.storage.read(operation.itemId);
            
            // Log redo/undo information
            ctx.logSequenceNumber = await this.log.writeRedoUndo({
                transactionId: globalTxId,
                itemId: operation.itemId,
                oldValue: oldItem.value,
                newValue: operation.value
            });
            
            // Apply modification (in buffer, not yet durable)
            await this.storage.write(operation.itemId, operation.value);
            
            // Track in write set
            ctx.writeSet.set(operation.itemId, {
                oldValue: oldItem.value,
                newValue: operation.value
            });
            
            return { success: true };
        }
    }
}

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

class ParticipantTransactionManager {
    /**
     * Handle PREPARE request from coordinator
     */
    async handlePrepare(globalTxId: string): Promise<'VOTE_COMMIT' | 'VOTE_ABORT'> {
        const ctx = this.activeTransactions.get(globalTxId);
        
        if (!ctx) {
            // Unknown transaction - vote abort
            return 'VOTE_ABORT';
        }
        
        if (ctx.state !== 'ACTIVE') {
            // Already processed - should not happen
            throw new Error(`Unexpected PREPARE for ${globalTxId} in state ${ctx.state}`);
        }
        
        // CHECK 1: Verify all constraints are satisfied
        const constraintResult = await this.checkConstraints(ctx);
        if (!constraintResult.satisfied) {
            return this.voteAbort(ctx, `Constraint violation: ${constraintResult.reason}`);
        }
        
        // CHECK 2: Check for deadlock involvement
        if (this.deadlockDetector.isInvolvedInDeadlock(globalTxId)) {
            return this.voteAbort(ctx, 'Selected as deadlock victim');
        }
        
        // CHECK 3: Verify sufficient resources
        const resourceCheck = await this.checkResources(ctx);
        if (!resourceCheck.available) {
            return this.voteAbort(ctx, `Insufficient resources: ${resourceCheck.reason}`);
        }
        
        // CHECK 4: Validation for optimistic concurrency control
        if (this.usesOCC) {
            const validationResult = await this.validateReadSet(ctx);
            if (!validationResult.valid) {
                return this.voteAbort(ctx, 'Validation failed - read set was modified');
            }
        }
        
        // All checks passed - vote COMMIT
        return this.voteCommit(ctx);
    }
    
    /**
     * Vote to commit the transaction
     */
    private async voteCommit(ctx: ParticipantTransactionContext): Promise<'VOTE_COMMIT'> {
        // CRITICAL: Force-write PREPARED record BEFORE sending vote
        // This record must contain enough info to redo OR undo the transaction
        await this.log.forceWrite({
            type: 'PREPARED',
            transactionId: ctx.globalTxId,
            coordinator: ctx.coordinator,
            writeSet: ctx.writeSet,  // For redo
            // Undo information was logged during execution
        });
        
        // Transition to PREPARED state
        ctx.state = 'PREPARED';
        
        // Keep all locks - do NOT release them!
        // Other transactions will block on these locks
        
        console.log(`Transaction ${ctx.globalTxId} entering PREPARED state`);
        
        return 'VOTE_COMMIT';
    }
    
    /**
     * Vote to abort the transaction
     */
    private async voteAbort(
        ctx: ParticipantTransactionContext, 
        reason: string
    ): Promise<'VOTE_ABORT'> {
        console.log(`Transaction ${ctx.globalTxId} voting ABORT: ${reason}`);
        
        // Log abort
        await this.log.forceWrite({
            type: 'ABORT',
            transactionId: ctx.globalTxId,
            reason: reason
        });
        
        // Rollback local changes
        await this.rollbackTransaction(ctx);
        
        // Release all locks
        await this.lockManager.releaseAll(ctx.globalTxId);
        ctx.heldLocks.clear();
        
        // Update state
        ctx.state = 'ABORTED';
        
        // Clean up
        this.activeTransactions.delete(ctx.globalTxId);
        
        return 'VOTE_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
106
107
108
109
110
111
112
113
114
115
116

class ParticipantTransactionManager {
    /**
     * Handle the global decision from coordinator
     */
    async handleGlobalDecision(
        globalTxId: string, 
        decision: 'GLOBAL_COMMIT' | 'GLOBAL_ABORT'
    ): Promise<void> {
        const ctx = this.activeTransactions.get(globalTxId);
        
        if (!ctx) {
            // Transaction not found - might have been cleaned up already
            // This is okay - idempotent handling
            console.log(`Decision for unknown tx ${globalTxId} - already completed`);
            return;
        }
        
        if (ctx.state === 'COMMITTED' || ctx.state === 'ABORTED') {
            // Already processed - idempotent
            console.log(`Decision for ${globalTxId} already processed`);
            return;
        }
        
        if (ctx.state !== 'PREPARED') {
            // Unexpected state - might be a late PREPARE race condition
            throw new Error(`Unexpected decision for ${globalTxId} in state ${ctx.state}`);
        }
        
        if (decision === 'GLOBAL_COMMIT') {
            await this.executeCommit(ctx);
        } else {
            await this.executeAbort(ctx);
        }
    }
    
    /**
     * Execute local commit after receiving GLOBAL_COMMIT
     */
    private async executeCommit(ctx: ParticipantTransactionContext): Promise<void> {
        console.log(`Committing transaction ${ctx.globalTxId}`);
        
        // Step 1: Force-write COMMIT record
        await this.log.forceWrite({
            type: 'COMMIT',
            transactionId: ctx.globalTxId,
            timestamp: Date.now()
        });
        
        // Step 2: Make changes permanent
        // In practice, this might flush buffer pool pages or
        // trigger checkpoint behavior depending on the storage engine
        await this.storage.makeChangesPermanent(ctx.globalTxId);
        
        // Step 3: Release all locks
        await this.lockManager.releaseAll(ctx.globalTxId);
        ctx.heldLocks.clear();
        
        // Step 4: Update state
        ctx.state = 'COMMITTED';
        
        // Step 5: Clean up
        this.activeTransactions.delete(ctx.globalTxId);
        
        console.log(`Transaction ${ctx.globalTxId} committed successfully`);
    }
    
    /**
     * Execute local abort after receiving GLOBAL_ABORT
     */
    private async executeAbort(ctx: ParticipantTransactionContext): Promise<void> {
        console.log(`Aborting transaction ${ctx.globalTxId}`);
        
        // Step 1: Force-write ABORT record
        await this.log.forceWrite({
            type: 'ABORT',
            transactionId: ctx.globalTxId,
            timestamp: Date.now()
        });
        
        // Step 2: Rollback all modifications using undo log
        await this.rollbackTransaction(ctx);
        
        // Step 3: Release all locks
        await this.lockManager.releaseAll(ctx.globalTxId);
        ctx.heldLocks.clear();
        
        // Step 4: Update state
        ctx.state = 'ABORTED';
        
        // Step 5: Clean up
        this.activeTransactions.delete(ctx.globalTxId);
        
        console.log(`Transaction ${ctx.globalTxId} aborted successfully`);
    }
    
    /**
     * Rollback transaction modifications using undo log
     */
    private async rollbackTransaction(ctx: ParticipantTransactionContext): Promise<void> {
        // Read undo log records in reverse order (most recent first)
        const undoRecords = await this.log.getUndoRecords(ctx.globalTxId);
        
        for (const record of undoRecords.reverse()) {
            // Restore old value
            await this.storage.write(record.itemId, record.oldValue);
            
            // Log the undo operation (for recovery if we crash during rollback)
            await this.log.write({
                type: 'UNDO',
                transactionId: ctx.globalTxId,
                itemId: record.itemId,
                restoredValue: record.oldValue
            });
        }
    }
}

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
131
132
133
134

class ParticipantRecovery {
    /**
     * Recover participant state after crash
     */
    async recover(): Promise<void> {
        console.log('Starting participant recovery...');
        
        // Scan log and identify in-flight transactions
        const transactions = await this.classifyTransactions();
        
        // Handle each transaction based on its state
        for (const [txId, state] of transactions) {
            await this.recoverTransaction(txId, state);
        }
        
        console.log('Participant recovery complete');
    }
    
    /**
     * Recover a single transaction
     */
    private async recoverTransaction(
        txId: string, 
        state: RecoveryTransactionState
    ): Promise<void> {
        switch (state.lastRecord) {
            case 'COMMIT':
                // Complete the commit
                await this.redoCommit(txId, state);
                break;
                
            case 'ABORT':
                // Complete the abort
                await this.undoAbort(txId, state);
                break;
                
            case 'PREPARED':
                // Re-enter prepared state and query coordinator
                await this.recoverPrepared(txId, state);
                break;
                
            default:
                // Only execution records - abort
                await this.abortIncomplete(txId, state);
        }
    }
    
    /**
     * Recover a transaction that was in PREPARED state
     */
    private async recoverPrepared(
        txId: string, 
        state: RecoveryTransactionState
    ): Promise<void> {
        console.log(`Recovering PREPARED transaction ${txId}`);
        
        // Re-acquire locks based on write set
        const writeSet = state.preparedRecord!.writeSet;
        for (const itemId of writeSet.keys()) {
            await this.lockManager.acquireLock(itemId, 'EXCLUSIVE', txId);
        }
        
        // Create transaction context in PREPARED state
        const ctx: ParticipantTransactionContext = {
            globalTxId: txId,
            localTxId: state.preparedRecord!.localTxId,
            coordinator: state.preparedRecord!.coordinator,
            state: 'PREPARED',
            heldLocks: new Set(), // Will be populated by lock acquisition
            readSet: new Map(),
            writeSet: writeSet,
            logSequenceNumber: state.preparedRecord!.lsn
        };
        
        this.activeTransactions.set(txId, ctx);
        
        // Query coordinator for decision
        await this.queryCoordinatorForDecision(txId, ctx);
    }
    
    /**
     * Query coordinator to learn transaction outcome
     */
    private async queryCoordinatorForDecision(
        txId: string, 
        ctx: ParticipantTransactionContext
    ): Promise<void> {
        const retryInterval = 5000; // 5 seconds
        
        while (ctx.state === 'PREPARED') {
            try {
                const decision = await this.sendDecisionQuery(
                    ctx.coordinator.endpoint, 
                    txId
                );
                
                if (decision === 'COMMIT') {
                    await this.executeCommit(ctx);
                } else if (decision === 'ABORT') {
                    await this.executeAbort(ctx);
                }
                // else: coordinator doesn't know yet, keep waiting
                
            } catch (error) {
                console.log(`Cannot reach coordinator for ${txId}: ${error}`);
                // Will retry after interval
            }
            
            if (ctx.state === 'PREPARED') {
                await this.sleep(retryInterval);
            }
        }
    }
    
    /**
     * Abort a transaction that never reached PREPARED state
     */
    private async abortIncomplete(
        txId: string, 
        state: RecoveryTransactionState
    ): Promise<void> {
        console.log(`Aborting incomplete transaction ${txId}`);
        
        // Undo any modifications
        await this.undoAbort(txId, state);
        
        // Log abort
        await this.log.forceWrite({
            type: 'ABORT',
            transactionId: txId,
            reason: 'Recovery: never prepared'
        });
    }
}