At-least-once #

Dalam sistem terdistribusi modern, kehilangan data adalah salah satu skenario kegagalan terburuk yang harus dihindari oleh para insinyur perangkat lunak. Untuk menjawab tantangan keandalan ini, RabbitMQ secara default dirancang untuk mendukung dan beroperasi di bawah spektrum jaminan pengiriman At-Least-Once (Paling Sedikit Sekali).

Prinsip dasar dari jaminan At-Least-Once adalah prioritas mutlak terhadap integritas data: broker menjamin bahwa setiap pesan yang diterbitkan akan dikirimkan dan diproses oleh konsumen setidaknya satu kali. Namun, untuk memberikan garansi tanpa kehilangan pesan ini, sistem terdistribusi harus membayar harga berupa kemungkinan terjadinya pesan duplikat (duplikasi). Di lingkungan produksi, kegagalan jaringan sejenak atau crash pada aplikasi konsumen di waktu yang tidak tepat dapat menyebabkan pesan yang sama terkirim berulang kali. Artikel ini akan membahas mendalam bagaimana RabbitMQ mencapai jaminan At-Least-Once, membedah skenario-skenario penyebab terjadinya duplikasi pesan, merumuskan strategi penanganan idempotensi di sisi konsumen, serta menyajikan implementasi kode Go yang tangguh.

Posisi At-Least-Once dalam Siklus Hidup Pesan #

Jaminan At-Least-Once dicapai dengan menerapkan siklus konfirmasi dua arah yang ketat pada sisi penerbitan (Publisher Confirms) dan sisi konsumsi (Consumer Acknowledgements).

flowchart TD
    Producer["Produsen (Publisher)"] -->|"1. Publish + Confirm Mode"| Exchange["Exchange"]
    Exchange -->|"2. Rute Pesan"| Queue["Queue (Durable)"]
    Queue -->|"3. Replikasi Konsensus (Quorum)"| Replicas["Node Follower"]
    
    Queue -->|"4. Push (basic.deliver)"| Consumer["Aplikasi Konsumen"]
    Consumer -->|"5. Proses Bisnis Sukses"| DB["Database (Atomic Commit)"]
    DB -->|"6. Kirim Manual ACK"| Queue
    Queue -->|"7. Hapus Pesan"| Delete["Selesai"]

    style Producer stroke:#0288d1,stroke-width:2px
    style Queue stroke:#388e3c,stroke-width:2px
    style Consumer stroke:#e65100,stroke-width:2px

Dalam model ini, pesan tidak akan pernah dihapus dari penyimpanan broker sebelum konsumen mengirimkan sinyal basic.ack eksplisit. Sinyal ini baru boleh dikirimkan setelah seluruh logika bisnis dan penyimpanan database di sisi konsumen diselesaikan dengan sukses.

Bagaimana RabbitMQ Mencapai Jaminan At-Least-Once? #

Keberhasilan jaminan At-Least-Once tidak bertumpu pada satu fitur tunggal, melainkan merupakan hasil kolaborasi erat antara produsen, broker RabbitMQ, dan konsumen. Ada tiga pilar utama yang menyusun jaminan ini:

1. Sisi Penerbitan (Producer): Publisher Confirms #

Tanpa adanya konfirmasi dari broker, produsen tidak memiliki cara untuk mengetahui apakah pesan yang dikirimnya benar-benar sampai di broker. Produsen mengaktifkan mode Publisher Confirms pada channel komunikasi. Broker akan mengirimkan ACK balik ke produsen hanya setelah pesan berhasil diterima, dirutekan ke antrean, dan (jika persistent) diamankan ke disk. Jika terjadi kendala jaringan sebelum broker menerima pesan, produsen akan mendeteksi hilangnya konfirmasi tersebut dan melakukan pengiriman ulang (retry).

2. Sisi Penyimpanan (Broker): Durable & Persistent #

Untuk memastikan pesan bertahan melewati restart broker atau kegagalan hardware, kita harus mengonfigurasi penyimpanan permanen:

  • Mendeklarasikan antrean dengan flag durable = true agar struktur antrean tidak terhapus saat broker dimulai ulang.
  • Mengirimkan pesan dengan properti delivery_mode = 2 (persistent) agar payload bodi pesan ditulis secara fisik ke berkas segmen disk (msg_store_persistent) menggunakan pemanggilan sistem fsync.
  • Jika menggunakan Quorum Queues, pesan direplikasikan secara konsisten ke mayoritas node klaster menggunakan log Raft sebelum broker memberikan konfirmasi sukses ke produsen.

3. Sisi Konsumsi (Consumer): Manual Acknowledgements #

Konsumen harus mematikan fitur konfirmasi otomatis (autoAck = false) dan secara eksplisit mengirimkan perintah basic.ack secara manual setelah seluruh pemrosesan data selesai. Selama ACK manual belum diterima oleh broker, broker akan menandai pesan sebagai Unacknowledged di memori RAM dan tetap mempertahankan salinan fisiknya.

Skenario Utama Terjadinya Pesan Duplikat (Duplicates) #

Karena sistem terdistribusi selalu dihadapkan pada ketidakpastian jaringan (network uncertainty), duplikasi pesan adalah konsekuensi logis yang tidak dapat dihindari demi mencegah hilangnya pesan.

Berikut adalah tiga skenario utama di mana duplikasi pesan terjadi di lingkungan produksi:

Skenario 1: Konsumen Crash Setelah Proses Bisnis, Sebelum Mengirim ACK #

Ini adalah skenario kegagalan paling umum dalam aplikasi microservices:

  1. Konsumen menerima pesan pesanan dari RabbitMQ.
  2. Konsumen sukses memproses pesanan dan memotong stok di database utama.
  3. Tepat sebelum konsumen sempat memanggil fungsi d.Ack(false) untuk dikirimkan melalui soket TCP ke broker, server konsumen mengalami crash atau kehabisan daya listrik.
  4. RabbitMQ mendeteksi bahwa koneksi TCP dengan konsumen tersebut terputus secara tiba-tiba.
  5. Karena status pesan tersebut masih Unacknowledged, RabbitMQ mengembalikan pesan tersebut (requeue) ke kepala antrean.
  6. Saat konsumen baru menyala kembali, ia akan menerima pesan pesanan yang sama untuk kedua kalinya, meskipun stok di database sebenarnya sudah dipotong pada percobaan pertama.

Skenario 2: Kegagalan Jaringan (Network Partition) #

Skenario ini terjadi ketika baik konsumen maupun broker sama-sama sehat, namun jalur komunikasi di antara mereka terganggu:

  1. Konsumen sukses memproses pesan dan memanggil d.Ack(false).
  2. Sinyal ACK dikirim ke jaringan, namun terjadi gangguan koneksi jaringan sejenak (network glitch) yang menyebabkan paket ACK tersebut hilang di tengah jalan dan tidak pernah sampai di broker.
  3. Broker menganggap konsumen tidak responsif karena koneksi TCP mengalami heartbeat timeout.
  4. Broker mengembalikan pesan ke antrean dan mendorongnya ke konsumen lain. Konsumen lain kini memproses pesan yang sebenarnya sudah sukses diselesaikan oleh konsumen pertama.

Skenario 3: Penutupan Channel Akibat Pengecualian Protokol (Exception) #

Jika konsumen melakukan kesalahan pengoperasian protokol AMQP (seperti mencoba meng-ACK pesan dengan delivery_tag yang salah, atau memanggil ACK pada channel yang berbeda):

  • Broker akan menutup secara paksa (channel exception) seluruh channel tersebut.
  • Penutupan channel secara otomatis membatalkan seluruh status pesan in-flight yang sedang diproses di channel tersebut, memaksa broker me-requeue semua pesan tersebut secara massal, memicu pemrosesan ganda pada pesan-pesan yang sebenarnya sedang berjalan dengan sukses di memori konsumen.

Strategi Penanganan Idempotensi di Sisi Konsumen #

Karena duplikasi pesan adalah kepastian arsitektural yang tidak dapat dihindari di bawah model At-Least-Once, aplikasi konsumen wajib dirancang bersifat Idempotent. Idempoten berarti aplikasi kita mampu menerima dan memproses pesan yang sama berulang kali, namun hanya menghasilkan efek samping (side effect) bisnis tepat satu kali.

Ada tiga strategi utama untuk membangun konsumen yang idempotent di produksi:

1. Unique Event ID & Deduplication Store #

Setiap pesan yang diterbitkan oleh produsen harus disemati pengidentifikasi UUID unik di dalam properti message_id atau di dalam bodi payload.

  • Cara Kerja: Konsumen memelihara basis data penyimpanan cepat terpusat (seperti Redis dengan waktu kedaluwarsa/TTL, atau tabel database relasional terdedikasi). Sebelum memproses pesan, konsumen memeriksa apakah ID pesan tersebut sudah ada di penyimpanan deduplikasi. Jika sudah ada, konsumen langsung mengabaikan pesan tersebut dan mengirimkan ACK ke broker tanpa memproses logika bisnis ulang.

2. Constraint Database (Indeks Unik) #

Memanfaatkan fitur integritas database relasional sebagai benteng pertahanan terakhir.

  • Cara Kerja: Jika pesan mewakili proses pembuatan data (misalnya pendaftaran pengguna baru), kita dapat membuat indeks unik (Unique Constraint) pada kolom alamat email di database. Jika terjadi duplikasi pesan, upaya penulisan kedua kali akan ditolak secara atomik oleh database dengan error pelanggaran duplikasi (duplicate key error). Konsumen menangkap error ini, menganggap pemrosesan sukses, dan mengirimkan ACK ke broker tanpa memicu kegagalan sistem.

3. Logika Bisnis Berbasis State Machine #

Menghindari operasi matematika relatif yang akumulatif dan menggantinya dengan operasi transisi status logis yang eksplisit.

  • Contoh Anti-Pattern: Mengirimkan perintah pesan Amount = Amount - 10000 untuk memotong saldo. Jika terjadi duplikasi, saldo pengguna akan terpotong dua kali (-20000).
  • Contoh Solusi Idempoten: Mengirimkan pesan berisi instruksi transisi status pesanan dari PENDING menjadi PAID pada baris ID transaksi pembayaran spesifik. Sebelum mengubah saldo, periksa apakah baris ID transaksi tersebut sudah memiliki status PAID di database. Jika statusnya sudah PAID, abaikan pesan duplikat tersebut dengan aman.

Implementasi Kode Go: Konsumen Tangguh dengan Deduplikasi Redis #

Berikut adalah contoh implementasi lengkap dalam bahasa Go untuk konsumen At-Least-Once. Kode ini menggunakan koneksi Redis untuk memeriksa dan menyimpan kunci deduplikasi pesan secara atomik sebelum mengeksekusi logika bisnis utama, menjamin pemrosesan yang aman terhadap bahaya duplikasi.

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/redis/go-redis/v9"
	amqp "github.com/rabbitmq/amqp091-go"
)

var (
	ctx         = context.Background()
	rdb         *redis.Client
	amqpURL     = "amqp://guest:guest@localhost:5672/"
	queueName   = "durable-order-payments"
	redisPrefix = "processed_message:"
)

func init() {
	// Inisialisasi Koneksi Redis Cache untuk Deduplication Store
	rdb = redis.NewClient(&redis.Options{
		Addr: "localhost:6379",
	})
}

func main() {
	// 1. Membuka Koneksi TCP ke RabbitMQ
	conn, err := amqp.Dial(amqpURL)
	if err != nil {
		log.Fatalf("Gagal terhubung ke RabbitMQ: %s", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Gagal membuka channel: %s", err)
	}
	defer ch.Close()

	// 2. Deklarasi Antrean Durable
	_, err = ch.QueueDeclare(
		queueName,
		true,  // durable: true (Antrean bertahan saat restart)
		false, // auto-delete
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal deklarasi antrean: %s", err)
	}

	// 3. Konfigurasi Prefetch QoS untuk Fair Dispatch
	err = ch.Qos(10, 0, false)
	if err != nil {
		log.Fatalf("Gagal menyetel QoS: %s", err)
	}

	// 4. Mendaftarkan Konsumen dengan manual ACK (autoAck = false)
	msgs, err := ch.Consume(
		queueName,
		"",    // consumer tag
		false, // autoAck: false (KUNCI UTAMA AT-LEAST-ONCE DI KONSUMEN!)
		false, // exclusive
		false, // no-local
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan konsumen: %s", err)
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	log.Println("[*] Konsumen At-Least-Once aktif. Tekan Ctrl+C untuk keluar.")

	go func() {
		for d := range msgs {
			// Memeriksa keunikan ID Pesan (Deduplikasi)
			messageID := d.MessageId
			if messageID == "" {
				log.Printf("[Peringatan] Pesan dengan Tag %d tidak memiliki MessageId. Mengabaikan pengecekan deduplikasi...", d.DeliveryTag)
				// JANGAN abaikan pesan di produksi; di sini kita hanya memberi peringatan
			}

			// Menggunakan operasi SETNX (Set if Not Exists) Redis secara atomik
			redisKey := fmt.Sprintf("%s%s", redisPrefix, messageID)
			// Kita setel kunci deduplikasi dengan waktu kedaluwarsa (TTL) 24 jam
			isUnique, err := rdb.SetNX(ctx, redisKey, "processed", 24*time.Hour).Result()
			
			if err != nil {
				log.Printf("[ERROR] Gagal berkomunikasi dengan Redis: %s. Melakukan requeue pesan...", err)
				// Jika Redis down, kembalikan pesan ke antrean agar dicoba lagi
				d.Nack(false, true)
				continue
			}

			if !isUnique {
				// PESAN DUPLIKAT TERDETEKSI: Abaikan proses bisnis, langsung kirim ACK
				log.Printf("[INFO] Pesan duplikat terdeteksi (ID: %s). Langsung mengirim ACK...", messageID)
				d.Ack(false)
				continue
			}

			// 5. Eksekusi Logika Bisnis Utama
			log.Printf("Memproses pembayaran transaksi untuk ID Pesan: %s", messageID)
			processSuccess := executePaymentLogic(d.Body)

			if !processSuccess {
				// GAGAL PROSES: Hapus kunci deduplikasi di Redis dan lakukan requeue
				log.Printf("[INFO] Gagal memproses transaksi bisnis. Menolak pesan...")
				rdb.Del(ctx, redisKey)
				d.Nack(false, true) // requeue = true
				continue
			}

			// 6. Mengirimkan Manual ACK setelah Sukses Pemrosesan & Database Commit
			err = d.Ack(false)
			if err != nil {
				log.Printf("[ERROR] Gagal mengirim ACK ke broker: %s", err)
				// Jika ACK gagal terkirim karena jaringan putus saat ini,
				// pesan akan di-requeue oleh broker dan terkirim ulang.
				// Namun pada percobaan kedua, Redis SETNX akan mendeteksi kunci "processed"
				// dan mencegah transaksi bisnis diproses ganda.
			} else {
				log.Printf("✓ Sukses memproses dan meng-ACK pesan: %s", messageID)
			}
		}
	}()

	<-sigChan
	log.Println("[*] Shutdown aman diselesaikan.")
}

func executePaymentLogic(payload []byte) bool {
	// Simulasi logika pemotongan saldo database relasional
	return true
}

Anti-Pattern vs Solusi Praktis di Lingkungan Produksi #

Hindari kesalahan fatal berikut saat mendesain sistem dengan jaminan At-Least-Once:

Anti-Pattern: Menggunakan Auto-Ack dan Memindahkan Requeue ke Sisi Aplikasi #

Seorang developer menonaktifkan manual ACK di RabbitMQ (autoAck = true) demi throughput cepat, namun dia menulis kode penanganan error internal di dalam aplikasi konsumen yang bertugas mempublikasikan ulang (re-publish) pesan yang gagal ke antrean yang sama secara manual.

Mengapa ini salah? #

Pola ini sangat rentan kehilangan data. Jika aplikasi konsumen crash di tengah-tengah pemrosesan (sebelum sempat mempublikasikan ulang pesan yang gagal tersebut), pesan tersebut hilang selamanya karena broker sudah menghapusnya sejak awal. Selain itu, mempublikasikan ulang pesan secara manual akan mengubah metadata asli pesan (seperti tanggal kirim asli, ID pesan asli, dan header perutean), serta mempersulit analisis jejak kesalahan (troubleshooting).

  • Solusi: Biarkan RabbitMQ yang mengelola requeue secara native. Gunakan manual konfirmasi (autoAck = false). Jika terjadi kegagalan pemrosesan sementara, cukup panggil d.Nack(false, true). Broker akan mengembalikan pesan ke antrean secara aman tanpa mengubah metadata asli pesan.

Ringkasan #

  • Garansi At-Least-Once — Jaminan bahwa setiap pesan akan dikirimkan dan diproses oleh konsumen minimal satu kali, meniadakan risiko kehilangan data secara diam-diam.
  • Risiko Duplikasi Pesan — Merupakan konsekuensi logis dari model ini. Duplikasi terjadi akibat crash pada konsumen setelah proses bisnis sukses namun sebelum ACK dikirim, atau akibat gangguan jaringan.
  • Pilar Pencapaian — Memerlukan Publisher Confirms di produsen, antrean durable dan pesan persistent di broker (atau quorum queue), serta manual ACK di konsumen.
  • Wajib Idempoten — Konsumen harus dirancang idempotent menggunakan taktik pengecekan kunci deduplikasi unik (misal via Redis), constraint indeks unik database, atau logika transisi state.
  • Urutan Eksekusi yang Benar — Selalu selesaikan logika bisnis dan commit transaksi database terlebih dahulu, baru kemudian panggil instruksi d.Ack ke broker.
  • Peran Quorum Queue — Quorum queue memperkuat At-Least-Once dengan memastikan data ter-replicate ke mayoritas node klaster secara konsisten sebelum produsen menerima konfirmasi.

← Sebelumnya: At-most-once   Berikutnya: Exactly-once →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact