Merge branch 'main' into overtrue/fix-read-version-quorum-fixture

* fix(multipart): serialize complete and abort

* test(multipart): order abort-first finalization

* fix(multipart): enforce quorum staging cleanup

* fix(multipart): remove stale mutable binding

.config/nextest.toml

@@ -1,17 +1,14 @@
 # nextest configuration for RustFS.
 #
-# Serialize two known load-sensitive / global-state-sharing ecstore test groups
-# so the full parallel nextest suite stops producing spurious failures
-# (backlog #937). These tests pass in isolation but flake under the loaded
-# parallel run for two distinct reasons:
+# Serialize the ecstore tests that share the process-wide disk registry or
+# exercise a multi-disk commit handoff across nextest process boundaries.
 #
 #   * store::bucket::tests::bucket_delete_* share process/global state (disk
 #     registry, lock client) and race make_bucket into InsufficientWriteQuorum
 #     when run concurrently with other ecstore tests.
 #   * bucket_lifecycle_ops::tests::concurrent_resend_same_part_commits_one_generation
-#     asserts a lock-acquire correctness property whose serialized cross-disk
-#     commits exceed the (already max'd, 60s) acquire deadline only when the
-#     suite saturates disk I/O.
+#     uses the shared multipart fixture and a deterministic uploadId-lock
+#     handoff, so it must not overlap another process mutating that fixture.
 #
 # serial_test's #[serial] attribute does NOT serialize these across runs:
 # nextest executes each test in its own process, where the in-process
@@ -100,13 +97,6 @@ path = "junit.xml"
 # profile's own overrides list, not the default profile's).
 # ===========================================================================
 
-# QUARANTINE: OPEN backlog#937 — concurrent_resend lock-acquire deadline flakes
-# under saturated disk I/O in the full parallel suite.
-[[profile.ci.overrides]]
-filter = 'package(rustfs-ecstore) & test(concurrent_resend_same_part_commits_one_generation)'
-test-group = 'ecstore-serial-flaky'
-retries = 2
-
 # QUARANTINE: OPEN backlog#937 — store::bucket::tests::bucket_delete_* race
 # make_bucket into InsufficientWriteQuorum via shared global state under load.
 [[profile.ci.overrides]]
@@ -114,6 +104,11 @@ filter = 'package(rustfs-ecstore) & test(/^store::bucket::tests::bucket_delete_(
 test-group = 'ecstore-serial-flaky'
 retries = 2
 
+# Keep the deterministic multipart handoff isolated across nextest processes.
+[[profile.ci.overrides]]
+filter = 'package(rustfs-ecstore) & test(concurrent_resend_same_part_commits_one_generation)'
+test-group = 'ecstore-serial-flaky'
+
 # QUARANTINE: OPEN rustfs#4690 — walk_dir stall-budget accounting test depends
 # on producer/consumer timing windows that stretch past the budget on loaded
 # CI runners (regression test for rustfs#4644; failed on a zero-Rust-diff PR).

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -11112,6 +11112,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn concurrent_resend_same_part_commits_one_generation() {
+        use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
         use crate::storage_api_contracts::object::ObjectIO as _;
 
         let (_paths, ecstore) = setup_test_env().await;
@@ -11133,51 +11134,36 @@ mod tests {
             })
             .collect();
 
-        // Two independent causes can produce a spurious lock-acquire timeout
-        // here, and both must stay covered:
-        // 1. A lost/stolen fast-lock wakeup could strand a waiter until the
-        //    deadline — fixed for real in fast_lock::shard by bounding each
-        //    notification wait (NOTIFY_WAIT_CAP re-polling).
-        // 2. Under the full nextest suite on loaded CI disks, the
-        //    *legitimately serialized* cross-disk commits can exceed the
-        //    acquire deadline all by themselves — observed on CI at the 5s
-        //    default and the 30s production default with six resends, and
-        //    again at 60s, which is a hard ceiling: fast_lock clamps every
-        //    requested timeout to MAX_ACQUIRE_TIMEOUT (60s), so raising the
-        //    env override higher is a no-op (the Timeout error still reports
-        //    the requested value). Keep the guard about the correctness
-        //    property, not disk latency: request the full 60s ceiling and cap
-        //    the queue depth at three resends, so the last waiter sits behind
-        //    at most two serialized commits (~12s each on the slowest observed
-        //    CI runner, comfortably inside the deadline). Three concurrent
-        //    resends still race the streaming phase and contend on the commit
-        //    lock, which is all the generation-mixing regression needs.
-        //    `#[serial]` keeps the process-wide env override isolated.
-        let results = temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60"))], async {
-            let mut tasks = tokio::task::JoinSet::new();
-            for payload in candidates.iter().cloned() {
-                let store = ecstore.clone();
-                let bucket = bucket.clone();
-                let upload_id = upload.upload_id.clone();
-                tasks.spawn(async move {
-                    let mut data = PutObjReader::from_vec(payload.clone());
-                    store
-                        .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
-                        .await
-                        .map(|info| (info, payload))
-                });
-            }
+        let commit_barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
+        let start = Arc::new(tokio::sync::Barrier::new(candidates.len() + 1));
+        let mut tasks = tokio::task::JoinSet::new();
+        for payload in candidates.iter().cloned() {
+            let store = ecstore.clone();
+            let bucket = bucket.clone();
+            let upload_id = upload.upload_id.clone();
+            let start = Arc::clone(&start);
+            tasks.spawn(async move {
+                start.wait().await;
+                let mut data = PutObjReader::from_vec(payload.clone());
+                store
+                    .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
+                    .await
+                    .map(|info| (info, payload))
+            });
+        }
+        start.wait().await;
 
-            // Every concurrent resend must succeed; the commit lock must never
-            // starve a waiter into a timeout.
-            let mut results = Vec::new();
-            while let Some(joined) = tasks.join_next().await {
-                let outcome = joined.expect("put_object_part task should not panic");
-                results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
-            }
-            results
-        })
-        .await;
+        // The first writer holds the uploadId commit lock while the other
+        // resends reach the same critical section. Releasing it proves the
+        // handoff without depending on saturated CI disk latency.
+        commit_barrier.wait_until_paused().await;
+        commit_barrier.release();
+
+        let mut results = Vec::new();
+        while let Some(joined) = tasks.join_next().await {
+            let outcome = joined.expect("put_object_part task should not panic");
+            results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
+        }
         assert_eq!(results.len(), candidates.len());
 
         // Exactly one generation is visible after the serialized commits, and its

crates/ecstore/src/set_disk/mod.rs

@@ -687,6 +687,8 @@ mod core;
 mod ctx;
 mod metadata;
 mod ops;
+#[cfg(test)]
+pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;

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

@@ -29,6 +29,12 @@ impl SetDisks {
     pub async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
         ListOperations::new(self.ctx()).delete_all(bucket, prefix).await
     }
+
+    pub(crate) async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
+        ListOperations::new(self.ctx())
+            .delete_all_with_quorum(bucket, prefix, write_quorum)
+            .await
+    }
 }
 
 /// List/prefix maintenance operations, borrowing the `SetDisks` core state
@@ -48,6 +54,14 @@ impl<'a> ListOperations<'a> {
     }
 
     pub(crate) async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
+        self.delete_all_inner(bucket, prefix, None).await
+    }
+
+    async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
+        self.delete_all_inner(bucket, prefix, Some(write_quorum)).await
+    }
+
+    async fn delete_all_inner(&self, bucket: &str, prefix: &str, write_quorum: Option<usize>) -> Result<()> {
         let disks = self.ctx.disks().read().await;
 
         let disks = disks.clone();
@@ -79,6 +93,9 @@ impl<'a> ListOperations<'a> {
                 Ok(_) => {
                     errors.push(None);
                 }
+                Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => {
+                    errors.push(None);
+                }
                 Err(e) => {
                     errors.push(Some(e));
                 }
@@ -97,6 +114,12 @@ impl<'a> ListOperations<'a> {
             );
         }
 
+        if let Some(write_quorum) = write_quorum
+            && let Some(err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, write_quorum)
+        {
+            return Err(err.into());
+        }
+
         Ok(())
     }
 }

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

@@ -26,6 +26,8 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
+#[cfg(test)]
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 use tokio::task::JoinSet;
 
@@ -33,7 +35,7 @@ const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
 #[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-enum MultipartCommitPause {
+pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
@@ -45,12 +47,13 @@ struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
     pause: MultipartCommitPause,
+    armed: AtomicBool,
     arrived: tokio::sync::Notify,
     release: tokio::sync::Notify,
 }
 
 #[cfg(test)]
-struct MultipartCommitBarrier {
+pub(crate) struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
@@ -60,11 +63,12 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc
 
 #[cfg(test)]
 impl MultipartCommitBarrier {
-    fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
         let state = Arc::new(MultipartCommitBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
             pause,
+            armed: AtomicBool::new(true),
             arrived: tokio::sync::Notify::new(),
             release: tokio::sync::Notify::new(),
         });
@@ -78,13 +82,13 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    async fn wait_until_paused(&self) {
+    pub(crate) async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("multipart completion should reach the deterministic commit barrier");
     }
 
-    fn release(&self) {
+    pub(crate) fn release(&self) {
         self.state.release.notify_one();
     }
 }
@@ -112,7 +116,9 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm
         .as_ref()
         .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
         .cloned();
-    if let Some(barrier) = barrier {
+    if let Some(barrier) = barrier
+        && barrier.armed.swap(false, Ordering::AcqRel)
+    {
         barrier.arrived.notify_one();
         barrier.release.notified().await;
     }
@@ -777,6 +783,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         mut max_parts: usize,
         opts: &ObjectOptions,
     ) -> Result<ListPartsInfo> {
+        let _upload_guard = self
+            .acquire_multipart_upload_read_lock("list_object_parts", bucket, object, upload_id, opts)
+            .await?;
         let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, false).await?;
 
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
@@ -1254,10 +1263,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let _upload_guard = self
             .acquire_multipart_upload_write_lock("abort_multipart_upload", bucket, object, upload_id, opts)
             .await?;
-        self.check_upload_id_exists(bucket, object, upload_id, false).await?;
+        let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
-        self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await
+        self.delete_all_with_quorum(
+            RUSTFS_META_MULTIPART_BUCKET,
+            &upload_id_path,
+            fi.write_quorum(self.default_write_quorum()),
+        )
+        .await
     }
     // complete_multipart_upload finished
     #[tracing::instrument(skip(self))]
@@ -1815,7 +1829,36 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         #[cfg(test)]
         pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-        drop(upload_guard);
+
+        let cleanup_store = self.clone();
+        let cleanup_upload_id_path = upload_id_path.clone();
+        let cleanup_bucket = bucket.to_owned();
+        let cleanup_object = object.to_owned();
+        let cleanup_upload_id = upload_id.to_owned();
+        let cleanup_handle = tokio::spawn(async move {
+            let _upload_guard = upload_guard;
+            if let Err(err) = cleanup_store
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
+                .await
+            {
+                warn!(
+                    bucket = %cleanup_bucket,
+                    object = %cleanup_object,
+                    upload_id = %cleanup_upload_id,
+                    error = ?err,
+                    "completed multipart upload staging cleanup did not reach write quorum"
+                );
+            }
+        });
+        if let Err(err) = cleanup_handle.await {
+            warn!(
+                bucket = %bucket,
+                object = %object,
+                upload_id = %upload_id,
+                error = ?err,
+                "completed multipart upload staging cleanup task failed"
+            );
+        }
         drop(object_lock_guard); // drop object lock guard to release the lock
 
         // backlog#1321: enqueue heal only when the committed replicas actually
@@ -1860,12 +1903,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             });
         }
 
-        let upload_id_path = upload_id_path.clone();
-        let store = self.clone();
-        let _cleanup_handle = tokio::spawn(async move {
-            let _ = store.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await;
-        });
-
         for (i, op_disk) in online_disks.iter().enumerate() {
             if let Some(disk) = op_disk
                 && disk.is_online().await
@@ -2177,6 +2214,122 @@ mod tests {
         )
     }
 
+    async fn assert_complete_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x47; 4096], &create_opts).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                .await
+        });
+        barrier.wait_until_paused().await;
+
+        let abort_store = set_disks.clone();
+        let abort_upload_id = upload_id.clone();
+        let abort = tokio::spawn(async move {
+            abort_store
+                .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+        assert!(!abort.is_finished(), "abort must wait for the completion upload lock");
+
+        barrier.release();
+        complete
+            .await
+            .expect("completion task should not panic")
+            .expect("completion should win the upload finalization");
+        let abort_err = abort
+            .await
+            .expect("abort task should not panic")
+            .expect_err("abort must observe the upload as finalized");
+        assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
+        set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("complete-first must leave the committed object readable");
+        assert!(matches!(
+            set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+    }
+
+    async fn assert_abort_first_linearizes(bucket: &'static str, object: &'static str, create_opts: ObjectOptions) {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x48; 4096], &create_opts).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone()));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let object_holder = set_disks
+            .new_ns_lock(bucket, object)
+            .await
+            .expect("object namespace lock should be created")
+            .get_write_lock(Duration::from_secs(5))
+            .await
+            .expect("test should hold the object lock");
+        let holder = set_disks
+            .new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path)
+            .await
+            .expect("upload namespace lock should be created")
+            .get_write_lock(Duration::from_secs(5))
+            .await
+            .expect("test should hold the upload lock");
+        signaling.wait_for_attempts(1).await;
+
+        let abort_store = set_disks.clone();
+        let abort_upload_id = upload_id.clone();
+        let abort = tokio::spawn(async move {
+            abort_store
+                .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                .await
+        });
+        drop(holder);
+
+        abort
+            .await
+            .expect("abort task should not panic")
+            .expect("abort should win the upload finalization");
+        drop(object_holder);
+        let complete_err = complete
+            .await
+            .expect("completion task should not panic")
+            .expect_err("completion must observe the aborted upload");
+        assert!(matches!(complete_err, StorageError::InvalidUploadID(..)));
+        let object_err = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect_err("abort-first must not publish an object");
+        assert!(matches!(object_err, StorageError::ObjectNotFound(..)));
+        assert!(matches!(
+            set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+    }
+
     async fn assert_quorum_minus_one_retry_preserves_completable_part(
         disk_count: usize,
         parity: usize,
@@ -3039,6 +3192,81 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn abort_and_complete_linearize_for_plain_sse_and_legacy_layouts() {
+        assert_complete_first_linearizes("multipart-complete-first-plain", "object", ObjectOptions::default()).await;
+        assert_abort_first_linearizes("multipart-abort-first-plain", "object", ObjectOptions::default()).await;
+
+        let encrypted_opts = ObjectOptions {
+            user_defined: HashMap::from([(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())]),
+            ..Default::default()
+        };
+        temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async {
+            assert_complete_first_linearizes("multipart-complete-first-sse", "object", encrypted_opts.clone()).await;
+            assert_abort_first_linearizes("multipart-abort-first-sse", "object", encrypted_opts.clone()).await;
+        })
+        .await;
+        temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("false"))], async {
+            assert_complete_first_linearizes("multipart-complete-first-legacy", "object", encrypted_opts.clone()).await;
+            assert_abort_first_linearizes("multipart-abort-first-legacy", "object", encrypted_opts).await;
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn abort_enforces_delete_write_quorum_boundary() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-abort-delete-quorum";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let quorum_upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("multipart upload should be created");
+
+        let saved_disks = {
+            let mut disks = set_disks.disks.write().await;
+            let saved = disks.clone();
+            disks[3] = None;
+            saved
+        };
+        set_disks
+            .abort_multipart_upload(bucket, object, &quorum_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("abort should succeed at the exact delete write quorum");
+        *set_disks.disks.write().await = saved_disks;
+        assert!(matches!(
+            set_disks
+                .check_upload_id_exists(bucket, object, &quorum_upload.upload_id, false)
+                .await,
+            Err(StorageError::InvalidUploadID(..))
+        ));
+
+        let below_quorum_upload = set_disks
+            .new_multipart_upload(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("second multipart upload should be created");
+        let saved_disks = {
+            let mut disks = set_disks.disks.write().await;
+            let saved = disks.clone();
+            disks[2] = None;
+            disks[3] = None;
+            saved
+        };
+        let err = set_disks
+            .abort_multipart_upload(bucket, object, &below_quorum_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect_err("abort must report a delete below write quorum");
+        assert!(matches!(err, StorageError::ErasureWriteQuorum));
+
+        *set_disks.disks.write().await = saved_disks;
+        set_disks
+            .check_upload_id_exists(bucket, object, &below_quorum_upload.upload_id, false)
+            .await
+            .expect("failed abort must leave quorum-visible staging on the restored disks");
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_revalidates_layout_candidate_after_upload_lock() {
@@ -3170,6 +3398,17 @@ mod tests {
             tokio::task::yield_now().await;
             assert!(!abort.is_finished(), "abort must wait until completion releases the upload lock");
 
+            let list_store = set_disks.clone();
+            let list_upload_id = upload_id.clone();
+            let list = tokio::spawn(async move {
+                list_store
+                    .list_object_parts(bucket, object, &list_upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default())
+                    .await
+            });
+            signaling.wait_for_attempts(3).await;
+            tokio::task::yield_now().await;
+            assert!(!list.is_finished(), "ListParts must wait until completion releases the upload lock");
+
             barrier.release();
             complete
                 .await
@@ -3180,10 +3419,68 @@ mod tests {
                 .expect("abort task should not panic")
                 .expect_err("the committed upload should no longer exist when abort acquires the lock");
             assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
+            let list_err = list
+                .await
+                .expect("ListParts task should not panic")
+                .expect_err("the committed upload should no longer exist when ListParts acquires the lock");
+            assert!(matches!(list_err, StorageError::InvalidUploadID(..)));
         })
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn complete_validates_parts_after_an_inflight_upload_part_commit() {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+        let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+        let bucket = "multipart-complete-put-part-race-bucket";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, original_parts) =
+            stage_upload_with_create_opts(&set_disks, bucket, object, &[0x49; 4096], &ObjectOptions::default()).await;
+        let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+        signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+        let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+        let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
+
+        let put_store = set_disks.clone();
+        let put_upload_id = upload_id.clone();
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![0x4a; 4096]);
+            put_store
+                .put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
+                .await
+        });
+        barrier.wait_until_paused().await;
+
+        let complete_store = set_disks.clone();
+        let complete_upload_id = upload_id.clone();
+        let complete = tokio::spawn(async move {
+            complete_store
+                .complete_multipart_upload(bucket, object, &complete_upload_id, original_parts, &ObjectOptions::default())
+                .await
+        });
+        signaling.wait_for_attempts(2).await;
+        tokio::task::yield_now().await;
+        assert!(!complete.is_finished(), "completion must wait for the UploadPart commit lock");
+
+        barrier.release();
+        put.await
+            .expect("UploadPart task should not panic")
+            .expect("UploadPart replacement should commit");
+        let err = complete
+            .await
+            .expect("completion task should not panic")
+            .expect_err("completion must reject the stale ETag after UploadPart wins");
+        assert!(matches!(err, StorageError::InvalidPart(..)));
+        set_disks
+            .list_object_parts(bucket, object, &upload_id, None, MAX_PARTS_COUNT, &ObjectOptions::default())
+            .await
+            .expect("failed completion must leave the upload retryable");
+    }
+
     #[tokio::test(start_paused = true)]
     #[serial]
     async fn complete_fences_upload_lock_loss_before_commit() {

crates/ecstore/src/set_disk/read.rs

@@ -1971,6 +1971,17 @@ mod metadata_cache_tests {
                 let mut fi = valid_test_fileinfo(object);
                 fi.mod_time = Some(OffsetDateTime::now_utc());
                 fi.erasure.index = fi.erasure.distribution[disk_index];
+                // `valid_test_fileinfo` carries a positive size with no parts —
+                // the shape `validate_collection_contents` rejects as
+                // `FileCorrupt` — so it can drive
+                // `get_object_with_fileinfo_rejects_positive_size_without_parts`.
+                // Metadata that has to survive a write needs the matching part.
+                fi.parts.push(ObjectPartInfo {
+                    number: 1,
+                    size: 1,
+                    actual_size: 1,
+                    ..Default::default()
+                });
                 disk.write_metadata(bucket, bucket, object, fi)
                     .await
                     .expect("metadata should be written before quorum read");

crates/ecstore/src/store/object.rs

@@ -2601,10 +2601,10 @@ mod tests {
     // (backlog#1304): restore entry no longer serializes on the object lock.
     // The replacement semantics — non-blocking reads during the copy-back and
     // fast rejection of a concurrent restore — are covered end-to-end by
-    // `restore_object_usecase_reports_ongoing_conflict_and_completion`
-    // (rustfs/src/app/lifecycle_transition_api_test.rs) and at the lock level
-    // by the accept-guard test below; restore-vs-reader data protection lives
-    // in the inner put_object/complete_multipart_upload commit locks.
+    // `restore_object_usecase_reports_ongoing_conflict`
+    // (rustfs/src/app/lifecycle_transition_api_test.rs), while the SetDisks
+    // transition matrix covers the final local commit. Restore-vs-reader data
+    // protection lives in the inner put_object/complete_multipart_upload locks.
     #[tokio::test]
     #[serial_test::serial]
     async fn restore_accept_guard_serializes_concurrent_accepts() {

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -60,7 +60,6 @@ use uuid::Uuid;
 static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
 static INIT: Once = Once::new();
 const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
-const RESTORE_COPY_BACK_WAIT_TIMEOUT: Duration = Duration::from_secs(60);
 const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
 const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
 const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
@@ -339,45 +338,6 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
     }
 }
 
-async fn wait_for_restore_completion(
-    ecstore: &Arc<ECStore>,
-    backend: &MockWarmBackend,
-    bucket: &str,
-    object: &str,
-    timeout: Duration,
-) -> Result<ObjectInfo, String> {
-    let deadline = tokio::time::Instant::now() + timeout;
-    let mut last_state = None;
-
-    loop {
-        if tokio::time::Instant::now() >= deadline {
-            let tier_gets = backend.get_count().await;
-            let op_log = backend.op_log().await;
-            return Err(format!(
-                "restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}",
-                last_state.unwrap_or_else(|| "no object info observed".to_string())
-            ));
-        }
-
-        match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
-            Ok(info) => {
-                if !info.restore_ongoing && info.restore_expires.is_some() {
-                    return Ok(info);
-                }
-                last_state = Some(format!(
-                    "restore_ongoing={}, restore_expires={:?}, transitioned_status={}",
-                    info.restore_ongoing, info.restore_expires, info.transitioned_object.status
-                ));
-            }
-            Err(err) => {
-                last_state = Some(format!("get_object_info failed: {err}"));
-            }
-        }
-
-        tokio::time::sleep(Duration::from_millis(500)).await;
-    }
-}
-
 // SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
 // runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
 // `env::set_var` / `env::remove_var` window.
@@ -2097,10 +2057,9 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() {
 /// POST restore(days=1) is accepted and flips the object to
 /// `x-amz-restore: ongoing-request="true"` while the mock tier GET barrier
 /// proves the background copy-back has reached the remote read; a second POST
-/// during that window is rejected with 409 `RestoreAlreadyInProgress`; once the
-/// copy-back completes the object reports `ongoing-request="false"` with a
-/// future expiry-date; and a full GET is then served from the local restored
-/// copy (the mock tier records no further `get` calls).
+/// during that window is rejected with 409 `RestoreAlreadyInProgress`.
+/// Synchronous SetDisks transition tests cover copy-back completion, restore
+/// metadata, and local byte-identical reads.
 ///
 /// Re-enabled in the serial lane by backlog#1304: the accept path now flips
 /// the ongoing flag under a short compare-and-set guard and the copy-back
@@ -2110,7 +2069,7 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() {
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
-async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
+async fn restore_object_usecase_reports_ongoing_conflict() {
     let (_disk_paths, ecstore) = setup_test_env().await;
     let usecase = DefaultObjectUsecase::from_global();
 
@@ -2177,35 +2136,6 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
     );
 
     get_barrier.release();
-
-    // Completion: ongoing flips to false and a future expiry-date appears.
-    let completed =
-        wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_COPY_BACK_WAIT_TIMEOUT).await;
-    let completed = completed.unwrap_or_else(|err| panic!("{err}"));
-
-    let now_secs = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .expect("clock before unix epoch")
-        .as_secs() as i64;
-    let expires = completed.restore_expires.expect("completed restore carries an expiry");
-    assert!(
-        expires.unix_timestamp() > now_secs,
-        "restore expiry-date must be in the future, got {expires}"
-    );
-    assert_eq!(
-        completed.transitioned_object.status, "complete",
-        "restore must not clear the transitioned state"
-    );
-
-    // The restored copy serves GET locally: no further tier GETs.
-    let tier_gets_after_restore = backend.get_count().await;
-    let data = read_object_bytes(&ecstore, bucket.as_str(), object).await;
-    assert_eq!(data, payload, "restored GET must return the original bytes");
-    assert_eq!(
-        backend.get_count().await,
-        tier_gets_after_restore,
-        "GET of a restored object must be served locally, not from the tier"
-    );
 }
 
 /// backlog#1304: the restore-accept compare-and-set itself, under real

scripts/README.md

@@ -90,7 +90,7 @@ their issue closes.
 | `manual_transition_journal_audit.sh` | dev-tool | Journal + metrics + log audit for manual transition jobs | — |
 | `manual_transition_mixed_rollout_matrix.sh` | dev-tool | Matrix generator for mixed-version rollout phases | — |
 | `manual_transition_mixed_rollout_runbook.sh` | dev-tool | Reusable mixed-version rollout runbook generator (external run) | — |
-| `manual_transition_mixed_version_docker_harness.sh` | dev-tool | Dedicated #1508 Docker harness for old/new manual-transition rollout evidence | `test_manual_transition_runbooks.sh` |
+| `manual_transition_mixed_version_docker_harness.sh` | dev-tool | Dedicated #1508 Docker harness for old/new manual-transition rollout evidence with strict/baseline/blocked result classification | `test_manual_transition_runbooks.sh` |
 | `monitor_manual_transition_ci.sh` | dev-tool | CI workflow/status watcher for manual transition follow-up monitoring | — |
 | `manual_transition_soak_matrix.sh` | dev-tool | Matrix generator for nightly stress windows | — |
 | `manual_transition_nightly_stress_runbook.sh` | dev-tool | Nightly stress entrypoint with failure snapshot templates | — |

scripts/manual_transition_mixed_version_docker_harness.sh

@@ -25,9 +25,13 @@ COLD_IMAGE="${COLD_IMAGE:-${NEW_IMAGE}}"
 BASE_PORT="${BASE_PORT:-19400}"
 OUT_DIR="${OUT_DIR:-${PROJECT_ROOT}/target/manual-transition-1508-docker/$(date +%Y%m%dT%H%M%S)}"
 KEEP_UP=false
-ROLLBACK_NEW2_TO_OLD=true
+ROLLBACK_PHASE="${ROLLBACK_PHASE:-after-terminal}"
+OLD_NODE_PHASE="${OLD_NODE_PHASE:-initial}"
 WAIT_TIMEOUT_SECS="${WAIT_TIMEOUT_SECS:-180}"
 POLL_SECONDS="${POLL_SECONDS:-180}"
+TRANSITION_WORKERS="${TRANSITION_WORKERS:-2}"
+TRANSITION_QUEUE_CAPACITY="${TRANSITION_QUEUE_CAPACITY:-64}"
+FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT="${FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT:-false}"
 
 HOT_ACCESS_KEY="${HOT_ACCESS_KEY:-mvadmin}"
 HOT_SECRET_KEY="${HOT_SECRET_KEY:-mvsecret}"
@@ -58,18 +62,31 @@ Options:
   --object-count <n>           Non-empty probe object count
   --tier <name>                Remote tier name
   --keep-up                    Leave Docker services running
-  --no-rollback                Do not replace node2 with old image after job admission
+  --rollback-phase <phase>     Rollback node2 timing: after-terminal, in-flight, none
+  --old-node-phase <phase>     Old node1 timing: initial, before-job
+  --no-rollback                Alias for --rollback-phase none
   -h, --help                   Show help
 
 Environment:
   PROJECT_NAME OLD_IMAGE NEW_IMAGE COLD_IMAGE BASE_PORT OUT_DIR KEEP_UP
   HOT_ACCESS_KEY HOT_SECRET_KEY COLD_ACCESS_KEY COLD_SECRET_KEY
   TIER_NAME TIER_BUCKET TIER_PREFIX JOB_BUCKET JOB_PREFIX OBJECT_COUNT
-  WAIT_TIMEOUT_SECS POLL_SECONDS AWS_SIGV4_SCOPE
+  ROLLBACK_PHASE WAIT_TIMEOUT_SECS POLL_SECONDS TRANSITION_WORKERS
+  TRANSITION_QUEUE_CAPACITY OLD_NODE_PHASE FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT AWS_SIGV4_SCOPE
 
 Artifacts:
   compose.yml, image inspect files, health/readiness logs, API responses,
   terminal status, old-node readback, container logs, summary.env.
+
+Result classifications:
+  strict_mixed_rollout_pass          Real old/new images, non-empty completed transition, zero failures
+  baseline_tiered_storage_pass      Same old/new image completed transition; useful baseline, not #1508 closure
+  blocked_manual_api_not_implemented Manual transition API returned 501 before job admission
+  blocked_manual_api_unavailable    Manual transition API did not return a usable job_id
+  blocked_cluster_readiness_failed  Docker cluster did not reach health/readiness before admission
+  blocked_empty_scan_or_lifecycle   Job completed without lifecycle-matching transition work
+  blocked_manual_job_preempted_by_lifecycle_queue Lifecycle/immediate transition queued work before the job
+  strict_mixed_rollout_fail         Mixed rollout ran but did not satisfy the strict #1508 gate
 USAGE
 }
 
@@ -111,6 +128,28 @@ parse_positive_int() {
   fi
 }
 
+validate_rollback_phase() {
+  case "$ROLLBACK_PHASE" in
+    after-terminal|in-flight|none)
+      ;;
+    *)
+      log_error "--rollback-phase must be one of: after-terminal, in-flight, none"
+      exit 1
+      ;;
+  esac
+}
+
+validate_old_node_phase() {
+  case "$OLD_NODE_PHASE" in
+    initial|before-job)
+      ;;
+    *)
+      log_error "--old-node-phase must be one of: initial, before-job"
+      exit 1
+      ;;
+  esac
+}
+
 parse_args() {
   while [[ $# -gt 0 ]]; do
     case "$1" in
@@ -150,8 +189,16 @@ parse_args() {
         KEEP_UP=true
         shift
         ;;
+      --rollback-phase)
+        ROLLBACK_PHASE="$(arg_value "$1" "${2:-}")"
+        shift 2
+        ;;
+      --old-node-phase)
+        OLD_NODE_PHASE="$(arg_value "$1" "${2:-}")"
+        shift 2
+        ;;
       --no-rollback)
-        ROLLBACK_NEW2_TO_OLD=false
+        ROLLBACK_PHASE=none
         shift
         ;;
       -h|--help)
@@ -198,12 +245,18 @@ cleanup() {
 }
 
 write_compose_file() {
+  local node1_image
   COMPOSE_FILE="${OUT_DIR}/compose.yml"
+  node1_image="$OLD_IMAGE"
+  if [[ "$OLD_NODE_PHASE" == "before-job" ]]; then
+    node1_image="$NEW_IMAGE"
+  fi
   cat >"$COMPOSE_FILE" <<EOF
 services:
   cold:
     image: ${COLD_IMAGE}
     hostname: cold
+    user: "0:0"
     environment:
       - RUSTFS_ADDRESS=:9000
       - RUSTFS_ACCESS_KEY=${COLD_ACCESS_KEY}
@@ -222,14 +275,20 @@ services:
       - rustfs-1508-net
 
   node1:
-    image: ${OLD_IMAGE}
+    image: ${node1_image}
     hostname: node1
+    user: "0:0"
     environment: &hot-env
       - RUSTFS_ADDRESS=:9000
       - RUSTFS_ACCESS_KEY=${HOT_ACCESS_KEY}
       - RUSTFS_SECRET_KEY=${HOT_SECRET_KEY}
       - RUSTFS_VOLUMES=$(hot_volumes)
       - RUSTFS_SCANNER_ENABLED=false
+      - RUSTFS_SCANNER_CYCLE=3600
+      - RUSTFS_SCANNER_START_DELAY_SECS=3600
+      - RUSTFS_MAX_TRANSITION_WORKERS=${TRANSITION_WORKERS}
+      - RUSTFS_TRANSITION_QUEUE_CAPACITY=${TRANSITION_QUEUE_CAPACITY}
+      - RUSTFS_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT=${FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT}
       - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true
       - RUSTFS_OBS_LOGGER_LEVEL=warn
     volumes:
@@ -245,6 +304,7 @@ services:
   node2:
     image: ${NEW_IMAGE}
     hostname: node2
+    user: "0:0"
     environment: *hot-env
     volumes:
       - node2_data_0:/data/rustfs0
@@ -259,6 +319,7 @@ services:
   node3:
     image: ${NEW_IMAGE}
     hostname: node3
+    user: "0:0"
     environment: *hot-env
     volumes:
       - node3_data_0:/data/rustfs0
@@ -273,6 +334,7 @@ services:
   node4:
     image: ${NEW_IMAGE}
     hostname: node4
+    user: "0:0"
     environment: *hot-env
     volumes:
       - node4_data_0:/data/rustfs0
@@ -433,16 +495,19 @@ seed_objects() {
 }
 
 start_transition_job() {
-  local query response job_id
+  local query response job_id run_http_code
   query="bucket=$(url_encode "$JOB_BUCKET")"
   query="${query}&prefix=$(url_encode "${JOB_PREFIX}/")"
   query="${query}&tier=$(url_encode "$TIER_NAME")"
   query="${query}&dryRun=false&maxObjects=${OBJECT_COUNT}&mode=async"
-  response="$(curl_hot POST "$(hot_endpoint 2)/rustfs/admin/v3/ilm/transition/run?${query}")"
-  printf '%s\n' "$response" >"${OUT_DIR}/run-response.json"
+  curl_hot POST "$(hot_endpoint 2)/rustfs/admin/v3/ilm/transition/run?${query}" \
+    -o "${OUT_DIR}/run-response.json" \
+    -w "%{http_code}\n" >"${OUT_DIR}/run-response.http_code" || true
+  response="$(cat "${OUT_DIR}/run-response.json" 2>/dev/null || true)"
+  run_http_code="$(cat "${OUT_DIR}/run-response.http_code" 2>/dev/null || true)"
   job_id="$(printf '%s' "$response" | jq -r '.job_id // empty')"
   if [[ -z "$job_id" ]]; then
-    log_error "manual transition response omitted job_id"
+    log_warn "manual transition response omitted job_id, http_code=${run_http_code:-unknown}"
     return 1
   fi
   printf '%s\n' "$job_id" >"${OUT_DIR}/job-id.txt"
@@ -451,19 +516,25 @@ start_transition_job() {
 replace_node2_with_old_image() {
   local network
   network="$(network_name)"
-  log_info "Replacing node2 with old image ${OLD_IMAGE} for in-flight rollback readback"
+  log_info "Replacing node2 with old image ${OLD_IMAGE} for ${ROLLBACK_PHASE} rollback readback"
   compose stop node2 >/dev/null
   compose rm -f node2 >/dev/null
   docker run -d \
     --name "${PROJECT_NAME}-node2-rollback-old" \
     --network "$network" \
     --network-alias node2 \
+    --user 0:0 \
     -p "$((BASE_PORT + 2)):9000" \
     -e RUSTFS_ADDRESS=:9000 \
     -e RUSTFS_ACCESS_KEY="$HOT_ACCESS_KEY" \
     -e RUSTFS_SECRET_KEY="$HOT_SECRET_KEY" \
     -e RUSTFS_VOLUMES="$(hot_volumes)" \
     -e RUSTFS_SCANNER_ENABLED=false \
+    -e RUSTFS_SCANNER_CYCLE=3600 \
+    -e RUSTFS_SCANNER_START_DELAY_SECS=3600 \
+    -e RUSTFS_MAX_TRANSITION_WORKERS="$TRANSITION_WORKERS" \
+    -e RUSTFS_TRANSITION_QUEUE_CAPACITY="$TRANSITION_QUEUE_CAPACITY" \
+    -e RUSTFS_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT="$FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT" \
     -e RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
     -e RUSTFS_OBS_LOGGER_LEVEL=warn \
     -v "${PROJECT_NAME}_node2_data_0:/data/rustfs0" \
@@ -473,6 +544,37 @@ replace_node2_with_old_image() {
     "$OLD_IMAGE" >/dev/null
 }
 
+replace_node1_with_old_image() {
+  local network
+  network="$(network_name)"
+  log_info "Replacing node1 with old image ${OLD_IMAGE} before manual transition job"
+  compose stop node1 >/dev/null
+  compose rm -f node1 >/dev/null
+  docker run -d \
+    --name "${PROJECT_NAME}-node1-before-job-old" \
+    --network "$network" \
+    --network-alias node1 \
+    --user 0:0 \
+    -p "$((BASE_PORT + 1)):9000" \
+    -e RUSTFS_ADDRESS=:9000 \
+    -e RUSTFS_ACCESS_KEY="$HOT_ACCESS_KEY" \
+    -e RUSTFS_SECRET_KEY="$HOT_SECRET_KEY" \
+    -e RUSTFS_VOLUMES="$(hot_volumes)" \
+    -e RUSTFS_SCANNER_ENABLED=false \
+    -e RUSTFS_SCANNER_CYCLE=3600 \
+    -e RUSTFS_SCANNER_START_DELAY_SECS=3600 \
+    -e RUSTFS_MAX_TRANSITION_WORKERS="$TRANSITION_WORKERS" \
+    -e RUSTFS_TRANSITION_QUEUE_CAPACITY="$TRANSITION_QUEUE_CAPACITY" \
+    -e RUSTFS_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT="$FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT" \
+    -e RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
+    -e RUSTFS_OBS_LOGGER_LEVEL=warn \
+    -v "${PROJECT_NAME}_node1_data_0:/data/rustfs0" \
+    -v "${PROJECT_NAME}_node1_data_1:/data/rustfs1" \
+    -v "${PROJECT_NAME}_node1_data_2:/data/rustfs2" \
+    -v "${PROJECT_NAME}_node1_data_3:/data/rustfs3" \
+    "$OLD_IMAGE" >/dev/null
+}
+
 poll_terminal_status() {
   local job_id="$1"
   local status_url status_json terminal_state
@@ -512,6 +614,85 @@ head_probe() {
     -w "%{http_code}\n" >"${OUT_DIR}/head-object.http_code" || true
 }
 
+image_id() {
+  local file="$1"
+  jq -r '.[0].Id // ""' "$file" 2>/dev/null || true
+}
+
+is_compat_readback_code() {
+  case "$1" in
+    200|501)
+      return 0
+      ;;
+    *)
+      return 1
+      ;;
+  esac
+}
+
+classify_result() {
+  local terminal_state="$1"
+  local transition_completed="$2"
+  local transition_failed="$3"
+  local tier_failure="$4"
+  local old_code="$5"
+  local rollback_code="$6"
+  local run_http_code="$7"
+  local lifecycle_config_found="$8"
+  local scanned="$9"
+  local eligible="${10}"
+  local skipped_already_transitioned="${11}"
+  local skipped_already_in_flight="${12}"
+  local old_image_id new_image_id images_are_mixed readback_ok
+
+  if [[ -f "${OUT_DIR}/readiness-failed" ]]; then
+    printf 'blocked_cluster_readiness_failed\n'
+    return
+  fi
+
+  old_image_id="$(image_id "${OUT_DIR}/old-image.inspect.json")"
+  new_image_id="$(image_id "${OUT_DIR}/new-image.inspect.json")"
+  images_are_mixed=false
+  if [[ "$OLD_IMAGE" != "$NEW_IMAGE" && -n "$old_image_id" && -n "$new_image_id" && "$old_image_id" != "$new_image_id" ]]; then
+    images_are_mixed=true
+  fi
+
+  readback_ok=false
+  if is_compat_readback_code "$old_code"; then
+    if [[ "$ROLLBACK_PHASE" == "none" ]] || is_compat_readback_code "$rollback_code"; then
+      readback_ok=true
+    fi
+  fi
+
+  if [[ "$run_http_code" == "501" ]]; then
+    printf 'blocked_manual_api_not_implemented\n'
+    return
+  fi
+  if [[ -z "$(cat "${OUT_DIR}/job-id.txt" 2>/dev/null || true)" ]]; then
+    printf 'blocked_manual_api_unavailable\n'
+    return
+  fi
+  if [[ "$terminal_state" == "completed" && ( "$lifecycle_config_found" != "true" || "$scanned" == "0" || "$eligible" == "0" || "$transition_completed" == "0" ) ]]; then
+    printf 'blocked_empty_scan_or_lifecycle\n'
+    return
+  fi
+  if [[ "$transition_completed" == "0" && "$eligible" != "0" && "$transition_failed" == "0" && "$tier_failure" == "0" ]]; then
+    if [[ "$skipped_already_transitioned" != "0" || "$skipped_already_in_flight" != "0" ]]; then
+      printf 'blocked_manual_job_preempted_by_lifecycle_queue\n'
+      return
+    fi
+  fi
+  if [[ "$terminal_state" == "completed" && "$transition_completed" != "0" && "$transition_failed" == "0" && "$tier_failure" == "0" ]]; then
+    if [[ "$images_are_mixed" == "true" && "$readback_ok" == "true" ]]; then
+      printf 'strict_mixed_rollout_pass\n'
+      return
+    fi
+    printf 'baseline_tiered_storage_pass\n'
+    return
+  fi
+  printf 'strict_mixed_rollout_fail\n'
+}
+
 collect_logs() {
   local service
   if [[ -z "$COMPOSE_FILE" || ! -f "$COMPOSE_FILE" ]]; then
@@ -520,37 +701,63 @@ collect_logs() {
   for service in cold node1 node2 node3 node4; do
     compose logs --no-color "$service" >"${OUT_DIR}/${service}.log" 2>/dev/null || true
   done
+  docker logs "${PROJECT_NAME}-node1-before-job-old" >"${OUT_DIR}/node1-before-job-old.log" 2>/dev/null || true
   docker logs "${PROJECT_NAME}-node2-rollback-old" >"${OUT_DIR}/node2-rollback-old.log" 2>/dev/null || true
 }
 
 summarize() {
-  local terminal_state transition_completed tier_failure transition_failed old_code rollback_code
+  local terminal_state transition_completed tier_failure transition_failed old_code rollback_code run_http_code lifecycle_config_found scanned eligible skipped_already_transitioned skipped_already_in_flight queue_queued queue_active result_classification
   terminal_state="$(cat "${OUT_DIR}/terminal-state.txt" 2>/dev/null || true)"
   transition_completed="$(jq -r '.report.transition_completed // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
   tier_failure="$(jq -r '.report.tier_failure // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
   transition_failed="$(jq -r '.report.transition_failed // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
   old_code="$(cat "${OUT_DIR}/old-node-status.http_code" 2>/dev/null || true)"
   rollback_code="$(cat "${OUT_DIR}/rollback-node2-status.http_code" 2>/dev/null || true)"
+  run_http_code="$(cat "${OUT_DIR}/run-response.http_code" 2>/dev/null || true)"
+  lifecycle_config_found="$(jq -r '.report.lifecycle_config_found // false' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf 'false')"
+  scanned="$(jq -r '.report.scanned // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  eligible="$(jq -r '.report.eligible // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  skipped_already_transitioned="$(jq -r '.report.skipped_already_transitioned // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  skipped_already_in_flight="$(jq -r '.report.skipped_already_in_flight // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  queue_queued="$(jq -r '.queue_snapshot.queued // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  queue_active="$(jq -r '.queue_snapshot.active // 0' "${OUT_DIR}/status-terminal.json" 2>/dev/null || printf '0')"
+  result_classification="$(classify_result "$terminal_state" "$transition_completed" "$transition_failed" "$tier_failure" "$old_code" "$rollback_code" "$run_http_code" "$lifecycle_config_found" "$scanned" "$eligible" "$skipped_already_transitioned" "$skipped_already_in_flight")"
   cat >"${OUT_DIR}/summary.env" <<EOF
 project_name=${PROJECT_NAME}
 old_image=${OLD_IMAGE}
 new_image=${NEW_IMAGE}
 cold_image=${COLD_IMAGE}
+old_image_id=$(image_id "${OUT_DIR}/old-image.inspect.json")
+new_image_id=$(image_id "${OUT_DIR}/new-image.inspect.json")
 base_port=${BASE_PORT}
 tier=${TIER_NAME}
 job_bucket=${JOB_BUCKET}
 job_prefix=${JOB_PREFIX}
 object_count=${OBJECT_COUNT}
+transition_workers=${TRANSITION_WORKERS}
+transition_queue_capacity=${TRANSITION_QUEUE_CAPACITY}
+force_immediate_transition_enqueue_timeout=${FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT}
+run_http_code=${run_http_code}
 terminal_state=${terminal_state}
+lifecycle_config_found=${lifecycle_config_found}
+scanned=${scanned}
+eligible=${eligible}
+skipped_already_transitioned=${skipped_already_transitioned}
+skipped_already_in_flight=${skipped_already_in_flight}
 transition_completed=${transition_completed}
 transition_failed=${transition_failed}
 tier_failure=${tier_failure}
+queue_queued=${queue_queued}
+queue_active=${queue_active}
 old_node_status_http_code=${old_code}
 rollback_node2_status_http_code=${rollback_code}
-rollback_new2_to_old=${ROLLBACK_NEW2_TO_OLD}
+rollback_phase=${ROLLBACK_PHASE}
+old_node_phase=${OLD_NODE_PHASE}
+rollback_new2_to_old=$([[ "$ROLLBACK_PHASE" == "none" ]] && printf 'false' || printf 'true')
+result_classification=${result_classification}
 EOF
   cat "${OUT_DIR}/summary.env"
-  if [[ "$terminal_state" != "completed" || "$transition_completed" == "0" || "$transition_failed" != "0" || "$tier_failure" != "0" ]]; then
+  if [[ "$result_classification" != "strict_mixed_rollout_pass" ]]; then
     log_error "strict #1508 mixed-version transition gate failed; see ${OUT_DIR}"
     return 1
   fi
@@ -560,6 +767,8 @@ main() {
   parse_args "$@"
   parse_positive_int "--base-port" "$BASE_PORT"
   parse_positive_int "--object-count" "$OBJECT_COUNT"
+  validate_rollback_phase
+  validate_old_node_phase
   require_cmd docker
   require_cmd curl
   require_cmd jq
@@ -574,20 +783,37 @@ main() {
   docker image inspect "$COLD_IMAGE" >"${OUT_DIR}/cold-image.inspect.json"
 
   compose up -d
-  wait_cluster_ready
+  if ! wait_cluster_ready; then
+    printf 'cluster readiness failed before manual transition admission\n' >"${OUT_DIR}/readiness-failed"
+    collect_logs
+    summarize
+    return 1
+  fi
 
   curl_cold PUT "$(cold_endpoint)/${TIER_BUCKET}" -o "${OUT_DIR}/create-cold-bucket.response" -w "%{http_code}\n" >"${OUT_DIR}/create-cold-bucket.http_code"
   create_bucket "$(hot_endpoint 2)" "$JOB_BUCKET" "${OUT_DIR}/create-hot-bucket"
   add_tier
-  put_lifecycle
   seed_objects
-  start_transition_job
-  if [[ "$ROLLBACK_NEW2_TO_OLD" == "true" ]]; then
-    replace_node2_with_old_image
+  put_lifecycle
+  if [[ "$OLD_NODE_PHASE" == "before-job" ]]; then
+    replace_node1_with_old_image
+    wait_http_ok "$(hot_endpoint 1)/health" "node1-old-live" || true
+    wait_http_ok "$(hot_endpoint 1)/health/ready" "node1-old-ready" || true
+  fi
+  if start_transition_job; then
+    if [[ "$ROLLBACK_PHASE" == "in-flight" ]]; then
+      replace_node2_with_old_image
+    fi
+    poll_terminal_status "$(cat "${OUT_DIR}/job-id.txt")" || true
+    if [[ "$ROLLBACK_PHASE" == "after-terminal" ]]; then
+      replace_node2_with_old_image
+      wait_http_ok "$(hot_endpoint 2)/health" "rollback-node2-live" || true
+    fi
+    capture_old_node_readback "$(cat "${OUT_DIR}/job-id.txt")"
+    head_probe
+  else
+    log_warn "Skipping terminal polling because no manual transition job was admitted"
   fi
-  poll_terminal_status "$(cat "${OUT_DIR}/job-id.txt")"
-  capture_old_node_readback "$(cat "${OUT_DIR}/job-id.txt")"
-  head_probe
   collect_logs
   summarize
 }

scripts/test_manual_transition_runbooks.sh

@@ -208,7 +208,16 @@ bash "$MIXED_DOCKER_HARNESS" --help >/tmp/manual_transition_mixed_version_docker
 rg -q "mixed_version_docker_harness" /tmp/manual_transition_mixed_version_docker_harness.help
 rg -q -- "--old-image" /tmp/manual_transition_mixed_version_docker_harness.help
 rg -q -- "--new-image" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q -- "--rollback-phase" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q -- "--old-node-phase" /tmp/manual_transition_mixed_version_docker_harness.help
 rg -q -- "--no-rollback" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "strict_mixed_rollout_pass" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "baseline_tiered_storage_pass" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "blocked_manual_api_not_implemented" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "blocked_cluster_readiness_failed" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "blocked_manual_job_preempted_by_lifecycle_queue" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "OLD_NODE_PHASE" /tmp/manual_transition_mixed_version_docker_harness.help
+rg -q "FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT" /tmp/manual_transition_mixed_version_docker_harness.help
 if bash "$FAILURE_SAMPLES" --endpoint http://127.0.0.1:9000 --sample >/tmp/manual_transition_failure_samples.err 2>&1; then
   echo "failure samples script should fail when --sample has no value" >&2
   exit 1