test(scanner): cover reset cleanup across store reopen (#7317)

Add a real ECStore reopen regression for scanner usage-state reset cleanup boundaries. The fixture seeds each partially completed cleanup state, recreates the store, then verifies the reset resumes without rewriting the bootstrap intent or deleting unrelated metadata.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner/tests.rs

@@ -5291,6 +5291,103 @@ async fn scanner_usage_state_reset_resumes_every_cleanup_boundary_without_rewrit
     }
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_usage_state_reset_resumes_real_store_cleanup_boundaries_after_reopen() {
+    let primary_path = DATA_USAGE_OBJ_NAME_PATH.as_str();
+    let cleanup_paths = [
+        format!("{primary_path}.bkp"),
+        LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
+        format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str().to_string(),
+    ];
+
+    for completed in 0..=cleanup_paths.len() {
+        let (_temp_dir, store) = setup_scanner_cycle_store().await;
+        let cycle = CurrentCycle {
+            current: 12,
+            next: 42,
+            cycle_completed: vec![Utc::now()],
+            started: Utc::now(),
+        };
+        save_config(
+            store.clone(),
+            DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+            encode_scanner_cycle_state(&cycle, 3).expect("cycle state should encode"),
+        )
+        .await
+        .expect("cycle state should persist");
+        let marker = scanner_usage_bootstrap_marker(std::time::SystemTime::UNIX_EPOCH, Some(3));
+        save_config(
+            store.clone(),
+            primary_path,
+            serde_json::to_vec(&marker).expect("usage reset marker should encode"),
+        )
+        .await
+        .expect("usage reset marker should persist");
+
+        for path in cleanup_paths.iter().skip(completed) {
+            let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
+            usage.scanner_epoch = Some(1);
+            usage.scanner_cycle = Some(12);
+            save_config(store.clone(), path, serde_json::to_vec(&usage).expect("cleanup slot should encode"))
+                .await
+                .expect("cleanup slot should persist");
+        }
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            save_config(store.clone(), path, b"retain".to_vec())
+                .await
+                .expect("unrelated state should persist before reopen");
+        }
+
+        let restarted = restart_scanner_cycle_store_from(&store).await;
+        let intent_before = read_config_with_revision(restarted.clone(), primary_path)
+            .await
+            .expect("reopened reset intent should be readable");
+
+        let result = reset_scanner_usage_state_for_full_rebuild(CancellationToken::new(), restarted.clone())
+            .await
+            .expect("reopened usage reset should complete");
+
+        assert_eq!(result.leader_epoch, 3, "boundary {completed}");
+        assert_eq!(result.next_cycle, 42, "boundary {completed}");
+        assert_eq!(result.reset_paths.len(), cleanup_paths.len() + 1 - completed, "boundary {completed}");
+        assert_eq!(
+            read_config_with_revision(restarted.clone(), primary_path)
+                .await
+                .expect("completed reset intent should remain readable"),
+            intent_before,
+            "boundary {completed}: resumed cleanup must not rewrite the reset intent"
+        );
+
+        let (floor, state) = persisted_usage_floor_for_startup(restarted.clone(), false)
+            .await
+            .expect("completed reset marker should remain resumable");
+        assert_eq!(floor.leader_epoch, 3, "boundary {completed}");
+        assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending, "boundary {completed}");
+        assert!(
+            persisted_usage_floor(restarted.clone()).await.is_err(),
+            "boundary {completed}: bootstrap marker must not become an authoritative floor"
+        );
+
+        for path in &cleanup_paths {
+            assert!(
+                matches!(read_config(restarted.clone(), path).await, Err(EcstoreError::ConfigNotFound)),
+                "boundary {completed}: reset should remove stale usage slot {path}"
+            );
+        }
+        for path in ["buckets/quota-reservations/ledger", "buckets/example/incarnation"] {
+            assert_eq!(
+                read_config(restarted.clone(), path)
+                    .await
+                    .expect("unrelated state should survive reopened reset"),
+                b"retain",
+                "boundary {completed}: reset must preserve non-scanner-state config"
+            );
+        }
+    }
+}
+
 #[tokio::test]
 async fn scanner_usage_state_reset_stops_usage_fence_after_owner_loss() {
     let store = Arc::new(MemoryConfigStore::default());