diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 680ba643f..0c0572aa1 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -71,6 +71,8 @@ const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state"; static CACHE_SAVE_METRICS_ONCE: Once = Once::new(); pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1; +pub const DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION: u16 = 1; +const DATA_USAGE_SCAN_CURSOR_MAX_BYTES: usize = 16 * 1024; #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) enum DataUsageCacheRevision { @@ -401,6 +403,54 @@ impl DataUsageScanCheckpoint { } } +/// Durable raw directory-page cursor for a bucket scan. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DataUsageRawEnumerationCursor { + pub version: u16, + pub parent: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_entry: Option, + pub entries_seen: u64, + pub page_digest: [u8; 32], +} + +impl DataUsageRawEnumerationCursor { + pub fn new(parent: String, last_entry: Option, entries_seen: u64, page_digest: [u8; 32]) -> Self { + Self { + version: DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION, + parent, + last_entry, + entries_seen, + page_digest, + } + } + + fn is_valid_for_bucket(&self, bucket: &str) -> bool { + self.version == DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION + && bucket != DATA_USAGE_ROOT + && path_is_in_bucket_scope(bucket, &self.parent) + && self.parent.len() <= DATA_USAGE_SCAN_CURSOR_MAX_BYTES + && self.page_digest != [0; 32] + && match &self.last_entry { + Some(last_entry) => { + !last_entry.is_empty() + && self.entries_seen > 0 + && last_entry.len() <= DATA_USAGE_SCAN_CURSOR_MAX_BYTES + && !last_entry.contains(SLASH_SEPARATOR) + } + None => self.entries_seen == 0, + } + } +} + +fn path_is_in_bucket_scope(bucket: &str, path: &str) -> bool { + path == bucket + || path + .strip_prefix(bucket) + .is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR)) +} + /// Durable scope of a bucket checkpoint, independent of namespace mutation counters. #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] @@ -550,6 +600,8 @@ pub struct DataUsageCacheInfo { #[serde(default)] pub scan_checkpoint: Option, #[serde(default)] + pub scan_raw_enumeration_cursor: Option, + #[serde(default)] pub scan_identity: Option, #[serde(default)] pub scan_progress: Option, @@ -608,6 +660,7 @@ impl Serialize for DataUsageCacheInfo { // Keep this metadata map-encoded so older readers can ignore fields // appended by newer scanner versions during rolling upgrades. let field_count = 16 + + usize::from(self.scan_raw_enumeration_cursor.is_some()) + usize::from(self.scan_identity.is_some()) + usize::from(self.scan_progress.is_some()) + usize::from(self.scan_coverage_receipt.is_some()) @@ -631,6 +684,9 @@ impl Serialize for DataUsageCacheInfo { state.serialize_entry("failed_objects", &self.failed_objects)?; state.serialize_entry("scan_resume_after", &self.scan_resume_after)?; state.serialize_entry("scan_checkpoint", &self.scan_checkpoint)?; + if let Some(cursor) = &self.scan_raw_enumeration_cursor { + state.serialize_entry("scan_raw_enumeration_cursor", cursor)?; + } if let Some(identity) = self.scan_identity { state.serialize_entry("scan_identity", &identity)?; } @@ -838,6 +894,7 @@ impl DataUsageCache { && self.info.snapshot_complete && self.info.scan_progress.is_none() && self.info.scan_checkpoint.is_none() + && self.info.scan_raw_enumeration_cursor.is_none() && self.info.scan_resume_after.is_none() && self.info.scan_coverage_receipt.is_none() && self.info.scan_plan_digest == Some(scan_plan_digest) @@ -862,10 +919,15 @@ impl DataUsageCache { self.info.pending_heals = pending_heals; self.info.size_reconciliation = size_reconciliation; } + if self.validated_raw_enumeration_cursor().is_none() { + self.info.scan_raw_enumeration_cursor = None; + } let cursor_is_valid = (self.info.scan_checkpoint.is_none() + && self.info.scan_raw_enumeration_cursor.is_none() && self.info.scan_resume_after.is_none() && self.info.scan_coverage_receipt.is_none()) - || self.validated_scan_frontier().is_some(); + || self.validated_scan_frontier().is_some() + || self.info.scan_raw_enumeration_cursor.is_some(); if !cursor_is_valid { self.info.scan_progress = None; } @@ -886,6 +948,7 @@ impl DataUsageCache { }); self.info.scan_resume_after = None; self.info.scan_checkpoint = None; + self.info.scan_raw_enumeration_cursor = None; self.info.scan_coverage_receipt = None; } // Old readers do not understand coverage sweeps. An absent plan makes @@ -954,6 +1017,15 @@ impl DataUsageCache { .then_some(receipt.through.as_str()) } + pub(crate) fn validated_raw_enumeration_cursor(&self) -> Option<&DataUsageRawEnumerationCursor> { + let cursor = self.info.scan_raw_enumeration_cursor.as_ref()?; + (self.info.scan_progress.is_some() + && self.info.scan_identity.is_some_and(|identity| identity.is_valid()) + && self.info.source.is_some() + && cursor.is_valid_for_bucket(&self.info.name)) + .then_some(cursor) + } + /// Seal only the frontier supplied by completed traversal, never a restored cursor. pub(crate) fn seal_scan_frontier(&mut self, frontier: Option<&str>) -> Result<(), serde_json::Error> { if self.info.scan_progress.is_none() { diff --git a/crates/scanner/src/data_usage_define/tests.rs b/crates/scanner/src/data_usage_define/tests.rs index 2f8c98723..e59ae00b6 100644 --- a/crates/scanner/src/data_usage_define/tests.rs +++ b/crates/scanner/src/data_usage_define/tests.rs @@ -1133,6 +1133,7 @@ fn test_data_usage_cache_info_unmarshal_old_msgpack_defaults_scan_resume_after() assert_eq!(decoded.failed_objects.get("bad-object"), Some(&11)); assert!(decoded.scan_resume_after.is_none()); assert!(decoded.scan_checkpoint.is_none()); + assert!(decoded.scan_raw_enumeration_cursor.is_none()); assert!(decoded.pending_heals.is_empty()); assert!(decoded.source.is_none()); assert!(!decoded.snapshot_complete); @@ -1172,6 +1173,12 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { skip_healing: true, failed_objects: HashMap::from([("bad-object".to_string(), 11)]), source: Some(DataUsageCacheSource::new(1, 2)), + scan_raw_enumeration_cursor: Some(DataUsageRawEnumerationCursor::new( + "bucket/prefix".to_string(), + Some("last-object".to_string()), + 7, + [7; 32], + )), snapshot_complete: true, scan_plan_digest: Some(TEST_PLAN_DIGEST), scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])), @@ -1192,6 +1199,14 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { let current = DataUsageCache::unmarshal(&buf).expect("Current reader failed to deserialize new cache"); assert_eq!(current.info.leader_epoch, 9); assert_eq!(current.info.source, Some(DataUsageCacheSource::new(1, 2))); + assert_eq!( + current + .info + .scan_raw_enumeration_cursor + .as_ref() + .map(|cursor| cursor.last_entry.as_deref()), + Some(Some("last-object")) + ); assert!(current.info.snapshot_complete); assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST)); assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32]))); @@ -1214,6 +1229,123 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { assert_eq!(decoded.cache.get("bucket").map(|entry| entry.objects), Some(3)); } +fn valid_scan_identity() -> DataUsageScanIdentity { + DataUsageScanIdentity { + version: 1, + bucket_incarnation: uuid::Uuid::new_v4(), + set_layout: TEST_PLAN_DIGEST, + publication_epoch: 5, + tier_registry_generation: 9, + scan_mode: HealScanMode::Normal, + } +} + +fn cache_with_raw_cursor(cursor: DataUsageRawEnumerationCursor) -> DataUsageCache { + DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + leader_epoch: 1, + source: Some(DataUsageCacheSource::new(1, 2)), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + scan_identity: Some(valid_scan_identity()), + tier_registry_generation: Some(9), + scan_progress: Some(DataUsageScanProgress { + started_plan: TEST_PLAN_DIGEST, + requested_plan: TEST_PLAN_DIGEST, + }), + scan_raw_enumeration_cursor: Some(cursor), + ..Default::default() + }, + ..Default::default() + } +} + +#[test] +fn raw_enumeration_cursor_validation_requires_bucket_identity_and_bounded_marker() { + let valid = DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), Some("entry-001".to_string()), 1, [8; 32]); + let cache = cache_with_raw_cursor(valid.clone()); + assert_eq!(cache.validated_raw_enumeration_cursor(), Some(&valid)); + + let page_begin = DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), None, 0, [8; 32]); + let cache = cache_with_raw_cursor(page_begin.clone()); + assert_eq!(cache.validated_raw_enumeration_cursor(), Some(&page_begin)); + + let seen_without_marker = DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), None, 1, [8; 32]); + assert!( + cache_with_raw_cursor(seen_without_marker) + .validated_raw_enumeration_cursor() + .is_none() + ); + + let mut missing_identity = cache_with_raw_cursor(valid.clone()); + missing_identity.info.scan_identity = None; + assert!(missing_identity.validated_raw_enumeration_cursor().is_none()); + + let mut outside_bucket = valid.clone(); + outside_bucket.parent = "other/raw".to_string(); + assert!( + cache_with_raw_cursor(outside_bucket) + .validated_raw_enumeration_cursor() + .is_none() + ); + + let mut future_version = valid.clone(); + future_version.version = DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION + 1; + assert!( + cache_with_raw_cursor(future_version) + .validated_raw_enumeration_cursor() + .is_none() + ); + + let mut zero_digest = valid.clone(); + zero_digest.page_digest = [0; 32]; + assert!( + cache_with_raw_cursor(zero_digest) + .validated_raw_enumeration_cursor() + .is_none() + ); + + let mut nested_marker = valid; + nested_marker.last_entry = Some("child/object".to_string()); + assert!( + cache_with_raw_cursor(nested_marker) + .validated_raw_enumeration_cursor() + .is_none() + ); + + let oversized_marker = + DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), Some("x".repeat(16 * 1024 + 1)), 1, [8; 32]); + assert!( + cache_with_raw_cursor(oversized_marker) + .validated_raw_enumeration_cursor() + .is_none() + ); +} + +#[test] +fn prepare_bucket_checkpoint_preserves_only_valid_raw_enumeration_cursor() { + let identity = valid_scan_identity(); + let source = DataUsageCacheSource::new(1, 2); + let cursor = DataUsageRawEnumerationCursor::new("bucket/raw".to_string(), Some("entry-001".to_string()), 1, [9; 32]); + let mut cache = cache_with_raw_cursor(cursor.clone()); + cache.info.scan_identity = Some(identity); + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity), + DataUsageCachePrepareOutcome::Reused + ); + assert_eq!(cache.info.scan_raw_enumeration_cursor, Some(cursor)); + + let invalid = DataUsageRawEnumerationCursor::new("other/raw".to_string(), Some("entry-001".to_string()), 1, [9; 32]); + let mut cache = cache_with_raw_cursor(invalid); + cache.info.scan_identity = Some(identity); + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity), + DataUsageCachePrepareOutcome::Reused + ); + assert!(cache.info.scan_raw_enumeration_cursor.is_none()); + assert!(cache.info.scan_progress.is_some()); +} + /// Deterministic, fully populated cache used to pin the persisted /// `.usage-cache.bin` wire bytes. Every map/set holds at most one element /// so the map-encoded `marshal_msg` output is byte-stable.