ci: allow macOS release builds to finish (#6976)

Co-authored-by: heihutu <heihutu@gmail.com>

.github/workflows/build.yml

@@ -244,7 +244,7 @@ jobs:
     needs: [ build-check, prepare-platform-matrix ]
     if: needs.build-check.outputs.should_build == 'true' && needs.prepare-platform-matrix.result == 'success'
     runs-on: ${{ matrix.os }}
-    timeout-minutes: 150
+    timeout-minutes: 180
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       # Release binaries ship without dial9 telemetry and therefore do not need

crates/ecstore/src/disk/disk_store.rs

@@ -2435,6 +2435,25 @@ mod tests {
         assert_eq!(window.acc_time, 18_000);
     }
 
+    #[test]
+    fn timed_action_slot_snapshot_skips_writer_owned_slot() {
+        let slot = TimedActionSlot::default();
+        slot.unix_sec.store(70, Ordering::Relaxed);
+        slot.count.store(2, Ordering::Relaxed);
+        slot.acc_time.store(18_000, Ordering::Relaxed);
+        slot.version.store(2, Ordering::Release);
+        assert_eq!(slot.snapshot(), Some((70, 2, 18_000)));
+
+        assert_eq!(slot.version.compare_exchange(2, 3, Ordering::AcqRel, Ordering::Relaxed), Ok(2));
+        slot.unix_sec.store(71, Ordering::Relaxed);
+        slot.count.store(1, Ordering::Relaxed);
+        slot.acc_time.store(11_000, Ordering::Relaxed);
+        assert_eq!(slot.snapshot(), None);
+
+        slot.version.store(4, Ordering::Release);
+        assert_eq!(slot.snapshot(), Some((71, 1, 11_000)));
+    }
+
     #[test]
     fn disk_health_metrics_snapshot_exports_waiting_errors_and_operation_windows() {
         let metrics = DiskHealthMetricEpoch::default();

crates/scanner/src/data_usage_define.rs

@@ -131,6 +131,39 @@ pub(crate) async fn read_config_with_revision<S: ScannerObjectIO>(
     }
 }
 
+pub(crate) fn usage_floor_primary_read_error_allows_backup(err: &Error) -> bool {
+    match err {
+        Error::FileCorrupt
+        | Error::CorruptedFormat
+        | Error::CorruptedBackend
+        | Error::PartMissingOrCorrupt
+        | Error::LessData
+        | Error::MoreData => true,
+        Error::Io(io_error) => {
+            matches!(io_error.kind(), std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof)
+                || error_chain_has_usage_floor_corruption_signature(io_error)
+        }
+        _ => false,
+    }
+}
+
+fn error_chain_has_usage_floor_corruption_signature(error: &(dyn std::error::Error + 'static)) -> bool {
+    let mut current = Some(error);
+    while let Some(err) = current {
+        let message = err.to_string();
+        if message.contains("InlineData value out of range")
+            || message.contains("InlineData key out of range")
+            || message.contains("insufficient data for metadata")
+            || message.contains("insufficient data for meta length")
+            || message.contains("insufficient data for CRC")
+        {
+            return true;
+        }
+        current = err.source();
+    }
+    false
+}
+
 /// Read only the object revision without materializing its body.
 pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
     match store

crates/scanner/src/scanner/cycle_state.rs

@@ -14,7 +14,9 @@
 /// Scanner cycle-state codec, persisted usage floors, and cycle-state persistence.
 use super::*;
 use crate::ScannerGetObjectReader;
-use crate::data_usage_define::{DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH};
+use crate::data_usage_define::{
+    DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup,
+};
 use crate::storage_api::owner::ObjectIO as _;
 use tokio::io::AsyncReadExt as _;
 
@@ -1717,6 +1719,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
         let backup_path = format!("{primary_path}.bkp");
         let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str();
         let mut recovered_primary_companion_epoch = None;
+        let mut primary_read_error = None;
         let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await {
             Ok((Some(data), revision)) => {
                 let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
@@ -1776,6 +1779,12 @@ pub(super) async fn persisted_usage_floor_for_startup(
                 }
             }
             Ok((None, _)) => None,
+            Err(err) if !is_v2_path && usage_floor_primary_read_error_allows_backup(&err) => {
+                primary_read_error = Some(format!("failed to read scanner usage epoch floor from {primary_path}: {err}"));
+                invalid_baseline_path.get_or_insert_with(|| primary_path.to_string());
+                unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string());
+                None
+            }
             Err(err) => {
                 return Err(ScannerError::Other(format!(
                     "failed to read scanner usage epoch floor from {primary_path}: {err}"
@@ -1855,6 +1864,14 @@ pub(super) async fn persisted_usage_floor_for_startup(
                 )));
             }
         }
+        if let Some(primary_read_error) = primary_read_error
+            && !any_found
+        {
+            return Err(ScannerError::Other(format!(
+                "{}; no valid scanner usage floor backup was available at {backup_path}",
+                primary_read_error
+            )));
+        }
         if any_found {
             if bootstrap_pending {
                 return Err(ScannerError::Other(

crates/scanner/src/scanner/leadership.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 /// Leader-lock claiming, usage-epoch fencing, and lock-loss handling.
 use super::*;
+use crate::data_usage_define::usage_floor_primary_read_error_allows_backup;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum ScannerLeadershipClaimReconcile {
@@ -142,20 +143,34 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
         }
     }
 
-    for path in [
-        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
-        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
-    ] {
-        let (legacy, _) = read_config_with_revision(storeapi.clone(), &path)
-            .await
-            .map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?;
+    let legacy_primary_path = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string();
+    let legacy_backup_path = format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let mut legacy_primary_read_error = None;
+    for path in [&legacy_primary_path, &legacy_backup_path] {
+        let legacy = match read_config_with_revision(storeapi.clone(), path).await {
+            Ok((legacy, _)) => legacy,
+            Err(err) if path == &legacy_primary_path && usage_floor_primary_read_error_allows_backup(&err) => {
+                legacy_primary_read_error = Some(format!("failed to read legacy scanner usage epoch fence from {path}: {err}"));
+                continue;
+            }
+            Err(err) => {
+                return Err(ScannerError::Other(format!(
+                    "failed to read legacy scanner usage epoch fence from {path}: {err}"
+                )));
+            }
+        };
         if let Some(legacy) = legacy.as_deref() {
-            let usage = decode_usage_snapshot_for_epoch_fence(legacy, &path, false)?;
+            let usage = decode_usage_snapshot_for_epoch_fence(legacy, path, false)?;
             if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
                 return Ok(Some(usage));
             }
         }
     }
+    if let Some(legacy_primary_read_error) = legacy_primary_read_error {
+        return Err(ScannerError::Other(format!(
+            "{legacy_primary_read_error}; no valid legacy scanner usage epoch fence backup was available at {legacy_backup_path}"
+        )));
+    }
     // A missing usage snapshot is an uninitialized state, not an empty
     // snapshot. Leadership fencing may proceed without creating a plausible
     // default; the first authoritative scanner publication will create it.
@@ -258,6 +273,12 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
             Some(epoch) if epoch == claimed_epoch => return Ok(()),
             Some(_) | None => {}
         }
+        // A validated pre-marker legacy baseline needs an explicit complete
+        // identity before acquiring an epoch. Otherwise the v2 reader would
+        // reject the fenced value on its next startup.
+        if !usage.usage_snapshot_bootstrap_pending {
+            usage.usage_snapshot_complete = true;
+        }
         usage.scanner_epoch = Some(claimed_epoch);
         let data = serde_json::to_vec(&usage)
             .map_err(|err| ScannerError::Other(format!("failed to encode scanner usage epoch fence: {err}")))?;

crates/scanner/src/scanner/tests.rs

@@ -574,6 +574,7 @@ struct MemoryConfigStore {
     objects: Mutex<HashMap<String, Vec<u8>>>,
     revisions: Mutex<HashMap<String, u64>>,
     insert_after_gets: Mutex<HashMap<String, Vec<u8>>>,
+    read_errors: Mutex<HashMap<String, EcstoreError>>,
     delayed_gets: Mutex<HashMap<String, Duration>>,
     non_regular_objects: Mutex<HashSet<String>>,
     fail_put_number: Mutex<HashMap<String, usize>>,
@@ -623,6 +624,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
         _opts: &ObjectOptions,
     ) -> EcstoreResult<GetObjectReader> {
         let key = memory_config_key(bucket, object);
+        if let Some(error) = self.read_errors.lock().await.get(&key).cloned() {
+            return Err(error);
+        }
         if let Some(delay) = self.delayed_gets.lock().await.remove(&key) {
             tokio::time::sleep(delay).await;
         }
@@ -2902,6 +2906,147 @@ async fn scanner_usage_floor_recovers_from_incomplete_v2_primary_using_fenced_ba
     );
 }
 
+async fn seed_legacy_primary_read_error_with_backup(store: &Arc<MemoryConfigStore>, error: EcstoreError, epoch: u64, cycle: u64) {
+    let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
+    let legacy_backup = format!("{legacy_primary}.bkp");
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(epoch);
+    backup.scanner_cycle = Some(cycle);
+
+    store
+        .read_errors
+        .lock()
+        .await
+        .insert(memory_config_key(RUSTFS_META_BUCKET, legacy_primary), error);
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, &legacy_backup),
+        serde_json::to_vec(&backup).expect("legacy backup usage snapshot should encode"),
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_recovers_legacy_backup_after_primary_decode_error() {
+    let store = Arc::new(MemoryConfigStore::default());
+    seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
+
+    let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
+        .await
+        .expect("valid legacy backup should recover the startup floor");
+    assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
+    assert_eq!(
+        floor,
+        PersistedUsageFloor {
+            next_cycle: 42,
+            leader_epoch: 19,
+        }
+    );
+    assert_eq!(
+        persisted_usage_floor(store)
+            .await
+            .expect("valid legacy backup should recover the authoritative floor"),
+        floor
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_does_not_bootstrap_over_corrupt_legacy_primary_without_backup() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
+    store
+        .read_errors
+        .lock()
+        .await
+        .insert(memory_config_key(RUSTFS_META_BUCKET, legacy_primary), EcstoreError::FileCorrupt);
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("corrupt legacy primary without a valid backup must remain fail-closed");
+    assert!(err.to_string().contains("no valid scanner usage floor backup"), "unexpected error: {err}");
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_does_not_fallback_to_legacy_after_corrupt_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let mut v2_backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    v2_backup.scanner_epoch = Some(8);
+    v2_backup.scanner_cycle = Some(11);
+    let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    legacy.scanner_epoch = Some(3);
+    legacy.scanner_cycle = Some(7);
+    store.read_errors.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        EcstoreError::FileCorrupt,
+    );
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())),
+        serde_json::to_vec(&v2_backup).expect("v2 backup usage snapshot should encode"),
+    );
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&legacy).expect("legacy usage snapshot should encode"),
+    );
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("corrupt v2 primary must not recover without a primary revision");
+    assert!(
+        err.to_string().contains(&format!(
+            "failed to read scanner usage epoch floor from {}",
+            DATA_USAGE_OBJ_NAME_PATH.as_str()
+        )),
+        "unexpected error: {err}"
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_keeps_transient_primary_read_error_fail_closed() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
+    seed_legacy_primary_read_error_with_backup(
+        &store,
+        EcstoreError::Io(std::io::Error::new(
+            std::io::ErrorKind::ConnectionReset,
+            "connection reset while reading usage primary",
+        )),
+        19,
+        41,
+    )
+    .await;
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("transient primary errors must not be converted into backup recovery");
+    assert!(
+        err.to_string()
+            .contains(&format!("failed to read scanner usage epoch floor from {legacy_primary}")),
+        "unexpected error: {err}"
+    );
+    assert!(
+        !err.to_string().contains("no valid scanner usage floor backup"),
+        "transient error should not enter corrupt-primary fallback: {err}"
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_keeps_outdated_primary_metadata_fail_closed() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
+    seed_legacy_primary_read_error_with_backup(&store, EcstoreError::OutdatedXLMeta, 19, 41).await;
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("outdated primary metadata must not be converted into backup recovery");
+    assert!(
+        err.to_string()
+            .contains(&format!("failed to read scanner usage epoch floor from {legacy_primary}")),
+        "unexpected error: {err}"
+    );
+    assert!(
+        !err.to_string().contains("no valid scanner usage floor backup"),
+        "outdated metadata should not enter corrupt-primary fallback: {err}"
+    );
+}
+
 #[tokio::test]
 async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -3004,6 +3149,19 @@ async fn scanner_leadership_fencing_recovers_incomplete_v2_primary_from_backup()
     assert_eq!(recovered.scanner_cycle, Some(103));
 }
 
+#[tokio::test]
+async fn scanner_usage_floor_leadership_fencing_recovers_legacy_backup_after_primary_decode_error() {
+    let store = Arc::new(MemoryConfigStore::default());
+    seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
+
+    let recovered = usage_snapshot_for_epoch_fence(store, None, false)
+        .await
+        .expect("a valid legacy backup should provide the fencing baseline")
+        .expect("the fencing baseline should be present");
+    assert_eq!(recovered.scanner_epoch, Some(19));
+    assert_eq!(recovered.scanner_cycle, Some(41));
+}
+
 #[tokio::test]
 async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -3955,6 +4113,162 @@ async fn scanner_defers_leadership_when_usage_snapshots_are_stably_absent() {
     assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err());
 }
 
+#[tokio::test]
+async fn scanner_usage_floor_leadership_claim_recovers_legacy_backup_after_primary_decode_error() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let ctx = CancellationToken::new();
+    seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
+
+    let mut revision = DataUsageCacheRevision::Missing;
+    let mut cycle = CurrentCycle::default();
+    let mut persisted_epoch = 19;
+    assert!(
+        claim_scanner_leadership(
+            &ctx,
+            store.clone(),
+            &mut cycle,
+            &mut revision,
+            &mut persisted_epoch,
+            false,
+            ScannerCycleResetPolicy::None,
+        )
+        .await
+    );
+
+    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("leadership claim should persist after legacy backup recovery");
+    let (_, claimed_epoch) = decode_scanner_cycle_state(&state).expect("leadership claim should decode");
+    assert_eq!(claimed_epoch, 20);
+    assert_eq!(persisted_epoch, 20);
+
+    let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("legacy backup recovery should publish a fenced v2 usage primary");
+    let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("fenced v2 usage primary should decode");
+    assert_eq!(usage.scanner_epoch, Some(20));
+    assert_eq!(usage.scanner_cycle, Some(41));
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn scanner_legacy_usage_backup_survives_fencing_and_restart_after_real_metadata_truncation() {
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+    let (temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
+    let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    usage.usage_snapshot_complete = false;
+    usage.scanner_cycle = Some(41);
+    let mut data = serde_json::to_vec(&usage).expect("legacy usage should encode");
+    data.resize(data.len() + 16 * 1024, b' ');
+    let legacy_path = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
+    let backup_path = format!("{legacy_path}.bkp");
+    for path in [legacy_path, backup_path.as_str()] {
+        save_config(store.clone(), path, data.clone())
+            .await
+            .expect("legacy usage fixture should persist");
+    }
+    let mut truncated_files = Vec::new();
+    for disk_index in 0..4 {
+        let path = temp_dir
+            .path()
+            .join(format!("pool0/disk{disk_index}"))
+            .join(RUSTFS_META_BUCKET)
+            .join(legacy_path)
+            .join("xl.meta");
+        let file = tokio::fs::OpenOptions::new()
+            .write(true)
+            .open(&path)
+            .await
+            .expect("legacy inline metadata should exist");
+        assert!(file.metadata().await.expect("metadata should be readable").len() > 4096);
+        file.set_len(4096).await.expect("fixture should truncate at a page boundary");
+        truncated_files.push((
+            path.clone(),
+            tokio::fs::read(&path)
+                .await
+                .expect("truncated evidence should remain readable"),
+        ));
+    }
+
+    let store = restart_scanner_cycle_store_from(&store).await;
+    let error = read_config_with_revision(store.clone(), legacy_path)
+        .await
+        .expect_err("truncated primary must fail in the real object reader");
+    assert!(
+        error.to_string().contains("InlineData value out of range"),
+        "unexpected truncated-primary error: {error}"
+    );
+    assert_eq!(
+        read_config_with_revision(store.clone(), &backup_path)
+            .await
+            .expect("backup should remain readable")
+            .0,
+        Some(data.clone()),
+    );
+    let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
+        .await
+        .expect("intact legacy backup must recover startup despite truncated primary");
+    assert_eq!(
+        floor,
+        PersistedUsageFloor {
+            next_cycle: 42,
+            leader_epoch: 0
+        }
+    );
+    assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
+
+    let baseline = read_data_usage_persist_baseline(store.clone())
+        .await
+        .expect("publication must also read the intact backup");
+    assert_eq!(baseline.data.as_deref(), Some(data.as_slice()));
+    assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
+    fence_scanner_usage_epoch_with_expected_epoch(&CancellationToken::new(), store.clone(), 7, None, false)
+        .await
+        .expect("legacy backup must be fenced into v2");
+    let fenced = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("fencing must publish a v2 usage primary");
+    let fenced = serde_json::from_slice::<DataUsageInfo>(&fenced).expect("fenced v2 usage primary should decode");
+    assert!(
+        fenced.usage_snapshot_complete,
+        "the fenced pre-marker baseline must become a complete v2 identity"
+    );
+
+    let store = restart_scanner_cycle_store_from(&store).await;
+    let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
+        .await
+        .expect("a restart after fencing must preserve the recovered floor");
+    assert_eq!(
+        floor,
+        PersistedUsageFloor {
+            next_cycle: 42,
+            leader_epoch: 7
+        }
+    );
+    assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
+    let restarted = restart_scanner_cycle_store_from(&store).await;
+    assert_eq!(
+        persisted_usage_floor(restarted)
+            .await
+            .expect("fenced floor must survive another restart"),
+        PersistedUsageFloor {
+            next_cycle: 42,
+            leader_epoch: 7
+        }
+    );
+    for (path, bytes) in truncated_files {
+        assert_eq!(tokio::fs::read(path).await.expect("legacy evidence must not be removed"), bytes);
+    }
+    assert_eq!(
+        read_config(store, &backup_path)
+            .await
+            .expect("legacy backup must remain intact"),
+        data
+    );
+    global_metrics().set_cycle(None).await;
+    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
+}
+
 #[tokio::test]
 async fn usage_bootstrap_pending_unblocks_first_leadership_claim() {
     let store = Arc::new(MemoryConfigStore::default());

crates/scanner/src/scanner/usage_store.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 /// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup.
 use super::*;
+use crate::data_usage_define::usage_floor_primary_read_error_allows_backup;
 use crate::storage_api::owner::ScannerPublicationCommitState;
 use std::collections::HashMap;
 use std::sync::atomic::AtomicBool;
@@ -53,6 +54,30 @@ pub(super) struct DataUsagePersistBaseline {
     pub(super) revision: DataUsageCacheRevision,
 }
 
+async fn read_usage_persist_candidate(
+    storeapi: Arc<impl ScannerObjectIO>,
+    path: &str,
+) -> Result<(Option<Vec<u8>>, DataUsageCacheRevision), EcstoreError> {
+    let primary = read_config_with_revision(storeapi.clone(), path).await;
+    if path != LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str() {
+        return primary;
+    }
+    let Err(primary_error) = &primary else {
+        return primary;
+    };
+    if !usage_floor_primary_read_error_allows_backup(primary_error) {
+        return primary;
+    }
+    let backup_path = format!("{path}.bkp");
+    let backup = read_config_with_revision(storeapi, &backup_path).await?;
+    if backup.0.as_deref().is_some_and(|data| {
+        serde_json::from_slice::<DataUsageInfo>(data).is_ok_and(|usage| data_usage_info_has_persisted_baseline_identity(&usage))
+    }) {
+        return Ok(backup);
+    }
+    primary
+}
+
 /// Read the bytes used as the baseline for a usage publication while keeping
 /// the v2 primary revision as the CAS fence. During an interrupted upgrade the
 /// primary can be valid JSON without a baseline identity; in that case a
@@ -68,7 +93,7 @@ pub(super) async fn read_data_usage_persist_baseline(
             LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
             format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
         ] {
-            let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
+            let (candidate, _) = read_usage_persist_candidate(storeapi.clone(), &path).await?;
             let Some(candidate) = candidate else {
                 continue;
             };
@@ -107,7 +132,7 @@ pub(super) async fn read_data_usage_persist_baseline(
         LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
         format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
     ] {
-        let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
+        let (candidate, _) = read_usage_persist_candidate(storeapi.clone(), &path).await?;
         let Some(candidate) = candidate else {
             continue;
         };

rustfs/src/admin/handlers/site_replication.rs

@@ -32,7 +32,9 @@ use crate::admin::storage_api::bucket::replication::{
 };
 use crate::admin::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets};
 use crate::admin::storage_api::bucket::utils::{deserialize, serialize};
-use crate::admin::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _};
+use crate::admin::storage_api::bucket::{
+    AdminObjectLockConfigExt as _, AdminReplicationConfigExt as _, AdminVersioningConfigExt as _,
+};
 use crate::admin::storage_api::contract::bucket::{
     BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp,
 };
@@ -61,18 +63,18 @@ use rustfs_madmin::{
     BucketBandwidth, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric, LDAPConfigSettings, LDAPSettings,
     OpenIDProviderSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus, ReplicateEditStatus,
     ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SR_IAM_ITEM_STS_ACC_LEGACY,
-    SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser, SRILMExpiryStatsSummary, SRInfo,
-    SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping, SRPolicyStatsSummary,
-    SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSessionPolicy, SRSiteSummary, SRStateEditReq, SRStateInfo, SRStatusInfo,
-    SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
+    SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser,
+    SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping,
+    SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSessionPolicy, SRSiteSummary, SRStateEditReq,
+    SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
 };
 use rustfs_policy::policy::{
     Policy,
     action::{Action, AdminAction},
 };
 use s3s::dto::{
-    DeleteMarkerReplicationStatus, DeleteReplicationStatus, ExistingObjectReplicationStatus, ReplicaModificationsStatus,
-    ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus,
+    DeleteMarkerReplicationStatus, DeleteReplicationStatus, ExistingObjectReplicationStatus, ObjectLockConfiguration,
+    ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, VersioningConfiguration,
 };
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use serde::Deserialize;
@@ -287,12 +289,21 @@ struct SiteReplicationAddPreflightInfo {
     endpoint: String,
     deployment_id: String,
     enabled: bool,
-    bucket_count: usize,
-    bucket_names: HashSet<String>,
+    buckets: BTreeMap<String, AddPreflightBucketCompat>,
     peer_deployment_ids: BTreeSet<String>,
     idp_settings: serde_json::Value,
 }
 
+/// The per-bucket facts the add preflight compares across sites when more
+/// than one requested site holds data (rustfs/backlog#2070). Only properties
+/// that cannot converge after the add belong here — everything else is
+/// reconciled by the bucket-metadata sync.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct AddPreflightBucketCompat {
+    versioning_enabled: bool,
+    object_lock_enabled: bool,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 struct SRPeerJoinResponse {
     peer: PeerInfo,
@@ -935,19 +946,65 @@ fn idp_settings_value(settings: &IDPSettings) -> S3Result<serde_json::Value> {
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize IDP settings failed: {e}")))
 }
 
+/// The merge-critical facts of one bucket a site reported in its add
+/// preflight metainfo. The configs arrive as the `build_sr_info` wire form
+/// (base64-encoded XML); an undecodable config fails the preflight instead of
+/// defaulting, so corruption cannot admit an unsafe merge.
+fn add_preflight_bucket_compat(endpoint: &str, bucket: &str, info: &SRBucketInfo) -> S3Result<AddPreflightBucketCompat> {
+    let versioning_enabled = info
+        .versioning
+        .as_deref()
+        .map(|raw| {
+            deserialize::<VersioningConfiguration>(&decode_bucket_meta_wire_value(raw))
+                .map(|config| config.enabled())
+                .map_err(|e| {
+                    s3_error!(
+                        InvalidRequest,
+                        "site `{endpoint}` reported an unreadable versioning config for bucket `{bucket}`: {e}"
+                    )
+                })
+        })
+        .transpose()?
+        .unwrap_or(false);
+    let object_lock_enabled = info
+        .object_lock_config
+        .as_deref()
+        .map(|raw| {
+            deserialize::<ObjectLockConfiguration>(&decode_bucket_meta_wire_value(raw))
+                .map(|config| config.enabled())
+                .map_err(|e| {
+                    s3_error!(
+                        InvalidRequest,
+                        "site `{endpoint}` reported an unreadable object-lock config for bucket `{bucket}`: {e}"
+                    )
+                })
+        })
+        .transpose()?
+        .unwrap_or(false);
+    Ok(AddPreflightBucketCompat {
+        versioning_enabled,
+        object_lock_enabled,
+    })
+}
+
 fn add_preflight_info_from_sr_info(
     site: &PeerSite,
     info: SRInfo,
     idp_settings: IDPSettings,
 ) -> S3Result<SiteReplicationAddPreflightInfo> {
-    let bucket_names = info.buckets.keys().cloned().collect();
+    let buckets = info
+        .buckets
+        .iter()
+        .map(|(bucket, bucket_info)| {
+            add_preflight_bucket_compat(&site.endpoint, bucket, bucket_info).map(|compat| (bucket.clone(), compat))
+        })
+        .collect::<S3Result<BTreeMap<_, _>>>()?;
     Ok(SiteReplicationAddPreflightInfo {
         name: if info.name.is_empty() { site.name.clone() } else { info.name },
         endpoint: site.endpoint.clone(),
         deployment_id: info.deployment_id,
         enabled: info.enabled,
-        bucket_count: info.buckets.len(),
-        bucket_names,
+        buckets,
         peer_deployment_ids: info.state.peers.keys().cloned().collect(),
         idp_settings: idp_settings_value(&idp_settings)?,
     })
@@ -1045,7 +1102,7 @@ fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], lo
         if info.deployment_id == local_peer.deployment_id {
             local_seen = true;
         }
-        if info.bucket_count > 0 {
+        if !info.buckets.is_empty() {
             non_empty_sites.push(info.name.clone());
         }
     }
@@ -1070,11 +1127,15 @@ fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], lo
     }
 
     if non_empty_sites.len() > 1 {
-        return Err(s3_error!(
-            InvalidRequest,
-            "site replication can be initialized with data on only one site; non-empty sites: {}",
-            non_empty_sites.join(", ")
-        ));
+        validate_nonempty_add_bucket_compatibility(infos)?;
+        info!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "nonempty_sites_admitted",
+            non_empty_sites = %non_empty_sites.join(", "),
+            "admin site replication state"
+        );
     }
 
     let requested: BTreeSet<String> = infos.iter().map(|info| info.deployment_id.clone()).collect();
@@ -1091,6 +1152,76 @@ fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], lo
     Ok(())
 }
 
+/// Operator recovery guidance for a rejected add between sites that both hold
+/// data — the only supported path is to empty one side and let a resync copy
+/// the objects back (rustfs/backlog#2070).
+const NONEMPTY_ADD_RECOVERY_HINT: &str = "to pair these sites, delete the conflicting bucket (or its data) on all but one \
+     site, re-run `replicate add`, then run `replicate resync` from the surviving site to restore the objects";
+
+/// Admission check for an add in which more than one requested site holds
+/// data — the DR re-pair case: two sites that were unpaired (or never
+/// finished a removal) both keep their buckets, and the historical
+/// unconditional "only one site may hold data" rejection made `replicate
+/// remove` a one-way door (rustfs/backlog#2070).
+///
+/// The add is admitted when every bucket name held by MORE than one requested
+/// site is provably safe to merge through the existing backfill/resync
+/// convergence:
+///
+/// - versioning must be Enabled on every holder: replication into a versioned
+///   bucket lands as another version, so a same-key object from the peer
+///   never destroys the local copy — while on an unversioned holder it would
+///   silently replace the only copy;
+/// - object-lock enablement must match across holders: lock cannot be toggled
+///   after bucket creation, so a mismatch never converges, and replicating
+///   locked objects into a lock-less bucket would strip their WORM guarantee.
+///
+/// A bucket held by a single site carries no merge risk — the post-add
+/// backfill creates it on the peers exactly as the historical
+/// one-non-empty-site path always has.
+fn validate_nonempty_add_bucket_compatibility(infos: &[SiteReplicationAddPreflightInfo]) -> S3Result<()> {
+    let mut holders: BTreeMap<&str, Vec<(&SiteReplicationAddPreflightInfo, AddPreflightBucketCompat)>> = BTreeMap::new();
+    for info in infos {
+        for (bucket, compat) in &info.buckets {
+            holders.entry(bucket.as_str()).or_default().push((info, *compat));
+        }
+    }
+
+    for (bucket, holders) in holders {
+        let [(first, first_compat), rest @ ..] = holders.as_slice() else {
+            continue;
+        };
+        if rest.is_empty() {
+            continue;
+        }
+        if let Some((conflicting, _)) = rest
+            .iter()
+            .find(|(_, compat)| compat.object_lock_enabled != first_compat.object_lock_enabled)
+        {
+            let (enabled_on, disabled_on) = if first_compat.object_lock_enabled {
+                (&first.name, &conflicting.name)
+            } else {
+                (&conflicting.name, &first.name)
+            };
+            return Err(s3_error!(
+                InvalidRequest,
+                "bucket `{bucket}` has object lock enabled on site `{enabled_on}` but not on site `{disabled_on}`, and \
+                 object lock cannot be changed after bucket creation; {NONEMPTY_ADD_RECOVERY_HINT}"
+            ));
+        }
+        if let Some((unversioned, _)) = holders.iter().find(|(_, compat)| !compat.versioning_enabled) {
+            return Err(s3_error!(
+                InvalidRequest,
+                "bucket `{bucket}` exists on more than one site but does not have versioning enabled on site `{}`, so \
+                 merging could silently overwrite objects; enable versioning on every site holding it, or {NONEMPTY_ADD_RECOVERY_HINT}",
+                unversioned.name
+            ));
+        }
+    }
+
+    Ok(())
+}
+
 fn site_replication_bootstrap_token(uri: &Uri) -> Option<String> {
     query_pairs(uri).get("bootstrapToken").cloned()
 }
@@ -5701,6 +5832,53 @@ fn sts_replication_compatibility_policy<'a>(claims: &HashMap<String, Value>, par
     (!claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) && !parent_policy_mapping.is_empty()).then_some(parent_policy_mapping)
 }
 
+/// Adopt only the fields a committed add computed onto the freshly loaded
+/// transaction state. Everything else is owned by writers that commit without
+/// touching `updated_at` (retry events, peer-edit generations, resync
+/// progress, the acks/clears of an already pending rotation), so the add's
+/// `updated_at` CAS cannot vouch for them — they keep the freshly loaded
+/// value, except `pending_remove`:
+///
+/// A committed add supersedes a half-finished removal THIS site started,
+/// exactly as an accepted join does on the receiving side (`apply_peer_join`,
+/// rustfs/rustfs#5963): the adopted topology IS the new membership, while the
+/// pending record only exists to keep notifying peers about the old one. Left
+/// in place, the reconcile tick would replay the stale `SRRemoveReq` against
+/// a freshly re-paired peer — `SRPeerRemoveHandler` applies it
+/// unconditionally — and dismantle the pairing this add just created
+/// (rustfs/backlog#2070). A removal that started AFTER the add's preflight
+/// snapshot moved `updated_at`, so the CAS refuses the commit before this
+/// runs.
+///
+/// The exhaustive destructure makes adding a state field a compile error here
+/// until it is classified.
+fn adopt_add_commit_state(state: &mut SiteReplicationState, next_state: SiteReplicationState) {
+    let SiteReplicationState {
+        name,
+        service_account_access_key,
+        service_account_secret_key: _,
+        service_account_parent,
+        peers,
+        updated_at,
+        resync_status: _,
+        pending_rotation: _,
+        pending_remove: _,
+        pending_endpoint_refresh: _,
+        retry_queue: _,
+        iam_deletion_replays: _,
+        sync_state_initialized,
+        edit_generation: _,
+        applied_edit_generations: _,
+    } = next_state;
+    state.name = name;
+    state.service_account_access_key = service_account_access_key;
+    state.service_account_parent = service_account_parent;
+    state.peers = peers;
+    state.updated_at = updated_at;
+    state.sync_state_initialized = sync_state_initialized;
+    state.pending_remove = None;
+}
+
 pub struct SiteReplicationAddHandler {}
 
 /// MinIO's `SRPeerJoin` replies with an empty body on success; synthesize the
@@ -5756,7 +5934,7 @@ impl Operation for SiteReplicationAddHandler {
         let bootstrap_buckets = preflight_infos
             .iter()
             .filter(|info| !same_identity_endpoint(&info.endpoint, &local_peer.endpoint))
-            .flat_map(|info| info.bucket_names.iter().cloned())
+            .flat_map(|info| info.buckets.keys().cloned())
             .collect();
         let add_in_progress_guard = SiteReplicationAddInProgressGuard::start(lifecycle_guard, bootstrap_buckets)?;
         let mut state = merge_add_sites(
@@ -5853,36 +6031,7 @@ impl Operation for SiteReplicationAddHandler {
                     "site replication state changed during peer join; the peers may already be joined — re-run replicate add"
                 ));
             }
-            // Adopt only the fields this add computed. Everything else is
-            // owned by writers that commit without touching `updated_at`
-            // (retry events, peer-edit generations, resync progress, the
-            // acks/clears of an already pending rotation or removal), so the
-            // CAS above cannot vouch for them — they keep the freshly loaded
-            // value. The exhaustive destructure makes adding a state field a
-            // compile error here until it is classified.
-            let SiteReplicationState {
-                name,
-                service_account_access_key,
-                service_account_secret_key: _,
-                service_account_parent,
-                peers,
-                updated_at,
-                resync_status: _,
-                pending_rotation: _,
-                pending_remove: _,
-                pending_endpoint_refresh: _,
-                retry_queue: _,
-                iam_deletion_replays: _,
-                sync_state_initialized,
-                edit_generation: _,
-                applied_edit_generations: _,
-            } = next_state;
-            state.name = name;
-            state.service_account_access_key = service_account_access_key;
-            state.service_account_parent = service_account_parent;
-            state.peers = peers;
-            state.updated_at = updated_at;
-            state.sync_state_initialized = sync_state_initialized;
+            adopt_add_commit_state(state, next_state);
             let edit_generation = next_peer_edit_generation(state);
             Ok((state.clone(), edit_generation))
         })
@@ -9421,18 +9570,29 @@ mod tests {
     }
 
     fn preflight_site(name: &str, endpoint: &str, deployment_id: &str, bucket_count: usize) -> SiteReplicationAddPreflightInfo {
+        // Site-prefixed names keep the generated buckets disjoint across
+        // sites; tests exercising shared-bucket merges insert their own.
+        let buckets = (0..bucket_count)
+            .map(|i| (format!("{name}-bucket-{i}"), versioned_bucket()))
+            .collect();
         SiteReplicationAddPreflightInfo {
             name: name.to_string(),
             endpoint: endpoint.to_string(),
             deployment_id: deployment_id.to_string(),
             enabled: false,
-            bucket_count,
-            bucket_names: HashSet::new(),
+            buckets,
             peer_deployment_ids: BTreeSet::new(),
             idp_settings: serde_json::json!({"provider": "same"}),
         }
     }
 
+    fn versioned_bucket() -> AddPreflightBucketCompat {
+        AddPreflightBucketCompat {
+            versioning_enabled: true,
+            object_lock_enabled: false,
+        }
+    }
+
     #[test]
     fn test_validate_add_preflight_topology_accepts_matching_sites() {
         let local_peer = PeerInfo {
@@ -9491,20 +9651,120 @@ mod tests {
         assert!(err.to_string().contains("IDP settings mismatch"));
     }
 
+    // rustfs/backlog#2070: two sites that both hold data (the DR re-pair
+    // case) must be admitted when their bucket sets are merge-safe, instead
+    // of the historical unconditional "only one site may hold data" rejection
+    // that made `replicate remove` a one-way door.
     #[test]
-    fn test_validate_add_preflight_topology_rejects_multiple_non_empty_sites() {
+    fn test_validate_add_preflight_topology_accepts_compatible_non_empty_sites() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 1);
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 1);
+        // The same bucket on both sites, versioning enabled on both: the
+        // exact shape a formerly paired cluster is left in after a remove.
+        local.buckets.insert("shared".to_string(), versioned_bucket());
+        remote.buckets.insert("shared".to_string(), versioned_bucket());
+        let infos = vec![local, remote];
+
+        validate_add_preflight_topology(&infos, &local_peer).expect("compatible non-empty sites should be admitted");
+    }
+
+    #[test]
+    fn test_validate_add_preflight_topology_accepts_disjoint_non_empty_sites() {
         let local_peer = PeerInfo {
             deployment_id: "local-dep".to_string(),
             ..peer("local", "https://local.example.com")
         };
         let infos = vec![
-            preflight_site("local", "https://local.example.com", "local-dep", 1),
-            preflight_site("remote", "https://remote.example.com", "remote-dep", 1),
+            preflight_site("local", "https://local.example.com", "local-dep", 2),
+            preflight_site("remote", "https://remote.example.com", "remote-dep", 2),
         ];
 
-        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("multiple non-empty sites should fail");
+        validate_add_preflight_topology(&infos, &local_peer).expect("disjoint non-empty sites should be admitted");
+    }
 
-        assert!(err.to_string().contains("only one site"));
+    #[test]
+    fn test_validate_add_preflight_topology_rejects_shared_bucket_object_lock_mismatch() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 0);
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 0);
+        local.buckets.insert(
+            "shared".to_string(),
+            AddPreflightBucketCompat {
+                versioning_enabled: true,
+                object_lock_enabled: true,
+            },
+        );
+        remote.buckets.insert("shared".to_string(), versioned_bucket());
+        let infos = vec![local, remote];
+
+        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("object-lock mismatch should fail");
+
+        let message = err.to_string();
+        assert!(
+            message.contains("bucket `shared` has object lock enabled on site `local`"),
+            "got: {message}"
+        );
+        // The rejection must carry the operator recovery steps, not a bare no.
+        assert!(message.contains("re-run `replicate add`"), "got: {message}");
+        assert!(message.contains("`replicate resync`"), "got: {message}");
+    }
+
+    #[test]
+    fn test_validate_add_preflight_topology_rejects_shared_unversioned_bucket() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 0);
+        let mut remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 0);
+        local.buckets.insert("shared".to_string(), versioned_bucket());
+        remote.buckets.insert(
+            "shared".to_string(),
+            AddPreflightBucketCompat {
+                versioning_enabled: false,
+                object_lock_enabled: false,
+            },
+        );
+        let infos = vec![local, remote];
+
+        let err = validate_add_preflight_topology(&infos, &local_peer).expect_err("shared unversioned bucket should fail");
+
+        let message = err.to_string();
+        assert!(
+            message.contains("bucket `shared`") && message.contains("versioning enabled on site `remote`"),
+            "got: {message}"
+        );
+        assert!(message.contains("re-run `replicate add`"), "got: {message}");
+    }
+
+    // A bucket held by a single site never blocks the add, whatever its
+    // configs: the backfill creates it on the peers exactly like the
+    // historical one-non-empty-site path.
+    #[test]
+    fn test_validate_add_preflight_topology_ignores_unshared_bucket_configs() {
+        let local_peer = PeerInfo {
+            deployment_id: "local-dep".to_string(),
+            ..peer("local", "https://local.example.com")
+        };
+        let mut local = preflight_site("local", "https://local.example.com", "local-dep", 1);
+        let remote = preflight_site("remote", "https://remote.example.com", "remote-dep", 1);
+        local.buckets.insert(
+            "local-only".to_string(),
+            AddPreflightBucketCompat {
+                versioning_enabled: false,
+                object_lock_enabled: true,
+            },
+        );
+        let infos = vec![local, remote];
+
+        validate_add_preflight_topology(&infos, &local_peer).expect("unshared buckets should not block the add");
     }
 
     #[test]
@@ -9524,6 +9784,91 @@ mod tests {
         assert!(err.to_string().contains("different site replication peer set"));
     }
 
+    // add_preflight_bucket_compat reads the build_sr_info wire form:
+    // base64-encoded XML for both the versioning and the object-lock config.
+    #[test]
+    fn test_add_preflight_bucket_compat_parses_wire_configs() {
+        let info = SRBucketInfo {
+            versioning: Some(
+                BASE64_STANDARD.encode_to_string(b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>"),
+            ),
+            object_lock_config: Some(BASE64_STANDARD.encode_to_string(
+                b"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>",
+            )),
+            ..Default::default()
+        };
+
+        let compat = add_preflight_bucket_compat("https://a.example.com", "b", &info).expect("wire configs should parse");
+
+        assert!(compat.versioning_enabled);
+        assert!(compat.object_lock_enabled);
+    }
+
+    #[test]
+    fn test_add_preflight_bucket_compat_absent_and_suspended_configs_are_disabled() {
+        let absent = add_preflight_bucket_compat("https://a.example.com", "b", &SRBucketInfo::default())
+            .expect("absent configs should parse");
+        assert!(!absent.versioning_enabled);
+        assert!(!absent.object_lock_enabled);
+
+        let suspended = SRBucketInfo {
+            versioning: Some(
+                BASE64_STANDARD
+                    .encode_to_string(b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>"),
+            ),
+            ..Default::default()
+        };
+        let compat =
+            add_preflight_bucket_compat("https://a.example.com", "b", &suspended).expect("suspended config should parse");
+        assert!(!compat.versioning_enabled, "suspended versioning is not merge-safe");
+    }
+
+    // rustfs/backlog#2070: a committed add must supersede this site's own
+    // half-finished removal (mirroring the join side, rustfs/rustfs#5963) —
+    // otherwise the reconcile tick replays the stale removal against the
+    // freshly re-paired peer and dismantles the new pairing.
+    #[test]
+    fn test_adopt_add_commit_state_clears_pending_remove() {
+        let mut state = SiteReplicationState {
+            pending_remove: Some(PendingRemove {
+                id: "remove-1".to_string(),
+                ..Default::default()
+            }),
+            edit_generation: 7,
+            ..Default::default()
+        };
+        let next_state = SiteReplicationState {
+            name: "local".to_string(),
+            peers: BTreeMap::from([
+                ("local-dep".to_string(), peer("local", "https://local.example.com")),
+                ("remote-dep".to_string(), peer("remote", "https://remote.example.com")),
+            ]),
+            updated_at: Some(OffsetDateTime::now_utc()),
+            sync_state_initialized: true,
+            ..Default::default()
+        };
+
+        adopt_add_commit_state(&mut state, next_state);
+
+        assert!(state.pending_remove.is_none(), "the committed add supersedes the removal");
+        assert_eq!(state.peers.len(), 2, "the add's topology is adopted");
+        assert_eq!(state.edit_generation, 7, "commit-owned fields keep the loaded value");
+    }
+
+    // Fail closed: a config this site cannot read must fail the preflight
+    // instead of defaulting into an unsafe admission.
+    #[test]
+    fn test_add_preflight_bucket_compat_rejects_undecodable_config() {
+        let info = SRBucketInfo {
+            versioning: Some(BASE64_STANDARD.encode_to_string(b"<VersioningConfiguration")),
+            ..Default::default()
+        };
+
+        let err = add_preflight_bucket_compat("https://a.example.com", "b", &info).expect_err("broken XML should fail");
+
+        assert!(err.to_string().contains("unreadable versioning config"));
+    }
+
     /// P1-15 review follow-up: the receiving side of the ordering fence. Two
     /// nodes of the sending site can fan out in the opposite order to their
     /// commits; the receiver decides ordering from the generation the sender

rustfs/src/admin/storage_api.rs

@@ -20,8 +20,8 @@ use time::OffsetDateTime;
 
 mod ecstore_bucket {
     pub(crate) use crate::storage::storage_api::ecstore_bucket::{
-        bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, quota, replication, target, utils,
-        versioning, versioning_sys,
+        bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, object_lock, quota, replication, target,
+        utils, versioning, versioning_sys,
     };
 }
 
@@ -185,6 +185,16 @@ impl AdminVersioningConfigExt for s3s::dto::VersioningConfiguration {
     }
 }
 
+pub(crate) trait AdminObjectLockConfigExt {
+    fn enabled(&self) -> bool;
+}
+
+impl AdminObjectLockConfigExt for s3s::dto::ObjectLockConfiguration {
+    fn enabled(&self) -> bool {
+        <s3s::dto::ObjectLockConfiguration as ecstore_bucket::object_lock::ObjectLockApi>::enabled(self)
+    }
+}
+
 pub(crate) mod bandwidth {
     pub(crate) mod monitor {
         pub(crate) type BandwidthDetails = super::super::ecstore_bucket::bandwidth::monitor::BandwidthDetails;
@@ -863,7 +873,9 @@ pub(crate) mod bucket {
     pub(crate) use super::replication;
     pub(crate) use super::target;
     pub(crate) use super::versioning_sys;
-    pub(crate) use super::{AdminReplicationConfigExt, AdminVersioningConfigExt, is_reserved_or_invalid_bucket};
+    pub(crate) use super::{
+        AdminObjectLockConfigExt, AdminReplicationConfigExt, AdminVersioningConfigExt, is_reserved_or_invalid_bucket,
+    };
 
     pub(crate) mod utils {
         pub(crate) use super::super::ecstore_utils::{deserialize, is_valid_object_prefix, serialize};

rustfs/src/app/object/put.rs

@@ -308,14 +308,14 @@ pub(crate) fn guard_put_object_body_read_timeout(
     })
 }
 
-struct PooledBufferReader {
+pub(super) struct PooledBufferReader {
     buffer: PooledBuffer,
     len: usize,
     pos: usize,
 }
 
 impl PooledBufferReader {
-    fn new(buffer: PooledBuffer, len: usize) -> Self {
+    pub(super) fn new(buffer: PooledBuffer, len: usize) -> Self {
         Self { buffer, len, pos: 0 }
     }
 }
@@ -631,7 +631,7 @@ fn select_put_path_with_concurrency(
 /// where the allocation cost is negligible (≤4KiB memcpy).
 const POOL_BYPASS_MAX_SIZE: usize = 4 * 1024;
 
-async fn read_small_put_body_into<R, B>(body: &mut R, buf: &mut B, size: usize) -> S3Result<()>
+pub(super) async fn read_small_put_body_into<R, B>(body: &mut R, buf: &mut B, size: usize) -> S3Result<()>
 where
     R: AsyncRead + Unpin,
     B: bytes::BufMut,
@@ -1084,8 +1084,6 @@ impl DefaultObjectUsecase {
         let ingress_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let should_compress =
             is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
-        let server_side_encryption_requested =
-            server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some();
 
         // Resolve the store through the request-bound server context
         // (backlog#1052 S6), not the process-global handle, so an embedded
@@ -1133,38 +1131,6 @@ impl DefaultObjectUsecase {
             base_buffer_size
         };
 
-        // Detect zero-copy opportunity before encryption/compression decisions
-        // Zero-copy is beneficial for large unencrypted, uncompressed objects
-        let enable_zero_copy = should_use_zero_copy(size, &req.headers);
-
-        if enable_zero_copy {
-            // Record zero-copy write attempt
-            counter!("rustfs_zero_copy_write_attempts_total").increment(1);
-            histogram!("rustfs_zero_copy_write_size_bytes").record(size as f64);
-            debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key);
-        }
-
-        let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) =
-            select_put_path_with_concurrency(
-                size,
-                &req.headers,
-                server_side_encryption_requested,
-                should_compress,
-                false,
-                concurrent_put_requests,
-            );
-        if use_zero_copy_eager_put_path {
-            counter!(buffered_write::ATTEMPTS_TOTAL).increment(1);
-            histogram!(buffered_write::ATTEMPT_SIZE_BYTES).record(size as f64);
-        }
-        rustfs_io_metrics::record_put_object_diagnostics(
-            put_path,
-            zero_copy_eager_put_path_status,
-            size,
-            buffer_size,
-            use_large_put_concurrency_tuning,
-        );
-
         let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
         rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start);
@@ -1203,6 +1169,32 @@ impl DefaultObjectUsecase {
             effective_kms_key_id = None;
         }
 
+        let server_side_encryption_requested =
+            effective_sse.is_some() || sse_customer_algorithm.is_some() || effective_kms_key_id.is_some();
+        let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) =
+            select_put_path_with_concurrency(
+                size,
+                &req.headers,
+                server_side_encryption_requested,
+                should_compress,
+                false,
+                concurrent_put_requests,
+            );
+        if use_zero_copy_eager_put_path {
+            counter!("rustfs_zero_copy_write_attempts_total").increment(1);
+            histogram!("rustfs_zero_copy_write_size_bytes").record(size as f64);
+            debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key);
+            counter!(buffered_write::ATTEMPTS_TOTAL).increment(1);
+            histogram!(buffered_write::ATTEMPT_SIZE_BYTES).record(size as f64);
+        }
+        rustfs_io_metrics::record_put_object_diagnostics(
+            put_path,
+            zero_copy_eager_put_path_status,
+            size,
+            buffer_size,
+            use_large_put_concurrency_tuning,
+        );
+
         // Validate SSE-C headers early: reject partial/invalid combinations per S3 spec
         validate_sse_headers_for_write(
             effective_sse.as_ref(),
@@ -1739,7 +1731,7 @@ impl DefaultObjectUsecase {
             rustfs_io_metrics::record_put_object(
                 duration_ms,
                 size,
-                enable_zero_copy, // Track if zero-copy was enabled
+                use_zero_copy_eager_put_path, // Track if zero-copy was enabled
             );
         }
 
@@ -1776,7 +1768,10 @@ mod tests {
     use super::*;
     use futures::StreamExt;
     use http::{HeaderMap, HeaderName, HeaderValue, Method};
-    use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule};
+    use s3s::dto::{
+        DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule, ServerSideEncryptionByDefault,
+        ServerSideEncryptionConfiguration, ServerSideEncryptionRule,
+    };
     use std::pin::Pin;
     use std::sync::Arc;
     use std::task::{Context, Poll};
@@ -2392,6 +2387,36 @@ mod tests {
         assert!(!use_small_eager);
     }
 
+    #[test]
+    fn select_put_path_treats_bucket_default_sse_as_encrypted() {
+        for (algorithm, kms_key_id) in [
+            (ServerSideEncryption::AES256, None),
+            (ServerSideEncryption::AWS_KMS, Some("bucket-key")),
+        ] {
+            let config = ServerSideEncryptionConfiguration {
+                rules: vec![ServerSideEncryptionRule {
+                    apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                        sse_algorithm: ServerSideEncryption::from_static(algorithm),
+                        kms_master_key_id: kms_key_id.map(|id| SSEKMSKeyId::from(id.to_string())),
+                    }),
+                    bucket_key_enabled: None,
+                }],
+            };
+            let (effective_sse, effective_kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);
+            let encryption_requested = effective_sse.is_some() || effective_kms_key_id.is_some();
+
+            for size in [512 * 1024, 2 * 1024 * 1024] {
+                let (path, status, use_zero_copy, use_small_eager) =
+                    select_put_path_with_concurrency(size, &HeaderMap::new(), encryption_requested, false, false, 256);
+
+                assert_eq!(path, "streaming");
+                assert_eq!(status, PUT_EAGER_STATUS_ENCRYPTED);
+                assert!(!use_zero_copy);
+                assert!(!use_small_eager);
+            }
+        }
+    }
+
     #[test]
     fn should_use_small_eager_put_path_allows_a_b_override_at_1mb() {
         let headers = HeaderMap::new();

rustfs/src/app/storage_api.rs

@@ -978,9 +978,12 @@ pub(crate) mod bucket {
 }
 
 pub(crate) mod concurrency {
+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::concurrency_consumer::SNOWBALL_MEMBER_COMMIT_LIMIT;
     pub(crate) use crate::storage::storage_api::concurrency_consumer::{
         ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission, GetObjectGuard, IoQueueStatus, IoStrategy,
-        PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
+        PutObjectGuard, SNOWBALL_STAGING_BYTES_LIMIT, get_concurrency_aware_buffer_size, get_concurrency_manager,
+        get_put_concurrency_aware_buffer_size,
     };
 }
 

rustfs/src/storage/concurrency/manager.rs

@@ -31,10 +31,12 @@ use rustfs_io_metrics::bandwidth::{BandwidthMonitor, BandwidthSnapshot};
 use rustfs_io_metrics::{MetricsCollector, PerformanceMetrics};
 use std::sync::{Arc, LazyLock, Mutex};
 use std::time::Duration;
-use tokio::sync::Semaphore;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::debug;
 
 const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32;
+pub(crate) const SNOWBALL_MEMBER_COMMIT_LIMIT: usize = 32;
+pub(crate) const SNOWBALL_STAGING_BYTES_LIMIT: usize = 4 * MI_B;
 
 /// Global concurrency manager instance
 pub(crate) static CONCURRENCY_MANAGER: LazyLock<ConcurrencyManager> = LazyLock::new(ConcurrencyManager::new);
@@ -69,6 +71,12 @@ pub struct ConcurrencyManager {
     metrics_collector: Arc<MetricsCollector>,
     /// Foreground write admission policy, resolved once at startup.
     foreground_write_admission_policy: ForegroundWriteAdmissionPolicy,
+    /// Snowball members are internal PUTs, so they use a separate global gate
+    /// from preparation through the independently owned post-commit tail.
+    snowball_member_commit_semaphore: Arc<Semaphore>,
+    /// Bounds the owned member bodies and metadata retained between TAR parsing
+    /// and storage commit across all extract requests.
+    snowball_staging_bytes_semaphore: Arc<Semaphore>,
 }
 
 impl std::fmt::Debug for ConcurrencyManager {
@@ -417,6 +425,8 @@ impl ConcurrencyManager {
             bandwidth_monitor,
             metrics_collector,
             foreground_write_admission_policy,
+            snowball_member_commit_semaphore: Arc::new(Semaphore::new(SNOWBALL_MEMBER_COMMIT_LIMIT)),
+            snowball_staging_bytes_semaphore: Arc::new(Semaphore::new(SNOWBALL_STAGING_BYTES_LIMIT)),
         }
     }
 
@@ -556,6 +566,40 @@ impl ConcurrencyManager {
             .await
     }
 
+    /// Admit a Snowball member through the foreground PUT policy using the
+    /// member's logical size. The outer archive has a separate preflight, while
+    /// every member shares the ordinary PUT gate and its wait/rejection policy.
+    pub(crate) async fn admit_snowball_foreground_write(
+        &self,
+        member_size: i64,
+    ) -> Result<ForegroundWriteAdmission, tokio::sync::AcquireError> {
+        self.foreground_write_admission_policy
+            .admit(ForegroundWriteAdmissionKind::PutObject, member_size)
+            .await
+    }
+
+    /// Acquire one global Snowball member lifecycle slot.
+    pub(crate) async fn acquire_snowball_member_commit(&self) -> Result<OwnedSemaphorePermit, tokio::sync::AcquireError> {
+        self.snowball_member_commit_semaphore.clone().acquire_owned().await
+    }
+
+    /// Try to acquire one global Snowball member lifecycle slot.
+    pub(crate) fn try_acquire_snowball_member_commit(&self) -> Option<OwnedSemaphorePermit> {
+        self.snowball_member_commit_semaphore.clone().try_acquire_owned().ok()
+    }
+
+    /// Try to reserve prepared-member bytes without waiting.
+    ///
+    /// A producer holding a non-empty micro-batch must use this method and
+    /// flush before waiting, otherwise several archives can each retain part of
+    /// the global budget while waiting forever for the remainder.
+    pub(crate) fn try_acquire_snowball_staging_bytes(&self, bytes: u32) -> Option<OwnedSemaphorePermit> {
+        self.snowball_staging_bytes_semaphore
+            .clone()
+            .try_acquire_many_owned(bytes)
+            .ok()
+    }
+
     /// Admit a multipart UploadPart request under the configured write gate.
     ///
     /// Multipart workloads can saturate memory and internode write streams with
@@ -1050,13 +1094,138 @@ impl Default for ConcurrencyManager {
 mod integration_tests {
     use super::super::io_schedule::{IoLoadLevel, IoPriority};
     use super::super::request_guard::GetObjectGuard;
-    use super::{ConcurrencyManager, ForegroundWriteAdmission, derive_large_put_admission_limit};
+    use super::{
+        ConcurrencyManager, ForegroundWriteAdmission, SNOWBALL_MEMBER_COMMIT_LIMIT, SNOWBALL_STAGING_BYTES_LIMIT,
+        derive_large_put_admission_limit,
+    };
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
     use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
     use serial_test::serial;
     use std::time::Duration;
 
+    #[test]
+    fn test_snowball_gates_are_global_bounded_and_reusable() {
+        let manager = ConcurrencyManager::new();
+        let clone = manager.clone();
+
+        let commit_permits = manager
+            .snowball_member_commit_semaphore
+            .clone()
+            .try_acquire_many_owned(u32::try_from(SNOWBALL_MEMBER_COMMIT_LIMIT).expect("Snowball commit limit must fit into u32"))
+            .expect("the exact Snowball commit limit must be available");
+        assert!(
+            clone.snowball_member_commit_semaphore.clone().try_acquire_owned().is_err(),
+            "a cloned manager must share the global commit gate"
+        );
+        drop(commit_permits);
+        assert!(clone.snowball_member_commit_semaphore.clone().try_acquire_owned().is_ok());
+
+        let staging_bytes = u32::try_from(SNOWBALL_STAGING_BYTES_LIMIT).expect("Snowball staging limit must fit into u32");
+        let staging_permit = manager
+            .try_acquire_snowball_staging_bytes(staging_bytes)
+            .expect("the exact Snowball staging budget must be available");
+        assert!(
+            clone.try_acquire_snowball_staging_bytes(1).is_none(),
+            "a cloned manager must share the global staging budget"
+        );
+        drop(staging_permit);
+        assert!(clone.try_acquire_snowball_staging_bytes(1).is_some());
+    }
+
+    #[tokio::test]
+    async fn test_snowball_members_share_the_strict_foreground_put_gate() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(true, 2, Duration::ZERO);
+        let outer = match manager
+            .admit_put_object(1)
+            .await
+            .expect("strict outer admission must remain open")
+        {
+            ForegroundWriteAdmission::Admitted(permit) => permit,
+            outcome => panic!("strict outer admission must return a permit: {outcome:?}"),
+        };
+        let member = match manager
+            .admit_snowball_foreground_write(1)
+            .await
+            .expect("strict member admission must remain open")
+        {
+            ForegroundWriteAdmission::Admitted(permit) => permit,
+            outcome => panic!("strict member admission must return a permit: {outcome:?}"),
+        };
+        assert!(
+            matches!(
+                manager
+                    .admit_snowball_foreground_write(1)
+                    .await
+                    .expect("strict member admission must remain open"),
+                ForegroundWriteAdmission::Rejected
+            ),
+            "a saturated Snowball member admission must preserve the zero-wait rejection policy"
+        );
+        assert!(
+            matches!(
+                manager.admit_put_object(1).await.expect("strict gate must remain usable"),
+                ForegroundWriteAdmission::Rejected
+            ),
+            "outer PUTs and Snowball members must exhaust the same strict gate"
+        );
+
+        drop(outer);
+        let replacement = match manager
+            .admit_snowball_foreground_write(1)
+            .await
+            .expect("released strict capacity must be reusable")
+        {
+            ForegroundWriteAdmission::Admitted(permit) => permit,
+            outcome => panic!("strict replacement admission must return a permit: {outcome:?}"),
+        };
+        drop((member, replacement));
+    }
+
+    #[tokio::test]
+    async fn test_snowball_members_use_their_size_for_the_large_foreground_put_gate() {
+        let min_size = 16 * 1024 * 1024;
+        let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, min_size, Duration::ZERO);
+
+        assert!(matches!(
+            manager
+                .admit_put_object((min_size - 1) as i64)
+                .await
+                .expect("small outer archive admission must remain open"),
+            ForegroundWriteAdmission::Disabled
+        ));
+        let large_member = match manager
+            .admit_snowball_foreground_write(min_size as i64)
+            .await
+            .expect("large Snowball member admission must remain open")
+        {
+            ForegroundWriteAdmission::Admitted(permit) => permit,
+            outcome => panic!("large Snowball member must consume the large PUT gate: {outcome:?}"),
+        };
+        assert!(matches!(
+            manager
+                .admit_snowball_foreground_write(min_size as i64)
+                .await
+                .expect("saturated Snowball member admission must remain open"),
+            ForegroundWriteAdmission::Rejected
+        ));
+        assert!(matches!(
+            manager
+                .admit_put_object(min_size as i64)
+                .await
+                .expect("ordinary large PUT admission must remain open"),
+            ForegroundWriteAdmission::Rejected
+        ));
+        assert!(matches!(
+            manager
+                .admit_snowball_foreground_write((min_size - 1) as i64)
+                .await
+                .expect("small Snowball member admission must remain open"),
+            ForegroundWriteAdmission::Disabled
+        ));
+        drop(large_member);
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_concurrency_manager_priority_queue_integration() {

rustfs/src/storage/concurrency/mod.rs

@@ -51,6 +51,9 @@ pub use io_schedule::{
 pub use request_guard::{GetObjectGuard, PutObjectGuard};
 
 // Concurrency manager
+#[cfg(test)]
+pub(crate) use manager::SNOWBALL_MEMBER_COMMIT_LIMIT;
+pub(crate) use manager::SNOWBALL_STAGING_BYTES_LIMIT;
 pub use manager::{ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission};
 
 // ============================================

rustfs/src/storage/storage_api.rs

@@ -125,9 +125,12 @@ pub(crate) mod access_consumer {
 }
 
 pub(crate) mod concurrency_consumer {
+    #[cfg(test)]
+    pub(crate) use super::super::concurrency::SNOWBALL_MEMBER_COMMIT_LIMIT;
     pub(crate) use super::super::concurrency::{
         ConcurrencyManager, DiskReadAdmission, ForegroundWriteAdmission, GetObjectGuard, IoQueueStatus, IoStrategy,
-        PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
+        PutObjectGuard, SNOWBALL_STAGING_BYTES_LIMIT, get_concurrency_aware_buffer_size, get_concurrency_manager,
+        get_put_concurrency_aware_buffer_size,
     };
 }
 

rustfs/src/table_catalog/tests.rs

@@ -16445,7 +16445,7 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
         "if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await",
         "validate_table_catalog_object_mutation(&bucket, &object).await?;",
         "validate_object_key(&key, \"PUT\")?;\n        validate_table_catalog_object_mutation(&bucket, &key).await?;",
-        "validate_table_catalog_object_mutation(&bucket, &fpath).await?;",
+        "extract_try!(validate_table_catalog_object_mutation(&bucket, &fpath).await);",
     ] {
         assert!(source.contains(expected), "missing object mutation guard: {expected}");
     }