fencekit

Redis-backed idempotency and fenced distributed locks for background jobs.

Celery with acks_late redelivers work after a worker crash. Redis gives you primitives (SET NX, Lua). fencekit wires them into three tested pieces:

Extracted from ChessMate (Celery + Redis + Postgres). See DESIGN.md for guarantees, non-guarantees, and crash semantics.

Install

pip install fencekit
pip install "fencekit[django]"   # optional Django QuerySet helper

Requires Redis 6+ (tested with Redis 7) and Python 3.10+.

Quick example

from datetime import timedelta
from redis import Redis

from fencekit import (
    DistributedLock,
    FenceGate,
    IdempotencyGuard,
    IdempotencyResultMissing,
    fenced_update,
    idempotency_key,
)

r = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
guard = IdempotencyGuard(r)
lock = DistributedLock(r)
fence = FenceGate(r)

def analyze_batch(job) -> dict | None:
    key = idempotency_key(
        {"game_ids": ["abc", "def"], "engine": "sf16"},
        namespace="analysis",
    )
    if not guard.try_begin(key, ttl=timedelta(hours=24)):
        try:
            return guard.get_result(key)  # prior outcome on redelivery
        except IdempotencyResultMissing:
            return None  # still pending or done without a memo

    handle = lock.acquire(f"analysis:{job.pk}", ttl=timedelta(minutes=5))
    try:
        fence.set_if_fresh(handle.token, f"analysis:{job.pk}:status", "running")
        fenced_update(
            type(job).objects.filter(pk=job.pk),
            handle.token,
            updates={"progress": 50, "status": "running"},
        )
        result = {"report_id": job.pk, "status": "done"}
        guard.mark_done(key, result=result, ttl=timedelta(hours=24))
        return result
    finally:
        lock.release(handle)

API overview

• idempotency_key(payload, namespace=...) — deterministic key from JSON-canonicalized payload
• IdempotencyGuard.try_begin / mark_done / get_result — at-most-once start; optional JSON memo on completion
• DistributedLock.acquire / release / extend — lease + monotonic fencing token (Lua)
• FenceGate.set_if_fresh — atomic fenced Redis string writes
• fenced_update(queryset, token, updates=...) — fenced Django/Postgres UPDATE in one statement

Typed public API (py.typed). No Celery adapter yet; wire the guard in your task body for now.

Guarantees (honest)

fencekit does not implement Redlock. Fencing only works when the storage layer checks the token in the same operation as the write.

Development

REM Windows (CMD)
python -m pip install -e ".[dev]"
python -m ruff check src tests
python -m mypy src
python -m pytest -m "not integration" -q

Full suite (Redis on localhost:6379):

set FENCEKIT_REDIS_URL=redis://localhost:6379/15
python -m pytest -q

# Linux/macOS with uv
uv sync --extra dev
uv run pytest

CI runs lint, type-check, and tests on Python 3.10–3.13 with Redis. Releases publish to PyPI via Trusted Publishing (OIDC, no long-lived token).

License

MIT