Concept: At-Least-Once, Heartbeats, and Idempotency

Kind: Concept (read, do not code). This is the chapter that arms the payoff — the last idea before the capstone.

You have built almost the whole cluster. The framed codec turns a byte stream into whole messages; the connection actor multiplexes one socket, matching Result frames back to the jobs that are waiting on them; the RemoteWorker hides all of that behind the same WorkerHandle the scheduler has dispatched through since Part VI. On a good day, a job goes out as an Assign frame and its answer comes back as a Result frame, and nobody upstream can tell the worker was on another machine.

This chapter is about the bad day — the one the entire course was designed around. A worker dies mid-job. Hold the sentence from the introduction in view: under worker failure, no eval is lost and none is double-counted. Those two properties pull against each other, and this chapter is where the tension finally resolves. To lose nothing, the coordinator must be willing to run a job again somewhere else. But running it again means it can genuinely run twice. The escape is that the store was built — back in Part IV — so that recording the same job's outcome twice is indistinguishable from recording it once. This chapter earns that claim.

We proceed in the order the ideas depend on each other: first how you even notice a worker died (it is harder than it sounds), then what you do about it (redeliver), then why that is safe (idempotency).

Why a clean close is not the only failure

Start with the failure you already handle. When a worker process exits — a panic, a kill, a clean shutdown — its TCP socket closes. On the coordinator side, conn.recv() returns Ok(None): end of stream. The connection actor sees that, breaks its loop, and fails every job still in flight with a retryable error. You built this in the previous chapter, and its test (worker_death_makes_dispatch_retryable) is green. A clean close is the easy death, because the operating system tells you about it.

Now the hard death. A worker's process freezes — it deadlocks, its event loop wedges, the box swaps itself to a standstill, or the network path silently drops every packet without ever delivering a FIN. The socket is still open from the coordinator's point of view. Nothing closes. conn.recv() does not return Ok(None); it does not return anything at all. It just blocks, forever, waiting for a frame that will never come. This is a half-open connection: one side considers the link alive and is waiting to read, while the other side is gone or catatonic and will never write.

If the connection actor only ever broke on Ok(None), a frozen worker would pin its in-flight jobs forever. Those jobs would never fail, so the scheduler would never redeliver them, so the run would never finish. "No eval is lost" would be violated in the most insidious way possible: nothing errors, the system just quietly stops making progress. A silent hang is worse than a loud crash.

The fix: define silence, then time it out

Since the transport won't tell you the worker is dead, you decide what "dead" means, and the honest definition is behavioural: a worker that has said nothing for too long is presumed dead. Not "disconnected" — silent. You pick a window — call it the heartbeat timeout — and if a whole window passes with no frame of any kind arriving on that connection, you reap it: break the loop, fail the in-flight jobs retryably, exactly as if it had closed cleanly. The half-open hang becomes an ordinary retryable failure, and the redelivery machinery you already have takes over.

In the connection actor this is one more branch in the select!. Alongside "a new job to dispatch" and "a frame arrived from the worker," you add "the inactivity timer fired." The subtle, load-bearing detail is that the timer is created fresh at the top of every loop iteration, so any activity — a job going out, a result coming back, a heartbeat — resets the clock by starting the next iteration with a brand-new timer. Only genuine silence for the entire window lets the timer win the select!. That is the whole reaper, and you will meet its toy below.

Heartbeats: filling the silence on purpose

There is an obvious problem with "silent for a window means dead": a healthy worker can be silent. If it is sitting idle with no jobs assigned, it has nothing to say, and it would be absurd to reap a perfectly good worker just because the coordinator handed it no work. Worse, even a busy worker on a long job might not produce a Result frame for a while.

The fix is that the worker takes responsibility for its own liveness. When it has nothing else to send, it periodically sends a heartbeat — a tiny frame whose only purpose is to prove "still here." The coordinator does not need to do anything with a heartbeat's contents; the mere fact that a frame arrived resets the inactivity timer, which is the entire point. A heartbeat is silence-insurance.

This is a rate relationship, and getting it right is the whole game. The worker's heartbeat interval must be comfortably shorter than the coordinator's inactivity timeout — several beats should fit inside one window. If a worker beats every 10 seconds against a 30-second timeout, then losing one or two beats to a slow network is harmless; it takes a sustained, three-window silence to trip the reaper. Make the interval too close to the timeout and you get false reaps — you kill live workers over ordinary jitter. Make the timeout enormous and you are slow to notice real deaths. The gap between "how often I promise to speak" and "how long you'll wait before giving up on me" is your tolerance for jitter.

At-least-once: what reaping does

Reaping is only worth anything because of what happens to the reaped worker's jobs. When the connection actor breaks — whether from a clean close, a malformed frame, or the inactivity timeout — it does one crucial thing on the way out: it drains its pending map and sends every waiting job a retryable ControlError::Worker. Each RemoteWorker::dispatch that was blocked awaiting a reply now wakes up with that error.

And a retryable error is precisely the signal the scheduler was built to act on, back in Part VI. The scheduler dispatched the job; the dispatch returned Err(e) with e.is_retryable() == true; so the scheduler puts the job back and hands it to another worker from the pool. The job that the dying worker never finished gets redelivered to a survivor. That is at-least-once delivery: the coordinator guarantees each job runs at least once by being willing to re-run any job whose worker did not confirm completion.

Notice the honesty in "at least." The coordinator cannot know why a dispatch failed. Maybe the worker died before touching the job — then redelivery is the only way the job ever runs. But maybe the worker actually finished the job and died (or was reaped) before its Result frame made it back across the wire. Or maybe it was merely slow, got reaped as "silent," and is still grinding away — and now a survivor is running the same job in parallel. From the coordinator's side these are indistinguishable: all it saw was a dispatch that did not return a confirmed outcome. So it must assume the job might not have run — and redeliver. The unavoidable cost of never losing a job is that a job can genuinely execute twice.

Idempotency: what makes the duplicate harmless

So a job can arrive at the store twice. The property that makes this a non-event is idempotency: an operation is idempotent when applying it twice has the same effect as applying it once. Recording a job outcome must be idempotent — the second recording of the same job must not advance the run's progress a second time.

You already built this, in Part IV, and it is worth revisiting why it was shaped the way it was — because this chapter is the reason. The store's record_job_outcome does not blindly bump the run's done-count. It first tries to move this specific job from not-done to done, and it advances the run only when that transition actually happened for the first time. A redelivered outcome finds the job already marked done, the transition is a no-op, and the done-count is left untouched. The same records land, but the run's progress moves exactly once. (Revisit the guard and its test at the idempotent-recording build step.)

This is the sentence the whole course has been walking toward, and now every clause is load-bearing: at-least-once delivery, made safe by idempotent recording. At-least-once (redeliver on any unconfirmed failure) guarantees no eval is lost. Idempotent recording (advance only on a job's first landing) guarantees none is double-counted. Neither property alone is enough; the pair is the payoff. Heartbeats and the reaper are what turn a silent half-open death into the "unconfirmed failure" that triggers redelivery in the first place — without them, the whole chain never starts.

Toy (a): a select! inactivity-timeout loop that reaps a silent peer

Here is the reaper, distilled to channels and timers — no sockets, so it runs anywhere. A watch task loops on a select! between "a beat arrived" and "the inactivity timer fired." The timer is rebuilt at the top of every iteration, so each beat resets the clock; only a full window of silence lets the timer win. This is the connection actor's heartbeat branch with the networking stripped away.

use std::time::Duration;
use tokio::sync::mpsc;

async fn watch(mut beats: mpsc::Receiver<&'static str>, timeout: Duration) -> &'static str {
    loop {
        // A fresh timer each iteration: any frame resets the clock, so only
        // genuine silence for the whole window trips it.
        let idle = tokio::time::sleep(timeout);
        tokio::select! {
            frame = beats.recv() => {
                match frame {
                    Some(b) => println!("saw {b} — clock reset"),
                    None => { println!("channel closed"); return "closed"; }
                }
            }
            _ = idle => {
                println!("silent past {timeout:?} — reaped");
                return "reaped";
            }
        }
    }
}

#[tokio::main]
async fn main() {
    // Peer A beats twice inside the window, then goes silent → reaped.
    let (tx, rx) = mpsc::channel(4);
    tokio::spawn(async move {
        for b in ["beat-1", "beat-2"] {
            tokio::time::sleep(Duration::from_millis(30)).await;
            let _ = tx.send(b).await;
        }
        // then freeze: hold tx, never send again
        tokio::time::sleep(Duration::from_secs(5)).await;
        drop(tx);
    });
    let verdict = watch(rx, Duration::from_millis(80)).await;
    println!("verdict A = {verdict}");

    // Peer B beats every 30ms, comfortably inside 80ms → stays alive.
    let (tx, rx) = mpsc::channel(4);
    let h = tokio::spawn(async move {
        for _ in 0..20 {
            tokio::time::sleep(Duration::from_millis(30)).await;
            if tx.send("beat").await.is_err() {
                return;
            }
        }
    });
    // Watch only long enough to prove it survives well past one window.
    let verdict =
        tokio::time::timeout(Duration::from_millis(200), watch(rx, Duration::from_millis(80)))
            .await
            .unwrap_or("still-alive");
    h.abort();
    println!("verdict B = {verdict}");
}

Predict the two verdicts before you read on. Peer A sends two beats 30ms apart — each well inside the 80ms window, each resetting the clock — then freezes; 80ms of silence later it is reaped. Peer B keeps beating every 30ms, so the timer never wins and watch never returns; the outer timeout gives up first and we call it still-alive. Real output:

saw beat-1 — clock reset
saw beat-2 — clock reset
silent past 80ms — reaped
verdict A = reaped
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
saw beat — clock reset
verdict B = still-alive

The two branches of the select! are exactly the coordinator's two questions about a worker: did you send me something? (reset the clock) and have you been silent too long? (reap). Heartbeats exist only to keep the answer to the first question "yes."

Toy (b): an idempotent counter that advances only on a key's first arrival

Now the other half — the thing that makes redelivery safe. This counter records outcomes keyed by job id and advances its progress only the first time a given id lands. A HashSet::insert returns true only when the key was new, which is the in-memory twin of the store's UPDATE ... WHERE status != 'done' affecting exactly one row.

use std::collections::HashSet;

struct Ledger {
    seen: HashSet<u64>, // which job ids have already landed
    done_count: u64,    // the run's progress
}

impl Ledger {
    fn new() -> Self {
        Ledger { seen: HashSet::new(), done_count: 0 }
    }

    /// Record an outcome. Advance the count ONLY the first time this id lands;
    /// a redelivered duplicate is recorded but does not double-count.
    fn record(&mut self, job_id: u64) {
        // `insert` returns true only if the id was NOT already present — the
        // in-memory twin of `UPDATE ... WHERE status != 'done'` affecting 1 row.
        if self.seen.insert(job_id) {
            self.done_count += 1;
            println!("job {job_id}: first landing — done_count = {}", self.done_count);
        } else {
            println!("job {job_id}: duplicate — ignored, done_count = {}", self.done_count);
        }
    }
}

fn main() {
    let mut ledger = Ledger::new();
    // Jobs 1 and 2 land once each; job 1 is REDELIVERED twice (a slow worker was
    // not dead, or a survivor re-ran it after a reap). At-least-once, made safe.
    ledger.record(1);
    ledger.record(2);
    ledger.record(1); // redelivery
    ledger.record(1); // another redelivery
    assert_eq!(ledger.done_count, 2, "two distinct jobs, counted once each");
    println!("final done_count = {} (not 4)", ledger.done_count);
}

Predict the final count: four record calls, but only two distinct ids, so the run advances exactly twice. Real output:

job 1: first landing — done_count = 1
job 2: first landing — done_count = 2
job 1: duplicate — ignored, done_count = 2
job 1: duplicate — ignored, done_count = 2
final done_count = 2 (not 4)

Four recordings, done_count == 2. That is idempotency in one method: the operation is safe to apply as many times as redelivery demands, because every application after the first is a no-op on the thing that matters. The real store enforces this transactionally in SQLite rather than with a HashSet, but the shape — decide firstness, act only on firstness — is identical.

One-for-one: the toy ↔ the real thing

• The watch loop's fresh sleep(timeout) per iteration ↔ connection_actor's let idle = tokio::time::sleep(heartbeat_timeout) at the top of the select! loop — a new timer each pass, so any frame resets the clock.
• The idle branch returning "reaped" ↔ the actor's _ = idle => break arm — genuine silence for the whole window ends the connection; the reap.
• The channel closing → "closed" ↔ conn.recv() returning Ok(None) → break — the clean death, distinct from the silent one but funnelled to the same exit.
• Peer B's beat loop ↔ the worker's heartbeat interval in run_session — periodic tiny frames whose only job is to reset the coordinator's clock.
• watch returning a verdict its caller observes ↔ the actor draining pending and sending each waiter a retryable ControlError::Worker — reaping produces the retryable error that triggers redelivery.
• Ledger::record advancing only when seen.insert is true ↔ record_job_outcome advancing done_count only when the WHERE status != 'done' update affected one row — advance on first landing only.
• Four record(1) calls leaving done_count == 2 ↔ recording_the_same_job_twice_counts_once — redelivery cannot double-count.

Hold the whole mapping against the payoff sentence. The reaper (toy a) converts a half-open freeze into a retryable failure; the scheduler turns that failure into a redelivery; the idempotent ledger (toy b) makes the redelivery harmless. At-least-once, made safe by idempotent recording — and next chapter you build both halves for real and prove them together.

Questions to lock

Genuinely stop on each. The capstone assumes you can answer all four cold.

1. Why isn't conn.recv() returning Ok(None) enough to detect every dead worker? Describe the failure it misses and what you add to catch it.
2. Why must the inactivity timer be recreated at the top of each loop iteration rather than once before the loop? What breaks if you build it once?
3. What is the required relationship between the worker's heartbeat interval and the coordinator's inactivity timeout, and what goes wrong at each extreme (interval too close to timeout; timeout enormous)?
4. State the payoff sentence and defend both halves: which mechanism guarantees "no eval is lost," which guarantees "none is double-counted," and why is neither sufficient alone?

Next: the capstone build — the heartbeat timeout in connection_actor, the worker's own heartbeat, and the integration test that kills a worker mid-run and proves no eval is lost and none double-counted.