Fault Tolerance and Recovery

"On one machine, failure was an event. On ten thousand machines, failure became the weather, and I had to learn to forecast it."

A Coordinator That Stopped Counting on Luck

In the previous section we partitioned data and parameters across machines so that no single node had to hold everything. That very partitioning is what now exposes us to failure. The moment a computation depends on $K$ machines instead of one, its success depends on all $K$ of them staying up long enough, and the more machines we recruit to go faster, the more ways the whole has to break. This is the central tension of scale-out: the same multiplicity that buys throughput also multiplies the failure surface. Fault tolerance is the discipline of buying back reliability that distribution spent, and it is not optional infrastructure bolted on at the end; it is a property you design for from the first line, because a training run that cannot survive a single preempted worker will not survive a week on a real cluster.

We proceed in three moves. First we name the ways a component can fail, from the merciful (it simply stops) to the malicious (it lies). Then we do the short calculation that turns "failures are rare" into "failures are certain," which is the reason this section exists at all. Finally we walk the recovery mechanisms, each one a different answer to the question "how do we make progress survive a death," and we trace each forward to the chapter where distributed AI puts it to work.

1. The Failure Models Worth Designing Against Beginner

Not all failures are the same, and the cost of tolerating one depends sharply on how badly it can misbehave. Distributed systems theory orders failure models from weakest (easiest to tolerate) to strongest (hardest), and choosing the right model for your setting is the difference between a design that is robust and one that is needlessly expensive. The four models that matter for distributed AI are crash-stop, omission, partition, and Byzantine, summarized in Table 2.4.1.

In the crash-stop (or fail-stop) model, a node works correctly until some instant and then halts permanently, producing no further output. A worker whose machine loses power, whose process is killed by the out-of-memory reaper, or whose spot instance is reclaimed by the cloud provider all behave this way: they were correct, then they were gone. This is the friendliest model, because a survivor can safely assume that anything it did not hear from the dead node simply never happened, and almost all production training systems are designed against exactly this assumption.

The omission model is slightly harder: a node or link drops some messages but not others, so a request or a gradient update can silently vanish without the sender crashing. Closely related, and the dominant headache in practice, is the network partition, in which a working group of nodes is split into subgroups that can each talk internally but cannot reach each other. The cruelty of a partition is that, from inside each subgroup, the nodes on the other side are indistinguishable from crashed: you cannot tell "it died" from "I cannot reach it." This ambiguity is the seed of the consistency dilemmas that Section 2.5 formalizes as the CAP trade-off.

The Byzantine model is the strongest and the rarest: a node may behave arbitrarily, sending different and even maliciously crafted messages to different peers, whether from a bug, corrupted memory, or an actual adversary. Tolerating Byzantine faults is expensive (classical agreement needs more than two-thirds of nodes honest) and most data-center training quite reasonably assumes it away. It returns as a first-class concern only where participants are untrusted, most prominently in federated learning, where a malicious client can poison the global model by reporting a corrupted gradient; the Byzantine-robust aggregation rules that defend against this are developed in Chapter 35.

2. Why Failure Becomes Certain: The Independent-Failure Math Intermediate

The reason fault tolerance is unavoidable at scale is not engineering pessimism; it is arithmetic. Suppose each node, over some window of operation, fails independently with a small probability $p$. A cluster has $K$ such nodes. The probability that a particular node survives the window is $1 - p$, and because the failures are independent, the probability that all $K$ survive is the product of the individual survival probabilities,

The quantity we actually care about, the probability that the job is disrupted by at least one failure, is the complement,

This expression is small for small $K$ but climbs toward $1$ as $K$ grows, and it climbs fast. For tiny $p$ we can read off the rate of climb with the approximation $(1-p)^K \approx e^{-pK}$, so the failure probability is roughly $1 - e^{-pK}$; once the product $pK$ reaches the order of $1$, disruption is more likely than not. The expected number of failed nodes is even simpler, $\mathbb{E}[\text{failures}] = pK$ by linearity of expectation, which makes the threshold concrete: when you have enough nodes that you expect on average one of them to fail in a window, you must plan for failures within that window. The code below tabulates both quantities for a fixed per-node failure probability across realistic cluster sizes.

import math

# Each node fails independently in a time window with probability p.
# With K independent nodes, P(no node fails) = (1-p)^K, so
# P(at least one fails) = 1 - (1-p)^K, and E[failures] = p*K.
p = 0.01  # 1% chance a single node fails during the window
print("Per-node failure probability p =", p)
print(f"{'K':>7} | {'P(>=1 fails)':>14} | {'E[failures]':>12}")
print("-" * 40)
for K in [1, 10, 100, 1000, 10000]:
    p_any  = 1.0 - (1.0 - p) ** K        # complement of all-survive
    e_fail = K * p                        # expected number of dead nodes
    print(f"{K:>7} | {p_any:>14.6f} | {e_fail:>12.2f}")

Per-node failure probability p = 0.01
      K |   P(>=1 fails) |  E[failures]
----------------------------------------
      1 |       0.010000 |         0.01
     10 |       0.095618 |         0.10
    100 |       0.633968 |         1.00
   1000 |       0.999957 |        10.00
  10000 |       1.000000 |       100.00

Read the table from top to bottom and the thesis of this entire section appears as a column. At one node, you can ignore failure and rerun on the rare bad day. At one hundred nodes, a job is disrupted more often than not, and you expect to lose one machine per window on average. At one thousand nodes, disruption is a near certainty on every run, and the question is no longer whether a worker dies during your job but how many and how often. A training run that takes hours on thousands of accelerators will, with overwhelming probability, outlive at least one of its workers, so the recovery machinery of the next section is not insurance against an unlikely event; it is the mechanism by which the job finishes at all.

3. Four Ways to Make Progress Survive a Death Intermediate

Given that failures are certain, recovery is the art of ensuring that a node's death does not erase the work the cluster has already done. Four mechanisms dominate distributed AI, and they are best understood as four different answers to one question: what do we save, and how do we rebuild what was lost? They are checkpoint/restart, deterministic re-execution, lineage-based recovery, and replication. Most real systems combine several of them.

Checkpoint/restart is the most general and the most widely used in deep learning. At intervals the job writes its entire recoverable state (model parameters, optimizer moments, the data-loader position, the random-number-generator seeds) to durable storage. When any worker dies, the job restarts every worker from the most recent checkpoint and replays forward. Nothing about the computation needs to be deterministic or re-derivable; you simply paid, in advance, to be able to rewind. The cost is the wasted work between the last checkpoint and the crash, plus the time spent writing checkpoints that are never used, which sets up the optimization problem we solve in Section 4. This mechanism is the backbone of fault-tolerant training, and the elastic variant that resizes the worker set on the fly without losing the checkpoint is the subject of Chapter 18.

Deterministic re-execution is cheaper when it applies. If a unit of work is a pure function of inputs that still exist somewhere, you do not need to have saved its output; you can simply run it again. This is the recovery model of MapReduce: each map and reduce task is deterministic over its input split, so when a worker dies, the scheduler reassigns its tasks to a survivor and reruns them from the durable input, with no global checkpoint at all. The completed work of other tasks is untouched. We develop this re-execution model and its shuffle in Chapter 6; its key limitation is that it demands determinism and re-readable inputs, which a long stateful training loop does not naturally provide.

Lineage-based recovery generalizes re-execution to a graph of transformations. Instead of remembering the data, the system remembers the recipe: the sequence of deterministic operations that produced each partition from durable source data. When a partition is lost, the system recomputes only that partition by replaying its slice of the recipe, not the whole job. This is exactly how Spark's Resilient Distributed Datasets recover; the lineage graph is the fault-tolerance mechanism, letting Spark avoid replicating intermediate data while still surviving the loss of any partition, as Chapter 7 details. Lineage is elegant precisely when the computation is a deep chain of deterministic transforms; it degrades when the chain is long and recomputation gets expensive, which is why Spark also offers optional checkpointing to truncate lineage.

Replication takes the opposite stance: rather than rebuild lost state, keep more than one copy of it so that a failure of one copy is invisible. A parameter shard held on three nodes survives the loss of two; a storage block written to three disks survives two disk failures. Replication trades storage and write bandwidth for instant recovery and continued availability during a failure, which is why it dominates the storage and serving tiers (the replicated parameter servers of Chapter 11 are a clean example) even as checkpointing and lineage dominate the compute tier. Table 2.4.2 places the four mechanisms side by side.

4. Tuning the Checkpoint Interval Advanced

Checkpointing has a sweet spot. Checkpoint too often and you spend the run writing snapshots you will never read; checkpoint too rarely and each crash discards a large block of work. We can find the optimum with a short model, due in its first-order form to Young and refined by Daly. Let $C$ be the time to write one checkpoint, let the job's mean time to failure be $M$, and suppose we checkpoint every $T$ seconds. The overhead has two parts: a checkpoint tax of $C/T$ (the fraction of time spent writing) and an expected re-do cost of $T/(2M)$ (a failure lands on average halfway through an interval, so we lose about $T/2$ of work, scaled by the failure rate $1/M$). The total wasted-time fraction is

Minimizing over $T$ by setting $f'(T) = -C/T^2 + 1/(2M) = 0$ gives the classic result that the optimal interval is the geometric balance of the two costs,

The optimum is intuitive: checkpoint more often when writes are cheap or failures are frequent (small $M$), and less often when writes are expensive or the system is reliable. The code below evaluates this for a job that writes a checkpoint in 30 seconds and has a 6-hour mean time to failure.

# Optimal checkpoint interval (Young/Daly first-order model).
# Overhead f(T) = C/T + T/(2M): checkpoint tax + expected re-done work.
# Minimized at T* = sqrt(2*C*M), giving min overhead sqrt(2C/M).
C = 30.0          # seconds to write one checkpoint
M = 6 * 3600.0    # job mean time to failure = 6 hours, in seconds
T_opt = math.sqrt(2.0 * C * M)
print(f"Checkpoint write cost C = {C:.0f} s, job MTTF M = {M/3600:.0f} h")
print(f"Optimal interval T* = sqrt(2*C*M) = {T_opt:.0f} s (~{T_opt/60:.1f} min)")

def overhead(T):
    return C / T + T / (2.0 * M)

print(f"\n{'T (s)':>8} | {'overhead %':>10}")
print("-" * 24)
for T in [120, 300, T_opt, 1200, 3600]:
    print(f"{T:>8.0f} | {overhead(T)*100:>10.3f}")
print(f"\nMinimum overhead at T* = {overhead(T_opt)*100:.3f} %")

Checkpoint write cost C = 30 s, job MTTF M = 6 h
Optimal interval T* = sqrt(2*C*M) = 1138 s (~19.0 min)

   T (s) | overhead %
------------------------
     120 |     25.278
     300 |     10.694
    1138 |      5.270
    1200 |      5.278
    3600 |      9.167

Minimum overhead at T* = 5.270 %

Two lessons sit inside Output 2.4.2. First, the cost of over-checkpointing is real and large: dropping from the optimal 19-minute interval to a nervous 2-minute interval quintuples the overhead, because the write tax $C/T$ explodes as $T$ shrinks. Second, the optimum is broad and forgiving; intervals of 19 and 20 minutes differ by a hundredth of a percent, so you do not need $M$ precisely, only its order of magnitude. As clusters grow, $M$ falls (more nodes, more failures, exactly the math of Section 2), which pulls $T^{\star}$ down and pushes the floor overhead $\sqrt{2C/M}$ up. That rising floor is precisely the pressure behind the research on cheaper checkpoints below.

import torch.distributed.checkpoint as dcp

# Save: each rank writes only its own shards in parallel to durable storage.
dcp.save(state_dict, checkpoint_id="ckpt/step_5000")

# Restart (possibly at a DIFFERENT world size): load reshards automatically.
dcp.load(state_dict, checkpoint_id="ckpt/step_5000")

We now have the full picture: the failure models that describe how a component can break, the independent-failure math that proves at least one will, and the four recovery mechanisms (with checkpoint/restart tuned by $T^{\star} = \sqrt{2CM}$) that turn a certain death into a survivable one. What we have quietly assumed throughout is that a survivor can always tell a dead node from a slow one and always agree with its peers on what state to recover to. A network partition breaks both assumptions, forcing a choice between staying consistent and staying available. That choice, and the staleness it produces in distributed training, is the subject of the next section.

The next section turns the partition ambiguity we kept setting aside, the inability to tell a dead node from an unreachable one, into a precise statement about what a distributed system can and cannot guarantee, and connects it to the staleness that distributed training tolerates on purpose. Continue to Section 2.5: Consistency Models: From Parameter Staleness to the CAP Trade-off.