rocketmq-broker/src/processor/pop_message_processor.rs (4)

150-152: Consider renaming expired_local_cache to lock_cache for clarity

The field expired_local_cache in QueueLockManager may not clearly represent its purpose. Since it stores all active locks (not just expired ones), consider renaming it to lock_cache or lock_map to improve code readability.

---

161-174: Optimize build_lock_key to return CheetahString directly

Currently, build_lock_key returns a String, which is then converted to a CheetahString in the try_lock and unlock methods. To avoid unnecessary conversions and improve performance, consider modifying build_lock_key to return a CheetahString directly.

Apply this diff to modify the function:

 pub fn build_lock_key(
     topic: &CheetahString,
     consumer_group: &CheetahString,
     queue_id: i32,
 ) -> CheetahString {
-    format!(
+    CheetahString::from_string(format!(
         "{}{}{}{}{}",
         topic,
         PopAckConstants::SPLIT,
         consumer_group,
         PopAckConstants::SPLIT,
         queue_id
-    )
+    ))
 }

---

224-225: Update log output to reflect actual lock count after cleanup

After modifying clean_unused_locks to return the number of remaining locks, adjust the log statement to accurately report the current lock cache size.

Apply this diff to update the log:

 let count = self.clean_unused_locks(60000).await;
-info!("QueueLockSize={}", count);
+info!("QueueLockSize after cleanup={}", count);

---

150-229: Add documentation comments to public methods

Adding documentation comments to the public methods of QueueLockManager will enhance code readability and maintainability, helping other developers understand their purpose and usage.

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

rocketmq-broker/src/processor/pop_message_processor.rs (2)

220-228: Make cleanup interval and expiry time configurable

The background task uses hardcoded values for:

• Cleanup interval: 60 seconds
• Lock expiry time: 60000 milliseconds

Consider making these values configurable through the QueueLockManager constructor or environment variables.

pub struct QueueLockManager {
    expired_local_cache: Arc<Mutex<HashMap<CheetahString, TimedLock>>>,
+   cleanup_interval_secs: u64,
+   lock_expiry_millis: u64,
}

impl QueueLockManager {
-   pub fn new() -> Self {
+   pub fn new(cleanup_interval_secs: Option<u64>, lock_expiry_millis: Option<u64>) -> Self {
        QueueLockManager {
            expired_local_cache: Arc::new(Mutex::new(HashMap::with_capacity(4096))),
+           cleanup_interval_secs: cleanup_interval_secs.unwrap_or(60),
+           lock_expiry_millis: lock_expiry_millis.unwrap_or(60000),
        }
    }

    pub fn start(self: Arc<Self>) {
        tokio::spawn(async move {
            loop {
-               tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
-               let count = self.clean_unused_locks(60000).await;
+               tokio::time::sleep(tokio::time::Duration::from_secs(self.cleanup_interval_secs)).await;
+               let count = self.clean_unused_locks(self.lock_expiry_millis).await;
                info!("QueueLockSize={}", count);
            }
        });
    }
}

---

331-390: Improve test coverage and naming conventions

1. Test naming:

   • Remove inconsistent "1" suffixes (e.g., try_lock_locks_successfully1).
   • Use descriptive names that indicate the scenario being tested.

2. Missing test cases:

   • Concurrent access scenarios
   • Edge cases for expired locks
   • Cleanup task behavior

Example of a concurrent test:

#[tokio::test]
async fn concurrent_lock_attempts_are_thread_safe() {
    let manager = Arc::new(QueueLockManager::new());
    let topic = CheetahString::from_static_str("test_topic");
    let consumer_group = CheetahString::from_static_str("test_group");
    let queue_id = 1;
    
    let mut handles = vec![];
    for _ in 0..10 {
        let manager = manager.clone();
        let topic = topic.clone();
        let consumer_group = consumer_group.clone();
        handles.push(tokio::spawn(async move {
            manager.try_lock(&topic, &consumer_group, queue_id).await
        }));
    }
    
    let results = futures::future::join_all(handles).await;
    let successful_locks = results.into_iter().filter(|r| r.unwrap()).count();
    assert_eq!(successful_locks, 1);
}

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

pub fn start(self: Arc<Self>) {
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
-           let count = self.clean_unused_locks(60000).await;
-           info!("QueueLockSize={}", count);
+           match self.clean_unused_locks(60000).await {
+               Ok(count) => info!("QueueLockSize={}", count),
+               Err(e) => error!("Failed to clean unused locks: {}", e),
+           }
        }
    });
}

pub async fn try_lock_with_key(&self, key: CheetahString) -> bool {
-    let mut cache = self.expired_local_cache.lock().await;
-    let lock = cache.entry(key).or_insert(TimedLock::new());
-    lock.try_lock()
+    let lock = {
+        let mut cache = self.expired_local_cache.lock().await;
+        cache.entry(key).or_insert(TimedLock::new()).clone()
+    };
+    lock.try_lock()
}

rocketmq-broker/src/processor/pop_message_processor.rs (4)

156-158: Consider making initial capacity configurable

The fixed capacity of 4096 might not be suitable for all use cases. Consider making it configurable through a constructor parameter.

-    pub fn new() -> Self {
+    pub fn new(initial_capacity: usize) -> Self {
         QueueLockManager {
-            expired_local_cache: Arc::new(Mutex::new(HashMap::with_capacity(4096))),
+            expired_local_cache: Arc::new(Mutex::new(HashMap::with_capacity(initial_capacity))),
         }
     }

---

161-174: Optimize string key generation

The current implementation creates intermediate String allocations. Consider using a pre-allocated buffer or string builder for better performance.

     pub fn build_lock_key(
         topic: &CheetahString,
         consumer_group: &CheetahString,
         queue_id: i32,
-    ) -> String {
-        format!(
-            "{}{}{}{}{}",
-            topic,
-            PopAckConstants::SPLIT,
-            consumer_group,
-            PopAckConstants::SPLIT,
-            queue_id
-        )
+    ) -> CheetahString {
+        let mut buffer = String::with_capacity(
+            topic.len() + consumer_group.len() + 10
+        );
+        buffer.push_str(topic.as_str());
+        buffer.push_str(PopAckConstants::SPLIT);
+        buffer.push_str(consumer_group.as_str());
+        buffer.push_str(PopAckConstants::SPLIT);
+        buffer.push_str(&queue_id.to_string());
+        CheetahString::from(buffer)
     }

---

217-225: Make cleanup interval configurable

The background task's interval is hardcoded to 60 seconds. Consider making it configurable to allow for different cleanup strategies.

-    pub fn start(self: Arc<Self>) {
+    pub fn start(self: Arc<Self>, cleanup_interval_secs: u64) {
         tokio::spawn(async move {
             loop {
-                tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
+                tokio::time::sleep(tokio::time::Duration::from_secs(cleanup_interval_secs)).await;
                 let count = self.clean_unused_locks(60000).await;
                 info!("QueueLockSize={}", count);
             }
         });
     }

---

345-362: Improve test naming consistency and coverage

1. Test names have inconsistent suffixes (e.g., try_lock_locks_successfully1).
2. Missing tests for concurrent access scenarios.

Consider:

1. Removing numeric suffixes from test names
2. Adding concurrent access tests:

#[tokio::test]
async fn try_lock_handles_concurrent_access() {
    let manager = Arc::new(QueueLockManager::new());
    let topic = CheetahString::from_static_str("test_topic");
    let consumer_group = CheetahString::from_static_str("test_group");
    let queue_id = 1;
    
    let mut handles = vec![];
    for _ in 0..10 {
        let manager = manager.clone();
        let topic = topic.clone();
        let consumer_group = consumer_group.clone();
        handles.push(tokio::spawn(async move {
            manager.try_lock(&topic, &consumer_group, queue_id).await
        }));
    }
    
    let results = futures::future::join_all(handles).await;
    assert_eq!(results.into_iter().filter(|r| r.unwrap()).count(), 1);
}

Configuration used: .coderabbit.yaml
Review profile: CHILL
Plan: Pro

     pub async fn clean_unused_locks(&self, used_expire_millis: u64) -> usize {
         let mut cache = self.expired_local_cache.lock().await;
-        let count = cache.len();
         cache.retain(|_, lock| get_current_millis() - lock.get_lock_time() <= used_expire_millis);
-        count
+        cache.len()
     }

#!/bin/bash
# Search for all lock acquisitions and releases
echo "Searching for lock acquisition patterns..."
rg -A 5 "try_lock|unlock" --type rust

# Search for potential nested locks
echo "Checking for potential nested locks..."
ast-grep --pattern 'try_lock($_) {
  $$$
  try_lock($_)
  $$$
}'