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

from cassandra.cluster import Cluster
from cassandra import ConsistencyLevel
 
class TunableConsistencyDemo:
    """
    Demonstrates using different consistency levels for different operations
    based on their requirements.
    """
    
    def __init__(self, hosts=['localhost']):
        self.cluster = Cluster(hosts)
        self.session = self.cluster.connect('myapp')
    
    def update_user_balance(self, user_id: str, new_balance: float):
        """
        Financial data: Use strong consistency.
        We cannot tolerate stale reads or lost writes.
        """
        statement = self.session.prepare(
            "UPDATE accounts SET balance = ? WHERE user_id = ?"
        )
        # QUORUM: Majority of replicas must acknowledge
        statement.consistency_level = ConsistencyLevel.QUORUM
        
        self.session.execute(statement, [new_balance, user_id])
        # Write is now durable on majority of nodes
    
    def get_user_balance(self, user_id: str) -> float:
        """
        Reading balance: Also use QUORUM to ensure we see latest write.
        WRITE(QUORUM) + READ(QUORUM) = linearizable
        """
        statement = self.session.prepare(
            "SELECT balance FROM accounts WHERE user_id = ?"
        )
        statement.consistency_level = ConsistencyLevel.QUORUM
        
        result = self.session.execute(statement, [user_id])
        return result.one().balance
    
    def post_social_update(self, user_id: str, content: str):
        """
        Social media post: Prioritize availability.
        It's okay if the post takes a moment to appear everywhere.
        """
        statement = self.session.prepare(
            "INSERT INTO posts (user_id, content, timestamp) VALUES (?, ?, toTimestamp(now()))"
        )
        # ONE: Just one replica is enough
        statement.consistency_level = ConsistencyLevel.ONE
        
        self.session.execute(statement, [user_id, content])
        # Fast response, eventual propagation to other replicas
    
    def get_user_feed(self, user_id: str, limit: int = 50):
        """
        Feed: Eventually consistent is fine.
        Missing a very recent post for a few seconds is acceptable.
        """
        statement = self.session.prepare(
            "SELECT * FROM posts WHERE user_id = ? LIMIT ?"
        )
        statement.consistency_level = ConsistencyLevel.ONE
        
        return self.session.execute(statement, [user_id, limit])
    
    def update_inventory(self, product_id: str, delta: int):
        """
        Inventory: Strong consistency to prevent overselling.
        Use lightweight transactions for atomic update.
        """
        # Lightweight Transaction (LWT) provides linearizable semantics
        # It's slower but essential for inventory
        statement = self.session.prepare(
            """
            UPDATE inventory 
            SET count = count + ?
            WHERE product_id = ?
            IF EXISTS
            """
        )
        # Serial consistency ensures linearizability for LWT
        statement.serial_consistency_level = ConsistencyLevel.SERIAL
        
        result = self.session.execute(statement, [delta, product_id])
        return result.was_applied
    
    def log_analytics_event(self, event_data: dict):
        """
        Analytics: Availability is paramount, consistency doesn't matter.
        Even if we lose a few events, it's fine.
        """
        statement = self.session.prepare(
            "INSERT INTO analytics_events (event_id, data) VALUES (?, ?)"
        )
        # ANY: Even hinted handoff counts - maximum availability
        statement.consistency_level = ConsistencyLevel.ANY
        
        self.session.execute(statement, [uuid.uuid4(), json.dumps(event_data)])
 
 
# Summary of consistency choices:
#
# Data Type              | Write Level  | Read Level   | Rationale
# -------------------------------------------------------------------------
# Account balance        | QUORUM       | QUORUM       | Financial accuracy
# Inventory counts       | SERIAL (LWT) | SERIAL       | Prevent overselling
# User profile           | QUORUM       | ONE          | Durable, fast reads
# Social posts           | ONE          | ONE          | Speed over freshness
# Analytics events       | ANY          | N/A          | Fire and forget
# Distributed locks      | SERIAL       | SERIAL       | Must be linearizable

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

from enum import Enum
from typing import Any
 
class ConsistencyLevel(Enum):
    STRONG = "strong"       # Read-your-writes, linearizable
    SESSION = "session"     # Consistent within user session
    EVENTUAL = "eventual"   # May return stale data
 
class PolyglotDataLayer:
    """
    Unified data layer that routes to appropriate stores
    based on data type and consistency requirements.
    """
    
    def __init__(self):
        # Different stores for different consistency needs
        self.postgres = PostgreSQLClient()      # Strong consistency
        self.cassandra = CassandraClient()      # Tunable consistency
        self.redis = RedisClient()              # Session/cache
        self.elasticsearch = ElasticsearchClient()  # Search (eventual)
    
    # ==========================================
    # STRONG CONSISTENCY OPERATIONS
    # ==========================================
    
    def get_account(self, user_id: str) -> Account:
        """User account: Strong consistency required."""
        return self.postgres.query(
            "SELECT * FROM accounts WHERE id = ?", 
            [user_id]
        )
    
    def transfer_funds(self, from_id: str, to_id: str, amount: float) -> bool:
        """Financial transaction: ACID required."""
        with self.postgres.transaction():
            self.postgres.execute(
                "UPDATE accounts SET balance = balance - ? WHERE id = ?",
                [amount, from_id]
            )
            self.postgres.execute(
                "UPDATE accounts SET balance = balance + ? WHERE id = ?",
                [amount, to_id]
            )
        return True  # Or raise on failure
    
    def reserve_inventory(self, product_id: str, quantity: int) -> bool:
        """Inventory: Strong consistency to prevent overselling."""
        # Use row-level locking for atomic decrement
        result = self.postgres.execute(
            """
            UPDATE inventory 
            SET available = available - ?
            WHERE product_id = ? AND available >= ?
            """,
            [quantity, product_id, quantity]
        )
        return result.rows_affected > 0
    
    # ==========================================
    # SESSION CONSISTENCY OPERATIONS
    # ==========================================
    
    def get_cart(self, user_id: str) -> Cart:
        """Shopping cart: Session consistency is sufficient."""
        # Redis with session affinity ensures user sees their updates
        cart_data = self.redis.get(f"cart:{user_id}")
        return Cart.from_json(cart_data) if cart_data else Cart()
    
    def add_to_cart(self, user_id: str, item: CartItem):
        """Add to cart: Fast, async replication is fine."""
        cart = self.get_cart(user_id)
        cart.add_item(item)
        self.redis.set(f"cart:{user_id}", cart.to_json(), ex=3600)
    
    def get_session(self, session_id: str) -> Session:
        """Session data: User only needs to see their own updates."""
        return self.redis.get(f"session:{session_id}")
    
    # ==========================================
    # EVENTUAL CONSISTENCY OPERATIONS
    # ==========================================
    
    def get_product(self, product_id: str) -> Product:
        """
        Product details: Eventual consistency is fine.
        Product changes are rare, and stale data briefly is OK.
        """
        return self.cassandra.query(
            "SELECT * FROM products WHERE id = ?",
            [product_id],
            consistency=ConsistencyLevel.ONE
        )
    
    def search_products(self, query: str) -> list[Product]:
        """Search: Eventual consistency (index may be slightly behind)."""
        return self.elasticsearch.search("products", query)
    
    def log_product_view(self, user_id: str, product_id: str):
        """Analytics event: Fire-and-forget, eventual is fine."""
        self.cassandra.execute(
            "INSERT INTO product_views (user_id, product_id, ts) VALUES (?, ?, ?)",
            [user_id, product_id, datetime.now()],
            consistency=ConsistencyLevel.ANY  # Maximum availability
        )
    
    def get_recommendations(self, user_id: str) -> list[Product]:
        """Recommendations: Computed periodically, staleness acceptable."""
        # Cached in Redis, refreshed every hour
        cached = self.redis.get(f"recs:{user_id}")
        if cached:
            return [Product.from_id(p) for p in json.loads(cached)]
        
        # Fallback to Cassandra with eventual consistency
        return self.cassandra.query(
            "SELECT * FROM user_recommendations WHERE user_id = ?",
            [user_id],
            consistency=ConsistencyLevel.ONE
        )
 
 
# The key insight: Different operations have legitimately different needs.
# Using strong consistency everywhere is wasteful.
# Using eventual consistency everywhere is dangerous.
# Polyglot consistency gives each operation what it actually needs.