fix: recover from a channel closed by the broker - #25
Conversation
A channel-level exception closes the channel while the connection stays healthy, so no reconnect signal is ever produced. Publishing to a missing exchange is the easy way to hit it, and the offending publish usually returns nil because the broker's 404 arrives asynchronously. The publisher was then left with a dead channel for the rest of the process: every later publish failed with 504 channel/connection is not open. The consumer recovered, but only when the consume loop's retry timer fired up to five seconds later. Both now register a NotifyClose watcher per established channel and re-establish as soon as the broker closes one. The watcher reports a death only while its channel is still the current one, so a channel replaced by a later setup and the graceful close done by Close do not trigger re-establishment; deaths coalesce into one pending signal. A channel that dies before its watcher registers is caught directly, since NotifyClose reports nothing once the channel is already shutting down. The publisher also closes the channel it replaces, as the consumer already did. That leaked before whenever a channel was replaced on a live connection, which this recovery makes routine. Consumer recovery stays owned by the consume loop, so it applies while Start/Consume is running; the retry timer remains for failures no channel close can signal, such as a deleted queue.
|
Warning Review limit reached
Next review available in: 45 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 (5)
WalkthroughThe library now detects broker-closed publisher and consumer channels and re-establishes them without reconnecting the underlying connection. It adds per-channel close watchers, immediate recovery signals, replacement-channel cleanup, integration tests, and documentation for recovery behavior. ChangesChannel Recovery
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant Publisher
participant Consumer
participant RabbitMQ
participant Connection
RabbitMQ-->>Publisher: closes publisher channel
Publisher->>Connection: re-establishes publisher channel
RabbitMQ-->>Consumer: closes consumer channel
Consumer->>Connection: re-establishes consumer channel
Connection-->>Publisher: publishing resumes
Connection-->>Consumer: consumption resumes
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 |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
Greptile SummaryThis PR fixes two longstanding gaps in channel lifecycle management: publishers were left permanently holding a dead channel after any broker-initiated channel close (e.g. a 404 on a missing exchange), and consumers recovered only after the 5-second retry timer rather than immediately.
Confidence Score: 5/5Safe to merge — the channel recovery logic is well-designed, edge cases (stale signals, graceful closes, TOCTOU on IsClosed) are all handled, and three integration tests exercise the new paths end-to-end. The stale-signal problem flagged in the previous review is now handled: the publisher's chDeadCh arm checks channelDead() before acting, and the consumer's unconditional setupChannel() on every arm makes a stale signal harmless. Replaced channels are now closed, godoc on both DeclareExchange methods is updated, and the previous comments are all addressed. No new correctness issues were found. Files Needing Attention: No files require special attention.
|
| Filename | Overview |
|---|---|
| rabbitmq.go | Adds watchChannelClose — the core mechanism that registers a per-channel goroutine to detect broker-initiated closes and signal re-establishment via a buffered dead channel. |
| publisher.go | Adds chDeadCh field, wires watchChannelClose into setupChannel, and adds a chDeadCh arm to handleReconnect with an explicit channelDead() staleness guard; also closes replaced channels to fix a resource leak. |
| consumer.go | Adds chDeadCh field, wires watchChannelClose into setupChannel, and adds a chDeadCh arm to waitForReconnect; the consumer intentionally skips the staleness guard because every arm of waitForReconnect calls setupChannel unconditionally. |
| integration_test.go | Adds three integration tests: publisher channel-exception recovery, stale-signal guard for the publisher, and consumer channel-exception recovery with retryDelay set to 10 min to prove the watcher — not the timer — drives recovery. |
| CHANGELOG.md | Adds v0.12.0 release notes documenting the channel-recovery fix and the publisher channel-leak fix. |
| README.md | Adds a Channel Recovery section explaining what triggers recovery, its scope limitations, and caveats about async 404 returns and confirms vs routing. |
Sequence Diagram
sequenceDiagram
participant Broker
participant Channel
participant watchChannelClose goroutine
participant chDeadCh (buf=1)
participant handleReconnect / waitForReconnect
Broker->>Channel: Channel-level exception (e.g. 404 NOT_FOUND)
Channel-->>watchChannelClose goroutine: closeCh <- amqpError (non-nil)
watchChannelClose goroutine->>watchChannelClose goroutine: isCurrent(ch)? yes
watchChannelClose goroutine->>chDeadCh (buf=1): signal (non-blocking)
chDeadCh (buf=1)->>handleReconnect / waitForReconnect: wakes select arm
alt Publisher path
handleReconnect / waitForReconnect->>handleReconnect / waitForReconnect: channelDead()? yes -> proceed
handleReconnect / waitForReconnect->>Channel: setupChannel() -> new channel
handleReconnect / waitForReconnect->>watchChannelClose goroutine: register new watcher
else Consumer path
handleReconnect / waitForReconnect->>Channel: setupChannel() -> new channel (unconditional)
handleReconnect / waitForReconnect->>watchChannelClose goroutine: register new watcher
end
Note over chDeadCh (buf=1): Stale signal scenario
Broker->>Channel: Connection loss (kills channel + fires reconnectCh)
Channel-->>watchChannelClose goroutine: both chDeadCh and reconnectCh signaled
handleReconnect / waitForReconnect->>Channel: reconnectCh arm wins -> setupChannel() first
Note over handleReconnect / waitForReconnect: chDeadCh arm fires next
handleReconnect / waitForReconnect->>handleReconnect / waitForReconnect: Publisher: channelDead()? no -> continue (skip)
handleReconnect / waitForReconnect->>handleReconnect / waitForReconnect: Consumer: setupChannel() (harmless extra cycle)
Reviews (2): Last reviewed commit: "fix: ignore a stale channel-death signal..." | Re-trigger Greptile
|
@coderabbitai full review |
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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.go`:
- Around line 375-419: Extract the duplicated channel-death watching and
signaling into one package-level watchChannelClose helper, preserving the
existing closed-channel, broker-error, current-channel, and nonblocking
notification behavior. In consumer.go lines 375-419, remove
Consumer.watchChannelClose and Consumer.signalChannelDead, then call the helper
from setupChannel with a c.mu-protected isCurrent closure. Apply the equivalent
change in publisher.go lines 231-275 using p.mu and the publisher-specific dead
channel, logger, and log prefix.
In `@integration_test.go`:
- Around line 2993-3013: Update the recovery assertions in the channel-exception
test to verify connection identity rather than calling conn.IsHealthy(). Before
triggering the exception, capture the underlying connection pointer using the
existing conn.mu/conn.conn pattern demonstrated by the nearby test, then after
publishing succeeds, capture it again and assert both pointers are identical
while retaining the health check before recovery.
In `@README.md`:
- Around line 165-167: Update the README guidance on publishing to distinguish
broker confirms from routing success: remove publisher confirms as a
routing-detection option, state that missing exchanges produce an asynchronous
channel-level NOT_FOUND, and recommend Mandatory with NotifyReturn for detecting
unroutable messages when an exchange exists without a matching queue.
🪄 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: 44f91d86-4277-4328-9a8e-3c5a4f7b0fb0
📒 Files selected for processing (5)
CHANGELOG.mdREADME.mdconsumer.gointegration_test.gopublisher.go
A connection loss signals both a reconnect and a channel death, so handleReconnect wakes on whichever arm it selects first, re-establishes the channel, and then reads the other signal — which by then refers to a channel that is already gone. Acting on it closed the channel just established, failing any confirm in flight on it. The dead-channel arm now checks whether the current channel is really dead instead of trusting the signal. The consumer's waitForReconnect keeps its signal as-is: it only runs when the consume loop needs a channel, so every arm ends in the same setup and a leftover signal costs nothing. Also fold the duplicated per-channel close watcher into one package-level watchChannelClose, correct the DeclareExchange docs that still said the channel is only re-established on connection recovery, and separate publisher confirms from routing in the README: a confirm means the broker accepted the message, not that it reached a queue.
2 participants
Summary
A channel-level exception closes the channel while the connection stays healthy, so no reconnect signal is ever produced. Publishing to a missing exchange is the easy way to hit it, and the offending publish usually returns nil because the broker's 404 arrives asynchronously.
The publisher was then left with a dead channel for the rest of the process: every later publish failed with 504 channel/connection is not open. The consumer recovered, but only when the consume loop's retry timer fired up to five seconds later.
Both now register a NotifyClose watcher per established channel and re-establish as soon as the broker closes one. The watcher reports a death only while its channel is still the current one, so a channel replaced by a later setup and the graceful close done by Close do not trigger re-establishment; deaths coalesce into one pending signal. A channel that dies before its watcher registers is caught directly, since NotifyClose reports nothing once the channel is already shutting down.
The publisher also closes the channel it replaces, as the consumer already did. That leaked before whenever a channel was replaced on a live connection, which this recovery makes routine.
Consumer recovery stays owned by the consume loop, so it applies while Start/Consume is running; the retry timer remains for failures no channel close can signal, such as a deleted queue.
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Documentation