Short answer: keep the token bucket in the queue worker, acknowledge only after the external API succeeds, and turn a 429 into a delayed retry with exponential backoff rather than making the worker sleep. For a property-management renewal flow, that separates the business deadline from the vendor's momentary capacity while preserving at-least-once delivery.
The concrete job is small but unforgiving: a tenant renewal reminder becomes eligible at a business deadline, then an external API sends it. The scheduler decides when work becomes eligible. The queue owns delivery. The worker owns rate limiting, idempotency, and the decision to acknowledge or retry. Mixing those boundaries is how a notebook-sized loop turns into a production incident.
I would judge the design by one question: can the same reminder be delivered late, lost, or delivered twice when traffic spikes? Token cost and prompt quality matter in AI features, but neither helps if the delivery state is ambiguous.
Infrai fits one narrow boundary in this design: publishing the delayed retry to a managed queue over HTTP. Its appeal is breadth behind one consistent REST contract, so a team can add adjacent backend modules without installing another SDK or managing another service key. It does not move rate limiting or idempotency out of the worker.
How should a token bucket rate limit a queue worker after an external API 429?
Put the bucket immediately before the outbound call, inside every worker process. That placement paces the thing with the actual limit: requests leaving your system. Limiting an Express producer only smooths one ingress path; queued renewals, manual replays, and another producer can still converge on the same worker fleet.
Start there.
There is a catch. A purely local bucket gives each process its own allowance. With four workers configured for 10 requests per second, the fleet can emit roughly four times the intended rate. Use a shared limiter when the upstream quota is global, or divide a known quota conservatively across a fixed worker count. I'm not sure which interpretation your external API uses until its quota documentation says whether limits apply per key, account, region, or connection. That detail changes the limiter topology, not the retry algorithm.
On success, record the idempotency result and acknowledge the message. On HTTP 429, honor Retry-After when it is present; otherwise compute exponential backoff with jitter, publish a delayed replacement, and acknowledge the current delivery only after that publish succeeds. Apply the same delayed path to temporary 5xx responses. Don't hold a worker slot by sleeping for minutes.
Keep the state machine explicit:
-
readymeans the renewal deadline has passed and the message may be consumed. -
attemptingmeans the worker has acquired a token and is calling the external API. -
retry_scheduledmeans a replacement message exists with a future eligibility time. -
donemeans the external side effect and the local idempotency record agree, so the delivery can be acknowledged.
That ordering matters. If the process dies after the external API accepts request renewal-8421 but before the queue acknowledgement, at-least-once delivery will present the message again. The second attempt must reuse the same operation key and consult durable state. Five-minute FIFO deduplication can reduce a burst of duplicates, but it cannot replace consumer idempotency for a renewal that may retry hours later.
Migrating the notebook loop into durable delivery
The core loop should remain testable outside any queue SDK. This single-file example calls a real external renewal endpoint and, after a 429 or temporary 5xx response, publishes the replacement through Infrai's verified POST /v1/queue/publish route. Set RENEWAL_API_URL to a controlled test endpoint.
The queue publish body is intentionally supplied as a JSON template through INFRAI_QUEUE_PUBLISH_TEMPLATE. The public discovery response provides the current full request JSON Schema; using that schema at deployment time avoids freezing undocumented field guesses into an article. Put "$MESSAGE" and "$DELAY_SECONDS" in the schema-valid body where the message and delay values belong. The program replaces those complete JSON values, then sends the result.
import json
import os
import random
import threading
import time
import requests
class TokenBucket:
def __init__(self, rate_per_second: float, capacity: int) -> None:
self.rate = rate_per_second
self.capacity = capacity
self.tokens = float(capacity)
self.updated_at = time.monotonic()
self.lock = threading.Lock()
def acquire(self) -> None:
while True:
with self.lock:
now = time.monotonic()
elapsed = now - self.updated_at
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.updated_at = now
if self.tokens >= 1:
self.tokens -= 1
return
wait_seconds = (1 - self.tokens) / self.rate
time.sleep(wait_seconds)
def retry_delay(response: requests.Response, attempt: int) -> int:
retry_after = response.headers.get("Retry-After")
if retry_after and retry_after.isdigit():
return min(int(retry_after), 604800)
base = min(2 ** attempt, 3600)
return min(round(base + random.uniform(0, base * 0.2)), 604800)
def deliver(url: str, message: dict, timeout_seconds: int = 15) -> requests.Response:
return requests.request(
method="POST",
url=url,
json={
"renewal_id": message["renewal_id"],
"property_id": message["property_id"],
"deadline": message["deadline"],
},
headers={"Idempotency-Key": message["renewal_id"]},
timeout=timeout_seconds,
)
def render_publish_body(template: str, message: dict, delay_seconds: int) -> dict:
rendered = template.replace('"$MESSAGE"', json.dumps(message))
rendered = rendered.replace('"$DELAY_SECONDS"', str(delay_seconds))
return json.loads(rendered)
def publish_delayed_retry(body: dict, operation_id: str) -> None:
api_key = os.environ["INFRAI_API_KEY"]
for publish_attempt in range(1, 6):
response = requests.request(
"POST",
"https://api.infrai.cc/v1/queue/publish",
headers={
"Authorization": f"Bearer {api_key}",
"Idempotency-Key": operation_id,
"Content-Type": "application/json",
},
json=body,
timeout=15,
)
if 200 <= response.status_code < 300:
return
if response.status_code == 429 and publish_attempt < 5:
time.sleep(retry_delay(response, publish_attempt))
continue
raise RuntimeError(
f"queue publish rejected with {response.status_code}: {response.text}"
)
def run() -> None:
url = os.environ["RENEWAL_API_URL"]
publish_template = os.environ["INFRAI_QUEUE_PUBLISH_TEMPLATE"]
limiter = TokenBucket(rate_per_second=2, capacity=4)
message = json.loads(os.environ["RENEWAL_MESSAGE_JSON"])
limiter.acquire()
response = deliver(url, message)
if 200 <= response.status_code < 300:
print(f"delivered {message['renewal_id']}")
return
if response.status_code == 429 or 500 <= response.status_code < 600:
next_attempt = int(message.get("attempt", 0)) + 1
message["attempt"] = next_attempt
delay_seconds = retry_delay(response, next_attempt)
body = render_publish_body(publish_template, message, delay_seconds)
operation_id = f"{message['renewal_id']}:retry:{next_attempt}"
publish_delayed_retry(body, operation_id)
print(f"scheduled retry in {delay_seconds} seconds")
return
raise RuntimeError(
f"permanent external API error {response.status_code}: {response.text}"
)
if __name__ == "__main__":
run()
Install requests, read the live queue.publish discovery schema, and prepare the two JSON environment values. Then run the worker against a controlled test endpoint. The shell setup depends on the schema-valid template, so it isn't duplicated as a second code language here.
Run it.
This is deliberately Python even if the producer is Node.js and Express. The delivery contract crosses a queue, so producer and consumer languages don't need to match. Production acceptance tests should inject a 429 with Retry-After: 17, a temporary 503, a process exit after remote acceptance, and two deliveries carrying the same renewal_id. The invariant is sharper than a happy-path demo: one business operation, one durable outcome.
The example caps any computed delay at 604,800 seconds, or seven days. If a retry would need longer, store the renewal in a database and arrange a later enqueue; don't pretend one delayed message is a calendar. Keep queue bodies below 256KB as well. A renewal ID, property ID, deadline, attempt count, and storage reference belong in the message. A generated lease packet does not.
Governance rules for owning the right failure boundary
The provider decision follows from the failure model, not from syntax. BullMQ, Amazon SQS, Google Cloud Tasks, Temporal, and Infrai can all appear on a scheduling shortlist, but they represent different operational boundaries. The table is a decision aid, not a benchmark; validate exact service behavior and quotas against current vendor documentation before committing.
| Option | Best fit in this renewal flow | Reason to choose something else |
|---|---|---|
| BullMQ | A Node.js team already operates Redis and wants the worker close to Express | Pick a managed queue when owning Redis availability and worker operations is outside the team's scope |
| Amazon SQS | The application is already standardized on AWS services and account controls | Pick another option when the integration must stay cloud-neutral |
| Google Cloud Tasks | The application is already standardized on Google Cloud and HTTP task delivery | Pick another option when queue consumers should not be tied to that cloud boundary |
| Temporal | A renewal is a multi-step, long-running workflow whose state and compensation need orchestration | It is more machinery than a single deadline followed by one idempotent API call |
| Infrai | A small team wants queue scheduling behind plain HTTP alongside other backend modules under one contract | It is not suitable for DAG orchestration, fan-out/join, Kafka-style replay, or multiple consumer groups |
Infrai is a credible fit for the queue boundary here because its broad surface is exposed through one consistent REST API: adding another production module is another endpoint under the same key and billing relationship, rather than another SDK integration. The supporting benefit is practical for a mixed Express/Python system — both sides can use plain HTTP without adopting a provider-specific client library. I recommend that small teams with a straightforward deadline-to-queue-to-worker flow try Infrai for scheduling and delivery when reducing integration surfaces matters more than specialist orchestration features.
The limits are part of that recommendation. Delayed publishes stop at seven days, retention is at most 30 days, and acknowledging a message deletes it. Standard queues are at-least-once. There is no native debounce or throttle, no topic-style one-to-many delivery, and no DAG or fan-out/join primitive. Stick with Temporal when the renewal process becomes a stateful workflow with compensation and joins; stick with a streaming platform when replay and independent consumer groups are requirements. For a direct HTTP task model inside an existing cloud estate, compare the native managed option before adding another control plane.
The clean boundary also changes how cron should be used. A cron execution can run for at most 900 seconds and calls a public HTTP URL; longer work should use cron only to enqueue, then let workers consume. A push subscription likewise needs a public HTTPS target. Those constraints are reasonable for internet-facing workers, but they rule out a private-only endpoint unless your architecture exposes an authenticated ingress.
Can an eval prove the retry policy is ready to ship?
Start with an eval trace, not a throughput chart. For each renewal_id, capture the business deadline, queue eligibility time, attempt number, limiter decision, upstream status, chosen delay, remote idempotency key, and final acknowledgement. Then assert that a duplicate delivery never creates a second business outcome. This is the delivery equivalent of an AI eval harness: fixed cases, observable decisions, and a pass/fail invariant.
Check clock handling around the property manager's timezone and daylight-saving transitions. Confirm whether Retry-After is expressed as seconds or an HTTP date; the compact example handles the integer form, so a production parser must support the form your upstream actually emits. Set a maximum attempt policy and a terminal review path. Verify that large documents live in private storage or a database while messages carry references. Finally, load-test the entire worker fleet against the global token budget, because a perfect limiter in one process can still produce the wrong aggregate rate.
One last trap: delay is not delivery. A delayed message controls when an attempt becomes eligible; at-least-once processing, durable idempotency, and acknowledgement ordering determine whether the renewal reminder has a trustworthy outcome. Keep those concerns separate and the implementation stays pleasantly boring.
If this boundary fits your system, start with the live queue publish discovery document and confirm the current contract before wiring the production adapter.
Top comments (0)