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

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):

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):

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:

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):

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):

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:

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:

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:

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:

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:

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:

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:

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):

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):

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)