fix(scanner): consume raw page owner resume oracle (#7391)

Treat persisted raw page owner entries as a validated set, not a read_dir-order prefix, so restart scans can consume committed owner pages without recounting them against the raw enumeration budget. Commit checkpointed partial pages, validate owner parent/generation/digests before the skip oracle, and fail closed on duplicate/corrupt page state.

Extend the real scanner restart driver to enforce fixed raw-entry and object budgets across fresh OS processes, and report owner-index coverage in each worker round.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/raw_page_index.rs

@@ -195,10 +195,10 @@ impl RawEnumerationPageIndex {
         if inner.generation != expected_generation {
             return Err(RawEnumerationPageIndexError::StaleGeneration);
         }
-        let mut entries = normalize_owner_entries(entries)?;
+        let entries = normalize_owner_entries(entries)?;
         let committed_entries = inner.validated_committed_entries()?;
         if inner.complete {
-            if entries != committed_entries {
+            if !entry_sets_match(&entries, &committed_entries) {
                 return Err(RawEnumerationPageIndexError::IdentityMismatch);
             }
             return Ok(RawEnumerationPageBuildOutcome {
@@ -207,23 +207,24 @@ impl RawEnumerationPageIndex {
             });
         }
 
-        if !entries.starts_with(&committed_entries) {
+        if source_complete && !entries_contain_all(&entries, &committed_entries) {
             return Err(RawEnumerationPageIndexError::IdentityMismatch);
         }
 
-        let mut indexed_entries = committed_entries.len();
+        let mut indexed_entries = committed_entries.clone();
         if let Some(building) = &inner.building {
             building.validate(
                 u64::try_from(inner.pages.len()).unwrap_or(u64::MAX),
                 u64::try_from(committed_entries.len()).unwrap_or(u64::MAX),
                 inner.page_entry_limit,
             )?;
-            if !entries[committed_entries.len()..].starts_with(&building.entries) {
+            if source_complete && !entries_contain_all(&entries, &building.entries) {
                 return Err(RawEnumerationPageIndexError::IdentityMismatch);
             }
-            indexed_entries = indexed_entries.saturating_add(building.entries.len());
+            indexed_entries.extend(building.entries.iter().cloned());
         }
-        if source_complete && indexed_entries == entries.len() && inner.building.is_none() {
+        let indexed_entries = normalize_owner_entries(indexed_entries)?;
+        if source_complete && entry_sets_match(&entries, &indexed_entries) && inner.building.is_none() {
             inner.complete = true;
             inner.generation = inner.generation.saturating_add(1);
             return Ok(RawEnumerationPageBuildOutcome {
@@ -232,6 +233,7 @@ impl RawEnumerationPageIndex {
             });
         }
 
+        let mut drop_empty_building = false;
         let ready_to_commit = {
             let page_index = u64::try_from(inner.pages.len()).unwrap_or(u64::MAX);
             let entries_start = u64::try_from(committed_entries.len()).unwrap_or(u64::MAX);
@@ -245,25 +247,39 @@ impl RawEnumerationPageIndex {
                 true
             } else {
                 let remaining_page_slots = inner.page_entry_limit.saturating_sub(building.entries.len());
-                let append_count = max_new_entries
-                    .min(remaining_page_slots)
-                    .min(entries.len().saturating_sub(indexed_entries));
+                let append_entries = entries
+                    .iter()
+                    .filter(|entry| indexed_entries.binary_search(entry).is_err())
+                    .take(max_new_entries.min(remaining_page_slots))
+                    .cloned()
+                    .collect::<Vec<_>>();
+                let append_count = append_entries.len();
                 if append_count == 0 {
                     if source_complete && !building.terminal {
                         building.terminal = true;
                         inner.generation = inner.generation.saturating_add(1);
+                    } else if building.entries.is_empty() {
+                        drop_empty_building = true;
                     }
                 } else {
-                    let source_entries = entries.len();
-                    building
-                        .entries
-                        .extend(entries.drain(indexed_entries..indexed_entries + append_count));
-                    building.terminal = source_complete && indexed_entries.saturating_add(append_count) == source_entries;
+                    building.entries.extend(append_entries);
+                    building.entries.sort();
+                    building.entries.dedup();
+                    let source_entries_indexed = source_complete && {
+                        let mut indexed_after_append = indexed_entries;
+                        indexed_after_append.extend(building.entries.iter().cloned());
+                        let indexed_after_append = normalize_owner_entries(indexed_after_append)?;
+                        entry_sets_match(&indexed_after_append, &entries)
+                    };
+                    building.terminal = source_entries_indexed;
                     inner.generation = inner.generation.saturating_add(1);
                 }
                 !building.entries.is_empty() && (building.terminal || building.entries.len() >= inner.page_entry_limit)
             }
         };
+        if drop_empty_building {
+            inner.building = None;
+        }
 
         Ok(RawEnumerationPageBuildOutcome {
             status: inner.status(),
@@ -370,6 +386,12 @@ impl RawEnumerationPageIndexInner {
         if self.complete && self.pages.last().is_some_and(|page| !page.terminal) {
             return Err(RawEnumerationPageIndexError::CorruptIndex);
         }
+        let mut unique_entries = entries.clone();
+        unique_entries.sort();
+        unique_entries.dedup();
+        if unique_entries.len() != entries.len() {
+            return Err(RawEnumerationPageIndexError::CorruptIndex);
+        }
         Ok(entries)
     }
 
@@ -497,6 +519,14 @@ fn entries_are_normalized(entries: &[String]) -> bool {
             .all(|window| window.first().zip(window.get(1)).is_some_and(|(left, right)| left < right))
 }
 
+fn entries_contain_all(entries: &[String], required: &[String]) -> bool {
+    required.iter().all(|entry| entries.binary_search(entry).is_ok())
+}
+
+fn entry_sets_match(left: &[String], right: &[String]) -> bool {
+    left.len() == right.len() && entries_contain_all(left, right)
+}
+
 fn raw_page_digest(parent: &str, building: &RawEnumerationPageBuilder) -> [u8; 32] {
     let mut digest = Sha256::new();
     update_digest(&mut digest, b"version", &RAW_PAGE_INDEX_VERSION.to_le_bytes());
@@ -751,6 +781,26 @@ mod tests {
         assert_eq!(page.entries(), entries(&["entry-a", "entry-b"]));
     }
 
+    #[test]
+    fn complete_owner_source_missing_committed_entry_fails_closed() {
+        let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");
+        let generation = owner.generation().expect("supported owner should expose generation");
+        owner
+            .ingest_owner_entries(entries(&["entry-a", "entry-b"]), 2, generation)
+            .expect("initial complete source should build a committed page");
+        let generation = owner.generation().expect("supported owner should expose generation");
+        owner
+            .commit_building_page(generation)
+            .expect("initial committed page should validate");
+
+        let generation = owner.generation().expect("supported owner should expose generation");
+        assert_eq!(
+            owner.ingest_owner_entries(entries(&["entry-a", "entry-c"]), 2, generation),
+            Err(RawEnumerationPageIndexError::IdentityMismatch),
+            "only complete source identity can prove a previously committed raw entry disappeared"
+        );
+    }
+
     #[test]
     fn terminal_marker_advances_generation_before_commit() {
         let initial_source = entries(&["entry-a", "entry-b", "entry-c"]);
@@ -866,6 +916,40 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deserialized_duplicate_entries_across_pages_fail_closed() {
+        let mut owner = RawEnumerationPageIndex::new("bucket", 1).expect("page owner should initialize");
+        for source in [entries(&["entry-b"]), entries(&["entry-b", "entry-a"])] {
+            let generation = owner.generation().expect("owner should expose generation");
+            let outcome = owner
+                .ingest_partial_owner_entries(source, 1, generation)
+                .expect("single entry page should build");
+            assert!(outcome.ready_to_commit);
+            let generation = owner.generation().expect("ready page should expose generation");
+            owner
+                .commit_building_page(generation)
+                .expect("single entry page should commit");
+        }
+
+        let encoded = rmp_serde::to_vec(&owner).expect("page index should encode");
+        let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("page index should decode");
+        let RawEnumerationPageIndexState::Supported(inner) = &mut decoded.state else {
+            panic!("decoded owner should be supported");
+        };
+        inner.pages[1] = inner.pages[0].clone();
+
+        assert_eq!(decoded.committed_entries(), Err(RawEnumerationPageIndexError::CorruptIndex));
+        assert_eq!(decoded.indexed_entries(), Err(RawEnumerationPageIndexError::CorruptIndex));
+        assert_eq!(
+            decoded.ingest_partial_owner_entries(
+                entries(&["entry-a", "entry-b"]),
+                1,
+                decoded.generation().expect("decoded owner should expose generation")
+            ),
+            Err(RawEnumerationPageIndexError::CorruptIndex)
+        );
+    }
+
     #[test]
     fn empty_owner_source_becomes_ready_without_empty_page_commit() {
         let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize");

crates/scanner/src/scanner_folder.rs

@@ -833,9 +833,17 @@ impl RawEnumerationProgress {
     }
 
     fn page_index(&self) -> Option<RawEnumerationPageIndex> {
-        self.page_index.clone().and_then(|index| match index.indexed_entries() {
-            Ok(entries) if !entries.is_empty() => Some(index),
-            _ => None,
+        self.page_index.clone().and_then(|mut index| {
+            if let Some(generation) = index.generation()
+                && matches!(index.status(), crate::raw_page_index::RawEnumerationPageOwnerStatus::Building { .. })
+                && index.commit_building_page(generation).is_err()
+            {
+                return None;
+            }
+            match index.indexed_entries() {
+                Ok(entries) if !entries.is_empty() => Some(index),
+                _ => None,
+            }
         })
     }
 }
@@ -1133,6 +1141,29 @@ impl FolderScanner {
         }
     }
 
+    fn raw_enumeration_committed_entry_oracle(&self, parent: &str) -> HashSet<String> {
+        let Some(index) = self.old_cache.validated_raw_enumeration_page_index() else {
+            return HashSet::new();
+        };
+        let generation_matches_parent = match index.status() {
+            crate::raw_page_index::RawEnumerationPageOwnerStatus::Building {
+                generation,
+                parent: index_parent,
+                ..
+            }
+            | crate::raw_page_index::RawEnumerationPageOwnerStatus::Ready {
+                generation,
+                parent: index_parent,
+                ..
+            } => generation > 0 && index_parent == parent,
+            crate::raw_page_index::RawEnumerationPageOwnerStatus::Unsupported => false,
+        };
+        if !generation_matches_parent {
+            return HashSet::new();
+        }
+        index.committed_entries().unwrap_or_default().into_iter().collect()
+    }
+
     fn finish_raw_enumeration_parent(&mut self, parent: &str) {
         self.raw_enumeration_progress.retain(|progress| {
             progress.parent != parent
@@ -1481,6 +1512,7 @@ impl FolderScanner {
             let mut pending_entry_progress = 0_u64;
             let mut last_entry_progress = Instant::now();
             let mut raw_enumeration_complete = false;
+            let raw_enumeration_committed_entries = self.raw_enumeration_committed_entry_oracle(&folder.name);
 
             loop {
                 let entry = match dir_reader.next_entry().await {
@@ -1519,20 +1551,23 @@ impl FolderScanner {
                     }
                     Err(e) => return Err(ScannerError::Io(e)),
                 };
-                #[cfg(test)]
-                tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
-                pending_entry_progress = pending_entry_progress.saturating_add(1);
-                if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
-                    || last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
-                {
-                    self.budget.record_entries_visited(pending_entry_progress);
-                    pending_entry_progress = 0;
-                    last_entry_progress = Instant::now();
-                }
                 let file_name = entry.file_name().to_string_lossy().to_string();
                 if file_name.is_empty() || file_name == "." || file_name == ".." {
                     continue;
                 }
+                let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
+                if !raw_entry_consumed_by_owner_index {
+                    #[cfg(test)]
+                    tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
+                    pending_entry_progress = pending_entry_progress.saturating_add(1);
+                    if pending_entry_progress >= SCANNER_ENTRY_PROGRESS_BATCH
+                        || last_entry_progress.elapsed() >= SCANNER_ENTRY_PROGRESS_INTERVAL
+                    {
+                        self.budget.record_entries_visited(pending_entry_progress);
+                        pending_entry_progress = 0;
+                        last_entry_progress = Instant::now();
+                    }
+                }
                 self.record_raw_enumeration_entry(&folder.name, &file_name);
                 let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
 

crates/scanner/src/scanner_folder/tests.rs

@@ -2733,8 +2733,15 @@ async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_
         1
     );
     assert_eq!(
-        page_index.committed_entries().expect("uncommitted raw page should validate"),
-        Vec::<String>::new()
+        page_index
+            .committed_entries()
+            .expect("checkpointed raw page should validate as committed coverage"),
+        vec![
+            raw_cursor
+                .last_entry
+                .clone()
+                .expect("checkpointed page should include the observed entry")
+        ]
     );
     assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Runtime));
 }
@@ -3410,7 +3417,27 @@ fn raw_enumeration_progress_waits_for_resume_index_floor_before_revalidation() {
 }
 
 #[test]
-fn raw_enumeration_progress_rejects_resume_index_after_floor_mismatch() {
+fn raw_enumeration_progress_checkpoint_commits_budgeted_page_for_oracle() {
+    let mut progress = RawEnumerationProgress::new("bucket", None);
+    progress.record_entry("entry-b");
+
+    let page_index = progress
+        .page_index()
+        .expect("checkpointed raw progress should retain a committed owner page");
+    let page_entries = page_index
+        .committed_entries()
+        .expect("checkpointed owner page should validate by digest");
+    assert_eq!(page_entries, vec!["entry-b".to_string()]);
+    assert_eq!(
+        page_index
+            .indexed_entries()
+            .expect("checkpointed owner index should validate"),
+        page_entries
+    );
+}
+
+#[test]
+fn raw_enumeration_progress_retains_resume_index_until_unordered_entries_reappear() {
     let mut index = RawEnumerationPageIndex::new("bucket", 2).expect("raw page index should initialize");
     let generation = index.generation().expect("raw page index should expose generation");
     index
@@ -3425,7 +3452,13 @@ fn raw_enumeration_progress_rejects_resume_index_after_floor_mismatch() {
 
     progress.record_entry("entry-c");
     assert!(
-        progress.page_index.is_none(),
-        "resume index must be discarded once enough current observations prove source drift"
+        progress.page_index.is_some(),
+        "partial observations must not discard the resume index before an unordered old entry can reappear"
+    );
+
+    progress.record_entry("entry-b");
+    assert!(
+        progress.page_index.is_some(),
+        "same source identity should keep the resume index even when read_dir order changes"
     );
 }

crates/scanner/src/scanner_folder/tests/enumeration_restart.rs

@@ -29,6 +29,9 @@ pub(in crate::scanner_folder) fn observe_raw_entry(dir: &str, name: &std::ffi::O
         let relative_dir = Path::new(dir)
             .strip_prefix(&observation.root)
             .unwrap_or_else(|_| Path::new(""));
+        if relative_dir.components().count() != 1 {
+            return;
+        }
         let entry_marker = relative_dir.join(name).to_string_lossy().to_string();
         observation.first_entry.get_or_insert_with(|| entry_marker.clone());
         observation.last_entry = Some(entry_marker);
@@ -130,7 +133,13 @@ async fn round(request: &Request) -> serde_json::Value {
     .await
     .expect("open synthetic disk in this process");
     let parent = CancellationToken::new();
-    let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default());
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        crate::scanner_budget::ScannerCycleBudgetConfig {
+            max_objects: Some(request.raw_entry_budget),
+            ..Default::default()
+        },
+    );
     let _observation_guard = install_raw_entry_budget(disk.path(), request.raw_entry_budget);
     let result = scan_data_folder(
         budget.token(),
@@ -157,6 +166,16 @@ async fn round(request: &Request) -> serde_json::Value {
     let reloaded = DataUsageCache::unmarshal(&read_bounded(&cache_path).await).expect("reload returned cache codec");
     let retained = reloaded.checked_flatten("bucket").expect("reloaded bucket root");
     let scanned = returned.checked_flatten("bucket").expect("returned bucket root");
+    let raw_page_index_committed_entries = reloaded
+        .validated_raw_enumeration_page_index()
+        .and_then(|index| index.committed_entries().ok())
+        .map(|entries| entries.len())
+        .unwrap_or(0);
+    let raw_page_index_indexed_entries = reloaded
+        .validated_raw_enumeration_page_index()
+        .and_then(|index| index.indexed_entries().ok())
+        .map(|entries| entries.len())
+        .unwrap_or(0);
     assert_eq!(
         (retained.objects, retained.versions, retained.size),
         (scanned.objects, scanned.versions, scanned.size)
@@ -169,6 +188,8 @@ async fn round(request: &Request) -> serde_json::Value {
         "objects_expected": request.objects, "raw_entry_budget": request.raw_entry_budget,
         "raw_entries": observation.entries, "raw_name_bytes": observation.name_bytes,
         "raw_first_entry": observation.first_entry, "raw_last_entry": observation.last_entry,
+        "raw_page_index_committed_entries": raw_page_index_committed_entries,
+        "raw_page_index_indexed_entries": raw_page_index_indexed_entries,
         "objects_processed": budget.progress().0,
         "objects_before": before, "objects_retained": retained.objects,
         "versions_retained": retained.versions, "bytes_retained": retained.size,
@@ -194,16 +215,16 @@ async fn enumeration_restart_worker() {
         let temp = tempfile::tempdir().expect("healthy fixture directory");
         let report = round(&Request {
             workspace: temp.path().to_path_buf(),
-            objects: 4,
+            objects: 8,
             raw_entry_budget: 16,
             round: 0,
         })
         .await;
         assert_eq!(report["outcome"], "complete");
         assert_eq!(report["snapshot_complete"], true);
-        assert_eq!(report["objects_retained"], 4);
-        assert_eq!(report["versions_retained"], 4);
-        assert_eq!(report["bytes_retained"], 4);
+        assert_eq!(report["objects_retained"], 8);
+        assert_eq!(report["versions_retained"], 8);
+        assert_eq!(report["bytes_retained"], 8);
         assert!(report["raw_entries"].as_u64().expect("observed entries") >= 8, "{report}");
     }
 }

crates/scanner/src/scanner_io/io_cache.rs

@@ -574,7 +574,6 @@ impl ScannerIOCache for SetDisks {
             let budget_clone = budget.clone();
             let store_clone_clone = self.clone();
             let bucket_result_tx_clone = bucket_result_tx.clone();
-            let disk_clone = disk.clone();
             let set_disk_inventory_clone = set_disk_inventory.clone();
             let disk_scan_semaphore_clone = disk_scan_semaphore.clone();
             let queued_disk_bucket_scans_clone = queued_disk_bucket_scans.clone();
@@ -622,10 +621,7 @@ impl ScannerIOCache for SetDisks {
                         BucketWorkGuard::new(remaining_bucket_work_clone.clone(), bucket_work_complete_clone.clone());
                     // Prefix hints are process-local. Never hand one to a
                     // remote or legacy-coordinator disk path.
-                    let prefix_scan_scope = disk_clone
-                        .is_local()
-                        .then(|| scope_clone.prefix_scope_for(&bucket.name))
-                        .flatten();
+                    let prefix_scan_scope = disk.is_local().then(|| scope_clone.prefix_scope_for(&bucket.name)).flatten();
 
                     metrics::histogram!(
                         METRIC_SCANNER_DISK_SCAN_WAIT_SECONDS,
@@ -680,7 +676,7 @@ impl ScannerIOCache for SetDisks {
                         };
                         remote_session_sequence = next_sequence;
                         let remote_outcome = crate::remote_scanner::scan_remote_bucket(
-                            &disk_clone,
+                            &disk,
                             ctx_clone.clone(),
                             budget_clone.clone(),
                             crate::remote_scanner::RemoteScannerScanSpec {
@@ -813,8 +809,8 @@ impl ScannerIOCache for SetDisks {
                         continue;
                     }
 
-                    let _local_admission = if disk_clone.is_local() {
-                        match crate::remote_scanner::try_admit_remote_scanner(&disk_clone) {
+                    let _local_admission = if disk.is_local() {
+                        match crate::remote_scanner::try_admit_remote_scanner(&disk) {
                             Ok(admission) => Some(admission),
                             Err(e) => {
                                 if requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await {
@@ -1055,7 +1051,7 @@ impl ScannerIOCache for SetDisks {
                     let before = cache.info.last_update;
 
                     let scan_ctx = ctx_clone.child_token();
-                    let scan = disk_clone.clone().nsscanner_disk(
+                    let scan = disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
                         budget_clone.clone(),
                         set_disk_inventory_clone.as_ref().clone(),

crates/scanner/tests/lifecycle_integration_test.rs

@@ -2552,7 +2552,7 @@ mod serial_tests {
                 .expect("Failed to upload multipart part");
             completed.push(CompletePart {
                 part_num: idx + 1,
-                etag: part.etag.clone(),
+                etag: part.etag,
                 ..Default::default()
             });
             offset += part_size;