Distributed Systems Design Patterns: Consensus, Replication &...
Distributed systems present unique challenges: network partitions, partial failures, variable latency, and consistency trade-offs. Design patterns help architects and engineers navigate these challenges with proven solutions. This guide covers the most important distributed systems patterns and their practical applications, drawing on experience from production systems at Google, Amazon, Netflix, and other industry leaders.
Consistency and Consensus Patterns
Leader Election
In distributed systems, many coordination tasks require a single decision-maker. Leader election patterns choose one node to coordinate tasks like assigning work, managing replicas, or controlling access to shared resources.
Algorithms: Raft and Paxos are the two dominant consensus-based leader election algorithms. Raft, developed by Diego Ongaro and John Ousterhout at Stanford, was designed specifically for understandability. It breaks consensus into three subproblems: leader election, log replication, and safety. Raft guarantees that at most one leader exists per term and that a leader’s log entries are committed only when replicated to a majority of nodes.
Production Implementations: etcd (used by Kubernetes) implements Raft. ZooKeeper implements Zab, a protocol similar to Paxos. Consul uses Raft. These systems provide leader and observe endpoints that clients use to discover the current leader.
Fault Tolerance: The leader sends periodic heartbeats to followers. When followers detect leader failure (a configurable election timeout passes without a heartbeat), they transition to candidate state and trigger a new election. The election timeout should be randomized to prevent multiple candidates from splitting votes.
Distributed Consensus
Consensus algorithms ensure multiple nodes agree on a value despite failures. They form the foundation of replicated state machines used in key-value stores, configuration services, and coordination services.
Paxos: Leslie Lamport’s Paxos was the first practical consensus algorithm. It guarantees safety (agreement, validity, non-triviality) under asynchronous network conditions with crash failures. Paxos is notoriously difficult to understand and implement correctly — Lamport himself acknowledged this in “Paxos Made Simple.”
Raft: Raft’s design goal was understandability. It uses a stronger leader model: the leader handles all client requests, manages log replication, and determines what entries are committed. Raft’s log structure ensures that once an entry is committed, all future leaders will contain that entry. Raft handles cluster membership changes through a joint consensus approach that allows nodes to be added or removed without downtime.
The CAP Theorem: Eric Brewer’s CAP theorem states that a distributed data store can provide at most two of three guarantees simultaneously: Consistency (all nodes see the same data at the same time), Availability (every request receives a response), and Partition Tolerance (the system continues to operate despite network partitions). Since network partitions are inevitable, architects must choose between CP (sacrifice availability) and AP (sacrifice consistency). The PACELC extension adds that even without partitions, there is a trade-off between latency and consistency.
Data Management Patterns
Sharding
Sharding distributes data across multiple databases by partitioning key ranges or using hash functions. Each shard contains a subset of the data, enabling horizontal scaling.
Hash-Based Sharding: A hash function maps each key to a shard ID. Consistent hashing minimizes re-sharding when nodes are added or removed. When a node joins, only its immediate neighbor’s data must be redistributed. Amazon Dynamo and Cassandra use consistent hashing.
Range-Based Sharding: Data is partitioned by key ranges (A-D, E-H, etc.). This preserves ordering for range queries but risks hot spots if the distribution is skewed. Bigtable and HBase use range-based sharding.
Routing: The application must route queries to the correct shard. Client-side routing embeds shard mapping in the application. Proxy-based routing (Vitess, Citus) intercepts queries and routes them transparently. Global secondary indexes store a second copy of data indexed by a different key, enabling efficient non-primary-key lookups across shards.
Replication
Replication copies data across multiple nodes for fault tolerance and read scalability.
Leader-Based Replication (single-leader, primary-secondary): All writes go to the leader, which propagates changes to followers. Followers can serve read queries. MySQL replication and PostgreSQL streaming replication use this model. Failover requires promoting a follower to leader, which introduces potential data loss depending on synchronization mode.
Multi-Leader Replication: Multiple nodes accept writes. This is common in multi-datacenter deployments where each datacenter has its own leader. Conflict resolution is the main challenge — last-writer-wins (LWW) based on timestamps is common but risks data loss. CRDTs (Conflict-Free Replicated Data Types) provide automatic conflict resolution for specific data structures.
Leaderless Replication: Any node can accept writes (Amazon Dynamo, Cassandra). Read repair and hinted handoff maintain consistency. Quorum-based reads and writes (R + W > N) ensure read-your-writes consistency.
CQRS
Command Query Responsibility Segregation separates read and write models. Commands handle mutations, writes go to the write model, and queries read from a potentially different read model. CQRS pairs naturally with event sourcing: the write model stores events, and the read model projects those events into query-optimized structures. This pattern is used at scale by companies like Event Store and Axon Framework.
Resilience Patterns
Circuit Breaker
The circuit breaker pattern prevents cascading failures by detecting when a downstream service is unhealthy and stopping requests to it. The circuit has three states: closed (normal operation, requests pass through), open (requests fail immediately without calling the downstream service), and half-open (testing if the service has recovered).
When failures exceed a configurable threshold (typically 5-10 consecutive failures), the circuit opens. After a timeout (typically 5-60 seconds), it transitions to half-open and allows a probe request. Successful probes close the circuit; failures reopen it. Netflix Hystrix and Resilience4j are production implementations.
Bulkhead
The bulkhead pattern isolates resources to prevent failure in one part of the system from overwhelming other parts. Named after ship compartments, bulkheads allocate dedicated thread pools, connection pools, or resources for each service or component. If one service fails, its bulkhead contains the failure, and other services continue unaffected. The pattern is essential in microservices architectures where one slow endpoint can exhaust the entire thread pool.
Retry with Backoff
Transient failures (network timeouts, temporary unavailability) are inevitable in distributed systems. The retry pattern re-executes failed operations with increasing delays. Exponential backoff multiplies the delay by a factor after each retry (e.g., 100ms, 200ms, 400ms, 800ms). Jitter adds randomness to prevent multiple clients from retrying simultaneously (the thundering herd problem). Amazon recommends jittered exponential backoff for AWS API calls.
Communication Patterns
Event-Driven Architecture
Event-driven systems communicate through asynchronous events rather than synchronous requests. Producers emit events without knowing which consumers will process them. This decoupling enables scalable, reactive systems.
Apache Kafka is the dominant event broker for production systems. It provides durable, ordered, partitioned logs with at-least-once delivery guarantees. Kafka’s log compaction retains the latest value for each key, enabling state restoration. RabbitMQ provides message queues with routing flexibility. AWS SQS and SNS provide managed serverless messaging.
Saga Pattern
Sagas manage distributed transactions across multiple services without two-phase commit (which does not scale in distributed systems). A saga is a sequence of local transactions, each with a compensating action that undoes it.
Choreography-based sagas: Each service publishes events after its local transaction completes. Other services subscribe to relevant events and execute their own transactions. This is simple but makes the saga flow harder to track.
Orchestration-based sagas: A centralized coordinator (orchestrator) tells each service what to do. The orchestrator tracks the saga state and triggers compensating actions on failure. This provides better visibility and control. Netflix Conductor, Camunda, and AWS Step Functions are orchestration platforms.
Observability Patterns
Distributed systems require comprehensive observability. Distributed tracing follows requests across services, identifying latency bottlenecks. OpenTelemetry is the industry standard for trace collection. Structured logging provides searchable event records — avoid unstructured log messages; use JSON with consistent field names. Metrics aggregation tracks system health — RED metrics (Rate, Errors, Duration) for services and USE metrics (Utilization, Saturation, Errors) for resources.
Frequently Asked Questions
How do I choose between strong consistency and eventual consistency?
Strong consistency (CP) guarantees that all nodes return the same data. Use it for financial transactions, user authentication, and inventory management. Eventual consistency (AP) allows temporary divergence but guarantees convergence. Use it for social feeds, product recommendations, and analytics. Consider read-to-write ratio: if writes dominate, strong consistency is more feasible. If reads dominate, eventual consistency with caching improves performance.
What is the difference between sharding and partitioning?
The terms are often used interchangeably. Sharding specifically refers to distributing data across separate database instances (usually on different servers). Partitioning typically refers to dividing a table within a single database instance. Both split data into smaller pieces; sharding implies horizontal scaling across machines.
When should I use an event-driven architecture over synchronous REST APIs?
Event-driven architecture excels when you need loose coupling, asynchronous processing, or the ability to replay past events. Use it for workflows spanning multiple services, real-time data processing, and audit logging. Synchronous REST APIs are simpler and appropriate for request-response patterns where latency tolerance is low.
How do I prevent cascading failures in microservices?
Combine circuit breakers (stop calling unhealthy services), bulkheads (isolate thread pools), retries with exponential backoff (handle transient failures), and load shedding (reject requests under overload). Monitor with RED metrics and set up alerts for error rate spikes.
What is the simplest consensus algorithm to implement?
Raft is significantly simpler than Paxos to implement correctly. Several university courses use Raft as a programming project. MIT 6.824’s Raft lab provides a structured implementation guide. However, for production use, prefer existing implementations (etcd, Consul) rather than writing your own consensus protocol.