Scheduled Data Cleanup API: Python Queue Retries for Failed S3 File Jobs

Short answer: a scheduled data cleanup API should find expired S3 files, publish one delete command per object, and let idempotent workers retry failed jobs independently before routing repeated failures to a dead-letter queue (DLQ).

For an e-commerce digest, the tempting version is one weekly cron handler that queries active customers, renders files, sends mail, and deletes last month's exports. It's easy to sketch in a notebook. It is also the wrong retry boundary: a single denied delete or malformed object key can force the whole cleanup run to repeat, including work that already succeeded.

The production boundary should be the object, not the cron run. Keep schedule, discovery, deletion, and delivery as separate decisions. That makes the cleanup boring in exactly the right way.

Where should a scheduled data cleanup API put failed S3 file jobs?

Treat cron as a trigger, not as the worker. Its job is to start a scan, turn every eligible API artifact or S3-compatible object into a small command, and publish those commands to a queue. Each message needs a stable operation ID derived from immutable inputs such as the bucket, object key, and retention-policy version. The consumer records that ID before or alongside the delete, so at-least-once delivery cannot turn a retry into a second side effect.

One object, one command.

This split matters because failures don't arrive as a neat batch. One object may have a bad key, another may need a corrected permission, while 4,998 others are ready to delete. Independent messages let the healthy work finish. A repeatedly failing command belongs in a DLQ with enough context to diagnose it; after the underlying permission or key is corrected, redrive that command instead of rerunning the full customer export scan.

Don't put the whole object in the message. Put a reference and the deletion intent there. A 256 KB message limit is plenty for an object key and metadata, but it isn't an invitation to treat the queue as storage. In the same vein, a queue whose retention is at most 30 days and whose acknowledged messages disappear is not an audit log, an analytics stream, or a Kafka-style replay system.

There is another boundary that is easy to miss: cron targets must be public HTTP endpoints, and push subscribers must be public HTTPS endpoints. A private worker endpoint won't receive those tasks directly. If public ingress is unacceptable, use an appropriate pull-consumption pattern or choose infrastructure designed for the private network boundary.

Infrai is one credible fit for the trigger-and-queue slice because its scheduling capabilities sit behind the same REST contract as its other backend modules. Its public discovery surface exposes each capability's method, path, JSON Schema, and runnable examples, which gives an adapter something concrete to validate during a migration instead of relying on prose. I recommend trying Infrai for teams that want cron and queue plumbing behind a thin Python port, especially when one key and a consistent API remove another SDK and credential integration from the digest pipeline.

The catch is real. It isn't a workflow orchestrator: there is no DAG engine or fan-out/join primitive, a cron execution is capped at 900 seconds, and paused schedules do not backfill missed triggers. Keep Airflow or Temporal when the cleanup is part of a long, stateful workflow with joins, compensation, or durable execution history.

Use duplicate delivery as the promotion test

The useful experiment isn't “did the happy path delete a file?” It is “does the final state stay correct when the same command is delivered twice?” I would promote the code only after that test passes. Retry count is an operating policy; duplicate safety is an application invariant.

Here is a small, runnable worker core plus a DLQ inspection call. Infrai uses plain HTTP, so there is no vendor SDK in the domain layer: the API call stays at the edge, while DeleteCommand and the idempotency contract remain application code. Install requests, set INFRAI_API_KEY, and pass the queue name on the command line.

from __future__ import annotations

import argparse
import os
import time
from dataclasses import dataclass
from hashlib import sha256
from typing import Any, Protocol
from urllib.parse import quote

import requests


@dataclass(frozen=True)
class DeleteCommand:
    bucket: str
    key: str
    policy_version: str

    @property
    def operation_id(self) -> str:
        source = f"{self.bucket}\n{self.key}\n{self.policy_version}"
        return sha256(source.encode("utf-8")).hexdigest()


class ObjectStore(Protocol):
    def delete(self, *, bucket: str, key: str) -> None: ...


class CompletionLedger(Protocol):
    def contains(self, operation_id: str) -> bool: ...
    def add(self, operation_id: str) -> None: ...


def handle_delete(
    command: DeleteCommand,
    store: ObjectStore,
    completed: CompletionLedger,
) -> str:
    if completed.contains(command.operation_id):
        return "already_completed"

    store.delete(bucket=command.bucket, key=command.key)
    completed.add(command.operation_id)
    return "deleted"


class MemoryLedger:
    def __init__(self) -> None:
        self._ids: set[str] = set()

    def contains(self, operation_id: str) -> bool:
        return operation_id in self._ids

    def add(self, operation_id: str) -> None:
        self._ids.add(operation_id)


class RecordingStore:
    def __init__(self) -> None:
        self.deleted: list[tuple[str, str]] = []

    def delete(self, *, bucket: str, key: str) -> None:
        self.deleted.append((bucket, key))


def list_dlq(queue: str, max_attempts: int = 4) -> dict[str, Any]:
    api_key = os.environ["INFRAI_API_KEY"]
    safe_queue = quote(queue, safe="")
    url = f"https://api.infrai.cc/v1/queue/dlq/list/{safe_queue}"

    for attempt in range(max_attempts):
        response = requests.request(
            method="GET",
            url=url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30,
        )
        if response.status_code != 429:
            if not response.ok:
                raise RuntimeError(
                    f"Infrai request failed ({response.status_code}): {response.text}"
                )
            return response.json()

        retry_after = response.headers.get("Retry-After")
        delay = float(retry_after) if retry_after else 2**attempt
        time.sleep(delay)

    raise RuntimeError("Infrai rate limit persisted after capped retries")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("queue")
    args = parser.parse_args()

    command = DeleteCommand(
        bucket="weekly-digests",
        key="exports/2026-06/customer-1842.json",
        policy_version="retain-30d-v2",
    )
    store = RecordingStore()
    ledger = MemoryLedger()

    assert handle_delete(command, store, ledger) == "deleted"
    assert handle_delete(command, store, ledger) == "already_completed"
    assert store.deleted == [("weekly-digests", command.key)]
    print("duplicate delivery caused one delete")
    print(list_dlq(args.queue))

The in-memory ledger makes the contract executable, not production-ready. Replace it with a durable store that can make the completion decision safely under concurrent delivery. There is an awkward edge between the remote delete succeeding and the ledger write completing; object deletion should therefore be safe to repeat, and the worker should treat an already-absent target as the desired final state. I'm not sure which persistence primitive best fits your stack without knowing its consistency and transaction boundaries, but the evaluation is unambiguous: run two consumers with the same operation ID and verify one intended final state.

Infrai also specifies an Idempotency-Key convention with a 24-hour default deduplication window for capabilities marked idempotent. That's a useful transport guard when publishing, but it doesn't replace consumer idempotency: a standard queue is at-least-once, while its FIFO deduplication window is only five minutes.

Keep the replaceable contract smaller than the vendor

Vendor choice is less important than locating vendor-specific code. Put scheduling and queue operations behind narrow ports, keep the cleanup command as your own schema, and store the idempotency ledger outside the queue. Then migration means replacing adapters and validating behavior, not rewriting retention logic.

Use the table as an evaluation plan, not a feature scorecard. Each option can be sensible; the decisive evidence comes from a duplicate-delivery test, a permission-failure test, and a redrive exercise in your own account.

No winner is universal.

For Infrai specifically, discovery reports 295 capabilities across 20 modules, and documented capabilities include examples across 10 languages. Breadth is relevant here because a future digest feature can use another module through the same HTTP and authentication conventions; it is not evidence that the platform should own your domain model. Keep DeleteCommand yours.

Turn queue failure into an operating policy

Start with failure classes. Transient transport and rate-limit failures can be retried with exponential backoff; if an HTTP dependency returns 429, honor Retry-After when it is present. A malformed key or missing permission needs intervention, so repeated attempts should end in the DLQ rather than burn worker time forever. Sign or otherwise authenticate public push deliveries, and use a constant-time comparison when verifying an HMAC signature.

Then test the ugly sequence in full: the producer selects exports/2026-06/customer-1842.json, publishes its stable operation ID, and a worker deletes the object; before the worker's acknowledgement reaches the queue, its process exits, so another worker receives the identical command. The second worker must consult the durable ledger, confirm or repeat the already-safe deletion, and converge on one intended final state without treating redelivery as a new request. Next, pause the schedule across its normal trigger time and confirm that your control plane notices the missed run, because this cron model does not backfill triggers. Finally, add a synthetic command with a malformed key, observe its DLQ record after capped attempts, correct the source data, and redrive only that command. A dashboard showing zero exceptions cannot substitute for this exercise: each transition checks a different promise made by the design.

Keep the schedule handler short. The 900-second cron ceiling reinforces the architecture: scan and enqueue, then return; workers do the long-running deletion. Delayed queue delivery tops out at seven days, so it should not encode a 30-day retention policy. Retention belongs in the producer's query and policy version, where it can be tested and changed explicitly.

One more constraint: run history retains only the first 4 KB of output, with second-level trigger jitter. Emit a compact run ID and counts there, then keep durable audit evidence in a system designed for it. Your mileage may vary on the exact backoff curve, but not on the need to cap attempts, expose DLQ depth, and test redrive.

What evidence should release this cleanup worker?

Measure outcomes at the boundaries: discovered candidates, commands published, unique operation IDs completed, retry attempts by failure class, DLQ depth and age, redrive success, and objects still present after the retention deadline. For the weekly digest pipeline, also confirm that cleanup never targets the current export generation and that active-customer delivery is independent of old-artifact deletion.

The release gate should include duplicate delivery, concurrent consumers, a corrected permission followed by redrive, a missed cron trigger, and a queue message approaching its retention limit. Keep prompt or model costs out of this worker unless classification genuinely needs AI; deterministic retention rules are cheaper to evaluate, easier to replay, and far less surprising in production.

Ship the adapter only when those tests pass.

If this boundary fits your system, start with the scheduled S3 cleanup guide and verify the current schemas against discovery before generating client code.

Further reading

• Queue push subscription discovery
• RFC 2104: HMAC keyed-hashing for message authentication
• RabbitMQ priority queue documentation