Cascading Rollback

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

Time  Transaction  Operation               Effect
────  ───────────  ────────────────────────────────────────────────
t1    T₁           BEGIN
t2    T₁           READ(A)                  Reads $1000
t3    T₁           WRITE(A) ← $800          A = $800 (uncommitted)
t4    T₂           BEGIN
t5    T₂           READ(A)                  Reads $800 [DIRTY READ from T₁]
t6    T₂           WRITE(B) ← A + $200      B = $1000 (uncommitted)
t7    T₃           BEGIN
t8    T₃           READ(B)                  Reads $1000 [DIRTY READ from T₂]
t9    T₃           WRITE(Report)            Report based on B=$1000
t10   T₄           BEGIN
t11   T₄           READ(A)                  Reads $800 [DIRTY READ from T₁]
t12   T₄           WRITE(Log) ← A           Log contains $800
 
─── CRITICAL EVENT ───
t13   T₁           ABORT                    Constraint violation!
 
─── CASCADE BEGINS ───
t14   T₂           CASCADING ABORT          Read from T₁ (aborted)
t15   T₃           CASCADING ABORT          Read from T₂ (aborted)
t16   T₄           CASCADING ABORT          Read from T₁ (aborted)
 
─── AFTERMATH ───
All work by T₁, T₂, T₃, T₄ is undone
A returns to $1000, B returns to $500
Report and Log never written

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

from dataclasses import dataclass
from typing import Dict
import math
 
@dataclass
class CascadeRiskAnalysis:
    """
    Analyze cascade risk based on workload characteristics.
    
    This model estimates the expected number of cascading aborts
    based on system parameters.
    """
    
    # System parameters
    transaction_rate: float      # Transactions per second
    avg_transaction_duration: float  # Seconds
    abort_rate: float            # Probability a transaction aborts
    dirty_read_rate: float       # Probability of reading uncommitted data
    
    def expected_dirty_reads_per_abort(self) -> float:
        """
        Estimate how many dirty reads occur during one transaction's lifetime.
        
        Dirty reads on a transaction's writes = 
            (read rate) × (time data is uncommitted) × (dirty read probability)
        """
        # Concurrent transactions during our transaction's lifetime
        concurrent_txns = self.transaction_rate * self.avg_transaction_duration
        
        # Expected dirty reads we cause (simplified model)
        return concurrent_txns * self.dirty_read_rate
    
    def expected_cascade_size(self) -> float:
        """
        Estimate the expected cascade size when a transaction aborts.
        
        Uses a simple model: each dirty read creates a direct cascade victim,
        and each victim may have their own dirty readers (recursive).
        """
        direct_victims = self.expected_dirty_reads_per_abort()
        
        # Recursive cascade (geometric series if rate < 1)
        if direct_victims < 1:
            # Converges: E[cascade] = direct / (1 - direct)
            return direct_victims / (1 - direct_victims) if direct_victims < 1 else float('inf')
        else:
            # Diverges - cascade avalanche territory
            return float('inf')
    
    def abort_impact_factor(self) -> float:
        """
        The multiplier effect: how many transactions are affected per abort.
        
        Impact Factor > 2 is concerning
        Impact Factor > 5 is dangerous
        Impact Factor > 10 often indicates system instability
        """
        return 1 + self.expected_cascade_size()
    
    def effective_throughput(self) -> float:
        """
        Estimate effective throughput considering cascading aborts.
        
        Base throughput × (1 - total abort rate including cascades)
        """
        direct_abort_rate = self.abort_rate
        cascade_abort_rate = direct_abort_rate * self.expected_cascade_size()
        total_abort_rate = min(1.0, direct_abort_rate + cascade_abort_rate)
        
        return 1.0 - total_abort_rate
 
 
def analyze_scenarios():
    """Compare cascade risk across different workload types."""
    
    scenarios = {
        "Low Risk (batch processing)": CascadeRiskAnalysis(
            transaction_rate=10,         # 10 TPS
            avg_transaction_duration=0.5, # 500ms
            abort_rate=0.01,             # 1% abort rate
            dirty_read_rate=0.01,        # Minimal dirty reads (good isolation)
        ),
        "Medium Risk (web application)": CascadeRiskAnalysis(
            transaction_rate=100,        # 100 TPS
            avg_transaction_duration=0.1, # 100ms
            abort_rate=0.02,             # 2% abort rate
            dirty_read_rate=0.05,        # Some dirty reads
        ),
        "High Risk (hot spot workload)": CascadeRiskAnalysis(
            transaction_rate=1000,       # 1000 TPS
            avg_transaction_duration=0.2, # 200ms (long transactions)
            abort_rate=0.05,             # 5% abort rate
            dirty_read_rate=0.20,        # Many dirty reads (READ UNCOMMITTED)
        ),
        "Cascade Avalanche Risk": CascadeRiskAnalysis(
            transaction_rate=500,
            avg_transaction_duration=0.5,
            abort_rate=0.10,
            dirty_read_rate=0.30,
        ),
    }
    
    print("=" * 70)
    print("CASCADE RISK ANALYSIS")
    print("=" * 70)
    
    for name, analysis in scenarios.items():
        print(f"
Scenario: {name}")
        print("-" * 50)
        print(f"  Expected dirty reads per abort: {analysis.expected_dirty_reads_per_abort():.2f}")
        cascade_size = analysis.expected_cascade_size()
        cascade_str = f"{cascade_size:.2f}" if cascade_size != float('inf') else "AVALANCHE"
        print(f"  Expected cascade size: {cascade_str}")
        impact = analysis.abort_impact_factor()
        impact_str = f"{impact:.2f}" if impact != float('inf') else "UNSTABLE"
        print(f"  Abort impact factor: {impact_str}")
        print(f"  Effective throughput: {analysis.effective_throughput():.1%}")
 
analyze_scenarios()

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

-- Cascade Detection Queries for PostgreSQL
 
-- 1. Monitor abort rate over time
SELECT 
    date_trunc('minute', now()) as minute,
    COUNT(*) FILTER (WHERE state = 'aborted') as aborted,
    COUNT(*) FILTER (WHERE state = 'committed') as committed,
    ROUND(
        COUNT(*) FILTER (WHERE state = 'aborted')::numeric /
        NULLIF(COUNT(*)::numeric, 0) * 100, 2
    ) as abort_rate_percent
FROM pg_stat_activity
GROUP BY 1;
 
-- 2. Find transactions with many rollbacks (cascade victims)
SELECT 
    query,
    calls as total_attempts,
    rows as rows_affected,
    (calls - rows) as potential_rollbacks,
    ROUND((calls - rows)::numeric / NULLIF(calls::numeric, 0) * 100, 2) 
        as failure_rate_percent
FROM pg_stat_statements
WHERE calls > 100
ORDER BY failure_rate_percent DESC
LIMIT 20;
 
-- 3. Detect lock wait patterns (cascade indicator)
SELECT 
    blocked_locks.pid AS blocked_pid,
    blocked_activity.usename AS blocked_user,
    blocking_locks.pid AS blocking_pid,
    blocking_activity.usename AS blocking_user,
    blocked_activity.query AS blocked_statement,
    blocking_activity.query AS current_statement_in_blocking_process,
    blocked_activity.state as blocked_state
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity 
    ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks 
    ON blocking_locks.locktype = blocked_locks.locktype
    AND blocking_locks.relation = blocked_locks.relation
    AND blocking_locks.page = blocked_locks.page
    AND blocking_locks.tuple = blocked_locks.tuple
    AND blocking_locks.virtualxid = blocked_locks.virtualxid
    AND blocking_locks.transactionid = blocked_locks.transactionid
    AND blocking_locks.classid = blocked_locks.classid
    AND blocking_locks.objid = blocked_locks.objid
    AND blocking_locks.objsubid = blocked_locks.objsubid
    AND blocking_locks.pid != blocked_locks.pid
JOIN pg_catalog.pg_stat_activity blocking_activity 
    ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;
 
-- 4. Rollback segment usage (cascade pressure indicator)
SELECT 
    name,
    setting,
    unit,
    source
FROM pg_settings
WHERE name LIKE '%wal%' OR name LIKE '%checkpoint%'
ORDER BY name;