mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 12:26:37 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a3c812527f |
@@ -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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -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");
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
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");
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
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");
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let client = kms_env.base_env.create_s3_client();
|
||||
let bucket = "copy-object-self-copy-bucket-default-sse-test";
|
||||
|
||||
@@ -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");
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let client = kms_env.base_env.create_s3_client();
|
||||
let bucket = "copy-object-version-restore-sse-test";
|
||||
|
||||
@@ -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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -24,6 +24,7 @@ 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
|
||||
@@ -34,7 +35,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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
@@ -102,7 +103,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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
@@ -136,7 +137,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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
@@ -207,7 +208,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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
@@ -252,7 +253,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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -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
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
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
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
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
|
||||
kms_env.wait_for_kms_ready().await.expect("KMS ready");
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
|
||||
|
||||
|
||||
@@ -19,7 +19,7 @@
|
||||
//! multipart upload behaviour.
|
||||
|
||||
use crate::common::{TEST_BUCKET, init_logging};
|
||||
use serial_test::serial;
|
||||
use tokio::time::{Duration, sleep};
|
||||
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?;
|
||||
|
||||
// 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?;
|
||||
// Allow Vault to finish initialising token auth and transit engine.
|
||||
sleep(Duration::from_secs(2)).await;
|
||||
|
||||
Ok(Self { env })
|
||||
}
|
||||
|
||||
@@ -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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).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?;
|
||||
kms_env.wait_for_kms_ready().await?;
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
|
||||
|
||||
let s3_client = kms_env.base_env.create_s3_client();
|
||||
kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
|
||||
|
||||
@@ -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, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
|
||||
ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
|
||||
ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
|
||||
RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
|
||||
DeleteVersionRequest, DeleteVersionsRequest, 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,28 +112,6 @@ 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());
|
||||
@@ -2428,6 +2406,8 @@ impl DiskAPI for RemoteDisk {
|
||||
return errors;
|
||||
}
|
||||
|
||||
// TODO(backlog): replace string errors with typed `StorageError` variants
|
||||
|
||||
let result = self
|
||||
.execute_with_timeout(
|
||||
|| async {
|
||||
@@ -2459,7 +2439,17 @@ impl DiskAPI for RemoteDisk {
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
decode_delete_versions_errors(response, versions.len())
|
||||
response
|
||||
.errors
|
||||
.iter()
|
||||
.map(|error| {
|
||||
if error.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(Error::other(error.to_string()))
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "trace", skip_all)]
|
||||
@@ -3770,63 +3760,6 @@ 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(());
|
||||
|
||||
@@ -784,6 +784,24 @@ 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>,
|
||||
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
|
||||
let writer = if is_inline_buffer {
|
||||
CustomWriter::new_inline_buffer()
|
||||
} else if let Some(disk) = disk {
|
||||
let length = if length > 0 {
|
||||
let length = length as usize;
|
||||
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
|
||||
} else {
|
||||
0
|
||||
};
|
||||
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
|
||||
|
||||
let file = disk.create_file("", volume, path, length).await?;
|
||||
#[cfg(feature = "hotpath")]
|
||||
@@ -820,6 +833,25 @@ 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>,
|
||||
}
|
||||
|
||||
@@ -722,10 +722,6 @@ 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 {
|
||||
|
||||
@@ -493,9 +493,6 @@ 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 {
|
||||
|
||||
@@ -146,29 +146,6 @@ 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
|
||||
@@ -575,7 +552,6 @@ impl NodeService {
|
||||
success: false,
|
||||
errors: Vec::new(),
|
||||
error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
|
||||
item_errors: Vec::new(),
|
||||
}));
|
||||
}
|
||||
};
|
||||
@@ -587,26 +563,30 @@ impl NodeService {
|
||||
success: false,
|
||||
errors: Vec::new(),
|
||||
error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
|
||||
item_errors: Vec::new(),
|
||||
}));
|
||||
}
|
||||
};
|
||||
|
||||
let (errors, item_errors) =
|
||||
encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
|
||||
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();
|
||||
|
||||
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(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
@@ -1632,8 +1612,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_delete_versions_errors, 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_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;
|
||||
@@ -1652,18 +1632,6 @@ 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() {
|
||||
|
||||
Reference in New Issue
Block a user