feat: queue worker (durable:subscriber migration) - #424
Merged
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughA new standalone Rust "queue" worker crate is introduced, implementing durable pub/sub queueing via a pluggable ChangesQueue Worker Implementation
Estimated code review effort: 5 (Critical) | ~150 minutes Sequence Diagram(s)sequenceDiagram
participant Main as queue::main
participant Boot as boot::start
participant Config as configuration
participant Adapter as SwappableAdapter
participant Trigger as QueueTriggerHandler
Main->>Boot: start(iii, config)
Boot->>Boot: guard_against_builtin_iii_queue()
Boot->>Adapter: build_adapter(config)
Boot->>Trigger: register durable:subscriber trigger
Main->>Config: register_config_trigger(adapter, trigger, config)
Config-->>Main: config-change trigger registered
Note over Config,Adapter: On configuration update
Config->>Config: swap_needed(old, new)
Config->>Adapter: build_adapter(new_config)
Config->>Adapter: replace(new_adapter, name)
Config->>Trigger: resubscribe_all(new_adapter)
sequenceDiagram
participant Publisher
participant Adapter as QueueAdapter
participant Store as QueueStore
participant Worker as Poller/Worker
participant Invoker
Publisher->>Adapter: enqueue(topic, payload)
Adapter->>Store: enqueue job
Worker->>Store: dequeue job
Worker->>Invoker: call(function_id, payload)
alt success
Invoker-->>Worker: Ok
Worker->>Store: ack(job_id)
else failure
Invoker-->>Worker: Err
Worker->>Store: nack(job, retries, backoff)
Store->>Store: move to DLQ if exhausted
end
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 |
Contributor
skill-check — worker0 verified, 38 skipped (no docs/).
Four for four. Nicely done. |
… boot, unique e2e queue names - store: drop redundant .into_iter() (clippy 1.96 useless_conversion) - boot: treat absent/Null adapter.config as builtin defaults instead of failing deserialization; regression tests for the no-config boot path (this was the interface-boot-smoke CI failure) - e2e: unique queue names per run — the engine re-delivers persisted durable:subscriber triggers on boot, so fixed names race against leftover consumers from earlier local runs
guibeira
force-pushed
the
feat/queue-worker
branch
from
July 6, 2026 22:52
0f14bb0 to
0609952
Compare
guibeira
marked this pull request as ready for review
July 6, 2026 22:56
…iority, backoff fields)
Port engine/src/workers/queue/adapters/rabbitmq/{types.rs,naming.rs}
verbatim (pure logic, no engine dependency) as the first step of the
RabbitMQ adapter port. Adds the `rabbitmq` cargo feature (default on,
gates `dep:lapin`) and wires queue/src/adapters/mod.rs to include the
new module tree only under that feature.
…apter Port the remaining engine/src/workers/queue/adapters/rabbitmq modules (topology.rs, retry.rs, publisher.rs, consumer.rs, worker.rs, adapter.rs) implementing the full QueueAdapter trait. Compiles and passes clippy under both --features rabbitmq (default) and --no-default-features. Adds a `max_priority: Option<u8>` field to the worker's minimal FunctionQueueConfig (queue/src/adapter.rs) so setup_function_queue can pass priority-queue declaration through, matching the engine's config shape.
Complete the RabbitMQ transport adapter port (types/naming from the first commit; topology/retry/publisher/consumer/worker/adapter from the second) with e2e coverage: - queue/tests/common/docker.rs: add a rabbitmq:3-alpine container starter (start_rabbitmq), generalizing the existing redis port-resolution helper. Readiness is polled via a real lapin AMQP connect/close loop (RabbitMQ opens its TCP port well before the AMQP handshake is actually served), not a bare TCP check. - queue/tests/e2e_rabbitmq.rs: four connect-or-skip scenarios against a live iii engine + docker RabbitMQ -- basic delivery; priority ordering (3 messages published highest-priority-last, pre-declared+pre-published before any consumer attaches, delivered 9/5/1); fifo mode (10 messages delivered in publish order); and retry-then-DLQ-then-redrive exercised through the function-queue trait methods directly (setup/publish/consume/ ack/nack_function_queue), since `resolve_dlq_name` for a bare (non-`__fn_queue::`) topic resolves to a queue name `subscribe()` never actually declares -- a mismatch inherited verbatim from the engine (its own rabbitmq_queue_integration.rs test suite never exercises DLQ ops against a bare subscribed topic either). All gates pass: cargo test (default + --no-default-features), cargo fmt --check, cargo clippy --all-targets (default + --no-default-features) -D warnings. e2e_rabbitmq's 4 scenarios verified passing against a real `iii` engine and a real rabbitmq:3-alpine container, not just compiling.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
🧹 Nitpick comments (11)
queue/src/store.rs (1)
63-67: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftSingle global lock across all topics.
SharedStoreguards every topic's queue/DLQ/stats behind oneMutex, so unrelated topics contend for the same lock on every operation. Combined with the full-state clone on each mutation (see other comment), this caps throughput as the number of topics/messages grows. Consider per-topic sharding (e.g.DashMap/Mutexper topic) if this store needs to scale beyond light/dev workloads.🤖 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/store.rs` around lines 63 - 67, The SharedStore design uses one global Mutex over all StoreData, which serializes queue/DLQ/stats access for every topic and creates unnecessary contention as traffic grows. Refactor the store around per-topic sharding in SharedStore and StoreData, such as a map of topic-specific mutexes or a DashMap-backed structure, so unrelated topics can mutate independently while keeping the existing queue/DLQ/stats behavior intact.queue/src/configuration.rs (1)
184-197: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winFragile string-matching for "config not found".
Classifying "not found" via
e.to_ascii_uppercase().contains("NOT_FOUND")tiestry_get_config_value(and transitivelyshould_seed_initial_value/fetch_config, which gate first-boot seeding and config reload) to the exact wording ofconfiguration::get's error message. If that message format changes, this worker would treat "not found" as a hard failure instead of the expected empty-config case.Please confirm whether the
configurationworker exposes a typed/structured error (e.g., an error code field) that could be matched on instead of a raw string substring.🤖 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/configuration.rs` around lines 184 - 197, `try_get_config_value` is using fragile string matching to detect the empty-config case, which is coupled to the exact wording returned by `trigger_with_retry`/`configuration::get`. Update this path to use a typed or structured error from the configuration worker (for example an error code or enum field) and match that instead of checking `to_ascii_uppercase().contains("NOT_FOUND")`; keep the `should_seed_initial_value` and `fetch_config` behavior the same by mapping only the structured “not found” case to `Ok(None)`.queue/src/trigger.rs (1)
149-178: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRegistrations lock held across
adapter.subscribe().await.Unlike
unregister/shutdown, which drop the lock before calling into the adapter,register_subscriberkeepsregistrationslocked for the duration ofself.adapter.subscribe(...).await. If subscribing blocks on real network/topology setup (Redis/RabbitMQ adapters aren't in this file), every concurrentregister_subscriber/unregister/registrations/shutdowncall — including a gracefulshutdown()— is serialized behind that one call.♻️ Suggested restructure
- let mut registrations = self.registrations.lock().await; - if registrations.contains_key(®istration.trigger_id) { - return Err(format!( - "durable:subscriber trigger '{}' is already registered", - registration.trigger_id - )); - } - + { + let registrations = self.registrations.lock().await; + if registrations.contains_key(®istration.trigger_id) { + return Err(format!( + "durable:subscriber trigger '{}' is already registered", + registration.trigger_id + )); + } + } let queue_config = merged_queue_config(®istration.spec); self.adapter .subscribe(...) .await; - registrations.insert(registration.trigger_id.clone(), registration); + self.registrations.lock().await.insert(registration.trigger_id.clone(), registration); Ok(())Note: re-check the duplicate condition after re-acquiring the lock to close the race between the two lock sections.
🤖 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 149 - 178, `register_subscriber` in `trigger.rs` holds the `registrations` mutex while awaiting `self.adapter.subscribe(...)`, which blocks other registration and shutdown paths. Refactor `register_subscriber` to acquire the lock only for the duplicate check, release it before calling `adapter.subscribe().await`, then re-acquire it afterward to insert the `RegisteredSubscriber`. Be sure to re-check `registrations.contains_key(®istration.trigger_id)` after the await to avoid the race introduced by splitting the lock scope, matching the approach used by `unregister` and `shutdown`.queue/src/adapters/builtin.rs (2)
496-518: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winJob invocation failure reason is dropped before nacking.
Err(_err)discards the invoker's error message without logging it, unlikeredis.rs's equivalent path (which logs viatracing::error!before continuing). This makes diagnosing why a job is retrying/heading to DLQ harder in production.🔧 Proposed fix
- Err(_err) => { + Err(err) => { + tracing::warn!(error = %err, function_id = %cfg.function_id, job_id = %job.id, "Job handler failed"); store .nack(&cfg.queue_name, job, cfg.max_retries, cfg.backoff_ms) .await }🤖 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 496 - 518, The process_job path in builtin.rs is dropping the invoker failure details in the Err(_err) branch before calling store.nack, unlike the equivalent handling in redis.rs. Update process_job to capture the error from invoke and log it with tracing::error! (or the same logging style used in the Redis adapter) before nacking the job, so the failure reason is preserved when retries or DLQ routing happen.
158-184: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winEnqueue errors are silently swallowed.
Both fan-out and bare-topic paths discard
store.enqueue'sResultwithlet _ = ...(lines 177, 181), with no logging. For aFileStore-backed queue, an I/O failure here would vanish without a trace, unlike other failure paths in this file which log viatracing::warn!/tracing::error!.🔧 Proposed fix
- for function_id in &function_ids { - let queue_name = internal_queue_name(topic, function_id); - let _ = self.store.enqueue(&queue_name, data.clone()).await; - } + for function_id in &function_ids { + let queue_name = internal_queue_name(topic, function_id); + if let Err(e) = self.store.enqueue(&queue_name, data.clone()).await { + tracing::error!(error = %e, topic = %topic, queue = %queue_name, "Failed to enqueue message"); + } + }🤖 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 158 - 184, The enqueue path in builtin.rs is swallowing store.enqueue failures in both the fan-out and bare-topic branches of enqueue, so add explicit error handling instead of discarding the Result. Update enqueue to log failures with tracing::warn! or tracing::error! and include the topic or derived queue_name plus the underlying error, matching the logging style used elsewhere in this file. Use the enqueue method, topic_functions fan-out logic, and internal_queue_name-derived queues to locate the two call sites.queue/src/adapters/rabbitmq/consumer.rs (1)
41-70: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNo unit tests for
JobParser.
redis.rsin this same PR extracts pure logic (resolve_redis_url,unsubscribe_locked) specifically to unit-test it without a live connection.extract_attempts/parse_from_deliveryare similarly pure/testable (given aDeliverywith headers) but have no coverage here.🤖 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/consumer.rs` around lines 41 - 70, Add unit tests for JobParser to cover the pure parsing logic in parse_from_delivery and extract_attempts. Create tests that build a Delivery with headers and verify attempts_made is set for the supported AMQPValue numeric variants, and that parse_from_delivery preserves the deserialized Job when headers are absent or irrelevant. Use the JobParser, parse_from_delivery, and extract_attempts symbols to locate the code and keep the tests focused on header parsing without needing a live RabbitMQ connection.queue/src/adapters/redis.rs (1)
61-66: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid locking
ConnectionManageron every publish.redis::aio::ConnectionManageris cloneable and meant to be shared concurrently;Arc<Mutex<_>>serializesenqueueandpublish_to_function_queuebehind one lock and removes the multiplexing benefit. Store it directly and clone it per call instead.🤖 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/redis.rs` around lines 61 - 66, The RedisAdapter currently wraps ConnectionManager in Arc<Mutex<_>>, which serializes publish/enqueue calls and defeats its concurrent cloneable usage. Update RedisAdapter to store the ConnectionManager directly, then adjust enqueue and publish_to_function_queue to clone and use a per-call ConnectionManager instead of locking it; keep the subscriber and subscriptions handling unchanged.queue/src/functions.rs (2)
50-55: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
RedriveSingleResult.redrivenis a misleading field name when reused for discard.
discard_messagereturnsRedriveSingleResult { .., redriven: u64::from(found) }, but the operation discards (not redrives) the message. Reusing a field literally named "redriven" to indicate a successful discard is confusing to any consumer parsing this response.♻️ Suggested rename
-#[derive(Debug, Clone, Serialize, JsonSchema, PartialEq, Eq)] -pub struct RedriveSingleResult { - pub queue: String, - pub message_id: String, - pub redriven: u64, -} +#[derive(Debug, Clone, Serialize, JsonSchema, PartialEq, Eq)] +pub struct RedriveSingleResult { + pub queue: String, + pub message_id: String, + pub found: u64, +}Also applies to: 235-249
🤖 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/functions.rs` around lines 50 - 55, The shared RedriveSingleResult type uses a field named redriven, but discard_message reuses it to report discard success, which is misleading for consumers. Rename or split the response shape around RedriveSingleResult and discard_message so the discard path returns a semantically correct field name (for example, discard/discarded) and update any constructors/usages that currently set redriven to u64::from(found) to match the new meaning.
313-333: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRoute DLQ paging through
dlq_peekinqueue/src/functions.rs:313-333
dlq_messagesalways starts from offset0and applies the requested page in memory. Calldlq_peek(&input.topic, input.offset, input.limit)here so the adapter sees the real page boundaries instead of fetching the full prefix first.🤖 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/functions.rs` around lines 313 - 333, The dlq_messages function is paginating in memory by calling adapter.dlq_messages with a count derived from offset + limit, which ignores the real page boundaries. Update dlq_messages to route the request through dlq_peek using input.topic, input.offset, and input.limit so the SwappableAdapter handles paging directly, and keep the existing validation and result mapping in place.queue/tests/e2e_redis.rs (1)
126-126: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated DLQ-not-supported string literal.
This duplicates the private
DLQ_NOT_SUPPORTEDconstant inqueue/src/adapters/redis.rs. Consider making that constantpub(crate)and importing it here instead of re-typing the literal, so the two can't silently drift apart.🤖 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_redis.rs` at line 126, The DLQ-not-supported message in the Redis E2E test is duplicated as a local string literal and can drift from the source of truth in RedisAdapter. Update the test to reuse the existing constant from RedisAdapter by making DLQ_NOT_SUPPORTED accessible within the crate (for example, pub(crate)) and importing it in queue/tests/e2e_redis.rs, so the test references the same symbol instead of re-typing the message.queue/tests/e2e_durability.rs (1)
85-97: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated polling helper across e2e test files.
wait_for_fireshere is duplicated verbatim inqueue/tests/e2e_redis.rs(lines 20-32), andqueue/tests/e2e_rabbitmq.rs'swait_until(lines 30-43) is a more generic variant of the same pattern. Consider moving a shared polling helper intoqueue/tests/common/mod.rsfor reuse across all three e2e suites.🤖 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_durability.rs` around lines 85 - 97, The polling helper logic is duplicated across the e2e test suites, so extract it into a shared helper in queue/tests/common/mod.rs and reuse it from e2e_durability::wait_for_fires, e2e_redis, and e2e_rabbitmq. Keep the shared helper generic enough to cover both the exact fire-count wait and the more general wait_until pattern, then replace the per-file copies with calls to that shared function.
🤖 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/adapter.rs`:
- Around line 229-259: The SwappableAdapter state can be observed out of sync
because replace() updates inner and name with separate RwLock writes. Refactor
SwappableAdapter so the adapter and its identity are stored and swapped together
atomically, and make current() and current_name() read from the same paired
state. Use the existing SwappableAdapter, current(), current_name(), and
replace() methods as the touchpoints, and ensure callers never see a new
QueueAdapter with a stale transport name or the reverse.
In `@queue/src/adapters/builtin.rs`:
- Around line 463-494: Detached per-job tasks in run_concurrent keep running
after unsubscribe/shutdown because only the outer polling task is aborted.
Update run_concurrent to track spawned job handlers with a local JoinSet (or
equivalent scoped task manager) instead of discarding the tokio::spawn
JoinHandle, so dropping/aborting the poller also cancels in-flight jobs. Keep
the change localized around run_concurrent, process_job, and the
unsubscribe/shutdown paths that currently abort the parent task.
In `@queue/src/adapters/rabbitmq/adapter.rs`:
- Around line 318-401: The subscription flow in the RabbitMQ adapter still has a
race because `already_subscribed` only checks under a read lock, then releases
it before `setup_topic`, `setup_subscriber_queue`, and `tokio::spawn`, so
concurrent calls for the same `topic:id` can create duplicate workers and
overwrite `SubscriptionInfo`. Fix `subscribe` in `adapter.rs` by reserving the
key with a write-locked `subscriptions` path up front, ideally using
`entry`/placeholder state before any async setup, and only spawning the
`Worker::run` task after the reservation is in place so the second caller exits
before consumer creation.
- Around line 93-102: The DLQ resolution currently falls back to a topic-level
name in RabbitMQAdapter::resolve_dlq_name, so per-function subscriber dead
letters are not found unless the input already uses the __fn_queue:: prefix.
Update the DLQ lookup path used by the browse/redrive/count APIs to resolve
subscriber queues with the function-aware name (iii.{topic}.{function_id}.dlq)
instead of the topic-only RabbitNames::new(topic).dlq(), and make sure the
dlq_topics aggregation reflects those durable subscriber DLQs as well.
In `@queue/src/boot.rs`:
- Around line 54-61: The registration result from
`iii.register_trigger_type(...)` is being ignored, so `start()` can succeed even
when `TRIGGER_TYPE` never registers. Update `start()` in `boot` to handle the
`Result` from `register_trigger_type` explicitly: log the failure with context
and return an error instead of continuing to build `BootHandle`. Use the
existing `trigger_handler`/`RegisterTriggerType::new(...)` call site to locate
the fix and make sure a failed `durable:subscriber` registration cannot be
reported as a healthy boot.
In `@queue/src/functions.rs`:
- Around line 251-284: The topic count fields are being hardcoded to zero in the
list_topics and topic_stats flows, which makes the returned admin data
misleading. Update TopicInfo and TopicStatsOutput construction in list_topics
and topic_stats to either pass through real subscriber/consumer counts from the
adapter results when available, or remove those fields from the response shape
if the adapter cannot provide them. Use the list_topics, topic_stats, TopicInfo,
and TopicStatsOutput symbols to locate the affected code and ensure the API no
longer returns fake counts.
In `@queue/src/store.rs`:
- Around line 204-227: The mutating store paths in enqueue and the related
dequeue/ack/nack/redrive/discard functions are cloning the full StoreData even
when no persistence is configured. Move the file_dir check earlier in each
function so the snapshot clone is only created when SharedStore.file_dir is
Some, and otherwise call persist_if_needed without a cloned snapshot or skip
that path entirely. Use the existing enqueue helper and the other mutating
methods as the places to apply this lazy-snapshot pattern consistently.
- Around line 96-108: FileStore::open, load_snapshot, and persist_if_needed are
performing synchronous filesystem work directly inside async paths, which can
block the Tokio runtime thread under load. Move the create/read/write/rename
directory and snapshot operations onto a blocking-safe path such as a dedicated
blocking task or async filesystem APIs, and keep the async methods non-blocking.
Update the file-backed persistence flow in persist_if_needed and the startup
load path in FileStore::open to use the same offloaded approach so all disk I/O
is handled consistently.
In `@queue/src/trigger.rs`:
- Around line 41-55: IiiInvoker::call is triggering downstream work with
timeout_ms set to None, leaving per-message processing unbounded and able to
hang the consumer indefinitely. Update the TriggerRequest construction in
IiiInvoker::call to pass a concrete timeout value instead of None, using the
same timeout discipline as trigger_with_retry in configuration.rs (prefer the
existing CONFIG_BUS_TIMEOUT_MS pattern or an equivalent per-message timeout
source). Keep the change localized to the IiiInvoker implementation and ensure
the returned Result still maps the trigger response/errors as before.
In `@queue/tests/common/docker.rs`:
- Around line 63-83: The container startup flow in start_redis and
start_rabbitmq can leak a running container if resolve_mapped_port returns None
before the Drop-guarded RedisContainer/RabbitMqContainer is created. Restructure
the logic so the container is wrapped in its guard immediately after obtaining
container_id, then call resolve_mapped_port through that guarded value (or
otherwise ensure cleanup still happens on early return). Mirror the same fix in
both helper functions that start containers, using the RedisContainer and
RabbitMqContainer types as the cleanup owners.
In `@queue/tests/e2e_rabbitmq.rs`:
- Around line 199-222: The RabbitMQ DLQ helpers currently resolve bare
subscribed topics to the wrong queue name, which can hit a non-existent DLQ and
poison the shared channel. Update the DLQ-related paths in RabbitMQAdapter,
especially resolve_dlq_name and the callers used by dlq_count, redrive_dlq,
redrive_dlq_message, and discard_dlq_message, so they use the subscriber-scoped
DLQ created by subscribe() via setup_subscriber_queue
(iii.{topic}.{function_id}.dlq) or clearly restrict these APIs to
function-queue-only names.
---
Nitpick comments:
In `@queue/src/adapters/builtin.rs`:
- Around line 496-518: The process_job path in builtin.rs is dropping the
invoker failure details in the Err(_err) branch before calling store.nack,
unlike the equivalent handling in redis.rs. Update process_job to capture the
error from invoke and log it with tracing::error! (or the same logging style
used in the Redis adapter) before nacking the job, so the failure reason is
preserved when retries or DLQ routing happen.
- Around line 158-184: The enqueue path in builtin.rs is swallowing
store.enqueue failures in both the fan-out and bare-topic branches of enqueue,
so add explicit error handling instead of discarding the Result. Update enqueue
to log failures with tracing::warn! or tracing::error! and include the topic or
derived queue_name plus the underlying error, matching the logging style used
elsewhere in this file. Use the enqueue method, topic_functions fan-out logic,
and internal_queue_name-derived queues to locate the two call sites.
In `@queue/src/adapters/rabbitmq/consumer.rs`:
- Around line 41-70: Add unit tests for JobParser to cover the pure parsing
logic in parse_from_delivery and extract_attempts. Create tests that build a
Delivery with headers and verify attempts_made is set for the supported
AMQPValue numeric variants, and that parse_from_delivery preserves the
deserialized Job when headers are absent or irrelevant. Use the JobParser,
parse_from_delivery, and extract_attempts symbols to locate the code and keep
the tests focused on header parsing without needing a live RabbitMQ connection.
In `@queue/src/adapters/redis.rs`:
- Around line 61-66: The RedisAdapter currently wraps ConnectionManager in
Arc<Mutex<_>>, which serializes publish/enqueue calls and defeats its concurrent
cloneable usage. Update RedisAdapter to store the ConnectionManager directly,
then adjust enqueue and publish_to_function_queue to clone and use a per-call
ConnectionManager instead of locking it; keep the subscriber and subscriptions
handling unchanged.
In `@queue/src/configuration.rs`:
- Around line 184-197: `try_get_config_value` is using fragile string matching
to detect the empty-config case, which is coupled to the exact wording returned
by `trigger_with_retry`/`configuration::get`. Update this path to use a typed or
structured error from the configuration worker (for example an error code or
enum field) and match that instead of checking
`to_ascii_uppercase().contains("NOT_FOUND")`; keep the
`should_seed_initial_value` and `fetch_config` behavior the same by mapping only
the structured “not found” case to `Ok(None)`.
In `@queue/src/functions.rs`:
- Around line 50-55: The shared RedriveSingleResult type uses a field named
redriven, but discard_message reuses it to report discard success, which is
misleading for consumers. Rename or split the response shape around
RedriveSingleResult and discard_message so the discard path returns a
semantically correct field name (for example, discard/discarded) and update any
constructors/usages that currently set redriven to u64::from(found) to match the
new meaning.
- Around line 313-333: The dlq_messages function is paginating in memory by
calling adapter.dlq_messages with a count derived from offset + limit, which
ignores the real page boundaries. Update dlq_messages to route the request
through dlq_peek using input.topic, input.offset, and input.limit so the
SwappableAdapter handles paging directly, and keep the existing validation and
result mapping in place.
In `@queue/src/store.rs`:
- Around line 63-67: The SharedStore design uses one global Mutex over all
StoreData, which serializes queue/DLQ/stats access for every topic and creates
unnecessary contention as traffic grows. Refactor the store around per-topic
sharding in SharedStore and StoreData, such as a map of topic-specific mutexes
or a DashMap-backed structure, so unrelated topics can mutate independently
while keeping the existing queue/DLQ/stats behavior intact.
In `@queue/src/trigger.rs`:
- Around line 149-178: `register_subscriber` in `trigger.rs` holds the
`registrations` mutex while awaiting `self.adapter.subscribe(...)`, which blocks
other registration and shutdown paths. Refactor `register_subscriber` to acquire
the lock only for the duplicate check, release it before calling
`adapter.subscribe().await`, then re-acquire it afterward to insert the
`RegisteredSubscriber`. Be sure to re-check
`registrations.contains_key(®istration.trigger_id)` after the await to avoid
the race introduced by splitting the lock scope, matching the approach used by
`unregister` and `shutdown`.
In `@queue/tests/e2e_durability.rs`:
- Around line 85-97: The polling helper logic is duplicated across the e2e test
suites, so extract it into a shared helper in queue/tests/common/mod.rs and
reuse it from e2e_durability::wait_for_fires, e2e_redis, and e2e_rabbitmq. Keep
the shared helper generic enough to cover both the exact fire-count wait and the
more general wait_until pattern, then replace the per-file copies with calls to
that shared function.
In `@queue/tests/e2e_redis.rs`:
- Line 126: The DLQ-not-supported message in the Redis E2E test is duplicated as
a local string literal and can drift from the source of truth in RedisAdapter.
Update the test to reuse the existing constant from RedisAdapter by making
DLQ_NOT_SUPPORTED accessible within the crate (for example, pub(crate)) and
importing it in queue/tests/e2e_redis.rs, so the test references the same symbol
instead of re-typing the message.
🪄 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
Run ID: ab94526a-adc2-4fc2-a516-692c31bc8e88
⛔ Files ignored due to path filters (1)
queue/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (39)
.github/workflows/create-tag.yml.github/workflows/release.ymlqueue/Cargo.tomlqueue/README.mdqueue/benches/enqueue_latency.mdqueue/docs/adr-001-store-location.mdqueue/iii.worker.yamlqueue/skills/SKILL.mdqueue/src/adapter.rsqueue/src/adapters/builtin.rsqueue/src/adapters/memory.rsqueue/src/adapters/mod.rsqueue/src/adapters/rabbitmq/adapter.rsqueue/src/adapters/rabbitmq/consumer.rsqueue/src/adapters/rabbitmq/mod.rsqueue/src/adapters/rabbitmq/naming.rsqueue/src/adapters/rabbitmq/publisher.rsqueue/src/adapters/rabbitmq/retry.rsqueue/src/adapters/rabbitmq/topology.rsqueue/src/adapters/rabbitmq/types.rsqueue/src/adapters/rabbitmq/worker.rsqueue/src/adapters/redis.rsqueue/src/boot.rsqueue/src/config.rsqueue/src/configuration.rsqueue/src/functions.rsqueue/src/lib.rsqueue/src/main.rsqueue/src/manifest.rsqueue/src/store.rsqueue/src/subscriber_config.rsqueue/src/trigger.rsqueue/tests/.gitkeepqueue/tests/common/docker.rsqueue/tests/common/engine.rsqueue/tests/common/mod.rsqueue/tests/e2e_durability.rsqueue/tests/e2e_rabbitmq.rsqueue/tests/e2e_redis.rs
Comment on lines
+229
to
+259
| pub struct SwappableAdapter { | ||
| inner: Arc<RwLock<Arc<dyn QueueAdapter>>>, | ||
| /// The active adapter's identity (e.g. `"builtin"`, `"redis"`, | ||
| /// `"rabbitmq"`), set by the factory (`crate::boot::build_adapter`'s | ||
| /// caller) at construction/replace time. Exposed via [`Self::current_name`] | ||
| /// for service functions (`list_topics`) that report which transport | ||
| /// backs a topic. | ||
| name: Arc<RwLock<String>>, | ||
| } | ||
|
|
||
| impl SwappableAdapter { | ||
| pub fn new(adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) -> Self { | ||
| Self { | ||
| inner: Arc::new(RwLock::new(adapter)), | ||
| name: Arc::new(RwLock::new(name.into())), | ||
| } | ||
| } | ||
|
|
||
| pub async fn current(&self) -> Arc<dyn QueueAdapter> { | ||
| self.inner.read().await.clone() | ||
| } | ||
|
|
||
| pub async fn current_name(&self) -> String { | ||
| self.name.read().await.clone() | ||
| } | ||
|
|
||
| pub async fn replace(&self, adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) { | ||
| *self.inner.write().await = adapter; | ||
| *self.name.write().await = name.into(); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
inner and name can be observed out of sync during hot-swap.
replace() updates the adapter and its name via two separate RwLock writes (Line 256, Line 257). A concurrent reader calling current() and current_name() can observe the new adapter paired with the stale name (or vice versa) in that window — relevant since callers use current_name() specifically to report which transport backs a topic.
🔒 Proposed fix: single lock for atomic adapter+name pairing
pub struct SwappableAdapter {
- inner: Arc<RwLock<Arc<dyn QueueAdapter>>>,
- name: Arc<RwLock<String>>,
+ state: Arc<RwLock<(Arc<dyn QueueAdapter>, String)>>,
}
impl SwappableAdapter {
pub fn new(adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) -> Self {
Self {
- inner: Arc::new(RwLock::new(adapter)),
- name: Arc::new(RwLock::new(name.into())),
+ state: Arc::new(RwLock::new((adapter, name.into()))),
}
}
pub async fn current(&self) -> Arc<dyn QueueAdapter> {
- self.inner.read().await.clone()
+ self.state.read().await.0.clone()
}
pub async fn current_name(&self) -> String {
- self.name.read().await.clone()
+ self.state.read().await.1.clone()
}
pub async fn replace(&self, adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) {
- *self.inner.write().await = adapter;
- *self.name.write().await = name.into();
+ *self.state.write().await = (adapter, name.into());
}
}📝 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.
Suggested change
| pub struct SwappableAdapter { | |
| inner: Arc<RwLock<Arc<dyn QueueAdapter>>>, | |
| /// The active adapter's identity (e.g. `"builtin"`, `"redis"`, | |
| /// `"rabbitmq"`), set by the factory (`crate::boot::build_adapter`'s | |
| /// caller) at construction/replace time. Exposed via [`Self::current_name`] | |
| /// for service functions (`list_topics`) that report which transport | |
| /// backs a topic. | |
| name: Arc<RwLock<String>>, | |
| } | |
| impl SwappableAdapter { | |
| pub fn new(adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) -> Self { | |
| Self { | |
| inner: Arc::new(RwLock::new(adapter)), | |
| name: Arc::new(RwLock::new(name.into())), | |
| } | |
| } | |
| pub async fn current(&self) -> Arc<dyn QueueAdapter> { | |
| self.inner.read().await.clone() | |
| } | |
| pub async fn current_name(&self) -> String { | |
| self.name.read().await.clone() | |
| } | |
| pub async fn replace(&self, adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) { | |
| *self.inner.write().await = adapter; | |
| *self.name.write().await = name.into(); | |
| } | |
| } | |
| pub struct SwappableAdapter { | |
| state: Arc<RwLock<(Arc<dyn QueueAdapter>, String)>>, | |
| } | |
| impl SwappableAdapter { | |
| pub fn new(adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) -> Self { | |
| Self { | |
| state: Arc::new(RwLock::new((adapter, name.into()))), | |
| } | |
| } | |
| pub async fn current(&self) -> Arc<dyn QueueAdapter> { | |
| self.state.read().await.0.clone() | |
| } | |
| pub async fn current_name(&self) -> String { | |
| self.state.read().await.1.clone() | |
| } | |
| pub async fn replace(&self, adapter: Arc<dyn QueueAdapter>, name: impl Into<String>) { | |
| *self.state.write().await = (adapter, name.into()); | |
| } | |
| } |
🤖 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 229 - 259, The SwappableAdapter state can
be observed out of sync because replace() updates inner and name with separate
RwLock writes. Refactor SwappableAdapter so the adapter and its identity are
stored and swapped together atomically, and make current() and current_name()
read from the same paired state. Use the existing SwappableAdapter, current(),
current_name(), and replace() methods as the touchpoints, and ensure callers
never see a new QueueAdapter with a stale transport name or the reverse.
Comment on lines
+463
to
+494
| async fn run_concurrent( | ||
| store: Arc<dyn QueueStore>, | ||
| invoker: Arc<dyn Invoker>, | ||
| cfg: PollerConfig, | ||
| concurrency: u32, | ||
| ) { | ||
| let semaphore = Arc::new(Semaphore::new(concurrency as usize)); | ||
| let cfg = Arc::new(cfg); | ||
|
|
||
| loop { | ||
| let permit = match Arc::clone(&semaphore).acquire_owned().await { | ||
| Ok(permit) => permit, | ||
| Err(_) => return, // semaphore closed; subscription is being torn down | ||
| }; | ||
|
|
||
| match store.dequeue(&cfg.queue_name).await { | ||
| Some(job) => { | ||
| let store = Arc::clone(&store); | ||
| let invoker = Arc::clone(&invoker); | ||
| let cfg = Arc::clone(&cfg); | ||
| tokio::spawn(async move { | ||
| process_job(&store, &invoker, &cfg, job).await; | ||
| drop(permit); | ||
| }); | ||
| } | ||
| None => { | ||
| drop(permit); | ||
| tokio::time::sleep(Duration::from_millis(cfg.poll_interval_ms)).await; | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
In-flight concurrent job handlers survive unsubscribe/shutdown.
run_concurrent spawns a detached tokio::spawn per dequeued job (lines 483-486) and discards its JoinHandle. unsubscribe/shutdown only call .abort() on the outer polling task (lines 264, 428), which stops new dequeues but does not cancel already-spawned job handlers — they keep running to completion, still calling invoker.call and store.ack/nack, even after the subscription is supposedly torn down. This matters for the adapter hot-swap flow described in the PR objectives, where old consumers are expected to fully stop before/while a new store/adapter takes over.
Since a tokio::task::JoinSet aborts all of its tasks when dropped, scoping the per-job spawns to a JoinSet local to run_concurrent would let the existing .abort() on the outer task cascade correctly.
🔧 Proposed fix using a scoped JoinSet
async fn run_concurrent(
store: Arc<dyn QueueStore>,
invoker: Arc<dyn Invoker>,
cfg: PollerConfig,
concurrency: u32,
) {
let semaphore = Arc::new(Semaphore::new(concurrency as usize));
let cfg = Arc::new(cfg);
+ let mut handlers = tokio::task::JoinSet::new();
loop {
let permit = match Arc::clone(&semaphore).acquire_owned().await {
Ok(permit) => permit,
Err(_) => return, // semaphore closed; subscription is being torn down
};
match store.dequeue(&cfg.queue_name).await {
Some(job) => {
let store = Arc::clone(&store);
let invoker = Arc::clone(&invoker);
let cfg = Arc::clone(&cfg);
- tokio::spawn(async move {
+ handlers.spawn(async move {
process_job(&store, &invoker, &cfg, job).await;
drop(permit);
});
}
None => {
drop(permit);
tokio::time::sleep(Duration::from_millis(cfg.poll_interval_ms)).await;
}
}
}
}Also applies to: 425-431
🤖 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 463 - 494, Detached per-job tasks
in run_concurrent keep running after unsubscribe/shutdown because only the outer
polling task is aborted. Update run_concurrent to track spawned job handlers
with a local JoinSet (or equivalent scoped task manager) instead of discarding
the tokio::spawn JoinHandle, so dropping/aborting the poller also cancels
in-flight jobs. Keep the change localized around run_concurrent, process_job,
and the unsubscribe/shutdown paths that currently abort the parent task.
Comment on lines
+54
to
+61
| let _ = iii.register_trigger_type( | ||
| RegisterTriggerType::new( | ||
| TRIGGER_TYPE, | ||
| "Durable queue subscriber", | ||
| trigger_handler.clone(), | ||
| ) | ||
| .trigger_request_format::<SubscriberSpec>(), | ||
| ); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Registration failure is silently swallowed.
iii.register_trigger_type(...)'s Result is discarded via let _ =. If this fails, start() still returns Ok(BootHandle { .. }) and the worker reports a healthy boot, but no durable:subscriber registration can ever succeed — silently defeating this worker's core purpose with no log line to diagnose it.
🛡️ Suggested fix
- let _ = iii.register_trigger_type(
+ iii.register_trigger_type(
RegisterTriggerType::new(
TRIGGER_TYPE,
"Durable queue subscriber",
trigger_handler.clone(),
)
.trigger_request_format::<SubscriberSpec>(),
- );
+ )
+ .map_err(|e| anyhow::anyhow!("failed to register durable:subscriber trigger type: {e}"))?;📝 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.
Suggested change
| let _ = iii.register_trigger_type( | |
| RegisterTriggerType::new( | |
| TRIGGER_TYPE, | |
| "Durable queue subscriber", | |
| trigger_handler.clone(), | |
| ) | |
| .trigger_request_format::<SubscriberSpec>(), | |
| ); | |
| iii.register_trigger_type( | |
| RegisterTriggerType::new( | |
| TRIGGER_TYPE, | |
| "Durable queue subscriber", | |
| trigger_handler.clone(), | |
| ) | |
| .trigger_request_format::<SubscriberSpec>(), | |
| ) | |
| .map_err(|e| anyhow::anyhow!("failed to register durable:subscriber trigger type: {e}"))?; |
🤖 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/boot.rs` around lines 54 - 61, The registration result from
`iii.register_trigger_type(...)` is being ignored, so `start()` can succeed even
when `TRIGGER_TYPE` never registers. Update `start()` in `boot` to handle the
`Result` from `register_trigger_type` explicitly: log the failure with context
and return an error instead of continuing to build `BootHandle`. Use the
existing `trigger_handler`/`RegisterTriggerType::new(...)` call site to locate
the fix and make sure a failed `durable:subscriber` registration cannot be
reported as a healthy boot.
Comment on lines
+96
to
+108
| impl FileStore { | ||
| pub async fn open(path: impl AsRef<Path>, _save_interval_ms: u64) -> anyhow::Result<Self> { | ||
| let dir = path.as_ref().to_path_buf(); | ||
| std::fs::create_dir_all(&dir)?; | ||
| let data = load_snapshot(&dir).await?; | ||
| Ok(Self { | ||
| shared: Arc::new(SharedStore { | ||
| inner: Mutex::new(data), | ||
| file_dir: Some(dir), | ||
| }), | ||
| }) | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Blocking filesystem I/O runs directly on the async runtime thread.
persist_if_needed, load_snapshot, and FileStore::open call std::fs::write/rename/read/create_dir_all synchronously inside async fns. Since persist_if_needed runs on every mutation in file-backed mode (by design, per this file's module doc), this blocks a Tokio worker thread on every enqueue/dequeue/ack/nack for the whole duration of the file write+rename, which can stall other concurrent async work on that thread under load.
🧵 Suggested fix: offload to a blocking-safe path
async fn persist_if_needed(shared: &SharedStore, snapshot: &StoreData) -> anyhow::Result<()> {
let Some(dir) = &shared.file_dir else {
return Ok(());
};
- std::fs::create_dir_all(dir)?;
- let path = dir.join(STORE_FILE_NAME);
- let tmp = dir.join(format!("{STORE_FILE_NAME}.tmp"));
- let bytes = serde_json::to_vec_pretty(snapshot)?;
- std::fs::write(&tmp, bytes)?;
- std::fs::rename(tmp, path)?;
- Ok(())
+ let dir = dir.clone();
+ let bytes = serde_json::to_vec_pretty(snapshot)?;
+ tokio::task::spawn_blocking(move || {
+ std::fs::create_dir_all(&dir)?;
+ let path = dir.join(STORE_FILE_NAME);
+ let tmp = dir.join(format!("{STORE_FILE_NAME}.tmp"));
+ std::fs::write(&tmp, bytes)?;
+ std::fs::rename(tmp, path)
+ })
+ .await??;
+ Ok(())
}Also applies to: 397-417
🤖 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/store.rs` around lines 96 - 108, FileStore::open, load_snapshot,
and persist_if_needed are performing synchronous filesystem work directly inside
async paths, which can block the Tokio runtime thread under load. Move the
create/read/write/rename directory and snapshot operations onto a blocking-safe
path such as a dedicated blocking task or async filesystem APIs, and keep the
async methods non-blocking. Update the file-backed persistence flow in
persist_if_needed and the startup load path in FileStore::open to use the same
offloaded approach so all disk I/O is handled consistently.
Comment on lines
+41
to
+55
| #[async_trait] | ||
| impl Invoker for IiiInvoker { | ||
| async fn call(&self, function_id: &str, payload: Value) -> Result<Option<Value>, String> { | ||
| self.iii | ||
| .trigger(TriggerRequest { | ||
| function_id: function_id.to_string(), | ||
| payload, | ||
| action: None, | ||
| timeout_ms: None, | ||
| }) | ||
| .await | ||
| .map(Some) | ||
| .map_err(|e| e.to_string()) | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
No timeout on the per-message invoker call.
IiiInvoker::call triggers the subscriber's function with timeout_ms: None. Since ack/nack decisions in the adapter presumably hinge on this call returning, a hung or slow downstream function blocks that consumer indefinitely with no bound. configuration.rs's trigger_with_retry uses an explicit CONFIG_BUS_TIMEOUT_MS for its own outbound calls — worth applying similar discipline here.
🛡️ Suggested fix
async fn call(&self, function_id: &str, payload: Value) -> Result<Option<Value>, String> {
self.iii
.trigger(TriggerRequest {
function_id: function_id.to_string(),
payload,
action: None,
- timeout_ms: None,
+ timeout_ms: Some(DEFAULT_INVOKE_TIMEOUT_MS),
})
.await
.map(Some)
.map_err(|e| e.to_string())
}📝 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.
Suggested change
| #[async_trait] | |
| impl Invoker for IiiInvoker { | |
| async fn call(&self, function_id: &str, payload: Value) -> Result<Option<Value>, String> { | |
| self.iii | |
| .trigger(TriggerRequest { | |
| function_id: function_id.to_string(), | |
| payload, | |
| action: None, | |
| timeout_ms: None, | |
| }) | |
| .await | |
| .map(Some) | |
| .map_err(|e| e.to_string()) | |
| } | |
| } | |
| #[async_trait] | |
| impl Invoker for IiiInvoker { | |
| async fn call(&self, function_id: &str, payload: Value) -> Result<Option<Value>, String> { | |
| self.iii | |
| .trigger(TriggerRequest { | |
| function_id: function_id.to_string(), | |
| payload, | |
| action: None, | |
| timeout_ms: Some(DEFAULT_INVOKE_TIMEOUT_MS), | |
| }) | |
| .await | |
| .map(Some) | |
| .map_err(|e| e.to_string()) | |
| } | |
| } |
🤖 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 41 - 55, IiiInvoker::call is triggering
downstream work with timeout_ms set to None, leaving per-message processing
unbounded and able to hang the consumer indefinitely. Update the TriggerRequest
construction in IiiInvoker::call to pass a concrete timeout value instead of
None, using the same timeout discipline as trigger_with_retry in
configuration.rs (prefer the existing CONFIG_BUS_TIMEOUT_MS pattern or an
equivalent per-message timeout source). Keep the change localized to the
IiiInvoker implementation and ensure the returned Result still maps the trigger
response/errors as before.
Comment on lines
+63
to
+83
| let container_id = String::from_utf8_lossy(&run_output.stdout) | ||
| .trim() | ||
| .to_string(); | ||
| if container_id.is_empty() { | ||
| return None; | ||
| } | ||
|
|
||
| let host_port = resolve_mapped_port(&container_id, "6379/tcp")?; | ||
|
|
||
| let container = RedisContainer { | ||
| container_id, | ||
| host_port, | ||
| }; | ||
|
|
||
| if !wait_until_tcp_ready(container.host_port) { | ||
| eprintln!("[skip] redis container did not become ready in time"); | ||
| return None; | ||
| } | ||
|
|
||
| Some(container) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Container leaked if resolve_mapped_port fails.
In both start_redis and start_rabbitmq, resolve_mapped_port(&container_id, ...)? runs before container_id is wrapped in the Drop-guarded RedisContainer/RabbitMqContainer. If it returns None, the function bails out via ? and the already-started container is never stopped, leaking a running container.
🐳 Proposed fix (redis path shown; mirror for rabbitmq)
let host_port = resolve_mapped_port(&container_id, "6379/tcp")?;
+ let host_port = match host_port {
+ Some(p) => p,
+ None => {
+ let _ = Command::new("docker")
+ .args(["stop", "-t", "1", &container_id])
+ .output();
+ return None;
+ }
+ };Also applies to: 168-187
🤖 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/common/docker.rs` around lines 63 - 83, The container startup
flow in start_redis and start_rabbitmq can leak a running container if
resolve_mapped_port returns None before the Drop-guarded
RedisContainer/RabbitMqContainer is created. Restructure the logic so the
container is wrapped in its guard immediately after obtaining container_id, then
call resolve_mapped_port through that guarded value (or otherwise ensure cleanup
still happens on early return). Mirror the same fix in both helper functions
that start containers, using the RedisContainer and RabbitMqContainer types as
the cleanup owners.
guibeira
force-pushed
the
feat/queue-worker
branch
from
July 7, 2026 18:44
ee41833 to
c43e958
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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/README.md`:
- Around line 104-110: The RabbitMQ transport description is overstating
capabilities because the shared AMQP channel can be closed by DLQ browse/redrive
and stop all consumers. Update the `rabbitmq` section in the README to either
remove the “full-featured” claim or narrow it until the channel isolation issue
is fixed, and make sure the wording matches the actual behavior of the RabbitMQ
adapter.
- Around line 99-102: The README’s Redis DLQ section refers to outdated API
names, so update the documentation to match the actual exported symbols used in
the function table. Replace the `dlq_count` and `dlq_peek` references with the
correct `engine::queue::dlq_topics` and `engine::queue::dlq_messages` names,
while keeping the RedisAdapter DLQ error text aligned with those documented
APIs.
In `@queue/skills/SKILL.md`:
- Around line 12-15: Clarify the queue worker documentation so it does not imply
unconditional persistence: update the description in the SKILL.md section for
the queue worker to state that durable message storage depends on the selected
storage mode. Mention that `durable:subscriber`, `iii::durable::publish`,
retries, DLQ, and redrive/discard behavior apply, but in-memory mode does not
retain pending jobs across restart or hot-swap.
🪄 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
Run ID: a5e0d4b3-d31c-4f37-aa82-f2c8091ee416
⛔ Files ignored due to path filters (1)
queue/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (37)
.github/workflows/create-tag.yml.github/workflows/release.ymlqueue/Cargo.tomlqueue/README.mdqueue/iii.worker.yamlqueue/skills/SKILL.mdqueue/src/adapter.rsqueue/src/adapters/builtin.rsqueue/src/adapters/memory.rsqueue/src/adapters/mod.rsqueue/src/adapters/rabbitmq/adapter.rsqueue/src/adapters/rabbitmq/consumer.rsqueue/src/adapters/rabbitmq/mod.rsqueue/src/adapters/rabbitmq/naming.rsqueue/src/adapters/rabbitmq/publisher.rsqueue/src/adapters/rabbitmq/retry.rsqueue/src/adapters/rabbitmq/topology.rsqueue/src/adapters/rabbitmq/types.rsqueue/src/adapters/rabbitmq/worker.rsqueue/src/adapters/redis.rsqueue/src/boot.rsqueue/src/config.rsqueue/src/configuration.rsqueue/src/functions.rsqueue/src/lib.rsqueue/src/main.rsqueue/src/manifest.rsqueue/src/store.rsqueue/src/subscriber_config.rsqueue/src/trigger.rsqueue/tests/.gitkeepqueue/tests/common/docker.rsqueue/tests/common/engine.rsqueue/tests/common/mod.rsqueue/tests/e2e_durability.rsqueue/tests/e2e_rabbitmq.rsqueue/tests/e2e_redis.rs
✅ Files skipped from review due to trivial changes (4)
- queue/tests/common/mod.rs
- queue/tests/.gitkeep
- .github/workflows/create-tag.yml
- .github/workflows/release.yml
🚧 Files skipped from review as they are similar to previous changes (31)
- queue/src/lib.rs
- queue/src/adapters/mod.rs
- queue/iii.worker.yaml
- queue/src/manifest.rs
- queue/Cargo.toml
- queue/tests/e2e_durability.rs
- queue/src/adapters/rabbitmq/retry.rs
- queue/src/adapters/rabbitmq/mod.rs
- queue/src/main.rs
- queue/src/config.rs
- queue/src/adapter.rs
- queue/src/adapters/rabbitmq/consumer.rs
- queue/src/adapters/rabbitmq/publisher.rs
- queue/src/adapters/rabbitmq/naming.rs
- queue/src/subscriber_config.rs
- queue/tests/e2e_redis.rs
- queue/src/adapters/memory.rs
- queue/src/adapters/rabbitmq/types.rs
- queue/src/adapters/rabbitmq/topology.rs
- queue/src/configuration.rs
- queue/src/trigger.rs
- queue/src/adapters/builtin.rs
- queue/tests/common/engine.rs
- queue/tests/e2e_rabbitmq.rs
- queue/src/functions.rs
- queue/src/boot.rs
- queue/tests/common/docker.rs
- queue/src/store.rs
- queue/src/adapters/rabbitmq/worker.rs
- queue/src/adapters/redis.rs
- queue/src/adapters/rabbitmq/adapter.rs
Comment on lines
+99
to
+102
| Every DLQ operation (`redrive`, `redrive_message`, `discard_message`, | ||
| `dlq_count`, `dlq_peek`) returns the error | ||
| `RedisAdapter does not support DLQ operations (pub/sub only)`, verbatim | ||
| from the engine. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Align the Redis DLQ error text with the documented API names.
This paragraph refers to dlq_count and dlq_peek, but the function table exposes engine::queue::dlq_topics and engine::queue::dlq_messages. As written, the docs point readers at APIs that do not exist here.
🤖 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/README.md` around lines 99 - 102, The README’s Redis DLQ section refers
to outdated API names, so update the documentation to match the actual exported
symbols used in the function table. Replace the `dlq_count` and `dlq_peek`
references with the correct `engine::queue::dlq_topics` and
`engine::queue::dlq_messages` names, while keeping the RedisAdapter DLQ error
text aligned with those documented APIs.
Comment on lines
+104
to
+110
| #### `rabbitmq` | ||
|
|
||
| Full-featured transport: retries, DLQ, redrive, message priority, and both | ||
| standard and FIFO queue modes. On by default (the `rabbitmq` cargo feature | ||
| is feature-default-on); building with `--no-default-features` drops it, | ||
| and selecting `adapter.name: rabbitmq` in that build fails boot with an | ||
| error naming the missing feature. |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Dial back the RabbitMQ claim or isolate the DLQ channel first.
The later gap note says a DLQ browse/redrive can close the shared AMQP channel and stop every consumer on the worker. Until that is fixed, this transport is not operationally "full-featured."
🤖 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/README.md` around lines 104 - 110, The RabbitMQ transport description
is overstating capabilities because the shared AMQP channel can be closed by DLQ
browse/redrive and stop all consumers. Update the `rabbitmq` section in the
README to either remove the “full-featured” claim or narrow it until the channel
isolation issue is fixed, and make sure the wording matches the actual behavior
of the RabbitMQ adapter.
Comment on lines
+12
to
+15
| The `queue` worker provides durable topic delivery for iii. Register a consumer | ||
| with a `durable:subscriber` trigger, publish with `iii::durable::publish`, and | ||
| the worker persists messages, retries failed deliveries, moves exhausted jobs to | ||
| DLQ, and exposes redrive/discard inspection functions. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Clarify that persistence depends on the selected storage mode.
The description says the worker "persists messages" unconditionally, but the documented in-memory mode loses pending jobs on restart/hot-swap. That overstates durability for the default/dev configuration.
🤖 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/skills/SKILL.md` around lines 12 - 15, Clarify the queue worker
documentation so it does not imply unconditional persistence: update the
description in the SKILL.md section for the queue worker to state that durable
message storage depends on the selected storage mode. Mention that
`durable:subscriber`, `iii::durable::publish`, retries, DLQ, and redrive/discard
behavior apply, but in-memory mode does not retain pending jobs across restart
or hot-swap.
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
1 participant
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
queueregistry worker, replacing the built-iniii-queue.durable:subscribertrigger type and all 8 queue/DLQ service functions with verbatim builtin ids.builtin(per-subscriber fan-out, concurrency, fifo),redis(pub/sub, DLQ-less by design),rabbitmq(full retry/DLX topology, priority queues, fifo; cargo feature, default on),memory(test-only). Adapter factory + hot-swap: config change builds the new adapter first (failure keeps the old one serving), swaps, and re-subscribes every trigger.SubscriberQueueConfigparity:type(fifo/concurrent),maxRetries,concurrency,visibilityTimeout,delaySeconds,backoffType,backoffDelayMs,maxPriority.iii-queueworker is connected.Behavior
durable:subscribertrigger type and all 8 queue/DLQ function ids verbatim.x-max-prioritypriority queues driven bypriority_field/maxPriority, standard and fifo modes.TriggerAction::Enqueueparity depends on the engine QueueEnqueuer cut (MOT-3829).Validation
cd queue && cargo test/--no-default-features/--features test-adapters— 99 unit tests + e2e.e2e_durability(delivery, DLQ+redrive, restart survival),e2e_redis(pub/sub delivery, DLQ error parity),e2e_rabbitmq(delivery, retry→DLQ→redrive, priority ordering 9/5/1, fifo ordering).cargo fmt --check,cargo clippy --all-targets -- -D warnings(both feature configs).validate_worker.py,build_skills_payload.py,--manifestsmoke.Notes
queue/v0.1.0is post-merge only, via the Create Tag workflow.test_python_queue/and is intentionally untracked.Summary by CodeRabbit
New Features
Documentation