diff --git a/.config/scanner-heal-required-tests.json b/.config/scanner-heal-required-tests.json index abf25420c..b7fb4b268 100644 --- a/.config/scanner-heal-required-tests.json +++ b/.config/scanner-heal-required-tests.json @@ -8,10 +8,26 @@ "suite": "e2e_test", "name": "heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_remote_shards_after_background_target_restart", "oracle": "background-target-restart.json", + "evidence": "process-restart", + "unclean_shutdown_marker": false, "min_objects": 9, "max_objects": 65, "topology": {"nodes": 4, "drives_per_node": 1}, "scope": "Target process restart, exact unversioned S3 bodies and replacement-disk shards; not power loss or EC8+4." + }, + "background-target-crash": { + "gate": "G14", + "task": "W21", + "lane": "e2e-nightly", + "suite": "e2e_test", + "name": "heal_erasure_disk_rebuild_test::tests::test_cluster_root_heal_recovers_remote_shards_after_background_target_crash", + "oracle": "background-target-crash.json", + "evidence": "process-crash-restart", + "unclean_shutdown_marker": true, + "min_objects": 9, + "max_objects": 65, + "topology": {"nodes": 4, "drives_per_node": 1}, + "scope": "Target process killed during partial background rebuild, real unclean-shutdown marker, exact unversioned S3 bodies and replacement-disk shards; not power loss or EC8+4." } }, "release_pending": { diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index 917462ef5..1114efda2 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -58,11 +58,22 @@ mod tests { struct ScannerHealEvidenceCase { id: &'static str, oracle: &'static str, + evidence: &'static str, + unclean_shutdown_marker: bool, } const BACKGROUND_TARGET_RESTART_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase { id: "background-target-restart", oracle: "background-target-restart.json", + evidence: "process-restart", + unclean_shutdown_marker: false, + }; + + const BACKGROUND_TARGET_CRASH_EVIDENCE: ScannerHealEvidenceCase = ScannerHealEvidenceCase { + id: "background-target-crash", + oracle: "background-target-crash.json", + evidence: "process-crash-restart", + unclean_shutdown_marker: true, }; struct RestartEvidenceContext { @@ -98,6 +109,8 @@ mod tests { || case.oracle.contains('/') || case.oracle.contains('\\') || case.oracle.contains("..") + || !matches!(case.evidence, "process-restart" | "process-crash-restart") + || (case.evidence == "process-crash-restart") != case.unclean_shutdown_marker { return Err("invalid scanner/heal evidence case".into()); } @@ -950,6 +963,16 @@ mod tests { .await? } + #[tokio::test(flavor = "multi_thread")] + async fn test_cluster_root_heal_recovers_remote_shards_after_background_target_crash() + -> Result<(), Box> { + timeout( + Duration::from_secs(420), + run_cluster_root_heal_interruption(InterruptionScenario::BackgroundTargetCrash), + ) + .await? + } + #[tokio::test(flavor = "multi_thread")] async fn test_cluster_root_heal_recovers_remote_shards_after_coordinator_restart() -> Result<(), Box> { @@ -986,21 +1009,27 @@ mod tests { enum InterruptionScenario { IsolatedTargetRestart, BackgroundTargetRestart, + BackgroundTargetCrash, BackgroundCoordinatorRestart, TargetEndpointBlackhole, } async fn run_cluster_root_heal_interruption(scenario: InterruptionScenario) -> Result<(), Box> { let server_binary = rustfs_binary_path(); - let evidence_run = if scenario == InterruptionScenario::BackgroundTargetRestart { - restart_evidence_run(&server_binary, BACKGROUND_TARGET_RESTART_EVIDENCE)? - } else { - None + let evidence_run = match scenario { + InterruptionScenario::BackgroundTargetRestart => { + restart_evidence_run(&server_binary, BACKGROUND_TARGET_RESTART_EVIDENCE)? + } + InterruptionScenario::BackgroundTargetCrash => { + restart_evidence_run(&server_binary, BACKGROUND_TARGET_CRASH_EVIDENCE)? + } + _ => None, }; let mut evidence_objects = Vec::new(); let (background_enabled, interruption_node, interruption_kind) = match scenario { InterruptionScenario::IsolatedTargetRestart => (false, 1, "target_restart"), InterruptionScenario::BackgroundTargetRestart => (true, 1, "background_target_restart"), + InterruptionScenario::BackgroundTargetCrash => (true, 1, "background_target_crash"), InterruptionScenario::BackgroundCoordinatorRestart => (true, 0, "coordinator_restart"), InterruptionScenario::TargetEndpointBlackhole => (false, 1, "target_endpoint_blackhole"), }; @@ -1067,6 +1096,7 @@ mod tests { .unwrap_or(4 * 1024 * 1024) .clamp(1024 * 1024, 16 * 1024 * 1024); let mut expected_manifests = Vec::with_capacity(online_object_count); + let mut unclean_shutdown_marker_observed = None; for index in 0..online_object_count { let key = format!("cluster/online/object-{index:04}.bin"); let payload_seed = u8::try_from(index + 1).expect("clamped object count must fit in u8"); @@ -1433,7 +1463,11 @@ mod tests { "Restored target endpoint forwarding" ); } else { - cluster.stop_node(interruption_node)?; + if scenario == InterruptionScenario::BackgroundTargetRestart { + cluster.stop_node_gracefully(interruption_node).await?; + } else { + cluster.stop_node(interruption_node)?; + } let stopped_count = metadata_count(&replaced_disk, bucket, &expected_manifests); assert!( stopped_count > 0 && stopped_count < expected_manifests.len(), @@ -1449,9 +1483,12 @@ mod tests { .join(".rustfs.sys") .join("unclean-shutdown"); if background_enabled { + let marker_exists = unclean_shutdown_marker.is_file(); + unclean_shutdown_marker_observed = Some(marker_exists); + let expected_marker = !matches!(scenario, InterruptionScenario::BackgroundTargetRestart); assert!( - unclean_shutdown_marker.is_file(), - "background restart must retain the real unclean-shutdown marker" + marker_exists == expected_marker, + "background restart/crash lane observed unexpected unclean-shutdown marker state" ); } else { match std::fs::remove_file(&unclean_shutdown_marker) { @@ -1651,13 +1688,14 @@ mod tests { "server build changed during restart" ); let evidence = serde_json::json!({ - "schema": 1, "case": evidence_context.case.id, "evidence": "process-restart", + "schema": 1, "case": evidence_context.case.id, "evidence": evidence_context.case.evidence, "run_id": evidence_context.run.run_id, "source_revision": evidence_context.run.source_revision, "test_build": compiled_test_identity(), "binary_sha256": evidence_context.run.binary.sha256, "test_binary_sha256": evidence_context.run.test_binary.sha256, "topology": {"nodes": cluster.nodes.len(), "drives_per_node": cluster.nodes[0].data_dirs.len()}, "pid_before": target_pid, "pid_after": restarted_pid, + "unclean_shutdown_marker": unclean_shutdown_marker_observed.unwrap_or(false), "objects": evidence_objects, "node_listings": node_listings, }); let data = serde_json::to_vec(&evidence)?; diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index e555573ae..3e72dcf5d 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -16899,6 +16899,87 @@ mod tests { assert!(err.to_string().contains("requires 60 bytes, but 59 bytes are available")); } + async fn single_pool_capacity_admission_test_store() -> (Vec, Arc) { + let (temp_dirs, store) = + crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta::default()).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + (temp_dirs, store) + } + + #[tokio::test] + #[serial_test::serial] + async fn single_pool_public_writes_skip_decommission_capacity_admission() { + let (_temp_dirs, store) = single_pool_capacity_admission_test_store().await; + let bucket = format!("single-pool-capacity-skip-{}", uuid::Uuid::new_v4()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create single-pool bucket before blocking pool metadata writes"); + let incarnation = store.bucket_incarnation_id(&bucket).await.expect("load bucket incarnation"); + store.pool_meta_save_gate.lock().await.block_writes_after_fence_loss(); + + let object = "ordinary-put.bin"; + let mut put_data = crate::object_api::PutObjReader::from_vec(b"ordinary single-pool body".to_vec()); + store + .put_object(&bucket, object, &mut put_data, &ObjectOptions::default()) + .await + .expect("single-pool ordinary PUT must not enter decommission capacity admission"); + store + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("single-pool ordinary PUT must remain readable"); + + let multipart_object = "ordinary-multipart.bin"; + let upload = store + .new_multipart_upload( + &bucket, + multipart_object, + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + ) + .await + .expect("single-pool MPU creation must not enter decommission capacity admission"); + let mut part_data = crate::object_api::PutObjReader::from_vec(b"single-pool multipart body".to_vec()); + let part = store + .put_object_part( + &bucket, + multipart_object, + &upload.upload_id, + 1, + &mut part_data, + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + ) + .await + .expect("single-pool UploadPart must not enter decommission capacity admission"); + store + .clone() + .complete_multipart_upload( + &bucket, + multipart_object, + &upload.upload_id, + vec![crate::storage_api_contracts::multipart::CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }], + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + ) + .await + .expect("single-pool CompleteMultipartUpload must not enter decommission capacity admission"); + store + .get_object_info(&bucket, multipart_object, &ObjectOptions::default()) + .await + .expect("single-pool completed MPU must remain readable"); + } + #[tokio::test] #[serial_test::serial] async fn multipart_mutations_locate_later_upload_before_reserved_pool_admission() { diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index 1b54b4899..7c6d927f3 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -29,7 +29,12 @@ use rustfs_heal::heal::{ storage::{ECStoreHealStorage, HealStorageAPI}, }; use serial_test::serial; -use std::{path::Path, process::Command, sync::Arc, time::Duration}; +use std::{ + path::{Path, PathBuf}, + process::{Command, Stdio}, + sync::Arc, + time::Duration, +}; mod storage_api; @@ -110,6 +115,48 @@ fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16] body } +fn scoped_journal_record( + kind: u8, + bucket: &str, + object: &str, + version: Option<[u8; 16]>, + attempts: u8, + pool_index: u32, + set_index: u32, +) -> Vec { + let mut body = vec![1u8, 2, kind, attempts]; + body.extend_from_slice(&1_700_000_000_000u64.to_le_bytes()); + match version { + Some(bytes) => { + body.push(1); + body.extend_from_slice(&bytes); + } + None => body.push(0), + } + body.extend_from_slice(&pool_index.to_le_bytes()); + body.extend_from_slice(&set_index.to_le_bytes()); + body.extend_from_slice( + &u32::try_from(bucket.len()) + .expect("fixture bucket length must fit journal format") + .to_le_bytes(), + ); + body.extend_from_slice( + &u32::try_from(object.len()) + .expect("fixture object length must fit journal format") + .to_le_bytes(), + ); + body.extend_from_slice(bucket.as_bytes()); + body.extend_from_slice(object.as_bytes()); + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&body); + body.extend_from_slice( + &u32::try_from(hasher.finalize()) + .expect("CRC32 must fit the journal checksum field") + .to_le_bytes(), + ); + body +} + fn write_journal_path_to_disks(disk_paths: &[std::path::PathBuf], relative_path: &str, data: &[u8]) { for path in disk_paths { let journal = path.join(META_BUCKET).join(relative_path); @@ -128,6 +175,12 @@ fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: .all(|path| Path::new(path).join(META_BUCKET).join(relative_path).exists()) } +fn journal_matches_on_all_disks(disk_paths: &[PathBuf], relative_path: &str, expected: &[u8]) -> bool { + disk_paths + .iter() + .all(|path| std::fs::read(path.join(META_BUCKET).join(relative_path)).is_ok_and(|actual| actual == expected)) +} + async fn wait_until(deadline: Duration, mut probe: F) -> bool where F: FnMut() -> Fut, @@ -259,6 +312,25 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() { !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() })); + + let scoped_v2 = scoped_journal_record(1, "scoped-v2-bucket", "scoped-v2-object", None, 0, 3, 7); + let stale_legacy = journal_record(1, "stale-legacy-bucket", "stale-legacy-object", None, 0); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &scoped_v2); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &stale_legacy); + assert_eq!( + mrf_queue::replay_journal_once(&manager).await, + 1, + "a scoped v2 authoritative epoch must not be merged with a stale v1 legacy mirror" + ); + assert_eq!( + manager.operations_snapshot().await.queued_by_source.mrf, + 3, + "only the three authoritative/scoped-only epochs should have reached the manager" + ); + assert!(disk_paths.iter().all(|path| { + !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + })); } /// If replay reaches a full heal-manager queue, the old journal remains the @@ -348,6 +420,99 @@ fn mrf_journal_child_process_fixture() { std::process::exit(77); } +#[test] +fn mrf_successor_flush_child_process_fixture() { + let Ok(root) = std::env::var("RUSTFS_MRF_SUCCESSOR_FLUSH_CHILD_ROOT") else { + return; + }; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("child runtime should build"); + runtime.block_on(async { + let (disk_paths, storage) = heal_env_at(Some(Path::new(&root))).await; + register_local_disks(&disk_paths, "mrf-successor-flush-child").await; + + let mut startup = journal_record(1, "successor-bucket", "first-object", None, 0); + startup.extend(journal_record(1, "successor-bucket", "second-object", None, 0)); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &startup); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &startup); + + let manager = Arc::new(HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )); + mrf_queue::spawn_mrf_consumer(manager.clone()); + let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + let flushed = wait_until(Duration::from_secs(10), || async { + manager.operations_snapshot().await.queued_by_source.mrf == 1 + && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor) + && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor) + }) + .await; + assert!( + flushed, + "child process must publish the pending successor snapshot before the delete phase" + ); + }); + std::process::exit(78); +} + +#[test] +#[cfg(unix)] +fn mrf_successor_flush_waiting_child_process_fixture() { + let Ok(root) = std::env::var("RUSTFS_MRF_SUCCESSOR_KILL_CHILD_ROOT") else { + return; + }; + let ready_path = std::env::var("RUSTFS_MRF_SUCCESSOR_KILL_READY") + .map(PathBuf::from) + .expect("ready marker path should be provided"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("child runtime should build"); + runtime.block_on(async { + let (disk_paths, storage) = heal_env_at(Some(Path::new(&root))).await; + register_local_disks(&disk_paths, "mrf-successor-kill-child").await; + + let mut startup = journal_record(1, "service-kill-bucket", "first-object", None, 0); + startup.extend(journal_record(1, "service-kill-bucket", "second-object", None, 0)); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &startup); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &startup); + + let manager = Arc::new(HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )); + mrf_queue::spawn_mrf_consumer(manager.clone()); + let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + let flushed = wait_until(Duration::from_secs(10), || async { + manager.operations_snapshot().await.queued_by_source.mrf == 1 + && journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor) + && journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &expected_successor) + }) + .await; + assert!( + flushed, + "child process must publish the pending successor snapshot before it can be killed" + ); + std::fs::write(&ready_path, b"ready").expect("write ready marker"); + loop { + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); +} + /// A journal published by a different OS process must remain a durable anchor /// when the restarted process can only admit a prefix of the replayed intents. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] @@ -390,3 +555,96 @@ async fn journal_replay_retains_child_process_anchor_when_manager_is_full() { "replay must retain the child-published journal until a successor snapshot can replace it" ); } + +/// If a process crashes after flushing a smaller successor snapshot but before +/// deleting the startup anchor, the restarted process must replay the +/// successor tail rather than losing it or merging it with stale records. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn journal_replay_survives_successor_flush_before_delete() { + let temp_dir = tempfile::tempdir().expect("successor-flush MRF root"); + let status = Command::new(std::env::current_exe().expect("test binary path")) + .arg("mrf_successor_flush_child_process_fixture") + .arg("--exact") + .arg("--nocapture") + .env("RUSTFS_MRF_SUCCESSOR_FLUSH_CHILD_ROOT", temp_dir.path()) + .status() + .expect("child MRF successor fixture should start"); + assert_eq!(status.code(), Some(78), "child process did not reach the successor flush boundary"); + + let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; + let expected_successor = journal_record(1, "successor-bucket", "second-object", None, 2); + assert!( + journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor), + "restarted process must see the pending successor snapshot" + ); + + let restarted = make_manager(storage); + let replayed = mrf_queue::replay_journal_once(&restarted).await; + assert_eq!(replayed, 1, "restart after successor flush must replay only the still-pending tail"); + assert_eq!( + restarted.operations_snapshot().await.queued_by_source.mrf, + 1, + "the successor tail must be accepted after restart" + ); + assert!( + disk_paths.iter().all(|path| { + !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + }), + "a fully consumed successor snapshot may be deleted after restart replay" + ); +} + +/// A service-style hard kill after successor flush must be equivalent to a +/// crash at the flush-before-delete boundary: restart may replay the smaller +/// successor snapshot, but must not lose or merge stale startup records. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[cfg(unix)] +async fn journal_replay_survives_service_kill_after_successor_flush() { + let temp_dir = tempfile::tempdir().expect("successor-kill MRF root"); + let ready = temp_dir.path().join("successor-flushed.ready"); + let mut child = Command::new(std::env::current_exe().expect("test binary path")) + .arg("mrf_successor_flush_waiting_child_process_fixture") + .arg("--exact") + .arg("--nocapture") + .env("RUSTFS_MRF_SUCCESSOR_KILL_CHILD_ROOT", temp_dir.path()) + .env("RUSTFS_MRF_SUCCESSOR_KILL_READY", &ready) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("child MRF successor fixture should start"); + let ready_seen = wait_until(Duration::from_secs(10), || { + let ready = ready.clone(); + async move { ready.exists() } + }) + .await; + assert!(ready_seen, "child process did not reach the successor flush boundary"); + child.kill().expect("kill child fixture"); + let status = child.wait().expect("wait for killed child fixture"); + assert!(!status.success(), "child fixture must be terminated instead of exiting cleanly"); + + let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; + let expected_successor = journal_record(1, "service-kill-bucket", "second-object", None, 2); + assert!( + journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &expected_successor), + "restarted process must see the successor snapshot produced before the kill" + ); + + let restarted = make_manager(storage); + let replayed = mrf_queue::replay_journal_once(&restarted).await; + assert_eq!(replayed, 1, "restart after service kill must replay only the still-pending tail"); + assert_eq!( + restarted.operations_snapshot().await.queued_by_source.mrf, + 1, + "the successor tail must be accepted after service kill restart" + ); + assert!( + disk_paths.iter().all(|path| { + !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists() + && !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists() + }), + "a fully consumed successor snapshot may be deleted after service-kill restart replay" + ); +} diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 0c0572aa1..d969452da 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -38,6 +38,7 @@ use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use tokio::time::{Duration, Instant, sleep, timeout}; use tracing::{debug, warn}; +use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageOwnerStatus}; use crate::storage_api::owner::HTTPPreconditions; use crate::{ BUCKET_META_PREFIX, EcstoreError as Error, EcstoreResult as StorageResult, RUSTFS_META_BUCKET, ReplicationConfig, @@ -601,6 +602,8 @@ pub struct DataUsageCacheInfo { pub scan_checkpoint: Option, #[serde(default)] pub scan_raw_enumeration_cursor: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scan_raw_enumeration_page_index: Option, #[serde(default)] pub scan_identity: Option, #[serde(default)] @@ -661,6 +664,7 @@ impl Serialize for DataUsageCacheInfo { // 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_raw_enumeration_page_index.is_some()) + usize::from(self.scan_identity.is_some()) + usize::from(self.scan_progress.is_some()) + usize::from(self.scan_coverage_receipt.is_some()) @@ -687,6 +691,9 @@ impl Serialize for DataUsageCacheInfo { if let Some(cursor) = &self.scan_raw_enumeration_cursor { state.serialize_entry("scan_raw_enumeration_cursor", cursor)?; } + if let Some(index) = &self.scan_raw_enumeration_page_index { + state.serialize_entry("scan_raw_enumeration_page_index", index)?; + } if let Some(identity) = self.scan_identity { state.serialize_entry("scan_identity", &identity)?; } @@ -895,6 +902,7 @@ impl DataUsageCache { && self.info.scan_progress.is_none() && self.info.scan_checkpoint.is_none() && self.info.scan_raw_enumeration_cursor.is_none() + && self.info.scan_raw_enumeration_page_index.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) @@ -922,12 +930,17 @@ impl DataUsageCache { if self.validated_raw_enumeration_cursor().is_none() { self.info.scan_raw_enumeration_cursor = None; } + if self.validated_raw_enumeration_page_index().is_none() { + self.info.scan_raw_enumeration_page_index = None; + } let cursor_is_valid = (self.info.scan_checkpoint.is_none() && self.info.scan_raw_enumeration_cursor.is_none() + && self.info.scan_raw_enumeration_page_index.is_none() && self.info.scan_resume_after.is_none() && self.info.scan_coverage_receipt.is_none()) || self.validated_scan_frontier().is_some() - || self.info.scan_raw_enumeration_cursor.is_some(); + || self.info.scan_raw_enumeration_cursor.is_some() + || self.info.scan_raw_enumeration_page_index.is_some(); if !cursor_is_valid { self.info.scan_progress = None; } @@ -949,6 +962,7 @@ impl DataUsageCache { self.info.scan_resume_after = None; self.info.scan_checkpoint = None; self.info.scan_raw_enumeration_cursor = None; + self.info.scan_raw_enumeration_page_index = None; self.info.scan_coverage_receipt = None; } // Old readers do not understand coverage sweeps. An absent plan makes @@ -1026,6 +1040,25 @@ impl DataUsageCache { .then_some(cursor) } + pub(crate) fn validated_raw_enumeration_page_index(&self) -> Option<&RawEnumerationPageIndex> { + let index = self.info.scan_raw_enumeration_page_index.as_ref()?; + if self.info.scan_progress.is_none() + || !self.info.scan_identity.is_some_and(|identity| identity.is_valid()) + || self.info.source.is_none() + || index.committed_entries().is_err() + || index.indexed_entries().is_err() + { + return None; + } + let parent = match index.status() { + RawEnumerationPageOwnerStatus::Unsupported => return None, + RawEnumerationPageOwnerStatus::Building { parent, .. } | RawEnumerationPageOwnerStatus::Ready { parent, .. } => { + parent + } + }; + path_is_in_bucket_scope(&self.info.name, &parent).then_some(index) + } + /// 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 e59ae00b6..e1986af06 100644 --- a/crates/scanner/src/data_usage_define/tests.rs +++ b/crates/scanner/src/data_usage_define/tests.rs @@ -1179,6 +1179,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { 7, [7; 32], )), + scan_raw_enumeration_page_index: Some(raw_page_index_fixture("bucket/prefix", &["entry-a"], false)), snapshot_complete: true, scan_plan_digest: Some(TEST_PLAN_DIGEST), scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])), @@ -1207,6 +1208,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() { .map(|cursor| cursor.last_entry.as_deref()), Some(Some("last-object")) ); + assert!(current.info.scan_raw_enumeration_page_index.is_some()); 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]))); @@ -1260,6 +1262,24 @@ fn cache_with_raw_cursor(cursor: DataUsageRawEnumerationCursor) -> DataUsageCach } } +fn raw_page_index_fixture(parent: &str, entries: &[&str], complete: bool) -> RawEnumerationPageIndex { + let mut index = RawEnumerationPageIndex::new(parent, 2).expect("raw page index should initialize"); + let generation = index.generation().expect("raw page index should expose generation"); + let outcome = if complete { + index.ingest_owner_entries(entries.iter().map(|entry| (*entry).to_string()), entries.len().max(1), generation) + } else { + index.ingest_partial_owner_entries(entries.iter().map(|entry| (*entry).to_string()), entries.len().max(1), generation) + } + .expect("raw page index fixture should ingest entries"); + if outcome.ready_to_commit { + let generation = index.generation().expect("raw page index should expose commit generation"); + index + .commit_building_page(generation) + .expect("raw page index fixture should commit ready page"); + } + index +} + #[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]); @@ -1346,6 +1366,44 @@ fn prepare_bucket_checkpoint_preserves_only_valid_raw_enumeration_cursor() { assert!(cache.info.scan_progress.is_some()); } +#[test] +fn prepare_bucket_checkpoint_preserves_only_valid_raw_page_index() { + let identity = valid_scan_identity(); + let source = DataUsageCacheSource::new(1, 2); + let page_index = raw_page_index_fixture("bucket/raw", &["entry-001"], false); + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + leader_epoch: 1, + source: Some(source), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + scan_identity: Some(identity), + tier_registry_generation: Some(9), + scan_progress: Some(DataUsageScanProgress { + started_plan: TEST_PLAN_DIGEST, + requested_plan: TEST_PLAN_DIGEST, + }), + scan_raw_enumeration_page_index: Some(page_index.clone()), + ..Default::default() + }, + ..Default::default() + }; + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity), + DataUsageCachePrepareOutcome::Reused + ); + assert_eq!(cache.info.scan_raw_enumeration_page_index, Some(page_index)); + + let invalid = raw_page_index_fixture("other/raw", &["entry-001"], false); + cache.info.scan_raw_enumeration_page_index = Some(invalid); + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 1, 1, source, TEST_PLAN_DIGEST, identity), + DataUsageCachePrepareOutcome::Reused + ); + assert!(cache.info.scan_raw_enumeration_page_index.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. diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index cfc63a3b7..986fb7d62 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -60,6 +60,7 @@ use uuid::Uuid; pub mod data_usage_define; pub mod error; pub mod prefix_usage; +pub mod raw_page_index; mod remote_scanner; pub mod runtime_config; pub mod scanner; diff --git a/crates/scanner/src/raw_page_index.rs b/crates/scanner/src/raw_page_index.rs new file mode 100644 index 000000000..f263197f2 --- /dev/null +++ b/crates/scanner/src/raw_page_index.rs @@ -0,0 +1,945 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; +use thiserror::Error; + +const RAW_PAGE_INDEX_VERSION: u16 = 1; +const RAW_PAGE_ENTRY_MAX_BYTES: usize = 16 * 1024; + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct RawEnumerationPageIndex { + state: RawEnumerationPageIndexState, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +enum RawEnumerationPageIndexState { + Unsupported, + Supported(RawEnumerationPageIndexInner), +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct RawEnumerationPageIndexInner { + generation: u64, + parent: String, + page_entry_limit: usize, + pages: Vec, + building: Option, + complete: bool, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct RawEnumerationPage { + version: u16, + parent: String, + page_index: u64, + entries_start: u64, + entries: Vec, + terminal: bool, + digest: [u8; 32], +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct RawEnumerationPageBuilder { + page_index: u64, + entries_start: u64, + entries: Vec, + terminal: bool, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct RawEnumerationPageBuildOutcome { + pub status: RawEnumerationPageOwnerStatus, + pub ready_to_commit: bool, +} + +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub enum RawEnumerationPageOwnerStatus { + Unsupported, + Building { + generation: u64, + parent: String, + page_index: u64, + indexed_entries: u64, + buffered_entries: usize, + }, + Ready { + generation: u64, + parent: String, + committed_pages: usize, + indexed_entries: u64, + complete: bool, + }, +} + +#[derive(Debug, Error, PartialEq, Eq)] +pub enum RawEnumerationPageIndexError { + #[error("raw enumeration page owner is unsupported")] + Unsupported, + #[error("raw enumeration page owner generation changed")] + StaleGeneration, + #[error("raw enumeration page identity changed before committed coverage")] + IdentityMismatch, + #[error("raw enumeration page owner requires a non-empty parent")] + EmptyParent, + #[error("raw enumeration page entry limit must be non-zero")] + EmptyPage, + #[error("raw enumeration page build budget must be non-zero")] + EmptyBudget, + #[error("raw enumeration page entry is invalid")] + InvalidEntry, + #[error("raw enumeration page has no staged entries")] + EmptyCommit, + #[error("raw enumeration page index is corrupt")] + CorruptIndex, +} + +impl RawEnumerationPageIndex { + pub fn unsupported() -> Self { + Self { + state: RawEnumerationPageIndexState::Unsupported, + } + } + + pub fn new(parent: impl Into, page_entry_limit: usize) -> Result { + let parent = parent.into(); + if parent.is_empty() { + return Err(RawEnumerationPageIndexError::EmptyParent); + } + if page_entry_limit == 0 { + return Err(RawEnumerationPageIndexError::EmptyPage); + } + Ok(Self { + state: RawEnumerationPageIndexState::Supported(RawEnumerationPageIndexInner { + generation: 0, + parent, + page_entry_limit, + pages: Vec::new(), + building: None, + complete: false, + }), + }) + } + + pub fn generation(&self) -> Option { + match &self.state { + RawEnumerationPageIndexState::Unsupported => None, + RawEnumerationPageIndexState::Supported(inner) => Some(inner.generation), + } + } + + pub fn status(&self) -> RawEnumerationPageOwnerStatus { + match &self.state { + RawEnumerationPageIndexState::Unsupported => RawEnumerationPageOwnerStatus::Unsupported, + RawEnumerationPageIndexState::Supported(inner) => inner.status(), + } + } + + pub fn ingest_owner_entries( + &mut self, + entries: I, + max_new_entries: usize, + expected_generation: u64, + ) -> Result + where + I: IntoIterator, + { + self.ingest_owner_entries_inner(entries, max_new_entries, expected_generation, true) + } + + pub fn ingest_partial_owner_entries( + &mut self, + entries: I, + max_new_entries: usize, + expected_generation: u64, + ) -> Result + where + I: IntoIterator, + { + self.ingest_owner_entries_inner(entries, max_new_entries, expected_generation, false) + } + + fn ingest_owner_entries_inner( + &mut self, + entries: I, + max_new_entries: usize, + expected_generation: u64, + source_complete: bool, + ) -> Result + where + I: IntoIterator, + { + if max_new_entries == 0 { + return Err(RawEnumerationPageIndexError::EmptyBudget); + } + let RawEnumerationPageIndexState::Supported(inner) = &mut self.state else { + return Err(RawEnumerationPageIndexError::Unsupported); + }; + if inner.generation != expected_generation { + return Err(RawEnumerationPageIndexError::StaleGeneration); + } + let mut entries = normalize_owner_entries(entries)?; + let committed_entries = inner.validated_committed_entries()?; + if inner.complete { + if entries != committed_entries { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + } + return Ok(RawEnumerationPageBuildOutcome { + status: inner.status(), + ready_to_commit: false, + }); + } + + if !entries.starts_with(&committed_entries) { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + } + + let mut indexed_entries = committed_entries.len(); + 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) { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + } + indexed_entries = indexed_entries.saturating_add(building.entries.len()); + } + if source_complete && indexed_entries == entries.len() && inner.building.is_none() { + inner.complete = true; + inner.generation = inner.generation.saturating_add(1); + return Ok(RawEnumerationPageBuildOutcome { + status: inner.status(), + ready_to_commit: 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); + let building = inner.building.get_or_insert_with(|| RawEnumerationPageBuilder { + page_index, + entries_start, + entries: Vec::new(), + terminal: false, + }); + if building.entries.len() >= inner.page_entry_limit { + 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)); + if append_count == 0 { + if source_complete && !building.terminal { + building.terminal = true; + inner.generation = inner.generation.saturating_add(1); + } + } 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; + inner.generation = inner.generation.saturating_add(1); + } + !building.entries.is_empty() && (building.terminal || building.entries.len() >= inner.page_entry_limit) + } + }; + + Ok(RawEnumerationPageBuildOutcome { + status: inner.status(), + ready_to_commit, + }) + } + + pub fn commit_building_page(&mut self, expected_generation: u64) -> Result { + let RawEnumerationPageIndexState::Supported(inner) = &mut self.state else { + return Err(RawEnumerationPageIndexError::Unsupported); + }; + if inner.generation != expected_generation { + return Err(RawEnumerationPageIndexError::StaleGeneration); + } + let entries_start = inner.validated_committed_entries()?.len(); + let Some(building) = inner.building.as_ref() else { + return Err(RawEnumerationPageIndexError::EmptyCommit); + }; + if building.entries.is_empty() { + return Err(RawEnumerationPageIndexError::EmptyCommit); + } + building.validate( + u64::try_from(inner.pages.len()).unwrap_or(u64::MAX), + u64::try_from(entries_start).unwrap_or(u64::MAX), + inner.page_entry_limit, + )?; + let Some(building) = inner.building.take() else { + return Err(RawEnumerationPageIndexError::EmptyCommit); + }; + let page = RawEnumerationPage::new(&inner.parent, building); + inner.complete = page.terminal; + inner.pages.push(page.clone()); + inner.generation = inner.generation.saturating_add(1); + Ok(page) + } + + pub fn page(&self, page_index: u64, expected_digest: [u8; 32]) -> Result<&RawEnumerationPage, RawEnumerationPageIndexError> { + let RawEnumerationPageIndexState::Supported(inner) = &self.state else { + return Err(RawEnumerationPageIndexError::Unsupported); + }; + let Some(page_index) = usize::try_from(page_index).ok() else { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + }; + let Some(page) = inner.pages.get(page_index) else { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + }; + page.validate(&inner.parent, u64::try_from(page_index).unwrap_or(u64::MAX), page.entries_start)?; + if page.digest != expected_digest { + return Err(RawEnumerationPageIndexError::IdentityMismatch); + } + Ok(page) + } + + pub fn committed_entries(&self) -> Result, RawEnumerationPageIndexError> { + match &self.state { + RawEnumerationPageIndexState::Unsupported => Ok(Vec::new()), + RawEnumerationPageIndexState::Supported(inner) => inner.validated_committed_entries(), + } + } + + pub fn indexed_entries(&self) -> Result, RawEnumerationPageIndexError> { + match &self.state { + RawEnumerationPageIndexState::Unsupported => Ok(Vec::new()), + RawEnumerationPageIndexState::Supported(inner) => inner.validated_indexed_entries(), + } + } +} + +impl RawEnumerationPageIndexInner { + fn status(&self) -> RawEnumerationPageOwnerStatus { + let indexed_entries = u64::try_from(self.indexed_entries()).unwrap_or(u64::MAX); + if let Some(building) = &self.building { + RawEnumerationPageOwnerStatus::Building { + generation: self.generation, + parent: self.parent.clone(), + page_index: building.page_index, + indexed_entries, + buffered_entries: building.entries.len(), + } + } else { + RawEnumerationPageOwnerStatus::Ready { + generation: self.generation, + parent: self.parent.clone(), + committed_pages: self.pages.len(), + indexed_entries, + complete: self.complete, + } + } + } + + fn validated_committed_entries(&self) -> Result, RawEnumerationPageIndexError> { + let mut entries = Vec::new(); + for (page_index, page) in self.pages.iter().enumerate() { + page.validate( + &self.parent, + u64::try_from(page_index).unwrap_or(u64::MAX), + u64::try_from(entries.len()).unwrap_or(u64::MAX), + )?; + if page.terminal && page_index + 1 != self.pages.len() { + return Err(RawEnumerationPageIndexError::CorruptIndex); + } + entries.extend(page.entries.iter().cloned()); + } + if self.complete && self.pages.last().is_some_and(|page| !page.terminal) { + return Err(RawEnumerationPageIndexError::CorruptIndex); + } + Ok(entries) + } + + fn validated_indexed_entries(&self) -> Result, RawEnumerationPageIndexError> { + let mut entries = self.validated_committed_entries()?; + if let Some(building) = &self.building { + building.validate( + u64::try_from(self.pages.len()).unwrap_or(u64::MAX), + u64::try_from(entries.len()).unwrap_or(u64::MAX), + self.page_entry_limit, + )?; + entries.extend(building.entries.iter().cloned()); + } + Ok(entries) + } + + fn indexed_entries(&self) -> usize { + self.pages.iter().map(|page| page.entries.len()).sum::() + + self.building.as_ref().map_or(0, |building| building.entries.len()) + } +} + +impl RawEnumerationPageBuilder { + fn validate(&self, page_index: u64, entries_start: u64, page_entry_limit: usize) -> Result<(), RawEnumerationPageIndexError> { + if self.page_index != page_index + || self.entries_start != entries_start + || self.entries.is_empty() + || self.entries.len() > page_entry_limit + || !entries_are_normalized(&self.entries) + { + return Err(RawEnumerationPageIndexError::CorruptIndex); + } + Ok(()) + } +} + +impl RawEnumerationPage { + pub fn version(&self) -> u16 { + self.version + } + + pub fn parent(&self) -> &str { + &self.parent + } + + pub fn page_index(&self) -> u64 { + self.page_index + } + + pub fn entries_start(&self) -> u64 { + self.entries_start + } + + pub fn entries(&self) -> &[String] { + &self.entries + } + + pub fn terminal(&self) -> bool { + self.terminal + } + + pub fn digest(&self) -> [u8; 32] { + self.digest + } + + fn new(parent: &str, building: RawEnumerationPageBuilder) -> Self { + let digest = raw_page_digest(parent, &building); + Self { + version: RAW_PAGE_INDEX_VERSION, + parent: parent.to_string(), + page_index: building.page_index, + entries_start: building.entries_start, + entries: building.entries, + terminal: building.terminal, + digest, + } + } + + fn validate(&self, parent: &str, page_index: u64, entries_start: u64) -> Result<(), RawEnumerationPageIndexError> { + let builder = RawEnumerationPageBuilder { + page_index: self.page_index, + entries_start: self.entries_start, + entries: self.entries.clone(), + terminal: self.terminal, + }; + if self.version != RAW_PAGE_INDEX_VERSION + || self.parent != parent + || self.page_index != page_index + || self.entries_start != entries_start + || self.entries.is_empty() + || !entries_are_normalized(&self.entries) + || self.digest != raw_page_digest(parent, &builder) + { + return Err(RawEnumerationPageIndexError::CorruptIndex); + } + Ok(()) + } +} + +fn normalize_owner_entries(entries: I) -> Result, RawEnumerationPageIndexError> +where + I: IntoIterator, +{ + let mut entries = entries.into_iter().map(validate_owner_entry).collect::, _>>()?; + entries.sort(); + entries.dedup(); + Ok(entries) +} + +fn validate_owner_entry(entry: String) -> Result { + if !owner_entry_is_valid(&entry) { + return Err(RawEnumerationPageIndexError::InvalidEntry); + } + Ok(entry) +} + +fn owner_entry_is_valid(entry: &str) -> bool { + !entry.is_empty() && entry != "." && entry != ".." && !entry.contains('/') && entry.len() <= RAW_PAGE_ENTRY_MAX_BYTES +} + +fn entries_are_normalized(entries: &[String]) -> bool { + entries.iter().all(|entry| owner_entry_is_valid(entry)) + && entries + .windows(2) + .all(|window| window.first().zip(window.get(1)).is_some_and(|(left, right)| 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()); + update_digest(&mut digest, b"parent", parent.as_bytes()); + update_digest(&mut digest, b"page_index", &building.page_index.to_le_bytes()); + update_digest(&mut digest, b"entries_start", &building.entries_start.to_le_bytes()); + update_digest(&mut digest, b"terminal", &[u8::from(building.terminal)]); + for entry in &building.entries { + update_digest(&mut digest, b"entry", entry.as_bytes()); + } + digest.finalize().into() +} + +fn update_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) { + digest.update(label); + digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_le_bytes()); + digest.update(value); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entries(values: &[&str]) -> Vec { + values.iter().map(|value| (*value).to_string()).collect() + } + + fn indexed_entries(status: &RawEnumerationPageOwnerStatus) -> u64 { + match status { + RawEnumerationPageOwnerStatus::Unsupported => 0, + RawEnumerationPageOwnerStatus::Building { indexed_entries, .. } + | RawEnumerationPageOwnerStatus::Ready { indexed_entries, .. } => *indexed_entries, + } + } + + #[test] + fn unsupported_owner_reports_unsupported_without_building_pages() { + let mut owner = RawEnumerationPageIndex::unsupported(); + + assert_eq!(owner.status(), RawEnumerationPageOwnerStatus::Unsupported); + assert_eq!( + owner.ingest_owner_entries(entries(&["entry-a"]), 1, 0), + Err(RawEnumerationPageIndexError::Unsupported) + ); + assert_eq!(owner.commit_building_page(0), Err(RawEnumerationPageIndexError::Unsupported)); + } + + #[test] + fn owner_page_builds_monotonically_across_small_budget_restarts() { + let source = entries(&["entry-c", "entry-a", "entry-e", "entry-b", "entry-d"]); + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let mut last_indexed_entries = 0; + + for _ in 0..8 { + let mut restarted_owner = owner.clone(); + let generation = restarted_owner + .generation() + .expect("supported owner should expose generation"); + let outcome = restarted_owner + .ingest_owner_entries(source.clone(), 1, generation) + .expect("owner page build should accept stable source entries"); + let now_indexed_entries = indexed_entries(&outcome.status); + assert!( + now_indexed_entries >= last_indexed_entries, + "owner-backed page build must not move coverage backward across restart" + ); + last_indexed_entries = now_indexed_entries; + if outcome.ready_to_commit { + let generation = restarted_owner + .generation() + .expect("supported owner should expose generation"); + restarted_owner + .commit_building_page(generation) + .expect("ready page should commit under matching generation"); + } + owner = restarted_owner; + if let RawEnumerationPageOwnerStatus::Ready { complete: true, .. } = owner.status() { + break; + } + } + + assert_eq!( + owner.status(), + RawEnumerationPageOwnerStatus::Ready { + generation: 8, + parent: "bucket".to_string(), + committed_pages: 3, + indexed_entries: 5, + complete: true, + } + ); + assert_eq!( + owner.committed_entries().expect("committed entries should validate"), + entries(&["entry-a", "entry-b", "entry-c", "entry-d", "entry-e"]) + ); + } + + #[test] + fn page_digest_identity_guards_consumption_and_source_drift() { + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let source = entries(&["entry-a", "entry-b", "entry-c"]); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(source, 2, generation) + .expect("first page build should succeed"); + assert!(outcome.ready_to_commit); + let generation = owner.generation().expect("supported owner should expose generation"); + let page = owner.commit_building_page(generation).expect("first page should commit"); + + assert_eq!( + owner + .page(page.page_index, page.digest) + .expect("committed page should be readable"), + &page + ); + let mut wrong_digest = page.digest; + wrong_digest[0] ^= 0xff; + assert_eq!( + owner.page(page.page_index, wrong_digest), + Err(RawEnumerationPageIndexError::IdentityMismatch) + ); + + let generation = owner.generation().expect("supported owner should expose generation"); + assert_eq!( + owner.ingest_owner_entries(entries(&["entry-a", "entry-x", "entry-c"]), 1, generation), + Err(RawEnumerationPageIndexError::IdentityMismatch) + ); + assert_eq!( + owner + .committed_entries() + .expect("committed entries should validate after source drift rejection"), + entries(&["entry-a", "entry-b"]) + ); + } + + #[test] + fn page_commit_cas_failure_and_precommit_crash_do_not_publish_coverage() { + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let source = entries(&["entry-a", "entry-b", "entry-c"]); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(source, 2, generation) + .expect("page build should stage entries"); + assert!(outcome.ready_to_commit); + assert_eq!( + owner + .committed_entries() + .expect("uncommitted staged entries should not publish coverage"), + Vec::::new() + ); + + assert_eq!(owner.commit_building_page(generation), Err(RawEnumerationPageIndexError::StaleGeneration)); + assert_eq!( + owner + .committed_entries() + .expect("failed CAS should not corrupt committed coverage"), + Vec::::new() + ); + + let mut restarted_owner = owner.clone(); + let generation = restarted_owner + .generation() + .expect("supported owner should expose generation"); + restarted_owner + .commit_building_page(generation) + .expect("persisted staged page should commit after restart with fresh CAS generation"); + assert_eq!( + restarted_owner + .committed_entries() + .expect("restarted committed entries should validate"), + entries(&["entry-a", "entry-b"]) + ); + } + + #[test] + fn serialized_building_page_resumes_and_commits_after_restart() { + let source = entries(&["entry-a", "entry-b", "entry-c"]); + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(source.clone(), 1, generation) + .expect("first budgeted page build should stage one entry"); + assert_eq!( + outcome.status, + RawEnumerationPageOwnerStatus::Building { + generation: 1, + parent: "bucket".to_string(), + page_index: 0, + indexed_entries: 1, + buffered_entries: 1, + } + ); + assert!(!outcome.ready_to_commit); + + let encoded = rmp_serde::to_vec(&owner).expect("building page index should encode"); + let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("building page index should decode"); + + let generation = decoded.generation().expect("decoded owner should expose generation"); + let outcome = decoded + .ingest_owner_entries(source, 1, generation) + .expect("decoded page owner should resume from staged coverage"); + assert!(outcome.ready_to_commit); + let generation = decoded.generation().expect("decoded owner should expose generation"); + let page = decoded + .commit_building_page(generation) + .expect("decoded ready page should commit"); + assert_eq!(page.entries(), entries(&["entry-a", "entry-b"])); + assert_eq!( + decoded + .page(page.page_index(), page.digest()) + .expect("committed decoded page should validate by digest") + .entries(), + entries(&["entry-a", "entry-b"]) + ); + } + + #[test] + fn partial_owner_source_does_not_mark_terminal_before_completion() { + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_partial_owner_entries(entries(&["entry-a"]), 1, generation) + .expect("partial source should stage one entry"); + assert_eq!( + outcome.status, + RawEnumerationPageOwnerStatus::Building { + generation: 1, + parent: "bucket".to_string(), + page_index: 0, + indexed_entries: 1, + buffered_entries: 1, + } + ); + assert!(!outcome.ready_to_commit); + assert_eq!( + owner.indexed_entries().expect("building page entries should validate"), + entries(&["entry-a"]) + ); + + let encoded = rmp_serde::to_vec(&owner).expect("partial owner should encode"); + let mut restarted: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("partial owner should decode"); + let generation = restarted.generation().expect("restarted owner should expose generation"); + let outcome = restarted + .ingest_owner_entries(entries(&["entry-a", "entry-b"]), 1, generation) + .expect("complete source should finish resumed building page"); + assert!(outcome.ready_to_commit); + let generation = restarted.generation().expect("finished owner should expose generation"); + let page = restarted + .commit_building_page(generation) + .expect("terminal resumed page should commit"); + assert!(page.terminal()); + assert_eq!(page.entries(), entries(&["entry-a", "entry-b"])); + } + + #[test] + fn terminal_marker_advances_generation_before_commit() { + let initial_source = entries(&["entry-a", "entry-b", "entry-c"]); + let current_source = entries(&["entry-a", "entry-b"]); + let mut owner = RawEnumerationPageIndex::new("bucket", 3).expect("page owner should initialize"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(initial_source, 2, generation) + .expect("budgeted page build should stage all source entries"); + assert_eq!(owner.generation(), Some(1)); + assert!(!outcome.ready_to_commit); + + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(current_source, 1, generation) + .expect("terminal-only build step should complete the staged page"); + assert!(outcome.ready_to_commit); + assert_eq!( + owner.generation(), + Some(2), + "terminal marker is a persisted builder state change and must advance the CAS generation" + ); + assert_eq!(owner.commit_building_page(generation), Err(RawEnumerationPageIndexError::StaleGeneration)); + + let generation = owner.generation().expect("supported owner should expose generation"); + let page = owner + .commit_building_page(generation) + .expect("fresh generation should commit terminal page"); + assert!(page.terminal()); + } + + #[test] + fn deserialized_corrupt_page_digest_fails_closed() { + let source = entries(&["entry-a", "entry-b", "entry-c"]); + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(source.clone(), 2, generation) + .expect("page build should stage entries"); + assert!(outcome.ready_to_commit); + let generation = owner.generation().expect("supported owner should expose generation"); + let page = owner.commit_building_page(generation).expect("ready page should commit"); + + let encoded = rmp_serde::to_vec(&owner).expect("committed page index should encode"); + let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("committed page index should decode"); + let RawEnumerationPageIndexState::Supported(inner) = &mut decoded.state else { + panic!("decoded owner should be supported"); + }; + inner.pages[0].digest[0] ^= 0xff; + + assert_eq!(decoded.committed_entries(), Err(RawEnumerationPageIndexError::CorruptIndex)); + assert_eq!( + decoded.page(page.page_index(), page.digest()), + Err(RawEnumerationPageIndexError::CorruptIndex) + ); + assert_eq!( + decoded.ingest_owner_entries(source, 1, decoded.generation().expect("decoded owner should expose generation")), + Err(RawEnumerationPageIndexError::CorruptIndex) + ); + + let mut complete_owner = RawEnumerationPageIndex::new("bucket", 2).expect("complete owner should initialize"); + let generation = complete_owner.generation().expect("complete owner should expose generation"); + let outcome = complete_owner + .ingest_owner_entries(entries(&["entry-a", "entry-b"]), 2, generation) + .expect("terminal page build should stage entries"); + assert!(outcome.ready_to_commit); + let generation = complete_owner.generation().expect("complete owner should expose generation"); + complete_owner + .commit_building_page(generation) + .expect("terminal page should commit"); + let encoded = rmp_serde::to_vec(&complete_owner).expect("complete page index should encode"); + let mut decoded_complete: RawEnumerationPageIndex = + rmp_serde::from_slice(&encoded).expect("complete page index should decode"); + let RawEnumerationPageIndexState::Supported(inner) = &mut decoded_complete.state else { + panic!("decoded complete owner should be supported"); + }; + inner.pages[0].digest[0] ^= 0xff; + assert_eq!( + decoded_complete.ingest_owner_entries( + entries(&["entry-a", "entry-b"]), + 1, + decoded_complete + .generation() + .expect("decoded complete owner should expose generation") + ), + Err(RawEnumerationPageIndexError::CorruptIndex) + ); + } + + #[test] + fn deserialized_corrupt_building_page_fails_closed_before_append() { + let source = entries(&["entry-a", "entry-b", "entry-c"]); + 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(source.clone(), 1, generation) + .expect("page build should stage one entry"); + + let encoded = rmp_serde::to_vec(&owner).expect("building page index should encode"); + let mut decoded: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("building page index should decode"); + let RawEnumerationPageIndexState::Supported(inner) = &mut decoded.state else { + panic!("decoded building owner should be supported"); + }; + inner.building.as_mut().expect("building page should be present").page_index = 9; + + assert_eq!( + decoded.ingest_owner_entries(source, 1, decoded.generation().expect("decoded owner should expose generation")), + Err(RawEnumerationPageIndexError::CorruptIndex) + ); + assert_eq!( + decoded.commit_building_page(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"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(Vec::new(), 1, generation) + .expect("empty owner source should be a valid complete index"); + + assert_eq!( + outcome.status, + RawEnumerationPageOwnerStatus::Ready { + generation: 1, + parent: "bucket".to_string(), + committed_pages: 0, + indexed_entries: 0, + complete: true, + } + ); + assert!(!outcome.ready_to_commit); + assert_eq!(owner.commit_building_page(1), Err(RawEnumerationPageIndexError::EmptyCommit)); + } + + #[test] + fn complete_owner_rejects_source_drift_after_restart() { + let mut owner = RawEnumerationPageIndex::new("bucket", 2).expect("page owner should initialize"); + let generation = owner.generation().expect("supported owner should expose generation"); + let outcome = owner + .ingest_owner_entries(entries(&["entry-a", "entry-b"]), 2, generation) + .expect("terminal page build should stage entries"); + assert!(outcome.ready_to_commit); + let generation = owner.generation().expect("supported owner should expose generation"); + owner.commit_building_page(generation).expect("terminal page should commit"); + + let encoded = rmp_serde::to_vec(&owner).expect("complete owner should encode"); + let mut restarted_owner: RawEnumerationPageIndex = rmp_serde::from_slice(&encoded).expect("complete owner should decode"); + let generation = restarted_owner + .generation() + .expect("restarted owner should expose generation"); + + assert_eq!( + restarted_owner.ingest_owner_entries(entries(&["entry-a", "entry-b", "entry-c"]), 1, generation), + Err(RawEnumerationPageIndexError::IdentityMismatch) + ); + } + + #[test] + fn owner_page_rejects_invalid_boundaries_without_advancing_generation() { + assert_eq!(RawEnumerationPageIndex::new("", 2), Err(RawEnumerationPageIndexError::EmptyParent)); + assert_eq!(RawEnumerationPageIndex::new("bucket", 0), Err(RawEnumerationPageIndexError::EmptyPage)); + + let mut owner = RawEnumerationPageIndex::new("bucket", 1).expect("page owner should initialize"); + assert_eq!( + owner.ingest_owner_entries(entries(&["entry-a"]), 0, 0), + Err(RawEnumerationPageIndexError::EmptyBudget) + ); + assert_eq!( + owner.ingest_owner_entries(entries(&["nested/name"]), 1, 0), + Err(RawEnumerationPageIndexError::InvalidEntry) + ); + assert_eq!(owner.generation(), Some(0)); + + let oversized = "x".repeat(RAW_PAGE_ENTRY_MAX_BYTES + 1); + assert_eq!( + owner.ingest_owner_entries(vec![oversized], 1, 0), + Err(RawEnumerationPageIndexError::InvalidEntry) + ); + assert_eq!(owner.generation(), Some(0)); + + let exact_boundary = "x".repeat(RAW_PAGE_ENTRY_MAX_BYTES); + let outcome = owner + .ingest_owner_entries(vec![exact_boundary.clone()], 1, 0) + .expect("max-sized direct entry should be accepted"); + assert!(outcome.ready_to_commit); + let page = owner.commit_building_page(1).expect("max-sized direct entry should commit"); + assert_eq!(page.entries(), &[exact_boundary]); + } +} diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 3518fa002..1e0edc362 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -25,6 +25,7 @@ use crate::data_usage_define::{ PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path, }; use crate::error::ScannerError; +use crate::raw_page_index::{RawEnumerationPageIndex, RawEnumerationPageIndexError}; use crate::runtime_config::{ scanner_alert_excess_folders, scanner_alert_excess_version_size, scanner_alert_excess_versions, scanner_yield_every_n_objects, }; @@ -90,6 +91,8 @@ const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000; const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60); const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32; const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30); +const SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT: usize = 128; +const SCANNER_RAW_ENUMERATION_PAGE_BUILD_BUDGET: usize = 1; // Erasure data directories contain direct part.N files; keep namespace probes bounded. const ERASURE_DATA_DIR_PROBE_ENTRY_LIMIT: usize = 64; const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024; @@ -751,17 +754,34 @@ struct RawEnumerationProgress { last_entry: Option, entries_seen: u64, digest: Sha256, + observed_entries: Vec, + revalidate_after_entries: usize, + page_index: Option, } impl RawEnumerationProgress { - fn new(parent: &str) -> Self { + fn new(parent: &str, page_index: Option) -> Self { let mut digest = Sha256::new(); update_raw_enumeration_digest(&mut digest, b"parent", parent.as_bytes()); + let mut revalidate_after_entries = 0; + let page_index = match page_index { + Some(index) => match index.indexed_entries() { + Ok(entries) => { + revalidate_after_entries = entries.len(); + Some(index) + } + Err(_) => None, + }, + None => RawEnumerationPageIndex::new(parent, SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT).ok(), + }; Self { parent: parent.to_string(), last_entry: None, entries_seen: 0, digest, + observed_entries: Vec::new(), + revalidate_after_entries, + page_index, } } @@ -769,19 +789,55 @@ impl RawEnumerationProgress { update_raw_enumeration_digest(&mut self.digest, b"entry", entry.as_bytes()); self.last_entry = Some(entry.to_string()); self.entries_seen = self.entries_seen.saturating_add(1); + self.observed_entries.push(entry.to_string()); + if let Some(index) = &mut self.page_index { + if self.observed_entries.len() < self.revalidate_after_entries { + return; + } + let result = index + .generation() + .ok_or(RawEnumerationPageIndexError::Unsupported) + .and_then(|generation| { + index.ingest_partial_owner_entries( + self.observed_entries.clone(), + SCANNER_RAW_ENUMERATION_PAGE_BUILD_BUDGET, + generation, + ) + }); + match result { + Ok(outcome) if outcome.ready_to_commit => { + if let Some(generation) = index.generation() + && index.commit_building_page(generation).is_err() + { + self.page_index = None; + } + } + Ok(_) => {} + Err(_) => { + self.page_index = None; + } + } + } } - fn into_cursor(self) -> Option { + fn cursor(&self) -> Option { if self.entries_seen == 0 { return None; } Some(DataUsageRawEnumerationCursor::new( - self.parent, - self.last_entry, + self.parent.clone(), + self.last_entry.clone(), self.entries_seen, - self.digest.finalize().into(), + self.digest.clone().finalize().into(), )) } + + 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, + }) + } } fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) { @@ -1049,6 +1105,19 @@ impl FolderScanner { if self.old_cache.info.scan_progress.is_none() { return; } + let page_index = self + .old_cache + .validated_raw_enumeration_page_index() + .filter(|index| match index.status() { + crate::raw_page_index::RawEnumerationPageOwnerStatus::Building { + parent: index_parent, .. + } + | crate::raw_page_index::RawEnumerationPageOwnerStatus::Ready { + parent: index_parent, .. + } => index_parent == parent, + crate::raw_page_index::RawEnumerationPageOwnerStatus::Unsupported => false, + }) + .cloned(); if let Some(position) = self .raw_enumeration_progress .iter() @@ -1056,7 +1125,8 @@ impl FolderScanner { { self.raw_enumeration_progress.truncate(position + 1); } else { - self.raw_enumeration_progress.push(RawEnumerationProgress::new(parent)); + self.raw_enumeration_progress + .push(RawEnumerationProgress::new(parent, page_index)); } if let Some(progress) = self.raw_enumeration_progress.last_mut() { progress.record_entry(entry); @@ -1073,11 +1143,11 @@ impl FolderScanner { }); } - fn take_raw_enumeration_cursor(&mut self) -> Option { - self.raw_enumeration_progress - .drain(..) - .next() - .and_then(RawEnumerationProgress::into_cursor) + fn take_raw_enumeration_resume_state(&mut self) -> (Option, Option) { + match self.raw_enumeration_progress.drain(..).next() { + Some(progress) => (progress.cursor(), progress.page_index()), + None => (None, None), + } } fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) { @@ -2686,6 +2756,7 @@ pub(crate) async fn scan_data_folder_scoped( new_cache.info.scan_resume_after = None; new_cache.info.scan_checkpoint = None; new_cache.info.scan_raw_enumeration_cursor = None; + new_cache.info.scan_raw_enumeration_page_index = None; new_cache.info.scan_coverage_receipt = None; if had_scan_checkpoint { global_metrics().record_scanner_checkpoint_cleared(); @@ -2703,9 +2774,10 @@ pub(crate) async fn scan_data_folder_scoped( let root_hash = hash_path(&cache.info.name); let root_has_progress = data_usage_root_has_progress(&root); let pending_heals_changed = scanner.pending_heals_changed; - let raw_enumeration_cursor = scanner.take_raw_enumeration_cursor(); - let carry_forward_cache = - (raw_enumeration_cursor.is_some() && !root_has_progress).then(|| scanner.old_cache.cache.clone()); + let (raw_enumeration_cursor, raw_enumeration_page_index) = scanner.take_raw_enumeration_resume_state(); + let carry_forward_cache = ((raw_enumeration_cursor.is_some() || raw_enumeration_page_index.is_some()) + && !root_has_progress) + .then(|| scanner.old_cache.cache.clone()); if root_has_progress { scanner.carry_forward_old_children(&root_hash, &mut root); } @@ -2722,8 +2794,15 @@ pub(crate) async fn scan_data_folder_scoped( new_cache.info.scan_resume_after = None; new_cache.info.scan_coverage_receipt = None; } + if raw_enumeration_page_index.is_some() { + new_cache.info.scan_raw_enumeration_page_index = raw_enumeration_page_index; + new_cache.info.scan_checkpoint = None; + new_cache.info.scan_resume_after = None; + new_cache.info.scan_coverage_receipt = None; + } if partial_cache_is_useful(&root, pending_heals_changed) || new_cache.info.scan_raw_enumeration_cursor.is_some() + || new_cache.info.scan_raw_enumeration_page_index.is_some() || !new_cache.info.size_reconciliation.is_empty() { if new_cache.root().is_some() { diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index cc1add61f..c9ee80c17 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -2722,6 +2722,20 @@ async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_ assert!(raw_cursor.last_entry.is_some()); assert_ne!(raw_cursor.page_digest, [0; 32]); assert_eq!(partial_cache.validated_raw_enumeration_cursor(), Some(raw_cursor)); + let page_index = partial_cache + .validated_raw_enumeration_page_index() + .expect("raw enumeration cancellation should persist a validated page index"); + assert_eq!( + page_index + .indexed_entries() + .expect("persisted raw page index entries should validate") + .len(), + 1 + ); + assert_eq!( + page_index.committed_entries().expect("uncommitted raw page should validate"), + Vec::::new() + ); assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Runtime)); } @@ -3370,3 +3384,48 @@ fn test_should_log_failed_object_samples_after_initial_limit() { assert!(!should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY + 1)); assert!(should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY * 2)); } + +#[test] +fn raw_enumeration_progress_waits_for_resume_index_floor_before_revalidation() { + 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 + .ingest_partial_owner_entries(["entry-a".to_string(), "entry-b".to_string()], 2, generation) + .expect("initial entries should build a page"); + let generation = index.generation().expect("raw page index should expose next generation"); + index.commit_building_page(generation).expect("initial page should commit"); + + let mut progress = RawEnumerationProgress::new("bucket", Some(index)); + progress.record_entry("entry-b"); + assert!( + progress.page_index.is_some(), + "resume index must not be dropped before the current run observes the old index floor" + ); + + progress.record_entry("entry-a"); + assert!( + progress.page_index.is_some(), + "same entry identity after the observation floor should keep the resume index" + ); +} + +#[test] +fn raw_enumeration_progress_rejects_resume_index_after_floor_mismatch() { + 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 + .ingest_partial_owner_entries(["entry-a".to_string(), "entry-b".to_string()], 2, generation) + .expect("initial entries should build a page"); + let generation = index.generation().expect("raw page index should expose next generation"); + index.commit_building_page(generation).expect("initial page should commit"); + + let mut progress = RawEnumerationProgress::new("bucket", Some(index)); + progress.record_entry("entry-a"); + assert!(progress.page_index.is_some()); + + progress.record_entry("entry-c"); + assert!( + progress.page_index.is_none(), + "resume index must be discarded once enough current observations prove source drift" + ); +} diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs index b54f36665..a141b412b 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture/segment_observation.rs @@ -1,6 +1,7 @@ //! Fixture-only range diagnostics. No result is supplied to a scan selector. use super::*; +use crate::DATA_USAGE_CACHE_KEY_FORMAT; use std::collections::BTreeSet; const MAX_SEGMENTS: usize = 4; @@ -15,6 +16,89 @@ enum ProposalError { InvalidKey, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ProducerKind { + Put, + Delete, + DeleteMarker, + Multipart, + Replication, + Tier, + DirectoryObject, +} + +impl ProducerKind { + const REQUIRED: [Self; 7] = [ + Self::Put, + Self::Delete, + Self::DeleteMarker, + Self::Multipart, + Self::Replication, + Self::Tier, + Self::DirectoryObject, + ]; +} + +#[derive(Clone, Debug)] +struct SegmentObservationEnvelope<'a> { + source: DataUsageCacheSource, + bucket_incarnation: uuid::Uuid, + key_format: u16, + baseline_scan_plan_digest: DataUsageScanPlanDigest, + process_epoch: &'a str, + generation_start: u64, + generation_end: u64, + restart_gap: bool, + overflow: bool, + producers: BTreeSet<&'a str>, + keys: &'a [&'a str], +} + +#[derive(Clone, Debug)] +struct SegmentObservationProof<'a> { + source: DataUsageCacheSource, + bucket_incarnation: uuid::Uuid, + key_format: u16, + baseline_scan_plan_digest: DataUsageScanPlanDigest, + process_epoch: &'a str, +} + +fn producer_name(kind: ProducerKind) -> &'static str { + match kind { + ProducerKind::Put => "put", + ProducerKind::Delete => "delete", + ProducerKind::DeleteMarker => "delete_marker", + ProducerKind::Multipart => "multipart", + ProducerKind::Replication => "replication", + ProducerKind::Tier => "tier", + ProducerKind::DirectoryObject => "directory_object", + } +} + +fn trusted_fixture_proposal( + envelope: &SegmentObservationEnvelope<'_>, + proof: &SegmentObservationProof<'_>, +) -> Result, ProposalError> { + if envelope.source != proof.source + || envelope.bucket_incarnation.is_nil() + || envelope.bucket_incarnation != proof.bucket_incarnation + || envelope.key_format != proof.key_format + || envelope.baseline_scan_plan_digest != proof.baseline_scan_plan_digest + || envelope.process_epoch != proof.process_epoch + || envelope.generation_start == 0 + || envelope.generation_end < envelope.generation_start + || envelope.restart_gap + || envelope.overflow + || !ProducerKind::REQUIRED + .iter() + .all(|producer| envelope.producers.contains(producer_name(*producer))) + { + return Err(ProposalError::InvalidKey); + } + + fixture_proposal(envelope.keys) +} + // Keys come from successful fixture writes, not a production mutation stream. fn fixture_proposal(keys: &[&str]) -> Result, ProposalError> { let mut segments = BTreeSet::new(); @@ -54,6 +138,78 @@ fn segment_observation_fixture_proposal_bounds() { } } +#[test] +fn segment_observation_trusted_proposal_requires_identity_and_complete_producer_coverage() { + let source = DataUsageCacheSource::new(2, 3); + let incarnation = uuid::Uuid::from_u128(0x12345678123456781234567812345678); + let baseline = DataUsageScanPlanDigest([9; 32]); + let producers = ProducerKind::REQUIRED + .iter() + .map(|producer| producer_name(*producer)) + .collect::>(); + let envelope = SegmentObservationEnvelope { + source, + bucket_incarnation: incarnation, + key_format: DATA_USAGE_CACHE_KEY_FORMAT, + baseline_scan_plan_digest: baseline, + process_epoch: "epoch-a", + generation_start: 11, + generation_end: 13, + restart_gap: false, + overflow: false, + producers, + keys: &["hot/one", "hot/two", "archive/delete-marker"], + }; + let proof = SegmentObservationProof { + source, + bucket_incarnation: incarnation, + key_format: DATA_USAGE_CACHE_KEY_FORMAT, + baseline_scan_plan_digest: baseline, + process_epoch: "epoch-a", + }; + + assert_eq!( + trusted_fixture_proposal(&envelope, &proof), + Ok(BTreeSet::from(["archive".to_string(), "hot".to_string()])) + ); + + let mut wrong_source = envelope.clone(); + wrong_source.source = DataUsageCacheSource::new(2, 4); + assert_eq!(trusted_fixture_proposal(&wrong_source, &proof), Err(ProposalError::InvalidKey)); + + let mut missing_incarnation = envelope.clone(); + missing_incarnation.bucket_incarnation = uuid::Uuid::nil(); + assert_eq!(trusted_fixture_proposal(&missing_incarnation, &proof), Err(ProposalError::InvalidKey)); + + let mut wrong_key_format = envelope.clone(); + wrong_key_format.key_format = DATA_USAGE_CACHE_KEY_FORMAT.saturating_add(1); + assert_eq!(trusted_fixture_proposal(&wrong_key_format, &proof), Err(ProposalError::InvalidKey)); + + let mut wrong_baseline = envelope.clone(); + wrong_baseline.baseline_scan_plan_digest = DataUsageScanPlanDigest([8; 32]); + assert_eq!(trusted_fixture_proposal(&wrong_baseline, &proof), Err(ProposalError::InvalidKey)); + + let mut wrong_epoch = envelope.clone(); + wrong_epoch.process_epoch = "epoch-b"; + assert_eq!(trusted_fixture_proposal(&wrong_epoch, &proof), Err(ProposalError::InvalidKey)); + + let mut restart_gap = envelope.clone(); + restart_gap.restart_gap = true; + assert_eq!(trusted_fixture_proposal(&restart_gap, &proof), Err(ProposalError::InvalidKey)); + + let mut overflow = envelope.clone(); + overflow.overflow = true; + assert_eq!(trusted_fixture_proposal(&overflow, &proof), Err(ProposalError::InvalidKey)); + + let mut generation_gap = envelope.clone(); + generation_gap.generation_end = generation_gap.generation_start - 1; + assert_eq!(trusted_fixture_proposal(&generation_gap, &proof), Err(ProposalError::InvalidKey)); + + let mut missing_producer = envelope.clone(); + missing_producer.producers.remove(producer_name(ProducerKind::Replication)); + assert_eq!(trusted_fixture_proposal(&missing_producer, &proof), Err(ProposalError::InvalidKey)); +} + fn cache_value(cache: &DataUsageCache) -> serde_json::Value { let mut value = serde_json::to_value(cache).expect("serialize the entire cache"); // Children are a HashSet: canonicalize only that unordered field, without @@ -181,10 +337,6 @@ async fn walk_and_save(observe: bool) -> (Vec, serde_json::Value) { 2, "the two non-proposed segments must still be walked" ); - eprintln!( - "segment fixture: proposed={proposed:?}, actual_segments={walked_segments:?}, actual_walk_callbacks={}, production_producer_coverage=unverified", - paths.len() - ); } else { assert!(proposed_walked.lock().expect("read disabled observations").is_empty()); } diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 2b5ca78bc..af8023964 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2750,12 +2750,12 @@ mod tests { GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, - HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, - LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, - LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest, - MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest, - ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest, - RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerDirtyUsageSnapshotRequest, + HealBucketRequest, HealControlRequest, HealControlResponse, ListBucketRequest, ListDirRequest, ListVolumesRequest, + LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, + LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, + MakeVolumeRequest, MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, + ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, + RenameDataRequest, RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ScannerDirtyUsageSnapshotRequest, ScannerPublicationLeaseReleaseRequest, ScannerPublicationLeaseRequest, ServerInfoRequest, SettlePartTransactionRequest, SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest, StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationAbortRequest, @@ -2992,6 +2992,162 @@ mod tests { (manager, request, metadata) } + #[derive(Clone, Copy)] + enum HealControlTransportFault { + None, + DropBeforeAdmission, + DropAfterAdmission, + } + + struct HealControlTransportFaultService { + manager: Arc, + fingerprint: String, + coordinator_epoch: u64, + fault: HealControlTransportFault, + } + + #[tonic::async_trait] + impl rustfs_protos::proto_gen::node_service::heal_control_service_server::HealControlService + for HealControlTransportFaultService + { + async fn heal_control(&self, request: Request) -> Result, Status> { + let command = request.get_ref().command.to_vec(); + let body = rustfs_protos::canonical_heal_control_request_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + ) + .map_err(|_| Status::invalid_argument("heal control request length cannot be represented"))?; + crate::storage::storage_api::verify_tonic_canonical_body_digest(&request, &body) + .map_err(|err| Status::permission_denied(format!("heal control authentication failed: {err}")))?; + if request.get_ref().version != rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION { + return Err(Status::failed_precondition("unsupported heal control protocol version")); + } + if request.get_ref().topology_fingerprint != self.fingerprint { + return Err(Status::failed_precondition("heal control topology does not match")); + } + if matches!(self.fault, HealControlTransportFault::DropBeforeAdmission) { + return Err(Status::unavailable("transport failed before heal admission")); + } + + let envelope = rustfs_protos::heal_control::decode_envelope(&command).map_err(Status::invalid_argument)?; + let result = + execute_heal_control_envelope_with_manager(envelope, self.coordinator_epoch, Some(Arc::clone(&self.manager))) + .await?; + if matches!(self.fault, HealControlTransportFault::DropAfterAdmission) { + return Err(Status::unavailable("transport failed after heal admission")); + } + + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + request.get_ref().version, + &self.fingerprint, + &command, + &result, + ) + .map_err(|_| Status::internal("heal control response length cannot be represented"))?; + let response_proof = crate::storage::storage_api::sign_tonic_rpc_response_proof(&canonical_response) + .map_err(|_| Status::internal("heal control response proof is unavailable"))?; + Ok(Response::new(HealControlResponse { + success: true, + result: result.into(), + error_info: None, + response_proof: response_proof.into(), + })) + } + } + + async fn connect_faulty_heal_control_client( + manager: Arc, + fingerprint: &str, + coordinator_epoch: u64, + fault: HealControlTransportFault, + ) -> Option> { + let listener = match TcpListener::bind("127.0.0.1:0").await { + Ok(listener) => listener, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None, + Err(err) => panic!("test listener should bind: {err}"), + }; + let addr = listener.local_addr().expect("listener local address should be available"); + let service = HealControlTransportFaultService { + manager, + fingerprint: fingerprint.to_string(), + coordinator_epoch, + fault, + }; + + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service( + HealControlServiceServer::new(service) + .max_decoding_message_size(rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE) + .max_encoding_message_size(rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE), + ) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + .expect("faulty heal control transport server should run"); + }); + + Some( + HealControlServiceClient::connect(format!("http://{addr}")) + .await + .expect("faulty heal control test client should connect"), + ) + } + + fn signed_heal_control_request(fingerprint: &str, command: Vec) -> Request { + let mut request = Request::new(HealControlRequest { + version: rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + topology_fingerprint: fingerprint.to_string(), + command: command.into(), + }); + request.set_timeout(rustfs_protos::heal_control_execution_timeout()); + let body = rustfs_protos::canonical_heal_control_request_body( + request.get_ref().version, + &request.get_ref().topology_fingerprint, + &request.get_ref().command, + ) + .expect("heal control transport request should encode"); + set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + request + } + + async fn call_heal_control_transport( + client: &mut HealControlServiceClient, + fingerprint: &str, + command: Vec, + ) -> Result, Status> { + let response = client + .heal_control(signed_heal_control_request(fingerprint, command.clone())) + .await? + .into_inner(); + if !response.success { + return Err(Status::unknown( + response + .error_info + .unwrap_or_else(|| "peer heal control failed without an error".to_string()), + )); + } + let canonical_response = rustfs_protos::canonical_heal_control_response_body( + rustfs_protos::HEAL_CONTROL_PROTOCOL_VERSION, + fingerprint, + &command, + &response.result, + ) + .map_err(|_| Status::internal("heal control response length cannot be represented"))?; + crate::storage::storage_api::verify_tonic_rpc_response_proof(&canonical_response, &response.response_proof) + .map_err(|err| Status::permission_denied(format!("heal control response proof failed: {err}")))?; + Ok(response.result.to_vec()) + } + + fn encode_transport_start( + request: rustfs_heal_contracts::heal_channel::HealChannelRequest, + metadata: rustfs_protos::heal_control::RequestMetadata, + ) -> Vec { + let envelope = rustfs_protos::heal_control::Envelope::start(request, metadata).expect("valid start envelope"); + rustfs_protos::heal_control::encode_envelope(&envelope).expect("valid start command should encode") + } + #[tokio::test] async fn heal_start_retry_exact_forced_envelope_returns_cached_admission() { let (manager, request, metadata) = heal_start_retry_fixture(); @@ -3103,6 +3259,146 @@ mod tests { )); } + #[tokio::test] + async fn heal_control_transport_pre_admission_loss_retries_original_deadline_envelope() { + let _ = rustfs_credentials::set_global_rpc_secret("heal-control-transport-fault-test-secret".to_string()); + let (manager, request, metadata) = heal_start_retry_fixture(); + let fingerprint = "transport-pre-admission-fingerprint"; + let command = encode_transport_start(request.clone(), metadata); + let mut lost_before_admission = match connect_faulty_heal_control_client( + Arc::clone(&manager), + fingerprint, + metadata.coordinator_epoch, + HealControlTransportFault::DropBeforeAdmission, + ) + .await + { + Some(client) => client, + None => return, + }; + + let lost = call_heal_control_transport(&mut lost_before_admission, fingerprint, command.clone()) + .await + .expect_err("transport loss before admission must be visible to the caller"); + assert_eq!(lost.code(), tonic::Code::Unavailable); + assert_eq!( + manager.operations_snapshot().await.queue_length, + 0, + "pre-admission transport loss must not create a canonical task" + ); + assert!(matches!( + manager.get_task_status(&request.id).await, + Err(rustfs_heal::Error::TaskNotFound { .. }) + )); + + let mut retry = connect_faulty_heal_control_client( + Arc::clone(&manager), + fingerprint, + metadata.coordinator_epoch, + HealControlTransportFault::None, + ) + .await + .expect("retry listener should bind"); + let accepted = call_heal_control_transport(&mut retry, fingerprint, command) + .await + .expect("original envelope should remain usable within its deadline"); + let outcome = rustfs_protos::heal_control::decode_result(&accepted) + .and_then(|result| result.into_outcome(&request.id, metadata.coordinator_epoch)) + .expect("accepted retry should carry a canonical receipt"); + assert!(matches!( + outcome, + rustfs_protos::heal_control::Outcome::Start { + task_id, + admission: rustfs_protos::heal_control::Admission::Accepted, + } if task_id == request.id + )); + assert_eq!(manager.operations_snapshot().await.queue_length, 1); + } + + #[tokio::test] + async fn heal_control_transport_post_admission_loss_replays_receipt_but_fresh_force_start_is_distinct() { + let _ = rustfs_credentials::set_global_rpc_secret("heal-control-transport-fault-test-secret".to_string()); + let (manager, request, metadata) = heal_start_retry_fixture(); + let fingerprint = "transport-post-admission-fingerprint"; + let first_id = request.id.clone(); + let first_command = encode_transport_start(request.clone(), metadata); + let mut lost_after_admission = match connect_faulty_heal_control_client( + Arc::clone(&manager), + fingerprint, + metadata.coordinator_epoch, + HealControlTransportFault::DropAfterAdmission, + ) + .await + { + Some(client) => client, + None => return, + }; + + let lost = call_heal_control_transport(&mut lost_after_admission, fingerprint, first_command.clone()) + .await + .expect_err("post-admission response loss must be visible to the caller"); + assert_eq!(lost.code(), tonic::Code::Unavailable); + assert_eq!( + manager.operations_snapshot().await.queue_length, + 1, + "post-admission response loss must leave exactly one canonical task" + ); + + let mut retry = connect_faulty_heal_control_client( + Arc::clone(&manager), + fingerprint, + metadata.coordinator_epoch, + HealControlTransportFault::None, + ) + .await + .expect("retry listener should bind"); + let replayed = call_heal_control_transport(&mut retry, fingerprint, first_command) + .await + .expect("exact transport retry should replay the original receipt"); + let replayed = rustfs_protos::heal_control::decode_result(&replayed) + .and_then(|result| result.into_outcome(&first_id, metadata.coordinator_epoch)) + .expect("replayed retry should carry a canonical receipt"); + assert!(matches!( + replayed, + rustfs_protos::heal_control::Outcome::Start { + task_id, + admission: rustfs_protos::heal_control::Admission::Accepted, + } if task_id == first_id + )); + assert_eq!( + manager.operations_snapshot().await.queue_length, + 1, + "exact replay must not duplicate a destructive forced start" + ); + + let mut fresh_request = request; + fresh_request.id = Uuid::new_v4().to_string(); + let fresh_id = fresh_request.id.clone(); + let fresh_metadata = rustfs_protos::heal_control::RequestMetadata { + nonce: *Uuid::new_v4().as_bytes(), + ..metadata + }; + let fresh_command = encode_transport_start(fresh_request, fresh_metadata); + let fresh = call_heal_control_transport(&mut retry, fingerprint, fresh_command) + .await + .expect("fresh forceStart should keep explicit new-start semantics"); + let fresh = rustfs_protos::heal_control::decode_result(&fresh) + .and_then(|result| result.into_outcome(&fresh_id, metadata.coordinator_epoch)) + .expect("fresh forceStart should carry its own receipt"); + assert!(matches!( + fresh, + rustfs_protos::heal_control::Outcome::Start { + task_id, + admission: rustfs_protos::heal_control::Admission::Accepted, + } if task_id == fresh_id && task_id != first_id + )); + assert_eq!( + manager.operations_snapshot().await.queue_length, + 2, + "a new forceStart request must be counted as a distinct canonical task" + ); + } + #[tokio::test] async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() { let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None)); diff --git a/scripts/check_test_wiring.py b/scripts/check_test_wiring.py index 5f91dbfc8..0d43c274d 100755 --- a/scripts/check_test_wiring.py +++ b/scripts/check_test_wiring.py @@ -10,7 +10,6 @@ import re import subprocess import sys import tempfile -import tomllib import unittest import uuid import xml.etree.ElementTree as ET @@ -19,6 +18,11 @@ from unittest import mock from pathlib import Path from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +try: + import tomllib +except ModuleNotFoundError: + import tomli as tomllib + from scanner_abba import MAX_JSON_BYTES, digest, number, read_json, require, sha, write_json @@ -894,6 +898,10 @@ def scanner_heal_oracle_names(root: Path) -> tuple[str, ...]: require(isinstance(oracle, str) and oracle.endswith(".json"), f"invalid oracle for {case_id}") path = Path(oracle) require(not path.is_absolute() and ".." not in path.parts, f"oracle path escapes run directory for {case_id}") + require(requirement.get("evidence") in ("process-restart", "process-crash-restart"), + f"invalid evidence for {case_id}") + require(type(requirement.get("unclean_shutdown_marker")) is bool, + f"invalid unclean-shutdown marker expectation for {case_id}") names.add(oracle) return tuple(sorted(names)) @@ -1024,9 +1032,11 @@ def check_scanner_heal_evidence(root: Path, directory: Path, case_id: str) -> li require(digest(path) == execution["artifacts"][requirement["oracle"]], "oracle hash mismatch") oracle = read_json(path) evidence_integer(oracle.get("schema"), "oracle schema", 1, 1) - require(oracle.get("evidence") == "process-restart", "not real process-restart evidence") + require(oracle.get("evidence") == requirement["evidence"], f"not real {requirement['evidence']} evidence") require(oracle.get("case") == name and oracle.get("run_id") == run["run_id"], "oracle belongs to another case/run") require(oracle.get("source_revision") == run["source_revision"], "oracle source mismatch") + require(oracle.get("unclean_shutdown_marker") is requirement["unclean_shutdown_marker"], + "unclean-shutdown marker evidence mismatch") built = oracle["test_build"] for key in ("source_revision", "dirty", "lock_blob", "features"): require(built[key] == expected_build[key], f"compiled test {key} mismatch") @@ -1240,7 +1250,7 @@ class SelfTests(unittest.TestCase): run_dir.mkdir() registry = read_json(ROOT / ".config/scanner-heal-required-tests.json") write_json(root / ".config/scanner-heal-required-tests.json", registry) - requirement = registry["cases"]["background-target-restart"] + requirements = registry["cases"] binary = directory / "fake-binary" binary.write_bytes(b"parser fixture, not a real build") binary.chmod(0o700) @@ -1251,14 +1261,23 @@ class SelfTests(unittest.TestCase): "lock_blob": "c" * 40, "features": "default"}, "started_at": datetime.now(timezone.utc).timestamp() - 1, "binary": build, "test_binary": build}) - write_json(run_dir / "listing.json", {"rust-suites": {requirement["suite"]: { - "binary-id": requirement["suite"], "binary-path": str(binary), "package-name": "e2e_test", "build-platform": "target", + suite = "e2e_test" + write_json(run_dir / "listing.json", {"rust-suites": {suite: { + "binary-id": suite, "binary-path": str(binary), "package-name": "e2e_test", "build-platform": "target", "testcases": { - requirement["name"]: {"ignored": False, "filter-match": {"status": "matches"}} - }}}}) + requirement["name"]: {"ignored": False, "filter-match": {"status": "matches"}} + for requirement in requirements.values() + } + }}}) (run_dir / "junit.xml").write_text( - f'') + "" + + "".join( + f'' + for requirement in requirements.values() + ) + + "" + ) physical = {"has_xl_meta": True, "version_id": None, "data_dir": "data-generation", "erasure_index": 1, "data_blocks": 2, "parity_blocks": 2, "expected_part_numbers": [1], "present_part_fingerprints": {"1": {"size": 12, "sha256": "c" * 64}}, @@ -1268,15 +1287,17 @@ class SelfTests(unittest.TestCase): "expected_physical": physical, "physical": physical} objects = [dict(obj, key=f"object-{index}") for index in range(9)] objects[-1] = dict(objects[-1], expected_physical=None) - write_json(run_dir / "background-target-restart.json", { - "schema": 1, "evidence": "process-restart", "case": "background-target-restart", - "run_id": "a" * 32, "source_revision": "b" * 40, - "test_build": {"source_revision": "b" * 40, "dirty": False, "lock_blob": "c" * 40, - "features": "default", "target": "aarch64-apple-darwin", "profile": "debug", "rustflags_hex": ""}, - "binary_sha256": build["sha256"], "test_binary_sha256": build["sha256"], - "topology": {"nodes": 4, "drives_per_node": 1}, "pid_before": 10, "pid_after": 11, - "objects": objects, "node_listings": [[item["key"] for item in objects]] * 4, - }) + for case_id, requirement in requirements.items(): + write_json(run_dir / requirement["oracle"], { + "schema": 1, "evidence": requirement["evidence"], "case": case_id, + "run_id": "a" * 32, "source_revision": "b" * 40, + "test_build": {"source_revision": "b" * 40, "dirty": False, "lock_blob": "c" * 40, + "features": "default", "target": "aarch64-apple-darwin", "profile": "debug", "rustflags_hex": ""}, + "binary_sha256": build["sha256"], "test_binary_sha256": build["sha256"], + "topology": requirement["topology"], "pid_before": 10, "pid_after": 11, + "unclean_shutdown_marker": requirement["unclean_shutdown_marker"], + "objects": objects, "node_listings": [[item["key"] for item in objects]] * 4, + }) finish_scanner_heal_receipt(run_dir, 0, root) return root, run_dir @@ -1332,6 +1353,19 @@ class SelfTests(unittest.TestCase): self.assertEqual(len(errors), 21) self.assertTrue(all(error.startswith("pending ") for error in errors)) + def test_scanner_heal_crash_case_rejects_restart_oracle(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root, run_dir = self.scanner_heal_fixture(Path(tmp)) + path = run_dir / "background-target-crash.json" + oracle = read_json(path) + oracle["evidence"] = "process-restart" + oracle["unclean_shutdown_marker"] = False + write_json(path, oracle) + (run_dir / "execution.json").unlink() + finish_scanner_heal_receipt(run_dir, 0, root) + + self.assertTrue(check_scanner_heal_evidence(root, run_dir, "background-target-crash")) + def test_scanner_heal_rejects_broken_execution_and_artifacts(self) -> None: for fault in ("exit", "missing", "zero", "skipped", "failed", "retry", "filtered", "ignored", "stale", "hash", "binary", "synthetic", "wrong-run", "same-pid", "body", "parts", "listing", "topology"): diff --git a/scripts/scanner_abba.py b/scripts/scanner_abba.py index e728dc965..5edbd3a08 100644 --- a/scripts/scanner_abba.py +++ b/scripts/scanner_abba.py @@ -72,7 +72,12 @@ def report_number(value): def digest(path): with Path(path).open("rb") as stream: - return hashlib.file_digest(stream, "sha256").hexdigest() + if hasattr(hashlib, "file_digest"): + return hashlib.file_digest(stream, "sha256").hexdigest() + hasher = hashlib.sha256() + while chunk := stream.read(1024 * 1024): + hasher.update(chunk) + return hasher.hexdigest() def read_json(path):