Better error handling

mkuratczyk · mkuratczyk · commit 1456893d4aca · 2026-05-14T13:39:29.000+02:00
Don't report consumers as ready until subscription succeeded
diff --git a/pkg/amqp091/consumer.go b/pkg/amqp091/consumer.go
@@ -61,7 +61,7 @@ func (c *Amqp091Consumer) Connect() {
 	c.Channel = nil
 	c.Connection = nil
 
-	for c.Connection == nil {
+	utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
 		uri := utils.NextURI(c.Config.ConsumerUri, &c.whichUri)
 		dialCfg := amqp091.Config{
 			Properties: amqp091.Table{
@@ -75,34 +75,26 @@ func (c *Amqp091Consumer) Connect() {
 		conn, err := amqp091.DialConfig(uri, dialCfg)
 		if err != nil {
 			log.Error("consumer connection failed", "id", c.Id, "error", err.Error())
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-time.After(config.ReconnectDelay):
-				continue
-			}
-		} else {
-			log.Debug("consumer connected", "id", c.Id, "uri", uri)
-			c.Connection = conn
+			return false
 		}
+		log.Debug("consumer connected", "id", c.Id, "uri", uri)
+		c.Connection = conn
+		return true
+	})
+
+	if c.Connection == nil {
+		return
 	}
 
-	for c.Channel == nil {
+	utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
 		channel, err := c.Connection.Channel()
 		if err != nil {
-			if err == context.Canceled {
-				return
-			}
 			log.Error("consumer failed to create a channel", "id", c.Id, "error", err.Error())
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-time.After(config.ReconnectDelay):
-			}
-		} else {
-			c.Channel = channel
+			return false
 		}
-	}
+		c.Channel = channel
+		return true
+	})
 }
 
 func (c *Amqp091Consumer) Subscribe() {
@@ -165,7 +157,19 @@ func (c *Amqp091Consumer) Subscribe() {
 }
 
 func (c *Amqp091Consumer) Start(consumerReady chan bool) {
-	c.Subscribe()
+	if !utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
+		c.Subscribe()
+		if c.Messages != nil {
+			return true
+		}
+		log.Error("subscription failed, reconnecting", "id", c.Id, "terminus", c.Terminus)
+		c.Connect()
+		return false
+	}) {
+		close(consumerReady)
+		c.Stop("context cancelled")
+		return
+	}
 	close(consumerReady)
 	log.Info("consumer started", "id", c.Id, "terminus", c.Terminus)
 	var oooTracker *utils.OutOfOrderTracker
@@ -174,13 +178,13 @@ func (c *Amqp091Consumer) Start(consumerReady chan bool) {
 	}
 
 	for i := 1; i <= c.Config.ConsumeCount; {
-		for c.Messages == nil {
-			select {
-			case <-c.ctx.Done():
+		if c.Messages == nil {
+			if !utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
+				c.Subscribe()
+				return c.Messages != nil
+			}) {
 				c.Stop("context cancelled")
 				return
-			default:
-				c.Subscribe()
 			}
 		}
 
diff --git a/pkg/amqp10/consumer.go b/pkg/amqp10/consumer.go
@@ -65,7 +65,7 @@ func (c *Amqp10Consumer) Connect() {
 	c.Session = nil
 	c.Connection = nil
 
-	for c.Connection == nil {
+	utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
 		uri := utils.NextURI(c.Config.ConsumerUri, &c.whichUri)
 		hostname, vhost := hostAndVHost(uri)
 		conn, err := amqp.Dial(c.ctx, uri, &amqp.ConnOptions{
@@ -80,33 +80,26 @@ func (c *Amqp10Consumer) Connect() {
 		})
 		if err != nil {
 			log.Error("consumer failed to connect", "id", c.Id, "error", err.Error())
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-time.After(config.ReconnectDelay):
-			}
-		} else {
-			log.Debug("consumer connected", "id", c.Id, "uri", uri)
-			c.Connection = conn
+			return false
 		}
+		log.Debug("consumer connected", "id", c.Id, "uri", uri)
+		c.Connection = conn
+		return true
+	})
+
+	if c.Connection == nil {
+		return
 	}
 
-	for c.Session == nil {
+	utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
 		session, err := c.Connection.NewSession(c.ctx, nil)
 		if err != nil {
-			if err == context.Canceled {
-				return
-			}
 			log.Error("consumer failed to create a session", "id", c.Id, "error", err.Error())
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-time.After(config.ReconnectDelay):
-			}
-		} else {
-			c.Session = session
+			return false
 		}
-	}
+		c.Session = session
+		return true
+	})
 }
 
 func (c *Amqp10Consumer) CreateReceiver(ctx context.Context) {
@@ -120,53 +113,45 @@ func (c *Amqp10Consumer) CreateReceiver(ctx context.Context) {
 		durability = amqp.DurabilityUnsettledState
 	}
 
-	for c.Receiver == nil && c.Session != nil {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			linkProperties := buildLinkProperties(c.Config, c.Id)
-			receiverOpts := &amqp.ReceiverOptions{
-				SourceDurability:          durability,
-				Credit:                    int32(c.Config.ConsumerCredits),
-				Properties:                linkProperties,
-				Filters:                   buildLinkFilters(c.Config),
-				RequestedSenderSettleMode: requestedSenderSettleMode(c.Config),
-				OnLinkStateProperties: func(props map[string]any) {
-					if active, ok := props["rabbitmq:active"]; ok {
-						if activeBool, ok := active.(bool); ok {
-							logArgs := []any{"id", c.Id}
-							if priority, ok := linkProperties["rabbitmq:priority"]; ok {
-								logArgs = append(logArgs, "priority", priority)
-							}
-							if activeBool {
-								log.Info("consumer is active", logArgs...)
-							} else {
-								log.Info("consumer is not active", logArgs...)
-							}
-						}
+	linkProperties := buildLinkProperties(c.Config, c.Id)
+	receiverOpts := &amqp.ReceiverOptions{
+		SourceDurability:          durability,
+		Credit:                    int32(c.Config.ConsumerCredits),
+		Properties:                linkProperties,
+		Filters:                   buildLinkFilters(c.Config),
+		RequestedSenderSettleMode: requestedSenderSettleMode(c.Config),
+		OnLinkStateProperties: func(props map[string]any) {
+			if active, ok := props["rabbitmq:active"]; ok {
+				if activeBool, ok := active.(bool); ok {
+					logArgs := []any{"id", c.Id}
+					if priority, ok := linkProperties["rabbitmq:priority"]; ok {
+						logArgs = append(logArgs, "priority", priority)
+					}
+					if activeBool {
+						log.Info("consumer is active", logArgs...)
+					} else {
+						log.Info("consumer is not active", logArgs...)
 					}
-				},
-			}
-			if c.Config.Amqp.Browse {
-				receiverOpts.SourceDistributionMode = "copy"
-			}
-			receiver, err := c.Session.NewReceiver(ctx, c.Terminus, receiverOpts)
-			if err != nil {
-				if err == context.Canceled {
-					return
-				}
-				log.Error("consumer failed to create a receiver", "id", c.Id, "error", err.Error())
-				select {
-				case <-ctx.Done():
-					return
-				case <-time.After(config.ReconnectDelay):
 				}
-			} else {
-				c.Receiver = receiver
 			}
-		}
+		},
+	}
+	if c.Config.Amqp.Browse {
+		receiverOpts.SourceDistributionMode = "copy"
 	}
+
+	utils.Retry(ctx, config.ReconnectDelay, func() bool {
+		if c.Session == nil {
+			return true // session gone; caller must reconnect
+		}
+		receiver, err := c.Session.NewReceiver(ctx, c.Terminus, receiverOpts)
+		if err != nil {
+			log.Error("consumer failed to create a receiver", "id", c.Id, "error", err.Error())
+			return false
+		}
+		c.Receiver = receiver
+		return true
+	})
 }
 
 func (c *Amqp10Consumer) Start(consumerReady chan bool) {
diff --git a/pkg/stomp/consumer.go b/pkg/stomp/consumer.go
@@ -58,7 +58,7 @@ func (c *StompConsumer) Connect() {
 	c.Subscription = nil
 	c.Connection = nil
 
-	for c.Connection == nil {
+	utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
 		uri := utils.NextURI(c.Config.ConsumerUri, &c.whichUri)
 		useTLS := strings.HasPrefix(uri, "stomp+ssl://") || strings.HasPrefix(uri, "stomps://")
 		parsedUri := utils.ParseURI(uri, "stomp", "61613")
@@ -69,9 +69,9 @@ func (c *StompConsumer) Connect() {
 		}
 
 		log.Debug("connecting to broker", "id", c.Id, "broker", parsedUri.Broker)
+		dialer := &net.Dialer{Timeout: dialTimeout}
 		var conn *stomp.Conn
 		var err error
-		dialer := &net.Dialer{Timeout: dialTimeout}
 		if useTLS {
 			netConn, tlsErr := tls.DialWithDialer(dialer, "tcp", parsedUri.Broker, &tls.Config{
 				InsecureSkipVerify: c.Config.InsecureSkipTLSVerify,
@@ -95,19 +95,13 @@ func (c *StompConsumer) Connect() {
 				}
 			}
 		}
-
 		if err != nil {
 			log.Error("consumer connection failed", "id", c.Id, "error", err.Error())
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-time.After(config.ReconnectDelay):
-				continue
-			}
-		} else {
-			c.Connection = conn
+			return false
 		}
-	}
+		c.Connection = conn
+		return true
+	})
 }
 
 func (c *StompConsumer) Subscribe() {
@@ -125,7 +119,19 @@ func (c *StompConsumer) Subscribe() {
 }
 
 func (c *StompConsumer) Start(consumerReady chan bool) {
-	c.Subscribe()
+	if !utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
+		c.Subscribe()
+		if c.Subscription != nil {
+			return true
+		}
+		log.Error("subscription failed, reconnecting", "id", c.Id, "destination", c.Topic)
+		c.Connect()
+		return false
+	}) {
+		close(consumerReady)
+		c.Stop("context cancelled")
+		return
+	}
 	close(consumerReady)
 	log.Info("consumer started", "id", c.Id, "destination", c.Topic)
 
@@ -135,13 +141,13 @@ func (c *StompConsumer) Start(consumerReady chan bool) {
 	}
 
 	for i := 1; i <= c.Config.ConsumeCount; {
-		for c.Subscription == nil {
-			select {
-			case <-c.ctx.Done():
+		if c.Subscription == nil {
+			if !utils.Retry(c.ctx, config.ReconnectDelay, func() bool {
+				c.Subscribe()
+				return c.Subscription != nil
+			}) {
 				c.Stop("context cancelled")
 				return
-			default:
-				c.Subscribe()
 			}
 		}
 
diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
@@ -2,6 +2,7 @@ package utils
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"net/url"
@@ -410,6 +411,22 @@ func (t *OutOfOrderTracker) Check(publisherID int, seq uint64) SequenceCheckResu
 	return SequenceCheckResult{Status: SequenceOK, LastSeq: lastSeq}
 }
 
+// Retry calls f repeatedly until f returns true or ctx is cancelled.
+// A fixed delay is inserted between each failed attempt.
+// Returns true if f succeeded, false if ctx was cancelled first.
+func Retry(ctx context.Context, delay time.Duration, f func() bool) bool {
+	for {
+		if f() {
+			return true
+		}
+		select {
+		case <-ctx.Done():
+			return false
+		case <-time.After(delay):
+		}
+	}
+}
+
 // PastTense converts message outcome verbs to their past tense form for logging.
 func PastTense(outcome string) string {
 	switch outcome {
