fix(swift): persist container and account metadata writes (#5398)

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.

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(())
 }