Compare commits

..

2 Commits

Author SHA1 Message Date
houseme 357cf866ac fix(api): merge duplicate storage error message branch
Combine the MaxVersionsExceeded and InternalError Io message cases so the mapping keeps the same S3 response semantics without tripping clippy's if_same_then_else lint.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
2026-09-08 02:05:40 +08:00
houseme d5b8f49c9d fix(heal): cleanup consumed MRF replay journals
Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them.

This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
2026-09-08 01:50:11 +08:00
4 changed files with 8 additions and 130 deletions
+8 -31
View File
@@ -846,16 +846,6 @@ impl RawEnumerationProgress {
}
})
}
fn has_checkpointable_page_index(&self) -> bool {
self.page_index().is_some()
}
fn checkpointable_entry_count(&self) -> usize {
self.page_index()
.and_then(|index| index.indexed_entries().ok())
.map_or(0, |entries| entries.len())
}
}
fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) {
@@ -1175,33 +1165,20 @@ impl FolderScanner {
}
fn finish_raw_enumeration_parent(&mut self, parent: &str) {
let scan_root = self.old_cache.info.name.as_str();
self.raw_enumeration_progress.retain(|progress| {
if progress.parent == parent {
return parent == scan_root && progress.has_checkpointable_page_index();
}
!progress
.parent
.strip_prefix(parent)
.is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR))
progress.parent != parent
&& !progress
.parent
.strip_prefix(parent)
.is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR))
});
}
fn take_raw_enumeration_resume_state(&mut self) -> (Option<DataUsageRawEnumerationCursor>, Option<RawEnumerationPageIndex>) {
if self.raw_enumeration_progress.is_empty() {
return (None, None);
match self.raw_enumeration_progress.drain(..).next() {
Some(progress) => (progress.cursor(), progress.page_index()),
None => (None, None),
}
let progress_index = self
.raw_enumeration_progress
.iter()
.enumerate()
.max_by_key(|(index, progress)| (progress.checkpointable_entry_count(), std::cmp::Reverse(*index)))
.map(|(index, _)| index)
.unwrap_or(0);
let progress = self.raw_enumeration_progress.swap_remove(progress_index);
self.raw_enumeration_progress.clear();
(progress.cursor(), progress.page_index())
}
fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) {
@@ -3512,80 +3512,6 @@ fn raw_enumeration_progress_checkpoint_commits_budgeted_page_for_oracle() {
);
}
#[tokio::test]
async fn raw_enumeration_root_page_survives_child_partial_boundary() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(temp_dir),
};
scanner.old_cache.info.name = "bucket".to_string();
let mut root_progress = RawEnumerationProgress::new("bucket", None);
root_progress.record_entry("object-0000");
root_progress.record_entry("object-0001");
scanner.raw_enumeration_progress.push(root_progress);
scanner.finish_raw_enumeration_parent("bucket");
assert_eq!(scanner.raw_enumeration_progress.len(), 1);
let root_index = scanner.raw_enumeration_progress[0]
.page_index()
.expect("completed scan root should retain its raw-page oracle");
assert_eq!(
root_index
.committed_entries()
.expect("retained root raw-page oracle should validate"),
vec!["object-0000".to_string(), "object-0001".to_string()]
);
let mut child_progress = RawEnumerationProgress::new("bucket/object-0000", None);
child_progress.record_entry("xl.meta");
scanner.raw_enumeration_progress.push(child_progress);
scanner.finish_raw_enumeration_parent("bucket/object-0000");
assert_eq!(
scanner
.raw_enumeration_progress
.iter()
.map(|progress| progress.parent.as_str())
.collect::<Vec<_>>(),
vec!["bucket"]
);
}
#[tokio::test]
async fn raw_enumeration_resume_state_keeps_largest_durable_quantum() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(temp_dir),
};
let mut root_progress = RawEnumerationProgress::new("bucket", None);
root_progress.record_entry("object-0000");
scanner.raw_enumeration_progress.push(root_progress);
let mut child_progress = RawEnumerationProgress::new("bucket/object-0000", None);
child_progress.record_entry("part-0000");
child_progress.record_entry("part-0001");
child_progress.record_entry("part-0002");
scanner.raw_enumeration_progress.push(child_progress);
let (cursor, page_index) = scanner.take_raw_enumeration_resume_state();
assert_eq!(
cursor.as_ref().expect("largest raw quantum should include a cursor").parent,
"bucket/object-0000"
);
assert_eq!(
page_index
.as_ref()
.expect("largest raw quantum should include a page index")
.indexed_entries()
.expect("selected page index should validate")
.len(),
3
);
assert!(scanner.raw_enumeration_progress.is_empty());
}
#[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");
@@ -85,20 +85,15 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
raise ValueError("no scanner restart reports were produced")
previous = None
made_enumeration_progress = False
made_raw_page_commit_progress = False
made_classification_progress = False
made_durable_progress = False
for index, report in enumerate(reports):
validate_report(report, round_number=index, pid=report["pid"], objects=objects, budget=budget)
if report["raw_page_index_parent"] == "bucket" and report["raw_page_index_committed_entries"] > 0:
made_raw_page_commit_progress = True
if previous is not None:
if report["objects_before"] != previous["objects_retained"]:
raise ValueError("durable retained coverage did not survive process restart")
if report["objects_retained"] < previous["objects_retained"]:
raise ValueError("durable retained coverage regressed across restart")
if replays_raw_window(previous, report):
raise ValueError("raw enumeration window replayed without durable coverage")
if (report["raw_page_index_parent"] == previous["raw_page_index_parent"]
and report["raw_page_index_committed_entries"] < previous["raw_page_index_committed_entries"]
and not previous["raw_page_index_complete"]):
@@ -109,8 +104,6 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
previous = report
if not made_enumeration_progress:
raise ValueError("restart proof did not exercise raw enumeration")
if not made_raw_page_commit_progress:
raise ValueError("restart proof did not commit a durable raw enumeration page")
if not made_classification_progress:
raise ValueError("restart proof did not exercise object classification")
if not made_durable_progress:
@@ -140,15 +140,6 @@ class ReportTests(unittest.TestCase):
advanced = dict(current, objects_retained=1)
self.assertFalse(replays_raw_window(previous, advanced))
def test_recoverable_quantum_rejects_replayed_raw_window(self):
previous = self.report()
previous.update(objects_retained=0, versions_retained=0, bytes_retained=0,
objects_processed=0, snapshot_complete=False, outcome="partial")
current = dict(previous, round=1, pid=124, objects_before=0)
with self.assertRaisesRegex(ValueError, "raw enumeration window replayed"):
validate_recoverable_quantum([previous, current], objects=4, budget=16, require_converged=False)
def test_recoverable_quantum_requires_three_stage_progress_and_convergence(self):
first = self.report()
first.update(round=0, pid=123, raw_entries=2, raw_page_index_committed_entries=2,
@@ -202,15 +193,6 @@ class ReportTests(unittest.TestCase):
with self.assertRaisesRegex(ValueError, "object classification"):
validate_recoverable_quantum([report], objects=4, budget=16, require_converged=False)
def test_recoverable_quantum_rejects_missing_raw_page_commit(self):
report = self.report()
report.update(snapshot_complete=False, outcome="partial",
raw_page_index_committed_entries=0, raw_page_index_indexed_entries=1,
objects_retained=1, versions_retained=1, bytes_retained=1)
with self.assertRaisesRegex(ValueError, "durable raw enumeration page"):
validate_recoverable_quantum([report], objects=4, budget=16, require_converged=False)
if __name__ == "__main__":
unittest.main()