test(ecstore): pin rename quorum tail visibility (#6301)

Add deterministic rename_data coverage for tail-disk success/failure, cancellation serialization, and strict quorum rollback visibility after disk reopen. This establishes the safety boundary before experimenting with write-quorum early ACK and background tail completion for backlog #925.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -5655,6 +5655,8 @@ pub(in crate::set_disk) mod rename_fanout_barrier {
 
 #[cfg(test)]
 mod tests {
+    use crate::disk::local::{DurabilityMode, durability_mode_override};
+
     use super::*;
     use std::io::Cursor;
     use tempfile::TempDir;
@@ -5862,6 +5864,34 @@ mod tests {
         (dirs, disks)
     }
 
+    async fn reopen_local_disk(dir: &TempDir) -> DiskStore {
+        let endpoint =
+            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
+        new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("disk should reopen from the same tempdir")
+    }
+
+    async fn prepare_rename_source_dirs(dirs: &[TempDir], disks: &[Option<DiskStore>], source: &str) {
+        for (dir, disk) in dirs.iter().zip(disks.iter()) {
+            let Some(disk) = disk else {
+                continue;
+            };
+            match disk.make_volume(RUSTFS_META_TMP_BUCKET).await {
+                Ok(()) | Err(DiskError::VolumeExists) => {}
+                Err(err) => panic!("temporary metadata volume should be available: {err:?}"),
+            }
+            std::fs::create_dir_all(dir.path().join(RUSTFS_META_TMP_BUCKET).join(source))
+                .expect("rename staging source dir should be created");
+        }
+    }
+
     #[test]
     fn metadata_slowtail_fault_delay_parses_and_filters_request() {
         temp_env::with_vars(
@@ -7503,6 +7533,19 @@ mod tests {
         (0..count).map(|_| metadata_test_fileinfo(object)).collect()
     }
 
+    fn rename_commit_fileinfos(object: &str, count: usize, etag: &str) -> Vec<FileInfo> {
+        (0..count)
+            .map(|idx| {
+                let mut file_info = metadata_test_fileinfo(object);
+                file_info.mod_time = Some(OffsetDateTime::now_utc());
+                file_info.erasure.index = idx + 1;
+                file_info.data = Some(Bytes::from_static(b"inline-body"));
+                file_info.metadata.insert("etag".to_string(), etag.to_string());
+                file_info
+            })
+            .collect()
+    }
+
     #[tokio::test]
     async fn rename_data_skips_offline_placeholder_when_validating_new_metadata() {
         let (_dirs, mut online_disks) = call_counter_local_disks("rename-validation-bucket", 1).await;
@@ -7774,6 +7817,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
     async fn rename_fanout_drains_after_caller_cancellation() {
         const DISKS: usize = 4;
         let bucket = "rename-cancel-bucket";
@@ -7818,6 +7862,189 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
+    async fn rename_data_waits_for_tail_disk_after_write_quorum() {
+        const DISKS: usize = 4;
+        let bucket = "rename-tail-success-bucket";
+        let object = "rename-tail-success-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        prepare_rename_source_dirs(&dirs, &disks, "source").await;
+        let file_infos = rename_commit_fileinfos(object, DISKS, "tail-success-etag");
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+
+        let rename = SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 3);
+        tokio::pin!(rename);
+        tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
+            tokio::select! {
+                () = barrier.wait_until_paused() => {}
+                result = &mut rename => panic!("rename_data returned before the armed fan-out barrier: {result:?}"),
+            }
+        })
+        .await
+        .expect("paused disk must reach the armed rename barrier");
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(50), &mut rename).await.is_err(),
+            "current rename_data waits for the paused fan-out disk even after the other three disks can reach write quorum"
+        );
+
+        barrier.release();
+        rename
+            .await
+            .expect("tail success must complete the rename after the barrier is released");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            let reopened = reopen_local_disk(dir).await;
+            let stored = reopened
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {idx} must contain the tail-success commit after reopen: {err:?}"));
+            assert_eq!(
+                stored.metadata.get("etag").map(String::as_str),
+                Some("tail-success-etag"),
+                "disk {idx} must expose the same committed metadata after tail success and reopen"
+            );
+        }
+
+        drop(dirs);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
+    async fn rename_data_commits_fresh_object_when_tail_disk_fails_after_write_quorum() {
+        const DISKS: usize = 4;
+        let bucket = "rename-tail-failure-fresh-bucket";
+        let object = "rename-tail-failure-fresh-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        prepare_rename_source_dirs(&dirs, &disks, "source").await;
+        let mut file_infos = rename_commit_fileinfos(object, DISKS, "fresh-etag");
+        file_infos[3] = FileInfo::default();
+
+        SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 3)
+            .await
+            .expect("three successful disks must satisfy write quorum despite one tail failure");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            let reopened = reopen_local_disk(dir).await;
+            let read = reopened.read_version("", bucket, object, "", &ReadOptions::default()).await;
+            if idx < 3 {
+                read.unwrap_or_else(|err| panic!("quorum disk {idx} must contain the fresh commit after reopen: {err:?}"));
+            } else {
+                assert!(
+                    matches!(read, Err(DiskError::FileNotFound | DiskError::FileVersionNotFound)),
+                    "failed tail disk must not expose a partial fresh commit after reopen: {read:?}"
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
+    async fn rename_data_overwrite_tail_failure_preserves_old_tail_version_after_reopen() {
+        const DISKS: usize = 4;
+        let bucket = "rename-tail-failure-overwrite-bucket";
+        let object = "rename-tail-failure-overwrite-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        prepare_rename_source_dirs(&dirs, &disks, "source").await;
+        let mut old = metadata_test_fileinfo(object);
+        old.mod_time = Some(OffsetDateTime::now_utc());
+        old.data = Some(Bytes::from_static(b"old-inline-body"));
+        old.metadata.insert("etag".to_string(), "old-etag".to_string());
+        for disk in disks.iter().flatten() {
+            disk.write_metadata(bucket, bucket, object, old.clone())
+                .await
+                .expect("old metadata should be written before overwrite");
+        }
+
+        let mut file_infos = rename_commit_fileinfos(object, DISKS, "new-etag");
+        file_infos[3] = FileInfo::default();
+
+        SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 3)
+            .await
+            .expect("three successful disks must satisfy overwrite quorum despite one tail failure");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            let reopened = reopen_local_disk(dir).await;
+            let stored = reopened
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {idx} must have a readable version after reopen: {err:?}"));
+            let expected_etag = if idx < 3 { "new-etag" } else { "old-etag" };
+            assert_eq!(
+                stored.metadata.get("etag").map(String::as_str),
+                Some(expected_etag),
+                "disk {idx} must keep the correct overwrite visibility after reopen"
+            );
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
+    async fn rename_data_strict_quorum_failure_rolls_back_fresh_object_after_reopen() {
+        let _mode = durability_mode_override::set(DurabilityMode::Strict);
+        const DISKS: usize = 4;
+        let bucket = "rename-strict-rollback-fresh-bucket";
+        let object = "rename-strict-rollback-fresh-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        prepare_rename_source_dirs(&dirs, &disks, "source").await;
+        let mut file_infos = rename_commit_fileinfos(object, DISKS, "fresh-rollback-etag");
+        file_infos[3] = FileInfo::default();
+
+        SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 4)
+            .await
+            .expect_err("three successful disks must fail a strict write quorum of four");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            let reopened = reopen_local_disk(dir).await;
+            let read = reopened.read_version("", bucket, object, "", &ReadOptions::default()).await;
+            assert!(
+                matches!(read, Err(DiskError::FileNotFound | DiskError::FileVersionNotFound)),
+                "disk {idx} must not expose a fresh object after strict quorum rollback and reopen: {read:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(rename_quorum_ack)]
+    async fn rename_data_strict_quorum_failure_restores_overwrite_after_reopen() {
+        let _mode = durability_mode_override::set(DurabilityMode::Strict);
+        const DISKS: usize = 4;
+        let bucket = "rename-strict-rollback-overwrite-bucket";
+        let object = "rename-strict-rollback-overwrite-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        prepare_rename_source_dirs(&dirs, &disks, "source").await;
+        let mut old = metadata_test_fileinfo(object);
+        old.mod_time = Some(OffsetDateTime::now_utc());
+        old.data = Some(Bytes::from_static(b"old-inline-body"));
+        old.metadata.insert("etag".to_string(), "old-rollback-etag".to_string());
+        for disk in disks.iter().flatten() {
+            disk.write_metadata(bucket, bucket, object, old.clone())
+                .await
+                .expect("old metadata should be written before overwrite rollback test");
+        }
+
+        let mut file_infos = rename_commit_fileinfos(object, DISKS, "new-rollback-etag");
+        file_infos[3] = FileInfo::default();
+
+        SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 4)
+            .await
+            .expect_err("three successful disks must fail a strict overwrite quorum of four");
+
+        for (idx, dir) in dirs.iter().enumerate() {
+            let reopened = reopen_local_disk(dir).await;
+            let stored = reopened
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {idx} must keep old metadata after strict rollback: {err:?}"));
+            assert_eq!(
+                stored.metadata.get("etag").map(String::as_str),
+                Some("old-rollback-etag"),
+                "disk {idx} must restore the old overwrite target after strict rollback and reopen"
+            );
+        }
+    }
+
     /// Demo / regression guard for the barrier on the commit (old-data-dir)
     /// cleanup fan-out. Serves the same #1312/#1319 "no background disk write
     /// after release" shape, on the reclamation path that runs *after* a write is