Publisher Confirms #

Dalam merancang arsitektur komunikasi asinkron yang andal menggunakan RabbitMQ, perhatian kita sering kali terpusat pada sisi konsumen: bagaimana menangani kesalahan, mengatur batas pra-ambil (prefetch), dan merancang proses yang idempoten. Namun, ada gerbang awal yang tidak kalah kritis yang menentukan kelangsungan hidup pesan: sisi Produsen (Publisher).

Tanpa jaminan di sisi produsen, sistem kita rentan terhadap hilangnya data secara diam-diam (silent data loss). Hanya karena fungsi pengiriman kode aplikasi kita tidak menghasilkan error, bukan berarti pesan tersebut telah sampai dan disimpan dengan aman oleh broker RabbitMQ. Di sinilah Publisher Confirms berperan sebagai protokol konfirmasi resmi dari broker kembali ke produsen. Artikel ini akan membahas tuntas mekanisme internal Publisher Confirms, perbedaannya dengan transaksi AMQP, interaksinya dengan media penyimpanan disk dan konsensus Quorum Queue, tiga pola sinkronisasi di aplikasi, serta implementasi produsen asinkron yang tangguh menggunakan bahasa Go.

Posisi Publisher Confirms dalam Siklus Hidup Pesan #

Publisher Confirms beroperasi di gerbang masuk paling awal dari daur hidup pesan, mengamankan jalur transmisi antara aplikasi produsen dan broker.

flowchart TD
    Producer["Produsen (Publisher)"] -->|"1. Publish (Sequence #1)"| Exchange["Exchange"]
    Exchange -->|"2. Rute ke Antrean"| Queue["Queue"]
    
    subgraph StorageConfirm["Siklus Penyimpanan & Konsensus"]
        Queue -->|"3. Replikasi (Quorum)"| Replicas["Replicas"]
        Queue -->|"4. Write & fsync"| Disk["Disk Fisik"]
    end
    
    Disk -->|"5. Generate ACK (Sequence #1)"| Channel["Channel Broker"]
    Replicas -->|"5. Generate ACK"| Channel
    Channel -->|"6. basic.ack (Sequence #1)"| Producer

Siklus penerbitan baru dianggap tuntas dan sah secara hukum sistem setelah produsen menerima paket frame basic.ack spesifik dari broker yang merujuk pada nomor urut (sequence number) pesan yang bersangkutan.

Mengapa Publisher Confirms Mutlak Diperlukan? #

Sebelum adanya fitur Publisher Confirms, protokol AMQP 0-9-1 mengandalkan penerbitan pesan berbasis kepercayaan (fire-and-forget). Produsen menulis data ke soket jaringan TCP lokal dan langsung menganggap proses sukses tanpa menunggu respon apa pun dari broker.

Pola ini sangat berbahaya karena beberapa alasan:

1. Kegagalan Perutean (Unroutable Message): Jika produsen mengirim pesan ke Exchange yang tidak memiliki Binding ke antrean mana pun, pesan akan dibuang secara diam-diam (silent drop) oleh broker tanpa memberi tahu produsen.
2. Crash Broker sebelum fsync: Broker dapat menerima pesan di memori RAM, namun mengalami crash atau kehilangan daya listrik sebelum sempat menuliskan pesan persistent tersebut ke disk.
3. Buffer Jaringan Luap: Soket TCP lokal produsen dapat melaporkan penulisan sukses, tetapi paket data tersebut terbuang di tengah jalan akibat kemacetan router jaringan (network congestion).

Publisher Confirms menyelesaikan masalah ini dengan memberikan mekanisme umpan balik (feedback loop) aktif dari broker kembali ke produsen untuk setiap pesan tunggal.

Mekanisme Internal Konfirmasi dan Sequence Number #

Ketika kita mengaktifkan Publisher Confirms pada channel komunikasi (melalui perintah confirm.select AMQP):

1. Channel tersebut akan dikunci ke dalam Confirm Mode.
2. Broker RabbitMQ mulai menghitung nomor urut integer 64-bit (sequence_number) yang dimulai dari 1 untuk setiap pesan yang diterbitkan melalui channel tersebut.
3. Produsen juga melacak sequence number ini secara lokal di memorinya.
4. Setelah memproses pesan, broker akan mengirimkan salah satu dari dua sinyal berikut kembali ke produsen:
   • basic.ack: Menandakan pesan sukses diterima dan diamankan.
   • basic.nack: Menandakan kegagalan broker. Ini bisa terjadi jika antrean tujuan penuh dan dikonfigurasi dengan overflow policy reject-publish, atau jika broker mengalami kegagalan sistem internal.

Interaksi dengan Tipe Penyimpanan Antrean #

Waktu pengiriman ACK oleh broker sangat ditentukan oleh tipe antrean dan properti pesan yang kita kirimkan:

• Antrean Transient / Pesan Transient: Broker langsung mengirimkan ACK seketika setelah pesan berhasil disalin ke memori RAM volatil broker. Operasi ini sangat cepat namun tidak tahan restart.
• Antrean Durable / Pesan Persistent: Broker baru mengirimkan ACK setelah payload pesan berhasil ditulis dan dikunci ke disk fisik via pemanggilan sistem fsync. Operasi ini membutuhkan waktu I/O disk milidetik.
• Quorum Queue: Broker baru mengirimkan ACK ke produsen setelah log pesan berhasil ditulis ke disk lokal pemimpin (Leader) dan berhasil direplikasikan serta di-commit ke disk mayoritas node replika klaster (Followers). Ini memberikan tingkat jaminan tertinggi.

Penanganan Timeout dan Kegagalan Jaringan Berat #

Menerapkan konfirmasi asinkron juga menuntut kita untuk siap menghadapi skenario terburuk di mana broker tidak kunjung mengirimkan jawaban ACK atau NACK dalam jangka waktu tertentu.

Mengapa Timeout Bisa Terjadi? #

Timeout konfirmasi biasanya dipicu oleh kelebihan beban I/O disk (disk I/O bottleneck) pada server broker. Ketika laju pesan persisten yang masuk melampaui kemampuan hardware harddisk untuk melakukan fsync, antrean buffer konfirmasi internal broker akan menumpuk. Broker akan menunda pengiriman ACK ke produsen untuk melindungi dirinya agar tidak kehabisan RAM.

Strategi Penanganan Timeout di Aplikasi #

Untuk menjaga stabilitas sistem, produsen kita tidak boleh membiarkan pelacakan pesan di memori lokal menggantung selamanya. Kita wajib menerapkan mekanisme batas waktu (timeout):

1. Membatasi Antrean In-Flight: Batasi jumlah maksimal pesan yang menanti konfirmasi (misalnya maksimal 5000 pesan). Jika batas terlampaui, tangguhkan penerbitan pesan baru (backpressure sisi produsen).
2. Pemantauan Timer: Setiap pesan yang dimasukkan ke peta pelacakan lokal harus diasosiasikan dengan pencatat waktu (timestamp).
3. Pembersihan & Retry: Jika dalam waktu misalnya 5 detik pesan tidak kunjung menerima ACK, asumsikan terjadi timeout. Hapus pesan dari peta pelacakan lokal, dan lakukan rute pengiriman ulang (republish) atau kembalikan status error ke pengguna hulu.

Pemulihan Koneksi dan Reset Sequence Number #

Ketika koneksi TCP antara produsen dan RabbitMQ terputus karena gangguan fisik, broker akan menghapus seluruh state channel yang diasosiasikan dengan koneksi tersebut.

Apa yang Terjadi pada Sequence Number Saat Reconnect? #

Begitu pustaka client mendeteksi kegagalan koneksi dan melakukan koneksi ulang (reconnect):

1. Kita harus membuka channel AMQP baru.
2. Membuka channel baru secara otomatis akan me-reset perhitungan sequence_number di broker kembali dari angka 1.
3. Tantangan: Peta pelacakan lokal produsen kita mungkin masih menyimpan sisa pesan dari channel lama dengan sequence number bernilai besar (misalnya 1520).
4. Solusi: Saat proses pemulihan channel terdeteksi, aplikasi produsen wajib membersihkan (clear) seluruh data in-flight di peta pelacakan lama secara aman, menganggap seluruh pesan tersebut gagal terkirim (karena status ACK mereka tidak bisa lagi dikirimkan oleh broker pada channel yang sudah hancur), dan mempublikasikan ulang pesan-pesan tersebut melalui channel baru dengan urutan sequence number yang baru dimulai dari 1.

Perbandingan: Publisher Confirms vs Transaksi AMQP (tx.select) #

RabbitMQ menyediakan opsi lain untuk keandalan produsen, yaitu Transaksi AMQP (tx.select, tx.commit, dan tx.rollback). Namun, transaksi AMQP memiliki kelemahan performa yang sangat besar di lingkungan produksi.

Transaksi AMQP memaksa broker menghentikan pemrosesan antrean dan melakukan sinkronisasi disk I/O penuh untuk setiap pesan secara sekuensial. Sebaliknya, Publisher Confirms memungkinkan produsen terus mempublikasikan pesan ke jaringan tanpa henti, sementara proses konfirmasi mengalir secara asinkron di belakang layar.

Tiga Pola Penerapan Publisher Confirms di Aplikasi #

Aplikasi produsen dapat mengimplementasikan Publisher Confirms menggunakan salah satu dari tiga pola berikut, disesuaikan dengan kebutuhan performa dan latensi kita:

1. Pola Sinkron Tunggal (Single Confirm) #

Produsen mengirim satu pesan, lalu memanggil fungsi blokir thread (seperti WaitForConfirms) untuk menunggu ACK dari broker sebelum mengirimkan pesan berikutnya.

• Karakteristik: Sederhana ditulis, namun performa sangat buruk karena latensi round-trip (RTT) jaringan membatasi kecepatan pengiriman.

2. Pola Batch (Batch Confirm) #

Produsen mengirimkan sekelompok pesan (misalnya 100 pesan) secara berurutan, lalu memanggil fungsi blokir thread sekali untuk menunggu konfirmasi dari seluruh 100 pesan tersebut.

• Karakteristik: Throughput jauh lebih baik daripada pola tunggal. Namun, jika broker mengirimkan NACK untuk satu pesan di tengah batch, produsen tidak tahu pesan mana yang gagal, sehingga terpaksa mempublikasikan ulang seluruh 100 pesan tersebut, memicu duplikasi data di antrean.

3. Pola Asinkron (Asynchronous Confirm) — Direkomendasikan #

Produsen mendaftarkan fungsi pendengar (listener callback) pada channel Go. Produsen menulis pesan ke jaringan secara non-blocking dan melanjutkan tugasnya. Ketika broker mengirim ACK/NACK, goroutine pendengar akan menangkapnya secara asinkron.

• Karakteristik: Throughput maksimal dan latensi terendah. Pola ini membutuhkan manajemen memori lokal di sisi produsen untuk mencatat daftar sequence number pesan yang masih berstatus in-flight menunggu konfirmasi.

Implementasi Kode Go: Produsen Asinkron dengan Penanganan Retry #

Berikut adalah contoh implementasi lengkap dalam bahasa Go untuk produsen asinkron. Kode ini mengaktifkan Publisher Confirms, melacak sequence number pesan menggunakan struktur data peta (map) lokal yang dilindungi mutex, mendengarkan konfirmasi asinkron secara non-blocking, dan mengelola antrean pengiriman ulang (retry) jika menerima sinyal NACK.

package main

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

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

const (
	amqpURL   = "amqp://guest:guest@localhost:5672/"
	queueName = "critical-transactions"
)

// Struktur pelacak pesan in-flight yang menunggu konfirmasi
type ConfirmTracker struct {
	mu       sync.Mutex
	inFlight map[uint64]amqp.Publishing
}

func NewConfirmTracker() *ConfirmTracker {
	return &ConfirmTracker{
		inFlight: make(map[uint64]amqp.Publishing),
	}
}

func (t *ConfirmTracker) Add(seq uint64, msg amqp.Publishing) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.inFlight[seq] = msg
}

func (t *ConfirmTracker) Confirm(seq uint64, multiple bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if multiple {
		for k := range t.inFlight {
			if k <= seq {
				delete(t.inFlight, k)
			}
		}
	} else {
		delete(t.inFlight, seq)
	}
}

func (t *ConfirmTracker) GetAndRemove(seq uint64) (amqp.Publishing, bool) {
	t.mu.Lock()
	defer t.mu.Unlock()
	msg, exists := t.inFlight[seq]
	if exists {
		delete(t.inFlight, seq)
	}
	return msg, exists
}

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. Mengaktifkan Publisher Confirms Mode
	err = ch.Confirm(false) // false karena kita ingin pemrosesan asinkron
	if err != nil {
		log.Fatalf("Gagal mengaktifkan Publisher Confirms: %s", err)
	}

	// 3. Deklarasi Antrean Durable
	_, err = ch.QueueDeclare(
		queueName,
		true, // durable
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		log.Fatalf("Gagal deklarasi antrean: %s", err)
	}

	// Mendapatkan channel untuk menerima notifikasi konfirmasi dari broker
	ackChan := make(chan amqp.Confirmation, 100)
	ch.NotifyPublish(ackChan)

	tracker := NewConfirmTracker()

	// 4. Goroutine Pendengar Konfirmasi Asinkron
	go func() {
		for confirm := range ackChan {
			if confirm.Ack {
				// broker sukses menerima dan mengamankan pesan
				log.Printf("✓ Menerima ACK dari broker untuk Sequence: %d (multiple: %t)", confirm.DeliveryTag, confirm.Ack)
				tracker.Confirm(confirm.DeliveryTag, confirm.Ack)
			} else {
				// broker gagal (NACK)
				log.Printf("✗ Menerima NACK dari broker untuk Sequence: %d", confirm.DeliveryTag)
				msg, exists := tracker.GetAndRemove(confirm.DeliveryTag)
				if exists {
					// Jalankan strategi retry: kirim kembali pesan tersebut
					log.Printf("[RETRY] Mempublikasikan ulang pesan sequence %d...", confirm.DeliveryTag)
					republishMessage(ch, tracker, msg)
				}
			}
		}
	}()

	// 5. Menerbitkan Pesan secara Non-blocking
	ctx := context.Background()
	for i := 1; i <= 50; i++ {
		payload := []byte(fmt.Sprintf(`{"tx_id":"TX-90081-%d","amount":350000}`, i))
		msg := amqp.Publishing{
			DeliveryMode: amqp.Persistent,
			ContentType:  "application/json",
			Body:         payload,
			MessageId:    fmt.Sprintf("msg_uuid_%d", i),
		}

		// Mengunci sequence number yang akan didapatkan berikutnya sebelum memanggil Publish
		nextSeqNum := ch.GetNextPublishSeqNo()
		tracker.Add(nextSeqNum, msg)

		err = ch.PublishWithContext(ctx,
			"", // Default Exchange
			queueName,
			false, // mandatory
			false, // immediate
			msg,
		)
		if err != nil {
			log.Printf("Gagal publish pesan %d: %s", i, err)
			tracker.GetAndRemove(nextSeqNum)
		} else {
			log.Printf("Pesan %d dikirim (Sequence: %d)", i, nextSeqNum)
		}
		time.Sleep(10 * time.Millisecond)
	}

	// Memberikan waktu agar seluruh konfirmasi asinkron selesai diterima sebelum shutdown
	time.Sleep(2 * time.Second)
}

func republishMessage(ch *amqp.Channel, tracker *ConfirmTracker, msg amqp.Publishing) {
	ctx := context.Background()
	nextSeqNum := ch.GetNextPublishSeqNo()
	tracker.Add(nextSeqNum, msg)

	err := ch.PublishWithContext(ctx,
		"",
		queueName,
		false,
		false,
		msg,
	)
	if err != nil {
		log.Printf("[RETRY ERROR] Gagal mengirim ulang: %s", err)
		tracker.GetAndRemove(nextSeqNum)
	}
}

Anti-Pattern vs Solusi Praktis di Lingkungan Produksi #

Hindari kesalahan fatal konfigurasi penerbitan berikut demi menjaga keandalan data klaster:

Anti-Pattern: Menganggap TCP Connection Write Sukses Sebagai Jaminan Pesan Sampai #

Menulis logika program produsen yang mengasumsikan transaksi pemesanan sukses dibuat hanya karena baris kode ch.PublishWithContext tidak mengmengembalikan error Go (err == nil).

Mengapa ini salah? #

Fungsi PublishWithContext mengembalikan status sukses segera setelah driver pustaka client sukses menuliskan biner data ke soket jaringan TCP sistem operasi lokal. Driver client tidak memiliki informasi apakah jaringan antara produsen dan broker sedang terputus di tengah jalan, apakah Exchange tujuan salah eja, ataukah broker RabbitMQ kehabisan RAM. Tanpa mengaktifkan Publisher Confirms, pesan kita berada di bawah spektrum At-Most-Once di mana kehilangan data secara diam-diam (silent data loss) sangat sering terjadi tanpa terdeteksi oleh sistem logging produsen.

• Solusi: Selalu aktifkan Publisher Confirms secara asinkron untuk seluruh transaksi bisnis krusial. Jika kita membutuhkan performa ekstrim, gunakan pola Transactional Outbox di mana pesan disimpan di tabel database lokal terlebih dahulu sebelum didorong ke RabbitMQ oleh worker khusus yang memantau penerimaan ACK/NACK dari broker.

Ringkasan #

• Keandalan Sisi Produsen — Publisher Confirms adalah protokol umpan balik aktif dari broker ke produsen untuk memastikan pesan sukses sampai dan disimpan oleh broker.
• Sequence Number Pelacak — Broker menetapkan nomor urut linier per channel untuk setiap pesan yang masuk, yang dicocokkan dengan metadata di sisi produsen.
• Sinyal ACK vs NACK — Sinyal basic.ack dikirim saat pesan sukses diamankan. Sinyal basic.nack dikirim jika antrean tujuan penuh atau terjadi error internal broker.
• Interaksi dengan Quorum Queue — Pada Quorum Queue, ACK baru dikirim ke produsen setelah log pesan sukses ditulis ke disk lokal pemimpin dan mayoritas pengikut klaster.
• Asynchronous Confirm Terbaik — Pola asinkron menggunakan callback listener menyajikan performa throughput maksimal dan latensi terendah dibandingkan pola sinkron atau batch confirms.
• Pencegahan Silent Data Loss — Tanpa Publisher Confirms, produsen buta terhadap kegagalan perutean dan pemadaman broker, menempatkan sistem pada risiko kehilangan data parah.

← Sebelumnya: Exactly-once   Berikutnya: Consumer Acknowledge →