Making Celery Rebuild Tasks Idempotent with Redis Locks

Part of the Choosing a Scheduler: Celery vs APScheduler vs cron guide.

Operative rule: the lock must outlive the slowest run and be refreshed while the task works — an expiring lock is worse than no lock, because it produces two concurrent rebuilds while everyone believes one is impossible.

Why Duplicate Runs Are Normal

Celery guarantees that a task is delivered at least once. For short tasks with early acknowledgement, duplicates are rare. For long rebuilds the usual configuration is acks_late=True — acknowledge only on completion, so a killed worker’s task is not lost — and that guarantee is exactly what makes duplicates ordinary: a deploy that restarts workers mid-rebuild returns the message to the queue and another worker starts it.

Two rebuilds running concurrently on the same output path is not a hypothetical. It produces interleaved writes, a corrupted archive, or two publishes racing to swap the pointer.

How a duplicate rebuild happens Worker one receives the task and begins a rebuild. A deploy restarts it before it acknowledges. The broker redelivers the message. Worker two begins the same rebuild while worker one's partial output is still on disk. Without a claim, both write to the same paths. A deploy during a rebuild is all it takes t0 · worker A receives the task and starts tiling acks_late means the message is still unacknowledged t0+9m · a deploy restarts worker A partial output remains on disk; the broker sees no acknowledgement t0+9m · the broker redelivers; worker B starts the same rebuild without a claim, two processes now write the same output paths This is the system working as designed — the fix belongs in the task, not in the broker.

Production-Ready Implementation

from __future__ import annotations

import contextlib
import socket
import time
import uuid

import redis
from celery import Celery, Task

app = Celery("maps", broker="redis://localhost:6379/0")
r = redis.Redis.from_url("redis://localhost:6379/1")

LOCK_TTL = 3 * 60 * 60          # 3 h — well above the slowest observed run
REFRESH_EVERY = 60


@contextlib.contextmanager
def claim(name: str, ttl: int = LOCK_TTL):
    """Acquire a named claim, or yield None if someone else holds it."""
    token = f"{socket.gethostname()}:{uuid.uuid4().hex[:8]}"
    acquired = r.set(name, token, nx=True, ex=ttl)
    if not acquired:
        yield None
        return
    try:
        yield token
    finally:
        # Release only if we still hold it — never delete another owner's lock.
        release = r.register_script(
            "if redis.call('get', KEYS[1]) == ARGV[1] then "
            "return redis.call('del', KEYS[1]) else return 0 end"
        )
        release(keys=[name], args=[token])


def refresh(name: str, token: str, ttl: int = LOCK_TTL) -> bool:
    """Extend the claim if we still own it. Call periodically during long work."""
    script = r.register_script(
        "if redis.call('get', KEYS[1]) == ARGV[1] then "
        "return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end"
    )
    return bool(script(keys=[name], args=[token, ttl]))


@app.task(bind=True, acks_late=True, max_retries=3, default_retry_delay=300)
def rebuild_map(self: Task, layer: str) -> dict:
    lock_name = f"lock:rebuild:{layer}"
    with claim(lock_name) as token:
        if token is None:
            # Not an error: another worker is already doing exactly this.
            return {"status": "skipped", "reason": "already running"}

        last_refresh = time.time()
        for step in build_steps(layer):
            step.run()
            if time.time() - last_refresh > REFRESH_EVERY:
                if not refresh(lock_name, token):
                    raise RuntimeError("lost the claim mid-run — aborting")
                last_refresh = time.time()

        return {"status": "ok", "layer": layer}

Three properties make this correct rather than merely lock-shaped. The lock is set with NX and a TTL in one command, so two workers cannot both believe they acquired it. Release is conditional on still owning it, so a task that overran cannot delete the lock a different worker now holds. And losing the claim mid-run aborts rather than continuing, because the alternative is two writers who both think they are alone.

Layers of Idempotency

A lock prevents concurrency; it does not make a repeat harmless. A worker that acquires the lock after a previous run died halfway still needs the second attempt to converge on the same result.

Four layers, four different protections The claim prevents two runs at once. Content-addressed output means a repeat writes identical bytes to identical paths. An atomic pointer swap means the last writer wins cleanly. Upsert semantics mean repeated ingestion converges instead of duplicating rows. The lock is the first layer, not the only one 1 · the claim prevents two runs at the same time 2 · content-addressed output a repeat writes identical bytes to an identical path 3 · atomic pointer swap last writer wins, and both wrote the same thing 4 · upsert, never blind insert repeated ingestion converges rather than duplicating rows

Sizing the TTL

Choosing a lock lifetime A TTL shorter than the slowest run lets a second worker start mid-rebuild, which is the exact failure the lock was meant to prevent. A TTL far longer than the run blocks all rebuilds until it expires when a worker dies. Refreshing periodically while working removes the trade entirely. Too short and too long both fail — refreshing avoids choosing TTL < slowest run the lock expires mid-rebuild and a second worker starts worse than no lock: everyone believes concurrency is impossible TTL ≫ slowest run a dead worker blocks every rebuild until the TTL expires a stale lock at 02:00 means no map until someone clears it by hand moderate TTL + refresh while working held while alive, released quickly when not a crashed worker's lock expires within one refresh interval

Verification Steps

  • Enqueue the same task twice in quick succession and confirm the second returns “skipped”.
  • Kill a worker mid-run and confirm the lock expires within one refresh interval, then a retry succeeds.
  • Confirm a task that overruns its TTL aborts rather than continuing to write.
  • Confirm the release script never deletes a lock held by another token.
  • Confirm the skipped result is recorded in the run record as a skip, not as a failure.

Common Errors & Fixes

Two rebuilds still overlap occasionally

The lock is set and then given a TTL in a second command. Use a single SET key value NX EX ttl; anything else has a window.

A stale lock blocks every rebuild

A worker died holding a long TTL. Refresh periodically and keep the TTL moderate, so the lock outlives a working task but not a dead one.

The task reports failure when it was simply skipped

A skip is a normal outcome. Return it as a result rather than raising, so retries and alerts are not triggered by correct behaviour.

Locks disappear after a Redis restart

The lock database has no persistence. Locks are advisory and short-lived, so this is usually acceptable — but the rebuild must still be safe to repeat, which is what the other three layers provide.

Naming the Lock Correctly

A lock protects whatever its name identifies, and getting the granularity wrong produces either false contention or no protection at all.

Too coarse is the common first mistake: a single lock:rebuild shared across every layer means a slow rebuild of one dataset blocks every other one, turning a parallel pipeline into a serial one for no benefit. Too fine is the opposite error: a lock keyed on the task’s arguments, including a timestamp, is unique to each invocation and therefore never contends with anything, which looks like it works and protects nothing.

The right key is whatever names the resource two runs would collide on — usually the output path. If two rebuilds write to builds/incidents/, the lock should be named for that, not for the task, the queue or the schedule. Keying on the output makes the protection obvious to read and correct by construction: two tasks that cannot write the same place do not contend, and two that can, do.

Tenanted dashboards need one more component. A rebuild for tenant A and one for tenant B write different outputs and should run concurrently, so the tenant belongs in the key. Forgetting it is the classic cause of a system that is mysteriously slower than its worker count suggests: everything is queueing behind one lock that nobody realised was shared.

Retries, Skips and What the Caller Sees

A task that skips because another worker holds the lock has not failed, and the distinction matters for everything downstream. If a skip is raised as an exception, Celery retries it, the retry skips again, and eventually the task is marked failed — producing an alert about a system that is working correctly.

Returning a structured result instead keeps that clean: the caller sees skipped, the run record notes it, and no alert fires. The one case that deserves attention is a skip that repeats: a lock held for hours means either a genuinely long run or a stale claim, and counting consecutive skips is a cheap way to notice the difference before it becomes an incident.

Gotchas & Edge Cases

  • A Redis instance shared with a cache may be configured to evict keys under memory pressure, which silently removes locks. Use a separate database with eviction disabled for anything that grants exclusivity.
  • A worker paused by the operating system — swapping, or a container throttled to near zero CPU — can lose its lock without noticing; the mid-run ownership check is what turns that into an abort rather than a corruption.
  • Task time limits must be longer than the lock TTL, or the worker is killed while still holding a claim it can no longer release.
  • acks_late is what makes duplicates likely and is still the right setting for long tasks; the alternative loses work outright when a worker dies.
  • Locks acquired inside a task must be released in a finally, and the release must be conditional on ownership — an unconditional delete is the classic way one task frees another’s claim.