Idempotency ও Exactly-once-এর মিথ

"""idempotent_endpoint.py — exactly-once EFFECT wrapper (~80 line)"""
import random, threading, time
from collections import defaultdict


class IdempotentEndpoint:
    def __init__(self, ttl_s=0.05):
        self.ttl = ttl_s
        self.mu = threading.Lock()
        self.results = {}      # key -> (expires_at, value)
        self.claims = set()    # claimed-but-running keys
        self.waiters = defaultdict(list)

    def _expired(self, key, now):
        rec = self.results.get(key)
        return rec is not None and now >= rec[0]

    def handle(self, key, effect):
        """returns (verdict, value) — effect ঠিক একবারই চলে"""
        while True:
            now = time.monotonic()
            with self.mu:
                if self._expired(key, now):
                    del self.results[key]           # TTL eviction
                if key in self.results:
                    return ('replayed', self.results[key][1])
                if key in self.claims:
                    evt = threading.Event()
                    self.waiters[key].append(evt)   # in-flight lock
                else:
                    self.claims.add(key)            # atomic claim!
                    break                            # we execute
            evt.wait(2)                              # outside lock
        try:
            value = effect()                         # ★ single effect ★
            now = time.monotonic()
            with self.mu:
                self.results[key] = (now + self.ttl, value)
                self.claims.discard(key)
                for w in self.waiters.pop(key, []):
                    w.set()
            return ('executed', value)
        except Exception:
            with self.mu:
                self.claims.discard(key)             # retry allowed
                for w in self.waiters.pop(key, []):
                    w.set()
            raise


# ── fuzz test: random clients, flaky effect, duplicate storms ──
def fuzz(trials=300, seed=99):
    import concurrent.futures as cf
    rng = random.Random(seed)
    ep = IdempotentEndpoint(ttl_s=60)                 # TTL > test-window
    committed = defaultdict(int)                      # key -> real effects

    def effect(key):
        if rng.random() > 0.8:                        # 20% provider-fail:
            raise RuntimeError('payment hiccup')      # কিছুই commit হয়নি
        committed[key] += 1                           # ★ committed effect ★
        return f'result-{key}'

    for i in range(trials):                           # retry-storm phase
        key = f'op-{rng.randint(0, 24)}'
        deadline = time.monotonic() + 0.5
        while True:
            try:
                ep.handle(key, lambda k=key: effect(k))
                break                                 # success/replay
            except RuntimeError:
                if time.monotonic() >= deadline:
                    break                             # gave up

    with cf.ThreadPoolExecutor(8) as ex:              # duplicate storm!
        futs = [ex.submit(ep.handle, 'op-storm',
                          lambda: effect('op-storm'))
                for _ in range(16)]
        verdicts = [f.result()[0] for f in futs]

    bad = {k: c for k, c in committed.items() if c != 1}
    assert not bad, f'exactly-once ভাঙল: {bad}'
    assert committed['op-storm'] == 1
    executed = sum(1 for v in verdicts if v == 'executed')
    print(f'{trials} trials + 16-thread storm on 1 key')
    print(f'storm verdicts: {executed} executed, '
          f'{16 - executed} replayed/in-flight-wait')
    print('PASS — প্রতিটা key-র effect ঠিক একবার')


fuzz()

নমুনা output:

300 trials + 16-thread storm on 1 key
storm verdicts: 1 executed, 15 replayed/in-flight-wait
PASS — প্রতিটা key-র effect ঠিক একবার

তিনটা পাঠ:

• claims.add আর results-check একই lock-এর ভেতরে — atomicity- টা এখানেই; আলাদা হলে race window ফিরে আসে।
• Effect-fail হলে claim ছেড়ে দেওয়া হয় — নইলে poison-request key-টা চিরতরে বন্দি করে ফেলত।
• Fuzz-এ effects[key] != 1 কখনো ধরা না পড়লেও confidence আসে thread-interleaving থেকে — production-এ এই assertion-টাই dedup-monitoring-এর ভিত্তি।

নিজে বাড়ান:

1. handle-কে multi-node বানান — claims-কে Redis SET NX PX-এ সরিয়ে নিয়ে দুই process-এর duplicate-execution আটকান
2. Response-body বড় হলে store-এ pointer (S3-key) রাখার pattern যোগ করুন — Stripe-এর কাছাকাছি
3. TTL-expire-পরবর্তী retry-এ ‘stale-key’ warning metric বাড়ান