fix(scanner): recover usage floor from fenced backups

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

crates/scanner/src/scanner/cycle_state.rs

@@ -1327,6 +1327,13 @@ pub(super) async fn persisted_usage_floor_for_startup(
     let mut floor = PersistedUsageFloor::default();
     let mut found_any = false;
     let mut bootstrap_pending = false;
+    // A valid JSON object without a baseline identity is not a floor and must
+    // never be treated as an empty one.  It can, however, be a partially
+    // written v2 primary left behind during an upgrade.  Keep its epoch as a
+    // fence while looking for a durable companion snapshot; if no companion
+    // is new enough, the caller still fails closed below.
+    let mut invalid_baseline_path: Option<String> = None;
+    let mut invalid_baseline_epoch: Option<u64> = None;
     let update_floor = |floor: &mut PersistedUsageFloor, usage: &DataUsageInfo, path: &str| -> Result<(), ScannerError> {
         floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default());
         if let Some(completed_cycle) = usage.scanner_cycle {
@@ -1353,9 +1360,9 @@ pub(super) async fn persisted_usage_floor_for_startup(
                     update_floor(&mut floor, &usage, primary_path)?;
                     None
                 } else if !data_usage_info_has_persisted_baseline_identity(&usage) {
-                    return Err(ScannerError::Other(format!(
-                        "scanner usage floor from {primary_path} has no persisted baseline identity"
-                    )));
+                    invalid_baseline_path.get_or_insert_with(|| primary_path.to_string());
+                    invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
+                    None
                 } else {
                     let epoch = usage.scanner_epoch.unwrap_or_default();
                     update_floor(&mut floor, &usage, primary_path)?;
@@ -1377,21 +1384,27 @@ pub(super) async fn persisted_usage_floor_for_startup(
                         "scanner usage bootstrap conflicts with a persisted backup".to_string(),
                     ));
                 }
-                any_found = true;
                 let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
                     ScannerError::Other(format!("failed to decode scanner usage floor from {backup_path}: {err}"))
                 })?;
                 if !data_usage_info_has_persisted_baseline_identity(&usage) {
-                    return Err(ScannerError::Other(format!(
-                        "scanner usage floor from {backup_path} has no persisted baseline identity"
-                    )));
-                }
-                let backup_epoch = usage.scanner_epoch.unwrap_or_default();
-                // A backup write from an older leader may complete after the
-                // primary epoch has been fenced. It must not advance the startup
-                // floor unless its epoch is at least as new as the primary.
-                if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch) {
-                    update_floor(&mut floor, &usage, &backup_path)?;
+                    invalid_baseline_path.get_or_insert_with(|| backup_path.clone());
+                    invalid_baseline_epoch = invalid_baseline_epoch.max(usage.scanner_epoch);
+                    // This is still persisted state, so it must not enable a
+                    // missing-state bootstrap.  Continue to a legacy pair in
+                    // case it contains a complete, fenced snapshot.
+                    any_found = false;
+                } else {
+                    let backup_epoch = usage.scanner_epoch.unwrap_or_default();
+                    // A backup write from an older leader may complete after the
+                    // primary epoch has been fenced. It must not advance the startup
+                    // floor unless its epoch is at least as new as the primary.
+                    if primary_epoch.is_none_or(|epoch| backup_epoch >= epoch)
+                        && invalid_baseline_epoch.is_none_or(|epoch| backup_epoch >= epoch)
+                    {
+                        update_floor(&mut floor, &usage, &backup_path)?;
+                        any_found = true;
+                    }
                 }
             }
             Ok((None, _)) => {}
@@ -1413,6 +1426,11 @@ pub(super) async fn persisted_usage_floor_for_startup(
     }
 
     if !found_any && !bootstrap_pending {
+        if let Some(path) = invalid_baseline_path {
+            return Err(ScannerError::Other(format!(
+                "persisted scanner usage floor from {path} has no authoritative baseline or newer valid backup"
+            )));
+        }
         if !allow_missing_for_bootstrap {
             return Err(ScannerError::Other(
                 "persisted scanner usage floor has no authoritative baseline".to_string(),

crates/scanner/src/scanner/leadership.rs

@@ -82,9 +82,30 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
     primary: Option<&[u8]>,
     allow_bootstrap_pending: bool,
 ) -> Result<Option<DataUsageInfo>, ScannerError> {
+    // A partially written v2 primary is not itself a baseline, but a durable
+    // companion may still provide one after an interrupted upgrade. Keep the
+    // primary epoch as a fence while checking those companions; malformed
+    // bytes and bootstrap markers retain their fail-closed behavior.
+    let mut invalid_primary_epoch = None;
     if let Some(primary) = primary {
-        return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending)
-            .map(Some);
+        let usage: DataUsageInfo = serde_json::from_slice(primary).map_err(|err| {
+            ScannerError::Other(format!(
+                "failed to decode scanner usage epoch fence from {}: {err}",
+                DATA_USAGE_OBJ_NAME_PATH.as_str()
+            ))
+        })?;
+        if data_usage_info_has_persisted_baseline_identity(&usage)
+            || (allow_bootstrap_pending && data_usage_info_is_bootstrap_pending(&usage))
+        {
+            return Ok(Some(usage));
+        }
+        if data_usage_info_is_bootstrap_pending(&usage) {
+            return Err(ScannerError::Other(format!(
+                "scanner usage epoch fence from {} has no persisted baseline identity",
+                DATA_USAGE_OBJ_NAME_PATH.as_str()
+            )));
+        }
+        invalid_primary_epoch = usage.scanner_epoch;
     }
 
     let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -92,7 +113,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
         .await
         .map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence backup: {err}")))?;
     if let Some(backup) = backup.as_deref() {
-        return decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false).map(Some);
+        let usage = decode_usage_snapshot_for_epoch_fence(backup, &backup_path, false)?;
+        if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
+            return Ok(Some(usage));
+        }
     }
 
     for path in [
@@ -103,7 +127,10 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
             .await
             .map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?;
         if let Some(legacy) = legacy.as_deref() {
-            return decode_usage_snapshot_for_epoch_fence(legacy, &path, false).map(Some);
+            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));
+            }
         }
     }
     // A missing usage snapshot is an uninitialized state, not an empty

crates/scanner/src/scanner/tests.rs

@@ -1923,6 +1923,116 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() {
     assert_eq!(epoch, 11);
 }
 
+#[tokio::test]
+async fn scanner_usage_floor_recovers_from_incomplete_v2_primary_using_fenced_backup() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+
+    // This shape is valid JSON from an interrupted v2 publication, but it is
+    // not a durable baseline because the snapshot is incomplete. It must not
+    // be converted into an empty floor.
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(103);
+
+    for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    assert_eq!(
+        persisted_usage_floor(store)
+            .await
+            .expect("valid backup should recover the usage floor"),
+        PersistedUsageFloor {
+            next_cycle: 104,
+            leader_epoch: 7,
+        }
+    );
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&primary).expect("usage snapshot should encode"),
+    );
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("an existing incomplete primary must remain fail-closed");
+    assert!(err.to_string().contains("no authoritative baseline"));
+}
+
+#[tokio::test]
+async fn scanner_usage_floor_rejects_backup_older_than_incomplete_v2_primary() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let primary = DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    };
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(6);
+    backup.scanner_cycle = Some(10_000);
+
+    for (path, usage) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), primary), (backup_path.as_str(), backup)] {
+        store.objects.lock().await.insert(
+            memory_config_key(RUSTFS_META_BUCKET, path),
+            serde_json::to_vec(&usage).expect("usage snapshot should encode"),
+        );
+    }
+
+    let err = persisted_usage_floor_for_startup(store, true)
+        .await
+        .expect_err("an older backup must not cross the incomplete primary epoch fence");
+    assert!(err.to_string().contains("no authoritative baseline"));
+}
+
+#[tokio::test]
+async fn scanner_leadership_fencing_recovers_incomplete_v2_primary_from_backup() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let primary = serde_json::to_vec(&DataUsageInfo {
+        scanner_epoch: Some(7),
+        scanner_cycle: Some(100),
+        usage_snapshot_complete: false,
+        ..Default::default()
+    })
+    .expect("incomplete usage snapshot should encode");
+    let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+    backup.scanner_epoch = Some(7);
+    backup.scanner_cycle = Some(103);
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, &backup_path),
+        serde_json::to_vec(&backup).expect("backup usage snapshot should encode"),
+    );
+
+    let recovered = usage_snapshot_for_epoch_fence(store, Some(&primary), false)
+        .await
+        .expect("a valid backup should provide the fencing baseline")
+        .expect("the fencing baseline should be present");
+    assert_eq!(recovered.scanner_epoch, Some(7));
+    assert_eq!(recovered.scanner_cycle, Some(103));
+}
+
 #[tokio::test]
 async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
     let store = Arc::new(MemoryConfigStore::default());