[ISSUE #5531]♻️Refactor controllers to use ArcMut for mutable access to RaftController #5532
Conversation
…to RaftController
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
WalkthroughThis PR refactors the RocketMQ controller architecture to support mutable state during lifecycle operations. It converts RaftController storage from Arc to ArcMut, updates startup/shutdown method signatures from immutable to mutable borrows, simplifies OpenRaftController's internal state management by removing unnecessary Arc<Mutex<>> wrappers, and propagates these changes through ControllerManager and all processor implementations. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25–30 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
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}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In @rocketmq-controller/src/controller/raft_controller.rs:
- Around line 62-67: The startup method should use the same ArcMut access
pattern used elsewhere: instead of calling controller.startup().await directly
on the &mut ArcMut<T> enum branches, call
controller.mut_from_ref().startup().await for both Self::OpenRaft(controller)
and Self::RaftRs(controller) so it matches the existing explicit mut_from_ref()
usage (referencing ArcMut, mut_from_ref, and the startup method on the inner
controller).
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
rocketmq-controller/src/controller.rsrocketmq-controller/src/controller/controller_manager.rsrocketmq-controller/src/controller/open_raft_controller.rsrocketmq-controller/src/controller/raft_controller.rsrocketmq-controller/src/controller/raft_rs_controller.rsrocketmq-controller/src/processor.rsrocketmq-controller/src/processor/broker_processor.rsrocketmq-controller/src/processor/topic_processor.rsrocketmq-controller/tests/raft_controller_test.rs
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-05-10T06:20:00.401Z
Learnt from: 578223592
Repo: mxsm/rocketmq-rust PR: 3240
File: rocketmq-namesrv/src/kvconfig/kvconfig_mananger.rs:69-71
Timestamp: 2025-05-10T06:20:00.401Z
Learning: In RocketMQ Rust, while dashmap is concurrent-safe, returning a direct reference to a DashMap can break encapsulation by allowing external code to modify the map without triggering persistence methods like `persist()`. Consider returning a read-only view, iterator methods, or a snapshot instead.
Applied to files:
rocketmq-controller/src/processor/broker_processor.rsrocketmq-controller/src/processor/topic_processor.rs
🧬 Code graph analysis (7)
rocketmq-controller/src/processor.rs (3)
rocketmq-controller/src/controller/controller_manager.rs (4)
new(139-211)config(575-577)raft(548-550)metadata(557-559)rocketmq-controller/src/processor/broker_processor.rs (4)
new(50-52)new(128-130)new(184-186)new(235-237)rocketmq-controller/src/processor/topic_processor.rs (3)
new(46-48)new(126-128)new(197-199)
rocketmq-controller/src/controller/raft_controller.rs (4)
rocketmq-controller/src/controller.rs (4)
startup(183-183)startup(523-525)shutdown(200-200)shutdown(527-529)rocketmq-controller/src/controller/open_raft_controller.rs (2)
startup(77-112)shutdown(114-147)rocketmq-controller/src/controller/controller_manager.rs (2)
controller(620-622)shutdown(457-510)rocketmq-controller/src/processor.rs (1)
shutdown(149-153)
rocketmq-controller/src/controller.rs (3)
rocketmq-controller/src/controller/open_raft_controller.rs (3)
startup(77-112)shutdown(114-147)new(66-73)rocketmq-controller/src/controller/raft_controller.rs (2)
startup(62-67)shutdown(69-74)rocketmq-controller/src/controller/raft_rs_controller.rs (3)
startup(47-50)shutdown(52-55)new(41-43)
rocketmq-controller/src/processor/broker_processor.rs (3)
rocketmq-controller/src/controller/controller_manager.rs (3)
new(139-211)metadata(557-559)raft(548-550)rocketmq-controller/src/processor.rs (2)
new(79-127)new(192-196)rocketmq-controller/src/processor/topic_processor.rs (3)
new(46-48)new(126-128)new(197-199)
rocketmq-controller/src/processor/topic_processor.rs (2)
rocketmq-controller/src/processor.rs (2)
new(79-127)new(192-196)rocketmq-controller/src/processor/broker_processor.rs (4)
new(50-52)new(128-130)new(184-186)new(235-237)
rocketmq-controller/src/controller/controller_manager.rs (3)
rocketmq-controller/src/controller/open_raft_controller.rs (1)
new(66-73)rocketmq-controller/src/processor.rs (2)
new(79-127)new(192-196)rocketmq-controller/src/controller/raft_controller.rs (1)
new_open_raft(51-53)
rocketmq-controller/src/controller/open_raft_controller.rs (1)
rocketmq-controller/src/controller.rs (4)
startup(183-183)startup(523-525)shutdown(200-200)shutdown(527-529)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (5)
- GitHub Check: Build & Test (ubuntu-latest)
- GitHub Check: Build & Test (macos-latest)
- GitHub Check: Build & Test (windows-latest)
- GitHub Check: Code Coverage
- GitHub Check: auto-approve
🔇 Additional comments (28)
rocketmq-controller/src/controller.rs (4)
183-183: LGTM: Trait signature updated to support stateful lifecycle operations.The change from
&selfto&mut selfis appropriate forstartup(), as implementations likeOpenRaftControllerneed to initialize and store mutable state (node, handle, shutdown_tx). This aligns with typical lifecycle patterns where startup involves state transitions.
200-200: LGTM: Shutdown signature correctly requires mutable access.Changing
shutdown()to take&mut selfis correct, as implementations need to take ownership of resources (viaOption::take()) to perform cleanup. This is consistent with the startup changes and standard Rust patterns for resource cleanup.
523-529: LGTM: MockController updated consistently.The MockController implementation correctly reflects the trait signature changes. The mock remains simple and appropriate for testing purposes.
611-614: LGTM: Test updated for new API.The test correctly uses a mutable binding to accommodate the updated trait methods. The test remains clear and validates the lifecycle operations properly.
rocketmq-controller/src/processor.rs (2)
68-68: LGTM: Field type updated to support mutable access.Changing from
Arc<RaftController>toArcMut<RaftController>enables processors to obtain mutable access to the controller when needed for lifecycle operations. This aligns with the broader refactoring objective.
79-127: LGTM: Constructor and processor instantiations updated consistently.The changes are applied uniformly across:
- Constructor parameter (line 79)
- All processor instantiations (lines 86, 90, 98, 110, 114, 118)
Each processor now receives
ArcMut<RaftController>viaraft.clone(), maintaining proper reference counting while enabling mutable access patterns.rocketmq-controller/tests/raft_controller_test.rs (3)
25-29: LGTM: Test updated for mutable lifecycle methods.The test correctly uses a mutable binding to accommodate the new
startup(&mut self)andshutdown(&mut self)signatures. The test logic remains unchanged and properly validates the controller lifecycle.
38-42: LGTM: RaftRs controller test updated consistently.The mutable binding is correctly applied, matching the pattern established in the OpenRaft test. Both test variants now properly handle the new trait requirements.
60-71: LGTM: Wrapper test updated for both variants.Both controller variants (OpenRaft and RaftRs) are instantiated with mutable bindings, ensuring the test validates the refactored API consistently across implementations.
rocketmq-controller/src/controller/raft_controller.rs (1)
69-74: Same dereferencing concern for shutdown.Similar to
startup(), verify that the direct method call onArcMutworks as intended. See comment on lines 62-67.rocketmq-controller/src/controller/raft_rs_controller.rs (1)
47-55: LGTM: Stub implementation updated consistently.The signature changes align with the
Controllertrait requirements. The stub implementations with TODO comments are appropriate for this stage of development.rocketmq-controller/src/processor/broker_processor.rs (4)
18-18: LGTM!The
ArcMutimport aligns with the PR objective to enable mutable access to theRaftControlleracross processors.
40-52: LGTM!The
RegisterBrokerProcessorcorrectly migrates toArcMut<RaftController>. Theis_leader()call works transparently throughArcMut'sDerefimplementation.
117-130: LGTM!The
UnregisterBrokerProcessorfollows the same consistent refactoring pattern.
224-237: LGTM!The
ElectMasterProcessorcompletes the consistent migration pattern across all broker processors requiring raft access.rocketmq-controller/src/processor/topic_processor.rs (4)
17-17: LGTM!The
ArcMutimport is correctly added for the topic processors.
35-48: LGTM!The
CreateTopicProcessorcorrectly migrates toArcMut<RaftController>, consistent with the broker processors.
115-128: LGTM!The
UpdateTopicProcessorfollows the same consistent refactoring pattern.
186-199: LGTM!The
DeleteTopicProcessorcompletes the consistent migration pattern across all topic processors.rocketmq-controller/src/controller/controller_manager.rs (5)
90-91: LGTM!The field type change to
ArcMut<RaftController>with updated documentation is consistent with the PR objective.
150-150: LGTM!The
ArcMut::new()initialization correctly wraps theRaftControllerfor shared mutable access.
376-383: LGTM!The
mut_from_ref()call is appropriate here sincestart()takesmut self: ArcMut<Self>, providing legitimate mutable access to start the Raft controller.
620-622: LGTM!The accessor correctly returns a reference to the
ArcMut<RaftController>, allowing callers to clone for shared access.
498-502: TheArcMut::mut_from_ref()pattern is safe in this context.While
ArcMutexplicitly requires callers to ensure no data races occur, theshutdown()method is protected by an atomic compare-exchange on therunningfield (line 459-462). This guarantees that only the first caller proceeds to invokemut_from_ref()on the raft_controller; subsequent callers return immediately. This serialization pattern eliminates concurrent access and satisfiesArcMut's safety requirements.rocketmq-controller/src/controller/open_raft_controller.rs (4)
55-63: Simplified state management by removingArc<Mutex<>>wrappers.The fields are now direct
Option<...>types instead ofArc<Mutex<Option<...>>>. This is cleaner and shifts synchronization responsibility to theArcMutwrapper at theControllerManagerlevel.
65-73: LGTM!The constructor is now cleaner with direct
Noneinitialization instead ofArc::new(Mutex::new(None))patterns.
77-112: LGTM!The
startup(&mut self)implementation is cleaner with direct field assignments. The mutable signature correctly reflects the semantic intent of modifying controller state during startup.
114-147: LGTM!The
shutdown(&mut self)implementation properly usesOption::take()for one-time resource consumption, maintaining the correct shutdown ordering: signal gRPC server → shutdown Raft node → await server task with timeout.
| async fn startup(&mut self) -> RocketMQResult<()> { | ||
| match self { | ||
| Self::OpenRaft(controller) => controller.startup().await, | ||
| Self::RaftRs(controller) => controller.startup().await, | ||
| } | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for ArcMut implementation to verify DerefMut trait and mut_from_ref usage patterns
echo "=== Searching for DerefMut implementation for ArcMut ==="
rg -n "impl.*DerefMut.*for.*ArcMut" -A 5
echo ""
echo "=== Searching for mut_from_ref function definition ==="
rg -n "fn mut_from_ref" -A 3
echo ""
echo "=== Searching for mut_from_ref usage in codebase ==="
rg -n "\.mut_from_ref\(\)" -C 2
echo ""
echo "=== Searching for ArcMut definition and trait implementations ==="
rg -n "struct ArcMut|impl.*for ArcMut" -A 3Repository: mxsm/rocketmq-rust
Length of output: 50374
Consider using mut_from_ref() for consistency with the same codebase pattern.
ArcMut correctly implements DerefMut, so the direct method call controller.startup().await on &mut ArcMut<T> works as intended. However, the same raft_controller field uses explicit mut_from_ref() calls elsewhere in controller_manager.rs (lines 376 and 498):
self.raft_controller.mut_from_ref().startup().await
self.raft_controller.mut_from_ref().shutdown().awaitAlign this code with the explicit pattern for consistency across the codebase.
🤖 Prompt for AI Agents
In @rocketmq-controller/src/controller/raft_controller.rs around lines 62 - 67,
The startup method should use the same ArcMut access pattern used elsewhere:
instead of calling controller.startup().await directly on the &mut ArcMut<T>
enum branches, call controller.mut_from_ref().startup().await for both
Self::OpenRaft(controller) and Self::RaftRs(controller) so it matches the
existing explicit mut_from_ref() usage (referencing ArcMut, mut_from_ref, and
the startup method on the inner controller).
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #5532 +/- ##
==========================================
- Coverage 38.44% 38.43% -0.01%
==========================================
Files 815 815
Lines 110567 110551 -16
==========================================
- Hits 42508 42492 -16
Misses 68059 68059 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
There was a problem hiding this comment.
LGTM - All CI checks passed ✅
rocketmq-rust-bot
added
approved
Which Issue(s) This PR Fixes(Closes)
Fixes #5531
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.