fix(object): always unregister deadlock-tracked get requests (#2275)

Signed-off-by: heihutu <heihutu@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: cxymds <Cxymds@qq.com>

rustfs/src/app/object_usecase.rs

@@ -121,6 +121,26 @@ use tokio_util::io::{ReaderStream, StreamReader};
 use tracing::{debug, error, info, instrument, warn};
 use uuid::Uuid;
 
+struct DeadlockRequestGuard {
+    deadlock_detector: Arc<crate::storage::deadlock_detector::DeadlockDetector>,
+    request_id: String,
+}
+
+impl DeadlockRequestGuard {
+    fn new(deadlock_detector: Arc<crate::storage::deadlock_detector::DeadlockDetector>, request_id: String) -> Self {
+        Self {
+            deadlock_detector,
+            request_id,
+        }
+    }
+}
+
+impl Drop for DeadlockRequestGuard {
+    fn drop(&mut self) {
+        self.deadlock_detector.unregister_request(&self.request_id);
+    }
+}
+
 pin_project! {
     struct ExtractArchiveEtagReader<R> {
         #[pin]
@@ -169,6 +189,31 @@ impl<R: AsyncRead> AsyncRead for ExtractArchiveEtagReader<R> {
     }
 }
 
+#[cfg(test)]
+mod deadlock_request_guard_tests {
+    use super::DeadlockRequestGuard;
+    use crate::storage::deadlock_detector::{DeadlockDetector, DeadlockDetectorConfig};
+    use std::sync::Arc;
+
+    #[test]
+    fn deadlock_request_guard_unregisters_on_drop() {
+        let detector = Arc::new(DeadlockDetector::new(DeadlockDetectorConfig {
+            enabled: true,
+            ..DeadlockDetectorConfig::default()
+        }));
+        let request_id = "test-request-id".to_string();
+
+        detector.register_request(&request_id, "test request");
+        assert_eq!(detector.tracked_count(), 1);
+
+        {
+            let _guard = DeadlockRequestGuard::new(Arc::clone(&detector), request_id.clone());
+            // `_guard` is dropped at the end of this scope, which should unregister the request.
+        }
+
+        assert_eq!(detector.tracked_count(), 0);
+    }
+}
 async fn maybe_enqueue_transition_immediate(obj_info: &ObjectInfo, src: LcEventSrc) {
     enqueue_transition_immediate(obj_info, src).await;
 }
@@ -1268,6 +1313,7 @@ impl DefaultObjectUsecase {
         let deadlock_detector = crate::storage::deadlock_detector::get_deadlock_detector();
         let request_id = wrapper.request_id().to_string();
         deadlock_detector.register_request(&request_id, format!("GetObject {}/{}", req.input.bucket, req.input.key));
+        let _deadlock_request_guard = DeadlockRequestGuard::new(deadlock_detector.clone(), request_id);
 
         // Check for request timeout before proceeding
         if wrapper.is_timeout() {
@@ -1278,7 +1324,6 @@ impl DefaultObjectUsecase {
                 elapsed_ms = wrapper.elapsed().as_millis(),
                 "GetObject request timed out before processing"
             );
-            deadlock_detector.unregister_request(&request_id);
             return Err(s3_error!(InternalError, "Request timeout before processing"));
         }
 
@@ -1470,7 +1515,6 @@ impl DefaultObjectUsecase {
                 elapsed_ms = wrapper.elapsed().as_millis(),
                 "GetObject request timed out while waiting for disk permit"
             );
-            deadlock_detector.unregister_request(&request_id);
             #[cfg(feature = "metrics")]
             metrics::counter!("rustfs.get.object.timeout.total", "stage" => "disk_permit").increment(1);
             return Err(s3_error!(InternalError, "Request timeout while waiting for disk permit"));
@@ -1576,7 +1620,6 @@ impl DefaultObjectUsecase {
                 elapsed_ms = wrapper.elapsed().as_millis(),
                 "GetObject request timed out before reading object"
             );
-            deadlock_detector.unregister_request(&request_id);
             #[cfg(feature = "metrics")]
             metrics::counter!("rustfs.get.object.timeout.total", "stage" => "before_read").increment(1);
             return Err(s3_error!(InternalError, "Request timeout before reading object"));
@@ -1964,9 +2007,6 @@ impl DefaultObjectUsecase {
             cache_key, response_content_length, total_duration, optimal_buffer_size
         );
 
-        // Unregister from deadlock detector
-        deadlock_detector.unregister_request(&request_id);
-
         let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
         let result = Ok(response);
         let _ = helper.complete(&result);