gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-23 12:49:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): bound orphan heal discovery fallback

Browse Source
This commit is contained in:
马登山
2026-08-22 19:35:56 +08:00
parent 27a921372c
commit f40f8179a4
5 changed files with 272 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/filemeta/src/metacache.rs
+102 -24
View File
@@ -42,6 +42,9 @@ use uuid::Uuid;
const SLASH_SEPARATOR: &str = "/";
pub const MAX_META_CACHE_HEAL_CANDIDATES: usize = 1024;
/// Keep truncation continuations bounded while still giving the scanner a
/// safe object-level retry for versions that did not fit in the candidate set.
pub const MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS: usize = 64;
#[derive(Clone, Debug, Default)]
pub struct MetadataResolutionParams {
@@ -102,6 +105,12 @@ impl MetaCacheHealCandidate {
pub struct MetaCacheHealDiscovery {
pub candidates: Vec<MetaCacheHealCandidate>,
pub unverified_count: usize,
pub truncated: bool,
/// Object names whose validated version set exceeded the candidate cap.
/// The scanner retries these names without a version and with destructive
/// healing disabled; this is an explicit bounded continuation, not a
/// version claim.
pub truncated_objects: Vec<String>,
}
impl MetaCacheEntry {
@@ -431,6 +440,8 @@ impl MetaCacheEntries {
let mut discovery = MetaCacheHealDiscovery {
candidates: Vec::<MetaCacheHealCandidate>::with_capacity(limit.min(self.0.len())),
unverified_count: 0,
truncated: false,
truncated_objects: Vec::with_capacity(MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS.min(limit)),
};
let mut seen: HashMap<(String, Option<Uuid>, MetaCacheHealCandidateKind), usize> =
HashMap::with_capacity(limit.min(self.0.len()));
@@ -553,24 +564,30 @@ impl MetaCacheEntries {
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 entry_seen.contains(&key) {
continue;
}
if let Some(index) = seen.get(&key).copied() {
entry_seen.insert(key);
discovery.candidates[index].replica_count = discovery.candidates[index].replica_count.saturating_add(1);
} else if discovery.candidates.len() >= limit {
// Keep the validated candidate list bounded, but retain a
// bounded object-level continuation so the scanner cannot
// silently lose every version of a busy object.
discovery.truncated = true;
if discovery.truncated_objects.len() < MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS
&& !discovery.truncated_objects.iter().any(|object| object == &candidate.object)
{
discovery.truncated_objects.push(candidate.object.clone());
}
// The remaining versions in this raw entry cannot add a
// bounded candidate; avoid parsing a very long history
// after the safe continuation has been recorded.
break;
} else {
entry_seen.insert(key.clone());
seen.insert(key, discovery.candidates.len());
discovery.candidates.push(candidate);
if discovery.candidates.len() >= limit {
return discovery;
}
}
}
}
@@ -755,20 +772,18 @@ 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') {
if bucket.is_empty()
|| entry.name.is_empty()
|| entry.is_dir()
|| entry.name.contains('\\')
|| entry.name.chars().any(char::is_control)
{
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;
}
}
// Keep the S3 key opaque. In particular, do not normalize or reject dot
// components or repeated separators; identity is checked against the
// decoded FileInfo below and the raw key must be preserved exactly.
true
}
@@ -1856,6 +1871,28 @@ mod tests {
);
}
#[test]
fn discover_heal_candidates_does_not_count_duplicate_versions_within_one_entry() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let mut meta = FileMeta::load(&metacache_entry_single_version(1, now, "duplicate").metadata)
.expect("duplicate fixture should decode");
meta.versions.push(meta.versions[0].clone());
let entry = MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("duplicate metadata should marshal"),
cached: Some(meta),
reusable: false,
};
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 16);
let candidate = discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(Uuid::from_u128(1)))
.expect("duplicate fixture should be discovered");
assert_eq!(candidate.replica_count, 1);
}
#[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");
@@ -1919,6 +1956,35 @@ mod tests {
}));
}
#[test]
fn discover_heal_candidates_rejects_delete_markers_without_ids() {
let mut marker_meta = FileMeta::new();
marker_meta
.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
deleted: true,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
..Default::default()
})
.expect("nil delete marker should be added");
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: marker_meta.marshal_msg().expect("nil marker metadata should marshal"),
cached: Some(marker_meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.kind == MetaCacheHealCandidateKind::DeleteMarker)
);
assert!(discovery.unverified_count >= 1);
}
#[test]
fn discover_heal_candidates_skips_free_versions() {
let object_id = Uuid::from_u128(100);
@@ -2056,6 +2122,11 @@ mod tests {
);
let discovery = entries.discover_heal_candidates("bucket", 5);
assert!(discovery.candidates.len() <= 5);
assert!(discovery.truncated, "bounded discovery must expose dropped candidates");
assert!(
discovery.truncated_objects.iter().any(|object| object == "object"),
"bounded discovery must expose an object-level safe continuation"
);
assert!(
!discovery
.candidates
@@ -2082,7 +2153,7 @@ mod tests {
"malformed and rejected metadata must remain observable during discovery"
);
for invalid_name in ["../object", "object//", "object\0name"] {
for invalid_name in ["object\\name", "object\u{0001}name", "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);
@@ -2091,6 +2162,13 @@ mod tests {
"invalid key should not become a heal candidate: {invalid_name:?}"
);
}
for valid_name in ["../object", "object//", "trailing/"] {
let mut entry = metacache_entry_single_version(401, now, valid_name);
entry.name = valid_name.to_string();
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 5);
assert_eq!(discovery.candidates.len(), 1, "raw S3 key should remain opaque: {valid_name:?}");
}
}
#[test]
crates/scanner/src/scanner_folder.rs
+78 -12
View File
@@ -43,7 +43,10 @@ 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::{MAX_META_CACHE_HEAL_CANDIDATES, MetaCacheEntries, MetaCacheEntry, MetaCacheHealCandidateKind};
use rustfs_filemeta::{
MAX_META_CACHE_HEAL_CANDIDATES, MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS, MetaCacheEntries, MetaCacheEntry,
MetaCacheHealCandidateKind,
};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration};
use time::OffsetDateTime;
@@ -100,6 +103,7 @@ const METRIC_SCANNER_HEAL_DISCOVERY_CANDIDATES_TOTAL: &str = "rustfs_scanner_hea
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 METRIC_SCANNER_HEAL_DISCOVERY_TRUNCATED_TOTAL: &str = "rustfs_scanner_heal_discovery_truncated_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
@@ -1877,6 +1881,7 @@ impl FolderScanner {
let mut partial_closed = false;
let mut finished_closed = false;
let mut seen_heal_candidates: HashSet<(String, Option<String>, MetaCacheHealCandidateKind)> = HashSet::new();
let mut seen_truncated_objects: HashSet<String> = HashSet::new();
loop {
if agreed_closed && partial_closed && finished_closed {
@@ -1917,8 +1922,12 @@ impl FolderScanner {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_UNVERIFIED_TOTAL).increment(
u64::try_from(discovery.unverified_count).unwrap_or(u64::MAX),
);
if discovery.truncated {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_TRUNCATED_TOTAL).increment(1);
}
for candidate in discovery.candidates {
let sub_quorum_candidate = candidate.replica_count < disks_quorum;
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
@@ -1929,25 +1938,82 @@ impl FolderScanner {
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);
}
let request = if candidate.is_unversioned() {
build_non_destructive_object_heal_request(
bucket.clone(),
candidate.object.clone(),
self.scan_mode,
HealChannelPriority::High,
)
} else {
build_object_heal_request(
bucket.clone(),
candidate.object.clone(),
version_id.clone(),
self.scan_mode,
HealChannelPriority::High,
)
};
(self.update_current_path)(&candidate.object).await;
let admission = self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(candidate.object.clone()),
version_id,
version_id.clone(),
request,
)
.await?;
if admission.is_admitted() {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL).increment(1);
} else if sub_quorum_candidate {
self.mark_pending_scanner_heal_reason(
PendingScannerHealKind::Object,
&bucket,
Some(&candidate.object),
version_id.as_deref(),
"sub_quorum_metadata",
);
}
found_objects = true;
}
// A bounded candidate union may overflow for an
// object with a very long version history. Keep
// that overflow explicit and issue one safe,
// versionless inspection request per object so
// the dropped versions are not silently treated
// as absent. This continuation is deliberately
// outside the versioned candidate cap and always
// disables destructive cleanup.
for object in discovery.truncated_objects {
if seen_truncated_objects.len() >= MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS
&& !seen_truncated_objects.contains(&object)
{
continue;
}
if !seen_truncated_objects.insert(object.clone()) {
continue;
}
let identity = (object.clone(), None, MetaCacheHealCandidateKind::UnversionedObject);
if !seen_heal_candidates.insert(identity) {
continue;
}
let request = build_non_destructive_object_heal_request(
bucket.clone(),
object.clone(),
self.scan_mode,
HealChannelPriority::High,
);
(self.update_current_path)(&object).await;
let admission = self
.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(object.clone()),
None,
request,
)
.await?;
if admission.is_admitted() {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL).increment(1);
}
crates/scanner/src/scanner_folder/item_actions.rs
+15
View File
@@ -90,6 +90,21 @@ pub(super) fn build_object_heal_request(
}
}
/// Build the versionless inspection request used when discovery cannot prove
/// a destructive version identity (for example an unversioned object or a
/// bounded candidate overflow). The explicit flag is the fail-closed safety
/// boundary; callers must not reconstruct it with the destructive default.
pub(super) fn build_non_destructive_object_heal_request(
bucket: String,
object: String,
scan_mode: HealScanMode,
priority: HealChannelPriority,
) -> HealChannelRequest {
let mut request = build_object_heal_request(bucket, object, None, scan_mode, priority);
request.remove_corrupted = Some(false);
request
}
#[cfg(test)]
pub(super) fn resolve_object_heal_entry(
entries: &MetaCacheEntries,
crates/scanner/src/scanner_folder/ledger.rs
+37 -9
View File
@@ -105,6 +105,29 @@ impl FolderScanner {
}
}
/// Preserve the discovery reason when a candidate could not be admitted
/// immediately. The existing string field is intentionally reused so the
/// scanner's map-encoded cache schema stays backward compatible.
pub(super) fn mark_pending_scanner_heal_reason(
&mut self,
kind: PendingScannerHealKind,
bucket: &str,
object: Option<&str>,
version_id: Option<&str>,
reason: &str,
) {
if let Some(entry) = self
.new_cache
.info
.pending_heals
.iter_mut()
.find(|entry| pending_scanner_heal_matches(entry, kind, bucket, object, version_id))
{
entry.last_admission_reason = reason.to_string();
self.sync_pending_heals();
}
}
pub(super) fn prune_pending_scanner_heals(&mut self) {
let now = Self::now_secs();
let before_expiry = self.new_cache.info.pending_heals.len();
@@ -305,17 +328,22 @@ 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| {
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);
build_non_destructive_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
} else {
build_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.version_id.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
}
request
}),
}
}
crates/scanner/src/scanner_folder/tests.rs
+40
View File
@@ -982,6 +982,21 @@ fn test_build_object_heal_request_omits_nil_version_id() {
assert_eq!(request.recreate_missing, Some(false));
}
#[test]
fn test_build_non_destructive_object_heal_request_disables_removal() {
let request = build_non_destructive_object_heal_request(
"bucket".to_string(),
"path/to/object".to_string(),
HealScanMode::Deep,
HealChannelPriority::High,
);
assert_eq!(request.object_version_id, None);
assert_eq!(request.remove_corrupted, Some(false));
assert_eq!(request.recreate_missing, Some(false));
assert_eq!(request.source, HealRequestSource::Scanner);
}
#[test]
fn test_build_bucket_heal_request_disables_recreate_for_scanner() {
let request = build_bucket_heal_request("bucket".to_string(), HealChannelPriority::Low);
@@ -1132,6 +1147,31 @@ fn test_pending_heal_reconstructs_unversioned_request_without_removal() {
assert_eq!(request.recreate_missing, Some(false));
}
#[tokio::test]
async fn test_pending_heal_reason_preserves_sub_quorum_discovery() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
scanner.update_pending_scanner_heal_after_admission(
PendingScannerHealKind::Object,
"bucket",
Some("object"),
Some("version-a"),
HealScanMode::Deep,
HealAdmissionResult::Full,
);
scanner.mark_pending_scanner_heal_reason(
PendingScannerHealKind::Object,
"bucket",
Some("object"),
Some("version-a"),
"sub_quorum_metadata",
);
assert_eq!(scanner.new_cache.info.pending_heals.len(), 1);
assert_eq!(scanner.new_cache.info.pending_heals[0].last_admission_reason, "sub_quorum_metadata");
}
#[test]
fn test_pending_heal_retry_candidates_respect_cap_and_order() {
let pending: Vec<PendingScannerHeal> = (0..(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET + 2))