diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 65f76ac2a..afd54825d 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1910,10 +1910,10 @@ where .as_ref() .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone())); let remote_lease_release_safe = Arc::new(AtomicBool::new(true)); - let mut usage_persist_outcome = match publication_defer_reason { + let mut usage_publication_result = match publication_defer_reason { Some(reason) => { drop(receiver); - DataUsagePersistOutcome::Deferred(reason) + DataUsagePublicationResult::from(DataUsagePersistOutcome::Deferred(reason)) } None => { // ScannerIO emits its complete or observational update only after @@ -1928,6 +1928,11 @@ where .as_ref() .map(|(_, grants)| grants.iter().map(|grant| grant.lease.token).collect()) .unwrap_or_default(); + let ack_expectation = scan_result + .as_ref() + .ok() + .filter(|result| result.has_dirty_usage_to_acknowledge()) + .and_then(ScannerCycleResult::publication_expectation); let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move { store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence( ctx_clone, @@ -1940,6 +1945,7 @@ where remote_lease_deadline, remote_lease_fence, ) + .with_ack_expectation(ack_expectation) .with_remote_lease_tokens(remote_lease_tokens) .with_lease_release_flag(remote_lease_release_safe_for_task), move || { @@ -1973,7 +1979,7 @@ where error = %err, "Scanner data usage persistence task failed" ); - DataUsagePersistOutcome::Failed + DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed) } DataUsagePersistTaskResult::Cancelled => { debug!( @@ -1985,7 +1991,7 @@ where state = "usage_persist_task_cancelled", "Scanner data usage persistence task cancelled" ); - DataUsagePersistOutcome::Failed + DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed) } DataUsagePersistTaskResult::TimedOut => { error!( @@ -1998,11 +2004,12 @@ where state = "usage_persist_task_timed_out", "Scanner data usage persistence task timed out" ); - DataUsagePersistOutcome::Failed + DataUsagePublicationResult::from(DataUsagePersistOutcome::Failed) } } } }; + let mut usage_persist_outcome = usage_publication_result.outcome(); let lease_expired = remote_publication_leases .as_ref() .is_some_and(|(_, grants)| grants.iter().any(|grant| !grant.lease.is_valid())); @@ -2202,8 +2209,9 @@ where }; } + usage_publication_result.restrict_outcome(usage_persist_outcome); let (completion_outcome, scanner_pending_maintenance_work, remote_dirty_usage_acknowledgements) = - finalize_scanner_cycle_result(scan_cycle_result, usage_persist_outcome); + finalize_scanner_cycle_result(scan_cycle_result, usage_publication_result); let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() { false } else if let Some(notification_system) = storeapi.scanner_notification_system() { @@ -3437,21 +3445,35 @@ fn scanner_cycle_completion_outcome( fn finalize_scanner_cycle_result( scan_cycle_result: crate::scanner_io::ScannerCycleResult, - usage_persist_outcome: DataUsagePersistOutcome, + publication: DataUsagePublicationResult, ) -> (ScannerCycleOutcome, bool, Vec) { + let (usage_persist_outcome, proof) = publication.into_parts(); let completion_outcome = scanner_cycle_completion_outcome_for_result(&scan_cycle_result, usage_persist_outcome); - let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work(); let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete && matches!( usage_persist_outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable - ); + ) + && scan_cycle_result.publication_expectation().as_ref().is_some_and(|expected| { + proof + .as_ref() + .is_some_and(|proof| proof.verified_version_for(expected).is_some()) + }); + let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work() + || (scan_cycle_result.has_dirty_usage_to_acknowledge() && !durable_complete_snapshot); let remote_dirty_usage_acknowledgements = if durable_complete_snapshot { - scan_cycle_result.acknowledge_durable_usage() + match proof { + Some(proof) => scan_cycle_result.acknowledge_durable_usage(&proof), + None => Vec::new(), + } } else { Vec::new() }; - (completion_outcome, pending_maintenance_work, remote_dirty_usage_acknowledgements) + ( + completion_outcome, + pending_maintenance_work || crate::scanner_io::dirty_usage_buckets_pending(), + remote_dirty_usage_acknowledgements, + ) } fn scanner_cycle_completion_outcome_for_result( @@ -3551,6 +3573,7 @@ use activity::*; use backlog::*; use cycle_state::*; use leadership::*; +pub(crate) use usage_store::RootPublicationProof; use usage_store::*; pub use activity::scanner_topology_digest; diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 04d548404..6d75ce26e 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -37,6 +37,7 @@ use tokio::time::{Duration, advance}; const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60; mod recovery_control; +mod scoped_ack_publication; async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc) { setup_scanner_cycle_store_with_usage_baseline(true).await @@ -6184,7 +6185,7 @@ async fn coordinator_classifies_an_expired_publication_lease() { .await; assert_eq!( - outcome, + outcome.outcome(), DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded) ); assert!(store.put_counts.lock().await.is_empty(), "expired lease must prevent a PUT"); @@ -7325,7 +7326,7 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() { #[test] #[serial] -fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() { +fn finalizing_a_saved_enum_without_proof_keeps_dirty_pending() { crate::scanner_io::clear_dirty_usage_bucket("photos"); crate::scanner_io::record_dirty_usage_bucket("photos"); let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); @@ -7337,17 +7338,19 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() { }; let unsaved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot.clone())) .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate); + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(unsaved, DataUsagePersistOutcome::NoUpdate.into()); assert_eq!(outcome, ScannerCycleOutcome::Failed); assert!(acknowledgements.is_empty()); assert!(crate::scanner_io::dirty_usage_buckets_pending()); let saved = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot)) - .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement.clone()]); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved); + .with_remote_dirty_usage_acknowledgements(vec![remote_acknowledgement]); + let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(saved, DataUsagePersistOutcome::Saved.into()); assert_eq!(outcome, ScannerCycleOutcome::Completed); - assert_eq!(acknowledgements, vec![remote_acknowledgement]); - assert!(!crate::scanner_io::dirty_usage_buckets_pending()); + assert!(acknowledgements.is_empty()); + assert!(pending); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + crate::scanner_io::clear_dirty_usage_bucket("photos"); } #[test] @@ -7359,7 +7362,7 @@ fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() { let deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot)); let (outcome, _, acknowledgements) = - finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); + finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement).into()); assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement)); assert!(acknowledgements.is_empty()); @@ -7379,7 +7382,7 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() { ) .with_observational_snapshot_published(true); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved); + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved.into()); assert_eq!(outcome, ScannerCycleOutcome::Partial); assert!(acknowledgements.is_empty()); @@ -7415,17 +7418,20 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() { #[test] #[serial] -fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() { +fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() { crate::scanner_io::clear_dirty_usage_bucket("photos"); crate::scanner_io::record_dirty_usage_bucket("photos"); let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot)); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable); + let (outcome, pending, acknowledgements) = + finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::AlreadyDurable.into()); assert_eq!(outcome, ScannerCycleOutcome::Completed); assert!(acknowledgements.is_empty()); - assert!(!crate::scanner_io::dirty_usage_buckets_pending()); + assert!(pending); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + crate::scanner_io::clear_dirty_usage_bucket("photos"); } #[test] @@ -7436,7 +7442,8 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() { let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); let durable = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot)); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable); + let (outcome, _, acknowledgements) = + finalize_scanner_cycle_result(durable, DataUsagePersistOutcome::PriorCycleDurable.into()); assert_eq!(outcome, ScannerCycleOutcome::Completed); assert!(acknowledgements.is_empty()); @@ -7452,7 +7459,7 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() { let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); let superseded = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Superseded, Some(dirty_snapshot)); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved); + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved.into()); assert_eq!(outcome, ScannerCycleOutcome::Superseded); assert!(acknowledgements.is_empty()); @@ -8881,7 +8888,7 @@ fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acqui ]); let (outcome, _, acknowledgements) = finalize_scanner_cycle_result( result, - DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")), + DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")).into(), ); assert_eq!( outcome, diff --git a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs new file mode 100644 index 000000000..39deae1e5 --- /dev/null +++ b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs @@ -0,0 +1,490 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +use super::super::usage_store::DataUsagePublicationResult; +use super::*; +use crate::scanner_io::ScannerBucketScanScope; +use rustfs_utils::path::path_join_buf; +use sha2::Digest; +use std::time::SystemTime; + +const PROOF_BUCKET: &str = "publication-proof-bucket"; +const PROOF_EPOCH: u64 = 7; +const PROOF_CYCLE: u64 = 11; + +async fn settle_namespace_commits(store: &ECStore) { + tokio::time::timeout(Duration::from_secs(30), async { + while store.scanner_data_usage_publication_blocked().await { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("fixture namespace commits must settle before collecting complete coverage"); +} + +async fn complete_candidate(store: &Arc, cycle: u64) -> (crate::scanner_io::ScannerCycleResult, DataUsageInfo) { + settle_namespace_commits(store).await; + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking( + &ctx, + ScannerCycleBudgetConfig { + max_objects: Some(8), + ..Default::default() + }, + ); + let (updates, mut receiver) = mpsc::channel(1); + let result = crate::scanner_io::nsscanner_with_storage_status_scoped( + store.as_ref(), + crate::scanner_io::ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle: cycle, + leader_epoch: PROOF_EPOCH, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: None, + observed_usage_candidate: None, + requires_full_scan: true, + service_cohort: None, + resolved_scope_observer: None, + }, + ) + .await + .expect("real scanner must produce the fixture candidate"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + let candidate = receiver.recv().await.expect("complete scanner snapshot"); + assert!(candidate.usage_snapshot_complete); + assert_eq!(candidate.usage_snapshot_converged, Some(true)); + assert_eq!(candidate.scanner_cycle, Some(cycle)); + assert_eq!(candidate.scanner_epoch, Some(PROOF_EPOCH)); + (result, candidate) +} + +async fn candidate_store() -> (tempfile::TempDir, Arc) { + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + let (directory, store) = setup_scanner_cycle_store_with_usage_baseline(false).await; + store + .make_bucket(PROOF_BUCKET, &crate::storage_api::scan::MakeBucketOptions::default()) + .await + .expect("create proof fixture bucket through the owner"); + let mut reader = PutObjReader::from_vec(b"proof".to_vec()); + store.pools[0].disk_set[0] + .put_object(PROOF_BUCKET, "initial", &mut reader, &ObjectOptions::default()) + .await + .expect("persist fixture object through the owner"); + crate::scanner_io::record_dirty_usage_bucket(PROOF_BUCKET); + settle_namespace_commits(&store).await; + (directory, store) +} + +async fn read_root(store: &Arc) -> (Option>, DataUsageCacheRevision) { + read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("read actual v2 root bytes and revision") +} + +async fn publish_candidate( + store: &Arc, + scan: &crate::scanner_io::ScannerCycleResult, + candidate: DataUsageInfo, + baseline: Option, +) -> DataUsagePublicationResult { + let expectation = scan.publication_expectation(); + assert!(expectation.is_some(), "only a real complete scan may supply the expectation"); + let (sender, receiver) = mpsc::channel(1); + sender.send(candidate).await.expect("enqueue the real scan candidate"); + drop(sender); + store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence( + CancellationToken::new(), + store.clone(), + receiver, + Some(PROOF_EPOCH), + baseline, + ScannerPublicationFence::new(scan.publication_epoch(), None, None).with_ack_expectation(expectation), + || async { None }, + ) + .await +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_companion_only_does_not_authorize_root_ack() { + for companion in [ + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + let (_directory, store) = candidate_store().await; + let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let bytes = serde_json::to_vec(&candidate).expect("actual candidate JSON"); + save_config(store.clone(), &companion, bytes.clone()) + .await + .expect("persist the companion on real disks"); + let baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("companion fallback baseline"); + assert_eq!(baseline.data.as_deref(), Some(bytes.as_slice())); + assert_eq!(baseline.revision, DataUsageCacheRevision::Missing); + assert_eq!(read_root(&store).await.0, None); + let dirty = crate::scanner_io::dirty_usage_buckets_for_tests(); + + let publication = publish_candidate(&store, &scan, candidate, Some(baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable); + let (_, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication); + assert!(pending, "unacknowledged durable companion work must remain pending"); + assert!(acknowledgements.is_empty()); + assert_eq!( + crate::scanner_io::dirty_usage_buckets_for_tests(), + dirty, + "a companion is not the v2 root target" + ); + assert_eq!(read_root(&store).await, (None, DataUsageCacheRevision::Missing)); + assert_eq!(read_config(store.clone(), &companion).await.expect("companion retained"), bytes); + } + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_actual_root_readback_accepts_semantic_json_equivalence() { + let (_directory, store) = candidate_store().await; + let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let canonical = serde_json::to_vec(&candidate).expect("candidate encoding"); + let mut value = serde_json::to_value(&candidate).expect("candidate value"); + value + .as_object_mut() + .expect("usage object") + .insert("fixture_unknown_field".into(), serde_json::json!({"retained": true})); + let different_bytes = serde_json::to_vec_pretty(&value).expect("noncanonical primary JSON"); + assert_ne!(different_bytes, canonical); + assert_eq!( + serde_json::from_slice::(&different_bytes).expect("semantic primary"), + candidate + ); + save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), different_bytes.clone()) + .await + .expect("persist actual primary representation"); + let before = read_root(&store).await; + assert!(matches!(&before.1, DataUsageCacheRevision::Etag(etag) if !etag.is_empty())); + let baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("real primary revision"); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + + let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable); + let (_, proof) = publication.into_parts(); + let proof = proof.expect("actual primary readback must produce its own root proof"); + let expected = scan.publication_expectation().expect("real scan expectation"); + let (etag, raw_digest) = proof.verified_version_for(&expected).expect("proof must bind this candidate"); + let DataUsageCacheRevision::Etag(expected_etag) = &before.1 else { panic!("actual root ETag") }; + assert_eq!(etag, expected_etag); + let expected_digest: [u8; 32] = sha2::Sha256::digest(&different_bytes).into(); + assert_eq!( + *raw_digest, expected_digest, + "proof must record actual bytes, not reserialized candidate bytes" + ); + // Obtain another proof through the same real readback path rather than + // fabricating a publication result from the inspected proof above. + let publication = publish_candidate(&store, &scan, candidate, None).await; + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication); + assert_eq!(outcome, ScannerCycleOutcome::Completed); + assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets"); + assert!( + !crate::scanner_io::dirty_usage_buckets_pending(), + "actual root bytes plus a real revision authorize this scan" + ); + assert_eq!(read_root(&store).await, before, "readback must not rewrite unknown fields or whitespace"); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_successful_root_cas_authorizes_its_scan() { + let (_directory, store) = candidate_store().await; + let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("initial root revision"); + assert_eq!(baseline.revision, DataUsageCacheRevision::Missing); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved); + let (bytes, revision) = read_root(&store).await; + assert!(matches!(revision, DataUsageCacheRevision::Etag(etag) if !etag.is_empty())); + assert_eq!( + serde_json::from_slice::(&bytes.expect("actual saved root")).expect("root JSON"), + candidate + ); + let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication); + assert_eq!(outcome, ScannerCycleOutcome::Completed); + assert!(!pending); + assert!(acknowledgements.is_empty(), "the single-node fixture has no remote targets"); + assert!( + !crate::scanner_io::dirty_usage_buckets_pending(), + "the real root CAS must authorize its matching scan" + ); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_observed_candidate_reuse_requires_a_new_root_proof() { + let (_directory, store) = candidate_store().await; + let bootstrap = scanner_usage_bootstrap_marker(SystemTime::now(), Some(PROOF_EPOCH)); + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&bootstrap).expect("bootstrap root encoding"), + ) + .await + .expect("persist authoritative bootstrap root"); + let (prior_scan, mut observed_candidate) = complete_candidate(&store, PROOF_CYCLE).await; + // Seed a complete but unconverged observation from real scanner coverage; + // the production writer attaches its authoritative baseline identity. + observed_candidate.usage_snapshot_converged = Some(false); + let observation = publish_candidate(&store, &prior_scan, observed_candidate, None).await; + let (outcome, proof) = observation.into_parts(); + assert_eq!(outcome, DataUsagePersistOutcome::Saved); + assert!(proof.is_none(), "an observational write cannot authorize a root ACK"); + let (root_before, revision_before) = read_root(&store).await; + let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()) + .await + .expect("read real persisted observation"); + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let (updates, mut receiver) = mpsc::channel(1); + let (observer, selected) = tokio::sync::oneshot::channel(); + let scan = crate::scanner_io::nsscanner_with_storage_status_scoped( + store.as_ref(), + crate::scanner_io::ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle: PROOF_CYCLE + 1, + leader_epoch: PROOF_EPOCH, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: root_before.clone().map(Bytes::from), + observed_usage_candidate: Some(Bytes::from(observed)), + requires_full_scan: false, + service_cohort: None, + resolved_scope_observer: Some(observer), + }, + ) + .await + .expect("observation-backed scope must run through the real scanner"); + let scope = selected.await.expect("production resolver decision"); + assert_eq!(scope.selected_buckets_for_tests(), Some(&HashSet::from([PROOF_BUCKET.to_string()]))); + assert_eq!(scan.status, ScannerCycleStatus::Complete); + let expectation = scan.publication_expectation().expect("reused coverage must be revalidated"); + assert!( + !expectation.same_candidate(&prior_scan.publication_expectation().expect("prior real candidate")), + "the observation cannot transfer the previous scan's expectation" + ); + assert_eq!(read_root(&store).await, (root_before, revision_before)); + assert!(crate::scanner_io::dirty_usage_buckets_pending()); + let candidate = receiver.recv().await.expect("new validated root candidate"); + assert_eq!(candidate.scanner_cycle, Some(PROOF_CYCLE + 1)); + assert_eq!(candidate.usage_snapshot_converged, Some(true)); + let publication = publish_candidate(&store, &scan, candidate, None).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved); + let (outcome, pending, acknowledgements) = finalize_scanner_cycle_result(scan, publication); + assert_eq!(outcome, ScannerCycleOutcome::Completed); + assert!(!pending); + assert!(acknowledgements.is_empty()); + assert!(!crate::scanner_io::dirty_usage_buckets_pending()); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_stale_root_cas_keeps_dirty_after_bucket_save() { + let (_directory, store) = candidate_store().await; + let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let mut bucket_cache = DataUsageCache::default(); + bucket_cache + .load(store.pools[0].disk_set[0].clone(), &path_join_buf(&[PROOF_BUCKET, DATA_USAGE_CACHE_NAME])) + .await + .expect("real bucket checkpoint must be persisted before root publication"); + assert!(bucket_cache.info.snapshot_complete); + assert_eq!( + bucket_cache + .checked_flatten(PROOF_BUCKET) + .expect("persisted bucket root") + .objects, + 1 + ); + let stale_baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("missing root revision"); + assert_eq!(stale_baseline.revision, DataUsageCacheRevision::Missing); + let mut competing = candidate.clone(); + competing.scanner_epoch = Some(PROOF_EPOCH + 1); + competing.scanner_cycle = Some(PROOF_CYCLE + 1); + for state in &mut competing.usage_snapshot_set_states { + state.scanner_epoch = Some(PROOF_EPOCH + 1); + state.scanner_cycle = Some(PROOF_CYCLE + 1); + } + let competing_bytes = serde_json::to_vec(&competing).expect("competing root"); + save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), competing_bytes.clone()) + .await + .expect("another publisher wins the actual root slot"); + let before = read_root(&store).await; + let dirty = crate::scanner_io::dirty_usage_buckets_for_tests(); + + let publication = publish_candidate(&store, &scan, candidate, Some(stale_baseline)).await; + assert_eq!( + publication.outcome(), + DataUsagePersistOutcome::Current, + "the old missing revision loses CAS and reconciles the newer root" + ); + let (_, _, acknowledgements) = finalize_scanner_cycle_result(scan, publication); + assert!(acknowledgements.is_empty()); + assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty); + assert_eq!( + read_root(&store).await, + before, + "bucket durability must not authorize replacing the winning root" + ); + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_cannot_transfer_proof_between_real_scan_results() { + let (_directory, store) = candidate_store().await; + let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let (second_scan, second_candidate) = complete_candidate(&store, PROOF_CYCLE).await; + assert_eq!(first_candidate.scanner_epoch, second_candidate.scanner_epoch); + assert_eq!(first_candidate.scanner_cycle, second_candidate.scanner_cycle); + assert_eq!(first_candidate.objects_total_count, second_candidate.objects_total_count); + let baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("initial root revision"); + let dirty = crate::scanner_io::dirty_usage_buckets_for_tests(); + let publication = publish_candidate(&store, &first_scan, first_candidate, Some(baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved); + assert!(read_root(&store).await.0.is_some(), "the first scan really published its root"); + + let (_, pending, acknowledgements) = finalize_scanner_cycle_result(second_scan, publication); + assert!(pending, "another scan's publication must not finish this scan's dirty maintenance work"); + assert!(acknowledgements.is_empty()); + assert_eq!( + crate::scanner_io::dirty_usage_buckets_for_tests(), + dirty, + "same counters and cycle cannot transfer another scan's proof" + ); + + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_stale_baseline_cannot_prove_a_replaced_root() { + let (_directory, store) = candidate_store().await; + let (first_scan, first_candidate) = complete_candidate(&store, PROOF_CYCLE).await; + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&first_candidate).expect("first candidate"), + ) + .await + .expect("persist the first candidate on real disks"); + let stale_baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("capture the genuine first root revision"); + let dirty = crate::scanner_io::dirty_usage_buckets_for_tests(); + let mut reader = PutObjReader::from_vec(b"second".to_vec()); + store.pools[0].disk_set[0] + .put_object(PROOF_BUCKET, "second", &mut reader, &ObjectOptions::default()) + .await + .expect("commit a real namespace change"); + assert_eq!( + crate::scanner_io::dirty_usage_buckets_for_tests(), + dirty, + "direct storage writes leave this fixture's scanner hint generation unchanged" + ); + let (_, replacement) = complete_candidate(&store, PROOF_CYCLE).await; + assert_eq!(first_candidate.scanner_epoch, replacement.scanner_epoch); + assert_eq!(first_candidate.scanner_cycle, replacement.scanner_cycle); + assert_eq!((first_candidate.objects_total_count, replacement.objects_total_count), (1, 2)); + save_config( + store.clone(), + DATA_USAGE_OBJ_NAME_PATH.as_str(), + serde_json::to_vec(&replacement).expect("replacement candidate"), + ) + .await + .expect("publish the replacement root"); + let current = read_root(&store).await; + assert_ne!(current.1, stale_baseline.revision); + + // The supplied baseline still equals candidate A, but the actual target + // now contains B. Compatibility's AlreadyDurable outcome is not proof. + let publication = publish_candidate(&store, &first_scan, first_candidate, Some(stale_baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::AlreadyDurable); + let (_, pending, acknowledgements) = finalize_scanner_cycle_result(first_scan, publication); + assert!(pending); + assert!(acknowledgements.is_empty()); + assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty); + assert_eq!(read_root(&store).await, current); + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn scoped_ack_publication_rejects_builder_mutation_after_real_root_publish() { + for mutation in ["remote_ack_target", "publication_epoch", "remote_lease_targets"] { + let (_directory, store) = candidate_store().await; + let (scan, candidate) = complete_candidate(&store, PROOF_CYCLE).await; + let baseline = read_data_usage_persist_baseline(store.clone()) + .await + .expect("initial root revision"); + let dirty = crate::scanner_io::dirty_usage_buckets_for_tests(); + let changed_generation = dirty + .get(PROOF_BUCKET) + .expect("the real scan has dirty work") + .checked_add(1) + .expect("bounded fixture generation"); + let changed_epoch = scan + .publication_epoch() + .expect("real scan publication epoch") + .checked_add(1) + .expect("bounded fixture epoch"); + let publication = publish_candidate(&store, &scan, candidate.clone(), Some(baseline)).await; + assert_eq!(publication.outcome(), DataUsagePersistOutcome::Saved, "{mutation}"); + let root_before = read_root(&store).await; + assert_eq!( + serde_json::from_slice::(root_before.0.as_deref().expect("actual saved root")) + .expect("persisted root JSON"), + candidate, + "{mutation}: the original candidate really reached root storage" + ); + + let changed = match mutation { + "remote_ack_target" => scan.with_remote_dirty_usage_acknowledgements(vec![ScannerDirtyUsageAcknowledgement { + host: "proof-peer:9000".to_string(), + instance_id: crate::scanner_activity_epoch().to_string(), + generation: changed_generation, + }]), + "publication_epoch" => scan.with_publication_epoch(Some(changed_epoch)), + "remote_lease_targets" => scan.with_remote_publication_lease_targets(vec![( + "proof-peer:9000".to_string(), + crate::scanner_activity_epoch().to_string(), + changed_generation, + )]), + _ => unreachable!("fixed mutation cases"), + }; + let (_, pending, acknowledgements) = finalize_scanner_cycle_result(changed, publication); + assert!( + acknowledgements.is_empty(), + "{mutation}: the old root proof must not authorize changed ACK work" + ); + assert!(pending, "{mutation}: changed maintenance work must remain pending"); + assert_eq!( + crate::scanner_io::dirty_usage_buckets_for_tests(), + dirty, + "{mutation}: the changed scan must not clear local dirty work" + ); + assert_eq!(read_root(&store).await, root_before, "{mutation}: the durable original root is retained"); + } + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 08a3c9996..5c4727732 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -34,6 +34,139 @@ pub(super) enum DataUsagePersistOutcome { Failed, } +#[derive(Debug)] +pub(crate) struct RootPublicationProof { + candidate: crate::scanner_io::ScannerPublicationExpectation, + root_version: (String, [u8; 32]), +} + +impl RootPublicationProof { + pub(crate) fn verified_version_for( + &self, + expected: &crate::scanner_io::ScannerPublicationExpectation, + ) -> Option<&(String, [u8; 32])> { + self.candidate.same_candidate(expected).then_some(&self.root_version) + } +} + +#[derive(Debug)] +pub(super) struct DataUsagePublicationResult { + outcome: DataUsagePersistOutcome, + proof: Option, +} + +impl From for DataUsagePublicationResult { + fn from(outcome: DataUsagePersistOutcome) -> Self { + Self { outcome, proof: None } + } +} + +impl DataUsagePublicationResult { + pub(super) fn outcome(&self) -> DataUsagePersistOutcome { + self.outcome + } + pub(super) fn restrict_outcome(&mut self, outcome: DataUsagePersistOutcome) { + if outcome != self.outcome { + self.proof = None; + } + self.outcome = outcome; + } + pub(super) fn into_parts(self) -> (DataUsagePersistOutcome, Option) { + (self.outcome, self.proof) + } +} + +fn root_ack_write_is_confirmed( + result: &std::result::Result, + state: Option, + written_etag: Option<&str>, +) -> bool { + result.is_ok() && state == Some(ScannerPublicationCommitState::Committed) && written_etag.is_some_and(|etag| !etag.is_empty()) +} + +#[cfg(test)] +mod root_publication_confirmation_tests { + use super::*; + + #[test] + fn root_publication_confirmation_requires_committed_state_and_write_revision() { + let saved = Ok::<(), ()>(()); + for state in [ + None, + Some(ScannerPublicationCommitState::Admitted), + Some(ScannerPublicationCommitState::InFlight), + Some(ScannerPublicationCommitState::AbortedBeforeCommit), + Some(ScannerPublicationCommitState::Indeterminate), + ] { + assert!(!root_ack_write_is_confirmed(&saved, state, Some("revision"))); + } + for etag in [None, Some("")] { + assert!(!root_ack_write_is_confirmed(&saved, Some(ScannerPublicationCommitState::Committed), etag)); + } + assert!(root_ack_write_is_confirmed( + &saved, + Some(ScannerPublicationCommitState::Committed), + Some("revision") + )); + } + + #[test] + fn root_publication_confirmation_does_not_carry_state_across_cas_attempts() { + let attempts = [ + (Err(()), Some(ScannerPublicationCommitState::Committed), Some("first")), + (Ok(()), Some(ScannerPublicationCommitState::AbortedBeforeCommit), Some("second")), + (Ok(()), None, Some("legacy")), + (Ok(()), Some(ScannerPublicationCommitState::Committed), Some("confirmed")), + ]; + let confirmations = attempts + .iter() + .map(|(result, state, etag)| root_ack_write_is_confirmed(result, *state, *etag)) + .collect::>(); + assert_eq!(confirmations, [false, false, false, true]); + } +} + +async fn read_root_publication_proof( + store: Arc, + ctx: &CancellationToken, + deadline: tokio::time::Instant, + epoch: u64, + expected: &crate::scanner_io::ScannerPublicationExpectation, + candidate: &DataUsageInfo, + written_etag: Option<&str>, +) -> Option { + let read = async { + let _admission = scanner_publication_admission_for_epoch(store.clone(), epoch).await?; + let (bytes, revision) = read_config_with_revision(store, DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .ok()?; + let bytes = bytes?; + let DataUsageCacheRevision::Etag(etag) = revision else { + return None; + }; + if etag.is_empty() || written_etag.is_some_and(|written| written != etag) { + return None; + } + let persisted: DataUsageInfo = serde_json::from_slice(&bytes).ok()?; + if &persisted != candidate { + return None; + } + let root_digest = Sha256::digest(&bytes).into(); + if ctx.is_cancelled() || tokio::time::Instant::now() >= deadline { + return None; + } + Some(RootPublicationProof { + candidate: expected.clone(), + root_version: (etag, root_digest), + }) + }; + tokio::select! { + biased; + _ = ctx.cancelled() => None, + result = tokio::time::timeout_at(deadline, read) => result.ok().flatten(), + } +} + fn remote_lease_expired(deadline: Option) -> bool { deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) } @@ -166,6 +299,7 @@ pub(super) struct ScannerPublicationFence { pub(super) scanner_publication_lease_fence: Option, pub(super) remote_lease_tokens: Vec, pub(super) lease_release_safe: Arc, + pub(super) ack_expectation: Option, } impl ScannerPublicationFence { @@ -180,6 +314,7 @@ impl ScannerPublicationFence { scanner_publication_lease_fence, remote_lease_tokens: Vec::new(), lease_release_safe: Arc::new(AtomicBool::new(true)), + ack_expectation: None, } } @@ -192,21 +327,26 @@ impl ScannerPublicationFence { self.lease_release_safe = lease_release_safe; self } + + pub(super) fn with_ack_expectation(mut self, expected: Option) -> Self { + self.ack_expectation = expected; + self + } } #[derive(Debug)] -pub(super) enum DataUsagePersistTaskResult { - Completed(DataUsagePersistOutcome), +pub(super) enum DataUsagePersistTaskResult { + Completed(T), Cancelled, TimedOut, JoinFailed(tokio::task::JoinError), } -pub(super) async fn wait_for_data_usage_persist_task( +pub(super) async fn wait_for_data_usage_persist_task( ctx: &CancellationToken, - task: &mut AbortOnDropHandle, + task: &mut AbortOnDropHandle, timeout: Duration, -) -> DataUsagePersistTaskResult { +) -> DataUsagePersistTaskResult { tokio::select! { biased; result = &mut *task => match result { @@ -320,6 +460,7 @@ where route_probe, ) .await + .outcome() } pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe_for_publication_epoch_and_lease_fence< @@ -333,7 +474,7 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel initial_baseline: Option, publication_fence: ScannerPublicationFence, route_probe: F, -) -> DataUsagePersistOutcome +) -> DataUsagePublicationResult where F: Fn() -> Fut + Send + Sync, Fut: Future> + Send, @@ -344,11 +485,15 @@ where scanner_publication_lease_fence, remote_lease_tokens, lease_release_safe, + ack_expectation, } = publication_fence; + let ack_deadline = scanner_publication_scope_deadline(data_usage_persist_timeout(), remote_lease_deadline); let mut outcome = DataUsagePersistOutcome::NoUpdate; + let mut proof = None; let mut next_baseline = initial_baseline; 'updates: while let Some(mut data_usage_info) = receiver.recv().await { + proof = None; let _activity_guard = ScannerActivityGuard::new(); if ctx.is_cancelled() { break; @@ -523,10 +668,14 @@ where continue; } }; - let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower)); + let data_digest: [u8; 32] = Sha256::digest(&data).into(); + let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(data_digest, hex_simd::AsciiCase::Lower)); let data = Bytes::from(data); let backup_due = !observational && data_usage_backup_due(&data_usage_info); let mut cas_retry = 0usize; + let mut ack_epoch = None; + let mut write_confirmed = false; + let mut written_etag = None; let save_outcome = loop { if ctx.is_cancelled() { break 'updates; @@ -557,6 +706,7 @@ where } else { None }; + ack_epoch = Some(publication_epoch_for_save); let (existing_data, revision) = match baseline { Some(baseline) => (baseline.data, baseline.revision), None => match read_config_with_revision(storeapi.clone(), target_path).await { @@ -645,7 +795,7 @@ where } let done_save = Metrics::time(Metric::SaveUsage); - let save_result = { + let (save_result, commit_state) = { let publication_scope = storeapi .scanner_data_usage_publication_commit_scope_with_release_flag( publication_epoch_for_save, @@ -681,24 +831,33 @@ where .await; drop(legacy_publication_admission); if let Some(scope) = publication_scope { - match scope.wait_for_completion().await { - ScannerPublicationCommitState::Committed | ScannerPublicationCommitState::AbortedBeforeCommit => { - save_result - } + let state = scope.wait_for_completion().await; + let result = match state { + ScannerPublicationCommitState::Committed => save_result, + ScannerPublicationCommitState::AbortedBeforeCommit => save_result, ScannerPublicationCommitState::Indeterminate | ScannerPublicationCommitState::Admitted | ScannerPublicationCommitState::InFlight => Err(EcstoreError::other( "scanner publication commit scope did not reach a safe terminal state", )), - } + }; + (result, Some(state)) } else { - save_result + (save_result, None) } }; done_save(); + let attempt_confirmed = root_ack_write_is_confirmed( + &save_result, + commit_state, + save_result.as_ref().ok().and_then(|info| info.etag.as_deref()), + ); + match save_result { Ok(object_info) => { + write_confirmed = attempt_confirmed; + written_etag = object_info.etag.as_ref().filter(|etag| !etag.is_empty()).cloned(); if !observational { next_baseline = object_info .etag @@ -909,9 +1068,27 @@ where break 'updates; } } + if !observational + && data_usage_info.usage_snapshot_converged == Some(true) + && matches!(outcome, DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable) + && (outcome == DataUsagePersistOutcome::AlreadyDurable || write_confirmed) + && let (Some(expected), Some(epoch)) = (ack_expectation.as_ref(), ack_epoch) + && expected.matches_encoded_candidate(&data_digest) + { + proof = read_root_publication_proof( + storeapi.clone(), + &ctx, + ack_deadline, + epoch, + expected, + &data_usage_info, + written_etag.as_deref(), + ) + .await; + } } - outcome + DataUsagePublicationResult { outcome, proof } } async fn cleanup_observed_data_usage_snapshot_for_epoch_and_lease( diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 3ec8cb5ab..b51e5d2f8 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -114,6 +114,11 @@ pub(crate) struct ScannerBucketScanScope { } impl ScannerBucketScanScope { + #[cfg(test)] + pub(crate) fn selected_buckets_for_tests(&self) -> Option<&HashSet> { + self.selected_buckets.as_deref() + } + fn is_default(&self) -> bool { self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none() } @@ -218,8 +223,8 @@ fn complete_scanner_cache_snapshot_plan_digest( fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option { let authoritative = serde_json::from_slice::(proof.authoritative_data?).ok()?; - if complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true).is_some() { - return Some(proof.scan_plan_digest); + if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, true) { + return Some(validated_digest); } // A complete but superseded observation may reuse its per-set cache only @@ -896,6 +901,7 @@ pub(crate) struct ScannerCycleResult { failed_dirty_usage: bool, pending_maintenance_work: bool, required_cycle_floor: Option, + publication_expectation: Option, } impl ScannerCycleResult { @@ -911,10 +917,12 @@ impl ScannerCycleResult { failed_dirty_usage: false, pending_maintenance_work: false, required_cycle_floor: None, + publication_expectation: None, } } pub(crate) fn with_publication_epoch(mut self, publication_epoch: Option) -> Self { + self.publication_expectation = None; self.publication_epoch = publication_epoch; self } @@ -924,6 +932,7 @@ impl ScannerCycleResult { } fn with_activity_digest(mut self, activity_digest: [u8; 32]) -> Self { + self.publication_expectation = None; self.activity_digest = Some(activity_digest); self } @@ -933,6 +942,7 @@ impl ScannerCycleResult { } pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self { + self.publication_expectation = None; self.observational_snapshot_published = published; self } @@ -942,16 +952,19 @@ impl ScannerCycleResult { } fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self { + self.publication_expectation = None; self.failed_dirty_usage = failed_dirty_usage; self } fn with_pending_maintenance_work(mut self, pending_maintenance_work: bool) -> Self { + self.publication_expectation = None; self.pending_maintenance_work = pending_maintenance_work; self } fn with_required_cycle_floor(mut self, required_cycle_floor: Option) -> Self { + self.publication_expectation = None; self.required_cycle_floor = required_cycle_floor; self } @@ -960,11 +973,13 @@ impl ScannerCycleResult { mut self, acknowledgements: Vec, ) -> Self { + self.publication_expectation = None; self.remote_dirty_usage_acknowledgements = acknowledgements; self } pub(crate) fn with_remote_publication_lease_targets(mut self, targets: Vec<(String, String, u64)>) -> Self { + self.publication_expectation = None; self.remote_publication_lease_targets = targets; self } @@ -973,7 +988,32 @@ impl ScannerCycleResult { &self.remote_publication_lease_targets } - pub(crate) fn acknowledge_durable_usage(self) -> Vec { + pub(crate) fn publication_expectation(&self) -> Option { + self.publication_expectation.clone() + } + + fn with_publication_expectation(mut self, expectation: Option) -> Self { + // Seal only after all coverage and acknowledgement inputs are final. + self.publication_expectation = expectation; + self + } + + pub(crate) fn acknowledge_durable_usage( + self, + proof: &crate::scanner::RootPublicationProof, + ) -> Vec { + if self.status != ScannerCycleStatus::Complete + || self + .publication_expectation + .as_ref() + .is_none_or(|expected| proof.verified_version_for(expected).is_none()) + { + return Vec::new(); + } + self.clear_verified_usage() + } + + fn clear_verified_usage(self) -> Vec { if let Some(snapshot) = self.dirty_usage_clear { clear_dirty_usage_buckets(&snapshot); } @@ -1013,6 +1053,7 @@ mod publish_gate_tests; #[cfg(test)] mod tests; +pub(crate) use cache::ScannerPublicationExpectation; use cache::*; use dirty_usage::*; use guards::*; diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index 5e47128b8..328b91ce9 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -308,6 +308,86 @@ impl<'a> ValidatedScannerSnapshot<'a> { } } +#[derive(Clone, Debug)] +pub(crate) struct ScannerPublicationExpectation { + candidate: Arc<([u8; 32], DataUsageScanPlanDigest)>, +} + +impl ScannerPublicationExpectation { + pub(crate) fn matches_encoded_candidate(&self, digest: &[u8; 32]) -> bool { + &self.candidate.0 == digest + } + + pub(crate) fn same_candidate(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.candidate, &other.candidate) && self.candidate.1 == other.candidate.1 + } +} + +pub(super) struct ValidatedUsageCandidate { + data: DataUsageInfo, + #[cfg(test)] + last_update: SystemTime, + coverage_digest: DataUsageScanPlanDigest, +} + +pub(super) fn empty_namespace_usage_candidate( + all_buckets: &[BucketInfo], + sources: &HashSet, + buckets_by_source: &HashMap>, + identity: ScannerSnapshotIdentity, +) -> Option { + if !all_buckets.is_empty() + || sources.is_empty() + || sources.len() != buckets_by_source.len() + || sources + .iter() + .any(|source| buckets_by_source.get(source).is_none_or(|buckets| !buckets.is_empty())) + { + return None; + } + let last_update = SystemTime::now(); + Some(ValidatedUsageCandidate { + data: DataUsageInfo { + last_update: Some(last_update), + scanner_cycle: Some(identity.cycle), + scanner_epoch: Some(identity.leader_epoch), + usage_snapshot_complete: true, + ..Default::default() + }, + #[cfg(test)] + last_update, + coverage_digest: identity.coverage_digest, + }) +} + +impl ValidatedUsageCandidate { + pub(super) fn prepare(mut self, status: ScannerCycleStatus) -> (DataUsageInfo, Option) { + self.data.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete); + let expectation = if status == ScannerCycleStatus::Complete { + struct DigestWriter(Sha256); + impl std::io::Write for DigestWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.update(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let mut writer = DigestWriter(Sha256::new()); + serde_json::to_writer(&mut writer, &self.data) + .ok() + .map(|()| ScannerPublicationExpectation { + candidate: Arc::new((writer.0.finalize().into(), self.coverage_digest)), + }) + } else { + None + }; + (self.data, expectation) + } +} + +#[cfg(test)] pub(super) fn completed_data_usage_info( results: &[DataUsageCache], scope: &ScannerSnapshotScope<'_>, @@ -316,6 +396,18 @@ pub(super) fn completed_data_usage_info( budget_elapsed: bool, cancelled: bool, ) -> Option<(DataUsageInfo, SystemTime)> { + completed_usage_candidate(results, scope, tier_registry_names, bucket_plan_complete, budget_elapsed, cancelled) + .map(|candidate| (candidate.data, candidate.last_update)) +} + +pub(super) fn completed_usage_candidate( + results: &[DataUsageCache], + scope: &ScannerSnapshotScope<'_>, + tier_registry_names: &[String], + bucket_plan_complete: bool, + budget_elapsed: bool, + cancelled: bool, +) -> Option { if !bucket_plan_complete { return None; } @@ -393,7 +485,12 @@ pub(super) fn completed_data_usage_info( usage_snapshot_set_states, ..Default::default() }; - Some((data_usage_info, merged_last_update)) + Some(ValidatedUsageCandidate { + data: data_usage_info, + #[cfg(test)] + last_update: merged_last_update, + coverage_digest: scope.identity.coverage_digest, + }) } fn tier_accounting_proof_is_publishable( diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 4eefe2910..d4094a311 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -338,12 +338,21 @@ where dirty_usage_status, activity_status, ); - let empty_usage = DataUsageInfo { - last_update: Some(SystemTime::now()), - scanner_cycle: Some(want_cycle), - usage_snapshot_complete: true, - ..Default::default() + let Some(candidate) = empty_namespace_usage_candidate( + &all_buckets, + &expected_sources, + &buckets_by_source, + ScannerSnapshotIdentity { + cycle: want_cycle, + leader_epoch, + plan_digest: scan_plan_digest, + coverage_digest: bucket_coverage_digest, + tier_registry_generation: Some(tier_registry_generation), + }, + ) else { + return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch)); }; + let (empty_usage, publication_expectation) = candidate.prepare(status); let observational_snapshot_published = if should_publish_observational_snapshot(status) { publish_observational_snapshot(&updates, empty_usage).await? } else { @@ -366,7 +375,8 @@ where .with_activity_digest(activity_digest) .with_observational_snapshot_published(observational_snapshot_published) .with_remote_publication_lease_targets(remote_publication_lease_targets) - .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)); + .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) + .with_publication_expectation(publication_expectation)); } let total_results = expected_sources.len(); @@ -595,7 +605,7 @@ where let (activity_status, remote_publication_lease_targets) = scanner_cycle_activity_status(store, distributed, &activity_before).await; let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::>(); - let completed_usage = completed_data_usage_info( + let completed_usage = completed_usage_candidate( &results, &ScannerSnapshotScope { sources: &expected_sources, @@ -636,7 +646,10 @@ where dirty_usage_status, activity_status, ); - let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage { + let mut publication_expectation = None; + let observational_snapshot_published = if let Some(candidate) = completed_usage { + let (data_usage_info, expectation) = candidate.prepare(cycle_status); + publication_expectation = expectation; if should_publish_observational_snapshot(cycle_status) { publish_observational_snapshot(&updates, data_usage_info).await? } else { @@ -674,5 +687,6 @@ where .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) .with_failed_dirty_usage(!failed_buckets.is_empty()) .with_pending_maintenance_work(pending_maintenance_work) - .with_required_cycle_floor(required_cycle_floor)) + .with_required_cycle_floor(required_cycle_floor) + .with_publication_expectation(publication_expectation)) } diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index e19fc2496..87e7b9c85 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -454,6 +454,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( .await .expect("initial object should persist"); } + wait_for_namespace_commit_tails(&store).await; let ctx = CancellationToken::new(); let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); let (updates, receiver) = mpsc::channel(1); @@ -503,6 +504,7 @@ async fn scoped_scan_same_cycle_maintenance_rewalks_after_root_delivery_failure( .put_object("cold-bucket", "new", &mut reader, &ScannerObjectOptions::default()) .await .expect("new cold object should persist"); + wait_for_namespace_commit_tails(&store).await; record_dirty_usage_bucket("hot-bucket"); if scan_mode == HealScanMode::Normal && !requires_full_scan { record_dirty_usage_bucket("cold-bucket"); @@ -994,8 +996,8 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() { assert!(dirty_usage_buckets().contains_key("temporarily-omitted")); assert_eq!(dirty_usage_snapshot_status(&snapshot), DirtyUsageSnapshotStatus::Current); - let acknowledgements = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())) - .acknowledge_durable_usage(); + let acknowledgements = + ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())).clear_verified_usage(); assert!(acknowledgements.is_empty()); assert!(!dirty_usage_buckets().contains_key("temporarily-omitted")); clear_dirty_usage_buckets_for_tests(); @@ -1101,7 +1103,7 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() { assert!(dirty_usage_buckets().contains_key("photos")); let confirmed = ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(snapshot.buckets.as_ref().clone())); - let acknowledgements = confirmed.acknowledge_durable_usage(); + let acknowledgements = confirmed.clear_verified_usage(); assert!(acknowledgements.is_empty()); assert!(!dirty_usage_buckets().contains_key("photos")); clear_dirty_usage_buckets_for_tests();