Merge remote-tracking branch 'origin/main' into cxymds/fix-read-version-test-fixture

.agents/skills/arch-checks/SKILL.md

@@ -10,16 +10,10 @@ never weaken a check to get green.
 
 ## `check_layer_dependencies.sh` — layer DAG in `rustfs/src`
 
-Enforces `composition (server, startup/init) → interface (admin,
-storage/ecfs, storage/s3_api) → app → infra`; no upward imports. Server source
-files are composition roots, while imports of their exported HTTP contracts
-are classified as interface dependencies. Known legacy violations live in
+Enforces `interface (admin, storage/ecfs, storage/s3_api) → app → infra`; no
+upward imports. Known legacy violations live in
 `scripts/layer-dependency-baseline.txt`.
 
-Dedicated `*_test.rs` and `tests/` modules are outside this production guard.
-Inline `#[cfg(test)]` imports remain checked under their source file's layer;
-move architecture-crossing test scaffolding into a dedicated test module.
-
 - **New violation**: restructure your change so the dependency points
   downward (move the shared type/function to the lower layer).
 - **You legitimately removed a baseline entry**: run

.config/nextest.toml

@@ -1,14 +1,17 @@
 # nextest configuration for RustFS.
 #
-# Serialize the ecstore tests that share the process-wide disk registry or
-# exercise a multi-disk commit handoff across nextest process boundaries.
+# 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:
 #
 #   * 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
-#     uses the shared multipart fixture and a deterministic uploadId-lock
-#     handoff, so it must not overlap another process mutating that fixture.
+#     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.
 #
 # serial_test's #[serial] attribute does NOT serialize these across runs:
 # nextest executes each test in its own process, where the in-process
@@ -97,6 +100,13 @@ 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]]
@@ -104,11 +114,6 @@ 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).

Cargo.lock

@@ -9732,7 +9732,6 @@ dependencies = [
  "rustfs-policy",
  "rustfs-rio",
  "rustfs-storage-api",
- "rustfs-test-utils",
  "rustfs-tls-runtime",
  "rustfs-trusted-proxies",
  "rustfs-utils",

crates/ecstore/src/api/mod.rs

@@ -128,7 +128,7 @@ pub mod bucket {
             get_object_lock_config, get_public_access_block_config, get_quota_config, get_replication_config,
             get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config,
             init_bucket_metadata_sys, list_bucket_targets, remove_bucket_metadata, set_bucket_metadata, update,
-            update_bucket_targets_under_transaction_lock, update_config_with,
+            update_bucket_targets_under_transaction_lock,
         };
     }
 

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

@@ -11112,7 +11112,6 @@ 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;
@@ -11134,36 +11133,51 @@ mod tests {
             })
             .collect();
 
-        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;
+        // 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))
+                });
+            }
 
-        // 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"));
-        }
+            // 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;
         assert_eq!(results.len(), candidates.len());
 
         // Exactly one generation is visible after the serialized commits, and its

crates/ecstore/src/bucket/metadata.rs

@@ -737,9 +737,6 @@ impl BucketMetadata {
             }
             BUCKET_TAGGING_CONFIG => {
                 self.tagging_config_xml = data;
-                // Drop the parsed form (like lifecycle above) so clearing the
-                // payload can't leave stale parsed tags to be cached.
-                self.tagging_config = None;
                 self.tagging_config_updated_at = updated;
             }
             BUCKET_QUOTA_CONFIG_FILE => {
@@ -1321,30 +1318,6 @@ mod test {
         assert!(bm.lifecycle_config.is_none());
     }
 
-    /// Companion to the lifecycle case above. `parse_all_configs` skips empty
-    /// XML rather than clearing, so without the explicit reset a cleared
-    /// tagging config would keep serving the previously parsed tags.
-    #[test]
-    fn tagging_update_config_clears_parsed_config_on_delete() {
-        let mut bm = BucketMetadata::new("test-bucket");
-        let tagging_xml = br#"<Tagging><TagSet><Tag><Key>env</Key><Value>prod</Value></Tag></TagSet></Tagging>"#;
-
-        bm.update_config(BUCKET_TAGGING_CONFIG, tagging_xml.to_vec())
-            .expect("tagging config should update");
-        bm.parse_all_configs().expect("tagging config should parse");
-        assert!(bm.tagging_config.is_some());
-
-        bm.update_config(BUCKET_TAGGING_CONFIG, Vec::new())
-            .expect("tagging config delete should update metadata");
-
-        assert!(bm.tagging_config_xml.is_empty());
-        assert!(bm.tagging_config.is_none());
-
-        // A re-parse must not resurrect them either.
-        bm.parse_all_configs().expect("cleared tagging should parse");
-        assert!(bm.tagging_config.is_none());
-    }
-
     #[tokio::test]
     async fn marshal_msg_complete_example() {
         // Create a complete BucketMetadata with various configurations

crates/ecstore/src/bucket/metadata_sys.rs

@@ -239,35 +239,6 @@ pub async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Ve
     bucket_meta_sys.update(bucket, BUCKET_TARGETS_FILE, data).await
 }
 
-/// Read-modify-write one bucket config file under the metadata system's
-/// outer write guard.
-///
-/// `mutate` sees the freshly loaded on-disk metadata and returns the
-/// replacement payload for `config_file` (empty clears it, like
-/// [`delete`]). Both the read and the persisted write happen inside the
-/// same guard that [`update`] uses, so within this process the rewrite can
-/// neither clobber a concurrent update to another config file nor lose a
-/// concurrent write to the same one — unlike caching a mutated clone of
-/// previously read metadata.
-///
-/// This guard is process-local. Writers on other nodes still race, exactly
-/// as they do for [`update`]: each rewrites the whole metadata file, so the
-/// later save wins. What this narrows is the window — from "as stale as the
-/// local cache" down to a single metadata read plus write.
-pub async fn update_config_with<F>(bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
-where
-    F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
-{
-    let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
-    let _targets_guard = if config_file == BUCKET_TARGETS_FILE {
-        Some(acquire_bucket_targets_transaction_lock(bucket).await?)
-    } else {
-        None
-    };
-    let mut bucket_meta_sys = bucket_meta_sys_lock.write().await;
-    bucket_meta_sys.update_config_with(bucket, config_file, mutate).await
-}
-
 pub async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result<rustfs_lock::NamespaceLockGuard> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let api = bucket_meta_sys_lock.read().await.object_store();
@@ -632,56 +603,30 @@ impl BucketMetadataSys {
             return Err(Error::other("errServerNotInitialized"));
         };
 
-        let mut bm = Self::load_bucket_metadata_for_update(store, bucket, parse).await?;
-
-        let updated = bm.update_config(config_file, data)?;
-
-        self.save(bm).await?;
-
-        Ok(updated)
-    }
-
-    /// See the free [`update_config_with`]: same load-mutate-persist cycle as
-    /// [`Self::update`], with the payload computed from the loaded metadata
-    /// instead of supplied up front. Loads through this system's own store so
-    /// the read and the persisted write target the same instance.
-    async fn update_config_with<F>(&mut self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
-    where
-        F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
-    {
-        let mut bm = Self::load_bucket_metadata_for_update(self.api.clone(), bucket, true).await?;
-
-        let data = mutate(&bm)?;
-        let updated = bm.update_config(config_file, data)?;
-
-        self.save(bm).await?;
-
-        Ok(updated)
-    }
-
-    /// Load a bucket's on-disk metadata as the base of a config rewrite.
-    /// Outside erasure setups a missing metadata file degrades to a fresh
-    /// default (legacy buckets without one); erasure setups fail instead of
-    /// fabricating state that a quorum may still hold.
-    async fn load_bucket_metadata_for_update(store: Arc<ECStore>, bucket: &str, parse: bool) -> Result<BucketMetadata> {
         if is_meta_bucketname(bucket) {
             return Err(Error::other("errInvalidArgument"));
         }
 
-        match load_bucket_metadata_parse(store, bucket, parse).await {
-            Ok(res) => Ok(res),
+        let mut bm = match load_bucket_metadata_parse(store, bucket, parse).await {
+            Ok(res) => res,
             Err(err) => {
                 if !runtime_sources::setup_is_erasure().await
                     && !runtime_sources::setup_is_dist_erasure().await
                     && is_err_bucket_not_found(&err)
                 {
-                    Ok(BucketMetadata::new(bucket))
+                    BucketMetadata::new(bucket)
                 } else {
                     error!("load bucket metadata failed: {}", err);
-                    Err(err)
+                    return Err(err);
                 }
             }
-        }
+        };
+
+        let updated = bm.update_config(config_file, data)?;
+
+        self.save(bm).await?;
+
+        Ok(updated)
     }
 
     async fn save(&self, bm: BucketMetadata) -> Result<()> {
@@ -1123,122 +1068,6 @@ mod tests {
 
         assert!(matches!(err, Error::Io(_)), "malformed persisted policy must surface its parse failure");
     }
-    /// A tagging rewrite through `update_config_with` (the Swift metadata
-    /// POST path) is persisted: it survives a metadata reload from disk, and
-    /// an emptied rewrite clears the config in the cached copy too instead of
-    /// leaving stale parsed tags behind.
-    #[tokio::test]
-    async fn update_config_with_persists_tagging_rewrite_across_disk_reload() {
-        use crate::bucket::metadata::BUCKET_TAGGING_CONFIG;
-        use s3s::dto::Tag;
-
-        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
-        let mut sys = BucketMetadataSys::new(ecstore);
-
-        let bucket = "swift-tagging-bucket";
-        sys.persist_and_set(BucketMetadata::new(bucket))
-            .await
-            .expect("initial metadata should persist");
-
-        let tagging = Tagging {
-            tag_set: vec![Tag {
-                key: Some("swift-meta-color".to_string()),
-                value: Some("blue".to_string()),
-            }],
-        };
-        let xml = crate::bucket::utils::serialize::<Tagging>(&tagging).expect("tagging should serialize");
-        sys.update_config_with(bucket, BUCKET_TAGGING_CONFIG, move |bm| {
-            assert!(bm.tagging_config.is_none(), "rewrite must see the on-disk state");
-            Ok(xml)
-        })
-        .await
-        .expect("tagging rewrite should persist");
-
-        // Simulate the disk-truth reload that used to lose Swift writes: drop
-        // the cached entry and lazily re-load from the metadata file.
-        sys.metadata_map.write().await.clear();
-        let (tags, _) = sys
-            .get_tagging_config(bucket)
-            .await
-            .expect("tagging must survive a reload from disk");
-        assert_eq!(tags.tag_set.len(), 1);
-        assert_eq!(tags.tag_set[0].key.as_deref(), Some("swift-meta-color"));
-        assert_eq!(tags.tag_set[0].value.as_deref(), Some("blue"));
-
-        // An emptied rewrite clears the config everywhere.
-        sys.update_config_with(bucket, BUCKET_TAGGING_CONFIG, |bm| {
-            assert!(bm.tagging_config.is_some(), "rewrite must see the persisted tags");
-            Ok(Vec::new())
-        })
-        .await
-        .expect("clearing rewrite should persist");
-        assert_eq!(
-            sys.get_tagging_config(bucket).await.unwrap_err(),
-            Error::ConfigNotFound,
-            "cleared tagging must not be served from the cache"
-        );
-        sys.metadata_map.write().await.clear();
-        assert_eq!(
-            sys.get_tagging_config(bucket).await.unwrap_err(),
-            Error::ConfigNotFound,
-            "cleared tagging must not reappear after a reload from disk"
-        );
-    }
-
-    /// The load and the persisted write share one write guard, so concurrent
-    /// rewrites of the same config compose instead of clobbering each other.
-    /// Moving the load outside that guard loses all but the last tag.
-    #[tokio::test]
-    async fn concurrent_update_config_with_calls_do_not_lose_writes() {
-        use crate::bucket::metadata::BUCKET_TAGGING_CONFIG;
-        use s3s::dto::Tag;
-
-        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
-        let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore)));
-
-        let bucket = "swift-tagging-concurrent";
-        sys.read()
-            .await
-            .persist_and_set(BucketMetadata::new(bucket))
-            .await
-            .expect("initial metadata should persist");
-
-        const WRITERS: usize = 8;
-        let mut handles = Vec::with_capacity(WRITERS);
-        for idx in 0..WRITERS {
-            let sys = sys.clone();
-            handles.push(tokio::spawn(async move {
-                sys.write()
-                    .await
-                    .update_config_with(bucket, BUCKET_TAGGING_CONFIG, move |bm| {
-                        // Each writer merges its own tag onto whatever is
-                        // currently persisted — the Swift rewrite shape.
-                        let mut tagging = bm.tagging_config.clone().unwrap_or_else(|| Tagging { tag_set: vec![] });
-                        tagging.tag_set.push(Tag {
-                            key: Some(format!("swift-meta-key{idx}")),
-                            value: Some(idx.to_string()),
-                        });
-                        crate::bucket::utils::serialize::<Tagging>(&tagging).map_err(|e| Error::other(e.to_string()))
-                    })
-                    .await
-            }));
-        }
-
-        for handle in handles {
-            handle
-                .await
-                .expect("writer task should join")
-                .expect("rewrite should persist");
-        }
-
-        let (tags, _) = sys
-            .read()
-            .await
-            .get_tagging_config(bucket)
-            .await
-            .expect("tagging should be readable");
-        assert_eq!(tags.tag_set.len(), WRITERS, "every concurrent rewrite must survive: {tags:?}");
-    }
 
     fn target(bucket: &str, id: &str) -> BucketTarget {
         BucketTarget {

crates/ecstore/src/set_disk/mod.rs

@@ -687,8 +687,6 @@ 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,12 +29,6 @@ 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
@@ -54,14 +48,6 @@ 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();
@@ -93,9 +79,6 @@ impl<'a> ListOperations<'a> {
                 Ok(_) => {
                     errors.push(None);
                 }
-                Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => {
-                    errors.push(None);
-                }
                 Err(e) => {
                     errors.push(Some(e));
                 }
@@ -114,12 +97,6 @@ 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,8 +26,6 @@ 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;
 
@@ -35,7 +33,7 @@ const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
 #[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-pub(crate) enum MultipartCommitPause {
+enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
@@ -47,13 +45,12 @@ struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
     pause: MultipartCommitPause,
-    armed: AtomicBool,
     arrived: tokio::sync::Notify,
     release: tokio::sync::Notify,
 }
 
 #[cfg(test)]
-pub(crate) struct MultipartCommitBarrier {
+struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
@@ -63,12 +60,11 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc
 
 #[cfg(test)]
 impl MultipartCommitBarrier {
-    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    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(),
         });
@@ -82,13 +78,13 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    pub(crate) async fn wait_until_paused(&self) {
+    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");
     }
 
-    pub(crate) fn release(&self) {
+    fn release(&self) {
         self.state.release.notify_one();
     }
 }
@@ -116,9 +112,7 @@ 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
-        && barrier.armed.swap(false, Ordering::AcqRel)
-    {
+    if let Some(barrier) = barrier {
         barrier.arrived.notify_one();
         barrier.release.notified().await;
     }
@@ -783,9 +777,6 @@ 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);
@@ -1263,15 +1254,10 @@ 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?;
-        let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
+        self.check_upload_id_exists(bucket, object, upload_id, false).await?;
         let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
 
-        self.delete_all_with_quorum(
-            RUSTFS_META_MULTIPART_BUCKET,
-            &upload_id_path,
-            fi.write_quorum(self.default_write_quorum()),
-        )
-        .await
+        self.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_id_path).await
     }
     // complete_multipart_upload finished
     #[tracing::instrument(skip(self))]
@@ -1829,36 +1815,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         #[cfg(test)]
         pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-
-        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(upload_guard);
         drop(object_lock_guard); // drop object lock guard to release the lock
 
         // backlog#1321: enqueue heal only when the committed replicas actually
@@ -1903,6 +1860,12 @@ 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
@@ -2214,122 +2177,6 @@ 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,
@@ -3192,81 +3039,6 @@ 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() {
@@ -3398,17 +3170,6 @@ 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
@@ -3419,68 +3180,10 @@ 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

@@ -1969,6 +1969,12 @@ mod metadata_cache_tests {
             let (dir, disk) = new_read_version_test_disk(bucket).await;
             if disk_index < readable_disks {
                 let mut fi = valid_test_fileinfo(object);
+                fi.parts.push(ObjectPartInfo {
+                    number: 1,
+                    size: 1,
+                    actual_size: 1,
+                    ..Default::default()
+                });
                 fi.mod_time = Some(OffsetDateTime::now_utc());
                 fi.erasure.index = fi.erasure.distribution[disk_index];
                 disk.write_metadata(bucket, bucket, object, fi)

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`
-    // (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.
+    // `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.
     #[tokio::test]
     #[serial_test::serial]
     async fn restore_accept_guard_serializes_concurrent_accepts() {

crates/protocols/Cargo.toml

@@ -134,7 +134,6 @@ socket2 = { workspace = true, optional = true, features = ["all"] }
 [dev-dependencies]
 tempfile = { workspace = true }
 proptest = "1"
-rustfs-test-utils = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter", "time"] }
 tokio = { workspace = true, features = ["test-util", "macros", "fs"] }
 

crates/protocols/src/swift/account.rs

@@ -16,11 +16,12 @@
 
 use super::storage_api::account::{BucketOperations, MakeBucketOptions};
 use super::{SwiftError, SwiftResult};
-use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, update_swift_bucket_tagging, validate_metadata};
+use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata};
 use rustfs_credentials::Credentials;
 use s3s::dto::{Tag, Tagging};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
+use time;
 
 /// Validate that the authenticated user has access to the requested account
 ///
@@ -147,33 +148,15 @@ pub async fn get_account_metadata(account: &str, _credentials: &Option<Credentia
 /// Updates account-level metadata such as TempURL keys.
 /// Only updates swift-account-meta-* tags, preserving other tags.
 ///
-/// The caller must own the account: this metadata holds the account's TempURL
-/// signing key, so writing it for someone else's account would let the writer
-/// mint valid pre-signed URLs against that account's objects. Reads
-/// ([`get_account_metadata`]) stay unauthenticated on purpose — TempURL
-/// signature validation has to resolve the key before any credentials exist.
-///
 /// # Arguments
 /// * `account` - Account identifier
 /// * `metadata` - Metadata key-value pairs to store (keys will be prefixed with `swift-account-meta-`)
-/// * `credentials` - Keystone credentials of the caller
+/// * `credentials` - S3 credentials
 pub async fn update_account_metadata(
     account: &str,
     metadata: &HashMap<String, String>,
-    credentials: &Option<Credentials>,
+    _credentials: &Option<Credentials>,
 ) -> SwiftResult<()> {
-    let Some(credentials) = credentials.as_ref() else {
-        return Err(SwiftError::Unauthorized(
-            "Keystone authentication required to update account metadata".to_string(),
-        ));
-    };
-    validate_account_access(account, credentials)?;
-
-    // These tags are persisted into the bucket metadata file, which every
-    // later config write rewrites in full — so unbounded metadata inflates
-    // the cost of unrelated writes for the life of the account.
-    validate_metadata(metadata)?;
-
     let bucket_name = get_account_metadata_bucket_name(account);
 
     let Some(store) = resolve_swift_object_store_handle() else {
@@ -190,30 +173,57 @@ pub async fn update_account_metadata(
             .map_err(|e| SwiftError::InternalServerError(format!("Failed to create account metadata bucket: {}", e)))?;
     }
 
-    // Rewrite the persisted tags: replace swift-account-meta-* tags with the
-    // new metadata while preserving other tags. An empty result clears the
-    // tagging config.
-    update_swift_bucket_tagging(bucket_name, |current| {
-        let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
+    // Load current bucket metadata
+    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
 
-        tagging.tag_set.retain(|tag| {
-            if let Some(key) = &tag.key {
-                !key.starts_with("swift-account-meta-")
-            } else {
-                true
-            }
-        });
+    let mut bucket_meta_clone = (*bucket_meta).clone();
 
-        for (key, value) in metadata {
-            tagging.tag_set.push(Tag {
-                key: Some(format!("swift-account-meta-{}", key)),
-                value: Some(value.clone()),
-            });
+    // Get existing tags, preserving non-Swift tags
+    let mut existing_tagging = bucket_meta_clone
+        .tagging_config
+        .clone()
+        .unwrap_or_else(|| Tagging { tag_set: vec![] });
+
+    // Remove old swift-account-meta-* tags while preserving other tags
+    existing_tagging.tag_set.retain(|tag| {
+        if let Some(key) = &tag.key {
+            !key.starts_with("swift-account-meta-")
+        } else {
+            true
         }
+    });
 
-        tagging
-    })
-    .await?;
+    // Add new metadata tags
+    for (key, value) in metadata {
+        existing_tagging.tag_set.push(Tag {
+            key: Some(format!("swift-account-meta-{}", key)),
+            value: Some(value.clone()),
+        });
+    }
+
+    let now = time::OffsetDateTime::now_utc();
+
+    if existing_tagging.tag_set.is_empty() {
+        // No tags remain; clear tagging config
+        bucket_meta_clone.tagging_config_xml = Vec::new();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = None;
+    } else {
+        // Serialize tags to XML
+        let tagging_xml = quick_xml::se::to_string(&existing_tagging)
+            .map_err(|e| SwiftError::InternalServerError(format!("Failed to serialize tags: {}", e)))?;
+
+        bucket_meta_clone.tagging_config_xml = tagging_xml.into_bytes();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = Some(existing_tagging);
+    }
+
+    // Save updated metadata
+    set_swift_bucket_metadata(bucket_name.clone(), bucket_meta_clone)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to save metadata: {}", e)))?;
 
     Ok(())
 }

crates/protocols/src/swift/container.rs

@@ -22,10 +22,7 @@ use super::storage_api::container::{
 };
 use super::types::Container;
 use super::{SwiftError, SwiftResult};
-use super::{
-    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
-    validate_metadata,
-};
+use super::{get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata};
 use rustfs_credentials::Credentials;
 use s3s::dto::{Tag, Tagging};
 use sha2::{Digest, Sha256};
@@ -486,11 +483,6 @@ pub async fn update_container_metadata(
     // Validate container name
     validate_container_name(container)?;
 
-    // These tags are persisted into the bucket metadata file, which every
-    // later config write rewrites in full — so unbounded metadata inflates
-    // the cost of unrelated writes for the life of the container.
-    validate_metadata(&metadata)?;
-
     // Create mapper with default config (tenant prefixing enabled)
     let mapper = ContainerMapper::default();
 
@@ -514,30 +506,57 @@ pub async fn update_container_metadata(
             }
         })?;
 
-    // Rewrite the persisted tags: replace swift-meta-* tags with the new
-    // metadata while preserving non-Swift tags. An empty result clears the
-    // tagging config.
-    update_swift_bucket_tagging(bucket_name, |current| {
-        let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
+    // Load current bucket metadata
+    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
 
-        tagging.tag_set.retain(|tag| {
-            if let Some(key) = &tag.key {
-                !key.starts_with("swift-meta-")
-            } else {
-                true // Keep tags with no key (shouldn't happen, but be safe)
-            }
-        });
+    let mut bucket_meta_clone = (*bucket_meta).clone();
 
-        if let Some(mut new_tagging) = swift_metadata_to_s3_tags(&metadata) {
-            tagging.tag_set.append(&mut new_tagging.tag_set);
+    // Get existing tags, preserving non-Swift tags
+    let mut existing_tagging = bucket_meta_clone
+        .tagging_config
+        .clone()
+        .unwrap_or_else(|| Tagging { tag_set: vec![] });
+
+    // Remove old swift-meta-* tags while preserving other tags
+    existing_tagging.tag_set.retain(|tag| {
+        if let Some(key) = &tag.key {
+            !key.starts_with("swift-meta-")
+        } else {
+            true // Keep tags with no key (shouldn't happen, but be safe)
         }
-        // If metadata.is_empty() and swift_metadata_to_s3_tags returns None,
-        // we've already removed swift-meta-* tags above, so only non-Swift
-        // tags remain
+    });
 
-        tagging
-    })
-    .await?;
+    // Add new Swift metadata tags if provided
+    if let Some(mut new_tagging) = swift_metadata_to_s3_tags(&metadata) {
+        // Merge: existing non-Swift tags + new Swift tags
+        existing_tagging.tag_set.append(&mut new_tagging.tag_set);
+    }
+    // If metadata.is_empty() and swift_metadata_to_s3_tags returns None,
+    // we've already removed swift-meta-* tags above, so only non-Swift tags remain
+
+    let now = time::OffsetDateTime::now_utc();
+
+    if existing_tagging.tag_set.is_empty() {
+        // No tags remain after removing swift-meta-* tags; clear tagging config
+        bucket_meta_clone.tagging_config_xml = Vec::new();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = None;
+    } else {
+        // Serialize the merged tags to XML
+        let tagging_xml = quick_xml::se::to_string(&existing_tagging)
+            .map_err(|e| SwiftError::InternalServerError(format!("Failed to serialize tags: {}", e)))?;
+
+        bucket_meta_clone.tagging_config_xml = tagging_xml.into_bytes();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = Some(existing_tagging);
+    }
+
+    // Save updated metadata
+    set_swift_bucket_metadata(bucket_name, bucket_meta_clone)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to save metadata: {}", e)))?;
 
     Ok(())
 }
@@ -800,23 +819,44 @@ pub async fn enable_versioning(
             }
         })?;
 
-    // Rewrite the persisted tags: replace any versioning tag with the new
-    // archive location while preserving all other tags.
-    update_swift_bucket_tagging(bucket_name, |current| {
-        let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
+    // Load current bucket metadata
+    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
 
-        tagging
-            .tag_set
-            .retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
+    let mut bucket_meta_clone = (*bucket_meta).clone();
 
-        tagging.tag_set.push(Tag {
-            key: Some("swift-versions-location".to_string()),
-            value: Some(archive_container.to_string()), // Store Swift container name, not S3 bucket name
-        });
+    // Get existing tags
+    let mut existing_tagging = bucket_meta_clone
+        .tagging_config
+        .clone()
+        .unwrap_or_else(|| Tagging { tag_set: vec![] });
 
-        tagging
-    })
-    .await?;
+    // Remove old versioning tag if present
+    existing_tagging
+        .tag_set
+        .retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
+
+    // Add new versioning tag
+    existing_tagging.tag_set.push(Tag {
+        key: Some("swift-versions-location".to_string()),
+        value: Some(archive_container.to_string()), // Store Swift container name, not S3 bucket name
+    });
+
+    let now = time::OffsetDateTime::now_utc();
+
+    // Serialize tags to XML
+    let tagging_xml = quick_xml::se::to_string(&existing_tagging)
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to serialize tags: {}", e)))?;
+
+    bucket_meta_clone.tagging_config_xml = tagging_xml.into_bytes();
+    bucket_meta_clone.tagging_config_updated_at = now;
+    bucket_meta_clone.tagging_config = Some(existing_tagging);
+
+    // Save updated metadata
+    set_swift_bucket_metadata(bucket_name, bucket_meta_clone)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to save metadata: {}", e)))?;
 
     Ok(())
 }
@@ -842,38 +882,50 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
     let mapper = ContainerMapper::default();
     let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
 
-    let Some(store) = resolve_swift_object_store_handle() else {
+    // Verify container exists
+    let Some(_store) = resolve_swift_object_store_handle() else {
         return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
     };
 
-    // Verify container exists. Without this the rewrite below would persist a
-    // fabricated default for a container that does not exist: the metadata
-    // loader turns "no metadata on disk" into a fresh BucketMetadata, and
-    // writing that creates an orphan .metadata.bin and caches a fabricated
-    // default as authoritative.
-    store
-        .get_bucket_info(&bucket_name, &BucketOptions::default())
+    // Load current bucket metadata
+    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
         .await
-        .map_err(|e| {
-            if e.to_string().contains("not found") || e.to_string().contains("NoSuchBucket") {
-                SwiftError::NotFound(format!("Container '{}' not found", container))
-            } else {
-                sanitize_storage_error("Container verification", e)
-            }
-        })?;
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
 
-    // Rewrite the persisted tags: drop the versioning tag while preserving
-    // all other tags. An empty result clears the tagging config.
-    update_swift_bucket_tagging(bucket_name, |current| {
-        let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
+    let mut bucket_meta_clone = (*bucket_meta).clone();
 
-        tagging
-            .tag_set
-            .retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
+    // Get existing tags
+    let mut existing_tagging = bucket_meta_clone
+        .tagging_config
+        .clone()
+        .unwrap_or_else(|| Tagging { tag_set: vec![] });
 
-        tagging
-    })
-    .await?;
+    // Remove versioning tag
+    existing_tagging
+        .tag_set
+        .retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
+
+    let now = time::OffsetDateTime::now_utc();
+
+    if existing_tagging.tag_set.is_empty() {
+        // No tags remain; clear tagging config
+        bucket_meta_clone.tagging_config_xml = Vec::new();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = None;
+    } else {
+        // Serialize remaining tags to XML
+        let tagging_xml = quick_xml::se::to_string(&existing_tagging)
+            .map_err(|e| SwiftError::InternalServerError(format!("Failed to serialize tags: {}", e)))?;
+
+        bucket_meta_clone.tagging_config_xml = tagging_xml.into_bytes();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = Some(existing_tagging);
+    }
+
+    // Save updated metadata
+    set_swift_bucket_metadata(bucket_name, bucket_meta_clone)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to save metadata: {}", e)))?;
 
     Ok(())
 }
@@ -998,37 +1050,65 @@ pub async fn set_container_acl(
             }
         })?;
 
-    // Rewrite the persisted tags: replace the ACL tags with the new grants
-    // while preserving all other tags. An empty result clears the tagging
-    // config.
-    update_swift_bucket_tagging(bucket_name, |current| {
-        let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
+    // Load current bucket metadata
+    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
 
-        tagging
-            .tag_set
-            .retain(|tag| tag.key.as_deref() != Some("swift-acl-read") && tag.key.as_deref() != Some("swift-acl-write"));
+    let mut bucket_meta_clone = (*bucket_meta).clone();
 
-        if let Some(read) = read_acl
-            && !read.trim().is_empty()
-        {
-            tagging.tag_set.push(Tag {
-                key: Some("swift-acl-read".to_string()),
-                value: Some(read.to_string()),
-            });
-        }
+    // Get existing tags
+    let mut existing_tagging = bucket_meta_clone
+        .tagging_config
+        .clone()
+        .unwrap_or_else(|| Tagging { tag_set: vec![] });
 
-        if let Some(write) = write_acl
-            && !write.trim().is_empty()
-        {
-            tagging.tag_set.push(Tag {
-                key: Some("swift-acl-write".to_string()),
-                value: Some(write.to_string()),
-            });
-        }
+    // Remove old ACL tags
+    existing_tagging
+        .tag_set
+        .retain(|tag| tag.key.as_deref() != Some("swift-acl-read") && tag.key.as_deref() != Some("swift-acl-write"));
 
-        tagging
-    })
-    .await?;
+    // Add new read ACL tag if provided
+    if let Some(read) = read_acl
+        && !read.trim().is_empty()
+    {
+        existing_tagging.tag_set.push(Tag {
+            key: Some("swift-acl-read".to_string()),
+            value: Some(read.to_string()),
+        });
+    }
+
+    // Add new write ACL tag if provided
+    if let Some(write) = write_acl
+        && !write.trim().is_empty()
+    {
+        existing_tagging.tag_set.push(Tag {
+            key: Some("swift-acl-write".to_string()),
+            value: Some(write.to_string()),
+        });
+    }
+
+    let now = time::OffsetDateTime::now_utc();
+
+    if existing_tagging.tag_set.is_empty() {
+        // No tags remain; clear tagging config
+        bucket_meta_clone.tagging_config_xml = Vec::new();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = None;
+    } else {
+        // Serialize tags to XML
+        let tagging_xml = quick_xml::se::to_string(&existing_tagging)
+            .map_err(|e| SwiftError::InternalServerError(format!("Failed to serialize tags: {}", e)))?;
+
+        bucket_meta_clone.tagging_config_xml = tagging_xml.into_bytes();
+        bucket_meta_clone.tagging_config_updated_at = now;
+        bucket_meta_clone.tagging_config = Some(existing_tagging);
+    }
+
+    // Save updated metadata
+    set_swift_bucket_metadata(bucket_name, bucket_meta_clone)
+        .await
+        .map_err(|e| SwiftError::InternalServerError(format!("Failed to save metadata: {}", e)))?;
 
     debug!(
         "Set ACLs for container {}/{}: read={:?}, write={:?}",

crates/protocols/src/swift/mod.rs

@@ -58,53 +58,10 @@ pub mod versioning;
 
 pub use errors::{SwiftError, SwiftResult};
 pub use router::{SwiftRoute, SwiftRouter};
-
-/// Maximum number of metadata headers allowed per resource (Swift standard)
-pub(crate) const MAX_METADATA_COUNT: usize = 90;
-
-/// Maximum size in bytes for a single metadata value (Swift standard)
-pub(crate) const MAX_METADATA_VALUE_SIZE: usize = 256;
-
-/// Validate metadata against Swift limits
-///
-/// Checks that:
-/// - Total number of metadata entries doesn't exceed MAX_METADATA_COUNT
-/// - Individual metadata values don't exceed MAX_METADATA_VALUE_SIZE
-///
-/// Applies to object, container and account metadata alike: all three are
-/// persisted, and container/account metadata additionally lands in the
-/// bucket metadata file that every later config write rewrites in full.
-///
-/// Returns error if limits are exceeded.
-pub(crate) fn validate_metadata(metadata: &std::collections::HashMap<String, String>) -> SwiftResult<()> {
-    // Check total metadata count
-    if metadata.len() > MAX_METADATA_COUNT {
-        return Err(SwiftError::BadRequest(format!(
-            "Too many metadata headers: {} (max: {})",
-            metadata.len(),
-            MAX_METADATA_COUNT
-        )));
-    }
-
-    // Check individual value sizes
-    for (key, value) in metadata.iter() {
-        if value.len() > MAX_METADATA_VALUE_SIZE {
-            return Err(SwiftError::BadRequest(format!(
-                "Metadata value for '{}' too large: {} bytes (max: {} bytes)",
-                key,
-                value.len(),
-                MAX_METADATA_VALUE_SIZE
-            )));
-        }
-    }
-
-    Ok(())
-}
-
 // Note: Container, Object, and SwiftMetadata types used by Swift implementation
 pub use storage_api::public_api::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader};
 pub(crate) use storage_api::public_api::{
-    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
+    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata,
 };
 #[allow(unused_imports)]
 pub use types::{Container, Object, SwiftMetadata};

crates/protocols/src/swift/object.rs

@@ -53,7 +53,7 @@ use super::account::validate_account_access;
 use super::container::ContainerMapper;
 use super::expiration_worker::{track_object_expiration, untrack_object_expiration};
 use super::storage_api::object::{BucketOperations, BucketOptions, HTTPRangeSpec, ObjectIO as _, ObjectOperations as _};
-use super::{SwiftError, SwiftResult, resolve_swift_object_store_handle, validate_metadata};
+use super::{SwiftError, SwiftResult, resolve_swift_object_store_handle};
 use axum::http::HeaderMap;
 use rustfs_credentials::Credentials;
 use rustfs_rio::HashReader;
@@ -68,6 +68,12 @@ const LOG_SUBSYSTEM_SWIFT_OBJECT: &str = "swift_object";
 const EVENT_SWIFT_OBJECT_STORAGE_STATE: &str = "swift_object_storage_state";
 const SWIFT_DELETE_AT_METADATA: &str = "x-delete-at";
 
+/// Maximum number of metadata headers allowed per object (Swift standard)
+const MAX_METADATA_COUNT: usize = 90;
+
+/// Maximum size in bytes for a single metadata value (Swift standard)
+const MAX_METADATA_VALUE_SIZE: usize = 256;
+
 /// Maximum object size in bytes (5GB - Swift default)
 const MAX_OBJECT_SIZE: i64 = 5 * 1024 * 1024 * 1024;
 
@@ -228,6 +234,38 @@ impl Default for ObjectKeyMapper {
     }
 }
 
+/// Validate metadata against Swift limits
+///
+/// Checks that:
+/// - Total number of metadata entries doesn't exceed MAX_METADATA_COUNT
+/// - Individual metadata values don't exceed MAX_METADATA_VALUE_SIZE
+///
+/// Returns error if limits are exceeded.
+fn validate_metadata(metadata: &HashMap<String, String>) -> SwiftResult<()> {
+    // Check total metadata count
+    if metadata.len() > MAX_METADATA_COUNT {
+        return Err(SwiftError::BadRequest(format!(
+            "Too many metadata headers: {} (max: {})",
+            metadata.len(),
+            MAX_METADATA_COUNT
+        )));
+    }
+
+    // Check individual value sizes
+    for (key, value) in metadata.iter() {
+        if value.len() > MAX_METADATA_VALUE_SIZE {
+            return Err(SwiftError::BadRequest(format!(
+                "Metadata value for '{}' too large: {} bytes (max: {} bytes)",
+                key,
+                value.len(),
+                MAX_METADATA_VALUE_SIZE
+            )));
+        }
+    }
+
+    Ok(())
+}
+
 fn metadata_delete_at(metadata: &HashMap<String, String>) -> Option<u64> {
     metadata
         .get(SWIFT_DELETE_AT_METADATA)

crates/protocols/src/swift/storage_api.rs

@@ -15,19 +15,14 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use rustfs_ecstore::api::bucket::metadata::BUCKET_TAGGING_CONFIG;
 pub(crate) use rustfs_ecstore::api::bucket::metadata::BucketMetadata as SwiftBucketMetadata;
-use rustfs_ecstore::api::bucket::metadata_sys::{get as get_swift_bucket_metadata_from_backend, update_config_with};
-use rustfs_ecstore::api::bucket::utils::serialize as serialize_bucket_config;
-use rustfs_ecstore::api::error::Error as SwiftStorageError;
+use rustfs_ecstore::api::bucket::metadata_sys::{
+    get as get_swift_bucket_metadata_from_backend, set_bucket_metadata as set_swift_bucket_metadata_in_backend,
+};
 pub(crate) use rustfs_ecstore::api::error::Result as SwiftStorageResult;
-use rustfs_ecstore::api::notification::get_global_notification_sys;
 pub(crate) use rustfs_ecstore::api::runtime::object_store_handle as resolve_swift_object_store_handle;
 use rustfs_ecstore::api::storage::ECStore as SwiftStore;
 use rustfs_storage_api as storage_contracts;
-use s3s::dto::Tagging;
-
-use super::{SwiftError, SwiftResult};
 
 pub(crate) mod account {
     pub(crate) use super::storage_contracts::{BucketOperations, MakeBucketOptions};
@@ -50,7 +45,7 @@ pub(crate) mod object {
 pub(crate) mod public_api {
     pub use super::{SwiftGetObjectReader, SwiftObjectInfo, SwiftObjectOptions, SwiftPutObjReader};
     pub(crate) use super::{
-        get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
+        get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, set_swift_bucket_metadata,
     };
 }
 
@@ -58,15 +53,6 @@ pub(crate) mod versioning {
     pub(crate) use super::storage_contracts::{ListOperations, ObjectOperations};
 }
 
-const LOG_COMPONENT_PROTOCOLS: &str = "protocols";
-const LOG_SUBSYSTEM_SWIFT_STORAGE: &str = "swift_storage";
-const EVENT_SWIFT_BUCKET_TAGGING_UPDATE: &str = "swift_bucket_tagging_update";
-
-/// Marks the refusal to rewrite an unreadable persisted tagging config, so the
-/// caller can turn it into an actionable client error rather than a generic
-/// storage failure. Carried through the ecstore error, which is a string type.
-const UNREADABLE_TAGGING_SENTINEL: &str = "swift: persisted tagging config could not be parsed";
-
 pub type SwiftGetObjectReader = <SwiftStore as storage_contracts::ObjectIO>::GetObjectReader;
 pub type SwiftObjectInfo = <SwiftStore as storage_contracts::ObjectOperations>::ObjectInfo;
 pub type SwiftObjectOptions = <SwiftStore as storage_contracts::ObjectOperations>::ObjectOptions;
@@ -76,80 +62,8 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul
     get_swift_bucket_metadata_from_backend(bucket).await
 }
 
-/// Rewrite the bucket's tagging config through the persisting
-/// bucket-metadata path.
-///
-/// `rewrite` sees the tag set currently persisted on disk (`None` when the
-/// bucket has none) and returns the full replacement; an empty tag set
-/// clears the config. The read-modify-write runs under the bucket metadata
-/// system's write guard — serialized against every other config update —
-/// and the result is written to the bucket metadata file before the cache
-/// is refreshed, so a Swift metadata POST survives process restarts and
-/// disk-truth reloads. Peers are then told to reload, matching what the S3
-/// handlers do after a config write.
-///
-/// Storage failures are logged in full and reported to the client as a
-/// generic error: these now carry real disk and quorum detail, which does not
-/// belong in a Swift response body. The one exception is an unreadable
-/// persisted config, which is reported specifically because the operator has
-/// to act on it.
-pub(crate) async fn update_swift_bucket_tagging<F>(bucket: String, rewrite: F) -> SwiftResult<()>
-where
-    F: FnOnce(Option<&Tagging>) -> Tagging + Send,
-{
-    let result = update_config_with(&bucket, BUCKET_TAGGING_CONFIG, |bm| {
-        // Merging onto an unparseable tag set would silently drop every tag
-        // the bucket has — including the container ACL and versioning tags —
-        // because the rewrite closures treat "no parsed tags" as "no tags".
-        // Refuse instead: the persisted config is intact, just unreadable.
-        if !bm.tagging_config_xml.is_empty() && bm.tagging_config.is_none() {
-            return Err(SwiftStorageError::other(UNREADABLE_TAGGING_SENTINEL));
-        }
-
-        let tagging = rewrite(bm.tagging_config.as_ref());
-        if tagging.tag_set.is_empty() {
-            Ok(Vec::new())
-        } else {
-            // The S3 XML serializer, not quick_xml: the metadata loader's
-            // parse step must be able to round-trip what we persist.
-            serialize_bucket_config(&tagging)
-                .map_err(|e| SwiftStorageError::other(format!("failed to serialize bucket tagging: {e}")))
-        }
-    })
-    .await;
-
-    if let Err(err) = result {
-        let unreadable = err.to_string().contains(UNREADABLE_TAGGING_SENTINEL);
-        tracing::error!(
-            event = EVENT_SWIFT_BUCKET_TAGGING_UPDATE,
-            component = LOG_COMPONENT_PROTOCOLS,
-            subsystem = LOG_SUBSYSTEM_SWIFT_STORAGE,
-            bucket = %bucket,
-            error = %err,
-            reason = if unreadable { "unreadable_persisted_config" } else { "storage_failure" },
-            result = "failed",
-            "swift bucket tagging update failed"
-        );
-        // A Swift-only client has no way to repair this itself, so say what
-        // happened and name the remedy instead of a bare storage error.
-        return Err(if unreadable {
-            SwiftError::Conflict(format!(
-                "The persisted tagging configuration for container store '{bucket}' cannot be parsed, so metadata cannot be updated without discarding it. Reset it with the S3 DeleteBucketTagging API."
-            ))
-        } else {
-            SwiftError::InternalServerError("Metadata update operation failed".to_string())
-        });
-    }
-
-    if let Some(notification_sys) = get_global_notification_sys() {
-        tokio::spawn(async move {
-            if let Err(err) = notification_sys.load_bucket_metadata(&bucket).await {
-                tracing::warn!(bucket = %bucket, error = %err, "failed to notify peers after swift bucket tagging update");
-            }
-        });
-    }
-
-    Ok(())
+pub(crate) async fn set_swift_bucket_metadata(bucket: String, metadata: SwiftBucketMetadata) -> SwiftStorageResult<()> {
+    set_swift_bucket_metadata_in_backend(bucket, metadata).await
 }
 
 pub(crate) async fn get_swift_bucket_usage() -> SwiftStorageResult<Option<HashMap<String, (u64, u64)>>> {

crates/protocols/tests/ecstore_test_compat/mod.rs

@@ -1,25 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! ECStore facade boundary for the protocols integration tests.
-//!
-//! The Swift tests need to drive bucket-metadata reloads the way the peer
-//! `LoadBucketMetadata` RPC does. Everything they touch from `rustfs_ecstore`
-//! is aliased here so the tests themselves hold no raw facade subpaths, the
-//! same boundary `crates/protocols/src/swift/storage_api.rs` provides for the
-//! Swift implementation.
-
-pub use rustfs_ecstore::api::bucket::metadata::load_bucket_metadata;
-pub use rustfs_ecstore::api::bucket::metadata_sys::{get as get_bucket_metadata, set_bucket_metadata};
-pub use rustfs_ecstore::api::runtime::object_store_handle;

crates/protocols/tests/swift_metadata_persistence.rs

@@ -1,312 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Regression tests: a Swift metadata POST must be persisted to the bucket
-//! metadata file, not just the in-memory cache. The metadata has to survive
-//! the disk-truth reloads performed by peer LoadBucketMetadata notifications
-//! and the periodic refresh loop — and, transitively, a process restart.
-
-#![cfg(feature = "swift")]
-
-use std::collections::HashMap;
-
-use rustfs_credentials::Credentials;
-use rustfs_protocols::swift::SwiftError;
-use rustfs_protocols::swift::container::{ContainerMapper, update_container_metadata};
-use rustfs_protocols::swift::{account, container};
-use rustfs_test_utils::TestECStoreEnv;
-use serde_json::json;
-use sha2::{Digest, Sha256};
-
-mod ecstore_test_compat;
-use ecstore_test_compat::{get_bucket_metadata, load_bucket_metadata, object_store_handle, set_bucket_metadata};
-
-fn keystone_credentials(project_id: &str) -> Credentials {
-    let mut claims = HashMap::new();
-    claims.insert("keystone_project_id".to_string(), json!(project_id));
-    claims.insert("keystone_roles".to_string(), json!(["member"]));
-
-    Credentials {
-        access_key: "keystone:swift-test".to_string(),
-        claims: Some(claims),
-        ..Default::default()
-    }
-}
-
-/// The account-metadata bucket name scheme from `swift::account`
-/// (`swift-account-{sha256(account)[0..16]}`), mirrored here so the test can
-/// reload that bucket's metadata from disk.
-fn account_metadata_bucket_name(account: &str) -> String {
-    let mut hasher = Sha256::new();
-    hasher.update(account.as_bytes());
-    let hash = hex::encode(hasher.finalize());
-    format!("swift-account-{}", &hash[0..16])
-}
-
-/// Replace the cached bucket metadata with what is actually on disk — the
-/// same thing a peer LoadBucketMetadata notification or the periodic refresh
-/// loop does. Before the fix this silently discarded every Swift metadata
-/// POST, because those writes only ever touched the cache.
-async fn reload_bucket_metadata_from_disk(bucket: &str) {
-    let store = object_store_handle().expect("test store should be published");
-    let bm = load_bucket_metadata(store, bucket)
-        .await
-        .expect("bucket metadata should load from disk");
-    set_bucket_metadata(bucket.to_string(), bm)
-        .await
-        .expect("reloaded metadata should install");
-}
-
-/// The `swift-meta-*` tags currently persisted for a container, keyed the way
-/// a Swift client sees them. `get_container_metadata` would be the natural
-/// reader, but it additionally requires a data-usage snapshot for the object
-/// count and byte total, which a bare test store has none of — and that is
-/// orthogonal to whether the metadata itself was persisted.
-async fn persisted_container_metadata(bucket: &str) -> HashMap<String, String> {
-    let bm = get_bucket_metadata(bucket).await.expect("bucket metadata should be cached");
-    let mut out = HashMap::new();
-    if let Some(tagging) = &bm.tagging_config {
-        for tag in &tagging.tag_set {
-            if let (Some(key), Some(value)) = (&tag.key, &tag.value)
-                && let Some(meta_key) = key.strip_prefix("swift-meta-")
-            {
-                out.insert(meta_key.to_string(), value.clone());
-            }
-        }
-    }
-    out
-}
-
-/// Every scenario that needs a store runs against ONE environment: the Swift
-/// handlers resolve the ambient object store and bucket-metadata system, so a
-/// second `TestECStoreEnv` in this process would race the first.
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn swift_metadata_writes_are_durable() {
-    let env = TestECStoreEnv::builder().prefix("swift_meta_persist").build().await;
-
-    posts_survive_disk_truth_reload(&env).await;
-    tag_writers_preserve_each_others_state(&env).await;
-    versioning_writes_reject_missing_containers().await;
-}
-
-async fn posts_survive_disk_truth_reload(env: &TestECStoreEnv) {
-    // --- Container metadata POST (X-Container-Meta-*) ---
-    let project_id = "swiftpersistproj";
-    let swift_account = format!("AUTH_{project_id}");
-    let credentials = keystone_credentials(project_id);
-    let swift_container = "photos";
-    let bucket = ContainerMapper::default().swift_to_s3_bucket(swift_container, project_id);
-    env.make_bucket(&bucket, false).await;
-
-    let mut metadata = HashMap::new();
-    metadata.insert("color".to_string(), "blue".to_string());
-    update_container_metadata(&swift_account, swift_container, &credentials, metadata)
-        .await
-        .expect("container metadata POST should succeed");
-
-    reload_bucket_metadata_from_disk(&bucket).await;
-
-    let container_meta = persisted_container_metadata(&bucket).await;
-    assert_eq!(
-        container_meta.get("color").map(String::as_str),
-        Some("blue"),
-        "container metadata POST must survive a disk-truth metadata reload"
-    );
-
-    // A follow-up POST replaces the Swift metadata and that replacement must
-    // survive a reload too (the rewrite merges against disk state, so the
-    // previous value must actually be gone).
-    let mut metadata = HashMap::new();
-    metadata.insert("season".to_string(), "summer".to_string());
-    update_container_metadata(&swift_account, swift_container, &credentials, metadata)
-        .await
-        .expect("second container metadata POST should succeed");
-
-    reload_bucket_metadata_from_disk(&bucket).await;
-
-    let container_meta = persisted_container_metadata(&bucket).await;
-    assert_eq!(container_meta.get("season").map(String::as_str), Some("summer"));
-    assert!(
-        !container_meta.contains_key("color"),
-        "replaced container metadata must not resurrect on reload"
-    );
-
-    // --- Container versioning POST (X-Versions-Location) ---
-    let archive_container = "photos-archive";
-    let archive_bucket = ContainerMapper::default().swift_to_s3_bucket(archive_container, project_id);
-    env.make_bucket(&archive_bucket, false).await;
-
-    container::enable_versioning(&swift_account, swift_container, archive_container, &credentials)
-        .await
-        .expect("enable versioning should succeed");
-
-    reload_bucket_metadata_from_disk(&bucket).await;
-
-    let location = container::get_versions_location(&swift_account, swift_container, &credentials)
-        .await
-        .expect("versions location should load");
-    assert_eq!(
-        location.as_deref(),
-        Some(archive_container),
-        "versions location must survive a disk-truth metadata reload"
-    );
-
-    // --- Account metadata POST (TempURL keys etc.) ---
-    let mut account_meta = HashMap::new();
-    account_meta.insert("temp-url-key".to_string(), "s3cr3t".to_string());
-    account::update_account_metadata(&swift_account, &account_meta, &Some(credentials.clone()))
-        .await
-        .expect("account metadata POST should succeed");
-
-    reload_bucket_metadata_from_disk(&account_metadata_bucket_name(&swift_account)).await;
-
-    let loaded = account::get_account_metadata(&swift_account, &None)
-        .await
-        .expect("account metadata should load");
-    assert_eq!(
-        loaded.get("temp-url-key").map(String::as_str),
-        Some("s3cr3t"),
-        "account metadata POST must survive a disk-truth metadata reload"
-    );
-}
-
-/// The rewrites all share one tag set, so a closure that ignored the current
-/// state would still pass a single-feature test. Drive ACLs, versioning and
-/// container metadata over the same container and assert each survives the
-/// others — and that clearing one leaves the rest alone.
-async fn tag_writers_preserve_each_others_state(env: &TestECStoreEnv) {
-    let project_id = "swiftcrosstagproj";
-    let swift_account = format!("AUTH_{project_id}");
-    let credentials = keystone_credentials(project_id);
-    let container = "shared";
-    let archive = "shared-archive";
-    let bucket = ContainerMapper::default().swift_to_s3_bucket(container, project_id);
-    env.make_bucket(&bucket, false).await;
-    env.make_bucket(&ContainerMapper::default().swift_to_s3_bucket(archive, project_id), false)
-        .await;
-
-    let mut metadata = HashMap::new();
-    metadata.insert("color".to_string(), "blue".to_string());
-    update_container_metadata(&swift_account, container, &credentials, metadata)
-        .await
-        .expect("container metadata POST should succeed");
-    container::enable_versioning(&swift_account, container, archive, &credentials)
-        .await
-        .expect("enable versioning should succeed");
-    container::set_container_acl(&swift_account, container, Some(".r:*"), Some("AUTH_other"), &credentials)
-        .await
-        .expect("set container ACL should succeed");
-
-    reload_bucket_metadata_from_disk(&bucket).await;
-
-    // All three writers' state coexists after a disk-truth reload.
-    let meta = persisted_container_metadata(&bucket).await;
-    assert_eq!(meta.get("color").map(String::as_str), Some("blue"));
-    assert_eq!(
-        container::get_versions_location(&swift_account, container, &credentials)
-            .await
-            .expect("versions location should load")
-            .as_deref(),
-        Some(archive)
-    );
-    let acl = container::get_container_acl(&swift_account, container, &credentials)
-        .await
-        .expect("container ACL should load");
-    assert!(!acl.read.is_empty(), "read ACL must survive the reload");
-    assert!(!acl.write.is_empty(), "write ACL must survive the reload");
-
-    // Disabling versioning drops only the versioning tag.
-    container::disable_versioning(&swift_account, container, &credentials)
-        .await
-        .expect("disable versioning should succeed");
-
-    reload_bucket_metadata_from_disk(&bucket).await;
-
-    assert_eq!(
-        container::get_versions_location(&swift_account, container, &credentials)
-            .await
-            .expect("versions location should load"),
-        None,
-        "disable_versioning must clear the versioning tag durably"
-    );
-    let meta = persisted_container_metadata(&bucket).await;
-    assert_eq!(
-        meta.get("color").map(String::as_str),
-        Some("blue"),
-        "disable_versioning must not disturb container metadata"
-    );
-    let acl = container::get_container_acl(&swift_account, container, &credentials)
-        .await
-        .expect("container ACL should load");
-    assert!(!acl.read.is_empty(), "disable_versioning must not disturb the ACL");
-}
-
-/// A container that does not exist must not get metadata persisted for it:
-/// the metadata loader turns "nothing on disk" into a fresh default, so an
-/// unguarded rewrite would create an orphan metadata file and cache a
-/// fabricated default as authoritative.
-async fn versioning_writes_reject_missing_containers() {
-    let project_id = "swiftmissingproj";
-    let swift_account = format!("AUTH_{project_id}");
-    let credentials = keystone_credentials(project_id);
-    let missing = "no-such-container";
-
-    let err = container::disable_versioning(&swift_account, missing, &credentials)
-        .await
-        .expect_err("disabling versioning on a missing container must fail");
-    assert!(
-        matches!(err, SwiftError::NotFound(_)),
-        "expected NotFound for a missing container, got {err:?}"
-    );
-
-    let bucket = ContainerMapper::default().swift_to_s3_bucket(missing, project_id);
-    assert!(
-        get_bucket_metadata(&bucket).await.is_err(),
-        "a rejected write must not have cached metadata for a nonexistent container"
-    );
-}
-
-/// Account metadata holds the account's TempURL signing key, and it is now
-/// durable — so a write for someone else's account would be a persistent,
-/// cluster-wide takeover of that account's pre-signed URLs, not a cache blip.
-/// The write path must reject both a foreign account and a missing token,
-/// while reads stay open for pre-auth TempURL signature validation.
-///
-/// Deliberately builds no store: both rejections must happen before the write
-/// path resolves storage at all, and a second `TestECStoreEnv` in this process
-/// would race the other test over the ambient store handle.
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn account_metadata_write_rejects_foreign_and_anonymous_callers() {
-    let victim_account = "AUTH_victimproject";
-    let attacker_credentials = keystone_credentials("attackerproject");
-
-    let mut poisoned = HashMap::new();
-    poisoned.insert("temp-url-key".to_string(), "attacker-key".to_string());
-
-    let err = account::update_account_metadata(victim_account, &poisoned, &Some(attacker_credentials))
-        .await
-        .expect_err("writing another account's metadata must be rejected");
-    assert!(
-        matches!(err, SwiftError::Forbidden(_)),
-        "cross-account metadata write must be Forbidden, got {err:?}"
-    );
-
-    let err = account::update_account_metadata(victim_account, &poisoned, &None)
-        .await
-        .expect_err("anonymous account metadata write must be rejected");
-    assert!(
-        matches!(err, SwiftError::Unauthorized(_)),
-        "anonymous metadata write must be Unauthorized, got {err:?}"
-    );
-}

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -60,6 +60,7 @@ 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";
@@ -338,6 +339,45 @@ 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.
@@ -2057,9 +2097,10 @@ 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`.
-/// Synchronous SetDisks transition tests cover copy-back completion, restore
-/// metadata, and local byte-identical reads.
+/// 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).
 ///
 /// 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
@@ -2069,7 +2110,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() {
+async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
     let (_disk_paths, ecstore) = setup_test_env().await;
     let usecase = DefaultObjectUsecase::from_global();
 
@@ -2136,6 +2177,35 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
     );
 
     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 with strict/baseline/blocked result classification | `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 | `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/check_layer_dependencies.sh

@@ -13,14 +13,7 @@ fi
 classify_source_layer() {
   local file="$1"
 
-  if [[ "$file" == rustfs/src/server/* ]] ||
-    [[ "$file" == rustfs/src/startup_*.rs ]] ||
-    [[ "$file" == rustfs/src/init.rs ]] ||
-    [[ "$file" == rustfs/src/main.rs ]] ||
-    [[ "$file" == rustfs/src/lib.rs ]] ||
-    [[ "$file" == rustfs/src/embedded.rs ]]; then
-    printf 'composition'
-  elif [[ "$file" == rustfs/src/app/* ]]; then
+  if [[ "$file" == rustfs/src/app/* ]]; then
     printf 'app'
   elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]]; then
     printf 'interface'
@@ -37,14 +30,6 @@ classify_target_layer() {
   local storage_path
 
   case "$root" in
-    init | main | lib | embedded | startup_*)
-      printf 'composition'
-      ;;
-    server)
-      # Server files are composition roots when they import lower layers, but
-      # their exported HTTP contracts belong to the interface boundary.
-      printf 'interface'
-      ;;
     app)
       printf 'app'
       ;;
@@ -68,9 +53,6 @@ classify_target_layer() {
 
 layer_rank() {
   case "$1" in
-    composition)
-      printf '4'
-      ;;
     interface)
       printf '3'
       ;;
@@ -86,48 +68,6 @@ layer_rank() {
   esac
 }
 
-is_reverse_dependency() {
-  local source_rank target_rank
-
-  source_rank="$(layer_rank "$1")"
-  target_rank="$(layer_rank "$2")"
-  (( source_rank < target_rank ))
-}
-
-assert_dependency_direction() {
-  local expected="$1"
-  local source="$2"
-  local target="$3"
-  local actual='allowed'
-
-  if is_reverse_dependency "$source" "$target"; then
-    actual='reverse'
-  fi
-  if [[ "$actual" != "$expected" ]]; then
-    printf 'Layer dependency guard self-test failed: %s -> %s (expected %s, got %s)\n' \
-      "$source" "$target" "$expected" "$actual" >&2
-    exit 1
-  fi
-}
-
-run_layer_model_self_tests() {
-  local server_source app_source storage_source server_target admin_target app_target
-
-  server_source="$(classify_source_layer rustfs/src/server/http.rs)"
-  app_source="$(classify_source_layer rustfs/src/app/bucket_usecase.rs)"
-  storage_source="$(classify_source_layer rustfs/src/storage/rpc/node_service.rs)"
-  server_target="$(classify_target_layer server::http)"
-  admin_target="$(classify_target_layer admin::router)"
-  app_target="$(classify_target_layer app::bucket_usecase)"
-
-  assert_dependency_direction 'allowed' "$server_source" "$admin_target"
-  assert_dependency_direction 'allowed' "$server_source" "$app_target"
-  assert_dependency_direction 'reverse' "$app_source" "$server_target"
-  assert_dependency_direction 'reverse' "$storage_source" "$server_target"
-  assert_dependency_direction 'reverse' "$app_source" "$admin_target"
-  assert_dependency_direction 'reverse' "$storage_source" "$admin_target"
-}
-
 normalize_import_group_item() {
   local prefix="$1"
   local item="$2"
@@ -242,11 +182,6 @@ normalize_import_path() {
 
 emit_crate_use_statements() {
   (cd "$ROOT_DIR" && rg --files -g '*.rs' rustfs/src | while IFS= read -r file; do
-    # The guard is file-scoped: dedicated test modules are excluded, while
-    # inline #[cfg(test)] imports remain subject to the source file's layer.
-    if [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]]; then
-      continue
-    fi
     perl -0777 -ne '
       while (/\buse\s+crate::.*?;/sg) {
         my $statement = $&;
@@ -259,26 +194,6 @@ emit_crate_use_statements() {
   done)
 }
 
-write_baseline_file() {
-  local entries="$1"
-
-  cat >"$BASELINE_FILE" <<'EOF'
-# Layer dependency baseline for the rustfs binary crate.
-#
-# The guard models production imports as:
-#   composition -> interface -> app -> infra
-#
-# Canonical dependency entry:
-#   dep|source_file|source_layer->target_layer|crate::imported_symbol
-#
-# Canonical conceptual cycle entry:
-#   cycle|left_layer<->right_layer
-EOF
-  cat "$entries" >>"$BASELINE_FILE"
-}
-
-run_layer_model_self_tests
-
 normalize_baseline_file() {
   local input="$1"
   local output="$2"
@@ -350,7 +265,10 @@ while IFS= read -r line; do
       printf '%s->%s\n' "$source_layer" "$target_layer" >>"$EDGES_RAW"
     fi
 
-    if is_reverse_dependency "$source_layer" "$target_layer"; then
+    source_rank="$(layer_rank "$source_layer")"
+    target_rank="$(layer_rank "$target_layer")"
+
+    if (( source_rank < target_rank )); then
       printf 'dep|%s|%s->%s|crate::%s\n' "$file" "$source_layer" "$target_layer" "$import_path" >>"$VIOLATIONS_RAW"
     fi
   done < <(normalize_import_path "$text")
@@ -374,7 +292,7 @@ done <"${TMP_DIR}/edges_sorted.txt" | sort -u >"${TMP_DIR}/cycles_sorted.txt"
 cat "${TMP_DIR}/violations_sorted.txt" "${TMP_DIR}/cycles_sorted.txt" | sort -u >"$CURRENT_BASELINE"
 
 if [[ "$MODE" == "update" ]]; then
-  write_baseline_file "$CURRENT_BASELINE"
+  cp "$CURRENT_BASELINE" "$BASELINE_FILE"
   echo "Updated baseline: $BASELINE_FILE"
   exit 0
 fi

scripts/layer-dependency-baseline.txt

@@ -1,44 +1,35 @@
 # Layer dependency baseline for the rustfs binary crate.
 #
-# The guard models production imports as:
-#   composition -> interface -> app -> infra
+# These are intra-crate module references within rustfs/ that cross the
+# conceptual layer boundaries (app, infra/server, interface/admin).
+# Since they live inside one Cargo crate, Rust doesn't enforce separation.
+# The list is maintained for architectural awareness during code review.
 #
-# Canonical dependency entry:
-#   dep|source_file|source_layer->target_layer|crate::imported_symbol
+# Format for dependency entries:
+#   status|source_file|direction|imported_symbol|reason
 #
-# Canonical conceptual cycle entry:
-#   cycle|left_layer<->right_layer
+# Format for conceptual cycles:
+#   status|cycle|direction_pair|reason
+#
+# Status:
+#   accepted - reviewed and intentionally allowed
+#   todo     - should be resolved in a future refactor
+
 cycle|app<->infra
 cycle|app<->interface
-cycle|composition<->infra
-cycle|composition<->interface
 cycle|infra<->interface
-dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::ENV_SCANNER_ENABLED
-dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::scanner_enabled_from_env
-dep|rustfs/src/app/admin_usecase.rs|app->interface|crate::server::DependencyReadiness
-dep|rustfs/src/app/admin_usecase.rs|app->interface|crate::server::collect_dependency_readiness_report
 dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_bucket_meta_hook
 dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_delete_bucket_hook
 dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_make_bucket_hook
-dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::server::RemoteAddr
-dep|rustfs/src/app/object_usecase.rs|app->interface|crate::server::convert_ecstore_object_info
-dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::DependencyReadiness
-dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::DependencyReadinessReport
-dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::ReadinessDegradedReason
-dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::snapshot_dependency_readiness_report
+dep|rustfs/src/init.rs|infra->interface|crate::admin
 dep|rustfs/src/runtime_sources.rs|infra->app|crate::app::context
-dep|rustfs/src/storage/access.rs|infra->interface|crate::server::RemoteAddr
-dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::server::cors
+dep|rustfs/src/server/http.rs|infra->interface|crate::admin
+dep|rustfs/src/server/layer.rs|infra->interface|crate::admin::console::is_console_path
 dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::storage::ecfs::ListObjectUnorderedQuery
-dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::convert_ecstore_object_info
-dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_audit_module_enabled
-dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_notify_module_enabled
-dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_audit_module_enabled
-dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_notify_module_enabled
-dep|rustfs/src/storage/rpc/http_service.rs|infra->interface|crate::server::RPC_PREFIX
+dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::FS
+dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::validate_object_lock_configuration_input
+dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::s3_api::common::rustfs_initiator
+dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::s3_api::common::rustfs_owner
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_dynamic_config_runtime_state
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_runtime_config_snapshot
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::site_replication::reload_site_replication_runtime_state
-dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::server::MODULE_SWITCHES_SIGNAL_SUBSYSTEM
-dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::heal_enabled_from_env
-dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::scanner_enabled_from_env

scripts/manual_transition_mixed_version_docker_harness.sh

@@ -25,13 +25,9 @@ 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_PHASE="${ROLLBACK_PHASE:-after-terminal}"
-OLD_NODE_PHASE="${OLD_NODE_PHASE:-initial}"
+ROLLBACK_NEW2_TO_OLD=true
 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}"
@@ -62,31 +58,18 @@ Options:
   --object-count <n>           Non-empty probe object count
   --tier <name>                Remote tier name
   --keep-up                    Leave Docker services running
-  --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
+  --no-rollback                Do not replace node2 with old image after job admission
   -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
-  ROLLBACK_PHASE WAIT_TIMEOUT_SECS POLL_SECONDS TRANSITION_WORKERS
-  TRANSITION_QUEUE_CAPACITY OLD_NODE_PHASE FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT AWS_SIGV4_SCOPE
+  WAIT_TIMEOUT_SECS POLL_SECONDS 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
 }
 
@@ -128,28 +111,6 @@ 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
@@ -189,16 +150,8 @@ 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_PHASE=none
+        ROLLBACK_NEW2_TO_OLD=false
         shift
         ;;
       -h|--help)
@@ -245,18 +198,12 @@ 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}
@@ -275,20 +222,14 @@ services:
       - rustfs-1508-net
 
   node1:
-    image: ${node1_image}
+    image: ${OLD_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:
@@ -304,7 +245,6 @@ services:
   node2:
     image: ${NEW_IMAGE}
     hostname: node2
-    user: "0:0"
     environment: *hot-env
     volumes:
       - node2_data_0:/data/rustfs0
@@ -319,7 +259,6 @@ services:
   node3:
     image: ${NEW_IMAGE}
     hostname: node3
-    user: "0:0"
     environment: *hot-env
     volumes:
       - node3_data_0:/data/rustfs0
@@ -334,7 +273,6 @@ services:
   node4:
     image: ${NEW_IMAGE}
     hostname: node4
-    user: "0:0"
     environment: *hot-env
     volumes:
       - node4_data_0:/data/rustfs0
@@ -495,19 +433,16 @@ seed_objects() {
 }
 
 start_transition_job() {
-  local query response job_id run_http_code
+  local query response job_id
   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"
-  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)"
+  response="$(curl_hot POST "$(hot_endpoint 2)/rustfs/admin/v3/ilm/transition/run?${query}")"
+  printf '%s\n' "$response" >"${OUT_DIR}/run-response.json"
   job_id="$(printf '%s' "$response" | jq -r '.job_id // empty')"
   if [[ -z "$job_id" ]]; then
-    log_warn "manual transition response omitted job_id, http_code=${run_http_code:-unknown}"
+    log_error "manual transition response omitted job_id"
     return 1
   fi
   printf '%s\n' "$job_id" >"${OUT_DIR}/job-id.txt"
@@ -516,25 +451,19 @@ start_transition_job() {
 replace_node2_with_old_image() {
   local network
   network="$(network_name)"
-  log_info "Replacing node2 with old image ${OLD_IMAGE} for ${ROLLBACK_PHASE} rollback readback"
+  log_info "Replacing node2 with old image ${OLD_IMAGE} for in-flight 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" \
@@ -544,37 +473,6 @@ 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
@@ -614,85 +512,6 @@ 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
@@ -701,63 +520,37 @@ 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 run_http_code lifecycle_config_found scanned eligible skipped_already_transitioned skipped_already_in_flight queue_queued queue_active result_classification
+  local terminal_state transition_completed tier_failure transition_failed old_code rollback_code
   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_phase=${ROLLBACK_PHASE}
-old_node_phase=${OLD_NODE_PHASE}
-rollback_new2_to_old=$([[ "$ROLLBACK_PHASE" == "none" ]] && printf 'false' || printf 'true')
-result_classification=${result_classification}
+rollback_new2_to_old=${ROLLBACK_NEW2_TO_OLD}
 EOF
   cat "${OUT_DIR}/summary.env"
-  if [[ "$result_classification" != "strict_mixed_rollout_pass" ]]; then
+  if [[ "$terminal_state" != "completed" || "$transition_completed" == "0" || "$transition_failed" != "0" || "$tier_failure" != "0" ]]; then
     log_error "strict #1508 mixed-version transition gate failed; see ${OUT_DIR}"
     return 1
   fi
@@ -767,8 +560,6 @@ 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
@@ -783,37 +574,20 @@ main() {
   docker image inspect "$COLD_IMAGE" >"${OUT_DIR}/cold-image.inspect.json"
 
   compose up -d
-  if ! wait_cluster_ready; then
-    printf 'cluster readiness failed before manual transition admission\n' >"${OUT_DIR}/readiness-failed"
-    collect_logs
-    summarize
-    return 1
-  fi
+  wait_cluster_ready
 
   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
-  seed_objects
   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"
+  seed_objects
+  start_transition_job
+  if [[ "$ROLLBACK_NEW2_TO_OLD" == "true" ]]; then
+    replace_node2_with_old_image
   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,16 +208,7 @@ 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