feat(iii-queue): consume the configuration worker for runtime config#1873
Merged
Conversation
Contributor
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthrough
ChangesQueue Worker Hot-Swap Configuration Integration
Sequence Diagram(s)sequenceDiagram
participant Engine
participant QueueWorker
participant configuration_rs as configuration.rs
participant ConfigWorker as Configuration Worker Bus
participant QueueAdapter
rect rgba(70, 130, 180, 0.5)
Note over Engine,QueueAdapter: Boot / start_background_tasks
QueueWorker->>configuration_rs: register_config(engine, seed)
configuration_rs->>ConfigWorker: configuration::get
ConfigWorker-->>configuration_rs: NOT_FOUND or existing value
configuration_rs->>ConfigWorker: register (schema + conditional initial_value)
QueueWorker->>configuration_rs: fetch_config(engine, fallback)
ConfigWorker-->>configuration_rs: stored value or null
configuration_rs-->>QueueWorker: validated QueueModuleConfig
QueueWorker->>QueueAdapter: setup_function_queue per queue config
QueueWorker->>configuration_rs: register_config_trigger(engine)
configuration_rs->>QueueWorker: on_config_change (catch-up apply)
end
rect rgba(34, 139, 34, 0.5)
Note over Engine,QueueAdapter: Runtime config change
ConfigWorker->>configuration_rs: configuration:updated event
configuration_rs->>QueueWorker: on_config_change
QueueWorker->>QueueWorker: apply_config under apply_lock
QueueWorker->>configuration_rs: fetch_config (live value)
alt adapter identity changed
QueueWorker->>QueueAdapter: rebuild adapter (hot-swap)
QueueWorker->>QueueAdapter: restart all consumers
else same adapter, queue diff
QueueWorker->>QueueAdapter: per-queue add/remove consumers
end
alt apply timeout
configuration_rs->>configuration_rs: spawn one-shot retry after delay
configuration_rs->>QueueWorker: on_config_change (retry)
end
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
engine/src/workers/queue/config.rs (1)
129-143:⚠️ Potential issue | 🟠 Major | ⚡ Quick winReject non-runnable queue configs at validation time.
validate()currently acceptsconcurrency: 0andtype: "fifo"with an emptymessage_group_field.
At Line 803 inengine/src/workers/queue/queue.rs,concurrency: 0createsSemaphore::new(0), which can stall processing indefinitely once messages arrive. Empty FIFO group-field values also pass validation but fail every enqueue path at runtime.Suggested patch
pub fn validate(&self) -> anyhow::Result<()> { for (name, queue_config) in &self.queue_configs { + if queue_config.concurrency == 0 { + anyhow::bail!( + "Queue '{}' has invalid concurrency '{}'. Must be >= 1", + name, + queue_config.concurrency + ); + } + if queue_config.r#type != "standard" && queue_config.r#type != "fifo" { anyhow::bail!( "Queue '{}' has invalid type '{}'. Must be 'standard' or 'fifo'", name, queue_config.r#type ); } - if queue_config.r#type == "fifo" && queue_config.message_group_field.is_none() { + if queue_config.r#type == "fifo" + && queue_config + .message_group_field + .as_deref() + .map(str::trim) + .filter(|v| !v.is_empty()) + .is_none() + { anyhow::bail!( "Queue '{}' is of type 'fifo' but 'message_group_field' is not set", name ); } } Ok(()) }🤖 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 `@engine/src/workers/queue/config.rs` around lines 129 - 143, The validate() method in the queue configuration is missing validation checks that would prevent invalid configurations from causing runtime failures. Add a check to reject queue configs where concurrency is set to 0, as this creates an unresponsive Semaphore at runtime. Additionally, strengthen the existing FIFO validation to also reject cases where message_group_field is an empty string (not just when it is None), since empty group field values pass the current validation but fail during enqueue operations. These checks should be added within the loop that iterates through self.queue_configs, similar to the existing type validation checks.
🤖 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 `@engine/src/workers/queue/adapters/builtin/adapter.rs`:
- Around line 478-481: The eviction of messages from the outstanding queue
occurs before the send operation completes, which can strand messages if the
send blocks and the receiver is dropped. Restructure the code block by moving
the eviction logic (the check on outstanding.len() and the pop_front() call) to
execute after the push_back and send operations succeed, rather than before.
This ensures that messages are only removed from outstanding after they have
been successfully sent and will not be left stranded in the delivery_map if the
channel blocks.
In `@engine/src/workers/queue/queue.rs`:
- Around line 956-973: The hot-swap logic has a race condition where set_config
is called before set_adapter, creating a window where concurrent enqueue
operations can see the new config with the old adapter, causing message
misrouting. Fix this by making adapter and config updates atomic together rather
than in separate steps. Refactor the code so that both set_config and
set_adapter operations (or their underlying state) are protected by a single
synchronization primitive or are updated as a single atomic unit, ensuring
concurrent calls always see either the old config+adapter pair or the new pair,
never a mixed state during the swap.
---
Outside diff comments:
In `@engine/src/workers/queue/config.rs`:
- Around line 129-143: The validate() method in the queue configuration is
missing validation checks that would prevent invalid configurations from causing
runtime failures. Add a check to reject queue configs where concurrency is set
to 0, as this creates an unresponsive Semaphore at runtime. Additionally,
strengthen the existing FIFO validation to also reject cases where
message_group_field is an empty string (not just when it is None), since empty
group field values pass the current validation but fail during enqueue
operations. These checks should be added within the loop that iterates through
self.queue_configs, similar to the existing type validation checks.
🪄 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: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 277cbb29-07a0-4e08-9dd2-64a980735d26
📒 Files selected for processing (6)
engine/src/workers/queue/adapters/builtin/adapter.rsengine/src/workers/queue/config.rsengine/src/workers/queue/configuration.rsengine/src/workers/queue/mod.rsengine/src/workers/queue/queue.rsengine/tests/queue_configuration_e2e.rs
Contributor
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 (2)
engine/src/workers/queue/queue.rs (2)
1119-1120:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRebuild the adapter before adopting fetched boot config.
Line 1120 only swaps
config. The adapter was already constructed from the seedconfig.yamlvalue, so if the stored configuration selects a different adapter/config, Line 1138 starts consumers on the stale adapter. The catch-up at Line 1166 will not repair this becauseapply_confignow sees the fetched config as both old and new.Suggested boot-path fix
Ok(config) => match config.validate() { - Ok(()) => self.set_config(config), + Ok(()) => { + let adapter_changed = Self::effective_adapter(&self.config_snapshot()) + != Self::effective_adapter(&config); + let adapter_ready = if adapter_changed { + match Self::resolve_adapter(&self.engine, &config).await { + Ok(adapter) => { + self.set_adapter(adapter); + true + } + Err(err) => { + tracing::warn!( + error = %err, + "iii-queue: stored configuration selects an unavailable adapter; continuing with static config" + ); + false + } + } + } else { + true + }; + if adapter_ready { + self.set_config(config); + } + } Err(err) => tracing::warn!(Also applies to: 1136-1138
🤖 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 `@engine/src/workers/queue/queue.rs` around lines 1119 - 1120, The adapter is initialized from the seed config.yaml before the fetched boot configuration is adopted at line 1120. When config.validate() succeeds and self.set_config(config) is called, the adapter must also be rebuilt with the new configuration before consumers are started around line 1138. Currently, consumers start on the stale adapter instance that was constructed from the original seed configuration. Rebuild the adapter after setting the new config by invoking the appropriate adapter reconstruction logic that uses the newly adopted configuration, ensuring the adapter reflects the fetched boot config before any consumers begin operating on it.
633-657:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftReplay durable topic subscriptions when replacing the adapter.
Line 649 stores
durable:subscriberstate in whichever adapter was current at registration. The full swap branch installs a fresh adapter and only restarts function-queue consumers fromqueue_configs, so existing topic subscriptions remain on the old adapter whileenqueuepublishes to the new one. Re-subscribe registered durable triggers onnew_adapter, or keep subscription state outside the adapter.Also applies to: 958-984
🤖 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 `@engine/src/workers/queue/queue.rs` around lines 633 - 657, When the adapter is replaced in the full swap operation, existing topic subscriptions registered via adapter.subscribe() remain on the old adapter while new enqueue operations publish to the new adapter, causing a mismatch. After installing the fresh adapter during the swap, re-subscribe all registered durable triggers on the new adapter to ensure subscriptions follow the adapter replacement. This may require iterating through existing trigger configurations and re-calling the subscription logic with the updated adapter reference.
🤖 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 `@engine/src/workers/queue/queue.rs`:
- Around line 1119-1120: The adapter is initialized from the seed config.yaml
before the fetched boot configuration is adopted at line 1120. When
config.validate() succeeds and self.set_config(config) is called, the adapter
must also be rebuilt with the new configuration before consumers are started
around line 1138. Currently, consumers start on the stale adapter instance that
was constructed from the original seed configuration. Rebuild the adapter after
setting the new config by invoking the appropriate adapter reconstruction logic
that uses the newly adopted configuration, ensuring the adapter reflects the
fetched boot config before any consumers begin operating on it.
- Around line 633-657: When the adapter is replaced in the full swap operation,
existing topic subscriptions registered via adapter.subscribe() remain on the
old adapter while new enqueue operations publish to the new adapter, causing a
mismatch. After installing the fresh adapter during the swap, re-subscribe all
registered durable triggers on the new adapter to ensure subscriptions follow
the adapter replacement. This may require iterating through existing trigger
configurations and re-calling the subscription logic with the updated adapter
reference.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: afc49591-5ca0-4489-a175-d45621c5dae3
📒 Files selected for processing (4)
engine/src/workers/queue/configuration.rsengine/src/workers/queue/queue.rsengine/src/workers/telemetry/mod.rsengine/tests/queue_configuration_e2e.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- engine/tests/queue_configuration_e2e.rs
- engine/src/workers/queue/configuration.rs
andersonleal
force-pushed
the
feat/iii-queue-configuration-worker
branch
from
d1dc7df to
d82fbad
Compare
andersonleal
force-pushed
the
feat/iii-queue-configuration-worker
branch
from
aa65873 to
d9b3737
Compare
Contributor
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 (3)
engine/src/workers/queue/queue.rs (3)
151-156:⚠️ Potential issue | 🟠 Major | ⚡ Quick winUse one
LiveStatesnapshot for config-derived keys and adapter calls.
resolve_queue_key()snapshots config internally, then these callers snapshot the adapter separately. A hot-swap between those reads can redrive/peek/discard against an adapter generation that does not match the config used to add__fn_queue::.Suggested direction
- fn resolve_queue_key(&self, name: &str) -> String { - if self.config_snapshot().queue_configs.contains_key(name) { + fn resolve_queue_key_for(config: &QueueModuleConfig, name: &str) -> String { + if config.queue_configs.contains_key(name) { format!("__fn_queue::{}", name) } else { name.to_string() } } + + fn resolve_queue_key(&self, name: &str) -> String { + let config = self.config_snapshot(); + Self::resolve_queue_key_for(config.as_ref(), name) + }Then use the same snapshot in each operation that needs both:
- let resolved = self.resolve_queue_key(&input.queue); - match self.adapter_snapshot().redrive_dlq(&resolved).await { + let live = self.live_snapshot(); + let resolved = Self::resolve_queue_key_for(live.config.as_ref(), &input.queue); + match live.adapter.redrive_dlq(&resolved).await {Apply the same pattern to
redrive_message,discard_message,console_topic_stats, andconsole_dlq_messages.Also applies to: 292-293, 344-347, 406-409, 493-500, 571-574
🤖 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 `@engine/src/workers/queue/queue.rs` around lines 151 - 156, The resolve_queue_key() method creates its own config snapshot internally, but callers then snapshot the adapter separately. This creates a race condition where the adapter generation may not match the config used to determine the queue key. Refactor resolve_queue_key() to accept a LiveState snapshot parameter instead of creating one internally, then update all callers (including redrive_message, discard_message, console_topic_stats, and console_dlq_messages) to obtain a single snapshot and pass it to resolve_queue_key() along with their adapter operations to ensure consistency between config and adapter generations.
645-669:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftReplay durable subscribers when the adapter is replaced.
register_trigger()materializes durable subscriptions inside the current adapter viasubscribe(), but the adapter hot-swap path only restarts function-queue consumers. Afterset_live(...), existing durable subscriber triggers are not subscribed on the new adapter, so subsequent durable publishes can stop faning out to registered subscribers.Replay registered
durable:subscribertriggers ontonew_adapterduring the hot-swap, or move durable subscription state outside the adapter so replacing the transport cannot drop it.Also applies to: 987-1021
🤖 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 `@engine/src/workers/queue/queue.rs` around lines 645 - 669, The `register_trigger()` method currently subscribes durable triggers to the current adapter via the `adapter.subscribe()` call, but when the adapter is hot-swapped during `set_live()`, these subscriptions are lost because they only exist in the old adapter instance. To fix this, you must either replay all existing durable subscriber triggers onto the new adapter immediately after the adapter replacement completes, or refactor to store durable subscription state outside the adapter so subscriptions persist across adapter replacements. If replaying, collect all registered durable subscriber triggers and call their subscribe logic on the new adapter instance after the swap occurs.
1149-1150:⚠️ Potential issue | 🟠 Major | ⚡ Quick winResolve the fetched startup adapter before adopting fetched config.
On boot/reload, a stored configuration can select a different adapter than
config.yaml, but this path only callsset_config(config). That leavesLiveStateasnew config + old adapter; the catch-upapply_config()will not fix it because it compares the already-adopted config to the fetched value and sees no adapter change.Suggested fix
- Ok(config) => match config.validate() { - Ok(()) => self.set_config(config), + Ok(config) => match config.validate() { + Ok(()) => { + if Self::effective_adapter(&self.config_snapshot()) + != Self::effective_adapter(&config) + { + match Self::resolve_adapter(&self.engine, &config).await { + Ok(adapter) => self.set_live(config, adapter), + Err(err) => tracing::warn!( + error = %err, + "iii-queue: stored configuration selects an unavailable adapter; continuing with static config" + ), + } + } else { + self.set_config(config); + } + } Err(err) => tracing::warn!( error = %err, "iii-queue: stored configuration is invalid; continuing with static config" ),🤖 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 `@engine/src/workers/queue/queue.rs` around lines 1149 - 1150, The issue is that when the fetched configuration is adopted via set_config(config) in the Ok(()) branch after config.validate(), the adapter is not resolved from the fetched configuration first. This causes LiveState to have a mismatch of new config combined with the old adapter. To fix this, before calling set_config(config), you need to resolve the startup adapter from the fetched config so that the adapter is properly updated to match the fetched configuration before adoption.
🤖 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 `@engine/src/workers/queue/queue.rs`:
- Around line 151-156: The resolve_queue_key() method creates its own config
snapshot internally, but callers then snapshot the adapter separately. This
creates a race condition where the adapter generation may not match the config
used to determine the queue key. Refactor resolve_queue_key() to accept a
LiveState snapshot parameter instead of creating one internally, then update all
callers (including redrive_message, discard_message, console_topic_stats, and
console_dlq_messages) to obtain a single snapshot and pass it to
resolve_queue_key() along with their adapter operations to ensure consistency
between config and adapter generations.
- Around line 645-669: The `register_trigger()` method currently subscribes
durable triggers to the current adapter via the `adapter.subscribe()` call, but
when the adapter is hot-swapped during `set_live()`, these subscriptions are
lost because they only exist in the old adapter instance. To fix this, you must
either replay all existing durable subscriber triggers onto the new adapter
immediately after the adapter replacement completes, or refactor to store
durable subscription state outside the adapter so subscriptions persist across
adapter replacements. If replaying, collect all registered durable subscriber
triggers and call their subscribe logic on the new adapter instance after the
swap occurs.
- Around line 1149-1150: The issue is that when the fetched configuration is
adopted via set_config(config) in the Ok(()) branch after config.validate(), the
adapter is not resolved from the fetched configuration first. This causes
LiveState to have a mismatch of new config combined with the old adapter. To fix
this, before calling set_config(config), you need to resolve the startup adapter
from the fetched config so that the adapter is properly updated to match the
fetched configuration before adoption.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 3df62e24-f0e9-43b3-bd58-9b90f6466dc4
📒 Files selected for processing (2)
engine/src/workers/queue/adapters/builtin/adapter.rsengine/src/workers/queue/queue.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- engine/src/workers/queue/adapters/builtin/adapter.rs
andersonleal
force-pushed
the
feat/iii-queue-configuration-worker
branch
from
d9b3737 to
eba161d
Compare
sergiofilhowz
previously approved these changes
Jun 18, 2026
Migrate the `iii-queue` worker to consume the builtin `configuration` worker for its settings, mirroring the `iii-http` integration (#1842). Queue config (adapter/transport selection and per-queue retries, concurrency, FIFO ordering, backoff) is now runtime-editable and hot-reloaded: the worker registers a JSON-schema'd entry, seeds it from the config.yaml block only on first boot, reads the live value (with `${VAR:default}` expansion) at startup, and hot-applies `configuration:updated` events without an engine restart. After first boot the configuration entry is the source of truth; config.yaml is seed-only and survives restarts. Scope decisions: - Full adapter hot-swap: a runtime change to `adapter` re-instantiates the transport and restarts every consumer (queued messages in the old transport do not migrate — accepted tradeoff, documented in-code). - Best-effort per-queue apply: a strict fetch+validate gate (any failure keeps the previous config); past the gate each queue's consumer restart is independent, so one failure is logged while the others apply and that queue keeps its previous consumer. Hardening: - Boot path validates the fetched config and falls back to the seed, matching the runtime gate (a value `apply_config` rejects can be stored via `configuration::set` since the schema can't express the fifo cross-field rule). - `destroyed` flag (set under `apply_lock` in `destroy`) makes a late apply — from an in-flight event or the one-shot retry — bail instead of re-spawning consumers on a torn-down worker. - Builtin adapter nacks still-outstanding deliveries when a consumer channel closes, so a restart returns buffered messages to the queue (at-least-once) instead of stranding them (no visibility-timeout reclaim otherwise). Tests: 7 unit (configuration), 10 e2e (seed/no-clobber/hot-add/hot-change/ adapter-swap/remove/invalid-keeps-previous/destroy-gate/env-expansion/ timeout-retry), plus a builtin-adapter nack-on-close test. Coverage: config.rs 100%, configuration.rs 97.7%, queue.rs 96.8%.
…argo fmt CI: `all_functions_on_bare_engine_are_iii_builtins` flagged the new `iii-queue::on-config-change` handler as a non-builtin function. `is_iii_builtin_function_id` allowlists `iii-http::` and `iii-state::` (which register the same config handler) but not `iii-queue::`; add the prefix and a matching unit assertion. Also apply `cargo fmt`.
Two valid findings from the PR review: 1. Builtin adapter could strand the oldest buffered delivery on consumer stop. The `outstanding` window was trimmed BEFORE the channel send; if the buffer was full and the send blocked, then the receiver was dropped, the just-evicted delivery_id was never nacked (stranded in delivery_map). Trim AFTER a successful send instead — a send only unblocks once the consumer makes room, so anything beyond the window has necessarily been received. Added a regression test (`consumer_stop_nacks_oldest_when_buffer_full`) that reproduces the stranding (prefetch=2, 3 jobs, drop receiver mid-blocked-send). 2. Adapter hot-swap exposed a mixed-state window: `set_config` ran before `set_adapter` with a consumer drain (awaits) between them, so a concurrent enqueue could read the new per-queue config while still publishing through the old, dying transport. Combine config + adapter into a single `LiveState` behind one `RwLock` and swap them atomically via `set_live`; `enqueue_to_function_queue` now reads both from one `live_snapshot()`, so no reader can observe a half-applied swap.
Adapter config now renders as a closed per-adapter discriminated union (keyed on `name`), each branch carrying a typed `config`, mirroring the iii-state integration — the console shows typed fields instead of an opaque, read-only object. Branches: builtin/redis/bridge, plus rabbitmq and an in-process `memory` test transport under their respective features. `configuration::set` rejects unknown adapter names at the schema gate. Also reject `concurrency: 0` (schema `range(min=1)` + `validate()` guard): it reaches `mpsc::channel(0)` and panics the consumer — crashing boot from a seed value, or the config-change handler mid-apply at runtime. Tests: closed-union schema tests (real jsonschema validator), unbuildable- adapter keeps-previous + unknown-name rejection e2e, strengthened consumer-stop nack tests (assert redelivery to the DLQ, not a silent drop), and adapter rebuild-failure coverage.
andersonleal
force-pushed
the
feat/iii-queue-configuration-worker
branch
from
eba161d to
342d709
Compare
This was referenced Jun 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Migrates the
iii-queueworker to consume the builtinconfigurationworker for its settings, mirroring theiii-httpintegration (#1842).Queue config — the backing adapter/transport plus per-queue retries, concurrency, FIFO ordering, and backoff — is now runtime-editable and hot-reloaded. The worker registers a JSON-schema'd entry, seeds it from the
config.yamlblock only on first boot, reads the live value (with${VAR:default}expansion) at startup, and hot-appliesconfiguration:updatedevents without an engine restart. After first boot the configuration entry is the source of truth;config.yamlis seed-only and survives restarts.Scope decisions
adapterre-instantiates the transport and restarts every consumer. Messages still queued in the old transport do not migrate (accepted tradeoff, documented in-code).Hardening (from review)
apply_configrejects at runtime can still be stored viaconfiguration::set(the JSON schema can't express the fifo cross-field rule), so without this a bad stored value would boot a broken consumer on restart.destroyedflag (set underapply_lockindestroy) makes a late apply — from an in-flight event or the one-shot retry task — bail instead of re-spawning consumers on a torn-down worker (the queue analogue ofHttpWorker'sshutdown_rxguard).Tests & coverage
configuration.rs), 1 builtin-adapter nack-on-close test, 10 e2e tests (queue_configuration_e2e.rs): seed-on-first-boot, no-clobber across restart, hot-add (live consumer), hot-change, adapter hot-swap (withArc::ptr_eqproving the transport was rebuilt), hot-remove, invalid-value-keeps-previous, boot-fallback-to-seed, destroy gate,${VAR}expansion, and timeout one-shot retry (paused clock).queue_integration(12) exercise the graceful-degradation path (no configuration worker present).llvm-covon the queue module:config.rs100%,configuration.rs97.7%,queue.rs96.8%, builtin adapter 97.3%. Residual is error/tracing branches and the unrelateddurable:subscriberpaths.Verification
Summary by CodeRabbit
iii-queue::prefix.