Merge branch 'main' into codex/fix-5166-metacache-timeout

* feat(hotpath): use mimalloc in counting allocator

* fix(hotpath): adapt mimalloc for allocation counting

---------

Co-authored-by: overtrue <anzhengchao@gmail.com>

crates/e2e_test/src/object_lock/object_lock_test.rs

@@ -26,6 +26,7 @@
 
 use super::common::*;
 use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
 use aws_sdk_s3::types::{
     CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
@@ -2120,6 +2121,127 @@ async fn test_multipart_default_retention_fixed_at_create() {
 // Versioning Auto-Enable Tests
 // ============================================================================
 
+#[tokio::test]
+#[serial]
+async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
+    init_logging();
+    info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
+
+    let mut env = ObjectLockTestEnvironment::new()
+        .await
+        .expect("failed to create Object Lock test environment");
+    env.start_rustfs().await.expect("failed to start RustFS");
+
+    let bucket = "test-object-lock-delete-cleanup";
+    let key = "unretained-object";
+
+    env.create_object_lock_bucket(bucket)
+        .await
+        .expect("failed to create Object Lock bucket");
+    let client = env.s3_client();
+
+    let put_response = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"unretained data"))
+        .send()
+        .await
+        .expect("failed to upload unretained object");
+    let object_version_id = put_response
+        .version_id()
+        .expect("Object Lock buckets must create versioned objects")
+        .to_string();
+
+    let delete_response = client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("failed to create delete marker");
+    assert_eq!(delete_response.delete_marker(), Some(true));
+    let delete_marker_version_id = delete_response
+        .version_id()
+        .expect("Deleting without a version ID must create a delete marker")
+        .to_string();
+
+    let get_error = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect_err("GET must not return an object hidden by a delete marker");
+    assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
+    assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
+
+    let listed_objects = client
+        .list_objects_v2()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list current objects");
+    assert!(
+        listed_objects.contents().iter().all(|object| object.key() != Some(key)),
+        "ListObjectsV2 must hide objects whose latest version is a delete marker"
+    );
+
+    let listed_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list object versions");
+    assert!(
+        listed_versions
+            .versions()
+            .iter()
+            .any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
+        "The data version must remain until it is explicitly deleted"
+    );
+    assert!(
+        listed_versions
+            .delete_markers()
+            .iter()
+            .any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
+        "ListObjectVersions must expose the delete marker"
+    );
+
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(object_version_id)
+        .send()
+        .await
+        .expect("failed to delete the data version");
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(delete_marker_version_id)
+        .send()
+        .await
+        .expect("failed to delete the delete marker");
+
+    let remaining_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list versions after cleanup");
+    assert!(remaining_versions.versions().is_empty());
+    assert!(remaining_versions.delete_markers().is_empty());
+
+    client
+        .delete_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
+}
+
 #[tokio::test]
 #[serial]
 async fn test_versioning_auto_enabled_with_object_lock() {

crates/e2e_test/src/reliant/tiering.rs

@@ -382,7 +382,6 @@ struct ManualTransitionRunReport {
     skipped_delete_marker: u64,
     skipped_directory: u64,
     skipped_replication: u64,
-    skipped_already_transitioned: u64,
     skipped_already_in_flight: u64,
     skipped_queue_full: u64,
     skipped_queue_closed: u64,
@@ -408,41 +407,6 @@ fn assert_completed_or_in_flight_partial(state: &str, report: &ManualTransitionR
     }
 }
 
-fn assert_conflict_winner_report(state: &str, report: &ManualTransitionRunReport, expected_objects: u64, context: &str) {
-    assert_completed_or_in_flight_partial(state, report, context);
-    if report.skipped_already_in_flight > 0 {
-        assert!(
-            report.scanned <= expected_objects,
-            "{context}: scanned more objects than the conflict scope contains: {report:#?}"
-        );
-        assert!(
-            report.eligible <= expected_objects,
-            "{context}: marked more objects eligible than the conflict scope contains: {report:#?}"
-        );
-        assert_eq!(
-            report.enqueued + report.skipped_already_in_flight,
-            report.eligible,
-            "{context}: partial in-flight accounting must cover every eligible object: {report:#?}"
-        );
-    } else {
-        assert_eq!(report.scanned, expected_objects, "{context}: {report:#?}");
-        assert_eq!(
-            report.eligible + report.skipped_already_transitioned,
-            expected_objects,
-            "{context}: {report:#?}"
-        );
-        assert_eq!(
-            report.enqueued + report.skipped_already_in_flight,
-            expected_objects,
-            "{context}: {report:#?}"
-        );
-    }
-    assert_eq!(
-        report.transition_completed, report.enqueued,
-        "{context}: winner must wait for all queued transitions: {report:#?}"
-    );
-}
-
 #[derive(Debug, Deserialize)]
 struct ManualTransitionQueueSnapshot {
     queue_capacity: u64,
@@ -1276,8 +1240,15 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes
     cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
 
     let mut hot = RustFSTestEnvironment::new().await?;
-    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
-        .await?;
+    hot.start_rustfs_server_with_env(
+        vec![],
+        &[
+            ("RUSTFS_SCANNER_ENABLED", "false"),
+            ("RUSTFS_SCANNER_CYCLE", "3600"),
+            (MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"),
+        ],
+    )
+    .await?;
     let hot_client = hot.create_s3_client();
     add_rustfs_tier(&hot, &cold).await?;
 
@@ -1347,20 +1318,21 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes
     assert_eq!(conflict.cancel_endpoint, status_endpoint);
     assert!(!conflict.scope_key.is_empty());
 
+    manual_transition_job_cancel(&hot, cancel_endpoint).await?;
+
     let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT).await?;
     assert_eq!(terminal.job_id, job_id);
+    assert_eq!(terminal.status, "cancelled", "terminal conflict winner response: {terminal:#?}");
     assert!(!terminal.report.dry_run);
     assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET);
     assert_eq!(terminal.report.prefix, accepted.report.prefix);
-    assert_conflict_winner_report(
-        &terminal.status,
-        &terminal.report,
-        MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
-        "terminal conflict winner response",
+    assert!(terminal.report.cancelled, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(terminal.report.scanned, 0, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(terminal.report.enqueued, 0, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(
+        terminal.report.transition_completed, 0,
+        "terminal conflict winner response: {terminal:#?}"
     );
-    assert_eq!(terminal.report.dry_run_eligible, 0, "terminal conflict winner response: {terminal:#?}");
-    assert_eq!(terminal.report.transition_failed, 0, "terminal conflict winner response: {terminal:#?}");
-    assert_eq!(terminal.report.tier_failure, 0, "terminal conflict winner response: {terminal:#?}");
     let after_remote_count = cold_tier_object_count(&cold_client).await?;
     assert!(after_remote_count >= before_remote_count);
     assert!(after_remote_count <= before_remote_count + MANUAL_ASYNC_CONFLICT_OBJECTS);

crates/scanner/src/scanner_folder.rs

@@ -75,7 +75,7 @@ const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500;
 const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000;
 const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4;
 const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
-const SCANNER_LIST_PATH_RAW_TIMEOUT: Duration = Duration::from_secs(60);
+const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60);
 const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32;
 const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
 const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
@@ -100,6 +100,21 @@ static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
 static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new();
 static SCANNER_ALERT_METRICS_ONCE: Once = Once::new();
 
+#[cfg(test)]
+type ListPathRawTimeoutSnapshot = (bool, Option<Duration>, Option<Duration>);
+
+fn scanner_abandoned_child_list_options() -> ListPathRawOptions {
+    // A complete heal walk scales with bucket size and may legitimately take
+    // longer than a fixed wall-clock budget. Keep the total duration unbounded;
+    // Retain the scanner's per-read stall budget and keep cancellation controlled
+    // by the scanner cycle token.
+    ListPathRawOptions {
+        skip_walkdir_total_timeout: true,
+        walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT),
+        ..Default::default()
+    }
+}
+
 pub fn data_usage_update_dir_cycles() -> u32 {
     rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES)
 }
@@ -1281,6 +1296,8 @@ pub struct FolderScanner {
     skip_heal: Arc<std::sync::atomic::AtomicBool>,
     local_disk: Arc<Disk>,
     pending_heals_changed: bool,
+    #[cfg(test)]
+    list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
 }
 
 impl FolderScanner {
@@ -2410,75 +2427,79 @@ impl FolderScanner {
                 let bucket_clone = bucket.clone();
                 let prefix_clone = prefix.clone();
                 let child_ctx_clone = child_ctx.clone();
+                #[cfg(test)]
+                let list_path_raw_options_observer = self.list_path_raw_options_observer.clone();
 
                 tokio::spawn(async move {
-                    if let Err(e) = list_path_raw(
-                        child_ctx_clone.clone(),
-                        ListPathRawOptions {
-                            disks,
-                            bucket: bucket_clone.clone(),
-                            path: prefix_clone.clone(),
-                            recursive: true,
-                            report_not_found: true,
-                            min_disks: disks_quorum,
-                            walkdir_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
-                            walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
-                            agreed: Some(Box::new(move |entry: MetaCacheEntry| {
-                                let entry_name = entry.name.clone();
-                                let agreed_tx = agreed_tx.clone();
-                                Box::pin(async move {
-                                    if let Err(e) = agreed_tx.send(entry_name).await {
-                                        error!(
-                                            target: "rustfs::scanner::folder",
-                                            event = EVENT_SCANNER_FOLDER_STATE,
-                                            component = LOG_COMPONENT_SCANNER,
-                                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                                            entry = %entry.name,
-                                            state = "list_path_agreed_send_failed",
-                                            error = %e,
-                                            "Scanner list_path_raw agreed callback failed"
-                                        );
-                                    }
-                                })
-                            })),
-                            partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
-                                let partial_tx = partial_tx.clone();
-                                Box::pin(async move {
-                                    if let Err(e) = partial_tx.send(entries).await {
-                                        error!(
-                                            target: "rustfs::scanner::folder",
-                                            event = EVENT_SCANNER_FOLDER_STATE,
-                                            component = LOG_COMPONENT_SCANNER,
-                                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                                            state = "list_path_partial_send_failed",
-                                            error = %e,
-                                            "Scanner list_path_raw partial callback failed"
-                                        );
-                                    }
-                                })
-                            })),
-                            finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
-                                let finished_tx = finished_tx.clone();
-                                let errs_clone = errs.to_vec();
-                                Box::pin(async move {
-                                    if let Err(e) = finished_tx.send(errs_clone).await {
-                                        error!(
-                                            target: "rustfs::scanner::folder",
-                                            event = EVENT_SCANNER_FOLDER_STATE,
-                                            component = LOG_COMPONENT_SCANNER,
-                                            subsystem = LOG_SUBSYSTEM_FOLDER,
-                                            state = "list_path_finished_send_failed",
-                                            error = %e,
-                                            "Scanner list_path_raw finished callback failed"
-                                        );
-                                    }
-                                })
-                            })),
-                            ..Default::default()
-                        },
-                    )
-                    .await
-                    {
+                    let options = ListPathRawOptions {
+                        disks,
+                        bucket: bucket_clone.clone(),
+                        path: prefix_clone.clone(),
+                        recursive: true,
+                        report_not_found: true,
+                        min_disks: disks_quorum,
+                        agreed: Some(Box::new(move |entry: MetaCacheEntry| {
+                            let entry_name = entry.name.clone();
+                            let agreed_tx = agreed_tx.clone();
+                            Box::pin(async move {
+                                if let Err(e) = agreed_tx.send(entry_name).await {
+                                    error!(
+                                        target: "rustfs::scanner::folder",
+                                        event = EVENT_SCANNER_FOLDER_STATE,
+                                        component = LOG_COMPONENT_SCANNER,
+                                        subsystem = LOG_SUBSYSTEM_FOLDER,
+                                        entry = %entry.name,
+                                        state = "list_path_agreed_send_failed",
+                                        error = %e,
+                                        "Scanner list_path_raw agreed callback failed"
+                                    );
+                                }
+                            })
+                        })),
+                        partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
+                            let partial_tx = partial_tx.clone();
+                            Box::pin(async move {
+                                if let Err(e) = partial_tx.send(entries).await {
+                                    error!(
+                                        target: "rustfs::scanner::folder",
+                                        event = EVENT_SCANNER_FOLDER_STATE,
+                                        component = LOG_COMPONENT_SCANNER,
+                                        subsystem = LOG_SUBSYSTEM_FOLDER,
+                                        state = "list_path_partial_send_failed",
+                                        error = %e,
+                                        "Scanner list_path_raw partial callback failed"
+                                    );
+                                }
+                            })
+                        })),
+                        finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
+                            let finished_tx = finished_tx.clone();
+                            let errs_clone = errs.to_vec();
+                            Box::pin(async move {
+                                if let Err(e) = finished_tx.send(errs_clone).await {
+                                    error!(
+                                        target: "rustfs::scanner::folder",
+                                        event = EVENT_SCANNER_FOLDER_STATE,
+                                        component = LOG_COMPONENT_SCANNER,
+                                        subsystem = LOG_SUBSYSTEM_FOLDER,
+                                        state = "list_path_finished_send_failed",
+                                        error = %e,
+                                        "Scanner list_path_raw finished callback failed"
+                                    );
+                                }
+                            })
+                        })),
+                        ..scanner_abandoned_child_list_options()
+                    };
+                    #[cfg(test)]
+                    if let Some(observer) = list_path_raw_options_observer {
+                        let _ = observer.send((
+                            options.skip_walkdir_total_timeout,
+                            options.walkdir_timeout,
+                            options.walkdir_stall_timeout,
+                        ));
+                    }
+                    if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await {
                         if is_missing_path_disk_error(&e) {
                             debug!(
                                 target: "rustfs::scanner::folder",
@@ -2820,6 +2841,8 @@ pub async fn scan_data_folder(
         skip_heal,
         local_disk,
         pending_heals_changed: false,
+        #[cfg(test)]
+        list_path_raw_options_observer: None,
     };
 
     // Check if context is cancelled
@@ -3026,6 +3049,7 @@ mod tests {
             skip_heal: Arc::new(AtomicBool::new(false)),
             local_disk: disk,
             pending_heals_changed: false,
+            list_path_raw_options_observer: None,
         };
 
         (scanner, temp_dir)
@@ -4210,6 +4234,8 @@ mod tests {
         scanner.heal_object_select = 1;
         scanner.disks = disks;
         scanner.disks_quorum = 2;
+        let (options_tx, mut options_rx) = mpsc::unbounded_channel();
+        scanner.list_path_raw_options_observer = Some(options_tx);
         scanner.old_cache.replace(
             &format!("{bucket}/{object}"),
             bucket,
@@ -4231,6 +4257,11 @@ mod tests {
             .expect("scan_folder should not hang after list_path_raw finishes")
             .expect("scan_folder should finish successfully");
 
+        let observed_options = tokio::time::timeout(Duration::from_secs(1), options_rx.recv())
+            .await
+            .expect("abandoned-child listing options should be observed promptly")
+            .expect("abandoned-child listing options channel should remain open");
+        assert_eq!(observed_options, (true, None, Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT)));
         let root = scanner
             .new_cache
             .checked_flatten(bucket)

rustfs/src/main.rs

@@ -12,9 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+use std::alloc::{GlobalAlloc, Layout};
+
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+#[derive(Default)]
+struct DefaultMiMalloc;
+
+#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
+// SAFETY: allocation and deallocation are forwarded unchanged to MiMalloc, so
+// MiMalloc's GlobalAlloc guarantees apply to every returned pointer and layout.
+#[allow(unsafe_code)]
+unsafe impl GlobalAlloc for DefaultMiMalloc {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        // SAFETY: the caller upholds GlobalAlloc's contract for layout.
+        unsafe { mimalloc::MiMalloc.alloc(layout) }
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
+        unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
+    }
+}
+
 #[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
 #[global_allocator]
-static GLOBAL: hotpath::CountingAllocator = hotpath::CountingAllocator::new();
+static GLOBAL: hotpath::CountingAllocator<DefaultMiMalloc> = hotpath::CountingAllocator::new();
 
 #[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
 #[global_allocator]
@@ -25,3 +48,15 @@ fn main() {
 
     rustfs::startup_entrypoint::run_process();
 }
+
+#[cfg(all(test, feature = "hotpath", feature = "hotpath-alloc"))]
+mod tests {
+    #[test]
+    #[allow(unsafe_code)]
+    fn hotpath_allocator_uses_mimalloc() {
+        let allocation = Box::new([0_u8; 64]);
+
+        // SAFETY: the live Box pointer is valid to inspect for heap ownership.
+        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
+    }
+}