[mxsm]

Which Issue(s) This PR Fixes(Closes)

• Fixes [Enhancement✨] enhance(message-store): enhance master store handling with thread-safe access and add send message back hook support #7073

Brief Description

How Did You Test This Change?

Summary by CodeRabbit

• Refactor
  • Enhanced thread-safety for message store configuration components, enabling safe cross-thread operation.
  • Implemented runtime configuration for message routing and message handling hooks with proper concurrency controls.

[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💥.

[coderabbitai]

Walkthrough

The PR enhances message store thread-safety by adding Send + Sync + 'static bounds to the MessageStoreInner trait and SendMessageBackHook trait. The LocalFileMessageStore now implements runtime-configurable master store and message back hook handling using RwLock-protected type-erased storage with downcasting support.

Changes

Cohort / File(s)	Summary
Trait Bounds Enhancement rocketmq-store/src/base/message_store.rs, rocketmq-store/src/hook/send_message_back_hook.rs	Added Send + Sync + 'static trait bounds to MessageStoreInner generic methods and required Send + Sync on SendMessageBackHook trait for safe cross-thread usage.
Implementation with Type-Safe Storage rocketmq-store/src/message_store/local_file_message_store.rs	Replaced todo!() with functional implementations using RwLock-protected storage. get/set_master_store_in_process use type-erased Arc<dyn Any + Send + Sync> with downcasting; get/set_send_message_back_hook directly store and retrieve the hook. Includes roundtrip tests verifying pointer equality and hook storage.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Poem

🐰 Locks and types now dance in sync,
Thread-safe stores won't miss a blink,
Hooks are caught, the masters bound,
Safe and sound, the logic's sound! ✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 28.57% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately captures the two main changes: enhancing master store with thread-safe access and adding send message back hook support, matching the actual code modifications.
Linked Issues check	✅ Passed	The PR implements both stated objectives: adds thread-safe bounds (Send + Sync + 'static) to master store accessors and implements send message back hook support with proper locking.
Out of Scope Changes check	✅ Passed	All changes are directly related to the stated objectives; the PR focuses solely on thread-safety enhancements and hook support without introducing unrelated modifications.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat-7073

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (2)

rocketmq-store/src/message_store/local_file_message_store.rs (2)

1991-2006: Store the master store directly instead of wrapping Arc<M> in another Arc.

Arc<M> already coerces to Arc<dyn Any + Send + Sync>. The extra Arc::new(...) adds another heap object and makes the downcast path harder to follow.

♻️ Suggested simplification

 fn get_master_store_in_process<M: MessageStore + Send + Sync + 'static>(&self) -> Option<Arc<M>> {
-    let guard = self
-        .master_store_in_process
-        .read()
-        .expect("master_store_in_process lock poisoned");
-    let erased = guard.as_ref()?.clone();
-    let boxed_master_store = erased.downcast::<Arc<M>>().ok()?;
-    Some(Arc::clone(boxed_master_store.as_ref()))
+    self.master_store_in_process
+        .read()
+        .expect("master_store_in_process lock poisoned")
+        .as_ref()
+        .cloned()?
+        .downcast::<M>()
+        .ok()
 }
@@
-    *guard = Some(Arc::new(master_store_in_process) as Arc<dyn Any + Send + Sync>);
+    *guard = Some(master_store_in_process as Arc<dyn Any + Send + Sync>);

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rocketmq-store/src/message_store/local_file_message_store.rs` around lines
1991 - 2006, The current set_master_store_in_process wraps an Arc<M> inside
another Arc (Arc::new(master_store_in_process)), which creates an extra heap
allocation and complicates downcasting in get_master_store_in_process; change
set_master_store_in_process to store the Arc<M> directly into
master_store_in_process by coercing it to Arc<dyn Any + Send + Sync> (i.e.,
assign Some(master_store_in_process as Arc<dyn Any + Send + Sync>)) so that
get_master_store_in_process can continue to downcast::<Arc<M>>() and return
Arc::clone(...) without the extra indirection; update the assignment in
set_master_store_in_process and keep the read/downcast logic in
get_master_store_in_process unchanged.

---

3337-3392: Please add a behavior-level test for the two new guarantees.

These round-trip assertions only show that the slots retain an Arc. They still miss the behaviors this PR is about: concurrent access to master_store_in_process and an actually invocable send-message-back hook.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@rocketmq-store/src/message_store/local_file_message_store.rs` around lines
3337 - 3392, Add behavior-level tests that exercise the semantics beyond Arc
pointer equality: for master_store_in_process, create multiple threads/tasks
that concurrently call set_master_store_in_process and
get_master_store_in_process::<LocalFileMessageStore>() to confirm no races and
that readers always see a valid Arc (e.g., repeatedly read and call a method on
the restored LocalFileMessageStore or compare internal state); for the
send-message-back hook, register a MockSendBackHook (impl SendMessageBackHook)
and invoke the actual code path that calls execute_send_message_back (or
directly call the method via the restored Arc from get_send_message_back_hook())
to assert the hook runs and returns the expected boolean, and include a
concurrent invocation test to ensure thread-safety for
get_send_message_back_hook/set_send_message_back_hook. Ensure tests reference
set_master_store_in_process,
get_master_store_in_process::<LocalFileMessageStore>,
set_send_message_back_hook, get_send_message_back_hook, and
SendMessageBackHook::execute_send_message_back so reviewers can locate the
changes.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@rocketmq-store/src/message_store/local_file_message_store.rs`:
- Around line 1991-2006: The current set_master_store_in_process wraps an Arc<M>
inside another Arc (Arc::new(master_store_in_process)), which creates an extra
heap allocation and complicates downcasting in get_master_store_in_process;
change set_master_store_in_process to store the Arc<M> directly into
master_store_in_process by coercing it to Arc<dyn Any + Send + Sync> (i.e.,
assign Some(master_store_in_process as Arc<dyn Any + Send + Sync>)) so that
get_master_store_in_process can continue to downcast::<Arc<M>>() and return
Arc::clone(...) without the extra indirection; update the assignment in
set_master_store_in_process and keep the read/downcast logic in
get_master_store_in_process unchanged.
- Around line 3337-3392: Add behavior-level tests that exercise the semantics
beyond Arc pointer equality: for master_store_in_process, create multiple
threads/tasks that concurrently call set_master_store_in_process and
get_master_store_in_process::<LocalFileMessageStore>() to confirm no races and
that readers always see a valid Arc (e.g., repeatedly read and call a method on
the restored LocalFileMessageStore or compare internal state); for the
send-message-back hook, register a MockSendBackHook (impl SendMessageBackHook)
and invoke the actual code path that calls execute_send_message_back (or
directly call the method via the restored Arc from get_send_message_back_hook())
to assert the hook runs and returns the expected boolean, and include a
concurrent invocation test to ensure thread-safety for
get_send_message_back_hook/set_send_message_back_hook. Ensure tests reference
set_master_store_in_process,
get_master_store_in_process::<LocalFileMessageStore>,
set_send_message_back_hook, get_send_message_back_hook, and
SendMessageBackHook::execute_send_message_back so reviewers can locate the
changes.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: ba5f71e3-5a85-46a3-9425-521eb4ea3b5f

📥 Commits

Reviewing files that changed from the base of the PR and between cbda46a and 88cd3fe.

📒 Files selected for processing (3)

• rocketmq-store/src/base/message_store.rs
• rocketmq-store/src/hook/send_message_back_hook.rs
• rocketmq-store/src/message_store/local_file_message_store.rs

[codecov]

Codecov Report

❌ Patch coverage is 87.69231% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.16%. Comparing base (cbda46a) to head (88cd3fe).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...tore/src/message_store/local_file_message_store.rs	87.69%	8 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #7074      +/-   ##
==========================================
+ Coverage   58.15%   58.16%   +0.01%     
==========================================
  Files        1080     1080              
  Lines      182594   182657      +63     
==========================================
+ Hits       106186   106246      +60     
- Misses      76408    76411       +3     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.