From 3e156ee61e90ca68de0a143ad9ebea90b350ec03 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 13 Sep 2026 22:00:28 +0800 Subject: [PATCH] fix(heal): bind bucket recovery to its original incarnation (#7744) * fix(heal): bind bucket recovery to its original incarnation * fix(heal): preserve stable bucket heal task errors --- .../e2e_test/src/reliant/grpc_lock_server.rs | 35 ++ crates/ecstore/src/cluster/rpc/http_auth.rs | 1 + .../ecstore/src/cluster/rpc/peer_s3_client.rs | 24 +- crates/ecstore/src/cluster/rpc/remote_disk.rs | 58 ++- crates/ecstore/src/disk/disk_store.rs | 30 +- crates/ecstore/src/disk/mod.rs | 22 +- crates/ecstore/src/set_disk/ops/heal.rs | 8 + crates/ecstore/src/store/heal.rs | 413 +++++++++++++++++- crates/ecstore/src/store/mod.rs | 1 + crates/ecstore/src/store/object.rs | 2 +- crates/ecstore/src/store/peer.rs | 103 +++++ crates/heal/src/error.rs | 5 +- crates/heal/src/heal/manager.rs | 11 +- crates/heal/src/heal/manager/root_recovery.rs | 47 +- crates/heal/src/heal/manager/tests.rs | 18 + .../src/heal/manager/tests/root_recovery.rs | 190 +++++++- .../heal/manager/tests/running_mainline.rs | 21 + crates/heal/src/heal/storage.rs | 200 ++++++--- crates/heal/src/heal/storage_api.rs | 3 +- crates/heal/src/heal/task.rs | 6 + crates/heal/src/heal/task/heal_bucket.rs | 49 ++- .../tests/heal_b920_subquorum_union_test.rs | 59 ++- .../heal/tests/shard_identity_receipt_test.rs | 18 + .../src/generated/proto_gen/node_service.rs | 246 +++++++++++ crates/protos/src/lib.rs | 28 +- crates/protos/src/node.proto | 11 + docs/README.md | 3 + docs/operations/bucket-heal-recovery.md | 52 +++ rustfs/src/storage/rpc/node_service.rs | 309 ++++++++++++- rustfs/src/storage/rpc/node_service/bucket.rs | 32 +- rustfs/src/storage/rpc/node_service/disk.rs | 94 +++- 31 files changed, 1978 insertions(+), 121 deletions(-) create mode 100644 docs/operations/bucket-heal-recovery.md diff --git a/crates/e2e_test/src/reliant/grpc_lock_server.rs b/crates/e2e_test/src/reliant/grpc_lock_server.rs index 0e6f62a7a..93c872965 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_server.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_server.rs @@ -360,6 +360,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn heal_bucket_at_incarnation( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn list_bucket( &self, _request: Request, @@ -409,6 +416,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn delete_at_incarnation( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn verify_file( &self, _request: Request, @@ -506,6 +520,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn rename_data_at_incarnation( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn make_volumes( &self, _request: Request, @@ -562,6 +583,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn write_metadata_at_incarnation( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn read_version( &self, _request: Request, @@ -583,6 +611,13 @@ impl NodeService for MinimalLockNodeService { Err(Status::unimplemented("lock-only test server")) } + async fn delete_version_at_incarnation( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("lock-only test server")) + } + async fn delete_retired_marker( &self, _request: Request, diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 32ab79a68..35a5b41b5 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -3012,6 +3012,7 @@ mod tests { fn rename_data_mutation_contract_binds_method_nonce_and_body() { ensure_test_rpc_secret(); let message = rustfs_protos::proto_gen::node_service::RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: "http://node-a:9000/data/rustfs0".to_string(), src_volume: ".rustfs.sys/multipart".to_string(), src_path: "uploads/object".to_string(), diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 6a929113e..29f5ac986 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -741,7 +741,10 @@ impl LocalPeerS3Client { movement_guard_held: bool, ) -> Result { let disks = self.local_disks_for_pools().await.into_iter().map(Some).collect(); - let store = runtime_sources::object_store_handle().filter(|store| Arc::ptr_eq(&store.ctx, &self.instance_ctx)); + let store = crate::store::bucket_heal_scope(bucket) + .map(|scope| scope.store.clone()) + .or_else(runtime_sources::object_store_handle) + .filter(|store| Arc::ptr_eq(&store.ctx, &self.instance_ctx)); #[cfg(not(test))] if store.is_none() { return Err(Error::other("bucket heal refused: pool metadata is unavailable for this instance")); @@ -1266,11 +1269,19 @@ impl PeerS3Client for RemotePeerS3Client { let options = encode_heal_bucket_rpc_options(*opts, fenced_pools)?; let mut client = self.get_client().await?; let mut request = Request::new(HealBucketRequest { + bucket_incarnation_id: crate::store::bucket_heal_scope(bucket) + .map(|scope| scope.incarnation.as_bytes().to_vec().into()) + .unwrap_or_default(), bucket: bucket.to_string(), options, }); set_tonic_mutation_body_digest(&mut request)?; - let response = client.heal_bucket(request).await?.into_inner(); + let response = if request.get_ref().bucket_incarnation_id.is_empty() { + client.heal_bucket(request).await? + } else { + client.heal_bucket_at_incarnation(request).await? + } + .into_inner(); if !response.success { return if let Some(err) = response.error { Err(err.into()) @@ -1520,6 +1531,9 @@ async fn heal_bucket_local_on_disks_with_pool_meta( pool_meta: Option<&RwLock>, dispatch_fenced_pools: &[usize], ) -> Result { + if let Some(scope) = crate::store::bucket_heal_scope(bucket) { + scope.check()?; + } let (fenced_disks, mut fenced_pool_idxs) = snapshot_heal_bucket_fence(&disks, pool_meta, dispatch_fenced_pools).await?; let fenced_disks = Arc::new(fenced_disks); let before_state = Arc::new(RwLock::new(vec![String::new(); disks.len()])); @@ -1633,6 +1647,9 @@ async fn heal_bucket_local_on_disks_with_pool_meta( if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) { return Err(err); } + if let Some(scope) = crate::store::bucket_heal_scope(&bucket) { + scope.check()?; + } mutation_disk.delete_volume(&bucket, false).await }) .await; @@ -1687,6 +1704,9 @@ async fn heal_bucket_local_on_disks_with_pool_meta( if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) { return Err(err); } + if let Some(scope) = crate::store::bucket_heal_scope(&bucket) { + scope.check()?; + } match mutation_disk.make_volume(&bucket).await { Ok(()) | Err(Error::VolumeExists) => Ok(()), Err(err) => Err(err), diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 8dd1af013..627d04a24 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -2122,6 +2122,9 @@ impl RemoteDisk { let file_info_bin = encode_file_info_msgpack(fi)?; let mut client = self.get_client().await?; let mut request = Request::new(RenameDataRequest { + bucket_incarnation_id: crate::store::bucket_heal_scope(dst_volume) + .map(|scope| scope.incarnation.as_bytes().to_vec().into()) + .unwrap_or_default(), disk: self.endpoint.to_string(), src_volume: src_volume.to_string(), src_path: src_path.to_string(), @@ -2134,7 +2137,8 @@ impl RemoteDisk { .unwrap_or_default(), }); let canonical_body = rustfs_protos::canonical_rename_data_request_body(request.get_ref()); - if scanner_publication_lease_token.is_some() { + let incarnation_bound = !request.get_ref().bucket_incarnation_id.is_empty(); + if scanner_publication_lease_token.is_some() || incarnation_bound { let canonical_body = canonical_body.map_err(|_| Error::other("rename_data request length cannot be represented"))?; crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &canonical_body).map_err(Error::other)?; @@ -2142,7 +2146,13 @@ impl RemoteDisk { attach_mutation_body_digest(&mut request, canonical_body, "rename_data")?; } - let response = client.rename_data(request).await?.into_inner(); + let response = if incarnation_bound { + // Older peers return Unimplemented before mutation; never downgrade. + client.rename_data_at_incarnation(request).await? + } else { + client.rename_data(request).await? + } + .into_inner(); if !response.success { return Err(response.error.unwrap_or_default().into()); @@ -2192,6 +2202,9 @@ impl RemoteDisk { let options = serde_json::to_string(&opt)?; let mut client = self.get_client().await?; let mut request = Request::new(DeleteRequest { + bucket_incarnation_id: crate::store::bucket_heal_scope(volume) + .map(|scope| scope.incarnation.as_bytes().to_vec().into()) + .unwrap_or_default(), disk: self.endpoint.to_string(), volume: volume.to_string(), path: path.to_string(), @@ -2201,7 +2214,7 @@ impl RemoteDisk { .unwrap_or_default(), }); let canonical_body = rustfs_protos::canonical_delete_request_body(request.get_ref()); - if scanner_publication_lease_token.is_some() { + if scanner_publication_lease_token.is_some() || !request.get_ref().bucket_incarnation_id.is_empty() { let canonical_body = canonical_body.map_err(|_| Error::other("delete request length cannot be represented"))?; crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &canonical_body).map_err(Error::other)?; @@ -2209,7 +2222,12 @@ impl RemoteDisk { attach_mutation_body_digest(&mut request, canonical_body, "delete")?; } - let response = client.delete(request).await?.into_inner(); + let response = if request.get_ref().bucket_incarnation_id.is_empty() { + client.delete(request).await? + } else { + client.delete_at_incarnation(request).await? + } + .into_inner(); if !response.success { return Err(response.error.unwrap_or_default().into()); @@ -2510,6 +2528,9 @@ impl DiskAPI for RemoteDisk { let mut client = self.get_client().await?; let mut request = Request::new(DeleteVersionRequest { + bucket_incarnation_id: crate::store::bucket_heal_scope(volume) + .map(|scope| scope.incarnation.as_bytes().to_vec().into()) + .unwrap_or_default(), disk: self.endpoint.to_string(), volume: volume.to_string(), path: path.to_string(), @@ -2520,12 +2541,20 @@ impl DiskAPI for RemoteDisk { opts_bin: opts_bin.into(), }); let canonical_body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()); - attach_mutation_body_digest(&mut request, canonical_body, "delete_version")?; + let incarnation_bound = !request.get_ref().bucket_incarnation_id.is_empty(); + if incarnation_bound { + let body = canonical_body.map_err(|_| Error::other("delete-version body length cannot be represented"))?; + crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &body).map_err(Error::other)?; + } else { + attach_mutation_body_digest(&mut request, canonical_body, "delete_version")?; + } - // Unknown RPC methods fail closed on older peers. Never retry a - // conditional delete through the legacy unconditional method. + // The marker-specific method rejects older peers; its body digest + // also binds any incarnation, so neither precondition can be lost. let response = if conditional_marker { client.delete_retired_marker(request).await? + } else if incarnation_bound { + client.delete_version_at_incarnation(request).await? } else { client.delete_version(request).await? } @@ -2797,6 +2826,9 @@ impl DiskAPI for RemoteDisk { let disk = self.disk_ref().await; let mut client = self.get_client().await?; let mut request = Request::new(WriteMetadataRequest { + bucket_incarnation_id: crate::store::bucket_heal_scope(volume) + .map(|scope| scope.incarnation.as_bytes().to_vec().into()) + .unwrap_or_default(), disk, volume: volume.to_string(), path: path.to_string(), @@ -2804,9 +2836,15 @@ impl DiskAPI for RemoteDisk { file_info_bin: file_info_bin.into(), }); let canonical_body = rustfs_protos::canonical_write_metadata_request_body(request.get_ref()); - attach_mutation_body_digest(&mut request, canonical_body, "write_metadata")?; - - let response = client.write_metadata(request).await?.into_inner(); + let response = if request.get_ref().bucket_incarnation_id.is_empty() { + attach_mutation_body_digest(&mut request, canonical_body, "write_metadata")?; + client.write_metadata(request).await? + } else { + let body = canonical_body.map_err(|_| Error::other("write metadata request length cannot be represented"))?; + crate::cluster::rpc::set_tonic_canonical_body_digest(&mut request, &body).map_err(Error::other)?; + client.write_metadata_at_incarnation(request).await? + } + .into_inner(); if !response.success { return Err(response.error.unwrap_or_default().into()); diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index af4933341..45390eefa 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -2072,11 +2072,17 @@ impl DiskAPI for LocalDiskWrapper { } async fn make_volume(&self, volume: &str) -> Result<()> { + // Scoped heal must drain directory creation before releasing its lifecycle owner. + let timeout = if crate::store::bucket_heal_scope(volume).is_some() { + Duration::ZERO + } else { + get_max_timeout_duration() + }; self.track_disk_health_mutation( "make_volume", DiskMetricMutation::Write, || async { self.disk.make_volume(volume).await }, - get_max_timeout_duration(), + timeout, ) .await } @@ -2224,21 +2230,39 @@ impl DiskAPI for LocalDiskWrapper { } async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result { + let scope = crate::store::bucket_heal_scope(volume); + if let Some(scope) = &scope { + scope.check()?; + } + let timeout = if scope.is_some() { + Duration::ZERO + } else { + get_max_timeout_duration() + }; self.track_disk_health_mutation( "delete_data_dir", DiskMetricMutation::Delete, || async { self.disk.delete_data_dir(volume, path, opts).await }, - get_max_timeout_duration(), + timeout, ) .await } async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { + let scope = crate::store::bucket_heal_scope(volume); + if let Some(scope) = &scope { + scope.check()?; + } + let timeout = if scope.is_some() { + Duration::ZERO + } else { + get_max_timeout_duration() + }; self.track_disk_health_mutation( "write_metadata", DiskMetricMutation::Write, || async { self.disk.write_metadata(org_volume, volume, path, fi).await }, - get_max_timeout_duration(), + timeout, ) .await } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index fc9fa50ef..11685bc1c 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -372,6 +372,14 @@ impl DiskAPI for Disk { force_del_marker: bool, opts: DeleteOptions, ) -> Result<()> { + if let Some(scope) = crate::store::bucket_heal_scope(volume) { + scope.check()?; + if let Disk::Local(local_disk) = self { + return local_disk + .delete_version_with_namespace_owner(volume, path, fi, force_del_marker, opts, Some(scope)) + .await; + } + } match self { Disk::Local(local_disk) => local_disk.delete_version(volume, path, fi, force_del_marker, opts).await, Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await, @@ -627,6 +635,12 @@ impl DiskAPI for Disk { #[tracing::instrument(level = "trace", skip_all)] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { + if let Some(scope) = crate::store::bucket_heal_scope(volume) { + scope.check()?; + if let Self::Local(disk) = self { + return disk.delete_with_namespace_owner(volume, path, opt, Some(scope)).await; + } + } match self { Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await, Disk::Remote(remote_disk) => remote_disk.delete(volume, path, opt).await, @@ -806,7 +820,13 @@ impl Disk { dst_volume: &str, dst_path: &str, ) -> Result { - self.rename_data_borrowed_with_fence(src_volume, src_path, fi, dst_volume, dst_path, None) + let Some(scope) = crate::store::bucket_heal_scope(dst_volume) else { + return self + .rename_data_borrowed_with_fence(src_volume, src_path, fi, dst_volume, dst_path, None) + .await; + }; + scope.check()?; + self.rename_data_borrowed_with_fence_and_guard(src_volume, src_path, fi, dst_volume, dst_path, None, Some(scope)) .await } diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 12605ec7f..35517a7d0 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -492,6 +492,7 @@ impl SetDisks { targets: &[String], ) -> disk::error::Result { let disks = self.get_disks_internal().await; + let mut target_disks = Vec::with_capacity(targets.len()); for target in targets { @@ -708,6 +709,10 @@ impl SetDisks { ); let disks = self.get_disks_internal().await; + let bucket_heal_scope = crate::store::bucket_heal_scope(bucket); + if let Some(scope) = &bucket_heal_scope { + scope.check()?; + } let mut result = HealResultItem { heal_item_type: HealItemType::Object.to_string(), @@ -1574,6 +1579,9 @@ impl SetDisks { let mut healed_disks = vec![None; out_dated_disks.len()]; for (index, outdated_disk) in out_dated_disks.iter().enumerate() { if let Some(disk) = outdated_disk { + if let Some(scope) = &bucket_heal_scope { + scope.check()?; + } rename_attempts += 1; // record the index of the updated disks parts_metadata[index].erasure.index = index + 1; diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index faea98823..2fc5fdbb0 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -14,6 +14,7 @@ use super::*; use crate::core::pools::{POOL_META_NAME, load_pool_meta_identity_observing}; +use crate::disk::{self, error::DiskError}; use crate::services::rebalance::{REBAL_META_NAME, RebalStatus}; use crate::set_disk::get_lock_acquire_timeout; use crate::storage_api_contracts::heal::HealOperations as _; @@ -28,6 +29,34 @@ const EVENT_HEAL_ABANDONED_PARTS: &str = "heal_abandoned_parts"; const EVENT_HEAL_FORMAT_COMPLETED: &str = "heal_format_completed"; const EVENT_HEAL_OBJECT_STARTED: &str = "heal_object_started"; +/// An explicit bucket-heal admission owns one lifecycle generation through its write tail. +pub(crate) struct BucketHealScope { + pub(crate) bucket: String, + pub(crate) incarnation: uuid::Uuid, + pub(crate) store: Arc, + fence: super::BucketIncarnationFenceGuard, +} + +impl BucketHealScope { + pub(crate) fn check(&self) -> disk::error::Result<()> { + if self.fence.is_lock_lost() { + return Err(DiskError::other("bucket heal incarnation fence was lost")); + } + Ok(()) + } +} + +tokio::task_local! { + static BUCKET_HEAL_SCOPE: Arc; +} + +pub(crate) fn bucket_heal_scope(bucket: &str) -> Option> { + BUCKET_HEAL_SCOPE + .try_with(|scope| (scope.bucket == bucket).then(|| scope.clone())) + .ok() + .flatten() +} + /// Storage-owned proof for the exact version and every selected erasure location. /// This is an in-process result, never reconstructed from admin drive telemetry. #[derive(Debug)] @@ -114,6 +143,92 @@ fn heal_format_fence_lost_error() -> Error { } impl ECStore { + pub(super) async fn run_bucket_heal_at_incarnation( + self: &Arc, + bucket: &str, + expected: uuid::Uuid, + opts: &HealOpts, + operation: F, + ) -> Result + where + T: Send + 'static, + F: FnOnce(Arc, String, HealOpts) -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, + { + if expected.is_nil() || self.ctx.lock_manager().is_disabled() { + return Err(Error::other( + "incarnation-bound bucket heal requires a valid identity and namespace locking", + )); + } + // Lock order: bucket lifecycle, then bucket/object and capacity locks used by healing. + let fence = self.acquire_bucket_incarnation_fence(bucket, expected).await?; + let scope = Arc::new(BucketHealScope { + bucket: bucket.to_owned(), + incarnation: expected, + store: self.clone(), + fence, + }); + let store = self.clone(); + let bucket = bucket.to_owned(); + let opts = *opts; + // Dropping a caller's cancellation/timeout waiter must not release the lifecycle + // owner while a storage operation is still committing. + tokio::spawn(BUCKET_HEAL_SCOPE.scope(scope.clone(), async move { + scope.check().map_err(Error::from)?; + let result = operation(store, bucket, opts).await; + scope.check().map_err(Error::from)?; + result + })) + .await + .map_err(|error| Error::other_with_context("bucket heal owner task failed", error))? + } + + pub async fn heal_bucket_at_incarnation( + self: &Arc, + bucket: &str, + expected: uuid::Uuid, + opts: &HealOpts, + ) -> Result { + self.run_bucket_heal_at_incarnation(bucket, expected, opts, |store, bucket, opts| async move { + store.heal_bucket(&bucket, &opts).await + }) + .await + } + + pub async fn heal_local_bucket_at_incarnation( + self: &Arc, + bucket: &str, + expected: uuid::Uuid, + opts: &HealOpts, + fenced_pools: Vec, + pools: Option>, + ) -> Result { + self.run_bucket_heal_at_incarnation(bucket, expected, opts, |store, bucket, opts| async move { + let peer = + crate::cluster::rpc::peer_s3_client::LocalPeerS3Client::new_with_instance_ctx(None, pools, store.ctx.clone()); + crate::cluster::rpc::peer_s3_client::PeerS3Client::heal_bucket_with_fence(&peer, &bucket, &opts, &fenced_pools) + .await + .map_err(Error::from) + }) + .await + } + + pub async fn heal_object_at_incarnation( + self: &Arc, + bucket: &str, + object: &str, + version_id: &str, + expected: uuid::Uuid, + opts: &HealOpts, + ) -> Result { + let object = object.to_owned(); + let version_id = version_id.to_owned(); + self.run_bucket_heal_at_incarnation(bucket, expected, opts, |store, bucket, opts| async move { + store.heal_object_with_proof(&bucket, &object, &version_id, &opts).await + }) + .await + } + async fn acquire_heal_format_fence( &self, ) -> Result<( @@ -506,15 +621,35 @@ impl ECStore { // Match object publication: bucket lifecycle before capacity and object // namespace locks. Keep the incarnation pinned through proof delivery. - let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?; + // Reuse the admission's owner: reacquiring a read lock behind a queued + // lifecycle writer would deadlock with the read lock we already hold. + let scope = bucket_heal_scope(bucket).filter(|scope| std::ptr::eq(scope.store.as_ref(), self)); + let guard = if let Some(scope) = &scope { + scope.check().map_err(Error::from)?; + None + } else { + Some(self.acquire_bucket_lifecycle_read_lock(bucket).await?) + }; + let fence_is_lost = || { + scope.as_ref().is_some_and(|scope| scope.fence.is_lock_lost()) + || guard.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost) + }; + let lifecycle_guard = scope + .as_ref() + .and_then(|scope| scope.fence.namespace_lock_guard()) + .or(guard.as_ref()) + .ok_or_else(|| Error::other("bucket heal requires a held lifecycle guard"))?; let retirement = crate::bucket::retirement::MarkerRetirementContext { store: crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await, - current_incarnation: self - .bucket_incarnation_id_from_disk(bucket) - .await - .ok() - .filter(|id| !id.is_nil()), - lifecycle_guard: &guard, + current_incarnation: if let Some(scope) = &scope { + Some(scope.incarnation) + } else { + self.bucket_incarnation_id_from_disk(bucket) + .await + .ok() + .filter(|id| !id.is_nil()) + }, + lifecycle_guard, }; let mut proofs = None; let (item, mut error) = self @@ -522,16 +657,20 @@ impl ECStore { .await?; // Read the authoritative incarnation only for an absence candidate. // The lifecycle guard has pinned it throughout the storage operation. - let incarnation = if proofs.is_some() && !guard.is_lock_lost() { - self.bucket_incarnation_id_from_disk(bucket) - .await - .ok() - .filter(|id| !id.is_nil()) + let incarnation = if proofs.is_some() && !fence_is_lost() { + if let Some(scope) = &scope { + Some(scope.incarnation) + } else { + self.bucket_incarnation_id_from_disk(bucket) + .await + .ok() + .filter(|id| !id.is_nil()) + } } else { None }; let absence = match (incarnation, proofs) { - (Some(incarnation), Some(proofs)) if !guard.is_lock_lost() => Some(HealObjectAbsenceProof { + (Some(incarnation), Some(proofs)) if !fence_is_lost() => Some(HealObjectAbsenceProof { bucket: bucket.to_owned(), object: object.to_owned(), version_id: version_id.to_owned(), @@ -924,6 +1063,254 @@ mod tests { .expect("minimal pool should build") } + #[tokio::test] + #[serial_test::serial] + async fn bucket_incarnation_heal_preserves_successor_shards_and_allows_fresh_repair() { + let (_root, store, shutdown) = multi_pool_heal_store().await; + let bucket = format!("heal-incarnation-{}", Uuid::new_v4().simple()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("original bucket"); + let old = store + .bucket_incarnation_id_from_disk(&bucket) + .await + .expect("original identity"); + store + .delete_bucket(&bucket, &DeleteBucketOptions::default()) + .await + .expect("normal bucket deletion"); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("same-name successor"); + let new = store + .bucket_incarnation_id_from_disk(&bucket) + .await + .expect("successor identity"); + assert_ne!(old, new); + let object = "successor-version"; + let version = Uuid::new_v4().to_string(); + let set = store.pools[0].get_disks(0); + let mut reader = PutObjReader::from_vec(vec![17; 512 * 1024]); + set.put_object( + &bucket, + object, + &mut reader, + &ObjectOptions { + versioned: true, + version_id: Some(version.clone()), + ..Default::default() + }, + ) + .await + .expect("write successor version"); + let missing = set.disks.read().await[0].clone().expect("missing target"); + let healthy = set.disks.read().await[1].clone().expect("healthy target"); + missing + .delete( + &bucket, + object, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + .expect("remove one successor shard"); + let baseline = healthy + .read_all(&bucket, &format!("{object}/xl.meta")) + .await + .expect("healthy baseline"); + let opts = HealOpts { + pool: Some(0), + set: Some(0), + scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep, + ..Default::default() + }; + // Positive control for the original bug: name-only healing repairs the successor. + let (_, error) = store + .heal_object(&bucket, object, &version, &opts) + .await + .expect("unbound control"); + assert!(error.is_none()); + assert!(missing.read_xl(&bucket, object, false).await.is_ok()); + missing + .delete( + &bucket, + object, + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + ) + .await + .expect("restore the same missing-shard condition"); + let fi = healthy + .read_version("", &bucket, object, &version, &Default::default()) + .await + .expect("successor metadata"); + let disk_ref = healthy.endpoint().to_string(); + assert!( + store + .write_local_metadata_at_incarnation(&disk_ref, (&bucket, object), fi.clone(), old) + .await + .is_err() + ); + assert!( + store + .delete_local_path_at_incarnation( + &disk_ref, + (&bucket, object), + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + old + ) + .await + .is_err() + ); + assert!( + store + .delete_local_version_at_incarnation( + &disk_ref, + (&bucket, object), + fi.clone(), + false, + DeleteOptions::default(), + old + ) + .await + .is_err() + ); + assert!( + store + .rename_local_data_at_incarnation(&disk_ref, (&bucket, object), &fi, (&bucket, "stale-destination"), old) + .await + .is_err() + ); + assert!(store.heal_bucket_at_incarnation(&bucket, old, &opts).await.is_err()); + assert!( + store + .heal_object_at_incarnation(&bucket, object, &version, old, &opts) + .await + .is_err() + ); + assert!( + missing.read_xl(&bucket, object, false).await.is_err(), + "obsolete admission must not publish a shard" + ); + assert_eq!( + healthy + .read_all(&bucket, &format!("{object}/xl.meta")) + .await + .expect("healthy after stale heal"), + baseline + ); + store + .heal_bucket_at_incarnation(&bucket, new, &opts) + .await + .expect("fresh bucket metadata repair"); + let result = store + .heal_object_at_incarnation(&bucket, object, &version, new, &opts) + .await + .expect("new admission repair"); + assert!(result.error.is_none(), "new admission must repair independently: {:?}", result.error); + assert!( + missing.read_xl(&bucket, object, false).await.is_ok(), + "new admission publishes the missing shard" + ); + store + .write_local_metadata_at_incarnation(&disk_ref, (&bucket, object), fi.clone(), new) + .await + .expect("fresh target metadata"); + store + .delete_local_version_at_incarnation(&disk_ref, (&bucket, object), fi, false, DeleteOptions::default(), new) + .await + .expect("fresh target version cleanup"); + healthy + .write_all(&bucket, "fresh-cleanup/data", bytes::Bytes::from_static(b"temporary")) + .await + .expect("cleanup fixture"); + store + .delete_local_path_at_incarnation( + &disk_ref, + (&bucket, "fresh-cleanup"), + DeleteOptions { + recursive: true, + immediate: true, + ..Default::default() + }, + new, + ) + .await + .expect("fresh target path cleanup"); + shutdown.cancel(); + } + + #[tokio::test] + #[serial_test::serial] + async fn bucket_incarnation_heal_owner_survives_cancelled_waiter_and_queued_writer() { + let (_root, store, shutdown) = multi_pool_heal_store().await; + let bucket = format!("heal-owner-{}", Uuid::new_v4().simple()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("bucket"); + let identity = store.bucket_incarnation_id_from_disk(&bucket).await.expect("identity"); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (drained_tx, drained_rx) = tokio::sync::oneshot::channel(); + let worker_store = store.clone(); + let worker_bucket = bucket.clone(); + let waiter = tokio::spawn(async move { + worker_store + .run_bucket_heal_at_incarnation( + &worker_bucket, + identity, + &HealOpts::default(), + move |store, bucket, opts| async move { + started_tx.send(()).expect("announce held generation"); + release_rx.await.expect("release commit"); + let result = store + .heal_object_with_proof(&bucket, "history.txt", &Uuid::new_v4().to_string(), &opts) + .await?; + let proof = result.absence.expect("absence proof must reuse the held lifecycle owner"); + assert_eq!(proof.bucket_incarnation_id, identity); + assert_eq!(proof.locations.len(), 2); + drained_tx.send(()).expect("announce drain"); + Ok(()) + }, + ) + .await + }); + started_rx.await.expect("operation acquired generation fence"); + waiter.abort(); + assert!(waiter.await.expect_err("waiter aborted").is_cancelled()); + let writer_store = store.clone(); + let writer_bucket = bucket.clone(); + let mut writer = Box::pin(writer_store.acquire_bucket_lifecycle_write_lock(&writer_bucket)); + assert!( + futures::poll!(&mut writer).is_pending(), + "cancellation must not release the live operation's generation" + ); + release_tx.send(()).expect("finish operation"); + tokio::time::timeout(std::time::Duration::from_secs(10), drained_rx) + .await + .expect("proof must not reacquire a lifecycle read lock behind the queued writer") + .expect("physical operation drains"); + let guard = tokio::time::timeout(std::time::Duration::from_secs(10), writer) + .await + .expect("writer resumes after drain") + .expect("lifecycle writer"); + drop(guard); + shutdown.cancel(); + } + async fn minimal_heal_store() -> ECStore { ECStore { id: Uuid::new_v4(), diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 6dfc2b482..2198f7d2b 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -419,6 +419,7 @@ mod bucket_fence; pub(crate) use bucket::await_bucket_namespace_operation; pub use bucket_fence::BucketIncarnationFenceGuard; mod heal; +pub(crate) use heal::bucket_heal_scope; pub use heal::{HealObjectAbsenceProof, HealObjectStorageResult}; mod heal_walk; pub use heal_walk::HealWalkVersion; diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 98372573d..caa7e36b6 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -8591,7 +8591,7 @@ mod tests { ); drop(failure); let cleaned = store - .heal_object_with_proof(bucket, object, &marker.to_string(), &heal) + .heal_object_at_incarnation(bucket, object, &marker.to_string(), current, &heal) .await .expect("cleanup retired marker"); assert!(cleaned.error.is_none(), "cleanup error: {:?}", cleaned.error); diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 8cbadeda8..eb79964db 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -80,6 +80,109 @@ fn validate_bootstrap_volume(volume: &str) -> DiskResult<()> { } impl ECStore { + pub async fn write_local_metadata_at_incarnation( + self: &Arc, + disk_ref: &str, + target: (&str, &str), + fi: FileInfo, + expected: Uuid, + ) -> DiskResult<()> { + let disk_ref = disk_ref.to_owned(); + let path = target.1.to_owned(); + self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move { + let (disk, id) = local_disk_candidate(&store.ctx, &disk_ref).await?; + let _owner = admit_local_disk(&store.ctx, &disk, id, true).await?; + disk.write_metadata("", &volume, &path, fi).await.map_err(Into::into) + }) + .await + .map_err(|error| DiskError::other(error.to_string())) + } + + pub async fn delete_local_path_at_incarnation( + self: &Arc, + disk_ref: &str, + target: (&str, &str), + opts: DeleteOptions, + expected: Uuid, + ) -> DiskResult<()> { + let disk_ref = disk_ref.to_owned(); + let path = target.1.to_owned(); + self.run_bucket_heal_at_incarnation(target.0, expected, &Default::default(), move |store, volume, _| async move { + let (disk, id) = local_disk_candidate(&store.ctx, &disk_ref).await?; + let owner = admit_local_disk(&store.ctx, &disk, id, true).await?; + let scope = super::bucket_heal_scope(&volume).ok_or_else(|| StorageError::other("missing bucket heal scope"))?; + scope.check()?; + disk.delete_with_namespace_owner(&volume, &path, opts, Some(Arc::new((scope, owner)))) + .await + .map_err(Into::into) + }) + .await + .map_err(|error| DiskError::other(error.to_string())) + } + + pub async fn delete_local_version_at_incarnation( + &self, + disk_ref: &str, + target: (&str, &str), + fi: FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + expected: Uuid, + ) -> DiskResult<()> { + if expected.is_nil() || opts.undo_write || self.ctx.lock_manager().is_disabled() { + return Err(DiskError::other("invalid incarnation-bound heal deletion")); + } + let fence = Arc::new( + self.acquire_bucket_incarnation_fence(target.0, expected) + .await + .map_err(|error| DiskError::other(error.to_string()))?, + ); + let (disk, id) = local_disk_candidate(&self.ctx, disk_ref).await?; + let owner = admit_local_disk(&self.ctx, &disk, id, true).await?; + if fence.is_lock_lost() { + return Err(DiskError::other("bucket heal incarnation fence was lost")); + } + disk.delete_version_with_namespace_owner(target.0, target.1, fi, force_del_marker, opts, Some(Arc::new((fence, owner)))) + .await + } + + /// Target-side admission for an incarnation-bound heal rename. The physical + /// namespace operations retain the lifecycle owner after RPC cancellation. + pub async fn rename_local_data_at_incarnation( + &self, + disk_ref: &str, + source: (&str, &str), + fi: &FileInfo, + destination: (&str, &str), + expected: Uuid, + ) -> DiskResult { + if expected.is_nil() || self.ctx.lock_manager().is_disabled() { + return Err(DiskError::other( + "incarnation-bound rename requires namespace locking and a non-nil identity", + )); + } + let guard = Arc::new( + self.acquire_bucket_incarnation_fence(destination.0, expected) + .await + .map_err(|error| DiskError::other(error.to_string()))?, + ); + if guard.is_lock_lost() { + return Err(DiskError::other("bucket heal incarnation fence was lost")); + } + rename_local_data_with_ctx( + &self.ctx, + disk_ref, + source, + fi, + destination, + RenameDataGuards { + external_guard: Some(guard), + ..Default::default() + }, + ) + .await + } + /// Execute on this instance's active local disk through the physical owner. pub async fn rename_local_data( &self, diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index effd74d9a..8b636c265 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -54,6 +54,9 @@ pub enum Error { #[error("Heal task execution failed: {message}")] TaskExecutionFailed { message: String }, + #[error("stale_bucket_incarnation: bucket {bucket} no longer belongs to this heal admission ({expected:?})")] + StaleBucketIncarnation { bucket: String, expected: Option }, + /// The current page already exhausted its local retry budget. Retrying /// the enclosing bucket would replay pages whose results were counted. #[error("Heal listing failed for bucket {bucket}: {source}")] @@ -98,7 +101,7 @@ impl Error { /// catches errors whose typed identity was destroyed upstream. pub(crate) fn is_recoverable_heal(&self) -> bool { match self { - Error::TaskCancelled | Error::TaskTimeout => false, + Error::TaskCancelled | Error::TaskTimeout | Error::StaleBucketIncarnation { .. } => false, Error::TransientSkip { .. } => true, // Lock failures classify by LockError's own taxonomy: only the // fatal variants (ResourceNotFound / PermissionDenied / diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 8d3539deb..40f7b7c5a 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -521,6 +521,7 @@ fn active_heal_for_dedup_key(active_heals: &HashMap>, key: fn request_matches_task(request: &HealRequest, task: &HealTask) -> bool { request.heal_type == task.heal_type + && request.bucket_incarnation_id == task.bucket_incarnation_id && request.options == task.options && request.priority == task.priority && request.source == task.source @@ -530,6 +531,7 @@ fn request_matches_task(request: &HealRequest, task: &HealTask) -> bool { fn request_matches_request(request: &HealRequest, existing: &HealRequest) -> bool { request.heal_type == existing.heal_type + && request.bucket_incarnation_id == existing.bucket_incarnation_id && request.options == existing.options && request.priority == existing.priority && request.source == existing.source @@ -1782,7 +1784,7 @@ impl HealManager { async fn submit_heal_request_with_receipt_alias_and_mrf_notice( &self, - request: HealRequest, + mut request: HealRequest, preserve_alias: bool, accept_same_request_id_replay: bool, mrf_notice_target: Option, @@ -1809,6 +1811,13 @@ impl HealManager { Vec::new() }; + if source == HealRequestSource::Admin + && let HealType::Bucket { bucket } = &request.heal_type + && request.bucket_incarnation_id.is_none() + { + request.bucket_incarnation_id = Some(self.storage.admit_bucket_incarnation(bucket).await?); + } + let config = self.config.read().await; let dedup_key = PriorityHealQueue::make_dedup_key(&request); diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index dc4b004b5..49c4daaa1 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -22,13 +22,15 @@ use super::*; use crate::heal::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes}; use crate::heal::{DiskStore, RUSTFS_META_BUCKET}; use serde::{Deserialize, Serialize}; +use uuid::Uuid; // The metadata bucket already exists and its parent is durable. Creating a // nested journal directory here would also require syncing every ancestor. const ROOT_RECOVERY_PREFIX: &str = "root-heal-"; const ROOT_TERMINAL_PREFIX: &str = "terminal-root-heal-"; const LEGACY_ROOT_RECOVERY_SCHEMA: u32 = 1; -const ROOT_RECOVERY_SCHEMA: u32 = 2; +const SCOPED_ROOT_RECOVERY_SCHEMA: u32 = 2; +const ROOT_RECOVERY_SCHEMA: u32 = 3; const ROOT_TERMINAL_SCHEMA: u32 = 1; const ROOT_TERMINAL_GC_SCAN_BUDGET: usize = 1024; const ROOT_TERMINAL_GC_DELETE_BUDGET: usize = 64; @@ -194,6 +196,8 @@ struct RootHealIntent { task_id: String, #[serde(default = "default_recovery_heal_type")] heal_type: RecoveryHealType, + #[serde(default, skip_serializing_if = "Option::is_none")] + bucket_incarnation_id: Option, #[serde(deserialize_with = "decode_options")] options: HealOptions, priority: HealPriority, @@ -268,6 +272,7 @@ impl RootHealIntent { schema: ROOT_RECOVERY_SCHEMA, task_id: request.id.clone(), heal_type: RecoveryHealType::from(&request.heal_type), + bucket_incarnation_id: request.bucket_incarnation_id, options: request.options.clone(), priority: request.priority, retry_attempts: request.retry_attempts, @@ -278,6 +283,7 @@ impl RootHealIntent { fn into_request(self) -> HealRequest { let mut request = HealRequest::new(self.heal_type.into(), self.options, self.priority); request.id = self.task_id; + request.bucket_incarnation_id = self.bucket_incarnation_id; request.source = HealRequestSource::Admin; request.retry_attempts = self.retry_attempts; request.created_at = self.created_at; @@ -359,10 +365,17 @@ fn decode_intent(task_id: &str, bytes: &[u8]) -> Result { return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}"))); } match intent.schema { - LEGACY_ROOT_RECOVERY_SCHEMA if intent.heal_type == RecoveryHealType::Cluster => {} + LEGACY_ROOT_RECOVERY_SCHEMA + if intent.heal_type == RecoveryHealType::Cluster && intent.bucket_incarnation_id.is_none() => {} + SCOPED_ROOT_RECOVERY_SCHEMA if intent.bucket_incarnation_id.is_none() => {} ROOT_RECOVERY_SCHEMA => {} _ => return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}"))), } + if !matches!(intent.heal_type, RecoveryHealType::Bucket { .. }) && intent.bucket_incarnation_id.is_some() { + return Err(Error::Other(format!( + "Unexpected bucket incarnation in root heal recovery record {task_id}" + ))); + } intent.heal_type.validate()?; Ok(intent) } @@ -617,7 +630,9 @@ impl RootHealRecovery { return Ok(()); }; let mut intent = decode_intent(&task.id, &expected)?; - if HealType::from(intent.heal_type.clone()) != task.heal_type { + if HealType::from(intent.heal_type.clone()) != task.heal_type + || intent.bucket_incarnation_id != task.bucket_incarnation_id + { return Err(Error::Other(format!("Root heal recovery owner changed for {}", task.id))); } let mut expected_options = intent.options.clone(); @@ -928,7 +943,31 @@ impl HealManager { // Decode every record before admitting anything. These are already // accepted responsibilities, so restore distinct IDs even when their // paths overlap or the configured admission capacity has changed. - let requests = self.root_recovery.pending().await?; + let pending = self.root_recovery.pending().await?; + let mut requests = Vec::with_capacity(pending.len()); + for request in pending { + if let HealType::Bucket { bucket } = &request.heal_type { + match self + .storage + .validate_bucket_incarnation(bucket, request.bucket_incarnation_id) + .await + { + Ok(()) => {} + Err(error @ Error::StaleBucketIncarnation { .. }) => { + let mut terminal = RootHealTerminal::cancelled(&request.id, &request.heal_type, request.options.clone()); + terminal.status = HealTaskStatus::Failed { + error: error.to_string(), + }; + self.root_recovery + .persist_terminal(&request.id, &request.heal_type, request.source, &terminal.into_completed()) + .await?; + continue; + } + Err(error) => return Err(error), + } + } + requests.push(request); + } let active = self.active_heals.lock().await; let mut queue = self.heal_queue.lock().await; let retrying = self.retrying_heals.lock().await; diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index b66b89972..415ce9781 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -28,6 +28,7 @@ use tempfile::TempDir; pub(super) mod admin_overlap; mod root_recovery; +use uuid::Uuid; mod running_mainline; use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo}; @@ -559,6 +560,23 @@ async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_han #[async_trait::async_trait] impl HealStorageAPI for MockStorage { + async fn admit_bucket_incarnation(&self, bucket: &str) -> Result { + if bucket.starts_with("incarnation-metadata-unavailable-") { + return Err(Error::Storage(EcstoreError::SlowDown)); + } + root_recovery::test_bucket_incarnation(bucket) + .filter(|id| !id.is_nil()) + .ok_or_else(|| Error::StaleBucketIncarnation { + bucket: bucket.to_owned(), + expected: None, + }) + } + + async fn heal_bucket_at_incarnation(&self, bucket: &str, expected: Uuid, opts: &HealOpts) -> Result { + self.validate_bucket_incarnation(bucket, Some(expected)).await?; + self.heal_bucket(bucket, opts).await + } + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result> { Ok(None) } diff --git a/crates/heal/src/heal/manager/tests/root_recovery.rs b/crates/heal/src/heal/manager/tests/root_recovery.rs index 3dfc481d0..463ad5750 100644 --- a/crates/heal/src/heal/manager/tests/root_recovery.rs +++ b/crates/heal/src/heal/manager/tests/root_recovery.rs @@ -17,6 +17,20 @@ use super::*; use crate::heal::RUSTFS_META_BUCKET; use std::collections::HashSet; +fn bucket_incarnations() -> &'static std::sync::Mutex>> { + static IDS: std::sync::OnceLock>>> = std::sync::OnceLock::new(); + IDS.get_or_init(Default::default) +} + +pub(super) fn test_bucket_incarnation(bucket: &str) -> Option { + bucket_incarnations() + .lock() + .expect("bucket incarnation fixture") + .get(bucket) + .copied() + .unwrap_or(Some(Uuid::from_u128(42))) +} + #[cfg(unix)] struct RestoreDirectoryMode { path: std::path::PathBuf, @@ -93,6 +107,9 @@ fn root_request() -> HealRequest { fn admin_request(heal_type: HealType) -> HealRequest { let mut request = HealRequest::new(heal_type, HealOptions::default(), HealPriority::High); request.source = HealRequestSource::Admin; + if let HealType::Bucket { bucket } = &request.heal_type { + request.bucket_incarnation_id = test_bucket_incarnation(bucket); + } request } @@ -1416,7 +1433,7 @@ async fn root_recovery_invalid_records_are_retained_without_partial_replay() { let original = disk.read_all(RUSTFS_META_BUCKET, &path).await.expect("read root record"); let mut value: serde_json::Value = serde_json::from_slice(&original).expect("record JSON"); match kind { - "schema" => value["schema"] = 3.into(), + "schema" => value["schema"] = 4.into(), "identity" => value["task_id"] = valid.id.clone().into(), "option" => value["options"]["future_delete_mode"] = true.into(), "no_lock" => value["options"]["no_lock"] = true.into(), @@ -1654,3 +1671,174 @@ async fn root_recovery_terminal_timeout_updates_only_existing_journal_before_sec ); } } + +#[tokio::test] +async fn bucket_incarnation_recovery_retires_old_owner_and_admits_successor() { + let (_temp, disk) = recovery_disk().await; + let bucket = format!("incarnation-replay-{}", Uuid::new_v4()); + let old = Uuid::new_v4(); + let new = Uuid::new_v4(); + bucket_incarnations() + .lock() + .expect("fixture") + .insert(bucket.clone(), Some(old)); + let manager = recovery_manager(vec![disk.clone()]); + let mut request = HealRequest::bucket(bucket.clone()); + request.source = HealRequestSource::Admin; + request.options.recursive = true; + let token = request.id.clone(); + manager + .submit_heal_request_with_receipt(request) + .await + .expect("admit original bucket"); + let pending = manager.root_recovery.pending().await.expect("read admission"); + assert_eq!(pending[0].bucket_incarnation_id, Some(old)); + let original = disk + .read_all(RUSTFS_META_BUCKET, &format!("root-heal-{token}.json")) + .await + .expect("raw admission"); + drop(manager); + + let same = recovery_manager(vec![disk.clone()]); + same.replay_root_heals().await.expect("same generation restart"); + let restored = same + .heal_queue + .lock() + .await + .requests() + .next() + .expect("restored owner") + .clone(); + assert_eq!(restored.id, token); + assert_eq!(restored.bucket_incarnation_id, Some(old)); + assert!(restored.options.recursive); + let retry = HealTask::from_request(restored, Arc::new(MockStorage)).retry_request(); + assert_eq!(retry.bucket_incarnation_id, Some(old)); + drop(same); + + bucket_incarnations() + .lock() + .expect("fixture") + .insert(bucket.clone(), Some(new)); + let restarted = recovery_manager(vec![disk.clone()]); + restarted.replay_root_heals().await.expect("retire obsolete admission"); + assert_eq!(restarted.heal_queue.lock().await.len(), 0); + assert!(matches!(restarted.get_task_status(&token).await.expect("old token remains queryable"), + HealTaskStatus::Failed { error } if error.starts_with("stale_bucket_incarnation:"))); + let mut fresh = HealRequest::bucket(bucket.clone()); + fresh.source = HealRequestSource::Admin; + let receipt = restarted + .submit_heal_request_with_receipt(fresh) + .await + .expect("new generation admission"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + assert_ne!(receipt.task_id, token); + assert_eq!( + restarted + .heal_queue + .lock() + .await + .requests() + .next() + .expect("new owner") + .bucket_incarnation_id, + Some(new) + ); + + // A terminal receipt must dominate a duplicate old journal after another crash. + disk.write_all(RUSTFS_META_BUCKET, &format!("root-heal-{token}.json"), original) + .await + .expect("restore old bytes"); + let again = recovery_manager(vec![disk]); + again.replay_root_heals().await.expect("restart with duplicate old bytes"); + assert!(again.heal_queue.lock().await.requests().all(|request| request.id != token)); + assert!(matches!( + again.get_task_status(&token).await.expect("terminal persists"), + HealTaskStatus::Failed { .. } + )); + bucket_incarnations().lock().expect("fixture").remove(&bucket); +} + +#[tokio::test] +async fn bucket_incarnation_legacy_admissions_are_queryable_without_rebinding() { + for identity in [None, Some(Uuid::nil())] { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk.clone()]); + let request = admin_request(HealType::Bucket { + bucket: format!("legacy-{}", Uuid::new_v4()), + }); + manager + .root_recovery + .persist(&request) + .await + .expect("capture server admission"); + let path = format!("root-heal-{}.json", request.id); + let bytes = disk.read_all(RUSTFS_META_BUCKET, &path).await.expect("read JSON"); + let mut legacy: serde_json::Value = serde_json::from_slice(&bytes).expect("decode JSON"); + if let Some(id) = identity { + legacy["bucket_incarnation_id"] = serde_json::json!(id); + } else { + legacy["schema"] = 2.into(); + legacy.as_object_mut().expect("root record").remove("bucket_incarnation_id"); + } + disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&legacy).expect("legacy bytes").into()) + .await + .expect("legacy journal"); + let restarted = recovery_manager(vec![disk]); + restarted.replay_root_heals().await.expect("retire unsafe legacy task"); + assert_eq!(restarted.heal_queue.lock().await.len(), 0); + assert!(matches!(restarted.get_task_status(&request.id).await.expect("legacy token"), + HealTaskStatus::Failed { error } if error.starts_with("stale_bucket_incarnation:"))); + } +} + +#[tokio::test] +async fn bucket_incarnation_queued_and_retrying_work_does_not_rebind() { + for successor in [None, Some(Uuid::new_v4())] { + let bucket = format!("queued-incarnation-{}", Uuid::new_v4()); + let old = Uuid::new_v4(); + bucket_incarnations() + .lock() + .expect("fixture") + .insert(bucket.clone(), Some(old)); + let request = admin_request(HealType::Bucket { bucket: bucket.clone() }); + let retry = HealTask::from_request(request.clone(), Arc::new(MockStorage)).retry_request(); + bucket_incarnations() + .lock() + .expect("fixture") + .insert(bucket.clone(), successor); + for pending in [request, retry] { + let task = HealTask::from_request(pending, Arc::new(MockStorage)); + let error = task.execute().await.expect_err("obsolete generation cannot execute"); + assert!(matches!(error, Error::StaleBucketIncarnation { expected: Some(id), .. } if id == old)); + assert!(!error.is_recoverable_heal(), "never retry against a successor"); + assert_eq!(task.get_outcome().await.counters.processed, 0); + } + bucket_incarnations().lock().expect("fixture").remove(&bucket); + } +} + +#[tokio::test] +async fn bucket_incarnation_metadata_failure_defers_replay_without_retiring_owner() { + let (_temp, disk) = recovery_disk().await; + let manager = recovery_manager(vec![disk]); + let request = admin_request(HealType::Bucket { + bucket: format!("incarnation-metadata-unavailable-{}", Uuid::new_v4()), + }); + manager + .root_recovery + .persist(&request) + .await + .expect("accepted responsibility"); + assert!(matches!(manager.replay_root_heals().await, Err(Error::Storage(EcstoreError::SlowDown)))); + assert_eq!(manager.root_recovery.pending().await.expect("pending owner").len(), 1); + assert!( + manager + .root_recovery + .completed(&request.id) + .await + .expect("terminal lookup") + .is_none() + ); + assert!(manager.heal_queue.lock().await.requests().next().is_none()); +} diff --git a/crates/heal/src/heal/manager/tests/running_mainline.rs b/crates/heal/src/heal/manager/tests/running_mainline.rs index f26e4bece..d7b827ce6 100644 --- a/crates/heal/src/heal/manager/tests/running_mainline.rs +++ b/crates/heal/src/heal/manager/tests/running_mainline.rs @@ -47,6 +47,27 @@ struct RunningStorage { #[async_trait::async_trait] impl HealStorageAPI for RunningStorage { + async fn admit_bucket_incarnation(&self, _: &str) -> Result { + Ok(Uuid::from_u128(42)) + } + + async fn heal_bucket_at_incarnation(&self, bucket: &str, expected: Uuid, opts: &HealOpts) -> Result { + self.validate_bucket_incarnation(bucket, Some(expected)).await?; + self.heal_bucket(bucket, opts).await + } + + async fn heal_object_at_incarnation( + &self, + bucket: &str, + object: &str, + version_id: Option<&str>, + expected: Uuid, + opts: &HealOpts, + ) -> Result { + self.validate_bucket_incarnation(bucket, Some(expected)).await?; + self.heal_object_with_receipt(bucket, object, version_id, opts).await + } + async fn get_object_meta(&self, _: &str, _: &str) -> Result> { Ok(None) } diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index 962055742..ee0da5bba 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -26,8 +26,8 @@ use super::outcome::{HealObjectDisposition, HealObjectIdentity, HealObjectKind, use super::progress::stable_generation; use super::storage_api::owner::{EcstoreHealLifecycleExpiryContext, ecstore_load_admin_data_usage_from_backend_cached}; use super::storage_api::storage::{ - BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _, - ObjectOperations as _, StorageAdminApi, + BucketInfo, BucketOperations, DiskSetSelector, EcstoreHealObjectStorageResult, HealOperations as _, ListOperations as _, + ObjectIO as _, ObjectOperations as _, StorageAdminApi, }; use super::{DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity}; pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader}; @@ -76,6 +76,16 @@ pub struct HealStorageObjectResult { pub receipt: Option, } +fn incarnation_storage_error(bucket: &str, expected: Uuid, error: StorageError) -> Error { + match error { + StorageError::BucketNotFound(_) => Error::StaleBucketIncarnation { + bucket: bucket.to_owned(), + expected: Some(expected), + }, + error => Error::Storage(error), + } +} + impl From<(HealResultItem, Option)> for HealStorageObjectResult { fn from((item, error): (HealResultItem, Option)) -> Self { Self { @@ -438,6 +448,46 @@ pub trait HealStorageAPI: Send + Sync { Ok(None) } + /// Admission must use authoritative metadata, not an outcome cache. + async fn admit_bucket_incarnation(&self, bucket: &str) -> Result { + self.bucket_incarnation_id(bucket) + .await? + .filter(|id| !id.is_nil()) + .ok_or_else(|| Error::StaleBucketIncarnation { + bucket: bucket.to_owned(), + expected: None, + }) + } + + async fn validate_bucket_incarnation(&self, bucket: &str, expected: Option) -> Result<()> { + let stale = || Error::StaleBucketIncarnation { + bucket: bucket.to_owned(), + expected, + }; + let expected = expected.filter(|id| !id.is_nil()).ok_or_else(stale)?; + match self.admit_bucket_incarnation(bucket).await { + Ok(current) if current == expected => Ok(()), + Ok(_) | Err(Error::StaleBucketIncarnation { .. }) => Err(stale()), + Err(error) => Err(error), + } + } + + /// Implementations must retain the bucket lifecycle fence through storage mutation. + async fn heal_bucket_at_incarnation(&self, _bucket: &str, _expected: Uuid, _opts: &HealOpts) -> Result { + Err(Error::other("storage does not support incarnation-bound bucket healing")) + } + + async fn heal_object_at_incarnation( + &self, + _bucket: &str, + _object: &str, + _version_id: Option<&str>, + _expected: Uuid, + _opts: &HealOpts, + ) -> Result { + Err(Error::other("storage does not support incarnation-bound object healing")) + } + /// Heal object using ecstore async fn heal_object( &self, @@ -573,6 +623,66 @@ impl ECStoreHealStorage { Self { ecstore } } + async fn object_result_with_receipt( + &self, + bucket: &str, + object: &str, + version_id: Option<&str>, + opts: &HealOpts, + result: EcstoreHealObjectStorageResult, + expected: Option, + ) -> HealStorageObjectResult { + let item = result.item; + let error = result.error.map(Error::Storage); + let receipt = if let Some(proof) = result.absence { + if error.is_none() + && !opts.dry_run + && proof.bucket == bucket + && proof.object == object + && proof.version_id == version_id.unwrap_or("") + && proof.pool_index == opts.pool + && proof.set_index == opts.set + && !proof.bucket_incarnation_id.is_nil() + && expected.is_none_or(|expected| expected == proof.bucket_incarnation_id) + && !proof.locations.is_empty() + && proof.locations.iter().all(|(pool, set)| { + opts.pool.is_none_or(|expected| expected == *pool) && opts.set.is_none_or(|expected| expected == *set) + }) + { + Some(HealObjectReceipt { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: proof.bucket, + object: proof.object, + version_id: version_id.map(ToOwned::to_owned), + bucket_incarnation_id: Some(proof.bucket_incarnation_id), + pool_index: proof.pool_index, + set_index: proof.set_index, + }, + // A committed cleanup repaired the stale replica. A replay + // observing an already absent version made no new repair. + disposition: if proof.removed { + HealObjectDisposition::Repaired + } else { + HealObjectDisposition::AuthoritativelyAbsent + }, + }) + } else { + None + } + } else if error.is_none() && !opts.dry_run && item.integrity_verified { + let bucket_incarnation_id = match expected { + Some(expected) => Some(expected), + None => self.ecstore.bucket_incarnation_id(bucket).await.ok(), + }; + bucket_incarnation_id + .and_then(|incarnation| verified_object_receipt(bucket, object, version_id, opts, &item, incarnation)) + } else { + None + }; + HealStorageObjectResult { item, error, receipt } + } + /// Read back an object's bytes, capped to bound memory. /// /// Private support for the reserved `ec_decode_rebuild` (HS-01); not part @@ -697,6 +807,42 @@ fn is_transient_object_exists_error(err: &StorageError) -> bool { #[async_trait] impl HealStorageAPI for ECStoreHealStorage { + async fn admit_bucket_incarnation(&self, bucket: &str) -> Result { + match self.ecstore.bucket_incarnation_id_from_disk(bucket).await { + Ok(id) if !id.is_nil() => Ok(id), + Ok(_) | Err(StorageError::BucketNotFound(_)) => Err(Error::StaleBucketIncarnation { + bucket: bucket.to_owned(), + expected: None, + }), + Err(error) => Err(Error::Storage(error)), + } + } + + async fn heal_bucket_at_incarnation(&self, bucket: &str, expected: Uuid, opts: &HealOpts) -> Result { + self.ecstore + .heal_bucket_at_incarnation(bucket, expected, opts) + .await + .map_err(|error| incarnation_storage_error(bucket, expected, error)) + } + + async fn heal_object_at_incarnation( + &self, + bucket: &str, + object: &str, + version_id: Option<&str>, + expected: Uuid, + opts: &HealOpts, + ) -> Result { + let result = self + .ecstore + .heal_object_at_incarnation(bucket, object, version_id.unwrap_or_default(), expected, opts) + .await + .map_err(|error| incarnation_storage_error(bucket, expected, error))?; + Ok(self + .object_result_with_receipt(bucket, object, version_id, opts, result, Some(expected)) + .await) + } + async fn get_object_meta(&self, bucket: &str, object: &str) -> Result> { debug!( target: "rustfs::heal::storage", @@ -1173,53 +1319,9 @@ impl HealStorageAPI for ECStoreHealStorage { .heal_object_with_proof(bucket, object, version_id.unwrap_or(""), opts) .await .map_err(Error::Storage)?; - let item = result.item; - let error = result.error.map(Error::Storage); - let receipt = if let Some(proof) = result.absence { - if error.is_none() - && !opts.dry_run - && proof.bucket == bucket - && proof.object == object - && proof.version_id == version_id.unwrap_or("") - && proof.pool_index == opts.pool - && proof.set_index == opts.set - && !proof.bucket_incarnation_id.is_nil() - && !proof.locations.is_empty() - && proof.locations.iter().all(|(pool, set)| { - opts.pool.is_none_or(|expected| expected == *pool) && opts.set.is_none_or(|expected| expected == *set) - }) - { - Some(HealObjectReceipt { - identity: HealObjectIdentity { - kind: HealObjectKind::Object, - bucket: proof.bucket, - object: proof.object, - version_id: version_id.map(ToOwned::to_owned), - bucket_incarnation_id: Some(proof.bucket_incarnation_id), - pool_index: proof.pool_index, - set_index: proof.set_index, - }, - // A committed cleanup repaired the stale replica. A replay - // observing an already absent version made no new repair. - disposition: if proof.removed { - HealObjectDisposition::Repaired - } else { - HealObjectDisposition::AuthoritativelyAbsent - }, - }) - } else { - None - } - } else if error.is_none() && !opts.dry_run && item.integrity_verified { - self.ecstore - .bucket_incarnation_id(bucket) - .await - .ok() - .and_then(|incarnation| verified_object_receipt(bucket, object, version_id, opts, &item, incarnation)) - } else { - None - }; - Ok(HealStorageObjectResult { item, error, receipt }) + Ok(self + .object_result_with_receipt(bucket, object, version_id, opts, result, None) + .await) } async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index 8138185be..b5eb6bd06 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/crates/heal/src/heal/storage_api.rs @@ -29,7 +29,7 @@ pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreErrorType, StorageEr pub(crate) use rustfs_ecstore::api::runtime::local_disk_map_read as ecstore_local_disk_map_read; pub(crate) use rustfs_ecstore::api::storage::{ ECStore as EcstoreStore, HealLifecycleExpiryContext as EcstoreHealLifecycleExpiryContext, - POOL_META_NAME as ECSTORE_POOL_META_NAME, + HealObjectStorageResult as EcstoreHealObjectStorageResult, POOL_META_NAME as ECSTORE_POOL_META_NAME, }; use rustfs_storage_api as storage_contracts; @@ -48,6 +48,7 @@ pub(crate) mod owner { } pub(crate) mod storage { + pub(crate) use super::EcstoreHealObjectStorageResult; pub(crate) use super::storage_contracts::{ BucketInfo, BucketOperations, DiskSetSelector, HealOperations, ListOperations, ObjectIO, ObjectOperations, StorageAdminApi, diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index b29b822b5..b66511596 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -310,6 +310,8 @@ pub struct HealRequest { pub id: String, /// Heal type pub heal_type: HealType, + /// Admission identity for an explicit administrator bucket heal. Never rebound on replay. + pub bucket_incarnation_id: Option, /// Heal options pub options: HealOptions, /// Priority @@ -337,6 +339,7 @@ impl HealRequest { Self { id: Uuid::new_v4().to_string(), heal_type, + bucket_incarnation_id: None, options, priority, source: HealRequestSource::Internal, @@ -401,6 +404,7 @@ pub struct HealTask { pub id: String, /// Heal type pub heal_type: HealType, + pub bucket_incarnation_id: Option, /// Heal options pub options: HealOptions, /// Priority inherited from the request @@ -472,6 +476,7 @@ impl HealTask { Self { id: request.id, heal_type: request.heal_type, + bucket_incarnation_id: request.bucket_incarnation_id, options: request.options, priority: request.priority, source: request.source, @@ -502,6 +507,7 @@ impl HealTask { HealRequest { id: self.id.clone(), heal_type: self.heal_type.clone(), + bucket_incarnation_id: self.bucket_incarnation_id, options: self.options.clone(), priority: self.priority, source: self.source, diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index ae80eeea0..dcec1e6b0 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -134,6 +134,10 @@ fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Optio impl HealTask { pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> { self.pace_mainline().await?; + if self.source == HealRequestSource::Admin && matches!(self.heal_type, HealType::Bucket { .. }) { + self.await_with_control(self.storage.validate_bucket_incarnation(bucket, self.bucket_incarnation_id)) + .await?; + } debug!( target: "rustfs::heal::task", event = EVENT_HEAL_BUCKET_STAGE, @@ -216,7 +220,13 @@ impl HealTask { set: self.options.set_index, }; - let heal_result = self.await_with_control(self.storage.heal_bucket(bucket, &heal_opts)).await; + let heal_result = match self.bucket_incarnation_id { + Some(expected) => { + self.await_with_control(self.storage.heal_bucket_at_incarnation(bucket, expected, &heal_opts)) + .await + } + None => self.await_with_control(self.storage.heal_bucket(bucket, &heal_opts)).await, + }; match heal_result { Ok(result) => { @@ -247,6 +257,7 @@ impl HealTask { } Err(Error::TaskCancelled) => Err(Error::TaskCancelled), Err(Error::TaskTimeout) => Err(Error::TaskTimeout), + Err(error @ Error::StaleBucketIncarnation { .. }) => Err(error), Err(e) => { error!( target: "rustfs::heal::task", @@ -482,7 +493,14 @@ impl HealTask { }; for (set_disk_id, heal_opts) in listing_scopes { - let bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, heal_opts.dry_run).await?; + let bucket_incarnation_id = match self.bucket_incarnation_id { + Some(expected) => { + self.await_with_control(self.storage.validate_bucket_incarnation(bucket, Some(expected))) + .await?; + Some(expected) + } + None => self.outcome_bucket_incarnation_id(bucket, heal_opts.dry_run).await?, + }; let mut continuation_token: Option = None; let mut deferred = DeferredWindow::default(); let mut inline_retry: Option = None; @@ -602,12 +620,26 @@ impl HealTask { Some(Error::other("heal object retry age exhausted")) } else { match self - .await_with_control(self.storage.heal_object_with_receipt( - bucket, - object, - item.version_id.as_deref(), - &heal_opts, - )) + .await_with_control(async { + match self.bucket_incarnation_id { + Some(expected) => { + self.storage + .heal_object_at_incarnation( + bucket, + object, + item.version_id.as_deref(), + expected, + &heal_opts, + ) + .await + } + None => { + self.storage + .heal_object_with_receipt(bucket, object, item.version_id.as_deref(), &heal_opts) + .await + } + } + }) .await { Ok(storage_result) if storage_result.error.is_none() => { @@ -654,6 +686,7 @@ impl HealTask { if let Some(err) = error { match err { + Error::StaleBucketIncarnation { .. } => return Err(err), Error::TaskCancelled | Error::TaskTimeout => { let disposition = if matches!(err, Error::TaskCancelled) { HealObjectDisposition::Cancelled diff --git a/crates/heal/tests/heal_b920_subquorum_union_test.rs b/crates/heal/tests/heal_b920_subquorum_union_test.rs index 052de72b1..d18eff21a 100644 --- a/crates/heal/tests/heal_b920_subquorum_union_test.rs +++ b/crates/heal/tests/heal_b920_subquorum_union_test.rs @@ -582,6 +582,7 @@ mod absence_receipt_regressions { use super::*; use rustfs_heal::heal::outcome::{HealDeferredReason, HealObjectDisposition}; use rustfs_heal::heal::{HealOptions, HealPriority, HealRequest, HealTask, HealType}; + use rustfs_heal_contracts::heal_channel::HealRequestSource; use storage_api::integration::{DiskAPI as _, DiskSetSelector, ObjectOperations as _, ReadOptions, StorageAdminApi as _}; const OBJECT: &str = "history.txt"; @@ -717,13 +718,23 @@ mod absence_receipt_regressions { #[test] #[serial] fn historical_absence_receipt_bucket_outcome_matches_c06() { + run_historical_absence_receipt_bucket_outcome(HealRequestSource::Internal); + } + + #[test] + #[serial] + fn historical_absence_receipt_bucket_at_incarnation_matches_c06() { + run_historical_absence_receipt_bucket_outcome(HealRequestSource::Admin); + } + + fn run_historical_absence_receipt_bucket_outcome(source: HealRequestSource) { // The real ECStore initialization and bucket traversal need the debug // server's stack budget, which exceeds libtest's default on Linux. const STACK_SIZE: usize = 8 * 1024 * 1024; std::thread::Builder::new() .name("absence-receipt-c06".to_owned()) .stack_size(STACK_SIZE) - .spawn(|| { + .spawn(move || { let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(4) .thread_stack_size(STACK_SIZE) @@ -731,31 +742,34 @@ mod absence_receipt_regressions { .build() .expect("C06 test runtime should build"); - runtime.block_on(historical_absence_receipt_bucket_outcome_matches_c06_inner()); + runtime.block_on(historical_absence_receipt_bucket_outcome_matches_c06_inner(source)); }) .expect("C06 test thread should spawn") .join() .expect("C06 test thread should finish"); } - async fn historical_absence_receipt_bucket_outcome_matches_c06_inner() { + async fn historical_absence_receipt_bucket_outcome_matches_c06_inner(source: HealRequestSource) { let bucket = "absence-receipt-c06"; let (_paths, store, storage, old, current) = stale_history(bucket).await; put_versioned(&store, bucket, "healthy.txt", b"already healthy").await; - let task = HealTask::from_request( - HealRequest::new( - HealType::Bucket { - bucket: bucket.to_owned(), - }, - HealOptions { - recursive: true, - scan_mode: HealScanMode::Deep, - ..Default::default() - }, - HealPriority::Normal, - ), - storage, + let mut request = HealRequest::new( + HealType::Bucket { + bucket: bucket.to_owned(), + }, + HealOptions { + recursive: true, + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + HealPriority::Normal, ); + request.source = source; + if source == HealRequestSource::Admin { + request.bucket_incarnation_id = Some(storage.admit_bucket_incarnation(bucket).await.expect("admin admission")); + } + let expected = request.bucket_incarnation_id; + let task = HealTask::from_request(request, storage.clone()); with_dangling_grace_disabled(task.execute()) .await .expect("C06 bucket traversal should complete"); @@ -769,6 +783,19 @@ mod absence_receipt_regressions { assert_eq!(outcome.counters.failed, 0); assert_eq!(outcome.counters.skipped, 2); assert_versions(&store, bucket, &old, ¤t).await; + + if let Some(expected) = expected { + let replay = storage + .heal_object_at_incarnation(bucket, OBJECT, Some(&old), expected, &deep_heal_opts()) + .await + .expect("bound replay should retain the exact-version absence proof"); + assert!(replay.error.is_none(), "bound replay failed: {:?}", replay.error); + let receipt = replay.receipt.expect("already absent history needs a receipt"); + assert_eq!(receipt.disposition, HealObjectDisposition::AuthoritativelyAbsent); + assert_eq!(receipt.identity.bucket_incarnation_id, Some(expected)); + assert_eq!(receipt.identity.version_id.as_deref(), Some(old.as_str())); + assert_versions(&store, bucket, &old, ¤t).await; + } } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] diff --git a/crates/heal/tests/shard_identity_receipt_test.rs b/crates/heal/tests/shard_identity_receipt_test.rs index fafa88256..04fd52e6b 100644 --- a/crates/heal/tests/shard_identity_receipt_test.rs +++ b/crates/heal/tests/shard_identity_receipt_test.rs @@ -82,6 +82,24 @@ async fn legacy_repair_reports_execution_without_strong_receipt() { assert_eq!(result.item.drives_healed(), Some(1)); assert!(!result.item.integrity_verified); assert!(result.receipt.is_none(), "physical repair does not prove original identity"); + let storage = ECStoreHealStorage::new(env.ecstore.clone()); + let incarnation = storage.admit_bucket_incarnation(bucket).await.expect("bucket admission"); + let scoped = storage + .heal_object_at_incarnation( + bucket, + object, + None, + incarnation, + &HealOpts { + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + .expect("incarnation-bound legacy scan"); + assert!(scoped.error.is_none(), "{:?}", scoped.error); + assert!(!scoped.item.integrity_verified); + assert!(scoped.receipt.is_none(), "bucket admission cannot certify legacy shard integrity"); let mut reader = env .ecstore .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default()) diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index dd4006391..7f7b9d2b1 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -27,6 +27,8 @@ pub struct HealBucketRequest { pub bucket: ::prost::alloc::string::String, #[prost(string, tag = "2")] pub options: ::prost::alloc::string::String, + #[prost(bytes = "bytes", tag = "3")] + pub bucket_incarnation_id: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct HealBucketResponse { @@ -146,6 +148,8 @@ pub struct DeleteRequest { /// the complete delete operation to its movement read admission. #[prost(bytes = "bytes", tag = "5")] pub scanner_publication_lease_token: ::prost::bytes::Bytes, + #[prost(bytes = "bytes", tag = "6")] + pub bucket_incarnation_id: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteResponse { @@ -407,6 +411,9 @@ pub struct RenameDataRequest { /// rename linearization point. #[prost(bytes = "bytes", tag = "8")] pub scanner_publication_lease_token: ::prost::bytes::Bytes, + /// Required by RenameDataAtIncarnation; never accepted by legacy RenameData. + #[prost(bytes = "bytes", tag = "9")] + pub bucket_incarnation_id: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct RenameDataResponse { @@ -605,6 +612,8 @@ pub struct WriteMetadataRequest { pub file_info: ::prost::alloc::string::String, #[prost(bytes = "bytes", tag = "5")] pub file_info_bin: ::prost::bytes::Bytes, + #[prost(bytes = "bytes", tag = "6")] + pub bucket_incarnation_id: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct WriteMetadataResponse { @@ -701,6 +710,8 @@ pub struct DeleteVersionRequest { pub file_info_bin: ::prost::bytes::Bytes, #[prost(bytes = "bytes", tag = "8")] pub opts_bin: ::prost::bytes::Bytes, + #[prost(bytes = "bytes", tag = "9")] + pub bucket_incarnation_id: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteVersionResponse { @@ -1860,6 +1871,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "HealBucket")); self.inner.unary(req, path, codec).await } + pub async fn heal_bucket_at_incarnation( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/HealBucketAtIncarnation"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "HealBucketAtIncarnation")); + self.inner.unary(req, path, codec).await + } pub async fn list_bucket( &mut self, request: impl tonic::IntoRequest, @@ -1965,6 +1991,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "Delete")); self.inner.unary(req, path, codec).await } + pub async fn delete_at_incarnation( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteAtIncarnation"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "DeleteAtIncarnation")); + self.inner.unary(req, path, codec).await + } pub async fn acquire_snapshot_lease( &mut self, request: impl tonic::IntoRequest, @@ -2206,6 +2247,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "RenameData")); self.inner.unary(req, path, codec).await } + pub async fn rename_data_at_incarnation( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenameDataAtIncarnation"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "RenameDataAtIncarnation")); + self.inner.unary(req, path, codec).await + } pub async fn make_volumes( &mut self, request: impl tonic::IntoRequest, @@ -2326,6 +2382,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "WriteMetadata")); self.inner.unary(req, path, codec).await } + pub async fn write_metadata_at_incarnation( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/WriteMetadataAtIncarnation"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "WriteMetadataAtIncarnation")); + self.inner.unary(req, path, codec).await + } pub async fn read_version( &mut self, request: impl tonic::IntoRequest, @@ -2386,6 +2457,21 @@ pub mod node_service_client { .insert(GrpcMethod::new("node_service.NodeService", "DeleteVersion")); self.inner.unary(req, path, codec).await } + pub async fn delete_version_at_incarnation( + &mut self, + request: impl tonic::IntoRequest, + ) -> std::result::Result, tonic::Status> { + self.inner + .ready() + .await + .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?; + let codec = tonic_prost::ProstCodec::default(); + let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteVersionAtIncarnation"); + let mut req = request.into_request(); + req.extensions_mut() + .insert(GrpcMethod::new("node_service.NodeService", "DeleteVersionAtIncarnation")); + self.inner.unary(req, path, codec).await + } pub async fn delete_retired_marker( &mut self, request: impl tonic::IntoRequest, @@ -3261,6 +3347,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn heal_bucket_at_incarnation( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn list_bucket( &self, request: tonic::Request, @@ -3289,6 +3379,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn delete_at_incarnation( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn acquire_snapshot_lease( &self, request: tonic::Request, @@ -3366,6 +3460,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn rename_data_at_incarnation( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn make_volumes( &self, request: tonic::Request, @@ -3398,6 +3496,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn write_metadata_at_incarnation( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn read_version( &self, request: tonic::Request, @@ -3414,6 +3516,10 @@ pub mod node_service_server { &self, request: tonic::Request, ) -> std::result::Result, tonic::Status>; + async fn delete_version_at_incarnation( + &self, + request: tonic::Request, + ) -> std::result::Result, tonic::Status>; async fn delete_retired_marker( &self, request: tonic::Request, @@ -3771,6 +3877,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/HealBucketAtIncarnation" => { + #[allow(non_camel_case_types)] + struct HealBucketAtIncarnationSvc(pub Arc); + impl tonic::server::UnaryService for HealBucketAtIncarnationSvc { + type Response = super::HealBucketResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::heal_bucket_at_incarnation(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = HealBucketAtIncarnationSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/ListBucket" => { #[allow(non_camel_case_types)] struct ListBucketSvc(pub Arc); @@ -3967,6 +4101,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/DeleteAtIncarnation" => { + #[allow(non_camel_case_types)] + struct DeleteAtIncarnationSvc(pub Arc); + impl tonic::server::UnaryService for DeleteAtIncarnationSvc { + type Response = super::DeleteResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::delete_at_incarnation(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = DeleteAtIncarnationSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/AcquireSnapshotLease" => { #[allow(non_camel_case_types)] struct AcquireSnapshotLeaseSvc(pub Arc); @@ -4418,6 +4580,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/RenameDataAtIncarnation" => { + #[allow(non_camel_case_types)] + struct RenameDataAtIncarnationSvc(pub Arc); + impl tonic::server::UnaryService for RenameDataAtIncarnationSvc { + type Response = super::RenameDataResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::rename_data_at_incarnation(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = RenameDataAtIncarnationSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/MakeVolumes" => { #[allow(non_camel_case_types)] struct MakeVolumesSvc(pub Arc); @@ -4642,6 +4832,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/WriteMetadataAtIncarnation" => { + #[allow(non_camel_case_types)] + struct WriteMetadataAtIncarnationSvc(pub Arc); + impl tonic::server::UnaryService for WriteMetadataAtIncarnationSvc { + type Response = super::WriteMetadataResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::write_metadata_at_incarnation(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = WriteMetadataAtIncarnationSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/ReadVersion" => { #[allow(non_camel_case_types)] struct ReadVersionSvc(pub Arc); @@ -4754,6 +4972,34 @@ pub mod node_service_server { }; Box::pin(fut) } + "/node_service.NodeService/DeleteVersionAtIncarnation" => { + #[allow(non_camel_case_types)] + struct DeleteVersionAtIncarnationSvc(pub Arc); + impl tonic::server::UnaryService for DeleteVersionAtIncarnationSvc { + type Response = super::DeleteVersionResponse; + type Future = BoxFuture, tonic::Status>; + fn call(&mut self, request: tonic::Request) -> Self::Future { + let inner = Arc::clone(&self.0); + let fut = async move { ::delete_version_at_incarnation(&inner, request).await }; + Box::pin(fut) + } + } + let accept_compression_encodings = self.accept_compression_encodings; + let send_compression_encodings = self.send_compression_encodings; + let max_decoding_message_size = self.max_decoding_message_size; + let max_encoding_message_size = self.max_encoding_message_size; + let inner = self.inner.clone(); + let fut = async move { + let method = DeleteVersionAtIncarnationSvc(inner); + let codec = tonic_prost::ProstCodec::default(); + let mut grpc = tonic::server::Grpc::new(codec) + .apply_compression_config(accept_compression_encodings, send_compression_encodings) + .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + let res = grpc.unary(method, req).await; + Ok(res) + }; + Box::pin(fut) + } "/node_service.NodeService/DeleteRetiredMarker" => { #[allow(non_camel_case_types)] struct DeleteRetiredMarkerSvc(pub Arc); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 3b416559b..82d301124 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -817,6 +817,9 @@ impl_canonical_mutation_body!( |request, body| { body.push_str(&request.bucket)?; body.push_str(&request.options)?; + if !request.bucket_incarnation_id.is_empty() { + body.push_bytes(&request.bucket_incarnation_id)?; + } } ); impl_canonical_mutation_body!( @@ -995,6 +998,10 @@ pub fn canonical_rename_data_request_body( if !request.scanner_publication_lease_token.is_empty() { body.push_bytes(&request.scanner_publication_lease_token)?; } + if !request.bucket_incarnation_id.is_empty() { + body.push_str("bucket-incarnation-v1")?; + body.push_bytes(&request.bucket_incarnation_id)?; + } Ok(body.finish()) } @@ -1010,6 +1017,9 @@ pub fn canonical_delete_version_request_body( body.push_str(&request.opts)?; body.push_bytes(&request.file_info_bin)?; body.push_bytes(&request.opts_bin)?; + if !request.bucket_incarnation_id.is_empty() { + body.push_bytes(&request.bucket_incarnation_id)?; + } Ok(body.finish()) } @@ -1041,6 +1051,10 @@ pub fn canonical_write_metadata_request_body( body.push_str(&request.path)?; body.push_str(&request.file_info)?; body.push_bytes(&request.file_info_bin)?; + if !request.bucket_incarnation_id.is_empty() { + body.push_str("bucket-incarnation-v1")?; + body.push_bytes(&request.bucket_incarnation_id)?; + } Ok(body.finish()) } @@ -1080,6 +1094,10 @@ pub fn canonical_delete_request_body( if !request.scanner_publication_lease_token.is_empty() { body.push_bytes(&request.scanner_publication_lease_token)?; } + if !request.bucket_incarnation_id.is_empty() { + body.push_str("bucket-incarnation-v1")?; + body.push_bytes(&request.bucket_incarnation_id)?; + } Ok(body.finish()) } @@ -1240,6 +1258,7 @@ mod disk_mutation_canonical_tests { #[test] fn rename_data_canonical_body_binds_every_field() { let baseline = RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: "disk-a".into(), src_volume: "src-vol".into(), src_path: "src-path".into(), @@ -1260,6 +1279,7 @@ mod disk_mutation_canonical_tests { |r: &mut RenameDataRequest| r.file_info_bin = vec![0x81, 0x02].into(), |r: &mut RenameDataRequest| r.file_info_bin = Vec::new().into(), |r: &mut RenameDataRequest| r.scanner_publication_lease_token = vec![0x01; 16].into(), + |r: &mut RenameDataRequest| r.bucket_incarnation_id = vec![0x01; 16].into(), ] { let mut request = baseline.clone(); mutate(&mut request); @@ -1290,6 +1310,7 @@ mod disk_mutation_canonical_tests { #[test] fn delete_version_canonical_body_binds_every_field() { let baseline = DeleteVersionRequest { + bucket_incarnation_id: Default::default(), disk: "disk-a".into(), volume: "vol".into(), path: "path".into(), @@ -1309,6 +1330,7 @@ mod disk_mutation_canonical_tests { |r: &mut DeleteVersionRequest| r.opts = "{\"o\":1}".into(), |r: &mut DeleteVersionRequest| r.file_info_bin = vec![0x82].into(), |r: &mut DeleteVersionRequest| r.opts_bin = Vec::new().into(), + |r: &mut DeleteVersionRequest| r.bucket_incarnation_id = vec![1; 16].into(), ] { let mut request = baseline.clone(); mutate(&mut request); @@ -1351,6 +1373,7 @@ mod disk_mutation_canonical_tests { // Mutating each field in turn and asserting all bodies differ catches a dropped or // duplicated `push_*` in these hand-written builders — an unbound field is tamperable. let write_metadata = WriteMetadataRequest { + bucket_incarnation_id: Default::default(), disk: "d".into(), volume: "v".into(), path: "p".into(), @@ -1364,6 +1387,7 @@ mod disk_mutation_canonical_tests { |r: &mut WriteMetadataRequest| r.path = "p2".into(), |r: &mut WriteMetadataRequest| r.file_info = "{\"a\":2}".into(), |r: &mut WriteMetadataRequest| r.file_info_bin = vec![0x82].into(), + |r: &mut WriteMetadataRequest| r.bucket_incarnation_id = vec![1; 16].into(), ] { let mut request = write_metadata.clone(); mutate(&mut request); @@ -1416,6 +1440,7 @@ mod disk_mutation_canonical_tests { assert_all_distinct(&bodies); let delete = DeleteRequest { + bucket_incarnation_id: Default::default(), disk: "d".into(), volume: "v".into(), path: "p".into(), @@ -1428,6 +1453,7 @@ mod disk_mutation_canonical_tests { |r: &mut DeleteRequest| r.volume = "v2".into(), |r: &mut DeleteRequest| r.path = "p2".into(), |r: &mut DeleteRequest| r.options = "{\"recursive\":true}".into(), + |r: &mut DeleteRequest| r.bucket_incarnation_id = vec![1; 16].into(), |r: &mut DeleteRequest| r.scanner_publication_lease_token = vec![0x01; 16].into(), ] { let mut request = delete.clone(); @@ -1766,7 +1792,7 @@ mod non_disk_mutation_canonical_tests { #[test] fn bucket_and_lock_canonical_bodies_bind_every_semantic_field() { - assert_fields_bound!(HealBucketRequest, { bucket: "bucket".into(), options: "opts".into() }); + assert_fields_bound!(HealBucketRequest, { bucket: "bucket".into(), options: "opts".into(), bucket_incarnation_id: vec![1; 16].into() }); assert_fields_bound!(MakeBucketRequest, { name: "bucket".into(), options: "opts".into() }); assert_fields_bound!(DeleteBucketRequest, { bucket: "bucket".into(), options: "opts".into() }); assert_fields_bound!(GenerallyLockRequest, { args: "lock".into() }); diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index cc957d0f7..8608c3636 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -48,6 +48,7 @@ message PingResponse { message HealBucketRequest { string bucket = 1; string options = 2; + bytes bucket_incarnation_id = 3; } message HealBucketResponse { @@ -128,6 +129,7 @@ message DeleteRequest { // Optional scanner publication lease token. When present, the target binds // the complete delete operation to its movement read admission. bytes scanner_publication_lease_token = 5; + bytes bucket_incarnation_id = 6; } message DeleteResponse { @@ -305,6 +307,8 @@ message RenameDataRequest { // legacy rename request body; a non-empty token is checked at the target's // rename linearization point. bytes scanner_publication_lease_token = 8; + // Required by RenameDataAtIncarnation; never accepted by legacy RenameData. + bytes bucket_incarnation_id = 9; } message RenameDataResponse { @@ -433,6 +437,7 @@ message WriteMetadataRequest { string path = 3; string file_info = 4; bytes file_info_bin = 5; + bytes bucket_incarnation_id = 6; } message WriteMetadataResponse { @@ -494,6 +499,7 @@ message DeleteVersionRequest { // both; receivers prefer the *_bin form and fall back to the JSON string when it is empty. bytes file_info_bin = 7; bytes opts_bin = 8; + bytes bucket_incarnation_id = 9; } message DeleteVersionResponse { @@ -1174,6 +1180,7 @@ service NodeService { /* -------------------------------meta service-------------------------- */ rpc Ping(PingRequest) returns (PingResponse) {}; // auth-policy: read-only rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {}; // auth-policy: body-bound + rpc HealBucketAtIncarnation(HealBucketRequest) returns (HealBucketResponse) {}; // auth-policy: body-bound rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {}; // auth-policy: read-only rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {}; // auth-policy: body-bound rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {}; // auth-policy: read-only @@ -1184,6 +1191,7 @@ service NodeService { rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound + rpc DeleteAtIncarnation(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {}; // auth-policy: body-bound @@ -1201,6 +1209,7 @@ service NodeService { rpc ListDir(ListDirRequest) returns (ListDirResponse) {}; // auth-policy: read-only rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {}; // auth-policy: streaming rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {}; // auth-policy: body-bound + rpc RenameDataAtIncarnation(RenameDataRequest) returns (RenameDataResponse) {}; // auth-policy: body-bound rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {}; // auth-policy: body-bound rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {}; // auth-policy: body-bound rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}; // auth-policy: read-only @@ -1209,10 +1218,12 @@ service NodeService { rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}; // auth-policy: body-bound rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {}; // auth-policy: read-only rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {}; // auth-policy: body-bound + rpc WriteMetadataAtIncarnation(WriteMetadataRequest) returns (WriteMetadataResponse) {}; // auth-policy: body-bound rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {}; // auth-policy: read-only rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {}; // auth-policy: read-only rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {}; // auth-policy: read-only rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound + rpc DeleteVersionAtIncarnation(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound rpc DeleteRetiredMarker(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {}; // auth-policy: body-bound rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {}; // auth-policy: read-only diff --git a/docs/README.md b/docs/README.md index af6c50329..2337461c6 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,6 +18,9 @@ operators should start with: | [Replication object size limits](operations/replication-object-size-limits.md) | Multipart routing, large-object limits, and retry characteristics. | | [Replication outbound transport](operations/replication-outbound-transport.md) | Integrity headers, generic target behavior, and transport knobs. | +For persisted administrator bucket tasks and bucket recreation, see +[Bucket heal recovery](operations/bucket-heal-recovery.md). + Other runbooks remain grouped by filename in [`operations/`](operations/); architecture pages link to the relevant runbook where a cross-boundary procedure is required. diff --git a/docs/operations/bucket-heal-recovery.md b/docs/operations/bucket-heal-recovery.md new file mode 100644 index 000000000..e6cd9bcea --- /dev/null +++ b/docs/operations/bucket-heal-recovery.md @@ -0,0 +1,52 @@ +# Bucket heal recovery and bucket recreation + +Explicit administrator bucket heals bind the bucket name to its authoritative, +non-nil incarnation UUID at admission. The same identity follows the task through +queueing, retries, shutdown persistence, and startup replay. Deleting a bucket and +creating another with the same name does not transfer the old task to the new bucket. + +## Recovery records and stale tokens + +Pending records use schema 3 in `.rustfs.sys/root-heal-.json` on the existing +coordinator disk. A bucket record includes `bucket_incarnation_id`. Options, task ID, +and the remaining retry budget retain their existing recovery semantics. + +A missing bucket, changed incarnation, or legacy bucket record without a usable +identity becomes a durable `Failed` terminal result containing +`stale_bucket_incarnation`. The original token remains queryable under the existing +terminal retention policy. Recovery never fills a legacy task with the identity of +the bucket currently using its name. Existing schema 1 cluster records and schema 2 +non-bucket records keep their recovery behavior. Unsupported or corrupt records +continue to defer recovery; transient metadata failures do not retire a valid task. + +If a task is stale, submit a new bucket heal for the current bucket and use the new +token. Do not edit recovery files to replace the incarnation. A retained terminal +record takes precedence over an old pending copy of the same task. + +## Execution and upgrades + +Each bucket metadata or object repair acquires the existing bucket lifecycle read +fence and validates the admitted identity. The guard remains held through the +storage operation and its mutation tail. Cancellation stops waiting for the result; +an already admitted physical mutation retains its owner until it drains. Bucket +deletion and recreation must use the normal lifecycle write path. + +Remote bucket repair, shard rename, metadata regeneration, version removal, and +path cleanup use dedicated `AtIncarnation` RPC methods. Their canonical body digest +binds the incarnation. A receiver validates it against its own storage instance; +missing or nil identities and unsigned bodies are rejected. Older nodes return +`Unimplemented`; senders do not retry through an unfenced method. Upgrade all +participating nodes before expecting these bucket heals to complete. Unscoped +RPC bodies retain their prior canonical encoding. Older coordinators do not +understand schema 3 pending records. Finish or cancel pending administrator tasks +before downgrading a coordinator; do not rewrite the schema to force replay. + +The lifecycle fence remains mandatory even when `nolock` disables the optional +object namespace lock. Deployments with namespace locking disabled cannot perform +incarnation-bound bucket healing. + +This contract covers explicit administrator **bucket** tasks. Cluster, prefix, +object, scanner, and replacement tasks retain their existing admission semantics. +It does not introduce a new object-generation protocol or claim rejection of every +already-dispatched native syscall after distributed lock loss; see the existing +[heal concurrency model](../architecture/heal-concurrency-model.md). diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 8b7300f6c..5ac8a306d 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -127,6 +127,19 @@ fn verify_node_mutation_body(request: &Request, ope .map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}"))) } +fn require_incarnation_body_digest(request: &Request) -> Result<(), Status> { + if request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| !value.is_empty() && value != "UNSIGNED-PAYLOAD") + { + Ok(()) + } else { + Err(Status::permission_denied("incarnation-bound mutation requires a body-bound digest")) + } +} + fn verify_node_signal_body(request: &Request, operation: &'static str) -> Result<(), Status> { let canonical_body = request .get_ref() @@ -1381,6 +1394,25 @@ impl Node for NodeService { async fn heal_bucket(&self, request: Request) -> Result, Status> { verify_node_mutation_body(&request, "heal bucket")?; + if !request.get_ref().bucket_incarnation_id.is_empty() { + return Err(Status::invalid_argument("incarnation-bound heal requires HealBucketAtIncarnation")); + } + self.handle_heal_bucket(request).await + } + + async fn heal_bucket_at_incarnation( + &self, + request: Request, + ) -> Result, Status> { + require_incarnation_body_digest(&request)?; + verify_node_signal_body(&request, "heal bucket at incarnation")?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_heal_bucket(request).await } @@ -1411,6 +1443,21 @@ impl Node for NodeService { } async fn delete(&self, request: Request) -> Result, Status> { + if !request.get_ref().bucket_incarnation_id.is_empty() { + return Err(Status::invalid_argument("incarnation-bound mutation requires its dedicated RPC")); + } + self.handle_delete(request).await + } + + async fn delete_at_incarnation(&self, request: Request) -> Result, Status> { + require_incarnation_body_digest(&request)?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_delete(request).await } @@ -1600,6 +1647,24 @@ impl Node for NodeService { } async fn rename_data(&self, request: Request) -> Result, Status> { + if !request.get_ref().bucket_incarnation_id.is_empty() { + return Err(Status::invalid_argument("incarnation-bound rename requires RenameDataAtIncarnation")); + } + self.handle_rename_data(request).await + } + + async fn rename_data_at_incarnation( + &self, + request: Request, + ) -> Result, Status> { + require_incarnation_body_digest(&request)?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_rename_data(request).await } @@ -1649,6 +1714,24 @@ impl Node for NodeService { } async fn write_metadata(&self, request: Request) -> Result, Status> { + if !request.get_ref().bucket_incarnation_id.is_empty() { + return Err(Status::invalid_argument("incarnation-bound mutation requires its dedicated RPC")); + } + self.handle_write_metadata(request).await + } + + async fn write_metadata_at_incarnation( + &self, + request: Request, + ) -> Result, Status> { + require_incarnation_body_digest(&request)?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_write_metadata(request).await } @@ -1668,6 +1751,24 @@ impl Node for NodeService { } async fn delete_version(&self, request: Request) -> Result, Status> { + if !request.get_ref().bucket_incarnation_id.is_empty() { + return Err(Status::invalid_argument("incarnation-bound delete requires DeleteVersionAtIncarnation")); + } + self.handle_delete_version(request, false).await + } + + async fn delete_version_at_incarnation( + &self, + request: Request, + ) -> Result, Status> { + require_incarnation_body_digest(&request)?; + if Uuid::from_slice(&request.get_ref().bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .is_none() + { + return Err(Status::invalid_argument("bucket incarnation must be a non-nil UUID")); + } self.handle_delete_version(request, false).await } @@ -2922,8 +3023,12 @@ mod tests { use tonic::{Request, Response, Status}; use uuid::Uuid; - const DISK_MUTATION_RPC_METHODS: [&str; 19] = [ + const DISK_MUTATION_RPC_METHODS: [&str; 23] = [ "renamedata", + "renamedataatincarnation", + "writemetadataatincarnation", + "deleteatincarnation", + "deleteversionatincarnation", "deleteversion", "deleteretiredmarker", "deleteversions", @@ -4126,6 +4231,7 @@ mod tests { fn delete_request_message(options: &str) -> DeleteRequest { DeleteRequest { + bucket_incarnation_id: Default::default(), disk: "http://node-a:9000/data/rustfs0".to_string(), volume: "bucket".to_string(), path: "object".to_string(), @@ -4258,6 +4364,7 @@ mod tests { assert_gated!( rename_data, RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: disk.clone(), src_volume: "src".into(), src_path: "sp".into(), @@ -4272,6 +4379,7 @@ mod tests { assert_gated!( delete_version, DeleteVersionRequest { + bucket_incarnation_id: Default::default(), disk: disk.clone(), volume: "v".into(), path: "p".into(), @@ -4283,9 +4391,26 @@ mod tests { }, rustfs_protos::canonical_delete_version_request_body ); + assert_gated!( + rename_data_at_incarnation, + RenameDataRequest { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + }, + rustfs_protos::canonical_rename_data_request_body + ); + assert_gated!( + delete_version_at_incarnation, + DeleteVersionRequest { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + }, + rustfs_protos::canonical_delete_version_request_body + ); assert_gated!( delete_retired_marker, DeleteVersionRequest { + bucket_incarnation_id: Default::default(), disk: disk.clone(), volume: "v".into(), path: "p".into(), @@ -4297,6 +4422,22 @@ mod tests { }, rustfs_protos::canonical_delete_version_request_body ); + assert_gated!( + write_metadata_at_incarnation, + WriteMetadataRequest { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + }, + rustfs_protos::canonical_write_metadata_request_body + ); + assert_gated!( + delete_at_incarnation, + DeleteRequest { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + }, + rustfs_protos::canonical_delete_request_body + ); assert_gated!( delete_versions, DeleteVersionsRequest { @@ -4312,6 +4453,7 @@ mod tests { assert_gated!( write_metadata, WriteMetadataRequest { + bucket_incarnation_id: Default::default(), disk: disk.clone(), volume: "v".into(), path: "p".into(), @@ -4346,6 +4488,7 @@ mod tests { assert_gated!( delete, DeleteRequest { + bucket_incarnation_id: Default::default(), disk: disk.clone(), volume: "v".into(), path: "p".into(), @@ -5271,11 +5414,107 @@ mod tests { assert!(!ping_response.body.is_empty()); } + #[tokio::test] + async fn bucket_incarnation_rpcs_require_identity_digest_and_dedicated_method() { + let service = create_test_node_service(); + macro_rules! assert_fenced { + ($fenced:ident, $legacy:ident, $message:ident, $canonical:path) => {{ + let message = $message { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + }; + for unsigned in [false, true] { + let mut request = Request::new(message.clone()); + if unsigned { + request + .metadata_mut() + .insert("x-rustfs-content-sha256", "UNSIGNED-PAYLOAD".parse().unwrap()); + } + mark_v2_authenticated(&mut request); + assert_eq!( + service + .$fenced(request) + .await + .expect_err("digest is mandatory even for authenticated peers") + .code(), + tonic::Code::PermissionDenied + ); + } + let mut legacy = Request::new(message.clone()); + let body = $canonical(legacy.get_ref()).unwrap(); + set_tonic_canonical_body_digest(&mut legacy, &body).unwrap(); + mark_v2_authenticated(&mut legacy); + assert_eq!( + service + .$legacy(legacy) + .await + .expect_err("cannot downgrade a fenced mutation") + .code(), + tonic::Code::InvalidArgument + ); + for invalid in [Vec::new(), vec![0; 16], vec![1; 15]] { + let mut request = Request::new(message.clone()); + request.get_mut().bucket_incarnation_id = invalid.into(); + let body = $canonical(request.get_ref()).unwrap(); + set_tonic_canonical_body_digest(&mut request, &body).unwrap(); + mark_v2_authenticated(&mut request); + assert_eq!( + service + .$fenced(request) + .await + .expect_err("missing, nil and malformed identities fail closed") + .code(), + tonic::Code::InvalidArgument + ); + } + let mut tampered = Request::new(message.clone()); + let body = $canonical(tampered.get_ref()).unwrap(); + set_tonic_canonical_body_digest(&mut tampered, &body).unwrap(); + tampered.get_mut().bucket_incarnation_id = vec![2; 16].into(); + mark_v2_authenticated(&mut tampered); + assert_eq!( + service + .$fenced(tampered) + .await + .expect_err("incarnation is authenticated") + .code(), + tonic::Code::PermissionDenied + ); + }}; + } + assert_fenced!( + write_metadata_at_incarnation, + write_metadata, + WriteMetadataRequest, + rustfs_protos::canonical_write_metadata_request_body + ); + assert_fenced!(delete_at_incarnation, delete, DeleteRequest, rustfs_protos::canonical_delete_request_body); + assert_fenced!( + heal_bucket_at_incarnation, + heal_bucket, + HealBucketRequest, + rustfs_protos::CanonicalMutationBody::canonical_body + ); + assert_fenced!( + rename_data_at_incarnation, + rename_data, + RenameDataRequest, + rustfs_protos::canonical_rename_data_request_body + ); + assert_fenced!( + delete_version_at_incarnation, + delete_version, + DeleteVersionRequest, + rustfs_protos::canonical_delete_version_request_body + ); + } + #[tokio::test] async fn test_heal_bucket_invalid_options() { let service = create_test_node_service(); let request = Request::new(HealBucketRequest { + bucket_incarnation_id: Default::default(), bucket: "test-bucket".to_string(), options: "invalid json".to_string(), }); @@ -5418,6 +5657,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(DeleteRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), volume: "test-volume".to_string(), path: "test-path".to_string(), @@ -5438,6 +5678,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(DeleteRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), volume: "test-volume".to_string(), path: "test-path".to_string(), @@ -5611,6 +5852,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), src_volume: "src-volume".to_string(), src_path: "src-path".to_string(), @@ -5634,6 +5876,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), src_volume: "src-volume".to_string(), src_path: "src-path".to_string(), @@ -6276,6 +6519,7 @@ mod tests { ); let mut request = Request::new(RenameDataRequest { + bucket_incarnation_id: Default::default(), disk: disk_id.to_string(), src_volume: volume.to_string(), src_path: staging.to_string(), @@ -6491,6 +6735,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(WriteMetadataRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), volume: "test-volume".to_string(), path: "test-path".to_string(), @@ -6511,6 +6756,7 @@ mod tests { let service = create_test_node_service(); let request = Request::new(WriteMetadataRequest { + bucket_incarnation_id: Default::default(), disk: "invalid-disk-path".to_string(), volume: "test-volume".to_string(), path: "test-path".to_string(), @@ -6628,6 +6874,60 @@ mod tests { } } + #[tokio::test] + async fn retired_marker_rpc_binds_bucket_incarnation() { + use crate::storage::storage_api::ecstore_disk::DeleteOptions; + use rustfs_filemeta::{FileInfo, MetaDeleteMarker}; + + let service = create_test_node_service(); + let mut marker = FileInfo { + deleted: true, + version_id: Some(Uuid::new_v4()), + mod_time: Some(time::OffsetDateTime::now_utc()), + ..Default::default() + }; + marker.set_delete_marker_incarnation(Uuid::new_v4()); + let message = DeleteVersionRequest { + bucket_incarnation_id: vec![1; 16].into(), + volume: "bucket".into(), + path: "marker.bin".into(), + file_info: serde_json::to_string(&FileInfo::default()).expect("file info"), + opts: serde_json::to_string(&DeleteOptions { + expected_delete_marker: Some(MetaDeleteMarker::from(marker)), + ..Default::default() + }) + .expect("marker precondition"), + ..Default::default() + }; + for (identity, digest_identity, expected) in [ + (vec![1; 16], None, tonic::Code::PermissionDenied), + (vec![0; 16], Some(vec![0; 16]), tonic::Code::InvalidArgument), + (vec![1; 15], Some(vec![1; 15]), tonic::Code::InvalidArgument), + (vec![2; 16], Some(vec![1; 16]), tonic::Code::PermissionDenied), + (Vec::new(), Some(vec![1; 16]), tonic::Code::PermissionDenied), + (vec![1; 16], Some(vec![1; 16]), tonic::Code::FailedPrecondition), + ] { + let mut request = Request::new(message.clone()); + if let Some(digest_identity) = digest_identity { + request.get_mut().bucket_incarnation_id = digest_identity.into(); + let body = rustfs_protos::canonical_delete_version_request_body(request.get_ref()).expect("canonical body"); + set_tonic_canonical_body_digest(&mut request, &body).expect("body digest"); + } + // Removing the wire field models a peer predating incarnation support: + // its canonical body must reject the new sender's signed identity. + request.get_mut().bucket_incarnation_id = identity.into(); + mark_v2_authenticated(&mut request); + assert_eq!( + service + .delete_retired_marker(request) + .await + .expect_err("incarnation and marker checks must precede disk mutation") + .code(), + expected + ); + } + } + #[tokio::test] async fn test_delete_version_invalid_disk() { let service = create_test_node_service(); @@ -7829,6 +8129,13 @@ mod tests { } assert_tampered!(heal_bucket, HealBucketRequest::default()); + assert_tampered!( + heal_bucket_at_incarnation, + HealBucketRequest { + bucket_incarnation_id: vec![1; 16].into(), + ..Default::default() + } + ); assert_tampered!(make_bucket, MakeBucketRequest::default()); assert_tampered!(delete_bucket, DeleteBucketRequest::default()); assert_tampered!(lock, GenerallyLockRequest::default()); diff --git a/rustfs/src/storage/rpc/node_service/bucket.rs b/rustfs/src/storage/rpc/node_service/bucket.rs index 60d69c119..e7f8f82ae 100644 --- a/rustfs/src/storage/rpc/node_service/bucket.rs +++ b/rustfs/src/storage/rpc/node_service/bucket.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::NodeService; +use super::{LocalMutationTarget, NodeService}; use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{ BucketOptions, DeleteBucketOptions, MakeBucketOptions, }; @@ -22,6 +22,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::{ use rustfs_protos::proto_gen::node_service::*; use tonic::{Request, Response, Status}; use tracing::debug; +use uuid::Uuid; impl NodeService { pub(super) async fn handle_delete_bucket_metadata( @@ -252,11 +253,30 @@ impl NodeService { } }; - match self - .local_peer - .heal_bucket_with_fence(&request.bucket, &options, &fenced_pools) - .await - { + let result = if request.bucket_incarnation_id.is_empty() { + self.local_peer + .heal_bucket_with_fence(&request.bucket, &options, &fenced_pools) + .await + } else { + let expected = Uuid::from_slice(&request.bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .ok_or_else(|| Status::invalid_argument("bucket incarnation must be a non-nil UUID"))?; + let LocalMutationTarget::Ready(store) = self.local_mutation_target() else { + return Err(Status::failed_precondition("bucket heal requires a ready storage instance")); + }; + store + .heal_local_bucket_at_incarnation( + &request.bucket, + expected, + &options, + fenced_pools, + self.local_peer.pools.clone(), + ) + .await + .map_err(|error| DiskError::other(error.to_string())) + }; + match result { Ok(_) => Ok(Response::new(HealBucketResponse { success: true, error: None, diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 4f40a208a..11d86cc90 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -70,14 +70,23 @@ impl LocalMutationTarget { fi: &FileInfo, destination: (&str, &str), scanner_token: Option, + bucket_incarnation: Option, ) -> Result { match self { Self::Ready(store) => { + if let Some(expected) = bucket_incarnation { + return store + .rename_local_data_at_incarnation(disk_ref, source, fi, destination, expected) + .await; + } store .rename_local_data(disk_ref, source, fi, destination, scanner_token) .await } Self::Bootstrap(target) => { + if bucket_incarnation.is_some() { + return Err(DiskError::other("incarnation-bound rename requires a ready storage instance")); + } target .rename_local_data(disk_ref, source, fi, destination, scanner_token) .await @@ -728,6 +737,15 @@ impl NodeService { request: Request, require_marker_condition: bool, ) -> Result, Status> { + if !request.get_ref().bucket_incarnation_id.is_empty() + && !request + .metadata() + .get("x-rustfs-content-sha256") + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value != "UNSIGNED-PAYLOAD") + { + return Err(Status::permission_denied("incarnation-bound delete requires a body-bound digest")); + } verify_disk_mutation_digest( &request, rustfs_protos::canonical_delete_version_request_body(request.get_ref()), @@ -769,7 +787,25 @@ impl NodeService { "retired marker preconditions cannot be combined with other mutations", )); } - let result = if opts.undo_write { + let result = if !request.bucket_incarnation_id.is_empty() { + let expected = Uuid::from_slice(&request.bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .ok_or_else(|| Status::invalid_argument("bucket incarnation must be a non-nil UUID"))?; + let LocalMutationTarget::Ready(store) = self.local_mutation_target() else { + return Err(Status::failed_precondition("bucket heal requires a ready storage instance")); + }; + store + .delete_local_version_at_incarnation( + &request.disk, + (&request.volume, &request.path), + file_info, + request.force_del_marker, + opts, + expected, + ) + .await + } else if opts.undo_write { if request.force_del_marker { Err(DiskError::other("undo_write cannot force a delete marker")) } else { @@ -1000,6 +1036,24 @@ impl NodeService { "write_metadata", )?; let request = request.into_inner(); + if !request.bucket_incarnation_id.is_empty() { + let expected = Uuid::from_slice(&request.bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .ok_or_else(|| Status::invalid_argument("bucket incarnation must be a non-nil UUID"))?; + let LocalMutationTarget::Ready(store) = self.local_mutation_target() else { + return Err(Status::failed_precondition("bucket heal requires a ready storage instance")); + }; + let value = decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") + .map_err(|error| Status::invalid_argument(error.to_string()))?; + let result = store + .write_local_metadata_at_incarnation(&request.disk, (&request.volume, &request.path), value, expected) + .await; + return Ok(Response::new(WriteMetadataResponse { + success: result.is_ok(), + error: result.err().map(Into::into), + })); + } if let Some(disk) = self.find_disk(&request.disk).await { let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info, @@ -1273,7 +1327,7 @@ impl NodeService { &self, request: Request, ) -> Result, Status> { - if !request.get_ref().scanner_publication_lease_token.is_empty() { + if !request.get_ref().scanner_publication_lease_token.is_empty() || !request.get_ref().bucket_incarnation_id.is_empty() { let has_body_digest = request .metadata() .get("x-rustfs-content-sha256") @@ -1290,6 +1344,20 @@ impl NodeService { )?; let request = request.into_inner(); let target = self.local_mutation_target(); + let bucket_incarnation = if request.bucket_incarnation_id.is_empty() { + None + } else { + let expected = Uuid::from_slice(&request.bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .ok_or_else(|| Status::invalid_argument("bucket incarnation must be a non-nil UUID"))?; + if !request.scanner_publication_lease_token.is_empty() { + return Err(Status::invalid_argument( + "heal incarnation and scanner publication lease cannot be combined", + )); + } + Some(expected) + }; #[cfg(feature = "e2e-test-hooks")] super::rename_target_capture_test_hook::wait(&target, &request).await; let decoded_file_info = match decode_rename_data_request_file_info(&request.file_info_bin, &request.file_info) { @@ -1323,6 +1391,7 @@ impl NodeService { &decoded_file_info.value, (&request.dst_volume, &request.dst_path), scanner_publication_lease_token, + bucket_incarnation, ) .await; #[cfg(feature = "e2e-test-hooks")] @@ -1684,6 +1753,27 @@ impl NodeService { } verify_disk_mutation_digest(&request, rustfs_protos::canonical_delete_request_body(request.get_ref()), "delete")?; let request = request.into_inner(); + if !request.bucket_incarnation_id.is_empty() { + if !request.scanner_publication_lease_token.is_empty() { + return Err(Status::invalid_argument("heal incarnation and scanner lease cannot be combined")); + } + let expected = Uuid::from_slice(&request.bucket_incarnation_id) + .ok() + .filter(|id| !id.is_nil()) + .ok_or_else(|| Status::invalid_argument("bucket incarnation must be a non-nil UUID"))?; + let LocalMutationTarget::Ready(store) = self.local_mutation_target() else { + return Err(Status::failed_precondition("bucket heal requires a ready storage instance")); + }; + let value = serde_json::from_str::(&request.options) + .map_err(|error| Status::invalid_argument(error.to_string()))?; + let result = store + .delete_local_path_at_incarnation(&request.disk, (&request.volume, &request.path), value, expected) + .await; + return Ok(Response::new(DeleteResponse { + success: result.is_ok(), + error: result.err().map(Into::into), + })); + } if let Some(disk) = self.find_disk(&request.disk).await { let options = match serde_json::from_str::(&request.options) { Ok(options) => options,