diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 19018263c..1b383a1eb 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -7976,6 +7976,325 @@ mod heal_result_report_tests { assert_eq!(survivor.size, 1024 * 1024, "survivor size must be intact"); } + #[derive(Clone, Copy)] + enum VersionCommitMutation { + Append, + DeleteHistorical, + DeleteCurrent, + } + + struct VersionCommitFixture { + version: Uuid, + data_dir: Uuid, + body: Vec, + parts: Vec>, + } + + async fn explicit_version_commit_fence(mutation: VersionCommitMutation) { + use crate::object_api::{ShardIntegrityWriteMode, WriteCompletion}; + use std::time::Duration; + use tokio::io::AsyncReadExt; + + for (disk_count, parity) in [(4, 2), (16, 4)] { + let (dirs, disks, set) = hermetic_set_disks_for_pool_with_default_parity_isolated(disk_count, 0, parity).await; + let label = match mutation { + VersionCommitMutation::Append => "append", + VersionCommitMutation::DeleteHistorical => "delete-historical", + VersionCommitMutation::DeleteCurrent => "delete-current", + }; + let prefix = if matches!(mutation, VersionCommitMutation::Append) { + "native_g02" + } else { + "native_g03" + }; + let bucket = format!("version-commit-{label}-{disk_count}"); + let object = "object.bin"; + set.make_bucket( + &bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("create isolated versioned bucket"); + let options = ObjectOptions { + versioned: true, + write_completion: WriteCompletion::TailDrained, + shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected), + ..Default::default() + }; + let mut versions = Vec::new(); + for byte in [0x11, 0x22, 0x33] { + let body = vec![byte; 1024 * 1024 + 123]; + let info = set + .put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &options) + .await + .expect("ACK original version after its full rename tail"); + let version = info.version_id.expect("independent version id"); + let meta = disks[0] + .read_version("", &bucket, object, &version.to_string(), &ReadOptions::default()) + .await + .expect("read original version metadata"); + assert!(meta.data.is_none(), "native version fence must reconstruct external data"); + assert_eq!(meta.erasure.data_blocks, disk_count - parity); + assert_eq!(meta.erasure.parity_blocks, parity); + let data_dir = meta.data_dir.expect("actual external generation"); + let mut parts = Vec::new(); + for dir in &dirs { + parts.push( + tokio::fs::read( + dir.path() + .join(&bucket) + .join(object) + .join(data_dir.to_string()) + .join("part.1"), + ) + .await + .expect("every original encoded shard after ACK"), + ); + } + versions.push(VersionCommitFixture { + version, + data_dir, + body, + parts, + }); + } + let selected = if matches!(mutation, VersionCommitMutation::DeleteCurrent) { + 2 + } else { + 0 + }; + let selected_version = versions[selected].version; + let selected_dir = versions[selected].data_dir; + let missing = dirs[0] + .path() + .join(&bucket) + .join(object) + .join(selected_dir.to_string()) + .join("part.1"); + tokio::fs::remove_file(&missing) + .await + .expect("remove exact selected-version shard"); + let (_pause, arrived, release) = ReadRepairPauseScope::install(&bucket, object); + let heal_set = set.clone(); + let heal_bucket = bucket.clone(); + let heal = tokio::spawn(async move { + heal_set + .heal_object( + &heal_bucket, + object, + &selected_version.to_string(), + &HealOpts { + scan_mode: HealScanMode::Deep, + read_repair: true, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), arrived.notified()) + .await + .expect("actual selected-version reconstruction reaches final commit fence"); + assert!(!heal.is_finished() && !missing.exists()); + println!("{prefix}: geometry={disk_count}/{parity} mutation={label} reconstruction_paused_before_final_rename"); + + if matches!(mutation, VersionCommitMutation::Append) { + let body = vec![0x44; 1024 * 1024 + 123]; + let info = set + .put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &options) + .await + .expect("new version ACK while historical reconstruction is paused"); + let version = info.version_id.expect("new acknowledged version id"); + assert!(versions.iter().all(|v| v.version != version)); + let meta = disks[0] + .read_version("", &bucket, object, &version.to_string(), &ReadOptions::default()) + .await + .expect("new physical version metadata"); + let data_dir = meta.data_dir.expect("new external generation"); + let mut parts = Vec::new(); + for dir in &dirs { + parts.push( + tokio::fs::read( + dir.path() + .join(&bucket) + .join(object) + .join(data_dir.to_string()) + .join("part.1"), + ) + .await + .expect("new version ACK has every physical shard"), + ); + } + versions.push(VersionCommitFixture { + version, + data_dir, + body, + parts, + }); + } else { + set.delete_object( + &bucket, + object, + ObjectOptions { + versioned: true, + version_id: Some(selected_version.to_string()), + object_lock_config_snapshot: Some(Arc::new(crate::set_disk::ObjectLockConfigSnapshot::new( + crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent, + ))), + ..Default::default() + }, + ) + .await + .expect("exact version delete ACK before releasing stale reconstruction"); + } + // Commit pause has dropped the Heal read guard. Acquire this key's + // write fence only after the foreground ACK, draining its physical + // cleanup before snapshots; release it before resuming the Heal. + let namespace = set.new_ns_lock(&bucket, object).await.expect("foreground drain fence"); + let drained = namespace + .get_write_lock(Duration::from_secs(30)) + .await + .expect("foreground physical tail drains"); + let mut committed_metadata = Vec::new(); + for dir in &dirs { + committed_metadata.push( + tokio::fs::read(dir.path().join(&bucket).join(object).join(STORAGE_FORMAT_FILE)) + .await + .expect("foreground committed metadata on every drive"), + ); + if !matches!(mutation, VersionCommitMutation::Append) { + assert!(!dir.path().join(&bucket).join(object).join(selected_dir.to_string()).exists()); + } + } + drop(drained); + assert!(!heal.is_finished()); + println!( + "{prefix}: geometry={disk_count}/{parity} mutation={label} foreground_ack_and_physical_tail_verified_before_release" + ); + release.notify_one(); + let outcome = tokio::time::timeout(Duration::from_secs(30), heal) + .await + .expect("Heal leaves its released final fence") + .expect("Heal task joins"); + if matches!(mutation, VersionCommitMutation::Append) { + assert!(outcome.expect("historical version remains healable").1.is_none()); + } else { + assert!( + matches!(outcome, Err(DiskError::FileVersionNotFound)), + "deleted version must remain typed absent: {outcome:?}" + ); + } + for (slot, dir) in dirs.iter().enumerate() { + if slot != 0 || !matches!(mutation, VersionCommitMutation::Append) { + assert_eq!( + tokio::fs::read(dir.path().join(&bucket).join(object).join(STORAGE_FORMAT_FILE)) + .await + .expect("metadata after Heal"), + committed_metadata[slot], + "Heal must preserve every committed successor metadata byte outside its repaired target" + ); + } + if !matches!(mutation, VersionCommitMutation::Append) { + assert!(!dir.path().join(&bucket).join(object).join(selected_dir.to_string()).exists()); + } + for version in &versions { + if !matches!(mutation, VersionCommitMutation::Append) && version.version == selected_version { + assert!(matches!( + disks[slot] + .read_version("", &bucket, object, &version.version.to_string(), &ReadOptions::default()) + .await, + Err(DiskError::FileVersionNotFound) + )); + continue; + } + let meta = disks[slot] + .read_version("", &bucket, object, &version.version.to_string(), &ReadOptions::default()) + .await + .expect("every acknowledged surviving version metadata"); + assert_eq!(meta.data_dir, Some(version.data_dir)); + assert_eq!(meta.size, version.body.len() as i64); + assert_eq!( + tokio::fs::read( + dir.path() + .join(&bucket) + .join(object) + .join(version.data_dir.to_string()) + .join("part.1") + ) + .await + .expect("physical shard before any GET"), + version.parts[slot], + "every surviving version must retain its exact original encoded shard" + ); + } + } + for version in &versions { + if !matches!(mutation, VersionCommitMutation::Append) && version.version == selected_version { + continue; + } + let mut reader = set + .get_object_reader( + &bucket, + object, + None, + Default::default(), + &ObjectOptions { + versioned: true, + version_id: Some(version.version.to_string()), + ..Default::default() + }, + ) + .await + .expect("read exact surviving version after physical proof"); + let mut actual = Vec::new(); + reader + .read_to_end(&mut actual) + .await + .expect("read complete surviving version"); + assert_eq!(actual, version.body); + } + let current = if matches!(mutation, VersionCommitMutation::DeleteCurrent) { + &versions[1] + } else { + versions.last().unwrap() + }; + let mut reader = set + .get_object_reader( + &bucket, + object, + None, + Default::default(), + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("read exact last surviving current generation"); + let mut actual = Vec::new(); + reader.read_to_end(&mut actual).await.expect("read full current generation"); + assert_eq!(actual, current.body); + println!( + "{prefix}: geometry={disk_count}/{parity} mutation={label} exact_version_outcome_physical_generations_and_full_bodies_verified" + ); + } + } + + #[tokio::test] + #[serial_test::serial] + async fn native_explicit_version_commit_preserves_acknowledged_new_version() { + explicit_version_commit_fence(VersionCommitMutation::Append).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn native_explicit_version_commit_refuses_deleted_historical_and_current_versions() { + explicit_version_commit_fence(VersionCommitMutation::DeleteHistorical).await; + explicit_version_commit_fence(VersionCommitMutation::DeleteCurrent).await; + } + #[tokio::test] #[serial_test::serial] async fn read_repair_commit_aborts_when_object_changes_before_final_rename() { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 79afc4244..1d1bad3c3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -21143,7 +21143,12 @@ mod put_object_tmp_cleanup_tests { .join("part.1"); assert!(tokio::fs::try_exists(&old_part).await.expect("original part exists")); - let barrier = rename_fanout_barrier::arm(&object, 0, rename_fanout_barrier::PHASE_CLEANUP); + // Cleanup slots follow the writer's shuffled erasure distribution, + // not the physical fixture order. Pause every real cleanup slot so + // no physical old shard can disappear before the proof below. + let barriers: Vec<_> = (0..disk_count) + .map(|slot| rename_fanout_barrier::arm(&object, slot, rename_fanout_barrier::PHASE_CLEANUP)) + .collect(); let writer_set = Arc::clone(&set); let writer_object = object.clone(); let writer_options = options.clone(); @@ -21154,9 +21159,13 @@ mod put_object_tmp_cleanup_tests { .put_object(bucket, &writer_object, &mut PutObjReader::from_vec(writer_payload), &writer_options) .await }); - tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) - .await - .expect("overwrite commits before its real old-data cleanup pause"); + tokio::time::timeout(Duration::from_secs(30), async { + for barrier in &barriers { + barrier.wait_until_paused().await; + } + }) + .await + .expect("overwrite commits before its real old-data cleanup pause"); assert!(!writer.is_finished(), "tail-drained ACK must remain pending at cleanup"); assert!( tokio::fs::try_exists(&old_part) @@ -21190,6 +21199,20 @@ mod put_object_tmp_cleanup_tests { .join(new_dir.to_string()) .join("part.1"); let expected_part = tokio::fs::read(&new_part).await.expect("exact new source bytes"); + let mut committed_parts = Vec::new(); + for dir in &dirs { + let original = dir.path().join(bucket).join(&object).join(old_dir.to_string()).join("part.1"); + assert!( + tokio::fs::try_exists(original) + .await + .expect("every old cleanup slot is paused") + ); + committed_parts.push( + tokio::fs::read(dir.path().join(bucket).join(&object).join(new_dir.to_string()).join("part.1")) + .await + .expect("every new encoded shard before fault"), + ); + } tokio::fs::remove_file(&new_part) .await .expect("remove only the newly committed target shard"); @@ -21236,14 +21259,31 @@ mod put_object_tmp_cleanup_tests { "native_g01: geometry={disk_count}/{parity} read_repair={read_repair} locked_heal_completed new_generation_part_rebuilt_before_get" ); - barrier.release(); + for barrier in &barriers { + barrier.release(); + } let acknowledged = tokio::time::timeout(Duration::from_secs(30), writer) .await .expect("cleanup drains") .expect("writer joins") .expect("overwrite ACK remains successful"); assert_eq!(acknowledged.size, i64::try_from(payload.len()).expect("fixture size fits i64")); - drop(barrier); + drop(barriers); + for (slot, dir) in dirs.iter().enumerate() { + let root = dir.path().join(bucket).join(&object); + assert!( + !tokio::fs::try_exists(root.join(old_dir.to_string())) + .await + .expect("old cleanup completed") + ); + assert_eq!( + tokio::fs::read(root.join(new_dir.to_string()).join("part.1")) + .await + .expect("new shard survives cleanup before GET"), + committed_parts[slot], + "old cleanup must preserve every exact committed encoded shard" + ); + } let mut reader = set .get_object_reader(bucket, &object, None, HeaderMap::new(), &options) .await diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index f146c122c..494d365be 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -3141,7 +3141,10 @@ impl DefaultObjectUsecase { // resolver; only the response headers, authorization and audit // summary are derived here, without a second KMS unwrap. Some(headers) => ( - Some(headers.server_side_encryption).filter(|_| headers.sse_customer_algorithm.is_none()), + headers + .sse_customer_algorithm + .is_none() + .then_some(headers.server_side_encryption), headers.sse_customer_algorithm, headers.sse_customer_key_md5, headers.ssekms_key_id, diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 56f488ff1..b92dcd36b 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -1476,9 +1476,11 @@ mod tests { fn rpc_unavailable_preserves_service_refusal_across_io_boundaries() { let status = || tonic::Status::unavailable("private transport detail must not reach the S3 response"); let storage = || StorageError::from(status()); + let original = storage(); + let cloned = original.clone(); for error in [ - ApiError::from(storage()), - ApiError::from(storage().clone()), + ApiError::from(original), + ApiError::from(cloned), ApiError::from(StorageError::Io(IoError::other(IoError::other(status())))), ApiError::from(IoError::other(status())), ApiError::from(IoError::other(storage())),