diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 8e2b62a92..dcf0f4d3c 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1907,6 +1907,10 @@ where }, ) .await; + let publication_plan = scan_result + .as_ref() + .map(ScannerCyclePublicationPlan::for_result) + .unwrap_or(ScannerCyclePublicationPlan::ProgressOnly); let publication_defer_reason = match &scan_result { Ok(result) if result @@ -1921,7 +1925,7 @@ where // fence remains enforced by the usage store and the cycle is advanced // as partial without acknowledging dirty usage. Ok(result) - if scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot()) + if publication_plan.is_observational() && matches!( result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) @@ -1930,18 +1934,11 @@ where None } Ok(result) => { - final_data_usage_publication_defer_reason( - storeapi.as_ref(), - result.status, - scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot()), - ) - .await + final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status, publication_plan.is_observational()).await } Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), }; - let observational_only = scan_result - .as_ref() - .is_ok_and(|result| scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot())); + let observational_only = publication_plan.is_observational(); let publication_epoch = scan_result.as_ref().ok().and_then(ScannerCycleResult::publication_epoch); let remote_publication_lease_targets = if publication_defer_reason.is_none() { scan_result @@ -1956,16 +1953,12 @@ where let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() { None } else if let Some(notification_system) = storeapi.scanner_notification_system() { - let publication_proof_ctx = cycle_budget.token(); - let lease_result = await_scanner_publication_proof( - &publication_proof_ctx, - cycle_info.current, - "lease_acquire", - || { + let lease_result = publication_plan + .await_leases(ctx, &cycle_budget, cycle_info.current, |observational_lease| { let targets = remote_publication_lease_targets.clone(); let notification_system = Arc::clone(¬ification_system); async move { - if observational_only { + if observational_lease { notification_system .acquire_observational_scanner_publication_leases(targets) .await @@ -1973,10 +1966,8 @@ where notification_system.acquire_scanner_publication_leases(targets).await } } - }, - |err| scanner_publication_lease_error_is_retryable(&err.to_string()), - ) - .await; + }) + .await; match lease_result { ScannerPublicationProofWait::Ready(grants) => { remote_publication_leases = Some((notification_system, grants)); @@ -1996,34 +1987,40 @@ where // it is never inserted into FileInfo metadata. Keep the representation // bounded and require one authenticated token per remote target so a // partial grant can never silently fall back to an unfenced rename. - let remote_lease_fence = remote_publication_leases.as_ref().and_then(|(_, grants)| { - if grants.len() != remote_publication_lease_targets.len() || grants.len() > SCANNER_PUBLICATION_LEASE_FENCE_MAX_ENTRIES { - return None; - } - let mut fence = BTreeMap::new(); - for grant in grants { - if grant.host.is_empty() || grant.host.len() > 1024 { + let remote_lease_fence = remote_publication_leases + .as_ref() + .filter(|_| publication_plan.has_usage_update()) + .and_then(|(_, grants)| { + if grants.len() != remote_publication_lease_targets.len() + || grants.len() > SCANNER_PUBLICATION_LEASE_FENCE_MAX_ENTRIES + { return None; } - if fence.insert(grant.host.clone(), grant.lease.token.to_string()).is_some() { + let mut fence = BTreeMap::new(); + for grant in grants { + if grant.host.is_empty() || grant.host.len() > 1024 { + return None; + } + if fence.insert(grant.host.clone(), grant.lease.token.to_string()).is_some() { + return None; + } + } + if remote_publication_lease_targets + .iter() + .any(|(host, _, _)| !fence.contains_key(host)) + { return None; } - } - if remote_publication_lease_targets - .iter() - .any(|(host, _, _)| !fence.contains_key(host)) - { - return None; - } - serde_json::to_string(&fence) - .ok() - .filter(|encoded| encoded.len() <= SCANNER_PUBLICATION_LEASE_FENCE_MAX_BYTES) - }); - let remote_lease_fence_defer_reason = (remote_publication_lease_fence_targets_are_required( - remote_publication_lease_targets.len(), - remote_publication_leases.is_some(), - remote_lease_fence.is_some(), - )) + serde_json::to_string(&fence) + .ok() + .filter(|encoded| encoded.len() <= SCANNER_PUBLICATION_LEASE_FENCE_MAX_BYTES) + }); + let remote_lease_fence_defer_reason = (publication_plan.has_usage_update() + && remote_publication_lease_fence_targets_are_required( + remote_publication_lease_targets.len(), + remote_publication_leases.is_some(), + remote_lease_fence.is_some(), + )) .then_some(ScannerCycleDeferReason::ActivityBaselineUnavailable); let publication_defer_reason = publication_defer_reason .or(remote_lease_defer_reason) @@ -2046,6 +2043,7 @@ where let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled(); let remote_lease_probe = remote_publication_leases .as_ref() + .filter(|_| publication_plan.has_usage_update()) .map(|(notification_system, grants)| (Arc::clone(notification_system), grants.clone())); let remote_lease_release_safe = Arc::new(AtomicBool::new(true)); let mut usage_publication_result = match publication_defer_reason { @@ -2053,6 +2051,10 @@ where drop(receiver); DataUsagePublicationResult::from(DataUsagePersistOutcome::Deferred(reason)) } + None if !publication_plan.has_usage_update() => { + drop(receiver); + DataUsagePublicationResult::from(DataUsagePersistOutcome::NoUpdate) + } None => { // ScannerIO emits its complete or observational update only after // all set workers finish. Persist after the final activity fence; @@ -2061,7 +2063,6 @@ where let ctx_clone = ctx.clone(); let route_probe_store = storeapi.clone(); let route_probe_observational_only = observational_only; - let remote_lease_fence = remote_lease_fence.clone(); let remote_lease_release_safe_for_task = Arc::clone(&remote_lease_release_safe); let remote_lease_tokens = remote_publication_leases .as_ref() @@ -2363,7 +2364,7 @@ where } usage_publication_result.restrict_outcome(usage_persist_outcome); - let partial_cause = if scan_cycle_result.has_observational_snapshot() + let partial_cause = if scan_cycle_result.has_usage_update() && matches!( scan_cycle_result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) @@ -3599,16 +3600,76 @@ where } } -fn scanner_cycle_uses_observational_publication(status: ScannerCycleStatus, has_published_snapshot: bool) -> bool { - // Incomplete cycles may emit only a non-authoritative snapshot. Its lease - // must allow ordinary namespace commits while retaining the movement fence. - has_published_snapshot - && matches!( - status, - ScannerCycleStatus::Superseded +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ScannerCyclePublicationPlan { + ProgressOnly, + Observational, + Authoritative, +} + +impl ScannerCyclePublicationPlan { + fn for_result(result: &ScannerCycleResult) -> Self { + match (result.has_usage_update(), result.status) { + (false, _) => Self::ProgressOnly, + ( + true, + ScannerCycleStatus::Superseded | ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) - | ScannerCycleStatus::Incomplete - ) + | ScannerCycleStatus::Incomplete, + ) => Self::Observational, + (true, _) => Self::Authoritative, + } + } + + fn has_usage_update(self) -> bool { + self != Self::ProgressOnly + } + + fn is_observational(self) -> bool { + self == Self::Observational + } + + fn uses_observational_lease(self) -> bool { + self != Self::Authoritative + } + + async fn await_leases( + self, + parent: &CancellationToken, + budget: &ScannerCycleBudget, + cycle: u64, + mut acquire: F, + ) -> ScannerPublicationProofWait + where + E: std::fmt::Display, + F: FnMut(bool) -> Fut, + Fut: Future>, + { + // Exhausting the walk budget must not cancel the movement proof needed + // to save partial progress. No usage update means namespace stability + // is unnecessary, but peer session/generation and movement checks stay. + let proof_ctx = if self == Self::ProgressOnly { + parent.clone() + } else { + budget.token() + }; + let proof = await_scanner_publication_proof( + &proof_ctx, + cycle, + "lease_acquire", + || acquire(self.uses_observational_lease()), + |err| scanner_publication_lease_error_is_retryable(&err.to_string()), + ); + if self == Self::ProgressOnly { + // Object/directory budgets can expire with no runtime deadline. + // Bound the control-plane drain independently of the walk budget. + tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, proof) + .await + .unwrap_or(ScannerPublicationProofWait::Cancelled) + } else { + proof.await + } + } } fn scanner_post_lease_activity_defer_reason( @@ -3807,7 +3868,7 @@ fn scanner_cycle_completion_outcome_for_result( ) -> ScannerCycleOutcome { let has_dirty_usage = scan_cycle_result.has_dirty_usage_to_acknowledge(); let has_failed_dirty_usage = scan_cycle_result.has_failed_dirty_usage(); - if scan_cycle_result.has_observational_snapshot() + if scan_cycle_result.has_usage_update() && matches!( scan_cycle_result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 91b691f6f..a86979458 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 cycle_budget_publication; pub(super) mod cycle_persist_failure; mod quota_reset_preservation; @@ -1689,7 +1690,11 @@ async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() let (_temp_dir, store) = setup_scanner_cycle_store().await; let pending = ecstore_hold_namespace_commit(store.as_ref()); - let complete_uses_observation = scanner_cycle_uses_observational_publication(ScannerCycleStatus::Complete, true); + let uses_observation = |status| { + ScannerCyclePublicationPlan::for_result(&ScannerCycleResult::new(status, None).with_usage_update_emitted(true)) + .is_observational() + }; + let complete_uses_observation = uses_observation(ScannerCycleStatus::Complete); assert!(!complete_uses_observation, "a converged Complete snapshot is authoritative"); assert_eq!( scanner_local_publication_defer_reason(store.as_ref(), complete_uses_observation).await, @@ -1697,15 +1702,11 @@ async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() "a Complete cycle must remain fenced while an ordinary namespace commit is pending" ); - assert!(scanner_cycle_uses_observational_publication(ScannerCycleStatus::Superseded, true)); - assert!(scanner_cycle_uses_observational_publication( - ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), - true - )); - assert!(!scanner_cycle_uses_observational_publication( - ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), - true - )); + assert!(uses_observation(ScannerCycleStatus::Superseded)); + assert!(uses_observation(ScannerCycleStatus::Deferred( + ScannerCycleDeferReason::ActivityBaselineUnavailable + ))); + assert!(!uses_observation(ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement))); drop(pending); } @@ -1714,10 +1715,11 @@ async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() async fn incomplete_observation_uses_observational_lease_during_pending_namespace_commit() { let (_temp_dir, store) = setup_scanner_cycle_store().await; let pending = ecstore_hold_namespace_commit(store.as_ref()); - let uses_observation = scanner_cycle_uses_observational_publication(ScannerCycleStatus::Incomplete, true); + let result = ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None); + assert!(!ScannerCyclePublicationPlan::for_result(&result).is_observational()); + let uses_observation = ScannerCyclePublicationPlan::for_result(&result.with_usage_update_emitted(true)).is_observational(); assert!(uses_observation); - assert!(!scanner_cycle_uses_observational_publication(ScannerCycleStatus::Incomplete, false)); assert_eq!( scanner_local_publication_defer_reason(store.as_ref(), uses_observation).await, None, @@ -7885,7 +7887,7 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() { ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), Some(dirty_snapshot), ) - .with_observational_snapshot_published(true); + .with_usage_update_emitted(true); let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(observed, DataUsagePersistOutcome::Saved.into()); diff --git a/crates/scanner/src/scanner/tests/cycle_budget_publication.rs b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs new file mode 100644 index 000000000..ee292e12e --- /dev/null +++ b/crates/scanner/src/scanner/tests/cycle_budget_publication.rs @@ -0,0 +1,246 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; + +#[tokio::test] +#[serial] +async fn exhausted_budget_still_proves_peer_movement_without_namespace_authority() { + let (_temp_dir, store) = setup_scanner_cycle_store_with_pool_count(true, 2).await; + let pending = ecstore_hold_namespace_commit(store.as_ref()); + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + budget.cancel_for_runtime(); + let generation = store.scanner_data_movement_generation(); + let result = ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_remote_publication_lease_targets(vec![( + "peer".into(), + "session".into(), + generation, + )]); + let plan = ScannerCyclePublicationPlan::for_result(&result); + let calls = AtomicUsize::new(0); + + let proof = plan + .await_leases(&ctx, &budget, 352, |observational| { + calls.fetch_add(1, Ordering::Relaxed); + store.acquire_scanner_publication_lease_with_purpose(generation, Duration::from_secs(60), observational) + }) + .await; + let ScannerPublicationProofWait::Ready((token, admitted_generation)) = proof else { + panic!("a stopped walk must still prove peer movement while the leader context is alive"); + }; + assert_eq!(calls.load(Ordering::Relaxed), 1); + assert!(!ctx.is_cancelled()); + store + .validate_scanner_publication_lease_with_purpose(token, admitted_generation, true) + .await + .expect("progress must retain a movement-only lease"); + assert!( + store + .validate_scanner_publication_lease_with_purpose(token, admitted_generation, false) + .await + .is_err(), + "the lease must not authorize an authoritative snapshot" + ); + assert!(store.release_scanner_publication_lease(token).await); + + let stale = plan + .await_leases(&ctx, &budget, 352, |observational| { + store.acquire_scanner_publication_lease_with_purpose(generation + 1, Duration::from_secs(60), observational) + }) + .await; + assert!(matches!(stale, ScannerPublicationProofWait::Rejected(_))); + drop(pending); + + store.pool_meta.write().await.pools[0].decommission = Some(crate::storage_api::owner::EcstorePoolDecommissionInfo { + failed: true, + ..Default::default() + }); + assert!(store.scanner_data_movement_pause_status().await.paused); + let current_generation = store.scanner_data_movement_generation(); + let moving = plan + .await_leases(&ctx, &budget, 352, |observational| { + store.acquire_scanner_publication_lease_with_purpose(current_generation, Duration::from_secs(60), observational) + }) + .await; + assert!( + matches!(moving, ScannerPublicationProofWait::Rejected(error) if error.to_string().contains("blocked by data movement")) + ); +} + +#[tokio::test] +async fn emitted_updates_keep_budget_cancellation_and_publication_purpose() { + for (status, expected_plan) in [ + (ScannerCycleStatus::Complete, ScannerCyclePublicationPlan::Authoritative), + (ScannerCycleStatus::Incomplete, ScannerCyclePublicationPlan::Observational), + (ScannerCycleStatus::Superseded, ScannerCyclePublicationPlan::Observational), + ( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), + ScannerCyclePublicationPlan::Observational, + ), + ( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), + ScannerCyclePublicationPlan::Authoritative, + ), + ] { + let result = ScannerCycleResult::new(status, None).with_usage_update_emitted(true); + let plan = ScannerCyclePublicationPlan::for_result(&result); + assert_eq!(plan, expected_plan); + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let ready = plan + .await_leases(&ctx, &budget, 352, |observational| async move { Ok::<_, String>(observational) }) + .await; + assert!(matches!(ready, ScannerPublicationProofWait::Ready(observational) if observational == plan.is_observational())); + budget.cancel_for_runtime(); + let calls = AtomicUsize::new(0); + let cancelled = plan + .await_leases(&ctx, &budget, 352, |_| { + calls.fetch_add(1, Ordering::Relaxed); + async { Ok::<(), String>(()) } + }) + .await; + assert!(matches!(cancelled, ScannerPublicationProofWait::Cancelled)); + assert_eq!(calls.load(Ordering::Relaxed), 0); + } +} + +#[tokio::test] +async fn progress_proof_honors_parent_cancellation() { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + ctx.cancel(); + let calls = AtomicUsize::new(0); + let proof = ScannerCyclePublicationPlan::ProgressOnly + .await_leases(&ctx, &budget, 352, |_| { + calls.fetch_add(1, Ordering::Relaxed); + async { Ok::<_, String>(()) } + }) + .await; + assert!(matches!(proof, ScannerPublicationProofWait::Cancelled)); + assert_eq!(calls.load(Ordering::Relaxed), 0); +} + +#[tokio::test(start_paused = true)] +async fn progress_proof_is_bounded_without_a_runtime_budget() { + for stalled in [false, true] { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new( + &ctx, + ScannerCycleBudgetConfig { + max_objects: Some(1), + ..Default::default() + }, + ); + budget.record_object_scanned(); + assert!(budget.budget_elapsed()); + let calls = AtomicUsize::new(0); + let started = Instant::now(); + let proof = ScannerCyclePublicationPlan::ProgressOnly + .await_leases(&ctx, &budget, 352, |_| { + calls.fetch_add(1, Ordering::Relaxed); + async move { + if stalled { + std::future::pending::<()>().await; + } + Err::<(), _>("scanner publication lease capacity is exhausted") + } + }) + .await; + assert!(matches!(proof, ScannerPublicationProofWait::Cancelled)); + assert!(calls.load(Ordering::Relaxed) > 0); + assert_eq!(started.elapsed(), SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT); + assert!(!ctx.is_cancelled()); + } +} + +#[tokio::test] +#[serial] +async fn coordinator_persists_budget_progress_without_usage_or_dirty_acknowledgement() { + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let bucket = "scanner-budget-progress"; + store + .make_bucket(bucket, &crate::storage_api::scan::MakeBucketOptions::default()) + .await + .expect("fixture bucket"); + for object in ["first", "second", "third"] { + store.pools[0].disk_set[0] + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(b"value".to_vec()), + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("fixture object"); + } + crate::scanner_io::record_dirty_usage_bucket(bucket); + let dirty_before = crate::scanner_io::dirty_usage_buckets_for_tests(); + let baseline = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("usage baseline"); + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking( + &ctx, + ScannerCycleBudgetConfig { + max_objects: Some(1), + ..Default::default() + }, + ); + let mut cycle = CurrentCycle { + next: 352, + ..Default::default() + }; + let mut revision = DataUsageCacheRevision::Missing; + let outcome = run_data_scanner_cycle_with_budget( + &ctx, + &store, + &mut cycle, + &mut revision, + 1, + Arc::clone(&budget), + ScannerCycleScheduling { + requires_full_scan: true, + service_cohort: None, + }, + ) + .await; + assert!(matches!(outcome, ScannerCycleOutcome::Partial)); + assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Objects)); + assert!(budget.progress().0 > 0); + assert!(budget.cycle_state_persisted()); + assert_eq!(cycle.next, 353); + assert!(cycle.cycle_completed.is_empty()); + assert!(matches!(revision, DataUsageCacheRevision::Etag(_))); + let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH) + .await + .expect("durable cycle progress"); + let (persisted, epoch) = decode_scanner_cycle_state(&state).expect("valid cycle state"); + assert_eq!(persisted.next, 353); + assert_eq!(epoch, 1); + assert_eq!( + read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) + .await + .expect("unchanged root"), + baseline + ); + assert!(read_config(store, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await.is_err()); + assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before); + global_metrics().set_cycle(None).await; + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); +} diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 6d0ee7b8f..7c2e169cc 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -1426,7 +1426,7 @@ pub(crate) struct ScannerCycleResult { pub(crate) status: ScannerCycleStatus, publication_epoch: Option, activity_digest: Option<[u8; 32]>, - observational_snapshot_published: bool, + usage_update_emitted: bool, dirty_usage_clear: Option, remote_dirty_usage_acknowledgements: Vec, distributed_segment_invalidation_evidence: Option, @@ -1444,7 +1444,7 @@ impl ScannerCycleResult { status, publication_epoch: None, activity_digest: None, - observational_snapshot_published: false, + usage_update_emitted: false, dirty_usage_clear, remote_dirty_usage_acknowledgements: Vec::new(), distributed_segment_invalidation_evidence: None, @@ -1477,14 +1477,14 @@ impl ScannerCycleResult { self.activity_digest } - pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self { + pub(crate) fn with_usage_update_emitted(mut self, emitted: bool) -> Self { self.publication_expectation = None; - self.observational_snapshot_published = published; + self.usage_update_emitted = emitted; self } - pub(crate) fn has_observational_snapshot(&self) -> bool { - self.observational_snapshot_published + pub(crate) fn has_usage_update(&self) -> bool { + self.usage_update_emitted } fn with_failed_dirty_usage(mut self, failed_dirty_usage: bool) -> Self { diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 577cb1094..e32a8dcef 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -457,12 +457,12 @@ where 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) { + let usage_update_emitted = if should_publish_observational_snapshot(status) { publish_observational_snapshot(&updates, empty_usage).await? } else { publish_usage_snapshot(&updates, status, empty_usage).await? }; - if !observational_snapshot_published { + if !usage_update_emitted { return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch)); } if status == ScannerCycleStatus::Complete { @@ -484,7 +484,7 @@ where return Ok(ScannerCycleResult::new(status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) - .with_observational_snapshot_published(observational_snapshot_published) + .with_usage_update_emitted(usage_update_emitted) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) .with_segment_reuse_activation_preflight(segment_reuse_activation_preflight) @@ -778,7 +778,7 @@ where activity_status, ); let mut publication_expectation = None; - let observational_snapshot_published = if let Some(candidate) = completed_usage { + let usage_update_emitted = 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) { @@ -816,7 +816,7 @@ where Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear) .with_publication_epoch(publication_epoch) .with_activity_digest(activity_digest) - .with_observational_snapshot_published(observational_snapshot_published) + .with_usage_update_emitted(usage_update_emitted) .with_remote_publication_lease_targets(remote_publication_lease_targets) .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) .with_distributed_segment_invalidation_evidence(distributed_segment_invalidation_evidence)