diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 10a98b215..f0bf767d2 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -3693,19 +3693,30 @@ mod tests { } } - async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec> { + async fn object_transaction_epochs( + disks: &[DiskStore], + bucket: &str, + object: &str, + write_quorum: usize, + ) -> Vec> { let mut epochs = Vec::with_capacity(disks.len()); for (disk_index, disk) in disks.iter().enumerate() { - let file_info = disk - .read_version("", bucket, object, "", &ReadOptions::default()) - .await - .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}")); + let file_info = match disk.read_version("", bucket, object, "", &ReadOptions::default()).await { + Ok(file_info) => file_info, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => continue, + Err(err) => panic!("disk {disk_index} should read object metadata: {err}"), + }; epochs.push( file_info .object_transaction_epoch() .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")), ); } + assert!( + epochs.len() >= write_quorum, + "object metadata should persist on write quorum: found {}, need {write_quorum}", + epochs.len() + ); epochs } @@ -3729,6 +3740,7 @@ mod tests { [ (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")), (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")), + (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false")), ], async { set_disks @@ -3777,7 +3789,11 @@ mod tests { ) .await; - let epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + disk_stores[0] + .delete_paths(bucket, &[format!("{object}/{STORAGE_FORMAT_FILE}")]) + .await + .expect("one lagging disk should be simulated"); + let epochs = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()).await; let first = epochs[0].expect("fenced multipart completion should persist an epoch"); assert!(!first.is_nil()); assert!(epochs.into_iter().all(|epoch| epoch == Some(first))); @@ -3811,7 +3827,7 @@ mod tests { ) .await .expect("initial fenced PUT should commit"); - let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object) + let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()) .await .into_iter() .next() @@ -3853,7 +3869,7 @@ mod tests { ) .await .expect("concurrent fenced PUT should advance the epoch"); - let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object) + let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()) .await .into_iter() .next() @@ -3868,7 +3884,8 @@ mod tests { .expect_err("stale epoch multipart completion must be rejected"); assert_eq!(err, StorageError::PreconditionFailed); - let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await; + let final_epochs = + object_transaction_epochs(&disk_stores, bucket, object, set_disks.default_write_quorum()).await; assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch))); let mut reader = set_disks .get_object_reader(