gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-21 18:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): bind bucket recovery to its original incarnation (#7744)

Browse Source 
* fix(heal): bind bucket recovery to its original incarnation

* fix(heal): preserve stable bucket heal task errors
This commit is contained in:
cxymds
2026-09-13 22:00:28 +08:00
committed by GitHub
parent 3ca3e26cec
commit 3e156ee61e
31 changed files with 1978 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/e2e_test/src/reliant/grpc_lock_server.rs
+35
View File
@@ -360,6 +360,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn heal_bucket_at_incarnation(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::HealBucketRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::HealBucketResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn list_bucket(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::ListBucketRequest>,
@@ -409,6 +416,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn delete_at_incarnation(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::DeleteRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::DeleteResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn verify_file(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::VerifyFileRequest>,
@@ -506,6 +520,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn rename_data_at_incarnation(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::RenameDataRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::RenameDataResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn make_volumes(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::MakeVolumesRequest>,
@@ -562,6 +583,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn write_metadata_at_incarnation(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::WriteMetadataRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::WriteMetadataResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn read_version(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::ReadVersionRequest>,
@@ -583,6 +611,13 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("lock-only test server"))
}
async fn delete_version_at_incarnation(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::DeleteVersionRequest>,
) -> Result<Response<rustfs_protos::proto_gen::node_service::DeleteVersionResponse>, Status> {
Err(Status::unimplemented("lock-only test server"))
}
async fn delete_retired_marker(
&self,
_request: Request<rustfs_protos::proto_gen::node_service::DeleteVersionRequest>,
crates/ecstore/src/cluster/rpc/http_auth.rs
+1
View File
@@ -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(),
crates/ecstore/src/cluster/rpc/peer_s3_client.rs
+22 -2
View File
@@ -741,7 +741,10 @@ impl LocalPeerS3Client {
movement_guard_held: bool,
) -> Result<HealResultItem> {
let disks = self.local_disks_for_pools().await.into_iter().map(Some).collect();
let store = runtime_sources::object_store_handle().filter(|store| Arc::ptr_eq(&store.ctx, &self.instance_ctx));
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<PoolMeta>>,
dispatch_fenced_pools: &[usize],
) -> Result<HealResultItem> {
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),
crates/ecstore/src/cluster/rpc/remote_disk.rs
+48 -10
View File
@@ -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());
crates/ecstore/src/disk/disk_store.rs
+27 -3
View File
@@ -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<DataDirDeleteStatus> {
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
}
crates/ecstore/src/disk/mod.rs
+21 -1
View File
@@ -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<RenameDataResp> {
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
}
crates/ecstore/src/set_disk/ops/heal.rs
+8
View File
@@ -492,6 +492,7 @@ impl SetDisks {
targets: &[String],
) -> disk::error::Result<bool> {
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;
crates/ecstore/src/store/heal.rs
+400 -13
View File
@@ -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<ECStore>,
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<BucketHealScope>;
}
pub(crate) fn bucket_heal_scope(bucket: &str) -> Option<Arc<BucketHealScope>> {
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<T, F, Fut>(
self: &Arc<Self>,
bucket: &str,
expected: uuid::Uuid,
opts: &HealOpts,
operation: F,
) -> Result<T>
where
T: Send + 'static,
F: FnOnce(Arc<Self>, String, HealOpts) -> Fut + Send + 'static,
Fut: std::future::Future<Output = Result<T>> + 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<Self>,
bucket: &str,
expected: uuid::Uuid,
opts: &HealOpts,
) -> Result<HealResultItem> {
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<Self>,
bucket: &str,
expected: uuid::Uuid,
opts: &HealOpts,
fenced_pools: Vec<usize>,
pools: Option<Vec<usize>>,
) -> Result<HealResultItem> {
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<Self>,
bucket: &str,
object: &str,
version_id: &str,
expected: uuid::Uuid,
opts: &HealOpts,
) -> Result<HealObjectStorageResult> {
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(),
crates/ecstore/src/store/mod.rs
+1
View File
@@ -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;
crates/ecstore/src/store/object.rs
+1 -1
View File
@@ -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);
crates/ecstore/src/store/peer.rs
+103
View File
@@ -80,6 +80,109 @@ fn validate_bootstrap_volume(volume: &str) -> DiskResult<()> {
}
impl ECStore {
pub async fn write_local_metadata_at_incarnation(
self: &Arc<Self>,
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<Self>,
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<RenameDataResp> {
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,
crates/heal/src/error.rs
+4 -1
View File
@@ -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<uuid::Uuid> },
/// 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 /
crates/heal/src/heal/manager.rs
+10 -1
View File
@@ -521,6 +521,7 @@ fn active_heal_for_dedup_key(active_heals: &HashMap<String, Arc<HealTask>>, 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<MrfRepairNoticeTarget>,
@@ -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);
crates/heal/src/heal/manager/root_recovery.rs
+43 -4
View File
@@ -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<Uuid>,
#[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<RootHealIntent> {
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;
crates/heal/src/heal/manager/tests.rs
+18
View File
@@ -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<Uuid> {
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<HealResultItem> {
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<Option<HealObjectInfo>> {
Ok(None)
}
crates/heal/src/heal/manager/tests/root_recovery.rs
+189 -1
View File
@@ -17,6 +17,20 @@ use super::*;
use crate::heal::RUSTFS_META_BUCKET;
use std::collections::HashSet;
fn bucket_incarnations() -> &'static std::sync::Mutex<HashMap<String, Option<Uuid>>> {
static IDS: std::sync::OnceLock<std::sync::Mutex<HashMap<String, Option<Uuid>>>> = std::sync::OnceLock::new();
IDS.get_or_init(Default::default)
}
pub(super) fn test_bucket_incarnation(bucket: &str) -> Option<Uuid> {
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());
}
crates/heal/src/heal/manager/tests/running_mainline.rs
+21
View File
@@ -47,6 +47,27 @@ struct RunningStorage {
#[async_trait::async_trait]
impl HealStorageAPI for RunningStorage {
async fn admit_bucket_incarnation(&self, _: &str) -> Result<Uuid> {
Ok(Uuid::from_u128(42))
}
async fn heal_bucket_at_incarnation(&self, bucket: &str, expected: Uuid, opts: &HealOpts) -> Result<HealResultItem> {
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<crate::heal::storage::HealStorageObjectResult> {
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<Option<HealObjectInfo>> {
Ok(None)
}
crates/heal/src/heal/storage.rs
+151 -49
View File
@@ -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<HealObjectReceipt>,
}
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<Error>)> for HealStorageObjectResult {
fn from((item, error): (HealResultItem, Option<Error>)) -> 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<Uuid> {
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<Uuid>) -> 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<HealResultItem> {
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<HealStorageObjectResult> {
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<Uuid>,
) -> 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<Uuid> {
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<HealResultItem> {
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<HealStorageObjectResult> {
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<Option<HealObjectInfo>> {
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<HealResultItem> {
crates/heal/src/heal/storage_api.rs
+2 -1
View File
@@ -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,
crates/heal/src/heal/task.rs
+6
View File
@@ -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<Uuid>,
/// 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<Uuid>,
/// 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,
crates/heal/src/heal/task/heal_bucket.rs
+41 -8
View File
@@ -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<String> = None;
let mut deferred = DeferredWindow::default();
let mut inline_retry: Option<DeferredObject> = 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
crates/heal/tests/heal_b920_subquorum_union_test.rs
+43 -16
View File
@@ -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, &current).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, &current).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
crates/heal/tests/shard_identity_receipt_test.rs
+18
View File
@@ -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())
crates/protos/src/generated/proto_gen/node_service.rs
+246
View File
@@ -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<super::HealBucketRequest>,
) -> std::result::Result<tonic::Response<super::HealBucketResponse>, 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<super::ListBucketRequest>,
@@ -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<super::DeleteRequest>,
) -> std::result::Result<tonic::Response<super::DeleteResponse>, 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<super::SnapshotLeaseRequest>,
@@ -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<super::RenameDataRequest>,
) -> std::result::Result<tonic::Response<super::RenameDataResponse>, 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<super::MakeVolumesRequest>,
@@ -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<super::WriteMetadataRequest>,
) -> std::result::Result<tonic::Response<super::WriteMetadataResponse>, 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<super::ReadVersionRequest>,
@@ -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<super::DeleteVersionRequest>,
) -> std::result::Result<tonic::Response<super::DeleteVersionResponse>, 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<super::DeleteVersionRequest>,
@@ -3261,6 +3347,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::HealBucketRequest>,
) -> std::result::Result<tonic::Response<super::HealBucketResponse>, tonic::Status>;
async fn heal_bucket_at_incarnation(
&self,
request: tonic::Request<super::HealBucketRequest>,
) -> std::result::Result<tonic::Response<super::HealBucketResponse>, tonic::Status>;
async fn list_bucket(
&self,
request: tonic::Request<super::ListBucketRequest>,
@@ -3289,6 +3379,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::DeleteRequest>,
) -> std::result::Result<tonic::Response<super::DeleteResponse>, tonic::Status>;
async fn delete_at_incarnation(
&self,
request: tonic::Request<super::DeleteRequest>,
) -> std::result::Result<tonic::Response<super::DeleteResponse>, tonic::Status>;
async fn acquire_snapshot_lease(
&self,
request: tonic::Request<super::SnapshotLeaseRequest>,
@@ -3366,6 +3460,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::RenameDataRequest>,
) -> std::result::Result<tonic::Response<super::RenameDataResponse>, tonic::Status>;
async fn rename_data_at_incarnation(
&self,
request: tonic::Request<super::RenameDataRequest>,
) -> std::result::Result<tonic::Response<super::RenameDataResponse>, tonic::Status>;
async fn make_volumes(
&self,
request: tonic::Request<super::MakeVolumesRequest>,
@@ -3398,6 +3496,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::WriteMetadataRequest>,
) -> std::result::Result<tonic::Response<super::WriteMetadataResponse>, tonic::Status>;
async fn write_metadata_at_incarnation(
&self,
request: tonic::Request<super::WriteMetadataRequest>,
) -> std::result::Result<tonic::Response<super::WriteMetadataResponse>, tonic::Status>;
async fn read_version(
&self,
request: tonic::Request<super::ReadVersionRequest>,
@@ -3414,6 +3516,10 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::DeleteVersionRequest>,
) -> std::result::Result<tonic::Response<super::DeleteVersionResponse>, tonic::Status>;
async fn delete_version_at_incarnation(
&self,
request: tonic::Request<super::DeleteVersionRequest>,
) -> std::result::Result<tonic::Response<super::DeleteVersionResponse>, tonic::Status>;
async fn delete_retired_marker(
&self,
request: tonic::Request<super::DeleteVersionRequest>,
@@ -3771,6 +3877,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/HealBucketAtIncarnation" => {
#[allow(non_camel_case_types)]
struct HealBucketAtIncarnationSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::HealBucketRequest> for HealBucketAtIncarnationSvc<T> {
type Response = super::HealBucketResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::HealBucketRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
@@ -3967,6 +4101,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/DeleteAtIncarnation" => {
#[allow(non_camel_case_types)]
struct DeleteAtIncarnationSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::DeleteRequest> for DeleteAtIncarnationSvc<T> {
type Response = super::DeleteResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::DeleteRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
@@ -4418,6 +4580,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/RenameDataAtIncarnation" => {
#[allow(non_camel_case_types)]
struct RenameDataAtIncarnationSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::RenameDataRequest> for RenameDataAtIncarnationSvc<T> {
type Response = super::RenameDataResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::RenameDataRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
@@ -4642,6 +4832,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/WriteMetadataAtIncarnation" => {
#[allow(non_camel_case_types)]
struct WriteMetadataAtIncarnationSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::WriteMetadataRequest> for WriteMetadataAtIncarnationSvc<T> {
type Response = super::WriteMetadataResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::WriteMetadataRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
@@ -4754,6 +4972,34 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/DeleteVersionAtIncarnation" => {
#[allow(non_camel_case_types)]
struct DeleteVersionAtIncarnationSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::DeleteVersionRequest> for DeleteVersionAtIncarnationSvc<T> {
type Response = super::DeleteVersionResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::DeleteVersionRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::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<T: NodeService>(pub Arc<T>);
crates/protos/src/lib.rs
+27 -1
View File
@@ -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() });
crates/protos/src/node.proto
+11
View File
@@ -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
docs/README.md
+3
View File
@@ -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.
docs/operations/bucket-heal-recovery.md
+52
View File
@@ -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-<task-id>.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).
rustfs/src/storage/rpc/node_service.rs
+308 -1
View File
@@ -127,6 +127,19 @@ fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, ope
.map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}")))
}
fn require_incarnation_body_digest<T>(request: &Request<T>) -> 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<T: CanonicalMutationBody>(request: &Request<T>, 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<HealBucketRequest>) -> Result<Response<HealBucketResponse>, 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<HealBucketRequest>,
) -> Result<Response<HealBucketResponse>, 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<DeleteRequest>) -> Result<Response<DeleteResponse>, 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<DeleteRequest>) -> Result<Response<DeleteResponse>, 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<RenameDataRequest>) -> Result<Response<RenameDataResponse>, 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<RenameDataRequest>,
) -> Result<Response<RenameDataResponse>, 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<WriteMetadataRequest>) -> Result<Response<WriteMetadataResponse>, 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<WriteMetadataRequest>,
) -> Result<Response<WriteMetadataResponse>, 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<DeleteVersionRequest>) -> Result<Response<DeleteVersionResponse>, 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<DeleteVersionRequest>,
) -> Result<Response<DeleteVersionResponse>, 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());
rustfs/src/storage/rpc/node_service/bucket.rs
+26 -6
View File
@@ -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,
rustfs/src/storage/rpc/node_service/disk.rs
+92 -2
View File
@@ -70,14 +70,23 @@ impl LocalMutationTarget {
fi: &FileInfo,
destination: (&str, &str),
scanner_token: Option<Uuid>,
bucket_incarnation: Option<Uuid>,
) -> Result<RenameDataResp, DiskError> {
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<DeleteVersionRequest>,
require_marker_condition: bool,
) -> Result<Response<DeleteVersionResponse>, 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::<FileInfo>(&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::<FileInfo>(&request.file_info_bin, &request.file_info, "FileInfo") {
Ok(file_info) => file_info,
@@ -1273,7 +1327,7 @@ impl NodeService {
&self,
request: Request<RenameDataRequest>,
) -> Result<Response<RenameDataResponse>, 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::<DeleteOptions>(&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::<DeleteOptions>(&request.options) {
Ok(options) => options,