Recovery and Self-Healing Systems in Distributed Systems

by
Home » Distributed Systems » Fault Tolerance & High Availability » Recovery and Self-Healing Systems in Distributed Systems

Distributed Systems Series — Part 4.5: Fault Tolerance & High Availability

Detection Is Not Recovery

Post 4.4 established how distributed systems detect failures — through heartbeats, timeouts, phi accrual detectors, and gossip protocols. Detection is the prerequisite. But detecting that a node has failed solves nothing by itself. The system must then do something about it — and doing something correctly, automatically, and quickly enough that users do not notice is the recovery problem.

At scale, failures happen faster than humans can respond to them manually. A distributed system running thousands of nodes across multiple regions will experience node failures, disk failures, process crashes, and network partitions continuously — not as rare events but as the steady state. Engineering teams that respond to each failure manually are operating a system that is too fragile for its scale. Self-healing — the ability to detect failures and restore correct operation automatically, without paging an engineer for every individual component failure — is the operational property that makes large-scale distributed systems manageable.

Self-healing does not mean autonomous. It means that the recovery actions the system takes automatically are well-defined, safe, and bounded — and that the boundary between automatic recovery and human intervention is explicit and deliberate. A system that attempts to recover automatically from every failure, including failures it cannot safely recover from without human judgment, will eventually make things worse. The engineering discipline is defining the right boundary: automatic for the common cases, human for the exceptional ones.

This post covers the four stages of recovery — detection, isolation, recovery, restoration — the specific self-healing mechanisms that production systems use at each stage, how write-ahead logging enables crash recovery without data loss, and where the automatic recovery boundary must be drawn.

The Four Stages of Recovery

Recovery is not a single action — it is a pipeline of four stages, each with its own mechanisms and failure modes. The diagram above shows the complete pipeline. Every stage must succeed for full recovery to complete.

Detection is covered in Post 4.4 — heartbeats, phi accrual, gossip, and Kubernetes probes. The output of detection is a suspicion or confirmed failure signal that triggers the next stage. The speed of detection determines the minimum possible RTO — no recovery can begin before detection completes.

Isolation prevents a failed or suspect component from causing further damage while recovery proceeds. A partitioned node that is not isolated may continue accepting writes that diverge from the rest of the cluster. A failed primary that is not fenced may respond to some client requests with stale state while a new primary is being established. Isolation is the safety gate that makes the subsequent recovery stages safe to execute.

Recovery restores the system to operational state — electing a new leader, rescheduling a failed container, promoting a replica to primary. This is the stage that most engineers think of as “recovery” — but it is only the third of four steps, and it cannot happen safely without isolation completing first.

Restoration brings the recovered component back to full health — catching up on missed log entries, re-replicating data to the appropriate replication factor, rejoining the cluster as a fully participating member. Restoration is often the slowest stage and the one most likely to be incomplete when the next failure occurs.

Write-Ahead Logging: The Foundation of Crash Recovery

Before examining self-healing mechanisms, the foundational technique that makes crash recovery possible deserves explicit treatment. Write-ahead logging (WAL) is described in Post 3.2 in the context of replication, but its role in crash recovery is equally important.

WAL operates on a simple principle: before any change is applied to a data structure, a record of the intended change is written to a durable, append-only log. The log is written to stable storage before the change is acknowledged to the caller. If the node crashes at any point during a write operation, the log contains a complete record of what was intended. On restart, the node replays the log from the last known good checkpoint — reapplying all changes that were logged but not yet applied to the primary data structure.

PostgreSQL’s WAL is the canonical production example. Every transaction writes a WAL record before modifying data pages. On crash recovery, PostgreSQL replays WAL records from the last checkpoint, restoring the database to the exact state it was in at the moment of the crash — including partially completed transactions, which are rolled back by replaying their corresponding rollback records. The recovery process is deterministic and correct regardless of where in a transaction the crash occurred.

WAL has two performance implications that affect recovery design. First, WAL writes must be synchronous — the log record must be durably flushed to storage before the operation is acknowledged. Asynchronous WAL writes (a common misconfiguration) sacrifice the crash recovery guarantee for performance. Second, WAL grows continuously and must be periodically checkpointed — the system applies WAL records to the primary data structure and records a checkpoint, after which WAL records before the checkpoint can be safely discarded. Checkpoints create a performance spike (flushing all dirty pages to storage) and must be tuned to balance recovery time against checkpoint overhead.

Raft’s replicated log is WAL at the distributed level. Each log entry represents an operation that has been durably written to a majority of nodes before being applied to the state machine. On leader failure, the new leader’s log contains all committed operations — exactly what WAL provides on a single node. The new leader can reconstruct the correct state by applying its log from the beginning, or more practically, from the most recent snapshot plus subsequent log entries.

Automatic Leader Re-election

Leader-based systems — Raft, ZooKeeper, most production databases — depend on a single leader for write ordering. When the leader fails, the system must elect a new one. The speed and correctness of this election determines the write availability gap that users experience.

As established in Post 3.7, Raft’s election mechanism uses randomised timeouts. When a follower’s election timeout expires without a heartbeat from the leader, it increments its term, transitions to candidate state, and sends RequestVote RPCs to all other nodes. A candidate wins when it receives votes from a majority. Voters grant their vote only if the candidate’s log is at least as up-to-date as their own — ensuring the new leader has all committed entries.

The write availability gap during election is the time between the leader’s last heartbeat and the moment the new leader commits its first write — typically the election timeout plus one network round trip. In a well-tuned single-region cluster with 150ms election timeout and 5ms network round-trip time, this gap is approximately 155-300ms. For a system with 99.99% monthly availability target and a 4.4-minute error budget, one election per week consumes approximately 1% of the budget — acceptable. One election per day would consume the entire budget in a week.

etcd’s leader re-election is the production implementation that Kubernetes depends on for cluster state consistency. Every Kubernetes control plane operation — creating pods, updating deployments, managing secrets — requires a write to etcd. When the etcd leader fails, Kubernetes control plane operations block until a new leader is elected. The Kubernetes production recommendation is a three-node etcd cluster (tolerates one failure) with a five-node cluster for critical production deployments (tolerates two failures). The election timeout should be tuned to the network round-trip time as described in Post 4.4.

ZooKeeper’s leader election uses the ZAB protocol — ZooKeeper Atomic Broadcast — rather than Raft. ZAB elects a new leader through a discovery and synchronisation phase that ensures the new leader has all committed transactions before it begins serving requests. The election process is typically complete within one to two seconds in a stable cluster, making ZooKeeper’s write unavailability window during leader failure slightly longer than a well-tuned Raft cluster but with stronger guarantees about log completeness at the start of the new leader’s term.

Automatic Container and Process Recovery

Container orchestration platforms implement self-healing at the process level — automatically restarting failed containers, rescheduling pods on healthy nodes, and maintaining the desired number of running instances without human intervention.

Kubernetes implements process-level recovery through its reconciliation loop — the core control plane pattern where controllers continuously compare desired state (what is specified in the cluster) with actual state (what is running) and take actions to converge them. When a pod fails, the ReplicaSet controller detects the discrepancy between desired replicas (3) and actual running replicas (2) and creates a new pod to restore the desired count. This reconciliation is continuous — the controller does not wait for an explicit failure notification but polls for state discrepancy on every iteration.

Pod restart policies control how individual container failures are handled before the scheduler reschedules onto a new node. The `Always` restart policy (the default for Deployments) restarts a container on the same node whenever it exits, regardless of exit code. The `OnFailure` policy restarts only on non-zero exit codes. The `Never` policy does not restart — appropriate for batch jobs where a failed run should be reported rather than retried.

Kubernetes exponential backoff prevents restart storms — a container that crashes immediately on start is not restarted immediately each time, which would saturate the node with failed containers. The backoff starts at 10 seconds and doubles on each failure, up to a maximum of 5 minutes. A container in this state is reported as CrashLoopBackOff — one of the most common Kubernetes states that engineers encounter, indicating a container that is failing immediately on startup and being held in backoff by the scheduler.

Node-level failures trigger pod rescheduling rather than pod restart. When the kubelet stops sending heartbeats and the node is marked NotReady (after the 40-second grace period established in Post 4.4), the node controller begins evicting pods from the node after the pod-eviction-timeout (default 5 minutes). Evicted pods are rescheduled onto healthy nodes by the scheduler. Stateless pods reschedule immediately — they carry no state that must be preserved. Stateful pods — those using PersistentVolumeClaims — may require the storage volume to be detached from the failed node and reattached to the new node before the pod can start, which adds minutes to the recovery time for stateful workloads.

Automatic Data Re-replication

When a node that held data replicas fails, the system’s replication factor drops below target — a cluster configured for three replicas now has two for some data range. Data re-replication restores the target replication factor automatically by creating new replicas on surviving nodes.

Cassandra implements data re-replication through its hinted handoff and repair mechanisms. When a node fails, writes that would have gone to that node are temporarily stored as hints on other nodes. When the failed node recovers, hints are replayed to bring it back to current state. If the node does not recover within the hint window (default 3 hours), the data must be fully streamed from surviving replicas during a repair operation. Cassandra’s `nodetool repair` triggers this process explicitly, but Cassandra also runs automatic background repair to prevent replica divergence from accumulating over time.

HDFS (Hadoop Distributed File System) implements re-replication through the NameNode’s block replication monitor. HDFS stores each data block on multiple DataNodes according to the configured replication factor (default 3). When a DataNode fails, the NameNode detects under-replicated blocks and schedules re-replication — copying blocks from surviving DataNodes to other DataNodes to restore the target replication factor. The re-replication rate is throttled to prevent the recovery traffic from overwhelming the cluster’s network bandwidth during an already-stressed recovery period.

CockroachDB’s range replication uses Raft — each data range is a Raft group, and when a Raft group loses a member, the remaining members elect a new leader and the cluster’s replication layer adds a new replica to restore the target count. CockroachDB’s leaseholder mechanism separates the Raft leader role (managing log replication) from the leaseholder role (serving reads) — a new leaseholder can be assigned to a surviving node without a full leader election, allowing reads to resume immediately while the Raft group completes the write recovery process.

Circuit Breaker Recovery: The Half-Open State

Circuit breakers, introduced in Post 2.2, prevent cascading failures by stopping requests to a failing downstream service. Their recovery mechanism — the half-open state — is a specific self-healing pattern that deserves explicit treatment here.

A circuit breaker operates in three states. In the closed state, requests flow normally. When the failure rate exceeds a threshold, the circuit opens — requests are rejected immediately without attempting the downstream call, protecting the downstream service from additional load and preventing the caller from accumulating latency. After a configured timeout, the circuit transitions to half-open — a single probe request is allowed through. If the probe succeeds, the circuit closes and normal traffic resumes. If the probe fails, the circuit returns to open and the timeout resets.

The half-open state is a self-healing mechanism because it automatically detects when a downstream service has recovered without requiring human intervention or explicit notification from the downstream service. The circuit breaker polls recovery through probe requests rather than relying on the downstream service to announce its health. This is important because a service that has just recovered may not yet be registered in the service discovery system or may not have sent a health check that has propagated to all callers.

Resilience4j’s CircuitBreaker implementation in Java and Netflix’s Hystrix (now in maintenance mode) both implement the half-open probe pattern. Istio’s service mesh circuit breaker implements it at the infrastructure level — the sidecar proxy tracks failure rates per upstream endpoint and opens circuits without requiring application code changes. This infrastructure-level implementation means circuit breaker self-healing applies consistently to all services in the mesh, not just those that have explicitly integrated a circuit breaker library.

Snapshot-Based Recovery: Avoiding Full Log Replay

WAL-based recovery requires replaying all log entries from a known checkpoint. For long-running systems with large transaction volumes, replaying the entire WAL from the beginning of time is impractical — it would take longer than the acceptable recovery time. Snapshot-based recovery solves this by periodically capturing a complete point-in-time image of the system state and storing it durably, allowing recovery to start from the most recent snapshot and replay only the WAL entries since that snapshot.

Raft uses snapshots to bound log replay time and log storage size. Each node periodically takes a snapshot of its state machine state — the complete key-value store contents for etcd, or the complete range data for CockroachDB — and records the log index at the time of the snapshot. WAL entries before the snapshot index can be safely discarded. A new node joining the cluster, or a recovered node that has fallen significantly behind, can receive the latest snapshot from the leader and then replay only the subsequent log entries rather than the full history.

etcd snapshots are triggered when the WAL reaches a configured size (default 10,000 entries). The snapshot is written to disk atomically — etcd uses copy-on-write to avoid blocking reads and writes during the snapshot operation. On recovery, etcd loads the most recent snapshot and replays WAL entries from the snapshot point forward. For a cluster with 10,000 entries between snapshots and a write rate of 1,000 entries per second, the maximum WAL replay time is 10 seconds — acceptable for most production recovery requirements.

Where Automatic Recovery Must Stop

Self-healing is powerful but not unlimited. Four failure scenarios require human intervention because automatic recovery would be incorrect, unsafe, or impossible without human judgment.

Correlated failures exceeding quorum. When more nodes fail simultaneously than the system’s quorum threshold — more than half of a Raft cluster, more than N-W nodes in a quorum system — automatic recovery cannot proceed safely. The system does not have enough surviving nodes to establish a safe majority. Attempting automatic recovery in this scenario risks split-brain — two groups of nodes each believing they have authority, producing diverged state. The correct response is to halt and require human intervention to assess which partition has the authoritative state and manually restore quorum. Kubernetes etcd documentation explicitly covers this scenario — recovery from a complete etcd cluster loss requires restoring from a known-good snapshot, which requires human judgment about which snapshot represents the correct state.

Data corruption. A node that has experienced silent data corruption — bit flips in storage, software bugs producing incorrect data, corrupted WAL entries — cannot safely self-heal by re-replicating its corrupted data to other nodes. Automatic re-replication of corrupted data would spread the corruption to previously correct replicas. Detecting data corruption requires checksums on every stored block (which PostgreSQL, HDFS, and CockroachDB implement) and human intervention to restore from a known-good backup when corruption is detected.

Byzantine failures. As established in Post 4.1, Byzantine failures — nodes sending incorrect or malicious messages — cannot be recovered from by standard crash-recovery mechanisms. A Byzantine node that is removed and restarted may continue exhibiting Byzantine behaviour. Recovery from Byzantine failures requires identifying and replacing the compromised component, which requires human investigation.

Capacity exhaustion. When a system has exhausted its available capacity — all nodes at maximum CPU, all disks at maximum utilisation, all available replicas committed — automatic recovery cannot restore service without adding capacity. Adding capacity (new nodes, new disks, expanded quotas) is a human decision that requires business judgment about cost and resource allocation. Automatic scaling can mitigate this for stateless workloads, but stateful workloads with fixed replication factors require deliberate capacity planning decisions.

The design principle from Vitillo’s Understanding Distributed Systems is directly applicable here: self-healing systems should be designed to handle the common cases automatically and to fail loudly and safely for the exceptional cases, presenting human operators with a clear diagnosis of what went wrong and what is needed to recover. A system that attempts to self-heal from every scenario — including scenarios it cannot safely recover from — will eventually make a catastrophic situation worse by taking automatic actions based on incomplete information.

Key Takeaways

  1. Recovery is a four-stage pipeline — detection, isolation, recovery, restoration — and each stage must complete correctly before the next can proceed safely; isolation before recovery is critical to prevent split-brain during the promotion window
  2. Write-ahead logging is the foundational crash recovery mechanism — by writing log records to durable storage before applying changes, WAL ensures deterministic, correct recovery regardless of where in a transaction a crash occurs
  3. Automatic leader re-election in Raft and ZooKeeper restores write availability within the election timeout window — typically 150ms to 2 seconds depending on cluster configuration — and the new leader’s log completeness guarantee ensures no committed writes are lost
  4. Container orchestrators implement process-level self-healing through continuous reconciliation — Kubernetes controllers continuously compare desired and actual state and take actions to converge them, with exponential backoff preventing restart storms
  5. Data re-replication restores the target replication factor after node loss — Cassandra uses hinted handoff and streaming repair, HDFS uses NameNode block replication monitoring, CockroachDB uses Raft group membership management
  6. Circuit breaker half-open state is a self-healing mechanism — probe requests automatically detect downstream service recovery without requiring explicit notification or human intervention
  7. Automatic recovery must stop at correlated failures beyond quorum, data corruption, Byzantine failures, and capacity exhaustion — these scenarios require human intervention, and systems should fail safely and loudly rather than attempting autonomous recovery that may worsen the situation

Frequently Asked Questions (FAQ)

What is self-healing in distributed systems?

Self-healing is the ability of a distributed system to detect failures and restore correct operation automatically, without requiring human intervention for each individual component failure. It encompasses automatic leader re-election when a primary fails, automatic container rescheduling when a process crashes, automatic data re-replication when a node is lost, and circuit breaker recovery when a downstream service restores. Self-healing does not mean fully autonomous — it means the boundary between automatic recovery and human intervention is explicit and deliberate, with automatic handling for common failures and safe failure for scenarios that require human judgment.

What is write-ahead logging and how does it enable crash recovery?

Write-ahead logging (WAL) is a technique where every change to a data structure is first recorded in a durable, append-only log before being applied. If a node crashes during a write, the log contains a complete record of the intended change. On restart, the node replays log records from the last checkpoint, restoring to the exact state at the moment of crash. WAL is used by PostgreSQL for database crash recovery, by Raft for distributed log replication, and by most production storage systems that require durability guarantees. The key requirement is that log writes must be synchronous — the record must be durably flushed to storage before the operation is acknowledged.

What is the write availability gap during Raft leader election?

The write availability gap is the period during which a Raft cluster does not accept writes — from when the old leader’s last heartbeat is missed to when the new leader commits its first write. It is approximately equal to the election timeout plus one network round trip. With a 150ms election timeout and 5ms single-region network round trip, the gap is approximately 155-300ms. With a 1000ms election timeout for a cross-region deployment, the gap is approximately 1 to 2 seconds. During this window, write requests receive errors or timeouts and must be retried by clients after the new leader is established.

What is CrashLoopBackOff in Kubernetes?

CrashLoopBackOff is a Kubernetes pod status indicating that a container is repeatedly crashing immediately on startup, and the scheduler is applying exponential backoff before attempting another restart. The backoff starts at 10 seconds and doubles on each failure up to 5 minutes. CrashLoopBackOff is not a failure of Kubernetes — it is Kubernetes correctly implementing self-healing behaviour for a container that cannot start successfully. The underlying cause is always in the container itself: a missing dependency, an incorrect configuration, an application bug that causes immediate exit, or resource limits that are too low for the container to start.

What is data re-replication and when does it trigger?

Data re-replication is the automatic process of creating new data replicas to restore the target replication factor after a node failure has reduced it. In Cassandra, re-replication is triggered when the hinted handoff window expires for a failed node that has not recovered — surviving nodes stream the missing data to replacement nodes. In HDFS, the NameNode’s block replication monitor continuously tracks block counts and schedules re-replication when under-replicated blocks are detected. Re-replication is throttled to avoid saturating cluster network bandwidth during recovery — a critical design consideration because re-replication traffic competes with normal application traffic on the same network links.

When should automatic recovery stop and human intervention begin?

Four scenarios require human intervention rather than automatic recovery. Correlated failures that exceed the system’s quorum threshold — when more nodes fail simultaneously than the fault tolerance boundary allows, automatic recovery risks split-brain and must halt. Data corruption — automatic re-replication of corrupted data spreads corruption to healthy replicas; detected corruption requires restoration from a known-good backup. Byzantine failures — restarting a node exhibiting Byzantine behaviour may not resolve it; the root cause requires human investigation. Capacity exhaustion — when all available resources are consumed, automatic recovery cannot restore service without adding capacity, which is a human decision involving cost and resource allocation.

Continue the Series

Series home: Distributed Systems — Concepts, Design & Real-World Engineering

Part 4 — Fault Tolerance & High Availability Overview

Previous: ← 4.4 — Failure Detection: Heartbeats, Timeouts and the Phi Accrual Detector

Next: 4.6 — Designing for High Availability: Patterns and Trade-offs →

Related posts from earlier in the series:

Discover more from Rahul Suryawanshi

Subscribe now to keep reading and get access to the full archive.

Continue reading