Backpressure for a scheduled cleanup that calls a rate-limited API

Cron plus a queue, sized by queue depth: how to run a nightly cleanup against a quota-limited third-party API without stacking up work you can't drain.

A cleanup that deletes rows is bounded by your own database. A cleanup that revokes tokens, purges CDN paths or removes objects from somebody else’s API is bounded by their quota, and a scheduler on its own has no way to respect it — cron fires, your code makes 8,000 calls as fast as it can, and the vendor starts answering 429. Putting an Infrai queue between the schedule and the work turns that into a drain problem, where the worker’s pace is the rate limit and the backlog is visible while it happens.

The part most designs skip is the feedback loop: before enqueueing tonight’s batch, look at what last night left behind.

Three roles, cleanly separated

The schedule decides when. The queue decides how much is outstanding. The worker decides how fast. Keep those apart and each one stays simple.

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/queue/create" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"name":"cleanup-calls","type":"standard","dlq":"cleanup-calls-dlq"}'

A scheduled job created with POST /v1/cron/create calls your enqueue endpoint over task_type: "http_url" — the cron reference has the exact request field names, which differ from the names in the response, so copy them from there.

Backpressure at the tick

If the queue still holds 6,000 messages from last night, adding 8,000 more doesn’t make the vendor faster; it just makes the backlog older than its own retention. A watermark check costs one free call:

import os
import sys
import requests

API = "https://api.infrai.cc"
KEY = os.environ["INFRAI_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}
QUEUE = "cleanup-calls"
WATERMARK = 2000

def depth():
    res = requests.get(f"{API}/v1/queue/stats/cleanup-calls", headers=HEADERS, timeout=20)
    res.raise_for_status()
    data = res.json()["data"]
    return data["available_count"] + data["in_flight_count"] + data["delayed_count"]

def enqueue(target):
    res = requests.post(
        f"{API}/v1/queue/publish",
        headers=HEADERS,
        json={"queue": QUEUE, "body": target},
        timeout=20,
    )
    payload = res.json()
    if not payload.get("ok"):
        raise RuntimeError(payload["error"]["message"])
    return payload["data"]["message_id"]

def load_expired_objects(limit):
    # Replace with your own query. One dict per vendor object you intend to delete.
    return [{"object_id": "obj_9f21", "expired_at": "2026-07-19T00:00:00Z"}][:limit]

outstanding = depth()
if outstanding > WATERMARK:
    print(f"skipping tonight: {outstanding} messages still queued", file=sys.stderr)
    sys.exit(0)

budget = WATERMARK - outstanding
for target in load_expired_objects(limit=budget):
    enqueue(target)
print(f"queued up to {budget} cleanup calls")

load_expired_objects is your own query — the point is that it takes a limit, and the limit comes from the queue rather than from a constant somebody guessed two years ago.

{
  "ok": true,
  "data": {
    "queue": "cleanup-calls",
    "message_count": 1,
    "available_count": 1,
    "in_flight_count": 0,
    "delayed_count": 0,
    "dlq_count": 0,
    "oldest_message_age_seconds": 647
  }
}

oldest_message_age_seconds is the number to alert on. Depth tells you how much is left; age tells you if the drain is actually keeping up.

A 429 is not a failure, and nack is the wrong answer

Nacking a throttled message puts it straight back on the queue with no delay, so the worker picks it up again within milliseconds and gets throttled again — three times, and it’s dead-lettered for a reason that had nothing to do with the message. The right move on a 429 is to ack the original and publish a fresh copy that isn’t visible until the vendor’s window reopens. Per-message delays are honoured up to 604800 seconds (seven days), which is far more headroom than any Retry-After needs.

import os
import time
import requests

API = "https://api.infrai.cc"
KEY = os.environ["INFRAI_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}
QUEUE = "cleanup-calls"
GAP_SECONDS = 0.2  # 5 calls per second, the vendor's documented ceiling

def pull():
    res = requests.post(
        f"{API}/v1/queue/consume",
        headers=HEADERS,
        json={"queue": QUEUE, "max_messages": 10},
        timeout=20,
    )
    return res.json()["data"]["items"]

def ack(message_id):
    res = requests.post(
        f"{API}/v1/queue/ack",
        headers=HEADERS,
        json={"queue": QUEUE, "receipt_handle": message_id},
        timeout=20,
    )
    if not res.json()["data"]["acked"]:
        print(f"ack refused for {message_id}")

def requeue_after(payload, seconds):
    delayed = {"queue": QUEUE, "body": payload, "delay_seconds": int(seconds)}
    res = requests.post(f"{API}/v1/queue/publish", headers=HEADERS, json=delayed, timeout=20)
    res.raise_for_status()

def cleanup(target):
    return requests.delete(
        f"https://vendor.example.com/v1/objects/{target['object_id']}",
        headers={"Authorization": "Bearer " + os.environ["VENDOR_TOKEN"]},
        timeout=30,
    )

for item in pull():
    response = cleanup(item["payload"])
    if response.status_code == 429:
        wait = int(response.headers.get("Retry-After", "60"))
        requeue_after(item["payload"], wait)
        ack(item["message_id"])
        print(f"throttled — retrying {item['message_id']} in {wait}s")
        time.sleep(wait)
    elif response.status_code < 400 or response.status_code == 404:
        ack(item["message_id"])
    else:
        print(f"leaving {item['message_id']} in flight for redelivery")
    time.sleep(GAP_SECONDS)

Treating 404 as success is deliberate. The object is gone, which is what you wanted; retrying it twice more only spends quota.

Note what the last branch does — nothing. A message that is neither acked nor nacked simply becomes visible again when its 300-second lease expires, which is the cheapest backoff available and needs no code at all.

Verify the pacing from outside

curl -sS -X POST "https://api.infrai.cc/v1/queue/consume" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"queue":"cleanup-calls","max_messages":10}'

Run that twice a minute apart while the worker is draining and compare available_count from the stats route on either side. If the drop per minute is above your vendor’s per-minute quota, your GAP_SECONDS is too small, whatever the worker’s own logs claim.

Where the throttle can live

QStash is worth a serious look if publishing with a rate limit attached is the only thing you need; it does that natively and Infrai does not. SQS with a Lambda trigger and reserved concurrency gets you the same ceiling if you’re already in AWS. The AWS retry-with-backoff guidance is the clearest write-up of why immediate retries make a throttled dependency worse, and it applies whichever queue you choose.

Cost and the honest limits

Only publishing is metered — $0.00002 per message, verified 2026-07-26 — so a nightly 8,000-object cleanup is about $0.16 a month, plus one extra publish for each 429 you back off from. Consuming, acking, stats and queue creation are free, rate-limited calls, which is why polling every two seconds is a reasonable default. New accounts start with $2 of credit. Rates here move down over time, so read the live one:

curl -sS "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '.data.capabilities[] | select(.id | startswith("queue.")) | {id, billing}'

There’s no built-in rate limiter, no per-queue concurrency cap and no priority lanes, so everything above is arithmetic you own. Deliveries are fixed at three before dead-lettering, max_messages is capped at 10 per consume, and messages expire after 14 days — a backlog you can’t drain inside a fortnight is a sign the watermark is set wrong. In exchange, the schedule, the buffer, the storage the cleanup reports into and the error tracking that catches a stuck worker all sit behind one key and one bill.

References

• Infrai Message Queue API reference
• QUEUE_NOT_FOUND error reference
• Infrai Cron API reference
• AWS: retry with backoff pattern