diff --git a/crates/config/src/server_config.rs b/crates/config/src/server_config.rs index c8d5bedb0..43f4ca341 100644 --- a/crates/config/src/server_config.rs +++ b/crates/config/src/server_config.rs @@ -105,6 +105,20 @@ impl KVS { self.insert(key, value); } } + + fn merged_with_defaults(&self, defaults: &KVS) -> KVS { + let mut merged = defaults.clone(); + + for kv in &self.0 { + if let Some(existing) = merged.0.iter_mut().find(|entry| entry.key == kv.key) { + *existing = kv.clone(); + } else { + merged.0.push(kv.clone()); + } + } + + merged + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -161,8 +175,23 @@ impl Config { } pub fn merge(&self) -> Config { - // TODO: merge default - self.clone() + if let Some(defaults) = DEFAULT_KVS.get() { + self.merge_with_defaults(defaults) + } else { + self.clone() + } + } + + fn merge_with_defaults(&self, defaults: &HashMap) -> Config { + let mut cfg = self.clone(); + + for (sub_sys, default_kvs) in defaults { + let targets = cfg.0.entry(sub_sys.clone()).or_default(); + let default_target = targets.entry(DEFAULT_DELIMITER.to_owned()).or_default(); + *default_target = default_target.merged_with_defaults(default_kvs); + } + + cfg } } @@ -238,6 +267,87 @@ mod tests { assert_eq!(loaded.merge(), loaded); } + #[test] + fn config_merge_fills_default_kvs_without_overwriting_user_values() { + let defaults = HashMap::from([ + ( + "merge_test_existing".to_string(), + KVS(vec![ + KV { + key: "keep".to_string(), + value: "default-keep".to_string(), + hidden_if_empty: false, + }, + KV { + key: "fill".to_string(), + value: "default-fill".to_string(), + hidden_if_empty: true, + }, + ]), + ), + ( + "merge_test_missing".to_string(), + KVS(vec![KV { + key: "enabled".to_string(), + value: "on".to_string(), + hidden_if_empty: false, + }]), + ), + ]); + let cfg = Config(HashMap::from([ + ( + "merge_test_existing".to_string(), + HashMap::from([( + DEFAULT_DELIMITER.to_string(), + KVS(vec![KV { + key: "keep".to_string(), + value: "user-keep".to_string(), + hidden_if_empty: false, + }]), + )]), + ), + ( + "unknown_subsystem".to_string(), + HashMap::from([( + DEFAULT_DELIMITER.to_string(), + KVS(vec![KV { + key: "custom".to_string(), + value: "value".to_string(), + hidden_if_empty: false, + }]), + )]), + ), + ])); + + let merged = cfg.merge_with_defaults(&defaults); + let existing = merged + .get_value("merge_test_existing", DEFAULT_DELIMITER) + .expect("existing subsystem should remain"); + assert_eq!(existing.lookup("keep"), Some("user-keep".to_string())); + assert_eq!(existing.lookup("fill"), Some("default-fill".to_string())); + assert!( + existing.0.iter().any(|kv| kv.key == "fill" && kv.hidden_if_empty), + "merged default entry should preserve default metadata" + ); + + let missing = merged + .get_value("merge_test_missing", DEFAULT_DELIMITER) + .expect("missing default subsystem should be added"); + assert_eq!(missing.lookup("enabled"), Some("on".to_string())); + + let unknown = merged + .get_value("unknown_subsystem", DEFAULT_DELIMITER) + .expect("unknown subsystem should be preserved"); + assert_eq!(unknown.lookup("custom"), Some("value".to_string())); + assert!( + cfg.get_value("merge_test_existing", DEFAULT_DELIMITER) + .expect("original existing subsystem should remain") + .lookup("fill") + .is_none(), + "merge should not mutate the source config" + ); + } + #[test] fn global_server_config_set_and_get_roundtrip() { let mut cfg = Config(HashMap::new()); diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index f780a141f..1161952c5 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -19,6 +19,7 @@ use super::replication_filemeta_boundary::{ ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, replication_statuses_map, version_purge_statuses_map, }; +use super::replication_lock_boundary::ReplicationLockTiming; use super::replication_logging::{EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION}; use super::replication_metadata_boundary::ReplicationMetadataStore; use super::replication_object_config::{ReplicationConfig, check_replicate_delete}; @@ -1028,10 +1029,40 @@ impl ReplicationPool { buckets: &[String], _cancellation_token: CancellationToken, ) -> Result<(), EcstoreError> { - // TODO: add leader_lock - // Make sure only one node running resync on the cluster - // Note: Leader lock implementation would be needed here - // let _lock_guard = global_leader_lock.get_lock().await?; + let load_resync_lock = match self + .storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), "replication/resync/load-resync.lock") + .await + { + Ok(lock) => lock, + Err(err) => { + warn!( + event = EVENT_REPLICATION_RESYNC_LOAD_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + error = ?err, + reason = "leader_lock_create_failed", + "Skipped replication resync metadata load" + ); + return Ok(()); + } + }; + let _load_resync_guard = match load_resync_lock + .get_write_lock(ReplicationLockTiming::acquire_timeout()) + .await + { + Ok(guard) => guard, + Err(_) => { + debug!( + event = EVENT_REPLICATION_RESYNC_LOAD_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + reason = "leader_lock_held_by_another_node", + "Another node is already loading replication resync metadata" + ); + return Ok(()); + } + }; let mut recovered_statuses = Vec::new(); let mut restart_opts = Vec::new(); @@ -1492,10 +1523,329 @@ async fn queue_replicate_deletes(batch: ReplicationHealResyncDeletes) -> Replica #[cfg(test)] mod tests { - use super::super::replication_resync_boundary::{decode_mrf_file, encode_mrf_file}; + use super::super::replication_resync_boundary::{decode_mrf_file, encode_mrf_file, encode_resync_file}; + use super::super::replication_storage_boundary::{ + DeletedObject, FileInfo, GetObjectReader, HTTPRangeSpec, ListOperations, ObjectIO, ObjectOperations, PutObjReader, + StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageNamespaceLocking, StorageObjectInfoOrErr, WalkOptions, + }; use super::*; + use std::fmt::{Debug, Formatter}; + use std::io::Cursor; + use std::sync::atomic::AtomicUsize; + use tokio::sync::Notify; use uuid::Uuid; + type TestListObjectsV2Info = StorageListObjectsV2Info; + type TestListObjectVersionsInfo = StorageListObjectVersionsInfo; + type TestObjectInfoOrErr = StorageObjectInfoOrErr; + + struct LoadResyncSharedState { + data: Vec, + lock_manager: Arc, + first_read_started: Notify, + read_count: AtomicUsize, + } + + struct LoadResyncNodeStore { + owner: String, + shared: Arc, + } + + impl LoadResyncNodeStore { + fn new(owner: &str, shared: Arc) -> Self { + Self { + owner: owner.to_string(), + shared, + } + } + } + + impl Debug for LoadResyncNodeStore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LoadResyncNodeStore").field("owner", &self.owner).finish() + } + } + + #[async_trait::async_trait] + impl ObjectIO for LoadResyncNodeStore { + type Error = EcstoreError; + type RangeSpec = HTTPRangeSpec; + type HeaderMap = http::HeaderMap; + type ObjectOptions = ObjectOptions; + type ObjectInfo = ObjectInfo; + type GetObjectReader = GetObjectReader; + type PutObjectReader = PutObjReader; + + async fn get_object_reader( + &self, + _bucket: &str, + object: &str, + _range: Option, + _h: Self::HeaderMap, + _opts: &Self::ObjectOptions, + ) -> Result { + if object != ReplicationMetadataStore::bucket_resync_file_path("load-resync-lock") { + return Err(EcstoreError::FileNotFound); + } + + let read_index = self.shared.read_count.fetch_add(1, Ordering::SeqCst); + if read_index == 0 { + self.shared.first_read_started.notify_waiters(); + tokio::time::sleep(Duration::from_millis(1_500)).await; + } + + let data = self.shared.data.clone(); + Ok(Self::GetObjectReader { + stream: Box::new(Cursor::new(data.clone())), + object_info: ObjectInfo { + size: data.len() as i64, + actual_size: data.len() as i64, + ..Default::default() + }, + buffered_body: None, + }) + } + + async fn put_object( + &self, + _bucket: &str, + _object: &str, + _data: &mut Self::PutObjectReader, + _opts: &Self::ObjectOptions, + ) -> Result { + Ok(ObjectInfo::default()) + } + } + + #[async_trait::async_trait] + impl ObjectOperations for LoadResyncNodeStore { + type Error = EcstoreError; + type ObjectInfo = ObjectInfo; + type ObjectOptions = ObjectOptions; + type FileInfo = FileInfo; + type ObjectToDelete = ObjectToDelete; + type DeletedObject = DeletedObject; + + async fn get_object_info( + &self, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn verify_object_integrity( + &self, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result<(), Self::Error> { + Err(EcstoreError::NotImplemented) + } + + async fn copy_object( + &self, + _src_bucket: &str, + _src_object: &str, + _dst_bucket: &str, + _dst_object: &str, + _src_info: &mut Self::ObjectInfo, + _src_opts: &Self::ObjectOptions, + _dst_opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn delete_object_version( + &self, + _bucket: &str, + _object: &str, + _fi: &Self::FileInfo, + _force_del_marker: bool, + ) -> Result<(), Self::Error> { + Err(EcstoreError::NotImplemented) + } + + async fn delete_object( + &self, + _bucket: &str, + _object: &str, + _opts: Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn delete_objects( + &self, + _bucket: &str, + _objects: Vec, + _opts: Self::ObjectOptions, + ) -> (Vec, Vec>) { + (Vec::new(), vec![Some(EcstoreError::NotImplemented)]) + } + + async fn put_object_metadata( + &self, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn get_object_tags( + &self, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn put_object_tags( + &self, + _bucket: &str, + _object: &str, + _tags: &str, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn delete_object_tags( + &self, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn add_partial(&self, _bucket: &str, _object: &str, _version_id: &str) -> Result<(), Self::Error> { + Err(EcstoreError::NotImplemented) + } + + async fn transition_object(&self, _bucket: &str, _object: &str, _opts: &Self::ObjectOptions) -> Result<(), Self::Error> { + Err(EcstoreError::NotImplemented) + } + + async fn restore_transitioned_object( + self: Arc, + _bucket: &str, + _object: &str, + _opts: &Self::ObjectOptions, + ) -> Result<(), Self::Error> { + Err(EcstoreError::NotImplemented) + } + } + + #[async_trait::async_trait] + impl ListOperations for LoadResyncNodeStore { + type Error = EcstoreError; + type ListObjectsV2Info = TestListObjectsV2Info; + type ListObjectVersionsInfo = TestListObjectVersionsInfo; + type ObjectInfoOrErr = TestObjectInfoOrErr; + type WalkOptions = WalkOptions; + type WalkCancellation = CancellationToken; + type WalkResultSender = Sender; + + async fn list_objects_v2( + self: Arc, + _bucket: &str, + _prefix: &str, + _continuation_token: Option, + _delimiter: Option, + _max_keys: i32, + _fetch_owner: bool, + _start_after: Option, + _incl_deleted: bool, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn list_object_versions( + self: Arc, + _bucket: &str, + _prefix: &str, + _marker: Option, + _version_marker: Option, + _delimiter: Option, + _max_keys: i32, + ) -> Result { + Err(EcstoreError::NotImplemented) + } + + async fn walk( + self: Arc, + _rx: Self::WalkCancellation, + _bucket: &str, + _prefix: &str, + _result: Self::WalkResultSender, + _opts: Self::WalkOptions, + ) -> Result<(), Self::Error> { + Ok(()) + } + } + + #[async_trait::async_trait] + impl StorageNamespaceLocking for LoadResyncNodeStore { + type Error = EcstoreError; + type NamespaceLock = rustfs_lock::NamespaceLockWrapper; + + async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { + let lock = + rustfs_lock::NamespaceLock::with_local_manager("load-resync-test".to_string(), self.shared.lock_manager.clone()); + Ok(rustfs_lock::NamespaceLockWrapper::new( + lock, + rustfs_lock::ObjectKey::new(bucket.to_string(), object.to_string()), + self.owner.clone(), + )) + } + } + + async fn new_test_replication_pool(storage: Arc) -> Arc> { + let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(1); + let (mrf_save_tx, mrf_save_rx) = mpsc::channel(1); + let (mrf_worker_kill_tx, _) = mpsc::channel(1); + let (mrf_stop_tx, _) = mpsc::channel(1); + + Arc::new(ReplicationPool { + active_workers: Arc::new(AtomicI32::new(0)), + active_lrg_workers: Arc::new(AtomicI32::new(0)), + active_mrf_workers: Arc::new(AtomicI32::new(0)), + storage, + priority: RwLock::new(ReplicationPoolOpts::default().priority), + max_workers: RwLock::new(WORKER_MAX_LIMIT), + max_l_workers: RwLock::new(LARGE_WORKER_COUNT), + stats: Arc::new(ReplicationStats::new()), + workers: RwLock::new(Vec::new()), + lrg_workers: RwLock::new(Vec::new()), + mrf_replica_tx, + mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), + mrf_save_tx, + mrf_save_rx: Mutex::new(Some(mrf_save_rx)), + mrf_worker_kill_tx, + mrf_stop_tx, + mrf_worker_size: AtomicI32::new(0), + task_handles: Mutex::new(Vec::new()), + resyncer: Arc::new(ReplicationResyncer::new().await), + }) + } + + fn load_resync_test_metadata() -> Vec { + let mut status = BucketReplicationResyncStatus::new(); + status.targets_map.insert( + "arn:test".to_string(), + TargetReplicationResyncStatus { + bucket: "load-resync-lock".to_string(), + resync_status: ResyncStatusType::ResyncCompleted, + ..Default::default() + }, + ); + encode_resync_file(&status).expect("test resync metadata should encode") + } + #[test] fn replication_queue_admission_combines_target_results() { let mut admission = ReplicationQueueAdmission::Skipped; @@ -1561,6 +1911,57 @@ mod tests { assert!(!should_auto_resume_resync(ResyncStatusType::ResyncFailed)); } + #[tokio::test] + async fn load_resync_leader_lock_allows_only_one_startup_recovery() { + temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async { + let shared = Arc::new(LoadResyncSharedState { + data: load_resync_test_metadata(), + lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), + first_read_started: Notify::new(), + read_count: AtomicUsize::new(0), + }); + let leader_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let skipped_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await; + + let leader = leader_pool.clone(); + let leader_task = tokio::spawn(async move { + let buckets = vec!["load-resync-lock".to_string()]; + leader.load_resync(&buckets, CancellationToken::new()).await + }); + + tokio::time::timeout(Duration::from_secs(1), shared.first_read_started.notified()) + .await + .expect("leader should start reading persisted resync metadata"); + + let buckets = vec!["load-resync-lock".to_string()]; + skipped_pool + .clone() + .load_resync(&buckets, CancellationToken::new()) + .await + .expect("contended load_resync should skip without failing startup"); + + leader_task + .await + .expect("leader load_resync task should not panic") + .expect("leader load_resync should succeed"); + + assert_eq!( + shared.read_count.load(Ordering::SeqCst), + 1, + "only the leader node should read persisted resync metadata" + ); + assert!( + leader_pool.resyncer.status_map.read().await.contains_key("load-resync-lock"), + "leader node should recover persisted resync status" + ); + assert!( + skipped_pool.resyncer.status_map.read().await.is_empty(), + "node that does not hold the leader lock must not populate status_map" + ); + }) + .await; + } + // ── MrfReplicateEntry encode/decode roundtrips ──────────────────────────── #[test] diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs index e06ab91fc..0b729d9ba 100644 --- a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_filemeta::FileInfo; +pub(crate) use rustfs_filemeta::FileInfo; use tokio_util::sync::CancellationToken; use super::replication_error_boundary::Error; diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index ca6744c86..8414a3dad 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -294,9 +294,43 @@ impl S3PeerSys { } } - if let Some(err) = reduce_write_quorum_errs(&errors, BUCKET_OP_IGNORED_ERRS, (errors.len() / 2) + 1) { + for i in 0..self.pools_count { + let per_pool_errs = pool_participant_errors(&self.clients, &errors, i); + if let Some(err) = reduce_pool_write_quorum_errs(&per_pool_errs) { + if !Error::is_err_object_not_found(&err) && !opts.no_recreate { + let make_bucket_opts = MakeBucketOptions::default(); + let mut rollback_futures = Vec::new(); + for (client, delete_err) in self.clients.iter().zip(errors.iter()) { + if delete_err.is_none() { + rollback_futures.push(client.make_bucket(bucket, &make_bucket_opts)); + } + } + for rollback_result in join_all(rollback_futures).await { + if let Err(rollback_err) = rollback_result { + warn!("delete_bucket rollback make_bucket failed: {rollback_err}"); + } + } + } + return Err(err); + } + } + + if self.pools_count == 0 + && let Some(err) = reduce_write_quorum_errs(&errors, BUCKET_OP_IGNORED_ERRS, (errors.len() / 2) + 1) + { if !Error::is_err_object_not_found(&err) && !opts.no_recreate { - let _ = self.make_bucket(bucket, &MakeBucketOptions::default()).await; + let make_bucket_opts = MakeBucketOptions::default(); + let mut rollback_futures = Vec::new(); + for (client, delete_err) in self.clients.iter().zip(errors.iter()) { + if delete_err.is_none() { + rollback_futures.push(client.make_bucket(bucket, &make_bucket_opts)); + } + } + for rollback_result in join_all(rollback_futures).await { + if let Err(rollback_err) = rollback_result { + warn!("delete_bucket rollback make_bucket failed: {rollback_err}"); + } + } } return Err(err); } @@ -586,11 +620,12 @@ impl PeerS3Client for LocalPeerS3Client { } } - // For errVolumeNotEmpty, do not delete; recreate only the entries already removed - - for (idx, err) in errs.into_iter().enumerate() { - if err.is_none() && recreate { - let _ = local_disks[idx].make_volume(bucket).await; + for (idx, err) in errs.iter().enumerate() { + if err.is_none() + && recreate + && let Err(rollback_err) = local_disks[idx].make_volume(bucket).await + { + warn!("local delete_bucket rollback make_volume failed: {rollback_err}"); } } @@ -598,7 +633,18 @@ impl PeerS3Client for LocalPeerS3Client { return Err(Error::VolumeNotEmpty); } - // TODO: reduceWriteQuorumErrs + if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, (local_disks.len() / 2) + 1) { + if !Error::is_err_object_not_found(&err) && !opts.no_recreate { + for (idx, delete_err) in errs.iter().enumerate() { + if delete_err.is_none() + && let Err(rollback_err) = local_disks[idx].make_volume(bucket).await + { + warn!("local delete_bucket rollback make_volume failed: {rollback_err}"); + } + } + } + return Err(err); + } Ok(()) } @@ -1117,6 +1163,7 @@ mod tests { use std::{ io, pin::Pin, + sync::atomic::{AtomicUsize, Ordering}, task::{Context, Poll}, }; use tempfile::TempDir; @@ -1143,6 +1190,8 @@ mod tests { pools: Option>, make_bucket_result: Result<()>, list_bucket_result: Result>, + delete_bucket_result: Result<()>, + make_bucket_calls: Arc, } #[async_trait] @@ -1152,6 +1201,7 @@ mod tests { } async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { + self.make_bucket_calls.fetch_add(1, Ordering::SeqCst); self.make_bucket_result.clone() } @@ -1160,7 +1210,7 @@ mod tests { } async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { - unreachable!("not used by quorum tests") + self.delete_bucket_result.clone() } async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { @@ -1184,15 +1234,39 @@ mod tests { test_peer_with_results(pools, Ok(()), list_bucket_result) } + fn test_peer_with_delete_bucket(pools: &[usize], delete_bucket_result: Result<()>) -> Client { + test_peer_with_delete_bucket_and_make_counter(pools, delete_bucket_result, Arc::new(AtomicUsize::new(0))) + } + + fn test_peer_with_delete_bucket_and_make_counter( + pools: &[usize], + delete_bucket_result: Result<()>, + make_bucket_calls: Arc, + ) -> Client { + test_peer_with_all_results(pools, Ok(()), Ok(Vec::new()), delete_bucket_result, make_bucket_calls) + } + fn test_peer_with_results( pools: &[usize], make_bucket_result: Result<()>, list_bucket_result: Result>, + ) -> Client { + test_peer_with_all_results(pools, make_bucket_result, list_bucket_result, Ok(()), Arc::new(AtomicUsize::new(0))) + } + + fn test_peer_with_all_results( + pools: &[usize], + make_bucket_result: Result<()>, + list_bucket_result: Result>, + delete_bucket_result: Result<()>, + make_bucket_calls: Arc, ) -> Client { Arc::new(Box::new(TestPeerS3Client { pools: Some(pools.to_vec()), make_bucket_result, list_bucket_result, + delete_bucket_result, + make_bucket_calls, })) } @@ -1536,4 +1610,80 @@ mod tests { assert_eq!(buckets.len(), 1); assert_eq!(buckets[0].name, bucket.name); } + + #[tokio::test] + async fn test_delete_bucket_fails_when_any_pool_misses_write_quorum() { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_delete_bucket(&[0], Ok(())), + test_peer_with_delete_bucket(&[0], Ok(())), + test_peer_with_delete_bucket(&[0], Err(Error::VolumeNotEmpty)), + test_peer_with_delete_bucket(&[0], Err(Error::VolumeNotEmpty)), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Ok(())), + ], + pools_count: 2, + }; + + let err = peer_sys + .delete_bucket("partially-deleted-bucket", &DeleteBucketOptions::default()) + .await + .expect_err("pool 0 should fail because it did not reach write quorum"); + + assert_eq!(err, Error::ErasureWriteQuorum); + } + + #[tokio::test] + async fn test_delete_bucket_succeeds_when_every_pool_reaches_write_quorum() { + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_delete_bucket(&[0], Ok(())), + test_peer_with_delete_bucket(&[0], Ok(())), + test_peer_with_delete_bucket(&[0], Ok(())), + test_peer_with_delete_bucket(&[0], Err(Error::DiskNotFound)), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Ok(())), + test_peer_with_delete_bucket(&[1], Err(Error::DiskNotFound)), + ], + pools_count: 2, + }; + + peer_sys + .delete_bucket("deleted-bucket", &DeleteBucketOptions::default()) + .await + .expect("each pool reached write quorum"); + } + + #[tokio::test] + async fn test_delete_bucket_rolls_back_only_successful_deletes_on_failure() { + let make_bucket_calls = (0..8).map(|_| Arc::new(AtomicUsize::new(0))).collect::>(); + let peer_sys = S3PeerSys { + clients: vec![ + test_peer_with_delete_bucket_and_make_counter(&[0], Ok(()), make_bucket_calls[0].clone()), + test_peer_with_delete_bucket_and_make_counter(&[0], Ok(()), make_bucket_calls[1].clone()), + test_peer_with_delete_bucket_and_make_counter(&[0], Err(Error::DiskAccessDenied), make_bucket_calls[2].clone()), + test_peer_with_delete_bucket_and_make_counter(&[0], Err(Error::DiskAccessDenied), make_bucket_calls[3].clone()), + test_peer_with_delete_bucket_and_make_counter(&[1], Err(Error::DiskAccessDenied), make_bucket_calls[4].clone()), + test_peer_with_delete_bucket_and_make_counter(&[1], Err(Error::DiskAccessDenied), make_bucket_calls[5].clone()), + test_peer_with_delete_bucket_and_make_counter(&[1], Err(Error::DiskAccessDenied), make_bucket_calls[6].clone()), + test_peer_with_delete_bucket_and_make_counter(&[1], Err(Error::DiskAccessDenied), make_bucket_calls[7].clone()), + ], + pools_count: 2, + }; + + let err = peer_sys + .delete_bucket("rolled-back-bucket", &DeleteBucketOptions::default()) + .await + .expect_err("delete failure should return the quorum error"); + + assert_eq!(err, Error::ErasureWriteQuorum); + let calls = make_bucket_calls + .iter() + .map(|call_count| call_count.load(Ordering::SeqCst)) + .collect::>(); + assert_eq!(calls, vec![1, 1, 0, 0, 0, 0, 0, 0]); + } } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 78b5671a1..6527b4cbb 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -43,8 +43,8 @@ use rustfs_filemeta::FileInfo; use rustfs_utils::path::SLASH_SEPARATOR; use serde_json::{Map, Value}; use std::collections::{HashMap, HashSet}; -use std::sync::Arc; use std::sync::LazyLock; +use std::sync::{Arc, RwLock}; use tracing::{debug, error, info, instrument, warn}; pub const CONFIG_PREFIX: &str = "config"; @@ -65,6 +65,7 @@ const DEFAULT_CONFIG_RECOVER_ON_CORRUPTION: bool = true; const LOG_COMPONENT_CONFIG: &str = "ecstore"; const LOG_SUBSYSTEM_CONFIG: &str = "config"; const EVENT_SERVER_CONFIG_DECODE_FAILED: &str = "server_config_decode_failed"; +const EVENT_SERVER_CONFIG_DECRYPT_FAILED: &str = "server_config_decrypt_failed"; const EVENT_SERVER_CONFIG_READ_FAILED: &str = "server_config_read_failed"; const EVENT_SERVER_CONFIG_HEAL_RESULT: &str = "server_config_heal_result"; const EVENT_SERVER_CONFIG_RECOVERED: &str = "server_config_recovered_after_heal"; @@ -81,6 +82,10 @@ fn config_corruption_recovery_enabled() -> bool { #[error("server config corrupt: {0}")] pub struct ServerConfigCorruptError(pub String); +#[derive(Debug, thiserror::Error)] +#[error("server config decrypt failed: {0}")] +struct ServerConfigDecryptError(pub String); + /// Returns true when `err` is a [`ServerConfigCorruptError`] produced by the /// server config decode path. Such failures are deterministic: the persisted /// blob itself is damaged and re-reading it cannot succeed. @@ -88,12 +93,43 @@ pub fn is_server_config_corrupt_error(err: &Error) -> bool { matches!(err, Error::Io(io_err) if io_err.get_ref().is_some_and(|inner| inner.is::())) } +fn is_server_config_decrypt_error(err: &Error) -> bool { + matches!(err, Error::Io(io_err) if io_err.get_ref().is_some_and(|inner| inner.is::())) +} + pub const STORAGE_CLASS_SUB_SYS: &str = "storage_class"; pub const COMMA_SEPARATED_LISTS: &[&str] = &[rustfs_config::oidc::OIDC_SCOPES, rustfs_config::oidc::OIDC_OTHER_AUDIENCES]; static CONFIG_BUCKET: LazyLock = LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{CONFIG_PREFIX}")); +type ServerConfigDecryptFn = crate::bucket::migration::LegacyBlobDecryptFn; + +static SERVER_CONFIG_DECRYPT_FN: LazyLock>> = LazyLock::new(|| RwLock::new(None)); + +pub fn register_server_config_decrypt_fn(decrypt_fn: ServerConfigDecryptFn) { + match SERVER_CONFIG_DECRYPT_FN.write() { + Ok(mut guard) => { + *guard = Some(decrypt_fn); + } + Err(err) => { + warn!("register server config decrypt function failed: {err}"); + } + } +} + +fn server_config_decrypt_fn() -> Option { + SERVER_CONFIG_DECRYPT_FN.read().ok().and_then(|guard| guard.clone()) +} + +#[cfg(test)] +fn replace_server_config_decrypt_fn_for_test(decrypt_fn: Option) -> Option { + SERVER_CONFIG_DECRYPT_FN + .write() + .ok() + .and_then(|mut guard| std::mem::replace(&mut *guard, decrypt_fn)) +} + static SUB_SYSTEMS_DYNAMIC: LazyLock> = LazyLock::new(|| { let mut h = HashSet::new(); h.insert(STORAGE_CLASS_SUB_SYS.to_owned()); @@ -1108,6 +1144,10 @@ where DeletedObject = DeletedObject, >, { + if let Some(decrypt) = &decrypt_fn { + register_server_config_decrypt_fn(decrypt.clone()); + } + let config_file = get_config_file(); match api .get_object_info( @@ -1253,7 +1293,6 @@ where // Try to read the configuration again match read_config_no_lock(api.clone(), &config_file).await { Ok(cfg_data) => { - // TODO: decrypt let cfg = decode_persisted_server_config(&cfg_data)?; return Ok(cfg.merge()); } @@ -1270,17 +1309,47 @@ where /// Decode the persisted server config blob, marking decode failures as /// deterministic corruption (see [`ServerConfigCorruptError`]). fn decode_persisted_server_config(data: &[u8]) -> Result { - decode_server_config_blob(data).map_err(|err| { - error!( - event = EVENT_SERVER_CONFIG_DECODE_FAILED, - component = LOG_COMPONENT_CONFIG, - subsystem = LOG_SUBSYSTEM_CONFIG, - size = data.len(), - error = %err, - "persisted server config cannot be decoded, object is corrupt" - ); - Error::other(ServerConfigCorruptError(err.to_string())) - }) + match decode_server_config_blob(data) { + Ok(cfg) => Ok(cfg), + Err(raw_decode_err) => { + let Some(decrypt) = server_config_decrypt_fn() else { + error!( + event = EVENT_SERVER_CONFIG_DECODE_FAILED, + component = LOG_COMPONENT_CONFIG, + subsystem = LOG_SUBSYSTEM_CONFIG, + size = data.len(), + error = %raw_decode_err, + "persisted server config cannot be decoded, object is corrupt" + ); + return Err(Error::other(ServerConfigCorruptError(raw_decode_err.to_string()))); + }; + + let Some(decrypted) = decrypt(data) else { + error!( + event = EVENT_SERVER_CONFIG_DECRYPT_FAILED, + component = LOG_COMPONENT_CONFIG, + subsystem = LOG_SUBSYSTEM_CONFIG, + size = data.len(), + error = %raw_decode_err, + "persisted server config cannot be decoded or decrypted" + ); + return Err(Error::other(ServerConfigDecryptError(raw_decode_err.to_string()))); + }; + + decode_server_config_blob(&decrypted).map_err(|err| { + error!( + event = EVENT_SERVER_CONFIG_DECODE_FAILED, + component = LOG_COMPONENT_CONFIG, + subsystem = LOG_SUBSYSTEM_CONFIG, + size = decrypted.len(), + encrypted_size = data.len(), + error = %err, + "decrypted persisted server config cannot be decoded, object is corrupt" + ); + Error::other(ServerConfigCorruptError(err.to_string())) + }) + } + } } /// Startup-only read of the server config with layered recovery: @@ -1387,10 +1456,9 @@ where } } -/// Availability failures that a startup retry loop can reasonably wait out: -/// falling back to a default config would mask them, so they are propagated. fn config_read_failure_is_retryable(err: &Error) -> bool { - err.is_quorum_error() + is_server_config_decrypt_error(err) + || err.is_quorum_error() || matches!( err, Error::DiskNotFound | Error::FaultyDisk | Error::FaultyRemoteDisk | Error::TooManyOpenFiles | Error::SlowDown @@ -2835,7 +2903,7 @@ mod tests { use super::{ ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS, ServerConfigCorruptError, config_read_failure_is_retryable, decode_persisted_server_config, fallback_server_config_after_corruption, is_server_config_corrupt_error, - read_config_without_migrate_with_recovery, + read_config_without_migrate_with_recovery, replace_server_config_decrypt_fn_for_test, }; use rustfs_common::heal_channel::HealOpts; use std::sync::Mutex; @@ -2919,6 +2987,24 @@ mod tests { } } + struct ServerConfigDecryptHookGuard { + previous: Option, + } + + impl ServerConfigDecryptHookGuard { + fn replace(decrypt_fn: crate::bucket::migration::LegacyBlobDecryptFn) -> Self { + Self { + previous: replace_server_config_decrypt_fn_for_test(Some(decrypt_fn)), + } + } + } + + impl Drop for ServerConfigDecryptHookGuard { + fn drop(&mut self) { + replace_server_config_decrypt_fn_for_test(self.previous.take()); + } + } + impl Debug for RecoveryMockStore { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.debug_struct("RecoveryMockStore").finish() @@ -3038,6 +3124,60 @@ mod tests { } } + fn encrypted_current_server_config_blob() -> Vec { + let mut cfg = Config::new(); + let kvs = storage_class_kvs_mut(&mut cfg); + kvs.insert("standard".to_string(), "EC:4".to_string()); + kvs.insert("rrs".to_string(), "EC:2".to_string()); + + let plain = encode_server_config_blob(&cfg, None).expect("encode current server config"); + rustfs_crypto::encrypt_data(b"root-secret-key", &plain).expect("encrypt current server config") + } + + #[tokio::test] + #[serial] + async fn test_read_config_decrypts_current_server_config_blob() { + let store = Arc::new(RecoveryMockStore::new( + RecoveryReadState::Blob(encrypted_current_server_config_blob()), + None, + )); + let decrypt_fn: crate::bucket::migration::LegacyBlobDecryptFn = + Arc::new(|data: &[u8]| rustfs_crypto::decrypt_data(b"root-secret-key", data).ok()); + let _decrypt_hook = ServerConfigDecryptHookGuard::replace(decrypt_fn); + + let cfg = read_config_without_migrate_with_recovery(store.clone()) + .await + .expect("encrypted current config should decrypt"); + + assert_eq!(store.heal_calls.load(Ordering::SeqCst), 0, "decrypt should avoid corruption recovery"); + let kvs = cfg + .get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER) + .expect("decrypted config should preserve storage_class"); + assert_eq!(kvs.get("standard"), "EC:4"); + assert_eq!(kvs.get("rrs"), "EC:2"); + } + + #[tokio::test] + #[serial] + async fn test_read_config_decrypt_failure_does_not_fallback_to_default() { + let store = Arc::new(RecoveryMockStore::new( + RecoveryReadState::Blob(encrypted_current_server_config_blob()), + None, + )); + let decrypt_fn: crate::bucket::migration::LegacyBlobDecryptFn = Arc::new(|_data: &[u8]| None); + let _decrypt_hook = ServerConfigDecryptHookGuard::replace(decrypt_fn); + + let err = read_config_without_migrate_with_recovery(store.clone()) + .await + .expect_err("decrypt failure must not fall back to the default config"); + + assert_eq!(store.heal_calls.load(Ordering::SeqCst), 1, "heal is still attempted before failing"); + assert!( + !is_server_config_corrupt_error(&err), + "decrypt failure must not be treated as defaultable corruption" + ); + } + #[tokio::test] async fn test_recovery_uses_healed_config_object() { // Heal reconstructs a valid config from parity: no fallback needed. diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index 92346aa0a..c29adecea 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -58,6 +58,38 @@ pub enum StorageError { CorruptedBackend, #[error("Too many open files")] TooManyOpenFiles, + #[error("maximum versions exceeded, please delete few versions to proceed")] + MaxVersionsExceeded, + #[error("inconsistent drive found")] + InconsistentDisk, + #[error("drive does not support O_DIRECT")] + UnsupportedDisk, + #[error("disk not a dir")] + DiskNotDir, + #[error("drive still did not complete the request")] + DiskOngoingReq, + #[error("path not found")] + PathNotFound, + #[error("bit-rot hash algorithm is invalid")] + BitrotHashAlgoInvalid, + #[error("Rename across devices not allowed, please fix your backend configuration")] + CrossDeviceLink, + #[error("less data available than what was requested")] + LessData, + #[error("more data was sent than what was advertised")] + MoreData, + #[error("outdated XL meta")] + OutdatedXLMeta, + #[error("part missing or corrupt")] + PartMissingOrCorrupt, + #[error("short write")] + ShortWrite, + #[error("source stalled")] + SourceStalled, + #[error("timeout")] + Timeout, + #[error("invalid path")] + InvalidPath, #[error("Volume not found")] VolumeNotFound, @@ -233,17 +265,17 @@ impl From for StorageError { fn from(e: DiskError) -> Self { match e { DiskError::Io(io_error) => StorageError::Io(io_error), - // DiskError::MaxVersionsExceeded => todo!(), + DiskError::MaxVersionsExceeded => StorageError::MaxVersionsExceeded, DiskError::Unexpected => StorageError::Unexpected, DiskError::CorruptedFormat => StorageError::CorruptedFormat, DiskError::CorruptedBackend => StorageError::CorruptedBackend, DiskError::UnformattedDisk => StorageError::UnformattedDisk, - // DiskError::InconsistentDisk => StorageError::InconsistentDisk, - // DiskError::UnsupportedDisk => StorageError::UnsupportedDisk, + DiskError::InconsistentDisk => StorageError::InconsistentDisk, + DiskError::UnsupportedDisk => StorageError::UnsupportedDisk, DiskError::DiskFull => StorageError::DiskFull, - // DiskError::DiskNotDir => StorageError::DiskNotDir, + DiskError::DiskNotDir => StorageError::DiskNotDir, DiskError::DiskNotFound => StorageError::DiskNotFound, - // DiskError::DiskOngoingReq => StorageError::DiskOngoingReq, + DiskError::DiskOngoingReq => StorageError::DiskOngoingReq, DiskError::DriveIsRoot => StorageError::DriveIsRoot, DiskError::FaultyRemoteDisk => StorageError::FaultyRemoteDisk, DiskError::FaultyDisk => StorageError::FaultyDisk, @@ -254,23 +286,26 @@ impl From for StorageError { DiskError::FileNameTooLong => StorageError::FileNameTooLong, DiskError::VolumeExists => StorageError::VolumeExists, DiskError::IsNotRegular => StorageError::IsNotRegular, - // DiskError::PathNotFound => StorageError::PathNotFound, + DiskError::PathNotFound => StorageError::PathNotFound, DiskError::VolumeNotFound => StorageError::VolumeNotFound, DiskError::VolumeNotEmpty => StorageError::VolumeNotEmpty, DiskError::VolumeAccessDenied => StorageError::VolumeAccessDenied, DiskError::FileAccessDenied => StorageError::FileAccessDenied, DiskError::FileCorrupt => StorageError::FileCorrupt, - // DiskError::BitrotHashAlgoInvalid => StorageError::BitrotHashAlgoInvalid, - // DiskError::CrossDeviceLink => StorageError::CrossDeviceLink, - // DiskError::LessData => StorageError::LessData, - // DiskError::MoreData => StorageError::MoreData, - // DiskError::OutdatedXLMeta => StorageError::OutdatedXLMeta, - // DiskError::PartMissingOrCorrupt => StorageError::PartMissingOrCorrupt, + DiskError::BitrotHashAlgoInvalid => StorageError::BitrotHashAlgoInvalid, + DiskError::CrossDeviceLink => StorageError::CrossDeviceLink, + DiskError::LessData => StorageError::LessData, + DiskError::MoreData => StorageError::MoreData, + DiskError::OutdatedXLMeta => StorageError::OutdatedXLMeta, + DiskError::PartMissingOrCorrupt => StorageError::PartMissingOrCorrupt, DiskError::NoHealRequired => StorageError::NoHealRequired, DiskError::MethodNotAllowed => StorageError::MethodNotAllowed, DiskError::ErasureReadQuorum => StorageError::ErasureReadQuorum, DiskError::ErasureWriteQuorum => StorageError::ErasureWriteQuorum, - _ => StorageError::Io(std::io::Error::other(e)), + DiskError::ShortWrite => StorageError::ShortWrite, + DiskError::SourceStalled => StorageError::SourceStalled, + DiskError::Timeout => StorageError::Timeout, + DiskError::InvalidPath => StorageError::InvalidPath, } } } @@ -286,6 +321,22 @@ impl From for DiskError { StorageError::MethodNotAllowed => DiskError::MethodNotAllowed, StorageError::StorageFull => DiskError::DiskFull, StorageError::SlowDown => DiskError::TooManyOpenFiles, + StorageError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded, + StorageError::InconsistentDisk => DiskError::InconsistentDisk, + StorageError::UnsupportedDisk => DiskError::UnsupportedDisk, + StorageError::DiskNotDir => DiskError::DiskNotDir, + StorageError::DiskOngoingReq => DiskError::DiskOngoingReq, + StorageError::PathNotFound => DiskError::PathNotFound, + StorageError::BitrotHashAlgoInvalid => DiskError::BitrotHashAlgoInvalid, + StorageError::CrossDeviceLink => DiskError::CrossDeviceLink, + StorageError::LessData => DiskError::LessData, + StorageError::MoreData => DiskError::MoreData, + StorageError::OutdatedXLMeta => DiskError::OutdatedXLMeta, + StorageError::PartMissingOrCorrupt => DiskError::PartMissingOrCorrupt, + StorageError::ShortWrite => DiskError::ShortWrite, + StorageError::SourceStalled => DiskError::SourceStalled, + StorageError::Timeout => DiskError::Timeout, + StorageError::InvalidPath => DiskError::InvalidPath, StorageError::ErasureReadQuorum => DiskError::ErasureReadQuorum, StorageError::ErasureWriteQuorum => DiskError::ErasureWriteQuorum, StorageError::TooManyOpenFiles => DiskError::TooManyOpenFiles, @@ -402,6 +453,22 @@ impl Clone for StorageError { StorageError::CorruptedBackend => StorageError::CorruptedBackend, StorageError::UnformattedDisk => StorageError::UnformattedDisk, StorageError::DiskNotFound => StorageError::DiskNotFound, + StorageError::MaxVersionsExceeded => StorageError::MaxVersionsExceeded, + StorageError::InconsistentDisk => StorageError::InconsistentDisk, + StorageError::UnsupportedDisk => StorageError::UnsupportedDisk, + StorageError::DiskNotDir => StorageError::DiskNotDir, + StorageError::DiskOngoingReq => StorageError::DiskOngoingReq, + StorageError::PathNotFound => StorageError::PathNotFound, + StorageError::BitrotHashAlgoInvalid => StorageError::BitrotHashAlgoInvalid, + StorageError::CrossDeviceLink => StorageError::CrossDeviceLink, + StorageError::LessData => StorageError::LessData, + StorageError::MoreData => StorageError::MoreData, + StorageError::OutdatedXLMeta => StorageError::OutdatedXLMeta, + StorageError::PartMissingOrCorrupt => StorageError::PartMissingOrCorrupt, + StorageError::ShortWrite => StorageError::ShortWrite, + StorageError::SourceStalled => StorageError::SourceStalled, + StorageError::Timeout => StorageError::Timeout, + StorageError::InvalidPath => StorageError::InvalidPath, StorageError::DriveIsRoot => StorageError::DriveIsRoot, StorageError::FaultyRemoteDisk => StorageError::FaultyRemoteDisk, StorageError::DiskAccessDenied => StorageError::DiskAccessDenied, @@ -491,6 +558,22 @@ impl StorageError { StorageError::CorruptedBackend => StorageErrorCode::CorruptedBackend, StorageError::UnformattedDisk => StorageErrorCode::UnformattedDisk, StorageError::DiskNotFound => StorageErrorCode::DiskNotFound, + StorageError::MaxVersionsExceeded => StorageErrorCode::MaxVersionsExceeded, + StorageError::InconsistentDisk => StorageErrorCode::InconsistentDisk, + StorageError::UnsupportedDisk => StorageErrorCode::UnsupportedDisk, + StorageError::DiskNotDir => StorageErrorCode::DiskNotDir, + StorageError::DiskOngoingReq => StorageErrorCode::DiskOngoingReq, + StorageError::PathNotFound => StorageErrorCode::PathNotFound, + StorageError::BitrotHashAlgoInvalid => StorageErrorCode::BitrotHashAlgoInvalid, + StorageError::CrossDeviceLink => StorageErrorCode::CrossDeviceLink, + StorageError::LessData => StorageErrorCode::LessData, + StorageError::MoreData => StorageErrorCode::MoreData, + StorageError::OutdatedXLMeta => StorageErrorCode::OutdatedXLMeta, + StorageError::PartMissingOrCorrupt => StorageErrorCode::PartMissingOrCorrupt, + StorageError::ShortWrite => StorageErrorCode::ShortWrite, + StorageError::SourceStalled => StorageErrorCode::SourceStalled, + StorageError::Timeout => StorageErrorCode::Timeout, + StorageError::InvalidPath => StorageErrorCode::InvalidPath, StorageError::DriveIsRoot => StorageErrorCode::DriveIsRoot, StorageError::FaultyRemoteDisk => StorageErrorCode::FaultyRemoteDisk, StorageError::DiskAccessDenied => StorageErrorCode::DiskAccessDenied, @@ -564,6 +647,22 @@ impl StorageError { StorageErrorCode::CorruptedBackend => Some(StorageError::CorruptedBackend), StorageErrorCode::UnformattedDisk => Some(StorageError::UnformattedDisk), StorageErrorCode::DiskNotFound => Some(StorageError::DiskNotFound), + StorageErrorCode::MaxVersionsExceeded => Some(StorageError::MaxVersionsExceeded), + StorageErrorCode::InconsistentDisk => Some(StorageError::InconsistentDisk), + StorageErrorCode::UnsupportedDisk => Some(StorageError::UnsupportedDisk), + StorageErrorCode::DiskNotDir => Some(StorageError::DiskNotDir), + StorageErrorCode::DiskOngoingReq => Some(StorageError::DiskOngoingReq), + StorageErrorCode::PathNotFound => Some(StorageError::PathNotFound), + StorageErrorCode::BitrotHashAlgoInvalid => Some(StorageError::BitrotHashAlgoInvalid), + StorageErrorCode::CrossDeviceLink => Some(StorageError::CrossDeviceLink), + StorageErrorCode::LessData => Some(StorageError::LessData), + StorageErrorCode::MoreData => Some(StorageError::MoreData), + StorageErrorCode::OutdatedXLMeta => Some(StorageError::OutdatedXLMeta), + StorageErrorCode::PartMissingOrCorrupt => Some(StorageError::PartMissingOrCorrupt), + StorageErrorCode::ShortWrite => Some(StorageError::ShortWrite), + StorageErrorCode::SourceStalled => Some(StorageError::SourceStalled), + StorageErrorCode::Timeout => Some(StorageError::Timeout), + StorageErrorCode::InvalidPath => Some(StorageError::InvalidPath), StorageErrorCode::DriveIsRoot => Some(StorageError::DriveIsRoot), StorageErrorCode::FaultyRemoteDisk => Some(StorageError::FaultyRemoteDisk), StorageErrorCode::DiskAccessDenied => Some(StorageError::DiskAccessDenied), @@ -1175,6 +1274,68 @@ mod tests { assert_eq!(storage_error, StorageError::FileNotFound); } + #[test] + fn test_disk_error_specific_variants_do_not_degrade_to_io() { + let cases = vec![ + (DiskError::MaxVersionsExceeded, StorageError::MaxVersionsExceeded), + (DiskError::InconsistentDisk, StorageError::InconsistentDisk), + (DiskError::UnsupportedDisk, StorageError::UnsupportedDisk), + (DiskError::DiskNotDir, StorageError::DiskNotDir), + (DiskError::DiskOngoingReq, StorageError::DiskOngoingReq), + (DiskError::PathNotFound, StorageError::PathNotFound), + (DiskError::BitrotHashAlgoInvalid, StorageError::BitrotHashAlgoInvalid), + (DiskError::CrossDeviceLink, StorageError::CrossDeviceLink), + (DiskError::LessData, StorageError::LessData), + (DiskError::MoreData, StorageError::MoreData), + (DiskError::OutdatedXLMeta, StorageError::OutdatedXLMeta), + (DiskError::PartMissingOrCorrupt, StorageError::PartMissingOrCorrupt), + (DiskError::ShortWrite, StorageError::ShortWrite), + (DiskError::SourceStalled, StorageError::SourceStalled), + (DiskError::Timeout, StorageError::Timeout), + (DiskError::InvalidPath, StorageError::InvalidPath), + ]; + + for (disk_error, expected_storage_error) in cases { + let storage_error: StorageError = disk_error.into(); + + assert!( + !matches!(&storage_error, StorageError::Io(_)), + "expected {expected_storage_error:?}, got generic Io" + ); + assert_eq!(storage_error, expected_storage_error); + } + } + + #[test] + fn test_storage_error_disk_preservation_codes_roundtrip() { + let errors = vec![ + StorageError::MaxVersionsExceeded, + StorageError::InconsistentDisk, + StorageError::UnsupportedDisk, + StorageError::DiskNotDir, + StorageError::DiskOngoingReq, + StorageError::PathNotFound, + StorageError::BitrotHashAlgoInvalid, + StorageError::CrossDeviceLink, + StorageError::LessData, + StorageError::MoreData, + StorageError::OutdatedXLMeta, + StorageError::PartMissingOrCorrupt, + StorageError::ShortWrite, + StorageError::SourceStalled, + StorageError::Timeout, + StorageError::InvalidPath, + ]; + + for original_error in errors { + let code = original_error.to_u32(); + let recovered_error = + StorageError::from_u32(code).unwrap_or_else(|| panic!("failed to recover error from code: {code:#x}")); + + assert_eq!(std::mem::discriminant(&original_error), std::mem::discriminant(&recovered_error)); + } + } + #[test] fn test_storage_error_from_io_error() { // Test direct IO error conversion diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index c965fb923..b85e4a68b 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -8,7 +8,9 @@ pub(crate) mod admin { } pub(crate) mod bucket { - pub(crate) use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions}; + pub(crate) use rustfs_storage_api::{ + BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp, + }; } pub(crate) mod error { diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 6c15d810f..27c28b93e 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -19,8 +19,11 @@ use crate::bucket::{ }; use crate::runtime::sources as runtime_sources; use crate::set_disk::get_lock_acquire_timeout; +use crate::storage_api_contracts::bucket::SRBucketDeleteOp; use crate::storage_api_contracts::namespace::NamespaceLocking as _; +const DELETED_BUCKETS_PREFIX: &str = ".deleted"; + fn should_override_created_from_metadata(created: OffsetDateTime) -> bool { created != OffsetDateTime::UNIX_EPOCH } @@ -67,7 +70,47 @@ fn bucket_delete_metadata_cleanup_prefixes(bucket: &str) -> [String; 2] { ] } +fn bucket_deleted_marker_prefix(bucket: &str) -> String { + format!("{BUCKET_META_PREFIX}/{DELETED_BUCKETS_PREFIX}/{bucket}") +} + +fn bucket_deleted_marker_volume(bucket: &str) -> String { + format!("{RUSTFS_META_BUCKET}/{}", bucket_deleted_marker_prefix(bucket)) +} + impl ECStore { + async fn mark_bucket_deleted(&self, bucket: &str) -> Result<()> { + let marker_volume = bucket_deleted_marker_volume(bucket); + + self.peer_sys + .make_bucket( + marker_volume.as_str(), + &MakeBucketOptions { + force_create: true, + ..Default::default() + }, + ) + .await + .map_err(|e| to_object_err(e.into(), vec![bucket]))?; + + Ok(()) + } + + async fn cleanup_deleted_bucket_metadata(&self, bucket: &str, include_deleted_marker: bool) -> Result<()> { + for prefix in bucket_delete_metadata_cleanup_prefixes(bucket) { + self.delete_all(RUSTFS_META_BUCKET, prefix.as_str()).await?; + } + + if include_deleted_marker { + let marker_prefix = bucket_deleted_marker_prefix(bucket); + self.delete_all(RUSTFS_META_BUCKET, marker_prefix.as_str()).await?; + } + + metadata_sys::remove_bucket_metadata(bucket).await?; + runtime_sources::delete_bucket_monitor_entry(bucket); + Ok(()) + } + #[instrument(skip(self))] pub(super) async fn handle_make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { if !is_meta_bucketname(bucket) @@ -229,11 +272,14 @@ impl ECStore { validate_table_bucket_delete_guard(bucket).await?; + let sr_mark_delete = opts.srdelete_op == SRBucketDeleteOp::MarkDelete; + let sr_purge = opts.srdelete_op == SRBucketDeleteOp::Purge; + // Check bucket is empty before deletion (per S3 API spec) // If bucket is not empty (contains actual objects with xl.meta files) and force // is not set, return BucketNotEmpty error. // Note: Empty directories (left after object deletion) should NOT count as objects. - if !opts.force { + if !opts.force && !sr_mark_delete { let local_disks = all_local_disk().await; for disk in local_disks.iter() { // Check if bucket directory contains any xl.meta files (actual objects) @@ -246,19 +292,18 @@ impl ECStore { } } + if sr_mark_delete { + self.mark_bucket_deleted(bucket).await?; + self.cleanup_deleted_bucket_metadata(bucket, false).await?; + return Ok(()); + } + self.peer_sys .delete_bucket(bucket, opts) .await .map_err(|e| to_object_err(e.into(), vec![bucket]))?; - // TODO: replication opts.srdelete_op - - // Delete internal metadata after the bucket is gone so stale catalog records cannot be reused - // if the same bucket name is created again. - for prefix in bucket_delete_metadata_cleanup_prefixes(bucket) { - self.delete_all(RUSTFS_META_BUCKET, prefix.as_str()).await?; - } - runtime_sources::delete_bucket_monitor_entry(bucket); + self.cleanup_deleted_bucket_metadata(bucket, sr_purge).await?; Ok(()) } } @@ -266,11 +311,124 @@ impl ECStore { #[cfg(test)] mod tests { use super::{ - bucket_delete_metadata_cleanup_prefixes, should_override_created_from_metadata, validate_table_bucket_delete_allowed, + bucket_delete_metadata_cleanup_prefixes, bucket_deleted_marker_prefix, bucket_deleted_marker_volume, + should_override_created_from_metadata, validate_table_bucket_delete_allowed, }; use crate::bucket::metadata::table_bucket_catalog_metadata_prefix; + use crate::bucket::metadata_sys; + use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::StorageError; + use crate::object_api::{ObjectOptions, PutObjReader}; + use crate::storage_api_contracts::{ + bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp}, + object::{ObjectIO as _, ObjectOperations as _}, + }; + use crate::store::{ECStore, init_local_disks}; + use crate::{ + disk::endpoint::Endpoint, + layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, + }; + use std::path::{Path, PathBuf}; + use std::sync::Arc; use time::OffsetDateTime; + use tokio::sync::OnceCell; + use tokio_util::sync::CancellationToken; + use uuid::Uuid; + + static BUCKET_DELETE_TEST_ENV: OnceCell<(Vec, Arc)> = OnceCell::const_new(); + + async fn setup_bucket_delete_test_env() -> (Vec, Arc) { + BUCKET_DELETE_TEST_ENV + .get_or_init(|| async { + let temp_dir = std::env::temp_dir().join(format!("rustfs_bucket_delete_test_{}", Uuid::new_v4())); + tokio::fs::create_dir_all(&temp_dir) + .await + .expect("test base directory should be created"); + + let disk_paths = (0..4) + .map(|disk_idx| temp_dir.join(format!("disk{disk_idx}"))) + .collect::>(); + + for disk_path in &disk_paths { + tokio::fs::create_dir_all(disk_path) + .await + .expect("disk directory should be created"); + } + + let mut endpoints = Vec::with_capacity(disk_paths.len()); + for (disk_idx, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = + Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + endpoints.push(endpoint); + } + + let endpoint_pools = EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "bucket-delete-test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }]); + + init_local_disks(endpoint_pools.clone()) + .await + .expect("local disks should initialize"); + let ecstore = + ECStore::new("127.0.0.1:0".parse().expect("test address"), endpoint_pools, CancellationToken::new()) + .await + .expect("ECStore should initialize"); + + if metadata_sys::get_global_bucket_metadata_sys().is_none() { + metadata_sys::init_bucket_metadata_sys(ecstore.clone(), Vec::new()).await; + } + + (disk_paths, ecstore) + }) + .await + .clone() + } + + async fn create_bucket_with_object(ecstore: &Arc, bucket: &str, object: &str) { + ecstore + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + + let mut reader = PutObjReader::from_vec(b"delete bucket semantics".to_vec()); + ecstore + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("object should be written"); + ecstore + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("object should be readable before bucket delete"); + } + + async fn any_disk_path_exists(disk_paths: &[PathBuf], relative_path: impl AsRef) -> bool { + for disk_path in disk_paths { + if tokio::fs::try_exists(disk_path.join(relative_path.as_ref())) + .await + .expect("test disk path should be stat-able") + { + return true; + } + } + false + } + + async fn any_disk_has_object_metadata(disk_paths: &[PathBuf], bucket: &str) -> bool { + for disk_path in disk_paths { + if super::has_xlmeta_files(&disk_path.join(bucket)).await { + return true; + } + } + false + } #[test] fn should_not_override_when_metadata_created_is_unix_epoch() { @@ -299,4 +457,104 @@ mod tests { assert!(prefixes.contains(&table_bucket_catalog_metadata_prefix("analytics"))); assert!(prefixes.contains(&"buckets/analytics".to_string())); } + + #[test] + fn bucket_delete_marker_path_uses_internal_deleted_bucket_metadata_prefix() { + assert_eq!(bucket_deleted_marker_prefix("analytics"), "buckets/.deleted/analytics"); + assert_eq!( + bucket_deleted_marker_volume("analytics"), + format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}/.deleted/analytics") + ); + } + + #[tokio::test] + async fn bucket_delete_mark_delete_marks_metadata_deleted_without_physical_object_delete() { + let (disk_paths, ecstore) = setup_bucket_delete_test_env().await; + let bucket = format!("bucket-mark-delete-{}", Uuid::new_v4().simple()); + let object = "object.txt"; + + create_bucket_with_object(&ecstore, &bucket, object).await; + assert!(metadata_sys::get(&bucket).await.is_ok()); + + ecstore + .delete_bucket( + &bucket, + &DeleteBucketOptions { + srdelete_op: SRBucketDeleteOp::MarkDelete, + ..Default::default() + }, + ) + .await + .expect("MarkDelete should not reject non-empty bucket data"); + + assert!( + any_disk_has_object_metadata(&disk_paths, &bucket).await, + "MarkDelete must not physically remove object xl.meta data" + ); + assert!( + any_disk_path_exists(&disk_paths, bucket_deleted_marker_volume(&bucket)).await, + "MarkDelete should persist the deleted-bucket marker" + ); + assert!( + metadata_sys::get(&bucket).await.is_err(), + "deleted bucket metadata must be removed from the local cache" + ); + } + + #[tokio::test] + async fn bucket_delete_purge_removes_bucket_data_and_internal_metadata() { + let (disk_paths, ecstore) = setup_bucket_delete_test_env().await; + let bucket = format!("bucket-purge-{}", Uuid::new_v4().simple()); + let object = "object.txt"; + let metadata_prefix = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}/{bucket}"); + + create_bucket_with_object(&ecstore, &bucket, object).await; + assert!(any_disk_path_exists(&disk_paths, &metadata_prefix).await); + + ecstore + .delete_bucket( + &bucket, + &DeleteBucketOptions { + force: true, + srdelete_op: SRBucketDeleteOp::Purge, + ..Default::default() + }, + ) + .await + .expect("Purge should force-delete bucket data"); + + assert!(!any_disk_path_exists(&disk_paths, &bucket).await, "Purge should remove the bucket volume"); + assert!( + !any_disk_path_exists(&disk_paths, &metadata_prefix).await, + "Purge should remove bucket metadata prefix" + ); + assert!( + metadata_sys::get(&bucket).await.is_err(), + "purged bucket metadata must be removed from the local cache" + ); + } + + #[tokio::test] + async fn bucket_delete_default_s3_delete_still_rejects_non_empty_bucket() { + let (disk_paths, ecstore) = setup_bucket_delete_test_env().await; + let bucket = format!("bucket-s3-delete-{}", Uuid::new_v4().simple()); + let object = "object.txt"; + + create_bucket_with_object(&ecstore, &bucket, object).await; + + let err = ecstore + .delete_bucket(&bucket, &DeleteBucketOptions::default()) + .await + .expect_err("default S3 DeleteBucket should reject non-empty buckets"); + + assert!(matches!(err, StorageError::BucketNotEmpty(name) if name == bucket)); + assert!( + any_disk_has_object_metadata(&disk_paths, &bucket).await, + "failed default S3 DeleteBucket must keep object data" + ); + assert!( + metadata_sys::get(&bucket).await.is_ok(), + "failed default S3 DeleteBucket must keep metadata cache" + ); + } } diff --git a/crates/storage-api/src/error.rs b/crates/storage-api/src/error.rs index c5bf13b65..0e7d03c7a 100644 --- a/crates/storage-api/src/error.rs +++ b/crates/storage-api/src/error.rs @@ -87,6 +87,22 @@ pub enum StorageErrorCode { RebalanceAlreadyRunning, OperationCanceled, NamespaceLockQuorumUnavailable, + MaxVersionsExceeded, + InconsistentDisk, + UnsupportedDisk, + DiskNotDir, + DiskOngoingReq, + PathNotFound, + BitrotHashAlgoInvalid, + CrossDeviceLink, + LessData, + MoreData, + OutdatedXLMeta, + PartMissingOrCorrupt, + ShortWrite, + SourceStalled, + Timeout, + InvalidPath, } impl StorageErrorCode { @@ -156,6 +172,22 @@ impl StorageErrorCode { Self::RebalanceAlreadyRunning => 0x40, Self::OperationCanceled => 0x41, Self::NamespaceLockQuorumUnavailable => 0x42, + Self::MaxVersionsExceeded => 0x43, + Self::InconsistentDisk => 0x44, + Self::UnsupportedDisk => 0x45, + Self::DiskNotDir => 0x46, + Self::DiskOngoingReq => 0x47, + Self::PathNotFound => 0x48, + Self::BitrotHashAlgoInvalid => 0x49, + Self::CrossDeviceLink => 0x4A, + Self::LessData => 0x4B, + Self::MoreData => 0x4C, + Self::OutdatedXLMeta => 0x4D, + Self::PartMissingOrCorrupt => 0x4E, + Self::ShortWrite => 0x4F, + Self::SourceStalled => 0x50, + Self::Timeout => 0x51, + Self::InvalidPath => 0x52, } } @@ -225,6 +257,22 @@ impl StorageErrorCode { 0x40 => Some(Self::RebalanceAlreadyRunning), 0x41 => Some(Self::OperationCanceled), 0x42 => Some(Self::NamespaceLockQuorumUnavailable), + 0x43 => Some(Self::MaxVersionsExceeded), + 0x44 => Some(Self::InconsistentDisk), + 0x45 => Some(Self::UnsupportedDisk), + 0x46 => Some(Self::DiskNotDir), + 0x47 => Some(Self::DiskOngoingReq), + 0x48 => Some(Self::PathNotFound), + 0x49 => Some(Self::BitrotHashAlgoInvalid), + 0x4A => Some(Self::CrossDeviceLink), + 0x4B => Some(Self::LessData), + 0x4C => Some(Self::MoreData), + 0x4D => Some(Self::OutdatedXLMeta), + 0x4E => Some(Self::PartMissingOrCorrupt), + 0x4F => Some(Self::ShortWrite), + 0x50 => Some(Self::SourceStalled), + 0x51 => Some(Self::Timeout), + 0x52 => Some(Self::InvalidPath), _ => None, } } @@ -301,9 +349,36 @@ mod tests { (StorageErrorCode::NamespaceLockQuorumUnavailable, 0x42), ]; + const DISK_PRESERVATION_ERROR_CODES: &[(StorageErrorCode, u32)] = &[ + (StorageErrorCode::MaxVersionsExceeded, 0x43), + (StorageErrorCode::InconsistentDisk, 0x44), + (StorageErrorCode::UnsupportedDisk, 0x45), + (StorageErrorCode::DiskNotDir, 0x46), + (StorageErrorCode::DiskOngoingReq, 0x47), + (StorageErrorCode::PathNotFound, 0x48), + (StorageErrorCode::BitrotHashAlgoInvalid, 0x49), + (StorageErrorCode::CrossDeviceLink, 0x4A), + (StorageErrorCode::LessData, 0x4B), + (StorageErrorCode::MoreData, 0x4C), + (StorageErrorCode::OutdatedXLMeta, 0x4D), + (StorageErrorCode::PartMissingOrCorrupt, 0x4E), + (StorageErrorCode::ShortWrite, 0x4F), + (StorageErrorCode::SourceStalled, 0x50), + (StorageErrorCode::Timeout, 0x51), + (StorageErrorCode::InvalidPath, 0x52), + ]; + #[test] fn storage_error_codes_roundtrip() { - for (error_code, raw_code) in ERROR_CODES { + for (error_code, raw_code) in ERROR_CODES.iter().chain(DISK_PRESERVATION_ERROR_CODES) { + assert_eq!(error_code.as_u32(), *raw_code); + assert_eq!(StorageErrorCode::from_u32(*raw_code), Some(*error_code)); + } + } + + #[test] + fn storage_error_disk_preservation_codes_roundtrip() { + for (error_code, raw_code) in DISK_PRESERVATION_ERROR_CODES { assert_eq!(error_code.as_u32(), *raw_code); assert_eq!(StorageErrorCode::from_u32(*raw_code), Some(*error_code)); }