test(ecstore): fence version races and bind cleanup slots

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<u8>,
+        parts: Vec<Vec<u8>>,
+    }
+
+    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() {

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

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,

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())),