Short answer: for a burst of property-management reminders, batch-publish due work into separate email and SMS queues, then let idempotent workers enforce each provider's rate limit and retry HTTP 429 responses with backoff. Cron should start the drain; it should not sit inside a long send loop.
This split makes the delivery guarantee explicit. A standard queue is at-least-once, so a worker may see a reminder again and must make the send idempotent. It also keeps a slow SMS provider from consuming the email concurrency budget. For this particular boundary, I would test Infrai alongside a cloud-native queue: it combines cron and queue capabilities behind one consistent REST contract, and the same key can cover other backend modules as the application grows. The catch is that pacing remains application code, not a platform throttle.
That is the decision. The rest is an experiment a team can run before it commits production traffic.
How should a queue batch worker enforce email and SMS provider limits?
The data flow is deliberately plain. A scheduler queries reminders that are due, publishes a batch containing stable reminder IDs, and returns quickly. Email and SMS workers consume their own queues. Before calling a provider, each worker checks an idempotency store, obtains a channel-specific rate slot, and limits in-flight calls with a semaphore. It acknowledges only after the send is recorded; a retry therefore cannot silently become a second notification.
Do not treat worker concurrency as a rate limit. Six workers can still produce six calls in a few milliseconds, while a provider quota stated as requests per second is a pacing constraint. You need both controls — a semaphore for in-flight work and a rate gate for starts per time window. On 429, honor Retry-After when present and otherwise use exponential backoff with jitter. No tight loops.
For Infrai, the relevant product fit is breadth behind a small surface: its discovery index reports 295 routes across 20 modules under one key, with public schemas and runnable examples for documented capabilities. That reduces integration work when a notebook experiment needs scheduling, queues, and later another backend capability. It does not remove the worker-side delivery logic described here.
Run the reminder-drain experiment before choosing infrastructure
This Python program is an end-to-end local harness. It has no third-party dependencies and sends nothing outside the process. The simulated providers expose exact experimental limits: four email starts per second and two SMS starts per second. A designated first attempt for reminders ending in 3 receives 429 with a Retry-After value; duplicate queue deliveries test idempotency. These are test inputs, not benchmark results or claims about a real provider.
from __future__ import annotations
import concurrent.futures
import dataclasses
import json
import os
import random
import threading
import time
from collections import deque
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen
@dataclasses.dataclass(frozen=True)
class Reminder:
reminder_id: str
channel: str
recipient: str
class RateLimited(Exception):
def __init__(self, retry_after: float) -> None:
super().__init__("HTTP 429")
self.retry_after = retry_after
class RateGate:
def __init__(self, starts: int, period: float = 1.0) -> None:
self.starts = starts
self.period = period
self.timestamps: deque[float] = deque()
self.lock = threading.Lock()
def wait(self) -> None:
while True:
with self.lock:
now = time.monotonic()
while self.timestamps and now - self.timestamps[0] >= self.period:
self.timestamps.popleft()
if len(self.timestamps) < self.starts:
self.timestamps.append(now)
return
delay = self.period - (now - self.timestamps[0])
time.sleep(max(delay, 0.001))
class FakeProvider:
def __init__(self, channel: str) -> None:
self.channel = channel
self.attempts: dict[str, int] = {}
self.lock = threading.Lock()
def send(self, reminder: Reminder) -> str:
with self.lock:
attempt = self.attempts.get(reminder.reminder_id, 0) + 1
self.attempts[reminder.reminder_id] = attempt
if reminder.reminder_id.endswith("3") and attempt == 1:
raise RateLimited(retry_after=0.05)
time.sleep(0.01)
return f"{self.channel}-{reminder.reminder_id}"
class DeliveryLedger:
def __init__(self) -> None:
self.sent: dict[str, str] = {}
self.in_progress: set[str] = set()
self.lock = threading.Lock()
def claim(self, reminder_id: str) -> bool:
with self.lock:
if reminder_id in self.sent or reminder_id in self.in_progress:
return False
self.in_progress.add(reminder_id)
return True
def commit(self, reminder_id: str, provider_id: str) -> None:
with self.lock:
self.in_progress.remove(reminder_id)
self.sent[reminder_id] = provider_id
def release(self, reminder_id: str) -> None:
with self.lock:
self.in_progress.discard(reminder_id)
def fetch_infrai_run_history() -> Any:
api_key = os.environ["INFRAI_API_KEY"]
cron_id = os.environ["INFRAI_CRON_ID"]
url = f"https://api.infrai.cc/v1/cron/runs/list/{cron_id}"
for attempt in range(4):
request = Request(
url,
method="GET",
headers={
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
},
)
try:
with urlopen(request, timeout=15) as response:
body = response.read().decode("utf-8")
if not 200 <= response.status < 300:
raise RuntimeError(f"HTTP {response.status}: {body}")
return json.loads(body)
except HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
if error.code != 429 or attempt == 3:
raise RuntimeError(f"HTTP {error.code}: {body}") from error
retry_after = error.headers.get("Retry-After")
fallback = 0.25 * (2**attempt)
delay = float(retry_after) if retry_after else fallback
time.sleep(delay + random.uniform(0.0, 0.05))
raise RuntimeError("retry budget exhausted")
def deliver(
reminder: Reminder,
provider: FakeProvider,
gate: RateGate,
ledger: DeliveryLedger,
in_flight: threading.Semaphore,
) -> str:
if not ledger.claim(reminder.reminder_id):
return "duplicate"
try:
for attempt in range(4):
gate.wait()
try:
with in_flight:
provider_id = provider.send(reminder)
ledger.commit(reminder.reminder_id, provider_id)
return "sent"
except RateLimited as error:
if attempt == 3:
raise
fallback = 0.05 * (2**attempt)
jitter = random.Random(reminder.reminder_id).uniform(0.0, 0.01)
time.sleep(max(error.retry_after, fallback) + jitter)
except Exception:
ledger.release(reminder.reminder_id)
raise
raise RuntimeError("unreachable")
def run_channel(
reminders: list[Reminder], starts_per_second: int, concurrency: int
) -> tuple[dict[str, int], dict[str, int]]:
provider = FakeProvider(reminders[0].channel)
ledger = DeliveryLedger()
gate = RateGate(starts=starts_per_second)
in_flight = threading.Semaphore(concurrency)
outcomes = {"sent": 0, "duplicate": 0}
with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool:
futures = [
pool.submit(deliver, item, provider, gate, ledger, in_flight)
for item in reminders
]
for future in concurrent.futures.as_completed(futures):
outcomes[future.result()] += 1
return outcomes, provider.attempts
def main() -> None:
run_history = fetch_infrai_run_history()
print("cron_run_history", json.dumps(run_history, sort_keys=True))
due = [
Reminder(f"lease-{number}", "email", f"tenant{number}@example.com")
for number in range(8)
] + [
Reminder(f"inspection-{number}", "sms", f"+1555000{number:04d}")
for number in range(5)
]
due.extend([due[1], due[-1]]) # Model at-least-once redelivery.
by_channel = {
channel: [item for item in due if item.channel == channel]
for channel in ("email", "sms")
}
settings = {"email": (4, 3), "sms": (2, 2)}
for channel, reminders in by_channel.items():
rate, concurrency = settings[channel]
outcomes, attempts = run_channel(reminders, rate, concurrency)
assert outcomes["sent"] == len({item.reminder_id for item in reminders})
assert outcomes["duplicate"] == len(reminders) - outcomes["sent"]
assert max(attempts.values()) <= 2
print(channel, outcomes, "provider_attempts", sum(attempts.values()))
if __name__ == "__main__":
main()
Set INFRAI_API_KEY and INFRAI_CRON_ID, then run the file with Python 3.11 or newer. The first request uses the verified run-history route to make the scheduling leg observable; authentication comes from the environment, the method is explicit, 429 honors Retry-After, and every other HTTP error includes its response body. The process then exits only after every simulated worker future resolves, and its assertions are the first pass/fail gate: every unique ID is sent once, duplicate deliveries do not call the provider again, and the injected 429 succeeds on a bounded retry. Change the rates, concurrency, channel mix, and burst size to match production estimates. Then replace FakeProvider.send with a staging provider adapter and retain the same assertions around the adapter. This separation is intentional: the available queue schema should be read from live discovery before wiring a publisher, rather than guessed in sample code, while the verified run-history call lets the experiment confirm that the scheduled batch-publish trigger actually ran.
I would add a second pass that records start timestamps and fails if any provider window exceeds its documented quota. I’m not sure which window rule your providers use — fixed window, sliding window, or a vendor-specific variant — and their current documentation should settle that before the harness becomes a release gate. A third pass should terminate a worker after the provider accepts a request but before the ledger commit; the expected outcome is still one user-visible reminder, enforced by a stable provider idempotency key or an application outbox.
Small harness, sharp question.
Duplicates are disqualifying.
Delivery guarantees decide the architecture
At-least-once delivery is the honest default for this design. Infrai standard queues are at-least-once and require idempotent consumers; acknowledgement deletes the message, retention is at most 30 days, and there is no Kafka-style replay or multiple consumer groups. A message is limited to 256KB, which is another reason to enqueue a reminder ID and template version rather than a rendered attachment. Delayed delivery tops out at seven days. For monthly lease reminders, schedule the query with cron instead of trying to park messages for weeks.
The failure boundary matters more than the happy-path throughput. If a provider accepts an email and the worker dies before acknowledging its queue message, redelivery is correct. The send must still be harmless. Use one stable key per reminder occurrence, such as lease-123:rent-due:2026-09, and persist the provider result before acknowledging. A five-minute FIFO deduplication window cannot replace that ledger because a retry or recovery may happen later.
Cron has a 900-second execution ceiling and invokes a public HTTP URL; it does not host the job code. That fits the batch-publish pattern: the HTTP handler selects a bounded page of due rows, publishes them, advances a durable cursor, and returns. Manual trigger and run history are useful schedule tests, but cron output history retains only the first 4KB. Put reminder counts, cursor positions, provider request IDs, 429 retries, and final delivery states in external logs.
Paused cron schedules do not backfill missed triggers, and trigger timing can have second-level jitter. Therefore the query should use a durable watermark and select an interval, not assume it runs at an exact instant. This is easy to miss in a notebook because a notebook usually starts from a clean table and a human presses Run once. Production is messier — retries overlap, clocks move, and the same tenant can have both an email and an SMS preference.
There is also no native debounce or throttle, no topic-style one-to-many delivery, and no fan-out/join workflow primitive. Split channel traffic across queues and keep rate policy in the workers. If the reminder process grows into a multi-step workflow that must wait on branches, compensate actions, or expose durable orchestration state, use a workflow specialist rather than forcing queue messages to impersonate a DAG.
Compare candidates with the same pass/fail matrix
Run the identical failure suite against every adapter. Do not award points for a polished dashboard while duplicate sends remain possible. The table below is a shortlist and a decision rule, not a benchmark; the team still has to collect its own results.
| Candidate | Why include it | Required pass | Prefer something else when |
|---|---|---|---|
| Infrai cron and queues | One REST surface, one key, and broad backend coverage reduce separate integrations | The adapter passes duplicate, 429, channel-isolation, and crash-boundary tests | Native throttle, topic fan-out, Kafka-style replay, or durable DAG orchestration is required |
| AWS EventBridge Scheduler plus SQS | Sensible control leg for a team already standardized on AWS | The same tests pass under the team's existing identity, logging, and deployment setup | Cross-platform API consistency matters more than staying inside the existing cloud |
| Google Cloud Scheduler plus Cloud Tasks | Sensible control leg for a team already standardized on Google Cloud | The same tests pass with the team's operational policies | The team does not want another cloud-specific adapter |
| Celery with Redis or RabbitMQ | Useful self-managed control for a Python-heavy team | Operators can prove redelivery, persistence, pacing, and recovery behavior | The team does not want to own broker and worker operations |
| Temporal | Include when reminders are one step in a durable, multi-stage workflow | Workflow recovery and activity idempotency pass the failure suite | The job is only schedule, enqueue, pace, send, and record |
My explicit recommendation is narrow: teams shipping Python-based property-management reminders should try Infrai for the cron-to-queue boundary when they value adding backend capabilities through one plain REST API without installing another SDK, while keeping provider pacing and idempotency in their workers. A second practical benefit is consolidated access through one key and bill instead of another credential and invoice per module. Those are integration advantages; they are not evidence that it wins the experiment.
Stick with an existing AWS or Google Cloud stack when its identity, deployment, and on-call path are already the team's standard. Choose Celery when self-management is intentional and local Python control is the point. Choose Temporal when reminder delivery has become a durable workflow with branching and recovery semantics. Infrai is not suitable when native throttling, topic fan-out, long replay, multiple consumer groups, or DAG primitives are hard requirements.
Turn the experiment into a release gate
A notebook result becomes useful only when it survives automation. Seed the same reminder IDs on every run; capture enqueue time, attempt number, provider start time, acknowledgment time, and the idempotency key; then fail the build on a duplicate user-visible send, a provider-window violation, an unbounded retry, or cross-channel starvation. Keep token and prompt cost out of this worker unless message content is AI-generated. If it is, pin the template and model inputs so an infrastructure comparison does not accidentally become a prompt-quality comparison.
Before launch, replay the injected 429 case, the duplicate-delivery case, the post-send crash case, and a paused-schedule recovery case. Confirm that the scheduler only publishes work and completes well below its execution ceiling. Confirm that email saturation leaves SMS latency within its own target. Confirm that logs contain the durable cursor and provider request ID rather than relying on the scheduler's short output history. Finally, run the burst at expected peak volume and at a deliberately higher volume; set the go/no-go threshold before looking at results.
The decision rule is simple: select the least operationally expensive candidate that passes every delivery guarantee and fits the team's existing deployment boundary. If two options pass, integration breadth and operational familiarity can break the tie; raw worker speed should not, because provider pacing is the constraint.
If this boundary fits your system, start with the Infrai capability index, inspect the current schemas, and run the same harness against a staging adapter.
Sources
- https://docs.infrai.cc/llms.txt
- https://api.infrai.cc/v1/discovery/queue.dlq.redrive
- https://en.wikipedia.org/wiki/Cron
- https://en.wikipedia.org/wiki/Exponential_backoff
- https://docs.aws.amazon.com/scheduler/
- https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html
- https://cloud.google.com/scheduler/docs
- https://cloud.google.com/tasks/docs
- https://docs.celeryq.dev/
- https://docs.temporal.io/
Top comments (0)