prompt
RabbitMQ Backlog Diagnosis and Retry Strategy Review
Turn RabbitMQ queue growth into an evidence-based containment and retry plan without creating a retry storm.
Engineering guide
Design webhook endpoints that tolerate duplicate delivery, retries, slow downstream work, and out-of-order events.
Act as a senior backend engineer reviewing a FastAPI webhook integration. Return: (1) the trust boundary and raw-body signature verification flow, (2) a database inbox schema with a unique provider event ID, processing status, attempt count, timestamps, and last error, (3) the transaction boundary for durable acceptance, (4) the HTTP acknowledgement rule, (5) an asynchronous processing and retry policy with exponential backoff and a terminal dead-letter state, (6) rules for duplicate and out-of-order events, (7) structured logs and metrics, and (8) pytest cases for duplicate delivery, reverse ordering, timeout, and worker crash recovery. Prefer PostgreSQL constraints over in-memory deduplication. Webhook context: [Provider, signature scheme, event examples, current FastAPI route, database, queue, downstream side effects, and provider timeout]
Treat the provider event ID as an idempotency key and enforce uniqueness in the database rather than relying on an in-memory cache.
Validate the signature, store a durable inbox record, and return success. Move slow work to a worker so the provider does not retry after a timeout.
Store provider timestamps or sequence numbers and make stale events no-ops instead of allowing older data to overwrite newer state.
Replay identical payloads, reverse event order, stop the worker after commit, and simulate downstream timeouts. Verify one durable business effect.
This local example uses SQLite so it runs immediately. Preserve the unique event ID constraint in PostgreSQL for production.
import hashlib
import hmac
import json
import os
import sqlite3
from fastapi import FastAPI, Header, HTTPException, Request, status
app = FastAPI()
secret = os.environ.get("WEBHOOK_SECRET", "local-secret").encode()
db = sqlite3.connect(os.environ.get("WEBHOOK_DB", ":memory:"), check_same_thread=False)
db.execute("""
CREATE TABLE IF NOT EXISTS webhook_inbox (
event_id TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
payload TEXT NOT NULL,
state TEXT NOT NULL DEFAULT 'accepted',
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
accepted_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""")
def valid_signature(body: bytes, supplied: str) -> bool:
expected = hmac.new(secret, body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, supplied)
@app.post("/webhooks/provider", status_code=status.HTTP_202_ACCEPTED)
async def receive_webhook(request: Request, x_signature: str = Header(default="")):
body = await request.body()
if not valid_signature(body, x_signature):
raise HTTPException(status_code=401, detail="Invalid signature")
event = json.loads(body)
cursor = db.execute(
"INSERT OR IGNORE INTO webhook_inbox(event_id,event_type,payload) VALUES(?,?,?)",
(str(event["id"]), str(event["type"]), body.decode()),
)
db.commit()
return {"accepted": True, "duplicate": cursor.rowcount == 0}import hashlib
import hmac
import json
from fastapi.testclient import TestClient
from main import app, secret
client = TestClient(app)
def signed_post(event):
body = json.dumps(event, separators=(",", ":")).encode()
signature = hmac.new(secret, body, hashlib.sha256).hexdigest()
return client.post("/webhooks/provider", content=body, headers={"x-signature": signature})
def test_duplicate_delivery():
event = {"id": "evt_duplicate_1", "type": "order.paid"}
assert signed_post(event).json()["duplicate"] is False
assert signed_post(event).json()["duplicate"] is True
def test_invalid_signature():
response = client.post("/webhooks/provider", json={"id": "evt_bad", "type": "order.paid"}, headers={"x-signature": "bad"})
assert response.status_code == 401Reusable resources
prompt
Turn RabbitMQ queue growth into an evidence-based containment and retry plan without creating a retry storm.