From d8e2b726f77f37ac8ab033c6a9a190d2d3b01f8e Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 07:59:24 +0530 Subject: [PATCH 1/2] feat(consumer): declarative topology restored on reconnect + fix consume-loop wedge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add ConsumerConfig.WithQueueConfig and WithBinding (new BindingConfig type): the consumer re-applies the configured queue and bindings on every channel setup — initially and after each reconnect. Previously a named exclusive/auto-delete queue was deleted by the broker on connection loss and never re-created or re-bound, silently killing the consumer. Fix the consume loop wedging forever when consuming fails while the connection is healthy (queue deleted / basic.cancel / precondition failure): no reconnect signal ever arrives, so waitForReconnect now also retries setup on a timer. Also close the previous channel when re-establishing on a healthy connection instead of leaking it, and drop a channel set up concurrently with Close. --- CHANGELOG.md | 29 +++++++ README.md | 25 ++++++ consumer.go | 158 +++++++++++++++++++++++++++++----- consumer_test.go | 59 +++++++++++++ integration_test.go | 201 ++++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 442 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0c2b3b..ca79242 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,35 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.5.0] - 2026-07-05 + +### Added + +- **Declarative consumer topology that survives reconnects.** + `ConsumerConfig.WithQueueConfig(QueueConfig)` and + `ConsumerConfig.WithBinding(exchange, routingKey, args)` (new `BindingConfig` + type) declare the consumed queue and its bindings as configuration. The + consumer re-applies them on **every** channel setup — initially and after + each reconnect — so caller-declared queues and bindings are restored + automatically after a connection loss. Previously, a named + exclusive/auto-delete queue was deleted by the broker when the connection + dropped and never re-created or re-bound, silently killing the consumer. + `WithBinding` also works with server-named (empty-name) queues, removing the + old "re-bind after a reconnect" caveat. When `QueueConfig` is set, its + `Name` takes precedence over `ConsumerConfig.Queue`. + +### Fixed + +- **The consume loop no longer wedges permanently when no reconnect signal is + coming.** Consuming can fail while the connection is healthy — queue deleted + (server-sent `basic.cancel`), `NOT_FOUND`, precondition failure — and the + loop previously blocked forever waiting for a reconnect signal that would + never arrive. It now retries channel setup every 5 seconds in addition to + reacting to reconnect signals. +- Re-establishing the consumer channel while the connection is healthy no + longer leaks the previous channel; it is closed when replaced. A channel + set up concurrently with `Close` is also closed instead of leaking. + ## [0.4.0] - 2026-07-02 ### Fixed diff --git a/README.md b/README.md index 0230eea..876b7e0 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ Requires Go 1.22+. ## Features - **Auto-reconnection** with exponential backoff for connections, publishers, and consumers +- **Declarative consumer topology** — queues and bindings restored automatically after reconnects - **Publisher confirms** for reliable message delivery - **Consumer middleware** (logging, recovery, retry — or bring your own) - **Concurrent consumers** with configurable worker goroutines @@ -222,6 +223,30 @@ err = consumer.Consume(ctx, func(ctx context.Context, d *rabbitmq.Delivery) erro Consumers automatically resume consuming after the connection recovers. +### Declarative Topology (survives reconnection) + +If the consumer's queue or bindings can be lost when the connection drops +(exclusive or auto-delete queues, bindings on server-named queues), declare +them as configuration instead of calling `DeclareQueue`/`BindQueue` manually. +The consumer re-applies this topology on every channel setup — initially and +after each reconnect: + +```go +consConfig := rabbitmq.DefaultConsumerConfig(). + WithQueueConfig(rabbitmq.DefaultQueueConfig("ws-fanout"). + WithDurable(false). + WithAutoDelete(true). + WithExclusive(true)). + WithBinding("events", "user.*", nil) + +consumer, err := rabbitmq.NewConsumer(conn, consConfig) +``` + +After a broker restart or network blip, the queue is re-declared and re-bound +automatically and consumption resumes. `WithBinding` also works for +server-named queues (empty queue name), which get a fresh name on each +reconnect. The bound exchange must already exist when the consumer is created. + ### Concurrent Consumers Process messages in parallel with multiple worker goroutines: diff --git a/consumer.go b/consumer.go index d82c777..bccc873 100644 --- a/consumer.go +++ b/consumer.go @@ -54,6 +54,15 @@ type ConsumerConfig struct { // Middleware is applied to the message handler in order. Middleware []Middleware + + // QueueConfig, when set, is declared on every channel setup — initially + // and after each reconnect — before consuming, so the queue and its + // arguments survive connection loss. Its Name takes precedence over Queue. + QueueConfig *QueueConfig + + // Bindings are applied to the consumed queue on every channel setup, + // initially and after each reconnect. + Bindings []BindingConfig } // DefaultConsumerConfig returns a default consumer configuration. @@ -136,12 +145,37 @@ func (c ConsumerConfig) WithMiddleware(mw ...Middleware) ConsumerConfig { return c } +// WithQueueConfig returns a new config that declares the given queue on every +// channel setup — initially and after each reconnect — so the queue and its +// arguments are restored after a connection loss. The name in qc takes +// precedence over any name set with WithQueue; an empty name declares a fresh +// server-named queue on each setup. +func (c ConsumerConfig) WithQueueConfig(qc QueueConfig) ConsumerConfig { + c.QueueConfig = &qc + c.Queue = qc.Name + return c +} + +// WithBinding returns a new config that binds the consumed queue to the given +// exchange on every channel setup — initially and after each reconnect. Call +// it multiple times to add several bindings. The exchange must exist when the +// consumer is created, and so must the queue if this consumer does not declare +// it (a non-empty queue name without WithQueueConfig). +func (c ConsumerConfig) WithBinding(exchange, routingKey string, args map[string]any) ConsumerConfig { + // Copy before appending so config copies never share a backing array. + bindings := make([]BindingConfig, len(c.Bindings), len(c.Bindings)+1) + copy(bindings, c.Bindings) + bindings = append(bindings, BindingConfig{Exchange: exchange, RoutingKey: routingKey, Args: args}) + c.Bindings = bindings + return c +} + // Consumer consumes messages from RabbitMQ. type Consumer struct { conn *Connection channel *Channel config ConsumerConfig - serverNamed bool // config.Queue was empty: declare a fresh server-named queue each setup + serverNamed bool // resolved queue name was empty: declare a fresh server-named queue each setup queue string // resolved queue name actually consumed from mu sync.RWMutex closed bool @@ -153,20 +187,30 @@ type Consumer struct { // NewConsumer creates a new consumer. // -// An empty config.Queue is allowed: the consumer declares a private, server-named -// queue (exclusive, auto-delete) and consumes from it. Read the assigned name -// with QueueName. Such a queue is re-declared with a new name on reconnect, so if -// you bind it to an exchange, re-bind after a reconnect. +// An empty queue name (config.Queue, or QueueConfig.Name when set) is allowed: +// the consumer declares a private, server-named queue (exclusive, auto-delete) +// and consumes from it. Read the assigned name with QueueName. +// +// Topology configured with WithQueueConfig and WithBinding is re-applied on +// every channel setup — initially and after each reconnect — so declared +// queues and bindings survive connection loss without manual re-declaration. func NewConsumer(conn *Connection, config ConsumerConfig) (*Consumer, error) { if conn == nil { return nil, ErrNilConnection } + // QueueConfig.Name wins over config.Queue (WithQueueConfig keeps them in + // sync, but the fields can also be set directly). + queueName := config.Queue + if config.QueueConfig != nil { + queueName = config.QueueConfig.Name + } + c := &Consumer{ conn: conn, config: config, - serverNamed: config.Queue == "", - queue: config.Queue, + serverNamed: queueName == "", + queue: queueName, reconnectCh: conn.subscribeReconnect(), log: conn.log, } @@ -179,7 +223,9 @@ func NewConsumer(conn *Connection, config ConsumerConfig) (*Consumer, error) { return c, nil } -// setupChannel creates a new channel and sets QoS. +// setupChannel creates a new channel, sets QoS, and applies the configured +// topology. It runs on initial setup and after every reconnect, replacing +// (and closing) any previous channel. func (c *Consumer) setupChannel() error { ch, err := c.conn.Channel() if err != nil { @@ -191,25 +237,67 @@ func (c *Consumer) setupChannel() error { return fmt.Errorf("failed to set QoS: %w", err) } + queueName, err := c.applyTopology(ch) + if err != nil { + _ = ch.Close() + return err + } + + c.mu.Lock() + if c.closed { + // The consumer was closed while this setup was in flight (e.g. a + // retry-timer race with Close); don't install a channel nobody will + // ever close. + c.mu.Unlock() + _ = ch.Close() + return ErrShuttingDown + } + old := c.channel + c.channel = ch + c.queue = queueName + c.mu.Unlock() + + // Close the channel being replaced; when re-setup happens on a healthy + // connection (e.g. after the queue was deleted) it would otherwise stay + // open and leak. + if old != nil { + _ = old.Close() + } + + return nil +} + +// applyTopology declares the configured queue and applies the configured +// bindings on the given channel, returning the resolved queue name. +func (c *Consumer) applyTopology(ch *Channel) (string, error) { queueName := c.config.Queue - if c.serverNamed { + + switch { + case c.config.QueueConfig != nil: + qc := c.config.QueueConfig + q, err := ch.ch.QueueDeclare(qc.Name, qc.Durable, qc.AutoDelete, qc.Exclusive, false /*noWait*/, amqp.Table(qc.buildArgs())) + if err != nil { + return "", fmt.Errorf("declare queue %q: %w", qc.Name, err) + } + queueName = q.Name + case c.serverNamed: // No queue named: declare a private, server-named queue that lives and // dies with this connection (exclusive + auto-delete). The previous one // is gone after a reconnect, so re-declare to get a fresh name. q, err := ch.ch.QueueDeclare("", false /*durable*/, true /*autoDelete*/, true /*exclusive*/, false /*noWait*/, nil) if err != nil { - _ = ch.Close() - return fmt.Errorf("declare server-named queue: %w", err) + return "", fmt.Errorf("declare server-named queue: %w", err) } queueName = q.Name } - c.mu.Lock() - c.channel = ch - c.queue = queueName - c.mu.Unlock() + for _, b := range c.config.Bindings { + if err := ch.ch.QueueBind(queueName, b.RoutingKey, b.Exchange, false /*noWait*/, amqp.Table(b.Args)); err != nil { + return "", fmt.Errorf("bind queue %q to exchange %q: %w", queueName, b.Exchange, err) + } + } - return nil + return queueName, nil } // QueueName returns the queue this consumer reads from. When NewConsumer was @@ -240,9 +328,16 @@ func (c *Consumer) Start(ctx context.Context) (<-chan *Delivery, error) { return outCh, nil } -// waitForReconnect waits for a reconnection signal or context cancellation. -// Returns true if reconnection was signaled (caller should continue the loop), -// or false if the loop should exit. +// consumeRetryDelay is how long the consume loop waits before retrying channel +// setup when no reconnect signal arrives. Consuming can fail while the +// connection is healthy (queue deleted, server-sent basic.cancel, precondition +// failure), in which case no reconnect signal will ever come. A variable so +// tests can shorten it. +var consumeRetryDelay = 5 * time.Second + +// waitForReconnect waits for a reconnection signal, a retry timeout, or +// context cancellation, then re-establishes the channel. Returns true if the +// caller should continue the consume loop, or false if it should exit. func (c *Consumer) waitForReconnect(ctx context.Context) bool { select { case <-ctx.Done(): @@ -251,11 +346,15 @@ func (c *Consumer) waitForReconnect(ctx context.Context) bool { if !ok { return false } - if err := c.setupChannel(); err != nil { - c.log.Errorf("consumer: failed to re-establish channel: %v", err) - } - return true + case <-time.After(consumeRetryDelay): + // No reconnect signal is coming if the connection is healthy but the + // queue is gone — retry setup on a timer so the loop never blocks + // forever. } + if err := c.setupChannel(); err != nil { + c.log.Errorf("consumer: failed to re-establish channel: %v", err) + } + return true } // consumeLoop runs the consume loop, automatically recovering on reconnection. @@ -584,6 +683,19 @@ type QueueInfo struct { Consumers int } +// BindingConfig describes a queue-to-exchange binding that the consumer +// applies on every channel setup (see ConsumerConfig.WithBinding). +type BindingConfig struct { + // Exchange is the exchange to bind to. + Exchange string + + // RoutingKey is the binding routing key. + RoutingKey string + + // Args are additional binding arguments. + Args map[string]any +} + // QueueConfig holds queue declaration configuration. type QueueConfig struct { // Name is the queue name. diff --git a/consumer_test.go b/consumer_test.go index dd7324f..86ee3d2 100644 --- a/consumer_test.go +++ b/consumer_test.go @@ -317,3 +317,62 @@ func TestQueueConfigWithQuorum(t *testing.T) { t.Errorf("expected x-queue-type=quorum, got %v", queueType) } } + +func TestConsumerConfigWithQueueConfig(t *testing.T) { + qc := DefaultQueueConfig("topo-q").WithAutoDelete(true).WithExclusive(true) + c := DefaultConsumerConfig().WithQueue("old-name").WithQueueConfig(qc) + + if c.QueueConfig == nil { + t.Fatal("expected QueueConfig to be set") + } + if c.QueueConfig.Name != "topo-q" { + t.Errorf("expected QueueConfig name topo-q, got %s", c.QueueConfig.Name) + } + if c.Queue != "topo-q" { + t.Errorf("expected Queue synced to topo-q, got %s", c.Queue) + } + + // The config stores a copy: mutating the original must not leak in. + qc.Name = "mutated" + if c.QueueConfig.Name != "topo-q" { + t.Errorf("expected stored copy to stay topo-q, got %s", c.QueueConfig.Name) + } +} + +func TestConsumerConfigWithBinding(t *testing.T) { + c := DefaultConsumerConfig(). + WithQueue("q"). + WithBinding("ex-1", "key.1", nil). + WithBinding("ex-2", "key.2", map[string]any{"x-match": "all"}) + + if len(c.Bindings) != 2 { + t.Fatalf("expected 2 bindings, got %d", len(c.Bindings)) + } + if c.Bindings[0].Exchange != "ex-1" || c.Bindings[0].RoutingKey != "key.1" { + t.Errorf("unexpected first binding: %+v", c.Bindings[0]) + } + if c.Bindings[1].Exchange != "ex-2" || c.Bindings[1].RoutingKey != "key.2" { + t.Errorf("unexpected second binding: %+v", c.Bindings[1]) + } + if c.Bindings[1].Args["x-match"] != "all" { + t.Errorf("unexpected second binding args: %+v", c.Bindings[1].Args) + } +} + +func TestConsumerConfigWithBindingCopySemantics(t *testing.T) { + base := DefaultConsumerConfig().WithQueue("q").WithBinding("ex", "base", nil) + + // Two configs diverging from the same base must not share bindings. + a := base.WithBinding("ex", "a", nil) + b := base.WithBinding("ex", "b", nil) + + if len(base.Bindings) != 1 { + t.Errorf("base mutated: expected 1 binding, got %d", len(base.Bindings)) + } + if a.Bindings[1].RoutingKey != "a" { + t.Errorf("expected a's second binding to be a, got %s", a.Bindings[1].RoutingKey) + } + if b.Bindings[1].RoutingKey != "b" { + t.Errorf("expected b's second binding to be b, got %s", b.Bindings[1].RoutingKey) + } +} diff --git a/integration_test.go b/integration_test.go index 9a0a874..8deb233 100644 --- a/integration_test.go +++ b/integration_test.go @@ -1807,7 +1807,7 @@ func TestIntegration_AnonymousQueueReconnect(t *testing.T) { // The consumer re-declares a fresh server-named queue on reconnect, so its // reported name changes; then delivery must resume on the new queue. newName := waitForReDeclaredQueue(t, consumer, origName) - publishUntilDelivered(t, ctx, pub, deliveryCh, newName, "after") + publishUntilDelivered(t, ctx, pub, deliveryCh, "", newName, "after") } // recvDelivery waits for one delivery, asserts its body equals want, and acks it. @@ -1843,17 +1843,21 @@ func waitForReDeclaredQueue(t *testing.T, consumer *Consumer, origName string) s return "" } -// publishUntilDelivered re-publishes want to queue until a matching delivery -// arrives, covering the brief window between queue re-declaration and the -// consume being re-established. It fails the test if ctx expires first. -func publishUntilDelivered(t *testing.T, ctx context.Context, pub *Publisher, deliveryCh <-chan *Delivery, queue, want string) { +// publishUntilDelivered re-publishes want to exchange/routingKey until a +// matching delivery arrives, covering the window between topology restoration +// and the consume being re-established. Transient publish errors (e.g. the +// publisher channel still re-establishing after a reconnect) are retried. +// It fails the test if ctx expires first. +func publishUntilDelivered(t *testing.T, ctx context.Context, pub *Publisher, deliveryCh <-chan *Delivery, exchange, routingKey, want string) { t.Helper() for { if ctx.Err() != nil { t.Fatalf("timed out waiting for %q", want) } - if err := pub.PublishToExchange(ctx, "", queue, NewTextMessage(want)); err != nil { - t.Fatalf("publish %q: %v", want, err) + if err := pub.PublishToExchange(ctx, exchange, routingKey, NewTextMessage(want)); err != nil { + // The publisher may still be re-establishing its channel; retry. + time.Sleep(250 * time.Millisecond) + continue } select { case d, ok := <-deliveryCh: @@ -1870,3 +1874,186 @@ func publishUntilDelivered(t *testing.T, ctx context.Context, pub *Publisher, de } } } + +// --- Declarative Topology --- + +func TestIntegration_DeclarativeTopology(t *testing.T) { + conn := integrationConn(t) + queue := uniqueQueue(t) + exchange := queue + ".ex" + + pub, err := NewPublisher(conn, DefaultPublisherConfig(). + WithExchange(exchange). + WithRoutingKey("evt.test")) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + if err := pub.DeclareExchange(exchange, ExchangeTopic, false, false, nil); err != nil { + t.Fatalf("declare exchange: %v", err) + } + t.Cleanup(func() { deleteExchange(t, conn, exchange) }) + + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(queue). + WithDurable(false). + WithAutoDelete(true). + WithExclusive(true)). + WithBinding(exchange, "evt.*", nil) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + + if consumer.QueueName() != queue { + t.Errorf("QueueName() = %q, want %q", consumer.QueueName(), queue) + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + deliveryCh, err := consumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start consumer: %v", err) + } + + if err := pub.Publish(ctx, NewTextMessage("via-binding")); err != nil { + t.Fatalf("publish: %v", err) + } + recvDelivery(t, ctx, deliveryCh, "via-binding") +} + +// TestIntegration_TopologyRestoredAfterReconnect verifies that a NAMED +// exclusive+auto-delete queue declared via WithQueueConfig, plus its binding, +// is restored after a connection loss. The broker deletes such a queue when +// the connection dies; without declarative topology the consumer would fail +// with NOT_FOUND on reconnect and stall forever. +func TestIntegration_TopologyRestoredAfterReconnect(t *testing.T) { + conn := integrationConn(t) + queue := uniqueQueue(t) + exchange := queue + ".ex" + + pub, err := NewPublisher(conn, DefaultPublisherConfig(). + WithExchange(exchange). + WithRoutingKey("evt.test")) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + if err := pub.DeclareExchange(exchange, ExchangeTopic, false, false, nil); err != nil { + t.Fatalf("declare exchange: %v", err) + } + t.Cleanup(func() { deleteExchange(t, conn, exchange) }) + + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(queue). + WithDurable(false). + WithAutoDelete(true). + WithExclusive(true)). + WithBinding(exchange, "evt.*", nil) + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + deliveryCh, err := consumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start consumer: %v", err) + } + + // Sanity: the declared queue + binding deliver before the drop. + if err := pub.Publish(ctx, NewTextMessage("before")); err != nil { + t.Fatalf("publish before reconnect: %v", err) + } + recvDelivery(t, ctx, deliveryCh, "before") + + // Force a reconnect by closing the underlying amqp connection out from + // under the wrapper; the broker deletes the exclusive queue and its binding. + conn.mu.RLock() + underlying := conn.conn + conn.mu.RUnlock() + if err := underlying.Close(); err != nil { + t.Fatalf("close underlying connection: %v", err) + } + + // The consumer must re-declare the queue, re-bind it, and resume delivery. + publishUntilDelivered(t, ctx, pub, deliveryCh, exchange, "evt.test", "after") +} + +// TestIntegration_ConsumeRecoversAfterQueueDeleted verifies the consume loop +// does not wedge when the queue is deleted while the CONNECTION stays healthy: +// the broker sends basic.cancel, the delivery channel closes, and no reconnect +// signal will ever arrive. The retry timer must re-declare the queue (via +// QueueConfig) and resume consuming. +func TestIntegration_ConsumeRecoversAfterQueueDeleted(t *testing.T) { + origDelay := consumeRetryDelay + consumeRetryDelay = 500 * time.Millisecond + t.Cleanup(func() { consumeRetryDelay = origDelay }) + + conn := integrationConn(t) + queue := uniqueQueue(t) + + cfg := DefaultConsumerConfig(). + WithQueueConfig(DefaultQueueConfig(queue)) // durable, so only an explicit delete removes it + consumer, err := NewConsumer(conn, cfg) + if err != nil { + t.Fatalf("failed to create consumer: %v", err) + } + t.Cleanup(func() { consumer.Close() }) + t.Cleanup(func() { deleteQueue(t, conn, queue) }) + + pub, err := NewPublisher(conn, DefaultPublisherConfig().WithRoutingKey(queue)) + if err != nil { + t.Fatalf("failed to create publisher: %v", err) + } + t.Cleanup(func() { pub.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + deliveryCh, err := consumer.Start(ctx) + if err != nil { + t.Fatalf("failed to start consumer: %v", err) + } + + if err := pub.Publish(ctx, NewTextMessage("before")); err != nil { + t.Fatalf("publish before delete: %v", err) + } + recvDelivery(t, ctx, deliveryCh, "before") + + // Delete the queue out from under the consumer; the connection stays up. + deleteQueue(t, conn, queue) + + // The retry timer must re-declare the queue and resume delivery. + publishUntilDelivered(t, ctx, pub, deliveryCh, "", queue, "after") +} + +// deleteExchange removes an exchange on a fresh channel; best-effort cleanup. +func deleteExchange(t *testing.T, conn *Connection, exchange string) { + t.Helper() + ch, err := conn.Channel() + if err != nil { + return + } + defer ch.Close() + _ = ch.Raw().ExchangeDelete(exchange, false, false) +} + +// deleteQueue removes a queue on a fresh channel; best-effort for cleanup, +// also used to delete a queue out from under a consumer mid-test. +func deleteQueue(t *testing.T, conn *Connection, queue string) { + t.Helper() + ch, err := conn.Channel() + if err != nil { + return + } + defer ch.Close() + _, _ = ch.Raw().QueueDelete(queue, false, false, false) +} From a260dea8e244825183dd36102fba6d3b128dddcd Mon Sep 17 00:00:00 2001 From: kartik Date: Sun, 5 Jul 2026 08:13:02 +0530 Subject: [PATCH 2/2] test(consumer): make consume-retry delay per-consumer to fix teardown data race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wedge-recovery test shortened a package-global consumeRetryDelay and restored it via t.Cleanup. Close() does not join the consumeLoop goroutine, so a queue-delete cleanup could trigger one more waitForReconnect read of the global concurrently with the restore write — a latent data race (timing-hidden from -race in practice). Replace the global with a per-Consumer retryDelay field, defaulted in NewConsumer and set by the test before Start (goroutine creation gives the happens-before edge). No shared mutable state remains, so there is nothing to restore or race on teardown. --- consumer.go | 16 +++++++++------- integration_test.go | 8 ++++---- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/consumer.go b/consumer.go index bccc873..fdf40a6 100644 --- a/consumer.go +++ b/consumer.go @@ -183,6 +183,7 @@ type Consumer struct { reconnectCh chan struct{} log Logger handlerWg sync.WaitGroup + retryDelay time.Duration // consume-loop retry delay; set before Start (see waitForReconnect) } // NewConsumer creates a new consumer. @@ -213,6 +214,7 @@ func NewConsumer(conn *Connection, config ConsumerConfig) (*Consumer, error) { queue: queueName, reconnectCh: conn.subscribeReconnect(), log: conn.log, + retryDelay: defaultConsumeRetryDelay, } if err := c.setupChannel(); err != nil { @@ -328,12 +330,12 @@ func (c *Consumer) Start(ctx context.Context) (<-chan *Delivery, error) { return outCh, nil } -// consumeRetryDelay is how long the consume loop waits before retrying channel -// setup when no reconnect signal arrives. Consuming can fail while the -// connection is healthy (queue deleted, server-sent basic.cancel, precondition -// failure), in which case no reconnect signal will ever come. A variable so -// tests can shorten it. -var consumeRetryDelay = 5 * time.Second +// defaultConsumeRetryDelay is the default for Consumer.retryDelay: how long the +// consume loop waits before retrying channel setup when no reconnect signal +// arrives. Consuming can fail while the connection is healthy (queue deleted, +// server-sent basic.cancel, precondition failure), in which case no reconnect +// signal will ever come. +const defaultConsumeRetryDelay = 5 * time.Second // waitForReconnect waits for a reconnection signal, a retry timeout, or // context cancellation, then re-establishes the channel. Returns true if the @@ -346,7 +348,7 @@ func (c *Consumer) waitForReconnect(ctx context.Context) bool { if !ok { return false } - case <-time.After(consumeRetryDelay): + case <-time.After(c.retryDelay): // No reconnect signal is coming if the connection is healthy but the // queue is gone — retry setup on a timer so the loop never blocks // forever. diff --git a/integration_test.go b/integration_test.go index 8deb233..d235133 100644 --- a/integration_test.go +++ b/integration_test.go @@ -1993,10 +1993,6 @@ func TestIntegration_TopologyRestoredAfterReconnect(t *testing.T) { // signal will ever arrive. The retry timer must re-declare the queue (via // QueueConfig) and resume consuming. func TestIntegration_ConsumeRecoversAfterQueueDeleted(t *testing.T) { - origDelay := consumeRetryDelay - consumeRetryDelay = 500 * time.Millisecond - t.Cleanup(func() { consumeRetryDelay = origDelay }) - conn := integrationConn(t) queue := uniqueQueue(t) @@ -2006,6 +2002,10 @@ func TestIntegration_ConsumeRecoversAfterQueueDeleted(t *testing.T) { if err != nil { t.Fatalf("failed to create consumer: %v", err) } + // Shorten the retry so the wedge-recovery path runs quickly. Set before + // Start so the write happens-before the consume goroutine reads it; being + // per-consumer, there is no shared global to restore (and race) on teardown. + consumer.retryDelay = 500 * time.Millisecond t.Cleanup(func() { consumer.Close() }) t.Cleanup(func() { deleteQueue(t, conn, queue) })