refactor(ecstore): fix bucket config static names

Swift container and account metadata handlers cloned the cached
BucketMetadata, set the tagging fields, and called set_bucket_metadata,
which only updates the in-memory cache map. Nothing reached
.metadata.bin, so every Swift metadata POST was lost on restart and
silently overwritten by the next disk-truth reload (a peer
LoadBucketMetadata notification or the 15-minute refresh loop) — while
the client had already been told 2xx.

Route these writes through a new metadata_sys::update_config_with: a
read-modify-write that loads the on-disk metadata and persists the
result under the same write guard metadata_sys::update uses, so the
rewrite merges against disk truth instead of a possibly stale cache and
cannot clobber a concurrent update to another config file. Peers are
notified afterwards, matching the S3 config handlers.

Persisting these writes required hardening the paths that now produce
durable state:

- Account metadata writes validate account ownership. This metadata
  holds the account's TempURL signing key, so an unauthenticated write
  for someone else's account would have become a durable, cluster-wide
  takeover of that account's pre-signed URLs. Reads stay open because
  TempURL signature validation runs before credentials exist.
- disable_versioning verifies the container exists. Without it the
  metadata loader's "no metadata on disk" default would be persisted,
  creating an orphan metadata file and caching a fabricated default as
  authoritative.
- Container and account metadata are size- and count-limited, reusing
  the Swift limits object metadata already enforces; these tags land in
  the bucket metadata file that every later config write rewrites whole.
- A rewrite refuses to run when the persisted tagging config is
  unreadable, instead of merging onto an empty set and wiping the
  container ACL and versioning tags. It reports 409 naming the remedy.
- Storage errors are logged in full and reported generically, since they
  now carry real disk and quorum detail.

The tagging arm of BucketMetadata::update_config also clears the parsed
config, as the lifecycle arm does: parse_all_configs skips empty XML
rather than clearing, so a cleared config kept serving the old tags.
Tagging is serialized with the S3 XML serializer the loader can parse
back, not quick_xml, whose output was never round-trippable.

Cargo.lock

@@ -9732,6 +9732,7 @@ 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_bucket_targets_under_transaction_lock, update_config_with,
         };
     }
 

crates/ecstore/src/bucket/metadata.rs

@@ -737,6 +737,9 @@ 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 => {
@@ -1318,6 +1321,30 @@ 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,6 +239,35 @@ 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();
@@ -603,24 +632,7 @@ impl BucketMetadataSys {
             return Err(Error::other("errServerNotInitialized"));
         };
 
-        if is_meta_bucketname(bucket) {
-            return Err(Error::other("errInvalidArgument"));
-        }
-
-        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)
-                {
-                    BucketMetadata::new(bucket)
-                } else {
-                    error!("load bucket metadata failed: {}", err);
-                    return Err(err);
-                }
-            }
-        };
+        let mut bm = Self::load_bucket_metadata_for_update(store, bucket, parse).await?;
 
         let updated = bm.update_config(config_file, data)?;
 
@@ -629,6 +641,49 @@ impl BucketMetadataSys {
         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),
+            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))
+                } else {
+                    error!("load bucket metadata failed: {}", err);
+                    Err(err)
+                }
+            }
+        }
+    }
+
     async fn save(&self, bm: BucketMetadata) -> Result<()> {
         if is_meta_bucketname(&bm.name) {
             return Err(Error::other("errInvalidArgument"));
@@ -1068,6 +1123,122 @@ 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/store/bucket.rs

@@ -310,12 +310,13 @@ impl ECStore {
         meta.set_created(opts.created_at);
 
         if opts.lock_enabled {
-            meta.object_lock_config_xml = crate::bucket::utils::serialize::<ObjectLockConfiguration>(&enableObjcetLockConfig)?;
-            meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&enableVersioningConfig)?;
+            meta.object_lock_config_xml =
+                crate::bucket::utils::serialize::<ObjectLockConfiguration>(&ENABLED_OBJECT_LOCK_CONFIG)?;
+            meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)?;
         }
 
         if opts.versioning_enabled {
-            meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&enableVersioningConfig)?;
+            meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)?;
         }
 
         await_bucket_namespace_operation(

crates/ecstore/src/store/mod.rs

@@ -427,11 +427,11 @@ impl ECStore {
 }
 
 lazy_static! {
-    static ref enableObjcetLockConfig: ObjectLockConfiguration = ObjectLockConfiguration {
+    static ref ENABLED_OBJECT_LOCK_CONFIG: ObjectLockConfiguration = ObjectLockConfiguration {
         object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
         ..Default::default()
     };
-    static ref enableVersioningConfig: VersioningConfiguration = VersioningConfiguration {
+    static ref ENABLED_VERSIONING_CONFIG: VersioningConfiguration = VersioningConfiguration {
         status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
         ..Default::default()
     };

crates/protocols/Cargo.toml

@@ -134,6 +134,7 @@ 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,12 +16,11 @@
 
 use super::storage_api::account::{BucketOperations, MakeBucketOptions};
 use super::{SwiftError, SwiftResult};
-use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, set_swift_bucket_metadata};
+use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, update_swift_bucket_tagging, validate_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
 ///
@@ -148,15 +147,33 @@ 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` - S3 credentials
+/// * `credentials` - Keystone credentials of the caller
 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 {
@@ -173,57 +190,30 @@ pub async fn update_account_metadata(
             .map_err(|e| SwiftError::InternalServerError(format!("Failed to create account metadata bucket: {}", e)))?;
     }
 
-    // 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)))?;
+    // 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![] });
 
-    let mut bucket_meta_clone = (*bucket_meta).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
-        }
-    });
-
-    // 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()),
+        tagging.tag_set.retain(|tag| {
+            if let Some(key) = &tag.key {
+                !key.starts_with("swift-account-meta-")
+            } else {
+                true
+            }
         });
-    }
 
-    let now = time::OffsetDateTime::now_utc();
+        for (key, value) in metadata {
+            tagging.tag_set.push(Tag {
+                key: Some(format!("swift-account-meta-{}", key)),
+                value: Some(value.clone()),
+            });
+        }
 
-    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)))?;
+        tagging
+    })
+    .await?;
 
     Ok(())
 }

crates/protocols/src/swift/container.rs

@@ -22,7 +22,10 @@ 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, set_swift_bucket_metadata};
+use super::{
+    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
+    validate_metadata,
+};
 use rustfs_credentials::Credentials;
 use s3s::dto::{Tag, Tagging};
 use sha2::{Digest, Sha256};
@@ -483,6 +486,11 @@ 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();
 
@@ -506,57 +514,30 @@ pub async fn update_container_metadata(
             }
         })?;
 
-    // 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)))?;
+    // 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![] });
 
-    let mut bucket_meta_clone = (*bucket_meta).clone();
+        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)
+            }
+        });
 
-    // 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 let Some(mut new_tagging) = swift_metadata_to_s3_tags(&metadata) {
+            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
 
-    // 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)))?;
+        tagging
+    })
+    .await?;
 
     Ok(())
 }
@@ -819,44 +800,23 @@ pub async fn enable_versioning(
             }
         })?;
 
-    // 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)))?;
+    // 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![] });
 
-    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.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
+        });
 
-    // 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)))?;
+        tagging
+    })
+    .await?;
 
     Ok(())
 }
@@ -882,50 +842,38 @@ 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);
 
-    // Verify container exists
-    let Some(_store) = resolve_swift_object_store_handle() else {
+    let Some(store) = resolve_swift_object_store_handle() else {
         return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
     };
 
-    // Load current bucket metadata
-    let bucket_meta = get_swift_bucket_metadata(&bucket_name)
+    // 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())
         .await
-        .map_err(|e| SwiftError::InternalServerError(format!("Failed to load bucket metadata: {}", e)))?;
+        .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)
+            }
+        })?;
 
-    let mut bucket_meta_clone = (*bucket_meta).clone();
+    // 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![] });
 
-    // Get existing tags
-    let mut existing_tagging = bucket_meta_clone
-        .tagging_config
-        .clone()
-        .unwrap_or_else(|| Tagging { tag_set: vec![] });
+        tagging
+            .tag_set
+            .retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
 
-    // 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)))?;
+        tagging
+    })
+    .await?;
 
     Ok(())
 }
@@ -1050,65 +998,37 @@ pub async fn set_container_acl(
             }
         })?;
 
-    // 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)))?;
+    // 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![] });
 
-    let mut bucket_meta_clone = (*bucket_meta).clone();
+        tagging
+            .tag_set
+            .retain(|tag| tag.key.as_deref() != Some("swift-acl-read") && tag.key.as_deref() != Some("swift-acl-write"));
 
-    // Get existing tags
-    let mut existing_tagging = bucket_meta_clone
-        .tagging_config
-        .clone()
-        .unwrap_or_else(|| Tagging { tag_set: vec![] });
+        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()),
+            });
+        }
 
-    // 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"));
+        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()),
+            });
+        }
 
-    // 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)))?;
+        tagging
+    })
+    .await?;
 
     debug!(
         "Set ACLs for container {}/{}: read={:?}, write={:?}",

crates/protocols/src/swift/mod.rs

@@ -58,10 +58,53 @@ 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, set_swift_bucket_metadata,
+    get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
 };
 #[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};
+use super::{SwiftError, SwiftResult, resolve_swift_object_store_handle, validate_metadata};
 use axum::http::HeaderMap;
 use rustfs_credentials::Credentials;
 use rustfs_rio::HashReader;
@@ -68,12 +68,6 @@ 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;
 
@@ -234,38 +228,6 @@ 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,14 +15,19 @@
 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, set_bucket_metadata as set_swift_bucket_metadata_in_backend,
-};
+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;
 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};
@@ -45,7 +50,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, set_swift_bucket_metadata,
+        get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
     };
 }
 
@@ -53,6 +58,15 @@ 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;
@@ -62,8 +76,80 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul
     get_swift_bucket_metadata_from_backend(bucket).await
 }
 
-pub(crate) async fn set_swift_bucket_metadata(bucket: String, metadata: SwiftBucketMetadata) -> SwiftStorageResult<()> {
-    set_swift_bucket_metadata_in_backend(bucket, metadata).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 get_swift_bucket_usage() -> SwiftStorageResult<Option<HashMap<String, (u64, u64)>>> {

crates/protocols/tests/ecstore_test_compat/mod.rs

@@ -0,0 +1,25 @@
+// 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

@@ -0,0 +1,312 @@
+// 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:?}"
+    );
+}