Enforcing Retention Windows With Scheduled Jobs #

A retention policy is a claim about the future; the sweep job is what makes it true. The gap between the two is where most compliance findings live — a documented 180-day window, a job that has silently failed for a month, and nobody aware because a job that does not run produces no errors. This guide builds a sweep that is batched, audited and monitored for absence, as part of data retention and GDPR deletion workflows in the Pipeline Storage, Deduplication & Monitoring section.

Problem Framing #

Four failure modes account for nearly every retention problem in practice.

None of the four produce an error, which is what makes them dangerous.

The job that stops running. A scheduler change, a credential expiry, a container image that no longer starts — and the only symptom is data quietly outliving its window.

The store nobody added. A new sink is created, records land in it, and it never enters the sweep’s configuration. The sweep runs successfully every night over the stores it knows about.

The delete that times out. An unbatched delete of several million rows takes a long lock, exceeds a timeout, rolls back, and leaves the policy unenforced while appearing to have been attempted.

Flagging instead of deleting. A deleted_at column is set, the row remains, and the data is still there for anyone querying the table directly or restoring a backup.

All four are addressed by the same handful of properties: derive the store list from the same registry that defines retention, delete in bounded batches, write an audit record that outlives the data, and alert on the outcome rather than on the job’s exit code.

Step-by-Step Implementation #

1. Derive the store list from the registry #

The final count is what catches a store the sweep silently skipped.

from dataclasses import dataclass

@dataclass(frozen=True)
class RetentionTarget:
    store: str
    kind: str                 # "table" | "partitioned" | "object" | "index"
    retention_class: str
    days: int
    timestamp_column: str

def targets_from_registry(registry) -> list[RetentionTarget]:
    """Every store the pipeline writes to, with its class. A store with no class is an error."""
    targets = []
    for store in registry.stores():
        if store.retention_class is None:
            raise ConfigError(f"{store.name}: no retention class assigned")
        targets.append(RetentionTarget(
            store=store.name, kind=store.kind,
            retention_class=store.retention_class,
            days=registry.retention_days(store.retention_class),
            timestamp_column=store.timestamp_column,
        ))
    return targets

Raising on a store with no retention class is what closes the “store nobody added” gap. A new sink cannot be registered without a class, and the sweep therefore covers everything the pipeline writes to by construction rather than by someone remembering.

2. Delete in bounded batches #

from datetime import datetime, timedelta, timezone

def sweep_table(conn, target: RetentionTarget, batch: int = 5000) -> dict:
    started = datetime.now(timezone.utc)
    cutoff = started - timedelta(days=target.days)
    total, rounds = 0, 0

    while True:
        deleted = conn.execute(
            f"""DELETE FROM {target.store}
                WHERE ctid IN (
                    SELECT ctid FROM {target.store}
                    WHERE {target.timestamp_column} < %s
                    LIMIT %s
                )""",
            (cutoff, batch),
        ).rowcount
        conn.commit()                       # commit each batch: progress is monotonic
        total += deleted
        rounds += 1
        if deleted < batch:
            break
        if rounds > 10_000:                 # guard against a pathological loop
            raise RuntimeError(f"{target.store}: sweep exceeded {rounds} batches")

    return {"store": target.store, "cutoff": cutoff.isoformat(), "deleted": total,
            "rounds": rounds, "started_at": started.isoformat(),
            "finished_at": datetime.now(timezone.utc).isoformat()}

Committing each batch is the property that makes an interrupted sweep useful: it leaves strictly fewer expired rows than it found. An unbatched delete that rolls back leaves exactly as many, having consumed the whole window.

For partitioned stores the same job should drop partitions rather than delete rows, which turns the operation into metadata work — the arrangement described in partitioning scraped datasets by crawl date.

3. Audit outside the data #

def run_sweep(conn, registry, audit_store) -> list[dict]:
    results = []
    for target in targets_from_registry(registry):
        try:
            result = SWEEPERS[target.kind](conn, target)
            result["status"] = "ok"
        except Exception as exc:
            result = {"store": target.store, "status": "failed", "error": str(exc)[:400]}
        result["retention_class"] = target.retention_class
        result["retention_days"] = target.days
        audit_store.write(event="retention_sweep", **result)
        metrics.retention_sweep.labels(
            store=target.store, status=result["status"]).inc()
        metrics.retention_deleted.labels(store=target.store).inc(result.get("deleted", 0))
        results.append(result)
    return results

The audit record must outlive the rows it describes, which means it lives in a separate store with a longer retention. It records what was removed and when — counts, cutoff, store — without retaining anything about the individuals whose records were removed, which is the correct balance: evidence that the control ran, not a shadow copy of the data it deleted.

Continuing after a per-store failure matters. Aborting the whole sweep because one store errored leaves every other store unswept, which multiplies one problem into many.

4. Alert on the outcome, not the job #

groups:
  - name: retention
    rules:
      # The job has not completed successfully within its expected interval.
      - alert: RetentionSweepStale
        expr: |
          time() - max by (store) (retention_sweep_last_success_timestamp_seconds) > 129600
        for: 30m
        labels: {severity: page}
        annotations:
          summary: "{{ $labels.store }} has not been swept for over 36 hours"

      # Stronger: rows past their expiry still exist after a sweep window.
      - alert: ExpiredRowsRemain
        expr: retention_expired_rows_present > 0
        for: 2h
        labels: {severity: page}
        annotations:
          summary: "{{ $labels.store }} still holds {{ $value }} rows past their retention window"

The second alert tests the result rather than the activity, which is what catches a job that runs successfully every night while silently skipping a store. Emitting retention_expired_rows_present requires a cheap counting query per store after each sweep — a single indexed count, well worth its cost.

def count_expired(conn, target: RetentionTarget) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(days=target.days)
    return conn.execute(
        f"SELECT count(*) FROM {target.store} WHERE {target.timestamp_column} < %s",
        (cutoff,),
    ).scalar()

Verification & Testing #

The quarantine and the landing zone are the two most often omitted.

def test_sweep_deletes_only_expired_rows(db, clock):
    seed(db, fetched_at=clock.days_ago(200))     # expired under a 180-day class
    seed(db, fetched_at=clock.days_ago(10))
    result = sweep_table(db, TARGET_180D)
    assert result["deleted"] == 1
    assert db.count("listings_curated") == 1

def test_interrupted_sweep_makes_progress(db, clock):
    seed_many(db, count=12_000, fetched_at=clock.days_ago(200))
    with interrupt_after_batches(2):
        with pytest.raises(Interrupted):
            sweep_table(db, TARGET_180D, batch=5000)
    assert db.count("listings_curated") == 2_000     # two batches committed

def test_store_without_a_class_fails_configuration(registry):
    registry.add_store("new_sink", retention_class=None)
    with pytest.raises(ConfigError, match="no retention class"):
        targets_from_registry(registry)

def test_audit_survives_the_data(db, audit_store, clock):
    seed(db, fetched_at=clock.days_ago(200))
    sweep_table(db, TARGET_180D)
    entries = audit_store.query(event="retention_sweep")
    assert entries[-1]["deleted"] == 1
    assert entries[-1]["cutoff"]

The third test is the one that prevents the most common finding: it fails at configuration time when someone adds a sink without deciding how long its contents live.

Compliance & Operational Guardrails #

Every store the pipeline writes to has a retention class; a store without one fails configuration.
Deletion is physical, not a flag — a soft-deleted row is still stored data.
The sweep is batched and commits incrementally so an interruption still makes progress.
Audit records outlive the data and record store, class, cutoff and count without personal data.
Alerts fire on expired rows remaining, not merely on the job’s exit status.
Backups have a stated rotation window; a restored backup is re-swept before use.

Common Mistakes #

Alerting only on job failure. A job that runs and skips a store never fails.
Unbatched deletes. They time out, roll back, and the window passes with nothing removed.
Soft deletes. The data remains readable to anyone querying the table directly.
Retention configured per store in code. It drifts from the assessment it is supposed to implement.
Forgetting the quarantine and the landing zone. Both hold the same data as the curated store and are routinely omitted.

Frequently Asked Questions #

How often should the sweep run? #

Daily is right for most windows. The interval only needs to be short relative to the retention period, and daily keeps the batch sizes modest. What matters more than frequency is that a missed run is detected — a weekly sweep with a working staleness alert is safer than a nightly one with none.

What about backups? #

They need a documented rotation window and the assurance that a restored backup is re-swept before use. A backup retained indefinitely is a store with infinite retention, however the policy is worded. Where the backup system cannot support targeted deletion, the rotation window becomes the effective maximum retention for everything in it, and that should be stated explicitly rather than discovered.

Does this handle erasure requests? #

No, and they should stay separate. Retention expiry is bulk and time-based; erasure is targeted and identity-based, and it needs the subject index rather than a timestamp column. Both must exist, and both must cover the same store list — the erasure path is covered in automating GDPR right-to-erasure deletions.

Should the sweep run against a replica or the primary? #

The primary, because deletion has to happen where the data authoritatively lives. Replicas will follow. What is worth doing on a replica is the verification query — counting rows past expiry — since it is read-only and its load is then kept off the primary entirely. If the replica lags, the count may be briefly stale, which for an hourly alert is immaterial.

What retention applies to the audit records themselves? #

Longer than the data they describe, and defined explicitly. The point of the audit record is to demonstrate that a control ran, which is a claim that must survive the deletion it documents. Because the records contain counts and cutoffs rather than personal data, a longer window carries little risk — but it still needs to be a stated period rather than an indefinite one, or the audit store becomes the one place with no retention policy at all.

Data Retention and GDPR Deletion Workflows — retention classes and what an erasure has to reach.
Automating GDPR Right-to-Erasure Deletions — the targeted counterpart to this bulk sweep.
Partitioning Scraped Datasets by Crawl Date — making expiry a metadata operation.