fix(scanner): persist partial progress after budget cancellation (#8578)

* fix(scanner): persist partial progress after budget cancellation Separate progress-only cycles from observational and authoritative usage updates. Keep peer movement leases alive under the parent cycle context for a bounded drain after the scan budget stops the walk, while preserving the existing strict publication fences for emitted usage updates. Skip root rename fences and the usage persister when no update was emitted. Cover real lease admission, cancellation, bounded retries, and durable partial-cycle advancement without usage publication or dirty acknowledgement. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(scanner): consolidate publication policy and ownership Classify emitted updates directly in the publication plan constructor and exercise that single policy from the namespace-admission tests. Share the parent cancellation state for read-only progress proof checks and move the one-use root fence into the persister, avoiding a child token allocation and a JSON string copy without changing lease or commit rules. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-27 06:18:43 +00:00 · 2026-10-15 23:27:26 +08:00
parent 69e987b4de
commit aae5cc4919
5 changed files with 390 additions and 81 deletions
@@ -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(&notification_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<T, E, F, Fut>(
+        self,
+        parent: &CancellationToken,
+        budget: &ScannerCycleBudget,
+        cycle: u64,
+        mut acquire: F,
+    ) -> ScannerPublicationProofWait<T, E>
+    where
+        E: std::fmt::Display,
+        F: FnMut(bool) -> Fut,
+        Fut: Future<Output = Result<T, E>>,
+    {
+        // 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)
@@ -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());

@@ -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();
+}
@@ -1426,7 +1426,7 @@ pub(crate) struct ScannerCycleResult {
    pub(crate) status: ScannerCycleStatus,
    publication_epoch: Option<u64>,
    activity_digest: Option<[u8; 32]>,
-    observational_snapshot_published: bool,
+    usage_update_emitted: bool,
    dirty_usage_clear: Option<DirtyUsageBuckets>,
    remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
    distributed_segment_invalidation_evidence: Option<DistributedSegmentInvalidationEvidence>,
@@ -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 {
@@ -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)