mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
fix(multipart): retain commit guards across request cancellation (#8571)
* fix(multipart): retain commit guards across request cancellation * fix(multipart): preserve commit task error context * fix(multipart): reclaim verified orphan upload staging (#8572) * fix(multipart): reclaim verified orphan upload staging * fix(multipart): share orphan guards across matching lock domains * fix(multipart): preserve orphan cleanup task error context * fix(ecstore): box guarded delete transition identity * fix(multipart): keep NoSuchUpload when orphan recovery cannot prove absence (#8587) Recovery is a side path of an idempotent Abort. An offline disk, lock contention, partially overlapping lock domains or unreadable bucket metadata all stop the staging-absence proof before any deletion; those used to propagate as 503/500, where the base answered NoSuchUpload. They now keep that answer and delete nothing; only failures inside the deletion task remain request failures, because a retry can finish them. --------- Co-authored-by: Chris <anzhengchao@gmail.com>
This commit is contained in:
cxymds
@@ -3149,6 +3149,88 @@ mod delete_recovery_timeout_tests {
|
||||
mod delete_recovery_wire_limit_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn boxed_current_delete_transition_preserves_existing_wire_payload() {
|
||||
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
enum PriorGeneration {
|
||||
Present {
|
||||
version_id: Option<Uuid>,
|
||||
data_dir: Option<Uuid>,
|
||||
mod_time: OffsetDateTime,
|
||||
delete_marker: bool,
|
||||
etag: String,
|
||||
#[serde(default)]
|
||||
transition: Option<CurrentDeleteTransition>,
|
||||
},
|
||||
}
|
||||
|
||||
let version_id = Some(Uuid::from_u128(1));
|
||||
let data_dir = Some(Uuid::from_u128(2));
|
||||
let mod_time = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1);
|
||||
for transition in [
|
||||
None,
|
||||
Some(CurrentDeleteTransition {
|
||||
status: rustfs_filemeta::TRANSITION_COMPLETE.to_owned(),
|
||||
tier: "WARM".into(),
|
||||
remote_object: "opaque/object+name".into(),
|
||||
remote_version: Some("opaque/version+id==".into()),
|
||||
version_state: rustfs_filemeta::TransitionVersionState::Exact,
|
||||
destination_id: Some([0xab; 32]),
|
||||
remote_only: true,
|
||||
}),
|
||||
] {
|
||||
let prior = PriorGeneration::Present {
|
||||
version_id,
|
||||
data_dir,
|
||||
mod_time,
|
||||
delete_marker: false,
|
||||
etag: "opaque-etag".into(),
|
||||
transition: transition.clone(),
|
||||
};
|
||||
let generation = CurrentDeleteGeneration::Present {
|
||||
version_id,
|
||||
data_dir,
|
||||
mod_time,
|
||||
delete_marker: false,
|
||||
etag: "opaque-etag".into(),
|
||||
transition: transition.map(Box::new),
|
||||
};
|
||||
let prior_encoded = rmp_serde::to_vec_named(&prior).expect("encode the previous unboxed generation shape");
|
||||
let encoded = rmp_serde::to_vec_named(&generation).expect("encode boxed transition generation");
|
||||
assert_eq!(encoded, prior_encoded);
|
||||
let decoded: CurrentDeleteGeneration =
|
||||
rustfs_protos::decode_delete_recovery_payload(&prior_encoded).expect("decode the previous generation payload");
|
||||
assert_eq!(decoded, generation);
|
||||
let prior_decoded: PriorGeneration =
|
||||
rustfs_protos::decode_delete_recovery_payload(&encoded).expect("decode boxed transition with the previous shape");
|
||||
assert_eq!(prior_decoded, prior);
|
||||
if matches!(generation, CurrentDeleteGeneration::Present { transition: None, .. }) {
|
||||
#[derive(Serialize)]
|
||||
enum GenerationWithoutTransition {
|
||||
Present {
|
||||
version_id: Option<Uuid>,
|
||||
data_dir: Option<Uuid>,
|
||||
mod_time: OffsetDateTime,
|
||||
delete_marker: bool,
|
||||
etag: String,
|
||||
},
|
||||
}
|
||||
let legacy = GenerationWithoutTransition::Present {
|
||||
version_id,
|
||||
data_dir,
|
||||
mod_time,
|
||||
delete_marker: false,
|
||||
etag: "opaque-etag".into(),
|
||||
};
|
||||
let legacy_encoded = rmp_serde::to_vec_named(&legacy).expect("encode a generation without the transition field");
|
||||
let legacy_decoded: CurrentDeleteGeneration =
|
||||
rustfs_protos::decode_delete_recovery_payload(&legacy_encoded).expect("default an omitted transition field");
|
||||
assert_eq!(legacy_decoded, generation);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn boxed_apply_version_preserves_existing_wire_payload() {
|
||||
#[derive(Serialize)]
|
||||
|
||||
@@ -1015,6 +1015,9 @@ pub struct ObjectOptions {
|
||||
/// Proves an upper layer holds the bucket lifecycle sentinel. A separate
|
||||
/// fence avoids recursively acquiring the read lock behind a queued writer.
|
||||
pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
|
||||
/// Retains the actual lifecycle guard for an owned multipart commit. A
|
||||
/// lock-loss fence alone does not keep a cancelled caller's lock alive.
|
||||
pub bucket_lifecycle_lock_owner: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
|
||||
pub replication_request: bool,
|
||||
/// True when the inbound request carried the
|
||||
/// `{x-rustfs-,x-minio-}source-proxy-request` header family with the
|
||||
@@ -1175,6 +1178,7 @@ impl std::fmt::Debug for ObjectOptions {
|
||||
.field("conditional_delete_generations", &self.conditional_delete_generations.is_some())
|
||||
.field("namespace_lock_fence", &self.namespace_lock_fence.is_some())
|
||||
.field("bucket_lifecycle_lock_fence", &self.bucket_lifecycle_lock_fence.is_some())
|
||||
.field("bucket_lifecycle_lock_owner", &self.bucket_lifecycle_lock_owner.is_some())
|
||||
.field("replication_request", &self.replication_request)
|
||||
.field("proxy_request", &self.proxy_request)
|
||||
.field("proxy_header_set", &self.proxy_header_set)
|
||||
|
||||
@@ -63,7 +63,6 @@ use crate::crash_inject::{self, CrashPoint};
|
||||
use crate::disk::DiskAPI;
|
||||
#[cfg(test)]
|
||||
use crate::disk::DiskOption;
|
||||
#[cfg(test)]
|
||||
use crate::disk::STORAGE_FORMAT_FILE;
|
||||
#[cfg(test)]
|
||||
use crate::disk::new_disk;
|
||||
@@ -231,6 +230,60 @@ pub(crate) struct StaleMultipartCleanupGuard {
|
||||
lock_guard: ObjectLockDiagGuard,
|
||||
}
|
||||
|
||||
/// Exact-key recovery is admitted only after every disk proves that the upload
|
||||
/// metadata is absent. Keep the checked disk generation and upload lock through
|
||||
/// deletion; a quorum miss is not proof that an upload is an orphan.
|
||||
/// Final object data, completion replay identity, and old-data cleanup receipts
|
||||
/// live under the destination bucket/object, never this staging path. This also
|
||||
/// permits removing an exact orphan from a previous bucket incarnation without
|
||||
/// claiming that its UploadID belongs to the current incarnation.
|
||||
pub(crate) struct OrphanMultipartCleanupGuard {
|
||||
set: Arc<SetDisks>,
|
||||
disks: Vec<Option<DiskStore>>,
|
||||
upload_path: String,
|
||||
lock_guard: Arc<ObjectLockDiagGuard>,
|
||||
exists: bool,
|
||||
}
|
||||
|
||||
impl OrphanMultipartCleanupGuard {
|
||||
pub(crate) async fn delete(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<bool> {
|
||||
fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
|
||||
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
|
||||
if !self.exists {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
let results = join_all(self.disks.iter().map(|disk| async {
|
||||
let Some(disk) = disk else {
|
||||
return Err(DiskError::DiskNotFound);
|
||||
};
|
||||
match disk
|
||||
.delete(
|
||||
RUSTFS_META_MULTIPART_BUCKET,
|
||||
&self.upload_path,
|
||||
crate::disk::DeleteOptions {
|
||||
recursive: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => Ok(()),
|
||||
result => result,
|
||||
}
|
||||
}))
|
||||
.await;
|
||||
self.set.record_capacity_scope_if_needed(None, &self.disks);
|
||||
let errors = results.into_iter().map(|result| result.err()).collect::<Vec<_>>();
|
||||
if let Some(err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, self.disks.len()) {
|
||||
return Err(to_object_err(err.into(), vec![bucket, object]));
|
||||
}
|
||||
fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
|
||||
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
|
||||
Ok(true)
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct DataMovementMultipartAbortGuard {
|
||||
upload_path: String,
|
||||
write_quorum: Option<usize>,
|
||||
@@ -557,6 +610,34 @@ fn multipart_bucket_incarnation_matches(metadata: &HashMap<String, String>, expe
|
||||
matches!(multipart_bucket_incarnation_id(metadata), Ok(Some(actual)) if actual == expected)
|
||||
}
|
||||
|
||||
fn orphan_multipart_upload_id_is_canonical(upload_id: &str) -> bool {
|
||||
let Ok(decoded) = base64_simd::URL_SAFE_NO_PAD.decode_to_vec(upload_id.as_bytes()) else {
|
||||
return false;
|
||||
};
|
||||
let Ok(decoded) = std::str::from_utf8(&decoded) else {
|
||||
return false;
|
||||
};
|
||||
let Some((_, raw)) = decoded.split_once('.') else {
|
||||
return false;
|
||||
};
|
||||
let Some((uuid, timestamp)) = raw.split_once('x') else {
|
||||
return false;
|
||||
};
|
||||
let Ok(id) = Uuid::parse_str(uuid) else {
|
||||
return false;
|
||||
};
|
||||
if id.is_nil() || id.to_string() != uuid {
|
||||
return false;
|
||||
}
|
||||
let Ok(timestamp_ns) = timestamp.parse::<i128>() else {
|
||||
return false;
|
||||
};
|
||||
if timestamp_ns.to_string() != timestamp {
|
||||
return false;
|
||||
}
|
||||
OffsetDateTime::from_unix_timestamp_nanos(timestamp_ns).is_ok()
|
||||
}
|
||||
|
||||
fn validate_multipart_bucket_incarnation(
|
||||
metadata: &HashMap<String, String>,
|
||||
bucket: &str,
|
||||
@@ -1246,6 +1327,114 @@ impl SetDisks {
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) async fn lock_orphan_multipart_cleanup(
|
||||
self: &Arc<Self>,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
upload_id: &str,
|
||||
opts: &ObjectOptions,
|
||||
held_guards: &[OrphanMultipartCleanupGuard],
|
||||
) -> Result<OrphanMultipartCleanupGuard> {
|
||||
let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
|
||||
if opts.data_movement
|
||||
|| opts.no_lock
|
||||
|| self.local_lock_manager.is_disabled()
|
||||
|| !orphan_multipart_upload_id_is_canonical(upload_id)
|
||||
{
|
||||
return Err(invalid_upload());
|
||||
}
|
||||
// This path is derived from the caller's exact bucket/key and a canonical
|
||||
// UUID/timestamp upload identity, never from the shared MPU directory list.
|
||||
let upload_path = Self::get_upload_id_dir(bucket, object, upload_id);
|
||||
// Pool labels do not partition the raw ObjectKey sent to lock clients.
|
||||
// Reuse the actual owner only for this exact path and proven backend;
|
||||
// independent distributed domains still need their own guards.
|
||||
let distributed = self.ctx.is_dist_erasure().await;
|
||||
let mut lock_guard = None;
|
||||
for held in held_guards {
|
||||
if held.upload_path != upload_path {
|
||||
continue;
|
||||
}
|
||||
let shared_domain = match (distributed, held.set.ctx.is_dist_erasure().await) {
|
||||
(false, false) => self.shares_namespace_lock_domain(&held.set).await,
|
||||
(true, true) => {
|
||||
let same = super::super::same_distributed_lock_domain(&self.lockers, &held.set.lockers);
|
||||
if !same
|
||||
&& self
|
||||
.lockers
|
||||
.iter()
|
||||
.any(|client| held.set.lockers.iter().any(|held_client| Arc::ptr_eq(client, held_client)))
|
||||
{
|
||||
// A guard for only part of another quorum is not proof
|
||||
// of ownership there, and acquiring it again may wait
|
||||
// on ourselves. Preserve all staging in this topology.
|
||||
return Err(Error::other("orphan multipart cleanup cannot cover partially overlapping lock domains"));
|
||||
}
|
||||
same
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
if shared_domain {
|
||||
lock_guard = Some(Arc::clone(&held.lock_guard));
|
||||
break;
|
||||
}
|
||||
}
|
||||
let lock_guard = match lock_guard {
|
||||
Some(guard) => guard,
|
||||
None => Arc::new(
|
||||
self.acquire_write_lock_diag("orphan_multipart_cleanup", RUSTFS_META_MULTIPART_BUCKET, &upload_path)
|
||||
.await?,
|
||||
),
|
||||
};
|
||||
fence_commit_on_lock_loss(Some(&lock_guard), "orphan_multipart_cleanup", &upload_path)?;
|
||||
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
|
||||
let disks = self.disks.read().await.clone();
|
||||
if disks.len() != self.set_drive_count || disks.is_empty() {
|
||||
return Err(Error::ErasureReadQuorum);
|
||||
}
|
||||
// Raw full fan-out cannot quorum-short-circuit or interpret an empty
|
||||
// version list as absence of the physical metadata file.
|
||||
let (_, errors) = Self::read_all_raw_file_info(&disks, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
|
||||
if errors.len() != disks.len() {
|
||||
return Err(Error::ErasureReadQuorum);
|
||||
}
|
||||
for error in errors {
|
||||
match error {
|
||||
Some(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
|
||||
// Even one valid metadata copy may still be recoverable. Never
|
||||
// convert a partial, corrupt, or unreachable upload into debris.
|
||||
None => return Err(invalid_upload()),
|
||||
Some(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
|
||||
}
|
||||
}
|
||||
let mut exists = false;
|
||||
for disk in &disks {
|
||||
let Some(disk) = disk else {
|
||||
return Err(Error::ErasureReadQuorum);
|
||||
};
|
||||
match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await {
|
||||
Ok(entries) => {
|
||||
// Raw metadata reads report a directory named xl.meta as
|
||||
// FileNotFound. Its physical presence is unknown/corrupt
|
||||
// metadata, not proof that this staging can be deleted.
|
||||
if entries.iter().any(|entry| entry.trim_end_matches('/') == STORAGE_FORMAT_FILE) {
|
||||
return Err(to_object_err(DiskError::FileCorrupt.into(), vec![bucket, object, upload_id]));
|
||||
}
|
||||
exists = true;
|
||||
}
|
||||
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
|
||||
Err(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
|
||||
}
|
||||
}
|
||||
Ok(OrphanMultipartCleanupGuard {
|
||||
set: self.clone(),
|
||||
disks,
|
||||
upload_path,
|
||||
lock_guard,
|
||||
exists,
|
||||
})
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) async fn list_multipart_uploads_for_incarnation(
|
||||
&self,
|
||||
@@ -1678,6 +1867,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
drop(admission_guard);
|
||||
}
|
||||
|
||||
let mut commit_owns_cleanup = false;
|
||||
let result: Result<PartInfo> = async {
|
||||
let erasure =
|
||||
Arc::new(coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size)
|
||||
@@ -1881,120 +2071,171 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
.await?;
|
||||
(Some(upload_guard), Some(part_guard))
|
||||
};
|
||||
let (commit_fi, _) = self
|
||||
.check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
|
||||
.await?;
|
||||
let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
|
||||
if commit_size_limit != multipart_size_limit {
|
||||
return Err(Error::InvalidArgument(
|
||||
"multipart upload".to_string(),
|
||||
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
|
||||
"size limit metadata changed or is missing".to_string(),
|
||||
));
|
||||
}
|
||||
ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
|
||||
ensure_multipart_bucket_incarnation(
|
||||
&self.ctx,
|
||||
&commit_fi,
|
||||
bucket,
|
||||
object,
|
||||
upload_id,
|
||||
opts.expected_bucket_incarnation_id,
|
||||
)
|
||||
.await?;
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
|
||||
fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
|
||||
fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
|
||||
if opts
|
||||
.namespace_lock_fence
|
||||
.as_ref()
|
||||
.is_some_and(NamespaceLockFence::is_lock_lost)
|
||||
{
|
||||
return Err(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "put_object_part_outer_lock",
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
|
||||
|
||||
if let Some(limit) = commit_size_limit {
|
||||
let current_size = self
|
||||
.current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
|
||||
.await?;
|
||||
let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
|
||||
admitted_multipart_size(current_size, candidate_size, limit)?;
|
||||
}
|
||||
|
||||
let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
|
||||
Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(
|
||||
&[self.pool_index],
|
||||
DecommissionCapacityAdmission::ExistingMultipart,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
|
||||
if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
|
||||
return Err(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "put_object_part_decommission_capacity",
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
let rename_result = self
|
||||
.rename_part(
|
||||
&shuffle_disks,
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
&tmp_part_path,
|
||||
RUSTFS_META_MULTIPART_BUCKET,
|
||||
&part_path,
|
||||
part_info_buff.into(),
|
||||
write_quorum,
|
||||
Some(MultipartWriteQuorumContext {
|
||||
stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
|
||||
// Only guards owned by this operation may outlive its caller. Internal
|
||||
// data-movement and borrowed-lock callers retain their synchronous
|
||||
// lifetime; a fence by itself cannot transfer the outer lock lease.
|
||||
let detach_commit_owner = _upload_commit_guard.is_some()
|
||||
&& opts.namespace_lock_fence.is_none()
|
||||
&& (opts.bucket_lifecycle_lock_fence.is_none() || opts.bucket_lifecycle_lock_owner.is_some());
|
||||
let commit_set = self.clone();
|
||||
let commit_bucket = bucket.to_owned();
|
||||
let commit_object = object.to_owned();
|
||||
let commit_upload_id = upload_id.to_owned();
|
||||
let commit_upload_id_path = upload_id_path.clone();
|
||||
let commit_tmp_part = tmp_part.clone();
|
||||
let commit_opts = opts.clone();
|
||||
commit_owns_cleanup = true;
|
||||
// The body is already staged. Keep every commit guard through the
|
||||
// prepare, publish, settlement/rollback and temporary cleanup I/O,
|
||||
// even when the HTTP request stops awaiting this task.
|
||||
let commit = async move {
|
||||
let _upload_commit_guard = _upload_commit_guard;
|
||||
let _part_commit_guard = _part_commit_guard;
|
||||
let _capped_staging_guard = _capped_staging_guard;
|
||||
let bucket = commit_bucket.as_str();
|
||||
let object = commit_object.as_str();
|
||||
let upload_id = commit_upload_id.as_str();
|
||||
let upload_id_path = commit_upload_id_path;
|
||||
let opts = &commit_opts;
|
||||
let result: Result<PartInfo> = async {
|
||||
let (commit_fi, _) = commit_set
|
||||
.check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
|
||||
.await?;
|
||||
let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
|
||||
if commit_size_limit != multipart_size_limit {
|
||||
return Err(Error::InvalidArgument(
|
||||
"multipart upload".to_string(),
|
||||
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
|
||||
"size limit metadata changed or is missing".to_string(),
|
||||
));
|
||||
}
|
||||
ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
|
||||
ensure_multipart_bucket_incarnation(
|
||||
&commit_set.ctx,
|
||||
&commit_fi,
|
||||
bucket,
|
||||
object,
|
||||
upload_id,
|
||||
part_number: Some(part_id),
|
||||
}),
|
||||
)
|
||||
opts.expected_bucket_incarnation_id,
|
||||
)
|
||||
.await?;
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
|
||||
fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
|
||||
fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
|
||||
if opts
|
||||
.namespace_lock_fence
|
||||
.as_ref()
|
||||
.is_some_and(NamespaceLockFence::is_lock_lost)
|
||||
{
|
||||
return Err(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "put_object_part_outer_lock",
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
|
||||
|
||||
if let Some(limit) = commit_size_limit {
|
||||
let current_size = commit_set
|
||||
.current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
|
||||
.await?;
|
||||
let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
|
||||
admitted_multipart_size(current_size, candidate_size, limit)?;
|
||||
}
|
||||
|
||||
let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
|
||||
Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(
|
||||
&[commit_set.pool_index],
|
||||
DecommissionCapacityAdmission::ExistingMultipart,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
|
||||
if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
|
||||
return Err(StorageError::NamespaceLockQuorumUnavailable {
|
||||
mode: "put_object_part_decommission_capacity",
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
required: 1,
|
||||
achieved: 0,
|
||||
});
|
||||
}
|
||||
let rename_result = commit_set
|
||||
.rename_part(
|
||||
&shuffle_disks,
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
&tmp_part_path,
|
||||
RUSTFS_META_MULTIPART_BUCKET,
|
||||
&part_path,
|
||||
part_info_buff.into(),
|
||||
write_quorum,
|
||||
Some(MultipartWriteQuorumContext {
|
||||
stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
|
||||
bucket,
|
||||
object,
|
||||
upload_id,
|
||||
part_number: Some(part_id),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
drop(decommission_capacity_guard);
|
||||
let _ = rename_result?;
|
||||
#[cfg(test)]
|
||||
observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
|
||||
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
|
||||
|
||||
let ret: PartInfo = PartInfo {
|
||||
etag: Some(etag.clone()),
|
||||
part_num: part_id,
|
||||
last_mod: Some(OffsetDateTime::now_utc()),
|
||||
size: w_size,
|
||||
actual_size,
|
||||
};
|
||||
|
||||
// error!("put_object_part ret {:?}", &ret);
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
.await;
|
||||
drop(decommission_capacity_guard);
|
||||
let _ = rename_result?;
|
||||
#[cfg(test)]
|
||||
observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
|
||||
|
||||
#[cfg(test)]
|
||||
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
|
||||
drop(_part_commit_guard);
|
||||
drop(_upload_commit_guard);
|
||||
|
||||
let ret: PartInfo = PartInfo {
|
||||
etag: Some(etag.clone()),
|
||||
part_num: part_id,
|
||||
last_mod: Some(OffsetDateTime::now_utc()),
|
||||
size: w_size,
|
||||
actual_size,
|
||||
if result.is_err()
|
||||
&& let Err(err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_part).await
|
||||
{
|
||||
warn!(
|
||||
event = EVENT_SET_DISK_MULTIPART,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
subsystem = LOG_SUBSYSTEM_SET_DISK,
|
||||
result = "cleanup_failed",
|
||||
bucket,
|
||||
object,
|
||||
error = ?err,
|
||||
"failed to clean multipart commit staging"
|
||||
);
|
||||
}
|
||||
result
|
||||
};
|
||||
|
||||
// error!("put_object_part ret {:?}", &ret);
|
||||
|
||||
Ok(ret)
|
||||
if detach_commit_owner {
|
||||
tokio::spawn(commit)
|
||||
.await
|
||||
.map_err(|err| Error::other_with_context("put_object_part commit task failed", err))?
|
||||
} else {
|
||||
commit.await
|
||||
}
|
||||
}
|
||||
.await;
|
||||
|
||||
if result.is_err()
|
||||
if !commit_owns_cleanup
|
||||
&& result.is_err()
|
||||
&& let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_part).await
|
||||
{
|
||||
warn!(tmp_part = %tmp_part, error = ?err, "failed to cleanup multipart temporary data");
|
||||
@@ -4233,6 +4474,300 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn orphan_multipart_identity_requires_canonical_upload_id() {
|
||||
let created = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10);
|
||||
let id = Uuid::new_v4();
|
||||
let timestamp = (created + time::Duration::seconds(1)).unix_timestamp_nanos();
|
||||
let encode = |raw: &str| runtime_sources::deployment_upload_id(raw);
|
||||
let upload_id = encode(&format!("{id}x{timestamp}"));
|
||||
assert!(orphan_multipart_upload_id_is_canonical(&upload_id));
|
||||
for raw in [
|
||||
format!("{}x{timestamp}", Uuid::nil()),
|
||||
format!("{}x{timestamp}", id.simple()),
|
||||
format!("{id}x0{timestamp}"),
|
||||
format!("{id}x+{timestamp}"),
|
||||
format!("{id}x{timestamp}/../another-upload"),
|
||||
format!("../{id}x{timestamp}"),
|
||||
format!("{id}x{}", i128::MAX),
|
||||
] {
|
||||
assert!(
|
||||
!orphan_multipart_upload_id_is_canonical(&encode(&raw)),
|
||||
"noncanonical upload identity must be rejected: {raw}"
|
||||
);
|
||||
}
|
||||
assert!(!orphan_multipart_upload_id_is_canonical("not-base64"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn orphan_multipart_cleanup_requires_every_disk_to_prove_metadata_absence() {
|
||||
let (_temp_dirs, disks, set) = hermetic_set_disks(4).await;
|
||||
let bucket = "orphan-multipart-proof";
|
||||
let object = "object";
|
||||
make_bucket_on_all(&disks, bucket).await;
|
||||
let upload = set
|
||||
.new_multipart_upload(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("create upload");
|
||||
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
|
||||
let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
|
||||
let leftover_path = format!("{upload_path}/part.late");
|
||||
for disk in &disks {
|
||||
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
|
||||
.await
|
||||
.expect("write late part fixture");
|
||||
}
|
||||
for disk in &disks[..3] {
|
||||
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
|
||||
.await
|
||||
.expect("remove majority metadata");
|
||||
}
|
||||
let opts = ObjectOptions::default();
|
||||
assert!(
|
||||
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
|
||||
.await
|
||||
.is_err(),
|
||||
"one surviving metadata copy must block orphan recovery"
|
||||
);
|
||||
disks[3]
|
||||
.write_all(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Bytes::from_static(b"corrupt metadata"))
|
||||
.await
|
||||
.expect("corrupt final metadata copy");
|
||||
assert!(
|
||||
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
|
||||
.await
|
||||
.is_err(),
|
||||
"corrupt metadata must not be treated as absence"
|
||||
);
|
||||
disks[3]
|
||||
.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
|
||||
.await
|
||||
.expect("remove final metadata copy");
|
||||
set.disks.write().await[3] = None;
|
||||
assert!(
|
||||
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
|
||||
.await
|
||||
.is_err(),
|
||||
"an unreachable disk must not be hidden by absence quorum"
|
||||
);
|
||||
set.disks.write().await[3] = Some(disks[3].clone());
|
||||
for disk in &disks {
|
||||
assert_eq!(
|
||||
disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
|
||||
.await
|
||||
.expect("failed preflight must preserve every part"),
|
||||
Bytes::from_static(b"late part")
|
||||
);
|
||||
}
|
||||
|
||||
let guard = set
|
||||
.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
|
||||
.await
|
||||
.expect("all disks now prove metadata absence");
|
||||
assert!(guard.delete(bucket, object, &opts).await.expect("recover exact staging"));
|
||||
drop(guard);
|
||||
for disk in &disks {
|
||||
assert!(matches!(
|
||||
disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
|
||||
Err(DiskError::FileNotFound | DiskError::VolumeNotFound)
|
||||
));
|
||||
}
|
||||
let guard = set
|
||||
.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
|
||||
.await
|
||||
.expect("already absent staging remains provably absent");
|
||||
assert!(
|
||||
!guard
|
||||
.delete(bucket, object, &opts)
|
||||
.await
|
||||
.expect("absence does not report recovery")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn orphan_multipart_cleanup_rejects_metadata_directory_on_one_disk() {
|
||||
let (temp_dirs, disks, set) = hermetic_set_disks(4).await;
|
||||
let bucket = "orphan-multipart-directory";
|
||||
let object = "object";
|
||||
make_bucket_on_all(&disks, bucket).await;
|
||||
let upload = set
|
||||
.new_multipart_upload(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("create upload");
|
||||
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
|
||||
let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
|
||||
let leftover_path = format!("{upload_path}/part.late");
|
||||
for disk in &disks {
|
||||
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
|
||||
.await
|
||||
.expect("remove upload metadata");
|
||||
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
|
||||
.await
|
||||
.expect("write leftover part");
|
||||
}
|
||||
let metadata_directory = temp_dirs[3].path().join(RUSTFS_META_MULTIPART_BUCKET).join(&metadata_path);
|
||||
tokio::fs::create_dir_all(&metadata_directory)
|
||||
.await
|
||||
.expect("replace one missing metadata file with a directory");
|
||||
let (_, errors) =
|
||||
SetDisks::read_all_raw_file_info(&set.disks.read().await, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
|
||||
assert!(
|
||||
errors.iter().all(|error| matches!(error, Some(DiskError::FileNotFound))),
|
||||
"raw metadata reads alone must reproduce the directory-as-absence case: {errors:?}"
|
||||
);
|
||||
assert!(
|
||||
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &ObjectOptions::default(), &[])
|
||||
.await
|
||||
.is_err(),
|
||||
"one physical metadata directory must block cleanup on every disk"
|
||||
);
|
||||
assert!(metadata_directory.is_dir(), "failed preflight must preserve the metadata directory");
|
||||
for disk in &disks {
|
||||
assert_eq!(
|
||||
disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
|
||||
.await
|
||||
.expect("failed preflight must preserve every leftover part"),
|
||||
Bytes::from_static(b"late part")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, PartialEq, Eq)]
|
||||
enum OrphanMultipartLockDomainCase {
|
||||
Shared,
|
||||
Independent,
|
||||
PartiallyOverlapping,
|
||||
}
|
||||
|
||||
async fn check_orphan_multipart_distributed_lock_owners(case: OrphanMultipartLockDomainCase) {
|
||||
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
|
||||
ctx.update_erasure_type(SetupType::DistErasure).await;
|
||||
let first_client: Arc<dyn LockClient> =
|
||||
Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
|
||||
let second_client: Arc<dyn LockClient> =
|
||||
Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
|
||||
let second_clients = match case {
|
||||
OrphanMultipartLockDomainCase::Shared => vec![first_client.clone()],
|
||||
OrphanMultipartLockDomainCase::Independent => vec![second_client],
|
||||
OrphanMultipartLockDomainCase::PartiallyOverlapping => vec![first_client.clone(), second_client],
|
||||
};
|
||||
let (first_dirs, first_disks, first_set) =
|
||||
hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, vec![first_client], ctx.clone()).await;
|
||||
let (second_dirs, second_disks, second_set) = hermetic_set_disks_with_lockers_and_ctx(4, 1, 2, second_clients, ctx).await;
|
||||
let bucket = "orphan-multipart-dist-locks";
|
||||
let object = "object";
|
||||
make_bucket_on_all(&first_disks, bucket).await;
|
||||
make_bucket_on_all(&second_disks, bucket).await;
|
||||
let upload_id = runtime_sources::deployment_upload_id(&format!(
|
||||
"{}x{}",
|
||||
Uuid::new_v4(),
|
||||
OffsetDateTime::now_utc().unix_timestamp_nanos()
|
||||
));
|
||||
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
|
||||
for directory in first_dirs.iter().chain(second_dirs.iter()) {
|
||||
let staging = directory.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path);
|
||||
tokio::fs::create_dir_all(&staging)
|
||||
.await
|
||||
.expect("create exact orphan staging");
|
||||
tokio::fs::write(staging.join("part.late"), b"late part")
|
||||
.await
|
||||
.expect("write orphan part");
|
||||
}
|
||||
let opts = ObjectOptions::default();
|
||||
let first_guard = first_set
|
||||
.lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, &[])
|
||||
.await
|
||||
.expect("first domain has an orphan");
|
||||
let second_result = tokio::time::timeout(
|
||||
Duration::from_secs(10),
|
||||
second_set.lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, std::slice::from_ref(&first_guard)),
|
||||
)
|
||||
.await
|
||||
.expect("lock-domain admission must not wait on a guard it already owns");
|
||||
if case == OrphanMultipartLockDomainCase::PartiallyOverlapping {
|
||||
let error = match second_result {
|
||||
Ok(_) => panic!("one existing quorum cannot authorize a partially overlapping domain"),
|
||||
Err(error) => error,
|
||||
};
|
||||
assert!(error.to_string().contains("partially overlapping lock domains"));
|
||||
for directory in first_dirs.iter().chain(second_dirs.iter()) {
|
||||
assert!(
|
||||
directory
|
||||
.path()
|
||||
.join(RUSTFS_META_MULTIPART_BUCKET)
|
||||
.join(&upload_path)
|
||||
.join("part.late")
|
||||
.exists(),
|
||||
"uncertain lock coverage must preserve every physical part"
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let second_guard = second_result.expect("the second pool has a proven lock domain");
|
||||
assert_eq!(
|
||||
Arc::ptr_eq(&first_guard.lock_guard, &second_guard.lock_guard),
|
||||
case == OrphanMultipartLockDomainCase::Shared,
|
||||
"only the same exact backend may share a real upload guard across pool labels"
|
||||
);
|
||||
let second_namespace = second_set
|
||||
.new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
|
||||
.await
|
||||
.unwrap();
|
||||
let mut second_writer = Box::pin(second_namespace.get_write_lock(Duration::from_secs(10)));
|
||||
assert!(futures::poll!(&mut second_writer).is_pending());
|
||||
if case == OrphanMultipartLockDomainCase::Shared {
|
||||
drop(first_guard);
|
||||
assert!(
|
||||
futures::poll!(&mut second_writer).is_pending(),
|
||||
"the second pool must retain the shared real owner after the first guard drops"
|
||||
);
|
||||
drop(second_guard);
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_secs(10), second_writer)
|
||||
.await
|
||||
.expect("the shared backend should release after the last owner")
|
||||
.expect("shared-domain writer resumes"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
let first_namespace = first_set
|
||||
.new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
|
||||
.await
|
||||
.unwrap();
|
||||
let mut first_writer = Box::pin(first_namespace.get_write_lock(Duration::from_secs(10)));
|
||||
assert!(futures::poll!(&mut first_writer).is_pending());
|
||||
drop(first_guard);
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_secs(10), first_writer)
|
||||
.await
|
||||
.expect("first domain should release independently")
|
||||
.expect("first writer resumes"),
|
||||
);
|
||||
assert!(futures::poll!(&mut second_writer).is_pending(), "the second real owner must remain held");
|
||||
drop(second_guard);
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_secs(10), second_writer)
|
||||
.await
|
||||
.expect("second domain should release with its owner")
|
||||
.expect("second writer resumes"),
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn orphan_multipart_cleanup_keeps_separate_distributed_lock_owners() {
|
||||
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Independent).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn orphan_multipart_cleanup_reuses_shared_distributed_owner_across_pools() {
|
||||
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Shared).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn orphan_multipart_cleanup_rejects_partially_overlapping_distributed_domains() {
|
||||
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::PartiallyOverlapping).await;
|
||||
}
|
||||
|
||||
async fn stage_upload_with_create_opts(
|
||||
set_disks: &Arc<SetDisks>,
|
||||
bucket: &str,
|
||||
@@ -6370,6 +6905,76 @@ mod tests {
|
||||
.expect("abort should delete the upload after UploadPart releases the lock");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial]
|
||||
async fn cancelled_put_object_part_keeps_upload_lock_through_commit() {
|
||||
for pause in [
|
||||
MultipartCommitPause::PutPartAfterCapacityAdmission,
|
||||
MultipartCommitPause::PutPartAfterRename,
|
||||
] {
|
||||
let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
|
||||
let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
|
||||
let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
|
||||
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
|
||||
let bucket = "multipart-cancelled-part-commit-bucket";
|
||||
let object = "object";
|
||||
make_bucket_on_all(&disk_stores, bucket).await;
|
||||
let upload = set_disks
|
||||
.new_multipart_upload(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("multipart upload should be created");
|
||||
let upload_id = upload.upload_id;
|
||||
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
|
||||
signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_path.clone()));
|
||||
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
|
||||
let barrier = MultipartCommitBarrier::install(bucket, object, pause);
|
||||
|
||||
let put_store = set_disks.clone();
|
||||
let put_upload_id = upload_id.clone();
|
||||
let put = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![0x45; 4096]);
|
||||
put_store
|
||||
.put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
barrier.wait_until_paused().await;
|
||||
put.abort();
|
||||
assert!(
|
||||
put.await
|
||||
.expect_err("the request waiter should remain cancellable")
|
||||
.is_cancelled()
|
||||
);
|
||||
|
||||
let abort_store = set_disks.clone();
|
||||
let abort_upload_id = upload_id.clone();
|
||||
let mut abort = tokio::spawn(async move {
|
||||
abort_store
|
||||
.abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
signaling.wait_for_attempts(2).await;
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), &mut abort).await.is_err(),
|
||||
"Abort must not acquire the upload lock while a cancelled request's commit is paused at {pause:?}"
|
||||
);
|
||||
barrier.release();
|
||||
tokio::time::timeout(Duration::from_secs(10), abort)
|
||||
.await
|
||||
.expect("Abort should finish after the commit drains")
|
||||
.expect("Abort task should not panic")
|
||||
.expect("Abort should delete the committed part staging");
|
||||
for disk in &disk_stores {
|
||||
assert!(
|
||||
matches!(
|
||||
disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
|
||||
Err(DiskError::FileNotFound)
|
||||
),
|
||||
"the upload directory must be absent after the commit and Abort finish"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial]
|
||||
async fn put_object_part_different_part_numbers_commit_concurrently() {
|
||||
@@ -9799,6 +10404,18 @@ mod tests {
|
||||
"the post-commit crash must leave the upload listable for reclamation"
|
||||
);
|
||||
|
||||
let cleanup = set_disks
|
||||
.lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
|
||||
.await
|
||||
.expect("the committed upload has no staging metadata on any disk");
|
||||
assert!(
|
||||
cleanup
|
||||
.delete(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("recover completed upload staging")
|
||||
);
|
||||
drop(cleanup);
|
||||
|
||||
// The commit rename already consumed the upload's xl.meta, so the
|
||||
// retry cannot read staging. It must still return the committed
|
||||
// object: AWS CompleteMultipartUpload is idempotent for the same
|
||||
@@ -9817,21 +10434,6 @@ mod tests {
|
||||
!upload_is_listed(&set_disks, bucket, object, &u_new).await,
|
||||
"the idempotent retry must reclaim staging left behind by the crash"
|
||||
);
|
||||
|
||||
// Reclaim the leftover exactly as the production tail does: delete_all
|
||||
// on the upload path (abort_multipart_upload cannot — the upload's
|
||||
// metadata is gone, so it too answers InvalidUploadID).
|
||||
let upload_dir = SetDisks::get_upload_id_dir(bucket, object, &u_new);
|
||||
set_disks
|
||||
.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_dir)
|
||||
.await
|
||||
.expect("sweeping the leftover upload staging should succeed");
|
||||
assert!(
|
||||
!upload_is_listed(&set_disks, bucket, object, &u_new).await,
|
||||
"reclaiming the upload must drop it from the listing"
|
||||
);
|
||||
let (body_after, _) = read_object(&set_disks, bucket, object).await;
|
||||
assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -9878,6 +10480,18 @@ mod tests {
|
||||
.expect("the failed post-commit completion should release its object guard"),
|
||||
);
|
||||
|
||||
let cleanup = set_disks
|
||||
.lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
|
||||
.await
|
||||
.expect("committed staging has no upload metadata");
|
||||
assert!(
|
||||
cleanup
|
||||
.delete(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("orphan recovery must only remove upload staging")
|
||||
);
|
||||
drop(cleanup);
|
||||
|
||||
let (body, _) = read_object(&set_disks, bucket, object).await;
|
||||
assert_eq!(body, new, "the committed replacement must remain readable after the crash");
|
||||
let mut receipts = 0;
|
||||
|
||||
@@ -5030,6 +5030,339 @@ mod tests {
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn repeated_abort_recovers_exact_orphan_before_decommission_drains() {
|
||||
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
|
||||
use sha2::Digest;
|
||||
|
||||
let temp_dir = tempfile::tempdir().expect("create orphan recovery store dir");
|
||||
let (_ctx, store, shutdown) =
|
||||
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-multipart-drain", &[4, 4])).await;
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("orphan-multipart-{}", uuid::Uuid::new_v4());
|
||||
let object = "exact-object.bin";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create bucket");
|
||||
let incarnation = store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation");
|
||||
let lifecycle_guard = store.acquire_bucket_lifecycle_read_lock(&bucket).await.expect("bucket guard");
|
||||
let mut upload_opts = ObjectOptions {
|
||||
expected_bucket_incarnation_id: Some(incarnation),
|
||||
..Default::default()
|
||||
};
|
||||
upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
|
||||
let upload = store.pools[0]
|
||||
.new_multipart_upload(&bucket, object, &upload_opts)
|
||||
.await
|
||||
.expect("create source upload");
|
||||
drop(lifecycle_guard);
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("initial abort");
|
||||
|
||||
let upload_sha =
|
||||
hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
|
||||
let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
|
||||
let staging_dir = |pool_index: usize, disk_index: usize| {
|
||||
temp_dir
|
||||
.path()
|
||||
.join(format!("pool{pool_index}/set0/disk{disk_index}"))
|
||||
.join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
|
||||
.join(&upload_sha)
|
||||
.join(&upload_uuid)
|
||||
};
|
||||
for disk_index in 0..3 {
|
||||
let path = staging_dir(0, disk_index);
|
||||
tokio::fs::create_dir_all(&path).await.expect("create late part directory");
|
||||
tokio::fs::write(path.join("part.1"), b"late part after abort")
|
||||
.await
|
||||
.expect("write late part fixture");
|
||||
}
|
||||
mark_test_pool_decommissioning(&store, 0).await;
|
||||
assert!(
|
||||
store
|
||||
.list_multipart_uploads(&bucket, "", None, None, None, 100)
|
||||
.await
|
||||
.expect("logical upload listing")
|
||||
.uploads
|
||||
.is_empty()
|
||||
);
|
||||
assert!(store.ensure_decommission_multipart_uploads_drained_for_test(0).await.is_err());
|
||||
assert!(
|
||||
store
|
||||
.abort_multipart_upload(&bucket, "different-key", &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.is_err(),
|
||||
"the same UploadID for a different key must not recover this path"
|
||||
);
|
||||
|
||||
// A minority metadata copy in another pool blocks recovery everywhere.
|
||||
let conflicting_metadata = staging_dir(1, 0).join(crate::disk::STORAGE_FORMAT_FILE);
|
||||
tokio::fs::create_dir_all(conflicting_metadata.parent().expect("metadata parent"))
|
||||
.await
|
||||
.expect("create conflicting pool staging");
|
||||
tokio::fs::write(&conflicting_metadata, b"unreadable metadata")
|
||||
.await
|
||||
.expect("write conflicting metadata");
|
||||
assert!(
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.is_err(),
|
||||
"no pool may be cleaned until every pool proves absence"
|
||||
);
|
||||
for disk_index in 0..3 {
|
||||
assert!(staging_dir(0, disk_index).join("part.1").exists());
|
||||
}
|
||||
tokio::fs::remove_file(conflicting_metadata)
|
||||
.await
|
||||
.expect("remove conflicting fixture");
|
||||
let (paused, release) = crate::store::multipart::pause_orphan_multipart_cleanup_for_test(&bucket);
|
||||
let recovery_store = store.clone();
|
||||
let recovery_bucket = bucket.clone();
|
||||
let recovery_upload = upload.upload_id.clone();
|
||||
let mut recovery = tokio::spawn(async move {
|
||||
recovery_store
|
||||
.abort_multipart_upload(&recovery_bucket, object, &recovery_upload, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
tokio::select! {
|
||||
paused = paused => paused.expect("cleanup owner announces arrival"),
|
||||
result = &mut recovery => panic!("recovery returned before physical cleanup: {result:?}"),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("recovery should reach physical cleanup");
|
||||
recovery.abort();
|
||||
assert!(recovery.await.expect_err("request waiter is cancellable").is_cancelled());
|
||||
let upload_lock = store.pools[0]
|
||||
.get_disks_by_key(object)
|
||||
.new_ns_lock(crate::disk::RUSTFS_META_MULTIPART_BUCKET, &format!("{upload_sha}/{upload_uuid}"))
|
||||
.await
|
||||
.expect("same upload namespace");
|
||||
let mut upload_writer = Box::pin(upload_lock.get_write_lock(Duration::from_secs(10)));
|
||||
let mut bucket_writer = Box::pin(store.acquire_bucket_lifecycle_write_lock(&bucket));
|
||||
assert!(
|
||||
futures::poll!(&mut upload_writer).is_pending(),
|
||||
"cancelled recovery must retain the upload lock"
|
||||
);
|
||||
assert!(
|
||||
futures::poll!(&mut bucket_writer).is_pending(),
|
||||
"cancelled recovery must retain the real bucket guard"
|
||||
);
|
||||
release.send(()).expect("allow physical cleanup to finish");
|
||||
let (upload_guard, bucket_guard) =
|
||||
tokio::time::timeout(Duration::from_secs(10), async { tokio::join!(upload_writer, bucket_writer) })
|
||||
.await
|
||||
.expect("both guards should release after physical cleanup");
|
||||
drop(upload_guard.expect("upload writer resumes"));
|
||||
drop(bucket_guard.expect("bucket writer resumes"));
|
||||
for pool_index in 0..2 {
|
||||
for disk_index in 0..4 {
|
||||
assert!(!staging_dir(pool_index, disk_index).exists(), "every checked staging path must be absent");
|
||||
}
|
||||
}
|
||||
store
|
||||
.ensure_decommission_multipart_uploads_drained_for_test(0)
|
||||
.await
|
||||
.expect("physical decommission gate should open after recovery");
|
||||
assert!(
|
||||
matches!(
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await,
|
||||
Err(StorageError::InvalidUploadID(_, _, _))
|
||||
),
|
||||
"an absent upload must not claim to have recovered anything"
|
||||
);
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn repeated_abort_keeps_no_such_upload_while_a_disk_is_offline() {
|
||||
use sha2::Digest;
|
||||
|
||||
let temp_dir = tempfile::tempdir().expect("create offline-disk recovery store dir");
|
||||
let (_ctx, store, shutdown) =
|
||||
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-multipart-offline", &[4, 4])).await;
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("orphan-multipart-{}", uuid::Uuid::new_v4());
|
||||
let object = "offline-object.bin";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create bucket");
|
||||
let incarnation = store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation");
|
||||
let lifecycle_guard = store.acquire_bucket_lifecycle_read_lock(&bucket).await.expect("bucket guard");
|
||||
let mut upload_opts = ObjectOptions {
|
||||
expected_bucket_incarnation_id: Some(incarnation),
|
||||
..Default::default()
|
||||
};
|
||||
upload_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
|
||||
let upload = store.pools[0]
|
||||
.new_multipart_upload(&bucket, object, &upload_opts)
|
||||
.await
|
||||
.expect("create source upload");
|
||||
drop(lifecycle_guard);
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("initial abort");
|
||||
|
||||
let upload_sha =
|
||||
hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
|
||||
let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
|
||||
let staging_dir = |disk_index: usize| {
|
||||
temp_dir
|
||||
.path()
|
||||
.join(format!("pool0/set0/disk{disk_index}"))
|
||||
.join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
|
||||
.join(&upload_sha)
|
||||
.join(&upload_uuid)
|
||||
};
|
||||
for disk_index in 0..3 {
|
||||
let path = staging_dir(disk_index);
|
||||
tokio::fs::create_dir_all(&path).await.expect("create late part directory");
|
||||
tokio::fs::write(path.join("part.1"), b"late part after abort")
|
||||
.await
|
||||
.expect("write late part fixture");
|
||||
}
|
||||
|
||||
// One offline disk means absence cannot be proven. The idempotent Abort
|
||||
// must keep answering NoSuchUpload instead of surfacing a disk error,
|
||||
// and must not delete anything.
|
||||
let set = store.pools[0].get_disks_by_key(object);
|
||||
let offline = set.disks.write().await[3].take();
|
||||
assert!(offline.is_some(), "fixture disk must have been online");
|
||||
assert!(
|
||||
matches!(
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await,
|
||||
Err(StorageError::InvalidUploadID(_, _, _))
|
||||
),
|
||||
"unprovable absence must keep NoSuchUpload while a disk is offline"
|
||||
);
|
||||
for disk_index in 0..3 {
|
||||
assert!(
|
||||
staging_dir(disk_index).join("part.1").exists(),
|
||||
"no staging may be deleted while a disk is offline"
|
||||
);
|
||||
}
|
||||
|
||||
set.disks.write().await[3] = offline;
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("recovery resumes once every disk is observable");
|
||||
for disk_index in 0..4 {
|
||||
assert!(!staging_dir(disk_index).exists(), "recovered staging must be absent");
|
||||
}
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn orphan_multipart_recovery_preserves_current_bucket_incarnation() {
|
||||
use sha2::Digest;
|
||||
|
||||
let temp_dir = tempfile::tempdir().expect("create bucket lifetime recovery store dir");
|
||||
let (_ctx, store, shutdown) =
|
||||
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "orphan-bucket-lifetime", &[4])).await;
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
|
||||
let bucket = format!("orphan-lifetime-{}", uuid::Uuid::new_v4());
|
||||
let object = "object.bin";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create original bucket");
|
||||
let original_incarnation = store.bucket_incarnation_id(&bucket).await.expect("original incarnation");
|
||||
let upload = store
|
||||
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("create original upload");
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("abort original upload");
|
||||
store
|
||||
.delete_bucket(&bucket, &crate::storage_api_contracts::bucket::DeleteBucketOptions::default())
|
||||
.await
|
||||
.expect("delete original bucket");
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("recreate bucket");
|
||||
assert_ne!(store.bucket_incarnation_id(&bucket).await.expect("new incarnation"), original_incarnation);
|
||||
let mut reader = PutObjReader::from_vec(b"current bucket object".to_vec());
|
||||
store
|
||||
.put_object(&bucket, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("write current bucket object");
|
||||
let current_upload = store
|
||||
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("create current incarnation upload for the same key");
|
||||
|
||||
let upload_sha =
|
||||
hex_simd::encode_to_string(sha2::Sha256::digest(format!("{bucket}/{object}").as_bytes()), hex_simd::AsciiCase::Lower);
|
||||
let upload_uuid = crate::runtime::sources::upload_uuid_suffix(&upload.upload_id);
|
||||
let leftover = temp_dir
|
||||
.path()
|
||||
.join("pool0/set0/disk0")
|
||||
.join(crate::disk::RUSTFS_META_MULTIPART_BUCKET)
|
||||
.join(upload_sha)
|
||||
.join(upload_uuid)
|
||||
.join("part.1");
|
||||
tokio::fs::create_dir_all(leftover.parent().expect("staging parent"))
|
||||
.await
|
||||
.expect("create old incarnation staging");
|
||||
tokio::fs::write(&leftover, b"previous bucket lifetime")
|
||||
.await
|
||||
.expect("write previous incarnation fixture");
|
||||
for opts in [
|
||||
ObjectOptions {
|
||||
expected_bucket_incarnation_id: Some(original_incarnation),
|
||||
..Default::default()
|
||||
},
|
||||
ObjectOptions {
|
||||
expected_bucket_incarnation_id: Some(uuid::Uuid::nil()),
|
||||
..Default::default()
|
||||
},
|
||||
] {
|
||||
assert!(
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &opts)
|
||||
.await
|
||||
.is_err(),
|
||||
"a stale or nil current-bucket identity must fail closed"
|
||||
);
|
||||
assert!(leftover.exists(), "failed bucket lifetime proof must preserve the path");
|
||||
}
|
||||
store
|
||||
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("exact metadata-less staging from an old incarnation is unreferenced");
|
||||
assert!(!leftover.exists());
|
||||
store
|
||||
.get_multipart_info(&bucket, object, ¤t_upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("recovery must preserve the current incarnation's active upload");
|
||||
let mut reader = store
|
||||
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("current object must remain readable");
|
||||
let mut body = Vec::new();
|
||||
reader.stream.read_to_end(&mut body).await.expect("read current object");
|
||||
assert_eq!(body, b"current bucket object");
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn active_multipart_upload_routes_before_faulted_suspended_source() {
|
||||
@@ -18764,6 +19097,70 @@ mod tests {
|
||||
assert_eq!(body, b"new-generation");
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial_test::serial]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn cancelled_ecstore_put_part_keeps_bucket_lifecycle_lock_through_commit() {
|
||||
for pause in [
|
||||
crate::set_disk::MultipartCommitPause::PutPartAfterCapacityAdmission,
|
||||
crate::set_disk::MultipartCommitPause::PutPartAfterRename,
|
||||
] {
|
||||
let temp_dir = tempfile::tempdir().expect("create cancelled multipart lifecycle store dir");
|
||||
let (_ctx, store, shutdown) =
|
||||
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "cancelled-multipart-lifecycle", &[4, 4]))
|
||||
.await;
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
|
||||
let bucket = format!("cancelled-part-lifecycle-{}", Uuid::new_v4());
|
||||
let object = "part.bin";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create bucket for multipart lifecycle cancellation");
|
||||
let upload = store
|
||||
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("create upload through the owning ECStore");
|
||||
let barrier = crate::set_disk::MultipartCommitBarrier::install(&bucket, object, pause);
|
||||
let put_store = Arc::clone(&store);
|
||||
let put_bucket = bucket.clone();
|
||||
let put_upload_id = upload.upload_id.clone();
|
||||
let put = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![0x45; 4096]);
|
||||
put_store
|
||||
.put_object_part(&put_bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
barrier.wait_until_paused().await;
|
||||
put.abort();
|
||||
assert!(
|
||||
put.await
|
||||
.expect_err("the ECStore request waiter must remain cancellable")
|
||||
.is_cancelled()
|
||||
);
|
||||
|
||||
let mut writer = Box::pin(store.acquire_bucket_lifecycle_write_lock(&bucket));
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), &mut writer).await.is_err(),
|
||||
"canceling UploadPart at {pause:?} must not release its bucket lifecycle owner before the commit drains"
|
||||
);
|
||||
barrier.release();
|
||||
let lifecycle_guard = tokio::time::timeout(Duration::from_secs(10), writer)
|
||||
.await
|
||||
.expect("the bucket lifecycle writer must resume after the part commit drains")
|
||||
.expect("acquire the bucket lifecycle write lock after cancellation");
|
||||
drop(lifecycle_guard);
|
||||
|
||||
let parts = store
|
||||
.list_object_parts(&bucket, object, &upload.upload_id, None, 100, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the cancelled request's owned part commit must finish successfully");
|
||||
assert_eq!(parts.parts.len(), 1, "the commit must publish exactly the uploaded part at {pause:?}");
|
||||
assert_eq!(parts.parts[0].part_num, 1);
|
||||
assert_eq!(parts.parts[0].size, 4096);
|
||||
shutdown.cancel();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(storage_class_env)]
|
||||
async fn pre_upgrade_multipart_upload_can_complete_and_abort_in_the_same_bucket_lifetime() {
|
||||
|
||||
@@ -32,6 +32,42 @@ fn data_movement_multipart_discovery_counts() -> &'static std::sync::Mutex<std::
|
||||
DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
type OrphanMultipartCleanupPause = (tokio::sync::oneshot::Sender<()>, tokio::sync::oneshot::Receiver<()>);
|
||||
|
||||
#[cfg(test)]
|
||||
static ORPHAN_MULTIPART_CLEANUP_PAUSES: std::sync::OnceLock<
|
||||
std::sync::Mutex<std::collections::HashMap<String, OrphanMultipartCleanupPause>>,
|
||||
> = std::sync::OnceLock::new();
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn pause_orphan_multipart_cleanup_for_test(
|
||||
bucket: &str,
|
||||
) -> (tokio::sync::oneshot::Receiver<()>, tokio::sync::oneshot::Sender<()>) {
|
||||
let (arrived_tx, arrived_rx) = tokio::sync::oneshot::channel();
|
||||
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
|
||||
let previous = ORPHAN_MULTIPART_CLEANUP_PAUSES
|
||||
.get_or_init(Default::default)
|
||||
.lock()
|
||||
.expect("orphan cleanup pause mutex should not poison")
|
||||
.insert(bucket.to_owned(), (arrived_tx, release_rx));
|
||||
assert!(previous.is_none(), "each bucket has at most one orphan cleanup pause");
|
||||
(arrived_rx, release_tx)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
async fn pause_orphan_multipart_cleanup(bucket: &str) {
|
||||
let pause = ORPHAN_MULTIPART_CLEANUP_PAUSES
|
||||
.get_or_init(Default::default)
|
||||
.lock()
|
||||
.expect("orphan cleanup pause mutex should not poison")
|
||||
.remove(bucket);
|
||||
if let Some((arrived, release)) = pause {
|
||||
let _ = arrived.send(());
|
||||
let _ = release.await;
|
||||
}
|
||||
}
|
||||
|
||||
fn decommission_multipart_target_clear_pending(opts: &ObjectOptions, target: Option<&ObjectInfo>) -> Result<bool> {
|
||||
let expected_mod_time = opts.mod_time.ok_or_else(|| Error::DecommissionCapacityBlocked {
|
||||
message: "multipart cleanup cannot prove exact target absence without a modification time".to_string(),
|
||||
@@ -654,7 +690,10 @@ impl ECStore {
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<PartInfo> {
|
||||
check_put_object_part_args(bucket, object, upload_id)?;
|
||||
let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
||||
let (mut opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
||||
if let Some(guard) = bucket_lifecycle_guard {
|
||||
opts.bucket_lifecycle_lock_owner = Some(Arc::new(guard));
|
||||
}
|
||||
if !self.single_pool() {
|
||||
opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
|
||||
}
|
||||
@@ -750,19 +789,12 @@ impl ECStore {
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<()> {
|
||||
check_abort_multipart_args(bucket, object, upload_id)?;
|
||||
let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
||||
let opts = &opts;
|
||||
|
||||
// TODO(backlog): defer DeleteUploadID to background for faster abort response
|
||||
|
||||
if self.single_pool() {
|
||||
return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
|
||||
}
|
||||
let (opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
|
||||
|
||||
for pool_idx in self.existing_multipart_pool_order().await {
|
||||
let pool = &self.pools[pool_idx];
|
||||
|
||||
let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await {
|
||||
let err = match pool.abort_multipart_upload(bucket, object, upload_id, &opts).await {
|
||||
Ok(_) => return Ok(()),
|
||||
Err(err) => {
|
||||
//
|
||||
@@ -775,7 +807,92 @@ impl ECStore {
|
||||
}
|
||||
}
|
||||
|
||||
Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
|
||||
self.recover_orphan_multipart_upload(bucket, object, upload_id, opts, bucket_lifecycle_guard)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn recover_orphan_multipart_upload(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
upload_id: &str,
|
||||
opts: ObjectOptions,
|
||||
bucket_lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
|
||||
) -> Result<()> {
|
||||
let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
|
||||
if self.ctx.lock_manager().is_disabled() {
|
||||
return Err(invalid_upload());
|
||||
}
|
||||
let Some(bucket_lifecycle_guard) = bucket_lifecycle_guard else {
|
||||
// A signal borrowed from an enclosing operation cannot keep its lock
|
||||
// alive when the request waiter disappears during physical cleanup.
|
||||
return Err(invalid_upload());
|
||||
};
|
||||
let Some(expected_incarnation) = opts.expected_bucket_incarnation_id.filter(|id| !id.is_nil()) else {
|
||||
return Err(invalid_upload());
|
||||
};
|
||||
// Recovery is a side path of an idempotent Abort. Until the first
|
||||
// deletion is issued, anything that stops staging absence from being
|
||||
// proven (an offline disk, lock contention, partially overlapping lock
|
||||
// domains, unreadable bucket metadata) keeps the upload reported as
|
||||
// gone, exactly as before recovery existed; only failures inside the
|
||||
// deletion task are request failures, because a retry can finish them.
|
||||
let (metadata, persisted) =
|
||||
match crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await {
|
||||
Ok(loaded) => loaded,
|
||||
Err(error) => {
|
||||
debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: bucket metadata unavailable");
|
||||
return Err(invalid_upload());
|
||||
}
|
||||
};
|
||||
if !persisted
|
||||
|| !metadata.bucket_creation_committed
|
||||
|| !metadata.bucket_incarnation_sidecar
|
||||
|| metadata.bucket_incarnation_id != expected_incarnation
|
||||
{
|
||||
return Err(invalid_upload());
|
||||
}
|
||||
|
||||
// Lock order: bucket lifecycle, then upload locks in pool-index order.
|
||||
// Prove metadata absence across every pool before deleting any staging.
|
||||
let mut cleanup_guards = Vec::with_capacity(self.pools.len());
|
||||
for pool in &self.pools {
|
||||
let guard = match pool
|
||||
.get_disks_by_key(object)
|
||||
.lock_orphan_multipart_cleanup(bucket, object, upload_id, &opts, &cleanup_guards)
|
||||
.await
|
||||
{
|
||||
Ok(guard) => guard,
|
||||
Err(error) => {
|
||||
if !is_err_invalid_upload_id(&error) {
|
||||
debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: staging absence not proven");
|
||||
}
|
||||
return Err(invalid_upload());
|
||||
}
|
||||
};
|
||||
cleanup_guards.push(guard);
|
||||
}
|
||||
let bucket = bucket.to_owned();
|
||||
let object = object.to_owned();
|
||||
let upload_id = upload_id.to_owned();
|
||||
tokio::spawn(async move {
|
||||
// Own both the real bucket guard and all upload guards until every
|
||||
// disk deletion settles, including after cancellation of the waiter.
|
||||
let _bucket_lifecycle_guard = bucket_lifecycle_guard;
|
||||
#[cfg(test)]
|
||||
pause_orphan_multipart_cleanup(&bucket).await;
|
||||
let mut recovered = false;
|
||||
for guard in &cleanup_guards {
|
||||
recovered |= guard.delete(&bucket, &object, &opts).await?;
|
||||
}
|
||||
if recovered {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(StorageError::InvalidUploadID(bucket, object, upload_id))
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|error| Error::other_with_context("orphan multipart cleanup task failed", error))?
|
||||
}
|
||||
|
||||
pub(crate) async fn abort_multipart_upload_for_data_movement(
|
||||
|
||||
Reference in New Issue
Block a user