From 2004001d47738e29f009b1e350841bba7aeac648 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 30 Aug 2026 00:05:11 +0800 Subject: [PATCH] fix(scanner): reuse recovered usage baseline for publication Co-Authored-By: heihutu --- crates/scanner/src/scanner.rs | 7 +- crates/scanner/src/scanner/tests.rs | 73 ++++++++++++++++++++ crates/scanner/src/scanner/usage_store.rs | 82 ++++++++++++++++++++++- 3 files changed, 155 insertions(+), 7 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index d4dfddb1a..788426b64 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1420,13 +1420,10 @@ async fn run_data_scanner_cycle_with_budget( mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement); }; - let usage_persist_baseline_result = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await; + let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await; drop(baseline_publication_guard); let usage_persist_baseline = match usage_persist_baseline_result { - Ok((data, revision)) => DataUsagePersistBaseline { - data: data.map(Bytes::from), - revision, - }, + Ok(baseline) => baseline, Err(err) => { error!( target: "rustfs::scanner", diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 88932cab9..c745c4c82 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -3324,6 +3324,79 @@ async fn test_observational_usage_defers_when_authoritative_baseline_is_missing( ); } +#[tokio::test] +async fn test_observational_usage_uses_fenced_backup_when_v2_primary_has_no_identity() { + let store = Arc::new(MemoryConfigStore::default()); + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + backup.scanner_epoch = Some(7); + backup.scanner_cycle = Some(103); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()), + serde_json::to_vec(&primary).expect("incomplete primary should encode"), + ); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())), + serde_json::to_vec(&backup).expect("backup baseline 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::(&observed).expect("observational snapshot should decode"); + assert_eq!(observed.usage_snapshot_authoritative_baseline, Some(backup.snapshot_identity())); +} + +#[tokio::test] +async fn usage_baseline_does_not_fall_back_to_older_legacy_snapshot() { + let store = Arc::new(MemoryConfigStore::default()); + let primary = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(100), + usage_snapshot_complete: false, + ..Default::default() + }; + let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0); + legacy.scanner_epoch = Some(6); + legacy.scanner_cycle = Some(103); + let primary_data = serde_json::to_vec(&primary).expect("incomplete primary should encode"); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()), + primary_data.clone(), + ); + store.objects.lock().await.insert( + memory_config_key(RUSTFS_META_BUCKET, LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + serde_json::to_vec(&legacy).expect("legacy baseline should encode"), + ); + + let baseline = read_data_usage_persist_baseline(store) + .await + .expect("baseline inspection should complete"); + assert_eq!(baseline.data.as_deref(), Some(primary_data.as_slice())); +} + #[tokio::test] #[serial] async fn test_usage_route_barrier_precedes_durable_reconciliation() { diff --git a/crates/scanner/src/scanner/usage_store.rs b/crates/scanner/src/scanner/usage_store.rs index 6edda880c..23bfe60d7 100644 --- a/crates/scanner/src/scanner/usage_store.rs +++ b/crates/scanner/src/scanner/usage_store.rs @@ -40,6 +40,84 @@ pub(super) struct DataUsagePersistBaseline { pub(super) revision: DataUsageCacheRevision, } +/// Read the bytes used as the baseline for a usage publication while keeping +/// the v2 primary revision as the CAS fence. During an interrupted upgrade the +/// primary can be valid JSON without a baseline identity; in that case a +/// same-or-newer durable companion may still be used, but an older legacy +/// snapshot must not cross the primary's epoch fence. +pub(super) async fn read_data_usage_persist_baseline( + storeapi: Arc, +) -> Result { + let (primary, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await?; + let Some(primary) = primary else { + for path in [ + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?; + let Some(candidate) = candidate else { + continue; + }; + let Ok(usage) = serde_json::from_slice::(&candidate) else { + continue; + }; + if data_usage_info_has_persisted_baseline_identity(&usage) { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(candidate)), + revision, + }); + } + } + return Ok(DataUsagePersistBaseline { data: None, revision }); + }; + + let Ok(primary_info) = serde_json::from_slice::(&primary) else { + // Preserve the original bytes and revision. A completed scan may + // replace the invalid primary under this CAS fence; an observation + // will still reject it below because it has no verifiable identity. + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }); + }; + if data_usage_info_has_persisted_baseline_identity(&primary_info) || data_usage_info_is_bootstrap_pending(&primary_info) { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }); + } + + let invalid_primary_epoch = primary_info.scanner_epoch; + for path in [ + format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), + LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(), + format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()), + ] { + let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?; + let Some(candidate) = candidate else { + continue; + }; + let Ok(usage) = serde_json::from_slice::(&candidate) else { + continue; + }; + let candidate_epoch = usage.scanner_epoch.unwrap_or_default(); + if data_usage_info_has_persisted_baseline_identity(&usage) + && invalid_primary_epoch.is_none_or(|epoch| candidate_epoch >= epoch) + { + return Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(candidate)), + revision, + }); + } + } + + Ok(DataUsagePersistBaseline { + data: Some(Bytes::from(primary)), + revision, + }) +} + /// Short-lived publication inputs captured for one usage persistence attempt. /// Keeping the movement epoch, lease deadline, and target fence together makes /// it explicit that they are one proof rather than independent options. @@ -281,8 +359,8 @@ where publication_epoch = Some(read_epoch); let authoritative_data = match next_baseline.as_ref() { Some(baseline) => baseline.data.clone(), - None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await { - Ok((data, _)) => data.map(Bytes::from), + None => match read_data_usage_persist_baseline(storeapi.clone()).await { + Ok(baseline) => baseline.data, Err(err) => { error!( target: "rustfs::scanner",