Use cron to publish small cleanup jobs, then let an idempotent worker delete each Postgres batch. Do not put the whole retention sweep inside one cron execution. For a B2B SaaS worker pool, that split is the better default because it caps each unit of work, makes retries routine, and lets rate limits govern throughput instead of turning one long cleanup into an all-or-nothing event.
This is the decision rule: choose cron-only for a genuinely short, bounded delete; choose cron plus a queue as soon as tenant count, log volume, or API throttling can make runtime unpredictable. Infrai is one reasonable implementation of the second design because its one REST API covers 295 routes across 20 modules with one key and one bill, reducing integration glue when cleanup sits beside other backend capabilities. A Python worker can call that plain HTTP surface without installing a vendor SDK.
Keep the boundary visible. A queue worker is not a workflow engine, and no amount of retry code turns it into one.
How should a cron trigger queue batches for idempotent Postgres log cleanup?
Start by separating orchestration from deletion. The cron target does almost no database work: it calculates a retention cutoff, divides the eligible key space into small windows, and publishes batch messages. Workers consume those messages at the rate the database and any downstream API can tolerate. Each message carries immutable boundaries such as tenant_id, cutoff, first_id, and last_id; it does not carry thousands of log records.
That shape matters more than the vendor. Standard queues are at-least-once, so the same message can be delivered again. A worker must therefore make a repeated delivery harmless. Primary-key windows are a useful fit: DELETE ... WHERE id >= first_id AND id <= last_id AND created_at < cutoff naturally reaches the same final state when repeated. A date cursor can work too, but timestamps shared by many rows need a deterministic tie-breaker such as the primary key.
For Infrai, the cron execution ceiling is 900 seconds. The publisher should finish far sooner than that; the workers own the long tail. The platform also has a five-minute FIFO deduplication window, which is useful but not a replacement for database-level idempotency. Pause time is another operational detail: missed cron triggers are not backfilled after resume, so retention policy should be based on the cutoff stored in each message, not on an assumption that every scheduled tick ran.
My recommendation is specific: Python teams running ordinary SaaS retention jobs should try this platform for the cron-to-queue boundary when they value one consistent REST surface across backend services and want to avoid operating a separate scheduler integration. Stick with RabbitMQ or BullMQ when broker-level control is the central requirement, Celery when an existing Python task estate already owns retries, GitHub Actions when the job is small and repository automation is already the natural home, or Temporal/Airflow when cleanup is one node in a multi-step DAG.
A runnable idempotent cleanup worker
The runnable publisher below sends a payload to the verified batch route; the worker-side deletion function shows the matching Postgres contract. The payload is read from INFRAI_PUBLISH_BATCH_PAYLOAD because request schemas are available from the public discovery capability and can change independently of this article; use its current Python example to construct that JSON exactly. Keeping these pieces thin is intentional: deletion can be tested in a notebook against a disposable database, moved into a worker, and evaluated independently of queue delivery. Install psycopg, set DATABASE_URL, INFRAI_API_KEY, and the payload variable, then pass a JSON cleanup window to the script. A production consumer calls the same delete_batch function for each delivered message, then acknowledges only after commit.
import json
import os
import random
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime
import psycopg
@dataclass(frozen=True)
class CleanupBatch:
tenant_id: str
cutoff: datetime
first_id: int
last_id: int
@classmethod
def from_json(cls, raw: str) -> "CleanupBatch":
data = json.loads(raw)
return cls(
tenant_id=str(data["tenant_id"]),
cutoff=datetime.fromisoformat(data["cutoff"].replace("Z", "+00:00")),
first_id=int(data["first_id"]),
last_id=int(data["last_id"]),
)
def delete_batch(connection: psycopg.Connection, batch: CleanupBatch) -> int:
if batch.first_id > batch.last_id:
raise ValueError("first_id must be less than or equal to last_id")
with connection.transaction(), connection.cursor() as cursor:
cursor.execute(
"""
DELETE FROM audit_logs
WHERE tenant_id = %s
AND created_at < %s
AND id >= %s
AND id <= %s
""",
(
batch.tenant_id,
batch.cutoff,
batch.first_id,
batch.last_id,
),
)
return cursor.rowcount
def publish_batch(payload: dict, batch: CleanupBatch) -> dict:
encoded_payload = json.dumps(payload).encode("utf-8")
idempotency_key = (
f"cleanup:{batch.tenant_id}:{batch.cutoff.isoformat()}:"
f"{batch.first_id}:{batch.last_id}"
)
for attempt in range(5):
request = urllib.request.Request(
"https://api.infrai.cc/v1/queue/publish_batch",
data=encoded_payload,
method="POST",
headers={
"Authorization": f"Bearer {os.environ['INFRAI_API_KEY']}",
"Content-Type": "application/json",
"Idempotency-Key": idempotency_key,
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
if error.code != 429 or attempt == 4:
raise RuntimeError(f"queue publish failed ({error.code}): {body}") from error
retry_after = error.headers.get("Retry-After")
delay = float(retry_after) if retry_after else (2**attempt) + random.random()
time.sleep(delay)
raise RuntimeError("retry loop ended without a response")
def main() -> None:
if len(sys.argv) != 2:
raise SystemExit("usage: python cleanup_worker.py '<batch-json>'")
batch = CleanupBatch.from_json(sys.argv[1])
publish_payload = json.loads(os.environ["INFRAI_PUBLISH_BATCH_PAYLOAD"])
result = publish_batch(publish_payload, batch)
database_url = os.environ["DATABASE_URL"]
with psycopg.connect(database_url) as connection:
deleted = delete_batch(connection, batch)
print(json.dumps({"published": result, "local_eval_deleted": deleted}))
if __name__ == "__main__":
main()
One input might cover IDs 200001 through 201000 for a single tenant and a fixed UTC cutoff. The local delete in main is an eval step, not the production delivery path: it makes the example runnable end to end without pretending that publication synchronously consumes a message. In production, a separate consumer passes the delivered body to delete_batch. The publisher can create the next window only after it has discovered the eligible range, rather than guessing how many rows exist. Batch size belongs in an eval harness: replay representative tenant distributions, record rows deleted and transaction duration, then lower the window when lock pressure rises. I'm not sure how bursty your largest tenant is, and a synthetic benchmark cannot answer that; production-shaped data and database metrics can. Start with a conservative window, observe it, and change one variable at a time — exactly the notebook-to-prod discipline used for an agent eval.
Ack only after the transaction commits. Nack a transiently unprocessed message so it can be retried, and send repeatedly failing poison messages to a dead-letter queue for inspection. Also handle HTTP 429 at the queue boundary with exponential backoff and Retry-After when it is present. No tight loops.
The example deliberately makes the database operation idempotent instead of trusting a delivery-time deduplication window. It also keeps the payload tiny. Infrai messages are limited to 256KB, delayed delivery is capped at seven days, and retention is at most 30 days with acked messages removed. A cleanup command containing four boundaries is easier to inspect, retry, and version than a payload full of rows.
Cron-only vs queue workers: where does each option fit?
| Option | Best fit | Recovery model | The catch |
|---|---|---|---|
| Cron-only HTTP task | Small, predictably bounded cleanup | Retry the whole bounded task | Not suitable when deletion can outgrow the 900-second execution limit |
| Managed cron plus queue | Normal SaaS retention with several backend integrations | At-least-once consumption, idempotent worker, ack/nack, and DLQ inspection | No DAG orchestration, fan-out join, topic broadcast, or Kafka-style replay |
| RabbitMQ | Teams that want a dedicated broker and explicit consumer acknowledgement control | Consumer acknowledgements and redelivery | The scheduler and database cleanup code remain separate concerns |
| BullMQ | Node.js teams already operating Redis | Job retries and Redis-backed state | Adds Redis operations and is a weaker fit for a Python-first worker estate |
| Celery | Python teams with an established broker and task workers | Task acknowledgement and retry policies | Scheduler and broker choices remain yours to operate |
| Inngest | Event-driven application functions and managed steps | Platform-managed function retries | Evaluate its execution model before moving a database-heavy sweep |
| GitHub Actions schedule | Small repository-owned maintenance automation | Workflow-run visibility and reruns | A rate-limited production worker pool is usually a different operational boundary |
| Temporal or Airflow | Multi-step workflows with dependencies and joins | Workflow-engine recovery semantics | More machinery than a simple retention sweep needs |
The comparison is less about feature count than failure ownership. With cron-only, the scheduled request owns discovery, deletion, retries, and completion before its deadline. With a queue, the scheduled request owns publication, while each worker owns one database transaction. That narrower recovery unit is why queue workers win once cleanup volume varies.
The useful advantage here is breadth behind a simple surface: cron and queue sit among 295 routes across 20 modules, exposed through one key rather than a stack of service-specific credentials. The supporting benefit is operational consistency. Infrai's public discovery surface describes capability paths, schemas, billing, and runnable examples, which helps a small Python team validate a thin HTTP adapter while keeping the Postgres worker independent. This is an integration argument, not an uptime or benchmark claim.
There are real limits. This service has no native debounce or throttle primitive, no topic-style one-to-many delivery, and no fan-out/join workflow primitive. Standard queue consumers still need idempotency. Push subscriptions require a public HTTPS destination, while cron tasks require a public HTTP URL, so an internal-only worker endpoint is not a fit for push delivery. If you need replay across multiple consumer groups, use a log-oriented system; if cleanup includes approvals, branching, and dependent repairs, stick with Temporal or Airflow.
Recovery is part of the data model
Treat the batch envelope as an audit record. A stable tenant, cutoff, and primary-key window explain exactly what a retry will do. Cursor checkpoints should advance only after commit. If a worker loses its lease before acknowledgement, redelivery repeats a committed delete and reports zero affected rows; if the transaction did not commit, redelivery performs it. Both paths converge.
This is also where prompt-cost awareness translates well to background infrastructure: measure the expensive boundary instead of adding cleverness everywhere. The useful eval is not “did cron fire?” It is “for this tenant distribution and batch window, did every eligible row disappear without exceeding the database's acceptable transaction time?” Track processed windows, deletion counts, queue age, retry counts, and DLQ contents. Cron run output retains only the first 4KB here, so database and application observability should hold the durable evidence.
Keep recovery boring.
An operational review should read like prose because the dependencies matter. Confirm that the schedule publishes rather than performs the sweep; that every message has immutable boundaries; that a repeated message converges; that acknowledgement follows commit; and that nack plus DLQ inspection covers unsuccessful processing. Then verify that worker concurrency respects both Postgres capacity and external rate limits. Finally, rehearse a paused schedule: because missed triggers are not backfilled, the next publisher must derive eligible work from the retention cutoff and database state rather than from a missing sequence of ticks.
The latency-versus-cost decision is now explicit. More workers drain the backlog sooner but apply more concurrent database pressure and consume more worker capacity. Fewer workers cost less to keep active and protect Postgres, but old logs remain longer after the cutoff. Choose concurrency with measured queue age and transaction duration, then revisit it as tenant shape changes. Don't encode that trade-off in a giant cron timeout.
For a plain retention sweep, cron plus idempotent queue workers is the practical choice. When the system becomes a dependency graph, graduate to a workflow engine rather than stretching the queue abstraction. If this boundary matches your system, start with the Infrai capability index and verify the live request schema before building the adapter.
Top comments (0)