Reliable Messaging with Redelivery

Introduction to the Lesson

Today, we address the "unhappy path." What happens if a consumer receives a message but crashes, gets stuck, or loses network connectivity before it can send the Ack? In our current implementation, the broker would wait forever, and the message would be effectively lost.

This lesson directly implements the learning outcome: Implement a redelivery mechanism for messages that are not acknowledged within a configured timeout.

By building this, we will complete the core logic for our "at-least-once" delivery guarantee. The broker will no longer just hope for an acknowledgment; it will actively monitor for it and take corrective action if it doesn't arrive.

We will cover:

1. The fundamental distributed systems principles of timeouts and retries.
2. The critical related concept of idempotency.
3. Architectural patterns for tracking in-flight messages, from simple to production-grade.
4. A practical Go implementation using a min-heap to efficiently manage message timeouts.

By the end of this session, our broker will be resilient to non-responsive consumers, automatically redelivering messages to ensure they are eventually processed.

1. Conceptual Foundation: Timeouts, Retries, and Reliability

At its core, a message sent to a consumer is a request that awaits a response (the Ack). In any distributed system, that response might never come. Building reliable systems means anticipating and handling this reality.

The following video provides an excellent, high-level overview of the core patterns for handling these challenges. As someone with a systems design interest, you'll find it frames our technical problem within broader architectural principles.

Key Takeaways for Our Broker:

• Detection (Timeout): When we send a message to a consumer, we start a timer. This timer isn't just for the consumer's benefit; it protects the broker from having resources (the message state) tied up indefinitely.
• Action (Redelivery): If the timer expires before we receive an Ack, we assume the consumer failed and redeliver the message. This ensures progress can continue.
• Consequence (Idempotency): Redelivery means a consumer might receive the same message more than once. For example, it might process the message, crash just before sending the Ack, and then receive the redelivered message upon restart. A truly robust consumer application would need to be idempotent—designed to handle duplicate messages without causing incorrect side effects (like charging a credit card twice). While we are focused on the broker today, this is a crucial concept to understand in the overall system design.

This flow is visualized perfectly in the sequence diagram below.

2. Designing the Timeout Tracking Mechanism

How can the broker efficiently track potentially thousands of in-flight messages, each with its own timeout?

A naive approach would be to store them in a list and periodically scan the entire list for expired messages. This is highly inefficient and doesn't scale. We need a data structure designed for this purpose.

Pattern 1: Priority Queue (Min-Heap)

A classic and effective solution is to use a priority queue, which is commonly implemented with a min-heap.

• When we send a message, we insert an object into the heap containing the message details and its expiration time (time.Now() + timeout).
• The heap is ordered by expiration time, so the message that will expire next is always at the root.
• A single background goroutine can then Peek at the root element, calculate the time.Sleep duration until it expires, and wake up to process it.

The Go standard library provides the container/heap package, making this straightforward to implement.

Pattern 2: Timer Wheel

For even higher performance at a massive scale (millions of timers), systems like Kafka and Netty use a Timer Wheel. This is a more complex data structure that avoids the heap's O(log N) complexity for insertions and deletions, achieving O(1) amortized time.

Pattern 3: Externalized Tracking via a Markers Topic

A third, very robust pattern involves externalizing the tracking logic. This is used by Splunk's kafka-mq-go library to add message-queue semantics on top of Kafka.

Our Decision: We will implement the min-heap approach. It achieves our learning goal perfectly, integrates cleanly into our existing Coordinator, and is an excellent exercise in using Go's standard library for a common systems programming task.

3. Implementation in Go

Let's integrate the min-heap into our Coordinator. The logic will be split into three parts: adding a message to track, removing it upon acknowledgment, and redelivering it on timeout.

Step 1: Define the Data Structures

First, in your coordinator package, define the structure for in-flight messages and the heap itself.

// coordinator/inflight.go

package coordinator

import (
	"container/heap"
	"time"

	"your_project/internal/protocol"
)

// InFlightMessage tracks a message that has been sent to a consumer but not yet acknowledged.
type InFlightMessage struct {
	Msg       *protocol.Message
	ExpiresAt time.Time
	// We need these to know where to redeliver or look up.
	GroupID   string
	Topic     string
	Partition int32

	// The index is needed by heap.Interface and is managed by the heap container.
	index int
}

// InFlightHeap is a min-heap of in-flight messages, ordered by expiration time.
type InFlightHeap []*InFlightMessage

func (h InFlightHeap) Len() int           { return len(h) }
func (h InFlightHeap) Less(i, j int) bool { return h[i].ExpiresAt.Before(h[j].ExpiresAt) }
func (h InFlightHeap) Swap(i, j int) {
	h[i], h[j] = h[j], h[i]
	h[i].index = i
	h[j].index = j
}

func (h *InFlightHeap) Push(x interface{}) {
	n := len(*h)
	item := x.(*InFlightMessage)
	item.index = n
	*h = append(*h, item)
}

func (h *InFlightHeap) Pop() interface{} {
	old := *h
	n := len(old)
	item := old[n-1]
	old[n-1] = nil  // avoid memory leak
	item.index = -1 // for safety
	*h = old[0 : n-1]
	return item
}

Step 2: Update the Coordinator

Now, augment the Coordinator to use this heap. We also need a map for efficient removal of messages when an Ack arrives.

// coordinator/coordinator.go

import (
    "container/heap"
    "fmt"
    "sync"
    "time"
)

type Coordinator struct {
	// ... existing fields
	
	mu              sync.Mutex
	inFlight        InFlightHeap
	// Map for O(1) lookup of in-flight messages by their unique ID.
	// Key: e.g., "topic-partition-offset"
	inFlightMap     map[string]*InFlightMessage
	redeliveryTimeout time.Duration
}

func NewCoordinator(storage Storage, redeliveryTimeout time.Duration) *Coordinator {
	c := &Coordinator{
		// ...
		inFlightMap:     make(map[string]*InFlightMessage),
		redeliveryTimeout: redeliveryTimeout,
	}
	heap.Init(&c.inFlight)
	go c.checkTimeouts() // Start the background checker
	return c
}

// Helper to generate a unique key for an in-flight message
func inFlightKey(topic string, partition int32, offset int64) string {
    return fmt.Sprintf("%s-%d-%d", topic, partition, offset)
}

Step 3: Implement the Core Logic

Tracking a message: When a message is sent to a consumer, call a new TrackInFlightMessage method.

// coordinator/coordinator.go

func (c *Coordinator) TrackInFlightMessage(msg *protocol.Message, groupID, topic string, partition int32) {
    c.mu.Lock()
    defer c.mu.Unlock()

    item := &InFlightMessage{
        Msg:       msg,
        ExpiresAt: time.Now().Add(c.redeliveryTimeout),
        GroupID:   groupID,
        Topic:     topic,
        Partition: partition,
    }

    key := inFlightKey(topic, partition, msg.Offset)
    c.inFlightMap[key] = item
    heap.Push(&c.inFlight, item)
}

Handling an Ack: Update HandleAck to remove the message from tracking before committing the offset.

// coordinator/coordinator.go

func (c *Coordinator) HandleAck(ctx context.Context, ack *protocol.Ack) {
    c.mu.Lock()
    key := inFlightKey(ack.Topic, ack.Partition, ack.Offset)
    item, ok := c.inFlightMap[key]
    if ok {
        // Remove from heap and map
        heap.Remove(&c.inFlight, item.index)
        delete(c.inFlightMap, key)
    }
    c.mu.Unlock()

    if !ok {
        // This can happen if the ack arrives after a message was already redelivered.
        // It's a normal race condition, so we just log it and ignore.
        log.Printf("Received ack for untracked or expired message: %s", key)
        return
    }

    // Proceed with offset commit as before
    // ...
}

Checking for Timeouts: This is the background goroutine's work.

// coordinator/coordinator.go

func (c *Coordinator) checkTimeouts() {
    // Check every second, or use a dynamic sleep based on the next expiry time.
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        c.mu.Lock()
        
        for c.inFlight.Len() > 0 {
            // Peek at the next message to expire
            next := c.inFlight[0]
            if time.Now().Before(next.ExpiresAt) {
                // Not expired yet, so nothing else in the heap is either.
                break
            }

            // It's expired. Pop it from the heap.
            item := heap.Pop(&c.inFlight).(*InFlightMessage)
            key := inFlightKey(item.Topic, item.Partition, item.Msg.Offset)
            delete(c.inFlightMap, key)

            log.Printf("Timeout for message %s. Redelivering.", key)
            
            // TODO: Trigger redelivery logic
            c.redeliver(item)
        }
        
        c.mu.Unlock()
    }
}

func (c *Coordinator) redeliver(item *InFlightMessage) {
    // Redelivery logic depends on your consumer group implementation.
    // The goal is to make this message available again to the group.
    
    // Find the consumer group
    group, ok := c.groups[item.GroupID]
    if !ok {
        return // Group is gone
    }
    
    // Find the partition assignment
    consumer, ok := group.consumersByPartition[item.Partition]
    if !ok {
        return // Partition no longer assigned to this group
    }

    // Re-queue the message for the consumer.
    // This will put it at the front of their personal queue.
    // Use a non-blocking send in case the consumer's channel is full.
    select {
    case consumer.messages <- item.Msg:
        log.Printf("Redelivered message %s to consumer %s", inFlightKey(item.Topic, item.Partition, item.Msg.Offset), consumer.id)
        // VERY IMPORTANT: We must re-track the redelivered message with a new timeout!
        c.TrackInFlightMessage(item.Msg, item.GroupID, item.Topic, item.Partition)
    default:
        // Consumer channel is full, they are likely stuck.
        // In a more advanced system, you might try another consumer
        // or disconnect this one. For now, we'll just log.
        log.Printf("Failed to redeliver message to consumer %s: channel full", consumer.id)
        // We could try to put it back in the heap with a small delay.
    }
}

Conclusion

Excellent work! You have now implemented a crucial piece of systems engineering that transforms our broker from a best-effort toy into a genuinely reliable component. By detecting and reacting to consumer timeouts, we ensure messages are not lost, completing our at-least-once delivery guarantee.

Key Takeaways:

• Timeouts are essential for system health, allowing the broker to reclaim resources and detect failed clients.
• Redelivery is the corrective action that ensures message processing can continue despite consumer failures.
• A min-heap is a highly efficient data structure (O(log N)) for managing active timers, far superior to linear scanning.
• Implementing redelivery requires careful state management: tracking in-flight messages, removing them upon success (Ack), and re-queuing them upon failure (timeout).
• Reliable delivery in the broker places a new responsibility on clients: designing them to be idempotent to handle potential duplicates gracefully.

Next Up

Our broker is now much more reliable, but it can still be overwhelmed. If producers send messages faster than consumers can process them, our broker's memory usage will grow indefinitely until it crashes.

In the next lesson, we will tackle this by introducing backpressure. We will use buffered channels to create a bounded work queue for incoming produce requests, giving the broker a mechanism to signal to producers that it is temporarily overloaded.