Worker Leases, Heartbeats, and the Zombie Task Problem

A scheduler cannot observe a worker directly. It sees messages arriving and messages stopping, and the absence of a message is not evidence of anything: the worker may have crashed, may be alive but slow, may be executing correctly while its network path to the scheduler is broken. A lease is the mechanism that converts this ambiguity into a decision, by attaching a deadline to the scheduler’s permission for the task to run and letting the worker’s continued reporting extend it.

The lease as a bounded permission

A lease is a grant with an expiry: the worker holds the right to execute a specific task attempt until time T, and after T the grant is void regardless of what the worker believes. The worker renews by sending a heartbeat before T, and the scheduler extends T in response. If no renewal arrives, the lease expires and the scheduler may grant the attempt to another worker.

Two properties make this workable. The lease is held by the scheduler, not the worker: the scheduler’s record of T is authoritative, and the worker’s belief about whether it still holds the lease is advisory. And the lease is bounded: it expires on its own without any action, so a scheduler that crashes and restarts can find expired leases and act on them without needing to have observed the worker’s failure.

The choice of lease duration is a trade between detection latency and heartbeat volume. A lease of thirty seconds with a heartbeat every ten seconds detects a dead worker within thirty seconds and generates six heartbeats per minute per running task. A lease of five minutes with a heartbeat every minute detects within five minutes and generates one heartbeat per minute. For a thousand concurrent tasks the first scheme is six thousand heartbeat messages per minute, which is a real load on the scheduler’s message path; the second is a thousand. The shorter lease detects failure faster and costs more; the longer one is cheaper and leaves failed tasks occupying their slots for longer.

The heartbeat interval must be well under the lease duration, because the worker’s clock and the scheduler’s clock are not the same clock. A worker that renews at exactly T minus one second, with a one-second network delay, renews after the lease expired. The usual ratio is a third: renew every ten seconds on a thirty-second lease, so that two consecutive renewals can be lost without the lease expiring.

Why an expired lease does not mean a stopped task

The scheduler’s decision to expire a lease is based on not having received a renewal. The worker’s task may nonetheless be running: a network partition between worker and scheduler stops the heartbeat without stopping the process, a long garbage collection pause stops the heartbeat without stopping the process, a worker overloaded by its own task’s CPU use may fail to send the heartbeat on time while the task proceeds.

This is the case the lease cannot resolve, and it is the case that produces two concurrent executions. The scheduler expires the lease, re-dispatches the attempt, and now two processes are executing the same task against the same target. If the task is not idempotent and not protected by a lock, both write, and the result is whatever the interleaving produces.

The situation is not hypothetical in the sense of being rare: any system that runs long tasks and enforces short leases will hit it whenever a worker’s heartbeat is delayed by the task itself. A task that blocks its own event loop, or that saturates its own network interface, delays its own heartbeat. The lease duration is therefore also a bound on how long a task may block its worker’s communication path, and tasks that do heavy synchronous work need either a longer lease or a heartbeat sent from a separate thread.

Fencing tokens

The standard defence is to make the target able to reject the older attempt. Each time a lease is granted, the scheduler issues a token that increases monotonically, whether a counter, a lease epoch, or a version number. The worker presents the token with every write, and the storage layer rejects any write whose token is lower than the highest it has seen.

class FencedStore:
    """Storage wrapper that rejects writes from superseded lease holders."""
    def __init__(self, backing):
        self.backing = backing
        self.highest = {}          # partition -> highest token seen

    def write(self, partition, token, rows):
        seen = self.highest.get(partition, 0)
        if token < seen:
            raise RuntimeError(f"token {token} superseded by {seen}")
        self.highest[partition] = token
        self.backing.write(partition, rows)

The critical property is that the comparison happens at the target, in the same place as the write, and the target remembers the highest token across attempts. A zombie holding token 7 that wakes up after the scheduler granted token 8 finds its write rejected, because the store has already recorded 8. The zombie’s work is wasted, which is the correct outcome; its write does not corrupt the live attempt’s output.

This requires the store to have a place to keep the highest token that is as durable as the data. A token counter held in the writer’s memory is useless, because the zombie’s memory says 7 and it will happily write. A token counter in the target table’s metadata, updated in the same transaction as the data write, is what makes the fence hold.

Implementations vary in how much they guarantee. A table format with a transactional log can enforce the fence on the commit, because the commit is a metadata operation that reads the current version. A plain object store has no such primitive, and the fence must be built by the writer, through a conditional put with an expected version or a separate lock service that the writer consults. Where neither is available, the fence degrades to a best-effort check, and the protection against zombies is correspondingly weaker.

The scheduler’s side of expiry

When a lease expires, the scheduler has three options, and they differ in what they assume about the zombie.

It can mark the attempt failed and re-dispatch. This maximises progress and maximises the chance of two concurrent executions. It is appropriate when the task is idempotent and fenced.

It can mark the attempt failed and wait for the worker to report before re-dispatching. This reduces the chance of concurrency but requires the scheduler to know when the worker has stopped, which is the original question it could not answer. In practice the wait is a fixed grace period longer than the lease, which is a guess.

It can leave the attempt in a running state with an expired lease and surface it for an operator. This is the safest and least useful option: no progress, but no duplication, and a human decides. Systems that choose this need an alert on leases expired beyond a threshold, or the run stalls silently.

The choice is usually made per task rather than globally, because the assumptions differ. A task that only reads has no reason to be fenced and can be re-dispatched freely. A task that writes a partition needs the fence. A task that calls an external system with side effects needs either the external system’s own idempotency key or the operator’s judgement.

Clock assumptions and their failure

Leases depend on a notion of time passing, and the scheduler’s clock is the authority. If the scheduler’s clock jumps forward, through an NTP correction or a virtual machine’s clock resynchronisation, every lease appears to have expired at once. The scheduler then re-dispatches everything that was running, and the resulting burst of duplicate attempts is bounded only by how quickly it can re-dispatch.

The mitigation is to use a monotonic clock for lease durations where the platform provides one, and to treat a large forward jump as a signal to re-evaluate rather than to act. Some systems store lease expiry as an absolute timestamp and compare against the current time; others store it as a duration and decrement. The absolute form is simpler and more vulnerable to jumps; the duration form requires a monotonic source and is immune to wall-clock corrections but harder to inspect.

Backward jumps are the mirror problem: a clock that steps back extends every lease, so dead workers are not detected until the clock catches up. Neither direction is common, and both are more common in virtualised environments than on hardware. The lease duration should be long enough that a clock correction of a few seconds does not expire a healthy lease, which for a thirty-second lease means a correction larger than the heartbeat interval is already a problem.

What the lease does not cover

The lease governs the scheduler’s relationship with the worker. It says nothing about the task’s relationship with its own inputs and outputs, and nothing about whether the worker is executing the task it claims to be executing. A worker that has lost its lease but continues to run will continue to read its input and write its output until its writes are rejected, and if the output path has no fence, it will write.

That gap is why fencing has to be implemented at the sink rather than at the scheduler. The scheduler can issue tokens and record leases, but the only place a stale write can be stopped is the place the write happens. The division is clean: the scheduler bounds how long it will wait and the sink bounds which attempt may commit. Neither is sufficient alone, and together they give the property that matters: at most one attempt’s output becomes visible for any given interval, even when two attempts ran.

This is the same commit boundary described in idempotent task boundaries, seen from the concurrency side rather than the repetition side. Once both are in place, the graph’s structure can be changed at run time without reopening the question, which is the subject of dynamic task mapping.