diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 58e7d8e12..cfcd37fe9 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/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());