# Distributed Systems

> Distributed Systems

- Skill: `viewway/distributed-systems` (Agent Skill)
- Install (CLI): `npx skillmds@latest add viewway/distributed-systems`
- Raw SKILL.md: https://api.skillmd.com/api/skills/viewway/distributed-systems/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: viewway (https://skillmd.com/u/viewway)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/viewway/distributed-systems

---

# Distributed Systems - SKILL.md

## Overview
Graduate-level distributed systems covering fundamental theory, consensus protocols, replication, messaging, storage, microservices architecture, and observability.

---

## Fundamental Theory

### Concept
Theoretical foundations that define the limits and possibilities of distributed computing.

### Principles

**CAP Theorem (Brewer's Theorem):**
- **Consistency:** Every read returns the most recent write or an error (linearizability)
- **Availability:** Every request receives a non-error response (no guarantee of freshness)
- **Partition Tolerance:** System continues to operate despite network partitions
- **Theorem:** During a network partition, you must choose between C and A
- **Practical implication:** All real systems must handle partitions → choose CP or AP

| System Type | C | A | Example |
|-------------|---|---|---------|
| CP | ✓ | ✗ | ZooKeeper, HBase, MongoDB (default) |
| AP | ✗ | ✓ | Cassandra, DynamoDB, CouchDB |
| CA | ✓ | ✓ | Impossible in distributed systems (single-node RDBMS) |

**FLP Impossibility Theorem:**
- In an asynchronous distributed system, consensus is impossible if even one process can crash
- No deterministic algorithm can guarantee consensus in bounded time
- **Practical workaround:** Use randomization (Paxos) or timeouts (Raft) to make progress

**Consistency Models (from strongest to weakest):**

| Model | Guarantee | Cost |
|-------|-----------|------|
| Linearizable | All operations appear atomic, globally ordered | Highest latency |
| Sequential | Operations of each process in order, total order | High latency |
| Causal | Causally related operations seen in order | Moderate overhead |
| Eventual | All replicas converge given no new updates | Low latency |

**Consistent Hashing:**
- Map both nodes and keys onto a hash ring (0 to 2^160-1)
- Each key is owned by the first node clockwise from its hash position
- **Virtual Nodes:** Each physical node maps to multiple positions on the ring
- **Advantage:** Adding/removing a node only affects K/N keys (K = total keys, N = nodes)
- **Used by:** DynamoDB, Cassandra, Riak, Akka

### Algorithms & Code

**Causal Consistency (Vector Clocks):**

```python
class VectorClock:
    def __init__(self, node_ids):
        self.clock = {nid: 0 for nid in node_ids}
    
    def increment(self, node_id):
        self.clock[node_id] += 1
    
    def merge(self, other):
        """Merge with another vector clock (take component-wise max)."""
        all_keys = set(self.clock) | set(other.clock)
        merged = {}
        for k in all_keys:
            merged[k] = max(self.clock.get(k, 0), other.clock.get(k, 0))
        self.clock = merged
        return self
    
    def happens_before(self, other):
        """Check if self → other (causal ordering)."""
        at_least_one_less = False
        for k in set(self.clock) | set(other.clock):
            if self.clock.get(k, 0) > other.clock.get(k, 0):
                return False
            if self.clock.get(k, 0) < other.clock.get(k, 0):
                at_least_one_less = True
        return at_least_one_less
    
    def concurrent(self, other):
        """Check if self || other (concurrent events)."""
        return not self.happens_before(other) and not other.happens_before(self)
    
    def __repr__(self):
        return f"VC({self.clock})"


class CausalStore:
    """Key-value store with causal consistency."""
    def __init__(self, node_id, node_ids):
        self.node_id = node_id
        self.data = {}  # key → (value, vector_clock)
        self.vc = VectorClock(node_ids)
    
    def write(self, key, value):
        self.vc.increment(self.node_id)
        self.data[key] = (value, VectorClock(dict(self.vc.clock)))
    
    def read(self, key):
        if key in self.data:
            return self.data[key]
        return None
    
    def sync(self, other_store, key):
        """Synchronize with another node for a key (resolve conflicts)."""
        if key not in other_store.data:
            return
        
        my_vc = self.data.get(key, (None, VectorClock([])))[1]
        their_vc = other_store.data[key][1]
        
        if my_vc.concurrent(their_vc):
            # Conflict! Apply last-writer-wins or merge
            self._resolve_conflict(key, self.data.get(key), other_store.data[key])
        elif their_vc.happens_before(my_vc):
            pass  # Our version is newer
        else:
            # Their version is newer
            self.data[key] = other_store.data[key]
            self.vc.merge(their_vc)
    
    def _resolve_conflict(self, key, my_entry, their_entry):
        """Last-writer-wins conflict resolution."""
        # In practice, use application-specific merge (CRDT)
        self.data[key] = their_entry  # Simple LWW
        self.vc.merge(their_entry[1])
```

**Consistent Hashing Ring:**

```python
import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, virtual_nodes=150):
        self.vnodes = virtual_nodes
        self.ring = []       # Sorted list of hash positions
        self.mapping = {}    # hash → node_id
        self.nodes = set()
    
    def _hash(self, key):
        return int(hashlib.sha256(key.encode()).hexdigest(), 16) % (2**128)
    
    def add_node(self, node_id):
        self.nodes.add(node_id)
        for i in range(self.vnodes):
            h = self._hash(f"{node_id}#vn{i}")
            bisect.insort(self.ring, h)
            self.mapping[h] = node_id
    
    def remove_node(self, node_id):
        self.nodes.discard(node_id)
        for i in range(self.vnodes):
            h = self._hash(f"{node_id}#vn{i}")
            idx = bisect.bisect_left(self.ring, h)
            if idx < len(self.ring) and self.ring[idx] == h:
                self.ring.pop(idx)
                del self.mapping[h]
    
    def get_node(self, key):
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect.bisect_right(self.ring, h)
        if idx == len(self.ring):
            idx = 0
        return self.mapping[self.ring[idx]]
    
    def get_replicas(self, key, n=3):
        """Get n distinct nodes for replication."""
        if not self.ring or n <= 0:
            return []
        h = self._hash(key)
        idx = bisect.bisect_right(self.ring, h)
        
        result = []
        seen = set()
        for i in range(len(self.ring)):
            pos = (idx + i) % len(self.ring)
            node = self.mapping[self.ring[pos]]
            if node not in seen:
                seen.add(node)
                result.append(node)
                if len(result) == n:
                    break
        return result
```

### Trade-offs

**Consistency Models:**
- **Strong (Linearizable):** Correct but slow (needs coordination)
- **Causal:** Good balance, vector clock overhead
- **Eventual:** Fast reads, stale data possible, conflict resolution needed

**Consistent Hashing:**
- **More virtual nodes:** Better distribution, more memory
- **Fewer virtual nodes:** Less overhead, potentially uneven distribution

### Applications

- **Distributed Caching:** Memcached consistent hashing
- **Database Sharding:** Cassandra token ring
- **CDN:** Routing requests to nearest edge node
- **Load Balancing:** Distributing requests across backends

---

## Consensus Protocols

### Concept
Algorithms that enable distributed nodes to agree on a single value or state, despite failures.

### Principles

**Paxos (Basic Paxos):**
- **Roles:** Proposer, Acceptor, Learner
- **Phases:**
  1. **Prepare:** Proposer sends numbered proposal to acceptors
  2. **Promise:** Acceptors promise not to accept lower-numbered proposals
  3. **Accept:** Proposer sends value, acceptors accept if number is highest seen
  4. **Learn:** Accepted value is communicated to learners

**Multi-Paxos:**
- Skip prepare phase when a stable leader exists
- Leader can directly send Accept requests
- Much more efficient for consecutive proposals

**Raft:**
- **Leader Election:** Term-based, randomized timeouts
  - Follower → Candidate (election timeout)
  - Candidate requests votes from all nodes
  - Majority vote → Leader
- **Log Replication:**
  - Leader appends to log, replicates to followers
  - Commit when majority have the entry
  - Followers apply committed entries to state machine
- **Safety:**
  - Election restriction: Candidate's log must be at least as up-to-date
  - Leader completeness: Committed entries are never lost
- **Membership Changes:** Joint consensus (old + new configuration)

**ZAB (ZooKeeper Atomic Broadcast):**
- Used by ZooKeeper for metadata management
- Similar to Raft: leader-based, term-based
- **Phases:** Discovery → Synchronization → Broadcast
- Primary use: Configuration management, leader election service

**PBFT (Practical Byzantine Fault Tolerance):**
- Tolerates f Byzantine (arbitrary) failures with 3f+1 nodes
- **Phases:** Pre-prepare → Prepare → Commit
- Each replica maintains a view number
- View change protocol for leader failure
- **Cost:** O(n²) message complexity per consensus round

### Algorithms & Code

**Raft Implementation (Core):**

```python
import random
import time
from enum import Enum

class NodeState(Enum):
    FOLLOWER = 'follower'
    CANDIDATE = 'candidate'
    LEADER = 'leader'

class LogEntry:
    def __init__(self, term, command):
        self.term = term
        self.command = command

class RaftNode:
    def __init__(self, node_id, peers):
        # Persistent state
        self.node_id = node_id
        self.current_term = 0
        self.voted_for = None
        self.log = [LogEntry(0, None)]  # Index 0 is sentinel
        
        # Volatile state
        self.commit_index = 0
        self.last_applied = 0
        self.state = NodeState.FOLLOWER
        self.leader_id = None
        
        # Leader state
        self.next_index = {}  # peer → next log index to send
        self.match_index = {}  # peer → highest replicated index
        
        # Election
        self.peers = peers
        self.votes_received = set()
        self.election_timeout = self._random_timeout()
        self.last_heartbeat = time.time()
    
    def _random_timeout(self):
        return random.uniform(150, 300)  # ms
    
    def tick(self):
        """Called periodically."""
        if self.state == NodeState.LEADER:
            self._send_heartbeats()
        else:
            if time.time() - self.last_heartbeat > self.election_timeout / 1000:
                self._start_election()
    
    def _start_election(self):
        self.current_term += 1
        self.state = NodeState.CANDIDATE
        self.voted_for = self.node_id
        self.votes_received = {self.node_id}
        self.last_heartbeat = time.time()
        
        last_log_idx = len(self.log) - 1
        last_log_term = self.log[-1].term
        
        for peer in self.peers:
            self._request_vote(peer, self.current_term, self.node_id,
                              last_log_idx, last_log_term)
    
    def handle_request_vote(self, term, candidate_id, last_log_idx, last_log_term):
        """Handle incoming RequestVote RPC."""
        if term < self.current_term:
            return {'term': self.current_term, 'vote_granted': False}
        
        if term > self.current_term:
            self._step_down(term)
        
        # Voting rules
        if self.voted_for is None or self.voted_for == candidate_id:
            # Log completeness check
            my_last_idx = len(self.log) - 1
            my_last_term = self.log[-1].term
            
            if last_log_term > my_last_term or \
               (last_log_term == my_last_term and last_log_idx >= my_last_idx):
                self.voted_for = candidate_id
                self.last_heartbeat = time.time()
                return {'term': self.current_term, 'vote_granted': True}
        
        return {'term': self.current_term, 'vote_granted': False}
    
    def handle_vote_response(self, term, vote_granted):
        """Handle response to our vote request."""
        if term > self.current_term:
            self._step_down(term)
            return
        
        if vote_granted and self.state == NodeState.CANDIDATE:
            self.votes_received.add(term)  # Simplified
            if len(self.votes_received) > (len(self.peers) + 1) // 2:
                self._become_leader()
    
    def _become_leader(self):
        self.state = NodeState.LEADER
        self.leader_id = self.node_id
        next_idx = len(self.log)
        for peer in self.peers:
            self.next_index[peer] = next_idx
            self.match_index[peer] = 0
    
    def handle_append_entries(self, term, leader_id, prev_log_idx, 
                               prev_log_term, entries, leader_commit):
        """Handle AppendEntries RPC."""
        if term < self.current_term:
            return {'term': self.current_term, 'success': False}
        
        self.last_heartbeat = time.time()
        
        if term > self.current_term:
            self._step_down(term)
        
        self.leader_id = leader_id
        
        # Log consistency check
        if prev_log_idx >= len(self.log):
            return {'term': self.current_term, 'success': False}
        
        if self.log[prev_log_idx].term != prev_log_term:
            # Log mismatch, delete conflicting entry and all following
            self.log = self.log[:prev_log_idx]
            return {'term': self.current_term, 'success': False}
        
        # Append new entries
        for i, entry in enumerate(entries):
            idx = prev_log_idx + 1 + i
            if idx < len(self.log):
                if self.log[idx].term != entry.term:
                    self.log = self.log[:idx]
                    self.log.append(entry)
            else:
                self.log.append(entry)
        
        # Update commit index
        if leader_commit > self.commit_index:
            self.commit_index = min(leader_commit, len(self.log) - 1)
            self._apply_committed()
        
        return {'term': self.current_term, 'success': True}
    
    def _send_heartbeats(self):
        """Send AppendEntries to all peers."""
        for peer in self.peers:
            prev_idx = self.next_index[peer] - 1
            prev_term = self.log[prev_idx].term if prev_idx < len(self.log) else 0
            entries = self.log[self.next_index[peer]:]
            
            self._append_entries(peer, self.current_term, self.node_id,
                                prev_idx, prev_term, entries, self.commit_index)
    
    def _apply_committed(self):
        """Apply committed entries to state machine."""
        while self.last_applied < self.commit_index:
            self.last_applied += 1
            entry = self.log[self.last_applied]
            # Apply entry.command to state machine
    
    def _step_down(self, term):
        self.current_term = term
        self.state = NodeState.FOLLOWER
        self.voted_for = None
    
    def propose(self, command):
        """Propose a new command (leader only)."""
        if self.state != NodeState.LEADER:
            return False
        
        entry = LogEntry(self.current_term, command)
        self.log.append(entry)
        
        # Replicate to followers
        self._send_heartbeats()
        
        # Check if committed (majority replicated)
        # ... (handle in append_entries responses)
        return True
```

**PBFT (Simplified):**

```python
class PBFTNode:
    def __init__(self, node_id, all_nodes, f=1):
        self.node_id = node_id
        self.all_nodes = all_nodes  # 3f + 1 nodes
        self.f = f
        self.view = 0
        self.sequence = 0
        self.log = []  # Messages seen
        self.prepared = {}   # (view, seq) → message
        self.committed = {}  # (view, seq) → message
    
    def request(self, operation):
        """Client sends request to primary."""
        msg = {'type': 'request', 'op': operation, 
               'timestamp': time.time(), 'client': self.node_id}
        primary = self.all_nodes[self.view % len(self.all_nodes)]
        primary.handle_request(msg)
    
    def handle_request(self, msg):
        """Primary handles client request."""
        if self.node_id != self.all_nodes[self.view % len(self.all_nodes)].node_id:
            return  # Not primary
        
        self.sequence += 1
        pre_prepare = {
            'type': 'pre-prepare', 'view': self.view,
            'sequence': self.sequence, 'digest': hash(str(msg)),
            'message': msg
        }
        self.log.append(pre_prepare)
        
        # Broadcast to all replicas
        for node in self.all_nodes:
            node.handle_pre_prepare(pre_prepare)
    
    def handle_pre_prepare(self, msg):
        """Replica handles pre-prepare from primary."""
        self.log.append(msg)
        
        prepare = {
            'type': 'prepare', 'view': msg['view'],
            'sequence': msg['sequence'], 'digest': msg['digest'],
            'node_id': self.node_id
        }
        
        # Send prepare to all nodes
        for node in self.all_nodes:
            node.handle_prepare(prepare)
    
    def handle_prepare(self, msg):
        """Handle prepare message from another replica."""
        key = (msg['view'], msg['sequence'])
        if key not in self.prepared:
            self.prepared[key] = []
        self.prepared[key].append(msg)
        
        # Check if we have 2f prepares (including our own)
        if len(self.prepared[key]) >= 2 * self.f:
            # Prepared! Send commit
            commit = {
                'type': 'commit', 'view': msg['view'],
                'sequence': msg['sequence'], 'digest': msg['digest'],
                'node_id': self.node_id
            }
            for node in self.all_nodes:
                node.handle_commit(commit)
    
    def handle_commit(self, msg):
        """Handle commit message."""
        key = (msg['view'], msg['sequence'])
        if key not in self.committed:
            self.committed[key] = []
        self.committed[key].append(msg)
        
        # Check if we have 2f+1 commits
        if len(self.committed[key]) >= 2 * self.f + 1:
            # Committed! Execute operation
            self._execute(msg)
    
    def _execute(self, msg):
        """Execute the committed operation."""
        # Apply to state machine
        pass
```

### Trade-offs

**Consensus Protocol Selection:**

| Protocol | Fault Tolerance | Message Complexity | Latency | Use Case |
|----------|-----------------|-------------------|---------|----------|
| Paxos | Crash (f < N/2) | O(N) (Multi-Paxos) | 2 RTT | Chubby, Spanner |
| Raft | Crash (f < N/2) | O(N) | 2 RTT | etcd, Consul, TiKV |
| ZAB | Crash (f < N/2) | O(N) | 2 RTT | ZooKeeper |
| PBFT | Byzantine (f < N/3) | O(N²) | 3 RTT | Blockchain, Hyperledger |

**Raft vs Paxos:**
- Raft: Easier to understand, leader-based, strong consistency
- Paxos: More flexible, harder to implement, proven correct

### Applications

- **Configuration Management:** etcd (Raft), ZooKeeper (ZAB)
- **Distributed Locking:** Consul, ZooKeeper
- **Database Replication:** TiKV (Raft), CockroachDB (Raft)
- **Blockchain:** PBFT variants for permissioned chains

---

## Distributed Transactions

### Concept
Protocols for executing transactions that span multiple nodes or services.

### Principles

**2PC (Two-Phase Commit):**
1. **Prepare Phase:** Coordinator asks all participants to prepare
2. **Commit/Abort Phase:** If all prepared → commit; if any abort → abort all
- **Blocking Problem:** If coordinator crashes after prepare, participants are locked
- **Solution:** 3PC adds pre-commit phase (requires failure detector)

**3PC (Three-Phase Commit):**
1. **CanCommit:** Coordinator checks if participants can commit
2. **PreCommit:** Participants prepare and acknowledge
3. **DoCommit:** Coordinator sends final commit
- **Advantage:** Non-blocking (if network doesn't partition)
- **Disadvantage:** More message rounds, still vulnerable to partitions

**TCC (Try-Confirm-Cancel):**
- **Try:** Reserve resources (check inventory, hold funds)
- **Confirm:** Commit the operation (deduct inventory, charge card)
- **Cancel:** Release reserved resources (release hold)
- Each service must implement try, confirm, cancel operations
- Used in: Alibaba Seata, service mesh transaction frameworks

**Saga Pattern:**
- **Choreography:** Each service emits events, next service reacts
- **Orchestration:** Central coordinator invokes services in sequence
- **Compensation:** Each step has a compensating action for rollback
- **Guarantee:** Eventually consistent (not ACID)

**Distributed Snapshot (Chandy-Lamport):**
- Record consistent global state without stopping the system
- Uses marker messages on each channel
- Algorithm:
  1. Initiator records its state, sends markers on all outgoing channels
  2. On receiving first marker: record state, send markers, start recording incoming channel
  3. On receiving subsequent markers: stop recording that channel
  4. Collected states form a consistent cut

### Algorithms & Code

**2PC Coordinator:**

```python
class Coordinator2PC:
    def __init__(self, participants, timeout=5.0):
        self.participants = participants
        self.timeout = timeout
        self.transaction_log = {}
    
    def execute(self, transaction_id, operations):
        """Execute distributed transaction using 2PC."""
        self.transaction_log[transaction_id] = {'status': 'preparing'}
        
        # Phase 1: Prepare
        votes = {}
        for participant, op in zip(self.participants, operations):
            try:
                vote = participant.prepare(transaction_id, op)
                votes[participant] = vote
            except TimeoutError:
                votes[participant] = 'abort'
            except Exception:
                votes[participant] = 'abort'
        
        # Phase 2: Commit or Abort
        if all(v == 'prepared' for v in votes.values()):
            decision = 'commit'
        else:
            decision = 'abort'
        
        self.transaction_log[transaction_id]['status'] = decision
        
        for participant in self.participants:
            try:
                if decision == 'commit':
                    participant.commit(transaction_id)
                else:
                    participant.abort(transaction_id)
            except Exception:
                # Retry until successful (participants must be able to commit eventually)
                self._retry(participant, transaction_id, decision)
        
        return decision
    
    def _retry(self, participant, txid, decision):
        """Retry commit/abort until successful."""
        for _ in range(3):
            try:
                if decision == 'commit':
                    participant.commit(txid)
                else:
                    participant.abort(txid)
                return
            except Exception:
                time.sleep(1)
        # Log for manual intervention
        print(f"CRITICAL: Failed to {decision} tx {txid} on {participant}")


class Participant2PC:
    def __init__(self, resource_manager):
        self.rm = resource_manager
        self.prepared = {}  # txid → undo_info
    
    def prepare(self, txid, operation):
        """Prepare to commit. Write undo log."""
        undo_info = self.rm.apply_temporarily(txid, operation)
        self.prepared[txid] = undo_info
        # Write prepare record to durable log
        return 'prepared'
    
    def commit(self, txid):
        """Commit the prepared transaction."""
        self.rm.make_permanent(txid)
        del self.prepared[txid]
    
    def abort(self, txid):
        """Abort and undo the transaction."""
        if txid in self.prepared:
            self.rm.undo(txid, self.prepared[txid])
            del self.prepared[txid]
```

**Saga Orchestrator with Compensation:**

```python
class SagaDefinition:
    def __init__(self):
        self.steps = []
    
    def add_step(self, name, action_fn, compensate_fn):
        self.steps.append({
            'name': name,
            'action': action_fn,
            'compensate': compensate_fn
        })
        return self

class SagaExecution:
    def __init__(self, definition):
        self.definition = definition
        self.completed_steps = []  # (step_index, result)
    
    def execute(self):
        """Execute saga with automatic compensation on failure."""
        for i, step in enumerate(self.definition.steps):
            try:
                result = step['action']()
                self.completed_steps.append((i, result))
            except Exception as e:
                print(f"Saga failed at step '{step['name']}': {e}")
                self._compensate()
                raise
        
        return {'status': 'completed', 'steps': len(self.definition.steps)}
    
    def _compensate(self):
        """Compensate completed steps in reverse order."""
        for i, result in reversed(self.completed_steps):
            step = self.definition.steps[i]
            try:
                step['compensate'](result)
            except Exception as e:
                print(f"Compensation failed for step '{step['name']}': {e}")
                # Log for manual intervention (dead letter queue)

# Example: Order processing saga
saga = SagaDefinition()
saga.add_step(
    'reserve_stock',
    action_fn=lambda: {'reservation_id': 'res-123'},
    compensate_fn=lambda res: release_stock(res['reservation_id'])
)
saga.add_step(
    'process_payment',
    action_fn=lambda: {'payment_id': 'pay-456'},
    compensate_fn=lambda res: refund_payment(res['payment_id'])
)
saga.add_step(
    'create_order',
    action_fn=lambda: {'order_id': 'ord-789'},
    compensate_fn=lambda res: cancel_order(res['order_id'])
)
saga.add_step(
    'schedule_delivery',
    action_fn=lambda: {'tracking_id': 'trk-012'},
    compensate_fn=lambda res: cancel_delivery(res['tracking_id'])
)

execution = SagaExecution(saga)
execution.execute()
```

**Chandy-Lamport Distributed Snapshot:**

```python
class DistributedSnapshot:
    def __init__(self, node_id, channels):
        self.node_id = node_id
        self.channels = channels  # Incoming channels from other nodes
        self.state = None
        self.channel_states = {ch: [] for ch in channels}
        self.markers_received = set()
        self.snapshot_complete = False
    
    def initiate_snapshot(self):
        """Start a snapshot (called by initiator)."""
        # Step 1: Record own state
        self.state = self._get_current_state()
        
        # Step 2: Send markers on all outgoing channels
        for channel in self.channels:
            self._send_marker(channel)
        
        # Step 3: Start recording incoming channels
        self.recording = True
    
    def handle_marker(self, channel, marker):
        """Handle incoming marker message."""
        if self.state is None:
            # First marker: record state, send markers, start recording
            self.state = self._get_current_state()
            self.markers_received.add(channel)
            
            for ch in self.channels:
                self._send_marker(ch)
            
            self.recording = True
        else:
            # Subsequent marker: stop recording this channel
            self.markers_received.add(channel)
        
        # Check if snapshot is complete
        if self.markers_received == set(self.channels):
            self.snapshot_complete = True
    
    def handle_message(self, channel, message):
        """Handle regular (non-marker) message."""
        if self.recording and channel not in self.markers_received:
            # Record message as part of channel state
            self.channel_states[channel].append(message)
        
        # Process message normally
        self._process(message)
    
    def _get_current_state(self):
        """Capture current application state."""
        return {'node': self.node_id, 'data': dict(self._application_state)}
    
    def _send_marker(self, channel):
        """Send marker on a channel."""
        channel.send({'type': 'marker', 'source': self.node_id})
    
    def _process(self, message):
        """Process application message."""
        pass
```

### Trade-offs

**Transaction Models:**

| Model | Consistency | Performance | Complexity | Failure Handling |
|-------|------------|-------------|------------|-----------------|
| 2PC | Strong | Low (blocking) | Medium | Coordinator failure blocks |
| TCC | Strong | Medium | High | Requires compensation logic |
| Saga | Eventual | High | Medium | Compensation chain |
| 1PC (per service) | Weak | Highest | Low | Manual reconciliation |

**Saga Choreography vs Orchestration:**
- **Choreography:** Decentralized, event-driven, harder to debug
- **Orchestration:** Centralized, easier to monitor, single point of failure

### Applications

- **E-Commerce:** Order processing across inventory, payment, shipping services
- **Banking:** Fund transfers across different banking systems
- **Travel Booking:** Flight + hotel + car reservation
- **Microservices:** Cross-service business transactions

---

## Replication

### Concept
Maintaining copies of data across multiple nodes for availability, durability, and read scaling.

### Principles

**Replication Topologies:**

| Type | Write Path | Read Path | Conflict Risk |
|------|-----------|-----------|--------------|
| Primary-Secondary | Write to primary, replicate | Read from any | None (single writer) |
| Multi-Master | Write to any | Read from any | High (concurrent writes) |
| Consensus (Raft) | Write to leader | Read from leader | None (consensus) |

**Read-Write Quorums (Dynamo-style):**
- N = replication factor
- W = write quorum (min writes acknowledged)
- R = read quorum (min reads to contact)
- **Consistency guarantee:** W + R > N → strong read-after-write consistency
- **Common configurations:**
  - Strong: R=2, W=2, N=3 (any 2 overlap)
  - Eventual: R=1, W=1, N=3 (fast but stale reads possible)

**Anti-Entropy:**
- **Read Repair:** Fix inconsistencies detected during reads
- **Hinted Handoff:** Store writes for temporarily down nodes
- **Merkle Tree:** Efficiently detect differences between replicas
  - Hash tree of data partitions
  - Compare root hashes; if different, compare children recursively
  - O(log n) comparisons to find divergent data

**Conflict Resolution Strategies:**
- **Last Writer Wins (LWW):** Use timestamp, simple but lossy
- **Application-level merge:** Custom logic (e.g., CRDTs)
- **CRDT (Conflict-free Replicated Data Types):**
  - **G-Counter:** Grow-only counter (max of per-node counts)
  - **PN-Counter:** Increment/decrement counter (two G-counters)
  - **G-Set:** Grow-only set (union of all elements)
  - **OR-Set:** Observed-remove set (add wins with unique tags)
  - **LWW-Register:** Last-writer-wins register

### Algorithms & Code

**Merkle Tree for Anti-Entropy:**

```python
import hashlib

class MerkleTree:
    def __init__(self, data_blocks, leaf_size=1024):
        self.data = data_blocks
        self.leaf_size = leaf_size
        self.tree = self._build_tree()
    
    def _hash(self, data):
        return hashlib.sha256(str(data).encode()).hexdigest()
    
    def _build_tree(self):
        """Build Merkle tree bottom-up."""
        # Create leaf nodes
        leaves = []
        for i in range(0, len(self.data), self.leaf_size):
            block = self.data[i:i+self.leaf_size]
            leaves.append(self._hash(block))
        
        # Build internal nodes
        tree = [leaves]
        while len(tree[-1]) > 1:
            level = tree[-1]
            parent_level = []
            for i in range(0, len(level), 2):
                left = level[i]
                right = level[i+1] if i+1 < len(level) else left
                parent_level.append(self._hash(left + right))
            tree.append(parent_level)
        
        return tree
    
    @property
    def root_hash(self):
        return self.tree[-1][0] if self.tree else None
    
    def diff(self, other):
        """Find differences with another Merkle tree."""
        if self.root_hash == other.root_hash:
            return []  # No differences
        
        return self._find_diffs(other, len(self.tree)-1, 0)
    
    def _find_diffs(self, other, level, index):
        """Recursively find differing leaf indices."""
        if level == 0:
            return [index]  # Leaf level, this block differs
        
        if self.tree[level][index] == other.tree[level][index]:
            return []  # Same hash, no differences below
        
        # Different hash, check children
        left_child = index * 2
        right_child = index * 2 + 1
        
        diffs = []
        if left_child < len(self.tree[level-1]):
            diffs.extend(self._find_diffs(other, level-1, left_child))
        if right_child < len(self.tree[level-1]):
            diffs.extend(self._find_diffs(other, level-1, right_child))
        
        return diffs


class AntiEntropySync:
    """Synchronize data between replicas using Merkle trees."""
    def __init__(self, node_id, data):
        self.node_id = node_id
        self.data = data  # Dict of key-value pairs
        self.merkle = MerkleTree(list(data.items()))
    
    def sync_with(self, other_node):
        """Synchronize with another node."""
        # Compare Merkle tree roots
        diffs = self.merkle.diff(other_node.merkle)
        
        if not diffs:
            return  # Already in sync
        
        # Exchange differing data
        for diff_idx in diffs:
            my_version = self._get_block(diff_idx)
            their_version = other_node._get_block(diff_idx)
            
            # Resolve conflict (LWW)
            for key in set(list(my_version.keys()) + list(their_version.keys())):
                my_ts = my_version.get(key, {}).get('timestamp', 0)
                their_ts = their_version.get(key, {}).get('timestamp', 0)
                
                if my_ts > their_ts:
                    other_node.data[key] = self.data[key]
                elif their_ts > my_ts:
                    self.data[key] = other_node.data[key]
        
        # Rebuild Merkle trees
        self.merkle = MerkleTree(list(self.data.items()))
        other_node.merkle = MerkleTree(list(other_node.data.items()))
```

**CRDT (G-Counter & PN-Counter):**

```python
class GCounter:
    """Grow-only counter CRDT."""
    def __init__(self, node_id, all_nodes):
        self.node_id = node_id
        self.counts = {node: 0 for node in all_nodes}
    
    def increment(self, amount=1):
        self.counts[self.node_id] += amount
    
    def value(self):
        return sum(self.counts.values())
    
    def merge(self, other):
        """Merge with another G-Counter (take max per node)."""
        for node in self.counts:
            self.counts[node] = max(self.counts[node], other.counts[node])


class PNCounter:
    """Positive-Negative counter CRDT."""
    def __init__(self, node_id, all_nodes):
        self.p = GCounter(node_id, all_nodes)  # Positive
        self.n = GCounter(node_id, all_nodes)  # Negative
        self.node_id = node_id
    
    def increment(self, amount=1):
        self.p.increment(amount)
    
    def decrement(self, amount=1):
        self.n.increment(amount)
    
    def value(self):
        return self.p.value() - self.n.value()
    
    def merge(self, other):
        self.p.merge(other.p)
        self.n.merge(other.n)


class ORSet:
    """Observed-Remove Set CRDT."""
    def __init__(self, node_id):
        self.node_id = node_id
        self.elements = {}  # element → set of unique tags
        self.tombstones = set()  # Removed tags
    
    def add(self, element):
        tag = (self.node_id, time.time())
        if element not in self.elements:
            self.elements[element] = set()
        self.elements[element].add(tag)
    
    def remove(self, element):
        if element in self.elements:
            self.tombstones |= self.elements[element]
            del self.elements[element]
    
    def contains(self, element):
        return element in self.elements and \
               bool(self.elements[element] - self.tombstones)
    
    def get_elements(self):
        return {e for e, tags in self.elements.items() 
                if tags - self.tombstones}
    
    def merge(self, other):
        """Merge with another OR-Set."""
        # Add: keep all tags
        for element, tags in other.elements.items():
            if element not in self.elements:
                self.elements[element] = set()
            self.elements[element] |= tags
        
        # Remove: keep all tombstones
        self.tombstones |= other.tombstones
        
        # Clean up: remove tombstoned elements
        for element in list(self.elements):
            remaining = self.elements[element] - self.tombstones
            if not remaining:
                del self.elements[element]
```

### Trade-offs

**Replication Factor (N):**
- Higher N: Better durability and availability, more storage and write cost
- Lower N: Less overhead, lower durability

**Quorum Configuration (R/W):**
- R+W > N: Strong consistency, higher latency
- R+W ≤ N: Eventual consistency, lower latency

**Synchronous vs Asynchronous Replication:**
- **Synchronous:** Zero data loss, higher write latency
- **Asynchronous:** Lower latency, potential data loss on failover

### Applications

- **Database Replication:** PostgreSQL streaming, MySQL GTID, MongoDB replica sets
- **Caching:** Redis replication, Memcached consistent hashing
- **File Systems:** HDFS replication (default 3x)
- **CDN:** Content replication to edge nodes

---

## Message Queues

### Concept
Asynchronous communication infrastructure enabling decoupled, reliable message delivery between services.

### Principles

**Apache Kafka:**
- **Topics:** Categories/feeds where messages are published
- **Partitions:** Ordered, immutable sequence of messages within a topic
- **Offsets:** Unique sequence ID per partition (consumer tracks position)
- **Consumer Groups:** Set of consumers cooperating to consume a topic
  - Each partition consumed by exactly one consumer in a group
  - Enables parallel consumption and replay
- **Replication:** Each partition replicated across brokers (leader + followers)
- **Exactly-Once Semantics:**
  - Idempotent producers (PID + sequence number)
  - Transactional producers (atomic writes across partitions)
  - Consumer read-process-write with transactions

**Kafka Architecture:**
```
Producer → [Broker 1: P0(L), P1(F)] → Consumer Group A
         → [Broker 2: P0(F), P1(L)] → Consumer Group B
         → [Broker 3: P2(L)]
```

**RabbitMQ:**
- **Exchange Types:**
  - **Direct:** Route by exact routing key
  - **Fanout:** Broadcast to all bound queues
  - **Topic:** Route by pattern (stock.*.price)
  - **Headers:** Route by message headers
- **Queue:** Buffer storing messages until consumed
- **Binding:** Link between exchange and queue with routing rules
- **Acknowledgments:** Consumer acks message after processing
- **Dead Letter Exchange:** Route failed/expired messages

**Apache Pulsar:**
- **Layered Architecture:** Separate compute (brokers) and storage (bookies)
- **Topics → Bundles → Brokers:** Automatic load balancing
- **Persistent vs Non-persistent:** Durable (BookKeeper) vs in-memory
- **Multi-tenancy:** Built-in tenant/namespace isolation
- **Geo-replication:** Cross-region replication

### Algorithms & Code

**Kafka Producer (Python):**

```python
from kafka import KafkaProducer, KafkaConsumer
import json

class ReliableKafkaProducer:
    def __init__(self, bootstrap_servers, transactional_id=None):
        config = {
            'bootstrap_servers': bootstrap_servers,
            'value_serializer': lambda v: json.dumps(v).encode(),
          

…(truncated)