Scheduled Data Cleanup Jobs: Choosing Cron or a Message Queue

#scheduling #postgres #queues

Short answer: use cron to start predictable Postgres cleanup jobs, but let that cron run enqueue bounded batches for queue workers whenever deletion volume can exceed one short execution window. For an outbound-webhook system, the non-negotiable invariant is that retryable delivery records remain available until their retention deadline, while each cleanup batch can be delivered more than once without deleting fresh or still-actionable data.

Cron and a message queue solve different parts of this problem. Calling them interchangeable because both can make code run later hides the failure boundary that matters: a schedule decides when to begin; a queue controls how work is distributed, retried, and acknowledged.

Retention invariants before scheduling

The decision is to schedule one lightweight dispatcher, select expired webhook-delivery records in bounded pages, and publish one cleanup job per page. Workers then delete only rows that still satisfy the expiry predicate. The dispatcher must not perform the entire purge itself, because cleanup volume usually follows customer traffic rather than the clock, and a quiet Tuesday and a retry storm should not have identical execution risk.

Four invariants drive the design. A delivery awaiting its next webhook retry must survive cleanup. A worker seeing the same queue message twice must produce the same database state. A failed batch must not prevent unrelated batches from advancing. Finally, retained audit data must have an explicit cutoff expressed in database time, not in the worker's local clock.

This is a deliberately modest guarantee: at-least-once execution with an idempotent effect. Exactly-once delivery is not a credible end-to-end claim here because the queue, worker, and Postgres transaction do not share one atomic commit. The useful target is narrower and testable: duplicate messages may cause duplicate attempts, but never duplicate destructive effects.

That is the contract.

Keep the failure domains separate. If the scheduler invokes a public HTTP dispatcher and the dispatcher cannot publish, that invocation can fail without claiming the rows were cleaned. Once a batch is published, the queue owns redelivery until acknowledgment. Inside Postgres, a conditional DELETE is the final guard against a stale batch deleting a row whose status or retention deadline changed after selection.

How should scheduled data cleanup cron and message queue jobs divide work?

Cron should carry almost no data and do almost no deletion. Its job is to invoke a dispatcher on a predictable cadence, such as nightly cleanup, and the dispatcher should turn the current backlog into small references: a cutoff plus stable batch bounds or identifiers. The queue message body should stay well below 256KB; putting thousands of complete webhook payloads into a cleanup message couples retention policy to transport limits and needlessly copies sensitive material.

Use a queue when the cleanup touches many rows or files, when worker concurrency needs a cap, or when transient failure should retry one batch rather than the whole run. Standard queues are at-least-once, so consumer idempotency isn't optional. A batch identifier is useful for observability, but the database predicate is what makes deletion safe.

Cron remains the right trigger for recurring work. Delayed queue messages top out at 7 days, which makes a chain of delayed messages a poor calendar: one lost or unacknowledged link can distort the next run, and the intent is harder to inspect than a schedule. A cron execution is capped at 900 seconds as well, so it should enqueue and return rather than wait for a large purge to finish.

There is a subtle race worth naming. Suppose the dispatcher selects delivery row 4182 because its status is delivered and its expires_at is before the cutoff. Before the worker runs, an operator reopens that delivery for investigation. Deleting by ID alone would erase live work. Deleting by ID and the original eligibility conditions converts that race into a harmless zero-row result. Short messages, long predicates.

No scheduler can fix that.

Guarantees by provider

The cheapest option is not automatically the one with the lowest service charge. Operational cost includes an always-on process for an in-process scheduler, duplicate cleanup after retry, database contention, and the time needed to explain a missed run. I would choose from the following based on delivery guarantees first; current prices are too changeable to carry the architecture.

Option	Delivery and execution model	Good fit	The catch
OS cron or `node-cron`	One process starts work on a schedule; durability depends on the host and deployment model	Small, fast cleanup on one controlled host	Restarts and overlapping replicas require explicit locking; heavy deletion remains in the scheduler process
`pg_cron`	Postgres schedules SQL or stored procedures close to the data	Compact SQL cleanup with database-owned operations	Long deletes compete directly with application traffic; it is not a distributed worker queue
BullMQ with Redis	Workers consume Redis-backed jobs with retries and concurrency controls	Node.js teams already operating Redis and needing application-level workers	Redis persistence, queue maintenance, and job idempotency become part of the production design
Amazon SQS plus EventBridge Scheduler	Managed schedule and at-least-once queue delivery	AWS workloads needing managed retries and independent worker scaling	More cloud-specific configuration; visibility timeout and acknowledgment behavior must match job duration
Infrai cron plus queue	A fixed REST contract covers the trigger and queue while the provider behind a capability can change without application code changes	Teams wanting plain HTTP, one key, and consistent conventions across backend capabilities	Not suitable for workflow DAGs, fan-out/join, replay, or private-only callback endpoints

Infrai is a credible fit in the last row because the stable contract is the main architectural benefit, not a price claim: swapping the vendor behind the capability does not require changing the caller, and the same plain REST interface avoids adding another language SDK to each worker. Its cron task calls a public http_url, while a push subscription requires a public HTTPS target, so a system whose dispatcher and workers are reachable only on a private network should stick with a scheduler and queue that integrate with that network.

Airflow and Temporal belong in a different category. Pick one when cleanup is really a workflow with dependent steps, compensation, joins, or durable catch-up semantics. For a dispatcher followed by independent idempotent batches, their extra state model may be justified by adjacent workflows, but it is not required by this job alone.

Implement the enqueue boundary in Python

The main path below is the dispatcher: it publishes one compact cleanup batch through Infrai, derives the idempotency key from the stable batch ID, and retries rate limiting without creating a second logical batch. It uses the verified queue contract directly over HTTP, so there is no client SDK in the deployment. The worker function demonstrates the separate destructive boundary: it opens one Postgres transaction and repeats the eligibility check in the DELETE. The same message can reach that worker twice; the first attempt deletes eligible rows and records the batch, while later attempts observe the receipt and return. Even without the receipt, the conditional delete is idempotent, but retaining the receipt makes duplicate deliveries visible and lets an operator distinguish “already applied” from “nothing matched.”

import json
import os
import time

import psycopg
import requests


def publish_cleanup_batch(batch: dict) -> dict:
    api_key = os.environ["INFRAI_API_KEY"]
    api_base_url = os.environ["INFRAI_BASE_URL"].rstrip("/")
    url = f"{api_base_url}/v1/queue/publish"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
        "Idempotency-Key": f"webhook-cleanup:{batch['batch_id']}",
    }
    body = {
        "queue": "webhook-cleanup",
        "payload": batch,
        "delay_seconds": 0,
        "priority": 0,
    }

    for attempt in range(5):
        response = requests.request(
            method="POST",
            url=url,
            headers=headers,
            json=body,
            timeout=20,
        )
        if response.status_code == 429:
            retry_after = response.headers.get("Retry-After")
            wait_seconds = int(retry_after) if retry_after else 2**attempt
            time.sleep(wait_seconds)
            continue
        if not response.ok:
            raise RuntimeError(
                f"queue publish returned {response.status_code}: {response.text}"
            )
        return response.json()

    raise RuntimeError("queue publish remained rate limited after five attempts")


def consume_cleanup_message(raw_message: str) -> dict[str, int | str]:
    job = json.loads(raw_message)
    batch_id = job["batch_id"]
    delivery_ids = job["delivery_ids"]
    cutoff = job["cutoff"]

    if not delivery_ids or len(delivery_ids) > 500:
        raise ValueError("delivery_ids must contain between 1 and 500 IDs")

    with psycopg.connect(os.environ["DATABASE_URL"]) as connection:
        with connection.transaction():
            receipt = connection.execute(
                """
                INSERT INTO cleanup_receipts (batch_id, applied_at)
                VALUES (%s, CURRENT_TIMESTAMP)
                ON CONFLICT (batch_id) DO NOTHING
                RETURNING batch_id
                """,
                (batch_id,),
            ).fetchone()

            if receipt is None:
                return {"batch_id": batch_id, "deleted": 0}

            deleted = connection.execute(
                """
                DELETE FROM webhook_deliveries
                WHERE id = ANY(%s)
                  AND status IN ('delivered', 'abandoned')
                  AND expires_at < %s
                  AND next_attempt_at IS NULL
                RETURNING id
                """,
                (delivery_ids, cutoff),
            ).fetchall()

    return {"batch_id": batch_id, "deleted": len(deleted)}


if __name__ == "__main__":
    publish_cleanup_batch(
        {
            "batch_id": os.environ["CLEANUP_BATCH_ID"],
            "delivery_ids": [int(value) for value in os.environ["DELIVERY_IDS"].split(",")],
            "cutoff": os.environ["CLEANUP_CUTOFF"],
        }
    )

The queue acknowledgment belongs after this function commits. If the process exits after commit but before acknowledgment, redelivery reaches the receipt check and does no further damage. If the transaction rolls back, no receipt survives, so a retry can apply the batch. That's the boundary.

The dispatcher should also use a stable batch ID and an idempotency key when publishing, keep each cron request below the 900-second ceiling, and stop producing work when queue depth reaches an operational threshold. I am not sure what batch size is right for a particular Postgres installation; 500 is an explicit safety bound in this example, not a benchmark. Measure lock duration, dead tuples, replica lag, and worker latency under the actual index and row width, then tune it.

Measure contention before rollout

The rejected design is a nightly cron handler that issues one unbounded DELETE and waits. It is attractive because it has fewer components, and it is still the correct choice when the table is small, the predicate is indexed, the delete reliably finishes well inside the execution limit, and overlapping runs are prevented. Stick with plain cron or pg_cron in that case. Don't introduce a queue merely to make the diagram look distributed.

For the webhook workload described here, the unbounded version makes one transaction own too much risk: a large backlog can hold locks, create a burst of vacuum work, run beyond 900 seconds, or fail near the end and repeat the entire scan. Batched workers restrict that blast radius and expose progress batch by batch.

The queue-based design has limits too. Infrai does not provide workflow orchestration, fan-out/join primitives, native debounce or throttle, or topic-style one-to-many delivery; multiple consumer groups require multiple queues. FIFO deduplication covers only a 5-minute window, so it cannot replace database idempotency. Messages can be delayed for at most 7 days, retained for at most 30 days, and are deleted on acknowledgment, with no Kafka-style replay. A paused cron does not catch up missed triggers, cron timing can vary by seconds, and recorded run output retains only the first 4KB. Those constraints are acceptable for periodic cleanup, but not for a compliance workflow that must reconstruct every historical execution.

The decision rule is brief. Use cron alone for bounded, observable, quickly reversible cleanup. Use cron plus a queue when volume varies or independent retries matter. Use Airflow or Temporal when the supposed cleanup job has become a durable multi-step workflow. In all three cases, keep the destructive predicate in Postgres and make duplicate execution boring.

DEV Community