diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 8a3499b58..b66b93ecf 100644 --- a/crates/filemeta/src/metacache.rs +++ b/crates/filemeta/src/metacache.rs @@ -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, 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, } impl MetaCacheEntry { @@ -431,6 +440,8 @@ impl MetaCacheEntries { let mut discovery = MetaCacheHealDiscovery { candidates: Vec::::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, 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] diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 4486c247b..c7dc3178d 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -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, MetaCacheHealCandidateKind)> = HashSet::new(); + let mut seen_truncated_objects: HashSet = 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); } diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index f5cb3507a..d5ee7b219 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/crates/scanner/src/scanner_folder/item_actions.rs @@ -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, diff --git a/crates/scanner/src/scanner_folder/ledger.rs b/crates/scanner/src/scanner_folder/ledger.rs index 040821b82..167a1425d 100644 --- a/crates/scanner/src/scanner_folder/ledger.rs +++ b/crates/scanner/src/scanner_folder/ledger.rs @@ -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 }), } } diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 67eafed19..436dd2db7 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -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 = (0..(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET + 2))