Test invalid builder rates #339
Conversation
Reviewer's GuideThis pull request refactors the PushQueues API into a fluent builder pattern with centralized validation (including invalid-rate checks), enhances dead-letter queue handling and log throttling, and overhauls tests and documentation to adopt the new builder interface, share fixtures, and improve examples and formatting. Sequence diagram for frame push with dead-letter queue and rate limitingsequenceDiagram
participant Producer
participant PushHandle
participant RateLimiter
participant PushQueues
participant DLQ
Producer->>PushHandle: push_high_priority(frame)
PushHandle->>PushQueues: reserve queue slot
alt Rate limiting enabled
PushHandle->>RateLimiter: acquire token
RateLimiter-->>PushHandle: token granted
end
alt Queue full
PushHandle->>DLQ: try_send(frame)
alt DLQ full or absent
PushHandle->>PushHandle: log warning (throttled)
end
else Queue available
PushHandle->>PushQueues: send frame
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. Note Reviews pausedUse the following commands to manage reviews:
Summary by CodeRabbit
WalkthroughReplace the legacy monolithic push implementation with a modular, builder-based push-queue subsystem; add builder, handle, errors and queues submodules; remove the old Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor App as Application
participant PB as PushQueues::<F>::builder()
participant B as PushQueuesBuilder<F>
participant Q as PushQueues<F>
participant H as PushHandle<F>
App->>PB: builder()
PB-->>App: PushQueuesBuilder<F>
App->>B: high_capacity(...), low_capacity(...), rate(...), dlq(...)
App->>B: build()
B-->>App: Result<(Q, H)>
note over Q,H #DDEBF7: Pair returned on success (or Err on invalid config)
sequenceDiagram
autonumber
actor Prod as Producer
participant H as PushHandle<F>
participant HP as high_prio_tx
participant LP as low_prio_tx
participant RL as RateLimiter (opt)
participant DLQ as DLQ Sender (opt)
alt push_high_priority
Prod->>H: push_high_priority(frame)
H->>HP: reserve/send (async)
else push_low_priority
Prod->>H: push_low_priority(frame)
H->>LP: reserve/send (async)
end
opt rate limiting enabled
H->>RL: throttle/reserve
end
alt try_push (non-blocking)
Prod->>H: try_push(frame, priority, policy)
H->>HP: try_send / full?
H->>LP: try_send / full?
alt Full + Drop policy
H->>DLQ: try_send(frame)
note right of H #FFF2CC: Log warn periodically on DLQ drops
else Full + ReturnError
H-->>Prod: Err(QueueFull)
end
end
sequenceDiagram
autonumber
participant Q as PushQueues<F>
participant HR as high_priority_rx
participant LR as low_priority_rx
participant Conn as Connection Actor
loop recv loop
Conn->>Q: recv().await
alt high priority available
Q->>HR: recv()
HR-->>Q: Some(F)
Q-->>Conn: (High, F)
else low priority available
Q->>LR: recv()
LR-->>Q: Some(F)
Q-->>Conn: (Low, F)
else both closed/empty
Q-->>Conn: None
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
✨ Finishing Touches🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Have the following now been resolved?
|
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
|
@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix: Excess Number of Function Arguments What lead to degradation? PushQueues.build_with_rate_dlq has 6 arguments, threshold = 4 Functions with many arguments indicate either a) low cohesion where the function has too many responsibilities, or b) a missing abstraction that encapsulates those arguments. The threshold for the Rust language is 4 function arguments. Start by investigating the responsibilities of the function. Make sure it doesn't do too many things, in which case it should be split into smaller and more cohesive functions. Consider the refactoring INTRODUCE PARAMETER OBJECT to encapsulate arguments that refer to the same logical concept. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
This comment was marked as resolved.
This comment was marked as resolved.
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 33
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (10)
tests/wireframe_protocol.rs (1)
66-79: Avoid explicit drop; ignore the unused binding in the pattern.Destructure to
_queuesinstead of callingdrop.- let (queues, handle) = queues; + let (_queues, handle) = queues; hooks.on_connection_setup(handle, &mut ConnectionContext); - drop(queues); // silence unused warningstests/connection_actor.rs (2)
1-554: Split this test module; it exceeds the 400-line limit.Extract fairness tests, shutdown/abort tests, and error propagation tests into separate files to meet the size guideline and improve focus.
I can draft a patch that introduces
tests/connection_actor_fairness.rs,tests/connection_actor_shutdown.rs, andtests/connection_actor_errors.rs, moving the respective test groups and shared fixtures.
397-410: Deflake the blocking assertion.The 50 ms timeout can be flaky under load. Increase the bound or switch to a non-blocking
try_push_high_priorityif available.- let blocked = timeout(Duration::from_millis(50), handle.push_high_priority(2)).await; + let blocked = timeout(Duration::from_millis(200), handle.push_high_priority(2)).await;docs/hardening-wireframe-a-guide-to-production-resilience.md (2)
354-366: Downgrade DLQ loss log from error to warning.Reflect the change to warning severity for DLQ loss events.
- tracing::error!("Push queue and DLQ are both full. Frame lost."); + tracing::warn!("Push queue and DLQ are both full. Frame lost.");Add a brief note that DLQ loss logging is throttled per handle.
336-341: Use Oxford -ize spelling in prose (“recognize”).Keep documentation spelling consistent.
-This explicit marker lets clients recognise that the logical stream has ended and +This explicit marker lets clients recognize that the logical stream has ended andtests/push_policies.rs (1)
68-71: Eliminate flakiness from timing-based assertionAvoid real-time timeouts in tests. Check queue emptiness without sleeping.
-assert!( - timeout(Duration::from_millis(20), queues.recv()) - .await - .is_err() -); +assert!(queues.try_recv().is_err(), "queue should be empty");tests/push.rs (3)
157-176: Fix deadlock: timeout with paused time never elapsesUsing time::timeout under time::pause causes a hang; neither the timeout nor the rate-limited push completes. Assert pending without time instead, then advance virtual time to unblock.
@@ - let attempt = match priority { - PushPriority::High => { - time::timeout(Duration::from_millis(10), handle.push_high_priority(2u8)).await - } - PushPriority::Low => { - time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8)).await - } - }; - - assert!(attempt.is_err(), "second push should block"); + use futures::FutureExt; + let fut = match priority { + PushPriority::High => handle.push_high_priority(2u8), + PushPriority::Low => handle.push_low_priority(2u8), + }; + tokio::pin!(fut); + tokio::task::yield_now().await; // register w/ scheduler + assert!( + fut.as_mut().now_or_never().is_none(), + "second push should be pending under rate limit" + );Add import near the top:
use tokio::time::{self, Duration}; +use futures::FutureExt;
206-213: Fix deadlock: shared limiter test also uses timeout under paused timeApply the same pending-assertion approach here.
- let attempt = time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8)).await; - assert!(attempt.is_err(), "second push should block across queues"); + use futures::FutureExt; + let mut fut = handle.push_low_priority(2u8); + tokio::pin!(fut); + tokio::task::yield_now().await; + assert!( + fut.as_mut().now_or_never().is_none(), + "second push should be pending across queues" + );
258-263: Fix deadlock: burst-limit test uses timeout under paused timeAvoid timeout + paused time; assert pending then advance.
- let res = time::timeout(Duration::from_millis(10), handle.push_high_priority(99)).await; - assert!( - res.is_err(), - "Push exceeding burst capacity should be rate limited" - ); + use futures::FutureExt; + let mut fut = handle.push_high_priority(99); + tokio::pin!(fut); + tokio::task::yield_now().await; + assert!( + fut.as_mut().now_or_never().is_none(), + "push exceeding burst capacity should be pending" + );docs/asynchronous-outbound-messaging-design.md (1)
19-19: Standardise on en-GB Oxford -ize spelling in docsAdopt -ize uniformly per project style (retain -our). Fix mixed -ise/-ize usages.
- required extra synchronisation + required extra synchronization - ### 3.1 Prioritised Message Queues + ### 3.1 Prioritized Message Queues - The prioritised write loop + The prioritized write loop - The flow diagram below summarises the fairness logic. + The flow diagram below summarizes the fairness logic. - clean, organised, and extensible configuration API + clean, organized, and extensible configuration APIAlso applies to: 63-66, 81-86, 158-162, 514-517
♻️ Duplicate comments (1)
src/push/queues/mod.rs (1)
77-84: Collapse long parameter list with a parameter object for DLQ logging.Reduce argument count and improve cohesion by bundling DLQ logging knobs into a small struct. This also addresses the CodeScene warning on “Excess Number of Function Arguments.”
@@ +#[derive(Clone, Copy, Debug)] +pub(super) struct DlqLogPolicy { + pub every_n: usize, + pub interval: Duration, +} @@ - dlq_log_every_n: usize, - dlq_log_interval: Duration, + dlq_log: DlqLogPolicy, ) -> Result<(Self, PushHandle<F>), PushConfigError> { @@ - dlq_last_log: Mutex::new( - Instant::now() - .checked_sub(dlq_log_interval) - .unwrap_or_else(Instant::now), - ), - dlq_log_every_n, - dlq_log_interval, + dlq_last_log: Mutex::new( + Instant::now() + .checked_sub(dlq_log.interval) + .unwrap_or_else(Instant::now), + ), + dlq_log_every_n: dlq_log.every_n, + dlq_log_interval: dlq_log.interval, }; @@ - Self::build_with_rate_dlq( - high_capacity, - low_capacity, - rate, - dlq, - 100, - Duration::from_secs(10), - ) + Self::build_with_rate_dlq( + high_capacity, + low_capacity, + rate, + dlq, + DlqLogPolicy { every_n: 100, interval: Duration::from_secs(10) }, + )AI coding-agent prompt: Refactor src/push/queues/mod.rs to introduce `DlqLogPolicy` as shown in the diff and update all call sites. Then run `make lint` and `make test`. If other internal helpers call `build_with_rate_dlq`, update them similarly. Ensure docs mention the new parameter object only for internal use (deprecated constructors remain unchanged externally).Also applies to: 206-214, 113-120
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
💡 Knowledge Base configuration:
- Jira integration is disabled
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (27)
README.md(1 hunks)docs/asynchronous-outbound-messaging-design.md(1 hunks)docs/efficiency-report.md(3 hunks)docs/hardening-wireframe-a-guide-to-production-resilience.md(1 hunks)docs/multi-packet-and-streaming-responses-design.md(2 hunks)src/connection.rs(2 hunks)src/push.rs(0 hunks)src/push/mod.rs(1 hunks)src/push/queues/builder.rs(1 hunks)src/push/queues/errors.rs(1 hunks)src/push/queues/handle.rs(1 hunks)src/push/queues/mod.rs(1 hunks)src/response.rs(1 hunks)src/session.rs(6 hunks)tests/advanced/concurrency_loom.rs(1 hunks)tests/advanced/interaction_fuzz.rs(1 hunks)tests/async_stream.rs(1 hunks)tests/connection_actor.rs(3 hunks)tests/correlation_id.rs(1 hunks)tests/push.rs(8 hunks)tests/push_policies.rs(8 hunks)tests/server.rs(1 hunks)tests/session_registry.rs(1 hunks)tests/stream_end.rs(3 hunks)tests/support.rs(1 hunks)tests/wireframe_protocol.rs(6 hunks)tests/world.rs(3 hunks)
💤 Files with no reviewable changes (1)
- src/push.rs
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
⚙️ CodeRabbit configuration file
**/*.rs: * Seek to keep the cyclomatic complexity of functions no more than 12.
Adhere to single responsibility and CQRS
Place function attributes after doc comments.
Do not use
returnin single-line functions.Move conditionals with >2 branches into a predicate function.
Avoid
unsafeunless absolutely necessary.Every module must begin with a
//!doc comment that explains the module's purpose and utility.Comments and docs must follow en-GB-oxendict (-ize / -our) spelling and grammar
Lints must not be silenced except as a last resort.
#[allow]is forbidden.- Only narrowly scoped
#[expect(lint, reason = "...")]is allowed.- No lint groups, no blanket or file-wide suppression.
- Include
FIXME:with link if a fix is expected.Where code is only used by specific features, it must be conditionally compiled or a conditional expectation for unused_code applied.
Use
rstestfixtures for shared setup and to avoid repetition between tests.Replace duplicated tests with
#[rstest(...)]parameterised cases.Prefer
mockallfor mocks/stubs.Prefer
.expect()over.unwrap()Ensure that any API or behavioural changes are reflected in the documentation in
docs/Ensure that any completed roadmap steps are recorded in the appropriate roadmap in
docs/Files must not exceed 400 lines in length
- Large modules must be decomposed
- Long match statements or dispatch tables should be decomposed by domain and collocated with targets
- Large blocks of inline data (e.g., test fixtures, constants or templates) must be moved to external files and inlined at compile-time or loaded at run-time.
Environment access (env::set_var and env::remove_var) are always unsafe in Rust 2024 and MUST be marked as such
- For testing of functionality depending upon environment variables, dependency injection and the
mockablecrate are the preferred option.- If mockable cannot be used, env mutations in tests ...
Files:
tests/server.rssrc/response.rssrc/push/queues/errors.rssrc/session.rstests/world.rssrc/connection.rstests/support.rssrc/push/mod.rstests/advanced/concurrency_loom.rstests/wireframe_protocol.rstests/stream_end.rstests/async_stream.rstests/session_registry.rssrc/push/queues/handle.rstests/connection_actor.rstests/push_policies.rssrc/push/queues/builder.rstests/correlation_id.rstests/advanced/interaction_fuzz.rstests/push.rssrc/push/queues/mod.rs
**/*.md
⚙️ CodeRabbit configuration file
**/*.md: * Avoid 2nd person or 1st person pronouns ("I", "you", "we")
- Use en-GB-oxendict (-ize / -our) spelling and grammar
- Headings must not be wrapped.
- Documents must start with a level 1 heading
- Headings must correctly increase or decrease by no more than one level at a time
- Use GitHub-flavoured Markdown style for footnotes and endnotes.
- Numbered footnotes must be numbered by order of appearance in the document.
Files:
docs/multi-packet-and-streaming-responses-design.mddocs/asynchronous-outbound-messaging-design.mdREADME.mddocs/hardening-wireframe-a-guide-to-production-resilience.mddocs/efficiency-report.md
🪛 LanguageTool
docs/efficiency-report.md
[style] ~98-~98: Using four (or more) nouns in a row may decrease readability.
Context: ... Connection Handling - Bottleneck: Connection actor event loop and fairness tracking - **Critical Path...
(FOUR_NN)
[style] ~99-~99: Using four (or more) nouns in a row may decrease readability.
Context: ... Critical Path: tokio::select! in connection actor - Optimization Priority: Medium—affects per-connection perfor...
(FOUR_NN)
🔍 Remote MCP
Here are a few concrete details from the diff that merit attention during review:
• The core constructor function PushQueues::build_with_rate_dlq now takes six parameters—
(high_capacity: usize, low_capacity: usize, rate: Option<usize>, dlq: Option<Sender<F>>, dlq_log_every_n: usize, dlq_log_interval: Duration)—
and the deprecated helpers (bounded_with_rate_dlq) call into it with two hard‐coded logging defaults (100, 10 s). This long parameter list suggests a refactor opportunity to bundle related settings into a PushQueueConfig struct rather than passing six separate args (github.com).
• The PushQueuesBuilder now offers a convenience method unlimited() that is simply shorthand for .rate(None), making it clearer when rate limiting is disabled (github.com).
• A dedicated shared fixture (tests/support.rs) now provides a builder() function returning a PushQueuesBuilder with unit capacities, avoiding repetition across tests (github.com).
• There is a new test builder_accepts_max_rate verifying that Some(MAX_PUSH_RATE) is accepted by the builder’s validation logic, closing the boundary‐case gap (github.com).
• The PushHandle send path has been tightened: after reserving capacity, the code now checks if permit.send(frame) returns a closed‐channel indicator and propagates Err(PushError::Closed) rather than silently dropping the frame (github.com).
🔇 Additional comments (14)
src/response.rs (1)
47-47: Standardize spelling to -ize; OK.The docstring uses “optimized”, which matches the en-GB Oxford -ize style. No behavioural change.
docs/efficiency-report.md (1)
14-16: Optimisation verified:LengthPrefixedProcessor::decodeusessrc.split_to(len).freeze().to_vec()at line 75 insrc/frame/processor.rs, matching the efficiency report.tests/server.rs (1)
46-46: Replace unwrap with expect; OK.The panic message improves failure diagnostics while preserving behaviour.
src/connection.rs (2)
101-106: Docs: switch to builder-based construction; OK.Example reflects the new API and uses
no_runappropriately.
132-145: Docs: updatenew()example to builder; OK.Matches the public API and house style.
tests/async_stream.rs (1)
26-31: LGTM: builder migration is correct.The builder initialisation and error handling read cleanly; the test semantics remain unchanged.
docs/hardening-wireframe-a-guide-to-production-resilience.md (1)
256-280: Fix ordering in example: reserve capacity before rate limiting.The PR states “Reserve queue slots before rate limiting”. Update the snippet to match the implementation behaviour.
async fn push(&self, frame: Frame) -> Result<(), PushError> { - // If a limiter is configured, wait for a token before queuing the frame. - if let Some(ref limiter) = self.limiter { - limiter.acquire(1).await; - } - - self.tx.send(frame).await.map_err(|_| /*...*/) + // Reserve capacity before waiting on the limiter. + let permit = self.tx.reserve().await.map_err(|_| /*...*/)?; + if let Some(ref limiter) = self.limiter { + limiter.acquire(1).await; + } + permit.send(frame).map_err(|_| /*...*/) }tests/push_policies.rs (1)
185-193: LGTM: DLQ warning assertionThe warning-level log assertion for DLQ failures matches the updated semantics.
tests/push.rs (1)
61-79: Keep: test correctly proves bursts succeed when unlimitedThe burst test is sound and will not hang because pushes complete immediately with unlimited rate.
src/push/queues/handle.rs (2)
67-75: Propagate closed error after send attemptReturned sender being closed implies the frame was not delivered. The check and error propagation look correct.
Add an async test that closes queues after reserving capacity but before send to ensure PushError::Closed is returned.
157-169: Fix compile error and division-by-zero in DLQ loggingis_multiple_of is not in scope and will not compile; modulo by a zero log_every_n would panic. Gate the count path and fall back to interval-based logging when n == 0.
- if dropped.is_multiple_of(log_every_n) - || now.duration_since(*last) > log_interval - { + let by_count = log_every_n > 0 && dropped % log_every_n == 0; + let by_interval = now.duration_since(*last) > log_interval; + if by_count || by_interval {Likely an incorrect or invalid review comment.
src/push/queues/builder.rs (1)
20-37: LGTM: Builder docs clearly state validation happens at build timeDocs align with guidelines and examples compile as no_run snippets.
docs/asynchronous-outbound-messaging-design.md (1)
356-365: LGTM: Public API diagrams match the builder-based surfaceDiagrams and method lists reflect PushQueues::builder() and PushQueuesBuilder.
Also applies to: 368-373
src/push/queues/mod.rs (1)
97-107: LGTM: channel wiring and rate-limiter construction.Channel setup and optional limiter initialisation look correct and side-effect free. Returning
(PushQueues, PushHandle)via a shared inner is clean.Also applies to: 121-128
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
✅ Actions performedReview triggered.
|
|
@coderabbitai pause |
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
✅ Actions performedReviews paused. |
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
leynos
force-pushed
the
codex/introduce-pushqueuesbuilder-with-deprecation-225o2i
branch
from
6e0a0d8 to
cc1a2c6
Compare
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
…Q logging - Avoid reserving tokens for probe-only futures by replacing with non-blocking loop plus short sleeps. - Use macros in push handle so captures warnings. - Log every DLQ drop by default to make tests deterministic. - Update docs to describe configurable DLQ logging cadence.
…eplace tracing macros with log across server/app/connection.\n- Emit both log and tracing panic events to satisfy tests.\n- Add regression test to guard rate limiter token reservation.\n- Keep DLQ logging default verbose for test determinism.
leynos
deleted the
codex/introduce-pushqueuesbuilder-with-deprecation-225o2i
branch
1 participant
Summary
Nonedisables throttling and frames are dropped when the DLQ is absent or fullis_invalid_ratepredicatePushErroras non-exhaustivebuild_via_builderhelperrate(None)disables throttling and share builder fixtures across push policy testsno_run, disable throttling in more tests, remove#[tokio::test]from doctests, and downgrade DLQ loss logs to warningsunwrapcalls forexpectin push handle examplesTesting
make fmtmake lintmake testmake markdownlintmake nixie(fails: KeyboardInterrupt)https://chatgpt.com/codex/tasks/task_e_68aeba8d630c8322bcab55de80b2ef06
Summary by Sourcery
Replace PushQueues’ legacy constructors with a flexible PushQueuesBuilder API and enforce robust rate‐limit and dead‐letter queue behavior across code, tests, and documentation
New Features:
Bug Fixes:
Enhancements:
Documentation:
Tests:
Chores: