gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 00:17:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): rebuild missing usage floor after upgrade (#6624)

Browse Source 
* fix(scanner): rebuild missing usage floor after upgrade

* fix(scanner): preserve missing-floor reset across conflicts

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
This commit is contained in:
Henry Guo
2026-08-26 12:30:53 +08:00
committed by GitHub
parent c7f201e6cb
commit 1590d9107b
4 changed files with 293 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner.rs
+54 -67
View File
@@ -403,7 +403,7 @@ pub(super) fn data_usage_info_has_persisted_baseline_identity(info: &DataUsageIn
&& u64::try_from(info.buckets_usage.len()).ok() == Some(info.buckets_count)
}
pub(super) fn data_usage_info_is_pristine_bootstrap_pending(info: &DataUsageInfo) -> bool {
pub(super) fn data_usage_info_is_bootstrap_pending(info: &DataUsageInfo) -> bool {
if info.last_update.is_none() || info.scanner_cycle.is_some() {
return false;
}
@@ -681,27 +681,30 @@ async fn initial_scanner_startup_usage_state(storeapi: &Arc<ECStore>) -> (bool,
(persisted_usage_cache_is_cold_for_startup(storeapi).await, has_buckets)
}
fn scanner_cycle_state_is_pristine(
cycle_info: &CurrentCycle,
leader_epoch: u64,
cycle_revision: &DataUsageCacheRevision,
) -> bool {
cycle_info.next == 0 && leader_epoch == 0 && matches!(cycle_revision, DataUsageCacheRevision::Missing)
}
fn scanner_may_bootstrap_missing_usage_floor(
cycle_info: &CurrentCycle,
leader_epoch: u64,
cycle_revision: &DataUsageCacheRevision,
) -> bool {
// The server becomes ready before the scanner starts, so a first bucket may
// already exist. The bootstrap marker is non-authoritative; only prior
// durable scanner progress must block its creation.
scanner_cycle_state_is_pristine(cycle_info, leader_epoch, cycle_revision)
}
fn scanner_may_resume_pristine_usage_bootstrap(cycle_info: &CurrentCycle) -> bool {
cycle_info.next == 0
fn prepare_cycle_for_usage_floor_bootstrap(
cycle_info: &mut CurrentCycle,
usage_floor: PersistedUsageFloor,
startup: PersistedUsageFloorStartup,
) -> (bool, bool) {
match startup {
PersistedUsageFloorStartup::Authoritative => (false, false),
PersistedUsageFloorStartup::Missing => {
// Cycle progress without its corresponding usage floor cannot
// prove namespace coverage. Restart from cycle zero while keeping
// the separately fenced leader epoch monotonic.
*cycle_info = CurrentCycle::default();
(true, true)
}
PersistedUsageFloorStartup::BootstrapPending => {
// An unfenced marker may have been written before an upgrade's old
// cycle state was replaced. A fenced marker belongs to the current
// scanner generation and may retain partial-cycle progress.
if usage_floor.leader_epoch == 0 {
*cycle_info = CurrentCycle::default();
}
(true, usage_floor.leader_epoch == 0)
}
}
}
pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
@@ -1220,7 +1223,7 @@ where
LockLost: Future<Output = ()>,
{
let fence_ctx = ctx.child_token();
let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false);
let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false, false);
tokio::pin!(claim);
tokio::pin!(lock_lost);
tokio::select! {
@@ -2186,47 +2189,30 @@ async fn run_data_scanner_with_maintenance_state(
return Err(err);
}
};
let may_bootstrap_missing_usage_floor = scanner_may_bootstrap_missing_usage_floor(&cycle_info, leader_epoch, &cycle_revision);
let (usage_floor, usage_floor_startup) =
match persisted_usage_floor_for_startup(storeapi.clone(), may_bootstrap_missing_usage_floor).await {
Ok(result) => result,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "usage_floor_load_failed",
error = %err,
"Scanner stopped because the persisted usage floor could not be loaded"
);
global_metrics().set_cycle(None).await;
return Ok(());
}
};
if usage_floor_startup == PersistedUsageFloorStartup::BootstrapPending
&& !scanner_may_resume_pristine_usage_bootstrap(&cycle_info)
{
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "usage_floor_bootstrap_conflict",
next_cycle = cycle_info.next,
"Scanner stopped because a pristine usage bootstrap conflicts with persisted cycle progress"
);
global_metrics().set_cycle(None).await;
return Ok(());
}
let (usage_floor, usage_floor_startup) = match persisted_usage_floor_for_startup(storeapi.clone(), true).await {
Ok(result) => result,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "usage_floor_load_failed",
error = %err,
"Scanner stopped because the persisted usage floor could not be loaded"
);
global_metrics().set_cycle(None).await;
return Ok(());
}
};
let (allow_usage_floor_bootstrap_pending, reset_usage_floor_bootstrap_cycle_on_conflict) =
prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, usage_floor_startup);
apply_persisted_usage_floor(&mut cycle_info, &mut leader_epoch, usage_floor);
let allow_pristine_bootstrap_pending = match usage_floor_startup {
PersistedUsageFloorStartup::Authoritative => false,
PersistedUsageFloorStartup::BootstrapPending => true,
match usage_floor_startup {
PersistedUsageFloorStartup::Authoritative | PersistedUsageFloorStartup::BootstrapPending => {}
PersistedUsageFloorStartup::Missing => {
if !may_bootstrap_missing_usage_floor || ctx.is_cancelled() || guard.is_lock_lost() {
if ctx.is_cancelled() || guard.is_lock_lost() {
global_metrics().set_cycle(None).await;
return Ok(());
}
@@ -2234,12 +2220,12 @@ async fn run_data_scanner_with_maintenance_state(
let bootstrap_ctx = ctx.child_token();
match await_scanner_cycle_with_lock_fence(
&bootstrap_ctx,
initialize_pristine_usage_baseline(storeapi.clone()),
initialize_usage_baseline_bootstrap(storeapi.clone()),
guard.lock_lost_notified(),
)
.await
{
Some(Ok(())) => true,
Some(Ok(())) => {}
Some(Err(err)) => {
error!(
target: "rustfs::scanner",
@@ -2249,7 +2235,7 @@ async fn run_data_scanner_with_maintenance_state(
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "usage_floor_bootstrap_failed",
error = %err,
"Scanner stopped because the pristine usage bootstrap could not be initialized"
"Scanner stopped because the usage baseline bootstrap could not be initialized"
);
global_metrics().set_cycle(None).await;
return Ok(());
@@ -2260,7 +2246,7 @@ async fn run_data_scanner_with_maintenance_state(
}
}
}
};
}
if ctx.is_cancelled() || guard.is_lock_lost() {
global_metrics().set_cycle(None).await;
@@ -2275,7 +2261,8 @@ async fn run_data_scanner_with_maintenance_state(
&mut cycle_info,
&mut cycle_revision,
&mut leader_epoch,
allow_pristine_bootstrap_pending,
allow_usage_floor_bootstrap_pending,
reset_usage_floor_bootstrap_cycle_on_conflict,
),
guard.lock_lost_notified(),
)
crates/scanner/src/scanner/cycle_state.rs
+9 -11
View File
@@ -1319,7 +1319,7 @@ pub(super) async fn persisted_usage_floor(
pub(super) async fn persisted_usage_floor_for_startup(
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
allow_missing_for_pristine_startup: bool,
allow_missing_for_bootstrap: bool,
) -> Result<(PersistedUsageFloor, PersistedUsageFloorStartup), ScannerError> {
let Some(read_epoch) = scanner_publication_epoch(storeapi.clone()).await else {
return Err(ScannerError::Other("scanner usage floor read is blocked by data movement".to_string()));
@@ -1345,11 +1345,9 @@ pub(super) async fn persisted_usage_floor_for_startup(
let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
ScannerError::Other(format!("failed to decode scanner usage floor from {primary_path}: {err}"))
})?;
if data_usage_info_is_pristine_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() {
if data_usage_info_is_bootstrap_pending(&usage) && primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str() {
if bootstrap_pending {
return Err(ScannerError::Other(
"multiple pristine scanner usage bootstrap markers were found".to_string(),
));
return Err(ScannerError::Other("multiple scanner usage bootstrap markers were found".to_string()));
}
bootstrap_pending = true;
update_floor(&mut floor, &usage, primary_path)?;
@@ -1376,7 +1374,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
Ok((Some(data), _)) => {
if bootstrap_pending {
return Err(ScannerError::Other(
"pristine scanner usage bootstrap conflicts with a persisted backup".to_string(),
"scanner usage bootstrap conflicts with a persisted backup".to_string(),
));
}
any_found = true;
@@ -1406,7 +1404,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
if any_found {
if bootstrap_pending {
return Err(ScannerError::Other(
"pristine scanner usage bootstrap conflicts with an authoritative usage floor".to_string(),
"scanner usage bootstrap conflicts with an authoritative usage floor".to_string(),
));
}
found_any = true;
@@ -1415,14 +1413,14 @@ pub(super) async fn persisted_usage_floor_for_startup(
}
if !found_any && !bootstrap_pending {
if !allow_missing_for_pristine_startup {
if !allow_missing_for_bootstrap {
return Err(ScannerError::Other(
"persisted scanner usage floor has no authoritative baseline".to_string(),
));
}
let Some(publication_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), read_epoch).await else {
return Err(ScannerError::Other(
"scanner usage floor changed before pristine state confirmation".to_string(),
"scanner usage floor changed before missing-state confirmation".to_string(),
));
};
for path in [
@@ -1435,12 +1433,12 @@ pub(super) async fn persisted_usage_floor_for_startup(
Ok((None, _)) => {}
Ok((Some(_), _)) => {
return Err(ScannerError::Other(format!(
"scanner usage floor changed while confirming pristine state: {path} appeared"
"scanner usage floor changed while confirming missing state: {path} appeared"
)));
}
Err(err) => {
return Err(ScannerError::Other(format!(
"failed to confirm pristine scanner usage floor at {path}: {err}"
"failed to confirm missing scanner usage floor at {path}: {err}"
)));
}
}
crates/scanner/src/scanner/leadership.rs
+35 -32
View File
@@ -63,14 +63,12 @@ pub(super) async fn reconcile_scanner_leadership_claim(
pub(super) fn decode_usage_snapshot_for_epoch_fence(
data: &[u8],
path: &str,
allow_pristine_bootstrap_pending: bool,
allow_bootstrap_pending: bool,
) -> Result<DataUsageInfo, ScannerError> {
let usage: DataUsageInfo = serde_json::from_slice(data)
.map_err(|err| ScannerError::Other(format!("failed to decode scanner usage epoch fence from {path}: {err}")))?;
if !data_usage_info_has_persisted_baseline_identity(&usage)
&& !(allow_pristine_bootstrap_pending
&& path == DATA_USAGE_OBJ_NAME_PATH.as_str()
&& data_usage_info_is_pristine_bootstrap_pending(&usage))
&& !(allow_bootstrap_pending && path == DATA_USAGE_OBJ_NAME_PATH.as_str() && data_usage_info_is_bootstrap_pending(&usage))
{
return Err(ScannerError::Other(format!(
"scanner usage epoch fence from {path} has no persisted baseline identity"
@@ -82,15 +80,11 @@ pub(super) fn decode_usage_snapshot_for_epoch_fence(
pub(super) async fn usage_snapshot_for_epoch_fence(
storeapi: Arc<impl ScannerObjectIO>,
primary: Option<&[u8]>,
allow_pristine_bootstrap_pending: bool,
allow_bootstrap_pending: bool,
) -> Result<Option<DataUsageInfo>, ScannerError> {
if let Some(primary) = primary {
return decode_usage_snapshot_for_epoch_fence(
primary,
DATA_USAGE_OBJ_NAME_PATH.as_str(),
allow_pristine_bootstrap_pending,
)
.map(Some);
return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending)
.map(Some);
}
let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -118,12 +112,12 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
Ok(None)
}
pub(super) async fn initialize_pristine_usage_baseline(
pub(super) async fn initialize_usage_baseline_bootstrap(
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
) -> Result<(), ScannerError> {
let Some(expected_epoch) = scanner_publication_epoch(storeapi.clone()).await else {
return Err(ScannerError::Other(
"pristine scanner usage baseline initialization is blocked by data movement".to_string(),
"scanner usage baseline bootstrap is blocked by data movement".to_string(),
));
};
let baseline = DataUsageInfo {
@@ -133,7 +127,7 @@ pub(super) async fn initialize_pristine_usage_baseline(
..Default::default()
};
let data = serde_json::to_vec(&baseline)
.map_err(|err| ScannerError::Other(format!("failed to encode pristine scanner usage baseline: {err}")))?;
.map_err(|err| ScannerError::Other(format!("failed to encode scanner usage baseline bootstrap: {err}")))?;
let save_result = save_config_with_publication_admission_for_epoch(
storeapi.clone(),
DATA_USAGE_OBJ_NAME_PATH.as_str(),
@@ -153,14 +147,14 @@ pub(super) async fn initialize_pristine_usage_baseline(
let (persisted, revision) = read_config_with_revision(storeapi, DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.map_err(|err| ScannerError::Other(format!("failed to reconcile pristine scanner usage bootstrap: {err}")))?;
.map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage bootstrap: {err}")))?;
if persisted.as_deref() == Some(data.as_slice()) && matches!(revision, DataUsageCacheRevision::Etag(_)) {
return Ok(());
}
Err(ScannerError::Other(match save_result {
Ok(_) => "pristine scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
Err(err) => format!("failed to persist pristine scanner usage bootstrap: {err}"),
Ok(_) => "scanner usage bootstrap returned no ETag and could not be confirmed".to_string(),
Err(err) => format!("failed to persist scanner usage bootstrap: {err}"),
}))
}
@@ -169,7 +163,7 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
claimed_epoch: u64,
expected_publication_epoch: Option<u64>,
allow_pristine_bootstrap_pending: bool,
allow_bootstrap_pending: bool,
) -> Result<(), ScannerError> {
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
@@ -193,7 +187,7 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
.await
.map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence: {err}")))?;
let Some(mut usage) =
usage_snapshot_for_epoch_fence(storeapi.clone(), primary.as_deref(), allow_pristine_bootstrap_pending).await?
usage_snapshot_for_epoch_fence(storeapi.clone(), primary.as_deref(), allow_bootstrap_pending).await?
else {
let Some(_publication_admission) = scanner_publication_admission_for_epoch(storeapi.clone(), read_epoch).await else {
if retry < SCANNER_PERSIST_CAS_RETRIES {
@@ -243,11 +237,8 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
.await
.map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage epoch fence: {err}")))?;
if let Some(persisted) = persisted {
let persisted = decode_usage_snapshot_for_epoch_fence(
&persisted,
DATA_USAGE_OBJ_NAME_PATH.as_str(),
allow_pristine_bootstrap_pending,
)?;
let persisted =
decode_usage_snapshot_for_epoch_fence(&persisted, DATA_USAGE_OBJ_NAME_PATH.as_str(), allow_bootstrap_pending)?;
match persisted.scanner_epoch {
Some(epoch) if epoch == claimed_epoch => return Ok(()),
Some(epoch) if epoch > claimed_epoch => {
@@ -277,14 +268,14 @@ pub(super) async fn complete_scanner_leadership_claim(
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
claimed_epoch: u64,
expected_publication_epoch: Option<u64>,
allow_pristine_bootstrap_pending: bool,
allow_bootstrap_pending: bool,
) -> bool {
if let Err(err) = fence_scanner_usage_epoch_with_expected_epoch(
ctx,
storeapi,
claimed_epoch,
expected_publication_epoch,
allow_pristine_bootstrap_pending,
allow_bootstrap_pending,
)
.await
{
@@ -310,7 +301,8 @@ pub(super) async fn claim_scanner_leadership(
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
persisted_epoch: &mut u64,
allow_pristine_bootstrap_pending: bool,
allow_bootstrap_pending: bool,
reset_bootstrap_cycle_on_conflict: bool,
) -> bool {
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
@@ -365,7 +357,7 @@ pub(super) async fn claim_scanner_leadership(
return false;
}
};
match usage_snapshot_for_epoch_fence(storeapi.clone(), usage_primary.as_deref(), allow_pristine_bootstrap_pending).await {
match usage_snapshot_for_epoch_fence(storeapi.clone(), usage_primary.as_deref(), allow_bootstrap_pending).await {
Ok(Some(_)) => {}
Ok(None) => {
warn!(
@@ -413,7 +405,7 @@ pub(super) async fn claim_scanner_leadership(
storeapi,
claimed_epoch,
Some(read_epoch),
allow_pristine_bootstrap_pending,
allow_bootstrap_pending,
)
.await;
}
@@ -435,11 +427,16 @@ pub(super) async fn claim_scanner_leadership(
storeapi,
claimed_epoch,
Some(read_epoch),
allow_pristine_bootstrap_pending,
allow_bootstrap_pending,
)
.await;
}
Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => continue,
Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => {
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
continue;
}
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
error!(
target: "rustfs::scanner",
@@ -486,18 +483,24 @@ pub(super) async fn claim_scanner_leadership(
storeapi,
claimed_epoch,
Some(read_epoch),
allow_pristine_bootstrap_pending,
allow_bootstrap_pending,
)
.await;
}
Ok(ScannerLeadershipClaimReconcile::Changed)
if retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
continue;
}
Ok(ScannerLeadershipClaimReconcile::Unchanged)
if precondition_failed && retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
continue;
}
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
crates/scanner/src/scanner/tests.rs
+195 -61
View File
@@ -2111,11 +2111,11 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state()
}
#[tokio::test]
async fn scanner_usage_floor_allows_only_explicit_pristine_bootstrap() {
async fn scanner_usage_floor_allows_only_explicit_missing_state_bootstrap() {
let store = Arc::new(MemoryConfigStore::default());
let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
.await
.expect("a verified pristine startup should use the empty floor");
.expect("a verified missing state should use the empty floor");
assert_eq!(floor, PersistedUsageFloor::default());
assert_eq!(state, PersistedUsageFloorStartup::Missing);
assert!(persisted_usage_floor_for_startup(store.clone(), false).await.is_err());
@@ -2126,7 +2126,7 @@ async fn scanner_usage_floor_allows_only_explicit_pristine_bootstrap() {
);
assert!(
persisted_usage_floor_for_startup(store, true).await.is_err(),
"pristine bootstrap must not hide corrupt persisted state"
"usage bootstrap must not hide corrupt persisted state"
);
}
@@ -2156,16 +2156,16 @@ async fn scanner_usage_floor_fails_closed_on_zero_byte_usage_objects() {
let err = persisted_usage_floor_for_startup(appearing, true)
.await
.expect_err("an empty usage object appearing during confirmation must prevent pristine bootstrap");
.expect_err("an empty usage object appearing during confirmation must prevent usage bootstrap");
assert!(
err.to_string().contains("changed while confirming pristine state"),
err.to_string().contains("changed while confirming missing state"),
"unexpected confirmation error for {path}: {err}"
);
}
}
#[tokio::test]
async fn scanner_usage_floor_requires_publication_admission_for_pristine_bootstrap() {
async fn scanner_usage_floor_requires_publication_admission_for_bootstrap() {
let store = Arc::new(MemoryConfigStore::default());
store.publication_admission_blocked.store(true, Ordering::Release);
@@ -2173,18 +2173,18 @@ async fn scanner_usage_floor_requires_publication_admission_for_pristine_bootstr
}
#[tokio::test]
async fn scanner_usage_floor_fails_closed_when_usage_appears_during_pristine_confirmation() {
async fn scanner_usage_floor_fails_closed_when_usage_appears_during_missing_confirmation() {
let store = Arc::new(MemoryConfigStore::default());
insert_usage_after_first_legacy_backup_read(store.as_ref()).await;
let err = persisted_usage_floor_for_startup(store, true)
.await
.expect_err("an appearing usage snapshot must prevent pristine bootstrap");
assert!(err.to_string().contains("changed while confirming pristine state"));
.expect_err("an appearing usage snapshot must prevent usage bootstrap");
assert!(err.to_string().contains("changed while confirming missing state"));
}
#[tokio::test]
async fn scanner_usage_floor_rejects_publication_change_during_pristine_confirmation() {
async fn scanner_usage_floor_rejects_publication_change_during_missing_confirmation() {
let store = Arc::new(MemoryConfigStore::default());
store.block_publication_after_admissions.store(2, Ordering::Release);
@@ -2192,42 +2192,139 @@ async fn scanner_usage_floor_rejects_publication_change_during_pristine_confirma
}
#[test]
fn scanner_pristine_cycle_state_requires_no_durable_progress() {
let cycle = CurrentCycle::default();
assert!(scanner_cycle_state_is_pristine(&cycle, 0, &DataUsageCacheRevision::Missing));
assert!(scanner_may_resume_pristine_usage_bootstrap(&cycle));
assert!(!scanner_cycle_state_is_pristine(
&CurrentCycle {
next: 1,
..Default::default()
},
0,
&DataUsageCacheRevision::Missing
));
assert!(!scanner_may_resume_pristine_usage_bootstrap(&CurrentCycle {
next: 1,
fn missing_usage_floor_discards_unfenced_cycle_progress() {
let mut cycle = CurrentCycle {
current: 11,
next: 12,
cycle_completed: vec![Utc::now()],
started: Utc::now(),
};
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(&mut cycle, PersistedUsageFloor::default(), PersistedUsageFloorStartup::Missing,),
(true, true)
);
assert_eq!(cycle.next, 0);
assert_eq!(cycle.current, 0);
assert!(cycle.cycle_completed.is_empty());
cycle.next = 12;
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(
&mut cycle,
PersistedUsageFloor::default(),
PersistedUsageFloorStartup::BootstrapPending,
),
(true, true)
);
assert_eq!(cycle.next, 0);
}
#[test]
fn fenced_usage_bootstrap_retains_partial_cycle_progress() {
let mut cycle = CurrentCycle {
next: 12,
..Default::default()
}));
assert!(!scanner_cycle_state_is_pristine(&cycle, 1, &DataUsageCacheRevision::Missing));
assert!(!scanner_cycle_state_is_pristine(
&cycle,
0,
&DataUsageCacheRevision::Etag("etag".to_string())
));
};
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(
&mut cycle,
PersistedUsageFloor {
next_cycle: 0,
leader_epoch: 7,
},
PersistedUsageFloorStartup::BootstrapPending,
),
(true, false)
);
assert_eq!(cycle.next, 12);
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(
&mut cycle,
PersistedUsageFloor {
next_cycle: 13,
leader_epoch: 7,
},
PersistedUsageFloorStartup::Authoritative,
),
(false, false)
);
assert_eq!(cycle.next, 12);
}
#[tokio::test]
async fn scanner_pristine_bootstrap_allows_first_bucket_to_win_startup() {
async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim() {
let store = Arc::new(MemoryConfigStore::default());
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
let stale_cycle = CurrentCycle {
next: 12,
..Default::default()
};
store.objects.lock().await.insert(
state_key.clone(),
encode_scanner_cycle_state(&stale_cycle, 4).expect("stale cycle state should encode"),
);
store.revisions.lock().await.insert(state_key, 7);
let ScannerCycleStateStartup::Ready {
cycle: mut cycle_info,
leader_epoch: mut persisted_epoch,
revision: mut cycle_revision,
} = load_scanner_cycle_state_for_startup(store.clone()).await
else {
panic!("valid persisted cycle state should load");
};
let (usage_floor, startup) = persisted_usage_floor_for_startup(store.clone(), true)
.await
.expect("stably missing usage floor should admit a bootstrap marker");
assert_eq!(startup, PersistedUsageFloorStartup::Missing);
let (allow_bootstrap_pending, reset_bootstrap_cycle_on_conflict) =
prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, startup);
apply_persisted_usage_floor(&mut cycle_info, &mut persisted_epoch, usage_floor);
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("missing usage floor should publish a pending marker");
assert!(
claim_scanner_leadership(
&CancellationToken::new(),
store.clone(),
&mut cycle_info,
&mut cycle_revision,
&mut persisted_epoch,
allow_bootstrap_pending,
reset_bootstrap_cycle_on_conflict,
)
.await
);
let persisted_cycle = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
.await
.expect("rebuilt cycle state should be persisted");
let (persisted_cycle, persisted_cycle_epoch) =
decode_scanner_cycle_state(&persisted_cycle).expect("rebuilt cycle state should decode");
assert_eq!(persisted_cycle.next, 0);
assert_eq!(persisted_cycle_epoch, 5);
let pending = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("fenced bootstrap marker should remain persisted");
let pending = serde_json::from_slice::<DataUsageInfo>(&pending).expect("bootstrap marker should decode");
assert!(data_usage_info_is_bootstrap_pending(&pending));
assert_eq!(pending.scanner_epoch, Some(5));
assert!(!data_usage_info_has_persisted_baseline_identity(&pending));
}
#[tokio::test]
async fn scanner_usage_bootstrap_allows_first_bucket_to_win_startup() {
let (_temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
let cycle = CurrentCycle::default();
let revision = DataUsageCacheRevision::Missing;
store
.make_bucket("first-user-bucket", &crate::storage_api::scan::MakeBucketOptions::default())
.await
.expect("test bucket should be created");
assert!(scanner_may_bootstrap_missing_usage_floor(&cycle, 0, &revision));
assert_eq!(
persisted_usage_floor_for_startup(store.clone(), true)
.await
@@ -2235,14 +2332,14 @@ async fn scanner_pristine_bootstrap_allows_first_bucket_to_win_startup() {
.1,
PersistedUsageFloorStartup::Missing
);
initialize_pristine_usage_baseline(store.clone())
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("first startup should persist its pending marker");
let pending = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("pending marker should be stored");
let pending = serde_json::from_slice::<DataUsageInfo>(&pending).expect("pending marker should decode");
assert!(data_usage_info_is_pristine_bootstrap_pending(&pending));
assert!(data_usage_info_is_bootstrap_pending(&pending));
assert!(!data_usage_info_has_persisted_baseline_identity(&pending));
assert_eq!(
persisted_usage_floor_for_startup(store.clone(), false)
@@ -2553,7 +2650,7 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf
);
let mut persisted_epoch = 8;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
.await
@@ -2565,6 +2662,42 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf
assert_eq!(store.put_counts.lock().await.get(&key), Some(&3));
}
#[tokio::test]
async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() {
let store = Arc::new(MemoryConfigStore::default());
let ctx = CancellationToken::new();
let mut revision = DataUsageCacheRevision::Missing;
let mut cycle = CurrentCycle {
next: 12,
..Default::default()
};
assert!(persist_scanner_cycle_state(&ctx, store.clone(), &mut cycle, &mut revision, 1).await);
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("missing usage floor should publish a pending marker");
cycle = CurrentCycle::default();
let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
let stale_cycle = CurrentCycle {
next: 14,
..Default::default()
};
store.interleaving_puts.lock().await.insert(
key,
(2, encode_scanner_cycle_state(&stale_cycle, 1).expect("stale cycle state should encode")),
);
let mut persisted_epoch = 1;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await);
let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH)
.await
.expect("rebuilt leadership claim should remain persisted");
let (claimed_cycle, claimed_epoch) = decode_scanner_cycle_state(&state).expect("claimed cycle state should decode");
assert_eq!(claimed_cycle.next, 0);
assert_eq!(claimed_epoch, 2);
}
#[tokio::test]
async fn test_leadership_claim_rejects_terminal_epoch() {
let store = Arc::new(MemoryConfigStore::default());
@@ -2576,7 +2709,7 @@ async fn test_leadership_claim_rejects_terminal_epoch() {
};
let mut persisted_epoch = u64::MAX - 1;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert_eq!(persisted_epoch, u64::MAX - 1);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -2592,64 +2725,64 @@ async fn scanner_defers_leadership_when_usage_snapshots_are_stably_absent() {
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err());
}
#[tokio::test]
async fn pristine_usage_bootstrap_pending_unblocks_first_leadership_claim() {
async fn usage_bootstrap_pending_unblocks_first_leadership_claim() {
let store = Arc::new(MemoryConfigStore::default());
initialize_pristine_usage_baseline(store.clone())
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("verified pristine startup should publish its pending marker");
.expect("verified missing state should publish its pending marker");
let ctx = CancellationToken::new();
let mut revision = DataUsageCacheRevision::Missing;
let mut cycle = CurrentCycle::default();
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false,).await);
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false,).await);
assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true).await);
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await);
let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("leadership claim should fence the pristine baseline");
let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("pristine bootstrap marker should remain valid");
assert!(data_usage_info_is_pristine_bootstrap_pending(&usage));
.expect("leadership claim should fence the bootstrap marker");
let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("bootstrap marker should remain valid");
assert!(data_usage_info_is_bootstrap_pending(&usage));
assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
assert_eq!(usage.scanner_epoch, Some(1));
}
#[tokio::test]
async fn existing_pristine_usage_bootstrap_is_resumed_after_restart() {
async fn existing_usage_bootstrap_is_resumed_after_restart() {
let store = Arc::new(MemoryConfigStore::default());
initialize_pristine_usage_baseline(store.clone())
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("verified pristine startup should publish its pending marker");
.expect("verified missing state should publish its pending marker");
let (floor, state) = persisted_usage_floor_for_startup(store.clone(), false)
.await
.expect("restart should recognize the pending pristine bootstrap");
.expect("restart should recognize the pending usage bootstrap");
assert_eq!(floor, PersistedUsageFloor::default());
assert_eq!(state, PersistedUsageFloorStartup::BootstrapPending);
assert!(persisted_usage_floor(store).await.is_err());
}
#[tokio::test]
async fn pristine_usage_bootstrap_reconciles_post_commit_error() {
async fn usage_bootstrap_reconciles_post_commit_error() {
let store = Arc::new(MemoryConfigStore::default());
let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
store.error_after_commit_put_number.lock().await.insert(key, 1);
initialize_pristine_usage_baseline(store.clone())
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("a committed pending marker should reconcile after a lost response");
let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("the reconciled pending marker should remain");
let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("pending marker should decode");
assert!(data_usage_info_is_pristine_bootstrap_pending(&usage));
assert!(data_usage_info_is_bootstrap_pending(&usage));
assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
assert_eq!(
persisted_usage_floor_for_startup(store.clone(), false)
@@ -2662,7 +2795,7 @@ async fn pristine_usage_bootstrap_reconciles_post_commit_error() {
}
#[tokio::test]
async fn pristine_usage_bootstrap_does_not_overwrite_concurrent_replacement() {
async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() {
let store = Arc::new(MemoryConfigStore::default());
let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
let replacement = serde_json::to_vec(&complete_usage_with_bucket_count(None, 1)).expect("replacement should encode");
@@ -2672,7 +2805,7 @@ async fn pristine_usage_bootstrap_does_not_overwrite_concurrent_replacement() {
.await
.insert(key, (1, replacement.clone()));
initialize_pristine_usage_baseline(store.clone())
initialize_usage_baseline_bootstrap(store.clone())
.await
.expect("the bootstrap write completed before the replacement");
assert_eq!(
@@ -2698,7 +2831,7 @@ async fn leadership_claim_defers_on_corrupt_usage_baseline_without_bloom_write()
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -2718,7 +2851,7 @@ async fn leadership_claim_defers_on_unidentified_usage_baseline_without_bloom_wr
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -2740,7 +2873,7 @@ async fn test_leadership_claim_confirms_commit_after_returned_error() {
let mut persisted_epoch = 0;
seed_usage_snapshot_for_leadership_claim(&store).await;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
.await
@@ -2796,7 +2929,7 @@ async fn test_leadership_claim_usage_fence_rejects_old_inflight_writer() {
..Default::default()
};
let mut persisted_epoch = 4;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false).await);
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
let (fenced_data, fenced_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
@@ -2862,6 +2995,7 @@ async fn cycle_budget_lease_takeover_rejects_old_generation() {
&mut replacement_revision,
&mut replacement_epoch,
false,
false,
)
.await
);