ACID Importance: The Complete Transaction Guarantee

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

# Example: Hybrid consistency architecture for e-commerce
 
class OrderService:
    def __init__(self):
        # PostgreSQL for transactional order data (ACID)
        self.order_db = PostgreSQLConnection(
            database="orders",
            isolation_level="SERIALIZABLE"
        )
        
        # Redis for cart data (ephemeral, can be lost)
        self.cart_cache = RedisConnection(cluster="carts")
        
        # Elasticsearch for product search (eventually consistent)
        self.search_index = ElasticsearchConnection(cluster="products")
    
    def place_order(self, customer_id: str, cart_items: list) -> Order:
        """
        Critical path: uses full ACID guarantees
        """
        with self.order_db.transaction() as txn:
            # All operations are atomic within this transaction
            order = self.create_order_record(txn, customer_id)
            
            for item in cart_items:
                # Check inventory with FOR UPDATE lock
                inventory = self.lock_inventory(txn, item.product_id)
                if inventory.quantity < item.quantity:
                    txn.rollback()
                    raise InsufficientInventoryError()
                
                self.deduct_inventory(txn, item.product_id, item.quantity)
                self.create_order_line(txn, order.id, item)
            
            self.create_payment_auth(txn, order.id, self.calculate_total(order))
            
            # Commit: all-or-nothing, durable, isolated
            txn.commit()
            
        # AFTER commit succeeds, update eventually consistent systems
        # These can fail without rolling back the order
        try:
            self.cart_cache.delete(f"cart:{customer_id}")
            self.search_index.update_inventory_async(order.items)
        except Exception:
            # Log but don't fail - eventual consistency will catch up
            logging.warning("Failed to update caches after order")
        
        return order

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

import logging
from contextlib import contextmanager
from typing import TypeVar, Callable
 
T = TypeVar('T')
 
class TransactionManager:
    """
    Centralized transaction management ensuring ACID compliance.
    """
    
    def __init__(self, connection_pool):
        self.pool = connection_pool
        self.logger = logging.getLogger(__name__)
    
    @contextmanager
    def transaction(self, isolation_level: str = "READ COMMITTED"):
        """
        Context manager for database transactions with proper ACID handling.
        
        Usage:
            with txn_manager.transaction("SERIALIZABLE") as conn:
                conn.execute("UPDATE ...")
                conn.execute("INSERT ...")
            # Automatic commit on success, rollback on exception
        """
        conn = self.pool.get_connection()
        try:
            conn.execute(f"SET TRANSACTION ISOLATION LEVEL {isolation_level}")
            conn.execute("BEGIN")
            
            yield conn
            
            conn.execute("COMMIT")
            self.logger.debug("Transaction committed successfully")
            
        except Exception as e:
            self.logger.warning(f"Transaction failed, rolling back: {e}")
            conn.execute("ROLLBACK")
            raise
            
        finally:
            self.pool.release(conn)
    
    def execute_with_retry(
        self, 
        operation: Callable[[], T], 
        max_retries: int = 3,
        isolation_level: str = "SERIALIZABLE"
    ) -> T:
        """
        Execute an operation with automatic retry on serialization failures.
        Essential for Serializable isolation level.
        """
        last_error = None
        
        for attempt in range(max_retries):
            try:
                with self.transaction(isolation_level) as conn:
                    return operation(conn)
                    
            except SerializationError as e:
                last_error = e
                delay = 0.1 * (2 ** attempt)  # Exponential backoff
                self.logger.info(
                    f"Serialization conflict, retry {attempt + 1}/{max_retries} "
                    f"after {delay}s"
                )
                time.sleep(delay)
                continue
                
            except Exception as e:
                # Don't retry on non-serialization errors
                raise
        
        raise MaxRetriesExceededError(
            f"Failed after {max_retries} attempts: {last_error}"
        )
 
 
# Usage Example
txn_manager = TransactionManager(db_pool)
 
def transfer_funds(from_acct: str, to_acct: str, amount: Decimal):
    """
    Transfer funds between accounts with full ACID guarantees.
    """
    def do_transfer(conn):
        # Lock source account
        balance = conn.execute(
            "SELECT balance FROM accounts WHERE id = %s FOR UPDATE",
            (from_acct,)
        ).fetchone()[0]
        
        if balance < amount:
            raise InsufficientFundsError()
        
        conn.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (amount, from_acct)
        )
        conn.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (amount, to_acct)
        )
        
        return True
    
    # Execute with retry for serialization conflicts
    return txn_manager.execute_with_retry(do_transfer)