gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 03:59:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): preserve known preflight rename rejections

Browse Source
This commit is contained in:
overtrue
2026-09-05 12:08:28 +08:00
parent 2cb0425380
commit 0658b228e1
5 changed files with 1324 additions and 1093 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/disk_store.rs
+77 -8
View File
@@ -317,6 +317,22 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
dst_path: &str,
external_guard: Option<Arc<dyn Send + Sync>>,
) -> Result<RenameDataResp> {
self.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
.await
.result
}
}
impl LocalDiskWrapper {
pub(in crate::disk) async fn rename_data_observed(
&self,
src_volume: &str,
src_path: &str,
fi: &FileInfo,
dst_volume: &str,
dst_path: &str,
external_guard: Option<Arc<dyn Send + Sync>>,
) -> super::RenameDataObservation {
let operation = self.clone();
let src_volume = src_volume.to_owned();
let src_path = src_path.to_owned();
@@ -333,22 +349,35 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
} else {
get_max_timeout_duration()
};
run_owned_mutation(external_guard, move || async move {
operation
let observed = run_owned_mutation(external_guard, move || async move {
let mut preflight_rejection = None;
let result = operation
.track_disk_health_mutation(
"rename_data",
DiskMetricMutation::Write,
|| async {
operation
.disk
.rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path)
.await
// Preserve the former DiskAPI future's single boxing boundary.
let observed =
Box::pin(
operation
.disk
.rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path),
)
.await;
preflight_rejection = observed.preflight_rejection;
observed.result
},
timeout_duration,
)
.await
.await;
// Health tracking must observe the real disk error, not an Ok tuple.
Ok(super::RenameDataObservation {
result,
preflight_rejection,
})
})
.await
.await;
observed.unwrap_or_else(|error| super::RenameDataObservation::unknown(Err(error)))
}
}
@@ -2588,6 +2617,46 @@ mod tests {
assert_eq!(wrapper.metrics_snapshot().api_calls.get("unknown"), Some(&1));
}
#[tokio::test]
async fn rename_preflight_evidence_preserves_health_errors_and_owned_reply() {
for source_exists in [false, true] {
for guarded in [false, true] {
let dir = tempfile::tempdir().expect("temp dir should be created");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8"))
.expect("endpoint should parse");
let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
if source_exists {
disk.make_volume("source").await.expect("source volume should exist");
}
let wrapper = LocalDiskWrapper::new(disk, false);
let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let external_guard = guarded.then(|| Arc::new(DropProbe(Arc::clone(&drops))) as Arc<dyn Send + Sync>);
let mut file_info = FileInfo::new("object", 1, 0);
file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
file_info.erasure.index = 1;
let observed = wrapper
.rename_data_observed("source", "object", &file_info, "missing-destination", "object", external_guard)
.await;
assert!(observed.rejected_before_publication(), "normal access rejection must carry proof");
assert!(matches!(observed.result, Err(DiskError::VolumeNotFound)));
let snapshot = wrapper.metrics_snapshot();
assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
assert_eq!(snapshot.total_writes, 0, "health tracking must not observe the rejection as Ok");
assert_eq!(drops.load(Ordering::SeqCst), usize::from(guarded));
wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
let observed = wrapper
.rename_data_observed("source", "object", &file_info, "missing-destination", "object", None)
.await;
assert!(!observed.rejected_before_publication(), "wrapper errors carry no local preflight proof");
assert!(matches!(observed.result, Err(DiskError::FaultyDisk)));
let snapshot = wrapper.metrics_snapshot();
assert_eq!(snapshot.total_errors_availability, 1);
assert_eq!(snapshot.total_writes, 0);
}
}
}
#[tokio::test]
async fn local_disk_health_wrapper_counts_returned_availability_errors() {
let dir = tempfile::tempdir().expect("temp dir should be created");
crates/ecstore/src/disk/local.rs
+998 -960
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/disk/mod.rs
+49
View File
@@ -75,6 +75,25 @@ use time::OffsetDateTime;
use tokio::io::{AsyncRead, AsyncWrite};
use uuid::Uuid;
/// Local preflight evidence stays outside DiskAPI and the RPC response format.
pub(crate) struct RenameDataObservation {
pub(crate) result: Result<RenameDataResp>,
preflight_rejection: Option<local::LocalRenamePreflightRejection>,
}
impl RenameDataObservation {
fn unknown(result: Result<RenameDataResp>) -> Self {
Self {
result,
preflight_rejection: None,
}
}
pub(crate) fn rejected_before_publication(&self) -> bool {
self.result.is_err() && self.preflight_rejection.is_some()
}
}
const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/";
pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token";
@@ -711,6 +730,36 @@ impl Disk {
.await
}
pub(crate) async fn rename_data_borrowed_with_fence_observed(
&self,
src_volume: &str,
src_path: &str,
fi: &FileInfo,
dst_volume: &str,
dst_path: &str,
scanner_publication_lease_token: Option<Uuid>,
) -> RenameDataObservation {
match self {
Disk::Local(local_disk) => {
local_disk
.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None)
.await
}
Disk::Remote(remote_disk) => RenameDataObservation::unknown(
remote_disk
.rename_data_borrowed_with_fence(
src_volume,
src_path,
fi,
dst_volume,
dst_path,
scanner_publication_lease_token,
)
.await,
),
}
}
pub(crate) async fn rename_data_borrowed_with_fence(
&self,
src_volume: &str,
crates/ecstore/src/set_disk/core/io_primitives.rs
+55 -27
View File
@@ -3841,9 +3841,17 @@ pub(in crate::set_disk) struct RenameTailOutcome {
const EVENT_SET_DISK_RENAME_ROLLBACK: &str = "set_disk_rename_rollback";
#[derive(Clone, Copy)]
enum RenameDispatchState {
NotDispatched,
RejectedBeforePublication,
MayHavePublished,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum RenameRollbackOutcome {
NotAttempted(DiskError),
RejectedBeforePublication(DiskError),
Indeterminate(DiskError),
Succeeded,
Failed(DiskError),
@@ -3855,6 +3863,7 @@ impl RenameRollbackOutcome {
fn stage(&self) -> &'static str {
match self {
Self::NotAttempted(_) => "rename_not_dispatched",
Self::RejectedBeforePublication(_) => "rename_rejected_before_publication",
Self::Indeterminate(_) => "rename_indeterminate",
Self::Succeeded => "undo_succeeded",
Self::Failed(_) => "undo_failed",
@@ -3940,14 +3949,14 @@ async fn rollback_failed_rename(
disks: &[Option<DiskStore>],
file_infos: Vec<FileInfo>,
errs: &[Option<DiskError>],
dispatched: &[bool],
dispatch_states: &[RenameDispatchState],
rollback_dirs: &[Option<Uuid>],
dst: (&str, &str),
receipt: Option<RenameRollbackReceipt>,
) {
let owned_disks = disks.to_vec();
let owned_errs = errs.to_vec();
let owned_dispatched = dispatched.to_vec();
let owned_dispatch_states = dispatch_states.to_vec();
let owned_dirs = rollback_dirs.to_vec();
let owned_dst = (dst.0.to_string(), dst.1.to_string());
let coordinator_failure_receipt = receipt.clone();
@@ -3956,7 +3965,7 @@ async fn rollback_failed_rename(
let rollback = tokio::spawn(async move {
let disks = owned_disks.as_slice();
let errs = owned_errs.as_slice();
let dispatched = owned_dispatched.as_slice();
let dispatch_states = owned_dispatch_states.as_slice();
let rollback_dirs = owned_dirs.as_slice();
let dst = (owned_dst.0.as_str(), owned_dst.1.as_str());
let mut file_infos = file_infos;
@@ -3967,8 +3976,13 @@ async fn rollback_failed_rename(
for (disk_index, disk) in disks.iter().enumerate() {
let rollback_dir = rollback_dirs[disk_index];
let outcome = match &errs[disk_index] {
Some(err) if dispatched[disk_index] => RenameRollbackOutcome::Indeterminate(err.clone()),
Some(err) => RenameRollbackOutcome::NotAttempted(err.clone()),
Some(err) => match dispatch_states[disk_index] {
RenameDispatchState::NotDispatched => RenameRollbackOutcome::NotAttempted(err.clone()),
RenameDispatchState::RejectedBeforePublication => {
RenameRollbackOutcome::RejectedBeforePublication(err.clone())
}
RenameDispatchState::MayHavePublished => RenameRollbackOutcome::Indeterminate(err.clone()),
},
None => RenameRollbackOutcome::Failed(DiskError::DiskNotFound),
};
outcomes.push(RenameRollbackDiskOutcome {
@@ -4545,7 +4559,7 @@ impl SetDisks {
let file_info = file_info.clone();
let successful_rename_completion_rank = successful_rename_completion_rank.clone();
tasks.spawn(async move {
let mut dispatched = false;
let mut dispatch_state = RenameDispatchState::NotDispatched;
let result = std::panic::AssertUnwindSafe(async {
#[allow(clippy::let_unit_value)]
let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object);
@@ -4571,9 +4585,9 @@ impl SetDisks {
}
let disk_wait_started = rustfs_io_metrics::put_stage_timer();
dispatched = true;
let result = disk
.rename_data_borrowed_with_fence(
dispatch_state = RenameDispatchState::MayHavePublished;
let observed = disk
.rename_data_borrowed_with_fence_observed(
&src_bucket,
&src_object,
&file_info,
@@ -4582,6 +4596,8 @@ impl SetDisks {
scanner_publication_lease_token,
)
.await;
let rejected_before_publication = observed.rejected_before_publication();
let result = observed.result;
#[cfg(test)]
if result.is_ok() {
rollback_fault_injection::after_rename(&dst_object, i)?;
@@ -4607,11 +4623,14 @@ impl SetDisks {
};
rustfs_io_metrics::record_put_rename_disk_wait_completion(position, duration_ms);
}
if rejected_before_publication {
dispatch_state = RenameDispatchState::RejectedBeforePublication;
}
result
})
.catch_unwind()
.await;
(i, dispatched, result)
(i, dispatch_state, result)
});
}
@@ -4622,7 +4641,7 @@ impl SetDisks {
let mut results_seen = 0usize;
let mut errs = vec![Some(DiskError::DiskNotFound); disk_count];
// Missing task results cannot prove that a disk mutation never ran.
let mut dispatched = vec![true; disk_count];
let mut dispatch_states = vec![RenameDispatchState::MayHavePublished; disk_count];
let mut disk_versions = vec![None; disk_count];
let mut data_dirs = vec![None; disk_count];
let mut cleanup_data_dirs = vec![None; disk_count];
@@ -4633,8 +4652,8 @@ impl SetDisks {
while let Some(joined) = tasks.join_next().await {
results_seen += 1;
match joined {
Ok((idx, was_dispatched, Ok(Ok(res)))) => {
dispatched[idx] = was_dispatched;
Ok((idx, dispatch_state, Ok(Ok(res)))) => {
dispatch_states[idx] = dispatch_state;
data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir);
cleanup_data_dirs[idx] = res.cleanup_data_dir;
disk_versions[idx] = res.sign;
@@ -4642,12 +4661,12 @@ impl SetDisks {
errs[idx] = None;
success_count += 1;
}
Ok((idx, was_dispatched, Ok(Err(err)))) => {
dispatched[idx] = was_dispatched;
Ok((idx, dispatch_state, Ok(Err(err)))) => {
dispatch_states[idx] = dispatch_state;
errs[idx] = Some(err);
}
Ok((idx, was_dispatched, Err(_))) => {
dispatched[idx] = was_dispatched;
Ok((idx, dispatch_state, Err(_))) => {
dispatch_states[idx] = dispatch_state;
errs[idx] = Some(DiskError::Unexpected);
fanout_panic += 1;
}
@@ -4698,7 +4717,7 @@ impl SetDisks {
&coordinator_disks,
file_infos,
&errs,
&dispatched,
&dispatch_states,
&data_dirs,
(&fanout_dst_bucket, &fanout_dst_object),
rollback_receipt,
@@ -4914,7 +4933,7 @@ impl SetDisks {
let publication_scope = scanner_publication_commit_scope.clone();
async move {
let mut dispatched = false;
let mut dispatch_state = RenameDispatchState::NotDispatched;
let result = std::panic::AssertUnwindSafe(async {
// Test-only introspection guard: counts this operation as
// in-flight for the whole body. Compiles to `()` in production.
@@ -4956,9 +4975,9 @@ impl SetDisks {
}
let disk_wait_started = rustfs_io_metrics::put_stage_timer();
dispatched = true;
let result = disk
.rename_data_borrowed_with_fence(
dispatch_state = RenameDispatchState::MayHavePublished;
let observed = disk
.rename_data_borrowed_with_fence_observed(
&src_bucket,
&src_object,
file_info,
@@ -4967,6 +4986,8 @@ impl SetDisks {
scanner_publication_lease_token,
)
.await;
let rejected_before_publication = observed.rejected_before_publication();
let result = observed.result;
#[cfg(test)]
if result.is_ok() {
rollback_fault_injection::after_rename(&dst_object, i)?;
@@ -4992,11 +5013,14 @@ impl SetDisks {
};
rustfs_io_metrics::record_put_rename_disk_wait_completion(position, duration_ms);
}
if rejected_before_publication {
dispatch_state = RenameDispatchState::RejectedBeforePublication;
}
result
})
.catch_unwind()
.await;
(dispatched, result)
(dispatch_state, result)
}
});
let results = join_all(futures).await;
@@ -5043,9 +5067,9 @@ impl SetDisks {
);
}
let mut dispatched = Vec::with_capacity(results.len());
for (idx, (was_dispatched, result)) in results.iter().enumerate() {
dispatched.push(*was_dispatched);
let mut dispatch_states = Vec::with_capacity(results.len());
for (idx, (dispatch_state, result)) in results.iter().enumerate() {
dispatch_states.push(*dispatch_state);
match result {
Ok(Ok(res)) => {
data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir);
@@ -5111,7 +5135,7 @@ impl SetDisks {
disks,
file_infos,
&errs,
&dispatched,
&dispatch_states,
&data_dirs,
(&dst_bucket, &dst_object),
rollback_receipt,
@@ -7075,6 +7099,7 @@ pub(in crate::set_disk) mod rollback_fault_injection {
Io,
Panic,
IoAfterRename,
VolumeNotFoundAfterRename,
PanicAfterRename,
CoordinatorPanic,
}
@@ -7123,6 +7148,7 @@ pub(in crate::set_disk) mod rollback_fault_injection {
.copied();
match fault {
Some((target, Fault::IoAfterRename)) if target == disk_index => Err(DiskError::FaultyDisk),
Some((target, Fault::VolumeNotFoundAfterRename)) if target == disk_index => Err(DiskError::VolumeNotFound),
Some((target, Fault::PanicAfterRename)) if target == disk_index => panic!("injected panic after rename mutation"),
_ => Ok(()),
}
@@ -10947,6 +10973,7 @@ mod tests {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
for fault in [
rollback_fault_injection::Fault::IoAfterRename,
rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
rollback_fault_injection::Fault::PanicAfterRename,
] {
let bucket = "rename-tail-unknown";
@@ -11014,6 +11041,7 @@ mod tests {
for fault in [
rollback_fault_injection::Fault::Io,
rollback_fault_injection::Fault::IoAfterRename,
rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
rollback_fault_injection::Fault::PanicAfterRename,
rollback_fault_injection::Fault::CoordinatorPanic,
] {
crates/ecstore/src/set_disk/ops/object.rs
+145 -98
View File
@@ -17505,27 +17505,69 @@ mod put_object_tmp_cleanup_tests {
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn put_object_failure_cleans_tmp_workspace_inline() {
let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "tmp-clean-missing-bucket";
let object = "orphan-object";
let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
let writer = Arc::clone(&set_disks);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
writer
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
write_completion,
..Default::default()
},
)
.await
});
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
.await
.expect("missing-bucket PUT must stage before rename");
let staged = non_trash_tmp_entries(&temp_dirs).await;
assert_eq!(staged.len(), 4, "every disk must have a staged workspace before rejection");
for workspace in staged {
let mut entries = tokio::fs::read_dir(&workspace)
.await
.expect("staged workspace should be readable");
let mut shards = 0;
while let Some(entry) = entries.next_entry().await.expect("staged data directory should be readable") {
if entry.file_type().await.expect("staged entry type").is_dir() {
let part = tokio::fs::metadata(entry.path().join("part.1"))
.await
.expect("staging must contain an actual erasure shard");
assert!(part.len() > 0, "the shard must be written before the missing-bucket failure");
shards += 1;
}
}
assert_eq!(shards, 1);
}
assert!(temp_dirs.iter().all(|dir| !dir.path().join(bucket).exists()));
barrier.release();
let err = tokio::time::timeout(Duration::from_secs(30), put)
.await
.expect("missing-bucket PUT must finish")
.expect("PUT task should join")
.expect_err("put_object into a missing bucket volume must fail");
assert!(matches!(err, StorageError::VolumeNotFound), "original disk error expected: {err}");
// The bucket volume is never created, so the shards are written into
// the tmp workspace and the commit fails at rename_data with a quorum
// error — exercising the failure-path cleanup.
let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
let err = set_disks
.put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default())
.await
.expect_err("put_object into a missing bucket volume must fail");
// No polling: the failure path must clean the tmp workspace inline,
// before put_object returns (backlog#864 / backlog#898 hardening).
let leftovers = non_trash_tmp_entries(&temp_dirs).await;
assert!(
leftovers.is_empty(),
"failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
);
drop(temp_dirs);
// No polling: known pre-publication rejection must clean staging
// inline, before PUT returns (backlog#864 / backlog#898).
let leftovers = non_trash_tmp_entries(&temp_dirs).await;
assert!(
leftovers.is_empty(),
"failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
);
}
})
.await;
}
#[tokio::test]
@@ -18373,87 +18415,92 @@ mod put_object_tmp_cleanup_tests {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
let (dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "put-incomplete-undo";
let object = "incomplete-undo-object";
make_completion_test_bucket(&disks, bucket).await;
let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
set.put_object(
bucket,
object,
&mut old_reader,
&ObjectOptions {
write_completion: WriteCompletion::TailDrained,
..Default::default()
},
)
.await
.expect("old generation should be completely committed");
wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await;
let old = disks[0]
.read_version("", bucket, object, "", &ReadOptions::default())
for fault in [
rollback_fault_injection::Fault::Io,
rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
] {
let (dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "put-incomplete-undo";
let object = "incomplete-undo-object";
make_completion_test_bucket(&disks, bucket).await;
let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
set.put_object(
bucket,
object,
&mut old_reader,
&ObjectOptions {
write_completion: WriteCompletion::TailDrained,
..Default::default()
},
)
.await
.expect("old metadata must be readable");
let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory");
let tasks = rename_fanout_barrier::observe_tasks(object);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
let _undo_fault = rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::Io);
let writer = Arc::clone(&set);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
writer
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
write_completion,
..Default::default()
},
)
.expect("old generation should be completely committed");
wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await;
let old = disks[0]
.read_version("", bucket, object, "", &ReadOptions::default())
.await
});
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
.await
.expect("overwrite must enter the actual rename fan-out before failure injection");
barrier.release();
let err = tokio::time::timeout(Duration::from_secs(30), put)
.await
.expect("incomplete undo must return without hanging")
.expect("PUT task should join")
.expect_err("two renamed disks cannot satisfy write quorum three");
assert!(
matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
"original quorum error expected: {err}"
);
assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return");
let leftovers = non_trash_tmp_entries(&dirs).await;
assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery");
let backups = dirs
.iter()
.filter(|dir| {
dir.path()
.join(bucket)
.join(object)
.join(old_data_dir.to_string())
.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
.exists()
})
.count();
assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup");
// The remaining three disks still serve the old generation;
// the failed minority must never become an acknowledged write.
let mut read = set
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("old generation must remain readable after incomplete rollback");
let mut body = Vec::new();
read.stream
.read_to_end(&mut body)
.await
.expect("old generation should stream");
assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]);
.expect("old metadata must be readable");
let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory");
let tasks = rename_fanout_barrier::observe_tasks(object);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
let _undo_fault = rollback_fault_injection::arm(object, 0, fault);
let writer = Arc::clone(&set);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
writer
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
write_completion,
..Default::default()
},
)
.await
});
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
.await
.expect("overwrite must enter the actual rename fan-out before failure injection");
barrier.release();
let err = tokio::time::timeout(Duration::from_secs(30), put)
.await
.expect("incomplete undo must return without hanging")
.expect("PUT task should join")
.expect_err("two renamed disks cannot satisfy write quorum three");
assert!(
matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
"original quorum error expected: {err}"
);
assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return");
let leftovers = non_trash_tmp_entries(&dirs).await;
assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery");
let backups = dirs
.iter()
.filter(|dir| {
dir.path()
.join(bucket)
.join(object)
.join(old_data_dir.to_string())
.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
.exists()
})
.count();
assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup");
// The remaining three disks still serve the old generation;
// the failed minority must never become an acknowledged write.
let mut read = set
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("old generation must remain readable after incomplete rollback");
let mut body = Vec::new();
read.stream
.read_to_end(&mut body)
.await
.expect("old generation should stream");
assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]);
}
}
})
.await;