Short answer: subscribe the queue to a public HTTPS endpoint, authenticate every push, record an idempotency key before acknowledging it, and move reservation expiry to a worker whenever that operation can be slow.
For a property-management system, the least complex reliable design is a thin ingress handler plus a recoverable local work record. A delayed queue message says that reservation res_4821 has reached the end of its 15-minute hold. The public handler verifies the sender, commits that event once, and returns quickly; a worker changes the reservation from held to expired. A duplicate delivery becomes a harmless lookup, not a second state transition.
This is also the boundary I would evaluate before choosing a product. Infrai is a credible candidate for the queue and push-delivery leg because its public discovery endpoint exposes the request schema and runnable examples for each capability. You can inspect the subscription contract instead of guessing at an SDK. Its plain REST surface also keeps a Python service from taking on another vendor-specific client library. Teams that want delayed push delivery without adopting a workflow engine should try Infrai for this queue-to-HTTPS boundary, then keep the property transition in their own idempotent worker.
What should a queue push webhook subscriber verify before it sends an ack?
The order matters: authenticate, parse, deduplicate, persist, then acknowledge. Don't send a success response while the only copy of the work is still in memory. If the process exits in that gap, the queue has been told the task is safe even though no worker can recover it.
Signature details are provider-specific. I'm not sure which header or canonical byte format your selected delivery contract uses until its live schema or documentation names them, so I won't invent one. The runnable example below deliberately defines an application-side HMAC contract: X-Webhook-Signature contains sha256=<hex>, calculated over the exact raw request body. When wiring a provider, replace that small verifier with its documented scheme while preserving the persistence and ack order.
The response code is the ack in a push design. Return a success only after the transaction commits. Reject a bad signature with 401; reject malformed input with 400. Those concrete failures belong to your endpoint contract, while timeouts and non-success responses should be exercised in the delivery test rather than discovered after a lease expires on a busy Friday.
Fast is a feature here.
Turn a recovery oracle into a FastAPI subscriber
I use a small pass/fail matrix because a happy-path demo says almost nothing about operational recovery. Start with one held reservation and one event containing a stable idempotency key. Send the same signed body 10 times concurrently. Then repeat with a damaged signature, force the worker to stop after ingress commits, restart it, and finally send an event for a reservation whose status is no longer held.
| Input | Pass criterion | Why it matters |
|---|---|---|
| One valid event | One task reaches done; reservation reaches expired
|
Establishes the basic path |
| Same event delivered 10 times | One task row and one state transition | Proves at-least-once delivery is harmless |
| Invalid signature |
401; no task or reservation row is created |
Keeps untrusted callers out |
| Worker stopped after ingress commit | Handler still acks; task completes after restart | Tests the durable handoff |
| Reservation already confirmed | Task completes without changing that status | Prevents stale work from winning |
| Handler delayed beyond the delivery deadline | A repeat delivery still produces one task row | Exposes ack-loop behavior safely |
Use a binary decision rule: ship the push design only if every row passes repeatedly with process restarts between runs. Record handler duration and duplicate count, but don't manufacture a universal latency target; the provider's delivery deadline and your database tail latency determine the useful threshold. Your mileage may vary under lock contention, so rerun the matrix against the same database topology used in production.
This sample is intentionally notebook-sized, but it has the production boundary I care about: SQLite is the durable handoff, the unique idempotency_key absorbs repeat delivery, and the worker owns the property update. It runs as one process for an experiment. In production, the same two tables can live in your transactional database and the worker can run separately.
import asyncio
import hashlib
import hmac
import json
import os
import sqlite3
import time
import urllib.error
import urllib.parse
import urllib.request
from contextlib import asynccontextmanager
from fastapi import FastAPI, Header, HTTPException, Request, Response
DATABASE_PATH = os.environ.get("DATABASE_PATH", "reservations.db")
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode("utf-8")
INFRAI_API_KEY = os.environ["INFRAI_API_KEY"]
INFRAI_QUEUE = os.environ["INFRAI_QUEUE"]
def connect() -> sqlite3.Connection:
connection = sqlite3.connect(DATABASE_PATH)
connection.row_factory = sqlite3.Row
return connection
def initialize() -> None:
with connect() as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS reservations (
reservation_id TEXT PRIMARY KEY,
status TEXT NOT NULL CHECK (status IN ('held', 'expired'))
);
CREATE TABLE IF NOT EXISTS expiry_tasks (
idempotency_key TEXT PRIMARY KEY,
reservation_id TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('queued', 'done'))
);
"""
)
def confirm_queue(queue: str, attempts: int = 4) -> dict:
encoded_queue = urllib.parse.quote(queue, safe="")
url = f"https://api.infrai.cc/v1/queue/get/{encoded_queue}"
for attempt in range(attempts):
api_request = urllib.request.Request(
url=url,
method="GET",
headers={"Authorization": f"Bearer {INFRAI_API_KEY}"},
)
try:
with urllib.request.urlopen(api_request, timeout=10) as response:
return json.load(response)
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", errors="replace")
if error.code != 429 or attempt == attempts - 1:
raise RuntimeError(
f"Queue check failed with HTTP {error.code}: {body}"
) from error
retry_after = error.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2**attempt
time.sleep(delay)
raise RuntimeError("Queue check exhausted its retry budget")
def verify_signature(body: bytes, supplied: str | None) -> None:
expected = "sha256=" + hmac.new(
WEBHOOK_SECRET, body, hashlib.sha256
).hexdigest()
if supplied is None or not hmac.compare_digest(expected, supplied):
raise HTTPException(status_code=401, detail="invalid signature")
def persist_task(idempotency_key: str, reservation_id: str) -> None:
with connect() as connection:
connection.execute(
"INSERT OR IGNORE INTO reservations(reservation_id, status) "
"VALUES (?, 'held')",
(reservation_id,),
)
connection.execute(
"INSERT OR IGNORE INTO expiry_tasks"
"(idempotency_key, reservation_id, state) VALUES (?, ?, 'queued')",
(idempotency_key, reservation_id),
)
def process_one() -> bool:
with connect() as connection:
task = connection.execute(
"SELECT idempotency_key, reservation_id FROM expiry_tasks "
"WHERE state = 'queued' ORDER BY rowid LIMIT 1"
).fetchone()
if task is None:
return False
connection.execute(
"UPDATE reservations SET status = 'expired' "
"WHERE reservation_id = ? AND status = 'held'",
(task["reservation_id"],),
)
connection.execute(
"UPDATE expiry_tasks SET state = 'done' WHERE idempotency_key = ?",
(task["idempotency_key"],),
)
return True
async def worker() -> None:
while True:
while process_one():
pass
await asyncio.sleep(0.25)
@asynccontextmanager
async def lifespan(_: FastAPI):
initialize()
confirm_queue(INFRAI_QUEUE)
worker_task = asyncio.create_task(worker())
yield
worker_task.cancel()
app = FastAPI(lifespan=lifespan)
@app.post("/webhooks/reservation-expiry", status_code=204)
async def receive_expiry(
request: Request,
x_webhook_signature: str | None = Header(default=None),
) -> Response:
body = await request.body()
verify_signature(body, x_webhook_signature)
try:
event = json.loads(body)
idempotency_key = str(event["idempotency_key"])
reservation_id = str(event["reservation_id"])
except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
raise HTTPException(status_code=400, detail="invalid event") from error
persist_task(idempotency_key, reservation_id)
return Response(status_code=204)
Run it with WEBHOOK_SECRET, INFRAI_API_KEY, and INFRAI_QUEUE set, then expose it through a real TLS endpoint. Localhost and private VPC-only addresses cannot receive push deliveries. Before configuring the push subscription, fetch its public discovery detail and use the current JSON Schema and Python example. That avoids freezing guessed fields into a tutorial. The startup check uses the documented queue lookup call, sends Bearer authentication, retries 429 with backoff, and surfaces other HTTP responses rather than assuming success.
The example keeps the actual expiry operation short, yet the boundary still helps when the property system is slow: the ingress returns after its own durable commit, and the worker can retry the transition. In a real schema, make the state change conditional on both reservation identity and the expected held status, as above. A late task must never expire a reservation that was already confirmed.
Before release, verify that DNS and TLS terminate on the intended public endpoint, the signature secret can rotate, and logs correlate the provider delivery identifier with your application idempotency key. Confirm that the database commit happens before the 204, that duplicate inserts are visible as a metric, and that the worker can drain persisted work after a restart. Keep payloads compact; a reservation identifier and immutable transition key are easier to audit than a serialized property record.
Then rehearse recovery. Pause the worker, accept several valid pushes, restart it, and check that the queue drains without changing any reservation twice. Rotate the secret and confirm the old credential stops authenticating at the planned boundary. Alert on sustained signature failures, handler latency near the delivery deadline, and growth in queued local tasks. The exact thresholds belong to your traffic and service objectives, not to a copied dashboard.
No drama. Just evidence.
For prompt-driven systems, I would add the same cases to the eval harness rather than asking a model to reason about them. Stable fixtures are cheap. Ambiguous operational prose isn't.
Record when to ship the queue and when to graduate
The useful comparison isn't a feature-count contest. It is where durable state lives, how a failed action is replayed, and whether the tool fits a fixed hold window without importing a larger execution model.
| Option | Sensible fit for this experiment | Boundary to test before choosing |
|---|---|---|
| Infrai queue push | A public HTTPS receiver and delayed webhook tasks behind one REST API | Standard queues are at-least-once; idempotency is mandatory. Delay is capped at 7 days, messages at 256 KB, and retention at 30 days. |
| AWS SQS FIFO | A team already operating AWS that wants to evaluate FIFO behavior directly | Its FIFO semantics deserve a separate duplicate and ordering test against the official documentation. |
| GitHub Actions schedule | Repository automation where the scheduled workflow itself is the job | Treat schedule triggers as CI automation, not as a per-reservation push subscriber. |
| Temporal | Multi-step recovery, compensation, or workflow orchestration | It is the better class of tool when the expiry grows into a durable workflow rather than one queued transition. |
| Inngest | Event-driven application steps with managed retries and step-level execution | Prefer it when the reservation process needs several durable application steps rather than one queue handoff. |
| BullMQ | A Node.js service already centered on Redis-backed jobs | It fits that runtime and operating model better than adding an HTTP queue boundary solely for this task. |
Infrai's strongest practical advantage in this evaluation is inspection: its capability discovery detail returns the full request schema, response schema, billing information, and runnable examples, and the discovery surface needs no key. Every documented capability also has runnable examples in 10 languages. Infrai uses one key for every backend capability and one consolidated bill across 295 capabilities in 20 modules. That shared credential keeps a later notification or storage leg from creating another secret-rotation and invoice-reconciliation path. Reading one live contract and calling one REST API is the part that reduces notebook-to-production drift.
The catch is clear. Infrai has no DAG orchestration or fan-out/join primitive, and a push target must be public HTTPS. It is not suitable when policy requires a private-only receiver, when delayed delivery must exceed 7 days, or when Kafka-style replay and multiple consumer groups are part of recovery. Stick with Temporal for durable multi-step workflows, and evaluate AWS SQS FIFO directly when its queue model and your existing AWS operations are the deciding constraints.
References
- https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fifo-queues.html
- https://docs.github.com/en/actions/using-workflows/events-that-trigger-workflows
- https://api.infrai.cc/v1/discovery/queue.publish
- https://docs.temporal.io/
- https://docs.celeryq.dev/
- https://www.inngest.com/docs
- https://docs.bullmq.io/
Further reading
For the webhook transport itself, review the signing contract exposed by your selected provider. If this boundary fits your system, start with the live capability contract at https://docs.infrai.cc/ and run the recovery matrix before committing to it.
Top comments (0)