mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-02 10:18:10 +00:00
fix(scanner): recover legacy usage floor from backup (#6964)
* fix(scanner): recover legacy usage floor from backup Allow scanner usage-floor startup and leadership fencing to use a valid legacy backup when the legacy primary read fails with a corruption-shaped error. Keep v2 primary read failures, stale metadata, transient I/O, and missing or invalid backups fail-closed. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(scanner): cover legacy backup fencing gaps (#6966) * fix(scanner): recover legacy usage from valid backup * fix(scanner): recover legacy usage floor from backup Allow scanner usage-floor startup and leadership fencing to use a valid legacy backup when the legacy primary read fails with a corruption-shaped error. Keep v2 primary read failures, stale metadata, transient I/O, and missing or invalid backups fail-closed. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(scanner): cover legacy backup fencing gaps --------- Co-authored-by: houseme <housemecn@gmail.com> Co-authored-by: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: Henry Guo <marshawcoco@gmail.com>
This commit is contained in:
@@ -131,6 +131,39 @@ pub(crate) async fn read_config_with_revision<S: ScannerObjectIO>(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) fn usage_floor_primary_read_error_allows_backup(err: &Error) -> bool {
|
||||||
|
match err {
|
||||||
|
Error::FileCorrupt
|
||||||
|
| Error::CorruptedFormat
|
||||||
|
| Error::CorruptedBackend
|
||||||
|
| Error::PartMissingOrCorrupt
|
||||||
|
| Error::LessData
|
||||||
|
| Error::MoreData => true,
|
||||||
|
Error::Io(io_error) => {
|
||||||
|
matches!(io_error.kind(), std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof)
|
||||||
|
|| error_chain_has_usage_floor_corruption_signature(io_error)
|
||||||
|
}
|
||||||
|
_ => false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn error_chain_has_usage_floor_corruption_signature(error: &(dyn std::error::Error + 'static)) -> bool {
|
||||||
|
let mut current = Some(error);
|
||||||
|
while let Some(err) = current {
|
||||||
|
let message = err.to_string();
|
||||||
|
if message.contains("InlineData value out of range")
|
||||||
|
|| message.contains("InlineData key out of range")
|
||||||
|
|| message.contains("insufficient data for metadata")
|
||||||
|
|| message.contains("insufficient data for meta length")
|
||||||
|
|| message.contains("insufficient data for CRC")
|
||||||
|
{
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
current = err.source();
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
/// Read only the object revision without materializing its body.
|
/// Read only the object revision without materializing its body.
|
||||||
pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
|
pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
|
||||||
match store
|
match store
|
||||||
|
|||||||
@@ -14,7 +14,9 @@
|
|||||||
/// Scanner cycle-state codec, persisted usage floors, and cycle-state persistence.
|
/// Scanner cycle-state codec, persisted usage floors, and cycle-state persistence.
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::ScannerGetObjectReader;
|
use crate::ScannerGetObjectReader;
|
||||||
use crate::data_usage_define::{DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH};
|
use crate::data_usage_define::{
|
||||||
|
DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup,
|
||||||
|
};
|
||||||
use crate::storage_api::owner::ObjectIO as _;
|
use crate::storage_api::owner::ObjectIO as _;
|
||||||
use tokio::io::AsyncReadExt as _;
|
use tokio::io::AsyncReadExt as _;
|
||||||
|
|
||||||
@@ -1717,6 +1719,7 @@ pub(super) async fn persisted_usage_floor_for_startup(
|
|||||||
let backup_path = format!("{primary_path}.bkp");
|
let backup_path = format!("{primary_path}.bkp");
|
||||||
let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str();
|
let is_v2_path = primary_path == DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
let mut recovered_primary_companion_epoch = None;
|
let mut recovered_primary_companion_epoch = None;
|
||||||
|
let mut primary_read_error = None;
|
||||||
let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await {
|
let primary_epoch = match read_config_with_revision(storeapi.clone(), primary_path).await {
|
||||||
Ok((Some(data), revision)) => {
|
Ok((Some(data), revision)) => {
|
||||||
let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
|
let usage = serde_json::from_slice::<DataUsageInfo>(&data).map_err(|err| {
|
||||||
@@ -1776,6 +1779,12 @@ pub(super) async fn persisted_usage_floor_for_startup(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
Ok((None, _)) => None,
|
Ok((None, _)) => None,
|
||||||
|
Err(err) if !is_v2_path && usage_floor_primary_read_error_allows_backup(&err) => {
|
||||||
|
primary_read_error = Some(format!("failed to read scanner usage epoch floor from {primary_path}: {err}"));
|
||||||
|
invalid_baseline_path.get_or_insert_with(|| primary_path.to_string());
|
||||||
|
unrecoverable_baseline_path.get_or_insert_with(|| primary_path.to_string());
|
||||||
|
None
|
||||||
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
return Err(ScannerError::Other(format!(
|
return Err(ScannerError::Other(format!(
|
||||||
"failed to read scanner usage epoch floor from {primary_path}: {err}"
|
"failed to read scanner usage epoch floor from {primary_path}: {err}"
|
||||||
@@ -1855,6 +1864,14 @@ pub(super) async fn persisted_usage_floor_for_startup(
|
|||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if let Some(primary_read_error) = primary_read_error
|
||||||
|
&& !any_found
|
||||||
|
{
|
||||||
|
return Err(ScannerError::Other(format!(
|
||||||
|
"{}; no valid scanner usage floor backup was available at {backup_path}",
|
||||||
|
primary_read_error
|
||||||
|
)));
|
||||||
|
}
|
||||||
if any_found {
|
if any_found {
|
||||||
if bootstrap_pending {
|
if bootstrap_pending {
|
||||||
return Err(ScannerError::Other(
|
return Err(ScannerError::Other(
|
||||||
|
|||||||
@@ -13,6 +13,7 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
/// Leader-lock claiming, usage-epoch fencing, and lock-loss handling.
|
/// Leader-lock claiming, usage-epoch fencing, and lock-loss handling.
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::data_usage_define::usage_floor_primary_read_error_allows_backup;
|
||||||
|
|
||||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||||
pub(super) enum ScannerLeadershipClaimReconcile {
|
pub(super) enum ScannerLeadershipClaimReconcile {
|
||||||
@@ -142,20 +143,34 @@ pub(super) async fn usage_snapshot_for_epoch_fence(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for path in [
|
let legacy_primary_path = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string();
|
||||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
let legacy_backup_path = format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str());
|
||||||
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
let mut legacy_primary_read_error = None;
|
||||||
] {
|
for path in [&legacy_primary_path, &legacy_backup_path] {
|
||||||
let (legacy, _) = read_config_with_revision(storeapi.clone(), &path)
|
let legacy = match read_config_with_revision(storeapi.clone(), path).await {
|
||||||
.await
|
Ok((legacy, _)) => legacy,
|
||||||
.map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?;
|
Err(err) if path == &legacy_primary_path && usage_floor_primary_read_error_allows_backup(&err) => {
|
||||||
|
legacy_primary_read_error = Some(format!("failed to read legacy scanner usage epoch fence from {path}: {err}"));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
return Err(ScannerError::Other(format!(
|
||||||
|
"failed to read legacy scanner usage epoch fence from {path}: {err}"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
};
|
||||||
if let Some(legacy) = legacy.as_deref() {
|
if let Some(legacy) = legacy.as_deref() {
|
||||||
let usage = decode_usage_snapshot_for_epoch_fence(legacy, &path, false)?;
|
let usage = decode_usage_snapshot_for_epoch_fence(legacy, path, false)?;
|
||||||
if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
|
if invalid_primary_epoch.is_none_or(|epoch| usage.scanner_epoch.unwrap_or_default() >= epoch) {
|
||||||
return Ok(Some(usage));
|
return Ok(Some(usage));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if let Some(legacy_primary_read_error) = legacy_primary_read_error {
|
||||||
|
return Err(ScannerError::Other(format!(
|
||||||
|
"{legacy_primary_read_error}; no valid legacy scanner usage epoch fence backup was available at {legacy_backup_path}"
|
||||||
|
)));
|
||||||
|
}
|
||||||
// A missing usage snapshot is an uninitialized state, not an empty
|
// A missing usage snapshot is an uninitialized state, not an empty
|
||||||
// snapshot. Leadership fencing may proceed without creating a plausible
|
// snapshot. Leadership fencing may proceed without creating a plausible
|
||||||
// default; the first authoritative scanner publication will create it.
|
// default; the first authoritative scanner publication will create it.
|
||||||
@@ -258,6 +273,12 @@ pub(super) async fn fence_scanner_usage_epoch_with_expected_epoch(
|
|||||||
Some(epoch) if epoch == claimed_epoch => return Ok(()),
|
Some(epoch) if epoch == claimed_epoch => return Ok(()),
|
||||||
Some(_) | None => {}
|
Some(_) | None => {}
|
||||||
}
|
}
|
||||||
|
// A validated pre-marker legacy baseline needs an explicit complete
|
||||||
|
// identity before acquiring an epoch. Otherwise the v2 reader would
|
||||||
|
// reject the fenced value on its next startup.
|
||||||
|
if !usage.usage_snapshot_bootstrap_pending {
|
||||||
|
usage.usage_snapshot_complete = true;
|
||||||
|
}
|
||||||
usage.scanner_epoch = Some(claimed_epoch);
|
usage.scanner_epoch = Some(claimed_epoch);
|
||||||
let data = serde_json::to_vec(&usage)
|
let data = serde_json::to_vec(&usage)
|
||||||
.map_err(|err| ScannerError::Other(format!("failed to encode scanner usage epoch fence: {err}")))?;
|
.map_err(|err| ScannerError::Other(format!("failed to encode scanner usage epoch fence: {err}")))?;
|
||||||
|
|||||||
@@ -574,6 +574,7 @@ struct MemoryConfigStore {
|
|||||||
objects: Mutex<HashMap<String, Vec<u8>>>,
|
objects: Mutex<HashMap<String, Vec<u8>>>,
|
||||||
revisions: Mutex<HashMap<String, u64>>,
|
revisions: Mutex<HashMap<String, u64>>,
|
||||||
insert_after_gets: Mutex<HashMap<String, Vec<u8>>>,
|
insert_after_gets: Mutex<HashMap<String, Vec<u8>>>,
|
||||||
|
read_errors: Mutex<HashMap<String, EcstoreError>>,
|
||||||
delayed_gets: Mutex<HashMap<String, Duration>>,
|
delayed_gets: Mutex<HashMap<String, Duration>>,
|
||||||
non_regular_objects: Mutex<HashSet<String>>,
|
non_regular_objects: Mutex<HashSet<String>>,
|
||||||
fail_put_number: Mutex<HashMap<String, usize>>,
|
fail_put_number: Mutex<HashMap<String, usize>>,
|
||||||
@@ -623,6 +624,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
|
|||||||
_opts: &ObjectOptions,
|
_opts: &ObjectOptions,
|
||||||
) -> EcstoreResult<GetObjectReader> {
|
) -> EcstoreResult<GetObjectReader> {
|
||||||
let key = memory_config_key(bucket, object);
|
let key = memory_config_key(bucket, object);
|
||||||
|
if let Some(error) = self.read_errors.lock().await.get(&key).cloned() {
|
||||||
|
return Err(error);
|
||||||
|
}
|
||||||
if let Some(delay) = self.delayed_gets.lock().await.remove(&key) {
|
if let Some(delay) = self.delayed_gets.lock().await.remove(&key) {
|
||||||
tokio::time::sleep(delay).await;
|
tokio::time::sleep(delay).await;
|
||||||
}
|
}
|
||||||
@@ -2902,6 +2906,147 @@ async fn scanner_usage_floor_recovers_from_incomplete_v2_primary_using_fenced_ba
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn seed_legacy_primary_read_error_with_backup(store: &Arc<MemoryConfigStore>, error: EcstoreError, epoch: u64, cycle: u64) {
|
||||||
|
let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
|
let legacy_backup = format!("{legacy_primary}.bkp");
|
||||||
|
let mut backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
|
||||||
|
backup.scanner_epoch = Some(epoch);
|
||||||
|
backup.scanner_cycle = Some(cycle);
|
||||||
|
|
||||||
|
store
|
||||||
|
.read_errors
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.insert(memory_config_key(RUSTFS_META_BUCKET, legacy_primary), error);
|
||||||
|
store.objects.lock().await.insert(
|
||||||
|
memory_config_key(RUSTFS_META_BUCKET, &legacy_backup),
|
||||||
|
serde_json::to_vec(&backup).expect("legacy backup usage snapshot should encode"),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_recovers_legacy_backup_after_primary_decode_error() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
|
||||||
|
|
||||||
|
let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
|
||||||
|
.await
|
||||||
|
.expect("valid legacy backup should recover the startup floor");
|
||||||
|
assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
|
||||||
|
assert_eq!(
|
||||||
|
floor,
|
||||||
|
PersistedUsageFloor {
|
||||||
|
next_cycle: 42,
|
||||||
|
leader_epoch: 19,
|
||||||
|
}
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
persisted_usage_floor(store)
|
||||||
|
.await
|
||||||
|
.expect("valid legacy backup should recover the authoritative floor"),
|
||||||
|
floor
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_does_not_bootstrap_over_corrupt_legacy_primary_without_backup() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
|
store
|
||||||
|
.read_errors
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.insert(memory_config_key(RUSTFS_META_BUCKET, legacy_primary), EcstoreError::FileCorrupt);
|
||||||
|
|
||||||
|
let err = persisted_usage_floor_for_startup(store, true)
|
||||||
|
.await
|
||||||
|
.expect_err("corrupt legacy primary without a valid backup must remain fail-closed");
|
||||||
|
assert!(err.to_string().contains("no valid scanner usage floor backup"), "unexpected error: {err}");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_does_not_fallback_to_legacy_after_corrupt_v2_primary() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
let mut v2_backup = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
|
||||||
|
v2_backup.scanner_epoch = Some(8);
|
||||||
|
v2_backup.scanner_cycle = Some(11);
|
||||||
|
let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
|
||||||
|
legacy.scanner_epoch = Some(3);
|
||||||
|
legacy.scanner_cycle = Some(7);
|
||||||
|
store.read_errors.lock().await.insert(
|
||||||
|
memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||||
|
EcstoreError::FileCorrupt,
|
||||||
|
);
|
||||||
|
store.objects.lock().await.insert(
|
||||||
|
memory_config_key(RUSTFS_META_BUCKET, &format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())),
|
||||||
|
serde_json::to_vec(&v2_backup).expect("v2 backup usage snapshot should encode"),
|
||||||
|
);
|
||||||
|
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 usage snapshot should encode"),
|
||||||
|
);
|
||||||
|
|
||||||
|
let err = persisted_usage_floor_for_startup(store, true)
|
||||||
|
.await
|
||||||
|
.expect_err("corrupt v2 primary must not recover without a primary revision");
|
||||||
|
assert!(
|
||||||
|
err.to_string().contains(&format!(
|
||||||
|
"failed to read scanner usage epoch floor from {}",
|
||||||
|
DATA_USAGE_OBJ_NAME_PATH.as_str()
|
||||||
|
)),
|
||||||
|
"unexpected error: {err}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_keeps_transient_primary_read_error_fail_closed() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
|
seed_legacy_primary_read_error_with_backup(
|
||||||
|
&store,
|
||||||
|
EcstoreError::Io(std::io::Error::new(
|
||||||
|
std::io::ErrorKind::ConnectionReset,
|
||||||
|
"connection reset while reading usage primary",
|
||||||
|
)),
|
||||||
|
19,
|
||||||
|
41,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let err = persisted_usage_floor_for_startup(store, true)
|
||||||
|
.await
|
||||||
|
.expect_err("transient primary errors must not be converted into backup recovery");
|
||||||
|
assert!(
|
||||||
|
err.to_string()
|
||||||
|
.contains(&format!("failed to read scanner usage epoch floor from {legacy_primary}")),
|
||||||
|
"unexpected error: {err}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!err.to_string().contains("no valid scanner usage floor backup"),
|
||||||
|
"transient error should not enter corrupt-primary fallback: {err}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_keeps_outdated_primary_metadata_fail_closed() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
let legacy_primary = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
|
seed_legacy_primary_read_error_with_backup(&store, EcstoreError::OutdatedXLMeta, 19, 41).await;
|
||||||
|
|
||||||
|
let err = persisted_usage_floor_for_startup(store, true)
|
||||||
|
.await
|
||||||
|
.expect_err("outdated primary metadata must not be converted into backup recovery");
|
||||||
|
assert!(
|
||||||
|
err.to_string()
|
||||||
|
.contains(&format!("failed to read scanner usage epoch floor from {legacy_primary}")),
|
||||||
|
"unexpected error: {err}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!err.to_string().contains("no valid scanner usage floor backup"),
|
||||||
|
"outdated metadata should not enter corrupt-primary fallback: {err}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() {
|
async fn scanner_usage_floor_does_not_bootstrap_over_incomplete_v2_primary() {
|
||||||
let store = Arc::new(MemoryConfigStore::default());
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
@@ -3004,6 +3149,19 @@ async fn scanner_leadership_fencing_recovers_incomplete_v2_primary_from_backup()
|
|||||||
assert_eq!(recovered.scanner_cycle, Some(103));
|
assert_eq!(recovered.scanner_cycle, Some(103));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_leadership_fencing_recovers_legacy_backup_after_primary_decode_error() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
|
||||||
|
|
||||||
|
let recovered = usage_snapshot_for_epoch_fence(store, None, false)
|
||||||
|
.await
|
||||||
|
.expect("a valid legacy backup should provide the fencing baseline")
|
||||||
|
.expect("the fencing baseline should be present");
|
||||||
|
assert_eq!(recovered.scanner_epoch, Some(19));
|
||||||
|
assert_eq!(recovered.scanner_cycle, Some(41));
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
|
async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
|
||||||
let store = Arc::new(MemoryConfigStore::default());
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
@@ -3955,6 +4113,162 @@ async fn scanner_defers_leadership_when_usage_snapshots_are_stably_absent() {
|
|||||||
assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err());
|
assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn scanner_usage_floor_leadership_claim_recovers_legacy_backup_after_primary_decode_error() {
|
||||||
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
let ctx = CancellationToken::new();
|
||||||
|
seed_legacy_primary_read_error_with_backup(&store, EcstoreError::other("InlineData value out of range"), 19, 41).await;
|
||||||
|
|
||||||
|
let mut revision = DataUsageCacheRevision::Missing;
|
||||||
|
let mut cycle = CurrentCycle::default();
|
||||||
|
let mut persisted_epoch = 19;
|
||||||
|
assert!(
|
||||||
|
claim_scanner_leadership(
|
||||||
|
&ctx,
|
||||||
|
store.clone(),
|
||||||
|
&mut cycle,
|
||||||
|
&mut revision,
|
||||||
|
&mut persisted_epoch,
|
||||||
|
false,
|
||||||
|
ScannerCycleResetPolicy::None,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
);
|
||||||
|
|
||||||
|
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||||
|
.await
|
||||||
|
.expect("leadership claim should persist after legacy backup recovery");
|
||||||
|
let (_, claimed_epoch) = decode_scanner_cycle_state(&state).expect("leadership claim should decode");
|
||||||
|
assert_eq!(claimed_epoch, 20);
|
||||||
|
assert_eq!(persisted_epoch, 20);
|
||||||
|
|
||||||
|
let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||||
|
.await
|
||||||
|
.expect("legacy backup recovery should publish a fenced v2 usage primary");
|
||||||
|
let usage = serde_json::from_slice::<DataUsageInfo>(&usage).expect("fenced v2 usage primary should decode");
|
||||||
|
assert_eq!(usage.scanner_epoch, Some(20));
|
||||||
|
assert_eq!(usage.scanner_cycle, Some(41));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[serial_test::serial]
|
||||||
|
async fn scanner_legacy_usage_backup_survives_fencing_and_restart_after_real_metadata_truncation() {
|
||||||
|
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||||
|
let (temp_dir, store) = setup_scanner_cycle_store_with_usage_baseline(false).await;
|
||||||
|
let mut usage = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0);
|
||||||
|
usage.usage_snapshot_complete = false;
|
||||||
|
usage.scanner_cycle = Some(41);
|
||||||
|
let mut data = serde_json::to_vec(&usage).expect("legacy usage should encode");
|
||||||
|
data.resize(data.len() + 16 * 1024, b' ');
|
||||||
|
let legacy_path = LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str();
|
||||||
|
let backup_path = format!("{legacy_path}.bkp");
|
||||||
|
for path in [legacy_path, backup_path.as_str()] {
|
||||||
|
save_config(store.clone(), path, data.clone())
|
||||||
|
.await
|
||||||
|
.expect("legacy usage fixture should persist");
|
||||||
|
}
|
||||||
|
let mut truncated_files = Vec::new();
|
||||||
|
for disk_index in 0..4 {
|
||||||
|
let path = temp_dir
|
||||||
|
.path()
|
||||||
|
.join(format!("pool0/disk{disk_index}"))
|
||||||
|
.join(RUSTFS_META_BUCKET)
|
||||||
|
.join(legacy_path)
|
||||||
|
.join("xl.meta");
|
||||||
|
let file = tokio::fs::OpenOptions::new()
|
||||||
|
.write(true)
|
||||||
|
.open(&path)
|
||||||
|
.await
|
||||||
|
.expect("legacy inline metadata should exist");
|
||||||
|
assert!(file.metadata().await.expect("metadata should be readable").len() > 4096);
|
||||||
|
file.set_len(4096).await.expect("fixture should truncate at a page boundary");
|
||||||
|
truncated_files.push((
|
||||||
|
path.clone(),
|
||||||
|
tokio::fs::read(&path)
|
||||||
|
.await
|
||||||
|
.expect("truncated evidence should remain readable"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
let store = restart_scanner_cycle_store_from(&store).await;
|
||||||
|
let error = read_config_with_revision(store.clone(), legacy_path)
|
||||||
|
.await
|
||||||
|
.expect_err("truncated primary must fail in the real object reader");
|
||||||
|
assert!(
|
||||||
|
error.to_string().contains("InlineData value out of range"),
|
||||||
|
"unexpected truncated-primary error: {error}"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
read_config_with_revision(store.clone(), &backup_path)
|
||||||
|
.await
|
||||||
|
.expect("backup should remain readable")
|
||||||
|
.0,
|
||||||
|
Some(data.clone()),
|
||||||
|
);
|
||||||
|
let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
|
||||||
|
.await
|
||||||
|
.expect("intact legacy backup must recover startup despite truncated primary");
|
||||||
|
assert_eq!(
|
||||||
|
floor,
|
||||||
|
PersistedUsageFloor {
|
||||||
|
next_cycle: 42,
|
||||||
|
leader_epoch: 0
|
||||||
|
}
|
||||||
|
);
|
||||||
|
assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
|
||||||
|
|
||||||
|
let baseline = read_data_usage_persist_baseline(store.clone())
|
||||||
|
.await
|
||||||
|
.expect("publication must also read the intact backup");
|
||||||
|
assert_eq!(baseline.data.as_deref(), Some(data.as_slice()));
|
||||||
|
assert_eq!(baseline.revision, DataUsageCacheRevision::Missing);
|
||||||
|
fence_scanner_usage_epoch_with_expected_epoch(&CancellationToken::new(), store.clone(), 7, None, false)
|
||||||
|
.await
|
||||||
|
.expect("legacy backup must be fenced into v2");
|
||||||
|
let fenced = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||||
|
.await
|
||||||
|
.expect("fencing must publish a v2 usage primary");
|
||||||
|
let fenced = serde_json::from_slice::<DataUsageInfo>(&fenced).expect("fenced v2 usage primary should decode");
|
||||||
|
assert!(
|
||||||
|
fenced.usage_snapshot_complete,
|
||||||
|
"the fenced pre-marker baseline must become a complete v2 identity"
|
||||||
|
);
|
||||||
|
|
||||||
|
let store = restart_scanner_cycle_store_from(&store).await;
|
||||||
|
let (floor, state) = persisted_usage_floor_for_startup(store.clone(), true)
|
||||||
|
.await
|
||||||
|
.expect("a restart after fencing must preserve the recovered floor");
|
||||||
|
assert_eq!(
|
||||||
|
floor,
|
||||||
|
PersistedUsageFloor {
|
||||||
|
next_cycle: 42,
|
||||||
|
leader_epoch: 7
|
||||||
|
}
|
||||||
|
);
|
||||||
|
assert_eq!(state, PersistedUsageFloorStartup::Authoritative);
|
||||||
|
let restarted = restart_scanner_cycle_store_from(&store).await;
|
||||||
|
assert_eq!(
|
||||||
|
persisted_usage_floor(restarted)
|
||||||
|
.await
|
||||||
|
.expect("fenced floor must survive another restart"),
|
||||||
|
PersistedUsageFloor {
|
||||||
|
next_cycle: 42,
|
||||||
|
leader_epoch: 7
|
||||||
|
}
|
||||||
|
);
|
||||||
|
for (path, bytes) in truncated_files {
|
||||||
|
assert_eq!(tokio::fs::read(path).await.expect("legacy evidence must not be removed"), bytes);
|
||||||
|
}
|
||||||
|
assert_eq!(
|
||||||
|
read_config(store, &backup_path)
|
||||||
|
.await
|
||||||
|
.expect("legacy backup must remain intact"),
|
||||||
|
data
|
||||||
|
);
|
||||||
|
global_metrics().set_cycle(None).await;
|
||||||
|
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn usage_bootstrap_pending_unblocks_first_leadership_claim() {
|
async fn usage_bootstrap_pending_unblocks_first_leadership_claim() {
|
||||||
let store = Arc::new(MemoryConfigStore::default());
|
let store = Arc::new(MemoryConfigStore::default());
|
||||||
|
|||||||
@@ -13,6 +13,7 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
/// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup.
|
/// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup.
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use crate::data_usage_define::usage_floor_primary_read_error_allows_backup;
|
||||||
use crate::storage_api::owner::ScannerPublicationCommitState;
|
use crate::storage_api::owner::ScannerPublicationCommitState;
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::sync::atomic::AtomicBool;
|
use std::sync::atomic::AtomicBool;
|
||||||
@@ -53,6 +54,30 @@ pub(super) struct DataUsagePersistBaseline {
|
|||||||
pub(super) revision: DataUsageCacheRevision,
|
pub(super) revision: DataUsageCacheRevision,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn read_usage_persist_candidate(
|
||||||
|
storeapi: Arc<impl ScannerObjectIO>,
|
||||||
|
path: &str,
|
||||||
|
) -> Result<(Option<Vec<u8>>, DataUsageCacheRevision), EcstoreError> {
|
||||||
|
let primary = read_config_with_revision(storeapi.clone(), path).await;
|
||||||
|
if path != LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str() {
|
||||||
|
return primary;
|
||||||
|
}
|
||||||
|
let Err(primary_error) = &primary else {
|
||||||
|
return primary;
|
||||||
|
};
|
||||||
|
if !usage_floor_primary_read_error_allows_backup(primary_error) {
|
||||||
|
return primary;
|
||||||
|
}
|
||||||
|
let backup_path = format!("{path}.bkp");
|
||||||
|
let backup = read_config_with_revision(storeapi, &backup_path).await?;
|
||||||
|
if backup.0.as_deref().is_some_and(|data| {
|
||||||
|
serde_json::from_slice::<DataUsageInfo>(data).is_ok_and(|usage| data_usage_info_has_persisted_baseline_identity(&usage))
|
||||||
|
}) {
|
||||||
|
return Ok(backup);
|
||||||
|
}
|
||||||
|
primary
|
||||||
|
}
|
||||||
|
|
||||||
/// Read the bytes used as the baseline for a usage publication while keeping
|
/// 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
|
/// 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
|
/// primary can be valid JSON without a baseline identity; in that case a
|
||||||
@@ -68,7 +93,7 @@ pub(super) async fn read_data_usage_persist_baseline(
|
|||||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
||||||
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||||
] {
|
] {
|
||||||
let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
|
let (candidate, _) = read_usage_persist_candidate(storeapi.clone(), &path).await?;
|
||||||
let Some(candidate) = candidate else {
|
let Some(candidate) = candidate else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
@@ -107,7 +132,7 @@ pub(super) async fn read_data_usage_persist_baseline(
|
|||||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
|
||||||
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||||
] {
|
] {
|
||||||
let (candidate, _) = read_config_with_revision(storeapi.clone(), &path).await?;
|
let (candidate, _) = read_usage_persist_candidate(storeapi.clone(), &path).await?;
|
||||||
let Some(candidate) = candidate else {
|
let Some(candidate) = candidate else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user