fix(scanner): rebuild unreadable usage floor instead of stopping leader iteration

backlog#2918 retires the usage floor as a quota proof. After dual-corruption
recovery has been attempted, an unreadable floor is reported and rebuilt
through a CAS overwrite with a bootstrap marker rather than stopping the
leader iteration for a supervisor retry. Transient I/O errors that prevent
reading the floor revision fall through to the existing stop-and-retry path.

The identity-less floor regression from rustfs/rustfs#8670 now verifies
automatic rebuild and recovery instead of bounded backoff.

crates/scanner/src/scanner.rs

@@ -1062,6 +1062,31 @@ 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;
 }
@@ -3090,6 +3115,27 @@ 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) => {
@@ -3102,7 +3148,7 @@ where
                 path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
                 state = "usage_floor_load_failed",
                 error = %err,
-                "Scanner stopped because the persisted usage floor could not be loaded"
+                "Scanner usage floor could not be loaded or rebuilt; stopping leader iteration for supervisor retry"
             );
             record_scanner_usage_floor_failure(error.clone());
             global_metrics().set_cycle(None).await;

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

@@ -41,20 +41,6 @@ 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 {
@@ -130,20 +116,19 @@ fn started_rebalance() -> EcstoreRebalanceMeta {
     }
 }
 
-/// 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.
+/// 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.
 #[tokio::test]
 #[serial]
-async fn usage_floor_load_failure_backs_off_and_recovers_without_restart() {
+async fn usage_floor_load_failure_rebuilds_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,
-        // whose error points the operator at the usage-state reset.
+        // Present but without a baseline identity: the shape #8670 reported.
         save_config(
             store.clone(),
             DATA_USAGE_OBJ_NAME_PATH.as_str(),
@@ -156,41 +141,11 @@ async fn usage_floor_load_failure_backs_off_and_recovers_without_restart() {
         let _cancel_on_drop = ctx.clone().drop_guard();
         init_data_scanner_with_storage(ctx.clone(), store.clone(), ScannerRuntimeGuard::new()).await;
 
-        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();
-        }
-
+        // backlog#2918: the unreadable floor is rebuilt automatically; the
+        // scanner must reach healthy without a manual repair or terminal pause.
         let healthy = wait_for_recovery_state("healthy").await;
-        assert_eq!(healthy.retry_count, 0, "a successful attempt clears the retry budget: {healthy:?}");
-        assert!(!healthy.retryable);
+        assert!(!healthy.retryable, "rebuild must clear the retry budget: {healthy:?}");
+        assert_ne!(healthy.state, "paused", "rebuild must not report a terminal pause");
         wait_for_persisted_cycle(&store, |cycle, _| cycle.next >= 1).await;
         assert_eq!(scanner_cycle_schedule_status().execution_role, "leader");
         stop_supervisor(ctx).await;