Pola Request-Reply (RPC over Message Broker) kerap dipilih saat sistem terdistribusi membutuhkan komunikasi asinkron antar-servis yang tetap mempertahankan semantik pertukaran data dua arah. Berbeda dari komunikasi titik-ke-titik via HTTP atau gRPC, abstraksi antrean (queue) memutus keterikatan langsung antara pengirim dan pemroses pesan. Namun, fleksibilitas ini menimbulkan dua persoalan operasional: reply queue leak (penumpukan antrean respons yatim piatu) dan worker deadlock (kegagalan pengiriman balasan yang membuat klien menggantung tanpa kepastian batas waktu).

Inspirasi Arsitektur: Exec MsgPort AmigaOS 2 ke Distributed RPC

Masalah pengiriman pesan asinkron dua arah bukanlah hal baru. Pada AmigaOS 2, subsistem kernel Exec mengelola komunikasi antar-proses (IPC) menggunakan struktur MsgPort dan Message secara efisien. Dalam model Exec AmigaOS:

  1. Task pengirim membuat port privat (CreateMsgPort()) untuk menampung balasan.
  2. Pesan disiapkan dengan field mn_ReplyPort yang mereferensikan port privat tersebut, lalu dikirim via PutMsg() ke port target.
  3. Task penerima memproses pesan, lalu wajib memanggil ReplyMsg(). Fungsi ini mengembalikan pointer pesan langsung ke mn_ReplyPort pemanggil tanpa perlu mengetahui identitas task pengirim.

Dalam sistem terdistribusi modern berbasis broker (seperti RabbitMQ atau broker kompatibel AMQP 0-9-1), field mn_ReplyPort bermutasi menjadi header reply_to, dan pointer identitas pesan menjadi correlation_id. Ketika kontrak dasar ini dilanggar—misalnya worker crash sebelum memanggil ekuivalen ReplyMsg(), atau reply port tidak pernah dihapus saat klien mati—sistem terdistribusi mengalami kegagalan operasional yang fatal.

Anatomi Kegagalan: Mengapa Sistem Mengalami Leak dan Deadlock?

1. Reply Queue Leak

Pada implementasi naif, klien membuat antrean sementara (temporary queue) per request atau per instance klien untuk menampung pesan balasan. Kebocoran terjadi ketika:

  • Klien terhenti secara mendadak (OOM crash, node restart) sebelum sempat mengeksekusi perintah queue_delete.
  • Koneksi terputus dan broker mempertahankan antrean karena flag auto_delete atau exclusive tidak dipasang dengan benar. Akibatnya ribuan antrean kosong menumpuk, menghabiskan memori broker (Erlang processes, file descriptors) hingga broker berhenti merespons.

2. Orphaned Correlation ID

Klien menerapkan timeout lokal (misalnya 5 detik). Jika worker membutuhkan waktu 6 detik, klien sudah memutus sesi tunggu dan membersihkan state internal correlation_id tersebut. Ketika balasan akhirnya tiba di reply queue, tidak ada lagi thread atau listener yang menunggu pesan tersebut. Pesan balasan menumpuk di antrean tanpa pernah dikonsumsi (unacknowledged/ready accumulation).

3. Worker Deadlock dan Consumer Starvation

Deadlock pada arsitektur worker queue biasanya bukan disebabkan oleh circular wait level OS, melainkan logical timeout block. Ketika worker mengalami Unhandled Exception (misalnya koneksi database terputus atau format payload rusak), alur eksekusi terhenti sebelum pesan balasan dikirimkan ke reply_to. Jika worker langsung melakukan basic.ack atau crash tanpa mengirim status kegagalan, klien pemanggil akan terus menunggu balasan hingga batas waktu habis, atau bahkan hang tanpa batas waktu jika klien tidak mengonfigurasi timeout lokal.

Solusi Arsitektur: Lifecycle, Fail-Safe Envelope, dan Timeout

1. Manajemen Siklus Hidup Reply Queue

Hindari pembuatan queue dinamis baru per request (anti-pattern). Gunakan salah satu dari dua pendekatan berikut:

  • Direct Reply-to (Pseudo-Queue): Fitur native RabbitMQ (amq.rabbitmq.reply-to). Klien mengonsumsi antrean virtual ini tanpa menciptakan antrean nyata di broker. Biaya deklarasi nol dan memory leak tereliminasi secara fisik.
  • Single Reply Queue per Klien dengan TTL: Jika broker tidak mendukung direct reply-to, gunakan satu antrean per instance klien dengan konfigurasi ketat: 
    x-expires = 60000        # Antrean otomatis terhapus jika tidak aktif selama 60 detik
exclusive = true         # Terikat langsung dengan lifecycle koneksi TCP klien
auto_delete = true       # Hapus ketika consumer terakhir disconnect

2. Fail-Safe Execution Envelope pada Worker

Worker wajib menjamin bahwa setiap pesan yang memiliki reply_to akan selalu menerima respons, baik saat eksekusi berhasil maupun saat terjadi catastrophic error. Kemas pesan balasan dalam envelope terstruktur:

{
  "status": "success" | "error",
  "data": null | { ... },
  "error": null | {
    "code": "INVALID_PAYLOAD",
    "message": "Detailed error message"
  }
}

Implementasi Worker Production-Ready (Python / Pika)

Berikut adalah implementasi worker AMQP yang menerapkan fail-safe envelope, penanganan unhandled exception secara atomik, dan validasi atribut perpesanan.

import json
import logging
import pika

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("WorkerRPC")

def process_task(payload: dict) -> dict:
    # Simulasi domain logic
    if "value" not in payload:
        raise ValueError("Field 'value' wajib disertakan.")
    return {"result": payload["value"] * 2}

def on_request(ch, method, props, body):
    correlation_id = props.correlation_id
    reply_to = props.reply_to

    # Validasi kepatuhan protokol RPC
    if not reply_to or not correlation_id:
        logger.warning("Pesan diabaikan: reply_to atau correlation_id tidak valid.")
        ch.basic_ack(delivery_tag=method.delivery_tag)
        return

    response_envelope = {
        "status": "error",
        "data": None,
        "error": None
    }

    try:
        payload = json.loads(body.decode("utf-8"))
        data = process_task(payload)
        response_envelope["status"] = "success"
        response_envelope["data"] = data
    except ValueError as ve:
        logger.error(f"Business logic error: {str(ve)}")
        response_envelope["error"] = {"code": "BAD_REQUEST", "message": str(ve)}
    except Exception as exc:
        logger.critical(f"Unhandled system error: {str(exc)}", exc_info=True)
        response_envelope["error"] = {"code": "INTERNAL_ERROR", "message": "Kesalahan internal worker."}
    finally:
        # Jaminan eksekusi: Reply SELALU dikirim kembali ke reply_to
        try:
            ch.basic_publish(
                exchange="",
                routing_key=reply_to,
                properties=pika.BasicProperties(
                    correlation_id=correlation_id,
                    content_type="application/json",
                    delivery_mode=1  # Transient: jangan tulis reply ke disk jika antrean mati
                ),
                body=json.dumps(response_envelope).encode("utf-8")
            )
        except Exception as pub_err:
            logger.error(f"Gagal mempublikasikan reply: {pub_err}")

        # Konfirmasi konsumsi request
        ch.basic_ack(delivery_tag=method.delivery_tag)

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
    channel = connection.channel()

    # Prefetch count wajib diatur untuk mencegah overload memory worker
    channel.basic_qos(prefetch_count=10)
    channel.queue_declare(queue="rpc_requests", durable=True)

    channel.basic_consume(queue="rpc_requests", on_message_callback=on_request)
    logger.info("Worker RPC aktif. Menunggu permintaan...")
    channel.start_consuming()

if __name__ == "__main__":
    main()

Observabilitas dan Monitoring Antrean

Untuk mencegah kebocoran antrean dan degradasi sistem yang tidak terdeteksi, pantau metrik broker berikut via Prometheus atau monitoring bawaan:

  • Queue Churn Rate (rabbitmq_queues_created_total vs rabbitmq_queues_deleted_total): Lonjakan tajam pembuatan antrean tanpa angka penghapusan yang seimbang adalah indikator mutlak terjadinya reply queue leak.
  • Ready Messages pada Reply Queues: Metrik rabbitmq_queue_messages_ready pada antrean temporer harus bernilai mendekati 0. Jika nilai meningkat, berarti klien mengalami timeout sebelum worker sempat mengembalikan balasan (orphaned message).
  • Queue Unacknowledged (rabbitmq_queue_messages_unacknowledged): Jika antrean request utama memiliki unacknowledged message yang stagnan, worker kemungkinan hang pada level socket/I/O saat memproses request tanpa trigger exception.
Rekomendasi Operasional: Selalu atur batas timeout di kedua sisi. Di sisi klien, pasang timeout pembacaan soket. Di sisi broker, lampirkan header expiration pada pesan request agar broker otomatis membuang request yang tertahan di antrean melebihi toleransi waktu pemanggil.