fix: restore consumer bindings deleted behind its back - #26
Conversation
Deleting an exchange destroys its bindings but leaves the consumer's queue, channel and consume entirely valid: no error, no basic.cancel, no channel close, and so nothing to trigger a re-setup. The consumer stayed alive bound to nothing for the rest of the process while publishes kept succeeding — a confirm means the broker accepted the message, not that it routed — so every message was silently dropped at the exchange. This is the half of channel-level recovery that v0.12.0 could not reach. A publisher notices because its next publish to the missing exchange draws a 404 that closes its channel; there is no equivalent signal on the consumer side, which made the result worse than no recovery at all: the topology looks healthy and nothing anywhere reports a problem. AMQP announces nothing here, so the only way to notice is to declare again. A consumer that declares topology now re-applies it every 30s by default, configurable (or disabled) with WithTopologyRefresh. Every declaration is idempotent, so a refresh is a no-op unless something is missing, and a consumer that declares no topology of its own never starts the loop. The refresh runs on its own short-lived channel rather than the consuming one: a declaration that cannot succeed — an exchange re-created with a different type, say — is then reported each interval instead of closing the channel deliveries are consumed on, which would drop every unacknowledged message with it. Server-named queues are deliberately not re-declared, since an empty name asks the broker for a new queue; channel setup still owns those. applyTopology is split into declareExchanges/declareQueue/applyBindings so setup and refresh share one definition of the declared topology.
|
Warning Review limit reached
Next review available in: 39 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
WalkthroughConsumers with declarative topology now refresh exchanges, queues, dead-letter topology, and bindings every 30 seconds by default. The interval can be configured or disabled. Refreshes use a dedicated channel and stop safely during consumer closure. ChangesConsumer topology refresh
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Consumer
participant RefreshLoop
participant RefreshChannel
participant RabbitMQ
Consumer->>RefreshLoop: Start enabled topology refresh
RefreshLoop->>RefreshChannel: Acquire dedicated channel
RefreshChannel->>RabbitMQ: Re-declare topology
RabbitMQ-->>RefreshChannel: Return declaration result
RefreshLoop-->>Consumer: Preserve consuming channel
Consumer->>RefreshLoop: Stop during close
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@consumer_test.go`:
- Around line 670-690: Add a test case in the existing topology refresh test
table with TopologyRefreshInterval explicitly set to zero, ensuring the guard
does not rewrite it before c.startTopologyRefresh(). Keep the long-interval
setup for other cases so the test remains deterministic, and verify the zero
interval starts the refresh loop through the default interval branch.
In `@consumer.go`:
- Around line 1079-1098: Update awaitTopologyRefreshStopped to accept the
caller’s context and select on ctx.Done() alongside completion and the existing
5-second timeout. Pass the CloseWithContext context at its call site, while
keeping Close’s context.Background() behavior so the fixed cap remains in effect
there.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 1e65cdef-242e-43fc-a0e1-8373df334577
📒 Files selected for processing (6)
CHANGELOG.mdREADME.mdconsumer.goconsumer_test.godoc.gointegration_test.go
Greptile SummaryThis PR adds periodic topology refresh to consumers, fixing a silent data-loss bug where deleting an exchange destroys its bindings but leaves the consumer's channel and
Confidence Score: 5/5Safe to merge — the change is additive, well-tested with both unit and integration coverage, and the shutdown ordering is correct. The refresh goroutine lifecycle is carefully managed: started only when topology is declared, stopped under the mutex before channel teardown, and bounded by a dual timeout so a blocked broker call cannot hold Close open. The dedicated refresh channel correctly isolates declaration failures from the consuming channel. Server-named queues are deliberately excluded from re-declaration with a matching stale-name check. Integration tests cover all key scenarios including the close/race path under -race. Files Needing Attention: No files require special attention. The consumer.go changes are the core of the PR and are well-structured.
|
| Filename | Overview |
|---|---|
| consumer.go | Core change: adds stopRefresh/refreshWg to Consumer, refactors applyTopology into three helpers, and introduces topologyRefreshLoop / refreshTopology / applyRefresh. Shutdown ordering (stopRefresh closed under lock, then awaitTopologyRefreshStopped with dual-timeout) is correct. Server-named queue guard in applyRefresh and stale-name detection are handled. |
| consumer_test.go | Adds unit tests for WithTopologyRefresh and startTopologyRefresh; covers enabled/disabled/zero-interval cases and goroutine lifecycle without a live broker. |
| integration_test.go | Adds 7 integration tests covering the primary bug scenario, foreign-exchange rebind, disabled refresh, server-named queue stability, channel isolation on failure, reconnect survival, and close/race. Test coverage is comprehensive. |
| CHANGELOG.md | Adds v0.13.0 entry documenting the fix and the new WithTopologyRefresh option with examples. |
| README.md | Adds "Topology refresh" subsection explaining the deletion scenario, the default 30 s interval, and the isolation guarantee of the dedicated channel. |
| doc.go | Package-level doc updated with one paragraph describing the periodic refresh and how to tune it. |
Sequence Diagram
sequenceDiagram
participant NC as NewConsumer
participant C as Consumer
participant RL as topologyRefreshLoop (goroutine)
participant RCh as refresh Channel
participant Broker as RabbitMQ Broker
participant CWC as CloseWithContext
NC->>C: setupChannel() — initial topology declared
NC->>C: startTopologyRefresh()
C->>RL: go topologyRefreshLoop(interval)
RL->>RCh: conn.Channel() — open dedicated refresh channel
loop every interval
RL->>RCh: applyRefresh(ch)
RCh->>Broker: ExchangeDeclare / QueueDeclare / QueueBind (idempotent)
alt topology intact
Broker-->>RCh: OK (no-op)
else binding/exchange deleted
Broker-->>RCh: OK (restored)
else wrong exchange type
Broker-->>RCh: PRECONDITION_FAILED
RCh->>RCh: ch.Close() — refresh channel spent
RL->>RL: log.Warnf — consuming channel unaffected
Note over RL: next tick opens a new refresh channel
else connection down
RCh-->>RL: ErrNotConnected / amqp.ErrClosed
RL->>RL: log.Debugf — deferred, self-correcting
end
end
CWC->>C: close(stopRefresh) [under mu]
CWC->>C: awaitTopologyRefreshStopped(ctx)
RL-->>RL: stopRefresh fires, return
RL->>RCh: deferred ch.Close()
C-->>CWC: refreshWg.Done() — close proceeds
Reviews (2): Last reviewed commit: "fix: apply topology refresh review feedb..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
- Close honours its context while waiting for the refresh loop to stop, as it already does for in-flight handlers, and is a true no-op for a consumer that never started one instead of spawning a goroutine per Close. - A bind that fails because a reconnect renamed the server-named queue between reading the name and the round-trip is logged at debug, not warned about: it is transient and the next tick binds the new name. Detected by re-reading the name after the failure, so a NOT_FOUND against a stable name — a missing exchange nobody re-creates, the case the refresh exists to report — still warns. - The refresh-loop test table covers an explicitly zero interval, exercising the branch that substitutes the default.
1 participant
Summary
Deleting an exchange destroys its bindings but leaves the consumer's queue, channel and consume entirely valid: no error, no basic.cancel, no channel close, and so nothing to trigger a re-setup. The consumer stayed alive bound to nothing for the rest of the process while publishes kept succeeding — a confirm means the broker accepted the message, not that it routed — so every message was silently dropped at the exchange.
This is the half of channel-level recovery that v0.12.0 could not reach. A publisher notices because its next publish to the missing exchange draws a 404 that closes its channel; there is no equivalent signal on the consumer side, which made the result worse than no recovery at all: the topology looks healthy and nothing anywhere reports a problem.
AMQP announces nothing here, so the only way to notice is to declare again. A consumer that declares topology now re-applies it every 30s by default, configurable (or disabled) with WithTopologyRefresh. Every declaration is idempotent, so a refresh is a no-op unless something is missing, and a consumer that declares no topology of its own never starts the loop.
The refresh runs on its own short-lived channel rather than the consuming one: a declaration that cannot succeed — an exchange re-created with a different type, say — is then reported each interval instead of closing the channel deliveries are consumed on, which would drop every unacknowledged message with it. Server-named queues are deliberately not re-declared, since an empty name asks the broker for a new queue; channel setup still owns those.
applyTopology is split into declareExchanges/declareQueue/applyBindings so setup and refresh share one definition of the declared topology.
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Documentation