lywsvip/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-03 06:37:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): reuse recovered usage baseline for publication

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-30 00:05:11 +08:00
parent 3a4a6b0161
commit 2004001d47
3 changed files with 155 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
crates/scanner/src/scanner.rs
View File

@@ -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",

73
crates/scanner/src/scanner/tests.rs
View File

@@ -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::<DataUsageInfo>(&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() {

82
crates/scanner/src/scanner/usage_store.rs
View File

@@ -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<impl ScannerObjectIO>,
) -> Result<DataUsagePersistBaseline, EcstoreError> {
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::<DataUsageInfo>(&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::<DataUsageInfo>(&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::<DataUsageInfo>(&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",