Merge remote-tracking branch 'origin/main' into HEAD

# Conflicts:
#	.config/e2e-full-selection.txt

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=ae412c27e2e43f3fda3e0c21ef616b48a6403354faab8db8d4154202e3c4eec5
+sha256-darwin=315bb13d9a2199fc87c9beabbdaf31672b30de8996a5eeb843d3d76f6a1540ca
 sha256-linux=ef3be856bd3257c2c369428f66a48ad073dad8187229fe95840a600a81edf22b

crates/config/src/constants/object.rs

@@ -168,19 +168,6 @@ pub const DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_WRITE);
 const _: () = assert!(!DEFAULT_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED);
 
-/// Request writing pool metadata version 2.
-///
-/// This remains ineffective until [`ENV_POOL_META_V2_FLEET_CONFIRMED`] is also enabled.
-pub const ENV_POOL_META_V2_WRITE: &str = "RUSTFS_POOL_META_V2_WRITE";
-pub const DEFAULT_POOL_META_V2_WRITE: bool = false;
-
-/// Operator-attested confirmation that every pool metadata reader and writer understands version 2.
-pub const ENV_POOL_META_V2_FLEET_CONFIRMED: &str = "RUSTFS_POOL_META_V2_FLEET_CONFIRMED";
-pub const DEFAULT_POOL_META_V2_FLEET_CONFIRMED: bool = false;
-
-const _: () = assert!(!DEFAULT_POOL_META_V2_WRITE);
-const _: () = assert!(!DEFAULT_POOL_META_V2_FLEET_CONFIRMED);
-
 // =============================================================================
 // Concurrent Request Fix - Timeout and Backpressure Configuration
 // =============================================================================
@@ -749,10 +736,4 @@ mod remote_version_state_tests {
             "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED"
         );
     }
-
-    #[test]
-    fn pool_meta_v2_gate_uses_stable_environment_names() {
-        assert_eq!(super::ENV_POOL_META_V2_WRITE, "RUSTFS_POOL_META_V2_WRITE");
-        assert_eq!(super::ENV_POOL_META_V2_FLEET_CONFIRMED, "RUSTFS_POOL_META_V2_FLEET_CONFIRMED");
-    }
 }

crates/e2e_test/src/kms/kms_fault_recovery_test.rs

@@ -393,11 +393,10 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
     Ok(())
 }
 
-/// Test KMS resilience to temporary resource constraints
+/// Test concurrent KMS encryption requests
 #[tokio::test]
-async fn test_kms_resource_constraints() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+async fn test_kms_concurrent_encryption_requests() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    info!("🧪 Testing KMS behavior under resource constraints");
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
@@ -431,29 +430,27 @@ async fn test_kms_resource_constraints() -> Result<(), Box<dyn std::error::Error
     }
 
     // Wait for all uploads to complete
-    let mut successful_uploads = 0;
-    let mut failed_uploads = 0;
+    let mut failures = Vec::new();
 
     for task in upload_tasks {
-        let (object_key, result) = task.await.unwrap();
+        let (object_key, result) = task.await?;
         match result {
             Ok(_) => {
-                successful_uploads += 1;
                 info!("✅ Rapid upload {} succeeded", object_key);
             }
             Err(e) => {
-                failed_uploads += 1;
                 warn!("❌ Rapid upload {} failed: {}", object_key, e);
+                failures.push(format!("{object_key}: {e}"));
             }
         }
     }
 
-    info!("📊 Rapid upload results: {} succeeded, {} failed", successful_uploads, failed_uploads);
-
-    // We expect most uploads to succeed even under load
-    assert!(successful_uploads >= 7, "Expected at least 7/10 rapid uploads to succeed");
+    assert!(
+        failures.is_empty(),
+        "all 10 concurrent KMS uploads must succeed; failures: {}",
+        failures.join("; ")
+    );
 
     kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
-    info!("✅ Resource constraints test completed successfully");
     Ok(())
 }

crates/e2e_test/src/reliant/conditional_writes.rs

@@ -1,26 +1,52 @@
 #![cfg(test)]
 
-use crate::common::{RustFSTestEnvironment, TEST_BUCKET, init_logging};
+use aws_config::meta::region::RegionProviderChain;
 use aws_sdk_s3::Client;
-use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
+use aws_sdk_s3::config::{Credentials, Region};
+use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
 use bytes::Bytes;
 use std::error::Error;
-use std::fmt::Debug;
 
-type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
+const ENDPOINT: &str = "http://localhost:9000";
+const ACCESS_KEY: &str = "rustfsadmin";
+const SECRET_KEY: &str = "rustfsadmin";
+const BUCKET: &str = "api-test";
 
-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:?}"
+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(),
     );
+    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
@@ -34,12 +60,7 @@ 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 + Send + Sync>> {
+async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, data: &[u8]) -> Result<String, Box<dyn Error>> {
     let response = client
         .put_object()
         .bucket(bucket)
@@ -48,164 +69,188 @@ async fn upload_object_with_metadata(
         .send()
         .await?;
 
-    response
-        .e_tag()
-        .map(str::to_owned)
-        .ok_or_else(|| std::io::Error::other("put object response did not include an ETag").into())
+    let etag = response.e_tag().unwrap_or("").to_string();
+    Ok(etag)
 }
 
-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())
+/// 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}")
 }
 
 #[tokio::test]
-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();
+#[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?;
 
-    let test_key = "conditional-put-ok";
+    let test_key = generate_test_key("conditional-put-ok");
     let initial_data = generate_test_data(1024); // 1KB test data
-    let matching_data = generate_test_data(2048); // 2KB updated data
-    let non_matching_data = generate_test_data(3072); // 3KB updated data
+    let updated_data = generate_test_data(2048); // 2KB updated data
 
     // Upload initial object and get its ETag
-    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
 
     // Test 1: PUT with matching If-Match condition (should succeed)
-    client
+    let response1 = client
         .put_object()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
-        .body(Bytes::from(matching_data.clone()).into())
+        .bucket(BUCKET)
+        .key(&test_key)
+        .body(Bytes::from(updated_data.clone()).into())
         .if_match(&initial_etag)
         .send()
-        .await?;
-    assert_eq!(object_body(&client, test_key).await?.as_ref(), matching_data);
+        .await;
+    assert!(response1.is_ok(), "PUT with matching If-Match should succeed");
 
     // Test 2: PUT with non-matching If-None-Match condition (should succeed)
     let fake_etag = "\"fake-etag-12345\"";
-    client
+    let response2 = client
         .put_object()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
-        .body(Bytes::from(non_matching_data.clone()).into())
+        .bucket(BUCKET)
+        .key(&test_key)
+        .body(Bytes::from(updated_data.clone()).into())
         .if_none_match(fake_etag)
         .send()
-        .await?;
-    assert_eq!(object_body(&client, test_key).await?.as_ref(), non_matching_data);
+        .await;
+    assert!(response2.is_ok(), "PUT with non-matching If-None-Match should succeed");
+
+    // Cleanup
+    cleanup_objects(&client, BUCKET, &[&test_key]).await;
 
     Ok(())
 }
 
 #[tokio::test]
-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();
+#[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?;
 
-    let test_key = "conditional-put-failed";
+    let test_key = generate_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, TEST_BUCKET, test_key, &initial_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, 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(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_match(fake_etag)
         .send()
         .await;
 
-    assert_s3_error_code(response1, "PreconditionFailed");
-    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
+    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:?}");
+        }
+    }
 
     // Test 2: PUT with matching If-None-Match condition (should fail with 412)
     let response2 = client
         .put_object()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .body(Bytes::from(updated_data.clone()).into())
         .if_none_match(&initial_etag)
         .send()
         .await;
 
-    assert_s3_error_code(response2, "PreconditionFailed");
-    assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
+    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;
 
     Ok(())
 }
 
 #[tokio::test]
-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();
+#[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?;
 
-    let key = "conditional-put-missing";
+    let key = "some_key";
+    cleanup_objects(&client, BUCKET, &[key]).await;
 
     // When the object does not exist, the If-Match condition should always fail
     let response1 = client
         .put_object()
-        .bucket(TEST_BUCKET)
+        .bucket(BUCKET)
         .key(key)
         .body(Bytes::from(generate_test_data(1024)).into())
         .if_match("*")
         .send()
         .await;
-    assert_s3_error_code(response1, "NoSuchKey");
+    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:?}");
+        }
+    }
 
     // When the object does not exist, the If-None-Match condition should be able to succeed
-    let created_data = generate_test_data(1024);
-    client
+    let response2 = client
         .put_object()
-        .bucket(TEST_BUCKET)
+        .bucket(BUCKET)
         .key(key)
-        .body(Bytes::from(created_data.clone()).into())
+        .body(Bytes::from(generate_test_data(1024)).into())
         .if_none_match("*")
         .send()
-        .await?;
-    assert_eq!(object_body(&client, key).await?.as_ref(), created_data);
+        .await;
+    assert!(response2.is_ok());
 
+    cleanup_objects(&client, BUCKET, &[key]).await;
     Ok(())
 }
 
 #[tokio::test]
-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();
+#[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?;
 
-    let test_key = "conditional-multipart-upload";
+    let test_key = generate_test_key("multipart-upload-ok");
     let test_data = generate_test_data(1024);
-    let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &test_data).await?;
+    let initial_etag = upload_object_with_metadata(&client, 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(TEST_BUCKET)
-        .key(test_key)
-        .send()
-        .await?;
+    let initiate_response = client.create_multipart_upload().bucket(BUCKET).key(&test_key).send().await?;
 
     let upload_id = initiate_response
         .upload_id()
@@ -213,13 +258,12 @@ async fn test_conditional_multi_part_upload() -> TestResult {
 
     // Upload parts
     for part_number in 1..=num_parts {
-        let part_data = vec![u8::try_from(part_number)?; part_size];
-        expected_data.extend_from_slice(&part_data);
+        let part_data = generate_test_data(part_size);
 
         let upload_part_response = client
             .upload_part()
-            .bucket(TEST_BUCKET)
-            .key(test_key)
+            .bucket(BUCKET)
+            .key(&test_key)
             .upload_id(upload_id)
             .part_number(part_number)
             .body(Bytes::from(part_data).into())
@@ -242,62 +286,57 @@ async fn test_conditional_multi_part_upload() -> TestResult {
     // Test 1: Multipart upload with wildcard If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match("*")
         .send()
         .await;
 
-    assert_s3_error_code(complete_response, "PreconditionFailed");
+    assert!(complete_response.is_err());
 
     // Test 2: Multipart upload with matching If-None-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_none_match(initial_etag.clone())
         .send()
         .await;
 
-    assert_s3_error_code(complete_response, "PreconditionFailed");
+    assert!(complete_response.is_err());
 
     // Test 3: Multipart upload with unmatching If-Match, should fail
     let complete_response = client
         .complete_multipart_upload()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .upload_id(upload_id)
         .multipart_upload(completed_upload.clone())
         .if_match("\"abcdef\"")
         .send()
         .await;
 
-    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)?);
+    assert!(complete_response.is_err());
 
     // Test 4: Multipart upload with matching If-Match, should succeed
-    client
+    let complete_response = client
         .complete_multipart_upload()
-        .bucket(TEST_BUCKET)
-        .key(test_key)
+        .bucket(BUCKET)
+        .key(&test_key)
         .upload_id(upload_id)
-        .multipart_upload(completed_upload)
+        .multipart_upload(completed_upload.clone())
         .if_match(initial_etag)
         .send()
-        .await?;
-    assert_eq!(object_body(&client, test_key).await?.as_ref(), expected_data);
+        .await;
+
+    assert!(complete_response.is_ok());
+
+    // Cleanup
+    cleanup_objects(&client, BUCKET, &[&test_key]).await;
 
     Ok(())
 }

crates/ecstore/src/api/mod.rs

@@ -161,7 +161,7 @@ pub mod bucket {
         pub mod objectlock_sys {
             pub use crate::bucket::object_lock::objectlock_sys::{
                 BucketObjectLockSys, ObjectLockBlockReason, add_years, check_object_lock_for_deletion,
-                check_retention_for_modification, is_retention_active, replication_write_may_pass_worm_gate,
+                check_retention_for_modification, is_retention_active,
             };
         }
     }
@@ -243,8 +243,8 @@ pub mod cache {
 
 pub mod capacity {
     pub use crate::core::pools::{
-        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
-        path2_bucket_object, path2_bucket_object_with_base_path,
+        PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free, path2_bucket_object,
+        path2_bucket_object_with_base_path,
     };
     pub use crate::store::utils::is_reserved_or_invalid_bucket;
 }
@@ -456,15 +456,15 @@ pub mod rpc {
         AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
         PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
         ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer,
-        check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers,
-        gen_tonic_replay_scope_headers, gen_tonic_signature_headers, gen_tonic_signature_interceptor,
-        node_service_time_out_client, node_service_time_out_client_no_auth, normalize_tonic_rpc_audience,
-        set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_ns_scanner_capability_with_tier_registry_generation,
-        sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
-        tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_ns_scanner_capability_with_tier_registry_generation,
-        verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response,
-        verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof,
-        verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
+        check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
+        gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth,
+        normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability,
+        sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
+        tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
+        verify_ns_scanner_capability, verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer,
+        verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
+        verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
+        verify_tonic_rpc_signature_with_bootstrap,
     };
 }
 

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -15,7 +15,7 @@
 use crate::bucket::metadata_sys::{ObjectLockConfigState, get_object_lock_config, get_object_lock_config_state};
 use crate::bucket::object_lock::objectlock;
 use crate::error::{Error, Result, StorageError};
-use crate::object_api::{ObjectInfo, ObjectOptions};
+use crate::object_api::ObjectInfo;
 use s3s::dto::{Date, DefaultRetention, ObjectLockConfiguration, ObjectLockLegalHoldStatus, ObjectLockRetentionMode};
 use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
 use std::sync::Arc;
@@ -136,50 +136,12 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
     let lhold = objectlock::get_object_legalhold_meta(user_defined);
     matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
 }
 
-/// Whether an authorized replication write (`ObjectOptions::replication_request`)
-/// may overwrite a locked destination version.
-///
-/// The source's lock state governs a replica (MinIO `checkPutObjectLockAllowed`
-/// skips the existing-version check for replicas), and a source-side hold
-/// release or retention change reaches this site only through this write. The
-/// overwrite is allowed only when the write carries the source timestamp of
-/// every category that currently locks the version, so receiver-side LWW
-/// (`merge_replication_metadata_lww`) judges each of them: a category locked
-/// more recently here is kept, otherwise the source's newer state wins. A write
-/// without that timestamp carries no source decision for the category — the
-/// metadata replace would lift the lock unjudged — so it stays WORM-rejected.
-///
-/// The locking categories come from the same authoritative evaluation as the
-/// commit-time WORM gate (`check_object_lock_for_deletion_with_state`): the
-/// bucket default retention locks a version that carries no explicit
-/// retention keys, so it is judged here too rather than read off the keys.
-/// Malformed persisted lock metadata or a non-authoritative bucket
-/// configuration is an error, never a pass.
-pub fn replication_write_may_pass_worm_gate(
-    state: &ObjectLockConfigState,
-    obj_info: &ObjectInfo,
-    opts: &ObjectOptions,
-) -> Result<bool> {
-    if !opts.replication_request {
-        return Ok(false);
-    }
-    if obj_info.delete_marker {
-        // Delete markers are never locked (same as the WORM gate).
-        return Ok(true);
-    }
-    let config = object_lock_config_from_state(state)?;
-    if legal_hold_locks(obj_info)? && opts.replication_legalhold_timestamp.is_none() {
-        return Ok(false);
-    }
-    let retention_locked = active_retention(config, obj_info)?.is_some();
-    Ok(!(retention_locked && opts.replication_retention_timestamp.is_none()))
-}
-
 /// Check if an object is locked based on its metadata.
 /// This is a common function used by both lifecycle evaluation and deletion checks.
 ///
@@ -277,101 +239,69 @@ pub(crate) fn check_object_lock_for_deletion_with_config(
         return Ok(None);
     }
 
-    if legal_hold_locks(obj_info)? {
-        return Ok(Some(ObjectLockBlockReason::LegalHold));
+    if let Some(status) = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) {
+        if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) {
+            return Ok(Some(ObjectLockBlockReason::LegalHold));
+        }
+        if !status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::OFF) {
+            return Err(Error::other("persisted object legal-hold metadata is invalid"));
+        }
     }
 
-    if let Some((mode_str, retain_until)) = active_retention(config, obj_info)?
-        && let Some(reason) = check_retention_blocks_deletion(mode_str, Some(retain_until), bypass_governance)
-    {
-        return Ok(Some(reason));
-    }
-
-    Ok(None)
-}
-
-/// A cleared retention / legal hold is persisted as empty strings (the MinIO
-/// on-disk shape, `parse_object_lock_retention`); read it as "no lock" rather
-/// than as corrupt metadata.
-fn persisted_lock_value<'a>(obj_info: &'a ObjectInfo, key: &str) -> Option<&'a String> {
-    obj_info.user_defined.get(key).filter(|value| !value.is_empty())
-}
-
-/// Whether the version's persisted legal hold is ON. Any other non-empty
-/// value than ON/OFF is malformed metadata and fails closed.
-fn legal_hold_locks(obj_info: &ObjectInfo) -> Result<bool> {
-    let Some(status) = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) else {
-        return Ok(false);
-    };
-    if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) {
-        return Ok(true);
-    }
-    if !status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::OFF) {
-        return Err(Error::other("persisted object legal-hold metadata is invalid"));
-    }
-    Ok(false)
-}
-
-/// The retention that currently locks the version, if any: the explicit
-/// persisted retention when the keys are present, otherwise the bucket
-/// default retention computed from the version's modification time. Returns
-/// `(mode, retain_until)` only while the retention is still active.
-fn active_retention<'a>(
-    config: Option<&'a ObjectLockConfiguration>,
-    obj_info: &ObjectInfo,
-) -> Result<Option<(&'a str, OffsetDateTime)>> {
-    let mode = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_MODE.as_str());
-    let retain_until = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
-    match (mode, retain_until) {
-        (None, None) => {}
+    let mode = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_MODE.as_str());
+    let retain_until = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
+    let explicit_ret = match (mode, retain_until) {
+        (None, None) => None,
         (Some(mode), Some(retain_until)) => {
             let mode =
                 objectlock::parse_ret_mode(mode).ok_or_else(|| Error::other("persisted object retention mode is invalid"))?;
             let retain_until = OffsetDateTime::parse(retain_until, &time::format_description::well_known::Iso8601::DEFAULT)
                 .map(Date::from)
                 .map_err(|_| Error::other("persisted object retention date is invalid"))?;
-            let mode_str = match mode.as_str() {
-                ObjectLockRetentionMode::COMPLIANCE => ObjectLockRetentionMode::COMPLIANCE,
-                ObjectLockRetentionMode::GOVERNANCE => ObjectLockRetentionMode::GOVERNANCE,
-                _ => return Err(Error::other("persisted object retention mode is invalid")),
-            };
-            return Ok(is_retention_active(mode_str, Some(&retain_until)).then(|| (mode_str, OffsetDateTime::from(retain_until))));
+            Some((mode, retain_until))
         }
         _ => return Err(Error::other("persisted object retention metadata is incomplete")),
+    };
+
+    if let Some((mode, retain_until)) = &explicit_ret {
+        let mode_str = mode.as_str();
+        if is_retention_active(mode_str, Some(retain_until))
+            && let Some(reason) =
+                check_retention_blocks_deletion(mode_str, Some(OffsetDateTime::from(retain_until.clone())), bypass_governance)
+        {
+            return Ok(Some(reason));
+        }
     }
 
-    let Some(default_retention) = config.and_then(|config| config.rule.as_ref()?.default_retention.as_ref()) else {
-        return Ok(None);
-    };
-    let Some(mode) = &default_retention.mode else {
-        return Ok(None);
-    };
-    let mode_str = mode.as_str();
-    if mode_str != ObjectLockRetentionMode::COMPLIANCE && mode_str != ObjectLockRetentionMode::GOVERNANCE {
-        return Ok(None);
-    }
-    // Calculate retention expiration date from object modification time
-    let mod_time = obj_info
-        .mod_time
-        .ok_or_else(|| Error::other("persisted object modification time is missing"))?;
-    let now = objectlock::utc_now_ntp();
-    let retain_until = if let Some(days) = default_retention.days {
-        mod_time.saturating_add(time::Duration::days(i64::from(days)))
-    } else {
-        let years = default_retention
-            .years
-            .ok_or_else(|| Error::other("persisted bucket Object Lock retention period is invalid"))?;
-        add_years(mod_time, years)
-    };
-    Ok((retain_until.unix_timestamp() > now.unix_timestamp()).then_some((mode_str, retain_until)))
-}
+    if explicit_ret.is_none()
+        && let Some(default_retention) = config.and_then(|config| config.rule.as_ref()?.default_retention.as_ref())
+        && let Some(mode) = &default_retention.mode
+    {
+        let mode_str = mode.as_str();
+        if mode_str == ObjectLockRetentionMode::COMPLIANCE || mode_str == ObjectLockRetentionMode::GOVERNANCE {
+            // Calculate retention expiration date from object modification time
+            let mod_time = obj_info
+                .mod_time
+                .ok_or_else(|| Error::other("persisted object modification time is missing"))?;
+            let now = objectlock::utc_now_ntp();
+            let retain_until = if let Some(days) = default_retention.days {
+                mod_time.saturating_add(time::Duration::days(i64::from(days)))
+            } else {
+                let years = default_retention
+                    .years
+                    .ok_or_else(|| Error::other("persisted bucket Object Lock retention period is invalid"))?;
+                add_years(mod_time, years)
+            };
 
-fn object_lock_config_from_state(state: &ObjectLockConfigState) -> Result<Option<&ObjectLockConfiguration>> {
-    match state {
-        ObjectLockConfigState::Configured { config, .. } => Ok(Some(config)),
-        ObjectLockConfigState::ConfirmedAbsent => Ok(None),
-        ObjectLockConfigState::Fabricated => Err(Error::other("bucket Object Lock metadata is not authoritative")),
+            if retain_until.unix_timestamp() > now.unix_timestamp()
+                && let Some(reason) = check_retention_blocks_deletion(mode_str, Some(retain_until), bypass_governance)
+            {
+                return Ok(Some(reason));
+            }
+        }
     }
+
+    Ok(None)
 }
 
 pub(crate) fn check_object_lock_for_deletion_with_state(
@@ -379,7 +309,13 @@ pub(crate) fn check_object_lock_for_deletion_with_state(
     obj_info: &ObjectInfo,
     bypass_governance: bool,
 ) -> Result<Option<ObjectLockBlockReason>> {
-    check_object_lock_for_deletion_with_config(object_lock_config_from_state(state)?, obj_info, bypass_governance)
+    match state {
+        ObjectLockConfigState::Configured { config, .. } => {
+            check_object_lock_for_deletion_with_config(Some(config), obj_info, bypass_governance)
+        }
+        ObjectLockConfigState::ConfirmedAbsent => check_object_lock_for_deletion_with_config(None, obj_info, bypass_governance),
+        ObjectLockConfigState::Fabricated => Err(Error::other("bucket Object Lock metadata is not authoritative")),
+    }
 }
 
 /// Compatibility wrapper for callers that predate fallible metadata lookup.
@@ -550,210 +486,6 @@ mod tests {
         }
     }
 
-    fn replication_opts(hold_ts: bool, retention_ts: bool) -> ObjectOptions {
-        ObjectOptions {
-            replication_request: true,
-            replication_legalhold_timestamp: hold_ts.then_some(OffsetDateTime::UNIX_EPOCH),
-            replication_retention_timestamp: retention_ts.then_some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        }
-    }
-
-    fn lock_metadata(entries: &[&[(&str, &str)]]) -> std::collections::HashMap<String, String> {
-        entries
-            .iter()
-            .flat_map(|entries| entries.iter())
-            .map(|(key, value)| (key.to_string(), value.to_string()))
-            .collect()
-    }
-
-    fn lock_object_info(user_defined: std::collections::HashMap<String, String>) -> ObjectInfo {
-        ObjectInfo {
-            user_defined: Arc::new(user_defined),
-            mod_time: Some(OffsetDateTime::now_utc()),
-            ..Default::default()
-        }
-    }
-
-    /// A replication write passes the WORM gate only when it carries the
-    /// source timestamp of every category that currently locks the version.
-    #[test]
-    fn replication_write_passes_worm_gate_only_with_every_locking_category_timestamp() {
-        use rustfs_utils::http::headers::{
-            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
-        };
-
-        let hold = [(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "ON")];
-        let retention = [
-            (AMZ_OBJECT_LOCK_MODE_LOWER, "GOVERNANCE"),
-            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, "2099-01-01T00:00:00Z"),
-        ];
-        let expired = [
-            (AMZ_OBJECT_LOCK_MODE_LOWER, "COMPLIANCE"),
-            (AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, "2000-01-01T00:00:00Z"),
-        ];
-        let absent = ObjectLockConfigState::ConfirmedAbsent;
-        let passes = |state: &ObjectLockConfigState, entries: &[&[(&str, &str)]], opts: &ObjectOptions| {
-            replication_write_may_pass_worm_gate(state, &lock_object_info(lock_metadata(entries)), opts)
-                .expect("well-formed lock metadata must be judged")
-        };
-
-        assert!(passes(&absent, &[&hold, &retention], &replication_opts(true, true)));
-        assert!(!passes(&absent, &[&hold, &retention], &replication_opts(true, false)));
-        assert!(!passes(&absent, &[&hold, &retention], &replication_opts(false, true)));
-
-        assert!(passes(&absent, &[&hold], &replication_opts(true, false)));
-        assert!(!passes(&absent, &[&hold], &replication_opts(false, true)));
-        assert!(passes(&absent, &[&retention], &replication_opts(false, true)));
-        assert!(!passes(&absent, &[&retention], &replication_opts(true, false)));
-
-        // Expired retention and a released hold no longer lock anything.
-        assert!(passes(
-            &absent,
-            &[&expired, &[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "OFF")]],
-            &replication_opts(false, false)
-        ));
-
-        // Never for a non-replication write, whatever it carries.
-        let local = ObjectOptions {
-            replication_request: false,
-            ..replication_opts(true, true)
-        };
-        assert!(!passes(&absent, &[&hold], &local));
-    }
-
-    /// The bucket default retention locks a version that carries no explicit
-    /// retention keys (`check_object_lock_for_deletion_with_config` judges it
-    /// from the modification time), so the replication bypass must demand the
-    /// retention source timestamp for it too — a tagging-only replication
-    /// write must not overwrite the default-protected version unjudged.
-    #[test]
-    fn replication_write_under_bucket_default_retention_requires_retention_timestamp() {
-        use rustfs_utils::http::headers::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER};
-
-        for mode in [ObjectLockRetentionMode::COMPLIANCE, ObjectLockRetentionMode::GOVERNANCE] {
-            let state = ObjectLockConfigState::Configured {
-                config: default_retention_config(mode),
-                updated_at: OffsetDateTime::now_utc(),
-            };
-            let no_keys = lock_object_info(std::collections::HashMap::new());
-            assert!(
-                check_object_lock_for_deletion_with_state(&state, &no_keys, false)
-                    .expect("default retention must be judged")
-                    .is_some(),
-                "{mode}: the gate must report the default retention lock"
-            );
-
-            let tagging_only = ObjectOptions {
-                replication_request: true,
-                replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-                ..Default::default()
-            };
-            assert!(
-                !replication_write_may_pass_worm_gate(&state, &no_keys, &tagging_only).expect("judged"),
-                "{mode}: a tagging-only replication write must not pass the default retention lock"
-            );
-            assert!(
-                replication_write_may_pass_worm_gate(&state, &no_keys, &replication_opts(false, true)).expect("judged"),
-                "{mode}: the retention source timestamp lets LWW judge the default retention"
-            );
-
-            // Default retention plus a legal hold: both categories need a timestamp.
-            let held = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "ON")]]));
-            assert!(!replication_write_may_pass_worm_gate(&state, &held, &replication_opts(false, true)).expect("judged"));
-            assert!(!replication_write_may_pass_worm_gate(&state, &held, &replication_opts(true, false)).expect("judged"));
-            assert!(replication_write_may_pass_worm_gate(&state, &held, &replication_opts(true, true)).expect("judged"));
-
-            // A version whose default retention has already expired (old
-            // mod_time) is not locked by the default any more.
-            let expired_default = ObjectInfo {
-                mod_time: Some(make_datetime(2000, 1, 1)),
-                ..lock_object_info(std::collections::HashMap::new())
-            };
-            assert!(replication_write_may_pass_worm_gate(&state, &expired_default, &tagging_only).expect("judged"));
-
-            // A delete marker is never locked, so there is nothing to judge.
-            let delete_marker = ObjectInfo {
-                delete_marker: true,
-                ..lock_object_info(std::collections::HashMap::new())
-            };
-            assert!(replication_write_may_pass_worm_gate(&state, &delete_marker, &tagging_only).expect("judged"));
-
-            // Cleared (empty) explicit keys fall back to the bucket default.
-            let cleared = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_MODE_LOWER, "")]]));
-            assert!(!replication_write_may_pass_worm_gate(&state, &cleared, &tagging_only).expect("judged"));
-        }
-    }
-
-    /// The replication bypass never judges from a non-authoritative bucket
-    /// state or malformed persisted lock metadata; both are errors, not a pass.
-    #[test]
-    fn replication_write_worm_gate_fails_closed_on_unverifiable_lock_state() {
-        use rustfs_utils::http::headers::AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER;
-
-        let opts = replication_opts(true, true);
-        let err = replication_write_may_pass_worm_gate(
-            &ObjectLockConfigState::Fabricated,
-            &lock_object_info(std::collections::HashMap::new()),
-            &opts,
-        )
-        .expect_err("fabricated bucket lock metadata must not be judged");
-        assert!(err.to_string().contains("not authoritative"));
-
-        let malformed = lock_object_info(lock_metadata(&[&[(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "MAYBE")]]));
-        let err = replication_write_may_pass_worm_gate(&ObjectLockConfigState::ConfirmedAbsent, &malformed, &opts)
-            .expect_err("malformed legal hold must not be judged");
-        assert!(err.to_string().contains("legal-hold"));
-
-        let state = ObjectLockConfigState::Configured {
-            config: default_retention_config(ObjectLockRetentionMode::COMPLIANCE),
-            updated_at: OffsetDateTime::now_utc(),
-        };
-        let no_mod_time = ObjectInfo::default();
-        let err = replication_write_may_pass_worm_gate(&state, &no_mod_time, &opts)
-            .expect_err("default retention without a modification time must not be judged");
-        assert!(err.to_string().contains("modification time"));
-    }
-
-    /// A local PutObjectRetention / PutObjectLegalHold "clear" persists the
-    /// lock keys as empty strings (the MinIO on-disk shape, see
-    /// `parse_object_lock_retention`); that is "no lock", not corruption, and
-    /// must not wedge later explicit-version PUTs or deletes
-    /// (rustfs/backlog#1953).
-    #[test]
-    fn deletion_treats_cleared_empty_lock_metadata_as_unlocked() {
-        use rustfs_utils::http::headers::{
-            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
-        };
-
-        let cases: [(&str, &[&str]); 3] = [
-            (
-                "cleared retention",
-                &[AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER],
-            ),
-            ("cleared legal hold", &[AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER]),
-            (
-                "all cleared",
-                &[
-                    AMZ_OBJECT_LOCK_MODE_LOWER,
-                    AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
-                    AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER,
-                ],
-            ),
-        ];
-
-        for (case, keys) in cases {
-            let user_defined = keys.iter().map(|key| (key.to_string(), String::new())).collect();
-            let obj_info = ObjectInfo {
-                user_defined: Arc::new(user_defined),
-                ..Default::default()
-            };
-
-            let result = check_object_lock_for_deletion_with_config(None, &obj_info, false);
-            assert!(matches!(result, Ok(None)), "{case}: empty lock keys must read as unlocked: {result:?}");
-        }
-    }
-
     #[test]
     fn deletion_rejects_invalid_persisted_legal_hold_metadata() {
         let mut user_defined = std::collections::HashMap::new();

crates/ecstore/src/cluster/rpc/mod.rs

@@ -51,9 +51,6 @@ pub use peer_rest_client::{
     SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity,
 };
 pub(crate) use peer_s3_client::heal_bucket_local_on_disks;
-pub use peer_s3_client::{
-    LocalPeerS3Client, PeerS3Client, S3PeerSys, ScannerBucketListing, ScannerSetBucketListing, decode_heal_bucket_rpc_options,
-    encode_heal_bucket_rpc_options,
-};
+pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, S3PeerSys, ScannerBucketListing, ScannerSetBucketListing};
 pub use remote_disk::RemoteDisk;
 pub use remote_locker::RemoteClient;

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -18,7 +18,6 @@ use crate::cluster::rpc::client::{
     node_service_time_out_client,
 };
 use crate::cluster::rpc::set_tonic_mutation_body_digest;
-use crate::core::pools::PoolMeta;
 use crate::disk::error::DiskError;
 use crate::disk::error::{Error, Result};
 use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
@@ -47,12 +46,7 @@ use std::sync::{
     Mutex as StdMutex,
     atomic::{AtomicBool, Ordering},
 };
-use std::{
-    collections::{BTreeSet, HashMap},
-    fmt::Debug,
-    sync::Arc,
-    time::Duration,
-};
+use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration};
 #[cfg(test)]
 use tokio::sync::Notify;
 use tokio::{net::TcpStream, sync::RwLock, time};
@@ -105,9 +99,6 @@ impl DeleteBucketEmptyScanBarrier {
 #[cfg(test)]
 static DELETE_BUCKET_EMPTY_SCAN_BARRIER: StdMutex<Option<Arc<DeleteBucketEmptyScanBarrier>>> = StdMutex::new(None);
 
-#[cfg(test)]
-static HEAL_BUCKET_PRE_MUTATION_BARRIER: StdMutex<Option<Arc<DeleteBucketEmptyScanBarrier>>> = StdMutex::new(None);
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 enum HealBucketOperation {
     Make,
@@ -180,15 +171,6 @@ pub(crate) fn install_delete_bucket_empty_scan_barrier() -> Arc<DeleteBucketEmpt
     barrier
 }
 
-#[cfg(test)]
-fn install_heal_bucket_pre_mutation_barrier() -> Arc<DeleteBucketEmptyScanBarrier> {
-    let barrier = Arc::new(DeleteBucketEmptyScanBarrier::default());
-    *HEAL_BUCKET_PRE_MUTATION_BARRIER
-        .lock()
-        .expect("heal bucket mutation barrier lock should not be poisoned") = Some(barrier.clone());
-    barrier
-}
-
 #[cfg(test)]
 async fn pause_after_delete_bucket_empty_scan() {
     let barrier = DELETE_BUCKET_EMPTY_SCAN_BARRIER
@@ -200,20 +182,6 @@ async fn pause_after_delete_bucket_empty_scan() {
     }
 }
 
-#[cfg(test)]
-async fn pause_before_heal_bucket_volume_mutation() {
-    let barrier = HEAL_BUCKET_PRE_MUTATION_BARRIER
-        .lock()
-        .expect("heal bucket mutation barrier lock should not be poisoned")
-        .take();
-    if let Some(barrier) = barrier {
-        barrier.pause().await;
-    }
-}
-
-#[cfg(not(test))]
-async fn pause_before_heal_bucket_volume_mutation() {}
-
 #[derive(Clone, Debug)]
 pub struct ScannerBucketListing {
     pub buckets: Vec<BucketInfo>,
@@ -285,45 +253,9 @@ fn resolve_heal_bucket_mode(opts: &mut HealOpts, pool_errs: &[Option<Error>]) ->
     Ok(())
 }
 
-#[derive(serde::Serialize, serde::Deserialize)]
-#[serde(deny_unknown_fields)]
-struct HealBucketRpcEnvelope {
-    options: HealOpts,
-    #[serde(rename = "fencedPools", default)]
-    fenced_pools: Vec<usize>,
-}
-
-pub fn encode_heal_bucket_rpc_options(opts: HealOpts, fenced_pools: &[usize]) -> Result<String> {
-    if fenced_pools.is_empty() {
-        return serde_json::to_string(&opts).map_err(Into::into);
-    }
-
-    serde_json::to_string(&HealBucketRpcEnvelope {
-        options: opts,
-        fenced_pools: fenced_pools.to_vec(),
-    })
-    .map_err(Into::into)
-}
-
-pub fn decode_heal_bucket_rpc_options(payload: &str) -> Result<(HealOpts, Vec<usize>)> {
-    match serde_json::from_str::<HealBucketRpcEnvelope>(payload) {
-        Ok(envelope) => Ok((envelope.options, envelope.fenced_pools)),
-        Err(envelope_err) => serde_json::from_str::<HealOpts>(payload)
-            .map(|options| (options, Vec::new()))
-            .map_err(|legacy_err| {
-                Error::other(format!(
-                    "decode heal bucket RPC options failed: envelope={envelope_err}; legacy={legacy_err}"
-                ))
-            }),
-    }
-}
-
 #[async_trait]
 pub trait PeerS3Client: Debug + Sync + Send + 'static {
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem>;
-    async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, _fenced_pools: &[usize]) -> Result<HealResultItem> {
-        self.heal_bucket(bucket, opts).await
-    }
     async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
     async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
     async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
@@ -377,10 +309,6 @@ impl S3PeerSys {
 
 impl S3PeerSys {
     pub async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
-        self.heal_bucket_with_fence(bucket, opts, &[]).await
-    }
-
-    pub async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result<HealResultItem> {
         let mut opts = *opts;
         let mut futures = Vec::with_capacity(self.clients.len());
         for client in self.clients.iter() {
@@ -403,7 +331,7 @@ impl S3PeerSys {
             let opts_clone = opts;
             let heal_bucket_results_clone = heal_bucket_results.clone();
             futures.push(async move {
-                match client.heal_bucket_with_fence(bucket, &opts_clone, fenced_pools).await {
+                match client.heal_bucket(bucket, &opts_clone).await {
                     Ok(res) => {
                         heal_bucket_results_clone.write().await[idx] = res;
                         None
@@ -707,18 +635,8 @@ impl PeerS3Client for LocalPeerS3Client {
     }
 
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
-        self.heal_bucket_with_fence(bucket, opts, &[]).await
-    }
-
-    async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result<HealResultItem> {
         let disks = self.local_disks_for_pools().await.into_iter().map(Some).collect();
-        let store = runtime_sources::object_store_handle().filter(|store| Arc::ptr_eq(&store.ctx, &self.instance_ctx));
-        #[cfg(not(test))]
-        if store.is_none() {
-            return Err(Error::other("bucket heal refused: pool metadata is unavailable for this instance"));
-        }
-        heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, store.as_ref().map(|store| &store.pool_meta), fenced_pools)
-            .await
+        heal_bucket_local_on_disks(bucket, opts, disks).await
     }
 
     async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
@@ -1161,13 +1079,9 @@ impl PeerS3Client for RemotePeerS3Client {
     }
 
     async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
-        self.heal_bucket_with_fence(bucket, opts, &[]).await
-    }
-
-    async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result<HealResultItem> {
         self.execute_with_timeout(
             || async {
-                let options = encode_heal_bucket_rpc_options(*opts, fenced_pools)?;
+                let options: String = serde_json::to_string(opts)?;
                 let mut client = self.get_client().await?;
                 let mut request = Request::new(HealBucketRequest {
                     bucket: bucket.to_string(),
@@ -1315,117 +1229,6 @@ pub(crate) async fn heal_bucket_local_on_disks(
     opts: &HealOpts,
     disks: Vec<Option<DiskStore>>,
 ) -> Result<HealResultItem> {
-    if let Some(store) = runtime_sources::object_store_handle() {
-        return heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, Some(&store.pool_meta), &[]).await;
-    }
-
-    #[cfg(test)]
-    return heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, None, &[]).await;
-
-    #[cfg(not(test))]
-    Err(Error::other("bucket heal refused: pool metadata is unavailable"))
-}
-
-fn disk_pool_index(disk: &DiskStore) -> Result<usize> {
-    usize::try_from(disk.endpoint().pool_idx)
-        .map_err(|_| Error::other(format!("invalid bucket-heal pool index {}", disk.endpoint().pool_idx)))
-}
-
-fn fenced_decommission_drive_state() -> DriveState {
-    DriveState::Unknown("skipped-decommission-suspended".to_string())
-}
-
-fn heal_bucket_fence_detail(fenced_pools: &BTreeSet<usize>) -> Option<String> {
-    if fenced_pools.is_empty() {
-        return None;
-    }
-    let pools = fenced_pools.iter().map(usize::to_string).collect::<Vec<_>>().join(", ");
-    Some(format!("skipped: bucket-volume heal fenced on decommission-suspended pool(s): {pools}"))
-}
-
-async fn snapshot_heal_bucket_fence(
-    disks: &[Option<DiskStore>],
-    pool_meta: Option<&RwLock<PoolMeta>>,
-    dispatch_fenced_pools: &[usize],
-) -> Result<(Vec<bool>, BTreeSet<usize>)> {
-    let mut fenced_disks = vec![false; disks.len()];
-    let mut fenced_pools = dispatch_fenced_pools.iter().copied().collect::<BTreeSet<_>>();
-    let pool_meta = match pool_meta {
-        Some(pool_meta) => Some(pool_meta.read().await),
-        None => None,
-    };
-    if let Some(pool_meta) = pool_meta.as_ref()
-        && let Some(pool_idx) = fenced_pools.iter().find(|pool_idx| **pool_idx >= pool_meta.pools.len())
-    {
-        return Err(Error::other(format!(
-            "bucket-heal dispatch fence pool index {pool_idx} is absent from {} pool metadata entries",
-            pool_meta.pools.len()
-        )));
-    }
-
-    for (disk_index, disk) in disks.iter().enumerate() {
-        let Some(disk) = disk else {
-            continue;
-        };
-        let pool_idx = disk_pool_index(disk)?;
-        if let Some(pool_meta) = pool_meta.as_ref() {
-            if pool_idx >= pool_meta.pools.len() {
-                return Err(Error::other(format!(
-                    "bucket-heal pool index {pool_idx} is absent from {} pool metadata entries",
-                    pool_meta.pools.len()
-                )));
-            }
-            if pool_meta.is_suspended(pool_idx) {
-                fenced_pools.insert(pool_idx);
-            }
-        }
-        if fenced_pools.contains(&pool_idx) {
-            fenced_disks[disk_index] = true;
-        }
-    }
-    Ok((fenced_disks, fenced_pools))
-}
-
-async fn run_heal_bucket_volume_mutation<F, Fut>(
-    disk: &DiskStore,
-    pool_meta: Option<&RwLock<PoolMeta>>,
-    operation: F,
-) -> Result<Option<usize>>
-where
-    F: FnOnce() -> Fut,
-    Fut: std::future::Future<Output = Result<()>>,
-{
-    let Some(pool_meta) = pool_meta else {
-        operation().await?;
-        return Ok(None);
-    };
-    let pool_idx = disk_pool_index(disk)?;
-    let pool_meta = pool_meta.read().await;
-    if pool_idx >= pool_meta.pools.len() {
-        return Err(Error::other(format!(
-            "bucket-heal pool index {pool_idx} is absent from {} pool metadata entries",
-            pool_meta.pools.len()
-        )));
-    }
-    if pool_meta.is_suspended(pool_idx) {
-        return Ok(Some(pool_idx));
-    }
-
-    // Keep the metadata read guard through the disk mutation so a decommission
-    // transition cannot pass between this state check and the destructive action.
-    operation().await?;
-    Ok(None)
-}
-
-async fn heal_bucket_local_on_disks_with_pool_meta(
-    bucket: &str,
-    opts: &HealOpts,
-    disks: Vec<Option<DiskStore>>,
-    pool_meta: Option<&RwLock<PoolMeta>>,
-    dispatch_fenced_pools: &[usize],
-) -> Result<HealResultItem> {
-    let (fenced_disks, mut fenced_pool_idxs) = snapshot_heal_bucket_fence(&disks, pool_meta, dispatch_fenced_pools).await?;
-    let fenced_disks = Arc::new(fenced_disks);
     let before_state = Arc::new(RwLock::new(vec![String::new(); disks.len()]));
     let after_state = Arc::new(RwLock::new(vec![String::new(); disks.len()]));
 
@@ -1435,14 +1238,7 @@ async fn heal_bucket_local_on_disks_with_pool_meta(
         let bucket = bucket.to_string();
         let bs_clone = before_state.clone();
         let as_clone = after_state.clone();
-        let fenced_disks = fenced_disks.clone();
         futures.push(async move {
-            if fenced_disks[index] {
-                let skipped = fenced_decommission_drive_state().to_string();
-                bs_clone.write().await[index] = skipped.clone();
-                as_clone.write().await[index] = skipped;
-                return None;
-            }
             let disk = match disk {
                 Some(disk) => disk,
                 None => {
@@ -1505,14 +1301,9 @@ async fn heal_bucket_local_on_disks_with_pool_meta(
                 state: state.to_string(),
             });
         }
-        if let Some(detail) = heal_bucket_fence_detail(&fenced_pool_idxs) {
-            res.detail = detail;
-        }
         return Ok(res);
     }
 
-    pause_before_heal_bucket_volume_mutation().await;
-
     let mut operation_error = errs
         .iter()
         .filter_map(|err| match err {
@@ -1524,35 +1315,26 @@ async fn heal_bucket_local_on_disks_with_pool_meta(
     if opts.remove && !bucket.starts_with(disk::RUSTFS_META_BUCKET) && !is_all_buckets_not_found(&errs) {
         let mut futures = Vec::new();
         for (index, disk) in disks.iter().enumerate() {
-            if fenced_disks[index] || matches!(errs[index].as_ref(), Some(Error::DiskNotFound | Error::VolumeNotFound)) {
+            if matches!(errs[index].as_ref(), Some(Error::DiskNotFound | Error::VolumeNotFound)) {
                 continue;
             }
             let Some(disk) = disk.clone() else {
                 continue;
             };
             let bucket = bucket.to_string();
-            let mutation_disk = disk.clone();
             futures.push(async move {
-                let result = run_heal_bucket_volume_mutation(&disk, pool_meta, || async move {
-                    if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) {
-                        return Err(err);
-                    }
-                    mutation_disk.delete_volume(&bucket, false).await
-                })
-                .await;
-                (index, result)
+                if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) {
+                    return (index, Err(err));
+                }
+                (index, disk.delete_volume(&bucket, false).await)
             });
         }
 
         for (index, result) in join_all(futures).await {
             match result {
-                Ok(None) | Err(Error::VolumeNotFound) => {
+                Ok(()) | Err(Error::VolumeNotFound) => {
                     after_state.write().await[index] = DriveState::Missing.to_string();
                 }
-                Ok(Some(pool_idx)) => {
-                    fenced_pool_idxs.insert(pool_idx);
-                    after_state.write().await[index] = fenced_decommission_drive_state().to_string();
-                }
                 Err(Error::VolumeNotEmpty) => {
                     warn!(
                         bucket,
@@ -1583,38 +1365,30 @@ async fn heal_bucket_local_on_disks_with_pool_meta(
             let bs_clone = before_state.clone();
             futures.push(async move {
                 if bs_clone.read().await[idx] == DriveState::Missing.to_string() {
-                    let Some(disk) = disk else {
-                        return (idx, Err(Error::DiskNotFound));
+                    let Some(disk) = disk.as_ref() else {
+                        return (idx, Some(Error::DiskNotFound));
                     };
-                    let mutation_disk = disk.clone();
-                    let result = run_heal_bucket_volume_mutation(&disk, pool_meta, || async move {
-                        if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) {
-                            return Err(err);
-                        }
-                        match mutation_disk.make_volume(&bucket).await {
-                            Ok(()) | Err(Error::VolumeExists) => Ok(()),
-                            Err(err) => Err(err),
-                        }
-                    })
-                    .await;
-                    return (idx, result);
+
+                    if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) {
+                        return (idx, Some(err));
+                    }
+                    match disk.make_volume(&bucket).await {
+                        Ok(()) | Err(Error::VolumeExists) => return (idx, None),
+                        Err(err) => return (idx, Some(err)),
+                    }
                 }
-                (idx, Ok(None))
+                (idx, None)
             });
         }
 
         for (index, result) in join_all(futures).await {
             match result {
-                Ok(None) => {
+                None => {
                     if before_state.read().await[index] == DriveState::Missing.to_string() {
                         after_state.write().await[index] = DriveState::Ok.to_string();
                     }
                 }
-                Ok(Some(pool_idx)) => {
-                    fenced_pool_idxs.insert(pool_idx);
-                    after_state.write().await[index] = fenced_decommission_drive_state().to_string();
-                }
-                Err(err) => {
+                Some(err) => {
                     after_state.write().await[index] = match &err {
                         Error::DiskNotFound => DriveState::Offline.to_string(),
                         _ => DriveState::Corrupt.to_string(),
@@ -1635,10 +1409,6 @@ async fn heal_bucket_local_on_disks_with_pool_meta(
         });
     }
 
-    if let Some(detail) = heal_bucket_fence_detail(&fenced_pool_idxs) {
-        res.detail = detail;
-    }
-
     match operation_error {
         Some(err) => Err(err),
         None => Ok(res),
@@ -1656,7 +1426,6 @@ async fn clone_drives() -> Vec<Option<DiskStore>> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
     use crate::disk::WalkDirOptions;
     use crate::disk::disk_store::LocalDiskWrapper;
     use crate::disk::endpoint::Endpoint;
@@ -1830,23 +1599,6 @@ mod tests {
         disks
     }
 
-    fn heal_bucket_pool_meta(suspended_pool: Option<usize>) -> PoolMeta {
-        PoolMeta {
-            pools: (0..2)
-                .map(|pool_idx| PoolStatus {
-                    id: pool_idx,
-                    cmd_line: format!("pool-{pool_idx}"),
-                    last_update: ::time::OffsetDateTime::UNIX_EPOCH,
-                    decommission: (suspended_pool == Some(pool_idx)).then(|| PoolDecommissionInfo {
-                        start_time: Some(::time::OffsetDateTime::UNIX_EPOCH),
-                        ..Default::default()
-                    }),
-                })
-                .collect(),
-            ..Default::default()
-        }
-    }
-
     fn test_remote_peer(addr: &str) -> RemotePeerS3Client {
         RemotePeerS3Client {
             pools: Some(vec![0]),
@@ -2158,127 +1910,6 @@ mod tests {
         reset_local_disk_test_state().await;
     }
 
-    #[tokio::test]
-    #[serial]
-    async fn heal_bucket_rechecks_decommission_before_recreating_volume() {
-        reset_local_disk_test_state().await;
-
-        let temp_dir = TempDir::new().expect("create temp dir for bucket-heal fence regression");
-        let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-mutation-fence").await;
-        let bucket = "fenced-recreate-bucket";
-        disks[0]
-            .make_volume(bucket)
-            .await
-            .expect("active pool should start with the bucket volume");
-
-        let pool_meta = Arc::new(RwLock::new(heal_bucket_pool_meta(None)));
-        let barrier = install_heal_bucket_pre_mutation_barrier();
-        let heal = tokio::spawn({
-            let disks = disks.clone();
-            let pool_meta = pool_meta.clone();
-            async move {
-                heal_bucket_local_on_disks_with_pool_meta(
-                    bucket,
-                    &HealOpts {
-                        recreate: true,
-                        ..Default::default()
-                    },
-                    disks.into_iter().map(Some).collect(),
-                    Some(pool_meta.as_ref()),
-                    &[],
-                )
-                .await
-            }
-        });
-
-        barrier.wait_until_paused().await;
-        pool_meta.write().await.pools[1].decommission = Some(PoolDecommissionInfo {
-            start_time: Some(::time::OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        });
-        barrier.release();
-
-        let result = heal
-            .await
-            .expect("bucket-heal task should join")
-            .expect("suspended pool should be reported as skipped");
-        assert!(result.detail.contains("skipped") && result.detail.contains('1'));
-        assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound)));
-
-        reset_local_disk_test_state().await;
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn heal_bucket_dispatch_fence_blocks_stale_active_peer_state() {
-        reset_local_disk_test_state().await;
-
-        let temp_dir = TempDir::new().expect("create temp dir for stale bucket-heal peer regression");
-        let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-dispatch-fence").await;
-        let bucket = "dispatch-fenced-bucket";
-        disks[0]
-            .make_volume(bucket)
-            .await
-            .expect("active pool should start with the bucket volume");
-        let stale_pool_meta = RwLock::new(heal_bucket_pool_meta(None));
-
-        let result = heal_bucket_local_on_disks_with_pool_meta(
-            bucket,
-            &HealOpts {
-                recreate: true,
-                ..Default::default()
-            },
-            disks.iter().cloned().map(Some).collect(),
-            Some(&stale_pool_meta),
-            &[1],
-        )
-        .await
-        .expect("dispatch fence should override stale active peer metadata");
-
-        assert!(result.detail.contains("skipped") && result.detail.contains('1'));
-        assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound)));
-
-        reset_local_disk_test_state().await;
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn heal_bucket_keeps_suspended_pool_volume_on_remove() {
-        reset_local_disk_test_state().await;
-
-        let temp_dir = TempDir::new().expect("create temp dir for bucket-heal delete fence regression");
-        let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-delete-fence").await;
-        let bucket = "fenced-remove-bucket";
-        for disk in &disks {
-            disk.make_volume(bucket)
-                .await
-                .expect("bucket volume should exist before heal");
-        }
-        let pool_meta = RwLock::new(heal_bucket_pool_meta(Some(1)));
-
-        let result = heal_bucket_local_on_disks_with_pool_meta(
-            bucket,
-            &HealOpts {
-                remove: true,
-                ..Default::default()
-            },
-            disks.iter().cloned().map(Some).collect(),
-            Some(&pool_meta),
-            &[],
-        )
-        .await
-        .expect("suspended pool should be skipped during bucket-volume removal");
-
-        assert!(result.detail.contains("skipped") && result.detail.contains('1'));
-        assert!(matches!(disks[0].stat_volume(bucket).await, Err(Error::VolumeNotFound)));
-        disks[1]
-            .stat_volume(bucket)
-            .await
-            .expect("suspended pool bucket volume must not be deleted");
-
-        reset_local_disk_test_state().await;
-    }
-
     #[tokio::test]
     #[serial]
     async fn heal_bucket_local_dry_run_reports_discovered_drive_states() {
@@ -2492,36 +2123,6 @@ mod tests {
         assert!(partial.recreate);
     }
 
-    #[test]
-    fn heal_bucket_rpc_envelope_preserves_legacy_compatibility_fail_closed() {
-        let opts = HealOpts {
-            recreate: true,
-            pool: Some(2),
-            ..Default::default()
-        };
-        let encoded = encode_heal_bucket_rpc_options(opts, &[1, 2]).expect("encode bucket-heal RPC envelope");
-
-        assert!(
-            serde_json::from_str::<HealOpts>(&encoded).is_err(),
-            "an old peer must reject the nested request instead of ignoring its dispatch fence"
-        );
-        let (decoded, fenced_pools) =
-            decode_heal_bucket_rpc_options(&encoded).expect("new peer should decode bucket-heal RPC envelope");
-        assert!(decoded.recreate);
-        assert_eq!(decoded.pool, Some(2));
-        assert_eq!(fenced_pools, vec![1, 2]);
-
-        let legacy = encode_heal_bucket_rpc_options(opts, &[]).expect("encode legacy HealOpts for an unfenced heal");
-        let old_peer_opts = serde_json::from_str::<HealOpts>(&legacy).expect("old peer should decode an unfenced heal request");
-        assert!(old_peer_opts.recreate);
-        assert_eq!(old_peer_opts.pool, Some(2));
-
-        let (decoded, fenced_pools) = decode_heal_bucket_rpc_options(&legacy).expect("new peer should accept a legacy request");
-        assert!(decoded.recreate);
-        assert_eq!(decoded.pool, Some(2));
-        assert!(fenced_pools.is_empty());
-    }
-
     #[tokio::test]
     async fn test_make_bucket_reduces_quorum_by_pool_participants() {
         let peer_sys = S3PeerSys {

crates/ecstore/src/set_disk/mod.rs

@@ -47,7 +47,6 @@ use crate::bucket::metadata_sys;
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::object_lock::objectlock_sys::{
     check_object_lock_for_deletion_with_config, check_object_lock_for_deletion_with_state, check_retention_for_modification,
-    replication_write_may_pass_worm_gate,
 };
 use crate::bucket::replication::{
     ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,

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

@@ -556,69 +556,6 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk::
 }
 
 impl SetDisks {
-    async fn discover_multipart_upload_paths(
-        &self,
-        orig_bucket: &str,
-        error_path: &str,
-    ) -> Result<(Vec<Option<DiskStore>>, Vec<String>, usize)> {
-        let disks = self.disks.read().await.clone();
-        if disks.is_empty() {
-            return Err(Error::ErasureReadQuorum);
-        }
-        let discovery_quorum = if self.default_parity_count == 0 {
-            disks.len()
-        } else {
-            (disks.len() / 2).max(1)
-        };
-        let mut discovery_errors = (0..disks.len()).map(|_| Some(DiskError::DiskNotFound)).collect::<Vec<_>>();
-        let mut candidate_counts = HashMap::<String, usize>::new();
-        let mut discovery_tasks = JoinSet::new();
-        for (index, disk) in disks.iter().enumerate() {
-            let disk = disk.clone();
-            let orig_bucket = orig_bucket.to_string();
-            discovery_tasks.spawn(async move {
-                let result = match disk {
-                    Some(disk) => multipart_upload_paths_on_disk(disk, &orig_bucket).await,
-                    None => Err(DiskError::DiskNotFound),
-                };
-                (index, result)
-            });
-        }
-
-        while let Some(task_result) = discovery_tasks.join_next().await {
-            let Ok((index, result)) = task_result else {
-                continue;
-            };
-            match result {
-                Ok(paths) => {
-                    discovery_errors[index] = None;
-                    for path in paths {
-                        *candidate_counts.entry(path).or_insert(0) += 1;
-                    }
-                }
-                Err(err) => discovery_errors[index] = Some(err),
-            }
-        }
-
-        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) {
-            return Err(to_object_err(err.into(), vec![orig_bucket, error_path]));
-        }
-
-        let mut candidate_paths = candidate_counts
-            .into_iter()
-            .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path))
-            .collect::<Vec<_>>();
-        candidate_paths.sort_unstable();
-        Ok((disks, candidate_paths, discovery_quorum))
-    }
-
-    pub(crate) async fn first_multipart_upload_path_for_decommission(&self, bucket: &str) -> Result<Option<String>> {
-        let (_, paths, _) = self
-            .discover_multipart_upload_paths(bucket, RUSTFS_META_MULTIPART_BUCKET)
-            .await?;
-        Ok(paths.into_iter().next())
-    }
-
     async fn acquire_multipart_upload_read_lock(
         &self,
         op: &'static str,
@@ -810,7 +747,53 @@ impl SetDisks {
         max_uploads: usize,
         expected_incarnation_id: Option<Uuid>,
     ) -> Result<ListMultipartsInfo> {
-        let (disks, candidate_paths, discovery_quorum) = self.discover_multipart_upload_paths(bucket, prefix).await?;
+        let disks = self.disks.read().await.clone();
+        if disks.is_empty() {
+            return Err(Error::ErasureReadQuorum);
+        }
+        let discovery_quorum = if self.default_parity_count == 0 {
+            disks.len()
+        } else {
+            (disks.len() / 2).max(1)
+        };
+        let mut discovery_errors = (0..disks.len()).map(|_| Some(DiskError::DiskNotFound)).collect::<Vec<_>>();
+        let mut candidate_counts = HashMap::<String, usize>::new();
+        let mut discovery_tasks = JoinSet::new();
+        for (index, disk) in disks.iter().enumerate() {
+            let disk = disk.clone();
+            let bucket = bucket.to_string();
+            discovery_tasks.spawn(async move {
+                let result = match disk {
+                    Some(disk) => multipart_upload_paths_on_disk(disk, &bucket).await,
+                    None => Err(DiskError::DiskNotFound),
+                };
+                (index, result)
+            });
+        }
+
+        while let Some(task_result) = discovery_tasks.join_next().await {
+            let Ok((index, result)) = task_result else {
+                continue;
+            };
+            match result {
+                Ok(paths) => {
+                    discovery_errors[index] = None;
+                    for path in paths {
+                        *candidate_counts.entry(path).or_insert(0) += 1;
+                    }
+                }
+                Err(err) => discovery_errors[index] = Some(err),
+            }
+        }
+
+        if let Some(err) = reduce_read_quorum_errs(&discovery_errors, OBJECT_OP_IGNORED_ERRS, discovery_quorum) {
+            return Err(to_object_err(err.into(), vec![bucket, prefix]));
+        }
+
+        let candidate_paths = candidate_counts
+            .into_iter()
+            .filter_map(|(path, count)| (count >= discovery_quorum).then_some(path))
+            .collect::<Vec<_>>();
         let listed_uploads = stream::iter(candidate_paths)
             .map(|upload_path| {
                 let disks = &disks;

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

@@ -2671,17 +2671,7 @@ impl SetDisks {
                         let object_lock_config = opts.object_lock_config_snapshot.as_deref().ok_or_else(|| {
                             Error::other("explicit-version PUT is missing its Object Lock configuration snapshot")
                         })?;
-                        // The WORM gate protects the locked version from local
-                        // overwrites; an authorized replication write passes it
-                        // only when the LWW merge below will judge every
-                        // locking category (see
-                        // `replication_write_may_pass_worm_gate`, which judges
-                        // the same authoritative lock state as the gate,
-                        // bucket default retention included). Gate first so
-                        // malformed lock metadata still fails closed.
-                        if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some()
-                            && !replication_write_may_pass_worm_gate(object_lock_config.state(), &existing, opts)?
-                        {
+                        if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some() {
                             return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
                         }
                         // Receiver-side LWW (rustfs/backlog#1953): reuse this
@@ -8599,259 +8589,6 @@ mod replication_lww_tests {
         );
         assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
     }
-
-    /// Destination version under an active legal hold at `hold_timestamp`,
-    /// plus an active COMPLIANCE retention (no retention timestamp).
-    async fn seed_locked_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str, hold_timestamp: &str) {
-        let mut local = HashMap::new();
-        local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "ON".to_string());
-        insert_str(&mut local, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, hold_timestamp.to_string());
-        local.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
-        local.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
-        put_version(set_disks, bucket, object, version_id, &versioned_opts(version_id, local)).await;
-    }
-
-    /// Inbound legal-hold release from a source that also carries the (same)
-    /// COMPLIANCE retention; the sender stamps a source timestamp for every
-    /// category the source version has.
-    fn inbound_legal_hold_release_opts(version_id: &str, timestamp: &str) -> ObjectOptions {
-        let mut inbound = HashMap::new();
-        inbound.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "OFF".to_string());
-        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, timestamp.to_string());
-        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
-        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
-        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
-        ObjectOptions {
-            replication_request: true,
-            replication_legalhold_timestamp: Some(parse_ts(timestamp)),
-            replication_retention_timestamp: Some(parse_ts(T_OLD)),
-            ..versioned_opts(version_id, inbound)
-        }
-    }
-
-    /// The source's lock state governs the replica: a legal-hold release (or a
-    /// retention change) can only reach this site through the authorized
-    /// replication write, so the commit-time WORM gate must not reject it
-    /// because the destination version is currently locked.
-    #[tokio::test]
-    async fn inbound_newer_legal_hold_release_updates_locked_version() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "lww-locked-release-newer";
-        let object = "object";
-        let version_id = Uuid::new_v4().to_string();
-        make_bucket(&disk_stores, bucket).await;
-        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
-
-        put_version(
-            &set_disks,
-            bucket,
-            object,
-            &version_id,
-            &inbound_legal_hold_release_opts(&version_id, T_NEW),
-        )
-        .await;
-
-        let info = version_info(&set_disks, bucket, object, &version_id).await;
-        assert_eq!(
-            info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
-            Some("OFF"),
-            "a newer source-side legal hold release must be applied to the locked replica"
-        );
-        assert_eq!(get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(), Some(T_NEW));
-        assert_eq!(
-            info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str),
-            Some("COMPLIANCE"),
-            "the untouched retention category must survive the write"
-        );
-    }
-
-    /// Skipping the WORM gate for replication writes must not weaken LWW: a
-    /// stale inbound release still loses to a hold applied more recently here.
-    #[tokio::test]
-    async fn inbound_stale_legal_hold_release_keeps_newer_local_hold() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "lww-locked-release-stale";
-        let object = "object";
-        let version_id = Uuid::new_v4().to_string();
-        make_bucket(&disk_stores, bucket).await;
-        seed_locked_version(&set_disks, bucket, object, &version_id, T_LOCAL).await;
-
-        put_version(
-            &set_disks,
-            bucket,
-            object,
-            &version_id,
-            &inbound_legal_hold_release_opts(&version_id, T_OLD),
-        )
-        .await;
-
-        let info = version_info(&set_disks, bucket, object, &version_id).await;
-        assert_eq!(
-            info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
-            Some("ON"),
-            "a stale inbound release must not lift a hold applied more recently on this site"
-        );
-        assert_eq!(
-            get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(),
-            Some(T_LOCAL)
-        );
-    }
-
-    /// A replication write that carries no source decision for a locking
-    /// category (here: tags changed at a source that never held the object)
-    /// must not lift the destination's hold by replacing the metadata
-    /// unjudged; it stays WORM-rejected like a local overwrite.
-    #[tokio::test]
-    async fn inbound_without_legal_hold_timestamp_stays_rejected_on_held_version() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "lww-locked-unjudged-category";
-        let object = "object";
-        let version_id = Uuid::new_v4().to_string();
-        make_bucket(&disk_stores, bucket).await;
-        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
-
-        let mut inbound = HashMap::new();
-        inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "k=v".to_string());
-        insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_NEW.to_string());
-        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
-        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2099-01-01T00:00:00Z".to_string());
-        let opts = ObjectOptions {
-            replication_request: true,
-            replication_tagging_timestamp: Some(parse_ts(T_NEW)),
-            replication_retention_timestamp: Some(parse_ts(T_NEW)),
-            replication_legalhold_timestamp: None,
-            ..versioned_opts(&version_id, inbound)
-        };
-        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
-        let err = set_disks
-            .put_object(bucket, object, &mut reader, &opts)
-            .await
-            .expect_err("a replication write without the legal-hold source timestamp must stay rejected");
-        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
-
-        let info = version_info(&set_disks, bucket, object, &version_id).await;
-        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("ON"));
-    }
-
-    fn default_retention_snapshot(mode: &'static str) -> Arc<ObjectLockConfigSnapshot> {
-        Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::Configured {
-            config: s3s::dto::ObjectLockConfiguration {
-                object_lock_enabled: Some(s3s::dto::ObjectLockEnabled::from_static(s3s::dto::ObjectLockEnabled::ENABLED)),
-                rule: Some(s3s::dto::ObjectLockRule {
-                    default_retention: Some(s3s::dto::DefaultRetention {
-                        mode: Some(s3s::dto::ObjectLockRetentionMode::from_static(mode)),
-                        days: Some(1),
-                        years: None,
-                    }),
-                }),
-            },
-            updated_at: OffsetDateTime::now_utc(),
-        }))
-    }
-
-    /// The bucket default retention locks a version that carries no explicit
-    /// retention keys. A tagging-only authorized replication write carries no
-    /// source retention decision, so it must stay WORM-rejected exactly like
-    /// it does for an explicitly retained version; with the retention source
-    /// timestamp the write passes and LWW judges it.
-    #[tokio::test]
-    async fn inbound_without_retention_timestamp_stays_rejected_under_bucket_default_retention() {
-        for mode in [
-            s3s::dto::ObjectLockRetentionMode::COMPLIANCE,
-            s3s::dto::ObjectLockRetentionMode::GOVERNANCE,
-        ] {
-            let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-            let bucket = "lww-locked-default-retention";
-            let object = "object";
-            let version_id = Uuid::new_v4().to_string();
-            make_bucket(&disk_stores, bucket).await;
-            seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
-            let seeded = version_info(&set_disks, bucket, object, &version_id).await;
-            assert!(
-                !seeded.user_defined.contains_key(AMZ_OBJECT_LOCK_MODE_LOWER),
-                "the seeded version must be protected by the bucket default only"
-            );
-
-            let tagging_only = ObjectOptions {
-                object_lock_config_snapshot: Some(default_retention_snapshot(mode)),
-                ..inbound_tagging_opts(&version_id, "site=remote", T_NEW)
-            };
-            let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
-            let err = set_disks
-                .put_object(bucket, object, &mut reader, &tagging_only)
-                .await
-                .expect_err("{mode}: a tagging-only replication write must not pass the bucket default retention lock");
-            assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "{mode}: unexpected error: {err}");
-            let info = version_info(&set_disks, bucket, object, &version_id).await;
-            assert_eq!(
-                info.user_tags.as_str(),
-                "site=local",
-                "{mode}: the default-protected version must be untouched"
-            );
-
-            let with_retention_decision = ObjectOptions {
-                replication_retention_timestamp: Some(parse_ts(T_NEW)),
-                ..tagging_only
-            };
-            put_version(&set_disks, bucket, object, &version_id, &with_retention_decision).await;
-            let info = version_info(&set_disks, bucket, object, &version_id).await;
-            assert_eq!(
-                info.user_tags.as_str(),
-                "site=remote",
-                "{mode}: with the retention source timestamp the newer inbound tags win"
-            );
-        }
-    }
-
-    /// The gate runs before the replication bypass, so malformed persisted
-    /// lock metadata still fails closed for an authorized replication write.
-    #[tokio::test]
-    async fn replication_write_on_malformed_lock_metadata_still_fails_closed() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "lww-locked-malformed";
-        let object = "object";
-        let version_id = Uuid::new_v4().to_string();
-        make_bucket(&disk_stores, bucket).await;
-        let mut local = HashMap::new();
-        local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "MAYBE".to_string());
-        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
-
-        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
-        let err = set_disks
-            .put_object(bucket, object, &mut reader, &inbound_legal_hold_release_opts(&version_id, T_NEW))
-            .await
-            .expect_err("malformed persisted lock metadata must fail the replication write closed");
-        assert!(!matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
-
-        let info = version_info(&set_disks, bucket, object, &version_id).await;
-        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("MAYBE"));
-    }
-
-    /// The bypass is scoped to authorized replication writes: the same
-    /// explicit-version PUT without `replication_request` stays WORM-rejected.
-    #[tokio::test]
-    async fn non_replication_overwrite_of_locked_version_is_still_rejected() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "lww-locked-plain-put";
-        let object = "object";
-        let version_id = Uuid::new_v4().to_string();
-        make_bucket(&disk_stores, bucket).await;
-        seed_locked_version(&set_disks, bucket, object, &version_id, T_OLD).await;
-
-        let opts = ObjectOptions {
-            replication_request: false,
-            ..inbound_legal_hold_release_opts(&version_id, T_NEW)
-        };
-        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
-        let err = set_disks
-            .put_object(bucket, object, &mut reader, &opts)
-            .await
-            .expect_err("a non-replication overwrite of a locked version must stay rejected");
-        assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected error: {err}");
-
-        let info = version_info(&set_disks, bucket, object, &version_id).await;
-        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str), Some("ON"));
-    }
 }
 
 #[cfg(test)]
@@ -14841,72 +14578,6 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, original_body);
     }
 
-    /// A local PutObjectRetention / PutObjectLegalHold clear persists empty
-    /// lock keys (`parse_object_lock_retention`). The commit-time WORM gate
-    /// must read that as unlocked: an explicit-version PUT (the inbound
-    /// replication transport) and a version delete both have to succeed
-    /// (rustfs/backlog#1953).
-    #[tokio::test]
-    async fn explicit_version_overwrite_and_delete_succeed_after_local_lock_clear() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "put-explicit-version-cleared-lock";
-        let object = "object";
-        for disk in &disk_stores {
-            disk.make_volume(bucket).await.expect("bucket volume should be created");
-        }
-
-        let mut initial_reader = PutObjReader::from_vec(b"original".to_vec());
-        let initial = set_disks
-            .put_object(
-                bucket,
-                object,
-                &mut initial_reader,
-                &ObjectOptions {
-                    versioned: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("initial version should be written");
-        let version_id = initial
-            .version_id
-            .expect("versioned PUT should return a version ID")
-            .to_string();
-        let version_opts = ObjectOptions {
-            versioned: true,
-            version_id: Some(version_id.clone()),
-            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
-            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
-            ..Default::default()
-        };
-        set_disks
-            .put_object_metadata(
-                bucket,
-                object,
-                &ObjectOptions {
-                    eval_metadata: Some(HashMap::from([
-                        (X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(), String::new()),
-                        (X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(), String::new()),
-                        (X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), String::new()),
-                    ])),
-                    ..version_opts.clone()
-                },
-            )
-            .await
-            .expect("cleared lock metadata should be written");
-
-        let mut replacement = PutObjReader::from_vec(b"replacement".to_vec());
-        set_disks
-            .put_object(bucket, object, &mut replacement, &version_opts)
-            .await
-            .expect("explicit-version PUT must not be wedged by cleared lock metadata");
-
-        set_disks
-            .delete_object(bucket, object, version_opts)
-            .await
-            .expect("version delete must not be wedged by cleared lock metadata");
-    }
-
     #[tokio::test]
     async fn version_only_copy_checks_the_destination_version_object_lock() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/ecstore/src/store/heal.rs

@@ -19,7 +19,6 @@ use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::heal::HealOperations as _;
 use crate::storage_api_contracts::namespace::NamespaceLocking as _;
 use rustfs_lock::NamespaceLockGuard;
-use std::collections::BTreeSet;
 use tracing::trace;
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -110,8 +109,7 @@ impl ECStore {
         }
 
         let mut pool_meta = PoolMeta::default();
-        let replica_state = pool_meta.load_no_lock_from_replicas(self.pools.clone()).await?;
-        replica_state.ensure_write_safe("heal format fence failed")?;
+        pool_meta.load_no_lock(metadata_pool.clone()).await?;
         if pool_meta.pools.len() != self.pools.len()
             || pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
                 pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
@@ -293,45 +291,7 @@ impl ECStore {
 
     #[instrument(skip(self))]
     pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
-        let mut fenced_pools = BTreeSet::new();
-        {
-            let pool_meta = self.pool_meta.read().await;
-            fenced_pools.extend((0..pool_meta.pools.len()).filter(|pool_idx| pool_meta.is_suspended(*pool_idx)));
-            if let Some(pool_idx) = opts.pool {
-                if pool_idx >= pool_meta.pools.len() {
-                    return Err(invalid_heal_pool_index(pool_idx, pool_meta.pools.len()));
-                }
-                if pool_meta.is_suspended(pool_idx) {
-                    let complete = pool_meta.pools[pool_idx]
-                        .decommission
-                        .as_ref()
-                        .is_some_and(|decommission| decommission.complete);
-                    return Err(if complete {
-                        StorageError::InvalidArgument(
-                            "heal".to_string(),
-                            "pool".to_string(),
-                            format!("heal pool {pool_idx} has completed decommission"),
-                        )
-                    } else {
-                        Error::SlowDown
-                    });
-                }
-            }
-        }
-
-        let dispatch_fenced_pools = fenced_pools.iter().copied().collect::<Vec<_>>();
-        let mut res = self
-            .peer_sys
-            .heal_bucket_with_fence(bucket, opts, &dispatch_fenced_pools)
-            .await?;
-        {
-            let pool_meta = self.pool_meta.read().await;
-            fenced_pools.extend((0..pool_meta.pools.len()).filter(|pool_idx| pool_meta.is_suspended(*pool_idx)));
-        }
-        if !fenced_pools.is_empty() {
-            let pools = fenced_pools.iter().map(usize::to_string).collect::<Vec<_>>().join(", ");
-            res.detail = format!("skipped: bucket-volume heal fenced on decommission-suspended pool(s): {pools}");
-        }
+        let res = self.peer_sys.heal_bucket(bucket, opts).await?;
 
         Ok(res)
     }
@@ -897,64 +857,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn scoped_heal_bucket_blocks_before_dispatch_when_pool_is_suspended() {
-        let mut store = minimal_heal_store().await;
-        store.pool_meta = RwLock::new(PoolMeta {
-            pools: vec![
-                PoolStatus {
-                    id: 0,
-                    cmd_line: "pool-0".to_string(),
-                    last_update: OffsetDateTime::UNIX_EPOCH,
-                    decommission: None,
-                },
-                PoolStatus {
-                    id: 1,
-                    cmd_line: "pool-1".to_string(),
-                    last_update: OffsetDateTime::UNIX_EPOCH,
-                    decommission: Some(PoolDecommissionInfo {
-                        start_time: Some(OffsetDateTime::UNIX_EPOCH),
-                        ..Default::default()
-                    }),
-                },
-            ],
-            ..Default::default()
-        });
-
-        let err = store
-            .handle_heal_bucket(
-                "bucket",
-                &HealOpts {
-                    pool: Some(1),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect_err("suspended pool must be blocked before bucket-heal fan-out");
-        assert_eq!(err, Error::SlowDown);
-
-        store.pool_meta.write().await.pools[1]
-            .decommission
-            .as_mut()
-            .expect("decommission state should exist")
-            .complete = true;
-        let err = store
-            .handle_heal_bucket(
-                "bucket",
-                &HealOpts {
-                    pool: Some(1),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect_err("completed pool must remain fenced from bucket heal");
-        assert!(
-            matches!(err, StorageError::InvalidArgument(_, ref field, ref reason)
-                if field == "pool" && reason.contains("completed decommission")),
-            "unexpected completed-pool error: {err:?}"
-        );
-    }
-
     #[tokio::test]
     #[serial_test::serial]
     async fn unscoped_heal_object_suspended_owner_semantics() {

crates/ecstore/src/store/init.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::*;
-use crate::core::pools::{PoolMetaReplicaState, local_decommission_queue_prefix, pool_meta_has_active_decommission};
+use crate::core::pools::{local_decommission_queue_prefix, pool_meta_has_active_decommission};
 use crate::error::is_err_decommission_running;
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -120,16 +120,13 @@ fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<()
     result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}")))
 }
 
-async fn load_pool_meta_for_startup<S>(pools: Vec<Arc<S>>) -> Result<(PoolMeta, PoolMetaReplicaState)>
+async fn load_pool_meta_for_startup<S>(pool: Arc<S>) -> Result<PoolMeta>
 where
     S: EcstoreObjectIO,
 {
     let mut meta = PoolMeta::default();
-    let replica_state = meta
-        .load_no_lock_from_replicas(pools)
-        .await
-        .map_err(|err| Error::other(format!("store init failed during load_pool_meta: {err}")))?;
-    Ok((meta, replica_state))
+    resolve_store_init_stage_result(meta.load_for_startup(pool).await, "load_pool_meta")?;
+    Ok(meta)
 }
 
 async fn save_validated_pool_meta_for_startup<S>(meta: &PoolMeta, pools: Vec<Arc<S>>) -> Result<()>
@@ -139,28 +136,6 @@ where
     resolve_store_init_stage_result(meta.save_for_startup(pools).await, "save_validated_pool_meta")
 }
 
-async fn persist_pool_meta_for_startup_if_safe<S>(
-    meta: &PoolMeta,
-    pools: Vec<Arc<S>>,
-    replica_state: PoolMetaReplicaState,
-    topology_update: bool,
-    elected_writer: bool,
-) -> Result<()>
-where
-    S: EcstoreObjectIO,
-{
-    if !elected_writer {
-        return Ok(());
-    }
-    if topology_update {
-        replica_state.ensure_write_safe("store init failed during save_validated_pool_meta")?;
-    }
-    if topology_update || (replica_state.needs_repair && replica_state.repair_write_safe) {
-        save_validated_pool_meta_for_startup(meta, pools).await?;
-    }
-    Ok(())
-}
-
 async fn resume_local_decommission_after_init(store: Arc<ECStore>, rx: CancellationToken, pool_indices: Vec<usize>) {
     for attempt in 0..=LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES {
         if rx.is_cancelled() {
@@ -475,26 +450,28 @@ impl ECStore {
     pub async fn init(self: &Arc<Self>, rx: CancellationToken) -> Result<()> {
         runtime_sources::ensure_boot_time().await;
 
-        let (meta, pool_meta_replica_state) = load_pool_meta_for_startup(self.pools.clone()).await?;
+        let meta = load_pool_meta_for_startup(
+            self.pools
+                .first()
+                .cloned()
+                .ok_or_else(|| Error::other("store init failed: no storage pools available"))?,
+        )
+        .await?;
         let update = meta.validate(self.pools.clone())?;
         let endpoints = runtime_sources::endpoint_pools_or_default();
         let should_persist_pool_meta = runtime_sources::first_cluster_node_is_local().await;
 
-        let installed_pool_meta = if update {
-            PoolMeta::new(&self.pools, &meta)
-        } else {
+        let installed_pool_meta = if !update {
             meta.clone()
+        } else {
+            let new_meta = PoolMeta::new(&self.pools, &meta);
+            // Only one local node should persist validated pool metadata here; otherwise
+            // distributed startup can race on the same lock and replay the prior init bug.
+            if should_persist_pool_meta {
+                save_validated_pool_meta_for_startup(&new_meta, self.pools.clone()).await?;
+            }
+            new_meta
         };
-        // Only one local node should persist validated pool metadata here; otherwise
-        // distributed startup can race on the same lock and replay the prior init bug.
-        persist_pool_meta_for_startup_if_safe(
-            &installed_pool_meta,
-            self.pools.clone(),
-            pool_meta_replica_state,
-            update,
-            should_persist_pool_meta,
-        )
-        .await?;
 
         {
             let mut pool_meta = self.pool_meta.write().await;
@@ -575,11 +552,10 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::{
-        LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, persist_pool_meta_for_startup_if_safe,
-        pool_first_endpoint_is_local, pool_meta_has_active_decommission, preflight_startup_rpc_secret_with,
-        resolve_startup_pool_defaults_with, resolve_store_init_stage_result, save_validated_pool_meta_for_startup,
-        should_auto_start_rebalance_after_init, should_retry_format_load, should_retry_local_decommission_resume,
-        wait_for_local_decommission_resume_delay,
+        LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local,
+        pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with,
+        resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
+        should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
     };
     #[cfg(feature = "test-util")]
     use crate::{
@@ -668,7 +644,7 @@ mod tests {
         future::Future,
         io::Cursor,
         sync::{
-            Arc, Mutex,
+            Arc,
             atomic::{AtomicBool, AtomicUsize, Ordering},
         },
         time::Duration,
@@ -677,40 +653,21 @@ mod tests {
     use tokio::io::AsyncReadExt;
     use tokio_util::sync::CancellationToken;
 
-    fn startup_pool_meta_payload(meta: &PoolMeta) -> Vec<u8> {
-        meta.encode_config_data_for_test().expect("pool metadata should encode")
-    }
-
     #[derive(Debug)]
     struct StartupPoolMetaStorage {
         read_payload: Vec<u8>,
-        read_error: bool,
         read_without_lock: AtomicBool,
         wrote_without_lock: AtomicBool,
         wrote_with_max_parity: AtomicBool,
-        written_payload: Mutex<Option<Vec<u8>>>,
     }
 
     impl StartupPoolMetaStorage {
         fn new(read_payload: Vec<u8>) -> Self {
             Self {
                 read_payload,
-                read_error: false,
                 read_without_lock: AtomicBool::new(false),
                 wrote_without_lock: AtomicBool::new(false),
                 wrote_with_max_parity: AtomicBool::new(false),
-                written_payload: Mutex::new(None),
-            }
-        }
-
-        fn unreadable() -> Self {
-            Self {
-                read_payload: Vec::new(),
-                read_error: true,
-                read_without_lock: AtomicBool::new(false),
-                wrote_without_lock: AtomicBool::new(false),
-                wrote_with_max_parity: AtomicBool::new(false),
-                written_payload: Mutex::new(None),
             }
         }
 
@@ -745,12 +702,6 @@ mod tests {
         ) -> Result<GetObjectReader> {
             assert!(opts.no_lock, "store init pool metadata load must not require namespace locks");
             self.read_without_lock.store(true, Ordering::SeqCst);
-            if self.read_error {
-                return Err(Error::other("pool metadata read quorum unavailable"));
-            }
-            if self.read_payload.is_empty() {
-                return Err(Error::FileNotFound);
-            }
 
             Ok(GetObjectReader {
                 stream: Box::new(Cursor::new(self.read_payload.clone())),
@@ -764,17 +715,13 @@ mod tests {
             &self,
             bucket: &str,
             object: &str,
-            data: &mut PutObjReader,
+            _data: &mut PutObjReader,
             opts: &ObjectOptions,
         ) -> Result<ObjectInfo> {
             assert!(opts.no_lock, "store init pool metadata save must not require namespace locks");
             self.wrote_without_lock.store(true, Ordering::SeqCst);
             self.wrote_with_max_parity.store(opts.max_parity, Ordering::SeqCst);
-            let mut payload = Vec::new();
-            data.stream.read_to_end(&mut payload).await?;
-            let size = payload.len();
-            *self.written_payload.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(payload);
-            Ok(self.object_info(bucket, object, size))
+            Ok(self.object_info(bucket, object, 0))
         }
     }
 
@@ -795,12 +742,10 @@ mod tests {
     async fn test_store_init_pool_meta_io_bypasses_namespace_lock_surface() {
         let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
 
-        let (loaded, replica_state) = load_pool_meta_for_startup(vec![storage.clone()])
+        let loaded = load_pool_meta_for_startup(storage.clone())
             .await
             .expect("startup pool metadata load should tolerate missing metadata without locks");
         assert!(loaded.pools.is_empty());
-        assert!(!replica_state.needs_repair);
-        assert!(replica_state.repair_write_safe);
         assert!(storage.read_without_lock.load(Ordering::SeqCst));
 
         let meta = PoolMeta {
@@ -815,69 +760,6 @@ mod tests {
         assert!(storage.wrote_with_max_parity.load(Ordering::SeqCst));
     }
 
-    #[tokio::test]
-    async fn test_store_init_pool_meta_falls_back_from_corrupt_first_replica() {
-        let corrupt = Arc::new(StartupPoolMetaStorage::new(vec![0, 1, 2]));
-        let expected = init_test_pool_meta(None);
-        let backup = Arc::new(StartupPoolMetaStorage::new(startup_pool_meta_payload(&expected)));
-
-        let (loaded, replica_state) = load_pool_meta_for_startup(vec![corrupt.clone(), backup.clone()])
-            .await
-            .expect("startup should select the validated backup replica");
-
-        assert!(replica_state.needs_repair);
-        assert!(replica_state.repair_write_safe);
-        assert_eq!(loaded.pools.len(), 1);
-        assert_eq!(loaded.pools[0].cmd_line, expected.pools[0].cmd_line);
-        assert!(corrupt.read_without_lock.load(Ordering::SeqCst));
-        assert!(backup.read_without_lock.load(Ordering::SeqCst));
-
-        persist_pool_meta_for_startup_if_safe(&loaded, vec![corrupt.clone(), backup.clone()], replica_state, false, true)
-            .await
-            .expect("the elected startup writer should repair validated corrupt replicas");
-
-        let corrupt_write = corrupt
-            .written_payload
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .clone()
-            .expect("corrupt replica should be repaired");
-        let backup_write = backup
-            .written_payload
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
-            .clone()
-            .expect("backup replica should receive the same canonical snapshot");
-        assert_eq!(corrupt_write, backup_write);
-        assert_ne!(corrupt_write, backup.read_payload);
-    }
-
-    #[tokio::test]
-    async fn test_store_init_pool_meta_does_not_repair_unreadable_replica() {
-        let valid = Arc::new(StartupPoolMetaStorage::new(startup_pool_meta_payload(&init_test_pool_meta(None))));
-        let unreadable = Arc::new(StartupPoolMetaStorage::unreadable());
-
-        let (loaded, replica_state) = load_pool_meta_for_startup(vec![valid.clone(), unreadable.clone()])
-            .await
-            .expect("startup should use a validated replica without overwriting an unreadable copy");
-        assert!(replica_state.needs_repair);
-        assert!(!replica_state.repair_write_safe);
-
-        persist_pool_meta_for_startup_if_safe(&loaded, vec![valid.clone(), unreadable.clone()], replica_state, false, true)
-            .await
-            .expect("an unreadable copy should defer repair when no topology write is needed");
-        assert!(!valid.wrote_without_lock.load(Ordering::SeqCst));
-        assert!(!unreadable.wrote_without_lock.load(Ordering::SeqCst));
-
-        let err =
-            persist_pool_meta_for_startup_if_safe(&loaded, vec![valid.clone(), unreadable.clone()], replica_state, true, true)
-                .await
-                .expect_err("a topology update must not overwrite an unreadable replica");
-        assert!(err.to_string().contains("cannot overwrite an unreadable replica"));
-        assert!(!valid.wrote_without_lock.load(Ordering::SeqCst));
-        assert!(!unreadable.wrote_without_lock.load(Ordering::SeqCst));
-    }
-
     #[test]
     fn test_pool_first_endpoint_is_local_respects_local_flag() {
         let mut local_endpoint = Endpoint::try_from("http://127.0.0.1:9000/data").expect("endpoint should parse");
@@ -1805,177 +1687,6 @@ mod tests {
         shutdown.cancel();
     }
 
-    #[tokio::test]
-    #[serial_test::serial(storage_class_env)]
-    async fn suspended_decommission_source_multipart_remains_operable_until_drained() {
-        let temp_dir = tempfile::tempdir().expect("create decommission multipart drain store dir");
-        let (_ctx, store, shutdown) =
-            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "decommission-multipart-drain", &[4, 4])).await;
-        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-
-        let bucket = format!("decommission-multipart-drain-{}", uuid::Uuid::new_v4());
-        let complete_object = "complete.bin";
-        let abort_object = "abort.bin";
-        store
-            .make_bucket(&bucket, &MakeBucketOptions::default())
-            .await
-            .expect("create decommission multipart drain bucket");
-
-        let incarnation = store.bucket_incarnation_id(&bucket).await.expect("read bucket incarnation");
-        let lifecycle_guard = store
-            .acquire_bucket_lifecycle_read_lock(&bucket)
-            .await
-            .expect("acquire multipart creation lifecycle fence");
-        let mut upload_opts = ObjectOptions {
-            expected_bucket_incarnation_id: Some(incarnation),
-            ..Default::default()
-        };
-        upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
-        let complete_upload = store.pools[0]
-            .new_multipart_upload(&bucket, complete_object, &upload_opts)
-            .await
-            .expect("create source upload to complete");
-        let abort_upload = store.pools[0]
-            .new_multipart_upload(&bucket, abort_object, &upload_opts)
-            .await
-            .expect("create source upload to abort");
-        drop(lifecycle_guard);
-
-        mark_test_pool_decommissioning(&store, 0).await;
-
-        let err = store
-            .ensure_decommission_multipart_uploads_drained_for_test(0)
-            .await
-            .expect_err("an unresolved source multipart upload must block final decommission");
-        let drain_error = err.to_string();
-        assert!(
-            drain_error.contains("still contains multipart upload") && drain_error.contains(&bucket),
-            "the drain error must identify both the upload path and user bucket: {drain_error}"
-        );
-
-        let listed = store
-            .list_multipart_uploads(&bucket, "", None, None, None, 100)
-            .await
-            .expect("list uploads from suspended decommission source");
-        assert!(
-            listed
-                .uploads
-                .iter()
-                .any(|upload| upload.upload_id.as_str() == complete_upload.upload_id.as_str()),
-            "the upload selected before suspension must remain visible"
-        );
-        store
-            .get_multipart_info(&bucket, complete_object, &complete_upload.upload_id, &ObjectOptions::default())
-            .await
-            .expect("read upload metadata from suspended decommission source");
-
-        let mut part_reader = PutObjReader::from_vec(b"multipart body".to_vec());
-        let part = store
-            .put_object_part(
-                &bucket,
-                complete_object,
-                &complete_upload.upload_id,
-                1,
-                &mut part_reader,
-                &ObjectOptions::default(),
-            )
-            .await
-            .expect("write part to suspended decommission source");
-        let parts = store
-            .list_object_parts(&bucket, complete_object, &complete_upload.upload_id, None, 100, &ObjectOptions::default())
-            .await
-            .expect("list parts from suspended decommission source");
-        assert_eq!(parts.parts.len(), 1);
-        assert_eq!(parts.parts[0].etag.as_deref(), part.etag.as_deref());
-
-        store
-            .clone()
-            .complete_multipart_upload(
-                &bucket,
-                complete_object,
-                &complete_upload.upload_id,
-                vec![crate::storage_api_contracts::multipart::CompletePart {
-                    part_num: part.part_num,
-                    etag: part.etag,
-                    ..Default::default()
-                }],
-                &ObjectOptions::default(),
-            )
-            .await
-            .expect("complete upload on suspended decommission source");
-        store
-            .abort_multipart_upload(&bucket, abort_object, &abort_upload.upload_id, &ObjectOptions::default())
-            .await
-            .expect("abort upload on suspended decommission source");
-
-        store
-            .ensure_decommission_multipart_uploads_drained_for_test(0)
-            .await
-            .expect("final decommission gate should open after all source uploads are resolved");
-        assert_pool_object_present(&store.pools[0], &bucket, complete_object).await;
-
-        shutdown.cancel();
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(storage_class_env)]
-    async fn active_multipart_upload_routes_before_faulted_suspended_source() {
-        let temp_dir = tempfile::tempdir().expect("create active-first multipart routing store dir");
-        let (_ctx, store, shutdown) =
-            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "active-first-multipart-routing", &[4, 4]))
-                .await;
-        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
-
-        let bucket = format!("active-first-multipart-routing-{}", uuid::Uuid::new_v4());
-        let object = "target-upload.bin";
-        store
-            .make_bucket(&bucket, &MakeBucketOptions::default())
-            .await
-            .expect("create active-first multipart routing bucket");
-
-        let incarnation = store.bucket_incarnation_id(&bucket).await.expect("read bucket incarnation");
-        let lifecycle_guard = store
-            .acquire_bucket_lifecycle_read_lock(&bucket)
-            .await
-            .expect("acquire multipart creation lifecycle fence");
-        let mut upload_opts = ObjectOptions {
-            expected_bucket_incarnation_id: Some(incarnation),
-            ..Default::default()
-        };
-        upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
-        let upload = store.pools[1]
-            .new_multipart_upload(&bucket, object, &upload_opts)
-            .await
-            .expect("create upload in active target pool");
-        drop(lifecycle_guard);
-
-        mark_test_pool_decommissioning(&store, 0).await;
-        let source_set = store.pools[0].get_disks_by_key(object);
-        let original_source_disks = {
-            let mut disks = source_set.disks.write().await;
-            let original = disks.clone();
-            disks.fill(None);
-            original
-        };
-
-        let source_result = store.pools[0]
-            .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
-            .await;
-        let routed_result = store
-            .get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
-            .await;
-        *source_set.disks.write().await = original_source_disks;
-
-        assert!(
-            matches!(&source_result, Err(StorageError::ErasureReadQuorum)),
-            "the suspended source must expose the injected hard read failure: {source_result:?}"
-        );
-        let routed = routed_result.expect("the active target UploadID must be resolved before the faulted suspended source");
-        assert_eq!(routed.upload_id, upload.upload_id);
-
-        shutdown.cancel();
-    }
-
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
     async fn delete_objects_skips_active_rebalance_source_pool() {

crates/ecstore/src/store/multipart.rs

@@ -196,25 +196,6 @@ async fn list_pool_multipart_uploads_for_incarnation(
 }
 
 impl ECStore {
-    async fn existing_multipart_pool_order(&self) -> Vec<usize> {
-        // A draining source must not hide a valid UploadID in an active target,
-        // while physical order within each phase preserves fail-closed errors.
-        let mut active = Vec::with_capacity(self.pools.len());
-        let mut draining = Vec::new();
-        for (idx, pool) in self.pools.iter().enumerate() {
-            if self.is_pool_rebalancing(pool.pool_idx).await {
-                continue;
-            }
-            if self.is_suspended(pool.pool_idx).await {
-                draining.push(idx);
-            } else {
-                active.push(idx);
-            }
-        }
-        active.extend(draining);
-        active
-    }
-
     #[allow(clippy::too_many_arguments)]
     pub async fn list_multipart_uploads_for_bucket_incarnation(
         &self,
@@ -309,8 +290,10 @@ impl ECStore {
                 .await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
             return match pool
                 .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts)
                 .await
@@ -370,8 +353,10 @@ impl ECStore {
         let mut common_prefixes = HashSet::new();
         let mut source_truncated = false;
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
             let res = list_pool_multipart_uploads_for_incarnation(
                 pool,
                 bucket,
@@ -538,8 +523,10 @@ impl ECStore {
                 .await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
             let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await {
                 Ok(res) => return Ok(res),
                 Err(err) => {
@@ -599,8 +586,10 @@ impl ECStore {
             return self.pools[0].get_multipart_info(bucket, object, upload_id, opts).await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
 
             return match pool.get_multipart_info(bucket, object, upload_id, opts).await {
                 Ok(res) => Ok(res),
@@ -635,8 +624,10 @@ impl ECStore {
             return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
 
             let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await {
                 Ok(_) => return Ok(()),
@@ -694,8 +685,10 @@ impl ECStore {
                 .await;
         }
 
-        for pool_idx in self.existing_multipart_pool_order().await {
-            let pool = &self.pools[pool_idx];
+        for pool in self.pools.iter() {
+            if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
+                continue;
+            }
 
             let pool = pool.clone();
             let err = match pool

crates/heal/src/heal/manager.rs

@@ -117,13 +117,13 @@ async fn pause_duplicate_admission_after_active_lock(request_id: &str) {
 type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub(super) struct MrfRepairNoticeTarget {
-    pub(super) bucket: Arc<str>,
-    pub(super) object: Arc<str>,
-    pub(super) version_id: Option<[u8; 16]>,
-    pub(super) kind: rustfs_common::mrf_channel::MrfKind,
-    pub(super) scope: Option<rustfs_common::mrf_channel::MrfScope>,
-    pub(super) lease: Option<rustfs_common::mrf_channel::MrfIngressLease>,
+struct MrfRepairNoticeTarget {
+    bucket: Arc<str>,
+    object: Arc<str>,
+    version_id: Option<[u8; 16]>,
+    kind: rustfs_common::mrf_channel::MrfKind,
+    scope: Option<rustfs_common::mrf_channel::MrfScope>,
+    lease: Option<rustfs_common::mrf_channel::MrfIngressLease>,
 }
 
 #[derive(Debug, Clone)]
@@ -1400,29 +1400,35 @@ impl HealManager {
             HealType::ECDecode { .. } => rustfs_common::mrf_channel::MrfKind::DecodeFailure,
             _ => rustfs_common::mrf_channel::MrfKind::PartialWrite,
         };
-        self.submit_mrf_heal_request_with_receipt_and_identity(
+        self.submit_mrf_heal_request_with_receipt_and_identity(request, bucket, object, version_id, kind, None, None)
+            .await
+    }
+
+    pub(crate) async fn submit_mrf_heal_request_with_receipt_and_identity(
+        &self,
+        request: HealRequest,
+        bucket: Arc<str>,
+        object: Arc<str>,
+        version_id: Option<[u8; 16]>,
+        kind: rustfs_common::mrf_channel::MrfKind,
+        scope: Option<rustfs_common::mrf_channel::MrfScope>,
+        lease: Option<rustfs_common::mrf_channel::MrfIngressLease>,
+    ) -> Result<HealAdmissionReceipt> {
+        self.submit_heal_request_with_receipt_alias_and_mrf_notice(
             request,
-            MrfRepairNoticeTarget {
+            true,
+            Some(MrfRepairNoticeTarget {
                 bucket,
                 object,
                 version_id,
                 kind,
-                scope: None,
-                lease: None,
-            },
+                scope,
+                lease,
+            }),
         )
         .await
     }
 
-    pub(super) async fn submit_mrf_heal_request_with_receipt_and_identity(
-        &self,
-        request: HealRequest,
-        mrf_notice_target: MrfRepairNoticeTarget,
-    ) -> Result<HealAdmissionReceipt> {
-        self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, true, Some(mrf_notice_target))
-            .await
-    }
-
     async fn submit_heal_request_with_receipt_alias_and_mrf_notice(
         &self,
         request: HealRequest,

crates/heal/src/heal/mrf_queue.rs

@@ -33,7 +33,7 @@
 //! than waiting for the failed-object TTL to re-scan the path.
 
 use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
-use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
+use crate::heal::manager::HealManager;
 use metrics::{counter, gauge};
 use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
 use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent};
@@ -491,14 +491,12 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
     let receipt = manager
         .submit_mrf_heal_request_with_receipt_and_identity(
             build_heal_request(intent),
-            MrfRepairNoticeTarget {
-                bucket: intent.bucket.clone(),
-                object: intent.object.clone(),
-                version_id: intent.version_id,
-                kind: intent.kind,
-                scope: intent.scope,
-                lease: intent.lease,
-            },
+            intent.bucket.clone(),
+            intent.object.clone(),
+            intent.version_id,
+            intent.kind,
+            intent.scope,
+            intent.lease,
         )
         .await?;
     Ok(receipt.result)

crates/protocols/Cargo.toml

@@ -183,7 +183,6 @@ subtle = { workspace = true, optional = true }
 socket2 = { workspace = true, optional = true, features = ["all"] }
 
 [dev-dependencies]
-http = { workspace = true }
 tempfile = { workspace = true }
 proptest = "1"
 rustfs-test-utils = { workspace = true }

crates/protocols/src/sftp/server.rs

@@ -127,7 +127,6 @@ const SFTP_COMPRESSION: &[russh::compression::Name] = &[russh::compression::NONE
 fn build_preferred() -> russh::Preferred {
     russh::Preferred {
         kex: Cow::Borrowed(SFTP_KEX),
-        host_key_certificates: Cow::Borrowed(&[]),
         key: Cow::Borrowed(SFTP_HOST_KEY_ALGORITHMS),
         cipher: Cow::Borrowed(SFTP_CIPHERS),
         mac: Cow::Borrowed(SFTP_MACS),

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -167,7 +167,7 @@ fn stale_quota_uses_complete_baseline_plus_positive_deltas() {
     let (observed, _) = observational_data_usage_info(&[current], &expected, &all_buckets, &[], TEST_PLAN_DIGEST, 8, 3)
         .expect("complete set data is a valid observational baseline");
     assert_eq!(observed.objects_total_size, 30);
-    assert!(observed.usage_snapshot_set_states[0].complete);
+    assert_eq!(observed.usage_snapshot_set_states[0].complete, true);
 }
 
 #[test]

docs/operations/rolling-restart.md

@@ -31,15 +31,6 @@ failure pattern reported in rustfs/rustfs#4304.
 > older build is not supported: older readers ignore the sidecar and can
 > report an object checksum in place of the requested part checksum.
 
-> [!WARNING]
-> Writing pool metadata version 2 remains inactive unless both
-> `RUSTFS_POOL_META_V2_WRITE=true` and
-> `RUSTFS_POOL_META_V2_FLEET_CONFIRMED=true`. Leave either setting disabled
-> until every node that can read or write `pool.bin` supports version 2. Once a node
-> observes or writes version 2 it will not downgrade the file, and older
-> binaries or rollback builds cannot read it. Unresolved decommission entries
-> fail closed instead of being written in the version 1 format.
-
 ## TL;DR
 
 - **Rolling restart (no downtime):** restart **one node at a time**, and wait

docs/testing/issue-2007-coalescer-delay.md

@@ -1,58 +0,0 @@
-# Backlog #2007 Coalescer Delay Validation
-
-`scripts/issue_2007_coalescer_prometheus_report.py` is a read-only Prometheus
-report helper for validating whether the GET metadata `ReadVersion` coalescer
-default can move from `200us` to `50us`.
-
-The benchmark itself is intentionally external to this helper: use the same
-main build, bucket/object set, workload, and
-`RUSTFS_BATCH_READ_VERSION_SERVER_PARALLELISM=4` for both cells. Only switch:
-
-```bash
-RUSTFS_GET_METADATA_READ_VERSION_COALESCE=auto
-RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS=200
-RUSTFS_GET_METADATA_READ_VERSION_COALESCE_DELAY_MICROS=50
-```
-
-After each measured workload window, collect a report from Prometheus:
-
-```bash
-scripts/issue_2007_coalescer_prometheus_report.py \
-  --query-url http://prometheus.example:9090 \
-  --profile delay-200us \
-  --window 180s \
-  --rustfs-selector 'server=~"node[5-8]"' \
-  --node-selector 'instance=~"node[5-8].*"'
-
-scripts/issue_2007_coalescer_prometheus_report.py \
-  --query-url http://prometheus.example:9090 \
-  --profile delay-50us \
-  --window 180s \
-  --rustfs-selector 'server=~"node[5-8]"' \
-  --node-selector 'instance=~"node[5-8].*"'
-```
-
-The output is Markdown and is suitable for attaching to the issue alongside the
-warp throughput, average latency, p95, p99, and TTFB p99 from the fixed
-workload run.
-
-Required RustFS signals:
-
-- `grpc_read_version` and `grpc_batch_read_version` outgoing request increases.
-- Coalescer batch distribution from
-  `rustfs_get_metadata_read_version_coalescer_total{event="attempted_batch"}`.
-- `batch_read_version_coalescer_wait`, `batch_read_version_rpc_roundtrip`,
-  `batch_read_version_disk_read`, and `batch_read_version_response_map` p99.
-
-Required host-cost signals:
-
-- CPU busy from `node_cpu_seconds_total`.
-- Network RX/TX from `node_network_receive_bytes_total` and
-  `node_network_transmit_bytes_total`.
-- Disk read await, average queue depth, and utilization from node-exporter disk
-  counters.
-
-If a section reports `UNAVAILABLE`, treat that evidence as missing rather than
-zero. Do not use a default-change PR until the `50us` cell has stable
-throughput/latency benefit and CPU, network, and disk cost are available and
-acceptable.

rustfs/src/admin/handlers/cluster_snapshot.rs

@@ -1358,7 +1358,7 @@ mod tests {
         assert_eq!(component.last_usage_save_result, "success");
         assert_eq!(component.last_success_unix_secs, Some(450));
 
-        let mut legacy_snapshot = snapshot;
+        let mut legacy_snapshot = snapshot.clone();
         legacy_snapshot.usage_freshness.last_durable_success_unix_secs = 0;
         let legacy_component = super::summarize_usage_freshness(&legacy_snapshot);
         assert_eq!(legacy_component.last_success_unix_secs, Some(456));

rustfs/src/app/admin_usecase.rs

@@ -16,8 +16,7 @@
 
 use super::storage_api::admin_usecase::admin::get_server_info;
 use super::storage_api::admin_usecase::capacity::{
-    DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity,
-    get_total_usable_capacity_free,
+    PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity, get_total_usable_capacity_free,
 };
 use super::storage_api::admin_usecase::contract::StorageAdminApi;
 use super::storage_api::admin_usecase::contract::bucket::{BucketOperations as _, BucketOptions};
@@ -108,8 +107,6 @@ pub struct AdminPoolDecommissionInfo {
     pub bytes_failed: usize,
     #[serde(rename = "waitingReason")]
     pub waiting_reason: Option<String>,
-    #[serde(rename = "unresolvedEntries", skip_serializing_if = "Vec::is_empty")]
-    pub unresolved_entries: Vec<DecommissionUnresolvedEntry>,
 }
 
 #[derive(Debug, Clone, serde::Serialize)]
@@ -622,7 +619,6 @@ impl DefaultAdminUsecase {
             bytes_done: info.bytes_done,
             bytes_failed: info.bytes_failed,
             waiting_reason,
-            unresolved_entries: info.unresolved_entries,
         }
     }
 
@@ -680,7 +676,7 @@ impl DefaultAdminUsecase {
 
 #[cfg(test)]
 mod tests {
-    use super::super::storage_api::admin_usecase::capacity::{DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus};
+    use super::super::storage_api::admin_usecase::capacity::{PoolDecommissionInfo, PoolStatus};
     use super::*;
     use time::OffsetDateTime;
     use tracing_subscriber::{Layer, Registry, layer::Context, prelude::*};
@@ -991,17 +987,6 @@ mod tests {
                     items_decommission_failed: 1,
                     bytes_done: 1024,
                     bytes_failed: 64,
-                    unresolved_entries: vec![DecommissionUnresolvedEntry {
-                        bucket: "bucket-a".to_string(),
-                        object: "prefix/unresolved.txt".to_string(),
-                        pool_index: 3,
-                        set_index: 1,
-                        source_generation: OffsetDateTime::UNIX_EPOCH,
-                        candidate_count: 2,
-                        disk_error_count: 1,
-                        observed_at: OffsetDateTime::UNIX_EPOCH,
-                        reason: "metadata_resolution_failed".to_string(),
-                    }],
                     ..Default::default()
                 }),
             },
@@ -1025,13 +1010,6 @@ mod tests {
         assert_eq!(value["decommissionInfo"]["objectsDecommissionedFailed"], 1);
         assert_eq!(value["decommissionInfo"]["bytesDecommissioned"], 1024);
         assert_eq!(value["decommissionInfo"]["bytesDecommissionedFailed"], 64);
-        assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["bucket"], "bucket-a");
-        assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["object"], "prefix/unresolved.txt");
-        assert_eq!(
-            value["decommissionInfo"]["unresolvedEntries"][0]["sourceGeneration"],
-            "1970-01-01T00:00:00Z"
-        );
-        assert_eq!(value["decommissionInfo"]["unresolvedEntries"][0]["reason"], "metadata_resolution_failed");
         assert_eq!(value["decommissionInfo"]["waitingReason"], "queued");
     }
 

rustfs/src/app/multipart_usecase.rs

@@ -524,10 +524,9 @@ impl DefaultMultipartUsecase {
                 .await
                 .map_err(ApiError::from)?,
         );
-        let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
         let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
             Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &current_opts)?;
+                validate_existing_object_lock_for_write(&existing_obj_info, &current_opts)?;
                 let physical_size = existing_obj_info.size.max(0) as u64;
                 let logical_size = quota_object_size(&existing_obj_info);
                 Some((physical_size, logical_size))
@@ -898,9 +897,7 @@ impl DefaultMultipartUsecase {
             .await
             .map_err(ApiError::from)?;
         match store.get_object_info(&bucket, &key, &current_opts).await {
-            Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)?
-            }
+            Ok(existing_obj_info) => validate_existing_object_lock_for_write(&existing_obj_info, &opts)?,
             Err(err) => {
                 if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
                     return Err(ApiError::from(err).into());

rustfs/src/app/object_usecase.rs

@@ -39,7 +39,7 @@ use super::storage_api::object_usecase::bucket::{
     metadata_sys,
     object_lock::{
         objectlock::{get_object_legalhold_meta, get_object_retention_meta},
-        objectlock_sys::{check_object_lock_for_deletion, is_retention_active, replication_write_may_pass_worm_gate},
+        objectlock_sys::{check_object_lock_for_deletion, is_retention_active},
     },
     predict_lifecycle_expiration,
     quota::{QuotaCheckResult, QuotaError, QuotaOperation},
@@ -3934,33 +3934,10 @@ fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool {
     opts.version_id.is_none() && opts.versioned && !opts.version_suspended
 }
 
-pub(crate) fn validate_existing_object_lock_for_write(
-    object_lock_config_state: &metadata_sys::ObjectLockConfigState,
-    existing_obj_info: &ObjectInfo,
-    opts: &ObjectOptions,
-) -> S3Result<()> {
+pub(crate) fn validate_existing_object_lock_for_write(existing_obj_info: &ObjectInfo, opts: &ObjectOptions) -> S3Result<()> {
     if put_like_write_creates_new_version(opts) {
         return Ok(());
     }
-    // An authorized replication write may replace the locked version only
-    // when the set layer's commit-lock LWW will judge every locking category,
-    // judged against the bucket's authoritative lock state (default retention
-    // included) exactly like the set-layer gate, which re-checks the same
-    // rule under the lock. A non-authoritative state or malformed lock
-    // metadata fails closed here.
-    if opts.replication_request {
-        let may_pass = replication_write_may_pass_worm_gate(object_lock_config_state, existing_obj_info, opts).map_err(|_| {
-            S3Error::with_message(S3ErrorCode::AccessDenied, "Object Lock state could not be verified.".to_string())
-        })?;
-        return if may_pass {
-            Ok(())
-        } else {
-            Err(S3Error::with_message(
-                S3ErrorCode::AccessDenied,
-                "Object is locked and the replication write carries no source lock decision for it.".to_string(),
-            ))
-        };
-    }
 
     let legal_hold = get_object_legalhold_meta(&existing_obj_info.user_defined);
     if legal_hold
@@ -6127,7 +6104,7 @@ impl DefaultObjectUsecase {
             };
             Some(match previous_current_info {
                 Ok(existing_obj_info) => {
-                    validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)?;
+                    validate_existing_object_lock_for_write(&existing_obj_info, &opts)?;
                     Some(if quota_enabled {
                         quota_object_size(&existing_obj_info).map_err(ApiError::from)?
                     } else {
@@ -7813,7 +7790,7 @@ impl DefaultObjectUsecase {
         }
         let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
             Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &dst_opts)?;
+                validate_existing_object_lock_for_write(&existing_obj_info, &dst_opts)?;
                 if let Some(expected) = expected_current_version_id.as_deref()
                     && existing_obj_info.version_id.unwrap_or_default().to_string() != expected
                 {
@@ -11013,28 +10990,9 @@ mod tests {
         assert_eq!(err.message(), Some(ERR_OBJECT_LOCK_RETENTION_HEADERS_MUST_BE_PAIRED));
     }
 
-    const NO_BUCKET_LOCK: metadata_sys::ObjectLockConfigState = metadata_sys::ObjectLockConfigState::ConfirmedAbsent;
-
-    fn bucket_default_retention_state(mode: &'static str) -> metadata_sys::ObjectLockConfigState {
-        metadata_sys::ObjectLockConfigState::Configured {
-            config: s3s::dto::ObjectLockConfiguration {
-                object_lock_enabled: Some(s3s::dto::ObjectLockEnabled::from_static(s3s::dto::ObjectLockEnabled::ENABLED)),
-                rule: Some(s3s::dto::ObjectLockRule {
-                    default_retention: Some(s3s::dto::DefaultRetention {
-                        mode: Some(ObjectLockRetentionMode::from_static(mode)),
-                        days: Some(1),
-                        years: None,
-                    }),
-                }),
-            },
-            updated_at: OffsetDateTime::now_utc(),
-        }
-    }
-
     fn object_info_with_lock_metadata(metadata: HashMap<String, String>) -> ObjectInfo {
         ObjectInfo {
             user_defined: Arc::new(metadata),
-            mod_time: Some(OffsetDateTime::now_utc()),
             ..Default::default()
         }
     }
@@ -11073,7 +11031,7 @@ mod tests {
             ..Default::default()
         };
 
-        validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts)
+        validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts)
             .expect("versioned put should create a new version");
     }
 
@@ -11085,18 +11043,14 @@ mod tests {
             ..Default::default()
         };
 
-        validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts)
+        validate_existing_object_lock_for_write(&legal_hold_object_info(), &opts)
             .expect("versioned put should create a new version");
     }
 
     #[test]
     fn validate_existing_object_lock_blocks_unversioned_compliance_overwrite() {
-        let err = validate_existing_object_lock_for_write(
-            &NO_BUCKET_LOCK,
-            &compliance_retained_object_info(),
-            &ObjectOptions::default(),
-        )
-        .expect_err("unversioned overwrite should still be blocked");
+        let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &ObjectOptions::default())
+            .expect_err("unversioned overwrite should still be blocked");
 
         assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
     }
@@ -11109,7 +11063,7 @@ mod tests {
             version_id: None,
             ..Default::default()
         };
-        let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts)
+        let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts)
             .expect_err("suspended versioning overwrite should still be blocked");
 
         assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
@@ -11122,86 +11076,12 @@ mod tests {
             version_id: Some(Uuid::new_v4().to_string()),
             ..Default::default()
         };
-        let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts)
+        let err = validate_existing_object_lock_for_write(&compliance_retained_object_info(), &opts)
             .expect_err("explicit version overwrite should still be blocked");
 
         assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
     }
 
-    /// The source's lock state governs the replica (rustfs/backlog#1953):
-    /// an authorized replication write carrying the locking category's source
-    /// timestamp may overwrite a locked version; the set layer's LWW then
-    /// decides per category.
-    #[test]
-    fn validate_existing_object_lock_allows_authorized_replication_overwrite() {
-        let opts = ObjectOptions {
-            versioned: true,
-            version_id: Some(Uuid::new_v4().to_string()),
-            replication_request: true,
-            replication_retention_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-            replication_legalhold_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        };
-
-        validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts)
-            .expect("replication write must bypass the destination COMPLIANCE lock");
-        validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts)
-            .expect("replication write must bypass the destination legal hold");
-    }
-
-    /// Without the locking category's source timestamp the LWW merge cannot
-    /// judge it, so the write stays rejected instead of lifting the lock.
-    #[test]
-    fn validate_existing_object_lock_rejects_replication_overwrite_without_lock_timestamp() {
-        let opts = ObjectOptions {
-            versioned: true,
-            version_id: Some(Uuid::new_v4().to_string()),
-            replication_request: true,
-            replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        };
-
-        let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &compliance_retained_object_info(), &opts)
-            .expect_err("COMPLIANCE lock must hold without a retention source timestamp");
-        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
-        let err = validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &legal_hold_object_info(), &opts)
-            .expect_err("legal hold must hold without a legal-hold source timestamp");
-        assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
-    }
-
-    /// The bucket default retention locks a version without explicit
-    /// retention keys; the pre-check judges the same authoritative state as
-    /// the set-layer gate, so a tagging-only replication write is rejected
-    /// and one carrying the retention source timestamp passes to LWW.
-    #[test]
-    fn validate_existing_object_lock_judges_bucket_default_retention_for_replication_overwrite() {
-        let default_protected = object_info_with_lock_metadata(HashMap::new());
-        let tagging_only = ObjectOptions {
-            versioned: true,
-            version_id: Some(Uuid::new_v4().to_string()),
-            replication_request: true,
-            replication_tagging_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-            ..Default::default()
-        };
-        let with_retention_decision = ObjectOptions {
-            replication_retention_timestamp: Some(OffsetDateTime::UNIX_EPOCH),
-            ..tagging_only.clone()
-        };
-
-        for mode in [ObjectLockRetentionMode::COMPLIANCE, ObjectLockRetentionMode::GOVERNANCE] {
-            let state = bucket_default_retention_state(mode);
-            let err = validate_existing_object_lock_for_write(&state, &default_protected, &tagging_only)
-                .expect_err("bucket default retention must hold without a retention source timestamp");
-            assert_eq!(err.code(), &S3ErrorCode::AccessDenied, "{mode}");
-            validate_existing_object_lock_for_write(&state, &default_protected, &with_retention_decision)
-                .expect("the retention source timestamp hands the default retention to LWW");
-        }
-
-        // Without a bucket default the same version is simply unlocked.
-        validate_existing_object_lock_for_write(&NO_BUCKET_LOCK, &default_protected, &tagging_only)
-            .expect("no bucket default, no lock");
-    }
-
     #[test]
     fn is_put_object_extract_requested_accepts_meta_header() {
         let mut headers = HeaderMap::new();

rustfs/src/app/storage_api.rs

@@ -31,7 +31,6 @@ pub(crate) mod admin {
 }
 
 pub(crate) mod capacity {
-    pub(crate) type DecommissionUnresolvedEntry = crate::storage::storage_api::ecstore_capacity::DecommissionUnresolvedEntry;
     pub(crate) type PoolDecommissionInfo = crate::storage::storage_api::ecstore_capacity::PoolDecommissionInfo;
     pub(crate) type PoolStatus = crate::storage::storage_api::ecstore_capacity::PoolStatus;
     pub(crate) type RebalStatus = crate::storage::storage_api::ecstore_rebalance::RebalStatus;
@@ -593,16 +592,6 @@ pub(crate) mod bucket {
                     retain_until_date,
                 )
             }
-
-            pub(crate) fn replication_write_may_pass_worm_gate(
-                state: &crate::storage::storage_api::ecstore_bucket::metadata_sys::ObjectLockConfigState,
-                obj_info: &crate::storage::storage_api::ObjectInfo,
-                opts: &crate::storage::storage_api::StorageObjectOptions,
-            ) -> Result<bool, crate::storage::storage_api::StorageError> {
-                crate::storage::storage_api::ecstore_bucket::object_lock::objectlock_sys::replication_write_may_pass_worm_gate(
-                    state, obj_info, opts,
-                )
-            }
         }
     }
 

rustfs/src/connect/inventory.rs

@@ -659,10 +659,6 @@ mod tests {
 
     use super::*;
 
-    fn create_directory(path: &Path) -> io::Result<()> {
-        fs::create_dir(path)
-    }
-
     #[test]
     fn inventory_directory_creation_is_synced_before_state_can_be_committed() {
         let temp = tempfile::tempdir().expect("tempdir");
@@ -703,7 +699,7 @@ mod tests {
         fs::create_dir(&root).expect("state root");
         let directory = root.join("inventory");
         let state = directory.join("state.json");
-        let error = prepare_inventory_directory_with(&directory, create_directory, |path| {
+        let error = prepare_inventory_directory_with(&directory, fs::create_dir, |path| {
             if path == directory {
                 Err(io::Error::other("injected leaf sync failure"))
             } else {
@@ -714,7 +710,7 @@ mod tests {
         assert!(matches!(error, InventoryError::StateIo { path, .. } if path == directory));
         assert!(!state.exists());
 
-        let error = prepare_inventory_directory_with(&directory, create_directory, |path| {
+        let error = prepare_inventory_directory_with(&directory, fs::create_dir, |path| {
             if path == root {
                 Err(io::Error::other("injected parent sync failure"))
             } else {
@@ -726,7 +722,7 @@ mod tests {
         assert!(!state.exists());
 
         let mut synced = Vec::new();
-        prepare_inventory_directory_with(&directory, create_directory, |path| {
+        prepare_inventory_directory_with(&directory, fs::create_dir, |path| {
             synced.push(path.to_path_buf());
             Ok(())
         })
@@ -741,7 +737,7 @@ mod tests {
         let directory = root.join("inventory");
 
         assert!(matches!(
-            prepare_inventory_directory_with(&directory, create_directory, |_| Ok(())),
+            prepare_inventory_directory_with(&directory, fs::create_dir, |_| Ok(())),
             Err(InventoryError::StateIo { path, .. }) if path == root
         ));
         assert!(!directory.exists());

rustfs/src/connect/offline/redaction.rs

@@ -173,6 +173,7 @@ pub enum RedactionError {
     NotRepresentable,
 }
 
+#[must_use]
 pub(super) fn redact(source: RedactionSource, document: &Map<String, Value>) -> Result<RedactionResult, RedactionError> {
     let encoded = serde_json::to_vec(document).map_err(|_| RedactionError::NotRepresentable)?;
     if encoded.len() > MAX_INPUT_BYTES {

rustfs/src/storage/rpc/node_service/bucket.rs

@@ -17,8 +17,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{
     BucketOptions, DeleteBucketOptions, MakeBucketOptions,
 };
 use crate::storage::storage_api::rpc_consumer::node_service::{
-    DiskError, StoragePeerS3ClientExt as _, decode_heal_bucket_rpc_options, reload_bucket_metadata, remove_bucket_metadata,
+    DiskError, StoragePeerS3ClientExt as _, reload_bucket_metadata, remove_bucket_metadata,
 };
+use rustfs_common::heal_channel::HealOpts;
 use rustfs_protos::proto_gen::node_service::*;
 use tonic::{Request, Response, Status};
 use tracing::debug;
@@ -238,7 +239,7 @@ impl NodeService {
     ) -> Result<Response<HealBucketResponse>, Status> {
         debug!("heal bucket");
         let request = request.into_inner();
-        let (options, fenced_pools) = match decode_heal_bucket_rpc_options(&request.options) {
+        let options = match serde_json::from_str::<HealOpts>(&request.options) {
             Ok(options) => options,
             Err(err) => {
                 return Ok(Response::new(HealBucketResponse {
@@ -248,11 +249,7 @@ impl NodeService {
             }
         };
 
-        match self
-            .local_peer
-            .heal_bucket_with_fence(&request.bucket, &options, &fenced_pools)
-            .await
-        {
+        match self.local_peer.heal_bucket(&request.bucket, &options).await {
             Ok(_) => Ok(Response::new(HealBucketResponse {
                 success: true,
                 error: None,

rustfs/src/storage/storage_api.rs

@@ -239,7 +239,6 @@ pub(crate) mod rpc_consumer {
     }
 
     pub(crate) mod node_service {
-        pub(crate) use super::super::ecstore_rpc::decode_heal_bucket_rpc_options;
         pub(crate) use super::super::storage_contracts::{
             SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION,
         };
@@ -399,7 +398,7 @@ pub(crate) mod ecstore_bucket {
 
 pub(crate) mod ecstore_capacity {
     pub(crate) use rustfs_ecstore::api::capacity::{
-        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
+        PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
         is_reserved_or_invalid_bucket,
     };
 }
@@ -521,7 +520,7 @@ pub(crate) mod ecstore_rpc {
     pub(crate) use rustfs_ecstore::api::rpc::{
         KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient,
         PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX,
-        check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience,
+        check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience,
         sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
         tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
         verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
@@ -1409,11 +1408,10 @@ where
 }
 
 pub(crate) trait StoragePeerS3ClientExt {
-    async fn heal_bucket_with_fence(
+    async fn heal_bucket(
         &self,
         bucket: &str,
         opts: &rustfs_common::heal_channel::HealOpts,
-        fenced_pools: &[usize],
     ) -> DiskResult<rustfs_madmin::heal_commands::HealResultItem>;
     async fn make_bucket(&self, bucket: &str, opts: &contract::bucket::MakeBucketOptions) -> DiskResult<()>;
     async fn list_bucket(&self, opts: &contract::bucket::BucketOptions) -> DiskResult<Vec<contract::bucket::BucketInfo>>;
@@ -1426,13 +1424,12 @@ pub(crate) trait StoragePeerS3ClientExt {
 }
 
 impl StoragePeerS3ClientExt for LocalPeerS3Client {
-    async fn heal_bucket_with_fence(
+    async fn heal_bucket(
         &self,
         bucket: &str,
         opts: &rustfs_common::heal_channel::HealOpts,
-        fenced_pools: &[usize],
     ) -> DiskResult<rustfs_madmin::heal_commands::HealResultItem> {
-        ecstore_rpc::PeerS3Client::heal_bucket_with_fence(self, bucket, opts, fenced_pools).await
+        ecstore_rpc::PeerS3Client::heal_bucket(self, bucket, opts).await
     }
 
     async fn make_bucket(&self, bucket: &str, opts: &contract::bucket::MakeBucketOptions) -> DiskResult<()> {

rustfs/tests/connect_registration_bootstrap.rs

@@ -158,15 +158,6 @@ impl Drop for TestServer {
     }
 }
 
-fn assert_reused_pending_requests(server: &TestServer) {
-    let requests = server.seen.lock().expect("seen lock");
-    assert_eq!(requests.len(), 4);
-    for request in &requests[1..] {
-        assert_eq!(request["requestId"], requests[0]["requestId"]);
-        assert_eq!(request["certificateRequest"], requests[0]["certificateRequest"]);
-    }
-}
-
 async fn server(state_directory: &std::path::Path, replies: Vec<Reply>) -> TestServer {
     let pki = Arc::new(TestPki::new());
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind test server");
@@ -384,7 +375,7 @@ async fn production_command_never_echoes_a_remote_reason() {
 async fn response_loss_reuses_the_pending_request_and_existing_credential_is_idempotent() {
     let temp = secure_tempdir();
     let state = temp.path().join("state");
-    let flaky_server = server(
+    let server = server(
         &state,
         vec![
             Reply::DropConnection,
@@ -394,21 +385,27 @@ async fn response_loss_reuses_the_pending_request_and_existing_credential_is_ide
         ],
     )
     .await;
-    let (root, token) = prepare_inputs(&temp, &flaky_server.root_pem);
+    let (root, token) = prepare_inputs(&temp, &server.root_pem);
 
-    while flaky_server.seen.lock().expect("seen lock").len() < 3 {
-        let error = register_from_protected_input(&flaky_server.endpoint, &root, &state, Some(&token))
+    while server.seen.lock().expect("seen lock").len() < 3 {
+        let error = register_from_protected_input(&server.endpoint, &root, &state, Some(&token))
             .await
             .expect_err("lost response must leave a retryable failure");
         assert!(!error.to_string().contains(TOKEN_SECRET));
         assert!(state.join("credential/registration.pending.json").is_file());
     }
 
-    let registered = register_from_protected_input(&flaky_server.endpoint, &root, &state, Some(&token))
+    let registered = register_from_protected_input(&server.endpoint, &root, &state, Some(&token))
         .await
         .expect("retry registration");
     assert_eq!(registered.device_uid, DEVICE_UID);
-    assert_reused_pending_requests(&flaky_server);
+    let requests = server.seen.lock().expect("seen lock");
+    assert_eq!(requests.len(), 4);
+    for request in &requests[1..] {
+        assert_eq!(request["requestId"], requests[0]["requestId"]);
+        assert_eq!(request["certificateRequest"], requests[0]["certificateRequest"]);
+    }
+    drop(requests);
 
     let idle = server(&state, vec![]).await;
     let idempotent = register_from_protected_input(&idle.endpoint, &root, &state, Some(&token))

scripts/README.md

@@ -73,7 +73,6 @@ their issue closes.
 | `run_pinned_paired_abba_bench.sh` | dev-tool | Pinned RustFS/MinIO paired ABBA benchmark orchestrator for backlog#1432 | `test_pinned_paired_abba_bench.sh` |
 | `run_get_codec_streaming_smoke.sh` | dev-tool | Local GET benchmark harness for the codec streaming read path | `docs/testing/ecstore-validation-suite-design.md` |
 | `run_get_1mib_abba_stage_metrics.sh` | dev-tool | Exact-1MiB isolated-host GET ABBA/stage-metrics harness for backlog#1434 | `test_get_1mib_abba_stage_metrics.sh` |
-| `issue_2007_coalescer_prometheus_report.py` | dev-tool | Read-only Prometheus report for GET metadata coalescer delay cost validation | `test_issue_2007_coalescer_prometheus_report.sh`; `docs/testing/issue-2007-coalescer-delay.md` |
 | `run_gt1g_get_http_matrix.sh` | dev-tool | >1 GiB GET HTTP matrix | `docs/testing/ecstore-validation-suite-design.md` |
 | `run_gt1g_multipart_put_matrix.sh` | dev-tool | >1 GiB multipart PUT matrix | `docs/testing/ecstore-validation-suite-design.md` |
 | `sample_remote_rustfs_rss.sh` | dev-tool | Remote RustFS PID CPU/RSS TSV sampler for hotpath profiling runs | `test_sample_remote_rustfs_rss.sh`; backlog#1647 |

scripts/issue_2007_coalescer_prometheus_report.py

@@ -1,329 +0,0 @@
-#!/usr/bin/env python3
-"""Read-only Prometheus report for rustfs/backlog#2007 coalescer delay runs.
-
-The script queries Prometheus' instant-query API and prints a Markdown summary
-for one already-completed workload window. It never writes to RustFS,
-Prometheus, or scrape targets.
-"""
-
-from __future__ import annotations
-
-import argparse
-import base64
-import json
-import math
-from dataclasses import dataclass
-from typing import Any, Iterable
-from urllib.error import HTTPError, URLError
-from urllib.parse import urlencode, urlparse
-from urllib.request import Request, urlopen
-
-
-RUSTFS_SELECTOR_HELP = "PromQL label selector applied to RustFS metrics, for example 'server=~\"node[5-8]\"'"
-NODE_SELECTOR_HELP = "PromQL label selector applied to node-exporter metrics, for example 'instance=~\"node[5-8].*\"'"
-
-
-@dataclass(frozen=True)
-class Sample:
-    labels: dict[str, str]
-    value: float
-
-
-@dataclass(frozen=True)
-class QueryResult:
-    name: str
-    query: str
-    samples: list[Sample]
-    error: str | None = None
-
-    def scalar_sum(self) -> float | None:
-        if self.error or not self.samples:
-            return None
-        return sum(sample.value for sample in self.samples)
-
-
-def query_url(value: str) -> str:
-    parsed = urlparse(value)
-    if parsed.path.rstrip("/").endswith("/api/v1/query"):
-        return value
-    return value.rstrip("/") + "/api/v1/query"
-
-
-def braces(selector: str = "", *pairs: tuple[str, str]) -> str:
-    labels = [selector.strip().strip("{}")] if selector.strip() else []
-    labels.extend(f'{key}="{value}"' for key, value in pairs)
-    return "{" + ",".join(label for label in labels if label) + "}"
-
-
-def braces_with_raw(selector: str = "", *raw_labels: str) -> str:
-    labels = [selector.strip().strip("{}")] if selector.strip() else []
-    labels.extend(raw_labels)
-    return "{" + ",".join(label for label in labels if label) + "}"
-
-
-def parse_vector(payload: dict[str, Any]) -> list[Sample]:
-    if payload.get("status") != "success":
-        raise RuntimeError(f"Prometheus returned non-success: {payload}")
-    data = payload.get("data", {})
-    if data.get("resultType") != "vector":
-        raise RuntimeError(f"Prometheus query did not return an instant vector: {payload}")
-    samples: list[Sample] = []
-    for item in data.get("result", []):
-        value = item.get("value", [None, "nan"])[1]
-        try:
-            parsed_value = float(value)
-        except (TypeError, ValueError):
-            parsed_value = math.nan
-        samples.append(Sample(dict(item.get("metric", {})), parsed_value))
-    return samples
-
-
-def fetch(endpoint: str, query: str, headers: dict[str, str], timeout: float) -> list[Sample]:
-    request = Request(f"{endpoint}?{urlencode({'query': query})}", headers=headers)
-    try:
-        with urlopen(request, timeout=timeout) as response:
-            payload = json.load(response)
-    except (HTTPError, URLError, TimeoutError) as error:
-        raise RuntimeError(f"Prometheus query failed for {query!r}: {error}") from error
-    return parse_vector(payload)
-
-
-def run_query(endpoint: str, headers: dict[str, str], timeout: float, name: str, query: str) -> QueryResult:
-    try:
-        return QueryResult(name=name, query=query, samples=fetch(endpoint, query, headers, timeout))
-    except RuntimeError as error:
-        return QueryResult(name=name, query=query, samples=[], error=str(error))
-
-
-def fmt_value(value: float | None, suffix: str = "", precision: int = 2) -> str:
-    if value is None or math.isnan(value):
-        return "UNAVAILABLE"
-    if math.isinf(value):
-        return "inf"
-    return f"{value:.{precision}f}{suffix}"
-
-
-def fmt_count(value: float | None) -> str:
-    if value is None or math.isnan(value):
-        return "UNAVAILABLE"
-    return f"{value:.0f}"
-
-
-def batch_distribution(samples: Iterable[Sample]) -> tuple[float, float, float, list[tuple[int, float]]]:
-    total_batches = 0.0
-    total_items = 0.0
-    single_item = 0.0
-    rows: list[tuple[int, float]] = []
-    for sample in samples:
-        raw_count = sample.labels.get("item_count", "")
-        if not raw_count.isdigit():
-            continue
-        item_count = int(raw_count)
-        count = sample.value
-        rows.append((item_count, count))
-        total_batches += count
-        total_items += item_count * count
-        if item_count == 1:
-            single_item += count
-    avg_batch_size = total_items / total_batches if total_batches else math.nan
-    single_item_ratio = single_item / total_batches if total_batches else math.nan
-    return total_batches, avg_batch_size, single_item_ratio, sorted(rows)
-
-
-def build_queries(window: str, rustfs_selector: str, coalescer_selector: str, node_selector: str) -> dict[str, str]:
-    read_version = braces(rustfs_selector, ("operation", "grpc_read_version"), ("backend", "grpc"))
-    batch_read_version = braces(rustfs_selector, ("operation", "grpc_batch_read_version"), ("backend", "grpc"))
-    coalescer = braces(coalescer_selector, ("event", "attempted_batch"))
-    cpu = braces(node_selector, ("mode", "idle"))
-    node = braces(node_selector)
-    net = braces_with_raw(node_selector, 'device!~"lo|docker.*|veth.*|br-.*|cni.*"')
-    disk = braces_with_raw(node_selector, 'device!~"loop.*|ram.*|dm-.*"')
-    return {
-        "grpc_read_version_requests": (
-            "sum(increase(rustfs_system_network_internode_operation_requests_outgoing_total"
-            f"{read_version}[{window}]))"
-        ),
-        "grpc_batch_read_version_requests": (
-            "sum(increase(rustfs_system_network_internode_operation_requests_outgoing_total"
-            f"{batch_read_version}[{window}]))"
-        ),
-        "coalescer_batches_by_item_count": (
-            "sum by (item_count) (increase(rustfs_get_metadata_read_version_coalescer_total"
-            f"{coalescer}[{window}]))"
-        ),
-        "coalescer_wait_p99_ms": (
-            "histogram_quantile(0.99, sum by (le) (rate("
-            "rustfs_system_network_internode_operation_stage_duration_ms_bucket"
-            f'{braces(rustfs_selector, ("operation", "grpc_batch_read_version"), ("backend", "grpc"), ("stage", "batch_read_version_coalescer_wait"))}'
-            f"[{window}])))"
-        ),
-        "batch_rpc_roundtrip_p99_ms": (
-            "histogram_quantile(0.99, sum by (le) (rate("
-            "rustfs_system_network_internode_operation_stage_duration_ms_bucket"
-            f'{braces(rustfs_selector, ("operation", "grpc_batch_read_version"), ("backend", "grpc"), ("stage", "batch_read_version_rpc_roundtrip"))}'
-            f"[{window}])))"
-        ),
-        "batch_disk_read_p99_ms": (
-            "histogram_quantile(0.99, sum by (le) (rate("
-            "rustfs_system_network_internode_operation_stage_duration_ms_bucket"
-            f'{braces(rustfs_selector, ("operation", "grpc_batch_read_version"), ("backend", "grpc"), ("stage", "batch_read_version_disk_read"))}'
-            f"[{window}])))"
-        ),
-        "batch_response_map_p99_ms": (
-            "histogram_quantile(0.99, sum by (le) (rate("
-            "rustfs_system_network_internode_operation_stage_duration_ms_bucket"
-            f'{braces(rustfs_selector, ("operation", "grpc_batch_read_version"), ("backend", "grpc"), ("stage", "batch_read_version_response_map"))}'
-            f"[{window}])))"
-        ),
-        "node_cpu_busy_percent": f"100 * (1 - avg(rate(node_cpu_seconds_total{cpu}[{window}])))",
-        "node_network_receive_bytes_per_sec": f"sum(rate(node_network_receive_bytes_total{net}[{window}]))",
-        "node_network_transmit_bytes_per_sec": f"sum(rate(node_network_transmit_bytes_total{net}[{window}]))",
-        "node_disk_read_await_ms": (
-            "1000 * sum(rate(node_disk_read_time_seconds_total"
-            f"{disk}[{window}])) / clamp_min(sum(rate(node_disk_reads_completed_total{disk}[{window}])), 1)"
-        ),
-        "node_disk_avg_queue_depth": (
-            "sum(rate(node_disk_io_time_weighted_seconds_total"
-            f"{disk}[{window}]))"
-        ),
-        "node_disk_util_percent": f"100 * sum(rate(node_disk_io_time_seconds_total{disk}[{window}]))",
-        "node_up": f"sum(up{node})",
-    }
-
-
-def render_report(args: argparse.Namespace, results: dict[str, QueryResult]) -> str:
-    read_version = results["grpc_read_version_requests"].scalar_sum()
-    batch_read_version = results["grpc_batch_read_version_requests"].scalar_sum()
-    total_rpc = (read_version or 0.0) + (batch_read_version or 0.0)
-    batch_ratio = batch_read_version / total_rpc if total_rpc else math.nan
-    total_batches, avg_batch_size, single_item_ratio, distribution = batch_distribution(
-        results["coalescer_batches_by_item_count"].samples
-    )
-
-    lines = [
-        f"## backlog#2007 coalescer cost report: {args.profile}",
-        "",
-        f"- Window: `{args.window}`",
-        f"- RustFS selector: `{args.rustfs_selector or '<none>'}`",
-        f"- Coalescer selector: `{args.coalescer_selector or '<none>'}`",
-        f"- Node selector: `{args.node_selector or '<none>'}`",
-        "",
-        "| Signal | Value |",
-        "|---|---:|",
-        f"| outgoing grpc_read_version requests | {fmt_count(read_version)} |",
-        f"| outgoing grpc_batch_read_version requests | {fmt_count(batch_read_version)} |",
-        f"| batch RPC share | {fmt_value(batch_ratio * 100 if not math.isnan(batch_ratio) else math.nan, '%')} |",
-        f"| coalescer batches | {fmt_count(total_batches)} |",
-        f"| avg coalesced batch size | {fmt_value(avg_batch_size)} |",
-        f"| single-item batch ratio | {fmt_value(single_item_ratio * 100 if not math.isnan(single_item_ratio) else math.nan, '%')} |",
-        f"| coalescer_wait p99 | {fmt_value(results['coalescer_wait_p99_ms'].scalar_sum(), ' ms')} |",
-        f"| batch rpc_roundtrip p99 | {fmt_value(results['batch_rpc_roundtrip_p99_ms'].scalar_sum(), ' ms')} |",
-        f"| batch disk_read p99 | {fmt_value(results['batch_disk_read_p99_ms'].scalar_sum(), ' ms')} |",
-        f"| batch response_map p99 | {fmt_value(results['batch_response_map_p99_ms'].scalar_sum(), ' ms')} |",
-        f"| node CPU busy | {fmt_value(results['node_cpu_busy_percent'].scalar_sum(), '%')} |",
-        f"| node network RX | {fmt_value(results['node_network_receive_bytes_per_sec'].scalar_sum(), ' B/s')} |",
-        f"| node network TX | {fmt_value(results['node_network_transmit_bytes_per_sec'].scalar_sum(), ' B/s')} |",
-        f"| node disk read await | {fmt_value(results['node_disk_read_await_ms'].scalar_sum(), ' ms')} |",
-        f"| node disk avg queue depth | {fmt_value(results['node_disk_avg_queue_depth'].scalar_sum())} |",
-        f"| node disk util | {fmt_value(results['node_disk_util_percent'].scalar_sum(), '%')} |",
-        f"| node-exporter up series | {fmt_count(results['node_up'].scalar_sum())} |",
-        "",
-        "### Batch distribution",
-        "",
-        "| item_count | batches |",
-        "|---:|---:|",
-    ]
-    if distribution:
-        lines.extend(f"| {item_count} | {fmt_count(count)} |" for item_count, count in distribution)
-    else:
-        lines.append("| UNAVAILABLE | UNAVAILABLE |")
-
-    unavailable = [result for result in results.values() if result.error or not result.samples]
-    if unavailable:
-        lines.extend(["", "### Unavailable queries", ""])
-        for result in unavailable:
-            reason = result.error or "no series returned"
-            lines.append(f"- `{result.name}`: {reason}")
-
-    if args.show_queries:
-        lines.extend(["", "### PromQL", ""])
-        for result in results.values():
-            lines.append(f"- `{result.name}`: `{result.query}`")
-
-    return "\n".join(lines)
-
-
-def run(args: argparse.Namespace) -> int:
-    headers = {"Accept": "application/json"}
-    if args.bearer:
-        headers["Authorization"] = f"Bearer {args.bearer}"
-    if args.basic:
-        headers["Authorization"] = "Basic " + base64.b64encode(args.basic.encode()).decode()
-
-    endpoint = query_url(args.query_url)
-    queries = build_queries(args.window, args.rustfs_selector, args.coalescer_selector, args.node_selector)
-    results = {
-        name: run_query(endpoint, headers, args.timeout, name, query)
-        for name, query in queries.items()
-    }
-    print(render_report(args, results))
-    return 0
-
-
-def self_test() -> None:
-    assert query_url("http://prom:9090") == "http://prom:9090/api/v1/query"
-    assert query_url("http://prom:9090/api/v1/query") == "http://prom:9090/api/v1/query"
-    assert braces('server=~"node[5-8]"', ("operation", "grpc_batch_read_version")) == (
-        '{server=~"node[5-8]",operation="grpc_batch_read_version"}'
-    )
-    assert braces_with_raw("", 'device!~"lo"') == '{device!~"lo"}'
-    assert braces_with_raw('instance=~"node.*"', 'device!~"lo"') == '{instance=~"node.*",device!~"lo"}'
-    payload = {
-        "status": "success",
-        "data": {
-            "resultType": "vector",
-            "result": [
-                {"metric": {"item_count": "1"}, "value": [1, "2"]},
-                {"metric": {"item_count": "4"}, "value": [1, "3"]},
-            ],
-        },
-    }
-    samples = parse_vector(payload)
-    total_batches, avg_batch_size, single_item_ratio, rows = batch_distribution(samples)
-    assert total_batches == 5
-    assert avg_batch_size == 2.8
-    assert single_item_ratio == 0.4
-    assert rows == [(1, 2.0), (4, 3.0)]
-    queries = build_queries("5m", "", "", "")
-    assert "increase(rustfs_get_metadata_read_version_coalescer_total" in queries["coalescer_batches_by_item_count"]
-    print("PASS: self-test")
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--query-url", help="Prometheus base URL or /api/v1/query endpoint")
-    parser.add_argument("--profile", default="unknown", help="Run label printed in the report, e.g. delay-200us or delay-50us")
-    parser.add_argument("--window", default="5m", help="PromQL range selector covering the measured workload window")
-    parser.add_argument("--rustfs-selector", default="", help=RUSTFS_SELECTOR_HELP)
-    parser.add_argument(
-        "--coalescer-selector",
-        default="",
-        help="PromQL label selector for rustfs_get_metadata_read_version_coalescer_total; leave empty if it has no server labels",
-    )
-    parser.add_argument("--node-selector", default="", help=NODE_SELECTOR_HELP)
-    parser.add_argument("--bearer")
-    parser.add_argument("--basic", help="username:password; prefer --bearer in shared shells")
-    parser.add_argument("--timeout", type=float, default=10.0)
-    parser.add_argument("--show-queries", action="store_true")
-    parser.add_argument("--self-test", action="store_true")
-    args = parser.parse_args()
-    if args.self_test:
-        self_test()
-        return 0
-    if not args.query_url:
-        parser.error("--query-url is required unless --self-test is used")
-    return run(args)
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/test_issue_2007_coalescer_prometheus_report.sh

@@ -1,25 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-
-"${SCRIPT_DIR}/issue_2007_coalescer_prometheus_report.py" --self-test >/dev/null
-
-output="$("${SCRIPT_DIR}/issue_2007_coalescer_prometheus_report.py" \
-  --query-url http://prometheus.example:9090 \
-  --profile delay-50us \
-  --window 180s \
-  --rustfs-selector 'server=~"node[5-8]"' \
-  --node-selector 'instance=~"node[5-8].*"' \
-  --show-queries \
-  --timeout 0.001 || true)"
-
-printf '%s\n' "$output" | rg -Fq '## backlog#2007 coalescer cost report: delay-50us'
-printf '%s\n' "$output" | rg -Fq 'Window: `180s`'
-printf '%s\n' "$output" | rg -Fq 'RustFS selector: `server=~"node[5-8]"`'
-printf '%s\n' "$output" | rg -Fq 'Node selector: `instance=~"node[5-8].*"`'
-printf '%s\n' "$output" | rg -Fq 'outgoing grpc_batch_read_version requests'
-printf '%s\n' "$output" | rg -Fq 'single-item batch ratio'
-printf '%s\n' "$output" | rg -Fq 'batch_read_version_response_map'
-printf '%s\n' "$output" | rg -Fq 'node_disk_read_time_seconds_total'
-printf '%s\n' "$output" | rg -Fq '### Unavailable queries'