Fix concurrency/correctness bugs and harden public API for v0.2.0 - #9
Merged
Conversation
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
Walkthroughv0.2.0 introduces sentinel nil errors, percent-encoded AMQP URL construction, honoring ConnectionTimeout for dials, publisher nil checks and return-handler refactor, atomic BatchPublisher draining with requeue on failure, consumer multi-start cancellation tracking, nil-safe message headers and middleware guards, plus tests and docs/tooling updates. ChangesRabbitMQ Wrapper v0.2.0 Release
Sequence Diagram(s)sequenceDiagram
participant Client
participant Config
participant Connection
participant Publisher
participant BatchPublisher
participant Broker
participant Consumer
Client->>Config: build Config
Config->>Connection: connectionURL()
Connection->>Broker: amqp.Dial (uses amqp.DefaultDial(ConnectionTimeout))
Client->>Publisher: NewPublisher(conn)
alt conn is nil
Publisher-->>Client: ErrNilConnection
else
Publisher-->>Client: publisher
end
Client->>Publisher: Publish(msg)
Publisher->>Publisher: validate msg != nil
alt msg is nil
Publisher-->>Client: ErrNilMessage
else
Publisher->>Broker: publish
Broker-->>Publisher: ack or return
Publisher->>Publisher: return listener forwards to latest onReturn handler
end
Client->>BatchPublisher: Add(msg)
Client->>BatchPublisher: PublishAndClear
BatchPublisher->>BatchPublisher: drain queue under lock
BatchPublisher->>Broker: publish drained messages
alt publish error
BatchPublisher->>BatchPublisher: requeue remaining ahead of new adds
end
Client->>Consumer: Start
Consumer->>Consumer: append cancel func to cancelFns
Client->>Consumer: Stop
Consumer->>Consumer: iterate cancelFns and cancel
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
Addresses findings from a full-repo review (each verified against the code and
covered by tests).
Bug fixes:
- RetryMiddleware: a negative retry count no longer skips the handler entirely
(previously the message was acked without ever being processed).
- Publisher.NotifyReturn: registers on the active channel immediately instead of
only taking effect after the next reconnection.
- Config.ConnectionTimeout: now honored when dialing (wired via amqp.DefaultDial);
it was previously ignored.
- Message.WithHeader/WithHeaders: initialize a nil Headers map instead of
panicking on a &Message{} literal.
- Consumer.Start: track all cancel funcs so repeated calls no longer leak the
previous consume goroutine.
- Config.connectionURL: percent-encode username/password/vhost via net/url so
credentials containing reserved characters produce a valid AMQP URI.
- BatchPublisher.PublishAndClear: atomically swap out the pending messages so
concurrent Add* calls are never cleared without being published; re-queue the
unpublished remainder on partial failure.
API hardening (new typed errors, backward compatible):
- Add ErrNilConnection and ErrNilMessage.
- NewConsumer, NewPublisher, PublishToExchange, PublishDelayed return typed
errors on nil arguments instead of panicking.
- LoggingMiddleware falls back to a no-op logger when passed nil.
- Guard against sync.WaitGroup misuse during concurrent Consume/Close.
Tooling/docs:
- Pin golangci-lint to v2.11.4 in both Makefile and CI; drop redundant gofmt
from the fmt target.
- Document integration-test command, new sentinel errors, and fix the stale
`make check` reference (now `make ci`).
- CHANGELOG: add the 0.2.0 section.
KARTIKrocks
force-pushed
the
release/v0.2.0
branch
from
June 13, 2026 14:03
4159d17 to
03498f2
Compare
Owner
Author
|
@coderabbitai full review |
✅ 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: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
Makefile (1)
42-45:⚠️ Potential issue | 🟡 MinorMake
fmtdepend onsetup.
goimportsis installed bysetup, butfmtrunsgoimports -w .andallcan invokefmtbeforelint(which is the only target that depends onsetup), somake fmt/make allcan fail on a fresh clone. Wirefmt: setup(or fall back togofmtifgoimportsisn’t available).♻️ Proposed fix
-fmt: +fmt: setup goimports -w .🤖 Prompt for 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. In `@Makefile` around lines 42 - 45, The Makefile's fmt target can run before setup installs goimports; modify the fmt target (named "fmt") to depend on the "setup" target (i.e., change its rule header to "fmt: setup") so goimports is available, and as an alternative add a fallback in the fmt recipe that uses "command -v goimports >/dev/null 2>&1 && goimports -w . || gofmt -w ." so formatting still works if goimports is missing.
🤖 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 421-427: The loop using time.Sleep in the test makes timing flaky;
replace it with synchronization so the producer and the flusher overlap
deterministically: run the producer that calls batch.Add(NewTextMessage(...)) in
a goroutine and coordinate with the code path that triggers PublishAndClear by
using channels/WaitGroup signals (e.g., send a "started flush" signal from the
flusher or wait for a "ready" signal before continuing adds) so you guarantee
interleaving without sleeps; update the test to wait for completion of both
producer and flusher goroutines and remove the time.Sleep(200 *
time.Microsecond) calls so the test is deterministic and race-free.
In `@message_test.go`:
- Around line 182-192: The test TestWithHeaderNilMap currently seeds the Headers
map via WithHeader before exercising WithHeaders, so update or add a test case
that constructs m := &Message{} with Headers nil and then calls
m.WithHeaders(map[string]any{"k2":"v2"}) (or call WithHeaders on a fresh Message
without first calling WithHeader) and assert that m.Headers["k2"] == "v2"; keep
the existing assertions for WithHeader behavior in a separate test or add a
second sub-case that first calls WithHeader then WithHeaders to cover both
paths, verifying that WithHeaders correctly initializes a nil Message.Headers
map and merges entries when map already exists (referencing
TestWithHeaderNilMap, WithHeaders, WithHeader, Message, and Headers).
In `@README.md`:
- Around line 182-187: The snippet creates a publisher with
rabbitmq.NewPublisher (assigned to publisher) but never closes it; after
checking err from NewPublisher, ensure you close the publisher to avoid leaking
the confirm channel by calling publisher.Close() (or publisher.Close(ctx) if the
API requires a context) — place this as a deferred call immediately after
successful creation (e.g., defer publisher.Close() / defer publisher.Close(ctx))
so the publisher is always closed even if publisher.Publish(ctx, msg) returns an
error.
---
Outside diff comments:
In `@Makefile`:
- Around line 42-45: The Makefile's fmt target can run before setup installs
goimports; modify the fmt target (named "fmt") to depend on the "setup" target
(i.e., change its rule header to "fmt: setup") so goimports is available, and as
an alternative add a fallback in the fmt recipe that uses "command -v goimports
>/dev/null 2>&1 && goimports -w . || gofmt -w ." so formatting still works if
goimports is missing.
🪄 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: 0171a5d6-510d-443d-b40b-a25871689d4b
📒 Files selected for processing (15)
.github/workflows/ci.ymlCHANGELOG.mdCONTRIBUTING.mdMakefileREADME.mdconfig_test.goconsumer.goconsumer_test.gointegration_test.gomessage.gomessage_test.gomiddleware.gomiddleware_test.gopublisher.gorabbitmq.go
Comment on lines
+421
to
+427
| // Pace the adds so they interleave with the publish window of | ||
| // PublishAndClear; without pacing the producer finishes before the | ||
| // first flush and the race is never exercised. | ||
| for i := 0; i < total; i++ { | ||
| batch.Add(NewTextMessage(fmt.Sprintf("msg-%d", i))) | ||
| time.Sleep(200 * time.Microsecond) | ||
| } |
There was a problem hiding this comment.
Replace sleep-based pacing with synchronization-driven overlap
Using time.Sleep(200 * time.Microsecond) to force interleaving makes this integration test timing-sensitive and flaky across environments. Prefer synchronization primitives (e.g., producer/flusher goroutines coordinated via channels + completion signals) to create overlap without fixed sleeps.
As per coding guidelines, **/*_test.go: “Verify tests are deterministic and free of races, sleeps, and flaky timing assumptions.”
🤖 Prompt for 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.
In `@integration_test.go` around lines 421 - 427, The loop using time.Sleep in the
test makes timing flaky; replace it with synchronization so the producer and the
flusher overlap deterministically: run the producer that calls
batch.Add(NewTextMessage(...)) in a goroutine and coordinate with the code path
that triggers PublishAndClear by using channels/WaitGroup signals (e.g., send a
"started flush" signal from the flusher or wait for a "ready" signal before
continuing adds) so you guarantee interleaving without sleeps; update the test
to wait for completion of both producer and flusher goroutines and remove the
time.Sleep(200 * time.Microsecond) calls so the test is deterministic and
race-free.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rabbitmq.go (1)
19-32: 🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick winDocument
ErrNilConnectionandErrNilMessageindividually.These are new exported API symbols, but the var-block comment does not explain either identifier in godoc. Add per-symbol comments so downstream users see their exact meaning in package docs.
As per coding guidelines,
**/*.go: “Require godoc comments on all exported identifiers.”🤖 Prompt for 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. In `@rabbitmq.go` around lines 19 - 32, Add godoc comments for the two new exported sentinels so their meanings appear in package docs: place a short comment immediately above ErrNilConnection (for example: "ErrNilConnection is returned when a function expecting an active *amqp.Connection was given a nil connection") and another immediately above ErrNilMessage (for example: "ErrNilMessage is returned when a publish/consume function was passed a nil message or delivery"), keeping them as part of the existing var block and using the exact exported names ErrNilConnection and ErrNilMessage.Source: Coding guidelines
🤖 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 `@config_test.go`:
- Around line 154-200: Add a test case exercising reserved characters in VHost
so TestConnectionURLEncoding also validates vhost path-escaping: add a table
entry (same slice used by TestConnectionURLEncoding) named like "reserved
characters in vhost" that uses Config{Host:"myhost", Port:5672, Username:"user",
Password:"pass", VHost:"/p@th:with?=chars"} and set the expected fields
accordingly (wantUser "user", wantPass "pass", wantHost "myhost", wantPort 5672,
wantVhost "p@th:with?=chars") so the test asserts that the code handling Config
-> connection URL properly escapes/decodes VHost; reference the Config struct
and the TestConnectionURLEncoding table to locate where to insert this case.
In `@integration_test.go`:
- Around line 455-471: Replace the current "received" counter loop (variables
received, deliveryCh, idle, ctx, total) with logic that reads each delivery from
deliveryCh, extracts the message ID from the message body (the "msg-%d" payload
referenced in the test), and records IDs in a map/set (e.g., map[int]bool); on
each delivery check if the ID is a duplicate and fail the test immediately if
so, otherwise mark it seen; continue until the set size == total, then verify
that all IDs 1..total are present and fail if any are missing; preserve the idle
timer/context timeout behavior but update failure messages to report seen IDs
and duplicates/missing IDs for clearer assertions.
In `@publisher.go`:
- Around line 155-186: The current approach creates a new amqp Return listener
goroutine every time Publisher.NotifyReturn is called, causing stacked
listeners; fix by making the channel-level listener single and have it dispatch
to the current Publisher handler. Concretely: change registerReturnHandler so it
no longer creates a listener per call but instead ensures one long-lived
listener is started once for a given Channel (e.g., add a boolean/flag or
listener field on type Channel and guard it with Channel's mutex), have that
single goroutine read from the amqp returnCh and, for each received amqp.Return,
acquire Publisher.onReturnMu and call the latest p.onReturn (or drop if nil);
update Publisher.NotifyReturn to only set p.onReturn under onReturnMu and start
the channel-level listener only if it hasn’t already been started (do not spawn
a new goroutine on each NotifyReturn), ensuring old goroutines are not created
and returned messages are dispatched to the latest handler.
---
Outside diff comments:
In `@rabbitmq.go`:
- Around line 19-32: Add godoc comments for the two new exported sentinels so
their meanings appear in package docs: place a short comment immediately above
ErrNilConnection (for example: "ErrNilConnection is returned when a function
expecting an active *amqp.Connection was given a nil connection") and another
immediately above ErrNilMessage (for example: "ErrNilMessage is returned when a
publish/consume function was passed a nil message or delivery"), keeping them as
part of the existing var block and using the exact exported names
ErrNilConnection and ErrNilMessage.
🪄 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: c45c88de-2975-4ef1-bd94-16d52e4178c8
📒 Files selected for processing (15)
.github/workflows/ci.ymlCHANGELOG.mdCONTRIBUTING.mdMakefileREADME.mdconfig_test.goconsumer.goconsumer_test.gointegration_test.gomessage.gomessage_test.gomiddleware.gomiddleware_test.gopublisher.gorabbitmq.go
- message_test: split TestWithHeaderNilMap into two sub-cases, each starting
from a fresh &Message{}, so WithHeaders exercises its own nil-map
initialization path (previously WithHeader seeded the map first).
- README: defer publisher.Close() in the Publisher Confirms example so it does
not leak the confirm channel.
- Makefile: make the fmt target depend on setup so goimports is installed before
it runs (consistent with lint: setup).
Skipped: the time.Sleep pacing in TestIntegration_BatchPublishAndClearConcurrent
is intentional. The race window is inside PublishAndClear's broker round-trip,
which boundary-level channel/WaitGroup handshakes cannot deterministically
reach; removing the pacing makes the producer finish before flushing starts and
stops exercising the race. The pass/fail assertion is deterministic for the
fixed code (no message can be lost), so the test is not flaky.
Owner
Author
|
@coderabbitai can you check the state of the comments with updated codebase, and also check the last commit. |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== Open/Unresolved Review Threads on PR `#9` ==="
gh pr view 9 --repo KARTIKrocks/rabbitwrap --json reviews,reviewDecision,state
echo ""
echo "=== Review Comments (inline) ==="
gh api repos/KARTIKrocks/rabbitwrap/pulls/9/comments \
--jq '.[] | {id: .id, path: .path, line: .line, body: .body[:200], created_at: .created_at, updated_at: .updated_at}'
echo ""
echo "=== Issue Comments on PR `#9` ==="
gh api repos/KARTIKrocks/rabbitwrap/issues/9/comments \
--jq '.[] | {id: .id, user: .user.login, body: .body[:300], created_at: .created_at}'
echo ""
echo "=== Last 3 Commits on PR Branch ==="
gh pr view 9 --repo KARTIKrocks/rabbitwrap --json commits \
--jq '.commits[-3:] | .[] | {sha: .oid, message: .messageHeadline, authoredDate: .authoredDate, author: .authors[0].name}'Length of output: 17852 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== Diff of last commit abaf63065a2911b7b8824277a6a40b2ba059953e ==="
gh api repos/KARTIKrocks/rabbitwrap/commits/abaf63065a2911b7b8824277a6a40b2ba059953e \
--jq '.files[] | "FILE: \(.filename)\nSTATUS: \(.status)\nADDITIONS: \(.additions)\nDELETIONS: \(.deletions)\nPATCH:\n\(.patch // "(binary or no patch)")\n---"'Length of output: 2404
✅ Addressed in last commit
🔴 Still open
The |
- publisher: NotifyReturn no longer spawns a listener per call, which stacked goroutines and left replaced handlers firing. setupChannel now starts a single per-channel return listener that dispatches to the current handler; NotifyReturn just swaps the handler. Behavior (immediate effect, persists across reconnects) is preserved. Covered by a new integration test that also asserts no stacking. - rabbitmq: add godoc for the new ErrNilConnection and ErrNilMessage sentinels. - config_test: add a TestConnectionURLEncoding case with reserved characters in the vhost to cover vhost path-escaping. - integration_test: track message IDs in the batch concurrency test to detect duplicates and missing IDs, not just a raw count.
Owner
Author
|
@coderabbitai, check again. |
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== Last 5 Commits on PR `#9` ==="
gh pr view 9 --repo KARTIKrocks/rabbitwrap --json commits \
--jq '.commits[-5:] | .[] | {sha: .oid, message: .messageHeadline, authoredDate: .authoredDate}'
echo ""
echo "=== Latest commit diff ==="
LAST_SHA=$(gh pr view 9 --repo KARTIKrocks/rabbitwrap --json commits --jq '.commits[-1].oid')
echo "Last SHA: $LAST_SHA"
gh api repos/KARTIKrocks/rabbitwrap/commits/$LAST_SHA \
--jq '.files[] | "FILE: \(.filename)\nSTATUS: \(.status)\nPATCH:\n\(.patch // "(none)")\n---"'Length of output: 8325 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "=== publisher.go: NotifyReturn and registerReturnHandler ==="
cat publisher.go | grep -n -A 30 "registerReturnHandler\|NotifyReturn"
echo ""
echo "=== rabbitmq.go: ErrNil exports ==="
grep -n -B2 -A2 "ErrNil" rabbitmq.go
echo ""
echo "=== config_test.go: VHost test cases ==="
grep -n "VHost\|vhost\|Vhost" config_test.go
echo ""
echo "=== integration_test.go: delivery loop ==="
grep -n -A 5 "seen\|map\[int\]\|Duplicate\|duplicate\|Body\|msgID" integration_test.go | tail -60Length of output: 5778
No open issues remain. The PR is good to merge. 🐇 |
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.
Addresses findings from a full-repo CodeRabbit review (each verified against the code and covered by tests).
Bug fixes:
API hardening (new typed errors, backward compatible):
Tooling/docs:
make checkreference (nowmake ci).Summary
Motivation
Fixes #
Changes
Checklist
make all)Summary by CodeRabbit
New Features
Bug Fixes
Changed
Tests
Documentation