System Design: CAP Theorem and Trade-offs

class DistributedKVStore:
    def __init__(self, node_id, peers, consistency_mode="CP"):
        self.node_id = node_id
        self.peers = peers  # List of peer node connections
        self.data = {}
        self.consistency_mode = consistency_mode
    
    def write(self, key, value):
        if self.consistency_mode == "CP":
            # Wait for majority acknowledgment before confirming
            acks = 1  # Count self
            for peer in self.peers:
                try:
                    peer.replicate(key, value, timeout=5)
                    acks += 1
                except NetworkTimeout:
                    continue
            
            if acks < (len(self.peers) + 1) // 2 + 1:
                raise UnavailableError("Cannot reach quorum")
            
            self.data[key] = value
            return "OK"
        
        else:  # AP mode
            # Write locally, replicate asynchronously
            self.data[key] = value
            for peer in self.peers:
                self._async_replicate(peer, key, value)
            return "OK"  # Always responds
    
    def read(self, key):
        if self.consistency_mode == "CP":
            # Read from quorum to ensure consistency
            values = [self.data.get(key)]
            for peer in self.peers:
                try:
                    values.append(peer.get(key, timeout=5))
                except NetworkTimeout:
                    continue
            
            if len(values) < (len(self.peers) + 1) // 2 + 1:
                raise UnavailableError("Cannot reach quorum")
            
            return self._resolve_latest(values)
        
        else:  # AP mode
            # Return local value immediately
            return self.data.get(key)

import redis

# CP-style configuration: Redis Cluster with WAIT
def cp_write(client, key, value):
    """
    Blocks until write is replicated to specified number of replicas.
    May timeout and fail if replicas are unreachable.
    """
    pipe = client.pipeline()
    pipe.set(key, value)
    pipe.wait(numreplicas=2, timeout=5000)  # Wait for 2 replicas, 5s timeout
    results = pipe.execute()
    
    replicas_confirmed = results[1]
    if replicas_confirmed < 2:
        raise Exception(f"Only {replicas_confirmed} replicas confirmed")
    return True

# AP-style configuration: Fire and forget
def ap_write(client, key, value):
    """
    Returns immediately after primary acknowledges.
    Replication happens asynchronously.
    """
    client.set(key, value)
    return True  # Always succeeds if primary is up

# Reading with different consistency requirements
def cp_read(client, key):
    """
    Read from primary only to ensure latest value.
    Fails if primary is unreachable.
    """
    return client.get(key)  # Redis reads from primary by default

def ap_read(client, key):
    """
    Allow reading from replicas for higher availability.
    May return stale data.
    """
    client.readonly()  # Enable replica reads
    return client.get(key)

// AP configuration: Fast, potentially inconsistent
db.orders.insertOne(
  { orderId: "12345", status: "pending" },
  { writeConcern: { w: 1 } }  // Acknowledge after primary write
)

// CP configuration: Slower, strongly consistent
db.orders.insertOne(
  { orderId: "12345", status: "pending" },
  { writeConcern: { w: "majority", j: true } }  // Wait for majority + journal
)

// Read concern affects consistency too
db.orders.find({ orderId: "12345" }).readConcern("majority")
// Only returns data acknowledged by majority of nodes

-- On primary: Configure synchronous replication
ALTER SYSTEM SET synchronous_commit = 'on';
ALTER SYSTEM SET synchronous_standby_names = 'replica1, replica2';

-- Transactions wait for replica acknowledgment
BEGIN;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;  -- Blocks until replica confirms

// CP configuration
Properties props = new Properties();
props.put("acks", "all");  // Wait for all in-sync replicas
props.put("min.insync.replicas", 2);
props.put("retries", Integer.MAX_VALUE);

// AP configuration  
Properties props = new Properties();
props.put("acks", "1");  // Only wait for leader
props.put("retries", 0);

class ShoppingCart:
    """
    AP shopping cart with vector clock conflict resolution.
    Multiple nodes can accept concurrent updates.
    """
    
    def __init__(self):
        self.items = {}
        self.vector_clock = {}  # {node_id: counter}
    
    def add_item(self, node_id, item_id, quantity):
        # Increment this node's clock
        self.vector_clock[node_id] = self.vector_clock.get(node_id, 0) + 1
        
        # Add item with clock timestamp
        self.items[item_id] = {
            "quantity": quantity,
            "clock": self.vector_clock.copy()
        }
    
    def merge(self, other_cart):
        """
        Merge concurrent updates from another node.
        Uses vector clocks to detect and resolve conflicts.
        """
        for item_id, other_item in other_cart.items.items():
            if item_id not in self.items:
                self.items[item_id] = other_item
            else:
                my_item = self.items[item_id]
                
                if self._clock_greater(other_item["clock"], my_item["clock"]):
                    # Other is newer, take it
                    self.items[item_id] = other_item
                elif self._clock_concurrent(other_item["clock"], my_item["clock"]):
                    # Concurrent updates: merge by taking max quantity
                    self.items[item_id] = {
                        "quantity": max(my_item["quantity"], other_item["quantity"]),
                        "clock": self._merge_clocks(my_item["clock"], other_item["clock"])
                    }
                # else: my item is newer, keep it