Transaction States

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

-- Sequence values are NOT rolled back
-- This is a common source of confusion
 
CREATE SEQUENCE order_id_seq;
 
-- Transaction 1:
BEGIN;
INSERT INTO orders (id, ...) VALUES (nextval('order_id_seq'), ...);
-- Gets order_id = 1
ROLLBACK;  -- Transaction aborted
 
-- Transaction 2:
BEGIN;  
INSERT INTO orders (id, ...) VALUES (nextval('order_id_seq'), ...);
-- Gets order_id = 2, NOT 1!
COMMIT;
 
-- Result: order_id 1 is "missing" from the orders table
-- This is a feature, not a bug - sequences are designed for 
-- high concurrency and don't participate in transaction rollback
 
-- Same behavior applies to:
-- Serial columns (use sequences internally)
-- Identity columns
-- Some auto-increment implementations

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

-- Demonstrating how abort releases locks and unblocks waiters
 
-- Session 1: Acquire exclusive lock
BEGIN;
UPDATE accounts SET balance = 1000 WHERE id = 100;
-- Holds X lock on row id=100
 
-- Session 2: Tries to access same row, BLOCKS
BEGIN;
UPDATE accounts SET balance = 2000 WHERE id = 100;
-- Waiting... (blocked by Session 1's X lock)
 
-- Session 3: Also waiting for same row
BEGIN;
SELECT * FROM accounts WHERE id = 100 FOR UPDATE;
-- Also waiting...
 
-- Session 1: Encounters error or decides to rollback
ROLLBACK;  -- Transitions: Failed → (rollback) → Aborted
-- All locks released!
 
-- Session 2 and 3: Now unblocked!
-- One of them acquires the lock and proceeds
-- The other waits for the new lock holder
 
-- After Session 2 completes, Session 3 can proceed

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

-- MVCC visibility after abort
 
-- Initial state: balance = 1000
 
-- Transaction T1:
BEGIN;
UPDATE accounts SET balance = 5000 WHERE id = 100;
-- Creates new row version with T1's transaction ID
-- Concurrent readers may or may not see this depending on snapshots
 
-- Transaction T2 (started after T1's update):
BEGIN ISOLATION LEVEL READ COMMITTED;
SELECT balance FROM accounts WHERE id = 100;
-- May see 5000 if T1 committed, or 1000 if T1 is uncommitted/aborted
-- Visibility check says: "Is T1 committed?" No → show old version
 
-- T1 aborts:
ROLLBACK;  -- T1 is now marked as aborted
 
-- T2's next read:
SELECT balance FROM accounts WHERE id = 100;
-- Returns 1000 - the original value
-- The new version created by T1 is marked as created by an aborted 
-- transaction and is therefore invisible (and will be vacuumed later)
 
COMMIT;
 
-- The row version T1 created still physically exists on disk
-- But it's effectively invisible - no transaction will ever see it
-- VACUUM will eventually remove it

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

// Recovery processing for transactions that need to abort
 
function recovery_undo_phase(active_transactions) {
    // active_transactions = transactions that were in Active or Failed 
    // state at crash time (no COMMIT or complete ABORT in log)
    
    undo_list = active_transactions.copy()
    
    // Process log from end to beginning
    current_lsn = log.end_position()
    
    while undo_list is not empty:
        record = read_log_record(current_lsn)
        
        if record.transaction_id in undo_list:
            
            if record.type == CLR:
                // This is a compensation record - skip to undo-next-lsn
                // The original operation was already undone before crash
                current_lsn = record.undo_next_lsn
                if record.undo_next_lsn == NULL:
                    // This transaction's rollback is complete
                    write_abort_record(record.transaction_id)
                    undo_list.remove(record.transaction_id)
                continue
                
            elif record.type == UPDATE:
                // Need to undo this update
                page = read_page(record.page_id)
                if page.lsn >= record.lsn:
                    // Page has this update; undo it
                    apply_undo(record.old_value)
                    write_clr(record.transaction_id, record)
                    
            elif record.type == BEGIN:
                // Reached the beginning of this transaction
                write_abort_record(record.transaction_id)
                undo_list.remove(record.transaction_id)
        
        current_lsn = record.prev_lsn
    
    // All previously active transactions are now Aborted
}

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

import time
import random
from enum import Enum
from dataclasses import dataclass
from typing import Callable, TypeVar, Optional
 
class AbortReason(Enum):
    DEADLOCK = "deadlock"
    SERIALIZATION = "serialization_failure"
    LOCK_TIMEOUT = "lock_timeout"
    CONSTRAINT = "constraint_violation"
    USER_CANCEL = "user_cancel"
    RESOURCE = "resource_exhaustion"
    CONNECTION = "connection_lost"
    UNKNOWN = "unknown"
 
@dataclass
class RetryPolicy:
    max_attempts: int
    initial_delay: float  # seconds
    max_delay: float
    exponential_base: float = 2.0
    jitter: bool = True
 
# Default policies by abort reason
DEFAULT_POLICIES = {
    AbortReason.DEADLOCK: RetryPolicy(max_attempts=5, initial_delay=0.01, max_delay=1.0),
    AbortReason.SERIALIZATION: RetryPolicy(max_attempts=5, initial_delay=0.01, max_delay=1.0),
    AbortReason.LOCK_TIMEOUT: RetryPolicy(max_attempts=3, initial_delay=0.5, max_delay=10.0),
    AbortReason.RESOURCE: RetryPolicy(max_attempts=3, initial_delay=1.0, max_delay=30.0),
    AbortReason.CONNECTION: RetryPolicy(max_attempts=3, initial_delay=0.5, max_delay=5.0),
    # These should not be retried automatically
    AbortReason.CONSTRAINT: None,
    AbortReason.USER_CANCEL: None,
    AbortReason.UNKNOWN: None,
}
 
T = TypeVar('T')
 
def execute_with_smart_retry(
    operation: Callable[[], T],
    classify_error: Callable[[Exception], AbortReason],
    custom_policies: Optional[dict] = None
) -> T:
    """
    Execute an operation with intelligent retry logic based on abort reason.
    
    :param operation: Function to execute (should manage its own transaction)
    :param classify_error: Function to classify an exception into AbortReason
    :param custom_policies: Optional custom retry policies
    :return: Result of successful operation
    :raises: Last exception if all retries exhausted or non-retriable error
    """
    policies = {**DEFAULT_POLICIES, **(custom_policies or {})}
    
    attempt = 0
    last_error = None
    
    while True:
        try:
            return operation()
            
        except Exception as e:
            last_error = e
            reason = classify_error(e)
            policy = policies.get(reason)
            
            if policy is None:
                # Non-retriable error
                print(f"Non-retriable abort: {reason.value}")
                raise
            
            attempt += 1
            if attempt > policy.max_attempts:
                print(f"Max retries ({policy.max_attempts}) exceeded for {reason.value}")
                raise
            
            # Calculate delay with exponential backoff
            delay = min(
                policy.initial_delay * (policy.exponential_base ** (attempt - 1)),
                policy.max_delay
            )
            
            # Add jitter to prevent thundering herd
            if policy.jitter:
                delay = delay * (0.5 + random.random())
            
            print(f"Abort ({reason.value}), attempt {attempt}/{policy.max_attempts}, "
                  f"retrying in {delay:.2f}s")
            
            time.sleep(delay)
    
    return None  # Should never reach here
 
 
# Example usage
def transfer_funds(from_id: int, to_id: int, amount: float):
    """Business operation that may be retried."""
    with get_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("BEGIN")
            # ... perform transfer ...
            conn.commit()
 
def classify_postgres_error(e: Exception) -> AbortReason:
    """Classify PostgreSQL errors into retry categories."""
    if hasattr(e, 'pgcode'):
        code = e.pgcode
        if code == '40001':
            return AbortReason.SERIALIZATION
        elif code == '40P01':
            return AbortReason.DEADLOCK
        elif code in ('23505', '23503', '23502', '23514'):
            return AbortReason.CONSTRAINT
        elif code.startswith('08'):
            return AbortReason.CONNECTION
    return AbortReason.UNKNOWN
 
# Execute with retry
result = execute_with_smart_retry(
    lambda: transfer_funds(1, 2, 100.00),
    classify_postgres_error
)

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

-- Calculate abort/rollback rate by database
SELECT 
    datname,
    xact_commit AS commits,
    xact_rollback AS rollbacks,
    ROUND(100.0 * xact_rollback / NULLIF(xact_commit + xact_rollback, 0), 2) 
        AS abort_percentage,
    conflicts AS replication_conflicts
FROM pg_stat_database
WHERE datname NOT LIKE 'template%'
ORDER BY abort_percentage DESC;
 
-- Healthy systems typically see < 1% abort rate
-- > 5% abort rate suggests problems
 
-- Monitor conflict-specific aborts (replication)
SELECT 
    datname,
    confl_tablespace,
    confl_lock,
    confl_snapshot,
    confl_bufferpin,
    confl_deadlock,
    (confl_tablespace + confl_lock + confl_snapshot + 
     confl_bufferpin + confl_deadlock) AS total_conflicts
FROM pg_stat_database_conflicts;
 
-- Dead tuple accumulation (indicates abort/update activity)
SELECT 
    schemaname,
    relname,
    n_live_tup,
    n_dead_tup,
    ROUND(100.0 * n_dead_tup / NULLIF(n_live_tup + n_dead_tup, 0), 2) 
        AS dead_tuple_pct,
    last_vacuum,
    last_autovacuum
FROM pg_stat_user_tables
WHERE n_dead_tup > 1000
ORDER BY n_dead_tup DESC;

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

-- Techniques to prevent unnecessary aborts
 
-- 1. Use ON CONFLICT instead of letting constraint abort
INSERT INTO users (email, name)
VALUES ('alice@example.com', 'Alice')
ON CONFLICT (email) DO UPDATE SET name = EXCLUDED.name;
-- Never aborts on duplicate email
 
-- 2. Use SELECT FOR UPDATE SKIP LOCKED for queue processing
-- Instead of waiting and risking deadlock/timeout:
BEGIN;
SELECT * FROM job_queue 
WHERE status = 'pending'
ORDER BY created_at
LIMIT 1
FOR UPDATE SKIP LOCKED;  -- Skip rows locked by other transactions
-- Process the job...
COMMIT;
 
-- 3. Use advisory locks for coordinating without row locks
SELECT pg_advisory_lock(12345);  -- Acquire advisory lock
-- Do work that would otherwise contend on rows
SELECT pg_advisory_unlock(12345);
 
-- 4. Check before attempting operations
DO $$
DECLARE
    current_balance DECIMAL;
BEGIN
    -- Check first (outside main logic)
    SELECT balance INTO current_balance
    FROM accounts WHERE id = 100;
    
    IF current_balance < 500 THEN
        RAISE EXCEPTION 'Insufficient balance';
    END IF;
    
    -- Now do the actual update
    UPDATE accounts SET balance = balance - 500 WHERE id = 100;
END $$;