Library Catalog Data Sync Pipelines

Automating Retention Schedule Enforcement with Celery Beat

This page answers one narrow operational question: why does patron circulation data keep outliving its lawful retention window even though you have a purge script, and how do you make the purge run on a schedule that is idempotent, overlap-safe, and provable at audit time? It sits under Data Retention Policies for Public Libraries, which defines the retention windows themselves, and within the broader Patron Validation & Privacy Data Routing architecture. If your retention purge is a script someone remembers to run each quarter, this is the failure you will eventually discover during an audit — and by then the data has already overstayed.

Celery Beat retention enforcement flow with a distributed lock and audit log A Celery Beat schedule fires a periodic retention task on a fixed cadence. The task first acquires a distributed Redis lock; if the lock is already held by a prior run it skips immediately rather than overlapping. Once the lock is held, the task purges only the records whose retention window has expired, deleting each idempotently, then writes an append-only audit entry recording the run. If the purge raises, an alert is emitted to on-call. The audit log at the right captures every run so gaps become visible. Beat schedule daily 02:00 crontab periodic task enforce_retention distributed lock no overlap purge due idempotent delete audit log append-only lock held skip this run on failure alert on-call

Problem Framing

The symptom is silence. Nobody reports an error, no alert fires, and the retention purge “works” — until an audit asks for proof that it ran. When you pull the purge history, the gaps are unmistakable: a run every quarter until the person who owned the script changed teams, then months of nothing, then a single catch-up run far too large to be routine.

text
$ psql -c "SELECT date_trunc('month', run_at) AS month, count(*) AS runs, \
           sum(rows_purged) AS purged FROM retention_purge_audit \
           GROUP BY 1 ORDER BY 1"
  month     | runs | purged
------------+------+--------
 2025-10-01 |    1 |   4213
 2026-01-01 |    1 |   3987
 2026-04-01 |    0 |      0   <-- no purge ran; data retained past its window
 2026-05-01 |    0 |      0   <-- still nothing
 2026-06-01 |    0 |      0   <-- still nothing
 2026-07-01 |    1 |  38120  <-- catch-up run; a quarter of overdue data at once

Every zero-row month is a compliance exposure: circulation records whose retention window closed but that were never purged, sitting in the database, discoverable, and technically unlawful to retain. The catch-up spike is worse than the gap — it proves the data survived months past its window before anything touched it. A purge that depends on a human remembering is a purge that will, eventually, be forgotten.

Root Cause

No scheduler owns the recurring purge. The work exists as a management command or a standalone script that a person runs by hand, or at best a bare cron one-liner on a single box. Each of those arrangements is missing the same four properties, and the absence of any one of them is enough to produce the gap above.

The retention windows are already defined correctly in Data Retention Policies for Public Libraries. What is missing is a component whose job is to run the enforcer on time, exactly once per interval, and to leave a provable record whether it succeeds or fails.

Solution

Move the purge into a Celery Beat periodic task. Beat is the scheduler; a worker executes the task; a distributed lock guarantees a single run at a time; the delete is idempotent per record; and every outcome — ran, skipped, failed — is written to the append-only audit table. This reuses the same broker and worker topology described in Using Celery for Distributed Catalog Ingestion; you are adding a scheduled task to an existing Celery deployment, not standing up new infrastructure.

First, declare the schedule so the cadence lives in the application, not in one host’s crontab:

python
from __future__ import annotations

from celery import Celery
from celery.schedules import crontab

app = Celery("librarycatalog")

app.conf.beat_schedule = {
    "enforce-retention-daily": {
        "task": "retention.tasks.enforce_retention",
        # Run every night at 02:00; small windows purged daily never
        # accumulate into a quarter-sized catch-up batch.
        "schedule": crontab(hour=2, minute=0),
        "options": {"expires": 3600},  # drop if a worker cannot start it within the hour
    },
}

Then the task itself. The lock is acquired first and released in a finally; the purge deletes only records whose window has expired; and each outcome is logged with structured extra={...} and recorded in the audit table.

python
from __future__ import annotations

import logging
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Iterator

from celery import shared_task
from django.db import transaction
from redis import Redis
from redis.exceptions import RedisError

logger = logging.getLogger("retention.enforce")

_redis = Redis.from_url("redis://localhost:6379/1")
_LOCK_KEY = "lock:retention:enforce"
_LOCK_TTL = 1800  # seconds; longer than a healthy run, short enough to self-heal


class LockNotAcquired(Exception):
    """A prior run still holds the retention lock."""


class RetentionPurgeError(Exception):
    """The purge could not complete and must alert on-call."""


@contextmanager
def distributed_lock(key: str, ttl: int) -> Iterator[None]:
    """Redis single-holder lock; auto-expires so a crashed run cannot wedge it."""
    acquired = _redis.set(key, "1", nx=True, ex=ttl)
    if not acquired:
        raise LockNotAcquired(key)
    try:
        yield
    finally:
        _redis.delete(key)


@shared_task(
    bind=True,
    name="retention.tasks.enforce_retention",
    autoretry_for=(RedisError,),
    retry_backoff=True,
    max_retries=3,
)
def enforce_retention(self, *, retention_days: int = 730) -> dict[str, int | str]:
    """Purge circulation records past their retention window, exactly once per run."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
    run_id = self.request.id

    try:
        with distributed_lock(_LOCK_KEY, _LOCK_TTL):
            purged = _purge_expired(cutoff=cutoff, run_id=run_id)
    except LockNotAcquired:
        logger.info(
            "retention_skip_overlap",
            extra={"run_id": run_id, "reason": "prior_run_active"},
        )
        return {"status": "skipped", "run_id": run_id}
    except Exception as exc:  # noqa: BLE001 — audit + alert on any failure
        logger.exception(
            "retention_run_failed",
            extra={"run_id": run_id, "cutoff": cutoff.isoformat()},
        )
        _record_audit(run_id=run_id, rows_purged=0, status="failed", detail=str(exc))
        raise RetentionPurgeError(run_id) from exc

    logger.info(
        "retention_run_complete",
        extra={"run_id": run_id, "rows_purged": purged, "cutoff": cutoff.isoformat()},
    )
    _record_audit(run_id=run_id, rows_purged=purged, status="ok", detail="")
    return {"status": "ok", "run_id": run_id, "rows_purged": purged}


def _purge_expired(*, cutoff: datetime, run_id: str) -> int:
    """Idempotent per-record delete: only rows strictly older than the cutoff."""
    from circulation.models import CirculationEvent

    with transaction.atomic():
        # Re-selecting < cutoff each run makes the operation naturally idempotent:
        # a re-run over already-purged rows simply matches nothing.
        qs = CirculationEvent.objects.filter(occurred_at__lt=cutoff)
        deleted, _ = qs.delete()
    logger.debug(
        "retention_batch_deleted",
        extra={"run_id": run_id, "deleted": deleted},
    )
    return deleted

The behavioural change is that the schedule now owns the work. Beat fires enforce_retention every night whether or not anyone is watching. The lock makes a second concurrent run a no-op instead of a race. The delete is idempotent because it selects occurred_at < cutoff on every run — a re-run after an interruption matches only rows that still qualify, so nothing is double-counted and nothing is missed. And because both success and failure write to retention_purge_audit, the gap that started this investigation becomes structurally impossible: a night with no ok row is itself a visible, alertable anomaly.

Compliance or Privacy Impact

Automating enforcement changes the retention posture in three concrete ways, each with a downstream effect worth tracking.

The task narrows the retention surface rather than widening it, but it shifts risk onto the correctness of retention_days and the cutoff query. A window set too long silently retains data past its lawful period with the scheduler’s full authority behind it, so changes to that value deserve the same code review and the verification below.

Verification

Confirm three properties: that only expired rows are deleted, that a concurrent run is skipped rather than racing, and that a failure is recorded and re-raised so it alerts.

python
import pytest
from datetime import datetime, timedelta, timezone

from redis import Redis

from retention.tasks import _purge_expired, distributed_lock, LockNotAcquired


def test_only_expired_rows_are_purged(db) -> None:
    from circulation.models import CirculationEvent

    now = datetime.now(timezone.utc)
    cutoff = now - timedelta(days=730)
    old = CirculationEvent.objects.create(occurred_at=cutoff - timedelta(days=1))
    fresh = CirculationEvent.objects.create(occurred_at=now - timedelta(days=10))

    deleted = _purge_expired(cutoff=cutoff, run_id="test")

    assert deleted == 1
    assert not CirculationEvent.objects.filter(pk=old.pk).exists()   # expired, gone
    assert CirculationEvent.objects.filter(pk=fresh.pk).exists()     # in window, kept


def test_purge_is_idempotent(db) -> None:
    from circulation.models import CirculationEvent

    cutoff = datetime.now(timezone.utc) - timedelta(days=730)
    CirculationEvent.objects.create(occurred_at=cutoff - timedelta(days=5))

    first = _purge_expired(cutoff=cutoff, run_id="run-1")
    second = _purge_expired(cutoff=cutoff, run_id="run-2")  # re-run over same window

    assert first == 1
    assert second == 0  # nothing left to purge; no double-count


def test_overlapping_run_is_skipped() -> None:
    with distributed_lock("lock:test:overlap", ttl=30):
        with pytest.raises(LockNotAcquired):
            with distributed_lock("lock:test:overlap", ttl=30):
                pytest.fail("second holder must not acquire the lock")

For a running deployment, add a continuous invariant on the audit table itself: a monitor that asserts a status='ok' row exists for every scheduled interval, and pages if a night is missing. That is the same detection that would have caught the original gap on the first missed night instead of at audit time a quarter later. Track the rows_purged distribution too — a sudden spike is the signature of a schedule that stopped and restarted, exactly the catch-up batch this task exists to prevent.