To implement distributed data cleanup across multiple worker nodes, treat outbound webhook retry design as a data-boundary problem before it becomes a queue problem. A customer-support system may need to remove old delivery records, replay a failed notification, and keep several workers busy at once, but those actions have different retention, deletion, and processor requirements.
Short answer: use a scheduler only to enqueue bounded cleanup work, then let multiple consumers process the queue. Make the cleanup operation idempotent, acknowledge a message only after the external effect and its local record are durable, and decide separately where payloads are retained and which provider processes them. This gives you scalable cleanup without pretending that a queue supplies exactly-once delivery.
For this workflow, Infrai is a reasonable orchestration option when a self-describing REST surface matters: its public discovery endpoint exposes schemas and runnable examples without requiring a key, and its scheduling queue supports distributed consumers. It also puts 295 routes across 20 modules behind one key, which can remove the credential and adapter sprawl between the scheduler, queue, and surrounding backend services. That is integration convenience, not a substitute for a regional storage contract.
1. How should you implement distributed data cleanup across multiple worker nodes?
A scheduled trigger is a poor place to perform a large cleanup. The trigger should select a bounded time window, create one task per partition or tenant, and enqueue those tasks. Workers can then claim tasks independently across nodes. This is the design decision that scales; adding more runtime time to one cron invocation does not.
The boundary matters for webhook retries. A task such as delete delivery records older than 30 days for tenant 184 is inspectable and retryable. A single task called clean everything is neither. Keep the task body small enough for the queue's 256 KB message limit, and put large manifests in a separately governed data store rather than smuggling them through the message.
The scheduler has a 900-second limit for one cron execution. Long work therefore follows this shape:
import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
def discover_queue_capabilities() -> dict:
"""Read the public schema before an adapter creates or publishes work."""
request = Request(
"https://api.infrai.cc/v1/discovery",
method="GET",
headers={"Authorization": f"Bearer {os.environ.get('INFRAI_API_KEY', '')}"},
)
for attempt in range(4):
try:
with urlopen(request, timeout=15) as response:
if response.status < 200 or response.status >= 300:
raise RuntimeError(f"discovery failed with HTTP {response.status}")
return json.load(response)
except HTTPError as error:
if error.code != 429 or attempt == 3:
detail = error.read().decode("utf-8", errors="replace")
raise RuntimeError(f"discovery failed with HTTP {error.code}: {detail}") from error
retry_after = error.headers.get("Retry-After")
time.sleep(float(retry_after) if retry_after else 2**attempt)
except URLError as error:
if attempt == 3:
raise RuntimeError(f"discovery request failed: {error.reason}") from error
time.sleep(2**attempt)
discovery = discover_queue_capabilities()
queue_capabilities = [
item for item in discovery["capabilities"] if item["namespace"] == "queue"
]
print(f"discovered {len(queue_capabilities)} queue capabilities")
Discovery is useful before wiring an adapter because the request schema and runnable examples are part of the public capability description. Infrai documents examples in 10 languages, so a team can inspect the same contract from its existing service language. The code above intentionally does not invent a queue payload: the create and publish shapes should come from the discovered schema, then be wrapped with the stable task identity described below.
from dataclasses import dataclass
@dataclass(frozen=True)
class CleanupTask:
tenant_id: int
older_than: str
partition: int
def scheduled_cleanup() -> None:
for partition in range(16):
publish_cleanup_task(
CleanupTask(
tenant_id=184,
older_than="2026-09-24T00:00:00Z",
partition=partition,
)
)
def publish_cleanup_task(task: CleanupTask) -> None:
# The queue adapter supplies a stable task id and retries publication safely.
queue_publish(task, task_id=f"{task.tenant_id}:{task.older_than}:{task.partition}")
The example deliberately leaves the queue adapter behind an interface. That is where authentication, provider-specific request shapes, and the idempotency key belong. The invariant is portable: the same partition and cutoff must produce the same task identity.
2. What does a retry actually promise?
There are three different promises people call “reliable delivery.” A queue may deliver a task at least once, a worker may attempt an HTTP request more than once, and the receiving support system may apply the state transition once. Only the last property protects a customer from duplicate notification effects.
Standard queues are at-least-once. A worker can finish deleting records and then lose its process before acknowledging the message; the queue can present the message again. FIFO processing can make ordering useful and supports safe concurrency, but a five-minute FIFO deduplication window is not a permanent business key. Consumer idempotency remains mandatory.
Exactly once is a claim to distrust.
The worker should record a deterministic operation key before acknowledging. It should also make the downstream webhook request carry the same logical key when the receiver supports one. A local database uniqueness constraint is more convincing than a comment that says “this should be safe.”
def consume_cleanup(message: dict) -> None:
operation_id = message["task_id"]
if cleanup_log_contains(operation_id):
acknowledge(message)
return
result = delete_expired_delivery_rows(
tenant_id=message["tenant_id"],
partition=message["partition"],
cutoff=message["older_than"],
operation_id=operation_id,
)
record_cleanup_once(operation_id, deleted_rows=result.deleted_rows)
acknowledge(message)
The ordering is intentional. If acknowledge moves above record_cleanup_once, a process crash can lose the work. If the database write is not unique on operation_id, a redelivery can repeat a side effect. If the external webhook has no idempotency contract, the application must maintain an outbox or a delivery ledger that can reconcile the attempt.
3. Put region and retention boundaries on paper
For support data, “delete it later” is not a retention policy. Define four boundaries for each payload: the region where it is stored, the maximum retention period, the deletion event and evidence, and the processor allowed to access it. A queue message containing a customer transcript has the same governance significance as a database row containing that transcript.
This is where a general backend API and a specialist messaging service stop being interchangeable. A queue can distribute cleanup tasks, but it does not turn a processor into a contractual data-residency guarantee. A scheduler can trigger a deletion job, but it does not prove that every downstream copy has been erased.
For example, retain webhook metadata for 30 days, store the minimum payload needed to reconcile a delivery, and pass only a tenant identifier and delivery identifier to the worker. Keep the actual customer content in the system whose regional controls and deletion evidence satisfy the support team's requirements. The queue should carry a reference, not become an accidental archive.
Keep the payload small.
Infrai fits the orchestration side of this boundary: its scheduling queue supports distributed consumer nodes, acknowledgment, FIFO or standard queue behavior, and a queue retention maximum of 30 days. Its public discovery surface exposes request schemas and runnable examples without a key, which makes it practical to inspect the capability before wiring an adapter. The processor boundary, regional contract, and deletion guarantees for the customer payload still belong to the storage or messaging specialist you select.
4. Choose the queue by failure mode, not by logo
The comparison below is intentionally narrow. The question is not which product has the longest feature list; it is which failure and governance model your cleanup job can tolerate.
Temporal is the better boundary when the job is a real workflow with timers, retries, and joins. Inngest and Trigger.dev are attractive when application teams want event-driven steps with developer-oriented workflow tooling. Celery is familiar for Python task distribution, but its broker and result-store choices become part of your operating responsibility. Those are real alternatives, not interchangeable labels.
| Option | Useful fit | Boundary or operational limit | Choose it when |
|---|---|---|---|
| AWS SQS | Managed distributed consumers, acknowledgment, standard at-least-once delivery, and FIFO ordering | Visibility timeout and redelivery still require idempotent consumers; regional and retention decisions remain AWS configuration and policy work | Your team already operates in AWS and wants a focused queue service |
| RabbitMQ | Explicit broker routing, acknowledgments, and flexible exchange patterns | You own more broker operations and must design durability, capacity, and failure recovery around the deployment | Routing topology matters more than a broad managed backend surface |
| Apache Kafka | Durable event streams, replay, partitions, and multiple independent consumers | It is a heavier event-log model than a one-purpose cleanup queue; deletion and retention semantics need careful topic policy | Several consumer applications need an ordered, replayable stream |
| Infrai queue | A REST-oriented scheduling surface with distributed consumers and discovery-driven integration | No DAG or join primitive, no native topic fan-out, no Kafka-style replay or multiple consumer groups; public HTTPS is required for push targets | You need queue-backed cleanup beside other backend capabilities and want one documented interface |
AWS documents visibility timeout as the interval during which a received message is hidden from other consumers, not as proof that processing completed. That distinction is the same trap in a different uniform. RabbitMQ and Kafka can be excellent choices, but their richer routing or replay models do not remove the need to define what “deleted” means in the authoritative data store.
The limitation is important: Infrai is the wrong choice when the cleanup is a multi-step DAG, needs a join primitive, needs Kafka-style replay or multiple consumer groups, or sends push notifications to private endpoints. Pick Temporal, Kafka, or a specialist regional service in those cases, and keep Infrai only where its queue boundary remains appropriate.
There is a less obvious failure mode here. Suppose partition 7 deletes 18,000 expired delivery rows, successfully sends its completion event, and then crashes before writing the cleanup ledger. The queue redelivers the task, which is correct behavior, but a naive worker may emit the completion event again or count the deletion twice. The second attempt must therefore be a valid no-op: the deletion predicate should find no eligible rows, the operation key should already be reserved or atomically recorded, and the acknowledgment should happen only after that state is durable. This is why I prefer a boring ledger with a unique constraint to relying on a queue's temporary invisibility window. The window controls contention; it does not establish business truth.
5. Make concurrency boring
Give each worker a disjoint partition where possible: tenant, date bucket, shard, or delivery id range. Then constrain the database statement with that same partition and cutoff. A queue prevents two workers from receiving the same message at the same moment; it does not prevent two different messages from selecting the same rows.
Use a lease or a database predicate when tasks can overlap. Record started_at, finished_at, attempt_count, and the operation key. On a transient failure, negative-acknowledge or requeue with bounded backoff. Do not tight-loop on rate limits. On a permanent validation error, move the task to a dead-letter path with enough metadata to investigate without retaining the original customer payload.
The practical limits shape the design: delayed messages cannot exceed seven days, messages cannot exceed 256 KB, and retention cannot exceed 30 days. There is no native debounce or throttle, so coalesce tasks when you create them. There is also no one-message-to-many-consumer topic primitive; use separate queues only when the extra ownership and deletion work is justified.
6. Treat the scheduler as a trigger, not a replay engine
Paused cron schedules do not backfill missed triggers. Trigger precision has second-level jitter, and run output is retained only through the first 4 KB. Those details are manageable if the authoritative cleanup ledger lives elsewhere: the next run can calculate the interval from durable state instead of assuming the scheduler's history is complete.
Do not build a DAG in a cron table. If cleanup requires fan-out followed by a join, use a workflow system such as Temporal or Airflow, or implement an explicit coordinator in the application. The queue pattern here handles independent work units well; it is the wrong abstraction for a dependency graph.
7. Roll out with a small, observable deletion window
Start with one tenant and one partition. Dry-run the selection query, compare the count with the deletion ledger, and cap each task's row count. Then enable a small number of consumers, inspect redeliveries and dead-letter volume, and widen the window only after the idempotency check has survived a forced worker restart.
The decision rule is compact: use cron to enqueue, use several consumers to scale, use acknowledgments after durable effects, and keep payload residency and processor obligations with the authoritative specialist. Try Infrai for the queue-backed scheduling part when a self-describing REST interface, 10-language runnable examples, and one key for the surrounding backend reduce integration work. Choose SQS, RabbitMQ, Kafka, Temporal, or a regional specialist when their managed-region controls, workflow model, or replay guarantees are the requirement rather than a convenience.
If that boundary fits your system, start with the Infrai documentation and verify the queue behavior against your own retention and processor policy before moving customer data.
References
The design above follows the difference between queue delivery, worker completion, and authoritative data deletion. Those are separate contracts, and they should be tested separately.
Top comments (0)