fix(queue): handle RabbitMQ function queue DLQs - #613
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughRabbitMQ DLQ operations now resolve configured function queues to internal topology names and use isolated channels for inspection, counting, redrive, and statistics. DLQ decoding accepts normalized records, and tests cover redrive behavior and missing-queue channel safety. ChangesRabbitMQ DLQ behavior
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant QueueAPI
participant RabbitMQAdapter
participant RabbitMQ
participant Consumer
QueueAPI->>RabbitMQAdapter: inspect function-queue DLQ
RabbitMQAdapter->>RabbitMQ: resolve internal DLQ name
RabbitMQAdapter->>RabbitMQ: open operation channel
RabbitMQ-->>RabbitMQAdapter: return DLQ data
RabbitMQAdapter-->>QueueAPI: return messages or count
Consumer->>RabbitMQ: continue using consumer channel
Possibly related issues
Possibly related PRs
Suggested reviewers: 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 |
skill-check — worker0 verified, 49 skipped (no docs/).
Four for four. Nicely done. |
guibeira
added
the
no-ticket
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
queue/src/adapters/rabbitmq/adapter.rs (1)
525-573: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winEnable and validate publisher confirms before acknowledging DLQ redrive messages.
operation_channel("DLQ redrive")creates a channel withoutconfirm_select, so the function-queuebasic_publishfuture resolves toNotRequested; if the publish fails, the DLQ delivery can still be removed and lost. Enable confirms on this channel and reject non-is_ack()confirmations before acknowledging.Proposed fix
let channel = self.operation_channel("DLQ redrive").await?; +channel + .confirm_select(ConfirmSelectOptions::default()) + .await + .map_err(|e| anyhow::anyhow!("Failed to enable redrive publisher confirms: {}", e))?; - channel +let confirmation = channel .basic_publish( &names.exchange(), queue_name, BasicPublishOptions::default(), &delivery.delivery.data, properties, ) .await .map_err(|e| anyhow::anyhow!("Failed to republish: {}", e))? .await .map_err(|e| anyhow::anyhow!("Failed to confirm: {}", e))?; +if !confirmation.is_ack() { + anyhow::bail!("RabbitMQ did not acknowledge DLQ redrive publish"); +}🤖 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 `@queue/src/adapters/rabbitmq/adapter.rs` around lines 525 - 573, Update the DLQ redrive flow around operation_channel("DLQ redrive") to enable publisher confirms on the channel before publishing. In the function-queue republish path, validate the confirmation returned by basic_publish and reject any non-is_ack() result before acknowledging the original DLQ delivery, preserving the existing error propagation.
🤖 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.
Outside diff comments:
In `@queue/src/adapters/rabbitmq/adapter.rs`:
- Around line 525-573: Update the DLQ redrive flow around operation_channel("DLQ
redrive") to enable publisher confirms on the channel before publishing. In the
function-queue republish path, validate the confirmation returned by
basic_publish and reject any non-is_ack() result before acknowledging the
original DLQ delivery, preserving the existing error propagation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: d91acbf0-cfa8-4270-9366-1eb8ea2b9fa5
📒 Files selected for processing (3)
queue/src/adapters/rabbitmq/adapter.rsqueue/src/functions.rsqueue/tests/e2e_rabbitmq.rs
1 participant
Summary
Root cause
list_topicsexposes named function queues using their bare name, but RabbitMQ DLQ operations interpreted that value as a regular topic. The resulting passive declaration targeted a queue that did not exist, causing RabbitMQ to close the shared AMQP channel. RabbitMQ also returned normalized DLQ objects while the public function only attempted to deserialize legacyJobvalues, silently dropping those messages.Impact
Named RabbitMQ function queues now appear in DLQ listings, their messages can be browsed and redriven using the public queue name, and a failed DLQ inspection cannot stop unrelated consumers.
Fixes iii-hq/iii#2017
Smoke test
The recording runs the full III flow against RabbitMQ through Docker Compose. It first reproduces all three failures with
queue/v0.2.2, then verifies the same flow with this branch.Validation
cargo fmt --all -- --checkgit diff --checkcargo test(124 unit tests, 2 durability E2E tests, 5 RabbitMQ E2E tests, and 2 Redis E2E tests)cargo clippy --all-targets --all-features -- -D warningsqueue/v0.2.2and this branch with RabbitMQ in Docker ComposeSummary by CodeRabbit
Bug Fixes
Reliability