The Midnight Disaster That Didn't Happen: Building Bulletproof Distributed Systems

The Midnight Disaster That Didn't Happen: Building Bulletproof Distributed Systems

At 3:47 AM, a payment processing system briefly runs two active schedulers simultaneously. One scheduler has already dispatched invoice charges to customers when the second scheduler starts its own processing cycle. Without proper coordination, the system double-charges customers, triggers fraud alerts, and creates a reconciliation nightmare that takes days to untangle.

This scenario plays out in production systems when distributed coordination fails. The solution lies in coordination services like Apache ZooKeeper and etcd, which provide the primitives necessary to prevent these failures: distributed locks, leader election, and configuration management backed by strong consistency guarantees.

This article walks through the system design of these coordination services, examining how they implement consensus protocols, provide distributed locking primitives, prevent split-brain scenarios, and deliver the guarantees that keep distributed systems running correctly.

The Foundation: Why Coordination Services Matter

Before diving into implementation details, let's understand what these services provide and why standard approaches fail in distributed environments.

A coordination service offers three core capabilities:

Distributed Locks and Leader Election: Ensures only one process performs critical operations at a time, even across multiple machines. This prevents the double-processing scenario from our opening example.

Configuration Management: Provides a consistent view of system configuration across all nodes, with atomic updates and change notifications.

Group Membership and Failure Detection: Tracks which nodes are alive and coordinates cluster membership changes without creating inconsistent views.

Traditional approaches using databases or file-based locking break down in distributed systems. Database-based locks can fail during network partitions, creating split-brain scenarios where two nodes both believe they hold the lock. File-based locking doesn't work across machines. Coordination services solve these problems by implementing consensus protocols that provide strong consistency guarantees.

Both ZooKeeper and etcd serve these purposes but take different architectural approaches. ZooKeeper uses the ZooKeeper Atomic Broadcast (ZAB) protocol and presents a hierarchical namespace similar to a filesystem. etcd uses the Raft consensus algorithm and provides a key-value store with hierarchical key organization. Understanding how they work requires examining the consensus layer that makes their guarantees possible.

Consensus: The Raft Protocol Deep Dive

Consensus protocols ensure multiple machines agree on a sequence of operations, even in the presence of failures. etcd implements the Raft consensus algorithm, chosen for its understandability and proven correctness.

Raft's Core Mechanism

Raft organizes time into terms, each beginning with a leader election. A term is a monotonically increasing counter that helps nodes detect stale information. Each term has at most one leader, and nodes reject messages from earlier terms.

A Raft cluster always has one of three states per node:

Leader: Accepts client requests, replicates log entries to followers, and sends periodic heartbeats.

Follower: Responds to requests from the leader and candidates. Followers do not initiate communication except when transitioning to candidate state.

Candidate: A follower becomes a candidate when it hasn't received heartbeats within the election timeout window and attempts to become the new leader.

The election timeout is randomized (typically between 150ms and 300ms based on Raft research) to prevent split votes. When a follower's election timer expires, it increments its term, votes for itself, and requests votes from other nodes.

A candidate wins the election if it receives votes from a majority of the cluster. In a 5-node cluster, this means 3 votes. The winning candidate immediately sends heartbeats to all other nodes, establishing its leadership and preventing new elections.

Log Replication and Commit

Once elected, the leader handles all client requests. Each request becomes an entry in the leader's log. The leader replicates this entry to followers through AppendEntries RPCs.

Here's the critical sequence:

1. Client sends write request to leader
2. Leader appends entry to its local log (uncommitted)
3. Leader sends AppendEntries RPCs to all followers in parallel
4. Followers append the entry to their logs and acknowledge
5. Leader waits for acknowledgments from a majority (including itself)
6. Leader marks the entry as committed and applies it to its state machine
7. Leader returns success to client
8. Leader includes commit index in subsequent AppendEntries RPCs
9. Followers apply committed entries to their state machines

This process ensures that once the leader tells a client a write succeeded, that write is durable. Even if the leader immediately fails, the committed entry exists on a majority of nodes, guaranteeing the next leader will have it.

The Safety Properties

Raft guarantees several critical properties:

Election Safety: At most one leader per term. This prevents split-brain at the consensus layer.

Leader Append-Only: Leaders never overwrite or delete entries in their logs, only append new ones.

Log Matching: If two logs contain an entry with the same index and term, they contain identical entries up to that index.

Leader Completeness: If an entry is committed in a given term, it will be present in the logs of leaders for all higher terms.

State Machine Safety: If a server has applied a log entry at a given index, no other server will apply a different log entry for that index.

These properties combine to create linearizability: the system behaves as if there's a single copy of the data, and operations appear to take effect instantaneously at some point between their invocation and response.

Handling Network Partitions

Network partitions test consensus protocols severely. Consider a 5-node cluster that splits into groups of 3 and 2 nodes.

The partition containing 3 nodes can form a majority quorum. If it contains the leader, operations continue normally. If it doesn't, the nodes elect a new leader after their election timeouts expire (research shows leader election typically completes within 200ms to 2 seconds).

The partition with 2 nodes cannot form a quorum. If it contains the old leader, that leader cannot commit new writes because it cannot replicate to a majority. Clients connected to this partition will experience timeouts. This is the correct behavior: the system sacrifices availability to maintain consistency.

When the partition heals, nodes in the minority partition discover they have a lower term number. They immediately step down, discard any uncommitted log entries, and sync with the current leader. This automatic reconciliation prevents divergent state.

Leader Election: Practical Implementation

Leader election builds on Raft's consensus layer to provide a higher-level primitive: ensuring exactly one node performs a particular role in a distributed system.

The Pattern

Consider a distributed job scheduler that must run exactly one active instance. Multiple scheduler processes run for redundancy, but only one should actively dispatch jobs.

Using etcd, each scheduler instance attempts to create a key with a lease:

Key: /scheduler/leader
Value: scheduler-instance-3.example.com:8080
Lease TTL: 10 seconds

The create operation succeeds for exactly one instance due to etcd's linearizable write semantics. That instance becomes the leader. Other instances watch this key, waiting for it to disappear.

The leader must continuously renew its lease (typically at 1/3 of the TTL interval, so every 3 seconds for a 10-second lease). If the leader crashes or experiences a network partition, it stops renewing the lease. After 10 seconds, etcd automatically deletes the key.

Watching instances immediately observe the key deletion and race to create a new leader key. Again, exactly one succeeds, and the system has a new leader. Research shows this transition typically completes within milliseconds after the lease expires, plus the time for a new election if needed.

Fencing Tokens and Split-Brain Prevention

A critical problem remains: what if the old leader didn't actually crash but was merely slow or partitioned? It might still be processing work, unaware it lost leadership. When it reconnects, you have two nodes both believing they're the leader.

Fencing tokens solve this problem. A fencing token is a monotonically increasing number that accompanies each leadership session. In etcd, the revision number serves this purpose. Every write to etcd increments the global revision counter.

When a scheduler acquires leadership, it records the revision number from its create operation. Let's say scheduler A acquires leadership at revision 1247. It includes this revision number with every job dispatch request.

Later, scheduler A becomes partitioned. Its lease expires, and scheduler B acquires leadership at revision 1253. Scheduler B now includes revision 1253 with its requests.

When scheduler A reconnects, it still believes it's the leader and tries to dispatch a job with revision 1247. The job queue service compares this against the highest revision it has seen (1253) and rejects the request. This prevents scheduler A from performing operations after losing leadership.

The pattern requires cooperation from downstream services: they must track the highest fencing token seen and reject requests with lower tokens. This additional complexity provides strong protection against split-brain scenarios.

Session Management and Failure Detection

etcd implements failure detection through leases (ZooKeeper uses sessions, a similar concept). A lease is a time-bound contract: the client promises to send keep-alive messages, and etcd promises to maintain keys attached to that lease.

Leases solve a fundamental distributed systems problem: distinguishing between a slow node and a dead node. Without leases, a temporary network hiccup could trigger unnecessary failovers. With leases, you explicitly configure the trade-off between false failure detection and recovery time.

A 10-second lease means:

• If a node crashes, leadership transfers within 10 seconds
• If a node experiences a network hiccup lasting less than 10 seconds, no failover occurs
• If a node experiences a 5-second garbage collection pause, no failover occurs (assuming keep-alives sent every 3 seconds)

Shorter leases reduce failover time but increase false positives. Longer leases increase failover time but reduce spurious leader changes. Production systems typically use leases between 5 and 30 seconds based on their specific reliability and latency requirements.

Distributed Locks: Beyond Simple Mutual Exclusion

Distributed locks build on the same primitives as leader election but provide a different semantic: temporary exclusive access to a resource rather than a continuous leadership role.

Lock Acquisition Protocol

A client acquiring a distributed lock in etcd follows this sequence:

1. Create a lease with appropriate TTL (e.g., 30 seconds)
2. Attempt to create a key with the lease attached: PUT /locks/resource-123 value=client-id lease=lease-id
3. If the create succeeds (the key didn't exist), the client holds the lock
4. If the create fails (the key exists), the client watches the key for deletion
5. When the key is deleted, return to step 2

The client must refresh the lease periodically (typically every 10 seconds for a 30-second lease). When the client finishes its work, it explicitly deletes the key and revokes the lease, allowing the next client to acquire the lock immediately.

Handling Lock Holder Failures

If the lock holder crashes or becomes partitioned, it stops refreshing its lease. After the TTL expires, etcd automatically deletes the lock key. Clients watching the key receive a notification within milliseconds, and the next client acquires the lock.

This automatic cleanup prevents indefinite lock holding by failed clients. However, it introduces a critical safety issue: the original lock holder might not actually be dead, just slow or partitioned. It could still be performing operations on the protected resource.

This is where fencing tokens become essential. When a client acquires a lock, it records the revision number from the create operation. Every operation on the protected resource includes this revision number. The resource (a database, file system, or other service) tracks the highest revision number it has seen and rejects operations with lower revisions.

For example, a client acquires a lock at revision 5000 and begins processing a batch job. The client becomes partitioned, its lease expires, and a new client acquires the lock at revision 5005. The new client starts processing the same batch job. When the original client reconnects and tries to write results with revision 5000, the database rejects the write because it has already seen operations from revision 5005.

Lock Queues and Fairness

The basic lock protocol doesn't guarantee fairness. If many clients compete for a lock, a newly arriving client might acquire it before clients that have been waiting longer. For applications requiring fair queuing, coordination services provide sequential keys.

In ZooKeeper, this is explicit: create an ephemeral sequential node under /locks/resource-123/. ZooKeeper automatically appends a monotonically increasing sequence number. Each client checks if its node has the lowest sequence number among children of the lock path. If so, it holds the lock. If not, it watches the node with the next-lower sequence number.

When the lock holder releases the lock (or its session expires), only the next client in line receives a notification. This prevents the thundering herd problem where all waiting clients wake up simultaneously when a lock is released.

etcd achieves similar behavior using transactions and prefix-based watches, though the implementation is more complex than ZooKeeper's built-in sequential nodes.

Lock Granularity and Performance

Lock granularity significantly impacts system performance. A single global lock serializes all operations, limiting throughput to what one node can handle. Fine-grained locks (one per resource) allow parallel processing but increase coordination overhead.

Research shows that coordination services like etcd can sustain approximately 10,000 writes per second per cluster. Each lock acquisition and release requires at least one write (two for explicit release). This means a single etcd cluster can handle roughly 5,000 lock operations per second.

For higher throughput, systems partition locks across multiple etcd clusters or use optimistic concurrency control instead of locks. For example, instead of locking a database row, use compare-and-swap operations that succeed only if the row hasn't changed since reading it.

Watches and Notifications: Reactive Coordination

Watches enable reactive programming patterns in distributed systems. Instead of polling for changes, clients register interest in specific keys or prefixes and receive notifications when changes occur.

Watch Implementation

etcd implements watches using gRPC streaming. A client opens a watch request specifying:

• A key or key prefix to watch
• An optional starting revision (to receive historical events)
• Whether to watch a single key or a prefix range

The etcd server maintains a list of active watches. When a write operation commits, etcd checks which watches match the affected keys and sends notifications to those clients.

The notification includes:

• The event type (PUT or DELETE)
• The key and new value (for PUT events)
• The revision number of the change
• The previous value (if requested)

This design provides several guarantees:

Reliability: Watches deliver all events from a starting revision forward. If a client disconnects and reconnects, it can resume from its last-seen revision without missing events.

Ordering: Events are delivered in revision order, matching the linearizable order of writes.

Atomicity: A transaction affecting multiple keys generates a single notification with all changes at the same revision.

Watch Performance Characteristics

Watches impose minimal overhead when idle. The server maintains a data structure (typically a tree or trie) mapping key prefixes to active watches. When a write commits, the server performs a single lookup to find matching watches.

For active workloads, watch notification latency depends on network round-trip time and server load. In healthy clusters with low latency networks, notifications typically arrive within single-digit milliseconds of the write committing.

However, watches can create thundering herds. If 10,000 clients watch the same key and it changes, the server must send 10,000 notifications simultaneously. This can overwhelm the server's network bandwidth or CPU capacity.

Well-designed systems use prefix watches to reduce the number of active watches and implement client-side batching or rate limiting when processing watch events.

Configuration Management with Watches

Watches enable efficient configuration management. Consider a distributed system with 1,000 service instances that need consistent configuration. The naive approach polls a configuration service periodically, generating significant load even when configuration rarely changes.

With watches, each instance:

1. Reads the current configuration from etcd at startup
2. Establishes a watch on the configuration key
3. Applies the configuration locally
4. Waits for watch notifications

When an operator updates the configuration, etcd sends notifications to all 1,000 instances. Each instance receives the new configuration within milliseconds and applies it. The system achieves consistent configuration across all instances with minimal overhead.

For large-scale deployments, staged rollouts prevent thundering herds. Instead of updating a single configuration key, use a versioned key structure:

/config/v1 → old configuration
/config/v2 → new configuration
/config/active → "v1"

Update the v2 key first (no instances watch it yet). Then gradually update instance configurations to watch v2 instead of v1. Finally, update the active pointer. This allows testing new configuration on a subset of instances before full deployment.

Membership and Split-Brain Prevention

Cluster membership management coordinates which nodes participate in the system and ensures all nodes have a consistent view of membership, even during failures and network partitions.

Dynamic Membership Changes

Early consensus protocols required static membership: the set of nodes participating in consensus was fixed at startup. Changing membership required stopping the entire cluster, updating configuration files, and restarting. This made operational tasks like replacing failed nodes or scaling the cluster extremely disruptive.

Raft implements dynamic membership changes through joint consensus. When adding or removing a node, the cluster transitions through an intermediate state where decisions require majorities from both the old and new configurations.

For example, changing from a 3-node cluster (A, B, C) to a 4-node cluster (A, B, C, D):

1. Leader proposes a configuration change entry containing both old (A, B, C) and new (A, B, C, D) configurations
2. During joint consensus, writes must be replicated to majorities of both configurations (2 of A,B,C and 3 of A,B,C,D)
3. Once the joint consensus entry commits, the leader proposes a second entry with only the new configuration
4. After this commits, the cluster operates with the new configuration only

This two-phase approach prevents split-brain scenarios during membership changes. At no point can the old configuration and new configuration independently form quorums that make conflicting decisions.

Learners and Read Scalability

Standard Raft nodes participate in voting and must receive all writes. This limits cluster size: research and production experience show that clusters larger than 7 nodes experience increased latency as the leader must wait for more nodes to acknowledge writes.

etcd 3.4 introduced learners: non-voting members that receive log entries but don't participate in quorum decisions. Learners enable read scaling without impacting write latency. A cluster might have 5 voting members (tolerating 2 failures) and 10 learners (serving read-only traffic).

Learners also simplify adding new nodes. A new node starts as a learner, catches up with the log, and then promotes to a voting member. This prevents the new node (which is far behind) from slowing down the cluster during catch-up.

Split-Brain Prevention at the Coordination Layer

While Raft prevents split-brain at the consensus layer (only one leader per term, requiring majority quorums), applications must prevent split-brain in their own logic. This is where leases and fencing tokens become critical.

Consider a database with primary-replica replication. The primary handles writes, and replicas serve reads. If the primary fails, a replica must be promoted to primary. Without coordination, a network partition might cause two replicas to both believe they should be primary, leading to divergent data.

Using etcd for coordination:

1. The primary holds a lease on /db/primary with its network address
2. Replicas watch this key
3. If the primary's lease expires, replicas race to create a new /db/primary key
4. The winner becomes the new primary
5. The old primary, if it reconnects, sees its lease expired and steps down

The critical safety mechanism: before accepting writes, the primary checks that it still holds the lease. If the lease expired (even if the primary hasn't received notification yet), it refuses writes and steps down. This prevents the old primary from accepting writes after a new primary has been elected.

Fencing tokens provide an additional safety layer. Each primary records the revision number when it acquired leadership. All writes include this revision. Replicas track the highest revision they've seen and reject writes from primaries with lower revisions.

Membership Failure Scenarios

Several failure scenarios test membership management:

Leader Failure: The most common case. Followers detect missing heartbeats within the election timeout window (typically 150-300ms). A new election completes within 200ms to 2 seconds according to research. During this window, the cluster cannot process writes but can serve stale reads.

Follower Failure: Has minimal impact. The leader continues replicating to remaining followers. Writes succeed as long as a majority of nodes (including the leader) are healthy. The failed follower can rejoin later and catch up by replaying log entries.

Network Partition (Majority Side): The partition containing a majority of nodes continues operating normally. If it contains the leader, no election is needed. If not, a new leader is elected within the election timeout window.

Network Partition (Minority Side): Cannot form a quorum. Writes fail. Reads may return stale data (if served locally) or fail (if linearizable reads are required). This is correct behavior: the system prioritizes consistency over availability.

Simultaneous Multi-Node Failure: If failures reduce the cluster below quorum size (e.g., 2 nodes fail in a 3-node cluster), the cluster cannot process writes. It remains unavailable until enough nodes recover to form a quorum. This is why production deployments typically use 5-node clusters (tolerating 2 failures) rather than 3-node clusters (tolerating only 1 failure).

Practical Considerations and Operational Patterns

Cluster Sizing and Topology

The research shows optimal cluster sizes balance fault tolerance against performance:

3 nodes: Tolerates 1 failure. Suitable for development or non-critical services. Write latency is lowest (leader waits for 1 follower acknowledgment).

5 nodes: Tolerates 2 failures. Recommended for production. Balances fault tolerance with performance.

7 nodes: Tolerates 3 failures. Used for critical infrastructure requiring maximum availability. Write latency increases (leader waits for 3 follower acknowledgments).

Clusters larger than 7 nodes are rarely justified. The performance impact of waiting for additional nodes typically outweighs the marginal increase in fault tolerance.

Geographic distribution requires careful consideration. Placing nodes across data centers increases fault tolerance against site failures but increases write latency (the leader must wait for cross-datacenter network round trips). Research recommends keeping all voting members within a single low-latency region (sub-5ms) and using learners in remote regions for read scaling.

Resource Requirements and Performance Tuning

Coordination services have modest resource requirements but are sensitive to storage latency:

Storage: SSDs are strongly recommended. Write-ahead log writes are synchronous (fsync after each write). HDD seek latency (5-10ms) becomes the bottleneck, limiting write throughput to 100-200 operations per second. SSDs reduce this to sub-millisecond latency, enabling the 10,000 writes per second throughput documented in research.

Memory: Typical deployments use 2-8 GB. etcd recommends keeping the database size below 8 GB. The entire keyspace should fit in memory for optimal performance. Larger datasets require more aggressive compaction or architectural changes (sharding coordination across multiple clusters).

CPU: Coordination services are not CPU-intensive under normal operation. 2-4 cores per node suffice for most workloads. CPU usage spikes during leader elections, snapshot operations, and compaction.

Network: Bandwidth requirements are low (typically under 10 Mbps per node). Latency matters more than bandwidth. Inter-node latency above 10ms noticeably impacts write latency. Packet loss or reordering can trigger spurious leader elections.

Monitoring and Observability

Production coordination services require comprehensive monitoring:

Leader Election Frequency: Should be rare. Frequent elections (more than one per hour) indicate instability: network issues, resource exhaustion, or misconfigured timeouts.

Proposal Commit Latency: Monitor both 50th percentile (should be under 10ms in healthy clusters) and 99th percentile (should be under 100ms). Increasing latency indicates disk I/O problems, network degradation, or overload.

Disk Sync Duration: Time spent in fsync calls directly impacts write latency. Values consistently above 10ms indicate storage problems.

Applied vs Committed Index: Followers maintain two indices: committed (replicated to a majority) and applied (executed by the state machine). Large gaps indicate followers falling behind.

Watch Delivery Latency: Time between a write committing and watch notifications being sent. Increasing latency indicates server overload or network congestion.

etcd exposes these metrics via a Prometheus endpoint. ZooKeeper provides JMX metrics and the four-letter word commands (mntr, stat, cons) for monitoring.

Backup and Disaster Recovery

Coordination services store critical system state. Losing this data can render an entire distributed system inoperable. Regular backups are essential:

Snapshot Mechanism: etcd periodically snapshots its state to disk. The snapshot includes the complete keyspace at a specific revision. Snapshots enable fast recovery without replaying the entire log.

Backup Strategy: Take periodic snapshots (hourly or daily) and store them in durable storage (object storage like S3). Retain multiple snapshots to enable point-in-time recovery.

Recovery Procedure:

1. Stop all cluster members
2. Restore the snapshot to all nodes
3. Start the cluster with a new cluster token (prevents old members from rejoining)
4. Verify cluster health before pointing applications at it

For critical systems, maintain a standby cluster in