Sistem chatbot stateful membutuhkan riwayat dialog sebelumnya untuk menentukan transisi state atau respons berbasis aturan (seperti pada state engine turunan ELIZA). Masalah muncul ketika pengguna mengirim pesan beruntun (burst messages) dalam interval milidetik. Jika antrean diproses oleh worker pool paralel, worker dapat mengeksekusi pesan di luar urutan (out-of-order) dan saling menimpa state di memori cache (lost update anomaly).

Anatomi Masalah: Pesan Konkuren dan Lost Update

Ketika dua pesan dari sesi yang sama (session_123) masuk secara bersamaan, dua worker berbeda mengambil masing-masing pesan secara paralel:

  1. Worker A membaca State Sesi v1 untuk Pesan 1 ("Halo, nama saya Budi").
  2. Worker B membaca State Sesi v1 untuk Pesan 2 ("Saya punya keluhan akun") sebelum Worker A selesai menulis.
  3. Worker B menyelesaikan pemrosesan dan menyimpan state v2_B (state berisi intent keluhan akun, tanpa nama Budi).
  4. Worker A menyelesaikan pemrosesan dan menimpa state dengan v2_A (state berisi nama Budi, intent keluhan akun hilang).

Dampaknya, histori percakapan korup dan respons dialog yang dikirimkan ke pengguna menjadi tidak koheren.

Komparasi: Sharded Partition Queue vs. Distributed Keyed Mutex

Dua pola arsitektur utama untuk menyelesaikan masalah konkurensi ini:

  • Sharded Partition Queue (Kafka / AWS Kinesis / RabbitMQ Consistent Hash): Pesan di-route ke partisi tertentu menggunakan hash dari session_id. Satu partisi hanya dikonsumsi oleh satu instance consumer secara serial.
    • Kelebihan: FIFO terjamin tanpa overhead penguncian (locking).
    • Kekurangan: Rawan head-of-line blocking jika satu pesan memakan waktu pemrosesan lama; rebalancing lambat saat autoscaling worker.
  • Distributed Keyed Mutex (Redis SETNX): Antrean worker tetap flat (misal via Redis Streams, BullMQ, atau RabbitMQ standar). Worker mengambil lock spesifik per session_id sebelum membaca dan memperbarui state.
    • Kelebihan: Distribusi beban kerja antar-worker merata, elastisitas horizontal instan tanpa penataan ulang partisi.
    • Kekurangan: Memerlukan mekanisme retry/backoff jika lock sedang ditahan worker lain.

Implementasi Keyed Mutex pada Worker (Node.js & Redis)

Pola implementasi aman menggunakan Redis SET resource_key token NX PX ttl untuk akuisisi, dan evaluasi Lua Script untuk pelepasan lock yang atomik (mencegah worker menghapus lock milik worker lain yang sudah kedaluwarsa).

import Redis from 'ioredis';
import { randomUUID } from 'crypto';

const redis = new Redis(process.env.REDIS_URL);

const RELEASE_LOCK_LUA = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])
else
  return 0
end
`;

interface SessionState {
  step: string;
  memory: Record<string, string>;
  history: string[];
}

export async function processDialogMessage(sessionId: string, userMessage: string): Promise<string> {
  const lockKey = `lock:session:${sessionId}`;
  const lockToken = randomUUID();
  const lockTtlMs = 5000;

  // 1. Acquire Keyed Mutex
  const acquired = await redis.set(lockKey, lockToken, 'PX', lockTtlMs, 'NX');
  if (!acquired) {
    // Kembalikan error agar antrean melakukan retry dengan backoff
    throw new Error(`CONCURRENCY_LOCK_ACTIVE: Session ${sessionId} is locked`);
  }

  try {
    // 2. Safe State Retrieval
    const stateKey = `session:${sessionId}:state`;
    const rawState = await redis.get(stateKey);
    const state: SessionState = rawState
      ? JSON.parse(rawState)
      : { step: 'ROOT', memory: {}, history: [] };

    // 3. ELIZA-like Rule Matching & State Mutation
    let reply = 'Maaf, saya tidak mengerti.';
    if (userMessage.match(/nama saya (.*)/i)) {
      const match = userMessage.match(/nama saya (.*)/i);
      const name = match ? match[1] : 'Kawan';
      state.memory.userName = name;
      state.step = 'AWAITING_QUERY';
      reply = `Halo ${name}, ada yang bisa dibantu?`;
    } else if (state.step === 'AWAITING_QUERY') {
      reply = `Mencatat keluhan Anda: "${userMessage}". Tim kami akan meninjau.`;
      state.step = 'COMPLETED';
    }

    state.history.push(`User: ${userMessage}`, `Bot: ${reply}`);

    // 4. Atomic Cache Update
    await redis.set(stateKey, JSON.stringify(state), 'EX', 86400);

    return reply;
  } finally {
    // 5. Safe Lock Release via Lua Script
    await redis.eval(RELEASE_LOCK_LUA, 1, lockKey, lockToken);
  }
}

Mitigasi Worker Crash dan Poison Pill

1. Worker Crash Saat Menahan Lock

Parameter PX 5000 memastikan lock otomatis hangus setelah 5 detik jika worker mengalami OOM (Out of Memory) atau termination mendadak. Aturan penentuan TTL: Lock TTL = (P99 Execution Time * 2) + Margin. Hindari penggunaan lock tanpa batas waktu kedaluwarsa.

2. Poison Pill ke Dead Letter Queue (DLQ)

Jika satu pesan menyebabkan parsing error fatal atau uncaught exception di dalam engine dialog, mekanisme retry antrean dapat menyebabkan pesan tersebut dieksekusi berulang kali dan memblokir sesi pengguna secara permanen.

  • Sisipkan metadata attempts pada message header.
  • Jika attempts > 3, gagalkan job, kirim payload ke antrean dialog-dlq, simpan notifikasi fallback ke pengguna, dan lepaskan lock sesi.
  • Worker DLQ dapat memproses logging terpusat tanpa menahan laju dialog reguler.

Skrip Verifikasi Konkurensi

Gunakan skrip eksekusi konkuren berikut untuk membuktikan tidak adanya lost update ketika pesan dikirim serentak:

import { processDialogMessage } from './worker';

async function runVerification() {
  const sessionId = 'test-session-concurrency';
  const messages = [
    'nama saya Budi',
    'saya ada kendala billing'
  ];

  console.log('Mengirim 2 pesan konkuren...');

  // Eksekusi paralel simultan
  const results = await Promise.allSettled(
    messages.map((msg) => processDialogMessage(sessionId, msg))
  );

  results.forEach((res, idx) => {
    if (res.status === 'fulfilled') {
      console.log(`Pesan [${idx}] Sukses: ${res.value}`);
    } else {
      console.log(`Pesan [${idx}] Ditolak/Ditahan (Lock Aktif): ${res.reason.message}`);
    }
  });
}

runVerification().catch(console.error);

Hasil eksekusi akan menunjukkan satu pesan sukses diproses sementara pesan lainnya ditolak dengan galat CONCURRENCY_LOCK_ACTIVE, siap dijadwalkan ulang oleh antrean tanpa merusak integritas state.