Lonjakan beban atau eksekusi jadwal serentak pada sistem worker pembayaran multi-akun sering memicu kondisi thundering herd. Dalam otomasi pembayaran (seperti integrasi agregator pihak ketiga bergaya protokol aBaiAutoplus), ratusan worker paralel dapat berebut memproses akun yang sama. Akibatnya terjadi tabrakan pembaruan token otentikasi (race condition token refresh), pelanggaran rate limit API billing eksternal, hingga eksekusi pembayaran ganda (double disbursement).

Akar Masalah: Concurrency Conflict pada Multi-Akun

Saat sistem menjalankan pemrosesan transaksi massal, tiga titik kegagalan utama kerap muncul:

  • Token Refresh Stampede: Akses serentak oleh worker ke satu sub-akun saat token OAuth/session kadaluarsa menyebabkan beberapa worker mengirim permintaan penyegaran token bersamaan. Token lama langsung diinvalidsi oleh server billing eksternal sebelum worker lain selesai menggunakannya.
  • Batas Kuota Agregator: Eksternal gateway membatasi transaksi paralel per identitas akun. Jika konkurensi tidak dibatasi per-akun, gateway mengembalikan galat HTTP 429 atau memutus koneksi TLS.
  • Partial Batch Failure: Pemrosesan pembayaran batch tanpa isolasi status atomik membuat status transaksi gantung saat salah satu item gagal atau worker mati di tengah jalan.

Strategi Solusi: Lock, TTL Renewal, dan Message Deduplication

Untuk mengatasi masalah di atas, arsitektur worker membutuhkan tiga lapisan pelindung:

  1. Distributed Lock granular: Lock berbasis ID akun spesifik menggunakan perintah Redis SET resource_key my_random_token NX PX ttl.
  2. Auto-Renewal TTL (Heartbeat/Watchdog): Mencegah rilis lock prematur saat pemrosesan batch API pihak ketiga memakan waktu lebih lama dari perkiraan.
  3. Queue Deduplication: Memblokir antrean ganda terhadap akun atau batch transaksi yang identik sebelum masuk ke broker.

Implementasi Distributed Lock dengan Auto-Renewal TTL

Kode di bawah mengimplementasikan context manager Redis distributed lock di Python dengan thread terpisah untuk memperpanjang TTL selama operasi I/O pembayaran berjalan:

import time
import uuid
import threading
import redis

class RedisDistributedLock:
    def __init__(self, client: redis.Redis, lock_key: str, expire_ms: int = 10000):
        self.client = client
        self.lock_key = lock_key
        self.expire_ms = expire_ms
        self.token = str(uuid.uuid4())
        self._running = False
        self._renew_thread = None

    def __enter__(self):
        acquired = self.client.set(self.lock_key, self.token, nx=True, px=self.expire_ms)
        if not acquired:
            raise BlockingIOError(f"Gagal memperoleh lock untuk: {self.lock_key}")
        
        self._running = True
        self._renew_thread = threading.Thread(target=self._auto_renew_loop, daemon=True)
        self._renew_thread.start()
        return self

    def _auto_renew_loop(self):
        interval = (self.expire_ms / 1000.0) / 2
        renew_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("pexpire", KEYS[1], ARGV[2])
        else
            return 0
        end
        """
        while self._running:
            time.sleep(interval)
            if self._running:
                res = self.client.eval(renew_script, 1, self.lock_key, self.token, self.expire_ms)
                if res == 0:
                    break

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._running = False
        if self._renew_thread and self._renew_thread.is_alive():
            self._renew_thread.join(timeout=1.0)
            
        release_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        self.client.eval(release_script, 1, self.lock_key, self.token)

Implementasi Task Celery dan Penanganan Partial Failure

Integrasikan lock di atas ke dalam task Celery. Deduplikasi task dilakukan di level ID akun sebelum transaksi diproses, dan status setiap transaksi batch dicatat individual ke database/cache state sync.

from celery import Celery
import redis
from .locks import RedisDistributedLock

app = Celery("payment_processor", broker="redis://localhost:6379/0")
redis_client = redis.Redis.from_url("redis://localhost:6379/1")

def execute_external_payment(account_id: str, item: dict) -> bool:
    # Simulasi interaksi ke payment gateway/aBaiAutoplus
    return True

@app.task(bind=True, max_retries=3, default_retry_delay=5)
def process_account_batch(self, account_id: str, batch_items: list):
    lock_key = f"lock:payment_account:{account_id}"
    results = {}

    try:
        with RedisDistributedLock(redis_client, lock_key, expire_ms=8000):
            for item in batch_items:
                item_id = item["transaction_id"]
                state_key = f"tx_state:{account_id}:{item_id}"
                
                # Cek idempotensi: abaikan item yang telah berstatus SUCCEEDED
                if redis_client.get(state_key) == b"SUCCESS":
                    results[item_id] = "SKIPPED_ALREADY_PAID"
                    continue

                try:
                    is_ok = execute_external_payment(account_id, item)
                    if is_ok:
                        redis_client.set(state_key, "SUCCESS", ex=86400)
                        results[item_id] = "SUCCESS"
                    else:
                        redis_client.set(state_key, "FAILED_GATEWAY_REJECT", ex=86400)
                        results[item_id] = "FAILED_GATEWAY_REJECT"
                except Exception as api_err:
                    redis_client.set(state_key, "FAILED_EXCEPTION", ex=86400)
                    results[item_id] = f"ERROR: {str(api_err)}"

    except BlockingIOError as lock_err:
        # Worker lain sedang mengoperasikan akun ini, re-queue task
        raise self.retry(exc=lock_err)

    return results

Trade-offs dan Pertimbangan Sistem

Penerapan distributed lock dan state sync memiliki sejumlah konsekuensi teknis:

  • Throughput vs Konsistensi: Serialisasi per-akun menurunkan total konkurensi pada akun yang sama, namun mencegah pemblokiran gateway (HTTP 429) dan duplikasi penarikan dana. Throughput horizontal tetap berjalan antar-akun yang berbeda.
  • Kompleksitas Watchdog Thread: Penggunaan daemon thread untuk perpanjangan TTL menambah penggunaan memori worker. Jika worker mengalami crash hard-kill (SIGKILL), lock bertahan paling lama sebesar TTL awal (misal: 10 detik) sebelum terlepas secara otomatis.
  • Idempotensi Agregator: Distributed lock di sisi klien tidak menghilangkan kewajiban menyertakan Idempotency-Key unik pada setiap payload ke API gateway eksternal.