Fix/per-publish confirm correlation, confirms off by default, anonymous consumer queues - #12
Merged
Merged
Conversation
…ff; anonymous consumer queues - Publisher confirms now use a per-publish DeferredConfirmation (delivery-tag correlated) instead of a shared NotifyPublish channel. Fixes miscorrelation under concurrent publishers, where a call could return on another message's ack/nack (silent loss of the confirm guarantee). Batch publishing inherits it. - DefaultPublisherConfig().ConfirmMode now defaults to false; opt in with WithConfirmMode(true, timeout). - NewConsumer accepts an empty queue name, declaring a private server-named (exclusive, auto-delete) queue; new Consumer.QueueName() reports the assigned name. Adds integration coverage for concurrent confirmed publishing and anonymous queues. Release v0.3.0.
An anonymous consumer (empty queue name) previously overwrote config.Queue with the generated name in setupChannel, so on reconnect the empty-name branch was skipped and it never re-declared. Because the queue is exclusive + auto-delete, the broker drops it when the connection dies, and the consumer would then Consume from a stale, non-existent queue and silently stop delivering — contradicting its own godoc. Track the original empty-ness (serverNamed) and the resolved name (queue) separately from config.Queue, so setupChannel re-declares a fresh server-named queue on every setup, including reconnects. Read the resolved name under the same lock as the channel to keep the consume loop race-free. Adds an integration test that forces a reconnect and asserts the anonymous consumer re-declares a new queue and resumes delivery.
- Pin goimports via a GOIMPORTS_VERSION variable (v0.47.0) instead of @latest, mirroring GOLANGCI_LINT_VERSION, for reproducible tool installs. - Bump golangci-lint v2.11.4 -> v2.12.2 (latest) in both the Makefile and ci.yml, keeping the two in sync. Lint is clean on the new version.
Owner
Author
|
@coderabbitai full review |
✅ Action performedFull review finished. |
WalkthroughPublisher confirm mode now defaults to disabled and uses per-publish deferred confirmations instead of a shared confirmation channel. Consumer supports empty queue names, resolving them to server-named auto-delete queues exposed via ChangesPublisher Confirms and Anonymous Queue Support
Estimated code review effort: 4 (Complex) | ~60 minutes CI and Build Tooling
Estimated code review effort: 1 (Trivial) | ~3 minutes Sequence Diagram(s)sequenceDiagram
participant Caller
participant Publisher
participant Channel
participant Broker
Caller->>Publisher: PublishToExchange(msg)
alt ConfirmMode disabled
Publisher->>Channel: PublishWithContext
Publisher-->>Caller: return nil
else ConfirmMode enabled
Publisher->>Channel: PublishWithDeferredConfirmWithContext
Channel->>Broker: publish
Broker-->>Channel: ack/nack
Publisher->>Publisher: WaitContext(timeout)
alt nack or timeout
Publisher-->>Caller: ErrNack / ErrTimeout
else acked
Publisher-->>Caller: return nil
end
end
sequenceDiagram
participant Caller
participant Consumer
participant Channel
participant Broker
Caller->>Consumer: NewConsumer(empty Queue)
Consumer->>Channel: setupChannel()
Channel->>Broker: declare server-named queue
Broker-->>Channel: assigned name
Channel->>Consumer: store c.queue
Caller->>Consumer: QueueName()
Consumer-->>Caller: c.queue
Consumer->>Channel: Consume(c.queue)
Related Issues: None referenced in the provided diff. Related PRs: None referenced in the provided diff. Suggested labels: breaking-change, enhancement, documentation Suggested reviewers: Maintainers familiar with Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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! |
NewConsumer no longer errors on an empty queue name — it declares a private server-named queue instead — so this test asserting the old "queue is required" behavior was failing in CI. The empty-queue path is covered by TestIntegration_AnonymousServerNamedQueue and TestIntegration_AnonymousQueueReconnect.
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://webproxy.poorya-velaei-d67.workers.dev/https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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 1814-1820: The reconnect test around consumer.QueueName polling
and the retry gate using time.Sleep/time.After is timing-dependent and flaky.
Replace the sleep-based loop with an explicit synchronization point in the test
flow, using the existing consumer/reconnect path to wait for a deterministic
signal that the consumer has reconnected and started consuming. Update the
affected test logic in integration_test.go (the reconnect/queue-name check and
the retry section) so it no longer relies on broker timing assumptions.
- Around line 1699-1703: The loop that waits on deliveryCh can falsely pass if
the channel closes early because the receive result is ignored. Update the
select branch in this delivery-counting test to check the boolean receive value
from deliveryCh and fail the test immediately if the channel is closed before
got reaches total. Use the surrounding delivery tracking logic in
integration_test.go to locate the select over deliveryCh and ctx.Done(), and
keep the ctx cancellation handling unchanged.
In `@Makefile`:
- Around line 1-2: The setup step is currently allowing preinstalled goimports
and golangci-lint binaries to bypass the pinned versions, so the version
constants may not be enforced. Update the setup logic in Makefile so it always
installs the pinned GOIMPORTS_VERSION and GOLANGCI_LINT_VERSION, or explicitly
verifies the installed binary version before skipping the go install, using the
existing setup targets and those version variables as the entry points.
In `@publisher.go`:
- Around line 255-261: The deferred-confirm wait path in dc.WaitContext
currently returns the raw error, losing operation context for caller-side
cancellation or channel-close failures. Update the error handling around
WaitContext to keep the existing ErrTimeout mapping and any ErrNack behavior,
but wrap all remaining errors with a clear publish/deferred-confirm operation
message before returning; use the nearby WaitContext and acked/err handling in
publisher.go to locate the change.
🪄 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: 46eb744e-57a2-4e7d-8f87-f46af2de69ae
📒 Files selected for processing (8)
.github/workflows/ci.ymlCHANGELOG.mdMakefileREADME.mdconsumer.goconsumer_test.gointegration_test.gopublisher.go
RabbitMQ 4 deprecated and rejects transient non-exclusive queues (transient_nonexcl_queues), so integration tests that declared queues with durable=false and exclusive=false failed with Exception (541) INTERNAL_ERROR. - Declare the test queues as durable (autoDelete still cleans them up); valid on both old and new brokers. - Bump the broker to rabbitmq:4-management-alpine in both docker-compose.yml and the ci.yml services block (they provision RabbitMQ independently) so CI exercises the current major version. - Harden the concurrent-confirm test: check the delivery-channel receive so an early channel close fails the test instead of falsely counting toward total. Verified: full integration suite passes against RabbitMQ 4.3.2.
WaitContext failures other than our own confirm-timeout were returned raw,
losing operation context. Wrap them ("await publish confirmation: %w"), keeping
the ErrTimeout mapping and ErrNack path; %w preserves errors.Is checks against
context.Canceled and context.DeadlineExceeded.
Owner
Author
|
@coderabbitai resolve |
KARTIKrocks
changed the title
✅ Action performedComments resolved. Approval is disabled; enable |
3 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Changed