Part III - Cluster and platform layer

Debugging, Observability, and Failure Semantics

Every lesson so far quietly assumed that when a task or actor dies, something sane happens. Lesson 12 promised the autoscaler would replace a lost node; lesson 10 promised training would resume from a checkpoint; lesson 13 promised a Checkpoint could flow from train to serve. This lesson cashes those promises. It builds the failure model the rest of the course leaned on — what Ray retries for free, what it reconstructs from lineage, and the precise places where both of those quietly stop working and leave you holding a corrupted result. Once you can name what is rebuildable and what is not, observability stops being dashboards-as-decoration and becomes the instrument that tells you which boundary just failed.

Book source

Cross-cutting material from Chapters 2, 9, and 10 - runtime components, clusters, AIR failure model, memory management, and autoscaling behavior. Calibrated to current Ray: @ray.remote(max_retries=, retry_exceptions=) for tasks, max_restarts/max_task_retries for actors, the Ray dashboard, the timeline export, and ray memory for object-store introspection.

Linear position

Prerequisite: Lesson 04 (runtime, scheduling, object store) — you know the object store holds immutable values addressed by ObjectRef, that the owner of a ref is the worker that created it, and that the store spills to disk under pressure. Lesson 03 (actors) — actor state lives in one process and dies with it.
New capability: a precise model of what Ray recovers automatically versus what you must make recoverable yourself, so you can decide where idempotency and checkpoints are mandatory rather than optional, and read the dashboard to localize a slowdown to a specific boundary.

The plan

Five moves. (1) Establish that Ray's recovery story is two mechanisms — task retry and lineage-based reconstruction — and define each. (2) Find the three cliffs where reconstruction silently fails: actor state, side effects, and a dead owner. (3) Derive the consequence: Ray gives you at-least-once execution, so your tasks must be idempotent, with a real side-effect bug and its fix. (4) Make the retry-vs-checkpoint trade-off quantitative with an expected-redo-work formula and a draggable knob. (5) Walk three concrete diagnoses on the dashboard — spilling, a straggler, a leaked ref — then the failure checklist.

1 · Two recovery mechanisms, defined

When a task fails — its worker crashed, its node was preempted, it raised — Ray has exactly two tools, and they are not the same tool.

Retry is the obvious one. A task marked @ray.remote(max_retries=3) that dies because its worker process crashed is simply resubmitted, up to three times, to (possibly) a different worker. By default Ray retries on system failures (worker/node death) but not on application exceptions your function raised; retry_exceptions=True opts into retrying those too. Retry is a coarse "run it again" — it assumes running it again will produce the same answer.

Reconstruction is the subtle one, and it is the payoff of the ownership/lineage model from lesson 04. Lineage is the record of how every object was produced: which task, with which input refs. When an object in the store is lost — evicted under memory pressure, or living on a node that died — and some still-running task now needs it, Ray does not error. It walks the lineage backward and re-executes the task that produced the object, recursively reconstructing any of that task's inputs that are also gone. The result is recomputed rather than retrieved. This is why a long chain of transformations can survive a node loss in the middle: Ray rebuilds the missing intermediate from its recipe.

retryA failed task is resubmitted (up to max_retries). Used when the task itself died. Assumes re-running yields the same effect.

reconstructionA still-needed object was lost. Ray re-executes the producing task using lineage, recursively rebuilding lost inputs. The task did not fail — its output vanished.

lineageThe dependency recipe Ray keeps for each object: producing task + input refs. The thing that makes reconstruction possible — and the thing that is gone when the owner dies.

Both mechanisms share one assumption: re-executing a task is equivalent to executing it the first time. That holds for a pure function of its inputs — def square(x): return x*x recomputes the same value no matter how many times you run it. The entire rest of this lesson is about the cases where that assumption is false, because those are the cases that corrupt your data while every dashboard stays green.

2 · The three cliffs where reconstruction fails

Reconstruction is powerful precisely because it is automatic, which is also why it is dangerous: it will happily re-run work that was not safe to re-run. There are three places the recipe-replay model breaks.

Actor state

An actor's value is not a pure function of its constructor inputs — it is the accumulation of every method call since. There is no lineage that reproduces "the counter is at 4,217" or "the optimizer has seen 12k batches." Restart the actor and you get a fresh object, not the lost state.

Side effects

A task that writes to a database, sends an email, or charges a card changed the outside world. Re-executing it does it again. Lineage tracks the returned object, not the row it inserted — so reconstruction double-writes.

Dead owner

Lineage is held by the owner of the ref (the worker that created it, lesson 04). If the owner dies, the recipe dies with it. Ray cannot reconstruct an object whose lineage is gone — you get an OwnerDiedError, not a silent rebuild.

The first cliff is why lesson 03 insisted actors are a single point of failure: their state is the whole point and it is exactly the thing reconstruction cannot rebuild. The third is the most surprising — it means the safety of ray.put(big_array) on the driver depends on the driver staying alive, and an object pinned far from its owner is a latent failure waiting for that owner's node to be preempted. The second cliff is the one that bites correctness rather than availability, so it gets its own section.

The real contract

Ray's automatic recovery is safe only for pure tasks and for actor restarts where the state is reconstructable from a checkpoint or an external source. For everything else — side-effecting tasks and stateful actors — recovery is something you design, and Ray's automatic retry/reconstruction is then a hazard you must make idempotent rather than a feature you can lean on. The default max_restarts=0 for actors is conservative on purpose: Ray would rather fail loudly than silently hand you a fresh actor where you expected the old state.

3 · At-least-once execution forces idempotency

Put retry and reconstruction together and you get the delivery guarantee Ray actually offers: at-least-once — a task that "succeeds" from your driver's point of view may have run its body more than once (a retry after a partial failure, a reconstruction of a lost output). It will never run zero times if you got a result, but it may run two or three. Contrast at-most-once (runs once or not at all, no duplicates, but you can lose work) and exactly-once (the expensive ideal almost no distributed system gives you for free).

At-least-once is a fine guarantee for pure work and a correctness bomb for side effects. A task is idempotent when running it twice has the same observable effect as running it once. Pure functions are trivially idempotent. Side-effecting tasks are not, unless you engineer them to be. Here is the bug:

# BUG: at-least-once + non-idempotent side effect = double charge.
# A retry or a lineage reconstruction can run this body twice.
@ray.remote(max_retries=3, retry_exceptions=True)
def settle_order(order_id, amount):
    db.insert("payments", {"order": order_id, "amount": amount})  # appends a row
    return order_id
# Run twice -> TWO payment rows for one order. Reconstruction never told you.

The fix is to make the write idempotent — keyed on something stable so a second run is a no-op, not a second effect. Two standard moves: a dedup key (a unique constraint or conditional insert keyed by order_id) or an idempotent upsert (write-if-absent to a deterministic location).

# FIX: keyed so a second execution is a no-op, not a second charge.
@ray.remote(max_retries=3, retry_exceptions=True)
def settle_order(order_id, amount):
    # upsert keyed by order_id: the unique key makes the 2nd write a no-op
    db.upsert("payments", key=order_id, value={"amount": amount})
    return order_id

# Same shape for object/file outputs: write to a deterministic path,
# commit atomically, and treat "already committed" as success.
@ray.remote(max_retries=3)
def write_partition(partition_id, rows):
    final = f"s3://bucket/job_42/part={partition_id}/data.parquet"
    if already_committed(final):       # idempotency check first
        return final
    tmp = final + ".tmp"
    write_parquet(tmp, rows)
    commit_atomically(tmp, final)      # atomic rename = single visible effect
    return final

For actors, the equivalent of idempotency is recoverable state: allow restarts and make the state rebuildable.

# Actor with restarts AND a checkpoint, so restart restores state.
@ray.remote(max_restarts=2, max_task_retries=1)
class Counter:
    def __init__(self, ckpt="s3://bucket/counter.json"):
        self.ckpt = ckpt
        self.n = load_or_zero(ckpt)    # rebuild state on (re)construction
    def incr(self):
        self.n += 1
        if self.n % 1000 == 0:
            save(self.ckpt, self.n)    # periodic checkpoint, not every call
        return self.n

Note the tension max_restarts exposes: a restarted actor reruns __init__, so it recovers to the last checkpoint, not the last call. The 0–999 increments since the last save are gone. How often to save is the next question, and it has a number.

4 · Retry cost vs checkpoint cost

Checkpointing is not free, and not checkpointing is not free either — you are choosing where to pay. The lever is the interval between checkpoints (or, for a lineage chain, how much work sits between persisted boundaries). The expected work you must redo after a failure is:

expected redo work ≈ failure_rate × work_since_last_checkpoint

and the total overhead you are trying to minimize is the redo plus the checkpoint I/O itself:

total ≈ (failures × work_lost_per_failure) + (num_checkpoints × checkpoint_cost)

Worked example — training loop. A training job runs 10 hours, with a per-hour node-failure probability of 2% (p = 0.02). A checkpoint costs 30 s of I/O. If you checkpoint every 1 hour, you write 10 checkpoints (300 s of I/O ≈ 0.083 h) and on a failure you lose, on average, half an interval = 0.5 h of compute. Expected lost compute over the run ≈ 10 × 0.02 × 0.5 h = 0.10 h. Total overhead ≈ 0.083 + 0.10 ≈ 0.18 h. Now checkpoint every 6 minutes (0.1 h): 100 checkpoints = 3000 s ≈ 0.83 h of I/O, but expected loss drops to 10 × 0.02 × 0.05 h = 0.01 h — total ≈ 0.84 h, worse, because the I/O now dominates. The sweet spot is roughly where the two terms balance; the widget lets you find it.

Worked example — reconstruction vs checkpoint. A 12-stage Ray Data pipeline keeps intermediates only in the object store, relying on lineage. A node holding stage-7 output dies. Reconstruction re-runs stages 1–7 for the lost blocks — say 7 × 40 s = 280 s of recompute. If instead you had materialized stage-7 to durable storage at a one-time cost of 50 s, recovery would have been a 5 s reload. So if node loss is likely and the chain is long, a single mid-pipeline checkpoint (50 s) buys back 275 s on the (probable) failure. If node loss is rare, you skip the 50 s and let lineage handle the unlikely case — that is exactly the bet lineage is designed to win.

Checkpoint interval — redo work vs checkpoint I/O

A job of fixed length runs under a per-hour failure rate. Each checkpoint costs I/O time; on a failure you lose, on average, half a checkpoint interval of compute. Drag the interval and watch the two costs trade: long intervals → cheap I/O but big redo on failure; short intervals → tiny redo but I/O dominates. The minimum total is the interval you want.

checkpoint interval (h): 1.00 failure rate (/h): 0.020 checkpoint I/O (s): 30

Checkpoints written

—

Checkpoint I/O cost

—

Expected redo work

—

Total overhead

—

Show the core JS

// fixed-length job, interval I (hours), failure rate p (/h), io = checkpoint cost (s)
const nCkpts   = jobLen / I;                 // how many checkpoints over the run
const ioCost   = nCkpts * (io / 3600);       // total I/O, in hours
const redoWork = jobLen * p * (I / 2);       // expected lost compute (half an interval)
const total    = ioCost + redoWork;          // minimize this over I

The shape is a classic U: total cost falls as you checkpoint more often (less redo), bottoms out, then rises as I/O takes over. The minimum moves right (checkpoint less often) when I/O is expensive, and left (checkpoint more often) when failures are frequent — which is the whole reason a 1000-GPU run checkpoints far more aggressively than a laptop job.

5 · Reading the dashboard: three diagnoses

The failure model tells you what can go wrong; observability tells you what did. Ray's instruments: the dashboard (live actors, tasks, per-node object-store memory, logs), the timeline (a Chrome-trace export you open in chrome://tracing to see every task as a bar — the way you spot stragglers and load skew), ray memory (a snapshot of every live ObjectRef, who owns it, and store occupancy), and metrics (pending tasks, object-store fill %, spill rate). Three concrete slowdowns, and how each shows up.

(a) mystery slowdownA job that ran fine suddenly crawls. On the dashboard the object-store memory is pinned near 100% and the spill rate is non-zero — the store is full and Ray is writing objects to disk (lesson 04). Throughput collapsed because reads now hit disk, not shared memory. Fix: smaller blocks, ray.put fewer/larger objects, or free refs sooner so the store drains.

(b) straggler in a reduceA straggler is one task far slower than its siblings, holding up a barrier. In the timeline view, every map bar finishes at ~2 s except one that runs 40 s while the reduce waits idle. Cause is usually data skew — one partition is 20× larger. Fix: rebalance partitions, or split the hot key.

(c) leaked ObjectRefObject-store memory climbs monotonically across iterations and never drops. ray memory shows a growing list of refs owned by the driver — a Python list is holding every ObjectRef you ever created, so reference counting never frees them. Fix: drop refs you no longer need (del / let them go out of scope) so the store can evict.

Worked number — the spill cliff. Say each node has a 30 GB object store and your pipeline holds 28 GB of live intermediates. Add one more 4 GB block and you are at 32 GB > 30 GB: Ray spills ~2 GB to disk. A shared-memory read of a block is effectively free (zero-copy, ~0 ms); a spilled read at, say, 500 MB/s costs 2 GB / 0.5 GB/s = 4 s per touch. If that block is read by 50 downstream tasks, that is 50 × 4 s = 200 s of pure disk latency that did not exist when everything fit in memory. This is why the slowdown is a cliff, not a slope: nothing degrades until you cross the store size, then a whole class of reads falls off the zero-copy path at once.

Observability has a cost too — every metric scraped and every trace span recorded is overhead, and a noisy dashboard slows incident response as much as no dashboard. Keep a small set of golden signals (object-store fill %, spill rate, pending-task backlog, per-actor restart count, p99 latency for Serve) bright and on the front page; push the rest into drill-downs you open only when a golden signal moves.

6 · Failure modes & checklist

Failure modes

Retries as a cure-all. Bumping max_retries to fix flakiness amplifies side effects (double writes) and cluster load on exactly the runs already in trouble. Fix the idempotency, not the retry count.
Actor restart treated as task retry. Setting max_restarts without checkpointing gives you a fresh actor where you needed the old state — the slowdown vanishes and your counter silently resets.
Dead-owner surprise. ray.put on a short-lived worker (or driver) whose node later dies: the object is unreconstructable because its lineage went with the owner.
Debugging only the driver trace. The driver shows a generic RayTaskError; the real cause (OOM kill, spill, node loss) is in worker logs or autoscaler events on the dashboard.
Leaked refs as a slow OOM. Holding every ObjectRef in a growing list pins the store; the job that worked at small scale OOMs the object store at large scale.

Implementation checklist

Is every remote function labeled pure, idempotent, or side-effecting — and do the side-effecting ones have a dedup key or atomic commit?
For each actor: empty restart, checkpoint-restore, rebuild-from-external, or fail-fast? Is max_restarts set consistently with that choice?
What is the checkpoint interval, and does the redo-vs-I/O math justify it for this failure rate?
Have you actually restored from a checkpoint, not just written one?
Which golden signals are on the dashboard front page: object-store fill %, spill rate, pending tasks, actor restarts, Serve p99?
Could a long lineage chain be cut by one mid-pipeline materialization to bound reconstruction cost?

Checkpoint exercise

Try it

Take one Ray workload from an earlier lesson (the Ray Data pipeline from 09, or the Train loop from 10) and write its failure table with a row per boundary: task failure, actor death, node loss, object spill, dead owner, checkpoint corruption, autoscaler delay. For each, state what is retried, what is reconstructed, and what fails the job. Then flip one assumption — raise the node-failure rate from 2%/h to 20%/h — and check which decisions change. (The checkpoint interval should shrink; a long lineage chain that was fine now needs a mid-pipeline materialization. If nothing in your design changes, you were not actually relying on the failure model.)

Where this points next

You now have the failure model the whole course assumed — and that model is the honest boundary of what Ray is. It gives you at-least-once tasks, lineage reconstruction, and actor restarts, and it explicitly does not give you exactly-once side effects or durable workflow state for free. That boundary is exactly the question lesson 15 takes up: where does Ray stop and another system begin? Knowing that reconstruction is gone when an owner dies, and that side effects need external idempotency, is what lets you reason about whether a workload belongs on Ray at all, or behind Spark, Temporal, or a Kubernetes-native stack.

Takeaway

Ray recovers with two mechanisms: it retries failed tasks and reconstructs lost objects by re-executing their producing task from lineage. Both assume re-execution is harmless, which fails at three cliffs — actor state (no lineage reproduces it), side effects (re-running double-writes), and a dead owner (lineage is gone). The net guarantee is at-least-once, so side-effecting tasks must be made idempotent with a dedup key or atomic commit, and stateful actors need max_restarts plus a checkpoint. How often to checkpoint is a U-shaped trade-off: redo work ≈ failure_rate × work_since_last_checkpoint, balanced against checkpoint I/O. Observability turns this model into action — the dashboard's object-store fill and spill rate catch the spill cliff, the timeline catches stragglers, and ray memory catches leaked refs.

Interview prompts

What are Ray's two recovery mechanisms, and how do they differ? (§1 — retry resubmits a failed task; reconstruction re-executes the producing task from lineage to rebuild a lost output. Retry is for a dead task, reconstruction for a dead object.)
Name the three cases where lineage reconstruction does not save you. (§2 — actor state (not a pure function of inputs), side-effecting tasks (re-run double-writes), and a dead owner (lineage held by the owner is gone → OwnerDiedError).)
What execution guarantee does Ray give, and what does it force on you? (§3 — at-least-once; a body may run more than once via retry or reconstruction, so side-effecting tasks must be idempotent — dedup key or atomic write-if-absent.)
Show a non-idempotent Ray task bug and fix it. (§3 — an append-style DB insert double-writes on retry; fix with an upsert keyed by a stable id, or write-to-tmp + atomic commit + already-committed short-circuit.)
How do you decide how often to checkpoint? (§4 — minimize redo (≈ failure_rate × half-interval × job length) plus checkpoint I/O; the U-shaped total moves left (more often) as failure rate rises and right as I/O cost rises.)
A job that ran fine suddenly crawls — how do you diagnose it? (§5 — check object-store fill % and spill rate on the dashboard; near-100% with non-zero spill means reads fell off the zero-copy path to disk — the spill cliff. Then check the timeline for stragglers and ray memory for leaked refs.)
Why is setting max_restarts on an actor not enough? (§3 — a restarted actor reruns __init__ and loses in-memory state; you must rebuild state from a checkpoint or external source, and you only recover to the last checkpoint, not the last call.)