Compare commits

..

1 Commits

Author SHA1 Message Date
马登山 27a921372c fix(scanner): preserve unversioned heal retries 2026-08-22 19:12:27 +08:00
28 changed files with 811 additions and 3144 deletions
-56
View File
@@ -901,10 +901,6 @@ pub struct Metrics {
scanner_cycle_max_duration_millis: AtomicU64,
scanner_cycle_max_objects: AtomicU64,
scanner_cycle_max_directories: AtomicU64,
scanner_cycle_timeout_total: AtomicU64,
scanner_cycle_recovery_required_total: AtomicU64,
scanner_cycle_last_progress_age_seconds: AtomicU64,
scanner_leader_lease_without_progress: AtomicBool,
scanner_bitrot_cycle_enabled: AtomicBool,
scanner_bitrot_cycle_millis: AtomicU64,
scanner_checkpoint: Mutex<Option<ScannerCheckpointReport>>,
@@ -1374,14 +1370,6 @@ pub struct ScannerMetricsReport {
#[serde(default)]
pub cycle_max_directories: u64,
#[serde(default)]
pub cycle_timeout_total: u64,
#[serde(default)]
pub cycle_recovery_required_total: u64,
#[serde(default)]
pub cycle_last_progress_age: u64,
#[serde(default)]
pub leader_lease_without_progress: bool,
#[serde(default)]
pub bitrot_cycle_enabled: bool,
#[serde(default)]
pub bitrot_cycle_seconds: f64,
@@ -1442,9 +1430,6 @@ const OTEL_SCANNER_BUCKETS_SCANNED: &str = "rustfs_scanner_buckets_scanned_total
const OTEL_SCANNER_CYCLES: &str = "rustfs_scanner_cycles_total";
const OTEL_SCANNER_CYCLE_DURATION_SECONDS: &str = "rustfs_scanner_cycle_duration_seconds";
const OTEL_SCANNER_BUCKET_DRIVE_DURATION_SECONDS: &str = "rustfs_scanner_bucket_drive_duration_seconds";
const OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL: &str = "rustfs_scanner_cycle_timeout_total";
const OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE: &str = "rustfs_scanner_cycle_last_progress_age";
const OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS: &str = "rustfs_scanner_leader_lease_without_progress";
fn scan_cycle_result_label(result: u8) -> &'static str {
match result {
@@ -1928,10 +1913,6 @@ impl Metrics {
scanner_cycle_max_duration_millis: AtomicU64::new(0),
scanner_cycle_max_objects: AtomicU64::new(0),
scanner_cycle_max_directories: AtomicU64::new(0),
scanner_cycle_timeout_total: AtomicU64::new(0),
scanner_cycle_recovery_required_total: AtomicU64::new(0),
scanner_cycle_last_progress_age_seconds: AtomicU64::new(0),
scanner_leader_lease_without_progress: AtomicBool::new(false),
scanner_bitrot_cycle_enabled: AtomicBool::new(false),
scanner_bitrot_cycle_millis: AtomicU64::new(0),
scanner_checkpoint: Mutex::new(None),
@@ -2431,29 +2412,12 @@ impl Metrics {
.store(cycle_max_objects.unwrap_or_default(), Ordering::Relaxed);
self.scanner_cycle_max_directories
.store(cycle_max_directories.unwrap_or_default(), Ordering::Relaxed);
self.scanner_leader_lease_without_progress.store(false, Ordering::Relaxed);
self.scanner_cycle_last_progress_age_seconds.store(0, Ordering::Relaxed);
metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(0.0);
metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(0.0);
self.scanner_bitrot_cycle_enabled
.store(bitrot_cycle.is_some(), Ordering::Relaxed);
self.scanner_bitrot_cycle_millis
.store(bitrot_cycle.map(duration_millis_saturated).unwrap_or_default(), Ordering::Relaxed);
}
pub fn record_scanner_cycle_timeout(&self, recovery_required: bool, progress_age: Duration) {
self.scanner_cycle_timeout_total.fetch_add(1, Ordering::Relaxed);
if recovery_required {
self.scanner_cycle_recovery_required_total.fetch_add(1, Ordering::Relaxed);
}
self.scanner_cycle_last_progress_age_seconds
.store(progress_age.as_secs(), Ordering::Relaxed);
self.scanner_leader_lease_without_progress.store(true, Ordering::Relaxed);
metrics::counter!(OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL).increment(1);
metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(progress_age.as_secs_f64());
metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(1.0);
}
pub fn record_scanner_set_scan_state(&self, concurrency_limit: Option<usize>, queued: Option<usize>, active: Option<usize>) {
if let Some(concurrency_limit) = concurrency_limit {
self.scanner_set_scan_concurrency_limit
@@ -3301,10 +3265,6 @@ impl Metrics {
m.cycle_max_duration_seconds = self.scanner_cycle_max_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0;
m.cycle_max_objects = self.scanner_cycle_max_objects.load(Ordering::Relaxed);
m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed);
m.cycle_timeout_total = self.scanner_cycle_timeout_total.load(Ordering::Relaxed);
m.cycle_recovery_required_total = self.scanner_cycle_recovery_required_total.load(Ordering::Relaxed);
m.cycle_last_progress_age = self.scanner_cycle_last_progress_age_seconds.load(Ordering::Relaxed);
m.leader_lease_without_progress = self.scanner_leader_lease_without_progress.load(Ordering::Relaxed);
m.bitrot_cycle_enabled = self.scanner_bitrot_cycle_enabled.load(Ordering::Relaxed);
m.bitrot_cycle_seconds = self.scanner_bitrot_cycle_millis.load(Ordering::Relaxed) as f64 / 1000.0;
m.scan_checkpoint = match self.scanner_checkpoint.lock() {
@@ -4966,20 +4926,4 @@ mod tests {
assert!(!report.bitrot_cycle_enabled);
assert_eq!(report.bitrot_cycle_seconds, 0.0);
}
#[tokio::test]
async fn scanner_cycle_timeout_metrics_reset_for_a_new_cycle() {
let metrics = Metrics::new();
metrics.record_scanner_cycle_timeout(true, Duration::from_secs(17));
let timed_out = metrics.report().await;
assert_eq!(timed_out.cycle_timeout_total, 1);
assert_eq!(timed_out.cycle_last_progress_age, 17);
assert!(timed_out.leader_lease_without_progress);
metrics.record_scanner_cycle_config(Duration::from_secs(60), None, Some(Duration::from_secs(1)), None, None);
let current = metrics.report().await;
assert_eq!(current.cycle_timeout_total, 1);
assert_eq!(current.cycle_last_progress_age, 0);
assert!(!current.leader_lease_without_progress);
}
}
-6
View File
@@ -84,12 +84,6 @@ Current guidance:
- `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` (canonical)
- `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` (canonical)
Scanner cycle budget controls:
- When `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` is unset, the finite default is 1800 seconds (30 minutes), matching the scanner benchmark guidance.
- An explicit `0` preserves the compatibility behavior of an unbounded runtime budget. Object and directory budgets likewise remain unbounded when explicitly set to `0`.
- A timed-out cycle cancels cooperative scanner work, then fences its leader epoch before releasing the lease. An uncooperative I/O operation is dropped after the bounded shutdown window; its cursor is not claimed to be durable and the scanner reports `recovery-required` when the worker cannot stop cooperatively, the cycle state was not confirmed durable, or epoch fencing cannot be persisted.
## Mmap read environment aliases
- `RUSTFS_OBJECT_MMAP_READ_ENABLE` (canonical)
+3 -6
View File
@@ -143,12 +143,9 @@ pub const ENV_SCANNER_MAX_WAIT_SECS: &str = "RUSTFS_SCANNER_MAX_WAIT_SECS";
/// Default scanner speed preset.
pub const DEFAULT_SCANNER_SPEED: &str = "default";
/// Default scanner cycle runtime budget when no override is configured.
///
/// An explicit `0` remains the compatibility escape hatch for an unbounded
/// cycle. Keeping the unset default finite prevents a stalled scanner I/O
/// operation from holding the leader lease forever.
pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 30 * 60;
/// Default scanner cycle runtime budget.
/// `0` keeps the existing unbounded per-cycle behavior.
pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 0;
/// Default scanner per-cycle object budget.
/// `0` keeps the existing unbounded per-cycle behavior.
@@ -256,10 +256,6 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
cycle_max_duration_seconds: metrics.cycle_max_duration_seconds,
cycle_max_objects: metrics.cycle_max_objects,
cycle_max_directories: metrics.cycle_max_directories,
cycle_timeout_total: metrics.cycle_timeout_total,
cycle_recovery_required_total: metrics.cycle_recovery_required_total,
cycle_last_progress_age: metrics.cycle_last_progress_age,
leader_lease_without_progress: metrics.leader_lease_without_progress,
bitrot_cycle_enabled: metrics.bitrot_cycle_enabled,
bitrot_cycle_seconds: metrics.bitrot_cycle_seconds,
scan_checkpoint: metrics.scan_checkpoint.map(|checkpoint| MadminScannerCheckpointReport {
@@ -615,10 +611,6 @@ mod test {
current_started: chrono_to_jiff_timestamp(current_started),
last_cycle_partial_source: "usage".to_string(),
last_cycle_partial_source_code: 1,
cycle_timeout_total: 3,
cycle_recovery_required_total: 2,
cycle_last_progress_age: 17,
leader_lease_without_progress: true,
partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot {
source: "usage".to_string(),
cycles: 2,
@@ -630,10 +622,6 @@ mod test {
assert_eq!(scanner.current_started, chrono_to_jiff_timestamp(current_started));
assert_eq!(scanner.last_cycle_partial_source, "usage");
assert_eq!(scanner.last_cycle_partial_source_code, 1);
assert_eq!(scanner.cycle_timeout_total, 3);
assert_eq!(scanner.cycle_recovery_required_total, 2);
assert_eq!(scanner.cycle_last_progress_age, 17);
assert!(scanner.leader_lease_without_progress);
let usage = scanner
.partial_cycles_by_source
.iter()
+502 -1
View File
@@ -21,6 +21,7 @@ use arc_swap::ArcSwapOption;
use rmp::Marker;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::str::from_utf8;
use std::{
fmt::Debug,
@@ -37,8 +38,10 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::spawn;
use tokio::sync::Mutex;
use tracing::{debug, warn};
use uuid::Uuid;
const SLASH_SEPARATOR: &str = "/";
pub const MAX_META_CACHE_HEAL_CANDIDATES: usize = 1024;
#[derive(Clone, Debug, Default)]
pub struct MetadataResolutionParams {
@@ -66,6 +69,41 @@ pub struct MetaCacheEntry {
pub reusable: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum MetaCacheHealCandidateKind {
Object,
DeleteMarker,
UnversionedObject,
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct MetaCacheHealCandidate {
pub object: String,
pub version_id: Option<Uuid>,
pub kind: MetaCacheHealCandidateKind,
/// Number of raw disk entries that carried this validated version.
pub replica_count: usize,
}
impl MetaCacheHealCandidate {
pub fn validated_version(&self) -> Option<Uuid> {
match self.kind {
MetaCacheHealCandidateKind::Object | MetaCacheHealCandidateKind::DeleteMarker => self.version_id,
MetaCacheHealCandidateKind::UnversionedObject => None,
}
}
pub fn is_unversioned(&self) -> bool {
self.kind == MetaCacheHealCandidateKind::UnversionedObject
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MetaCacheHealDiscovery {
pub candidates: Vec<MetaCacheHealCandidate>,
pub unverified_count: usize,
}
impl MetaCacheEntry {
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
let mut wr = Vec::new();
@@ -370,6 +408,176 @@ impl MetaCacheEntries {
})
}
/// Discover validated object/delete-marker versions and safe unversioned
/// inspection candidates in the raw entries without applying read quorum.
/// This is intentionally separate from [`Self::resolve`]: a sub-quorum
/// version is a valid heal target even though it must not participate in
/// normal reads or writes.
///
/// The validated list is bounded and deduplicated by object, version id,
/// and metadata kind; each candidate retains the number of raw disk
/// entries that carried it so callers can classify sub-quorum versions.
/// Entries whose xl.meta cannot be decoded are counted separately for
/// discovery accounting; they never become versionless destructive heal
/// requests and do not consume the validated quota. An
/// [`MetaCacheHealCandidateKind::UnversionedObject`] is always consumed by
/// a non-destructive scanner request.
pub fn discover_heal_candidates(&self, bucket: &str, max_candidates: usize) -> MetaCacheHealDiscovery {
let limit = max_candidates.min(MAX_META_CACHE_HEAL_CANDIDATES);
if limit == 0 || bucket.is_empty() {
return MetaCacheHealDiscovery::default();
}
let mut discovery = MetaCacheHealDiscovery {
candidates: Vec::<MetaCacheHealCandidate>::with_capacity(limit.min(self.0.len())),
unverified_count: 0,
};
let mut seen: HashMap<(String, Option<Uuid>, MetaCacheHealCandidateKind), usize> =
HashMap::with_capacity(limit.min(self.0.len()));
for entry in self.0.iter().flatten() {
if !valid_heal_candidate_name(bucket, entry) {
continue;
}
let meta = match FileMeta::load(&entry.metadata) {
Ok(meta) => meta,
Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
let mut entry_seen = HashSet::new();
for shallow in meta.versions {
let version = match shallow.parse_version_meta() {
Ok(version) if version.valid() => version,
Ok(_) | Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if version.free_version() {
continue;
}
let payload_header = version.header();
if normalize_version_id(shallow.header.version_id) != normalize_version_id(payload_header.version_id)
|| shallow.header.version_type != payload_header.version_type
{
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
let (kind, version_id) = match version.version_type {
VersionType::Object
if version.object.is_some() && version.delete_marker.is_none() && version.legacy_object.is_none() =>
{
match version.object.as_ref().and_then(|object| object.version_id) {
Some(id) if !id.is_nil() => (MetaCacheHealCandidateKind::Object, Some(id)),
Some(_) | None => (MetaCacheHealCandidateKind::UnversionedObject, None),
}
}
VersionType::Delete
if version.delete_marker.is_some() && version.object.is_none() && version.legacy_object.is_none() =>
{
let Some(id) = version.delete_marker.as_ref().and_then(|marker| marker.version_id) else {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
};
if id.is_nil() {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
(MetaCacheHealCandidateKind::DeleteMarker, Some(id))
}
VersionType::Legacy
if version.legacy_object.is_some() && version.object.is_none() && version.delete_marker.is_none() =>
{
let Some(legacy) = version.legacy_object.as_ref() else {
continue;
};
if legacy.version_id.is_empty() {
(MetaCacheHealCandidateKind::UnversionedObject, None)
} else {
let Ok(id) = Uuid::parse_str(&legacy.version_id) else {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
};
if id.is_nil() {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
(MetaCacheHealCandidateKind::Object, Some(id))
}
}
_ => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if normalize_version_id(payload_header.version_id) != version_id {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
// `all_parts=true` is the trust-boundary check for versioned
// candidates. A null/legacy object may still need the old
// non-destructive inspection fallback when its part arrays
// are parseable but incomplete; never use that fallback for
// a candidate carrying a real version id.
let file_info = match version.clone().into_fileinfo(bucket, &entry.name, true) {
Ok(file_info) => file_info,
Err(_) if version_id.is_none() && matches!(kind, MetaCacheHealCandidateKind::UnversionedObject) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
match version.into_fileinfo(bucket, &entry.name, false) {
Ok(file_info) => file_info,
Err(_) => continue,
}
}
Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if file_info.volume != bucket || file_info.name != entry.name {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
let candidate = MetaCacheHealCandidate {
object: entry.name.clone(),
version_id,
kind,
replica_count: 1,
};
let key = (candidate.object.clone(), candidate.version_id, candidate.kind.clone());
if !entry_seen.contains(&key) {
// Keep per-entry dedupe bounded as well as the global
// candidate union. Once the cap is reached, only keys
// already present in the global map may update replica
// counts; novel versions are accounting-only.
if entry_seen.len() >= limit && !seen.contains_key(&key) {
continue;
}
entry_seen.insert(key.clone());
}
if let Some(index) = seen.get(&key).copied() {
discovery.candidates[index].replica_count = discovery.candidates[index].replica_count.saturating_add(1);
} else {
seen.insert(key, discovery.candidates.len());
discovery.candidates.push(candidate);
if discovery.candidates.len() >= limit {
return discovery;
}
}
}
}
discovery
}
fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option<MetaCacheEntry> {
if self.0.is_empty() {
debug!(
@@ -546,6 +754,28 @@ impl MetaCacheEntries {
}
}
fn valid_heal_candidate_name(bucket: &str, entry: &MetaCacheEntry) -> bool {
if bucket.is_empty() || entry.name.is_empty() || entry.is_dir() || entry.name.contains('\0') {
return false;
}
// Validate raw key components without normalizing them. A dot component
// could otherwise escape the bucket when the key is later mapped back to
// a disk path; a final empty component is retained for valid keys ending
// in '/'.
let mut components = entry.name.split('/').peekable();
while let Some(component) = components.next() {
if component == "." || component == ".." || (component.is_empty() && components.peek().is_some()) {
return false;
}
}
true
}
fn normalize_version_id(version_id: Option<Uuid>) -> Option<Uuid> {
version_id.filter(|id| !id.is_nil())
}
#[derive(Debug, Default)]
pub struct MetaCacheEntriesSortedResult {
pub entries: Option<MetaCacheEntriesSorted>,
@@ -991,7 +1221,7 @@ impl<T: Clone + Debug + Send + Sync + 'static> Cache<T> {
mod tests {
use super::*;
use crate::test_data::create_real_xlmeta;
use crate::{FileMetaVersion, MetaDeleteMarker, TRANSITION_COMPLETE};
use crate::{FileMetaVersion, MetaDeleteMarker, MetaObjectV1, MetaObjectV1Erasure, MetaObjectV1Stat, TRANSITION_COMPLETE};
use std::collections::HashMap;
use std::io::Cursor;
use std::sync::{
@@ -1592,6 +1822,277 @@ mod tests {
);
}
#[test]
fn discover_heal_candidates_keeps_sub_quorum_versions_and_deduplicates() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let entries = MetaCacheEntries(vec![
Some(metacache_entry_single_version(1, now, "one")),
Some(metacache_entry_single_version(2, now, "two")),
Some(metacache_entry_single_version(2, now, "two")),
Some(metacache_entry_single_version(3, now, "three")),
]);
let discovery = entries.discover_heal_candidates("bucket", 16);
let ids: std::collections::HashSet<Uuid> = discovery
.candidates
.iter()
.filter_map(|candidate| candidate.version_id)
.collect();
assert_eq!(
ids,
[Uuid::from_u128(1), Uuid::from_u128(2), Uuid::from_u128(3)]
.into_iter()
.collect()
);
assert_eq!(discovery.candidates.len(), 3, "duplicate tied versions must be emitted once");
assert_eq!(
discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(Uuid::from_u128(2)))
.expect("duplicate version should be discovered")
.replica_count,
2
);
}
#[test]
fn discover_heal_candidates_covers_divergent_quorum_boundaries_n2_n4_n6() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
for (disk_count, quorum) in [(2usize, 1usize), (4, 2), (6, 3)] {
let target_id = Uuid::from_u128(0x1000 + disk_count as u128);
for target_replicas in [quorum.saturating_sub(1), quorum, quorum + 1] {
let entries = (0..disk_count)
.map(|disk| {
let version_id = if disk < target_replicas {
target_id
} else {
Uuid::from_u128(0x2000 + disk as u128)
};
Some(metacache_entry_single_version(version_id.as_u128(), now, "divergent"))
})
.collect();
let discovery = MetaCacheEntries(entries).discover_heal_candidates("bucket", 32);
let target = discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(target_id));
assert_eq!(target.is_some(), target_replicas > 0, "N={disk_count}, replicas={target_replicas}");
if let Some(target) = target {
assert_eq!(target.replica_count, target_replicas);
}
}
}
}
#[test]
fn discover_heal_candidates_separates_delete_markers_and_preserves_unversioned_objects() {
let mut marker_meta = FileMeta::new();
marker_meta
.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(Uuid::from_u128(99)),
deleted: true,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
..Default::default()
})
.expect("delete marker should be added");
let marker = MetaCacheEntry {
name: "object".to_string(),
metadata: marker_meta.marshal_msg().expect("delete marker metadata should marshal"),
cached: Some(marker_meta),
reusable: false,
};
let unversioned_entry = metacache_entry_with_mod_time(
OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"),
"unversioned",
);
let discovery = MetaCacheEntries(vec![Some(marker), Some(unversioned_entry)]).discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::DeleteMarker && candidate.version_id == Some(Uuid::from_u128(99))
}));
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
}
#[test]
fn discover_heal_candidates_skips_free_versions() {
let object_id = Uuid::from_u128(100);
let free_id = Uuid::from_u128(101);
let mut meta = FileMeta::new();
meta.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(object_id),
transition_status: TRANSITION_COMPLETE.to_string(),
transitioned_objname: "remote/object".to_string(),
transition_version_id: Some(Uuid::from_u128(102)),
transition_tier: "WARM".to_string(),
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
})
.expect("transitioned object should be added");
let mut delete = FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(object_id),
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
};
delete.set_tier_free_version_id(&free_id.to_string());
meta.delete_version(&delete).expect("free version should be persisted");
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("free version metadata should marshal"),
cached: Some(meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.is_empty());
}
#[test]
fn discover_heal_candidates_preserves_unversioned_legacy_object() {
let legacy = MetaObjectV1 {
version: "1.0.1".to_string(),
format: "xl".to_string(),
stat: MetaObjectV1Stat {
size: 1,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
name: "object".to_string(),
..Default::default()
},
erasure: MetaObjectV1Erasure {
data_blocks: 4,
parity_blocks: 2,
index: 1,
distribution: vec![1, 2, 3, 4, 5, 6],
..Default::default()
},
..Default::default()
};
let version = FileMetaVersion {
version_type: VersionType::Legacy,
legacy_object: Some(legacy),
..Default::default()
};
let mut meta = FileMeta::new();
meta.versions
.push(FileMetaShallowVersion::try_from(version).expect("legacy metadata should marshal"));
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("legacy metadata should marshal"),
cached: Some(meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
}
#[test]
fn discover_heal_candidates_rejects_nil_and_malformed_metadata_and_is_bounded() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let mut nil = metacache_entry_single_version(1, now, "nil");
let mut nil_meta = FileMeta::load(&nil.metadata).expect("nil fixture should decode");
let mut nil_version = nil_meta.versions[0]
.parse_version_meta()
.expect("nil fixture version should decode");
nil_version.object.as_mut().expect("object fixture").version_id = Some(Uuid::nil());
nil_meta.versions[0] = FileMetaShallowVersion::try_from(nil_version).expect("nil fixture should marshal");
nil.metadata = nil_meta.marshal_msg().expect("nil fixture metadata should marshal");
let mut mismatched = metacache_entry_single_version(2, now, "mismatched");
let mut mismatched_meta = FileMeta::load(&mismatched.metadata).expect("mismatched fixture should decode");
mismatched_meta.versions[0].header.version_id = Some(Uuid::from_u128(200));
mismatched.metadata = mismatched_meta.marshal_msg().expect("mismatched metadata should marshal");
let mut short_parts = metacache_entry_single_version(3, now, "short-parts");
let mut short_parts_meta = FileMeta::load(&short_parts.metadata).expect("short-parts fixture should decode");
let mut short_parts_version = short_parts_meta.versions[0]
.parse_version_meta()
.expect("short-parts fixture version should decode");
let object = short_parts_version.object.as_mut().expect("object fixture");
object.part_numbers = vec![1];
object.part_actual_sizes = vec![1];
object.part_sizes.clear();
short_parts_meta.versions[0] = FileMetaShallowVersion::try_from(short_parts_version).expect("short-parts should marshal");
short_parts.metadata = short_parts_meta.marshal_msg().expect("short-parts metadata should marshal");
let mut short_unversioned = metacache_entry_with_mod_time(now, "short-unversioned");
let mut short_unversioned_meta =
FileMeta::load(&short_unversioned.metadata).expect("short-unversioned fixture should decode");
let mut short_unversioned_version = short_unversioned_meta.versions[0]
.parse_version_meta()
.expect("short-unversioned version should decode");
let unversioned_object = short_unversioned_version.object.as_mut().expect("unversioned object fixture");
unversioned_object.part_numbers = vec![1];
unversioned_object.part_actual_sizes = vec![1];
unversioned_object.part_sizes.clear();
short_unversioned_meta.versions[0] =
FileMetaShallowVersion::try_from(short_unversioned_version).expect("short-unversioned should marshal");
short_unversioned.metadata = short_unversioned_meta
.marshal_msg()
.expect("short-unversioned metadata should marshal");
let mut malformed = nil.clone();
malformed.name = "malformed".to_string();
malformed.metadata = vec![1, 2, 3];
let entries = MetaCacheEntries(
std::iter::once(Some(nil))
.chain(std::iter::once(Some(mismatched)))
.chain(std::iter::once(Some(short_parts)))
.chain(std::iter::once(Some(short_unversioned)))
.chain(std::iter::once(Some(malformed)))
.chain((0..32).map(|id| Some(metacache_entry_single_version(id + 10, now, "bounded"))))
.collect(),
);
let discovery = entries.discover_heal_candidates("bucket", 5);
assert!(discovery.candidates.len() <= 5);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::nil()))
);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::from_u128(2)))
);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::from_u128(3)))
);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
assert!(
discovery.unverified_count >= 1,
"malformed and rejected metadata must remain observable during discovery"
);
for invalid_name in ["../object", "object//", "object\0name"] {
let mut entry = metacache_entry_single_version(400, now, invalid_name);
entry.name = invalid_name.to_string();
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 5);
assert!(
discovery.candidates.is_empty(),
"invalid key should not become a heal candidate: {invalid_name:?}"
);
}
}
#[test]
fn resolve_rejects_partial_latest_and_returns_committed_previous_metadata() {
let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
-16
View File
@@ -689,14 +689,6 @@ pub struct ScannerMetrics {
pub cycle_max_objects: u64,
#[serde(rename = "cycle_max_directories", default)]
pub cycle_max_directories: u64,
#[serde(rename = "cycle_timeout_total", default)]
pub cycle_timeout_total: u64,
#[serde(rename = "cycle_recovery_required_total", default)]
pub cycle_recovery_required_total: u64,
#[serde(rename = "cycle_last_progress_age", default)]
pub cycle_last_progress_age: u64,
#[serde(rename = "leader_lease_without_progress", default)]
pub leader_lease_without_progress: bool,
#[serde(rename = "bitrot_cycle_enabled", default)]
pub bitrot_cycle_enabled: bool,
#[serde(rename = "bitrot_cycle_seconds", default)]
@@ -772,8 +764,6 @@ impl ScannerMetrics {
self.cycle_max_duration_seconds = other.cycle_max_duration_seconds;
self.cycle_max_objects = other.cycle_max_objects;
self.cycle_max_directories = other.cycle_max_directories;
self.cycle_last_progress_age = other.cycle_last_progress_age;
self.leader_lease_without_progress = other.leader_lease_without_progress;
self.bitrot_cycle_enabled = other.bitrot_cycle_enabled;
self.bitrot_cycle_seconds = other.bitrot_cycle_seconds;
}
@@ -867,12 +857,6 @@ impl ScannerMetrics {
.saturating_add(other.last_cycle_replication_checks);
self.last_cycle_usage_saves = self.last_cycle_usage_saves.saturating_add(other.last_cycle_usage_saves);
self.failed_cycles = self.failed_cycles.saturating_add(other.failed_cycles);
self.cycle_timeout_total = self.cycle_timeout_total.saturating_add(other.cycle_timeout_total);
self.cycle_recovery_required_total = self
.cycle_recovery_required_total
.saturating_add(other.cycle_recovery_required_total);
self.cycle_last_progress_age = self.cycle_last_progress_age.max(other.cycle_last_progress_age);
self.leader_lease_without_progress |= other.leader_lease_without_progress;
self.superseded_cycles = self.superseded_cycles.saturating_add(other.superseded_cycles);
self.partial_cycles_unknown = self.partial_cycles_unknown.saturating_add(other.partial_cycles_unknown);
self.partial_cycles_runtime = self.partial_cycles_runtime.saturating_add(other.partial_cycles_runtime);
-33
View File
@@ -125,34 +125,6 @@ pub(crate) async fn read_config_with_revision<S: ScannerObjectIO>(
}
}
/// 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> {
match store
.get_object_reader(
RUSTFS_META_BUCKET,
path,
None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
{
Ok(reader) => reader
.object_info
.etag
.filter(|etag| !etag.is_empty())
.map(DataUsageCacheRevision::Etag)
.ok_or_else(|| StorageError::other(format!("scanner config object {path} has no ETag"))),
Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
Ok(DataUsageCacheRevision::Missing)
}
Err(err) => Err(err),
}
}
#[derive(Clone, Debug)]
pub(crate) struct DataUsageCacheRevisions {
main: DataUsageCacheRevision,
@@ -174,11 +146,6 @@ pub static LEGACY_DATA_USAGE_OBJ_NAME_PATH: LazyLock<String> =
pub static DATA_USAGE_BLOOM_NAME_PATH: LazyLock<String> =
LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_BLOOM_NAME}"));
/// Durable companion object for a cycle-state object which cannot be decoded.
/// The primary object is deliberately never replaced or deleted by recovery.
pub static DATA_USAGE_BLOOM_RECOVERY_PATH: LazyLock<String> =
LazyLock::new(|| format!("{}.recovery-required.json", DATA_USAGE_BLOOM_NAME_PATH.as_str()));
pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}.background-heal.json"));
@@ -74,7 +74,7 @@ impl DataUsageCache {
let loaded = Self::load_cache(store.clone(), name).await?;
let backup = match loaded.backup_revision {
Some(revision) => Some(revision),
None => match read_config_revision(store, &backup_path).await {
None => match Self::revision_for_path(store, &backup_path).await {
Ok(revision) => Some(revision),
Err(err) => {
counter!(METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL).increment(1);
@@ -336,6 +336,33 @@ impl DataUsageCache {
}
}
async fn revision_for_path<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
match store
.get_object_reader(
RUSTFS_META_BUCKET,
path,
None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
{
Ok(reader) => reader
.object_info
.etag
.filter(|etag| !etag.is_empty())
.map(DataUsageCacheRevision::Etag)
.ok_or_else(|| StorageError::other(format!("scanner cache object {path} has no ETag"))),
Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
Ok(DataUsageCacheRevision::Missing)
}
Err(err) => Err(err),
}
}
pub(super) fn cache_save_timeout() -> Duration {
crate::runtime_config::scanner_cache_save_timeout()
}
+1 -4
View File
@@ -75,10 +75,7 @@ pub use remote_scanner::{
};
pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config};
pub use rustfs_common::last_minute;
pub use scanner::{
ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, init_data_scanner,
reset_scanner_cycle_recovery, scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_topology_digest,
};
pub use scanner::{ScannerCycleScheduleStatus, init_data_scanner, scanner_cycle_schedule_status, scanner_topology_digest};
pub use scanner_io::{
ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,
+23 -95
View File
@@ -125,10 +125,7 @@ impl Default for ScannerRuntimeConfig {
cycle_interval_source: ScannerRuntimeConfigSource::Default,
bitrot_cycle: Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS)),
bitrot_cycle_source: ScannerRuntimeConfigSource::Default,
cycle_budget: ScannerCycleBudgetConfig {
max_duration: Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
..Default::default()
},
cycle_budget: ScannerCycleBudgetConfig::default(),
cycle_max_duration_source: ScannerRuntimeConfigSource::Default,
cycle_max_objects_source: ScannerRuntimeConfigSource::Default,
cycle_max_directories_source: ScannerRuntimeConfigSource::Default,
@@ -377,10 +374,7 @@ fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<()
}
validate_optional_config_u64(scanner_kvs, SCANNER_START_DELAY, "")?;
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE, "")?;
if let Some(value) = config_value(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)?;
}
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)?;
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS)?;
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES)?;
if let Some(value) = config_value(heal_kvs, HEAL_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
@@ -442,46 +436,19 @@ fn lookup_max_wait(
Ok((speed.max_sleep(), speed_source))
}
fn lookup_cycle_duration(kvs: Option<&KVS>) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
match rustfs_utils::get_env_parse_outcome::<u64>(ENV_SCANNER_CYCLE_MAX_DURATION_SECS) {
rustfs_utils::EnvParseOutcome::Parsed(secs) => {
return cycle_duration_from_secs(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, secs)
.map(|duration| (duration, ScannerRuntimeConfigSource::Env));
}
rustfs_utils::EnvParseOutcome::Invalid => {
// Do not include the raw environment value in the typed error:
// deployments occasionally put sensitive material in inherited
// environment snapshots. The key still identifies the control.
return Err(invalid_value(
ENV_SCANNER_CYCLE_MAX_DURATION_SECS,
"<invalid>",
"expected unsigned integer seconds",
));
}
rustfs_utils::EnvParseOutcome::Absent => {}
fn lookup_optional_seconds(
kvs: Option<&KVS>,
key: &'static str,
env_key: &'static str,
default: u64,
) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
if let Some(secs) = rustfs_utils::get_env_opt_u64(env_key) {
return Ok((Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Env));
}
if let Some(value) = config_value(kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
return cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)
.map(|duration| (duration, ScannerRuntimeConfigSource::Config));
if let Some(value) = config_value(kvs, key, default) {
return parse_config_u64(key, value).map(|secs| (Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Config));
}
Ok((
Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
ScannerRuntimeConfigSource::Default,
))
}
fn cycle_duration_from_secs(key: &'static str, secs: u64) -> Result<Option<Duration>, ScannerRuntimeConfigError> {
if secs == 0 {
return Ok(None);
}
let duration = Duration::from_secs(secs);
if std::time::Instant::now().checked_add(duration).is_none() {
return Err(invalid_value(key, "<overflow>", "duration exceeds the timer range"));
}
Ok(Some(duration))
Ok((None, ScannerRuntimeConfigSource::Default))
}
fn lookup_start_delay(kvs: Option<&KVS>) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
@@ -586,7 +553,12 @@ pub(crate) fn lookup_scanner_runtime_config(
(speed.cycle_interval(), speed_source)
};
let (cycle_max_duration, cycle_max_duration_source) = lookup_cycle_duration(scanner_kvs)?;
let (cycle_max_duration, cycle_max_duration_source) = lookup_optional_seconds(
scanner_kvs,
SCANNER_CYCLE_MAX_DURATION,
ENV_SCANNER_CYCLE_MAX_DURATION_SECS,
DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS,
)?;
let (cycle_max_objects, cycle_max_objects_source) = lookup_count_budget(
scanner_kvs,
SCANNER_CYCLE_MAX_OBJECTS,
@@ -891,10 +863,10 @@ mod tests {
use rustfs_config::server_config::{Config as ServerConfig, KVS};
use rustfs_config::{
DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE,
SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION,
SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE,
SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE,
SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
};
use std::collections::HashMap;
use std::time::Duration;
@@ -969,50 +941,6 @@ mod tests {
});
}
#[test]
fn scanner_unset_budget_uses_safe_default_but_explicit_zero_is_unbounded() {
let config = server_config_with_scanner(&[]);
with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
assert_eq!(resolved.cycle_budget.max_duration, Some(Duration::from_secs(1800)));
assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Default);
});
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "0")]);
with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
assert_eq!(resolved.cycle_budget.max_duration, None);
assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Config);
});
}
#[test]
fn cycle_budget_invalid_or_overflow_config_is_rejected() {
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("invalid"), || {
let error = lookup_scanner_runtime_config(None).expect_err("invalid duration env must be rejected");
assert!(error.to_string().contains(ENV_SCANNER_CYCLE_MAX_DURATION_SECS));
assert!(error.to_string().contains("<invalid>"));
assert!(!error.to_string().contains(": invalid ("));
});
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551616"), || {
assert!(lookup_scanner_runtime_config(None).is_err());
});
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551615"), || {
assert!(lookup_scanner_runtime_config(None).is_err());
});
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "not-a-duration")]);
assert!(lookup_scanner_runtime_config(Some(&config)).is_err());
}
#[test]
fn scanner_runtime_config_validation_rejects_overflow_persisted_duration() {
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "18446744073709551615")]);
let error = validate_scanner_runtime_config(&config)
.expect_err("persisted duration that exceeds the timer range must be rejected");
assert!(error.to_string().contains(SCANNER_CYCLE_MAX_DURATION));
}
#[test]
fn scanner_runtime_config_normalizes_persisted_default_speed() {
let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]);
+63 -281
View File
@@ -20,7 +20,7 @@ use std::sync::{Arc, LazyLock, RwLock};
use crate::data_usage_define::{
BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DATA_USAGE_OBSERVED_OBJ_NAME_PATH,
DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_revision, read_config_with_revision,
DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision,
};
use crate::runtime_config::{
ScannerRuntimeConfig, ScannerRuntimeConfigSource, refresh_scanner_runtime_config_from_global, scanner_bitrot_cycle,
@@ -52,10 +52,11 @@ use rustfs_config::{
};
use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
use rustfs_data_usage::observed_data_usage_is_newer;
use rustfs_lock::NamespaceLockGuard;
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use tokio::sync::{Notify, mpsc};
#[cfg(test)]
use tokio::sync::Notify;
use tokio::sync::mpsc;
use tokio::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;
@@ -103,13 +104,6 @@ const CLEAN_IDLE_BACKOFF_FACTOR: u32 = 2;
/// unavailable peer cannot drive a tight retry loop.
const SCANNER_RETRY_BASE_INTERVAL: Duration = Duration::from_secs(5);
const SCANNER_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30 * 60);
/// A transient backend outage remains self-healing after the short retry
/// budget is exhausted, but the probe is intentionally sparse until storage
/// recovers or an operator reset wakes the scanner.
const SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL: Duration = Duration::from_secs(5 * 60);
/// Permanent recovery states still get a sparse status probe so a reset that
/// races the wait registration cannot leave the scanner asleep forever.
const SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL: Duration = Duration::from_secs(5 * 60);
const SCANNER_LEADER_LOCK_POLL_INTERVAL: Duration = Duration::from_secs(1);
#[cfg(not(test))]
const SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
@@ -131,12 +125,6 @@ type ScannerCycleStatePersistTestHook = (u64, Arc<Notify>);
static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock<StdMutex<Option<ScannerCycleStatePersistTestHook>>> =
LazyLock::new(|| StdMutex::new(None));
static SCANNER_CYCLE_RECOVERY_WAKE: LazyLock<Notify> = LazyLock::new(Notify::new);
pub(super) fn notify_scanner_cycle_recovery_wake() {
SCANNER_CYCLE_RECOVERY_WAKE.notify_one();
}
#[cfg(test)]
struct ScannerCycleStatePersistTestHookGuard;
@@ -588,21 +576,19 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
tokio::time::sleep(sleep_time).await;
}
let mut transient_backoff = ScannerRetryBackoff::default();
let mut recovery_retry_count = 0_u32;
loop {
if ctx_clone.is_cancelled() {
break;
}
let run_result = run_data_scanner_with_maintenance_state(
if let Err(e) = run_data_scanner_with_maintenance_state(
ctx_clone.clone(),
storeapi_clone.clone(),
startup_features,
startup_maintenance_generation,
)
.await;
if let Err(e) = &run_result {
.await
{
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
@@ -613,52 +599,11 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
"Scanner runtime iteration failed"
);
}
let recovery_status = scanner_cycle_recovery_status();
if recovery_status.retryable {
recovery_retry_count = recovery_retry_count.saturating_add(1);
let _ = record_scanner_cycle_recovery_retry(recovery_retry_count);
} else {
recovery_retry_count = 0;
}
let recovery_status = scanner_cycle_recovery_status();
if recovery_status.state == "paused" {
transient_backoff.record_retryable_cycle(false);
tokio::select! {
_ = ctx_clone.cancelled() => break,
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
_ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL) => {},
}
recovery_retry_count = 0;
continue;
}
if !recovery_status.retryable
&& matches!(recovery_status.state.as_str(), "blocked" | "recovery-required" | "cleanup-pending")
{
transient_backoff.record_retryable_cycle(false);
tokio::select! {
_ = ctx_clone.cancelled() => break,
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
_ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL) => {},
}
continue;
}
let retry_delay = if recovery_status.retryable || run_result.is_err() {
transient_backoff.record_retryable_cycle(true);
transient_backoff
.retry_interval(scanner_cycle_interval())
.unwrap_or(SCANNER_RETRY_BASE_INTERVAL)
} else {
transient_backoff.record_retryable_cycle(false);
randomized_cycle_delay()
};
// Backoff before retrying after lock contention or scanner-level failures.
// Keep this cancellation-aware so shutdown is not delayed by backoff sleep.
tokio::select! {
_ = ctx_clone.cancelled() => break,
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
_ = tokio::time::sleep(retry_delay) => {}
_ = tokio::time::sleep(randomized_cycle_delay()) => {}
}
}
});
@@ -1038,116 +983,20 @@ fn data_usage_persist_timeout() -> Duration {
DataUsageCache::persistence_timeout()
}
#[cfg(not(test))]
const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30);
#[cfg(test)]
const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_millis(50);
async fn fence_scanner_epoch_after_cycle_timeout<Store, LockLost>(
ctx: &CancellationToken,
storeapi: Arc<Store>,
cycle_info: &mut CurrentCycle,
cycle_revision: &mut DataUsageCacheRevision,
leader_epoch: &mut u64,
lock_lost: LockLost,
) -> bool
where
Store: ScannerObjectIO,
LockLost: Future<Output = ()>,
{
let fence_ctx = ctx.child_token();
let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch);
tokio::pin!(claim);
tokio::pin!(lock_lost);
tokio::select! {
biased;
_ = &mut lock_lost => {
fence_ctx.cancel();
false
}
result = tokio::time::timeout(SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT, &mut claim) => {
result.unwrap_or(false) && !fence_ctx.is_cancelled()
}
}
}
struct ScannerCycleDeadlineState<'a> {
cycle_info: &'a mut CurrentCycle,
cycle_revision: &'a mut DataUsageCacheRevision,
leader_epoch: &'a mut u64,
cycle_budget: &'a ScannerCycleBudget,
}
fn cycle_timeout_requires_recovery(worker_stopped: bool, cycle_state_persisted: bool, generation_fenced: bool) -> bool {
!worker_stopped || !cycle_state_persisted || !generation_fenced
}
async fn handle_scanner_cycle_deadline<Store>(
ctx: &CancellationToken,
storeapi: Arc<Store>,
state: ScannerCycleDeadlineState<'_>,
worker_stopped: bool,
guard: &mut NamespaceLockGuard,
) where
Store: ScannerObjectIO,
{
let fenced = fence_scanner_epoch_after_cycle_timeout(
ctx,
storeapi,
state.cycle_info,
state.cycle_revision,
state.leader_epoch,
guard.lock_lost_notified(),
)
.await;
let cycle_state_persisted = state.cycle_budget.cycle_state_persisted();
let recovery_required = cycle_timeout_requires_recovery(worker_stopped, cycle_state_persisted, fenced);
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "cycle_timeout",
worker_stopped,
cycle_state_persisted,
generation_fenced = fenced,
recovery_required,
"Scanner cycle deadline expired; durable cursor/generation fencing completed when possible"
);
global_metrics().record_scanner_cycle_timeout(recovery_required, state.cycle_budget.progress_age());
// Stop renewing before releasing the lease. A new leader can then claim the
// higher persisted generation instead of inheriting the expired worker.
guard.release();
global_metrics().set_cycle(None).await;
}
async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard: &mut ScannerCycleMetricsGuard) {
cycle_info.current = 0;
global_metrics().clear_current_scan_mode();
cycle_metrics_guard.finish(cycle_info.clone()).await;
}
#[cfg(test)]
#[instrument(skip_all)]
#[hotpath::measure]
async fn run_data_scanner_cycle(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
cycle_info: &mut CurrentCycle,
cycle_revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
) -> ScannerCycleOutcome {
let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config());
run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await
}
#[instrument(skip_all)]
#[hotpath::measure]
async fn run_data_scanner_cycle_with_budget(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
cycle_info: &mut CurrentCycle,
cycle_revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
cycle_budget: Arc<ScannerCycleBudget>,
) -> ScannerCycleOutcome {
let _activity_guard = ScannerActivityGuard::new();
if let Err(err) = refresh_scanner_runtime_config_from_global() {
@@ -1163,11 +1012,7 @@ async fn run_data_scanner_cycle_with_budget(
}
let configured_cycle_interval = scanner_cycle_interval();
let configured_bitrot_cycle = scanner_bitrot_cycle();
let cycle_budget_config = ScannerCycleBudgetConfig {
max_duration: cycle_budget.max_duration(),
max_objects: cycle_budget.max_objects(),
max_directories: cycle_budget.max_directories(),
};
let cycle_budget_config = scanner_cycle_budget_config();
let usage_persist_timeout = data_usage_persist_timeout();
global_metrics().record_scanner_cycle_config(
configured_cycle_interval,
@@ -1238,6 +1083,7 @@ async fn run_data_scanner_cycle_with_budget(
let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
let done_cycle = Metrics::time(Metric::ScanCycle);
let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config);
let scan_result = storeapi
.clone()
.nsscanner_with_status(
@@ -1377,7 +1223,7 @@ async fn run_data_scanner_cycle_with_budget(
"Scanner cycle is recovering to a newer durable cache generation"
);
emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
let persisted = persist_required_scanner_cycle_floor(
return if persist_required_scanner_cycle_floor(
ctx,
storeapi.clone(),
cycle_info,
@@ -1386,9 +1232,8 @@ async fn run_data_scanner_cycle_with_budget(
required_cycle,
&mut cycle_metrics_guard,
)
.await;
return if persisted {
cycle_budget.mark_cycle_state_persisted();
.await
{
ScannerCycleOutcome::Partial
} else {
ScannerCycleOutcome::Failed
@@ -1446,7 +1291,7 @@ async fn run_data_scanner_cycle_with_budget(
scan_cycle_partial_reason(budget_reason),
scan_cycle_partial_source(budget_reason),
);
let persisted = finalize_partial_scan_cycle(
return if finalize_partial_scan_cycle(
ctx,
storeapi.clone(),
cycle_info,
@@ -1454,9 +1299,8 @@ async fn run_data_scanner_cycle_with_budget(
leader_epoch,
&mut cycle_metrics_guard,
)
.await;
return if persisted {
cycle_budget.mark_cycle_state_persisted();
.await
{
ScannerCycleOutcome::Partial
} else {
ScannerCycleOutcome::Failed
@@ -1531,7 +1375,7 @@ async fn run_data_scanner_cycle_with_budget(
);
}
emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
let persisted = finalize_partial_scan_cycle(
return if finalize_partial_scan_cycle(
ctx,
storeapi.clone(),
cycle_info,
@@ -1539,9 +1383,8 @@ async fn run_data_scanner_cycle_with_budget(
leader_epoch,
&mut cycle_metrics_guard,
)
.await;
return if persisted {
cycle_budget.mark_cycle_state_persisted();
.await
{
ScannerCycleOutcome::Partial
} else {
ScannerCycleOutcome::Failed
@@ -1582,7 +1425,6 @@ async fn run_data_scanner_cycle_with_budget(
)
.await
{
cycle_budget.mark_cycle_state_persisted();
emit_scan_cycle_superseded(cycle_start.elapsed());
return ScannerCycleOutcome::Superseded;
}
@@ -1615,7 +1457,6 @@ async fn run_data_scanner_cycle_with_budget(
emit_scan_cycle_complete(false, cycle_start.elapsed());
return ScannerCycleOutcome::Failed;
}
cycle_budget.mark_cycle_state_persisted();
done_cycle();
emit_scan_cycle_complete(true, cycle_start.elapsed());
@@ -1680,7 +1521,7 @@ async fn run_data_scanner_with_maintenance_state(
) -> Result<(), ScannerError> {
reset_scanner_cycle_schedule();
// Acquire leader lock (write lock) to ensure only one scanner runs
let mut guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
let guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await {
Ok(guard) => {
record_scanner_leader_lock_state("acquired");
@@ -1765,22 +1606,40 @@ async fn run_data_scanner_with_maintenance_state(
observe_scanner_activity(&storeapi, distributed, &mut scanner_activity_seen).await;
}
let (mut cycle_info, mut leader_epoch, mut cycle_revision) =
match load_scanner_cycle_state_for_startup(storeapi.clone()).await {
ScannerCycleStateStartup::Ready {
cycle,
leader_epoch,
revision,
} => (cycle, leader_epoch, revision),
ScannerCycleStateStartup::Blocked => {
global_metrics().set_cycle(None).await;
return Ok(());
}
ScannerCycleStateStartup::Transient(err) => {
global_metrics().set_cycle(None).await;
return Err(err);
}
};
let (buf, mut cycle_revision) = match read_config_with_revision(storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await {
Ok((buf, revision)) => (buf.unwrap_or_default(), revision),
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "revision_load_failed",
error = %err,
"Scanner cycle state revision load failed"
);
global_metrics().set_cycle(None).await;
return Ok(());
}
};
let (mut cycle_info, mut leader_epoch) = match decode_scanner_cycle_state_for_startup(&buf) {
Ok(state) => state,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "cycle_decode_failed",
error = %err,
"Scanner stopped because persisted cycle state is invalid"
);
global_metrics().set_cycle(None).await;
return Ok(());
}
};
let usage_floor = match persisted_usage_floor(storeapi.clone()).await {
Ok(floor) => floor,
Err(err) => {
@@ -1845,49 +1704,13 @@ async fn run_data_scanner_with_maintenance_state(
return Ok(());
}
let cycle_ctx = ctx.child_token();
let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config());
let initial_outcome = match await_scanner_cycle_with_budget_fence(
let initial_outcome = await_scanner_cycle_with_lock_fence(
&cycle_ctx,
&cycle_budget,
run_data_scanner_cycle_with_budget(
&cycle_ctx,
&storeapi,
&mut cycle_info,
&mut cycle_revision,
leader_epoch,
cycle_budget.clone(),
),
run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch),
guard.lock_lost_notified(),
)
.await
{
ScannerCycleWaitOutcome::Completed(outcome) => outcome,
ScannerCycleWaitOutcome::LockLost => {
record_scanner_leader_lock_lost("Scanner leader lock lost during the initial cycle").await;
global_metrics().set_cycle(None).await;
return Ok(());
}
ScannerCycleWaitOutcome::Cancelled => {
global_metrics().set_cycle(None).await;
return Ok(());
}
ScannerCycleWaitOutcome::Deadline { worker_stopped } => {
handle_scanner_cycle_deadline(
&ctx,
storeapi.clone(),
ScannerCycleDeadlineState {
cycle_info: &mut cycle_info,
cycle_revision: &mut cycle_revision,
leader_epoch: &mut leader_epoch,
cycle_budget: &cycle_budget,
},
worker_stopped,
&mut guard,
)
.await;
return Ok(());
}
};
.unwrap_or(ScannerCycleOutcome::Failed);
superseded_backoff.record_retryable_cycle(initial_outcome == ScannerCycleOutcome::Superseded);
deferred_backoff.record_retryable_cycle(matches!(initial_outcome, ScannerCycleOutcome::Deferred(_)));
dirty_usage_generation_seen = dirty_generation_before_cycle;
@@ -2093,49 +1916,13 @@ async fn run_data_scanner_with_maintenance_state(
}
let dirty_generation_before_cycle = dirty_usage_generation();
let cycle_ctx = ctx.child_token();
let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config());
let outcome = match await_scanner_cycle_with_budget_fence(
let outcome = await_scanner_cycle_with_lock_fence(
&cycle_ctx,
&cycle_budget,
run_data_scanner_cycle_with_budget(
&cycle_ctx,
&storeapi,
&mut cycle_info,
&mut cycle_revision,
leader_epoch,
cycle_budget.clone(),
),
run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch),
guard.lock_lost_notified(),
)
.await
{
ScannerCycleWaitOutcome::Completed(outcome) => outcome,
ScannerCycleWaitOutcome::LockLost => {
record_scanner_leader_lock_lost("Scanner leader lock lost during a scanner cycle").await;
global_metrics().set_cycle(None).await;
return Ok(());
}
ScannerCycleWaitOutcome::Cancelled => {
global_metrics().set_cycle(None).await;
return Ok(());
}
ScannerCycleWaitOutcome::Deadline { worker_stopped } => {
handle_scanner_cycle_deadline(
&ctx,
storeapi.clone(),
ScannerCycleDeadlineState {
cycle_info: &mut cycle_info,
cycle_revision: &mut cycle_revision,
leader_epoch: &mut leader_epoch,
cycle_budget: &cycle_budget,
},
worker_stopped,
&mut guard,
)
.await;
return Ok(());
}
};
.unwrap_or(ScannerCycleOutcome::Failed);
superseded_backoff.record_retryable_cycle(outcome == ScannerCycleOutcome::Superseded);
deferred_backoff.record_retryable_cycle(matches!(outcome, ScannerCycleOutcome::Deferred(_)));
dirty_usage_generation_seen = dirty_generation_before_cycle;
@@ -2432,12 +2219,7 @@ pub(crate) use activity::{
pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
#[cfg(test)]
pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test;
pub use cycle_state::{
ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, reset_scanner_cycle_recovery, scanner_cycle_recovery_status,
};
pub(crate) use cycle_state::{
current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup,
};
pub(crate) use cycle_state::{current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence};
pub use heal_info::{BackgroundHealInfo, read_background_heal_info, save_background_heal_info};
pub use usage_store::store_data_usage_in_backend;
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -196,7 +196,7 @@ pub(super) async fn claim_scanner_leadership(
if ctx.is_cancelled() {
return false;
}
let Some(claimed_epoch) = persisted_epoch.checked_add(1).filter(|epoch| *epoch < u64::MAX) else {
let Some(claimed_epoch) = persisted_epoch.checked_add(1) else {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
File diff suppressed because it is too large Load Diff
+15 -146
View File
@@ -14,16 +14,17 @@
use std::sync::{
Arc,
atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
atomic::{AtomicU8, AtomicU64, Ordering},
};
use tokio::time::{Duration, Instant};
use std::time::Instant;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
const BUDGET_REASON_NONE: u8 = 0;
const BUDGET_REASON_RUNTIME: u8 = 1;
const BUDGET_REASON_OBJECTS: u8 = 2;
const BUDGET_REASON_DIRECTORIES: u8 = 3;
const PROGRESS_CLOCK_SAMPLE_INTERVAL: u64 = 128;
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct ScannerCycleBudgetConfig {
@@ -62,51 +63,29 @@ pub struct ScannerCycleBudget {
token: CancellationToken,
reason: Arc<AtomicU8>,
started_at: Instant,
deadline: Option<Instant>,
max_duration: Option<Duration>,
max_objects: Option<u64>,
max_directories: Option<u64>,
track_progress: bool,
track_unbounded_counts: bool,
objects_scanned: AtomicU64,
directories_started: AtomicU64,
entries_visited: AtomicU64,
last_progress_millis: AtomicU64,
cycle_state_persisted: AtomicBool,
}
impl ScannerCycleBudget {
#[cfg(test)]
pub(crate) fn new(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
Self::new_inner(parent, config, false, false)
Self::new_inner(parent, config, false)
}
pub(crate) fn new_with_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
Self::new_inner(parent, config, true, true)
Self::new_inner(parent, config, true)
}
pub(crate) fn new_with_runtime_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
let track_progress = config.max_duration.is_some();
Self::new_inner(parent, config, track_progress, false)
}
fn new_inner(
parent: &CancellationToken,
config: ScannerCycleBudgetConfig,
track_progress: bool,
track_unbounded_counts: bool,
) -> Arc<Self> {
fn new_inner(parent: &CancellationToken, config: ScannerCycleBudgetConfig, track_progress: bool) -> Arc<Self> {
let token = parent.child_token();
let reason = Arc::new(AtomicU8::new(BUDGET_REASON_NONE));
let started_at = Instant::now();
let deadline = config.max_duration.map(|duration| match started_at.checked_add(duration) {
Some(deadline) => deadline,
// Runtime config rejects this range, but keep programmatic callers
// fail-closed instead of panicking or silently disabling the wall clock.
None => started_at,
});
if let Some(deadline) = deadline {
if let Some(duration) = config.max_duration {
let parent = parent.clone();
let token_wait = token.clone();
let token_cancel = token.clone();
@@ -115,7 +94,7 @@ impl ScannerCycleBudget {
tokio::select! {
_ = parent.cancelled() => {}
_ = token_wait.cancelled() => {}
_ = tokio::time::sleep_until(deadline) => {
_ = tokio::time::sleep(duration) => {
Self::cancel_for_reason(&reason, &token_cancel, ScannerCycleBudgetReason::Runtime);
}
}
@@ -125,18 +104,14 @@ impl ScannerCycleBudget {
Arc::new(Self {
token,
reason,
started_at,
deadline,
started_at: Instant::now(),
max_duration: config.max_duration,
max_objects: config.max_objects,
max_directories: config.max_directories,
track_progress,
track_unbounded_counts,
objects_scanned: AtomicU64::new(0),
directories_started: AtomicU64::new(0),
entries_visited: AtomicU64::new(0),
last_progress_millis: AtomicU64::new(0),
cycle_state_persisted: AtomicBool::new(false),
})
}
@@ -156,14 +131,6 @@ impl ScannerCycleBudget {
self.max_duration
}
pub(crate) fn deadline(&self) -> Option<Instant> {
self.deadline
}
pub(crate) fn cancel_for_runtime(&self) {
self.cancel_for(ScannerCycleBudgetReason::Runtime);
}
pub(crate) fn max_objects(&self) -> Option<u64> {
self.max_objects
}
@@ -206,43 +173,15 @@ impl ScannerCycleBudget {
self.entries_visited.load(Ordering::Relaxed)
}
pub(crate) fn mark_cycle_state_persisted(&self) {
self.cycle_state_persisted.store(true, Ordering::Release);
}
pub(crate) fn cycle_state_persisted(&self) -> bool {
self.cycle_state_persisted.load(Ordering::Acquire)
}
pub(crate) fn progress_age(&self) -> Duration {
let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
let last_progress = self.last_progress_millis.load(Ordering::Relaxed);
Duration::from_millis(elapsed_millis.saturating_sub(last_progress))
}
fn record_progress_sample(&self, event: u64) {
// Clock reads are sampled at batch/count boundaries; the scanner's
// per-object path does not add a second progress atomic.
if event == 0 || (event != 1 && !event.is_multiple_of(PROGRESS_CLOCK_SAMPLE_INTERVAL)) {
return;
}
let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
self.last_progress_millis.store(elapsed_millis, Ordering::Relaxed);
}
pub(crate) fn record_entries_visited(&self, entries_visited: u64) {
if self.track_progress {
let entries = saturating_fetch_add(&self.entries_visited, entries_visited);
self.record_progress_sample(entries);
saturating_fetch_add(&self.entries_visited, entries_visited);
}
}
pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) {
if self.track_progress || self.max_objects.is_some() {
let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned);
if self.track_progress {
self.record_progress_sample(objects);
}
if self.max_objects.is_some_and(|max_objects| objects >= max_objects) {
self.cancel_for(ScannerCycleBudgetReason::Objects);
}
@@ -250,12 +189,9 @@ impl ScannerCycleBudget {
if self.track_progress || self.max_directories.is_some() {
let directories = saturating_fetch_add(&self.directories_started, directories_started);
if self.track_progress {
self.record_progress_sample(directories);
}
if self
.max_directories
.is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories))
.is_some_and(|max_directories| directories > max_directories)
{
self.cancel_for(ScannerCycleBudgetReason::Directories);
}
@@ -271,17 +207,14 @@ impl ScannerCycleBudget {
}
pub(crate) fn try_start_directory(&self) -> bool {
if self.max_directories.is_none() && !self.track_unbounded_counts {
if !self.track_progress && self.max_directories.is_none() {
return true;
}
let directories = saturating_fetch_add(&self.directories_started, 1);
if self.track_progress {
self.record_progress_sample(directories);
}
if self
.max_directories
.is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories))
.is_some_and(|max_directories| directories > max_directories)
{
self.cancel_for(ScannerCycleBudgetReason::Directories);
return false;
@@ -291,14 +224,11 @@ impl ScannerCycleBudget {
}
pub(crate) fn record_object_scanned(&self) {
if self.max_objects.is_none() && !self.track_unbounded_counts {
if !self.track_progress && self.max_objects.is_none() {
return;
}
let objects = saturating_fetch_add(&self.objects_scanned, 1);
if self.track_progress {
self.record_progress_sample(objects);
}
if self.max_objects.is_some_and(|max_objects| objects >= max_objects) {
self.cancel_for(ScannerCycleBudgetReason::Objects);
}
@@ -329,13 +259,6 @@ fn saturating_fetch_add(value: &AtomicU64, delta: u64) -> u64 {
}
}
fn directory_budget_exhausted(directories: u64, max_directories: u64) -> bool {
// Saturation hides a remote max+1 update when the configured limit is the
// largest representable counter. Treat that boundary as exhausted rather
// than allowing work to continue indefinitely.
directories > max_directories || (directories == u64::MAX && max_directories == u64::MAX)
}
impl Drop for ScannerCycleBudget {
fn drop(&mut self) {
self.token.cancel();
@@ -478,35 +401,6 @@ mod tests {
assert_eq!(directory_budget.reason(), Some(ScannerCycleBudgetReason::Directories));
}
#[test]
fn directory_budget_fails_closed_when_progress_saturates() {
let parent = CancellationToken::new();
let budget = ScannerCycleBudget::new(
&parent,
ScannerCycleBudgetConfig {
max_directories: Some(u64::MAX),
..Default::default()
},
);
budget.record_remote_progress(0, u64::MAX);
assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Directories));
assert!(budget.token().is_cancelled());
let local_budget = ScannerCycleBudget::new(
&parent,
ScannerCycleBudgetConfig {
max_directories: Some(u64::MAX),
..Default::default()
},
);
local_budget.record_remote_progress(0, u64::MAX - 1);
assert!(!local_budget.budget_elapsed());
assert!(!local_budget.try_start_directory());
assert_eq!(local_budget.reason(), Some(ScannerCycleBudgetReason::Directories));
}
#[test]
fn explicit_progress_tracking_counts_unbounded_remote_work_without_cancelling() {
let parent = CancellationToken::new();
@@ -567,29 +461,4 @@ mod tests {
assert!(object_limited.requires_serial_progress_accounting());
assert!(directory_limited.requires_serial_progress_accounting());
}
#[tokio::test(start_paused = true)]
async fn progress_age_uses_virtual_time_and_sampled_progress() {
let parent = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_runtime_progress_tracking(
&parent,
ScannerCycleBudgetConfig {
max_duration: Some(Duration::from_secs(60)),
..Default::default()
},
);
tokio::time::advance(Duration::from_secs(5)).await;
assert_eq!(budget.progress_age(), Duration::from_secs(5));
budget.record_entries_visited(1);
assert_eq!(budget.progress_age(), Duration::ZERO);
tokio::time::advance(Duration::from_secs(2)).await;
for _ in 0..126 {
budget.record_entries_visited(1);
}
assert_eq!(budget.progress_age(), Duration::from_secs(2));
budget.record_entries_visited(1);
assert_eq!(budget.progress_age(), Duration::ZERO);
}
}
+56 -68
View File
@@ -43,7 +43,7 @@ use rustfs_common::metrics::{
UpdateCurrentPathFn, current_path_updater, global_metrics,
};
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_filemeta::{MAX_META_CACHE_HEAL_CANDIDATES, MetaCacheEntries, MetaCacheEntry, MetaCacheHealCandidateKind};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration};
use time::OffsetDateTime;
@@ -96,6 +96,10 @@ const METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL: &str = "rustfs_scanner_ex
const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders_total";
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const METRIC_SCANNER_HEAL_DISCOVERY_CANDIDATES_TOTAL: &str = "rustfs_scanner_heal_discovery_candidates_total";
const METRIC_SCANNER_HEAL_DISCOVERY_SUB_QUORUM_TOTAL: &str = "rustfs_scanner_heal_discovery_sub_quorum_total";
const METRIC_SCANNER_HEAL_DISCOVERY_UNVERIFIED_TOTAL: &str = "rustfs_scanner_heal_discovery_unverified_total";
const METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL: &str = "rustfs_scanner_heal_discovery_queued_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
@@ -883,7 +887,7 @@ impl FolderScanner {
object: Option<String>,
version_id: Option<String>,
request: HealChannelRequest,
) -> Result<(), ScannerError> {
) -> Result<HealAdmissionResult, ScannerError> {
let candidate_type = pending_scanner_heal_candidate_type(kind);
let priority = request.priority;
let scan_mode = request.scan_mode.unwrap_or(self.scan_mode);
@@ -911,7 +915,7 @@ impl FolderScanner {
error = %err,
"Scanner deferred heal request after channel error"
);
return Ok(());
return Ok(HealAdmissionResult::Full);
}
};
self.update_pending_scanner_heal_after_admission(
@@ -923,7 +927,7 @@ impl FolderScanner {
result,
);
if result.is_admitted() {
return Ok(());
return Ok(result);
}
record_high_priority_heal_escalation(candidate_type, priority, result);
@@ -944,7 +948,7 @@ impl FolderScanner {
state = "high_priority_not_admitted",
"Scanner high-priority heal admission failed"
);
Ok(())
Ok(result)
}
pub fn set_heal_object_select(&mut self, prob: u32) {
@@ -1736,14 +1740,7 @@ impl FolderScanner {
break;
}
let mut resolver = MetadataResolutionParams {
dir_quorum: self.disks_quorum,
obj_quorum: self.disks_quorum,
bucket: "".to_string(),
strict: false,
..Default::default()
};
let mut previous_bucket = String::new();
for name in abandoned_children {
if !self.should_heal().await {
break;
@@ -1751,7 +1748,7 @@ impl FolderScanner {
let (bucket, prefix) = path2_bucket_object(name.as_str());
if bucket != resolver.bucket {
if bucket != previous_bucket {
self.send_required_scanner_heal_request(
PendingScannerHealKind::Bucket,
bucket.clone(),
@@ -1760,10 +1757,9 @@ impl FolderScanner {
build_bucket_heal_request(bucket.clone(), HealChannelPriority::High),
)
.await?;
previous_bucket = bucket.clone();
}
resolver.bucket = bucket.clone();
let child_ctx = ctx.child_token();
let (agreed_tx, mut agreed_rx) = mpsc::channel::<String>(1);
@@ -1880,6 +1876,7 @@ impl FolderScanner {
let mut agreed_closed = false;
let mut partial_closed = false;
let mut finished_closed = false;
let mut seen_heal_candidates: HashSet<(String, Option<String>, MetaCacheHealCandidateKind)> = HashSet::new();
loop {
if agreed_closed && partial_closed && finished_closed {
@@ -1904,65 +1901,56 @@ impl FolderScanner {
break;
}
let Some(entry) = resolve_object_heal_entry(&entries, resolver.clone()) else {
continue;
};
let discovery = entries.discover_heal_candidates(&bucket, MAX_META_CACHE_HEAL_CANDIDATES);
counter!(METRIC_SCANNER_HEAL_DISCOVERY_CANDIDATES_TOTAL)
.increment(u64::try_from(discovery.candidates.len()).unwrap_or(u64::MAX));
counter!(METRIC_SCANNER_HEAL_DISCOVERY_SUB_QUORUM_TOTAL).increment(
u64::try_from(
discovery
.candidates
.iter()
.filter(|candidate| candidate.replica_count < disks_quorum)
.count(),
)
.unwrap_or(u64::MAX),
);
counter!(METRIC_SCANNER_HEAL_DISCOVERY_UNVERIFIED_TOTAL).increment(
u64::try_from(discovery.unverified_count).unwrap_or(u64::MAX),
);
(self.update_current_path)(&entry.name).await;
if entry.is_dir() {
continue;
}
let fivs = match entry.file_info_versions(&bucket) {
Ok(fivs) => fivs,
Err(e) => {
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %bucket,
entry = %entry.name,
state = "file_info_versions_failed",
error = %e,
"Scanner list_path_raw failed to resolve file versions"
);
self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(entry.name.clone()),
None,
build_object_heal_request(
bucket.clone(),
entry.name.clone(),
None,
self.scan_mode,
HealChannelPriority::High,
),
)
.await?;
found_objects = true;
for candidate in discovery.candidates {
let version_id = candidate.validated_version().map(|id| id.to_string());
let identity = (candidate.object.clone(), version_id.clone(), candidate.kind.clone());
if seen_heal_candidates.len() >= MAX_META_CACHE_HEAL_CANDIDATES
&& !seen_heal_candidates.contains(&identity)
{
continue;
}
};
for fiv in fivs.versions {
let version_id = fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) });
self.send_required_scanner_heal_request(
if !seen_heal_candidates.insert(identity) {
continue;
}
let mut request = build_object_heal_request(
bucket.clone(),
candidate.object.clone(),
version_id.clone(),
self.scan_mode,
HealChannelPriority::High,
);
if candidate.is_unversioned() {
request.remove_corrupted = Some(false);
}
(self.update_current_path)(&candidate.object).await;
let admission = self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(entry.name.clone()),
version_id.clone(),
build_object_heal_request(
bucket.clone(),
entry.name.clone(),
version_id,
self.scan_mode,
HealChannelPriority::High,
),
Some(candidate.object.clone()),
version_id,
request,
)
.await?;
if admission.is_admitted() {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL).increment(1);
}
found_objects = true;
}
@@ -13,6 +13,8 @@
// limitations under the License.
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
use super::*;
#[cfg(test)]
use rustfs_filemeta::MetadataResolutionParams;
/// Cached folder information for scanning
#[derive(Clone, Debug)]
@@ -88,6 +90,7 @@ pub(super) fn build_object_heal_request(
}
}
#[cfg(test)]
pub(super) fn resolve_object_heal_entry(
entries: &MetaCacheEntries,
resolver: MetadataResolutionParams,
+6 -2
View File
@@ -305,13 +305,17 @@ pub(super) fn build_pending_scanner_heal_request(entry: &PendingScannerHeal) ->
match entry.kind {
PendingScannerHealKind::Bucket => Some(build_bucket_heal_request(entry.bucket.clone(), HealChannelPriority::High)),
PendingScannerHealKind::Object => entry.object.as_ref().map(|object| {
build_object_heal_request(
let mut request = build_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.version_id.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
);
if entry.version_id.is_none() {
request.remove_corrupted = Some(false);
}
request
}),
}
}
+66 -6
View File
@@ -17,7 +17,7 @@ use crate::SCANNER_SLEEPER;
use super::*;
use crate::storage_api::VersionPurgeStatusType;
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
use rustfs_filemeta::{FileInfo, FileMeta};
use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams};
use std::io::Write;
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
@@ -1121,6 +1121,17 @@ fn test_pending_heal_reconstructs_object_request_with_version() {
assert_eq!(request.source, HealRequestSource::Scanner);
}
#[test]
fn test_pending_heal_reconstructs_unversioned_request_without_removal() {
let pending = pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 1, 1);
let request = build_pending_scanner_heal_request(&pending).expect("unversioned object request should rebuild");
assert!(request.object_version_id.is_none());
assert_eq!(request.remove_corrupted, Some(false));
assert_eq!(request.recreate_missing, Some(false));
}
#[test]
fn test_pending_heal_retry_candidates_respect_cap_and_order() {
let pending: Vec<PendingScannerHeal> = (0..(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET + 2))
@@ -1323,6 +1334,20 @@ fn metadata_for_object(bucket: &str, object: &str) -> Vec<u8> {
meta.marshal_msg().expect("test metadata should marshal")
}
fn metadata_for_object_version(bucket: &str, object: &str, version_id: Option<Uuid>) -> Vec<u8> {
let mut file_info = FileInfo::new(object, 4, 2);
file_info.volume = bucket.to_string();
file_info.name = object.to_string();
file_info.version_id = version_id;
file_info.versioned = version_id.is_some();
file_info.mod_time = Some(OffsetDateTime::now_utc());
file_info.size = 1;
let mut meta = FileMeta::new();
meta.add_version(file_info).expect("test metadata version should be accepted");
meta.marshal_msg().expect("test metadata should marshal")
}
async fn write_test_object_metadata(root: &std::path::Path, bucket: &str, object: &str) {
write_test_object_metadata_bytes(root, bucket, object, &metadata_for_object(bucket, object)).await;
}
@@ -1726,12 +1751,21 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
let heal_starts = Arc::new(AtomicUsize::new(0));
let heal_starts_clone = heal_starts.clone();
let healed_versions = Arc::new(Mutex::new(Vec::<Option<String>>::new()));
let healed_versions_clone = healed_versions.clone();
let mut heal_rx =
rustfs_common::heal_channel::init_heal_channel().expect("heal channel should initialize once for scanner tests");
let _heal_responder = tokio::spawn(async move {
while let Some(command) = heal_rx.recv().await {
if let rustfs_common::heal_channel::HealChannelCommand::Start { response_tx, .. } = command {
if let rustfs_common::heal_channel::HealChannelCommand::Start {
request, response_tx, ..
} = command
{
heal_starts_clone.fetch_add(1, Ordering::Relaxed);
healed_versions_clone
.lock()
.expect("heal version capture lock should not be poisoned")
.push(request.object_version_id);
let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
}
}
@@ -1739,13 +1773,18 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
let bucket = "src-archive";
let object = "snapshots/37b3f20d941e2f5e6d99114d9bb2f3e67a8a2e5c9c4c5a1b0d6e7f8091a2b3c4";
let metadata = metadata_for_object(bucket, object);
write_test_object_metadata_bytes(&temp_dir, bucket, object, &metadata).await;
let orphan_version = Uuid::from_u128(0x1934);
let shared_version = Uuid::from_u128(0x1935);
let orphan_metadata = metadata_for_object_version(bucket, object, Some(orphan_version));
let shared_metadata = metadata_for_object_version(bucket, object, Some(shared_version));
write_test_object_metadata_bytes(&temp_dir, bucket, object, &orphan_metadata).await;
let mut expected_metadata = vec![(temp_dir.join(bucket).join(object).join("xl.meta"), orphan_metadata.clone())];
let mut disks = vec![scanner.local_disk.clone()];
for disk_name in ["disk2", "disk3", "disk4"] {
let disk_root = temp_dir.join(disk_name);
write_test_object_metadata_bytes(&disk_root, bucket, object, &metadata).await;
write_test_object_metadata_bytes(&disk_root, bucket, object, &shared_metadata).await;
expected_metadata.push((disk_root.join(bucket).join(object).join("xl.meta"), shared_metadata.clone()));
let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("failed to create extra disk endpoint");
let disk = new_disk(
&endpoint,
@@ -1794,8 +1833,29 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
.new_cache
.checked_flatten(bucket)
.expect("healed cache must contain canonical child links");
assert_eq!(root.objects, 1);
// The fixture intentionally exposes two divergent version histories, so
// the scanner keeps both logical versions visible while discovering heals.
assert_eq!(root.objects, 2);
assert!(heal_starts.load(Ordering::Relaxed) > 0, "test must execute the heal child-link path");
let orphan_version_text = orphan_version.to_string();
assert!(
healed_versions
.lock()
.expect("heal version capture lock should not be poisoned")
.iter()
.any(|version| version.as_deref() == Some(orphan_version_text.as_str())),
"sub-quorum orphan version must be submitted as an exact heal candidate"
);
for (path, expected) in expected_metadata {
assert_eq!(
tokio::fs::read(&path)
.await
.expect("scanner discovery must not delete metadata"),
expected,
"scanner discovery must not modify candidate metadata: {}",
path.display()
);
}
}
#[tokio::test]
-1
View File
@@ -48,7 +48,6 @@ use time::OffsetDateTime;
use tokio::sync::{Mutex, Notify, Semaphore, mpsc};
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tokio_util::task::AbortOnDropHandle;
use tracing::{debug, error, warn};
use crate::ScannerObjectInfo as ObjectInfo;
+4 -4
View File
@@ -314,7 +314,7 @@ impl ScannerIOCache for SetDisks {
let ctx_clone = ctx.clone();
let completed_bucket_count = Arc::new(AtomicUsize::new(0));
let completed_bucket_count_clone = completed_bucket_count.clone();
let collect_bucket_results_fut = AbortOnDropHandle::new(tokio::spawn(async move {
let collect_bucket_results_fut = tokio::spawn(async move {
let mut cancelled = false;
loop {
@@ -333,7 +333,7 @@ impl ScannerIOCache for SetDisks {
}
}
}
}));
});
let mut futs = Vec::new();
@@ -365,7 +365,7 @@ impl ScannerIOCache for SetDisks {
NamespaceScannerWorkerMode::RemoteV4(server_epoch) => Some(server_epoch),
NamespaceScannerWorkerMode::Coordinator => None,
};
futs.push(AbortOnDropHandle::new(tokio::spawn(async move {
futs.push(tokio::spawn(async move {
let remote_session_id = uuid::Uuid::new_v4();
let mut remote_session_sequence = 0_u64;
loop {
@@ -1038,7 +1038,7 @@ impl ScannerIOCache for SetDisks {
);
}
}
})));
}));
}
drop(bucket_tx);
drop(bucket_result_tx);
+2 -2
View File
@@ -242,7 +242,7 @@ impl ScannerIOCycle for ECStore {
results[results_index_clone] = result;
}
});
wait_futs.push(AbortOnDropHandle::new(receiver_fut));
wait_futs.push(receiver_fut);
let scan_plan = ScannerBucketScanPlan {
buckets: set_buckets,
@@ -318,7 +318,7 @@ impl ScannerIOCycle for ECStore {
record_set_scan_failure(&mut first_err, e);
}
});
wait_futs.push(AbortOnDropHandle::new(scanner_fut));
wait_futs.push(scanner_fut);
}
}
+2 -2
View File
@@ -268,7 +268,7 @@ where
.parse::<T>()
.map_err(|_| {
log_once(&format!("env_invalid_value:{used_key}"), || {
format!("Invalid {} value for {used_key}. Treating as unset.", type_name::<T>())
format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::<T>())
});
})
.ok()
@@ -570,7 +570,7 @@ where
Ok(parsed) => EnvParseOutcome::Parsed(parsed),
Err(_) => {
log_once(&format!("env_invalid_value:{used_key}"), || {
format!("Invalid {} value for {used_key}. Treating as unset.", type_name::<T>())
format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::<T>())
});
EnvParseOutcome::Invalid
}
+1 -20
View File
@@ -52,7 +52,7 @@ The `/v3/scanner/status` response reports each effective runtime value with a
| `scanner.max_wait` | `RUSTFS_SCANNER_MAX_WAIT_SECS` | seconds | preset-derived | Caps one scanner sleep. |
| `scanner.cycle` | `RUSTFS_SCANNER_CYCLE` | seconds | preset-derived | Sets the interval between scanner cycles. |
| `scanner.start_delay` | `RUSTFS_SCANNER_START_DELAY_SECS` | seconds | unset | Sets startup delay and, for compatibility, the cycle interval when `scanner.cycle` is unset. |
| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `1800` | Caps one cycle's runtime. An explicit `0` disables this budget. |
| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `0` | Caps one cycle's runtime. `0` disables this budget. |
| `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` | Caps objects processed by one cycle. `0` disables this budget. |
| `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` | Caps directories entered by one cycle. `0` disables this budget. |
| `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
@@ -70,21 +70,6 @@ sleep multiplier, maximum wait, and cycle interval. Use `scanner.delay`,
`scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis
needs a precise override.
When the cycle duration control is unset, RustFS uses a finite 1800-second
(30-minute) default, matching the scanner benchmark guidance. An explicit `0`
preserves the compatibility behavior of an unbounded cycle; object and
directory budgets likewise remain unbounded when explicitly set to `0`. Invalid
or overflowing duration environment values are configuration errors rather than
silent fallback values.
When a finite deadline expires, RustFS cancels cooperative scanner work and
waits only for the existing bounded shutdown window. A non-yielding I/O future
is dropped after that window. RustFS then attempts a higher leadership epoch so
late cycle, usage, cache, and remote writes from the old generation fail closed.
If the worker cannot stop cooperatively, the cycle state was not confirmed
durable, or that epoch fence cannot be durably persisted, the scanner reports
`recovery-required`; it does not claim an uncooperative cursor was saved.
An explicit `scanner.cycle` or `RUSTFS_SCANNER_CYCLE` is a minimum inter-cycle
cadence: dirty-usage notifications do not bypass that configured interval.
The default adaptive policy continues to use dirty-usage notifications to wake
@@ -159,10 +144,6 @@ metrics.maintenance_control.primary_control
metrics.source_work
metrics.replication_repair
metrics.scan_checkpoint
metrics.cycle_timeout_total
metrics.cycle_last_progress_age
metrics.leader_lease_without_progress
metrics.cycle_recovery_required_total
```
## Reading Pacing Pressure
-1
View File
@@ -126,7 +126,6 @@ mod tests {
let _list_remote_target_handler = replication::ListRemoteTargetHandler {};
let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {};
let _scanner_status_handler = scanner::ScannerStatusHandler {};
let _scanner_cycle_state_reset_handler = scanner::ScannerCycleStateResetHandler {};
let _ilm_expiry_status_handler = scanner::IlmExpiryStatusHandler {};
let _manual_transition_handler = ilm_transition::ManualTransitionRunHandler {};
let _manual_transition_status_handler = ilm_transition::ManualTransitionJobStatusHandler {};
+2 -95
View File
@@ -13,11 +13,8 @@
// limitations under the License.
use crate::admin::auth::authorize_admin_request;
use crate::admin::handlers::supervise_admin_mutation;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::{
app_context_from_req, current_object_store_handle_for_context, current_scanner_metrics_report,
};
use crate::admin::runtime_sources::current_scanner_metrics_report;
use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
use crate::server::ADMIN_PREFIX;
use chrono::Utc;
@@ -25,13 +22,11 @@ use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_common::metrics::{ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport};
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
use rustfs_credentials::Credentials;
use rustfs_policy::policy::action::{Action, AdminAction};
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
use serde::Serialize;
const JSON_CONTENT_TYPE: &str = "application/json";
@@ -43,13 +38,6 @@ struct ScannerStatusResponse {
metrics: ScannerMetricsReport,
cycle_schedule: rustfs_scanner::ScannerCycleScheduleStatus,
runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus,
cycle_recovery: rustfs_scanner::ScannerCycleRecoveryStatus,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ScannerCycleResetRequest {
mode: String,
}
#[derive(Debug, Serialize)]
@@ -129,7 +117,6 @@ fn scanner_status_response(
metrics,
cycle_schedule,
runtime_config,
cycle_recovery: rustfs_scanner::scanner::scanner_cycle_recovery_status(),
}
}
@@ -157,11 +144,6 @@ pub fn register_scanner_route(r: &mut S3Router<AdminOperation>) -> std::io::Resu
format!("{ADMIN_PREFIX}/v3/scanner/status").as_str(),
AdminOperation(&ScannerStatusHandler {}),
)?;
r.insert(
Method::POST,
format!("{ADMIN_PREFIX}/v3/scanner/cycle-state/reset").as_str(),
AdminOperation(&ScannerCycleStateResetHandler {}),
)?;
r.insert(
Method::GET,
format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(),
@@ -181,13 +163,6 @@ async fn validate_scanner_status_request(req: &S3Request<Body>) -> S3Result<Cred
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await
}
async fn validate_scanner_reset_request(req: &S3Request<Body>) -> S3Result<Credentials> {
if req.credentials.is_none() {
return Err(s3_error!(InvalidRequest, "missing credentials"));
}
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)]).await
}
fn json_response(body: Vec<u8>) -> S3Result<S3Response<(StatusCode, Body)>> {
let mut headers = HeaderMap::new();
let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE)
@@ -217,37 +192,6 @@ impl Operation for ScannerStatusHandler {
pub struct IlmExpiryStatusHandler {}
pub struct ScannerCycleStateResetHandler {}
#[async_trait::async_trait]
impl Operation for ScannerCycleStateResetHandler {
async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let _cred = validate_scanner_reset_request(&req).await?;
let body = req
.input
.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
.await
.map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
let reset = serde_json::from_slice::<ScannerCycleResetRequest>(&body)
.map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
if reset.mode != "full-rescan" {
return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "reset mode must be full-rescan"));
}
let context = app_context_from_req(&req)
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
let store = current_object_store_handle_for_context(Some(context.as_ref()))
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
supervise_admin_mutation("scanner cycle state reset", async move {
rustfs_scanner::scanner::reset_scanner_cycle_recovery(CancellationToken::new(), store)
.await
.map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?;
Ok::<_, S3Error>(())
})
.await?;
json_response(br#"{"status":"reset","mode":"full-rescan"}"#.to_vec())
}
}
#[async_trait::async_trait]
impl Operation for IlmExpiryStatusHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -293,38 +237,6 @@ mod tests {
assert_eq!(err.message(), Some("missing credentials"));
}
#[tokio::test]
async fn scanner_reset_gate_rejects_missing_credentials() {
let req = S3Request {
input: Body::from(String::new()),
method: Method::POST,
uri: http::Uri::from_static("/rustfs/admin/v3/scanner/cycle-state/reset"),
headers: HeaderMap::new(),
extensions: http::Extensions::new(),
credentials: None,
region: None,
service: None,
trailing_headers: None,
};
let err = validate_scanner_reset_request(&req)
.await
.expect_err("a reset request without credentials must be rejected");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(err.message(), Some("missing credentials"));
}
#[test]
fn admin_reset_requires_full_rescan_or_verified_cursor() {
let full_rescan: ScannerCycleResetRequest =
serde_json::from_str(r#"{"mode":"full-rescan"}"#).expect("full rescan must be accepted");
assert_eq!(full_rescan.mode, "full-rescan");
let cursor: ScannerCycleResetRequest =
serde_json::from_str(r#"{"mode":"cursor"}"#).expect("mode validation belongs to the handler");
assert_ne!(cursor.mode, "full-rescan");
assert!(serde_json::from_str::<ScannerCycleResetRequest>(r#"{"mode":"full-rescan","cursor":"untrusted"}"#).is_err());
}
#[test]
fn scanner_disabled_reason_reports_startup_env_key() {
assert_eq!(scanner_disabled_reason(true), None);
@@ -392,11 +304,6 @@ mod tests {
assert_eq!(encoded["cycle_schedule"]["effective_interval_seconds"], 0);
assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_enabled"], false);
assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_multiplier"], 1);
assert_eq!(encoded["cycle_recovery"]["state"], "healthy");
assert_eq!(
encoded["cycle_recovery"]["quarantine_path"],
rustfs_scanner::DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()
);
}
#[test]
-12
View File
@@ -428,12 +428,6 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive),
admin(
HttpMethod::Post,
"/rustfs/admin/v3/scanner/cycle-state/reset",
CONFIG_UPDATE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/rustfs/admin/v3/ilm/expiry/status",
@@ -2026,12 +2020,6 @@ mod tests {
assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", SET_TIER);
}
#[test]
fn route_policy_requires_config_update_for_scanner_cycle_reset() {
assert_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", CONFIG_UPDATE);
assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", SERVER_INFO);
}
#[test]
fn route_policy_uses_tier_actions_for_transition_routes() {
assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER);
@@ -243,7 +243,6 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
admin_route(Method::GET, "/v3/config"),
admin_route(Method::PUT, "/v3/config"),
admin_route(Method::GET, "/v3/scanner/status"),
admin_route(Method::POST, "/v3/scanner/cycle-state/reset"),
admin_route(Method::GET, "/v3/audit/target/list"),
admin_route_sample(
Method::PUT,
@@ -880,7 +879,6 @@ fn test_register_routes_cover_representative_admin_paths() {
assert_route(&router, Method::GET, &admin_path("/v3/config"));
assert_route(&router, Method::PUT, &admin_path("/v3/config"));
assert_route(&router, Method::GET, &admin_path("/v3/scanner/status"));
assert_route(&router, Method::POST, &admin_path("/v3/scanner/cycle-state/reset"));
assert_route(&router, Method::GET, &admin_path("/v3/ilm/expiry/status"));
assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run"));
assert_route(
@@ -1369,7 +1367,6 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
(Method::GET, compat_admin_alias_path("/v3/config")),
(Method::PUT, compat_admin_alias_path("/v3/config")),
(Method::GET, compat_admin_alias_path("/v3/scanner/status")),
(Method::POST, compat_admin_alias_path("/v3/scanner/cycle-state/reset")),
(Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")),
] {
assert!(