From 2ba7f95547aad01c8947d37d134e7e6218c2f209 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 18:15:46 +0800 Subject: [PATCH] test(heal): write EC84 distributed restart oracle (#7481) Bind the distributed EC8+4 restart evidence lane to a scanner/heal oracle artifact so release validation can consume the real nextest run instead of accepting only a passing test. Require the registry to assert 8+4 erasure geometry for the three-node, four-drive case. Co-authored-by: zhi22915 --- .config/scanner-heal-required-tests.json | 2 + crates/e2e_test/src/distributed/heal_test.rs | 183 ++++++++++++++++++- 2 files changed, 184 insertions(+), 1 deletion(-) diff --git a/.config/scanner-heal-required-tests.json b/.config/scanner-heal-required-tests.json index ac6583d0c..2f265da53 100644 --- a/.config/scanner-heal-required-tests.json +++ b/.config/scanner-heal-required-tests.json @@ -43,6 +43,8 @@ "min_objects": 5, "max_objects": 5, "topology": {"nodes": 3, "drives_per_node": 4}, + "erasure": {"data_blocks": 8, "parity_blocks": 4}, + "erasure_set_drive_count": 12, "scope": "3-node x 4-drive single-set EC8+4, graceful target restart, preformatted replacement drive, exact unversioned S3 bodies and physical target shards; not mixed-version, multi-pool or long-window ABBA." }, "background-target-restart-ec8-4": { diff --git a/crates/e2e_test/src/distributed/heal_test.rs b/crates/e2e_test/src/distributed/heal_test.rs index 01d3f099a..f0e0516cd 100644 --- a/crates/e2e_test/src/distributed/heal_test.rs +++ b/crates/e2e_test/src/distributed/heal_test.rs @@ -12,12 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::harness::{DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_object, unique_bucket, wait_until}; +use super::harness::{ + DistCluster, DistLayout, TestResult, assert_inventory, get_object_bytes, payload_for, put_object, sha256_hex, unique_bucket, + wait_until, +}; use crate::chaos::{VersionShardCensus, census_object_version_on_disk, signed_admin_post}; use crate::common::init_logging; use aws_sdk_s3::Client; use aws_sdk_s3::primitives::ByteStream; +use serde_json::Value; +use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, HashSet}; +use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -25,6 +31,8 @@ const EC84_NODE_COUNT: usize = 3; const EC84_DRIVES_PER_NODE: usize = 4; const EC84_DATA_BLOCKS: usize = 8; const EC84_PARITY_BLOCKS: usize = 4; +const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart"; +const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json"; #[derive(Clone)] struct ExpectedShard { @@ -33,6 +41,79 @@ struct ExpectedShard { baseline: VersionShardCensus, } +struct ScannerHealEvidenceContext { + directory: PathBuf, + run: Value, +} + +fn file_sha256(path: &Path) -> TestResult { + let mut file = std::fs::File::open(path)?; + let mut digest = Sha256::new(); + let mut buffer = [0_u8; 64 * 1024]; + loop { + let read = file.read(&mut buffer)?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + Ok(digest.finalize().iter().map(|byte| format!("{byte:02x}")).collect()) +} + +fn compiled_test_identity() -> Value { + serde_json::json!({ + "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"), + "dirty": env!("RUSTFS_E2E_BUILD_DIRTY") != "false", + "lock_blob": env!("RUSTFS_E2E_BUILD_LOCK"), + "features": env!("RUSTFS_E2E_BUILD_FEATURES"), + "target": env!("RUSTFS_E2E_BUILD_TARGET"), + "profile": env!("RUSTFS_E2E_BUILD_PROFILE"), + "rustflags_hex": env!("RUSTFS_E2E_BUILD_RUSTFLAGS_HEX"), + }) +} + +fn string_field<'a>(value: &'a Value, path: &str) -> TestResult<&'a str> { + let mut current = value; + for segment in path.split('.') { + current = current + .get(segment) + .ok_or_else(|| format!("scanner/heal run receipt missing {path}"))?; + } + current + .as_str() + .filter(|text| !text.is_empty()) + .ok_or_else(|| format!("scanner/heal run receipt has invalid {path}").into()) +} + +fn scanner_heal_evidence_context() -> TestResult> { + let Some(directory) = std::env::var_os("RUSTFS_SCANNER_HEAL_RUN_DIR") else { + return Ok(None); + }; + let directory = PathBuf::from(directory); + let receipt = directory.join("run.json"); + if receipt.metadata()?.len() > 1024 * 1024 { + return Err("oversized scanner/heal execution receipt".into()); + } + let run: Value = serde_json::from_slice(&std::fs::read(receipt)?)?; + let built = compiled_test_identity(); + for key in ["source_revision", "dirty", "lock_blob", "features"] { + if built[key] != run["test_build"][key] { + return Err(format!("compiled test identity differs for {key}").into()); + } + } + let binary_path = PathBuf::from(string_field(&run, "binary.path")?); + if file_sha256(&binary_path)? != string_field(&run, "binary.sha256")? { + return Err("server binary must match the run receipt".into()); + } + if file_sha256(&std::env::current_exe()?)? != string_field(&run, "test_binary.sha256")? { + return Err("test executable must match the run receipt".into()); + } + if directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE).exists() { + return Err("scanner/heal oracle already exists; create a new execution receipt".into()); + } + Ok(Some(ScannerHealEvidenceContext { directory, run })) +} + fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult { if census.data_blocks != Some(EC84_DATA_BLOCKS) || census.parity_blocks != Some(EC84_PARITY_BLOCKS) { return Err(format!("object {key} did not use EC8+4 geometry: {census:?}").into()); @@ -49,6 +130,76 @@ fn assert_ec84_geometry(census: &VersionShardCensus, key: &str) -> TestResult { Ok(()) } +async fn write_scanner_heal_evidence( + context: ScannerHealEvidenceContext, + dist: &DistCluster, + bucket: &str, + expected: &[ExpectedShard], + outage_key: &str, + outage_body: &[u8], + replaced_drive: &Path, + pid_before: u32, + pid_after: u32, + node_listings: Vec>, +) -> TestResult { + let verifier = dist.client(0)?; + let mut objects = Vec::new(); + for item in expected { + let actual = get_object_bytes(&verifier, bucket, &item.key).await?; + let physical = census_object_version_on_disk(replaced_drive, bucket, &item.key, None)?; + objects.push(serde_json::json!({ + "key": item.key, + "version_id": null, + "expected_bytes": item.body.len(), + "actual_bytes": actual.len(), + "expected_sha256": sha256_hex(&item.body), + "actual_sha256": sha256_hex(&actual), + "expected_physical": item.baseline, + "physical": physical, + })); + } + let actual = get_object_bytes(&verifier, bucket, outage_key).await?; + let physical = census_object_version_on_disk(replaced_drive, bucket, outage_key, None)?; + objects.push(serde_json::json!({ + "key": outage_key, + "version_id": null, + "expected_bytes": outage_body.len(), + "actual_bytes": actual.len(), + "expected_sha256": sha256_hex(outage_body), + "actual_sha256": sha256_hex(&actual), + "expected_physical": null, + "physical": physical, + })); + + let evidence = serde_json::json!({ + "schema": 1, + "case": EC84_TARGET_DRIVE_RESTART_CASE, + "evidence": "process-restart", + "run_id": string_field(&context.run, "run_id")?, + "source_revision": string_field(&context.run, "source_revision")?, + "test_build": compiled_test_identity(), + "binary_sha256": string_field(&context.run, "binary.sha256")?, + "test_binary_sha256": string_field(&context.run, "test_binary.sha256")?, + "topology": {"nodes": EC84_NODE_COUNT, "drives_per_node": EC84_DRIVES_PER_NODE}, + "pid_before": pid_before, + "pid_after": pid_after, + "unclean_shutdown_marker": false, + "objects": objects, + "node_listings": node_listings, + }); + let data = serde_json::to_vec(&evidence)?; + if data.len() > 1024 * 1024 { + return Err("scanner/heal oracle exceeds the 1 MiB artifact budget".into()); + } + let mut output = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(context.directory.join(EC84_TARGET_DRIVE_RESTART_ORACLE))?; + output.write_all(&data)?; + output.sync_all()?; + Ok(()) +} + fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> TestResult { for key in keys { let census = census_object_version_on_disk(drive, bucket, key, None)?; @@ -87,6 +238,7 @@ async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult TestResult { init_logging(); + let evidence_context = scanner_heal_evidence_context()?; let mut dist = DistCluster::start_with_env( DistLayout::ThreeByFourEc84, &[ @@ -116,6 +268,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res let format_path = replaced_drive.join(".rustfs.sys").join("format.json"); let format_json = std::fs::read(&format_path)?; + let target_pid_before = dist.cluster.nodes[replaced_node] + .process + .as_ref() + .ok_or("target process is absent before graceful restart")? + .id(); dist.cluster.stop_node_gracefully(replaced_node).await?; let retired_drive = PathBuf::from(format!("{}.retired", replaced_drive.display())); std::fs::rename(&replaced_drive, &retired_drive)?; @@ -138,6 +295,11 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res .await?; dist.cluster.start_node(replaced_node).await?; + let target_pid_after = dist.cluster.nodes[replaced_node] + .process + .as_ref() + .ok_or("target process is absent after restart")? + .id(); let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#; let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url); @@ -167,6 +329,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res .chain(std::iter::once((outage_key.to_string(), outage_body.clone()))) .collect::>(); let expected_keys = inventory.keys().cloned().collect::>(); + let mut node_listings = Vec::new(); for node_index in 0..dist.cluster.nodes.len() { let client = dist.client(node_index)?; assert_inventory(&client, &bucket, &inventory).await?; @@ -177,6 +340,24 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res .filter_map(|object| object.key().map(str::to_owned)) .collect::>(); assert_eq!(observed, expected_keys, "node {node_index} listing diverged after EC8+4 heal"); + let mut observed = observed.into_iter().collect::>(); + observed.sort(); + node_listings.push(observed); + } + if let Some(context) = evidence_context { + write_scanner_heal_evidence( + context, + &dist, + &bucket, + &expected, + outage_key, + &outage_body, + &replaced_drive, + target_pid_before, + target_pid_after, + node_listings, + ) + .await?; } Ok(())