test(ecstore): cover post-apply rollback error (#5054)

Refs rustfs/backlog#1355

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

crates/ecstore/src/disk/local.rs

@@ -1471,6 +1471,8 @@ static RENAME_DATA_FAIL_COMMIT_RENAME: std::sync::Mutex<Option<String>> = std::s
 static LOCAL_INLINE_ROLLBACK_HARDLINK_FAILURE: std::sync::Mutex<Option<PathBuf>> = std::sync::Mutex::new(None);
 #[cfg(test)]
 static DELETE_VERSION_FAIL_AFTER_DATA_STAGED: std::sync::Mutex<Vec<String>> = std::sync::Mutex::new(Vec::new());
+#[cfg(test)]
+static DELETE_VERSION_FAIL_AFTER_COMMIT: std::sync::Mutex<Vec<(PathBuf, String)>> = std::sync::Mutex::new(Vec::new());
 
 #[cfg(test)]
 fn set_rename_data_fail_before_old_metadata_backup(dst_path: &str) {
@@ -1508,6 +1510,14 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
         .push(path.to_string());
 }
 
+#[cfg(test)]
+pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
+    DELETE_VERSION_FAIL_AFTER_COMMIT
+        .lock()
+        .expect("test failpoint lock should not be poisoned")
+        .push((root.to_path_buf(), path.to_string()));
+}
+
 #[cfg(test)]
 fn should_fail_before_old_metadata_backup(dst_path: &str) -> bool {
     let mut target = RENAME_DATA_FAIL_BEFORE_OLD_METADATA_BACKUP
@@ -1573,6 +1583,22 @@ fn should_fail_after_delete_data_staged(path: &str) -> bool {
     }
 }
 
+#[cfg(test)]
+fn should_fail_after_delete_commit(root: &Path, path: &str) -> bool {
+    let mut targets = DELETE_VERSION_FAIL_AFTER_COMMIT
+        .lock()
+        .expect("test failpoint lock should not be poisoned");
+    if let Some(index) = targets
+        .iter()
+        .position(|(target_root, target_path)| target_root == root && target_path == path)
+    {
+        targets.remove(index);
+        true
+    } else {
+        false
+    }
+}
+
 #[cfg(not(test))]
 fn should_fail_before_old_metadata_backup(_dst_path: &str) -> bool {
     false
@@ -1598,6 +1624,11 @@ fn should_fail_after_delete_data_staged(_path: &str) -> bool {
     false
 }
 
+#[cfg(not(test))]
+fn should_fail_after_delete_commit(_root: &Path, _path: &str) -> bool {
+    false
+}
+
 fn log_startup_disk_io_error(stage: &str, path: &Path, err: &IoError) {
     warn!(
         event = EVENT_DISK_LOCAL_STARTUP_CLEANUP,
@@ -7834,6 +7865,10 @@ impl DiskAPI for LocalDisk {
             .await);
         }
 
+        if should_fail_after_delete_commit(self.root.as_path(), path) {
+            return Err(DiskError::Unexpected);
+        }
+
         Ok(())
     }
     #[tracing::instrument(level = "trace", skip_all)]

crates/ecstore/src/set_disk/ops/object.rs

@@ -3829,6 +3829,97 @@ mod transition_commit_failure_tests {
         assert_eq!(backend.object_count().await, 1);
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn local_commit_post_apply_error_rolls_back_the_errored_disk() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-post-apply-rollback-bucket";
+        let object = "object.bin";
+        let payload = b"post-apply delete errors must still roll back the changed disk".repeat(1024);
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        let original = set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("source object should be written");
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        let opts = ObjectOptions {
+            no_lock: true,
+            transition: TransitionOptions {
+                status: TRANSITION_PENDING.to_string(),
+                tier: tier_name,
+                etag: original.etag.clone().unwrap_or_default(),
+                ..Default::default()
+            },
+            version_id: original.version_id.map(|version| version.to_string()),
+            mod_time: original.mod_time,
+            ..Default::default()
+        };
+
+        let barrier = TransitionCommitBarrier::install_after_lease_check(bucket, object);
+        let transition_set = Arc::clone(&set_disks);
+        let transition = tokio::spawn(async move { transition_set.transition_object(bucket, object, &opts).await });
+        barrier.wait_until_paused().await;
+        assert_eq!(backend.put_count().await, 1, "remote candidate must exist before local commit");
+
+        crate::disk::local::set_delete_version_fail_after_commit(disk_stores[0].path().as_path(), object);
+        let saved_disk = {
+            let mut disks = set_disks.disks.write().await;
+            let saved = disks[1].take();
+            assert!(saved.is_some(), "test setup should start with the second disk online");
+            saved
+        };
+        barrier.release();
+        let result = transition.await.expect("transition task should not panic");
+
+        result.expect_err("post-apply disk error plus one offline disk must fail write quorum");
+        let errored_disk_fi = disk_stores[0]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("rollback must restore metadata on the disk that returned the post-apply error");
+        assert_ne!(
+            errored_disk_fi.transition_status, TRANSITION_COMPLETE,
+            "rollback must not skip the disk that applied delete_version but returned an error"
+        );
+        disk_stores[0]
+            .check_parts(bucket, object, &errored_disk_fi)
+            .await
+            .expect("rollback must restore the errored disk's staged data directory");
+
+        set_disks.disks.write().await[1] = saved_disk;
+        assert_eq!(
+            backend.remove_count().await,
+            0,
+            "ambiguous local commit failure must retain the remote candidate for reconciliation"
+        );
+        assert_eq!(backend.object_count().await, 1);
+
+        let mut restored = Vec::new();
+        set_disks
+            .get_object_reader(
+                bucket,
+                object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("source must remain readable after rolling back the post-apply error")
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("the local source body should drain");
+        assert_eq!(restored, payload);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn production_transition_replacement_revokes_commit_and_cleans_with_old_driver() {