feat(consumer): declarative topology restored on reconnect + fix consume-loop wedge#16
Conversation
…ume-loop wedge 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.
|
@coderabbitai full review |
|
Warning Review limit reached
Next review available in: 47 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)
WalkthroughThis PR adds declarative consumer topology to ChangesDeclarative Topology Feature
Estimated code review effort: 3 (Moderate) | ~30 minutes Sequence Diagram(s)sequenceDiagram
participant NewConsumer
participant setupChannel
participant applyTopology
participant Channel
NewConsumer->>setupChannel: setup channel (initial or reconnect)
setupChannel->>applyTopology: declare queue + apply bindings
applyTopology->>Channel: QueueDeclare
applyTopology->>Channel: QueueBind (per BindingConfig)
applyTopology-->>setupChannel: resolved queue name
setupChannel->>Channel: close prior channel if replaced
setupChannel-->>NewConsumer: channel ready, c.queue updated
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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 |
✅ Action performedFull review finished. |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@integration_test.go`:
- Around line 1995-1998: The test setup in
TestIntegration_ConsumeRecoversAfterQueueDeleted restores consumeRetryDelay too
early because Close() does not wait for the consumeLoop goroutine, which can
still be inside waitForReconnect. Update the test to keep the temporary retry
delay in place until the consume loop has fully exited, using a synchronization
point tied to the consumeLoop/handler lifecycle rather than only t.Cleanup.
Refer to consumeRetryDelay, Close, and consumeLoop when adjusting the teardown
ordering.
🪄 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: b9604005-a93c-4dbf-9c5d-596cfd2fc120
📒 Files selected for processing (5)
CHANGELOG.mdREADME.mdconsumer.goconsumer_test.gointegration_test.go
… data race 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.
2 participants
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.
Summary
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Documentation