test(e2e): activate conditional write regressions

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
-sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
+sha256-darwin=03bdfb6a9d6e25d744c385f1461e651f05e4da78e5b0ead2adb3e8b2463e3834
+sha256-linux=78c46adad135231017fb8679fb91877b5ae3ec6ce1f75d3d045d2c1076c12a49

CLAUDE.md

@@ -30,8 +30,7 @@ make build-docker BUILD_OS=ubuntu22.04
 - Crate membership: `Cargo.toml` `[workspace].members`
 - Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
 - Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
-- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
-  matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
+- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
 - Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
   policy: [docs/testing/README.md](docs/testing/README.md)
 - Tier/ILM transition debugging (xl.meta inspection, versionId tracing):

CONTRIBUTING.md

@@ -70,8 +70,6 @@ make pre-pr
 
 > For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
 
-> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
-
 ### 🔒 Automated Pre-commit Hooks
 #### What `make pre-commit` and `make pre-pr` actually run
 

Cargo.lock

@@ -1858,9 +1858,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.4.4"
+version = "1.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
+checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -2522,6 +2522,12 @@ dependencies = [
  "subtle",
 ]
 
+[[package]]
+name = "cty"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
+
 [[package]]
 name = "curve25519-dalek"
 version = "4.1.3"
@@ -5982,6 +5988,15 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
 
+[[package]]
+name = "libmimalloc-sys"
+version = "0.1.49"
+source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
+dependencies = [
+ "cc",
+ "cty",
+]
+
 [[package]]
 name = "libredox"
 version = "0.1.20"
@@ -6382,6 +6397,14 @@ dependencies = [
  "synstructure 0.13.2",
 ]
 
+[[package]]
+name = "mimalloc"
+version = "0.1.52"
+source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
+dependencies = [
+ "libmimalloc-sys",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -9139,11 +9162,13 @@ dependencies = [
  "insta",
  "jiff",
  "libc",
+ "libmimalloc-sys",
  "libsystemd",
  "matchit 0.9.2",
  "md-5 0.11.0",
  "metrics",
  "metrics-util",
+ "mimalloc",
  "mime_guess",
  "opentelemetry",
  "opentelemetry_sdk",
@@ -9179,8 +9204,6 @@ dependencies = [
  "rustfs-lock",
  "rustfs-log-analyzer",
  "rustfs-madmin",
- "rustfs-mimalloc",
- "rustfs-mimalloc-sys",
  "rustfs-notify",
  "rustfs-object-capacity",
  "rustfs-object-data-cache",
@@ -9852,24 +9875,6 @@ dependencies = [
  "tokio",
 ]
 
-[[package]]
-name = "rustfs-mimalloc"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
-dependencies = [
- "rustfs-mimalloc-sys",
-]
-
-[[package]]
-name = "rustfs-mimalloc-sys"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
-dependencies = [
- "cc",
-]
-
 [[package]]
 name = "rustfs-notify"
 version = "1.0.0-rc.3"

Cargo.toml

@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
 dav-server = "0.11.0"
 
 # Performance Analysis and Memory Profiling
-rustfs-mimalloc = { version = "0.5.0" }
-rustfs-mimalloc-sys = { version = "0.5.0" }
+mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
+libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
 hotpath = { version = "0.23.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

crates/e2e_test/src/reliant/conditional_writes.rs

@@ -1,52 +1,26 @@
 #![cfg(test)]
 
-use aws_config::meta::region::RegionProviderChain;
+use crate::common::{RustFSTestEnvironment, TEST_BUCKET, init_logging};
 use aws_sdk_s3::Client;
-use aws_sdk_s3::config::{Credentials, Region};
-use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
 use bytes::Bytes;
 use std::error::Error;
+use std::fmt::Debug;
 
-const ENDPOINT: &str = "http://localhost:9000";
-const ACCESS_KEY: &str = "rustfsadmin";
-const SECRET_KEY: &str = "rustfsadmin";
-const BUCKET: &str = "api-test";
+type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
 
-async fn create_aws_s3_client() -> Result<Client, Box<dyn Error>> {
-    let region_provider = RegionProviderChain::default_provider().or_else(Region::new("us-east-1"));
-    let shared_config = aws_config::defaults(aws_config::BehaviorVersion::latest())
-        .region(region_provider)
-        .credentials_provider(Credentials::new(ACCESS_KEY, SECRET_KEY, None, None, "static"))
-        .endpoint_url(ENDPOINT)
-        .load()
-        .await;
-
-    let client = Client::from_conf(
-        aws_sdk_s3::Config::from(&shared_config)
-            .to_builder()
-            .force_path_style(true)
-            .build(),
+fn assert_s3_error_code<T, E>(result: Result<T, SdkError<E>>, expected: &str)
+where
+    T: Debug,
+    E: ProvideErrorMetadata + Debug,
+{
+    let error = result.expect_err("conditional request must fail");
+    assert_eq!(
+        error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some(expected),
+        "unexpected conditional request error: {error:?}"
     );
-    Ok(client)
-}
-
-/// Setup test bucket, creating it if it doesn't exist
-async fn setup_test_bucket(client: &Client) -> Result<(), Box<dyn Error>> {
-    match client.create_bucket().bucket(BUCKET).send().await {
-        Ok(_) => {}
-        Err(SdkError::ServiceError(e)) => {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            if !error_code.eq("BucketAlreadyExists") {
-                return Err(e.into());
-            }
-        }
-        Err(e) => {
-            return Err(e.into());
-        }
-    }
-    Ok(())
 }
 
 /// Generate test data of specified size
@@ -60,7 +34,12 @@ fn generate_test_data(size: usize) -> Vec<u8> {
 }
 
 /// Upload an object and return its ETag
-async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, data: &[u8]) -> Result<String, Box<dyn Error>> {
+async fn upload_object_with_metadata(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    data: &[u8],
+) -> Result<String, Box<dyn Error + Send + Sync>> {
     let response = client
         .put_object()
         .bucket(bucket)
@@ -69,188 +48,164 @@ async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, d
         .send()
         .await?;
 
-    let etag = response.e_tag().unwrap_or("").to_string();
-    Ok(etag)
+    response
+        .e_tag()
+        .map(str::to_owned)
+        .ok_or_else(|| std::io::Error::other("put object response did not include an ETag").into())
 }
 
-/// Cleanup test objects from bucket
-async fn cleanup_objects(client: &Client, bucket: &str, keys: &[&str]) {
-    for key in keys {
-        let _ = client.delete_object().bucket(bucket).key(*key).send().await;
-    }
-}
-
-/// Generate unique test object key
-fn generate_test_key(prefix: &str) -> String {
-    use std::time::{SystemTime, UNIX_EPOCH};
-    let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
-    format!("{prefix}-{timestamp}")
+async fn object_body(client: &Client, key: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
+    let response = client.get_object().bucket(TEST_BUCKET).key(key).send().await?;
+    Ok(response.body.collect().await?.into_bytes())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_okay() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_okay() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("conditional-put-ok");
+    let test_key = "conditional-put-ok";
     let initial_data = generate_test_data(1024); // 1KB test data
-    let updated_data = generate_test_data(2048); // 2KB updated data
+    let matching_data = generate_test_data(2048); // 2KB updated data
+    let non_matching_data = generate_test_data(3072); // 3KB updated data
 
     // Upload initial object and get its ETag
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
 
     // Test 1: PUT with matching If-Match condition (should succeed)
-    let response1 = client
+    client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
-        .body(Bytes::from(updated_data.clone()).into())
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .body(Bytes::from(matching_data.clone()).into())
         .if_match(&initial_etag)
         .send()
-        .await;
-    assert!(response1.is_ok(), "PUT with matching If-Match should succeed");
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), matching_data);
 
     // Test 2: PUT with non-matching If-None-Match condition (should succeed)
     let fake_etag = "\"fake-etag-12345\"";
-    let response2 = client
+    client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
-        .body(Bytes::from(updated_data.clone()).into())
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .body(Bytes::from(non_matching_data.clone()).into())
         .if_none_match(fake_etag)
         .send()
-        .await;
-    assert!(response2.is_ok(), "PUT with non-matching If-None-Match should succeed");
-
-    // Cleanup
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), non_matching_data);
 
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_failed() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_failed() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("conditional-put-failed");
+    let test_key = "conditional-put-failed";
     let initial_data = generate_test_data(1024);
     let updated_data = generate_test_data(2048);
 
     // Upload initial object and get its ETag
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
 
     // Test 1: PUT with non-matching If-Match condition (should fail with 412)
     let fake_etag = "\"fake-etag-should-not-match\"";
     let response1 = client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_match(fake_etag)
         .send()
         .await;
 
-    assert!(response1.is_err(), "PUT with non-matching If-Match should fail");
-    if let Err(e) = response1 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("PreconditionFailed", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
+    assert_s3_error_code(response1, "PreconditionFailed");
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
 
     // Test 2: PUT with matching If-None-Match condition (should fail with 412)
     let response2 = client
         .put_object()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_none_match(&initial_etag)
         .send()
         .await;
 
-    assert!(response2.is_err(), "PUT with matching If-None-Match should fail");
-    if let Err(e) = response2 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("PreconditionFailed", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
-
-    // Cleanup - only need to clean up the initial object since failed PUTs shouldn't create objects
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+    assert_s3_error_code(response2, "PreconditionFailed");
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
 
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_put_when_object_does_not_exist() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_put_when_object_does_not_exist() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let key = "some_key";
-    cleanup_objects(&client, BUCKET, &[key]).await;
+    let key = "conditional-put-missing";
 
     // When the object does not exist, the If-Match condition should always fail
     let response1 = client
         .put_object()
-        .bucket(BUCKET)
+        .bucket(TEST_BUCKET)
         .key(key)
         .body(Bytes::from(generate_test_data(1024)).into())
         .if_match("*")
         .send()
         .await;
-    assert!(response1.is_err());
-    if let Err(e) = response1 {
-        if let SdkError::ServiceError(e) = e {
-            let e = e.into_err();
-            let error_code = e.meta().code().unwrap_or("");
-            assert_eq!("NoSuchKey", error_code);
-        } else {
-            panic!("Unexpected error: {e:?}");
-        }
-    }
+    assert_s3_error_code(response1, "NoSuchKey");
 
     // When the object does not exist, the If-None-Match condition should be able to succeed
-    let response2 = client
+    let created_data = generate_test_data(1024);
+    client
         .put_object()
-        .bucket(BUCKET)
+        .bucket(TEST_BUCKET)
         .key(key)
-        .body(Bytes::from(generate_test_data(1024)).into())
+        .body(Bytes::from(created_data.clone()).into())
         .if_none_match("*")
         .send()
-        .await;
-    assert!(response2.is_ok());
+        .await?;
+    assert_eq!(object_body(&client, key).await?.as_ref(), created_data);
 
-    cleanup_objects(&client, BUCKET, &[key]).await;
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::Error>> {
-    let client = create_aws_s3_client().await?;
-    setup_test_bucket(&client).await?;
+async fn test_conditional_multi_part_upload() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    env.create_test_bucket(TEST_BUCKET).await?;
+    let client = env.create_s3_client();
 
-    let test_key = generate_test_key("multipart-upload-ok");
+    let test_key = "conditional-multipart-upload";
     let test_data = generate_test_data(1024);
-    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &test_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &test_data).await?;
 
     let part_size = 5 * 1024 * 1024; // 5MB per part (minimum for multipart)
     let num_parts = 3;
     let mut parts = Vec::new();
+    let mut expected_data = Vec::with_capacity(part_size * usize::try_from(num_parts)?);
 
     // Initiate multipart upload
-    let initiate_response = client.create_multipart_upload().bucket(BUCKET).key(&test_key).send().await?;
+    let initiate_response = client
+        .create_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .send()
+        .await?;
 
     let upload_id = initiate_response
         .upload_id()
@@ -258,12 +213,13 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
 
     // Upload parts
     for part_number in 1..=num_parts {
-        let part_data = generate_test_data(part_size);
+        let part_data = vec![u8::try_from(part_number)?; part_size];
+        expected_data.extend_from_slice(&part_data);
 
         let upload_part_response = client
             .upload_part()
-            .bucket(BUCKET)
-            .key(&test_key)
+            .bucket(TEST_BUCKET)
+            .key(test_key)
             .upload_id(upload_id)
             .part_number(part_number)
             .body(Bytes::from(part_data).into())
@@ -286,57 +242,62 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
     // Test 1: Multipart upload with wildcard If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match("*")
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
 
     // Test 2: Multipart upload with matching If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match(initial_etag.clone())
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
 
     // Test 3: Multipart upload with unmatching If-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_match("\"abcdef\"")
         .send()
         .await;
 
-    assert!(complete_response.is_err());
+    assert_s3_error_code(complete_response, "PreconditionFailed");
+
+    let staged_parts = client
+        .list_parts()
+        .bucket(TEST_BUCKET)
+        .key(test_key)
+        .upload_id(upload_id)
+        .send()
+        .await?;
+    assert_eq!(staged_parts.parts().len(), usize::try_from(num_parts)?);
 
     // Test 4: Multipart upload with matching If-Match, should succeed
-    let complete_response = client
+    client
         .complete_multipart_upload()
-        .bucket(BUCKET)
-        .key(&test_key)
+        .bucket(TEST_BUCKET)
+        .key(test_key)
         .upload_id(upload_id)
-        .multipart_upload(completed_upload.clone())
+        .multipart_upload(completed_upload)
         .if_match(initial_etag)
         .send()
-        .await;
-
-    assert!(complete_response.is_ok());
-
-    // Cleanup
-    cleanup_objects(&client, BUCKET, &[&test_key]).await;
+        .await?;
+    assert_eq!(object_body(&client, test_key).await?.as_ref(), expected_data);
 
     Ok(())
 }

crates/ecstore/src/set_disk/ops/object.rs

@@ -2124,13 +2124,26 @@ impl SetDisks {
 
             let put_object_size = known_put_object_storage_size(data.size());
             let shard_file_size_raw = erasure.shard_file_size(put_object_size);
-            let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
+            let is_inline_buffer =
+                storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
 
             let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
             let shard_file_size = shard_file_size_raw;
             let shard_size = erasure.shard_size();
             let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
             let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
+            {
+                use std::io::Write;
+                let msg = format!(
+                    "INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
+                    bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
+                    is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
+                );
+                if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
+                    let _ = f.write_all(msg.as_bytes());
+                }
+                let _ = std::io::stderr().write_all(msg.as_bytes());
+            }
             rustfs_io_metrics::record_put_object_path(write_path.metric_label());
             let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
             let (mut writers, errors) = if direct_inline_commit {

crates/ecstore/src/store/object.rs

@@ -3194,7 +3194,7 @@ impl ECStore {
 
         // Default return value
         let mut del_objects = vec![DeletedObject::default(); objects.len()];
-        let accounting = vec![None; objects.len()];
+        let mut accounting = vec![None; objects.len()];
 
         let mut del_errs = Vec::with_capacity(objects.len());
         for _ in 0..objects.len() {

crates/ecstore/src/store/rebalance/support.rs

@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
             .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
             .collect::<Vec<_>>();
 
-        latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
+        latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
 
         let Some(winner) = latest_candidates.first() else {
             return Err(Error::ErasureReadQuorum);

crates/heal/src/heal/manager.rs

@@ -14,7 +14,7 @@
 
 use crate::heal::{
     progress::{HealProgress, HealStatistics},
-    resume::{ReplacementPhase, ResumeGc, ResumeManager, ResumeState, ResumeUtils},
+    resume::{ReplacementPhase, ResumeManager, ResumeState, ResumeUtils},
     storage::HealStorageAPI,
     task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType, demote_to_debug_when},
 };
@@ -53,11 +53,9 @@ const EVENT_HEAL_MAINLINE_THROTTLE: &str = "heal_mainline_throttle";
 const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state";
 const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state";
 const EVENT_HEAL_UNCLEAN_SHUTDOWN: &str = "heal_unclean_shutdown";
-const EVENT_HEAL_RESUME_GC: &str = "heal_resume_gc";
 const LEGACY_ROOT_HEAL_PATH: &str = ".";
 const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3;
 const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30);
-const RESUME_GC_INTERVAL: Duration = Duration::from_secs(60 * 60);
 
 // Admission/scheduler outcomes for per-object requests (Object/Metadata/
 // ECDecode) log via demote_to_debug_when! — MRF, autoheal, and scanner
@@ -1152,49 +1150,6 @@ impl HealManager {
         Ok(())
     }
 
-    /// Start the bounded resume-state inspector.  Destructive GC remains
-    /// disabled until the durable owner/CAS contract from backlog#1927 is
-    /// available; this task therefore cannot remove an active or stale file.
-    async fn start_resume_gc(&self) {
-        let cancel = self.cancel_token.clone();
-        tokio::spawn(async move {
-            let mut gc_by_disk = HashMap::<String, ResumeGc>::new();
-            let mut ticker = interval(RESUME_GC_INTERVAL);
-            loop {
-                tokio::select! {
-                    _ = cancel.cancelled() => break,
-                    _ = ticker.tick() => {
-                        let disks = {
-                            let local_disk_map = local_disk_map_read().await;
-                            local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
-                        };
-                        for disk in disks {
-                            let disk_key = disk.endpoint().to_string();
-                            let gc = gc_by_disk.entry(disk_key).or_default();
-                            tokio::select! {
-                                _ = cancel.cancelled() => return,
-                                result = gc.inspect_disk(&disk) => {
-                                    if let Err(error) = result {
-                                        warn!(
-                                            target: "rustfs::heal::manager",
-                                            event = EVENT_HEAL_RESUME_GC,
-                                            component = LOG_COMPONENT_HEAL,
-                                            subsystem = LOG_SUBSYSTEM_MANAGER,
-                                            state = "inspect_failed",
-                                            endpoint = %disk.endpoint(),
-                                            error = %error,
-                                            "Heal resume GC inspection failed"
-                                        );
-                                    }
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        });
-    }
-
     /// Create new HealManager
     pub fn new(storage: Arc<dyn HealStorageAPI>, config: Option<HealConfig>) -> Self {
         Self::new_with_workload_provider(storage, config, None)
@@ -1260,9 +1215,6 @@ impl HealManager {
         // competing task for the same set.
         self.process_unclean_shutdown().await;
 
-        // Inspect resume artifacts in a bounded, fail-closed background task.
-        self.start_resume_gc().await;
-
         // start auto disk scanner to heal unformatted disks
         if self.config.read().await.enable_auto_heal {
             self.start_auto_disk_scanner().await?;

crates/heal/src/heal/resume.rs

@@ -28,12 +28,10 @@ use super::{
 };
 
 mod checkpoint;
-mod gc;
 mod replacement;
 mod utils;
 
 pub use checkpoint::{CheckpointManager, ResumeCheckpoint};
-pub(crate) use gc::ResumeGc;
 pub(crate) use replacement::replacement_target_identities_match;
 use replacement::replacement_targets_match_identities;
 pub use replacement::{

crates/heal/src/heal/resume/gc.rs

@@ -1,666 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Bounded inspection of heal resume artifacts.
-//!
-//! The durable owner/CAS and quarantine primitives belong to backlog #1927 and
-//! are not part of the current base revision. This module is therefore
-//! deliberately inspect-only. In particular, it must never turn an age check
-//! into a delete: ordinary heal writers still publish raw files on this base,
-//! so a GC-side compare-and-delete would not fence a concurrent claim.
-
-use metrics::counter;
-use std::{
-    collections::BTreeMap,
-    path::{Component, Path},
-    time::{SystemTime, UNIX_EPOCH},
-};
-use tokio::io::AsyncReadExt;
-
-use super::super::{BUCKET_META_PREFIX, DiskError, DiskStore, RUSTFS_META_BUCKET, storage_api::owner::EcstoreDiskAPI};
-use super::{
-    LEGACY_REPLACEMENT_RECOVERY_MARKER_FILE, REPLACEMENT_COMPLETION_PROOF_FILE, REPLACEMENT_INTENT_FILE,
-    REPLACEMENT_INTENT_SEAL_FILE, RESUME_CHECKPOINT_FILE, RESUME_PROGRESS_FILE, RESUME_STATE_FILE, ResumeCheckpoint, ResumeState,
-    checkpoint::CURRENT_CHECKPOINT_SCHEMA,
-};
-use crate::{Error, Result};
-
-const DEFAULT_ENTRY_BUDGET: usize = 256;
-const DEFAULT_BYTE_BUDGET: usize = 4 * 1024 * 1024;
-const GC_METRIC: &str = "rustfs_heal_resume_gc_inspected_total";
-const GC_ERROR_METRIC: &str = "rustfs_heal_resume_gc_inspect_errors_total";
-
-#[derive(Debug, Clone, Copy)]
-pub(crate) struct ResumeGcConfig {
-    /// Maximum number of directory entries considered in one disk pass.
-    pub(crate) max_entries: usize,
-    /// Maximum number of bytes read in one disk pass.
-    pub(crate) max_bytes: usize,
-}
-
-impl Default for ResumeGcConfig {
-    fn default() -> Self {
-        Self {
-            max_entries: DEFAULT_ENTRY_BUDGET,
-            max_bytes: DEFAULT_BYTE_BUDGET,
-        }
-    }
-}
-
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
-pub(crate) struct ResumeGcReport {
-    /// Directory entries visited (including malformed entries).
-    pub(crate) inspected: usize,
-    pub(crate) active_skipped: usize,
-    pub(crate) orphaned: usize,
-    /// Records that must be handed to #1927's quarantine owner.
-    pub(crate) quarantine_required: usize,
-    pub(crate) generation_skipped: usize,
-    pub(crate) clock_skew: usize,
-    pub(crate) read_errors: usize,
-    pub(crate) retained: usize,
-    /// True while #1927's durable claim/quarantine capability is unavailable.
-    pub(crate) destructive_disabled: bool,
-    pub(crate) budget_exhausted: bool,
-}
-
-#[derive(Debug, Default)]
-pub(crate) struct ResumeGc {
-    config: ResumeGcConfig,
-    /// Alternate the first namespace so a full ordinary page cannot starve
-    /// replacement recovery when the list API has no continuation token.
-    recovery_first: bool,
-}
-
-impl ResumeGc {
-    #[cfg(test)]
-    fn with_config(config: ResumeGcConfig) -> Self {
-        Self {
-            config,
-            recovery_first: false,
-        }
-    }
-
-    /// Inspect one bounded page from each resume namespace.
-    ///
-    /// The caller owns scheduling and cancellation. A malformed or unreadable
-    /// artifact is reported and retained so a later pass can retry it; no
-    /// individual artifact error aborts the rest of the bounded page.
-    pub(crate) async fn inspect_disk(&mut self, disk: &DiskStore) -> Result<ResumeGcReport> {
-        let mut report = ResumeGcReport {
-            destructive_disabled: true,
-            ..ResumeGcReport::default()
-        };
-        if self.config.max_entries == 0 || self.config.max_bytes == 0 {
-            report.budget_exhausted = true;
-            return Ok(report);
-        }
-
-        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
-        let mut bytes_read = 0usize;
-        let recovery_first = self.recovery_first;
-        self.recovery_first = !self.recovery_first;
-        if recovery_first {
-            inspect_namespace(self.config, disk, &replacement_prefix(), true, now, &mut bytes_read, &mut report).await?;
-            if !report.budget_exhausted {
-                inspect_namespace(self.config, disk, BUCKET_META_PREFIX, false, now, &mut bytes_read, &mut report).await?;
-            }
-        } else {
-            inspect_namespace(self.config, disk, BUCKET_META_PREFIX, false, now, &mut bytes_read, &mut report).await?;
-            if !report.budget_exhausted {
-                inspect_namespace(self.config, disk, &replacement_prefix(), true, now, &mut bytes_read, &mut report).await?;
-            }
-        }
-
-        counter!(GC_METRIC).increment(u64::try_from(report.inspected).unwrap_or(u64::MAX));
-        counter!(GC_ERROR_METRIC).increment(u64::try_from(report.read_errors).unwrap_or(u64::MAX));
-        Ok(report)
-    }
-}
-
-#[derive(Debug, Default, Clone, Copy)]
-struct ArtifactSet {
-    state: bool,
-    checkpoint: bool,
-    progress: bool,
-    replacement_intent: bool,
-    proof: bool,
-    seal: bool,
-    legacy_marker: bool,
-    temporary: bool,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum ArtifactKind {
-    State,
-    Checkpoint,
-    Progress,
-    ReplacementIntent,
-    Proof,
-    Seal,
-    LegacyMarker,
-}
-
-impl ArtifactSet {
-    fn add(&mut self, kind: ArtifactKind, temporary: bool) {
-        self.temporary |= temporary;
-        match kind {
-            ArtifactKind::State => self.state = true,
-            ArtifactKind::Checkpoint => self.checkpoint = true,
-            ArtifactKind::Progress => self.progress = true,
-            ArtifactKind::ReplacementIntent => self.replacement_intent = true,
-            ArtifactKind::Proof => self.proof = true,
-            ArtifactKind::Seal => self.seal = true,
-            ArtifactKind::LegacyMarker => self.legacy_marker = true,
-        }
-    }
-}
-
-#[derive(Debug, Clone, Copy)]
-struct InspectOptions {
-    max_bytes: usize,
-    now: u64,
-}
-
-struct InspectProgress<'a> {
-    bytes_read: &'a mut usize,
-    report: &'a mut ResumeGcReport,
-}
-
-fn replacement_prefix() -> String {
-    super::replacement_recovery_dir().to_string_lossy().into_owned()
-}
-
-async fn inspect_namespace(
-    config: ResumeGcConfig,
-    disk: &DiskStore,
-    prefix: &str,
-    replacement: bool,
-    now: u64,
-    bytes_read: &mut usize,
-    report: &mut ResumeGcReport,
-) -> Result<()> {
-    let remaining = config.max_entries.saturating_sub(report.inspected);
-    if remaining == 0 {
-        report.budget_exhausted = true;
-        return Ok(());
-    }
-    let count = i32::try_from(remaining).unwrap_or(i32::MAX);
-    let mut entries = match EcstoreDiskAPI::list_dir(disk.as_ref(), "", RUSTFS_META_BUCKET, prefix, count).await {
-        Ok(entries) => entries,
-        Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(()),
-        Err(error) => return Err(error.into()),
-    };
-    entries.sort_unstable();
-
-    let mut artifacts = BTreeMap::<String, ArtifactSet>::new();
-    for entry in entries {
-        if report.inspected >= config.max_entries {
-            report.budget_exhausted = true;
-            break;
-        }
-        report.inspected += 1;
-        let Some((task_id, kind, temporary)) = artifact_name(&entry, replacement) else {
-            report.quarantine_required += 1;
-            report.retained += 1;
-            continue;
-        };
-        artifacts.entry(task_id).or_default().add(kind, temporary);
-    }
-    if report.inspected >= config.max_entries {
-        report.budget_exhausted = true;
-    }
-
-    for (task_id, artifacts) in artifacts {
-        if *bytes_read >= config.max_bytes {
-            report.budget_exhausted = true;
-            break;
-        }
-        let options = InspectOptions {
-            max_bytes: config.max_bytes,
-            now,
-        };
-        let mut progress = InspectProgress { bytes_read, report };
-        inspect_task(options, disk, prefix, replacement, &task_id, artifacts, &mut progress).await?;
-    }
-    Ok(())
-}
-
-async fn inspect_task(
-    options: InspectOptions,
-    disk: &DiskStore,
-    prefix: &str,
-    replacement: bool,
-    task_id: &str,
-    artifacts: ArtifactSet,
-    progress: &mut InspectProgress<'_>,
-) -> Result<()> {
-    let legacy_replacement = !replacement && !artifacts.state && artifacts.replacement_intent;
-    let state_suffix = if replacement || legacy_replacement {
-        REPLACEMENT_INTENT_FILE
-    } else {
-        RESUME_STATE_FILE
-    };
-    let state_path = artifact_path(prefix, task_id, state_suffix)?;
-    let state = match read_bounded(disk, &state_path, options.max_bytes, progress.bytes_read).await {
-        ReadOutcome::Missing => {
-            progress.report.orphaned += 1;
-            progress.report.retained += 1;
-            return Ok(());
-        }
-        ReadOutcome::TooLarge => {
-            progress.report.quarantine_required += 1;
-            progress.report.retained += 1;
-            progress.report.budget_exhausted = true;
-            return Ok(());
-        }
-        ReadOutcome::Error => {
-            progress.report.read_errors += 1;
-            progress.report.retained += 1;
-            return Ok(());
-        }
-        ReadOutcome::Bytes(bytes) => bytes,
-    };
-
-    let parsed: ResumeState = match serde_json::from_slice(&state) {
-        Ok(state) => state,
-        Err(_) => {
-            progress.report.quarantine_required += 1;
-            progress.report.retained += 1;
-            return Ok(());
-        }
-    };
-    if parsed.schema_version > super::CURRENT_RESUME_SCHEMA || parsed.task_id != task_id {
-        progress.report.quarantine_required += 1;
-        progress.report.retained += 1;
-        return Ok(());
-    }
-    if persistent_age_seconds(options.now, parsed.last_update).is_none() {
-        progress.report.clock_skew += 1;
-        progress.report.retained += 1;
-        return Ok(());
-    }
-    if let Some(generation) = parsed.replacement_generation.as_deref()
-        && !claim_generation_matches(Some(generation), Some(task_id))
-    {
-        progress.report.generation_skipped += 1;
-        progress.report.retained += 1;
-        return Ok(());
-    }
-
-    if !replacement && artifacts.checkpoint {
-        let checkpoint_path = artifact_path(prefix, task_id, RESUME_CHECKPOINT_FILE)?;
-        match read_bounded(disk, &checkpoint_path, options.max_bytes, progress.bytes_read).await {
-            ReadOutcome::Bytes(bytes) => match serde_json::from_slice::<ResumeCheckpoint>(&bytes) {
-                Ok(checkpoint) if checkpoint.schema_version <= CURRENT_CHECKPOINT_SCHEMA && checkpoint.task_id == task_id => {}
-                _ => {
-                    progress.report.quarantine_required += 1;
-                    progress.report.retained += 1;
-                }
-            },
-            ReadOutcome::Missing => {
-                progress.report.orphaned += 1;
-                progress.report.retained += 1;
-            }
-            ReadOutcome::TooLarge => {
-                progress.report.quarantine_required += 1;
-                progress.report.retained += 1;
-                progress.report.budget_exhausted = true;
-            }
-            ReadOutcome::Error => {
-                progress.report.read_errors += 1;
-                progress.report.retained += 1;
-            }
-        }
-    }
-
-    if artifacts.state && artifacts.replacement_intent {
-        // A task cannot have two authoritative state records in one namespace;
-        // preserve both until the durable owner can resolve the generation.
-        progress.report.quarantine_required += 1;
-    }
-
-    if !parsed.completed {
-        progress.report.active_skipped += 1;
-    }
-    // The state and all associated evidence remain recoverable until #1927
-    // supplies a common generation/CAS transition and quarantine owner.
-    progress.report.retained += 1;
-    Ok(())
-}
-
-enum ReadOutcome {
-    Bytes(Vec<u8>),
-    Missing,
-    TooLarge,
-    Error,
-}
-
-async fn read_bounded(disk: &DiskStore, path: &str, max_bytes: usize, bytes_read: &mut usize) -> ReadOutcome {
-    let remaining = max_bytes.saturating_sub(*bytes_read);
-    if remaining == 0 {
-        return ReadOutcome::TooLarge;
-    }
-    let read_len = remaining.saturating_add(1);
-    let reader = match EcstoreDiskAPI::read_file(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
-        Ok(reader) => reader,
-        Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return ReadOutcome::Missing,
-        Err(_) => return ReadOutcome::Error,
-    };
-    let mut bytes = Vec::with_capacity(read_len.min(64 * 1024));
-    let Ok(read_len) = u64::try_from(read_len) else {
-        return ReadOutcome::TooLarge;
-    };
-    if reader.take(read_len).read_to_end(&mut bytes).await.is_err() {
-        return ReadOutcome::Error;
-    }
-    if bytes.len() > remaining {
-        *bytes_read = max_bytes;
-        return ReadOutcome::TooLarge;
-    }
-    *bytes_read = bytes_read.saturating_add(bytes.len());
-    ReadOutcome::Bytes(bytes)
-}
-
-fn artifact_path(prefix: &str, task_id: &str, suffix: &str) -> Result<String> {
-    if super::validate_resume_task_id(task_id).is_err() {
-        return Err(Error::other("invalid resume task id"));
-    }
-    Path::new(prefix)
-        .join(format!("{task_id}_{suffix}"))
-        .to_str()
-        .map(str::to_owned)
-        .ok_or_else(|| Error::other("invalid resume artifact path"))
-}
-
-/// Parse one directory entry without ever accepting a path component supplied
-/// by a client. DiskAPI filters symlinks, but this check also protects remote
-/// implementations and future mutating callers from traversal/reparse names.
-fn artifact_name(entry: &str, replacement: bool) -> Option<(String, ArtifactKind, bool)> {
-    let path = Path::new(entry);
-    if entry.is_empty() || path.components().count() != 1 || !matches!(path.components().next(), Some(Component::Normal(_))) {
-        return None;
-    }
-    let (stem, temporary) = entry
-        .strip_suffix(".tmp")
-        .map(|stem| (stem, true))
-        .or_else(|| entry.strip_suffix(".bak").map(|stem| (stem, true)))
-        .unwrap_or((entry, false));
-    let suffixes: &[(&str, ArtifactKind)] = if replacement {
-        &[
-            (REPLACEMENT_INTENT_FILE, ArtifactKind::ReplacementIntent),
-            (REPLACEMENT_COMPLETION_PROOF_FILE, ArtifactKind::Proof),
-            (REPLACEMENT_INTENT_SEAL_FILE, ArtifactKind::Seal),
-        ]
-    } else {
-        &[
-            (RESUME_STATE_FILE, ArtifactKind::State),
-            (RESUME_CHECKPOINT_FILE, ArtifactKind::Checkpoint),
-            (RESUME_PROGRESS_FILE, ArtifactKind::Progress),
-            (LEGACY_REPLACEMENT_RECOVERY_MARKER_FILE, ArtifactKind::LegacyMarker),
-            (REPLACEMENT_INTENT_FILE, ArtifactKind::ReplacementIntent),
-            (REPLACEMENT_COMPLETION_PROOF_FILE, ArtifactKind::Proof),
-            (REPLACEMENT_INTENT_SEAL_FILE, ArtifactKind::Seal),
-        ]
-    };
-    suffixes.iter().find_map(|(suffix, kind)| {
-        stem.strip_suffix(&format!("_{suffix}"))
-            .filter(|task_id| super::validate_resume_task_id(task_id).is_ok())
-            .map(|task_id| (task_id.to_string(), *kind, temporary))
-    })
-}
-
-fn persistent_age_seconds(now: u64, updated: u64) -> Option<u64> {
-    now.checked_sub(updated)
-}
-
-fn claim_generation_matches(observed: Option<&str>, expected: Option<&str>) -> bool {
-    expected.is_none() || observed == expected
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::heal::{DiskOption, Endpoint, new_disk};
-    use tempfile::TempDir;
-    use uuid::Uuid;
-
-    async fn test_disk() -> (TempDir, DiskStore) {
-        let temp = TempDir::new().expect("test disk directory");
-        let endpoint = Endpoint::try_from(temp.path().to_string_lossy().as_ref()).expect("test endpoint");
-        let disk = new_disk(
-            &endpoint,
-            &DiskOption {
-                cleanup: false,
-                health_check: false,
-            },
-        )
-        .await
-        .expect("test disk");
-        match disk.make_volume(RUSTFS_META_BUCKET).await {
-            Ok(()) | Err(DiskError::VolumeExists) => {}
-            Err(error) => panic!("metadata volume: {error}"),
-        }
-        match disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await {
-            Ok(()) | Err(DiskError::VolumeExists) => {}
-            Err(error) => panic!("resume volume: {error}"),
-        }
-        (temp, disk)
-    }
-
-    async fn write_state(disk: &DiskStore, state: &ResumeState) {
-        let path = format!("{BUCKET_META_PREFIX}/{}_{}", state.task_id, RESUME_STATE_FILE);
-        disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(state).unwrap().into())
-            .await
-            .expect("resume state");
-    }
-
-    async fn write_replacement_state(disk: &DiskStore, state: &ResumeState) {
-        let volume = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}/ahm-replacement");
-        match disk.make_volume(&volume).await {
-            Ok(()) | Err(DiskError::VolumeExists) => {}
-            Err(error) => panic!("replacement volume: {error}"),
-        }
-        let path = format!("{}/{}_{}", replacement_prefix(), state.task_id, REPLACEMENT_INTENT_FILE);
-        disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(state).unwrap().into())
-            .await
-            .expect("replacement state");
-    }
-
-    #[tokio::test]
-    async fn production_gc_does_not_delete_claimed_resume_state() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        write_state(&disk, &ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![])).await;
-        let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
-        assert_eq!(report.active_skipped, 1);
-        assert!(
-            disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
-                .await
-                .is_ok()
-        );
-    }
-
-    #[tokio::test]
-    async fn production_gc_generation_mismatch_is_skip() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
-        state.replacement_generation = Some(Uuid::new_v4().to_string());
-        write_state(&disk, &state).await;
-        assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().generation_skipped, 1);
-    }
-
-    #[cfg(unix)]
-    #[tokio::test]
-    async fn production_gc_rejects_symlink_or_outside_prefix() {
-        let (_temp, disk) = test_disk().await;
-        let id = Uuid::new_v4().to_string();
-        let root = EcstoreDiskAPI::path(disk.as_ref());
-        let outside = root.join("outside-resume-state");
-        std::fs::write(&outside, b"must remain").expect("outside fixture");
-        let symlink = root
-            .join(RUSTFS_META_BUCKET)
-            .join(BUCKET_META_PREFIX)
-            .join(format!("{id}_{RESUME_STATE_FILE}"));
-        std::os::unix::fs::symlink(&outside, &symlink).expect("symlink fixture");
-        let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
-        assert_eq!(report.inspected, 0, "symlinks are not eligible artifacts");
-        assert!(outside.exists());
-        assert!(artifact_name(&format!("{id}_{RESUME_STATE_FILE}"), false).is_some());
-        assert!(artifact_name(&format!("../{id}_{RESUME_STATE_FILE}"), false).is_none());
-        assert!(artifact_name(&format!("{id}/link_{RESUME_STATE_FILE}"), false).is_none());
-    }
-
-    #[cfg(not(unix))]
-    #[test]
-    fn production_gc_rejects_symlink_or_outside_prefix() {
-        let id = Uuid::new_v4().to_string();
-        assert!(artifact_name(&format!("{id}_{RESUME_STATE_FILE}"), false).is_some());
-        assert!(artifact_name(&format!("../{id}_{RESUME_STATE_FILE}"), false).is_none());
-        assert!(artifact_name(&format!("{id}/link_{RESUME_STATE_FILE}"), false).is_none());
-    }
-
-    #[tokio::test]
-    async fn production_gc_delete_failure_leaves_recoverable_state() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
-        state.mark_completed();
-        write_state(&disk, &state).await;
-        ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
-        assert!(
-            disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
-                .await
-                .is_ok()
-        );
-    }
-
-    #[tokio::test]
-    async fn production_gc_handles_clock_skew_and_restart() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let mut state = ResumeState::new(task_id, "set".into(), "disk".into(), vec![]);
-        state.last_update = u64::MAX;
-        write_state(&disk, &state).await;
-        assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().clock_skew, 1);
-        assert!(persistent_age_seconds(1, 2).is_none());
-    }
-
-    #[tokio::test]
-    async fn production_gc_100k_states_respects_budget() {
-        let (_temp, disk) = test_disk().await;
-        for _ in 0..8 {
-            let state = ResumeState::new(Uuid::new_v4().to_string(), "set".into(), "disk".into(), vec![]);
-            write_state(&disk, &state).await;
-        }
-        let config = ResumeGcConfig {
-            max_entries: 2,
-            max_bytes: usize::MAX,
-        };
-        let report = ResumeGc::with_config(config).inspect_disk(&disk).await.unwrap();
-        assert!(report.inspected <= 2);
-        assert!(report.budget_exhausted);
-    }
-
-    #[tokio::test]
-    async fn production_gc_recovery_namespace_is_not_starved() {
-        let (_temp, disk) = test_disk().await;
-        let ordinary = ResumeState::new(Uuid::new_v4().to_string(), "set".into(), "disk".into(), vec![]);
-        write_state(&disk, &ordinary).await;
-        let replacement_id = Uuid::new_v4().to_string();
-        let mut replacement = ResumeState::new(replacement_id, "set".into(), "disk".into(), vec![]);
-        replacement.replacement_generation = Some(replacement.task_id.clone());
-        write_replacement_state(&disk, &replacement).await;
-
-        let mut gc = ResumeGc::with_config(ResumeGcConfig {
-            max_entries: 1,
-            max_bytes: usize::MAX,
-        });
-        assert_eq!(gc.inspect_disk(&disk).await.unwrap().inspected, 1);
-        let second = gc.inspect_disk(&disk).await.unwrap();
-        assert_eq!(second.inspected, 1, "the next bounded pass must start at recovery");
-        assert_eq!(second.active_skipped, 1);
-    }
-
-    #[tokio::test]
-    async fn production_gc_pairs_orphan_checkpoint_and_resume() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let checkpoint = ResumeCheckpoint::new(task_id.clone());
-        let path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-        disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&checkpoint).unwrap().into())
-            .await
-            .expect("checkpoint");
-        assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().orphaned, 1);
-    }
-
-    #[tokio::test]
-    async fn production_gc_does_not_delete_slow_active_task() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let mut state = ResumeState::new(task_id, "set".into(), "disk".into(), vec![]);
-        state.last_update = 1;
-        write_state(&disk, &state).await;
-        assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().active_skipped, 1);
-    }
-
-    #[tokio::test]
-    async fn production_gc_disables_on_mixed_version_capability() {
-        assert!(claim_generation_matches(None, None));
-        assert!(!claim_generation_matches(Some("new"), Some("old")));
-        // No #1927 capability means this implementation has no delete path.
-        assert!(ResumeGcConfig::default().max_entries > 0);
-        let (_temp, disk) = test_disk().await;
-        let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
-        assert!(report.destructive_disabled);
-    }
-
-    #[tokio::test]
-    async fn production_gc_future_schema_is_not_mtime_deleted() {
-        let (_temp, disk) = test_disk().await;
-        let task_id = Uuid::new_v4().to_string();
-        let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
-        state.schema_version = super::super::CURRENT_RESUME_SCHEMA + 1;
-        write_state(&disk, &state).await;
-        let report = ResumeGc::default().inspect_disk(&disk).await.unwrap();
-        assert_eq!(report.quarantine_required, 1);
-        assert!(
-            disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
-                .await
-                .is_ok()
-        );
-    }
-
-    #[tokio::test]
-    async fn production_gc_quarantine_cleanup_is_bounded() {
-        let (_temp, disk) = test_disk().await;
-        for _ in 0..4 {
-            let task_id = Uuid::new_v4().to_string();
-            let path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}");
-            disk.write_all(RUSTFS_META_BUCKET, &path, b"corrupt".to_vec().into())
-                .await
-                .expect("corrupt state");
-        }
-        let report = ResumeGc::with_config(ResumeGcConfig {
-            max_entries: 2,
-            max_bytes: 1024,
-        })
-        .inspect_disk(&disk)
-        .await
-        .expect("inspect");
-        assert!(report.quarantine_required <= 2);
-        assert!(report.budget_exhausted);
-    }
-}

deny.toml

@@ -43,6 +43,9 @@ allow-git = [
     # RustFS fork carrying presigned expiry and constant-time authentication fixes.
     # owner: rustfs-maintainers  review: 2026-10
     "https://github.com/rustfs/s3s.git",
+    # MiMalloc fork pinned for hotpath allocation counting support.
+    # owner: houseme  review: 2026-10
+    "https://github.com/xonatius/mimalloc_rust.git",
 ]
 
 [bans]

docs/testing/ci-gates.md

@@ -1,149 +0,0 @@
-# CI gate matrix
-
-This file is the source of truth for which validation runs on each event, its
-configured wall-clock budget, and whether it can block a merge. Test taxonomy,
-naming, and nextest serialization rules remain in [README.md](README.md); e2e
-membership and counts remain in
-[e2e-suite-inventory.md](e2e-suite-inventory.md).
-
-The distinction between **required** and **report-only** is load-bearing:
-a failing job blocks a merge only when its exact check name is present in the
-live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
-does not make a job required by itself.
-
-## Required merge checks
-
-The live `main` ruleset (`6436880`) currently requires exactly these contexts:
-
-| Required context | Producer | Validation |
-|---|---|---|
-| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
-| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
-| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
-
-For pull requests limited to the paths excluded by the main CI workflow,
-`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
-`Test and Lint` under the same names. It runs the real quick checks and the
-planning-document guard; it does not claim that Rust compilation or runtime
-tests ran. Despite the workflow name, these paths also include selected deploy,
-workflow, and lock files.
-
-Verify the live rule rather than trusting this snapshot before changing merge
-policy:
-
-```bash
-gh api repos/rustfs/rustfs/rulesets/6436880 \
-  --jq '.rules[] | select(.type == "required_status_checks") | .parameters'
-```
-
-The ruleset currently has `strict_required_status_checks_policy=false`.
-`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
-them, but `End-to-End Tests (full merge gate)` is not currently a required
-context. Therefore the repository is prepared to test a merge-queue SHA, but
-the workflow alone does not prove that every merge passed that lane.
-
-## Pull request and merge matrix
-
-Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
-means the result is visible and actionable but is not in the live required
-context list.
-
-| Event | Validation | Budget | Merge status | Reproduction |
-|---|---|---:|---|---|
-| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
-| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
-| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
-| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
-| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
-| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
-| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
-| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
-| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
-| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
-| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
-| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
-| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
-| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
-| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
-| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
-| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
-| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
-
-The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
-instead of adding a second ad-hoc selector. Before a profile runs,
-`scripts/check_test_wiring.py` compares its exact membership to the committed
-digest so a silent test drop fails closed.
-
-## Scheduled and manual validation
-
-Scheduled lanes are independent fault domains. They do not block a pull
-request, but their workflow-local gate can fail the run and scheduled failures
-are routed to the shared failure-issue action. The scheduled-validation
-watchdog and freshness workflow separately detect incomplete runs and missing
-schedules.
-
-| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
-|---|---|---:|---|---|
-| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
-| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
-| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
-| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
-| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
-| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
-| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
-| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
-| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
-| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
-| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
-| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
-| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
-
-Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
-runs are debugging evidence and intentionally do not open scheduled-failure
-issues. A manual performance run may explicitly allow a known regression; that
-override must not be treated as an ordinary passing baseline.
-
-## Release validation
-
-Release validation is post-merge and tag-driven; it does not substitute for a
-pull-request gate.
-
-| Event | Validation | Budget | Result |
-|---|---|---:|---|
-| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
-| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
-| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
-| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
-| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
-
-Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
-are backfill/debug paths and do not prove the automatic `workflow_run` chain.
-
-## Evidence requirements
-
-A green check is useful only when it proves the intended behavior ran:
-
-- Record the exact commit SHA and run URL.
-- Separate product failure from runner prerequisites, service readiness, and
-  cancellation. Repair the precondition, then rerun the exact workload.
-- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
-  server logs where the workflow provides them.
-- For a bug fix or a new fault checker, provide sensitivity evidence: the old
-  behavior or an intentional mutation must fail the new oracle, and the fixed
-  behavior must pass it.
-- Never promote a report-only lane to required from one green run. Require at
-  least 14 days and 30 representative pull requests with at least 99% complete
-  execution, then update the ruleset and this table together.
-
-## Change checklist
-
-Update this file in the same pull request when any of these change:
-
-- workflow triggers, job names, timeouts, or nextest profile ownership;
-- required status contexts or strict/merge-queue policy;
-- scheduled cadence, alert routing, artifact contract, or local reproduction;
-- report-only versus gating semantics.
-
-Do not copy per-module test counts here. Update
-[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
-digest instead.

docs/testing/e2e-suite-inventory.md

@@ -84,7 +84,7 @@
 | protocols | 16 | 🌙 |
 | quota_test | 14 |  |
 | reliability_disk_fault_test | 4 |  |
-| reliant | 25 | 19 ✅ |
+| reliant | 29 | 19 ✅ |
 | replication_extension_test | 75 | 20 ✅ +55 🌙 |
 | security_boundary_test | 4 |  |
 | server_startup_failfast_test | 1 |  |

rustfs/Cargo.toml

@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 # Data structures
 hashbrown = { workspace = true, features = ["serde", "rayon"] }
-rustfs-mimalloc = { workspace = true }
+mimalloc = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libsystemd.workspace = true
 
 [target.'cfg(not(target_os = "windows"))'.dependencies]
-rustfs-mimalloc-sys.workspace = true
+libmimalloc-sys.workspace = true
 
 [dev-dependencies]
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }

rustfs/src/allocator_reclaim.rs

@@ -369,8 +369,14 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
 }
 
 #[cfg(not(target_os = "windows"))]
+#[allow(unsafe_code)]
 fn collect_allocator_memory(force: bool) -> Result<(), String> {
-    rustfs_mimalloc::MiMalloc::collect(force);
+    // SAFETY: `mi_collect` is provided by the active global allocator backend
+    // on this target family. It is explicitly intended to reclaim retained
+    // pages/segments and does not require additional invariants from the caller.
+    unsafe {
+        libmimalloc_sys::mi_collect(force);
+    }
     Ok(())
 }
 

rustfs/src/main.rs

@@ -26,22 +26,22 @@ struct MiMallocAllocator;
 unsafe impl GlobalAlloc for MiMallocAllocator {
     unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
         // SAFETY: the caller upholds GlobalAlloc's contract for layout.
-        unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
+        unsafe { mimalloc::MiMalloc.alloc(layout) }
     }
 
     unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
         // SAFETY: the caller upholds GlobalAlloc's contract for layout.
-        unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
+        unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
     }
 
     unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
         // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
-        unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
+        unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
     }
 
     unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
         // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
-        unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
+        unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
     }
 }
 
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
 
 #[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
 #[global_allocator]
-static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
 fn main() {
     let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,9 +71,8 @@ mod tests {
         allocation.extend_from_slice(&[7_u8; 64]);
 
         assert_eq!(allocation.len(), 64);
-        let heap = rustfs_mimalloc::heap::Heap::main();
         // SAFETY: the live Vec pointer is valid to inspect for heap ownership.
-        assert!(unsafe { heap.contains(allocation.as_ptr()) });
+        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
     }
 
     #[test]
@@ -86,13 +85,12 @@ mod tests {
         let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
         let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
         let allocator = super::MiMallocAllocator;
-        let heap = rustfs_mimalloc::heap::Heap::main();
 
         // SAFETY: The pointer is checked for null before use and later released
         // through the same allocator with the corresponding layout.
         let ptr = unsafe { allocator.alloc_zeroed(layout) };
         assert!(!ptr.is_null());
-        assert!(unsafe { heap.contains(ptr) });
+        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
         assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
 
         // SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -104,7 +102,7 @@ mod tests {
             panic!("mimalloc realloc failed in allocator smoke test");
         }
 
-        assert!(unsafe { heap.contains(grown_ptr) });
+        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
         // SAFETY: `grown_ptr` was reallocated by `allocator` and is released
         // with the matching grown layout.
         unsafe { allocator.dealloc(grown_ptr, grown_layout) };

rustfs/src/memory_observability.rs

@@ -17,7 +17,10 @@ use rustfs_io_metrics::{
     record_cpu_usage, record_memory_usage, record_process_memory_split,
 };
 use serde::Serialize;
+#[cfg(any(test, not(target_os = "windows")))]
 use serde_json::Value;
+#[cfg(not(target_os = "windows"))]
+use std::ffi::CStr;
 use std::path::Path;
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
@@ -228,18 +231,7 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
     read_cgroup_v2().or_else(read_cgroup_v1)
 }
 
-fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
-    let json = rustfs_mimalloc::MiMalloc::stats_json();
-    if json.is_empty() {
-        return None;
-    }
-    let observation = parse_mimalloc_stats_json(&json)?;
-    Some(AllocatorMemorySnapshot {
-        backend: crate::allocator_reclaim::allocator_backend(),
-        observation,
-    })
-}
-
+#[cfg(any(test, not(target_os = "windows")))]
 fn numeric_json_value(value: &Value) -> Option<u64> {
     match value {
         Value::Number(number) => number
@@ -250,6 +242,7 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
     }
 }
 
+#[cfg(any(test, not(target_os = "windows")))]
 fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
     match value {
         Value::Object(fields) => fields
@@ -261,6 +254,7 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
     }
 }
 
+#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
     match value {
         Value::Object(fields) => {
@@ -277,10 +271,12 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
     }
 }
 
+#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
     mimalloc_stat_field(value, metric, "current")
 }
 
+#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
     metrics
         .iter()
@@ -289,6 +285,7 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
         .filter(|value| *value > 0)
 }
 
+#[cfg(any(test, not(target_os = "windows")))]
 fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
     let value = serde_json::from_str::<Value>(stats_json).ok()?;
     let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -315,6 +312,33 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
     }
 }
 
+#[cfg(not(target_os = "windows"))]
+#[allow(unsafe_code)]
+fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
+    // SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
+    // mimalloc when called with a null input buffer. The mimalloc API requires
+    // freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
+    let observation = unsafe {
+        let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
+        if stats_ptr.is_null() {
+            return None;
+        }
+
+        let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
+        libmimalloc_sys::mi_free(stats_ptr.cast());
+        observation?
+    };
+    Some(AllocatorMemorySnapshot {
+        backend: crate::allocator_reclaim::allocator_backend(),
+        observation,
+    })
+}
+
+#[cfg(target_os = "windows")]
+fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
+    None
+}
+
 fn configured_memory_observability_interval_secs() -> u64 {
     rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
 }
@@ -542,13 +566,6 @@ mod tests {
         assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
     }
 
-    #[test]
-    fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
-        let snapshot = super::read_allocator_memory_snapshot();
-        #[cfg(not(target_os = "windows"))]
-        assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
-    }
-
     #[test]
     fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
         let snapshot = build_memory_observability_status_snapshot(false, 15, false);