Build: Redelivery and Heartbeat Reaping — the Capstone

Maps to: Phase 6 (cluster — the capstone). Kind: Build.

This is the climax. Every arc of this course has been laying track toward one sentence — under worker failure, no eval is lost and none is double-counted — and this chapter is where you lay the last rail and run a train over it. You already have a RemoteWorker, a connection actor that multiplexes one socket, and a worker binary that registers and answers Assign frames. What is missing is the machinery that survives a worker freezing rather than closing cleanly, and — the whole point — a test that kills a worker mid-run and proves the two properties hold. You add three things: an inactivity timeout in the connection actor, a heartbeat from the worker, and the capstone integration test. When it goes green, the system does the thing it was built to do.

Objective

Add half-open failure detection to the cluster and prove the payoff end to end. Concretely: give connection_actor a fresh inactivity timer per loop iteration that any frame resets and whose expiry reaps the connection; add DEFAULT_HEARTBEAT_TIMEOUT and the serve_workers_with seam that injects a short timeout for tests; make the worker binary emit a periodic Heartbeat; then write the integration test that dispatches a full run across two networked workers, kills one mid-job, and asserts every job completed exactly once. By the end the workspace is at 38 tests, clippy is clean, and both binaries build.

Scaffold

Modify (no new crates this chapter):

• crates/panoptes-control/src/remote.rs — add the DEFAULT_HEARTBEAT_TIMEOUT constant, thread a heartbeat_timeout: Duration through serve_workers_with → handle_connection → connection_actor, and add the inactivity branch to the actor's select!. serve_workers stays as the public entry point that calls serve_workers_with(listener, pool, DEFAULT_HEARTBEAT_TIMEOUT). Two new unit tests go in the existing #[cfg(test)] mod tests.
• crates/panoptes-worker/src/lib.rs — add a heartbeat: Duration parameter to run_session and a tokio::time::interval branch in its select! that sends a Heartbeat frame each tick. Add a DEFAULT_HEARTBEAT constant (the worker's beat interval — several times shorter than the coordinator's timeout).
• crates/panoptes-control/tests/distributed.rs — new integration test file holding the capstone a_dying_worker_loses_no_evals_and_double_counts_none.

Dependencies this chapter exercises: tokio::time (sleep, interval, timeout) for the timers; tokio::select! for the reaper branch; the Scheduler and its WorkerSource from Part VI; the Store's idempotent record_job_outcome from Part IV. No new crate dependencies.

Expected result:

• cargo test -p panoptes-control remote:: → the two new unit tests pass (a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive) alongside the existing remote tests.
• cargo test -p panoptes-control --test distributed → 1 test passes (a_dying_worker_loses_no_evals_and_double_counts_none).
• cargo test across the workspace → 38 tests pass.

The spec (givens)

The inactivity timeout in connection_actor

The actor already select!s over two branches: pull a new Dispatch while below capacity, and receive a frame from the socket. Add a third branch — the reaper. The exact shape matters:

loop {
    // A fresh timer each iteration: any frame or dispatch resets the clock,
    // so only genuine silence for the whole window trips it.
    let idle = tokio::time::sleep(heartbeat_timeout);
    tokio::select! {
        maybe = jobs.recv(), if pending.len() < capacity => { /* ... existing ... */ }
        frame = conn.recv() => { /* ... existing; a Heartbeat is a no-op ... */ }
        _ = idle => break, // silent past the timeout — presumed dead
    }
}

Three things are load-bearing and easy to get wrong:

1. The timer is created inside the loop, at the top of each iteration. Because select! drops the losing futures, every time a job branch or a frame branch wins, the next iteration builds a brand-new sleep. That is what makes any activity reset the clock. Build the timer once above the loop and it will fire on wall-clock schedule regardless of traffic — reaping busy, healthy workers. (This is exactly toy (a) from the concept chapter.)
2. A Heartbeat frame is handled but does nothing. Its arrival already reset the clock by driving another loop iteration; there is no state to update. The match arm exists only so the frame is consumed rather than treated as a protocol error.
3. On the idle branch you break — you fall out of the loop into the same cleanup the clean-close path uses: drain pending and send each waiter a retryable ControlError::Worker. Reaping and clean close funnel to one exit, so a frozen worker becomes an ordinary redelivery.

→ Answer key

DEFAULT_HEARTBEAT_TIMEOUT and serve_workers_with

/// How long a connection may go silent — no result, no heartbeat — before the
/// coordinator declares the worker dead. Catches a *half-open* connection.
pub const DEFAULT_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(30);

pub async fn serve_workers(listener: TcpListener, pool: SharedPool) {
    serve_workers_with(listener, pool, DEFAULT_HEARTBEAT_TIMEOUT).await
}

pub async fn serve_workers_with(
    listener: TcpListener,
    pool: SharedPool,
    heartbeat_timeout: Duration,
);

serve_workers is what production calls — 30 seconds is generous, tuned so a worker that beats every few seconds tolerates ordinary jitter. serve_workers_with exists for testability: a test cannot wait 30 real seconds to watch a reap, so it passes something like 80ms. This is the same injection trick as the base_url seam in Part III — a production default with a test-time override — applied to time instead of a URL. Thread heartbeat_timeout down through handle_connection into connection_actor unchanged.

→ Answer key

The worker's heartbeat in run_session

Give run_session a heartbeat: Duration and add a beat branch to its select!:

let mut beat = tokio::time::interval(heartbeat);
beat.tick().await; // the first tick is immediate — skip it

loop {
    tokio::select! {
        _ = beat.tick() => {
            conn.send(&Message::Heartbeat { worker_id: worker_id.to_string() }).await?;
        }
        frame = conn.recv() => { /* ... existing Assign handling ... */ }
    }
}

Two givens. First, skip the immediate first tick — tokio::time::interval fires once right away, and you do not want a beat before the session is even doing anything; beat.tick().await once before the loop discards it. Second, DEFAULT_HEARTBEAT (the worker's interval) must be several times shorter than DEFAULT_HEARTBEAT_TIMEOUT (the coordinator's window): 5 seconds against 30 means six beats fit in a window, so losing a few to a slow network never causes a false reap. The rate relationship is the correctness argument from the concept chapter — set it wrong and you either reap live workers or notice deaths slowly.

→ Answer key

Why redelivery is safe — the idempotency you already built

When the reaped worker's in-flight jobs fail retryably, the scheduler redelivers each to a survivor. But a redelivered job can also have been run by the worker that died — a worker that finished, shipped its Result, and was reaped before a later beat, or one that was merely slow. So the same job's outcome can reach the store twice. That is harmless only because record_job_outcome is idempotent: it advances the run's done_count only when the job transitions to done for the first time (UPDATE jobs SET status='done' ... WHERE status != 'done', then bump the run only if that affected one row). The second recording is a no-op on progress. You do not build anything new here — you depend on the guard you built in Part IV. If that test (recording_the_same_job_twice_counts_once) were not green, the capstone below could not pass. Re-read the idempotent-recording step if the guard is fuzzy.

→ Answer key

The capstone test wiring

a_dying_worker_loses_no_evals_and_double_counts_none stands up the whole system and breaks it on purpose. The pieces:

• A coordinator: bind a TcpListener on 127.0.0.1:0, make a SharedPool, and tokio::spawn(serve_workers(listener, pool)).
• A reliable worker helper: connect, Register with capacity 4, then loop forever answering each Assign with one ResponseRecord per vignette.
• A dying worker helper: connect, Register with capacity 1, receive exactly one Assign, then drop the socket without answering — a crash mid-job.
• Spawn one of each, wait_for_workers(&pool, 2), then submit a real run: a manifest of 3 vignettes, models = ["claude", "gpt"], epochs = 2 → 4 jobs × 3 vignettes = 12 records expected. Insert it into an in-memory Store, build a Scheduler::with_source(store, Arc::new(pool), &out), and drive it with scheduler.tick().await.
• Assert: the run ends RunStatus::Done, done_count == 4 ("every job must complete exactly once"), and store.run_results(run.id) returns exactly 12 records.

The dying worker takes a job and vanishes; that job fails retryably (via the socket close here — the reaper covers the frozen variant, proven separately by the unit test); the scheduler redelivers it to the reliable worker. If redelivery double-counted, done_count would exceed 4 or run_results would exceed 12. It does neither. That is the sentence, executed.

→ Answer key

Concepts exercised

• A tokio::select! inactivity-timeout branch with a per-iteration timer (the reaper).
• A production default plus a test-time override for a Duration — the timing analogue of the base_url seam.
• A periodic tokio::time::interval heartbeat, with the immediate first tick skipped.
• The heartbeat-interval-vs-timeout rate relationship as a correctness property.
• At-least-once redelivery driven entirely by the retryable error taxonomy from Part II.
• Idempotent recording (Part IV) as the safety net that makes at-least-once non-destructive.
• A full-stack integration test: real sockets, a real scheduler, a real store, an induced failure.

The build loop (you drive)

Test 1 — a_silent_worker_is_reaped (in remote.rs, #[tokio::test])

1. Write the failing test. Stand up a coordinator with a short timeout via serve_workers_with(listener, pool, Duration::from_millis(80)). Spawn a fake worker that Registers, then sleeps for 5 seconds holding the socket open — silent, never a beat, never a close. wait_for_workers(&pool, 1). Then assert the pool empties: timeout(2s, ...) a loop that waits while !pool.is_empty().
2. Predict: with the reaper branch not yet added, what does the actor's select! do while the worker sits silent — and does the pool ever empty? Why would this test hang rather than fail cleanly?
3. Run — it hangs/times out (no reaper yet).
4. Implement the third select! branch and thread heartbeat_timeout through. How you arrange it is yours; the givens above pin what.
5. Run green, commit.

Test 2 — heartbeats_keep_a_worker_alive (in remote.rs, #[tokio::test])

1. Write the failing test. Coordinator with Duration::from_millis(120). Spawn a worker that Registers and then beats every 30ms (send a Heartbeat frame, sleep(30ms), repeat ~10 times). wait_for_workers(&pool, 1), then sleep(200ms) — well past a single 120ms window — and assert pool.len() == 1.
2. Predict: 30ms beats against a 120ms window — how many beats land per window, and why does no single stretch of silence reach 120ms? What would you change to make this same worker get falsely reaped?
3. Run — confirm it passes once the reaper resets its clock on each frame. If it fails, your timer is not being rebuilt per iteration (given #1).
4. Run green, commit.

Test 3 — a_dying_worker_loses_no_evals_and_double_counts_none (new file tests/distributed.rs, #[tokio::test])

1. Write the failing test. Wire it exactly as the capstone givens describe: coordinator, one reliable worker, one dying worker, a 4-job run over 3 vignettes, drive with scheduler.tick(), assert Done / done_count == 4 / 12 records.
2. Predict: the dying worker takes one job and vanishes. Trace the redelivered job's path: which error does its dispatch return, what does the scheduler do with a retryable error, and which store property stops the redelivery from pushing done_count to 5? Name the test in Part IV that guarantees that property.
3. Run — it fails only if a wire is loose (the worker binary must beat, the scheduler must redeliver, the store must be idempotent). If done_count comes back as 5 or records as more than 12, the idempotency guard is the thing to inspect.
4. Run green, commit. This is the payoff — take a beat.

Done when

cargo test across the workspace shows 38 passing tests, including a_silent_worker_is_reaped, heartbeats_keep_a_worker_alive, and the integration capstone a_dying_worker_loses_no_evals_and_double_counts_none. cargo clippy --workspace --all-targets is clean, and both binaries — the coordinator and the worker — build. When that capstone line goes green, the coordinator does, for real, the thing the introduction promised on page one: it runs an eval run to completion across machines, and when a machine dies mid-job, it loses no eval and double-counts none.