fix(publisher): improve PublishDelayed to correctly handle message delays and add tests#15
Conversation
…lays and add tests
WalkthroughPublishDelayed is reworked to delay delivery through deterministic, TTL-configured holding queues with dead-letter routing to the real destination, rather than mutating message expiration. Adds ChangesDelay Ladder and Holding-Queue Publishing
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant Caller
participant Publisher as Publisher (PublishDelayed)
participant HoldingQueue as Holding Queue (TTL/DLX)
participant Destination as Real Destination Queue
Caller->>Publisher: PublishDelayed(msg, delay)
Publisher->>Publisher: snapDelay(delay) -> rung or ErrDelayTooLong
Publisher->>Publisher: delayQueueName(exchange, routingKey, rung)
Publisher->>HoldingQueue: declareDelayQueue(TTL, DLX, x-expires)
Publisher->>HoldingQueue: Publish message (default exchange)
Note over HoldingQueue: Message expires after TTL
HoldingQueue-->>Destination: Dead-lettered to real destination
Destination-->>Caller: Message delivered after ~rung duration
Estimated code review effort: 3 (Moderate) | ~25 minutes 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! |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 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 1339-1342: The timing assertion in the integration test is too
strict and can flake under CI jitter; adjust the check around the elapsed time
calculation so it uses a wider lower-bound margin or adds an upper-bound
assertion as well. Update the test logic near the elapsed/Since(start)
validation to keep catching near-instant delivery regressions without relying on
a 900ms floor that is too close to the intended ~1s delay.
In `@publisher_test.go`:
- Around line 9-34: The table-driven checks in TestSnapDelay_Ladder should be
converted to named t.Run subtests so each delay case is isolated and easier to
triage. Update the loop in TestSnapDelay_Ladder to wrap each test case in t.Run
using a descriptive case name, and keep the existing snapDelay assertions inside
the subtest so failures are reported per input and can later support
t.Parallel() if needed.
- Around line 36-47: The delayQueueName test is missing coverage for exchange
collisions, so a regression that ignores or merges the exchange input could slip
through. Update TestDelayQueueName_Deterministic to add a case that keeps the
routing key and delay the same but uses two different exchanges, and assert the
returned names differ. Use delayQueueName in publisher_test.go as the anchor for
the new check.
In `@publisher.go`:
- Around line 328-335: Add a brief godoc caveat to delayQueueName explaining
that the exchange/routingKey hash is intentionally truncated to 64 bits and that
a rare collision would cause QueueDeclare to fail with PRECONDITION_FAILED
rather than misroute messages; keep the implementation unchanged and make the
accepted residual collision risk explicit near delayQueueName and
PublishDelayed.
- Around line 363-374: PublishDelayed is doing a synchronous QueueDeclare on
every call via declareDelayQueue, which adds an extra broker round-trip for hot
retry/backoff paths. Update PublishDelayed and declareDelayQueue to cache queue
declarations by queueName with a lastDeclaredAt timestamp, skip redeclaring when
the queue was declared recently, and only refresh periodically often enough to
preserve the x-expires reset behavior while reducing repeated RPCs.
- Around line 382-416: Document the new AMQP permission requirement introduced
by PublishDelayed: because declareDelayQueue now calls QueueDeclare for the
rabbitwrap.delay.* queues, deployments need configure/declare access for those
queue names. Update the CHANGELOG or migration notes to call out this upgrade
impact and reference PublishDelayed and declareDelayQueue so operators know
restrictive allow-lists must be widened.
- Around line 355-380: PublishDelayed is only partially honoring ctx because
declareDelayQueue still calls QueueDeclare without any cancellation path, so a
slow broker can block past the caller’s deadline before PublishToExchange runs.
Update PublishDelayed and/or declareDelayQueue to make the queue-declaration
step cancellable using ctx (or stop treating it as fully context-aware), and if
full cancellation isn’t possible with amqp091-go, add godoc on PublishDelayed
explaining the partial context behavior.
🪄 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: e719d8e4-aef9-4914-8569-d0d058ebe742
📒 Files selected for processing (5)
CHANGELOG.mdintegration_test.gopublisher.gopublisher_test.gorabbitmq.go
| elapsed := time.Since(start) | ||
| if elapsed < 900*time.Millisecond { | ||
| t.Errorf("message delivered after %s, expected it to be delayed ~1s", elapsed) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Tight lower-bound threshold risks flakiness.
The 900ms floor against a targeted ~1s delay leaves only a 100ms margin for scheduling jitter, broker/network latency, and CI load — a common source of flaky integration timing assertions. Consider widening the margin (e.g. 700–800ms) or asserting against an upper bound too, so a regression that skips rounding entirely (delivering near-instantly) is still reliably caught without tightening the lower bound so much that healthy runs occasionally fail.
As per path instructions, integration tests should be "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 1339 - 1342, The timing assertion in the
integration test is too strict and can flake under CI jitter; adjust the check
around the elapsed time calculation so it uses a wider lower-bound margin or
adds an upper-bound assertion as well. Update the test logic near the
elapsed/Since(start) validation to keep catching near-instant delivery
regressions without relying on a 900ms floor that is too close to the intended
~1s delay.
Source: Path instructions
| func TestSnapDelay_Ladder(t *testing.T) { | ||
| tests := []struct { | ||
| delay time.Duration | ||
| want time.Duration | ||
| }{ | ||
| {50 * time.Millisecond, 1 * time.Second}, | ||
| {1 * time.Second, 1 * time.Second}, | ||
| {2 * time.Second, 5 * time.Second}, | ||
| {45 * time.Second, 1 * time.Minute}, | ||
| {1 * time.Hour, 1 * time.Hour}, | ||
| } | ||
| for _, tt := range tests { | ||
| got, err := snapDelay(tt.delay) | ||
| if err != nil { | ||
| t.Errorf("snapDelay(%s) unexpected error: %v", tt.delay, err) | ||
| continue | ||
| } | ||
| if got != tt.want { | ||
| t.Errorf("snapDelay(%s) = %s, want %s", tt.delay, got, tt.want) | ||
| } | ||
| } | ||
|
|
||
| if _, err := snapDelay(2 * time.Hour); !errors.Is(err, ErrDelayTooLong) { | ||
| t.Errorf("snapDelay(2h) error = %v, want ErrDelayTooLong", err) | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Table-driven test doesn't use t.Run subtests.
Each case is exercised inline without t.Run(...), so a single failing case fails the whole TestSnapDelay_Ladder without isolating which input failed by name (the error message includes the input, mitigating this somewhat). Not blocking, but subtests would improve failure triage and enable t.Parallel() per case.
🤖 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 `@publisher_test.go` around lines 9 - 34, The table-driven checks in
TestSnapDelay_Ladder should be converted to named t.Run subtests so each delay
case is isolated and easier to triage. Update the loop in TestSnapDelay_Ladder
to wrap each test case in t.Run using a descriptive case name, and keep the
existing snapDelay assertions inside the subtest so failures are reported per
input and can later support t.Parallel() if needed.
| func TestDelayQueueName_Deterministic(t *testing.T) { | ||
| if a, b := delayQueueName("ex", "key", time.Second), delayQueueName("ex", "key", time.Second); a != b { | ||
| t.Errorf("delayQueueName not deterministic: %q != %q", a, b) | ||
| } | ||
| // Different destinations and delays must not collide. | ||
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key2", time.Second) { | ||
| t.Error("delayQueueName collided on different routing key") | ||
| } | ||
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key", 5*time.Second) { | ||
| t.Error("delayQueueName collided on different delay") | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Missing exchange-collision coverage.
delayQueueName hashes exchange + routingKey, but the test only varies routing key and delay — it never checks that different exchanges (same routing key) produce distinct names. Given the naming scheme derives its uniqueness solely from this hash, a regression collapsing the two inputs (e.g. dropping the separator or the exchange) wouldn't be caught.
Suggested addition
if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key", 5*time.Second) {
t.Error("delayQueueName collided on different delay")
}
+ if delayQueueName("ex", "key", time.Second) == delayQueueName("ex2", "key", time.Second) {
+ t.Error("delayQueueName collided on different exchange")
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func TestDelayQueueName_Deterministic(t *testing.T) { | |
| if a, b := delayQueueName("ex", "key", time.Second), delayQueueName("ex", "key", time.Second); a != b { | |
| t.Errorf("delayQueueName not deterministic: %q != %q", a, b) | |
| } | |
| // Different destinations and delays must not collide. | |
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key2", time.Second) { | |
| t.Error("delayQueueName collided on different routing key") | |
| } | |
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key", 5*time.Second) { | |
| t.Error("delayQueueName collided on different delay") | |
| } | |
| } | |
| func TestDelayQueueName_Deterministic(t *testing.T) { | |
| if a, b := delayQueueName("ex", "key", time.Second), delayQueueName("ex", "key", time.Second); a != b { | |
| t.Errorf("delayQueueName not deterministic: %q != %q", a, b) | |
| } | |
| // Different destinations and delays must not collide. | |
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key2", time.Second) { | |
| t.Error("delayQueueName collided on different routing key") | |
| } | |
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex", "key", 5*time.Second) { | |
| t.Error("delayQueueName collided on different delay") | |
| } | |
| if delayQueueName("ex", "key", time.Second) == delayQueueName("ex2", "key", time.Second) { | |
| t.Error("delayQueueName collided on different exchange") | |
| } | |
| } |
🤖 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 `@publisher_test.go` around lines 36 - 47, The delayQueueName test is missing
coverage for exchange collisions, so a regression that ignores or merges the
exchange input could slip through. Update TestDelayQueueName_Deterministic to
add a case that keeps the routing key and delay the same but uses two different
exchanges, and assert the returned names differ. Use delayQueueName in
publisher_test.go as the anchor for the new check.
| // delayQueueName builds the deterministic name of the holding queue for a given | ||
| // destination and delay. The destination is hashed so the name stays within | ||
| // length limits and is safe regardless of the characters in the exchange or | ||
| // routing key, while remaining stable across calls (so the queue is reused). | ||
| func delayQueueName(exchange, routingKey string, delay time.Duration) string { | ||
| h := sha256.Sum256([]byte(exchange + "\x00" + routingKey)) | ||
| return fmt.Sprintf("rabbitwrap.delay.%d.%s", delay.Milliseconds(), hex.EncodeToString(h[:8])) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔵 Trivial | 💤 Low value
Truncated hash leaves a small residual collision surface.
Truncating SHA-256 to 8 bytes (64 bits) for (exchange, routingKey) naming is fine for practically all deployments, but if two distinct destinations ever collide, the resulting QueueDeclare for the second destination will fail with a broker PRECONDITION_FAILED (mismatched x-dead-letter-routing-key/x-dead-letter-exchange), silently breaking PublishDelayed for that destination rather than misdelivering. Worth a one-line godoc caveat, or just noting this residual risk is accepted.
🤖 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 `@publisher.go` around lines 328 - 335, Add a brief godoc caveat to
delayQueueName explaining that the exchange/routingKey hash is intentionally
truncated to 64 bits and that a rare collision would cause QueueDeclare to fail
with PRECONDITION_FAILED rather than misroute messages; keep the implementation
unchanged and make the accepted residual collision risk explicit near
delayQueueName and PublishDelayed.
| func (p *Publisher) PublishDelayed(ctx context.Context, msg *Message, delay time.Duration) error { | ||
| if msg == nil { | ||
| return ErrNilMessage | ||
| } | ||
| msg.WithTTL(delay) | ||
| return p.Publish(ctx, msg) | ||
| if delay <= 0 { | ||
| return p.Publish(ctx, msg) | ||
| } | ||
|
|
||
| snapped, err := snapDelay(delay) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| exchange := p.config.Exchange | ||
| routingKey := p.config.RoutingKey | ||
| queueName := delayQueueName(exchange, routingKey, snapped) | ||
|
|
||
| if err := p.declareDelayQueue(queueName, exchange, routingKey, snapped); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Publish into the holding queue via the default exchange, which routes by | ||
| // queue name. On TTL expiry the broker dead-letters the message to the real | ||
| // destination configured on the queue. | ||
| return p.PublishToExchange(ctx, "", queueName, msg) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does amqp091-go's Channel.QueueDeclare support context cancellation or a context-aware variant?
💡 Result:
No, the amqp091-go library's Channel.QueueDeclare method does not support context cancellation, nor is there a context-aware variant (such as QueueDeclareWithContext) [1][2]. The library maintains a synchronous RPC-like design for its channel operations [3]. While there have been ongoing discussions and proposals within the project to refactor the internal state machine to support context-based timeouts and cancellation for all RPC operations, these changes have not been implemented for QueueDeclare [4]. Attempts to add context support to various methods in the library have faced significant architectural challenges [5][6][7]. Specifically, because the AMQP protocol requires synchronous, serialized RPCs on a channel, cancelling a blocked operation like QueueDeclare would leave the channel in an indeterminate state that cannot be safely reused, typically requiring the entire channel to be closed [4]. Consequently, the current API remains synchronous, and developers are advised that even for methods labeled *WithContext (like PublishWithContext), the context is often not propagated to the underlying network call in a way that truly cancels the server-side operation [6][7].
Citations:
- 1: https://github.com/rabbitmq/amqp091-go/blob/2774d3391846d4c9d415254d75684926fb3946af/channel.go
- 2: https://github.com/rabbitmq/amqp091-go/blob/main/channel.go
- 3: https://github.com/rabbitmq/amqp091-go/blob/main/doc.go
- 4: Proposal: refactor internal state machine rabbitmq/amqp091-go#316
- 5: Add Contexts where appropriate rabbitmq/amqp091-go#124
- 6: Documentation needed (
PublishWithContextdoes not use context) rabbitmq/amqp091-go#195 - 7: fix: respect context cancellation on publishing with context operations rabbitmq/amqp091-go#330
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the relevant symbols and surrounding docs.
rg -n "func \\(p \\*Publisher\\) PublishDelayed|declareDelayQueue|QueueDeclare|PublishToExchange|PublishWithContext|Publish\\(" publisher.go
printf '\n--- publisher.go excerpt ---\n'
sed -n '1,260p' publisher.go
printf '\n--- publisher.go excerpt 2 ---\n'
sed -n '260,460p' publisher.goRepository: KARTIKrocks/rabbitwrap
Length of output: 15903
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check how other public methods describe ctx behavior and whether the delay path is documented.
rg -n "ctx|context" publisher.go README.md *.mdRepository: KARTIKrocks/rabbitwrap
Length of output: 4101
PublishDelayed is only partially context-aware.
ctx reaches the final publish, but declareDelayQueue still blocks on QueueDeclare, which has no context-aware variant in amqp091-go. A slow broker can therefore ignore the caller’s deadline/cancel before the delayed publish even starts. Either make that step cancellable or document the partial context behavior in the godoc.
🤖 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 `@publisher.go` around lines 355 - 380, PublishDelayed is only partially
honoring ctx because declareDelayQueue still calls QueueDeclare without any
cancellation path, so a slow broker can block past the caller’s deadline before
PublishToExchange runs. Update PublishDelayed and/or declareDelayQueue to make
the queue-declaration step cancellable using ctx (or stop treating it as fully
context-aware), and if full cancellation isn’t possible with amqp091-go, add
godoc on PublishDelayed explaining the partial context behavior.
| snapped, err := snapDelay(delay) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| exchange := p.config.Exchange | ||
| routingKey := p.config.RoutingKey | ||
| queueName := delayQueueName(exchange, routingKey, snapped) | ||
|
|
||
| if err := p.declareDelayQueue(queueName, exchange, routingKey, snapped); err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
Synchronous QueueDeclare on every delayed publish.
Every call to PublishDelayed now does a broker round-trip via declareDelayQueue before publishing, whereas the previous TTL-on-message approach needed only the publish itself. This is explained as intentional (redeclaring resets x-expires), but for a hot retry/backoff path this doubles RPCs per delayed message.
Consider caching (queueName -> lastDeclaredAt) and skipping redeclare if it happened recently (e.g., within some fraction of x-expires), still redeclaring periodically to keep the idle-expiry lease alive. This preserves the reset semantics while cutting broker round-trips for bursty callers.
🤖 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 `@publisher.go` around lines 363 - 374, PublishDelayed is doing a synchronous
QueueDeclare on every call via declareDelayQueue, which adds an extra broker
round-trip for hot retry/backoff paths. Update PublishDelayed and
declareDelayQueue to cache queue declarations by queueName with a lastDeclaredAt
timestamp, skip redeclaring when the queue was declared recently, and only
refresh periodically often enough to preserve the x-expires reset behavior while
reducing repeated RPCs.
| // declareDelayQueue idempotently declares the holding queue for PublishDelayed. | ||
| // Re-declaring on every publish also resets the queue's idle-expiry timer, so an | ||
| // actively used delay queue stays alive while unused ones are reaped. | ||
| func (p *Publisher) declareDelayQueue(name, dlExchange, dlRoutingKey string, delay time.Duration) error { | ||
| p.mu.RLock() | ||
| if p.closed { | ||
| p.mu.RUnlock() | ||
| return ErrShuttingDown | ||
| } | ||
| ch := p.channel | ||
| p.mu.RUnlock() | ||
|
|
||
| if ch == nil { | ||
| return ErrChannelClosed | ||
| } | ||
|
|
||
| // x-expires must comfortably exceed the message TTL, otherwise the broker | ||
| // could delete the queue (and any in-flight message) before it is | ||
| // dead-lettered. | ||
| expires := delay * 2 | ||
| if minExpires := delay + 30*time.Second; expires < minExpires { | ||
| expires = minExpires | ||
| } | ||
|
|
||
| args := amqp.Table{ | ||
| "x-message-ttl": delay.Milliseconds(), | ||
| "x-dead-letter-exchange": dlExchange, | ||
| "x-dead-letter-routing-key": dlRoutingKey, | ||
| "x-expires": expires.Milliseconds(), | ||
| } | ||
|
|
||
| if _, err := ch.ch.QueueDeclare(name, true /*durable*/, false /*autoDelete*/, false /*exclusive*/, false /*noWait*/, args); err != nil { | ||
| return fmt.Errorf("%w: declare delay queue: %v", ErrPublishFailed, err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
🧹 Nitpick | 🔵 Trivial
New broker permission requirement on upgrade.
Unlike the previous TTL-on-message implementation, PublishDelayed now requires the connecting user to have configure/declare permission on queue names matching rabbitwrap.delay.*. Existing deployments with restrictive AMQP permission grants (queue name allow-lists) will start failing PublishDelayed calls after upgrading until permissions are widened. Worth calling out explicitly in the CHANGELOG/migration notes if not already covered there.
🤖 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 `@publisher.go` around lines 382 - 416, Document the new AMQP permission
requirement introduced by PublishDelayed: because declareDelayQueue now calls
QueueDeclare for the rabbitwrap.delay.* queues, deployments need
configure/declare access for those queue names. Update the CHANGELOG or
migration notes to call out this upgrade impact and reference PublishDelayed and
declareDelayQueue so operators know restrictive allow-lists must be widened.
2 participants
Summary
Added
DelayLadder()returns the fixed set of delays supported byPublishDelayed(1s, 5s, 10s, 30s, 1m, 5m, 15m, 30m, 1h). A requested delay is rounded up
to the nearest rung so a message is never delivered early, keeping the number
of holding queues on the broker bounded.
ErrDelayTooLongsentinel error, returned byPublishDelayedwhen therequested delay exceeds the largest ladder rung.
Fixes #
PublishDelayednow actually delays delivery. Previously it set aper-message TTL and published straight to the destination, so the message was
either consumed immediately (no delay) or, if unconsumed, expired and was
silently dropped — the documented "TTL and dead letter exchange" mechanism was
half-implemented (the DLX was missing). It now publishes into a dedicated
holding queue whose queue-level TTL equals the delay and whose dead-letter
exchange/routing key point at the real destination; the message is
dead-lettered onward when its TTL expires. Using a queue-level TTL (one holding
queue per delay rung) avoids the head-of-line blocking that per-message TTL
suffers. Idle holding queues are auto-deleted by the broker (
x-expires).Verified against RabbitMQ 3.13 and 4.3; requires no broker plugin.
PublishDelayedno longer mutates the caller's*Message(it previouslyrewrote the message's
Expiration).Changes
PublishDelayeddelay semantics. A requested delay is now rounded up tothe nearest
DelayLadder()rung (minimum 1s); a delay<= 0publishesimmediately, and a delay above the largest rung returns
ErrDelayTooLong.Delivery timing is best-effort (at or shortly after the target, never before),
suitable for retry backoff rather than precise scheduling.
Motivation
Checklist
make all)Summary by CodeRabbit