fix(scanner): classify publication lease deferrals

Distinguish persistence budget and lease deadline deferrals from unavailable activity baselines, and ensure lease-gate deferrals update usage metrics. Keep the fixed lease gate fail-closed while storage-owned commit scope work remains pending.

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/scanner/src/scanner.rs

@@ -1205,6 +1205,10 @@ fn data_usage_persist_timeout() -> Duration {
     DataUsageCache::persistence_timeout()
 }
 
+fn scanner_publication_lease_budget_allows_persistence(timeout: Duration) -> bool {
+    timeout < Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS)
+}
+
 #[cfg(not(test))]
 const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30);
 #[cfg(test)]
@@ -1479,7 +1483,6 @@ async fn run_data_scanner_cycle_with_budget(
         Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await,
         Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
     };
-    let publication_deferred = publication_defer_reason.is_some();
     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
@@ -1493,11 +1496,11 @@ async fn run_data_scanner_cycle_with_budget(
     let mut remote_publication_leases = None;
     let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() {
         None
-    } else if usage_persist_timeout >= Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS) {
+    } else if !scanner_publication_lease_budget_allows_persistence(usage_persist_timeout) {
         // The lease is intentionally fixed-duration and has no renewal path.
         // Refuse a persistence budget that could outlive it instead of
         // allowing the peer to admit movement while a local PUT is in flight.
-        Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+        Some(ScannerCycleDeferReason::PublicationLeaseBudgetExceeded)
     } else if let Some(notification_system) = storeapi.notification_system() {
         match notification_system
             .acquire_scanner_publication_leases(remote_publication_lease_targets.clone())
@@ -1557,8 +1560,13 @@ async fn run_data_scanner_cycle_with_budget(
         .or(remote_lease_defer_reason)
         .or(remote_lease_fence_defer_reason);
     let publication_defer_reason = (!remote_lease_covers_persistence)
-        .then_some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
+        .then_some(ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded)
         .or(publication_defer_reason);
+    // Include reasons discovered while acquiring or validating remote leases.
+    // In particular, the static budget gate above is reached after the scan
+    // result is classified, so computing this flag earlier would suppress its
+    // deferred metric.
+    let publication_deferred = publication_defer_reason.is_some();
     let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled();
     let remote_lease_probe = remote_publication_leases
         .as_ref()

crates/scanner/src/scanner/tests.rs

@@ -4412,6 +4412,8 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
     for reason in [
         ScannerCycleDeferReason::DataMovement,
         ScannerCycleDeferReason::ActivityBaselineUnavailable,
+        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded,
+        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded,
     ] {
         let deferred = DataUsagePersistOutcome::Deferred(reason);
         assert_eq!(
@@ -4581,6 +4583,25 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() {
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
+#[test]
+fn scanner_publication_lease_budget_has_a_strict_ttl_boundary() {
+    let ttl = Duration::from_millis(SCANNER_PUBLICATION_LEASE_TTL_MS);
+
+    assert!(scanner_publication_lease_budget_allows_persistence(
+        ttl.saturating_sub(Duration::from_millis(1))
+    ));
+    assert!(!scanner_publication_lease_budget_allows_persistence(ttl));
+    assert!(!scanner_publication_lease_budget_allows_persistence(ttl + Duration::from_millis(1)));
+    assert_eq!(
+        ScannerCycleDeferReason::PublicationLeaseBudgetExceeded.as_str(),
+        "publication_lease_budget_exceeded"
+    );
+    assert_eq!(
+        ScannerCycleDeferReason::PublicationLeaseDeadlineExceeded.as_str(),
+        "publication_lease_deadline_exceeded"
+    );
+}
+
 #[tokio::test]
 async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
     let ctx = CancellationToken::new();

crates/scanner/src/scanner_io.rs

@@ -574,6 +574,14 @@ pub(crate) async fn scanner_set_disk_inventory(set: &SetDisks) -> Vec<Arc<Disk>>
 pub(crate) enum ScannerCycleDeferReason {
     ActivityBaselineUnavailable,
     DataMovement,
+    /// The configured persistence budget cannot fit within the fixed remote
+    /// publication-lease TTL. This is a deterministic configuration/contract
+    /// mismatch, not evidence that a peer activity probe failed.
+    PublicationLeaseBudgetExceeded,
+    /// A granted lease's absolute deadline cannot cover the persistence
+    /// operation. This can occur even when the configured budget fits the
+    /// nominal TTL because lease acquisition consumed part of the window.
+    PublicationLeaseDeadlineExceeded,
 }
 
 impl ScannerCycleDeferReason {
@@ -581,6 +589,8 @@ impl ScannerCycleDeferReason {
         match self {
             Self::ActivityBaselineUnavailable => "activity_baseline_unavailable",
             Self::DataMovement => "data_movement",
+            Self::PublicationLeaseBudgetExceeded => "publication_lease_budget_exceeded",
+            Self::PublicationLeaseDeadlineExceeded => "publication_lease_deadline_exceeded",
         }
     }
 }