test(e2e): isolate mc mirror fixture in full suite

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/store/multipart.rs

@@ -19,6 +19,7 @@ use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::multipart::MultipartOperations as _;
 use crate::storage_api_contracts::object::ObjectOperations as _;
 use futures::{StreamExt, stream};
+use rustfs_utils::path::encode_dir_object_ref;
 use std::collections::HashSet;
 
 const MULTIPART_LIST_SET_CONCURRENCY: usize = 4;
@@ -1168,11 +1169,9 @@ impl ECStore {
         let oi = if require_all_pools {
             lookup_opts.metadata_cache_safe = false;
             lookup_opts.suppress_read_repair = true;
-            if let Err(err) = check_object_args(bucket, object) {
-                return Some(err);
-            }
-            let encoded_object = encode_dir_object(object);
-            self.get_latest_object_info_for_conditional_write(bucket, &encoded_object, &lookup_opts)
+            // Only completion uses this branch, after validating bucket and object.
+            let encoded_object = encode_dir_object_ref(object);
+            self.get_latest_object_info_for_conditional_write(bucket, encoded_object.as_ref(), &lookup_opts)
                 .await
                 .and_then(|(info, _)| {
                     lookup_opts.precondition_check(&info)?;

crates/scanner/src/scanner.rs

@@ -1062,31 +1062,6 @@ fn prepare_cycle_for_usage_floor_bootstrap(
     }
 }
 
-/// Overwrite an unreadable usage floor with a bootstrap marker using the
-/// file's current revision for CAS. A successful publish leaves a
-/// `BootstrapPending` floor that the normal load path picks up on retry.
-/// Returns `Err` when the floor is not even readable (transient I/O) so the
-/// caller keeps the supervisor-retry behavior instead of rebuilding.
-async fn rebuild_unreadable_usage_floor(
-    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
-) -> Result<(), ScannerError> {
-    let Some(expected_epoch) = scanner_publication_epoch(storeapi.clone()).await else {
-        return Err(ScannerError::Other("scanner usage floor rebuild is blocked by data movement".to_string()));
-    };
-    let (_data, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .map_err(|err| ScannerError::Other(format!("scanner usage floor revision unreadable for rebuild: {err}")))?;
-    publish_scanner_usage_bootstrap_primary(
-        storeapi,
-        &revision,
-        expected_epoch,
-        None,
-        ScannerUsageBootstrapPublishContext::Recovery,
-        || true,
-    )
-    .await
-}
-
 pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
     init_data_scanner_with_storage(ctx, storeapi, ScannerRuntimeGuard::new()).await;
 }
@@ -3115,27 +3090,6 @@ where
             Err(err) => usage_floor_result = Err(err),
         }
     }
-    // backlog#2918: after dual-corruption recovery has been attempted an
-    // unreadable floor is no longer a quota proof. Report the failure, then
-    // rebuild through a CAS overwrite before giving up. Transient I/O errors
-    // fail the rebuild and fall through to the supervisor-retry stop below.
-    if usage_floor_result.is_err() {
-        let error = usage_floor_result.as_ref().unwrap_err().to_string();
-        error!(
-            target: "rustfs::scanner",
-            event = EVENT_SCANNER_PERSIST_STATE,
-            component = LOG_COMPONENT_SCANNER,
-            subsystem = LOG_SUBSYSTEM_RUNTIME,
-            path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
-            state = "usage_floor_load_failed",
-            error = %error,
-            "Scanner usage floor is unreadable after recovery; attempting rebuild"
-        );
-        record_scanner_usage_floor_failure(error);
-        if rebuild_unreadable_usage_floor(storeapi.clone()).await.is_ok() {
-            usage_floor_result = persisted_usage_floor_for_startup(storeapi.clone(), true).await;
-        }
-    }
     let (usage_floor, usage_floor_startup) = match usage_floor_result {
         Ok(result) => result,
         Err(err) => {
@@ -3148,7 +3102,7 @@ where
                 path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
                 state = "usage_floor_load_failed",
                 error = %err,
-                "Scanner usage floor could not be loaded or rebuilt; stopping leader iteration for supervisor retry"
+                "Scanner stopped because the persisted usage floor could not be loaded"
             );
             record_scanner_usage_floor_failure(error.clone());
             global_metrics().set_cycle(None).await;

crates/scanner/src/scanner/tests/gate_self_heal.rs

@@ -41,6 +41,20 @@ fn reset_recovery_status() {
     .expect("test must be able to reset the local recovery observation");
 }
 
+async fn wait_for_recovery_attempt(attempt: u32) -> ScannerCycleRecoveryStatus {
+    tokio::time::timeout(WAIT_BUDGET, async {
+        loop {
+            let status = scanner_cycle_recovery_status();
+            if status.retry_count >= u64::from(attempt) {
+                return status;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("recovery attempt {attempt} was never recorded: {:?}", scanner_cycle_recovery_status()))
+}
+
 async fn wait_for_recovery_state(state: &str) -> ScannerCycleRecoveryStatus {
     tokio::time::timeout(WAIT_BUDGET, async {
         loop {
@@ -116,19 +130,20 @@ fn started_rebalance() -> EcstoreRebalanceMeta {
     }
 }
 
-/// rustfs/rustfs#8670 + backlog#2918: a usage floor that is present but
-/// untrusted must not stop the scanner. The unreadable floor is reported and
-/// rebuilt through a CAS overwrite so the scanner recovers without a process
-/// restart or a terminal `paused, retryable=false` observation.
+/// rustfs/rustfs#8670: a usage floor that is present but untrusted must not
+/// drive the supervisor into a terminal `paused, retryable=false` observation.
+/// Every attempt stays retryable with a bounded backoff, and repairing the
+/// floor recovers the scanner without a process restart.
 #[tokio::test]
 #[serial]
-async fn usage_floor_load_failure_rebuilds_and_recovers_without_restart() {
+async fn usage_floor_load_failure_backs_off_and_recovers_without_restart() {
     temp_env::async_with_vars(SUPERVISOR_ENV, async {
         crate::runtime_config::refresh_scanner_runtime_config_for_tests();
         crate::scanner_io::clear_dirty_usage_buckets_for_tests();
         reset_recovery_status();
         let (_dir, store) = setup_scanner_cycle_store().await;
-        // Present but without a baseline identity: the shape #8670 reported.
+        // Present but without a baseline identity: the shape #8670 reported,
+        // whose error points the operator at the usage-state reset.
         save_config(
             store.clone(),
             DATA_USAGE_OBJ_NAME_PATH.as_str(),
@@ -141,11 +156,41 @@ async fn usage_floor_load_failure_rebuilds_and_recovers_without_restart() {
         let _cancel_on_drop = ctx.clone().drop_guard();
         init_data_scanner_with_storage(ctx.clone(), store.clone(), ScannerRuntimeGuard::new()).await;
 
-        // backlog#2918: the unreadable floor is rebuilt automatically; the
-        // scanner must reach healthy without a manual repair or terminal pause.
+        let attempts = MAX_SCANNER_CYCLE_RECOVERY_RETRIES + 2;
+        let mut first_detected = None;
+        for attempt in 1..=attempts {
+            let status = wait_for_recovery_attempt(attempt).await;
+            assert_eq!(status.classification.as_deref(), Some("usage_floor_load_failed"), "attempt {attempt}");
+            assert!(status.retryable, "attempt {attempt} must stay retryable: {status:?}");
+            assert_ne!(status.state, "paused", "attempt {attempt} must not report a terminal pause");
+            if attempt >= MAX_SCANNER_CYCLE_RECOVERY_RETRIES {
+                assert_eq!(status.state, "backoff", "attempt {attempt}: {status:?}");
+            } else {
+                assert_eq!(status.state, "usage_floor_load_failed", "attempt {attempt}: {status:?}");
+            }
+            assert!(
+                status
+                    .reason
+                    .as_deref()
+                    .is_some_and(|reason| reason.contains("usage-state/reset") && reason.contains("next automatic attempt")),
+                "the operator hint and the next attempt must both be visible: {status:?}"
+            );
+            assert!(status.next_attempt_at_unix_secs.is_some(), "attempt {attempt}: {status:?}");
+            match first_detected {
+                None => first_detected = status.first_detected_at_unix_secs,
+                Some(first) => assert_eq!(status.first_detected_at_unix_secs, Some(first), "attempt {attempt}"),
+            }
+            if attempt == attempts {
+                save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), valid_usage_floor(0, 0))
+                    .await
+                    .expect("repaired usage floor should persist");
+            }
+            notify_scanner_cycle_recovery_wake();
+        }
+
         let healthy = wait_for_recovery_state("healthy").await;
-        assert!(!healthy.retryable, "rebuild must clear the retry budget: {healthy:?}");
-        assert_ne!(healthy.state, "paused", "rebuild must not report a terminal pause");
+        assert_eq!(healthy.retry_count, 0, "a successful attempt clears the retry budget: {healthy:?}");
+        assert!(!healthy.retryable);
         wait_for_persisted_cycle(&store, |cycle, _| cycle.next >= 1).await;
         assert_eq!(scanner_cycle_schedule_status().execution_role, "leader");
         stop_supervisor(ctx).await;