From 0c6314babc7749f7bdcf98d1faa937b50a1459f1 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 19:30:15 +0800 Subject: [PATCH] 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 --- crates/scanner/src/raw_page_index.rs | 114 +++++++++++++++--- crates/scanner/src/scanner_folder.rs | 61 ++++++++-- crates/scanner/src/scanner_folder/tests.rs | 43 ++++++- .../tests/enumeration_restart.rs | 31 ++++- crates/scanner/src/scanner_io/io_cache.rs | 14 +-- .../tests/lifecycle_integration_test.rs | 2 +- .../diagnose_scanner_enumeration_restart.py | 10 +- ...st_diagnose_scanner_enumeration_restart.py | 17 +++ 8 files changed, 242 insertions(+), 50 deletions(-) diff --git a/crates/scanner/src/raw_page_index.rs b/crates/scanner/src/raw_page_index.rs index f263197f2..e861bd5da 100644 --- a/crates/scanner/src/raw_page_index.rs +++ b/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::>(); + 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"); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 1e0edc362..30a03ea48 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -833,9 +833,17 @@ impl RawEnumerationProgress { } fn page_index(&self) -> Option { - 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 { + 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; diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index c9ee80c17..772c11d8b 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/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::::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" ); } diff --git a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs index f04759491..48d06f885 100644 --- a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs +++ b/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}"); } } diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index df28c55f0..4cd995ebf 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/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(), diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 784bffd81..b87069d67 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/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; diff --git a/scripts/diagnose_scanner_enumeration_restart.py b/scripts/diagnose_scanner_enumeration_restart.py index 88f39a485..de9810af0 100644 --- a/scripts/diagnose_scanner_enumeration_restart.py +++ b/scripts/diagnose_scanner_enumeration_restart.py @@ -30,15 +30,21 @@ def validate_report(report, *, round_number, pid, objects, budget): if type(report.get(key)) is not int or report[key] != value: raise ValueError(f"worker report mismatch: {key}") for key in ("raw_entries", "raw_name_bytes", "objects_before", "objects_retained", - "versions_retained", "bytes_retained", "objects_processed"): + "versions_retained", "bytes_retained", "objects_processed", + "raw_page_index_committed_entries", "raw_page_index_indexed_entries"): if type(report.get(key)) is not int or not 0 <= report[key] <= 1048576: raise ValueError(f"invalid bounded counter: {key}") - if report["raw_entries"] == 0: + made_budgeted_object_progress = report["objects_processed"] > 0 + if report["raw_entries"] == 0 and not made_budgeted_object_progress: raise ValueError("nonempty fixture must observe raw entries; budget hook may not have run") if report["raw_entries"] > budget: raise ValueError("raw-entry budget exceeded; no unbudgeted tail is permitted") + if report["objects_processed"] > budget: + raise ValueError("object budget exceeded; no unbudgeted scan tail is permitted") for key in ("raw_first_entry", "raw_last_entry"): value = report.get(key) + if report["raw_entries"] == 0 and made_budgeted_object_progress and value is None: + continue if type(value) is not str or not 0 < len(value.encode("utf-8")) <= 512: raise ValueError(f"invalid raw entry marker: {key}") if type(report.get("snapshot_complete")) is not bool: diff --git a/scripts/test_diagnose_scanner_enumeration_restart.py b/scripts/test_diagnose_scanner_enumeration_restart.py index 208b1d040..718a6cf17 100644 --- a/scripts/test_diagnose_scanner_enumeration_restart.py +++ b/scripts/test_diagnose_scanner_enumeration_restart.py @@ -10,6 +10,8 @@ class ReportTests(unittest.TestCase): return dict(schema=1, round=0, pid=123, objects_expected=4, raw_entry_budget=16, raw_entries=8, raw_name_bytes=64, objects_before=0, objects_retained=4, versions_retained=4, bytes_retained=4, objects_processed=4, + raw_page_index_committed_entries=4, + raw_page_index_indexed_entries=4, raw_first_entry="bucket/object-0000", raw_last_entry="bucket/object-0003/xl.meta", snapshot_complete=True, outcome="complete") @@ -44,6 +46,11 @@ class ReportTests(unittest.TestCase): with self.assertRaises(ValueError): self.validate(report) + report = self.report() + report["objects_processed"] = 17 + with self.assertRaises(ValueError): + self.validate(report) + def test_missing_or_oversized_raw_marker_rejected(self): for value in (None, True, "", "x" * 513): with self.subTest(value=value): @@ -55,9 +62,19 @@ class ReportTests(unittest.TestCase): def test_complete_coverage_without_entry_observation_rejected(self): report = self.report() report["raw_entries"] = 0 + report["objects_processed"] = 0 + report["raw_page_index_committed_entries"] = 3 with self.assertRaises(ValueError): self.validate(report) + def test_budgeted_object_progress_can_consume_all_raw_entries(self): + report = self.report() + report["raw_entries"] = 0 + report["raw_first_entry"] = None + report["raw_last_entry"] = None + report["objects_processed"] = 1 + self.validate(report) + def test_missing_wrong_type_and_negative_counter_rejected(self): for value in (None, True, -1, "8", 1048577): with self.subTest(value=value):