Publisher Confirms #

When designing reliable asynchronous communication architectures with RabbitMQ, our attention is often centered on the consumer side: how to handle errors, set prefetch limits, and design idempotent processes. However, there’s an equally critical entry gate that determines message survival: the Producer (Publisher) side.

Without guarantees on the producer side, our system is vulnerable to silent data loss. Just because our application code’s send function doesn’t produce an error doesn’t mean the message has arrived and been safely stored by the RabbitMQ broker. This is where Publisher Confirms plays its role as the official confirmation protocol from the broker back to the producer. This article thoroughly discusses the internal mechanics of Publisher Confirms, its difference from AMQP transactions, its interaction with disk storage and Quorum Queue consensus, three application synchronization patterns, and robust asynchronous producer implementations using Go.

The Position of Publisher Confirms in the Message Lifecycle #

Publisher Confirms operates at the earliest entry gate of the message lifecycle, securing the transmission path between the producer application and the broker.

flowchart TD
    Producer["Producer (Publisher)"] -->|"1. Publish (Sequence #1)"| Exchange["Exchange"]
    Exchange -->|"2. Route to Queue"| Queue["Queue"]
    
    subgraph StorageConfirm["Storage & Consensus Cycle"]
        Queue -->|"3. Replication (Quorum)"| Replicas["Replicas"]
        Queue -->|"4. Write & fsync"| Disk["Physical Disk"]
    end
    
    Disk -->|"5. Generate ACK (Sequence #1)"| Channel["Broker Channel"]
    Replicas -->|"5. Generate ACK"| Channel
    Channel -->|"6. basic.ack (Sequence #1)"| Producer

A publishing cycle is only considered complete and system-legally valid after the producer receives a specific basic.ack frame packet from the broker referencing the sequence number of the message in question.

Why Are Publisher Confirms Absolutely Necessary? #

Before the Publisher Confirms feature existed, the AMQP 0-9-1 protocol relied on trust-based publishing (fire-and-forget). Producers wrote data to the local TCP network socket and immediately assumed success without waiting for any response from the broker.

This pattern is very dangerous for several reasons:

1. Routing Failures (Unroutable Messages): If a producer sends a message to an Exchange with no Binding to any queue, the message is silently dropped by the broker without informing the producer.
2. Broker Crash before fsync: The broker can receive the message in RAM memory, but crash or lose power before writing that persistent message to disk.
3. Network Buffer Overflow: The producer’s local TCP socket can report a successful write, but the data packet is discarded mid-journey due to network congestion in routers.

Publisher Confirms solves this problem by providing an active feedback loop mechanism from the broker back to the producer for every single message.

Internal Confirmation Mechanics and Sequence Numbers #

When we enable Publisher Confirms on a communication channel (through the confirm.select AMQP command):

1. That channel is locked into Confirm Mode.
2. The RabbitMQ broker starts counting 64-bit integer sequence numbers (sequence_number) starting from 1 for every message published through that channel.
3. The producer also tracks this sequence number locally in its memory.
4. After processing a message, the broker sends one of two signals back to the producer:
   • basic.ack: Signals the message was successfully received and secured.
   • basic.nack: Signals a broker failure. This can happen if the destination queue is full and configured with the reject-publish overflow policy, or if the broker experiences an internal system failure.

Interaction with Queue Storage Types #

The broker’s ACK send time is largely determined by the queue type and message properties we send:

• Transient Queue / Transient Message: The broker sends the ACK immediately after the message is successfully copied to the broker’s volatile RAM memory. This operation is very fast but not restart-resistant.
• Durable Queue / Persistent Message: The broker only sends the ACK after the message payload is successfully written and locked to physical disk via fsync system calls. This operation takes milliseconds of disk I/O time.
• Quorum Queue: The broker only sends the ACK to the producer after the message log is successfully written to the Leader’s local disk and successfully replicated and committed to the disk of a majority of cluster replica nodes (Followers). This provides the highest guarantee level.

Timeout Handling and Heavy Network Failures #

Implementing asynchronous confirmations also demands we be ready for the worst-case scenario where the broker doesn’t send an ACK or NACK answer within a certain period.

Why Can Timeouts Happen? #

Confirmation timeouts are usually triggered by disk I/O bottlenecks on the broker server. When the incoming persistent message rate exceeds the hard disk hardware’s ability to perform fsync, the broker’s internal confirmation buffer queue piles up. The broker delays sending ACKs to producers to protect itself from running out of RAM.

Timeout Handling Strategies in Applications #

To maintain system stability, our producer must not let message tracking in local memory hang forever. We must apply a timeout mechanism:

1. Limit the In-Flight Queue: Limit the maximum number of messages awaiting confirmation (e.g., maximum 5000 messages). If the limit is exceeded, suspend new message publishing (producer-side backpressure).
2. Timer Monitoring: Every message inserted into the local tracking map must be associated with a timestamp.
3. Cleanup & Retry: If a message doesn’t receive an ACK within, say, 5 seconds, assume a timeout occurred. Remove the message from the local tracking map, and perform a republish route or return an error status to the upstream user.

Connection Recovery and Sequence Number Reset #

When the TCP connection between the producer and RabbitMQ drops due to physical disruptions, the broker deletes all channel state associated with that connection.

What Happens to Sequence Numbers on Reconnect? #

Once the client library detects the connection failure and reconnects:

1. We must open a new AMQP channel.
2. Opening a new channel automatically resets the broker’s sequence_number count back to 1.
3. Challenge: Our producer’s local tracking map may still hold leftover messages from the old channel with large sequence number values (e.g., 1520).
4. Solution: When channel recovery is detected, the producer application must safely clear all in-flight data in the old tracking map, consider all those messages failed to send (because their ACK status can no longer be delivered by the broker on the destroyed channel), and republish those messages through the new channel with a new sequence number order starting from 1.

Comparison: Publisher Confirms vs AMQP Transactions (tx.select) #

RabbitMQ provides another option for producer reliability: AMQP Transactions (tx.select, tx.commit, and tx.rollback). However, AMQP transactions have a very large performance weakness in production environments.

AMQP transactions force the broker to stop queue processing and perform full disk I/O synchronization for every message sequentially. Conversely, Publisher Confirms lets producers keep publishing messages to the network continuously, while the confirmation process flows asynchronously behind the scenes.

Three Publisher Confirms Implementation Patterns in Applications #

Producer applications can implement Publisher Confirms using one of three patterns below, adjusted to our performance and latency needs:

1. Single Synchronous Pattern (Single Confirm) #

The producer sends one message, then calls a thread-blocking function (like WaitForConfirms) to wait for the broker’s ACK before sending the next message.

• Characteristics: Simple to write, but very poor performance because network round-trip (RTT) latency limits sending speed.

2. Batch Pattern (Batch Confirm) #

The producer sends a group of messages (e.g., 100 messages) sequentially, then calls a thread-blocking function once to wait for confirmation of all 100 messages.

• Characteristics: Much better throughput than the single pattern. However, if the broker sends a NACK for one message in the middle of the batch, the producer doesn’t know which message failed, so it’s forced to republish all 100 messages, triggering data duplication in the queue.

3. Asynchronous Pattern (Asynchronous Confirm) — Recommended #

The producer registers a listener callback function on the Go channel. The producer writes messages to the network non-blocking and continues its work. When the broker sends ACK/NACK, the listener goroutine catches it asynchronously.

• Characteristics: Maximum throughput and lowest latency. This pattern requires local memory management on the producer side to record the list of sequence numbers of messages still in-flight awaiting confirmation.

Go Code Implementation: Asynchronous Producer with Retry Handling #

Here is a complete Go language implementation example for an asynchronous producer. This code enables Publisher Confirms, tracks message sequence numbers using a mutex-protected local map data structure, listens for asynchronous confirmations non-blocking, and manages a retry resend queue when receiving NACK signals.

package main

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

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

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

// In-flight message tracker structure awaiting confirmation
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. Open a TCP Connection to 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. Enable Publisher Confirms Mode
	err = ch.Confirm(false) // false because we want asynchronous processing
	if err != nil {
		log.Fatalf("Gagal mengaktifkan Publisher Confirms: %s", err)
	}

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

	// Get the channel to receive confirmation notifications from the broker
	ackChan := make(chan amqp.Confirmation, 100)
	ch.NotifyPublish(ackChan)

	tracker := NewConfirmTracker()

	// 4. Asynchronous Confirmation Listener Goroutine
	go func() {
		for confirm := range ackChan {
			if confirm.Ack {
				// broker successfully received and secured the message
				log.Printf("✓ Menerima ACK dari broker untuk Sequence: %d (multiple: %t)", confirm.DeliveryTag, confirm.Ack)
				tracker.Confirm(confirm.DeliveryTag, confirm.Ack)
			} else {
				// broker failed (NACK)
				log.Printf("✗ Menerima NACK dari broker untuk Sequence: %d", confirm.DeliveryTag)
				msg, exists := tracker.GetAndRemove(confirm.DeliveryTag)
				if exists {
					// Run the retry strategy: resend that message
					log.Printf("[RETRY] Mempublikasikan ulang pesan sequence %d...", confirm.DeliveryTag)
					republishMessage(ch, tracker, msg)
				}
			}
		}
	}()

	// 5. Publish Messages 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),
		}

		// Lock the sequence number that will be obtained next before calling 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)
	}

	// Give time for all asynchronous confirmations to be received before 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-Patterns vs Practical Solutions in Production #

Avoid the following fatal publishing configuration mistakes to maintain cluster data reliability:

Anti-Pattern: Assuming a Successful TCP Connection Write Guarantees the Message Arrived #

Writing producer program logic that assumes an order transaction was successfully created just because the ch.PublishWithContext code line didn’t return a Go error (err == nil).

Why is this wrong? #

The PublishWithContext function returns success status as soon as the client library driver successfully writes the binary data to the local operating system TCP network socket. The client driver has no information about whether the network between the producer and broker is disconnected mid-journey, whether the destination Exchange is misspelled, or whether the RabbitMQ broker ran out of RAM. Without enabling Publisher Confirms, our messages are under the At-Most-Once spectrum where silent data loss very often occurs undetected by the producer’s logging system.

• Solution: Always enable asynchronous Publisher Confirms for all critical business transactions. If we need extreme performance, use the Transactional Outbox pattern where messages are stored in a local database table first before being pushed to RabbitMQ by a dedicated worker monitoring ACK/NACK receipts from the broker.

Summary #

• Producer-Side Reliability — Publisher Confirms is an active feedback protocol from the broker to the producer ensuring messages successfully arrive and are stored by the broker.
• Sequence Number Tracking — The broker assigns a linear sequence number per channel for every incoming message, matched with metadata on the producer side.
• ACK vs NACK Signals — The basic.ack signal is sent when the message is successfully secured. The basic.nack signal is sent if the destination queue is full or an internal broker error occurs.
• Interaction with Quorum Queues — On Quorum Queues, ACKs are only sent to producers after the message log is successfully written to the leader’s local disk and a majority of cluster followers.
• Asynchronous Confirm Is Best — The asynchronous pattern using callback listeners delivers maximum throughput performance and lowest latency compared to synchronous or batch confirm patterns.
• Preventing Silent Data Loss — Without Publisher Confirms, producers are blind to routing failures and broker outages, placing the system at severe data loss risk.

← Previous: Exactly-once   Next: Consumer Acknowledge →