fix(kms): use .expect() instead of ? in test functions that return ()

7 call sites of wait_for_kms_ready() used ? in async test functions
that return () instead of Result. Changed to .expect("KMS ready").

crates/e2e_test/src/kms/bucket_default_encryption_test.rs

@@ -37,7 +37,7 @@ async fn test_bucket_default_sse_s3_put_object() -> Result<(), Box<dyn std::erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -159,7 +159,7 @@ async fn test_bucket_default_sse_kms_put_object() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -278,7 +278,7 @@ async fn test_bucket_default_sse_kms_multipart_crc32() -> Result<(), Box<dyn std
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -475,7 +475,7 @@ async fn test_explicit_encryption_overrides_bucket_default() -> Result<(), Box<d
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -570,7 +570,7 @@ async fn test_sse_kms_without_key_id_populates_default() -> Result<(), Box<dyn s
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/copy_object_self_copy_sse_test.rs

@@ -61,7 +61,7 @@ async fn test_metadata_replace_self_copy_of_sse_object_stays_decryptable() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     let client = kms_env.base_env.create_s3_client();
     // Deliberately an UNVERSIONED bucket: that is the branch where the store layer can service
@@ -160,7 +160,7 @@ async fn test_metadata_replace_self_copy_dropping_sse_rewrites_plaintext() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     let client = kms_env.base_env.create_s3_client();
     // Unversioned, and deliberately WITHOUT a bucket default-encryption rule, so the copy below
@@ -256,7 +256,7 @@ async fn test_metadata_replace_self_copy_under_bucket_default_sse_stays_decrypta
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     let client = kms_env.base_env.create_s3_client();
     let bucket = "copy-object-self-copy-bucket-default-sse-test";

crates/e2e_test/src/kms/copy_object_version_restore_sse_test.rs

@@ -56,7 +56,7 @@ async fn test_self_copy_of_historical_sse_s3_version_is_readable() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     let client = kms_env.base_env.create_s3_client();
     let bucket = "copy-object-version-restore-sse-test";

crates/e2e_test/src/kms/encryption_metadata_test.rs

@@ -87,7 +87,7 @@ async fn test_head_reports_managed_metadata_for_sse_s3() -> Result<(), Box<dyn s
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -147,7 +147,7 @@ async fn test_head_reports_managed_metadata_for_sse_kms_and_copy() -> Result<(),
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -250,7 +250,7 @@ async fn test_multipart_upload_writes_encrypted_data() -> Result<(), Box<dyn std
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_comprehensive_test.rs

@@ -24,7 +24,6 @@ use super::common::{
     test_sse_kms_encryption, test_sse_s3_encryption,
 };
 use crate::common::{TEST_BUCKET, init_logging};
-use tokio::time::{Duration, sleep};
 use tracing::info;
 
 /// Comprehensive test: Full KMS workflow with all encryption types
@@ -35,7 +34,7 @@ async fn test_comprehensive_kms_full_workflow() -> Result<(), Box<dyn std::error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -103,7 +102,7 @@ async fn test_comprehensive_stress_test() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -137,7 +136,7 @@ async fn test_comprehensive_key_isolation() -> Result<(), Box<dyn std::error::Er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -208,7 +207,7 @@ async fn test_comprehensive_concurrent_operations() -> Result<(), Box<dyn std::e
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -253,7 +252,7 @@ async fn test_comprehensive_performance_benchmark() -> Result<(), Box<dyn std::e
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_edge_cases_test.rs

@@ -44,7 +44,7 @@ async fn test_kms_zero_byte_file_encryption() -> Result<(), Box<dyn std::error::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -117,7 +117,7 @@ async fn test_kms_single_byte_file_encryption() -> Result<(), Box<dyn std::error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -209,7 +209,7 @@ async fn test_kms_multipart_boundary_conditions() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -284,7 +284,7 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -371,7 +371,7 @@ async fn test_kms_concurrent_encryption() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = Arc::new(kms_env.base_env.create_s3_client());
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -478,7 +478,7 @@ async fn test_kms_key_validation_security() -> Result<(), Box<dyn std::error::Er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_fault_recovery_test.rs

@@ -37,7 +37,7 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -127,7 +127,7 @@ async fn test_kms_corrupted_key_files() -> Result<(), Box<dyn std::error::Error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -218,7 +218,7 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -401,7 +401,7 @@ async fn test_kms_resource_constraints() -> Result<(), Box<dyn std::error::Error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_local_test.rs

@@ -46,7 +46,7 @@ async fn test_local_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -127,7 +127,7 @@ async fn test_local_kms_key_isolation() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -227,7 +227,7 @@ async fn test_local_kms_large_file() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -309,7 +309,7 @@ async fn test_local_kms_multipart_upload() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait for KMS initialization
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await.expect("KMS ready");
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 

crates/e2e_test/src/kms/kms_vault_test.rs

@@ -19,7 +19,7 @@
 //! multipart upload behaviour.
 
 use crate::common::{TEST_BUCKET, init_logging};
-use tokio::time::{Duration, sleep};
+use serial_test::serial;
 use tracing::{error, info};
 
 use super::common::{
@@ -45,8 +45,8 @@ impl VaultKmsTestContext {
 
         start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?;
 
-        // Allow Vault to finish initialising token auth and transit engine.
-        sleep(Duration::from_secs(2)).await;
+        // Wait for KMS to finish initialising.
+        super::common::wait_for_kms_ready(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?;
 
         Ok(Self { env })
     }

crates/e2e_test/src/kms/multipart_encryption_test.rs

@@ -33,7 +33,7 @@ async fn test_step1_basic_single_file_encryption() -> Result<(), Box<dyn std::er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -89,7 +89,7 @@ async fn test_step2_basic_multipart_upload_without_encryption() -> Result<(), Bo
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -187,7 +187,7 @@ async fn test_step3_multipart_upload_with_sse_s3() -> Result<(), Box<dyn std::er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -310,7 +310,7 @@ async fn test_step4_large_multipart_upload_with_encryption() -> Result<(), Box<d
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -435,7 +435,7 @@ async fn test_step5_all_encryption_types_multipart() -> Result<(), Box<dyn std::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
+    kms_env.wait_for_kms_ready().await?;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

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

@@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
     BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
-    DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
-    MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
-    ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
-    RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
+    ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
+    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
+    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
     SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
     WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
@@ -112,6 +112,28 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
 
+fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
+    if !response.item_errors.is_empty() {
+        if response.item_errors.len() != expected_len {
+            return vec![Some(Error::other("malformed delete_versions item errors")); expected_len];
+        }
+        return response
+            .item_errors
+            .into_iter()
+            .map(|error| (error.code != 0).then(|| error.into()))
+            .collect();
+    }
+
+    if response.errors.len() != expected_len {
+        return vec![Some(Error::other("malformed delete_versions errors")); expected_len];
+    }
+    response
+        .errors
+        .into_iter()
+        .map(|error| (!error.is_empty()).then(|| Error::other(error)))
+        .collect()
+}
+
 fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
     if !response.success {
         return Err(response.error.unwrap_or_default().into());
@@ -2406,8 +2428,6 @@ impl DiskAPI for RemoteDisk {
             return errors;
         }
 
-        // TODO(backlog): replace string errors with typed `StorageError` variants
-
         let result = self
             .execute_with_timeout(
                 || async {
@@ -2439,17 +2459,7 @@ impl DiskAPI for RemoteDisk {
             }
             return errors;
         }
-        response
-            .errors
-            .iter()
-            .map(|error| {
-                if error.is_empty() {
-                    None
-                } else {
-                    Some(Error::other(error.to_string()))
-                }
-            })
-            .collect()
+        decode_delete_versions_errors(response, versions.len())
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -3760,6 +3770,63 @@ mod tests {
 
     static INIT: Once = Once::new();
 
+    #[test]
+    fn delete_versions_response_preserves_typed_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string(), String::new()],
+                error: None,
+                item_errors: vec![
+                    rustfs_protos::proto_gen::node_service::Error {
+                        code: DiskError::FileNotFound.to_u32(),
+                        error_info: "file not found".to_string(),
+                    },
+                    rustfs_protos::proto_gen::node_service::Error::default(),
+                ],
+            },
+            2,
+        );
+
+        assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None]));
+    }
+
+    #[test]
+    fn delete_versions_response_accepts_legacy_string_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["legacy error".to_string(), String::new()],
+                error: None,
+                item_errors: Vec::new(),
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error"));
+        assert!(errors[1].is_none());
+    }
+
+    #[test]
+    fn delete_versions_response_rejects_misaligned_item_errors() {
+        let errors = decode_delete_versions_errors(
+            DeleteVersionsResponse {
+                success: true,
+                errors: vec!["file not found".to_string()],
+                error: None,
+                item_errors: vec![rustfs_protos::proto_gen::node_service::Error {
+                    code: DiskError::FileNotFound.to_u32(),
+                    error_info: "file not found".to_string(),
+                }],
+            },
+            2,
+        );
+
+        assert_eq!(errors.len(), 2);
+        assert!(errors.iter().all(Option::is_some));
+    }
+
     #[test]
     fn disk_mutation_digest_marks_rolling_compatibility() {
         let mut request = Request::new(());

crates/ecstore/src/io_support/bitrot.rs

@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
-/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
-///
-/// A known length is grown by one checksum per shard so the on-disk file size
-/// matches what the bitrot writer emits. A negative length is the
-/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
-/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
-/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
-/// a fixed body length when locating the authenticated trailer. Clamping it
-/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
-/// because a genuinely empty object still means an empty body.
-fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
-    if length <= 0 {
-        return length;
-    }
-    let length = length as usize;
-    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-}
-
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
+        let length = if length > 0 {
+            let length = length as usize;
+            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+        } else {
+            0
+        };
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -833,25 +820,6 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
-    #[test]
-    fn bitrot_create_file_size_grows_known_length_by_checksums() {
-        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
-        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
-        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
-    }
-
-    #[test]
-    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
-        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
-        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
-        // put_file_stream receiver relies on a non-positive size to parse the auth
-        // trailer from the stream tail, so the sentinel must survive untouched.
-        assert_eq!(
-            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
-            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
-        );
-    }
-
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/protos/src/generated/proto_gen/node_service.rs

@@ -722,6 +722,10 @@ pub struct DeleteVersionsResponse {
     pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
+    /// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+    /// when present and fall back to strings for peers that predate this field. Code zero means success.
+    #[prost(message, repeated, tag = "4")]
+    pub item_errors: ::prost::alloc::vec::Vec<Error>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleRequest {

crates/protos/src/node.proto

@@ -493,6 +493,9 @@ message DeleteVersionsResponse {
   bool success = 1;
   repeated string errors = 2;
   optional Error error = 3;
+  // Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
+  // when present and fall back to strings for peers that predate this field. Code zero means success.
+  repeated Error item_errors = 4;
 }
 
 message ReadMultipleRequest {

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

@@ -146,6 +146,29 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
     encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
 }
 
+fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
+    let mut errors = Vec::with_capacity(disk_errors.len());
+    let mut item_errors = Vec::with_capacity(disk_errors.len());
+    for error in disk_errors {
+        match error {
+            Some(error) => {
+                let code = match &error {
+                    DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(),
+                    _ => error.to_u32(),
+                };
+                let error_info = error.to_string();
+                errors.push(error_info.clone());
+                item_errors.push(Error { code, error_info });
+            }
+            None => {
+                errors.push(String::new());
+                item_errors.push(Error::default());
+            }
+        }
+    }
+    (errors, item_errors)
+}
+
 fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
     let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
     value
@@ -552,6 +575,7 @@ impl NodeService {
                             success: false,
                             errors: Vec::new(),
                             error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
+                            item_errors: Vec::new(),
                         }));
                     }
                 };
@@ -563,30 +587,26 @@ impl NodeService {
                         success: false,
                         errors: Vec::new(),
                         error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
+                        item_errors: Vec::new(),
                     }));
                 }
             };
 
-            let errors = disk
-                .delete_versions(&request.volume, versions, opts)
-                .await
-                .into_iter()
-                .map(|error| match error {
-                    Some(e) => e.to_string(),
-                    None => "".to_string(),
-                })
-                .collect();
+            let (errors, item_errors) =
+                encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
 
             Ok(Response::new(DeleteVersionsResponse {
                 success: true,
                 errors,
                 error: None,
+                item_errors,
             }))
         } else {
             Ok(Response::new(DeleteVersionsResponse {
                 success: false,
                 errors: Vec::new(),
                 error: Some(DiskError::other("cannot find disk".to_string()).into()),
+                item_errors: Vec::new(),
             }))
         }
     }
@@ -1612,8 +1632,8 @@ impl NodeService {
 mod tests {
     use super::{
         compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
-        encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
-        encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
+        encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
+        encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
     };
     use crate::storage::rpc::node_service::make_server;
     use crate::storage::storage_api::ReadMultipleResp;
@@ -1632,6 +1652,18 @@ mod tests {
         count: u32,
     }
 
+    #[test]
+    fn delete_versions_response_dual_writes_typed_item_errors() {
+        let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
+        let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]);
+
+        assert!(errors[0].starts_with("io error "));
+        assert!(errors[1].is_empty());
+        assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32());
+        assert_eq!(item_errors[0].error_info, errors[0]);
+        assert_eq!(item_errors[1].code, 0);
+    }
+
     #[tokio::test]
     #[serial]
     async fn handle_read_version_records_attribution_for_missing_disk() {