"feat(ci): add fault-tolerance degradation suite to the functional chain" (#7667)

Revert "feat(ci): add fault-tolerance degradation suite to the functional chain (#7663)"

This reverts commit 7c47c48e85.

.github/workflows/rustfs-kms-test.yml

@@ -4,9 +4,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

.github/workflows/rustfs-replication-test.yml

@@ -18,9 +18,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

.github/workflows/rustfs-s3-compat-test.yml

@@ -4,9 +4,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

.github/workflows/rustfs-security-test.yml

@@ -18,9 +18,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

.github/workflows/rustfs-storage-test.yml

@@ -4,9 +4,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

.github/workflows/rustfs-tier-test.yml

@@ -4,9 +4,9 @@ on:
   workflow_dispatch:
     inputs:
       rustfs_version:
-        description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'RustFS release tag to test. Leave empty for nightly.'
         required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: ''
       package_url:
         description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
         required: false

crates/config/src/constants/body_limits.rs

@@ -64,6 +64,14 @@ pub const MAX_HEAL_REQUEST_SIZE: usize = 1024 * 1024; // 1 MB
 /// memory exhaustion from malicious or misconfigured remote services.
 pub const MAX_S3_CLIENT_RESPONSE_SIZE: usize = 10 * 1024 * 1024; // 10 MB
 
+/// Maximum body size accepted by a single `PutObject` or `UploadPart` request (5 GiB).
+/// Used for: the s3s streaming-body limit and the request-header admission check.
+/// Rationale: matches the AWS S3 single-PUT / single-part ceiling. Larger objects
+/// must use multipart upload. The header check rejects an oversize
+/// `Content-Length` before any body byte is read so the client gets
+/// `EntityTooLarge` immediately instead of streaming 5 GiB into a mid-stream failure.
+pub const MAX_SINGLE_PUT_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024; // 5 GiB
+
 /// Maximum size for OIDC provider response bodies (1 MB)
 /// Used for: discovery documents, JWKS documents and token endpoint responses
 /// Rationale: a hostile or compromised identity provider must not be able to exhaust

crates/e2e_test/src/replication_extension_test.rs

@@ -6383,6 +6383,18 @@ async fn test_site_replication_edit_and_status_peer_state_real_three_node() -> R
     let relayed_key = "after-edit-from-relay.txt";
     let relayed_payload = b"site replication after endpoint edit from relay".to_vec();
 
+    // The first joining receiver owns data before the third site has the
+    // shared account. Initial probes and backfill must wait for every join.
+    target_client.create_bucket().bucket(bucket).send().await?;
+    enable_bucket_versioning(&target_env, bucket).await?;
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key(baseline_key)
+        .body(ByteStream::from(baseline_payload.clone()))
+        .send()
+        .await?;
+
     let add_status = site_replication_add(
         &source_env,
         &[
@@ -6410,7 +6422,10 @@ async fn test_site_replication_edit_and_status_peer_state_real_three_node() -> R
         ],
     )
     .await?;
-    assert!(add_status.success, "unexpected site add result: {:?}", add_status);
+    assert!(
+        add_status.success && add_status.err_detail.is_empty() && add_status.initial_sync_error_message.is_empty(),
+        "unexpected site add result: {add_status:?}"
+    );
 
     let source_info = wait_for_site_replication_enabled(&source_env, 3).await?;
     let _target_info = wait_for_site_replication_enabled(&target_env, 3).await?;
@@ -6421,19 +6436,11 @@ async fn test_site_replication_edit_and_status_peer_state_real_three_node() -> R
         .find(|peer| peer.endpoint == target_env.url)
         .ok_or("target peer missing from source site replication info")?;
 
-    source_client.create_bucket().bucket(bucket).send().await?;
-    enable_bucket_versioning(&source_env, bucket).await?;
-    wait_for_bucket_on_target(&target_client, bucket).await?;
-    wait_for_bucket_on_target(&relay_client, bucket).await?;
-    source_client
-        .put_object()
-        .bucket(bucket)
-        .key(baseline_key)
-        .body(ByteStream::from(baseline_payload.clone()))
-        .send()
-        .await?;
-    let replicated_baseline = wait_for_object_on_target(&target_client, bucket, baseline_key).await?;
-    assert_eq!(replicated_baseline, baseline_payload);
+    for client in [&source_client, &relay_client] {
+        wait_for_bucket_on_target(client, bucket).await?;
+        let backfilled = wait_for_object_on_target(client, bucket, baseline_key).await?;
+        assert_eq!(backfilled, baseline_payload);
+    }
 
     let old_target_address = target_env.address.clone();
     let new_target_port = RustFSTestEnvironment::find_available_port().await?;

crates/ecstore/src/api/mod.rs

@@ -202,7 +202,9 @@ pub mod bucket {
     }
 
     pub mod migration {
-        pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config};
+        pub use crate::bucket::migration::{
+            LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_iam_config,
+        };
     }
 
     pub mod object_lock {

crates/ecstore/src/bucket/migration.rs

@@ -17,7 +17,7 @@
 use crate::bucket::metadata::BUCKET_METADATA_FILE;
 use crate::bucket::replication::ReplicationMigrationBridge;
 use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
-use crate::error::Error;
+use crate::error::{Error, Result, is_err_strict_not_found, is_err_strict_volume_not_found};
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
 use crate::storage_api_contracts::{
     bucket::{BucketOperations, BucketOptions},
@@ -33,7 +33,7 @@ use rustfs_utils::path::SLASH_SEPARATOR;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tracing::{debug, info, warn};
+use tracing::{debug, info};
 
 /// IAM config prefix under meta bucket (e.g. config/iam/).
 const IAM_CONFIG_PREFIX: &str = "config/iam";
@@ -53,6 +53,39 @@ type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
 type ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, Error>;
 type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
 
+#[derive(Clone, Debug, thiserror::Error)]
+enum MigrationMetadataError {
+    #[error("empty legacy metadata: {0}")]
+    Empty(String),
+    #[error("incompatible legacy metadata: {0}")]
+    Incompatible(String),
+}
+
+impl From<MigrationMetadataError> for Error {
+    fn from(error: MigrationMetadataError) -> Self {
+        let message = match &error {
+            MigrationMetadataError::Empty(_) => "empty legacy metadata",
+            MigrationMetadataError::Incompatible(_) => "incompatible legacy metadata",
+        };
+        // Keep the record path in the typed source, not in the quorum grouping key.
+        Self::other_with_context(message, error)
+    }
+}
+
+/// Converts a migration failure at the startup boundary, rendering the safe
+/// record path while leaving storage-layer error grouping stable.
+pub fn migration_startup_error(error: Error) -> std::io::Error {
+    if let Error::Io(io_error) = &error
+        && let Some(metadata_error) = io_error
+            .get_ref()
+            .and_then(|context| context.source())
+            .and_then(|source| source.downcast_ref::<MigrationMetadataError>())
+    {
+        return std::io::Error::other(metadata_error.clone());
+    }
+    std::io::Error::other(error)
+}
+
 /// Callback used to decrypt an at-rest config blob during MinIO -> RustFS migration.
 ///
 /// MinIO encrypts IAM identity/service-account files and the server config at rest
@@ -211,7 +244,7 @@ fn normalize_bucket_meta_blob(path: &str, data: &[u8]) -> std::result::Result<Op
 /// Uses list_bucket (from disk volumes) to get bucket names, since list_objects_v2 on the legacy
 /// meta bucket may not work (legacy format differs from object layer expectations).
 /// Skips buckets that already exist in RustFS (idempotent).
-pub async fn try_migrate_bucket_metadata<S>(store: Arc<S>)
+pub async fn try_migrate_bucket_metadata<S>(store: Arc<S>) -> Result<()>
 where
     S: BucketOperations<Error = crate::error::Error>
         + ObjectIO<
@@ -231,25 +264,18 @@ where
             DeletedObject = DeletedObject,
         >,
 {
-    let buckets_list = match store
+    let buckets_list = store
         .list_bucket(&BucketOptions {
             no_metadata: true,
             ..Default::default()
         })
-        .await
-    {
-        Ok(b) => b,
-        Err(e) => {
-            warn!("list buckets failed (skip migration): {e}");
-            return;
-        }
-    };
+        .await?;
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|b| b.name).collect();
 
     if buckets.is_empty() {
         debug!("No migrating bucket metadata found");
-        return;
+        return Ok(());
     }
 
     debug!("Found {} migrating bucket metadata, migrating...", buckets.len());
@@ -263,26 +289,40 @@ where
 
     for bucket in buckets {
         let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}");
-        migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await;
+        migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await?;
 
         let resync_path = format!(
             "{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{REPLICATION_META_DIR}{SLASH_SEPARATOR}{RESYNC_META_FILE}"
         );
-        migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await;
+        migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await?;
+    }
+    Ok(())
+}
+
+async fn migration_target_exists<S: EcstoreObjectOperations>(store: &S, path: &str) -> Result<bool> {
+    match store
+        .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
+        .await
+    {
+        Ok(_) => Ok(true),
+        Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => Ok(false),
+        Err(err) => Err(err),
     }
 }
 
-async fn migrate_one_if_missing<S>(store: Arc<S>, opts: &ObjectOptions, headers: &HeaderMap, path: &str, label: &str)
+async fn migrate_one_if_missing<S>(
+    store: Arc<S>,
+    opts: &ObjectOptions,
+    headers: &HeaderMap,
+    path: &str,
+    label: &str,
+) -> Result<()>
 where
     S: EcstoreObjectIO + EcstoreObjectOperations,
 {
-    if store
-        .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
-        .await
-        .is_ok()
-    {
+    if migration_target_exists(store.as_ref(), path).await? {
         debug!("{label} already exists in RustFS, skip");
-        return;
+        return Ok(());
     }
 
     let mut rd = match store
@@ -290,43 +330,31 @@ where
         .await
     {
         Ok(r) => r,
-        Err(e) => {
-            debug!("read migrating {label}: {e}");
-            return;
-        }
+        // Ordinary RustFS deployments have no legacy bucket, and optional
+        // legacy settings (such as replication resync) may not exist.
+        Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => return Ok(()),
+        Err(err) => return Err(err),
     };
 
-    let data = match rd.read_all().await {
-        Ok(d) if !d.is_empty() => d,
-        Ok(_) => return,
-        Err(e) => {
-            debug!("read migrating {label} body: {e}");
-            return;
-        }
-    };
-
-    let data = match normalize_bucket_meta_blob(path, &data) {
-        Ok(Some(normalized)) => normalized,
-        Ok(None) => data,
-        Err(e) => {
-            warn!("skip {label} migration due to incompatible format: {e}");
-            return;
-        }
-    };
+    let data = rd.read_all().await?;
+    if data.is_empty() {
+        return Err(MigrationMetadataError::Empty(path.to_owned()).into());
+    }
+    let data = normalize_bucket_meta_blob(path, &data)
+        .map_err(|_| MigrationMetadataError::Incompatible(path.to_owned()))?
+        .unwrap_or(data);
 
     let mut put_data = PutObjReader::from_vec(data);
-    if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await {
-        warn!("write {label}: {e}");
-    } else {
-        info!("Migrated {label}");
-    }
+    store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await?;
+    info!("Migrated {label}");
+    Ok(())
 }
 
 /// Migrates IAM config from legacy meta bucket `config/iam/` to RustFS meta bucket.
 /// Lists all objects under the IAM prefix in the source, copies each to the target if not present.
 /// Skips objects that already exist in RustFS (idempotent).
-/// If list_objects_v2 on the legacy bucket fails (e.g. format differs), migration is skipped.
-pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>)
+/// An absent legacy bucket is a no-op; migration errors prevent startup readiness.
+pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>) -> Result<()>
 where
     S: ListOperations<
             Error = crate::error::Error,
@@ -366,47 +394,36 @@ where
     loop {
         let list_result = match store
             .clone()
-            .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation, None, 500, false, None, false)
+            .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation.clone(), None, 500, false, None, false)
             .await
         {
             Ok(r) => r,
-            Err(e) => {
-                debug!("list IAM config from legacy bucket failed (skip migration): {e}");
-                return;
-            }
+            Err(err) if is_err_strict_volume_not_found(&err) => return Ok(()),
+            Err(err) => return Err(err),
         };
 
         for obj in list_result.objects {
             let path = &obj.name;
-            if path.is_empty() || path.ends_with('/') {
+            // Unsupported records must not trigger target lookups, reads, or decryption.
+            if path != IAM_FORMAT_FILE_PATH
+                && !is_identity_path(path)
+                && !is_group_path(path)
+                && !is_policy_doc_path(path)
+                && !is_policy_mapping_path(path)
+            {
                 continue;
             }
-            if store
-                .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default())
-                .await
-                .is_ok()
-            {
+            if migration_target_exists(store.as_ref(), path).await? {
                 debug!("IAM config already exists in RustFS, skip: {path}");
                 continue;
             }
-            let mut rd = match store
+            let mut rd = store
                 .get_object_reader(MIGRATING_META_BUCKET, path, None, h.clone(), &opts)
-                .await
-            {
-                Ok(r) => r,
-                Err(e) => {
-                    debug!("read migrating IAM config {path}: {e}");
-                    continue;
-                }
-            };
-            let data = match rd.read_all().await {
-                Ok(d) if !d.is_empty() => d,
-                Ok(_) => continue,
-                Err(e) => {
-                    debug!("read migrating IAM config {path} body: {e}");
-                    continue;
-                }
-            };
+                .await?;
+            let data = rd.read_all().await?;
+            if data.is_empty() {
+                return Err(MigrationMetadataError::Empty(path.to_owned()).into());
+            }
             // MinIO encrypts IAM identity/service-account files at rest. Decrypt
             // before normalizing; fall back to the raw bytes when no key applies
             // (plaintext blobs, or nothing to decrypt) so existing behavior holds.
@@ -420,22 +437,17 @@ where
                     debug!("skip unsupported IAM config path during migration: {path}");
                     continue;
                 }
-                Err(e) => {
-                    warn!("skip IAM config migration due to incompatible format, path: {path}, err: {e}");
-                    continue;
-                }
+                // Parser errors may contain credential data. Report only the path.
+                Err(_) => return Err(MigrationMetadataError::Incompatible(path.to_owned()).into()),
             };
             let mut put_data = PutObjReader::from_vec(data);
-            if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await {
-                warn!("write IAM config {path}: {e}");
-            } else {
-                info!("Migrated IAM config: {path}");
-                total_migrated += 1;
-            }
+            store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await?;
+            info!("Migrated IAM config: {path}");
+            total_migrated += 1;
         }
 
-        continuation = list_result.next_continuation_token.or(list_result.continuation_token);
-        if !list_result.is_truncated || continuation.is_none() {
+        continuation = next_iam_migration_page(list_result.is_truncated, continuation, list_result.next_continuation_token)?;
+        if continuation.is_none() {
             break;
         }
     }
@@ -443,10 +455,74 @@ where
     if total_migrated > 0 {
         info!("IAM migration complete: {} object(s) migrated", total_migrated);
     }
+    Ok(())
+}
+
+fn next_iam_migration_page(truncated: bool, previous: Option<String>, next: Option<String>) -> Result<Option<String>> {
+    if !truncated {
+        return Ok(None);
+    }
+    let next = next.filter(|token| !token.is_empty());
+    if next.is_none() || next == previous {
+        return Err(Error::other("legacy IAM migration listing did not advance"));
+    }
+    Ok(next)
 }
 
 #[cfg(test)]
 mod tests {
+    #[test]
+    fn migration_errors_group_by_cause_and_retain_typed_record_context() {
+        use super::{Error, MigrationMetadataError};
+
+        for (make_error, message) in [
+            (
+                MigrationMetadataError::Empty as fn(String) -> MigrationMetadataError,
+                "empty legacy metadata",
+            ),
+            (MigrationMetadataError::Incompatible, "incompatible legacy metadata"),
+        ] {
+            let first: Error = make_error("buckets/first/.metadata.bin".into()).into();
+            let second: Error = make_error("buckets/second/.metadata.bin".into()).into();
+            assert_eq!(first, second, "record paths must not fragment error grouping");
+            assert_eq!(first.clone(), second, "cloning must preserve error grouping");
+
+            let io_error = std::io::Error::from(first);
+            let detail = io_error
+                .get_ref()
+                .and_then(|context| context.source())
+                .expect("record context must remain in the error source");
+            assert!(detail.downcast_ref::<MigrationMetadataError>().is_some());
+            assert!(detail.to_string().contains("buckets/first/.metadata.bin"));
+
+            let startup_error = super::migration_startup_error(make_error("buckets/startup/.metadata.bin".into()).into());
+            assert!(
+                startup_error
+                    .get_ref()
+                    .is_some_and(|source| source.is::<MigrationMetadataError>())
+            );
+            assert_eq!(startup_error.to_string(), format!("{message}: buckets/startup/.metadata.bin"));
+        }
+        assert_ne!(
+            Error::from(MigrationMetadataError::Empty("record".into())),
+            Error::from(MigrationMetadataError::Incompatible("record".into())),
+            "different migration failures must remain distinguishable"
+        );
+    }
+
+    #[test]
+    fn truncated_iam_listing_cannot_report_completed_migration() {
+        use super::next_iam_migration_page;
+        assert_eq!(next_iam_migration_page(false, Some("old".into()), None).expect("final page"), None);
+        assert_eq!(
+            next_iam_migration_page(true, Some("old".into()), Some("next".into())).expect("advancing page"),
+            Some("next".into())
+        );
+        for next in [None, Some(String::new()), Some("old".into())] {
+            assert!(next_iam_migration_page(true, Some("old".into()), next).is_err());
+        }
+    }
+
     use super::{normalize_bucket_meta_blob, normalize_iam_config_blob};
     use crate::bucket::replication::{
         BucketReplicationResyncStatus, ReplicationMigrationBridge, ResyncStatusType, TargetReplicationResyncStatus,
@@ -659,6 +735,13 @@ mod tests {
             .collect();
         crate::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), existing).await;
 
+        super::try_migrate_bucket_metadata(ecstore.clone())
+            .await
+            .expect("fresh stores do not require a legacy metadata bucket");
+        super::try_migrate_iam_config(ecstore.clone(), None)
+            .await
+            .expect("fresh stores do not require a legacy IAM bucket");
+
         let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}interop{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}");
         let put_opts = ObjectOptions::default();
 
@@ -680,8 +763,31 @@ mod tests {
             .await
             .expect("seed .minio.sys bucket metadata");
 
-        // --- Run the real startup migration. ---
-        super::try_migrate_bucket_metadata(ecstore.clone()).await;
+        // A partial import must report failure, even if the main bucket
+        // metadata copied successfully before an incompatible resync record.
+        let resync_path = format!("{BUCKET_META_PREFIX}/interop/.replication/resync.bin");
+        ecstore
+            .put_object(
+                MIGRATING_META_BUCKET,
+                &resync_path,
+                &mut PutObjReader::from_vec(b"invalid resync metadata".to_vec()),
+                &put_opts,
+            )
+            .await
+            .expect("seed malformed legacy resync metadata");
+        assert!(
+            super::try_migrate_bucket_metadata(ecstore.clone()).await.is_err(),
+            "incompatible native metadata must not be reported as a completed migration"
+        );
+        ecstore
+            .delete_object(MIGRATING_META_BUCKET, &resync_path, ObjectOptions::default())
+            .await
+            .expect("remove invalid optional legacy resync record");
+
+        // Retry the real startup migration after repairing the source.
+        super::try_migrate_bucket_metadata(ecstore.clone())
+            .await
+            .expect("native bucket metadata migration completes");
 
         // --- The migrated `.rustfs.sys` blob must carry every MinIO config, ---
         // byte-identical to the source (typed XML/JSON parsing of these fields is

crates/ecstore/src/disk/local.rs

@@ -11879,14 +11879,59 @@ mod test {
     /// stale deterministically, instead of sleeping and hoping the filesystem
     /// timestamp granularity (or a backward wall-clock step) cooperates.
     fn backdate_mtime(path: &Path, age: Duration) {
-        use std::fs::{File, FileTimes};
+        use std::fs::{FileTimes, OpenOptions};
         let mtime = std::time::SystemTime::now() - age;
-        File::open(path)
+        let mut options = OpenOptions::new();
+        options.read(true);
+        #[cfg(windows)]
+        {
+            use std::os::windows::fs::OpenOptionsExt;
+            use windows_sys::Win32::Storage::FileSystem::{FILE_FLAG_BACKUP_SEMANTICS, FILE_WRITE_ATTRIBUTES};
+
+            // Directories need backup semantics, and changing mtime needs attribute-write access.
+            options
+                .access_mode(FILE_WRITE_ATTRIBUTES)
+                .custom_flags(FILE_FLAG_BACKUP_SEMANTICS);
+        }
+        options
+            .open(path)
             .expect("path should open to backdate its mtime")
             .set_times(FileTimes::new().set_modified(mtime))
             .expect("mtime should rewind into the past");
     }
 
+    #[test]
+    fn cleanup_tmp_on_startup_backdate_mtime_preserves_files_and_directory_contents() {
+        use std::time::SystemTime;
+
+        let root = tempfile::tempdir().expect("create timestamp fixture root");
+        let directory = root.path().join("directory");
+        let file = directory.join("payload");
+        std::fs::create_dir(&directory).expect("create timestamp fixture directory");
+        std::fs::write(&file, b"unchanged payload").expect("write timestamp fixture payload");
+        let age = Duration::from_secs(60);
+        // Filesystems may round stored timestamps; do not require subsecond precision or sleep.
+        let rounding = Duration::from_secs(2);
+
+        for path in [&file, &directory] {
+            let earliest = SystemTime::now() - age - rounding;
+            backdate_mtime(path, age);
+            let latest = SystemTime::now() - age + rounding;
+            let modified = std::fs::metadata(path)
+                .expect("read backdated path metadata")
+                .modified()
+                .expect("read backdated modification time");
+            assert!(modified >= earliest && modified <= latest, "mtime must be backdated for {path:?}");
+        }
+
+        let moved = root.path().join("moved");
+        std::fs::rename(&directory, &moved).expect("mtime helper must release its handles before cleanup");
+        assert_eq!(
+            std::fs::read(moved.join("payload")).expect("read preserved payload"),
+            b"unchanged payload"
+        );
+    }
+
     #[tokio::test]
     async fn startup_cleanup_barrier_and_tmp_trash_cleanup_cover_noop_and_delete_paths() {
         use tempfile::tempdir;

crates/iam/tests/minio_iam_migration_test.rs

@@ -100,6 +100,34 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
         .await;
     env.make_bucket(LEGACY_META_BUCKET, false).await;
 
+    for (path, body) in [
+        ("config/iam/empty.json", Vec::new()),
+        ("config/iam/users/ignored/extra.json", b"not JSON".to_vec()),
+    ] {
+        env.put_object_bytes(LEGACY_META_BUCKET, path, body).await;
+    }
+    try_migrate_iam_config(
+        env.ecstore.clone(),
+        Some(std::sync::Arc::new(|_| panic!("unsupported IAM records must not be decrypted"))),
+    )
+    .await
+    .expect("unsupported IAM records, including empty objects, must be skipped");
+
+    let format_path = "config/iam/format.json";
+    for body in [Vec::new(), b"invalid IAM format".to_vec()] {
+        env.put_object_bytes(LEGACY_META_BUCKET, format_path, body).await;
+        let error = try_migrate_iam_config(env.ecstore.clone(), None)
+            .await
+            .expect_err("empty or incompatible supported IAM metadata must prevent startup readiness");
+        let io_error = std::io::Error::from(error);
+        let detail = io_error
+            .get_ref()
+            .and_then(|context| context.source())
+            .expect("failure must retain the supported record in its source");
+        assert!(detail.to_string().contains(format_path), "failure must identify the supported record");
+    }
+    seed_legacy_iam_object(&env, format_path, &json!({"version": 1})).await;
+
     let regular_source = json!({
         "version": 1,
         "credentials": {
@@ -155,7 +183,12 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
     )
     .await;
 
-    try_migrate_iam_config(env.ecstore.clone(), None).await;
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("legacy IAM migration completes after source repair");
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("completed legacy IAM migration is idempotent");
 
     let store = ObjectStore::new(env.ecstore);
     assert_identity_survives(

rustfs/src/admin/handlers/account.rs

@@ -38,7 +38,7 @@ use super::supervise_admin_mutation;
 use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_action_credentials, current_ready_iam_handle, object_store_from_req};
-use crate::admin::service::caller_identity::CallerIdentity;
+use crate::admin::service::caller_identity::{CallerIdentity, oidc_profile_fields};
 use crate::admin::storage_api::s3::{self, Body, S3ErrorCode, S3Request, S3Response, S3Result};
 use crate::admin::utils::read_compatible_admin_body;
 use crate::auth::constant_time_eq;
@@ -73,6 +73,16 @@ pub fn register_account_route(r: &mut S3Router<AdminOperation>) -> std::io::Resu
 /// `GET /rustfs/admin/v3/account/info`
 pub struct SelfAccountInfoHandler {}
 
+#[derive(Debug, serde::Serialize)]
+struct SelfAccountInfoResponse {
+    #[serde(flatten)]
+    account: SelfAccountInfo,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    username: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    email: Option<String>,
+}
+
 #[async_trait::async_trait]
 impl Operation for SelfAccountInfoHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -124,17 +134,22 @@ impl Operation for SelfAccountInfoHandler {
             None => return Err(s3::error(S3ErrorCode::ServiceUnavailable, "the object store is not ready")),
         };
 
-        let info = SelfAccountInfo {
-            access_key: caller.access_key.clone(),
-            identity_type: caller.identity_type,
-            session_access_key: caller.session_access_key.clone(),
-            is_admin: caller.is_owner,
-            status,
-            member_of,
-            policies,
-            credentials_source: caller.credentials_source,
-            mutable: caller.mutability(),
-            mfa,
+        let (username, email) = oidc_profile_fields(&caller.credentials);
+        let info = SelfAccountInfoResponse {
+            account: SelfAccountInfo {
+                access_key: caller.access_key.clone(),
+                identity_type: caller.identity_type,
+                session_access_key: caller.session_access_key.clone(),
+                is_admin: caller.is_owner,
+                status,
+                member_of,
+                policies,
+                credentials_source: caller.credentials_source,
+                mutable: caller.mutability(),
+                mfa,
+            },
+            username,
+            email,
         };
 
         admin_json_response(req.uri.path(), &caller.credentials.secret_key, StatusCode::OK, &info)
@@ -548,6 +563,38 @@ fn validate_new_secret_key(request: &ChangePasswordRequest) -> S3Result<()> {
 mod tests {
     use super::*;
     use crate::server::ADMIN_PREFIX;
+    use rustfs_madmin::account::{AccountMutability, CredentialsSource};
+
+    #[test]
+    fn self_account_info_response_adds_oidc_display_fields_without_changing_base_type() {
+        let mut response = SelfAccountInfoResponse {
+            account: SelfAccountInfo {
+                access_key: "virtual-parent".to_string(),
+                identity_type: IdentityType::Sts,
+                session_access_key: Some("temporary-key".to_string()),
+                is_admin: false,
+                status: "enabled".to_string(),
+                member_of: Vec::new(),
+                policies: Vec::new(),
+                credentials_source: CredentialsSource::Iam,
+                mutable: AccountMutability::default(),
+                mfa: AccountMfaSummary::default(),
+            },
+            username: Some("oidc-user".to_string()),
+            email: Some("oidc-user@example.test".to_string()),
+        };
+
+        let value = serde_json::to_value(&response).expect("serialize account response");
+        assert_eq!(value["access_key"], "virtual-parent");
+        assert_eq!(value["username"], "oidc-user");
+        assert_eq!(value["email"], "oidc-user@example.test");
+
+        response.username = None;
+        response.email = None;
+        let legacy_shape = serde_json::to_value(&response).expect("serialize account response without OIDC fields");
+        assert!(!legacy_shape.as_object().unwrap().contains_key("username"));
+        assert!(!legacy_shape.as_object().unwrap().contains_key("email"));
+    }
 
     fn change_request(current: &str, new: &str) -> ChangePasswordRequest {
         ChangePasswordRequest {

rustfs/src/admin/handlers/account_info.rs

@@ -15,6 +15,7 @@
 use crate::admin::auth::authenticate_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_action_credentials, object_store_from_req};
+use crate::admin::service::caller_identity::oidc_profile_fields;
 use crate::admin::storage_api::bucket::versioning_sys::BucketVersioningSys;
 use crate::admin::storage_api::contract::admin::StorageAdminApi;
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
@@ -52,6 +53,16 @@ pub struct AccountInfo {
 
 pub struct AccountInfoHandler {}
 
+#[derive(Debug, Serialize)]
+struct AccountInfoResponse {
+    #[serde(flatten)]
+    account: rustfs_madmin::AccountInfo,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    username: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    email: Option<String>,
+}
+
 pub fn register_account_info_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
     r.insert(
         Method::GET,
@@ -242,6 +253,7 @@ impl Operation for AccountInfoHandler {
         let policy_str = serde_json::to_string(&effective_policy)
             .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse policy failed"))?;
 
+        let (username, email) = oidc_profile_fields(&cred);
         let mut account_info = rustfs_madmin::AccountInfo {
             account_name,
             server: StorageAdminApi::backend_info(store.as_ref()).await,
@@ -288,8 +300,12 @@ impl Operation for AccountInfoHandler {
             }
         }
 
-        let data = serde_json::to_vec(&account_info)
-            .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?;
+        let data = serde_json::to_vec(&AccountInfoResponse {
+            account: account_info,
+            username,
+            email,
+        })
+        .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?;
 
         let mut header = HeaderMap::new();
         header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
@@ -305,6 +321,25 @@ mod tests {
     use rustfs_policy::policy::BucketPolicy;
     use s3s::dto::{Destination, ReplicationRule};
 
+    #[test]
+    fn accountinfo_response_adds_optional_oidc_display_fields() {
+        let mut response = AccountInfoResponse {
+            account: rustfs_madmin::AccountInfo::default(),
+            username: Some("oidc-user".to_string()),
+            email: Some("oidc-user@example.test".to_string()),
+        };
+
+        let value = serde_json::to_value(&response).expect("serialize accountinfo response");
+        assert_eq!(value["username"], "oidc-user");
+        assert_eq!(value["email"], "oidc-user@example.test");
+
+        response.username = None;
+        response.email = None;
+        let legacy_shape = serde_json::to_value(&response).expect("serialize accountinfo response without OIDC fields");
+        assert!(!legacy_shape.as_object().unwrap().contains_key("username"));
+        assert!(!legacy_shape.as_object().unwrap().contains_key("email"));
+    }
+
     #[test]
     fn test_account_info_structure() {
         // Test AccountInfo struct creation and serialization

rustfs/src/admin/handlers/site_replication.rs

@@ -44,7 +44,8 @@ use crate::admin::utils::{empty_response, json_response, read_compatible_admin_b
 use crate::error::ApiError;
 use crate::server::ADMIN_PREFIX;
 use crate::site_replication::identity::{
-    canonical_endpoint, is_https_endpoint, mark_unknown_peer_sync_enabled, same_identity_endpoint, site_identity_key,
+    canonical_endpoint, deployment_id_for_endpoint, is_https_endpoint, mark_unknown_peer_sync_enabled, same_identity_endpoint,
+    site_identity_key,
 };
 use crate::storage::storage_api::{lock_bucket_targets_metadata, with_config_object_write_lock};
 use base64_simd::URL_SAFE_NO_PAD;
@@ -312,6 +313,8 @@ struct SRPeerJoinResponse {
     peer: PeerInfo,
     #[serde(rename = "initialSyncErrorMessage", default, skip_serializing_if = "String::is_empty")]
     initial_sync_error_message: String,
+    #[serde(rename = "initialSyncDeferred", default, skip_serializing_if = "std::ops::Not::not")]
+    initial_sync_deferred: bool,
     /// Whether the receiving site actually applied this join.
     ///
     /// Three-valued on purpose. `None` means the peer did not report — MinIO
@@ -330,6 +333,8 @@ struct SRPeerJoinEnvelope {
     request: SRPeerJoinReq,
     #[serde(rename = "deferSyncStateEnable", default, skip_serializing_if = "std::ops::Not::not")]
     defer_sync_state_enable: bool,
+    #[serde(rename = "deferInitialSync", default, skip_serializing_if = "std::ops::Not::not")]
+    defer_initial_sync: bool,
 }
 
 #[derive(Debug, Default)]
@@ -2792,6 +2797,19 @@ fn prune_in_sync_status_details(status: &mut SRStatusInfo, opts: &SRStatusOption
     }
 }
 
+fn peer_states_from_infos(
+    site_infos: BTreeMap<String, SRInfo>,
+    reachable_peers: &HashSet<String>,
+) -> BTreeMap<String, SRStateInfo> {
+    // Failed metainfo fetches leave default entries in site_infos for comparison;
+    // they must not become fabricated peer state. PeerErrors describes the failure.
+    site_infos
+        .into_iter()
+        .filter(|(deployment_id, _)| reachable_peers.contains(deployment_id))
+        .map(|(deployment_id, info)| (deployment_id, info.state))
+        .collect()
+}
+
 async fn build_status_info(state: &SiteReplicationState, local_peer: &PeerInfo, uri: &Uri) -> S3Result<SRStatusInfo> {
     let opts = sr_status_options(uri);
     let mut local_info = Some(filter_sr_info(build_sr_info(state, local_peer).await?, &opts));
@@ -2920,17 +2938,7 @@ async fn build_status_info(state: &SiteReplicationState, local_peer: &PeerInfo,
     }
 
     if opts.peer_state {
-        for (deployment_id, peer) in &state.peers {
-            status.peer_states.insert(
-                deployment_id.clone(),
-                SRStateInfo {
-                    name: peer.name.clone(),
-                    peers: state.peers.clone(),
-                    updated_at: state.updated_at,
-                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
-                },
-            );
-        }
+        status.peer_states = peer_states_from_infos(site_infos, &reachable_peers);
     }
 
     Ok(status)
@@ -2940,6 +2948,7 @@ fn merge_add_sites(
     mut state: SiteReplicationState,
     local_peer: PeerInfo,
     sites: Vec<PeerSite>,
+    preflight_infos: &[SiteReplicationAddPreflightInfo],
     service_account_access_key: String,
     service_account_parent: String,
     replicate_ilm_expiry: bool,
@@ -2949,11 +2958,36 @@ fn merge_add_sites(
     state.service_account_parent = service_account_parent;
     state.updated_at = Some(OffsetDateTime::now_utc());
     state.peers = build_join_peers(&state, &local_peer, sites, replicate_ilm_expiry);
+    // Every join must carry the verified identities, including peers that
+    // have not joined yet. Fixing only the coordinator after each reply
+    // leaves the other sites holding endpoint-derived placeholders.
+    for info in preflight_infos {
+        if let Some(mut peer) = existing_peer_for_endpoint(&state, &info.endpoint) {
+            peer.deployment_id = info.deployment_id.clone();
+            state = reconcile_peer_with_actual_identity(state, peer);
+        }
+    }
     state
 }
 
 fn update_peer(mut state: SiteReplicationState, incoming: PeerInfo, ilm_expiry_override: Option<bool>) -> SiteReplicationState {
     let mut peer = normalize_peer_info(incoming);
+    // An older sender may still hold a placeholder after this site has
+    // learned the real ID. Do not let that delivery downgrade the identity.
+    if peer.deployment_id == deployment_id_for_endpoint(&peer.endpoint)
+        && let Some(existing) = state.peers.values().find(|existing| {
+            same_identity_endpoint(&existing.endpoint, &peer.endpoint)
+                && existing.deployment_id != deployment_id_for_endpoint(&existing.endpoint)
+        })
+    {
+        peer.deployment_id = existing.deployment_id.clone();
+    }
+    // Remove the placeholder before persistence normalizes duplicate
+    // endpoints; otherwise map ordering can discard the real identity.
+    state.peers.retain(|_, existing| {
+        !same_identity_endpoint(&existing.endpoint, &peer.endpoint)
+            || existing.deployment_id != deployment_id_for_endpoint(&existing.endpoint)
+    });
     if let Some(enabled) = ilm_expiry_override {
         peer.replicate_ilm_expiry = enabled;
     }
@@ -3539,6 +3573,13 @@ fn align_peer_edit_deployment_id(state: &SiteReplicationState, incoming: &mut Pe
         return;
     };
     if matches.next().is_none() {
+        if same_identity_endpoint(&peer.endpoint, &incoming.endpoint)
+            && peer.deployment_id == deployment_id_for_endpoint(&peer.endpoint)
+            && !incoming.deployment_id.is_empty()
+            && incoming.deployment_id != deployment_id_for_endpoint(&incoming.endpoint)
+        {
+            return;
+        }
         incoming.deployment_id = peer.deployment_id.clone();
     }
 }
@@ -6710,6 +6751,7 @@ fn parse_peer_join_response(body: &[u8], fallback_peer: PeerInfo) -> Result<SRPe
         return Ok(SRPeerJoinResponse {
             peer: fallback_peer,
             initial_sync_error_message: String::new(),
+            initial_sync_deferred: false,
             applied: None,
         });
     }
@@ -6801,6 +6843,7 @@ impl Operation for SiteReplicationAddHandler {
                     current_state,
                     local_peer.clone(),
                     sites.clone(),
+                    &preflight_infos,
                     service_account_access_key.clone(),
                     admin_access_key,
                     replicate_ilm_expiry,
@@ -6815,66 +6858,86 @@ impl Operation for SiteReplicationAddHandler {
                         updated_at: state.updated_at,
                     },
                     defer_sync_state_enable: true,
+                    defer_initial_sync: true,
                 };
                 let peer_join_path = with_site_replication_bootstrap_token(
                     SITE_REPLICATION_PEER_JOIN_PATH,
                     &add_in_progress_guard.token.to_string(),
                 );
 
-                let mut joined_endpoints = HashSet::new();
+                // Install every site's service account before any receiver probes or
+                // backfills to a peer that may not have joined yet. Reuse the join
+                // snapshot for the second pass without repeating IAM/topology writes.
+                // Only peers acknowledging deferral get a second request; older
+                // receivers retain their one-pass behavior and reported errors.
+                let initial_sync_path = format!("{peer_join_path}&initial-sync=true");
                 let mut initial_sync_errors = SiteReplicationErrorSummary::default();
-                for (site, preflight) in sites.iter().zip(preflight_infos.iter()) {
-                    if same_identity_endpoint(&site.endpoint, &local_peer.endpoint)
-                        || !joined_endpoints.insert(site_identity_key(&site.endpoint))
-                    {
-                        continue;
-                    }
+                let mut deferred_endpoints = HashSet::new();
+                for (path, defer_initial_sync) in [(&peer_join_path, true), (&initial_sync_path, false)] {
+                    let mut joined_endpoints = HashSet::new();
+                    for (site, preflight) in sites.iter().zip(preflight_infos.iter()) {
+                        if same_identity_endpoint(&site.endpoint, &local_peer.endpoint)
+                            || (!defer_initial_sync && !deferred_endpoints.contains(&site_identity_key(&site.endpoint)))
+                            || !joined_endpoints.insert(site_identity_key(&site.endpoint))
+                        {
+                            continue;
+                        }
 
-                    let mut peer_join_req = join_req.clone();
-                    peer_join_req.request.svc_acct_parent = site.access_key.clone();
-                    let connection = PeerConnection::try_from(site)?;
-                    let body = PeerAdminRequest::put(&connection, &peer_join_path, &site.access_key)
-                        .send(&site.secret_key, &peer_join_req)
-                        .await?;
+                        let mut peer_join_req = join_req.clone();
+                        peer_join_req.defer_initial_sync = defer_initial_sync;
+                        peer_join_req.request.svc_acct_parent = site.access_key.clone();
+                        let connection = PeerConnection::try_from(site)?;
+                        let body = PeerAdminRequest::put(&connection, path, &site.access_key)
+                            .send(&site.secret_key, &peer_join_req)
+                            .await?;
 
-                    let mut fallback_peer = existing_peer_for_endpoint(&state, &site.endpoint)
-                        .unwrap_or_else(|| normalize_peer_site(site.clone(), replicate_ilm_expiry));
-                    fallback_peer.deployment_id = preflight.deployment_id.clone();
-                    let join_response = parse_peer_join_response(&body, fallback_peer).map_err(|e| {
-                        S3Error::with_message(
-                            S3ErrorCode::InternalError,
-                            format!("parse peer join response from {} failed: {e}", site.endpoint),
-                        )
-                    })?;
-                    if !join_response.initial_sync_error_message.is_empty() {
-                        initial_sync_errors.push(format!("{}: {}", site.endpoint, join_response.initial_sync_error_message));
+                        let mut fallback_peer = existing_peer_for_endpoint(&state, &site.endpoint)
+                            .unwrap_or_else(|| normalize_peer_site(site.clone(), replicate_ilm_expiry));
+                        fallback_peer.deployment_id = preflight.deployment_id.clone();
+                        let join_response = parse_peer_join_response(&body, fallback_peer).map_err(|e| {
+                            S3Error::with_message(
+                                S3ErrorCode::InternalError,
+                                format!("parse peer join response from {} failed: {e}", site.endpoint),
+                            )
+                        })?;
+                        if join_response.initial_sync_deferred {
+                            if defer_initial_sync {
+                                deferred_endpoints.insert(site_identity_key(&site.endpoint));
+                            } else {
+                                initial_sync_errors.push(format!("{}: peer did not complete initial sync", site.endpoint));
+                            }
+                        }
+                        if !join_response.initial_sync_error_message.is_empty() {
+                            initial_sync_errors.push(format!("{}: {}", site.endpoint, join_response.initial_sync_error_message));
+                        }
+                        // An explicit no-op join. The peer answered 200 but wrote nothing —
+                        // its persisted state is already newer than the snapshot it was
+                        // sent — so the add is only PARTIALLY configured and saying
+                        // "configured successfully" would be a lie (rustfs/rustfs#5963).
+                        // `None` (a MinIO peer, or one older than the field) is not a
+                        // no-op signal and is deliberately not reported.
+                        if join_response.applied == Some(false) {
+                            let phase = if defer_initial_sync { "join" } else { "initial sync" };
+                            initial_sync_errors.push(format!(
+                                "{}: peer did not apply the {phase} (its site replication state is newer than the snapshot it was sent); \
+                                 the site is not configured against this peer",
+                                site.endpoint
+                            ));
+                        }
+                        state = reconcile_peer_with_actual_identity(state, join_response.peer);
+                        let reconciled_peer = existing_peer_for_endpoint(&state, &site.endpoint).ok_or_else(|| {
+                            S3Error::with_message(
+                                S3ErrorCode::InternalError,
+                                format!("peer join response from {} did not identify the requested site", site.endpoint),
+                            )
+                        })?;
+                        validate_proposed_peer(&reconciled_peer).map_err(|err| {
+                            S3Error::with_message(
+                                S3ErrorCode::InvalidRequest,
+                                format!("invalid peer join response from {}: {err}", site.endpoint),
+                            )
+                        })?;
                     }
-                    // An explicit no-op join. The peer answered 200 but wrote nothing —
-                    // its persisted state is already newer than the snapshot it was
-                    // sent — so the add is only PARTIALLY configured and saying
-                    // "configured successfully" would be a lie (rustfs/rustfs#5963).
-                    // `None` (a MinIO peer, or one older than the field) is not a
-                    // no-op signal and is deliberately not reported.
-                    if join_response.applied == Some(false) {
-                        initial_sync_errors.push(format!(
-                            "{}: peer did not apply the join (its site replication state is newer than the snapshot it was sent); \
-                             the site is not configured against this peer",
-                            site.endpoint
-                        ));
-                    }
-                    state = reconcile_peer_with_actual_identity(state, join_response.peer);
-                    let reconciled_peer = existing_peer_for_endpoint(&state, &site.endpoint).ok_or_else(|| {
-                        S3Error::with_message(
-                            S3ErrorCode::InternalError,
-                            format!("peer join response from {} did not identify the requested site", site.endpoint),
-                        )
-                    })?;
-                    validate_proposed_peer(&reconciled_peer).map_err(|err| {
-                        S3Error::with_message(
-                            S3ErrorCode::InvalidRequest,
-                            format!("invalid peer join response from {}: {err}", site.endpoint),
-                        )
-                    })?;
                 }
 
                 mark_unknown_peer_sync_enabled(&mut state.peers);
@@ -7148,6 +7211,24 @@ impl Operation for SiteReplicationNetPerfHandler {
 
 pub struct SRPeerJoinHandler {}
 
+fn ensure_initial_sync_join_current(state: &SiteReplicationState, join_req: &SRPeerJoinReq) -> S3Result<()> {
+    if !state.enabled()
+        || join_req.updated_at.is_none()
+        || state.updated_at != join_req.updated_at
+        || state.service_account_access_key.is_empty()
+        || state.service_account_access_key != join_req.svc_acct_access_key
+        || state.pending_remove.is_some()
+        || state.pending_rotation.is_some()
+        || pending_endpoint_refresh(state).is_some()
+    {
+        return Err(s3_error!(
+            InvalidRequest,
+            "site replication changed before initial sync; re-run replicate add"
+        ));
+    }
+    Ok(())
+}
+
 /// What the join admission decided about an incoming peer join. The verdict —
 /// and the committed state the back-fill afterwards needs — travel out of
 /// [`admit_peer_join`] instead of being answered where they are decided.
@@ -7310,6 +7391,7 @@ fn superseded_join_response(peer: PeerInfo) -> SRPeerJoinResponse {
     SRPeerJoinResponse {
         peer,
         initial_sync_error_message: String::new(),
+        initial_sync_deferred: false,
         applied: Some(false),
     }
 }
@@ -7319,6 +7401,7 @@ fn applied_join_response(peer: PeerInfo, initial_sync_error_message: String) ->
     SRPeerJoinResponse {
         peer,
         initial_sync_error_message,
+        initial_sync_deferred: false,
         applied: Some(true),
     }
 }
@@ -7328,17 +7411,30 @@ impl Operation for SRPeerJoinHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationAddAction).await?;
         let bootstrap_token = site_replication_bootstrap_token(&req.uri);
+        let initial_sync_only = query_pairs(&req.uri).get("initial-sync").is_some_and(|value| value == "true");
         let local_endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
         // The body is fully read before the admission takes the lifecycle
         // guard: a sender that stalls mid-body must not block this node's
         // add/remove/rotate/reconciler.
         let join_envelope: SRPeerJoinEnvelope = read_site_replication_json(req, &cred.secret_key, true).await?;
         let defer_sync_state_enable = join_envelope.defer_sync_state_enable;
+        let defer_initial_sync = join_envelope.defer_initial_sync;
         let join_req = join_envelope.request;
         validate_join_peer_snapshot(&join_req.peers)?;
 
-        let committed =
-            admit_peer_join(local_endpoint, join_req, defer_sync_state_enable, apply_peer_join_service_account).await?;
+        let _initial_sync_guard = if initial_sync_only {
+            Some(SiteReplicationLifecycleGuard::acquire().await?)
+        } else {
+            None
+        };
+        let committed = if initial_sync_only {
+            let state = load_site_replication_state().await?;
+            ensure_initial_sync_join_current(&state, &join_req)?;
+            let local_peer = local_peer_at_endpoint(local_endpoint, &state);
+            PeerJoinOutcome::Applied(Box::new(state), local_peer)
+        } else {
+            admit_peer_join(local_endpoint, join_req, defer_sync_state_enable, apply_peer_join_service_account).await?
+        };
         // Committed; the reverse-reachability probe and the bucket back-fill
         // run outside the transaction — their transport helpers' retry-event
         // bookkeeping re-enters it (P1-15).
@@ -7355,6 +7451,12 @@ impl Operation for SRPeerJoinHandler {
                 return json_response(StatusCode::OK, &superseded_join_response(peer));
             }
         };
+        if defer_initial_sync && !initial_sync_only {
+            let mut response =
+                applied_join_response(state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer), String::new());
+            response.initial_sync_deferred = true;
+            return json_response(StatusCode::OK, &response);
+        }
         // Fix 1 (receiving side): ensure the joining peer also sets up replication for any
         // buckets it already owns so the reverse direction works from the start. Per-bucket
         // failures are logged (BUG2) so a reverse-direction back-fill gap is observable.
@@ -8482,9 +8584,92 @@ impl Operation for SRRotateServiceAccountHandler {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::site_replication::identity::deployment_id_for_endpoint;
     use rustfs_madmin::SRSessionPolicy;
 
+    #[test]
+    fn peer_states_preserve_each_sites_actual_membership_and_metadata() {
+        let local = SRStateInfo {
+            name: "local".to_string(),
+            peers: BTreeMap::from([(
+                "actual-remote".to_string(),
+                PeerInfo {
+                    deployment_id: "actual-remote".to_string(),
+                    endpoint: "http://remote:9000".to_string(),
+                    ..Default::default()
+                },
+            )]),
+            updated_at: Some(OffsetDateTime::UNIX_EPOCH),
+            api_version: Some("1".to_string()),
+        };
+        let remote = SRStateInfo {
+            name: "remote-reported-name".to_string(),
+            peers: BTreeMap::from([(
+                "legacy-placeholder".to_string(),
+                PeerInfo {
+                    deployment_id: "legacy-placeholder".to_string(),
+                    endpoint: "http://local:9000".to_string(),
+                    ..Default::default()
+                },
+            )]),
+            updated_at: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10)),
+            api_version: None,
+        };
+        let infos = BTreeMap::from([
+            (
+                "local".to_string(),
+                SRInfo {
+                    state: local.clone(),
+                    ..Default::default()
+                },
+            ),
+            (
+                "remote".to_string(),
+                SRInfo {
+                    state: remote.clone(),
+                    ..Default::default()
+                },
+            ),
+        ]);
+        let states = peer_states_from_infos(infos, &HashSet::from(["local".to_string(), "remote".to_string()]));
+        assert_eq!(states.len(), 2);
+        assert_eq!(serde_json::to_value(&states["local"]).unwrap(), serde_json::to_value(local).unwrap());
+        assert_eq!(serde_json::to_value(&states["remote"]).unwrap(), serde_json::to_value(remote).unwrap());
+    }
+
+    #[test]
+    fn peer_states_omit_unreachable_peers_instead_of_defaulting_them() {
+        let infos = BTreeMap::from([
+            ("local".to_string(), SRInfo::default()),
+            ("offline".to_string(), SRInfo::default()),
+        ]);
+        let states = peer_states_from_infos(infos, &HashSet::from(["local".to_string()]));
+        assert_eq!(states.len(), 1);
+        assert!(states.contains_key("local"));
+        assert!(!states.contains_key("offline"));
+    }
+
+    #[test]
+    fn peer_states_preserve_a_reachable_peers_empty_membership() {
+        let states = peer_states_from_infos(
+            BTreeMap::from([(
+                "remote".to_string(),
+                SRInfo {
+                    enabled: false,
+                    state: SRStateInfo {
+                        name: "remote".to_string(),
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                },
+            )]),
+            &HashSet::from(["remote".to_string()]),
+        );
+        assert_eq!(states["remote"].name, "remote");
+        assert!(states["remote"].peers.is_empty());
+        assert!(states["remote"].updated_at.is_none());
+        assert!(states["remote"].api_version.is_none());
+    }
+
     /// A peer the status probe could not reach must render as offline.
     ///
     /// Regression: `build_metrics_summary` used to hardcode `online: true` and
@@ -10916,6 +11101,7 @@ mod tests {
                 secret_key: "remote-sk".to_string(),
                 ..PeerSite::default()
             }],
+            &[],
             "svc-ak".to_string(),
             "root".to_string(),
             true,
@@ -10949,6 +11135,7 @@ mod tests {
                     ..PeerSite::default()
                 },
             ],
+            &[],
             "svc-ak".to_string(),
             "root".to_string(),
             true,
@@ -12155,6 +12342,173 @@ mod tests {
         assert!(normalized.contains_key("hash-remote"));
     }
 
+    #[test]
+    fn test_peer_identity_join_snapshot_uses_verified_ids() {
+        let actual = ["site-a", "site-b", "site-c"].map(|name| PeerInfo {
+            deployment_id: format!("{name}-deployment"),
+            ..peer(name, &format!("https://{name}.example.com:9000"))
+        });
+        let preflight = actual
+            .iter()
+            .map(|peer| preflight_site("reported-name", &peer.endpoint, &peer.deployment_id, 0))
+            .collect::<Vec<_>>();
+        let sites = actual
+            .iter()
+            .map(|peer| PeerSite {
+                name: peer.name.clone(),
+                endpoint: peer.endpoint.clone(),
+                skip_tls_verify: true,
+                ca_cert_pem: "requested-ca".to_string(),
+                ..Default::default()
+            })
+            .collect();
+        let state = merge_add_sites(
+            SiteReplicationState::default(),
+            actual[0].clone(),
+            sites,
+            &preflight,
+            "svc-ak".to_string(),
+            "root".to_string(),
+            false,
+        );
+        assert_eq!(state.peers.len(), actual.len());
+        for expected in &actual {
+            let stored = state
+                .peers
+                .get(&expected.deployment_id)
+                .expect("verified ID in initial join map");
+            assert_eq!(stored.name, expected.name);
+            assert_eq!(stored.endpoint, expected.endpoint);
+            assert!(stored.skip_tls_verify);
+            assert_eq!(stored.ca_cert_pem, "requested-ca");
+        }
+        for local in &actual[1..] {
+            let mut joined = SiteReplicationState::default();
+            apply_peer_join(
+                &mut joined,
+                local,
+                SRPeerJoinReq {
+                    peers: state.peers.clone(),
+                    ..Default::default()
+                },
+                true,
+            );
+            assert_eq!(joined.peers.keys().collect::<Vec<_>>(), state.peers.keys().collect::<Vec<_>>());
+        }
+    }
+
+    #[test]
+    fn test_peer_identity_legacy_edit_does_not_restore_placeholder() {
+        let actual = PeerInfo {
+            deployment_id: "actual-remote".to_string(),
+            ..peer("remote", "http://remote.example.com:9000")
+        };
+        for name in ["remote", ""] {
+            let state = SiteReplicationState {
+                peers: BTreeMap::from([(actual.deployment_id.clone(), actual.clone())]),
+                ..Default::default()
+            };
+            let mut incoming = PeerInfo {
+                deployment_id: deployment_id_for_endpoint("https://REMOTE.example.com:9000/"),
+                sync_state: SyncStatus::Enable,
+                ..peer(name, "https://REMOTE.example.com:9000/")
+            };
+            align_peer_edit_deployment_id(&state, &mut incoming);
+            let state = update_peer(state, incoming, None);
+            assert_eq!(state.peers.len(), 1);
+            assert_eq!(state.peers[&actual.deployment_id].sync_state, SyncStatus::Enable);
+        }
+    }
+
+    #[test]
+    fn test_peer_identity_finalization_repairs_legacy_three_site_join() {
+        let actual = ["site-a", "site-b", "site-c"].map(|name| PeerInfo {
+            deployment_id: format!("{name}-deployment"),
+            ..peer(name, &format!("http://{name}.example.com:9000"))
+        });
+        let sites = actual
+            .iter()
+            .map(|peer| PeerSite {
+                name: peer.name.clone(),
+                endpoint: peer.endpoint.clone(),
+                ..Default::default()
+            })
+            .collect();
+        let mut coordinator = merge_add_sites(
+            SiteReplicationState::default(),
+            actual[0].clone(),
+            sites,
+            &[],
+            "svc-ak".to_string(),
+            "root".to_string(),
+            true,
+        );
+        let join = SRPeerJoinReq {
+            peers: coordinator.peers.clone(),
+            ..Default::default()
+        };
+        for remote in &actual[1..] {
+            coordinator = reconcile_peer_with_actual_identity(coordinator, remote.clone());
+        }
+        mark_unknown_peer_sync_enabled(&mut coordinator.peers);
+
+        for local in &actual[1..] {
+            let mut state = SiteReplicationState::default();
+            apply_peer_join(&mut state, local, join.clone(), true);
+            for mut incoming in coordinator.peers.values().cloned() {
+                align_peer_edit_deployment_id(&state, &mut incoming);
+                state = apply_internal_peer_edit(state, local, incoming, None).expect("finalize peer identity");
+            }
+            assert_eq!(state.peers.len(), actual.len(), "finalization must not retain placeholder peers");
+            for expected in &actual {
+                let stored = existing_peer_for_endpoint(&state, &expected.endpoint).expect("peer remains configured");
+                assert_eq!(stored.deployment_id, expected.deployment_id, "observer: {}", local.name);
+                assert_eq!(stored.sync_state, SyncStatus::Enable);
+            }
+        }
+    }
+
+    #[test]
+    fn test_peer_identity_edit_replaces_placeholder_for_canonical_endpoint() {
+        let local = PeerInfo {
+            deployment_id: "local-deployment".to_string(),
+            ..peer("local", "https://local.example.com:9000")
+        };
+        let endpoint = "http://remote.example.com:9000";
+        let placeholder = PeerInfo {
+            deployment_id: deployment_id_for_endpoint(endpoint),
+            ..peer("remote", endpoint)
+        };
+        for deployment_id in ["00000000-0000-4000-8000-000000000001", "ffffffff-ffff-4fff-bfff-ffffffffffff"] {
+            for name in ["remote", ""] {
+                for already_present in [false, true] {
+                    let mut incoming = PeerInfo {
+                        deployment_id: deployment_id.to_string(),
+                        sync_state: SyncStatus::Enable,
+                        ..peer(name, "https://REMOTE.example.com:9000/")
+                    };
+                    let mut state = SiteReplicationState {
+                        peers: BTreeMap::from([
+                            (local.deployment_id.clone(), local.clone()),
+                            (placeholder.deployment_id.clone(), placeholder.clone()),
+                        ]),
+                        ..Default::default()
+                    };
+                    if already_present {
+                        state.peers.insert(incoming.deployment_id.clone(), incoming.clone());
+                    }
+                    align_peer_edit_deployment_id(&state, &mut incoming);
+                    let state = apply_internal_peer_edit(state, &local, incoming, None).expect("repair peer identity");
+                    assert_eq!(state.peers.len(), 2, "repair must replace, not duplicate, the placeholder");
+                    assert!(!state.peers.contains_key(&placeholder.deployment_id));
+                    assert!(state.peers.contains_key(deployment_id));
+                    let normalized = normalize_peer_map_by_identity(state.peers);
+                    assert!(normalized.contains_key(deployment_id), "normalization must retain the actual ID");
+                }
+            }
+        }
+    }
+
     #[test]
     fn test_reconcile_peer_with_actual_identity_replaces_endpoint_hash_key() {
         let mut state = SiteReplicationState::default();
@@ -13477,6 +13831,7 @@ mod tests {
             assert_eq!(response.peer.deployment_id, "remote-deployment");
             assert_eq!(response.peer.endpoint, "https://remote.example.com");
             assert!(response.initial_sync_error_message.is_empty());
+            assert!(!response.initial_sync_deferred);
             assert_eq!(
                 response.applied, None,
                 "a MinIO empty-body success reports nothing; it must not read as a no-op join"
@@ -13486,6 +13841,7 @@ mod tests {
         let json = serde_json::to_vec(&SRPeerJoinResponse {
             peer: peer("actual", "https://actual.example.com"),
             initial_sync_error_message: "sync failed".to_string(),
+            initial_sync_deferred: false,
             applied: Some(true),
         })
         .expect("serialize join response");
@@ -14338,6 +14694,69 @@ mod tests {
         assert_eq!(value.get("deferSyncStateEnable"), Some(&Value::Bool(true)));
     }
 
+    #[test]
+    fn test_initial_sync_join_requires_the_committed_snapshot() {
+        let now = OffsetDateTime::now_utc();
+        let mut state = SiteReplicationState {
+            updated_at: Some(now),
+            service_account_access_key: "replicator".to_string(),
+            peers: BTreeMap::from([
+                ("a".to_string(), peer("a", "https://a.example.com")),
+                ("b".to_string(), peer("b", "https://b.example.com")),
+            ]),
+            ..Default::default()
+        };
+        let mut request = SRPeerJoinReq {
+            updated_at: Some(now),
+            svc_acct_access_key: "replicator".to_string(),
+            ..Default::default()
+        };
+        ensure_initial_sync_join_current(&state, &request).expect("same committed join");
+        for timestamp in [None, Some(now - time::Duration::SECOND), Some(now + time::Duration::SECOND)] {
+            request.updated_at = timestamp;
+            assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        }
+        request.updated_at = Some(now);
+        request.svc_acct_access_key = "another-replicator".to_string();
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        request.svc_acct_access_key.clone_from(&state.service_account_access_key);
+        state.pending_remove = Some(PendingRemove::default());
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        state.pending_remove = None;
+        state.pending_rotation = Some(PendingRotation::default());
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        state.pending_rotation = None;
+        state.pending_endpoint_refresh = Some(PendingEndpointRefresh::default());
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        state.pending_endpoint_refresh = None;
+        state.service_account_access_key.clear();
+        request.svc_acct_access_key.clear();
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+        state.service_account_access_key = "replicator".to_string();
+        request.svc_acct_access_key.clone_from(&state.service_account_access_key);
+        state.peers.clear();
+        assert!(ensure_initial_sync_join_current(&state, &request).is_err());
+    }
+
+    #[test]
+    fn test_join_initial_sync_deferral_preserves_legacy_requests() {
+        let legacy: SRPeerJoinEnvelope = serde_json::from_str("{}").expect("legacy join");
+        assert!(!legacy.defer_initial_sync);
+        assert!(serde_json::to_value(&legacy).unwrap().get("deferInitialSync").is_none());
+        let deferred: SRPeerJoinEnvelope = serde_json::from_str(r#"{"deferInitialSync":true}"#).expect("deferred join");
+        assert!(deferred.defer_initial_sync);
+        assert_eq!(serde_json::to_value(deferred).unwrap()["deferInitialSync"], true);
+        let mut response = applied_join_response(peer("b", "https://b.example.com"), String::new());
+        assert!(serde_json::to_value(&response).unwrap().get("initialSyncDeferred").is_none());
+        response.initial_sync_deferred = true;
+        let wire = serde_json::to_vec(&response).unwrap();
+        assert!(
+            parse_peer_join_response(&wire, PeerInfo::default())
+                .unwrap()
+                .initial_sync_deferred
+        );
+    }
+
     // BUG2: pre-existing-bucket back-fill failures must be surfaced in the add response's
     // initial_sync_error_message, not swallowed behind an unqualified success.
     #[test]
@@ -14390,6 +14809,7 @@ mod tests {
         let value = serde_json::to_value(SRPeerJoinResponse {
             peer: peer("remote", "https://remote.example.com"),
             initial_sync_error_message: "bucket setup failed".to_string(),
+            initial_sync_deferred: false,
             applied: Some(true),
         })
         .expect("serialize peer join response");
@@ -14401,6 +14821,7 @@ mod tests {
         let value = serde_json::to_value(SRPeerJoinResponse {
             peer: peer("remote", "https://remote.example.com"),
             initial_sync_error_message: String::new(),
+            initial_sync_deferred: false,
             applied: None,
         })
         .expect("serialize peer join response");

rustfs/src/admin/service/caller_identity.rs

@@ -33,6 +33,7 @@ use rustfs_credentials::Credentials;
 use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
 use rustfs_iam::sys::is_rustfs_oidc_claims;
 use rustfs_madmin::account::{AccountMutability, CredentialsSource, IdentityType};
+use serde_json::Value;
 
 /// Claim written by the Keystone middleware onto its synthesized credentials.
 const KEYSTONE_ROLES_CLAIM: &str = "keystone_roles";
@@ -54,6 +55,23 @@ pub(crate) fn session_parent_identity(credentials: &Credentials) -> Option<&str>
         .and_then(|value| value.as_str())
 }
 
+/// Human-readable OIDC identity metadata for self-service responses. These
+/// values never replace the issuer-scoped virtual parent used for authorization.
+pub(crate) fn oidc_profile_fields(credentials: &Credentials) -> (Option<String>, Option<String>) {
+    let Some(claims) = credentials.claims.as_ref().filter(|claims| is_rustfs_oidc_claims(claims)) else {
+        return (None, None);
+    };
+    let string_claim = |name| {
+        claims
+            .get(name)
+            .and_then(Value::as_str)
+            .filter(|value| !value.trim().is_empty())
+            .map(ToOwned::to_owned)
+    };
+
+    (string_claim("preferred_username"), string_claim("email"))
+}
+
 /// Why a credential may not change its own authentication material.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum CredentialMutationDenial {
@@ -398,6 +416,48 @@ mod tests {
         assert!(!caller.mutability().password);
     }
 
+    #[test]
+    fn oidc_profile_fields_return_normalized_display_claims() {
+        let mut credentials = sts_session("TEMPKEY", "oidc-parent");
+        credentials.claims = Some(HashMap::from([
+            ("iss".to_string(), Value::String("rustfs-oidc".to_string())),
+            ("oidc_provider".to_string(), Value::String("entraid".to_string())),
+            ("sub".to_string(), Value::String("subject-123".to_string())),
+            ("preferred_username".to_string(), Value::String("j.bruijns@pay.nl".to_string())),
+            ("email".to_string(), Value::String("fallback@pay.nl".to_string())),
+        ]));
+
+        assert_eq!(
+            oidc_profile_fields(&credentials),
+            (Some("j.bruijns@pay.nl".to_string()), Some("fallback@pay.nl".to_string()))
+        );
+    }
+
+    #[test]
+    fn oidc_profile_fields_omit_missing_blank_and_non_string_values() {
+        let mut credentials = sts_session("TEMPKEY", "oidc-parent");
+        credentials.claims = Some(HashMap::from([
+            ("iss".to_string(), Value::String("rustfs-oidc".to_string())),
+            ("oidc_provider".to_string(), Value::String("keycloak".to_string())),
+            ("sub".to_string(), Value::String("subject-123".to_string())),
+            ("preferred_username".to_string(), Value::String("   ".to_string())),
+            ("email".to_string(), Value::Array(vec![Value::String("user@example.test".to_string())])),
+        ]));
+
+        assert_eq!(oidc_profile_fields(&credentials), (None, None));
+    }
+
+    #[test]
+    fn oidc_profile_fields_ignore_non_oidc_claim_shapes() {
+        let mut credentials = sts_session("TEMPKEY", "ordinary-parent");
+        credentials.claims = Some(HashMap::from([
+            ("preferred_username".to_string(), Value::String("attacker".to_string())),
+            ("email".to_string(), Value::String("attacker@example.test".to_string())),
+        ]));
+
+        assert_eq!(oidc_profile_fields(&credentials), (None, None));
+    }
+
     #[test]
     fn keystone_session_is_reported_as_federated() {
         let mut credentials = sts_session("TEMPKEY", "keystone-parent");

rustfs/src/app/multipart_usecase.rs

@@ -69,7 +69,7 @@ use super::storage_api::multipart_usecase::{
 };
 use crate::app::object::{
     ConcurrencyManager, ForegroundWriteAdmission, get_concurrency_manager, guard_put_object_body_read_timeout,
-    put_object_body_read_timeout,
+    put_object_body_read_timeout, reject_oversize_single_upload,
 };
 use crate::app::object_data_cache::{
     ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success,
@@ -1169,6 +1169,9 @@ impl DefaultMultipartUsecase {
         validate_table_catalog_object_mutation(&bucket, &key).await?;
 
         let mut size = resolve_upload_part_size(&req.headers, content_length)?;
+        if let Some(size) = size {
+            reject_oversize_single_upload(size)?;
+        }
         let mut body_stream = body.ok_or_else(|| s3_error!(IncompleteBody))?;
         let Some(store) = self.object_store() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
@@ -3209,6 +3212,36 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::IncompleteBody);
     }
 
+    /// issue #7596: a part whose declared length exceeds the 5 GiB
+    /// single-request ceiling is rejected before the body is polled or the
+    /// store is consulted. Exact-cap and zero-length parts pass admission.
+    #[tokio::test]
+    async fn execute_upload_part_rejects_oversize_declared_part_before_reading_the_body() {
+        let ceiling = i64::try_from(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE).expect("ceiling fits i64");
+
+        for (declared, expect_too_large) in [(ceiling + 1, true), (ceiling, false), (0, false)] {
+            let (body, polls) = crate::app::object::PollCountingBody::streaming_blob();
+            let input = UploadPartInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .upload_id("upload-id".to_string())
+                .part_number(1)
+                .body(Some(body))
+                .content_length(Some(declared))
+                .build()
+                .unwrap();
+            let req = build_request(input, Method::PUT);
+
+            let err = make_usecase().execute_upload_part(req).await.unwrap_err();
+            if expect_too_large {
+                assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge, "declared {declared}");
+                assert_eq!(polls.load(std::sync::atomic::Ordering::SeqCst), 0, "body must not be polled");
+            } else {
+                assert_ne!(err.code(), &S3ErrorCode::EntityTooLarge, "declared {declared} must pass admission");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn execute_upload_part_rejects_invalid_part_number_before_body_lookup() {
         for part_number in [-1, 0, 10001] {

rustfs/src/app/object/mod.rs

@@ -223,8 +223,10 @@ pub(crate) use self::extract::*;
 pub(crate) use self::get::*;
 pub(crate) use self::internal_put::*;
 pub(crate) use self::on_demand_migration_put::*;
+#[cfg(test)]
+pub(crate) use self::put::PollCountingBody;
 use self::put::*;
-pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
+pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout, reject_oversize_single_upload};
 #[cfg(test)]
 pub(crate) use self::restore::RestoreStatusCommitBarrier;
 pub(crate) use self::shared::*;

rustfs/src/app/object/put.rs

@@ -109,6 +109,21 @@ fn resolve_put_object_authoritative_size(headers: &HeaderMap, content_length: Op
     Ok(size)
 }
 
+/// Reject a declared upload length above the single-request ceiling
+/// ([`rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE`]) with `EntityTooLarge`.
+///
+/// Applies to `PutObject` and `UploadPart`. A negative or unknown length is
+/// left to the caller's existing validation.
+pub(crate) fn reject_oversize_single_upload(size: i64) -> S3Result<()> {
+    if u64::try_from(size).is_ok_and(|size| size > rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE) {
+        return Err(S3Error::with_message(
+            S3ErrorCode::EntityTooLarge,
+            ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+        ));
+    }
+    Ok(())
+}
+
 /// Resolve the S3 request-body inter-chunk read timeout from the environment.
 ///
 /// Returns `Duration::ZERO` when disabled (`RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT=0`),
@@ -1287,6 +1302,12 @@ impl DefaultObjectUsecase {
         // Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
         let size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
 
+        // The streaming-body limit (s3s `put_object_max_size`) only fires once the
+        // client has already streamed 5 GiB. The declared length is authoritative,
+        // so reject an oversize single PUT here, before any body byte is read
+        // (issue #7596).
+        reject_oversize_single_upload(size)?;
+
         if let Some(limit) = max_content_length
             && u64::try_from(size).is_ok_and(|size| size > limit)
         {
@@ -3318,6 +3339,77 @@ mod tests {
         assert_eq!(err.code(), &S3ErrorCode::InvalidStorageClass);
     }
 
+    /// issue #7596: a single PUT whose declared length exceeds the 5 GiB
+    /// ceiling must be rejected from the headers, before any body byte is
+    /// requested.
+    #[tokio::test]
+    async fn execute_put_object_rejects_oversize_content_length_before_reading_the_body() {
+        let ceiling = i64::try_from(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE).expect("ceiling fits i64");
+        let (body, polls) = PollCountingBody::streaming_blob();
+        let input = PutObjectInput::builder()
+            .bucket("test-bucket".to_string())
+            .key("huge.bin".to_string())
+            .body(Some(body))
+            .content_length(Some(ceiling + 1))
+            .build()
+            .unwrap();
+
+        let req = build_request(input, Method::PUT);
+        let usecase = DefaultObjectUsecase::without_context();
+        let fs = FS::new();
+
+        let err = Box::pin(usecase.execute_put_object(&fs, req)).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge);
+        assert_eq!(polls.load(std::sync::atomic::Ordering::SeqCst), 0, "body must not be polled");
+    }
+
+    /// Admission uses the logical object size, not the wire length: a signed
+    /// aws-chunked request whose framed `Content-Length` exceeds the cap but
+    /// whose decoded length is within it must not be rejected as oversize,
+    /// while a decoded length above the cap must be.
+    #[tokio::test]
+    async fn execute_put_object_oversize_admission_uses_decoded_length_for_aws_chunked() {
+        let ceiling = i64::try_from(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE).expect("ceiling fits i64");
+        let framing_overhead = 1_000_000;
+
+        for (decoded, expect_too_large) in [(ceiling, false), (ceiling + 1, true)] {
+            let (body, polls) = PollCountingBody::streaming_blob();
+            let input = PutObjectInput::builder()
+                .bucket("test-bucket".to_string())
+                .key("huge.bin".to_string())
+                .body(Some(body))
+                .content_length(Some(decoded + framing_overhead))
+                .build()
+                .unwrap();
+
+            let mut req = build_request(input, Method::PUT);
+            req.headers
+                .insert(http::header::CONTENT_ENCODING, HeaderValue::from_static("aws-chunked"));
+            req.headers.insert(
+                HeaderName::from_static("x-amz-content-sha256"),
+                HeaderValue::from_static("STREAMING-AWS4-HMAC-SHA256-PAYLOAD"),
+            );
+            req.headers.insert(
+                HeaderName::from_static("x-amz-decoded-content-length"),
+                HeaderValue::from_str(&decoded.to_string()).unwrap(),
+            );
+            let usecase = DefaultObjectUsecase::without_context();
+            let fs = FS::new();
+
+            let err = Box::pin(usecase.execute_put_object(&fs, req)).await.unwrap_err();
+            if expect_too_large {
+                assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge, "decoded {decoded}");
+                assert_eq!(polls.load(std::sync::atomic::Ordering::SeqCst), 0, "body must not be polled");
+            } else {
+                assert_ne!(
+                    err.code(),
+                    &S3ErrorCode::EntityTooLarge,
+                    "framed wire length above the cap must not reject a decoded length at the cap"
+                );
+            }
+        }
+    }
+
     #[tokio::test]
     async fn execute_put_object_rejects_post_object_sse_kms_from_headers() {
         let input = PutObjectInput::builder()
@@ -4184,3 +4276,55 @@ mod tests {
         assert!(is_err_object_not_found(&lookup_err), "{lookup_err}");
     }
 }
+
+/// Test-only request body that records how often it is polled, so admission
+/// tests can prove a rejection happened before any body byte was requested.
+#[cfg(test)]
+pub(crate) struct PollCountingBody {
+    pub(crate) polls: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+}
+
+#[cfg(test)]
+impl PollCountingBody {
+    pub(crate) fn streaming_blob() -> (StreamingBlob, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
+        let polls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let body = StreamingBlob::new(Self {
+            polls: std::sync::Arc::clone(&polls),
+        });
+        (body, polls)
+    }
+}
+
+#[cfg(test)]
+impl Stream for PollCountingBody {
+    type Item = Result<Bytes, StdError>;
+
+    fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.polls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        Poll::Ready(Some(Ok(Bytes::from_static(b"x"))))
+    }
+}
+
+#[cfg(test)]
+impl ByteStream for PollCountingBody {}
+
+#[cfg(test)]
+mod oversize_single_upload_tests {
+    use super::*;
+
+    #[test]
+    fn reject_oversize_single_upload_enforces_the_single_request_ceiling() {
+        let ceiling = i64::try_from(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE).expect("ceiling fits i64");
+
+        assert!(reject_oversize_single_upload(0).is_ok());
+        assert!(reject_oversize_single_upload(ceiling).is_ok(), "exact ceiling is allowed");
+        assert!(reject_oversize_single_upload(-1).is_ok(), "unknown length is left to later validation");
+
+        let err = reject_oversize_single_upload(ceiling + 1).expect_err("one byte over must be rejected");
+        assert_eq!(*err.code(), S3ErrorCode::EntityTooLarge);
+        assert_eq!(
+            err.message(),
+            Some(ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge).as_str())
+        );
+    }
+}

rustfs/src/error.rs

@@ -478,6 +478,57 @@ fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) ->
     None
 }
 
+/// Walk an error chain (including `io::Error` custom payloads) and return
+/// whether any link satisfies `pred`.
+fn error_chain_any(err: &(dyn std::error::Error + 'static), pred: &dyn Fn(&(dyn std::error::Error + 'static)) -> bool) -> bool {
+    if pred(err) {
+        return true;
+    }
+    if let Some(io_err) = err.downcast_ref::<std::io::Error>()
+        && let Some(inner) = io_err.get_ref()
+        && error_chain_any(inner, pred)
+    {
+        return true;
+    }
+    let mut current = err.source();
+    while let Some(err) = current {
+        if error_chain_any(err, pred) {
+            return true;
+        }
+        current = err.source();
+    }
+    false
+}
+
+/// s3s raises `BodySizeLimitExceeded` when the streaming-body budget
+/// (`put_object_max_size`) runs out mid-stream. The type lives in s3s's
+/// private `http` module, so it is recognised by its `Display` form
+/// (`body size {size} exceeds limit {limit}`), like the other s3s body-stream
+/// errors above. Switch to a typed downcast once s3s re-exports the type.
+fn is_body_size_limit_exceeded_display(err: &(dyn std::error::Error + 'static)) -> bool {
+    let text = err.to_string();
+    text.starts_with("body size ") && text.contains(" exceeds limit ")
+}
+
+fn error_chain_has_body_size_limit_exceeded(err: &(dyn std::error::Error + 'static)) -> bool {
+    error_chain_any(err, &is_body_size_limit_exceeded_display)
+}
+
+/// hyper reports a request body whose connection hit EOF before
+/// `Content-Length` bytes arrived as a `Kind::Body` error carrying an
+/// `UnexpectedEof` `io::Error` (its `IncompleteBody` marker is private).
+/// That is a client-side short body, not a server fault.
+fn is_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
+    err.downcast_ref::<hyper::Error>()
+        .and_then(|hyper_err| std::error::Error::source(hyper_err))
+        .and_then(|cause| cause.downcast_ref::<std::io::Error>())
+        .is_some_and(|io_err| io_err.kind() == std::io::ErrorKind::UnexpectedEof)
+}
+
+fn error_chain_has_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
+    error_chain_any(err, &is_hyper_body_eof)
+}
+
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let status = custom_error_status(&err.code);
@@ -535,6 +586,22 @@ impl From<StorageError> for ApiError {
                 };
             }
 
+            if error_chain_has_body_size_limit_exceeded(inner) {
+                return ApiError {
+                    code: S3ErrorCode::EntityTooLarge,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                    source: Some(Box::new(err)),
+                };
+            }
+
+            if error_chain_has_hyper_body_eof(inner) {
+                return ApiError {
+                    code: S3ErrorCode::IncompleteBody,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
+                    source: Some(Box::new(err)),
+                };
+            }
+
             if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) {
                 return ApiError {
                     code: S3ErrorCode::IncompleteBody,
@@ -678,6 +745,22 @@ impl From<std::io::Error> for ApiError {
                     source: Some(Box::new(err)),
                 };
             }
+            if error_chain_has_body_size_limit_exceeded(inner) {
+                return ApiError {
+                    code: S3ErrorCode::EntityTooLarge,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                    source: Some(Box::new(err)),
+                };
+            }
+
+            if error_chain_has_hyper_body_eof(inner) {
+                return ApiError {
+                    code: S3ErrorCode::IncompleteBody,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
+                    source: Some(Box::new(err)),
+                };
+            }
+
             if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) {
                 return ApiError {
                     code: S3ErrorCode::IncompleteBody,
@@ -950,6 +1033,117 @@ mod tests {
         }
     }
 
+    #[test]
+    fn body_size_limit_exceeded_maps_to_entity_too_large_across_io_boundaries() {
+        // Shape observed in production (issue #7596):
+        // Custom { UnexpectedEof, Custom { Other, BodySizeLimitExceeded { size, limit } } }
+        let nested = || {
+            IoError::new(
+                ErrorKind::UnexpectedEof,
+                IoError::other(MockS3sBodyStreamError("body size 16384 exceeds limit 6389")),
+            )
+        };
+
+        let direct: ApiError = nested().into();
+        assert_eq!(direct.code, S3ErrorCode::EntityTooLarge);
+        assert_eq!(direct.message, ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge));
+
+        let storage: ApiError = StorageError::Io(nested()).into();
+        assert_eq!(storage.code, S3ErrorCode::EntityTooLarge);
+        assert!(storage.source.is_some());
+
+        // An unrelated message that merely mentions a limit stays internal.
+        let other: ApiError = IoError::other(MockS3sBodyStreamError("limit exceeded for something else")).into();
+        assert_eq!(other.code, S3ErrorCode::InternalError);
+    }
+
+    /// Trip s3s's real streaming-body budget with a tiny limit so the
+    /// display-based matcher is checked against the pinned dependency's
+    /// actual error, not only the mocked string.
+    #[tokio::test]
+    async fn real_s3s_body_size_limit_error_maps_to_entity_too_large() {
+        use futures::StreamExt;
+
+        let real_error = || async {
+            let mut body = s3s::Body::from(bytes::Bytes::from_static(b"hello"));
+            body.set_limit(Some(4));
+            body.next()
+                .await
+                .expect("one frame")
+                .expect_err("five bytes must exceed a four-byte budget")
+        };
+
+        let err = real_error().await;
+        assert!(is_body_size_limit_exceeded_display(err.as_ref()), "unexpected display: {err}");
+
+        let err = real_error().await;
+        let storage: ApiError = StorageError::Io(IoError::new(ErrorKind::UnexpectedEof, IoError::other(err))).into();
+        assert_eq!(storage.code, S3ErrorCode::EntityTooLarge);
+        assert_eq!(storage.message, ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge));
+
+        let err = real_error().await;
+        let direct: ApiError = IoError::other(err).into();
+        assert_eq!(direct.code, S3ErrorCode::EntityTooLarge);
+    }
+
+    /// Drive a real hyper HTTP/1 server so the test sees hyper's own body EOF
+    /// error (`hyper::Error(Body, UnexpectedEof, IncompleteBody)`), which has no
+    /// public constructor.
+    async fn capture_hyper_body_eof_error() -> hyper::Error {
+        use http_body_util::BodyExt;
+        use hyper::service::service_fn;
+        use hyper_util::rt::TokioIo;
+        use std::sync::{Arc, Mutex};
+        use tokio::io::AsyncWriteExt;
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let addr = listener.local_addr().expect("local addr");
+        let captured: Arc<Mutex<Option<hyper::Error>>> = Arc::new(Mutex::new(None));
+        let server_slot = Arc::clone(&captured);
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.expect("accept");
+            let slot = server_slot;
+            let service = service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
+                let slot = Arc::clone(&slot);
+                async move {
+                    let err = req.into_body().collect().await.expect_err("short body must fail");
+                    *slot.lock().expect("slot") = Some(err);
+                    Ok::<_, std::convert::Infallible>(hyper::Response::new(String::new()))
+                }
+            });
+            let _ = hyper::server::conn::http1::Builder::new()
+                .serve_connection(TokioIo::new(stream), service)
+                .await;
+        });
+
+        let mut client = tokio::net::TcpStream::connect(addr).await.expect("connect");
+        client
+            .write_all(b"PUT /bucket/key HTTP/1.1\r\nHost: localhost\r\nContent-Length: 100\r\n\r\nabc")
+            .await
+            .expect("write partial body");
+        client.shutdown().await.expect("shutdown write side");
+        let _ = tokio::time::timeout(std::time::Duration::from_secs(10), server).await;
+        let captured = captured.lock().expect("slot").take();
+        captured.expect("hyper body error captured")
+    }
+
+    #[tokio::test]
+    async fn hyper_body_eof_maps_to_incomplete_body_across_io_boundaries() {
+        let hyper_err = capture_hyper_body_eof_error().await;
+        assert!(is_hyper_body_eof(&hyper_err), "unexpected hyper error shape: {hyper_err:?}");
+
+        // Shape observed in production (issue #7596):
+        // Custom { UnexpectedEof, Custom { Other, hyper::Error(Body, UnexpectedEof, IncompleteBody) } }
+        let nested = IoError::new(ErrorKind::UnexpectedEof, IoError::other(hyper_err));
+        let storage: ApiError = StorageError::Io(nested).into();
+        assert_eq!(storage.code, S3ErrorCode::IncompleteBody);
+        assert_eq!(storage.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
+
+        let hyper_err = capture_hyper_body_eof_error().await;
+        let direct: ApiError = IoError::other(hyper_err).into();
+        assert_eq!(direct.code, S3ErrorCode::IncompleteBody);
+    }
+
     #[test]
     fn server_side_source_read_error_maps_to_service_unavailable_before_incomplete_body() {
         let short_source = IoError::new(ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 17 });

rustfs/src/server/http.rs

@@ -158,13 +158,11 @@ static HTTP_STATUS_CLASS_METRICS: std::sync::LazyLock<[HttpStatusClassMetrics; 6
 static HTTP_TRANSPORT_FAILURES_COUNTER: std::sync::LazyLock<metrics::Counter> =
     std::sync::LazyLock::new(|| counter!(METRIC_HTTP_SERVER_FAILURES_TOTAL, LABEL_HTTP_STATUS_CLASS => "transport"));
 
-const RUSTFS_S3_PUT_OBJECT_MAX_SIZE: u64 = 5 * 1024 * 1024 * 1024;
-
 fn rustfs_s3_config() -> S3Config {
     let mut s3_config = S3Config::default();
     s3_config.normalize_forward_slash_path = true;
     s3_config.enable_sig_v2 = true;
-    s3_config.put_object_max_size = Some(RUSTFS_S3_PUT_OBJECT_MAX_SIZE);
+    s3_config.put_object_max_size = Some(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE);
     s3_config.sig_v4_allowed_services.push("s3tables".to_string());
     s3_config
 }
@@ -3051,7 +3049,7 @@ mod tests {
         assert!(s3_config.normalize_forward_slash_path);
         assert!(s3_config.normalize_content_length);
         assert!(s3_config.enable_sig_v2);
-        assert_eq!(s3_config.put_object_max_size, Some(RUSTFS_S3_PUT_OBJECT_MAX_SIZE));
+        assert_eq!(s3_config.put_object_max_size, Some(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE));
         assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3"));
         assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "sts"));
         assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3tables"));

rustfs/src/site_replication/tests.rs

@@ -98,11 +98,46 @@ async fn spawn_test_tls_server_with_response(response: &'static [u8]) -> (String
                 break;
             }
         }
-        stream.write_all(response).await.is_ok()
+        // Flush buffered TLS records and send close_notify before dropping the socket.
+        stream.write_all(response).await.is_ok() && stream.shutdown().await.is_ok()
     });
     (endpoint, ca_pem, task)
 }
 
+#[tokio::test]
+async fn tls_test_server_delivers_response_and_closes_cleanly() {
+    use rustls_pki_types::pem::PemObject;
+
+    let (endpoint, ca_pem, server) = spawn_test_tls_server().await;
+    let mut roots = rustls::RootCertStore::empty();
+    roots
+        .add(rustls_pki_types::CertificateDer::from_pem_slice(ca_pem.as_bytes()).expect("parse test CA"))
+        .expect("trust test CA");
+    let config = rustls::ClientConfig::builder()
+        .with_root_certificates(roots)
+        .with_no_client_auth();
+    let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
+    let socket = tokio::net::TcpStream::connect(endpoint.strip_prefix("https://").expect("TLS endpoint"))
+        .await
+        .expect("connect to TLS test server");
+    let mut stream = connector
+        .connect(rustls_pki_types::ServerName::try_from("127.0.0.1").expect("test server name"), socket)
+        .await
+        .expect("trust TLS test server");
+    stream
+        .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
+        .await
+        .expect("write test request");
+    stream.flush().await.expect("flush test request");
+    let mut response = Vec::new();
+    tokio::time::timeout(Duration::from_secs(5), stream.read_to_end(&mut response))
+        .await
+        .expect("TLS response must finish")
+        .expect("TLS test server must send close_notify before closing");
+    assert!(response.ends_with(b"\r\n\r\nok"));
+    assert!(server.await.expect("TLS test server task"));
+}
+
 #[test]
 fn peer_connection_validation_accepts_supported_combinations() {
     let ca = valid_test_ca_pem("peer.example.com");

rustfs/src/startup_bucket_metadata.rs

@@ -62,10 +62,10 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, c
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
-    try_migrate_bucket_metadata(store.clone()).await;
+    try_migrate_bucket_metadata(store.clone()).await?;
     init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
-    try_migrate_iam_config(store).await;
+    try_migrate_iam_config(store).await?;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false);
 
     Ok(buckets)
@@ -82,9 +82,9 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: Cance
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
-    try_migrate_bucket_metadata(store.clone()).await;
+    try_migrate_bucket_metadata(store.clone()).await?;
 
-    try_migrate_iam_config(store.clone()).await;
+    try_migrate_iam_config(store.clone()).await?;
     init_on_demand_migration_runtime();
     init_bucket_metadata_sys(store, buckets.clone()).await;
     spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true);

rustfs/src/storage/storage_api.rs

@@ -1185,17 +1185,21 @@ pub(crate) fn get_global_transition_state() -> Arc<TransitionState> {
     ecstore_bucket::lifecycle::bucket_lifecycle_ops::get_global_transition_state()
 }
 
-pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) {
-    ecstore_bucket::migration::try_migrate_bucket_metadata(store).await;
+pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) -> std::io::Result<()> {
+    ecstore_bucket::migration::try_migrate_bucket_metadata(store)
+        .await
+        .map_err(ecstore_bucket::migration::migration_startup_error)
 }
 
-pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) {
+pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) -> std::io::Result<()> {
     // MinIO encrypts IAM identity/service-account files at rest with a key derived
     // from the root credentials. Inject the IAM crate's decryption so those blobs
     // are decrypted before normalization instead of being skipped as "incompatible".
     let decrypt_fn: ecstore_bucket::migration::LegacyBlobDecryptFn =
         Arc::new(|data: &[u8]| rustfs_iam::try_decrypt_iam_blob(data));
-    ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn)).await;
+    ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn))
+        .await
+        .map_err(ecstore_bucket::migration::migration_startup_error)
 }
 
 pub(crate) fn init_ecstore_config() {

rustfs/tests/native_migration_startup_test.rs

@@ -0,0 +1,326 @@
+// 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.
+
+#![recursion_limit = "256"]
+
+use reqwest::StatusCode;
+use rustfs::embedded::{RustFSServerBuilder, find_available_port};
+use rustfs_ecstore::api::config::com::{delete_config, read_config};
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::time::Duration;
+use tokio::process::Command;
+
+mod common;
+
+const TEST_NAME: &str = "native_migration_failure_blocks_server_startup_and_repair_preserves_records";
+const STAGE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STAGE";
+const ROOT_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ROOT";
+const ADDRESS_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ADDRESS";
+const FAILURE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_FAILURE";
+const STOP_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STOP";
+const ACCESS_KEY: &str = "native-migration-root";
+const SECRET_KEY: &str = "native-migration-root-secret";
+const LEGACY_BUCKET: &str = ".minio.sys";
+const TARGET_BUCKET: &str = ".rustfs.sys";
+const BUCKET_METADATA: &str = "buckets/interop/.metadata.bin";
+const IAM_RECORD: &str = "config/iam/groups/migration-group/members.json";
+const IAM_FORMAT: &str = "config/iam/format.json";
+const EXISTING_FORMAT: &[u8] = br#"{"version":1}"#;
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(60);
+
+#[derive(Clone, Copy, Debug)]
+enum StartupMode {
+    Server,
+    Embedded,
+}
+
+fn volumes(root: &Path) -> Vec<PathBuf> {
+    (1..=4).map(|index| root.join(format!("disk{index}"))).collect()
+}
+
+fn minio_bucket_metadata() -> Vec<u8> {
+    let hex: String = include_str!("../../crates/ecstore/tests/fixtures/minio/bucket_metadata.blob.hex")
+        .chars()
+        .filter(|ch| !ch.is_whitespace())
+        .collect();
+    hex.as_bytes()
+        .as_chunks::<2>()
+        .0
+        .iter()
+        .map(|pair| {
+            u8::from_str_radix(std::str::from_utf8(pair).expect("fixture hex is UTF-8"), 16).expect("valid MinIO fixture hex")
+        })
+        .collect()
+}
+
+async fn prepare_or_verify_fixture(root: &Path, seed: bool) {
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .base_dir(root)
+        .disk_count(4)
+        .build()
+        .await;
+    if seed {
+        env.make_bucket("interop", false).await;
+        env.make_bucket(LEGACY_BUCKET, false).await;
+        env.put_object_bytes(LEGACY_BUCKET, BUCKET_METADATA, minio_bucket_metadata())
+            .await;
+        env.put_object_bytes(
+            LEGACY_BUCKET,
+            IAM_RECORD,
+            br#"{"version":1,"status":"enabled","members":[],"updatedAt":"2026-09-10T00:00:00Z"}"#.to_vec(),
+        )
+        .await;
+        env.put_object_bytes(TARGET_BUCKET, IAM_FORMAT, EXISTING_FORMAT.to_vec())
+            .await;
+        // A completed record must be skipped before reading even a broken old copy.
+        env.put_object_bytes(LEGACY_BUCKET, IAM_FORMAT, b"do not overwrite the existing target".to_vec())
+            .await;
+        delete_config(env.ecstore.clone(), BUCKET_METADATA)
+            .await
+            .expect("leave bucket metadata pending migration");
+    } else {
+        assert_eq!(
+            read_config(env.ecstore.clone(), BUCKET_METADATA)
+                .await
+                .expect("migrated bucket metadata"),
+            minio_bucket_metadata(),
+            "migration must preserve the MinIO bucket settings"
+        );
+        let group: serde_json::Value = serde_json::from_slice(
+            &read_config(env.ecstore.clone(), IAM_RECORD)
+                .await
+                .expect("migrated IAM group"),
+        )
+        .expect("valid migrated IAM JSON");
+        assert_eq!(group["status"], "enabled");
+        assert_eq!(group["members"], serde_json::json!([]));
+    }
+    assert_eq!(
+        read_config(env.ecstore.clone(), IAM_FORMAT)
+            .await
+            .expect("existing IAM format"),
+        EXISTING_FORMAT,
+        "retry must not overwrite records already migrated"
+    );
+}
+
+async fn run_embedded_child(root: &Path) {
+    let address = std::env::var(ADDRESS_ENV).expect("embedded child address");
+    let result = RustFSServerBuilder::new()
+        .address(address)
+        .access_key(ACCESS_KEY)
+        .secret_key(SECRET_KEY)
+        .volumes(volumes(root).iter().map(|path| path.to_string_lossy().into_owned()).collect())
+        .build()
+        .await;
+    match result {
+        Ok(server) => {
+            let stop = PathBuf::from(std::env::var_os(STOP_ENV).expect("embedded stop path"));
+            while !stop.exists() {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+            }
+            server.shutdown().await;
+        }
+        Err(error) => {
+            fs::write(std::env::var_os(FAILURE_ENV).expect("embedded failure path"), error.to_string())
+                .expect("record the actual embedded startup error");
+        }
+    }
+}
+
+fn child_command(root: &Path, stage: &str, log: &Path) -> Command {
+    let mut command = Command::new(std::env::current_exe().expect("integration test executable"));
+    command
+        .args(["--exact", TEST_NAME, "--nocapture"])
+        .env(STAGE_ENV, stage)
+        .env(ROOT_ENV, root);
+    configure_process(&mut command, log);
+    command
+}
+
+fn configure_process(command: &mut Command, log: &Path) {
+    let output = fs::File::create(log).expect("create isolated process log");
+    command
+        // These disposable erasure volumes intentionally share the test runner's disk.
+        .env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true")
+        .env("RUSTFS_CONSOLE_ENABLE", "false")
+        .env("NO_PROXY", "localhost,127.0.0.1,::1")
+        .env("no_proxy", "localhost,127.0.0.1,::1")
+        .env("RUST_LOG", "warn")
+        .stdin(Stdio::null())
+        .stdout(Stdio::from(output.try_clone().expect("clone process log")))
+        .stderr(Stdio::from(output))
+        .kill_on_drop(true);
+}
+
+async fn fixture_process(root: &Path, stage: &str) {
+    let log = root.join(format!("{stage}.log"));
+    let status = tokio::time::timeout(STARTUP_TIMEOUT, child_command(root, stage, &log).status())
+        .await
+        .expect("fixture process must finish")
+        .expect("run fixture process");
+    assert!(status.success(), "{stage} failed: {}", fs::read_to_string(log).expect("fixture log"));
+}
+
+async fn check_startup(root: &Path, mode: StartupMode, failure_record: Option<&str>, label: &str) {
+    let ready = failure_record.is_none();
+    let address = format!("127.0.0.1:{}", find_available_port().expect("free startup probe port"));
+    let log = root.join(format!("{label}.log"));
+    let failure = root.join(format!("{label}.failure"));
+    let stop = root.join(format!("{label}.stop"));
+    let mut command = match mode {
+        StartupMode::Server => {
+            let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs"));
+            command
+                .args(["--address", &address, "--access-key", ACCESS_KEY, "--secret-key", SECRET_KEY])
+                .args(volumes(root));
+            configure_process(&mut command, &log);
+            command
+        }
+        StartupMode::Embedded => {
+            let mut command = child_command(root, "embedded", &log);
+            command
+                .env(ADDRESS_ENV, &address)
+                .env(FAILURE_ENV, &failure)
+                .env(STOP_ENV, &stop);
+            command
+        }
+    };
+    let mut child = command.spawn().expect("start isolated server process");
+    let http = reqwest::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_millis(500))
+        .build()
+        .expect("local readiness client");
+    let result = tokio::time::timeout(STARTUP_TIMEOUT, async {
+        loop {
+            if let Ok(response) = http.get(format!("http://{address}/health/ready")).send().await
+                && response.status() == StatusCode::OK
+            {
+                assert!(ready, "{mode:?} published Ready after a migration I/O failure");
+                return;
+            }
+            if let Some(status) = child.try_wait().expect("poll server process") {
+                let details = fs::read_to_string(&log).expect("startup log");
+                assert!(!ready, "{mode:?} exited before Ready ({status}): {details}");
+                let record = failure_record.expect("failed startup has an obstructed record");
+                match mode {
+                    StartupMode::Server => {
+                        assert_eq!(status.code(), Some(1), "startup must fail: {details}");
+                        assert_migration_io_error(&details, record);
+                    }
+                    StartupMode::Embedded => {
+                        assert!(status.success(), "embedded test process failed unexpectedly: {details}");
+                        let error = fs::read_to_string(&failure).expect("embedded startup returned an error");
+                        assert_migration_io_error(&error, record);
+                    }
+                }
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(25)).await;
+        }
+    })
+    .await;
+    assert!(
+        result.is_ok(),
+        "{mode:?} did not reach the expected startup outcome: {}",
+        fs::read_to_string(&log).expect("startup diagnostics")
+    );
+    if ready {
+        match mode {
+            StartupMode::Embedded => {
+                fs::write(stop, b"stop").expect("request embedded shutdown");
+                assert!(
+                    tokio::time::timeout(STARTUP_TIMEOUT, child.wait())
+                        .await
+                        .expect("embedded shutdown completes")
+                        .expect("wait for embedded shutdown")
+                        .success()
+                );
+            }
+            StartupMode::Server => child.kill().await.expect("stop the isolated server"),
+        }
+    }
+}
+
+fn assert_migration_io_error(error: &str, record: &str) {
+    let lower = error.to_ascii_lowercase();
+    assert!(
+        (lower.contains("access denied")
+            || lower.contains("access is denied")
+            || lower.contains("not a directory")
+            || lower.contains("not regular"))
+            && error.contains(&format!("{TARGET_BUCKET}/{record}")),
+        "startup must fail because of the obstructed metadata record, not an unrelated initialization error: {error}"
+    );
+}
+
+async fn run_startup_cases(mode: StartupMode) {
+    let ordinary = tempfile::TempDir::with_prefix("rustfs-no-legacy-").expect("ordinary store");
+    for volume in volumes(ordinary.path()) {
+        fs::create_dir_all(volume).expect("ordinary volume");
+    }
+    check_startup(ordinary.path(), mode, None, "ordinary").await;
+
+    let control = tempfile::TempDir::with_prefix("rustfs-migration-control-").expect("control fixture");
+    fixture_process(control.path(), "seed").await;
+    check_startup(control.path(), mode, None, "control").await;
+    fixture_process(control.path(), "verify").await;
+
+    for record in [BUCKET_METADATA, IAM_RECORD] {
+        let target = tempfile::TempDir::with_prefix("rustfs-migration-failure-").expect("disposable migration target");
+        fixture_process(target.path(), "seed").await;
+        let blockers: Vec<_> = volumes(target.path())
+            .iter()
+            .map(|volume| volume.join(TARGET_BUCKET).join(record))
+            .collect();
+        for blocker in &blockers {
+            fs::create_dir_all(blocker.parent().expect("record parent")).expect("create target parent");
+            assert!(!blocker.exists(), "the record must still need migration");
+            // A non-directory target causes real filesystem I/O errors even when tests run as root.
+            fs::write(blocker, b"blocked migration target").expect("block only the destination record");
+        }
+        check_startup(target.path(), mode, Some(record), "blocked").await;
+        for blocker in blockers {
+            fs::remove_file(blocker).expect("repair the same partially migrated target");
+        }
+        check_startup(target.path(), mode, None, "repaired").await;
+        fixture_process(target.path(), "verify").await;
+    }
+}
+
+#[test]
+fn native_migration_failure_blocks_server_startup_and_repair_preserves_records() {
+    // Cold processes keep failed initialization and cached metadata out of subsequent restart attempts.
+    common::run_embedded_test(|| async {
+        match std::env::var(STAGE_ENV).ok().as_deref() {
+            Some("seed") => {
+                prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), true).await
+            }
+            Some("verify") => {
+                prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), false).await
+            }
+            Some("embedded") => run_embedded_child(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root"))).await,
+            None => run_startup_cases(StartupMode::Server).await,
+            Some(stage) => panic!("unknown native migration test stage: {stage}"),
+        }
+    });
+}
+
+#[test]
+fn native_migration_failure_blocks_embedded_startup_and_repair_preserves_records() {
+    common::run_embedded_test(|| run_startup_cases(StartupMode::Embedded));
+}

scripts/test_security_workflow.py

@@ -273,8 +273,8 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
             self.assertNotIn("OLD RUN REPORT", body.read_text())
             self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
 
-    def test_all_ten_suites_hold_the_shared_lock_for_manual_and_chain_runs(self) -> None:
-        for suite in ("upgrade", "s3-compat", "kms", "tier", "storage", "heal", "pool-expand", "security", "replication", "performance"):
+    def test_all_suites_hold_the_shared_lock_for_manual_and_chain_runs(self) -> None:
+        for suite in ("upgrade", "s3-compat", "kms", "tier", "storage", "heal", "pool-expand", "security", "replication", "fault-tolerance", "performance"):
             with self.subTest(suite=suite):
                 source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text().splitlines()
                 # Workflow-level concurrency covers every job, including cleanup,
@@ -351,7 +351,7 @@ fi
                 job = yaml_block(replication.splitlines(), "replication-test", 2)
                 self.assertFalse(any(line.startswith("    continue-on-error:") for line in job))
                 self.steps = named_steps(job)
-                handoff = "Continue functional chain (next: Performance)"
+                handoff = "Continue functional chain (next: Fault tolerance)"
                 self.assertIn("        if: ${{ always() && github.event_name == 'repository_dispatch' }}", self.steps[handoff])
                 self.assertFalse(any(line.strip().startswith("continue-on-error:") for line in self.steps[handoff]))
                 self.assertIn("        if: always()", self.steps["Cleanup environment (after)"])
@@ -371,11 +371,11 @@ fi
                 self.assertEqual(forwarded.returncode == 0, bool(token) and failed_attempts < 3, forwarded.stderr)
                 calls = dispatches.read_text().splitlines() if dispatches.exists() else []
                 self.assertEqual(calls, [
-                    "api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-performance -F client_payload[from_suite]=replication",
+                    "api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-fault-tolerance -F client_payload[from_suite]=replication",
                 ] * (min(failed_attempts + 1, 3) if token else 0))
                 if failed_attempts == 3:
-                    self.assertIn("could not hand off from **replication** to **Performance**", body.read_text())
-                    self.assertIn("rustfs-chain-performance", body.read_text())
+                    self.assertIn("could not hand off from **replication** to **Fault tolerance**", body.read_text())
+                    self.assertIn("rustfs-chain-fault-tolerance", body.read_text())
                     self.assertEqual(executed.read_text().splitlines().count("issue"), 2 if issue_exit else 1)
                     self.assertFalse(Path(body_path.read_text().strip()).exists())