mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-27 15:37:02 +00:00
fix(scanner): delay deep verify for fresh objects (#3329)
Gate scanner-triggered deep heal behind a short new-object cooldown and retry transient bitrot shard size mismatches during local verification. Add targeted tests for cooldown mode selection and size-mismatch classification, and log when deep scans are downgraded or size-mismatch retries occur.
This commit is contained in:
@@ -66,6 +66,10 @@ const DELETED_OBJECTS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 5);
|
|||||||
const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60);
|
const STALE_TMP_OBJECT_EXPIRY: Duration = Duration::from_secs(24 * 60 * 60);
|
||||||
const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old";
|
const RUSTFS_META_TMP_OLD_BUCKET: &str = ".rustfs.sys/tmp-old";
|
||||||
const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
|
const STARTUP_CLEANUP_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
|
||||||
|
const ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_COUNT";
|
||||||
|
const ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: &str = "RUSTFS_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS";
|
||||||
|
const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_COUNT: u64 = 2;
|
||||||
|
const DEFAULT_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS: u64 = 100;
|
||||||
|
|
||||||
fn log_startup_disk_io_error(stage: &str, path: &Path, err: &IoError) {
|
fn log_startup_disk_io_error(stage: &str, path: &Path, err: &IoError) {
|
||||||
warn!(
|
warn!(
|
||||||
@@ -132,6 +136,21 @@ fn record_file_cache_reclaim_error(kind: &'static str) {
|
|||||||
counter!("rustfs_page_cache_reclaim_requests_total", "kind" => kind.to_string(), "result" => "err".to_string()).increment(1);
|
counter!("rustfs_page_cache_reclaim_requests_total", "kind" => kind.to_string(), "result" => "err".to_string()).increment(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn bitrot_size_mismatch_retry_count() -> usize {
|
||||||
|
rustfs_utils::get_env_u64(ENV_BITROT_SIZE_MISMATCH_RETRY_COUNT, DEFAULT_BITROT_SIZE_MISMATCH_RETRY_COUNT) as usize
|
||||||
|
}
|
||||||
|
|
||||||
|
fn bitrot_size_mismatch_retry_delay() -> Duration {
|
||||||
|
Duration::from_millis(rustfs_utils::get_env_u64(
|
||||||
|
ENV_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS,
|
||||||
|
DEFAULT_BITROT_SIZE_MISMATCH_RETRY_DELAY_MS,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_bitrot_size_mismatch_error(err: &std::io::Error) -> bool {
|
||||||
|
err.to_string().contains("bitrot shard file size mismatch")
|
||||||
|
}
|
||||||
|
|
||||||
impl FileCacheReclaimReader {
|
impl FileCacheReclaimReader {
|
||||||
fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self {
|
fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self {
|
||||||
#[cfg(target_os = "macos")]
|
#[cfg(target_os = "macos")]
|
||||||
@@ -1368,16 +1387,42 @@ impl LocalDisk {
|
|||||||
sum: &[u8],
|
sum: &[u8],
|
||||||
shard_size: usize,
|
shard_size: usize,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let file = super::fs::open_file(part_path, O_RDONLY).await.map_err(to_file_error)?;
|
let retry_count = bitrot_size_mismatch_retry_count();
|
||||||
|
let retry_delay = bitrot_size_mismatch_retry_delay();
|
||||||
|
|
||||||
let meta = file.metadata().await.map_err(to_file_error)?;
|
for attempt in 0..=retry_count {
|
||||||
let file_size = meta.len() as usize;
|
let file = super::fs::open_file(part_path, O_RDONLY).await.map_err(to_file_error)?;
|
||||||
|
let meta = file.metadata().await.map_err(to_file_error)?;
|
||||||
|
let file_size = meta.len() as usize;
|
||||||
|
|
||||||
bitrot_verify(Box::new(file), file_size, part_size, algo, Bytes::copy_from_slice(sum), shard_size)
|
match bitrot_verify(
|
||||||
|
Box::new(file),
|
||||||
|
file_size,
|
||||||
|
part_size,
|
||||||
|
algo.clone(),
|
||||||
|
Bytes::copy_from_slice(sum),
|
||||||
|
shard_size,
|
||||||
|
)
|
||||||
.await
|
.await
|
||||||
.map_err(to_file_error)?;
|
{
|
||||||
|
Ok(()) => return Ok(()),
|
||||||
|
Err(err) if attempt < retry_count && is_bitrot_size_mismatch_error(&err) => {
|
||||||
|
info!(
|
||||||
|
path = %part_path.display(),
|
||||||
|
expected_size = part_size,
|
||||||
|
actual_size = file_size,
|
||||||
|
retry_attempt = attempt + 1,
|
||||||
|
retry_count,
|
||||||
|
retry_delay_ms = retry_delay.as_millis(),
|
||||||
|
"bitrot verify observed shard size mismatch; retrying"
|
||||||
|
);
|
||||||
|
tokio::time::sleep(retry_delay).await;
|
||||||
|
}
|
||||||
|
Err(err) => return Err(to_file_error(err).into()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
Ok(())
|
Err(DiskError::other("bitrot size mismatch retry loop exhausted"))
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_recursion::async_recursion]
|
#[async_recursion::async_recursion]
|
||||||
@@ -4452,4 +4497,10 @@ mod test {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_is_bitrot_size_mismatch_error_only_matches_target_message() {
|
||||||
|
assert!(is_bitrot_size_mismatch_error(&std::io::Error::other("bitrot shard file size mismatch")));
|
||||||
|
assert!(!is_bitrot_size_mismatch_error(&std::io::Error::other("bitrot hash mismatch")));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -62,6 +62,7 @@ use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket;
|
|||||||
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType};
|
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType};
|
||||||
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
||||||
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
|
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
|
||||||
|
use time::OffsetDateTime;
|
||||||
use tokio::select;
|
use tokio::select;
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
@@ -75,10 +76,12 @@ const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
|
|||||||
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
|
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
|
||||||
const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES";
|
const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES";
|
||||||
const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB";
|
const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB";
|
||||||
|
const ENV_SCANNER_DEEP_VERIFY_COOLDOWN_SECS: &str = "RUSTFS_SCANNER_DEEP_VERIFY_COOLDOWN_SECS";
|
||||||
const ENV_FAILED_OBJECT_TTL_SECS: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS";
|
const ENV_FAILED_OBJECT_TTL_SECS: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS";
|
||||||
const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX";
|
const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX";
|
||||||
const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400;
|
const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400;
|
||||||
const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000;
|
const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000;
|
||||||
|
const DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS: u64 = 60;
|
||||||
const METRIC_SCANNER_INLINE_HEAL_TOTAL: &str = "rustfs_scanner_inline_heal_total";
|
const METRIC_SCANNER_INLINE_HEAL_TOTAL: &str = "rustfs_scanner_inline_heal_total";
|
||||||
const METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL: &str = "rustfs_scanner_excess_object_versions_total";
|
const METRIC_SCANNER_EXCESS_OBJECT_VERSIONS_TOTAL: &str = "rustfs_scanner_excess_object_versions_total";
|
||||||
const METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL: &str = "rustfs_scanner_excess_object_version_size_total";
|
const METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL: &str = "rustfs_scanner_excess_object_version_size_total";
|
||||||
@@ -96,6 +99,34 @@ pub fn heal_object_select_prob() -> u32 {
|
|||||||
rustfs_utils::get_env_u32(ENV_HEAL_OBJECT_SELECT_PROB, DEFAULT_HEAL_OBJECT_SELECT_PROB)
|
rustfs_utils::get_env_u32(ENV_HEAL_OBJECT_SELECT_PROB, DEFAULT_HEAL_OBJECT_SELECT_PROB)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn deep_verify_cooldown() -> Duration {
|
||||||
|
Duration::from_secs(rustfs_utils::get_env_u64(
|
||||||
|
ENV_SCANNER_DEEP_VERIFY_COOLDOWN_SECS,
|
||||||
|
DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn object_is_within_deep_verify_cooldown(mod_time: Option<OffsetDateTime>, now: OffsetDateTime, cooldown: Duration) -> bool {
|
||||||
|
let Some(mod_time) = mod_time else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
let Ok(cooldown) = time::Duration::try_from(cooldown) else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
mod_time > now - cooldown
|
||||||
|
}
|
||||||
|
|
||||||
|
fn effective_object_heal_scan_mode(heal_bitrot: bool, mod_time: Option<OffsetDateTime>, now: OffsetDateTime) -> HealScanMode {
|
||||||
|
if !heal_bitrot {
|
||||||
|
return HealScanMode::Normal;
|
||||||
|
}
|
||||||
|
if object_is_within_deep_verify_cooldown(mod_time, now, deep_verify_cooldown()) {
|
||||||
|
HealScanMode::Normal
|
||||||
|
} else {
|
||||||
|
HealScanMode::Deep
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn scanner_inline_heal_enabled() -> bool {
|
fn scanner_inline_heal_enabled() -> bool {
|
||||||
scanner_inline_heal_enabled_from_value(std::env::var(rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE).ok().as_deref())
|
scanner_inline_heal_enabled_from_value(std::env::var(rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE).ok().as_deref())
|
||||||
}
|
}
|
||||||
@@ -828,11 +859,25 @@ impl ScannerItem {
|
|||||||
oi.version_id.unwrap_or_default()
|
oi.version_id.unwrap_or_default()
|
||||||
);
|
);
|
||||||
|
|
||||||
let scan_mode = if self.heal_bitrot {
|
let now = OffsetDateTime::now_utc();
|
||||||
HealScanMode::Deep
|
let scan_mode = effective_object_heal_scan_mode(self.heal_bitrot, oi.mod_time, now);
|
||||||
} else {
|
if self.heal_bitrot && scan_mode != HealScanMode::Deep {
|
||||||
HealScanMode::Normal
|
let cooldown = deep_verify_cooldown();
|
||||||
};
|
let age_secs = oi.mod_time.map(|mod_time| {
|
||||||
|
let age = now - mod_time;
|
||||||
|
age.whole_seconds().max(0)
|
||||||
|
});
|
||||||
|
info!(
|
||||||
|
bucket = %self.bucket,
|
||||||
|
object = %self.object_path(),
|
||||||
|
version_id = %oi.version_id.unwrap_or_default(),
|
||||||
|
object_age_secs = age_secs.unwrap_or_default(),
|
||||||
|
cooldown_secs = cooldown.as_secs(),
|
||||||
|
original_scan_mode = %HealScanMode::Deep.as_str(),
|
||||||
|
effective_scan_mode = %scan_mode.as_str(),
|
||||||
|
"scanner deep heal downgraded to normal during new-object cooldown"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
let result = send_scanner_heal_request(
|
let result = send_scanner_heal_request(
|
||||||
"object",
|
"object",
|
||||||
@@ -2544,6 +2589,26 @@ mod tests {
|
|||||||
assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING));
|
assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_effective_object_heal_scan_mode_keeps_normal_when_bitrot_disabled() {
|
||||||
|
let now = OffsetDateTime::now_utc();
|
||||||
|
assert_eq!(effective_object_heal_scan_mode(false, Some(now), now), HealScanMode::Normal);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_effective_object_heal_scan_mode_downgrades_recent_object_to_normal() {
|
||||||
|
let now = OffsetDateTime::now_utc();
|
||||||
|
let recent = now - time::Duration::seconds(5);
|
||||||
|
assert_eq!(effective_object_heal_scan_mode(true, Some(recent), now), HealScanMode::Normal);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_effective_object_heal_scan_mode_keeps_old_object_deep() {
|
||||||
|
let now = OffsetDateTime::now_utc();
|
||||||
|
let old = now - time::Duration::seconds((DEFAULT_SCANNER_DEEP_VERIFY_COOLDOWN_SECS as i64) + 5);
|
||||||
|
assert_eq!(effective_object_heal_scan_mode(true, Some(old), now), HealScanMode::Deep);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_heal_priority_label_matches_priority_names() {
|
fn test_heal_priority_label_matches_priority_names() {
|
||||||
assert_eq!(heal_priority_label(HealChannelPriority::Low), "low");
|
assert_eq!(heal_priority_label(HealChannelPriority::Low), "low");
|
||||||
|
|||||||
Reference in New Issue
Block a user