fencekit

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

Celery with acks_late redelivers work after a worker crash. Redis gives you SET NX and Lua. fencekit wraps those into 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
pip install "fencekit[celery]"   # optional Celery task decorator
pip install "fencekit[otel]"     # optional OpenTelemetry hook factory

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

Quick example

from datetime import timedelta
from redis import Redis

from fencekit import (
    BeginOutcome,
    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",
    )
    resource = f"analysis:{job.pk}"
    outcome = guard.try_begin_or_reclaim(
        key, lock=lock, lock_resource=resource, ttl=timedelta(hours=24)
    )
    if outcome == BeginOutcome.ALREADY_DONE:
        try:
            return guard.get_result(key)
        except IdempotencyResultMissing:
            return None
    if outcome == BeginOutcome.IN_PROGRESS:
        return None  # active worker holds the lock

    handle = lock.acquire(resource, ttl=timedelta(minutes=5), owner_id=guard.owner_id)
    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 / try_begin_or_reclaim / mark_done / get_result: at-most-once start; reclaim stale pending after crash; optional JSON memo on completion.

DistributedLock.acquire / release / extend: lease plus 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). Optional Celery helper: fencekit.celery.idempotent_task.

Observability (optional)

Pass :class:~fencekit.hooks.FenceKitHooks to IdempotencyGuard, DistributedLock, and FenceGate, or use the OTel factory:

from fencekit import DistributedLock, FenceKitHooks, IdempotencyGuard
from fencekit.otel import otel_hooks

hooks = otel_hooks()  # pip install "fencekit[otel]"
guard = IdempotencyGuard(redis, hooks=hooks)
lock = DistributedLock(redis, hooks=hooks)

Hook callbacks must not raise; fencekit swallows errors so metrics cannot break jobs.

Celery (optional)

from datetime import timedelta

from fencekit.celery import idempotent_task
from fencekit import DistributedLock, IdempotencyGuard, idempotency_key

guard = IdempotencyGuard(redis)
lock = DistributedLock(redis)

@shared_task(bind=True)
@idempotent_task(
    guard,
    lock,
    key=lambda batch_id: idempotency_key({"batch_id": batch_id}, namespace="analysis"),
    lock_resource=lambda batch_id: f"analysis:{batch_id}",
    idempotency_ttl=timedelta(hours=24),
    lock_ttl=timedelta(minutes=5),
    retry_on_in_progress=True,
)
def analyze_batch(self, batch_id: str) -> dict:
    ...

Comparison

Short view. Sources and nuance: docs/COMPARISON.md.

fencekit complements celery-once. celery-once dedupes scheduling; fencekit rejects writes from a worker whose lock TTL already expired.

Local demo (no cloud)

Redis via Docker on your machine. No hosted services, no monthly bill.

docker compose -f examples/reference/docker-compose.yml up -d
pip install -e .
python examples/reference/demo_stale_fence.py
python examples/reference/demo_idempotency.py

See examples/reference/README.md.

Guarantees

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