(MOT-4214) feat(queue): per-binding queue subscriptions with metadata delivery - #681
Conversation
Each durable:subscriber registration now gets its own adapter
subscription keyed by a stable durable identity, and every delivery
carries the trigger's stored metadata (the harness __binding pointer)
via the new Invoker::call_delivery — without it every harness-binding
delivery is an unresolvable fire (discovery run 4 lost all 18 messages).
Subscription identity is derived at the trigger layer:
metadata.__binding when present (harness bindings), else the target
function id (plain SDK subscribers — the pre-existing queue naming, so
live deployments keep their backlogs across the upgrade). SDK trigger
ids are fresh UUIDs per registration and never key durable state;
duplicate same-key registrations share one live subscription.
Hardening from the workflow code review (15 findings, all addressed):
- unsubscribe is non-destructive on every adapter: cancel the consumer,
drain the in-flight delivery to completion, keep the queue/DLQ/store
state a same-id resubscribe reattaches to. It fires on every routine
subscriber disconnect, so deleting broker queues or purging the store
there wiped exactly the data the trigger exists to keep (store.purge
removed again with its only caller).
- shutdown stays abort-based (process exit): in-flight jobs are owned by
store inflight-recovery / broker redelivery; draining could hang exit
behind one blocked invocation.
- builtin enqueue snapshots the routing set before store I/O instead of
holding the adapter-wide mutex across FileStore writes; the race test
asserts observable behavior instead of lock internals.
- rabbitmq/redis subscribe hold the write lock from duplicate-check
through insert, closing a TOCTOU that leaked an unstoppable duplicate
consumer; with queue deletion gone, the unsubscribe/resubscribe race
that killed a freshly re-created queue is dissolved.
- rabbitmq worker drains its JoinSet instead of abort_all; FIFO
deliveries borrow instead of deep-cloning metadata per message;
Worker::run carries the repo-conventional too_many_arguments allow.
- IiiInvoker::call delegates to call_delivery; redis unsubscribe_locked
and its HashMap-only tests are inlined/removed; the priority e2e
predeclares the subscriber-id topology subscribe actually consumes.
Compat: RabbitMQ topics consumed by harness bindings leave the old
shared iii.{topic}.{function_id}.queue/.dlq bound to the fanout
exchange with no consumer after upgrade (bindings now use per-binding
queues); delete those manually. Their backlog is messages the old code
was already dropping as unresolvable. Plain-SDK subscriber queue names
are unchanged.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughThe queue subscription API now accepts optional metadata and stable subscription identifiers. Builtin, RabbitMQ, and Redis adapters use per-subscription routing, metadata-aware delivery, duplicate protection, and updated cancellation or unsubscribe behavior. Tests cover fan-out, durability, metadata propagation, and lifecycle handling. ChangesSubscription contracts and trigger wiring
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant TriggerManager
participant QueueAdapter
participant Worker
participant Invoker
TriggerManager->>QueueAdapter: subscribe with stable key and metadata
QueueAdapter->>Worker: start subscription worker
Worker->>Invoker: call_delivery with payload and metadata
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 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, 52 skipped (no docs/).
Four for four. Nicely done. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
queue/src/adapters/rabbitmq/adapter.rs (1)
486-526: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
unsubscribecan block forever on a stuck invocation.
sub_info.task_handle.awaitwaits for the worker to finish its drain loop. The worker drains all in-flightprocess_deliverytasks with no time bound. A single handler that never returns therefore blocksunsubscribe, andunsubscriberuns on the trigger registration path (queue/src/trigger.rsline 492). The caller has no way to recover.Bound the drain with a timeout and abort on expiry.
🛡️ Proposed fix
- let _ = sub_info.task_handle.await; + // Bound the drain: a handler that never returns must not block + // the control plane forever. + let drain = tokio::time::timeout( + std::time::Duration::from_secs(30), + sub_info.task_handle, + ); + if let Err(_elapsed) = drain.await { + tracing::warn!( + topic = %topic, + id = %id, + "Timed out draining in-flight deliveries; unacked messages will be redelivered" + ); + }Note:
tokio::time::timeoutconsumes theJoinHandle, so the task is detached rather than aborted on expiry. If you want the task stopped, keep a clone of theAbortHandlebefore awaiting.🤖 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 486 - 526, Update unsubscribe around sub_info.task_handle.await to bound worker shutdown with a timeout, preserving the existing wait when it completes normally. Capture the task’s AbortHandle before awaiting, abort the worker if the timeout expires, and log the timeout or cancellation outcome so unsubscribe always returns even when process_delivery remains stuck.queue/src/adapters/builtin.rs (1)
396-423: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
dlq_peekpagination over multiple subscription queues is not stable.
subscription_idsis aHashSet, so the concatenation order of per-queue results varies between calls. With more than one subscription on the topic, two successivedlq_peek(topic, offset, limit)calls with the same arguments can return different rows, andoffsetskips into an arbitrary concatenation rather than a globally ordered list. A UI that pages through the DLQ can then show duplicates and miss messages. Sorting the collected jobs by a stable field (for exampleenqueued_at_msthenid) beforeskip/takemakes the page deterministic.🔧 Proposed stabilization
- Ok(jobs + let mut jobs = jobs; + jobs.sort_by(|left, right| { + left.enqueued_at_ms + .cmp(&right.enqueued_at_ms) + .then_with(|| left.id.cmp(&right.id)) + }); + Ok(jobs .into_iter() .skip(offset as usize)🤖 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/builtin.rs` around lines 396 - 423, Update dlq_peek to sort the combined jobs from multiple subscription queues by stable fields, such as enqueued_at_ms followed by id, before applying skip(offset) and take(limit). Preserve the existing single-queue retrieval and pagination behavior while ensuring HashSet iteration order cannot affect page contents.
🧹 Nitpick comments (5)
queue/tests/e2e_rabbitmq.rs (1)
171-171: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winEvery RabbitMQ subscribe call passes
metadata: None.The new metadata argument is the main behavior added in this layer, but no RabbitMQ end-to-end test exercises it. The Redis and builtin suites carry the metadata assertions. Add one case here that subscribes with
Some(json!({"__binding": "..."}))and asserts the handler receives it throughInvoker::call_delivery. This also protects themetadata.as_ref()borrow path inWorker::runfor bothQueueMode::FifoandQueueMode::Standard.Also applies to: 416-416, 493-493
🤖 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/tests/e2e_rabbitmq.rs` at line 171, Add RabbitMQ end-to-end coverage for subscription metadata at the existing subscribe call sites and corresponding handler assertions, using Some(json!({"__binding": "..."})) and verifying Invoker::call_delivery receives it. Cover both QueueMode::Fifo and QueueMode::Standard paths while preserving the existing no-metadata cases.queue/src/adapters/rabbitmq/adapter.rs (1)
387-421: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueThe write lock now covers two broker round trips.
subsis held acrosssetup_topicandsetup_subscriber_queue. Both calls perform AMQP I/O. If the broker stalls, every other subscription operation stalls with it:unsubscribe,list_topics, andshutdownall need the sameRwLock. Duplicate protection only needs the key reserved, not the topology work.Consider inserting a placeholder or a per-key "in progress" marker, releasing the lock for the topology calls, and re-acquiring it for the final insert.
🤖 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 387 - 421, Adjust the subscription flow around the subscriptions write lock so it does not remain held during setup_topic and setup_subscriber_queue AMQP calls. Reserve the (topic, id) key with an in-progress marker while holding the lock, release it for topology setup, then re-acquire the lock to replace the marker with the completed subscription or remove it on failure, preserving duplicate-subscribe protection and cleanup.queue/src/adapter.rs (1)
198-208: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument the new
metadataparameter.The doc comment describes
idandfunction_idonly.metadatanow carries subscription identity information that adapters must forward to each delivery (seequeue/src/trigger.rsRegisteredSubscriber::metadata). Add one line so implementors know the contract is "deliver verbatim with every message".📝 Proposed doc addition
/// Register a subscriber (`id`) on `topic` that invokes `function_id` /// for each delivered message. + /// `metadata` is the subscription's stored trigger metadata; adapters + /// must deliver it verbatim with every message via + /// [`crate::trigger::Invoker::call_delivery`]. async fn subscribe(🤖 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/adapter.rs` around lines 198 - 208, Update the `subscribe` documentation to describe the `metadata` parameter, stating that it contains subscription identity information and must be forwarded verbatim with every delivered message, consistent with `RegisteredSubscriber::metadata`.queue/src/trigger.rs (1)
323-341: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueDuplicate-key resubscription picks an arbitrary registration's config.
registrations()collectsHashMapvalues, so iteration order is unspecified. When two registrations share(queue, subscription_key)with differentqueue_configvalues, the surviving config after a hot swap can differ from the one chosen atregister_subscribertime ("first registration's config wins", documented on lines 260-262). To make the choice stable, sort the snapshot bytrigger_idbefore deduplicating.♻️ Proposed stabilization
pub async fn resubscribe_all(&self) { let mut seen = HashSet::new(); - for registration in self.registrations().await { + let mut registrations = self.registrations().await; + registrations.sort_by(|left, right| left.trigger_id.cmp(&right.trigger_id)); + for registration in registrations {🤖 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/trigger.rs` around lines 323 - 341, Update resubscribe_all to sort the registrations snapshot by trigger_id before deduplicating by queue and subscription key, ensuring the same first-registration config chosen during register_subscriber is reused. Preserve the existing deduplication and subscription behavior after ordering the snapshot.queue/src/adapters/builtin.rs (1)
326-339: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the "resolve topic to subscription queues" pattern.
The same snapshot-then-aggregate block is repeated in
redrive_dlq(326-339),redrive_dlq_message(341-359),discard_dlq_message(361-379),dlq_count(381-394),dlq_peek(396-423), andtopic_stats(464-481). Each copy re-implements the lock snapshot, theSome(ids) if !ids.is_empty()guard, and the bare-topic fallback. A single helper keeps the fallback rule in one place, so a future change cannot leave one operation behind.♻️ Proposed helper
impl BuiltinAdapter { /// The internal queue names a bare topic resolves to, or `None` when the /// topic has no subscribers and operations target the bare topic name. async fn subscription_queue_names(&self, topic: &str) -> Option<Vec<String>> { let ids = self.topic_subscriptions.lock().await.get(topic).cloned()?; if ids.is_empty() { return None; } Some( ids.iter() .map(|id| internal_queue_name(topic, id)) .collect(), ) } }async fn redrive_dlq(&self, topic: &str) -> anyhow::Result<u64> { - let subscription_ids = self.topic_subscriptions.lock().await.get(topic).cloned(); - match subscription_ids { - Some(subscription_ids) if !subscription_ids.is_empty() => { - let mut total = 0u64; - for subscription_id in &subscription_ids { - let queue_name = internal_queue_name(topic, subscription_id); - total += self.store.redrive_dlq(&queue_name).await; - } - Ok(total) - } - _ => Ok(self.store.redrive_dlq(topic).await), - } + let Some(queue_names) = self.subscription_queue_names(topic).await else { + return Ok(self.store.redrive_dlq(topic).await); + }; + let mut total = 0u64; + for queue_name in &queue_names { + total += self.store.redrive_dlq(queue_name).await; + } + Ok(total) }🤖 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/builtin.rs` around lines 326 - 339, Extract the repeated topic-to-subscription resolution into a BuiltinAdapter helper such as subscription_queue_names, returning internal queue names for non-empty subscriber lists and None for missing or empty lists. Update redrive_dlq, redrive_dlq_message, discard_dlq_message, dlq_count, dlq_peek, and topic_stats to use this helper while preserving their bare-topic fallback behavior.
🤖 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 `@queue/src/adapters/rabbitmq/worker.rs`:
- Around line 226-233: Update the unexpected termination path in the worker run
flow so a consumer stream ending without explicit cancellation notifies the
RabbitMQ adapter to remove the matching (topic, subscription_id) record from
subscriptions, allowing a later subscribe to recreate the consumer. Preserve the
existing cancellation behavior and warning log, and reuse the adapter’s existing
unsubscribe or cleanup mechanism rather than only logging the termination.
In `@queue/src/trigger.rs`:
- Around line 389-403: Update unregister so the registrations mutex guard is
released before awaiting adapter.unsubscribe: remove the registration and
compute still_shared while holding the lock, then drop the guard and perform the
unsubscribe only when needed. Apply the same release-before-await pattern to
register_subscriber’s adapter.subscribe path if practical, preserving existing
registration and subscription behavior.
In `@queue/tests/e2e_rabbitmq.rs`:
- Line 64: Update the article in the doc comment line that begins with `///
real` to use the correct grammatical article before the word "idempotent".
Change "a idempotent" to "an idempotent" since idempotent begins with a vowel
sound.
In `@queue/tests/e2e_redis.rs`:
- Around line 177-189: In the delivery assertions after wait_for_fires, add an
exact count assertion that deliveries.len() equals 2 before the existing all and
any content checks, preserving the current payload and metadata validations.
---
Outside diff comments:
In `@queue/src/adapters/builtin.rs`:
- Around line 396-423: Update dlq_peek to sort the combined jobs from multiple
subscription queues by stable fields, such as enqueued_at_ms followed by id,
before applying skip(offset) and take(limit). Preserve the existing single-queue
retrieval and pagination behavior while ensuring HashSet iteration order cannot
affect page contents.
In `@queue/src/adapters/rabbitmq/adapter.rs`:
- Around line 486-526: Update unsubscribe around sub_info.task_handle.await to
bound worker shutdown with a timeout, preserving the existing wait when it
completes normally. Capture the task’s AbortHandle before awaiting, abort the
worker if the timeout expires, and log the timeout or cancellation outcome so
unsubscribe always returns even when process_delivery remains stuck.
---
Nitpick comments:
In `@queue/src/adapter.rs`:
- Around line 198-208: Update the `subscribe` documentation to describe the
`metadata` parameter, stating that it contains subscription identity information
and must be forwarded verbatim with every delivered message, consistent with
`RegisteredSubscriber::metadata`.
In `@queue/src/adapters/builtin.rs`:
- Around line 326-339: Extract the repeated topic-to-subscription resolution
into a BuiltinAdapter helper such as subscription_queue_names, returning
internal queue names for non-empty subscriber lists and None for missing or
empty lists. Update redrive_dlq, redrive_dlq_message, discard_dlq_message,
dlq_count, dlq_peek, and topic_stats to use this helper while preserving their
bare-topic fallback behavior.
In `@queue/src/adapters/rabbitmq/adapter.rs`:
- Around line 387-421: Adjust the subscription flow around the subscriptions
write lock so it does not remain held during setup_topic and
setup_subscriber_queue AMQP calls. Reserve the (topic, id) key with an
in-progress marker while holding the lock, release it for topology setup, then
re-acquire the lock to replace the marker with the completed subscription or
remove it on failure, preserving duplicate-subscribe protection and cleanup.
In `@queue/src/trigger.rs`:
- Around line 323-341: Update resubscribe_all to sort the registrations snapshot
by trigger_id before deduplicating by queue and subscription key, ensuring the
same first-registration config chosen during register_subscriber is reused.
Preserve the existing deduplication and subscription behavior after ordering the
snapshot.
In `@queue/tests/e2e_rabbitmq.rs`:
- Line 171: Add RabbitMQ end-to-end coverage for subscription metadata at the
existing subscribe call sites and corresponding handler assertions, using
Some(json!({"__binding": "..."})) and verifying Invoker::call_delivery receives
it. Cover both QueueMode::Fifo and QueueMode::Standard paths while preserving
the existing no-metadata cases.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 37118b0f-d244-4a91-b41e-b48b2b09f4fa
📒 Files selected for processing (17)
queue/src/adapter.rsqueue/src/adapters/builtin.rsqueue/src/adapters/memory.rsqueue/src/adapters/rabbitmq/adapter.rsqueue/src/adapters/rabbitmq/naming.rsqueue/src/adapters/rabbitmq/publisher.rsqueue/src/adapters/rabbitmq/retry.rsqueue/src/adapters/rabbitmq/topology.rsqueue/src/adapters/rabbitmq/worker.rsqueue/src/adapters/redis.rsqueue/src/configuration.rsqueue/src/functions.rsqueue/src/runtime.rsqueue/src/trigger.rsqueue/tests/e2e_durability.rsqueue/tests/e2e_rabbitmq.rsqueue/tests/e2e_redis.rs
Close the two coverage gaps worth closing from the review pass: - rabbitmq e2e (docker-only, no engine): unsubscribe detaches the consumer, publishes while detached buffer on the still-bound durable queue, and a same-id resubscribe drains them — the broker-side half of the non-destructive-unsubscribe contract, previously untested anywhere. - trigger unit test: resubscribe_all collapses registrations sharing a subscription key to one subscribe against the swapped-in adapter. The remaining untested fixes are the subscribe TOCTOU lock scopes — inherently racy to pin, left structural.
Review follow-up (MOT-4214). CodeRabbit flagged that unregister holds the registrations lock across adapter.unsubscribe, which awaited the consumer drain — up to one whole invocation — stalling every trigger register/unregister behind one slow job. Releasing the lock (the suggested fix) would trade that for a race: a same-key register can run between the handler-lock release and the adapter's map removal, no-op on the adapter's duplicate check, and end up registered with no live consumer. Fixed at the root instead: builtin and rabbitmq unsubscribe now DETACH their worker task (drop the JoinHandle) rather than await it — the in-flight delivery still completes and acks, unsubscribe becomes a broker round-trip, and the handler lock stays held so the no-race property is preserved. The drain unit test now pins the detach contract (prompt return; delivery still completes and acks). Also from review: exact delivery-count assert in the redis fan-out e2e (a duplicate subscription now fails it) and a doc-comment typo. The consumer-stream-end-leaves-dead-record observation is pre-existing (same shape before this branch) and tracked as GH-22 (channel-level error recovery for the shared Arc<Channel>).
andersonleal
changed the title
1 participant
What
Each
durable:subscriberregistration now gets its own adapter subscription, and every delivery carries the trigger's stored metadata via the newInvoker::call_delivery. For harness-managed bindings that metadata is the__bindingpointer the delivery hop resolves — without it every binding delivery is an unresolvable fire (discovery run 4 lost all 18 messages to exactly that). Two bindings on the same function each get their own queue, their own copy, and their own metadata.Durable identity
Subscription identity is derived at the trigger layer (
RegisteredSubscriber::subscription_key):metadata.__bindingwhen present (harness bindings) — stable across restarts by definition.function_idotherwise (plain SDK subscribers) — the pre-existing queue naming, so live deployments keep their queue names and backlogs across the upgrade.SDK trigger ids are fresh UUIDs per
register_triggercall and never key durable state. Duplicate registrations sharing a key share one live subscription (the adapter unsubscribes only when the last one goes), so leftover re-registrations don't multiply deliveries.Lifecycle semantics
store.purgeis removed along with its only caller).Review hardening
A
/code-review maxpass over the branch produced 15 verified findings; all are fixed here. Highlights beyond the identity/lifecycle work above:subscribehold the write lock from duplicate-check through insert, closing a TOCTOU that leaked an unstoppable duplicate consumer; with queue deletion gone, the unsubscribe/resubscribe race that deleted a freshly re-created queue is dissolved.enqueuesnapshots the routing set before store I/O instead of holding the adapter-wide mutex across blocking FileStore writes; the race test asserts observable behavior instead of lock internals.abort_all; FIFO deliveries borrow instead of deep-cloning metadata per message.IiiInvoker::calldelegates tocall_delivery; redisunsubscribe_lockedand its HashMap-only tests are inlined/removed; the priority e2e predeclares the subscriber-id topologysubscribeactually consumes.Compat
RabbitMQ topics consumed by harness bindings leave the old shared
iii.{topic}.{function_id}.queue/.dlqbound to the fanout exchange with no consumer after upgrade — delete those queues manually or they keep accumulating fanout copies. Their backlog is messages the old code was already dropping as unresolvable. Plain-SDK subscriber queue names are unchanged, so their backlogs carry over untouched.Verification
cargo test --all-features: 130 unit + durability/redis-shape tests pass. Docker-backed e2e skipped locally (no docker socket access);missing_dlq_inspection_does_not_close_consumer_channelhard-requires docker and fails on any docker-less machine — pre-existing behavior, gate untouched.cargo clippy --all-targets --all-features -- -D warnings: clean.cargo fmt --check: clean.Summary by CodeRabbit
New Features
Bug Fixes
Tests
Ticket
Queue-worker slice of MOT-4214 (trigger-owning workers dropping the fire-time metadata sidecar).