Idempotency-Key Storage with Redis TTL

Most bank feed connectors guarantee at-least-once delivery, not exactly-once — a webhook retry after a timed-out acknowledgment, a batch file reprocessed after a crashed worker, or a polling client that overlaps its own window will all resend a transaction the pipeline already ingested. The fix is to store the canonical source_hash for every accepted record in a fast key-value store keyed by that hash, and to reject anything that arrives a second time before its retention window closes. Redis is a natural fit: SET key value NX EX <ttl> is atomic, sub-millisecond, and gives you set-if-absent semantics in a single round trip. This page implements that dedup layer end to end — hash computation, the atomic insert, TTL alignment to a statutory retention window, and a fail-safe for when Redis itself is unavailable. It sits under Secure API Token Management because the same connector credentials that authenticate the feed also gate which records are trusted enough to be deduplicated against, and it pairs with Streaming Bank Webhooks with Idempotent Consumers, which covers the consumer loop that calls into this store on every message.

Prerequisites

Step 1 — Compute the canonical source hash

The hash must be stable across retries of the same underlying event but sensitive to any change in the fields that make a transaction unique. Hash the normalized payload, not the raw bytes — a webhook retry often re-serializes JSON key order or re-wraps whitespace.

python
import hashlib
import json
from decimal import Decimal
from typing import Any

def source_hash(payload: dict[str, Any]) -> str:
    canonical = {
        "account_id": payload["account_id"],
        "bank_txn_id": payload["bank_txn_id"],
        "amount": str(Decimal(payload["amount"])),   # Decimal, never float
        "currency": payload["currency"],
        "posted_at": payload["posted_at"],            # ISO-8601 UTC string
    }
    blob = json.dumps(canonical, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

Sorting keys and fixing separators removes serialization noise. Casting amount through Decimal before hashing means "100.50" and "100.5000" collapse to the same hash instead of silently producing two dedup entries for one transaction.

Step 2 — Insert atomically with SET NX EX

The insert and the existence check must happen as one operation, or two workers racing on the same retry can both observe “not present” and both ingest. Redis’s SET key value NX EX ttl sets the key only if it is absent and applies the expiry in the same atomic command — no separate EXISTS call, no lock.

python
import redis
from datetime import datetime, timezone

class IdempotencyStore:
    def __init__(self, client: redis.Redis, key_prefix: str = "dedup:bankfeed:",
                 ttl_seconds: int = 2557 * 86400) -> None:
        self._r = client
        self._prefix = key_prefix
        self._ttl = ttl_seconds

    def _key(self, h: str) -> str:
        return f"{self._prefix}{h}"

    def try_insert(self, h: str, trace_id: str) -> bool:
        """Returns True if h is novel (insert succeeded), False if it is a duplicate."""
        marker = f"{trace_id}|{datetime.now(timezone.utc).isoformat()}"
        return bool(self._r.set(self._key(h), marker, nx=True, ex=self._ttl))

redis-py’s set(..., nx=True, ex=ttl) maps directly onto SET key value NX EX ttl — one network round trip, atomic on the server, no client-side race window between check and write.

Step 3 — Interpret the result and route the record

A True return from try_insert means the hash was novel and the caller should proceed with ingestion; False means another worker (or a prior attempt) already claimed that hash, so the record is a confirmed duplicate and must be suppressed rather than re-posted to the ledger.

python
import logging

log = logging.getLogger("ingestion.dedup")

def process_bank_event(store: IdempotencyStore, payload: dict, trace_id: str) -> str:
    h = source_hash(payload)
    novel = store.try_insert(h, trace_id)
    decision = "ACCEPTED" if novel else "DUPLICATE_SUPPRESSED"
    log.info(
        "dedup.checked",
        extra={"trace_id": trace_id, "source_hash": h, "match_decision": decision},
    )
    if not novel:
        return decision
    # ingest_transaction(payload) would run here
    return decision

Every decision — accepted or suppressed — is logged with the same trace_id/source_hash/match_decision triple, so an auditor can reconstruct exactly which retry was rejected and why, without needing to replay the Redis state itself.

Step 4 — Align the TTL to the retention window, not the feed’s retry window

Setting the TTL too short is the single most common cause of silent double-ingestion: once a key expires, a late retry looks novel again. The TTL should track the longest period a duplicate could plausibly arrive — which for regulated bank feeds is usually the same statutory retention period applied to the underlying financial records, not the connector’s short-lived retry budget.

python
from dataclasses import dataclass

@dataclass(frozen=True)
class RetentionPolicy:
    statutory_years: int = 7

    @property
    def ttl_seconds(self) -> int:
        # 7 years -> 2557 days, accounting for leap years at ~365.25/yr
        return int(self.statutory_years * 365.25 * 86400)

policy = RetentionPolicy(statutory_years=7)
store = IdempotencyStore(client=redis.Redis(host="localhost", port=6379, db=0),
                          ttl_seconds=policy.ttl_seconds)

A 7-year policy resolves to roughly 2557 days (221,126,400 seconds). Keeping the TTL derived from a named policy object — rather than a bare integer scattered through the codebase — makes it auditable: a reviewer can see the retention rule that produced the number stored in Redis.

Step 5 — Fail safe when Redis is unavailable

A dedup store that is down must never silently let duplicates through, and it must never block ingestion indefinitely either. A circuit breaker trips after repeated connection failures, and the configured fail_mode decides whether the pipeline halts (closed, safest for money movement) or degrades to a secondary check (open, for lower-risk feeds).

python
import time
from redis.exceptions import RedisError

class DedupUnavailable(Exception):
    pass

class BreakerState:
    def __init__(self, threshold: int = 5, reset_after_s: float = 30.0) -> None:
        self.threshold = threshold
        self.reset_after_s = reset_after_s
        self._failures = 0
        self._opened_at: float | None = None

    def record_failure(self) -> None:
        self._failures += 1
        if self._failures >= self.threshold:
            self._opened_at = time.monotonic()

    def record_success(self) -> None:
        self._failures = 0
        self._opened_at = None

    def is_tripped(self) -> bool:
        if self._opened_at is None:
            return False
        if time.monotonic() - self._opened_at > self.reset_after_s:
            self._opened_at = None   # half-open: allow a probe
            return False
        return True

def guarded_try_insert(store: IdempotencyStore, breaker: BreakerState, h: str,
                        trace_id: str, fail_mode: str = "closed") -> bool:
    if breaker.is_tripped():
        log.info("dedup.checked", extra={
            "trace_id": trace_id, "source_hash": h, "match_decision": "REDIS_UNAVAILABLE",
        })
        if fail_mode == "closed":
            raise DedupUnavailable("breaker open; refusing to ingest without dedup guarantee")
        return True  # fail_mode == "open": treat as novel, accept the small duplicate risk
    try:
        result = store.try_insert(h, trace_id)
        breaker.record_success()
        return result
    except RedisError:
        breaker.record_failure()
        if fail_mode == "closed":
            raise DedupUnavailable("Redis error; refusing to ingest without dedup guarantee")
        return True

For bank-feed ingestion the correct default is fail_mode="closed": it is far cheaper to pause ingestion and drain a backlog than to double-post a settled transaction and have to reverse it in the ledger later.

Configuration boundary table

Parameter Default Valid range Notes
dedup_ttl_days 2557 30–3653 Aligned to the statutory retention window (7 yr default)
key_prefix dedup:bankfeed: non-empty string Namespaces keys per feed to avoid cross-tenant collisions
lock_timeout_ms 50 10–500 Client socket timeout for the SET NX call itself
breaker_threshold 5 1–20 Consecutive Redis failures before the breaker trips
fail_mode closed closed | open closed halts ingestion; open accepts duplicate risk

Verification and testing

Use fakeredis or a disposable Redis container so the test is deterministic and doesn’t depend on network state. The fixture asserts that a second insert of the identical hash is suppressed, and that a genuinely different payload is accepted.

python
import fakeredis

def test_second_insert_is_suppressed():
    client = fakeredis.FakeRedis()
    store = IdempotencyStore(client=client, ttl_seconds=86400)

    payload = {
        "account_id": "ACC-01", "bank_txn_id": "TXN-9981",
        "amount": "1250.00", "currency": "USD",
        "posted_at": "2026-07-15T09:03:00+00:00",
    }
    h = source_hash(payload)

    first = process_bank_event(store, payload, trace_id="trace-a")
    assert first == "ACCEPTED"

    second = process_bank_event(store, payload, trace_id="trace-b")  # retry, same payload
    assert second == "DUPLICATE_SUPPRESSED"

    # A materially different transaction on the same account must not collide.
    other = dict(payload, bank_txn_id="TXN-9982", amount="75.10")
    assert process_bank_event(store, other, trace_id="trace-c") == "ACCEPTED"

    assert client.ttl(store._key(h)) > 0

Run this fixture in CI on every change to source_hash or IdempotencyStore — a regression here is invisible in normal operation until a retry storm hits production and duplicates start appearing in the ledger days later.

Troubleshooting

  • DUPLICATE_SUPPRESSED appearing for records that are actually distinct. Root cause: source_hash is derived from too few fields, or amount is hashed as a raw string instead of a normalized Decimal, so two different amounts formatted identically collide. Fix: widen the canonical field set in Step 1 and always route amount through Decimal before serializing.
  • TTL_TOO_SHORT — the same transaction re-ingests weeks later. Root cause: dedup_ttl_days was set to a short operational value (e.g. 7 days) instead of the statutory retention window. Fix: derive the TTL from a RetentionPolicy object as in Step 4, not a magic number tuned for a different purpose.
  • REDIS_UNAVAILABLE — ingestion stalls entirely during a Redis outage. Root cause: fail_mode="closed" correctly halts ingestion, but no alert fired and the backlog grew unbounded. Fix: pair the breaker with a paging alert on breaker.is_tripped() and monitor queue depth so the outage is triaged before the backlog becomes unmanageable.
  • HASH_COLLISION — two genuinely distinct transactions share a source_hash. Root cause: SHA-256 collisions are not the practical risk; a truncated or non-canonical hash input is. Fix: verify the canonical payload includes bank_txn_id (the bank’s own unique identifier) alongside amount and date, not amount/date alone.
  • Duplicate keys surviving past their TTL after a Redis failover. Root cause: a replica promoted without AOF/RDB persistence enabled lost the expiry metadata along with the keyspace. Fix: enable persistence (appendonly yes) on the Redis deployment backing this store, or accept the small re-ingestion window and rely on the ledger’s own uniqueness constraint as a second line of defense.

Part of Secure API Token Management within Core Architecture & Bank Feed Ingestion.