Implement run state consolidation in connection actor#146
Conversation
Reviewer's GuideThis PR refactors the ConnectionActor to consolidate its run state by swapping out a PushQueues struct for optional mpsc receivers, replacing multiple boolean flags with a RunState enum and source counters, and updating the polling and shutdown logic to operate directly on these Option receivers; it also marks the consolidation milestone as complete in the roadmap. Class diagram for consolidated run state in ConnectionActorclassDiagram
class ConnectionActor {
- Option<mpsc::Receiver<F>> high_rx
- Option<mpsc::Receiver<F>> low_rx
- Option<FrameStream<F, E>> response
- CancellationToken shutdown
- ProtocolHooks<F> hooks
- FairnessConfig fairness
- usize high_counter
- Option<Instant> high_start
+ set_fairness(fairness: FairnessConfig)
+ poll_sources(state: &mut ActorState, out: &mut Vec<F>)
+ process_shutdown(state: &mut ActorState)
+ process_high(res: Option<F>, state: &mut ActorState, out: &mut Vec<F>)
+ process_low(res: Option<F>, state: &mut ActorState, out: &mut Vec<F>)
+ process_response(res: Option<Result<F, WireframeError<E>>>, state: &mut ActorState, out: &mut Vec<F>)
+ start_shutdown(state: &mut ActorState)
+ after_high(out: &mut Vec<F>, state: &mut ActorState)
+ after_low()
+ handle_response(res: Option<Result<F, WireframeError<E>>>, state: &mut ActorState, out: &mut Vec<F>)
}
class ActorState {
- RunState run_state
- usize closed_sources
- usize total_sources
+ mark_closed()
+ start_shutdown()
+ is_active() bool
+ is_shutting_down() bool
+ is_done() bool
}
class RunState {
<<enum>>
Active
ShuttingDown
Finished
}
ConnectionActor --> ActorState
ActorState --> RunState
Class diagram for removed PushQueues and PushClosed structuresclassDiagram
class PushQueues {
- mpsc::Receiver<F> high_priority_rx
- mpsc::Receiver<F> low_priority_rx
}
class PushClosed {
- bool high
- bool low
}
%% These classes were removed from ConnectionActor and ActorState
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Summary by CodeRabbit
Summary by CodeRabbit
WalkthroughThe Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant ConnectionActor
participant HighPriorityQueue
participant LowPriorityQueue
participant ResponseStream
loop Actor Poll Loop
alt High-priority frame available
HighPriorityQueue->>ConnectionActor: Send frame
ConnectionActor->>ConnectionActor: Process frame
end
alt Low-priority frame available
LowPriorityQueue->>ConnectionActor: Send frame
ConnectionActor->>ConnectionActor: Process frame
end
alt Response available
ResponseStream->>ConnectionActor: Send response
ConnectionActor->>ConnectionActor: Process response
end
alt All sources closed
ConnectionActor->>ConnectionActor: Mark as Finished
end
end
Possibly related issues
Possibly related PRs
Suggested reviewers
Poem
📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (1)
🧰 Additional context used📓 Path-based instructions (2)`**/*.rs`: Every Rust module must begin with a module level (`//!`) comment expl...
📄 Source: CodeRabbit Inference Engine (AGENTS.md) List of files the instruction was applied to:
`**/*.rs`: * Seek to keep the cyclomatic complexity of functions no more than 12...
⚙️ Source: CodeRabbit Configuration File List of files the instruction was applied to:
🧬 Code Graph Analysis (1)src/connection.rs (1)
⏰ Context from checks skipped due to timeout of 90000ms (2)
🔇 Additional comments (6)
✨ 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. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey @leynos - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `src/connection.rs:159` </location>
<code_context>
}
- res = Self::next_response(&mut self.response), if !state.shutting_down && !state.resp_closed => {
+ res = Self::next_response(&mut self.response), if resp_available && !state.is_shutting_down() => {
self.process_response(res, state, out)?;
}
</code_context>
<issue_to_address>
Check for shutdown before polling response to avoid unnecessary work.
The select branch for `next_response` now checks both `resp_available` and `!state.is_shutting_down()`. Ensure that this logic matches the intended shutdown semantics and does not introduce a race where a response is polled after shutdown has started.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| } | ||
|
|
||
| res = Self::next_response(&mut self.response), if !state.shutting_down && !state.resp_closed => { | ||
| res = Self::next_response(&mut self.response), if resp_available && !state.is_shutting_down() => { |
There was a problem hiding this comment.
question (review_instructions): Check for shutdown before polling response to avoid unnecessary work.
The select branch for next_response now checks both resp_available and !state.is_shutting_down(). Ensure that this logic matches the intended shutdown semantics and does not introduce a race where a response is polled after shutdown has started.
Review instructions:
Path patterns: **/*
Instructions:
Create code-review comments for ALL issues. Avoid making general observations or non-specific feedback if at all possible.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
📜 Review details
Configuration used: CodeRabbit UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (2)
docs/asynchronous-outbound-messaging-roadmap.md(1 hunks)src/connection.rs(9 hunks)
🧰 Additional context used
📓 Path-based instructions (7)
`docs/**/*.md`: Use the markdown files within the `docs/` directory as a knowled...
docs/**/*.md: Use the markdown files within thedocs/directory as a knowledge base and source of truth for project requirements, dependency choices, and architectural decisions.
Proactively update the relevant file(s) in thedocs/directory to reflect the latest state when new decisions are made, requirements change, libraries are added/removed, or architectural patterns evolve.
Documentation in thedocs/directory must use en-GB-oxendict spelling and grammar, except for the word 'license'.
📄 Source: CodeRabbit Inference Engine (AGENTS.md)
List of files the instruction was applied to:
docs/asynchronous-outbound-messaging-roadmap.md
`**/*.md`: Validate Markdown files using `markdownlint *.md **/*.md`. Run `mdfor...
**/*.md: Validate Markdown files usingmarkdownlint *.md **/*.md.
Runmdformat-allafter any documentation changes to format all Markdown files and fix table markup.
Validate Markdown Mermaid diagrams using thenixieCLI by runningnixie *.md **/*.md.
Markdown paragraphs and bullet points must be wrapped at 80 columns.
Code blocks in Markdown files must be wrapped at 120 columns.
Tables and headings in Markdown files must not be wrapped.
📄 Source: CodeRabbit Inference Engine (AGENTS.md)
List of files the instruction was applied to:
docs/asynchronous-outbound-messaging-roadmap.md
`docs/*`: Each file in the docs/ directory provides design, roadmap, testing, op...
docs/*: Each file in the docs/ directory provides design, roadmap, testing, operations, or reference documentation for the project. Use these files as a reference for understanding the project's architecture, development plans, testing strategies, operational guidelines, and documentation conventions.
📄 Source: CodeRabbit Inference Engine (docs/contents.md)
List of files the instruction was applied to:
docs/asynchronous-outbound-messaging-roadmap.md
`docs/**/*.md`: Use British English spelling based on the Oxford English Diction...
docs/**/*.md: Use British English spelling based on the Oxford English Dictionary, except retain US spelling in API names (e.g., 'color').
Use the Oxford comma in lists.
Write headings in sentence case and use Markdown heading levels in order without skipping.
Follow markdownlint recommendations for Markdown formatting.
Always use fenced code blocks with a language identifier; use 'plaintext' for non-code text.
Use '-' as the first level bullet and renumber lists when items change.
Prefer inline links using 'text' or angle brackets around the URL.
Expand any uncommon acronym on first use, e.g., Continuous Integration (CI).
Wrap paragraphs at 80 columns, code at 120 columns, and do not wrap tables.
Use footnotes referenced with '[^label]'.
When embedding figures, use '' and provide concise alt text describing the content.
Add a short description before each Mermaid diagram for screen readers.
📄 Source: CodeRabbit Inference Engine (docs/documentation-style-guide.md)
List of files the instruction was applied to:
docs/asynchronous-outbound-messaging-roadmap.md
`**/*.md`: * Avoid 2nd person or 1st person pronouns ("I", "you", "we") * Use en...
**/*.md: * Avoid 2nd person or 1st person pronouns ("I", "you", "we")
- Use en-oxendic spelling and grammar.
- Paragraphs and bullets must be wrapped to 80 columns, except where a long URL would prevent this (in which case, silence MD013 for that line)
- Code blocks should be wrapped to 120 columns.
- Headings must not be wrapped.
⚙️ Source: CodeRabbit Configuration File
List of files the instruction was applied to:
docs/asynchronous-outbound-messaging-roadmap.md
`**/*.rs`: Every Rust module must begin with a module level (`//!`) comment expl...
**/*.rs: Every Rust module must begin with a module level (//!) comment explaining the module's purpose and utility.
Document public Rust APIs using Rustdoc comments (///) so documentation can be generated with cargo doc.
Place function attributes after doc comments.
📄 Source: CodeRabbit Inference Engine (AGENTS.md)
List of files the instruction was applied to:
src/connection.rs
`**/*.rs`: * Seek to keep the cyclomatic complexity of functions no more than 12...
**/*.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 must use en-GB-oxendict 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.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/
⚙️ Source: CodeRabbit Configuration File
List of files the instruction was applied to:
src/connection.rs
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: build-test
🔇 Additional comments (6)
src/connection.rs (5)
45-46: LGTM! Clean refactoring to support state consolidation.The change from a single
PushQueues<F>field to separate optional receivers improves state tracking and aligns perfectly with the run state consolidation objective.
78-79: Correct extraction of receivers from PushQueues.The receivers are properly extracted and wrapped in
Some, maintaining the initial state where both queues are available.
114-114: Proper state initialisation with response stream awareness.The state correctly tracks whether a response stream exists, which is essential for accurate source counting.
213-223: Well-structured shutdown sequence.The shutdown logic properly closes all receivers and drops the response stream whilst tracking the closed state. The use of
take()for the response stream is particularly elegant.
231-244: Robust fairness logic with proper Option handling.The updated fairness check correctly handles the optional
low_rxand properly manages the disconnection case by setting it toNoneand marking the source as closed.docs/asynchronous-outbound-messaging-roadmap.md (1)
18-18: Documentation correctly reflects completed implementation.The roadmap update accurately marks the "Run state consolidation" task as complete, which aligns with the refactoring implemented in
src/connection.rs.
1 participant
Summary
Optionreceivers and a closed source counterTesting
make lintmake testhttps://chatgpt.com/codex/tasks/task_e_6860120c72908322ba068e3d0b8b4f8d
Summary by Sourcery
Implement run state consolidation in ConnectionActor by replacing queue-based state with optional receivers and a unified run-state counter, refactoring shutdown and polling logic accordingly and updating the roadmap documentation.
Bug Fixes:
Enhancements:
Documentation: