fix(scanner): fit usage persistence within publication lease (#6967)

Lower the default scanner cache save timeout so the derived usage persistence budget stays inside the effective distributed publication lease window.

Add focused regressions for the default publication budget and bootstrap-pending observational baselines, and update operator docs with the new default.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/config/src/constants/scanner.rs

@@ -198,11 +198,11 @@ pub const ENV_SCANNER_IDLE_MODE: &str = "RUSTFS_SCANNER_IDLE_MODE";
 /// Environment variable that controls scanner cache save timeout in seconds.
 /// The scanner enforces a minimum value of `1`.
 /// - Unit: seconds (u64).
-/// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=30`
+/// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=14`
 pub const ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS: &str = "RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS";
 
 /// Default scanner cache save timeout in seconds.
-pub const DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS: u64 = 30;
+pub const DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS: u64 = 14;
 
 /// Environment variable that caps concurrent scanner set tasks.
 /// A value of `0` keeps the existing topology-based concurrency.

crates/scanner/src/data_usage_define/tests.rs

@@ -2241,6 +2241,7 @@ fn test_cache_save_timeout_uses_default_when_env_missing() {
             DataUsageCache::cache_save_timeout(),
             Duration::from_secs(rustfs_config::DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS)
         );
+        assert_eq!(DataUsageCache::persistence_timeout(), Duration::from_millis(52_350));
     });
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }

crates/scanner/src/scanner/tests.rs

@@ -4507,6 +4507,47 @@ async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_iden
     assert_eq!(observed.usage_snapshot_authoritative_baseline, Some(backup.snapshot_identity()));
 }
 
+#[tokio::test]
+#[serial]
+async fn test_observational_usage_uses_bootstrap_pending_primary_as_baseline() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let primary = DataUsageInfo {
+        last_update: Some(std::time::SystemTime::UNIX_EPOCH),
+        scanner_epoch: Some(7),
+        usage_snapshot_converged: Some(false),
+        usage_snapshot_bootstrap_pending: true,
+        ..Default::default()
+    };
+    assert!(data_usage_info_is_bootstrap_pending(&primary));
+    store.objects.lock().await.insert(
+        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
+        serde_json::to_vec(&primary).expect("bootstrap primary should encode"),
+    );
+
+    let (sender, receiver) = mpsc::channel(1);
+    let mut observation = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1);
+    observation.usage_snapshot_converged = Some(false);
+    sender.send(observation).await.expect("observation should enqueue");
+    drop(sender);
+
+    let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+        CancellationToken::new(),
+        store.clone(),
+        receiver,
+        None,
+        None,
+        || async { false },
+    )
+    .await;
+
+    assert_eq!(outcome, DataUsagePersistOutcome::Saved);
+    let observed = read_config(store, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())
+        .await
+        .expect("observational snapshot should be persisted");
+    let observed = serde_json::from_slice::<DataUsageInfo>(&observed).expect("observational snapshot should decode");
+    assert_eq!(observed.usage_snapshot_authoritative_baseline, Some(primary.snapshot_identity()));
+}
+
 #[tokio::test]
 async fn usage_baseline_does_not_fall_back_to_older_legacy_snapshot() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -5924,6 +5965,21 @@ fn data_usage_persist_wait_covers_cache_retries_and_backup() {
     crate::runtime_config::refresh_scanner_runtime_config_for_tests();
 }
 
+#[test]
+#[serial]
+fn default_data_usage_persist_wait_fits_publication_lease_window() {
+    with_var_unset(rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, || {
+        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
+        let effective_publication_lease_window =
+            Duration::from_millis(crate::storage_api::ECSTORE_SCANNER_PUBLICATION_LEASE_TTL_MS)
+                .saturating_sub(Duration::from_secs(5));
+
+        assert_eq!(data_usage_persist_timeout(), Duration::from_millis(52_350));
+        assert!(data_usage_persist_timeout() < effective_publication_lease_window);
+    });
+    crate::runtime_config::refresh_scanner_runtime_config_for_tests();
+}
+
 #[tokio::test]
 async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
     let ctx = CancellationToken::new();

crates/scanner/src/storage_api.rs

@@ -111,6 +111,8 @@ pub(crate) use rustfs_ecstore::api::runtime::{
 pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
 pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
 #[cfg(test)]
+pub(crate) use rustfs_ecstore::api::storage::SCANNER_PUBLICATION_LEASE_TTL_MS as ECSTORE_SCANNER_PUBLICATION_LEASE_TTL_MS;
+#[cfg(test)]
 pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
 use rustfs_storage_api as storage_contracts;