Fault Tolerance and Recovery

"I am a checkpoint on a disk, hoping I am never needed but glad I exist. Every hour I grow a little staler, and every hour I am a little more grateful that nothing has gone wrong yet."

A Checkpoint That Has Not Been Restored From, Yet

In the previous section we measured the failures: across a large fleet, the question is not whether a component dies during a long job but how many die and how often. This section turns to the mechanisms that survive those deaths. They are not specific to AI; checkpointing, replication, and erasure coding are decades-old systems techniques. What is specific to AI is the shape of the workload they protect. A web server is a pool of independent request handlers, and losing one costs one request. A large-scale training run is a single tightly coupled computation whose workers advance in lockstep through a sequence of synchronized steps, so losing one worker corrupts the global state and forces every worker back to the last point where the global state was consistent. That coupling is why a chapter on reliability must treat the cost of recovery, not just its correctness, as the central concern.

1. Checkpoint and Restart: The Workhorse Beginner

The simplest durable recovery mechanism is to periodically write the system's state to stable storage and, after a failure, reload the most recent such snapshot and resume. For training this state is concrete: the model parameters, the optimizer state (momentum buffers, Adam's running moments), the data-loader position, and the step counter and learning-rate schedule. A checkpoint that omits the optimizer state restarts a different optimization trajectory and is a common silent bug. The recovery contract is that a restored job is statistically indistinguishable from one that never failed, which requires the snapshot to be a globally consistent cut: every worker's state taken at the same logical step, with no all-reduce half-finished across the boundary. In synchronous training this consistency is free, because the natural checkpoint point is between steps, when no collective is in flight.

The cost of checkpointing has two parts that pull in opposite directions, and naming them is the whole design problem. Checkpoint too rarely and each failure throws away a large slab of work, the dashed segment in Figure 35.2.1. Checkpoint too often and the time spent writing snapshots, which buys nothing when no failure occurs, dominates. Let $C$ be the wall-clock cost of writing one checkpoint and $\tau$ the interval of useful work between checkpoints. If a failure strikes uniformly at random within an interval, the expected wasted work per failure is half the interval, $\tau/2$, plus the restart cost. The expected fraction of time lost to checkpointing overhead, combining the periodic write cost and the per-failure rollback, is approximately

where $R$ is the restart latency and $\text{MTBF}$ is the mean wall-clock time between job failures. The first term falls as $\tau$ grows; the second rises. Minimizing their sum gives the classic Young/Daly optimal interval $\tau^\star \approx \sqrt{2\,C\cdot\text{MTBF}}$, the same result we derived for spot-instance and preemption economics in Section 33.8. The point worth internalizing is that the optimum scales with the square root of the MTBF: a fleet that fails twice as often should checkpoint only about $1.4$ times as frequently, not twice as frequently.

Several refinements attack the cost $C$ directly. Asynchronous checkpointing snapshots the parameters into a staging buffer (often pinned host memory) and lets training continue while a background thread flushes that buffer to storage, hiding most of $C$ behind useful compute. In-memory and redundant checkpointing keeps a copy of each worker's state in the memory of a peer rather than on slow shared storage, so a single failure recovers from a neighbor at memory speed and only correlated failures fall back to disk. Sharded checkpointing is mandatory for huge models: a trillion-parameter model's state is terabytes, and no single node can write it, so each shard owner writes its own slice in parallel, turning a serial terabyte write into $N$ concurrent gigabyte writes. These are the mechanisms that make $\tau^\star$ achievable in practice; Chapter 18 is the deep dive on engineering them into a real training loop.

import torch.distributed.checkpoint as dcp
from torch.distributed.checkpoint.state_dict import (
    get_state_dict, set_state_dict)

# SAVE: every rank writes only its own shard, in parallel, to shared storage.
model_sd, optim_sd = get_state_dict(model, optimizer)
dcp.save({"model": model_sd, "optim": optim_sd}, checkpoint_id="ckpt/step_42000")

# LOAD: re-shards automatically, even onto a different world size after a restart.
model_sd, optim_sd = get_state_dict(model, optimizer)
dcp.load({"model": model_sd, "optim": optim_sd}, checkpoint_id="ckpt/step_42000")
set_state_dict(model, optimizer, model_state_dict=model_sd, optim_state_dict=optim_sd)

2. Replication and Standby: No Rollback at All Intermediate

Checkpointing accepts a rollback as the price of cheap durability. For components that must not lose any work, the alternative is to run a live redundant copy that can take over with little or no lost state. The general technique is state machine replication: model the component as a deterministic state machine, feed every replica the identical sequence of inputs through a consensus protocol, and each replica independently reaches the identical state, so any survivor can answer for a casualty. The agreement on the input order is exactly the consensus problem of Chapter 2; a replicated controller or scheduler that must never forget a decision is the canonical place it appears in an AI cluster.

Replication is graded by how warm the standby is, which trades recovery time against standing cost. A hot standby processes the same inputs in real time and holds current state, so failover is near-instant but doubles the running cost of that component. A warm standby stays loaded and periodically synchronized but does not do live work, so failover takes seconds to reload the gap. A cold standby is provisioned only after the failure is detected, the cheapest option and the slowest to recover. Table 35.2.1 places these alongside the other primitives so the trade space is visible at once. The right choice depends on what the component is: a training worker is cheap to restart from a checkpoint, so cold or no standby is correct, whereas the cluster's single scheduler or parameter-server coordinator may warrant a hot replicated quorum because its loss stalls everything.

3. Redundancy and Erasure Coding for Stored State Intermediate

Checkpoints and datasets live on storage, and that storage is itself a distributed system that loses disks and nodes. The crude protection is full replication: keep $r$ copies of every byte, survive $r-1$ losses, and pay an $r\times$ storage cost. For petabyte-scale checkpoint and dataset stores that overhead is brutal, so production systems use erasure coding instead. An $(n,k)$ code splits each object into $k$ data shards and computes $n-k$ parity shards, storing all $n$ on distinct nodes; any $k$ of the $n$ suffice to reconstruct the object, so the scheme tolerates up to $n-k$ simultaneous shard losses. The storage overhead is

A common choice such as a $(14,10)$ Reed-Solomon code tolerates four losses at only $1.4\times$ overhead, against the $3\times$ of triple replication for comparable durability. That is the same reason distributed file systems for AI storage, covered in Chapter 8, default to erasure coding for cold data and reserve replication for hot data where read latency matters (reconstructing from parity costs a decode). The demo below makes the recovery concrete on small integers: it encodes $k$ data values into $n$ shards, deletes $n-k$ of them, and rebuilds the originals from the survivors alone.

import numpy as np
from numpy.polynomial import polynomial as P

# A toy (n, k) Reed-Solomon-style code over the reals: treat the k data values as
# the coefficients of a degree-(k-1) polynomial, then store its value at n points.
# ANY k of those n samples determine the polynomial, hence recover the data.
k, n = 4, 7                                  # 4 data shards, 3 parity shards
data = np.array([12.0, -5.0, 8.0, 3.0])     # the k values we must protect
xs = np.arange(1, n + 1, dtype=float)        # n distinct evaluation points
shards = P.polyval(xs, data)                 # n encoded shards (data + parity)

# Lose n - k = 3 shards. Keep an arbitrary surviving subset of size k.
survivors = [0, 2, 5, 6]                      # indices of shards that did NOT die
sx, sy = xs[survivors], shards[survivors]
recovered = np.polynomial.polynomial.polyfit(sx, sy, k - 1)  # solve for coeffs

print("original data :", data)
print("lost shards   :", [i for i in range(n) if i not in survivors])
print("recovered     :", np.round(recovered, 6))
print("max abs error :", f"{np.max(np.abs(recovered - data)):.2e}")
print("overhead n/k  :", f"{n / k:.2f}x", "  tolerates", n - k, "losses")

original data : [12. -5.  8.  3.]
lost shards   : [1, 3, 4]
recovered     : [12. -5.  8.  3.]
max abs error : 5.45e-13
overhead n/k  : 1.75x   tolerates 3 losses

4. Lineage: Recompute Instead of Store Intermediate

There is a fourth primitive that stores almost nothing and pays at recovery time instead. If a piece of state was produced by a deterministic computation from inputs that still exist, you do not need to checkpoint that state at all: record the recipe that produced it, and on loss simply re-run the recipe. This is lineage-based recovery, and its purest expression is the resilient distributed dataset (RDD) of Spark, introduced in Chapter 7 and built on the re-execution model of MapReduce in Chapter 6. An RDD remembers the deterministic transformations (the lineage graph) that created it from durable inputs; when a partition is lost, the framework recomputes exactly that partition by replaying its slice of the graph, with no checkpoint and no replica. The cost is bounded because only the lost partition is recomputed, not the whole dataset.

Lineage is the right tool precisely when the computation is cheap relative to storing its output and is deterministic. A feature-engineering pipeline, a data-cleaning stage, or a deterministic preprocessing step over a fixed corpus fits perfectly. It fits poorly when the computation is stochastic (a training step depends on random sampling and floating-point reduction order), expensive (recomputing fifty thousand gradient steps is absurd), or its inputs no longer exist. That is why training uses checkpoints while the data pipeline that feeds it uses lineage, and a real system layers both: the pipeline of Chapter 8 recovers data partitions by lineage, and the trainer that consumes them recovers by checkpoint. Knowing which half of a system can afford to recompute and which must persist is the practical content of fault-tolerance design.

5. Health Checking, Failover, and Supervised Restart Advanced

Every mechanism above presupposes that a failure was detected and a recovery was triggered, and detection is its own problem because a slow node and a dead node look alike for a while. Production systems detect liveness with health checks: periodic heartbeats or probes, with a timeout past which a component is declared dead. The timeout is a genuine trade-off, the practical face of the failure-detector impossibility from Chapter 2: too short and a momentary stall triggers a needless and expensive failover, too long and the system limps on a dead component. When a death is confirmed, automatic failover promotes a standby or, in training, triggers a coordinated rollback and restart. Deciding the failover correctly when the cluster itself may be partitioned is again a consensus question, which is why robust control planes route failover decisions through the same quorum that orders their state.

For training specifically, the recovery is wrapped in a supervisor that owns the restart policy. PyTorch's elastic launcher, torchrun, is exactly this supervisor: it runs a rendezvous so workers discover each other, monitors the worker processes, and on a failure restarts the whole worker group from the last checkpoint up to a configured retry budget. Crucially it is elastic, meaning it can re-form the group at a different size when a node is permanently lost and a spare is unavailable, which is the bridge from "tolerate failure" to "tolerate failure without idling the rest of the cluster." This elastic supervised restart is the practical payoff of the whole section, and Chapter 18 develops the rendezvous, membership change, and state-resharding it requires in full.

# Detect a dead worker, restart the WHOLE group from the last checkpoint,
# up to 3 times; re-form the group at any size between 8 and 16 nodes.
torchrun \
  --nnodes=8:16 \
  --nproc_per_node=8 \
  --max-restarts=3 \
  --rdzv-backend=c10d \
  --rdzv-endpoint=$HEAD_NODE:29500 \
  train.py --resume-from=latest

The simulation below ties the whole section together quantitatively. It models the 1000-GPU synchronous job whose every failure rolls back all workers, sweeps the checkpoint interval, and measures effective throughput, the fraction of wall-clock that turned into committed work. It then prints the Young/Daly optimum for comparison, so we can check whether the theory predicts the measured sweet spot.

import random

# A 1000-GPU synchronous training job. Any single worker failure forces ALL
# workers to roll back to the last consistent checkpoint, so every failure
# wastes the work done since that checkpoint plus the time to restart.

random.seed(7)
TOTAL_STEPS   = 100_000     # useful steps the job must complete
STEP_SECONDS  = 1.0         # wall-clock per step (all workers, synchronous)
CKPT_SECONDS  = 60.0        # cost to write one consistent checkpoint
RESTART_SECONDS = 120.0     # detect failure + respawn + reload checkpoint
# With 1000 GPUs each failing rarely, the JOB fails often. Per-step job-failure
# probability aggregates every worker's hazard into one number.
P_FAIL_PER_STEP = 1.0 / 8000.0

def simulate(ckpt_interval):
    completed = 0          # useful steps durably committed (survive a rollback)
    since_ckpt = 0         # useful steps done since the last checkpoint
    wall = 0.0             # total wall-clock seconds
    while completed < TOTAL_STEPS:
        wall += STEP_SECONDS                  # advance one step
        since_ckpt += 1
        if random.random() < P_FAIL_PER_STEP:
            wall += RESTART_SECONDS           # rollback: discard work since ckpt
            since_ckpt = 0
            continue
        if since_ckpt >= ckpt_interval:
            wall += CKPT_SECONDS              # pay to persist
            completed += since_ckpt           # now durable
            since_ckpt = 0
    return wall

print(f"{'interval':>9} {'wall (h)':>9} {'efficiency':>11} {'overhead %':>11}")
ideal = TOTAL_STEPS * STEP_SECONDS
for interval in (50, 200, 1000, 5000, 20000):
    wall = simulate(interval)
    eff = ideal / wall
    overhead = 100.0 * (wall - ideal) / ideal
    print(f"{interval:>9} {wall/3600:>9.2f} {eff:>11.3f} {overhead:>11.1f}")

# Young/Daly optimal interval (Ch 33.8): tau* = sqrt(2 * C * MTBF).
mtbf = STEP_SECONDS / P_FAIL_PER_STEP         # mean wall-seconds between failures
tau_star = (2 * CKPT_SECONDS * mtbf) ** 0.5   # optimal seconds between checkpoints
print(f"\nMTBF (s)            : {mtbf:.0f}")
print(f"Young/Daly tau* (s) : {tau_star:.0f}")
print(f"  -> ~{tau_star/STEP_SECONDS:.0f} steps between checkpoints")

 interval  wall (h)  efficiency  overhead %
       50     61.38       0.453       121.0
      200     36.91       0.753        32.9
     1000     31.34       0.886        12.8
     5000     35.16       0.790        26.6
    20000    132.40       0.210       376.6

MTBF (s)            : 8000
Young/Daly tau* (s) : 980
  -> ~980 steps between checkpoints

The agreement between the simulated peak near 1000 steps and the analytic $\tau^\star \approx 980$ is the section's payoff: the checkpoint interval is not a knob to be tuned blindly but a quantity computed from the cost of a checkpoint and the failure rate of the fleet. As the fleet grows and the MTBF shrinks, $\tau^\star$ shrinks with the square root of the MTBF, and the only way to keep overhead low is to drive down $C$ with the sharded asynchronous and in-memory techniques of Section 1.