Streaming Bank Webhooks with Idempotent Consumers

Most bank and payment-processor webhooks carry no ordering guarantee and no exactly-once delivery promise: a single settlement event can arrive twice on two different TCP connections, arrive late after a provider-side retry storm, or arrive ahead of an earlier event for the same account. Treating the HTTP handler as the place where matching happens is what causes double-posted ledger entries. This page builds the pattern used on the Real-Time vs Batch Ingestion path: verify the signature, acknowledge in milliseconds, and push all correctness work — deduplication, ordering, and audit logging — into an idempotent consumer that runs behind a queue.

Prerequisites

The provider’s webhook secret (or JWKS endpoint for asymmetric signing) is available to the service at runtime, never hardcoded.
A shared Redis instance (or cluster) reachable from every consumer replica, used for the dedup key and — per Idempotency Key Storage with Redis TTL — for outbound idempotency keys as well.
Python 3.10+ with pydantic v2, an ASGI framework (FastAPI/Starlette-style), and redis.asyncio installed.
Monetary fields typed as Decimal; all timestamps timezone-aware UTC.
A structured logger configured to emit trace_id, source_hash, and match_decision on every consumer decision.
A downstream normalized-ledger-entry contract that the transaction matching engine can consume once an event is deduplicated.

Step 1 — Verify the webhook HMAC signature

Compute the signature over the raw request body, not a re-serialized copy — most providers sign the exact bytes they sent, and re-encoding JSON before verifying breaks the comparison. Reject anything outside max_skew_s of the provider’s timestamp header to shrink the window an attacker has to replay a captured payload.

python

import hmac
import hashlib
import time

class SignatureError(Exception):
    def __init__(self, code: str, detail: str) -> None:
        self.code = code
        super().__init__(detail)

def verify_signature(
    raw_body: bytes,
    signature_header: str,
    timestamp_header: str,
    secret: bytes,
    max_skew_s: int = 300,
) -> None:
    try:
        sent_at = int(timestamp_header)
    except ValueError as exc:
        raise SignatureError("BAD_SIGNATURE", "non-numeric timestamp header") from exc

    if abs(time.time() - sent_at) > max_skew_s:
        raise SignatureError("BAD_SIGNATURE", "timestamp outside max_skew_s")

    signed_payload = f"{timestamp_header}.".encode() + raw_body
    expected = hmac.new(secret, signed_payload, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, signature_header):
        raise SignatureError("BAD_SIGNATURE", "signature mismatch")

hmac.compare_digest is mandatory here — a naive == comparison on the hex digest leaks timing information that lets an attacker brute-force the signature byte by byte.

Step 2 — Respond fast and hand off asynchronously

The endpoint’s only job is to authenticate the request and get it off the wire. Any matching, deduplication, or ledger work performed inline turns the provider’s retry timeout into your outage: a slow database call at 2 a.m. makes the bank’s webhook dispatcher believe delivery failed, triggering a retry storm on top of an already-degraded system. Enqueue and return 202 before doing anything else.

python

import asyncio
from fastapi import FastAPI, Request, Response
from starlette.responses import JSONResponse

app = FastAPI()
inbound_queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=2000)

@app.post("/webhooks/bank-feed")
async def receive_webhook(request: Request) -> Response:
    raw_body = await request.body()
    try:
        verify_signature(
            raw_body,
            request.headers.get("X-Signature", ""),
            request.headers.get("X-Timestamp", ""),
            secret=b"replace-with-secret-manager-value",
        )
    except SignatureError as exc:
        return JSONResponse({"error": exc.code}, status_code=401)

    envelope = {"raw": raw_body, "received_at": time.time()}
    try:
        await asyncio.wait_for(inbound_queue.put(envelope), timeout=0.2)
    except asyncio.TimeoutError:
        # Queue is saturated; fail fast so the provider's own retry backoff kicks in.
        return JSONResponse({"error": "SLOW_ACK"}, status_code=503)

    return JSONResponse({"status": "accepted"}, status_code=202)

Returning 503 on a full queue is deliberate: it is cheaper to let the provider’s retry policy absorb a burst than to block the event loop waiting for queue space, which would stall every other in-flight request on the same worker.

Step 3 — Consume with idempotent dedup on `source_hash`

The consumer runs independently of the HTTP worker, pulling envelopes off the queue and computing a stable source_hash from the event’s business fields — not the raw bytes, since a provider’s retry envelope can differ byte-for-byte (different request id, different wrapper timestamp) while representing the same underlying settlement. SETNX (via redis.set(..., nx=True)) atomically claims the hash; a second attempt within the TTL window returns False and the event is dropped as a duplicate.

python

import json
from decimal import Decimal
from datetime import datetime, timezone
from uuid import uuid4
import logging
import redis.asyncio as redis

log = logging.getLogger("webhooks.consumer")
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def compute_source_hash(event: dict) -> str:
    canonical = json.dumps(
        {
            "event_id": event["event_id"],
            "account_id": event["account_id"],
            "amount": str(Decimal(event["amount"])),
            "posted_at": event["posted_at"],
        },
        sort_keys=True,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

async def claim_event(source_hash: str, trace_id: str, dedup_ttl: int = 86_400) -> bool:
    claimed = await r.set(f"webhook:dedup:{source_hash}", trace_id, nx=True, ex=dedup_ttl)
    return bool(claimed)

async def consume(queue: asyncio.Queue[dict]) -> None:
    while True:
        envelope = await queue.get()
        event = json.loads(envelope["raw"])
        trace_id = str(uuid4())
        source_hash = compute_source_hash(event)

        if not await claim_event(source_hash, trace_id):
            log.info(
                "webhook.duplicate",
                extra={
                    "trace_id": trace_id,
                    "source_hash": source_hash,
                    "match_decision": "DUPLICATE",
                },
            )
            queue.task_done()
            continue

        await handle_new_event(event, trace_id, source_hash)
        queue.task_done()

Step 4 — Handle out-of-order and duplicate delivery

Webhook dispatchers fan requests across multiple connections, so ordering per account is not guaranteed even when the provider’s own event log is ordered. If the payload carries a per-account sequence_no, buffer briefly and flush in order; if it doesn’t, accept that ledger entries land unordered and let the transaction matching engine’s date-window tolerance rules reconcile posting order downstream rather than blocking ingestion on it.

python

from collections import defaultdict
import heapq

class ReorderBuffer:
    def __init__(self, window_s: float = 2.0) -> None:
        self._window_s = window_s
        self._heaps: dict[str, list[tuple[int, dict]]] = defaultdict(list)
        self._deadlines: dict[str, float] = {}

    def add(self, account_id: str, sequence_no: int, event: dict) -> list[dict]:
        heap = self._heaps[account_id]
        heapq.heappush(heap, (sequence_no, event))
        self._deadlines.setdefault(account_id, time.monotonic() + self._window_s)

        ready: list[dict] = []
        if time.monotonic() >= self._deadlines[account_id]:
            while heap:
                ready.append(heapq.heappop(heap)[1])
            del self._deadlines[account_id]
        return ready

This buffer trades a small, bounded latency (window_s) for in-order delivery per account. Events that never resolve their sequence gap within the window are flushed anyway and logged with a match_decision of OUT_OF_ORDER_FLUSHED so the audit trail records that ordering was not guaranteed for that batch.

Step 5 — Handle the replay window explicitly

dedup_ttl is the only thing separating a legitimate resend after an outage from a suppressed replay: while the Redis key lives, a repeat source_hash is dropped as REPLAYED_EVENT; once it expires, the same event is treated as new. Set dedup_ttl comfortably longer than the provider’s documented maximum retry window (check their webhook reliability docs — many retry for 24–72 hours), and rely on the transaction matching engine’s exact-match hash comparison as a second line of defense for anything that slips past an expired dedup key.

python

async def handle_new_event(event: dict, trace_id: str, source_hash: str) -> None:
    normalized = {
        "account_id": event["account_id"],
        "amount": Decimal(event["amount"]).quantize(Decimal("0.01")),
        "posted_at": datetime.fromisoformat(event["posted_at"]).astimezone(timezone.utc),
        "source_hash": source_hash,
    }
    # Hand off to the matching engine's ingest queue here.
    log.info(
        "webhook.accepted",
        extra={
            "trace_id": trace_id,
            "source_hash": source_hash,
            "match_decision": "NEW",
            "account_id": normalized["account_id"],
            "amount": str(normalized["amount"]),
        },
    )

Step 6 — Emit one structured audit record per decision

Every branch — accepted, duplicate, out-of-order, or rejected at the signature check — should produce exactly one log line carrying trace_id, source_hash, and match_decision. That triple is what lets an auditor reconstruct, months later, why a specific inbound event was or was not posted to the ledger, without needing access to Redis state that has since expired.

python

def audit(trace_id: str, source_hash: str | None, match_decision: str, **extra: object) -> None:
    log.info(
        "webhook.audit",
        extra={
            "trace_id": trace_id,
            "source_hash": source_hash,
            "match_decision": match_decision,
            "logged_at": datetime.now(timezone.utc).isoformat(),
            **extra,
        },
    )

Configuration boundary table

Parameter	Default	Valid range	Notes
`hmac_alg`	`sha256`	`sha256` / `sha512`	Must match the provider’s documented signing algorithm exactly
`max_skew_s`	`300`	`30`–`900`	Reject webhooks whose timestamp header falls outside this window
`dedup_ttl`	`86400`	`3600`–`604800`	Redis TTL on the `source_hash` `SETNX` key; defines the replay window
`queue_maxsize`	`2000`	`100`–`20000`	Backpressure ceiling before the endpoint returns `503`
`ack_timeout_ms`	`200`	`50`–`2000`	Max time the endpoint blocks on a queue put before failing fast

Verification and testing

Use a fake Redis client (or fakeredis.aioredis) so the dedup test runs without a live dependency, and assert that a replayed webhook is suppressed on the second delivery while the first produces a NEW decision.

python

import pytest

class FakeRedis:
    def __init__(self) -> None:
        self._store: dict[str, str] = {}

    async def set(self, key: str, value: str, nx: bool = False, ex: int | None = None) -> bool:
        if nx and key in self._store:
            return False
        self._store[key] = value
        return True

@pytest.mark.asyncio
async def test_replayed_webhook_is_suppressed() -> None:
    fake = FakeRedis()
    event = {
        "event_id": "evt_9f2c",
        "account_id": "ACC-1042",
        "amount": "1250.005",
        "posted_at": "2026-07-15T14:03:00+00:00",
    }
    source_hash = compute_source_hash(event)

    first = await fake.set(f"webhook:dedup:{source_hash}", "trace-1", nx=True, ex=86_400)
    second = await fake.set(f"webhook:dedup:{source_hash}", "trace-2", nx=True, ex=86_400)

    assert first is True    # first delivery: NEW
    assert second is False  # replayed delivery: suppressed

Run the consumer under a soak test that replays a captured provider payload at 5x normal traffic to confirm queue_maxsize and ack_timeout_ms shed load before the process falls behind, and alert when p95_ack_latency_ms > 200 or dedup_hit_rate moves outside its historical baseline — a sudden spike usually means a provider-side retry storm, not an attack.

Troubleshooting

BAD_SIGNATURE — every webhook from a provider suddenly fails verification. Root cause: usually a rotated signing secret that hasn’t propagated to the service, or middleware that re-parses and re-serializes the body before verify_signature runs, changing the byte sequence. Fix: verify against the untouched raw body captured at the top of the handler, and confirm the active secret matches the provider’s current key.
REPLAYED_EVENT — the same source_hash is claimed and rejected repeatedly. Root cause: this is often correct behavior — the provider retries because it never saw your 202 in time. Fix: confirm the endpoint returns 202 well under the provider’s ack timeout; if it does and replays persist, check for a proxy or load balancer swallowing the response.
OUT_OF_ORDER — later events post before earlier ones for the same account. Root cause: no sequence_no in the payload, or window_s on the ReorderBuffer set too low for the provider’s real jitter. Fix: widen window_s incrementally while watching p95_ack_latency_ms, or accept unordered ingestion and rely on downstream date-window tolerance matching.
SLOW_ACK — the endpoint returns 503 under normal load. Root cause: queue_maxsize is undersized for the consumer’s throughput, or a slow consumer (blocking I/O inside handle_new_event) is draining the queue too slowly. Fix: raise queue_maxsize, scale consumer replicas horizontally, or profile handle_new_event for blocking calls that should be async.
NAIVE_TIMESTAMP — TypeError when comparing posted_at to a UTC deadline downstream. Root cause: datetime.fromisoformat on a payload missing an explicit offset produces a naive datetime. Fix: validate and coerce to UTC at the ingestion boundary in Step 5, before the event ever reaches the matching engine.

Part of Real-Time vs Batch Ingestion within Core Architecture & Bank Feed Ingestion.