From 7c2c2abbce135f2780bb40d5d424492238a49110 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Tue, 22 Sep 2026 22:57:06 +0800 Subject: [PATCH 1/4] feat: add evidence-bound ACK durability calibration --- docs/DURABILITY_QUALIFICATION.md | 109 ++++++ fault/examples/ack-put-relaxed.yaml | 23 ++ fault/examples/ack-put-strict.yaml | 15 + src/bin/s3chaos.rs | 14 + src/fault/ack_calibration.rs | 525 ++++++++++++++++++++++++++++ src/fault/acknowledged_mutation.rs | 186 ++++++++++ src/fault/artifact_validation.rs | 72 ++++ src/fault/config.rs | 2 + src/fault/mod.rs | 1 + src/fault/runner/ack.rs | 96 +++++ src/fault/spec.rs | 8 + src/fault/suite.rs | 44 +++ src/fault/suite_plan.rs | 17 +- 13 files changed, 1111 insertions(+), 1 deletion(-) create mode 100644 docs/DURABILITY_QUALIFICATION.md create mode 100644 fault/examples/ack-put-relaxed.yaml create mode 100644 fault/examples/ack-put-strict.yaml create mode 100644 src/fault/ack_calibration.rs diff --git a/docs/DURABILITY_QUALIFICATION.md b/docs/DURABILITY_QUALIFICATION.md new file mode 100644 index 0000000..e56885d --- /dev/null +++ b/docs/DURABILITY_QUALIFICATION.md @@ -0,0 +1,109 @@ +# Durability live qualification + +A calibration result belongs to one detector, candidate image digest, workload, +and storage layout. Catalog `gate-candidate` is not evidence of calibration. +Fresh-volume replacement and bitrot remain Planned until their supported +variants have separate live evidence. + +## ACK positive and negative controls + +Prepare an explicitly authorized dedicated DM lab using [DM_FLAKEY.md](DM_FLAKEY.md). +Pin the context, namespace, Tenant, static PVs, node/device allowlist and candidate +image. Keep the same image digest, EC geometry, filesystem, mount options and +host writeback/journal settings for both controls. Record the RustFS, Operator, +and S3Chaos commits and image digests with the lab report. Never run the pair in +a loop: static volumes require supervised inspection and a fresh fixture between +attempts. + +Use the same explicit seed for both controls: + +```bash +export RUSTFS_FAULT_TEST_SEED=424242 +make fault-suite-validate SUITE=fault/examples/ack-put-strict.yaml +make fault-suite-validate SUITE=fault/examples/ack-put-relaxed.yaml +make fault-suite-run SUITE=fault/examples/ack-put-strict.yaml +``` + +`ackCalibration: strict` sets both `RUSTFS_DURABILITY_MODE=strict` and +`RUSTFS_NEW_BUCKET_DURABILITY_MODE=strict`. The latter matters: newer RustFS +builds seed newly created buckets with their own relaxed override by default. +Conflicting or duplicate ambient entries in `RUSTFS_FAULT_TEST_SERVER_ENV` are +rejected. The run records the actual Pod modes and resolved image identities, +and requires the signed bucket-durability GET to return an explicit matching +override before preparing the fault. An unsupported endpoint, inherited/null +mode, wrong bucket, changed Pod UID or mixed image digest stops calibration. +Only these two non-secret environment values are retained. + +Inspect and validate the exact emitted artifacts before cleanup. Confirm the +recorded context, namespace and Tenant, pin +`RUSTFS_FAULT_TEST_EXPECTED_CONTEXT`, and follow the DM recovery/cleanup runbook. +After preparing a fresh approved fixture with matching lab settings, run: + +```bash +make fault-suite-run SUITE=fault/examples/ack-put-relaxed.yaml +``` + +The relaxed control must produce the declared product failure tied to the +acknowledged key/version. A timeout, backend failure, unrelated missing key, +invalid artifact, or successful relaxed checker is not a detector hit. A single +failed drive can be masked by EC redundancy; if both modes pass, this detector +is unqualified for the tested layout. Do not weaken the oracle or report that +pair as calibrated. Investigate the crash window and failure scope separately. + +After both supervised runs, compare their exact emitted suite roots: + +```bash +cargo run --quiet --bin s3chaos -- fault-ack-calibration-analyze \ + '' '' +``` + +The analyzer revalidates native artifacts rather than trusting prior validation +reports or suite status. It requires strict PASS and relaxed observed ACK state +loss, independent run identities, the same candidate digest, detector, payload +and seed, ACK timing, recovery policy, EC geometry, cluster/storage class, and +pre-crash filesystem/mount options. Host kernel writeback settings and image +provenance must additionally be preserved in the operator's lab report; the +analyzer does not attest those external settings. Its output is evidence for +this pair, not automatic catalog promotion or proof of physical power loss. + +Repeat with individually reviewed single-attempt suites for overwrite, +delete-marker, zero-byte PUT and multipart completion. Choose the exact expected +loss classification before running the negative control. Do not change the +expectation after seeing an unrelated failure. No result for PUT qualifies the +other mutation types. + +## Fresh-volume and bitrot variants + +Use the existing supervised qualification entrypoint and its exact target JSON, +helper image, dedicated Local-PV configuration and identity checks. See +`make fault-qualify-list` and the matching manifests under `fault/planned/`. +Preflight requires an explicit context, namespace and Tenant; it must succeed +before any host mutation. Run only one of these commands per fresh prepared +fixture, preserving and analyzing its evidence before the next: + +```bash +make fault-qualify QUALIFICATION_CASE=fresh-volume-replacement-automatic-replacement +make fault-qualify QUALIFICATION_CASE=fresh-volume-replacement-admin-deep +make fault-qualify QUALIFICATION_CASE=on-disk-bitrot-automatic-scanner +make fault-qualify QUALIFICATION_CASE=on-disk-bitrot-admin-deep +``` + +These are four separate supervised runs, not a batch script. After each run: + +```bash +make fault-qualify-analyze RUN_ROOT='' +``` + +Retain the plan/result, pinned images/topology, full native artifact validation, +old/new drive generation and emptiness evidence, matching heal operation and +terminal state, exact-version mapping, and force-read isolation/restore proof. +Bitrot also needs the actual mutation receipt and matching checksum detection +within the corruption window. Missing detection, background-repair races, +unsupported diagnostics and incomplete restore are unqualified or failed, never +PASS. Stop on an unexpected failure and preserve live evidence before cleanup. + +The earlier external replacement versioned-marker PASS recorded in backlog +#2347 does not qualify its unversioned branch or both current K8s variants. Each +current variant requires its own complete run on the selected image. Promote a +Planned scenario only after all advertised variants have valid live receipts, +static gates and independent review; retain unresolved product defects. diff --git a/fault/examples/ack-put-relaxed.yaml b/fault/examples/ack-put-relaxed.yaml new file mode 100644 index 0000000..959dcbd --- /dev/null +++ b/fault/examples/ack-put-relaxed.yaml @@ -0,0 +1,23 @@ +apiVersion: rustfs.com/s3chaos/v1alpha1 +kind: FaultSuite +metadata: + name: ack-put-relaxed-control +budgets: + stopOnFirstFailure: true + continueOnSeverities: [] +artifacts: + required: strict +scenarios: + - name: dm-drop-writes-after-ack-put + ackCalibration: relaxed + workload: + objects: 64 + concurrency: 4 + expectedFailure: + classification: committed_version_missing + severity: fail_correctness + responsibilityDomain: product + evidenceRefs: + - checker-pre-recommit-report.json + - fault-evidence.json + - run-events.jsonl diff --git a/fault/examples/ack-put-strict.yaml b/fault/examples/ack-put-strict.yaml new file mode 100644 index 0000000..9486901 --- /dev/null +++ b/fault/examples/ack-put-strict.yaml @@ -0,0 +1,15 @@ +apiVersion: rustfs.com/s3chaos/v1alpha1 +kind: FaultSuite +metadata: + name: ack-put-strict-control +budgets: + stopOnFirstFailure: true + continueOnSeverities: [] +artifacts: + required: strict +scenarios: + - name: dm-drop-writes-after-ack-put + ackCalibration: strict + workload: + objects: 64 + concurrency: 4 diff --git a/src/bin/s3chaos.rs b/src/bin/s3chaos.rs index fe3b61e..4c3e6dd 100644 --- a/src/bin/s3chaos.rs +++ b/src/bin/s3chaos.rs @@ -69,6 +69,7 @@ async fn main() -> Result<()> { "fault-suite-run" => run_signal_aware(run_fault_suite(args)).await, "fault-suite-template" => print_fault_suite_template(), "fault-suite-validate" => validate_fault_suite(args), + "fault-ack-calibration-analyze" => analyze_ack_calibration(args), "fault-validate-artifacts" => validate_fault_artifacts_command(args), "fault-run-spec-equal" => validate_fault_run_spec_equivalence(args), "protocol-catalog-json" => print_protocol_catalog_json(), @@ -106,6 +107,7 @@ fn print_help() -> Result<()> { println!(" fault-suite-run "); println!(" fault-suite-template"); println!(" fault-suite-validate "); + println!(" fault-ack-calibration-analyze "); println!(" fault-validate-artifacts [--validation-summary-tsv]"); println!(" fault-run-spec-equal "); println!(" protocol-catalog-json"); @@ -682,6 +684,18 @@ fn validate_fault_suite(mut args: impl Iterator) -> Result<()> { Ok(()) } +fn analyze_ack_calibration(mut args: impl Iterator) -> Result<()> { + let strict = args.next().context("requires strict suite root")?; + let relaxed = args.next().context("requires relaxed suite root")?; + ensure!(args.next().is_none(), "accepts exactly two suite roots"); + let report = s3chaos::fault::ack_calibration::validate_ack_calibration_pair( + std::path::Path::new(&strict), + std::path::Path::new(&relaxed), + )?; + println!("{}", serde_json::to_string_pretty(&report)?); + Ok(()) +} + fn validate_fault_artifacts_command(mut args: impl Iterator) -> Result<()> { let scenario = args .next() diff --git a/src/fault/ack_calibration.rs b/src/fault/ack_calibration.rs new file mode 100644 index 0000000..44e609f --- /dev/null +++ b/src/fault/ack_calibration.rs @@ -0,0 +1,525 @@ +// Copyright 2025 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. + +//! Offline comparison of two supervised ACK calibration attempts. + +use anyhow::{Context, Result, ensure}; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; +use std::{ + fs, + path::{Path, PathBuf}, +}; + +use crate::fault::{ + acknowledged_mutation::{ACK_CALIBRATION_ARTIFACT, AckCalibrationEvidence, AckCalibrationMode}, + artifact_validation::{ + ArtifactValidationOptions, validate_expected_failure_artifacts, + validate_fault_artifacts_for_planned_attempt_and_write_report, + }, + recovery_health::RecoveryHealthReport, + spec::FaultRunSpec, + suite_plan::FaultSuitePlan, +}; + +#[derive(Serialize)] +pub struct AckCalibrationReport { + pub scenario: String, + pub strict_run_id: String, + pub relaxed_run_id: String, + pub image_digest: String, + pub loss_classification: String, + pub strict_suite_root: PathBuf, + pub relaxed_suite_root: PathBuf, + pub checked_scope: Vec<&'static str>, + pub external_settings_not_attested: Vec<&'static str>, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct SuiteSummary { + run_id: String, + status: String, + attempts: Vec, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct AttemptSummary { + run_id: String, + scenario: String, + status: String, + started_at_ms: u64, + ended_at_ms: u64, +} + +#[derive(Clone)] +struct Control { + root: PathBuf, + spec: FaultRunSpec, + mode: AckCalibrationEvidence, + health: RecoveryHealthReport, + mount: serde_json::Value, + loss: Option, +} + +fn read(path: &Path) -> Result { + serde_json::from_slice(&fs::read(path).with_context(|| format!("read {}", path.display()))?) + .with_context(|| format!("parse {}", path.display())) +} + +fn load_control(root: &Path, mode: AckCalibrationMode) -> Result { + let root = fs::canonicalize(root)?; + let plan: FaultSuitePlan = read(&root.join("suite-plan.json"))?; + let summary: SuiteSummary = read(&root.join("suite-summary.json"))?; + ensure!( + plan.attempts.len() == 1 && summary.attempts.len() == 1, + "calibration requires separate single-attempt supervised suites" + ); + ensure!( + summary.run_id == plan.run_id && summary.status == "succeeded", + "calibration suite did not succeed" + ); + let attempt = &plan.attempts[0]; + let observed = &summary.attempts[0]; + ensure!( + attempt.run_id.as_deref() == Some(observed.run_id.as_str()) + && observed.scenario == attempt.scenario + && observed.started_at_ms <= observed.ended_at_ms, + "calibration attempt identity or time window differs from plan" + ); + ensure!( + observed.status + == if mode == AckCalibrationMode::Strict { + "succeeded" + } else { + "expected-failure" + }, + "calibration control has the wrong outcome" + ); + let case = fs::canonicalize(&attempt.artifacts.case_dir)?; + ensure!( + case.starts_with(&root), + "calibration case escaped suite root" + ); + let spec: FaultRunSpec = read(&case.join("run-spec.json"))?; + ensure!( + spec.metadata.run_id == observed.run_id + && spec.scenario.name == attempt.scenario + && spec.scenario.case_name == attempt.case_name + && spec.scenario.ack_trigger == attempt.ack_trigger, + "calibration run spec differs from suite plan" + ); + ensure!( + spec.scenario + .ack_trigger + .as_ref() + .and_then(|trigger| trigger.calibration_mode) + == Some(mode), + "calibration run spec lacks requested mode" + ); + let options = ArtifactValidationOptions { + scenario: attempt.scenario.clone(), + artifact_root: case.clone(), + expected_workload_objects: spec.workload.object_count, + expected_workload_concurrency: spec.workload.concurrency, + expected_workload_versioning: spec.workload.versioning, + expected_rustfs_pod_count: spec.recovery.expected_rustfs_pod_count, + expected_stable_window_seconds: spec.recovery.stable_pod_window_seconds, + expected_recovery_stability_reread_seconds: spec.recovery.recovery_stability_reread_seconds, + expected_rustfs_volume_path: String::new(), + }; + let validated = + validate_fault_artifacts_for_planned_attempt_and_write_report(&options, &observed.run_id)?; + ensure!( + validated.run_succeeded == (mode == AckCalibrationMode::Strict), + "calibration artifact verdict differs from control" + ); + let loss = if mode == AckCalibrationMode::Relaxed { + let expected = attempt + .expected_failure + .as_ref() + .context("relaxed control lacks expectedFailure")?; + let failure = validate_expected_failure_artifacts( + &root, + &case, + &observed.run_id, + &attempt.scenario, + &attempt.case_name, + observed.started_at_ms, + observed.ended_at_ms, + )?; + expected.validate_observed( + failure.summary.classification(), + failure.summary.severity(), + failure.summary.responsibility_domain(), + failure.summary.primary_evidence_refs(), + )?; + let classification = failure.summary.classification(); + ensure!( + matches!( + classification, + "committed_version_missing" + | "delete_marker_missing" + | "deleted_object_resurrected" + ), + "relaxed calibration requires observed ACK state loss, not availability or harness failure" + ); + Some(classification.to_string()) + } else { + ensure!( + attempt.expected_failure.is_none(), + "strict control must require PASS" + ); + None + }; + let mode = read(&case.join(ACK_CALIBRATION_ARTIFACT))?; + let health = read(&case.join("recovery-health.json"))?; + let crash: serde_json::Value = read(&case.join("dm-crash-boundary.json"))?; + let mount = crash + .get("mount_before") + .context("calibration crash boundary lacks pre-crash mount")? + .clone(); + Ok(Control { + root, + spec, + mode, + health, + mount, + loss, + }) +} + +pub fn validate_ack_calibration_pair( + strict_root: &Path, + relaxed_root: &Path, +) -> Result { + let strict = load_control(strict_root, AckCalibrationMode::Strict)?; + let relaxed = load_control(relaxed_root, AckCalibrationMode::Relaxed)?; + compare_controls(strict, relaxed) +} + +fn compare_controls(strict: Control, relaxed: Control) -> Result { + ensure!( + strict.root != relaxed.root && strict.spec.metadata.run_id != relaxed.spec.metadata.run_id, + "calibration controls must be independent runs" + ); + ensure!( + strict.spec.scenario.name == relaxed.spec.scenario.name + && strict.spec.scenario.detector == relaxed.spec.scenario.detector + && strict.spec.workload == relaxed.spec.workload + && strict.spec.recovery == relaxed.spec.recovery, + "calibration controls differ in detector, payload/seed, or recovery contract" + ); + let strict_trigger = strict + .spec + .scenario + .ack_trigger + .as_ref() + .context("strict ACK trigger missing")?; + let relaxed_trigger = relaxed + .spec + .scenario + .ack_trigger + .as_ref() + .context("relaxed ACK trigger missing")?; + ensure!( + strict_trigger.mutation == relaxed_trigger.mutation + && strict_trigger.operation_timeout_ms == relaxed_trigger.operation_timeout_ms + && strict_trigger.max_ack_to_fault_ms == relaxed_trigger.max_ack_to_fault_ms, + "calibration controls use different ACK timing contracts" + ); + ensure!( + strict.spec.cluster.context == relaxed.spec.cluster.context + && strict.spec.cluster.storage_class == relaxed.spec.cluster.storage_class, + "calibration controls use different cluster or storage classes" + ); + let left = &strict.health.baseline; + let right = &relaxed.health.baseline; + ensure!( + left.standard_parity == right.standard_parity + && left.total_sets == right.total_sets + && left.drives_per_set == right.drives_per_set + && left.server_endpoints.len() == right.server_endpoints.len(), + "calibration controls use different EC geometry" + ); + for key in ["filesystem", "options"] { + let left = strict.mount[key] + .as_str() + .filter(|value| !value.is_empty()) + .context("strict mount contract missing")?; + let right = relaxed.mount[key] + .as_str() + .filter(|value| !value.is_empty()) + .context("relaxed mount contract missing")?; + ensure!( + left == right, + "calibration controls use different filesystem or mount options" + ); + } + let digest = |control: &Control| -> Result { + control + .mode + .pods + .first() + .and_then(|pod| pod.image_id.rsplit_once("sha256:")) + .map(|(_, digest)| digest.to_ascii_lowercase()) + .context("calibration image digest missing") + }; + let image_digest = digest(&strict)?; + ensure!( + image_digest == digest(&relaxed)?, + "calibration controls use different candidate image digests" + ); + Ok(AckCalibrationReport { + scenario: strict.spec.scenario.name, + strict_run_id: strict.spec.metadata.run_id, + relaxed_run_id: relaxed.spec.metadata.run_id, + image_digest, + loss_classification: relaxed.loss.context("relaxed loss missing")?, + strict_suite_root: strict.root, + relaxed_suite_root: relaxed.root, + checked_scope: vec![ + "native-artifacts", + "strict-pass", + "relaxed-ack-state-loss", + "image-digest", + "detector", + "workload-and-seed", + "ack-timing", + "recovery-policy", + "ec-geometry", + "cluster-and-storage-class", + "filesystem-and-mount-options", + ], + external_settings_not_attested: vec![ + "host-kernel-writeback-settings", + "image-source-provenance", + "physical-power-loss", + ], + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::fault::{ + acknowledged_mutation::{AckBucketMode, AckCalibrationPod}, + config::FaultTestConfig, + plan::FaultPlan, + recovery_health::RecoveryHealthBaseline, + scenarios::{FaultScenario, apply_catalog_defaults, scenario_spec}, + workload::WorkloadPlan, + }; + + fn control(mode: AckCalibrationMode) -> Control { + let mut config = FaultTestConfig::for_test("lab", "dm-storage"); + config.scenario = "dm-drop-writes-after-ack-put".into(); + config.ack_calibration = Some(mode); + apply_catalog_defaults(&mut config).unwrap(); + let scenario = FaultScenario::from_config(&config).unwrap(); + let catalog = scenario_spec(&scenario.name).unwrap(); + let plan = FaultPlan::from_scenario(&scenario, catalog).unwrap(); + let workload = WorkloadPlan::seeded(42, scenario.object_count, config.workload.concurrency); + let spec = FaultRunSpec::resolved( + &config, + &scenario, + catalog, + &plan, + &workload, + mode.as_str(), + "bucket", + ); + Control { + root: PathBuf::from(mode.as_str()), + mode: AckCalibrationEvidence { + scenario: scenario.name, + run_id: mode.as_str().into(), + mode, + observed_at_ms: 1, + bucket_response: AckBucketMode { + bucket: "bucket".into(), + mode: Some(mode.as_str().into()), + }, + pods: vec![AckCalibrationPod { + name: "pod".into(), + uid: "uid".into(), + image: "candidate".into(), + image_id: format!("containerd://sha256:{}", "a".repeat(64)), + process_mode: mode.as_str().into(), + new_bucket_mode: mode.as_str().into(), + }], + }, + health: RecoveryHealthReport { + scenario: config.scenario, + run_id: mode.as_str().into(), + baseline: RecoveryHealthBaseline { + observed_at_ms: 1, + deployment_id: "deployment".into(), + standard_parity: 2, + total_sets: vec![1], + drives_per_set: vec![4], + server_endpoints: vec!["server".into()], + drive_uuids: vec![], + }, + started_at_ms: 2, + completed_at_ms: 3, + timeout_seconds: 10, + attempts: 1, + first_healthy_at_ms: Some(3), + observation: None, + readiness: vec![], + violations: vec![], + passed: true, + }, + spec, + mount: serde_json::json!({"filesystem": "ext4", "options": "rw,relatime,data=ordered"}), + loss: (mode == AckCalibrationMode::Relaxed).then(|| "committed_version_missing".into()), + } + } + + #[test] + fn pair_requires_matching_image_workload_topology_and_crash_contract() { + let strict = control(AckCalibrationMode::Strict); + let relaxed = control(AckCalibrationMode::Relaxed); + assert!(compare_controls(strict.clone(), relaxed.clone()).is_ok()); + let mutations: &[fn(&mut Control)] = &[ + |control| control.mode.pods[0].image_id = format!("sha256:{}", "b".repeat(64)), + |control| control.spec.workload.plan.seed += 1, + |control| control.health.baseline.standard_parity += 1, + |control| control.mount["options"] = serde_json::json!("rw,data=writeback"), + |control| { + control + .spec + .scenario + .ack_trigger + .as_mut() + .unwrap() + .max_ack_to_fault_ms += 1 + }, + |control| control.spec.cluster.context = "another-lab".into(), + |control| control.loss = None, + ]; + for mutate in mutations { + let mut mismatch = relaxed.clone(); + mutate(&mut mismatch); + assert!(compare_controls(strict.clone(), mismatch).is_err()); + } + let mut same_run = relaxed; + same_run.spec.metadata.run_id = strict.spec.metadata.run_id.clone(); + assert!(compare_controls(strict, same_run).is_err()); + } + + fn write_control_header(base_dir: &Path, mode: AckCalibrationMode) -> PathBuf { + use crate::fault::{suite::FaultSuite, suite_plan::build_fault_suite_plan_expansion}; + let yaml = if mode == AckCalibrationMode::Strict { + include_str!("../../fault/examples/ack-put-strict.yaml") + } else { + include_str!("../../fault/examples/ack-put-relaxed.yaml") + }; + let suite: FaultSuite = serde_yaml_ng::from_str(yaml).unwrap(); + let mut config = FaultTestConfig::for_test("lab", "dm-storage"); + config.cluster.artifacts_dir = base_dir.into(); + let expansion = build_fault_suite_plan_expansion( + suite.resolve().unwrap(), + config, + format!("suite-{}", mode.as_str()), + ) + .unwrap(); + let root = PathBuf::from(&expansion.plan.artifact_root); + let attempt = &expansion.plan.attempts[0]; + let case = Path::new(&attempt.artifacts.case_dir); + fs::create_dir_all(case).unwrap(); + fs::write( + root.join("suite-plan.json"), + serde_json::to_vec(&expansion.plan).unwrap(), + ) + .unwrap(); + fs::write(root.join("suite-summary.json"), serde_json::to_vec(&serde_json::json!({ + "runId": expansion.plan.run_id, "status": "succeeded", "attempts": [{ + "runId": attempt.run_id, "scenario": attempt.scenario, + "status": if mode == AckCalibrationMode::Strict { "succeeded" } else { "expected-failure" }, + "startedAtMs": 1, "endedAtMs": 100 + }] + })).unwrap()).unwrap(); + let mut fixture = control(mode); + fixture.spec.metadata.run_id = attempt.run_id.clone().unwrap(); + fixture.spec.scenario.ack_trigger = attempt.ack_trigger.clone(); + fs::write( + case.join("run-spec.json"), + serde_json::to_vec(&fixture.spec).unwrap(), + ) + .unwrap(); + root + } + + #[test] + fn load_control_rejects_relaxed_pass_and_changed_mode_or_scenario() { + let dir = tempfile::tempdir().unwrap(); + let root = write_control_header(dir.path(), AckCalibrationMode::Relaxed); + let summary_path = root.join("suite-summary.json"); + let summary: serde_json::Value = read(&summary_path).unwrap(); + let mut passed = summary.clone(); + passed["attempts"][0]["status"] = serde_json::json!("succeeded"); + fs::write(&summary_path, serde_json::to_vec(&passed).unwrap()).unwrap(); + assert!( + load_control(&root, AckCalibrationMode::Relaxed) + .err() + .unwrap() + .to_string() + .contains("wrong outcome") + ); + fs::write(&summary_path, serde_json::to_vec(&summary).unwrap()).unwrap(); + let plan: FaultSuitePlan = read(&root.join("suite-plan.json")).unwrap(); + let spec_path = Path::new(&plan.attempts[0].artifacts.case_dir).join("run-spec.json"); + let original: FaultRunSpec = read(&spec_path).unwrap(); + for mode in [Some(AckCalibrationMode::Strict), None] { + let mut wrong = original.clone(); + wrong + .scenario + .ack_trigger + .as_mut() + .unwrap() + .calibration_mode = mode; + fs::write(&spec_path, serde_json::to_vec(&wrong).unwrap()).unwrap(); + assert!( + load_control(&root, AckCalibrationMode::Relaxed) + .err() + .unwrap() + .to_string() + .contains("differs from suite plan") + ); + } + let mut wrong = original.clone(); + wrong.scenario.name = "dm-drop-writes-after-ack-overwrite".into(); + fs::write(&spec_path, serde_json::to_vec(&wrong).unwrap()).unwrap(); + assert!( + load_control(&root, AckCalibrationMode::Relaxed) + .err() + .unwrap() + .to_string() + .contains("differs from suite plan") + ); + fs::write(&spec_path, serde_json::to_vec(&original).unwrap()).unwrap(); + assert!( + load_control(&root, AckCalibrationMode::Relaxed).is_err(), + "success labels cannot replace native evidence" + ); + } + + #[test] + fn missing_native_artifacts_never_qualify_a_pair() { + let empty = tempfile::tempdir().unwrap(); + assert!(validate_ack_calibration_pair(empty.path(), empty.path()).is_err()); + } +} diff --git a/src/fault/acknowledged_mutation.rs b/src/fault/acknowledged_mutation.rs index 8db0708..16c6b02 100644 --- a/src/fault/acknowledged_mutation.rs +++ b/src/fault/acknowledged_mutation.rs @@ -51,6 +51,152 @@ impl AcknowledgedMutationKind { } } +/// A calibration request selects both process and newly created bucket policy. +/// It does not itself establish that the detector has been live-qualified. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum AckCalibrationMode { + Strict, + Relaxed, +} + +impl AckCalibrationMode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Strict => "strict", + Self::Relaxed => "relaxed", + } + } + + pub(crate) fn configure(self, server_env: &mut Vec<(String, String)>) -> Result<()> { + for name in [ + "RUSTFS_DURABILITY_MODE", + "RUSTFS_NEW_BUCKET_DURABILITY_MODE", + ] { + let configured = server_env + .iter() + .filter(|(key, _)| key == name) + .collect::>(); + anyhow::ensure!( + configured.len() <= 1 && configured.iter().all(|(_, value)| value == self.as_str()), + "ACK calibration requires exactly one unambiguous {name}={} value", + self.as_str() + ); + if configured.is_empty() { + server_env.push((name.to_string(), self.as_str().to_string())); + } + } + Ok(()) + } +} + +pub(crate) const ACK_CALIBRATION_ARTIFACT: &str = "ack-calibration-mode.json"; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct AckCalibrationPod { + pub name: String, + pub uid: String, + pub image: String, + pub image_id: String, + pub process_mode: String, + pub new_bucket_mode: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct AckBucketMode { + pub bucket: String, + pub mode: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct AckCalibrationEvidence { + pub scenario: String, + pub run_id: String, + pub mode: AckCalibrationMode, + pub observed_at_ms: u64, + pub bucket_response: AckBucketMode, + pub pods: Vec, +} + +pub(crate) fn calibration_pod( + pod: &serde_json::Value, + mode: AckCalibrationMode, + image: &str, +) -> Result { + use anyhow::{Context, ensure}; + let containers = pod + .pointer("/spec/containers") + .and_then(serde_json::Value::as_array) + .context("calibration Pod lacks containers")?; + let container = containers + .iter() + .find(|container| container["name"] == "rustfs") + .context("calibration Pod lacks rustfs container")?; + let env = container["env"] + .as_array() + .context("calibration Pod lacks explicit modes")?; + let observed_mode = |name: &str| -> Result { + let values = env + .iter() + .filter(|entry| entry["name"] == name) + .collect::>(); + ensure!( + values.len() == 1 && values[0]["valueFrom"].is_null(), + "calibration Pod requires one literal {name}" + ); + let value = values[0]["value"] + .as_str() + .context("calibration mode is not literal")?; + ensure!( + value == mode.as_str(), + "calibration Pod {name} differs from requested mode" + ); + Ok(value.to_string()) + }; + ensure!( + container["image"] == image, + "calibration Pod image differs from run image" + ); + let statuses = pod + .pointer("/status/containerStatuses") + .and_then(serde_json::Value::as_array) + .context("calibration Pod lacks container statuses")?; + let status = statuses + .iter() + .find(|status| status["name"] == "rustfs") + .context("calibration Pod lacks rustfs image identity")?; + ensure!(status["ready"] == true, "calibration Pod is not Ready"); + let image_id = status["imageID"] + .as_str() + .context("calibration Pod lacks imageID")?; + let digest = image_id + .rsplit_once("sha256:") + .map(|(_, digest)| digest) + .context("calibration requires a resolved sha256 image identity")?; + ensure!( + digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()), + "calibration image digest is invalid" + ); + Ok(AckCalibrationPod { + name: pod + .pointer("/metadata/name") + .and_then(serde_json::Value::as_str) + .context("calibration Pod lacks name")? + .to_string(), + uid: pod + .pointer("/metadata/uid") + .and_then(serde_json::Value::as_str) + .context("calibration Pod lacks UID")? + .to_string(), + image: image.to_string(), + image_id: image_id.to_string(), + process_mode: observed_mode("RUSTFS_DURABILITY_MODE")?, + new_bucket_mode: observed_mode("RUSTFS_NEW_BUCKET_DURABILITY_MODE")?, + }) +} + /// A single mutation with no concurrent traffic, post-ACK verification read, /// or retry. Avoiding activity after the ACK prevents the calibration workload /// from accidentally flushing the metadata whose loss it is meant to detect. @@ -552,6 +698,46 @@ mod tests { workload::{ObjectSpec, S3WorkloadClient, StagedMultipartCleanupGuard}, }; + #[test] + fn calibration_sets_both_modes_and_rejects_conflicting_or_duplicate_policy() { + use super::AckCalibrationMode; + for mode in [AckCalibrationMode::Strict, AckCalibrationMode::Relaxed] { + let mut env = Vec::new(); + mode.configure(&mut env).unwrap(); + assert_eq!(env.len(), 2); + assert!(env.iter().all(|(_, value)| value == mode.as_str())); + mode.configure(&mut env).unwrap(); + assert_eq!(env.len(), 2); + env.push(env[0].clone()); + assert!(mode.configure(&mut env).is_err()); + } + let mut env = vec![("RUSTFS_NEW_BUCKET_DURABILITY_MODE".into(), "relaxed".into())]; + assert!(AckCalibrationMode::Strict.configure(&mut env).is_err()); + } + + #[test] + fn calibration_observation_rejects_bucket_policy_and_image_drift() { + use super::{AckCalibrationMode, calibration_pod}; + let pod = serde_json::json!({ + "metadata": {"name": "server-0", "uid": "pod-uid"}, + "spec": {"containers": [{"name": "rustfs", "image": "candidate", "env": [ + {"name": "RUSTFS_DURABILITY_MODE", "value": "strict"}, + {"name": "RUSTFS_NEW_BUCKET_DURABILITY_MODE", "value": "strict"} + ]}]}, + "status": {"containerStatuses": [{"name": "rustfs", "ready": true, + "imageID": format!("containerd://sha256:{}", "a".repeat(64))}]} + }); + assert!(calibration_pod(&pod, AckCalibrationMode::Strict, "candidate").is_ok()); + let mut mismatch = pod.clone(); + mismatch["spec"]["containers"][0]["env"][1]["value"] = serde_json::json!("relaxed"); + assert!(calibration_pod(&mismatch, AckCalibrationMode::Strict, "candidate").is_err()); + assert!(calibration_pod(&pod, AckCalibrationMode::Strict, "other-image").is_err()); + let mut missing_digest = pod; + missing_digest["status"]["containerStatuses"][0]["imageID"] = + serde_json::json!("candidate:latest"); + assert!(calibration_pod(&missing_digest, AckCalibrationMode::Strict, "candidate").is_err()); + } + enum MockReply { Ok, Hang, diff --git a/src/fault/artifact_validation.rs b/src/fault/artifact_validation.rs index 5d3c0e8..1b2f129 100644 --- a/src/fault/artifact_validation.rs +++ b/src/fault/artifact_validation.rs @@ -7319,6 +7319,76 @@ fn validate_ack_triggered_dm_artifacts( require_boundary, require_recovered, } = context; + if let Some(mode) = run_spec + .scenario + .ack_trigger + .as_ref() + .and_then(|trigger| trigger.calibration_mode) + { + use crate::fault::acknowledged_mutation::{ + ACK_CALIBRATION_ARTIFACT, AckCalibrationEvidence, + }; + let calibration: AckCalibrationEvidence = + read_json(&locate_artifact(root, case_name, ACK_CALIBRATION_ARTIFACT)?)?; + ensure!( + calibration.scenario == scenario + && calibration.run_id == run_id + && calibration.mode == mode, + "ACK calibration identity differs from run spec" + ); + ensure!( + calibration.bucket_response.bucket == bucket + && calibration.bucket_response.mode.as_deref() == Some(mode.as_str()), + "ACK calibration bucket override differs from requested mode" + ); + let target: TargetProof = + read_json(&locate_artifact(root, case_name, "target-proof.json")?)?; + ensure!( + target.run_id == run_id && target.scenario == scenario, + "ACK calibration target identity differs from run" + ); + ensure!( + !calibration.pods.is_empty() && calibration.pods.len() == target.resolved_pods.len(), + "ACK calibration must cover every proven Pod" + ); + let identities = calibration + .pods + .iter() + .map(|pod| (&pod.name, &pod.uid)) + .collect::>(); + ensure!( + identities.len() == calibration.pods.len() + && target + .resolved_pods + .iter() + .all(|pod| identities.contains(&(&pod.name, &pod.uid))), + "ACK calibration Pod identities differ from target proof" + ); + ensure!( + calibration + .pods + .iter() + .all(|pod| pod.image == run_spec.cluster.rustfs_image + && pod.image_id == calibration.pods[0].image_id + && pod + .image_id + .rsplit_once("sha256:") + .is_some_and(|(_, digest)| digest.len() == 64 + && digest.bytes().all(|byte| byte.is_ascii_hexdigit())) + && pod.process_mode == mode.as_str() + && pod.new_bucket_mode == mode.as_str()), + "ACK calibration Pod mode or image identity differs from run spec" + ); + ensure!( + calibration.observed_at_ms > 0 + && events + .iter() + .find(|event| event.stage == "fault-prepare" + && event.status == RunEventStatus::Started) + .is_some_and(|event| calibration.observed_at_ms <= event.at_ms), + "ACK calibration mode was not observed before fault preparation" + ); + } validate_history_scope_and_order(history, scenario, run_id, bucket)?; let preparation_started = events .iter() @@ -14673,6 +14743,7 @@ mod tests { let mut ack_spec = run_spec.clone(); ack_spec.scenario.ack_trigger = Some(crate::fault::spec::FaultRunAckTriggerSpec { + calibration_mode: None, mutation: crate::fault::acknowledged_mutation::AcknowledgedMutationKind::Put, operation_timeout_ms: 30_000, max_ack_to_fault_ms: 1_000, @@ -16946,6 +17017,7 @@ mod tests { }; let planned = FaultRunAckTriggerSpec { + calibration_mode: None, mutation: AcknowledgedMutationKind::Put, operation_timeout_ms: 30_000, max_ack_to_fault_ms: 5, diff --git a/src/fault/config.rs b/src/fault/config.rs index f92d5ed..cf22c37 100644 --- a/src/fault/config.rs +++ b/src/fault/config.rs @@ -113,6 +113,7 @@ pub struct FaultTestConfig { pub prefill_concurrency: usize, pub workload_seed: Option, pub request_timeout: Duration, + pub ack_calibration: Option, pub ack_operation_timeout: Duration, pub max_ack_to_fault: Duration, pub recovery_stability_reread: Duration, @@ -358,6 +359,7 @@ impl FaultTestConfig { "RUSTFS_FAULT_TEST_REQUEST_TIMEOUT_SECONDS", DEFAULT_REQUEST_TIMEOUT_SECONDS, )?), + ack_calibration: None, ack_operation_timeout: Duration::from_millis(ack_operation_timeout_ms), max_ack_to_fault: Duration::from_millis(max_ack_to_fault_ms), recovery_stability_reread: Duration::from_secs(recovery_stability_reread_seconds), diff --git a/src/fault/mod.rs b/src/fault/mod.rs index 8efd4a2..becdb0a 100644 --- a/src/fault/mod.rs +++ b/src/fault/mod.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod ack_calibration; pub mod acknowledged_mutation; pub mod admin_decommission; pub mod admin_rebalance; diff --git a/src/fault/runner/ack.rs b/src/fault/runner/ack.rs index 67c9617..52b0459 100644 --- a/src/fault/runner/ack.rs +++ b/src/fault/runner/ack.rs @@ -73,10 +73,22 @@ impl FaultRun<'_> { .prepare_quiet_mutation(&prepared.s3, staged_uploads) .await?; let result = async { + let calibration_pods = if self.config.ack_calibration.is_some() { + Some(self.observe_ack_calibration(&prepared.endpoint).await?) + } else { + None + }; let target = self .deadline .run(self.prove_target(&prepared.endpoint, preflight_phases)) .await?; + if let Some(pods) = calibration_pods { + ensure!( + pods.len() == target.pods_before.len() + && pods.iter().all(|pod| target.pods_before.contains(pod)), + "calibration Pod identities changed before target proof" + ); + } self.deadline.check()?; let prepared_fault = match self.prepare_ack_fault(&target) { Ok(prepared) => prepared, @@ -125,6 +137,90 @@ impl FaultRun<'_> { result } + async fn observe_ack_calibration( + &self, + endpoint: &str, + ) -> Result> { + use crate::fault::acknowledged_mutation::{ + ACK_CALIBRATION_ARTIFACT, AckBucketMode, AckCalibrationEvidence, calibration_pod, + }; + let mode = self + .config + .ack_calibration + .context("calibration mode missing")?; + let kubectl = crate::framework::kubectl::Kubectl::new(&self.config.cluster) + .namespaced(&self.config.cluster.test_namespace); + let selector = format!("rustfs.tenant={}", self.config.cluster.tenant_name); + let output = kubectl + .command(["get", "pods", "-l", &selector, "-o", "json"]) + .run_bounded(self.deadline.bounded_timeout(self.config.request_timeout)?) + .await?; + ensure!(output.code == Some(0), "cannot observe calibration Pods"); + let raw: serde_json::Value = serde_json::from_str(&output.stdout)?; + let pods = raw["items"] + .as_array() + .context("calibration Pod list is missing")? + .iter() + .map(|pod| calibration_pod(pod, mode, &self.config.cluster.rustfs_image)) + .collect::>>()?; + ensure!(!pods.is_empty(), "calibration requires observed Pods"); + ensure!( + pods.iter().all(|pod| pod.image_id == pods[0].image_id), + "calibration requires a homogeneous RustFS image digest" + ); + let (access_key, secret_key) = crate::framework::resources::test_credentials(); + let admin = crate::rustfs::RustfsAdminTransport::new( + endpoint, + "us-east-1", + access_key, + secret_key, + None, + "ack-calibration", + )?; + let response = self + .deadline + .run(admin.request( + http::Method::GET, + &format!("/rustfs/admin/v3/bucket-durability/{}", self.context.bucket), + &[], + Vec::new(), + None, + )) + .await?; + ensure!( + response.status == 200, + "calibration bucket durability GET failed: {}", + response.status + ); + let bucket_response: AckBucketMode = serde_json::from_slice(&response.body)?; + ensure!( + bucket_response.bucket == self.context.bucket + && bucket_response.mode.as_deref() == Some(mode.as_str()), + "effective bucket override differs from requested ACK calibration mode" + ); + let evidence = AckCalibrationEvidence { + scenario: self.scenario.name.clone(), + run_id: self.context.run_id.clone(), + mode, + observed_at_ms: now_ms(), + bucket_response, + pods, + }; + self.collector.write_text( + self.scenario.case_name, + ACK_CALIBRATION_ARTIFACT, + &serde_json::to_string_pretty(&evidence)?, + )?; + Ok(evidence + .pods + .into_iter() + .map(|pod| crate::fault::reporting::PodIdentity { + name: pod.name, + uid: pod.uid, + }) + .collect()) + } + async fn prepare_quiet_mutation( &self, s3: &S3WorkloadClient, diff --git a/src/fault/spec.rs b/src/fault/spec.rs index 1bb6375..49078df 100644 --- a/src/fault/spec.rs +++ b/src/fault/spec.rs @@ -135,6 +135,8 @@ pub struct FaultRunScenarioSpec { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct FaultRunAckTriggerSpec { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub calibration_mode: Option, pub mutation: crate::fault::acknowledged_mutation::AcknowledgedMutationKind, pub operation_timeout_ms: u64, pub max_ack_to_fault_ms: u64, @@ -274,6 +276,11 @@ impl FaultRunSpec { .required .push(DM_FILESYSTEM_CHECK_ARTIFACT.to_string()); } + if config.ack_calibration.is_some() { + artifacts + .required + .push(crate::fault::acknowledged_mutation::ACK_CALIBRATION_ARTIFACT.to_string()); + } Self { api_version: FAULT_RUN_API_VERSION.to_string(), kind: FAULT_RUN_KIND.to_string(), @@ -305,6 +312,7 @@ impl FaultRunSpec { ack_trigger: acknowledged_mutation_kind(&scenario.name).map(|mutation| { FaultRunAckTriggerSpec { mutation, + calibration_mode: config.ack_calibration, operation_timeout_ms: config.ack_operation_timeout.as_millis() as u64, max_ack_to_fault_ms: config.max_ack_to_fault.as_millis() as u64, } diff --git a/src/fault/suite.rs b/src/fault/suite.rs index 49fc152..80be953 100644 --- a/src/fault/suite.rs +++ b/src/fault/suite.rs @@ -79,6 +79,8 @@ pub struct FaultSuiteBudgets { #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct FaultSuiteScenario { pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ack_calibration: Option, /// Closed planned-case selector. Parsing it keeps qualification suites /// reviewable while ordinary suite resolution still rejects Planned /// catalog entries before any execution plan is produced. @@ -301,6 +303,8 @@ pub struct ResolvedFaultSuiteBudgets { #[serde(rename_all = "camelCase")] pub struct ResolvedFaultSuiteScenario { pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub ack_calibration: Option, pub execution_type: String, #[serde(skip_serializing_if = "Option::is_none")] pub storage_recovery_case: Option, @@ -545,6 +549,20 @@ impl ResolvedFaultSuiteScenario { if let Some(expected_failure) = &scenario.expected_failure { expected_failure.validate(&scenario.name)?; } + if let Some(mode) = scenario.ack_calibration { + use crate::fault::acknowledged_mutation::AckCalibrationMode; + ensure!( + crate::fault::scenarios::acknowledged_mutation_kind(&scenario.name).is_some(), + "ackCalibration requires an ACK-triggered scenario" + ); + ensure!( + match mode { + AckCalibrationMode::Strict => scenario.expected_failure.is_none(), + AckCalibrationMode::Relaxed => scenario.expected_failure.is_some(), + }, + "strict ACK calibration requires PASS; relaxed requires a typed expectedFailure" + ); + } let params = scenario.params.clone().unwrap_or_default(); ensure!( spec.param_schema != crate::fault::scenarios::FaultParameterSchema::QuorumIo @@ -600,6 +618,7 @@ impl ResolvedFaultSuiteScenario { "injection" } .to_string(), + ack_calibration: scenario.ack_calibration, storage_recovery_case: scenario.storage_recovery_case, params, repetitions: scenario.repetitions, @@ -870,6 +889,31 @@ mod tests { }; use std::collections::{BTreeMap, BTreeSet}; + #[test] + fn calibration_requires_ack_scenario_and_opposite_control_outcomes() { + let yaml = |scenario: &str, mode: &str, expected: bool| { + format!( + "apiVersion: rustfs.com/s3chaos/v1alpha1\nkind: FaultSuite\nmetadata:\n name: calibration\nscenarios:\n - name: {scenario}\n ackCalibration: {mode}\n{}", + if expected { + " expectedFailure:\n classification: committed_version_missing\n severity: fail_correctness\n responsibilityDomain: product\n evidenceRefs: [checker-pre-recommit-report.json, fault-evidence.json, run-events.jsonl]\n" + } else { + "" + } + ) + }; + for (scenario, mode, expected, valid) in [ + ("dm-drop-writes-after-ack-put", "strict", false, true), + ("dm-drop-writes-after-ack-put", "relaxed", true, true), + ("dm-drop-writes-after-ack-put", "strict", true, false), + ("dm-drop-writes-after-ack-put", "relaxed", false, false), + ("io-eio", "strict", false, false), + ] { + let suite: FaultSuite = + serde_yaml_ng::from_str(&yaml(scenario, mode, expected)).unwrap(); + assert_eq!(suite.resolve().is_ok(), valid, "{scenario} {mode}"); + } + } + #[test] fn resolves_valid_fault_suite() { let suite = serde_yaml_ng::from_str::( diff --git a/src/fault/suite_plan.rs b/src/fault/suite_plan.rs index 924aaa7..513738f 100644 --- a/src/fault/suite_plan.rs +++ b/src/fault/suite_plan.rs @@ -582,6 +582,7 @@ impl FaultSuitePlanAttempt { ack_trigger: acknowledged_mutation_kind(&input.scenario.name).map(|mutation| { FaultRunAckTriggerSpec { mutation, + calibration_mode: input.config.ack_calibration, operation_timeout_ms: input.config.ack_operation_timeout.as_millis() as u64, max_ack_to_fault_ms: input.config.max_ack_to_fault.as_millis() as u64, } @@ -635,7 +636,17 @@ impl FaultSuitePlanAttempt { artifacts: FaultSuitePlanArtifacts { attempt_dir: input.attempt_dir.display().to_string(), case_dir: case_dir.display().to_string(), - required: FaultRunArtifactSpec::required_names_for_scenario(&input.scenario.name), + required: { + let mut required = + FaultRunArtifactSpec::required_names_for_scenario(&input.scenario.name); + if input.config.ack_calibration.is_some() { + required.push( + crate::fault::acknowledged_mutation::ACK_CALIBRATION_ARTIFACT + .to_string(), + ); + } + required + }, event_stream: "run-events.jsonl".to_string(), }, budget: input.budget, @@ -754,6 +765,10 @@ fn scenario_config( // ordinary catalog scenarios from the same process environment and must // never inherit that authorization into its attempts. config.qualify_planned_admin = false; + config.ack_calibration = scenario.ack_calibration; + if let Some(mode) = config.ack_calibration { + mode.configure(&mut config.cluster.rustfs_env)?; + } config.scenario = scenario.name.clone(); config.scenario_parameters = scenario.params.clone(); config.storage_recovery_case = scenario.storage_recovery_case; From 7eb603586c90785f87c06bff5fd70fb36c861cfc Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Tue, 22 Sep 2026 23:20:41 +0800 Subject: [PATCH 2/4] fix: bind ACK calibration to immutable runtime evidence --- docs/DURABILITY_QUALIFICATION.md | 3 +- src/fault/ack_calibration.rs | 49 ++- src/fault/acknowledged_mutation.rs | 36 ++ src/fault/artifact_validation.rs | 577 ++++++++++++++++++++++++++++- src/fault/runner/ack.rs | 23 +- src/fault/suite_plan.rs | 3 + 6 files changed, 671 insertions(+), 20 deletions(-) diff --git a/docs/DURABILITY_QUALIFICATION.md b/docs/DURABILITY_QUALIFICATION.md index e56885d..0f8fee5 100644 --- a/docs/DURABILITY_QUALIFICATION.md +++ b/docs/DURABILITY_QUALIFICATION.md @@ -9,7 +9,8 @@ variants have separate live evidence. Prepare an explicitly authorized dedicated DM lab using [DM_FLAKEY.md](DM_FLAKEY.md). Pin the context, namespace, Tenant, static PVs, node/device allowlist and candidate -image. Keep the same image digest, EC geometry, filesystem, mount options and +image as `name@sha256:`; mutable tags are rejected for calibration. +Keep the same pinned image and observed runtime digest, EC geometry, filesystem, mount options and host writeback/journal settings for both controls. Record the RustFS, Operator, and S3Chaos commits and image digests with the lab report. Never run the pair in a loop: static volumes require supervised inspection and a fresh fixture between diff --git a/src/fault/ack_calibration.rs b/src/fault/ack_calibration.rs index 44e609f..889cfef 100644 --- a/src/fault/ack_calibration.rs +++ b/src/fault/ack_calibration.rs @@ -107,7 +107,19 @@ fn load_control(root: &Path, mode: AckCalibrationMode) -> Result { }, "calibration control has the wrong outcome" ); - let case = fs::canonicalize(&attempt.artifacts.case_dir)?; + let relative_case = Path::new(&attempt.artifacts.case_dir) + .strip_prefix(&plan.artifact_root) + .context("calibration case is outside persisted suite root")?; + ensure!( + relative_case.components().count() == 2 + && relative_case.file_name().and_then(|name| name.to_str()) + == Some(attempt.case_name.as_str()) + && relative_case + .components() + .all(|component| matches!(component, std::path::Component::Normal(_))), + "calibration case contains unsafe path components" + ); + let case = fs::canonicalize(root.join(relative_case))?; ensure!( case.starts_with(&root), "calibration case escaped suite root" @@ -128,9 +140,24 @@ fn load_control(root: &Path, mode: AckCalibrationMode) -> Result { == Some(mode), "calibration run spec lacks requested mode" ); + ensure!( + spec.workload.mode == attempt.workload.mode + && spec.workload.object_count == attempt.workload.objects + && spec.workload.concurrency == attempt.workload.concurrency + && spec.workload.versioning == attempt.workload.versioning + && spec.workload.seed == attempt.workload.seed + && spec.workload.operation_mix == attempt.workload.operation_mix + && spec.workload.prefill_concurrency == attempt.workload.prefill_concurrency + && spec.workload.request_timeout_seconds == attempt.workload.request_timeout_seconds + && spec.cluster.rustfs_image == plan.cluster.rustfs_image, + "calibration workload or candidate image differs from suite plan" + ); let options = ArtifactValidationOptions { scenario: attempt.scenario.clone(), - artifact_root: case.clone(), + artifact_root: case + .parent() + .context("calibration case lacks attempt root")? + .to_path_buf(), expected_workload_objects: spec.workload.object_count, expected_workload_concurrency: spec.workload.concurrency, expected_workload_versioning: spec.workload.versioning, @@ -241,7 +268,8 @@ fn compare_controls(strict: Control, relaxed: Control) -> Result Result Result<()> { + anyhow::ensure!( + image + .rsplit_once("@sha256:") + .is_some_and(|(name, digest)| !name.is_empty() + && !name.chars().any(char::is_whitespace) + && digest.len() == 64 + && digest.bytes().all(|byte| byte.is_ascii_hexdigit())), + "ACK calibration requires a RustFS image pinned as name@sha256:<64 hex digits>" + ); + Ok(()) +} + pub(crate) const ACK_CALIBRATION_ARTIFACT: &str = "ack-calibration-mode.json"; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -99,6 +112,7 @@ pub(crate) struct AckCalibrationPod { pub uid: String, pub image: String, pub image_id: String, + pub container_id: String, pub process_mode: String, pub new_bucket_mode: String, } @@ -192,6 +206,11 @@ pub(crate) fn calibration_pod( .to_string(), image: image.to_string(), image_id: image_id.to_string(), + container_id: status["containerID"] + .as_str() + .filter(|value| !value.is_empty()) + .context("calibration Pod lacks containerID")? + .to_string(), process_mode: observed_mode("RUSTFS_DURABILITY_MODE")?, new_bucket_mode: observed_mode("RUSTFS_NEW_BUCKET_DURABILITY_MODE")?, }) @@ -698,6 +717,22 @@ mod tests { workload::{ObjectSpec, S3WorkloadClient, StagedMultipartCleanupGuard}, }; + #[test] + fn calibration_image_requires_an_immutable_digest_reference() { + assert!( + super::require_calibration_image(&format!("rustfs/rustfs@sha256:{}", "a".repeat(64))) + .is_ok() + ); + for image in [ + "rustfs/rustfs:latest", + "rustfs/rustfs@sha256:abc", + "@sha256:", + "rustfs/rustfs@sha512:abc", + ] { + assert!(super::require_calibration_image(image).is_err()); + } + } + #[test] fn calibration_sets_both_modes_and_rejects_conflicting_or_duplicate_policy() { use super::AckCalibrationMode; @@ -725,6 +760,7 @@ mod tests { {"name": "RUSTFS_NEW_BUCKET_DURABILITY_MODE", "value": "strict"} ]}]}, "status": {"containerStatuses": [{"name": "rustfs", "ready": true, + "containerID": "containerd://process-1", "imageID": format!("containerd://sha256:{}", "a".repeat(64))}]} }); assert!(calibration_pod(&pod, AckCalibrationMode::Strict, "candidate").is_ok()); diff --git a/src/fault/artifact_validation.rs b/src/fault/artifact_validation.rs index 1b2f129..1218779 100644 --- a/src/fault/artifact_validation.rs +++ b/src/fault/artifact_validation.rs @@ -7328,6 +7328,9 @@ fn validate_ack_triggered_dm_artifacts( use crate::fault::acknowledged_mutation::{ ACK_CALIBRATION_ARTIFACT, AckCalibrationEvidence, }; + crate::fault::acknowledged_mutation::require_calibration_image( + &run_spec.cluster.rustfs_image, + )?; let calibration: AckCalibrationEvidence = read_json(&locate_artifact(root, case_name, ACK_CALIBRATION_ARTIFACT)?)?; ensure!( @@ -7358,11 +7361,16 @@ fn validate_ack_triggered_dm_artifacts( .collect::>(); ensure!( identities.len() == calibration.pods.len() - && target - .resolved_pods - .iter() - .all(|pod| identities.contains(&(&pod.name, &pod.uid))), - "ACK calibration Pod identities differ from target proof" + && target.resolved_pods.iter().all(|pod| identities + .contains(&(&pod.name, &pod.uid)) + && calibration + .pods + .iter() + .any(|observed| observed.name == pod.name + && !observed.container_id.is_empty() + && pod.rustfs_container_id.as_deref() + == Some(observed.container_id.as_str()))), + "ACK calibration Pod/container identities differ from target proof" ); ensure!( calibration @@ -7381,6 +7389,10 @@ fn validate_ack_triggered_dm_artifacts( ); ensure!( calibration.observed_at_ms > 0 + && events + .iter() + .find(|event| event.stage == "run" && event.status == RunEventStatus::Started) + .is_some_and(|event| event.at_ms <= calibration.observed_at_ms) && events .iter() .find(|event| event.stage == "fault-prepare" @@ -17997,6 +18009,561 @@ mod tests { } } + fn write_native_ack_calibration_control( + root: &std::path::Path, + relaxed: bool, + ) -> std::path::PathBuf { + use crate::fault::host_storage::{ + DM_FILESYSTEM_CHECK_ARTIFACT, HOST_STORAGE_CLEANUP_ARTIFACT, + }; + use crate::fault::{ + acknowledged_mutation::AckCalibrationMode, suite::FaultSuite, + suite_plan::build_fault_suite_plan_expansion, + }; + use std::path::{Path, PathBuf}; + let mode = if relaxed { + AckCalibrationMode::Relaxed + } else { + AckCalibrationMode::Strict + }; + let yaml = if relaxed { + include_str!("../../fault/examples/ack-put-relaxed.yaml") + } else { + include_str!("../../fault/examples/ack-put-strict.yaml") + }; + let yaml = yaml + .replace("checker-pre-recommit-report.json", "checker-report.json") + .replace("objects: 64", "objects: 12"); + let suite: FaultSuite = serde_yaml_ng::from_str(&yaml).unwrap(); + let mut config = FaultTestConfig::for_test("real-cluster", "rustfs-fault-dm"); + config.cluster.artifacts_dir = root.to_path_buf(); + config.cluster.rustfs_image = format!("rustfs/rustfs@sha256:{}", "a".repeat(64)); + config.workload_seed = Some(42); + config.max_ack_to_fault = Duration::from_millis(5); + let expansion = build_fault_suite_plan_expansion( + suite.resolve().unwrap(), + config, + format!("suite-{}", mode.as_str()), + ) + .unwrap(); + let suite_root = PathBuf::from(&expansion.plan.artifact_root); + let attempt = &expansion.plan.attempts[0]; + let run_id = attempt.run_id.as_deref().unwrap(); + let mut config = expansion.attempts[0].config.clone(); + config.workload = crate::fault::config::FaultWorkloadProfile::new(12, 4).unwrap(); + config.prefill_concurrency = 4; + config.dm_name = Some("rustfs-fault-dm".into()); + config.dm_node = Some("worker-a".into()); + config.dm_mount_path = Some("/data/rustfs-fault/dm-volume".into()); + let scenario = FaultScenario::from_config(&config).unwrap(); + let catalog = scenario_spec(&scenario.name).unwrap(); + let plan = FaultPlan::from_scenario(&scenario, catalog).unwrap(); + let workload = WorkloadPlan::seeded(config.workload_seed.unwrap(), 12, 4); + let run_spec = FaultRunSpec::resolved( + &config, &scenario, catalog, &plan, &workload, run_id, "bucket", + ); + let case_dir = PathBuf::from(&attempt.artifacts.case_dir); + let attempt_dir = Path::new(&attempt.artifacts.attempt_dir); + write_success_artifacts(attempt_dir, &scenario.name); + fs::rename( + attempt_dir.join("fault_io_eio_preserves_committed_objects"), + &case_dir, + ) + .unwrap(); + for entry in fs::read_dir(&case_dir).unwrap() { + let path = entry.unwrap().path(); + if path + .extension() + .is_some_and(|ext| ext == "json" || ext == "jsonl" || ext == "yaml") + { + let raw = fs::read_to_string(&path) + .unwrap() + .replace("run-00000000-0000-4000-8000-000000000001", run_id) + .replace( + "fault_io_eio_preserves_committed_objects", + scenario.case_name, + ) + .replace("fast-csi", "rustfs-fault-dm"); + fs::write(path, raw).unwrap(); + } + } + for name in ["workload-summary.json", "recommit-report.json"] { + fs::remove_file(case_dir.join(name)).unwrap(); + } + let mut workload_json = json!(workload); + workload_json["scenario"] = json!(scenario.name); + workload_json["run_id"] = json!(run_id); + write_json(&case_dir, "workload-plan.json", &workload_json); + write_json(&case_dir, "run-spec.json", &json!(run_spec)); + fs::write( + case_dir.join("run-spec.yaml"), + serde_yaml_ng::to_string(&run_spec).unwrap(), + ) + .unwrap(); + let mut metadata: Value = read_json(&case_dir.join("run-metadata.json")).unwrap(); + metadata["backend"] = json!("device-mapper"); + metadata["target"] = json!("dedicated-block-device"); + metadata["fault_duration_seconds"] = json!(config.duration.as_secs()); + metadata["percent"] = json!(config.percent); + metadata["workload_versioning"] = json!(true); + write_json(&case_dir, "run-metadata.json", &metadata); + let mut target_proof = TargetProof::from_plan(&config, &scenario, catalog, &plan, run_id) + .with_resolved_pod_proofs((0..4).map(|index| { + let mut pod = TargetResolvedPodProof::new(format!("p{index}"), format!("u{index}")) + .with_node(if index == 0 { + "worker-a".into() + } else { + format!("node-{index}") + }) + .with_ready(true); + pod.rustfs_container_id = Some(format!("containerd://process-{index}")); + pod = pod.with_persistent_volume_claims(vec![TargetPersistentVolumeClaimProof { + name: format!("data-rustfs-{index}"), + uid: format!("pvc-uid-{index}"), + volume_name: Some(if index == 0 { + "pv-a".into() + } else { + format!("pv-{index}") + }), + storage_class: Some("rustfs-fault-dm".into()), + persistent_volume: Some(TargetPersistentVolumeProof { + name: if index == 0 { + "pv-a".into() + } else { + format!("pv-{index}") + }, + uid: if index == 0 { + "pv-uid-a".into() + } else { + format!("pv-uid-{index}") + }, + source: Some("local".into()), + required_node_affinity: None, + node: Some(if index == 0 { + "storage-host-a".into() + } else { + format!("node-{index}") + }), + device_or_path: Some(if index == 0 { + "/data/rustfs-fault/dm-volume".into() + } else { + format!("/data/rustfs-fault/volume-{index}") + }), + }), + }]); + pod + })); + target_proof.generated_at_ms = 6; + write_json(&case_dir, "target-proof.json", &json!(target_proof)); + let mut preflight: Value = read_json(&case_dir.join("preflight-summary.json")).unwrap(); + preflight["phases"] + .as_array_mut() + .unwrap() + .push(json!({"name":"host-storage-mutation-proof","status":"passed","checks":[]})); + write_json(&case_dir, "preflight-summary.json", &preflight); + let host_proof = HostStorageMutationProof::prove_device_mapper( + HostStorageMutationIntent { + scenario: scenario.name.clone(), + fault_name: run_spec.faults[0].name.clone(), + fault_kind: run_spec.faults[0].kind.clone(), + run_id: run_id.to_string(), + context: config.cluster.context.clone(), + namespace: config.cluster.test_namespace.clone(), + tenant: config.cluster.tenant_name.clone(), + observer_namespace: "rustfs-fault-observers".to_string(), + observer_pod: "observer-worker-a".to_string(), + backend_specific_destructive_opt_in: true, + allowlist: HostStorageAllowlist { + nodes: vec!["worker-a".to_string()], + devices: vec!["/dev/mapper/rustfs-fault-dm".to_string()], + persistent_volumes: vec!["pv-a".to_string()], + }, + fault_table: Some("0 1024 flakey /dev/loop0 0 0 86400 1 drop_writes".to_string()), + }, + HostStorageTargetObservation { + node: "worker-a".to_string(), + node_uid: "node-uid-a".to_string(), + node_labels: BTreeMap::from([( + "kubernetes.io/hostname".to_string(), + "storage-host-a".to_string(), + )]), + pod: "p0".to_string(), + pod_uid: "u0".to_string(), + volume_name: "data".to_string(), + persistent_volume_claim: "data-rustfs-0".to_string(), + persistent_volume_claim_uid: "pvc-uid-0".to_string(), + persistent_volume_claim_phase: "Bound".to_string(), + persistent_volume: "pv-a".to_string(), + persistent_volume_uid: "pv-uid-a".to_string(), + persistent_volume_phase: "Bound".to_string(), + persistent_volume_claim_ref: HostStoragePersistentVolumeClaimRef { + namespace: "rustfs-fault-test".to_string(), + name: "data-rustfs-0".to_string(), + uid: "pvc-uid-0".to_string(), + }, + node_selector: HostStorageNodeSelector { + key: "kubernetes.io/hostname".to_string(), + operator: "In".to_string(), + values: vec!["storage-host-a".to_string()], + }, + container_mount_path: "/data/rustfs0".to_string(), + persistent_volume_path: "/data/rustfs-fault/dm-volume".to_string(), + mapper_name: "rustfs-fault-dm".to_string(), + logical_device: "/dev/mapper/rustfs-fault-dm".to_string(), + canonical_device: "/dev/dm-0".to_string(), + mount_source: "/dev/mapper/rustfs-fault-dm".to_string(), + mount_canonical_source: "/dev/dm-0".to_string(), + filesystem: "ext4".to_string(), + recovery_table: "0 1024 linear /dev/loop0 0".to_string(), + observed_at_ms: 7, + }, + ) + .expect("host proof"); + let recovery_snapshot = json!({ + "stage": "recovered", + "mapper_name": "rustfs-fault-dm", + "canonical_device": "/dev/dm-0", + "suspended": false, + "observed_at_ms": 57, + "helper_pod": crate::fault::host_storage::helper_pod_name(run_id), + "mapping": { + "node": "worker-a", + "node_uid": "node-uid-a", + "node_labels": {"kubernetes.io/hostname": "storage-host-a"}, + "pod": "p0", + "pod_uid": "u0", + "volume_name": "data", + "pvc": "data-rustfs-0", + "pvc_uid": "pvc-uid-0", + "pvc_phase": "Bound", + "pv": "pv-a", + "pv_uid": "pv-uid-a", + "pv_phase": "Bound", + "pv_claim_ref": { + "namespace": "rustfs-fault-test", + "name": "data-rustfs-0", + "uid": "pvc-uid-0" + }, + "node_selector": { + "key": "kubernetes.io/hostname", + "operator": "In", + "values": ["storage-host-a"] + }, + "container_mount_path": "/data/rustfs0", + "mount_path": "/data/rustfs-fault/dm-volume" + }, + "table": "0 1024 linear /dev/loop0 0", + "status": "0 1024 linear" + }); + + write_json(&case_dir, HOST_STORAGE_PROOF_ARTIFACT, &json!(host_proof)); + write_json( + &case_dir, + HOST_STORAGE_CLEANUP_ARTIFACT, + &json!({ + "schemaVersion":1,"scenario":scenario.name,"faultName":run_spec.faults[0].name,"runId":run_id, + "observedAtMs":58,"node":"worker-a","persistentVolume":"pv-a","mapperName":"rustfs-fault-dm", + "logicalDevice":"/dev/mapper/rustfs-fault-dm","canonicalDevice":"/dev/dm-0","mountCanonicalSource":"/dev/dm-0", + "filesystemMounted":true,"nodeQuarantined":false,"recoveryTableSha256":host_proof.target.recovery_table_sha256 + }), + ); + let mut active = recovery_snapshot.clone(); + active["stage"] = json!("active"); + active["observed_at_ms"] = json!(12); + active["table"] = json!(host_proof.tables.fault_table); + let pods = (0..4) + .map(|index| json!({"name":format!("p{index}"),"uid":format!("u{index}")})) + .collect::>(); + let mut after = pods.clone(); + after[0]["uid"] = json!("u0-new"); + write_json( + &case_dir, + "fault-evidence.json", + &json!({ + "scenario":scenario.name,"run_id":run_id,"backend":"device-mapper","target":"dedicated-block-device", + "injected":true,"active_during_workload":false,"recovered":true,"require_client_disruption":false,"client_disruptions":0, + "pods_before":pods,"pods_after":after,"active_snapshots":[{"stage":"active","resource_kind":"device-mapper","dm_status":active}], + "workload_snapshots":[],"fault_prepare_started_at_ms":6,"fault_apply_started_at_ms":11,"fault_active_at_ms":12, + "fault_delete_started_at_ms":50,"recovery_started_at_ms":50,"recovery_ended_at_ms":70,"dm_recovery_snapshot":recovery_snapshot + }), + ); + write_json( + &case_dir, + DM_FILESYSTEM_CHECK_ARTIFACT, + &json!({ + "schemaVersion":1,"scenario":scenario.name,"faultName":host_proof.fault_name,"runId":run_id, + "node":"worker-a","persistentVolume":"pv-a","mapperName":"rustfs-fault-dm","logicalDevice":"/dev/mapper/rustfs-fault-dm", + "canonicalDevice":"/dev/dm-0","mountPath":"/data/rustfs-fault/dm-volume","filesystem":"ext4","checker":"/usr/sbin/e2fsck", + "arguments":["-f","-n","/dev/mapper/rustfs-fault-dm"],"startedAtMs":52,"completedAtMs":54,"exitCode":0, + "stdout":"clean","stderr":"","clean":true,"mountedForRecovery":true,"unmountedForCheck":true,"remountedAfterCheck":true,"remountedAtMs":55 + }), + ); + let mount = json!({"source":"/dev/mapper/rustfs-fault-dm","canonical_source":"/dev/dm-0","filesystem":"ext4","options":"rw,relatime"}); + write_json( + &case_dir, + "dm-crash-boundary.json", + &json!({ + "scenario":scenario.name,"run_id":run_id,"started_at_ms":13,"completed_at_ms":20,"old_pod_uid":"u0", + "replacement_pod_uid":null,"filesystem_unmounted":true,"mapper_mounts_absent":true,"mount_before":mount, + "fault":{"table":host_proof.tables.fault_table} + }), + ); + write_json( + &case_dir, + "dm-crash-recovered.json", + &json!({ + "scenario":scenario.name,"run_id":run_id,"recovered_at_ms":59,"taint_removed":true,"mount":mount, + "expected_table":host_proof.tables.recovery_table,"fault":{"table":host_proof.tables.recovery_table} + }), + ); + let mut fixture = AckFailureFixture::new(AcknowledgedMutationKind::Put); + fixture.restore_healthy_observations(); + for record in &mut fixture.history { + record.run_id = Some(run_id.into()); + record.key = record + .key + .as_ref() + .map(|value| value.replace("run-1", run_id)); + if let Some(keys) = &mut record.listed_keys { + for key in keys { + *key = key.replace("run-1", run_id); + } + } + if let Some(versions) = &mut record.listed_versions { + for version in versions { + version.key = version.key.replace("run-1", run_id); + } + } + if record.kind != OperationKind::Put { + record.started_at_ms += 100; + record.ended_at_ms += 100; + } + } + fixture.ack.run_id = run_id.into(); + fixture.ack.trigger_key = fixture.ack.trigger_key.replace("run-1", run_id); + fixture.report.run_id = run_id.into(); + fixture.refresh_audit(); + fixture.report.passed = true; + write_json( + &case_dir, + "checker-pre-recommit-report.json", + &json!(fixture.report), + ); + let suffix = fixture.history[fixture.prefix_len..].to_vec(); + fixture.prefix_len = fixture.history.len(); + for mut record in suffix { + record.id.push_str("-final"); + record.started_at_ms += 30; + record.ended_at_ms += 30; + record.started_sequence = Some(fixture.history.len() as u64 * 2 + 1); + record.ended_sequence = Some(fixture.history.len() as u64 * 2 + 2); + fixture.history.push(record); + } + if relaxed { + fixture.fail_data_version("trigger-version", true); + } + fixture.refresh_audit(); + fixture.report.operation_cohorts.clear(); + fixture.report.fault_window_relations.clear(); + for record in &fixture.history[..fixture.prefix_len] { + if let Some(cohort) = record.durability_cohort { + *fixture + .report + .operation_cohorts + .entry(cohort.as_str().into()) + .or_default() += 1; + } + if let Some(relation) = record.fault_window_relation { + *fixture + .report + .fault_window_relations + .entry(relation.as_str().into()) + .or_default() += 1; + } + } + fixture.report.passed = !relaxed; + write_json(&case_dir, "checker-report.json", &json!(fixture.report)); + write_json( + &case_dir, + "ack-to-fault-evidence.json", + &json!({ + "scenario":scenario.name,"run_id":run_id,"trigger_operation_id":fixture.ack.trigger_operation_id, + "trigger_kind":"put","trigger_key":fixture.ack.trigger_key,"trigger_version_id":"trigger-version", + "trigger_acknowledged_at_ms":11,"fault_activated_at_ms":12,"ack_to_fault_ms":1,"max_ack_to_fault_ms":5, + "crash_boundary_started_at_ms":13,"crash_boundary_next_sequence":3,"ack_to_crash_boundary_ms":2 + }), + ); + fs::write( + case_dir.join("history.jsonl"), + fixture + .history + .iter() + .map(|record| serde_json::to_string(record).unwrap()) + .collect::>() + .join("\n"), + ) + .unwrap(); + let baseline: Value = read_json(&case_dir.join(RECOVERY_HEALTH_ARTIFACT)).unwrap(); + let mut events = vec![]; + for (at, stage, status) in [ + (1, "run", "started"), + (5, "recovery-health-baseline", "succeeded"), + (6, "fault-prepare", "started"), + (7, "host-storage-mutation-preflight", "succeeded"), + (9, "fault-prepare", "succeeded"), + (11, "fault-apply", "started"), + (12, "fault-apply", "succeeded"), + (12, "ack-trigger", "succeeded"), + (20, "crash-recovery-boundary", "succeeded"), + (70, "recovery-evidence", "succeeded"), + (71, "post-recovery-write", "started"), + (200, "post-recovery-write", "succeeded"), + (201, "checker-pre-recommit", "started"), + (220, "checker-pre-recommit", "succeeded"), + (230, "checker-final", "started"), + (250, "checker-final", "succeeded"), + ] { + let mut event = json!({"at_ms":at,"scenario":scenario.name,"run_id":run_id,"stage":stage,"status":status,"message":"fixture"}); + if stage == "recovery-health-baseline" { + event["details"] = baseline["baseline"].clone(); + } + events.push(event); + } + if relaxed { + events.push(json!({"at_ms":251,"scenario":scenario.name,"run_id":run_id,"stage":"checker-verdict","status":"failed","message":"loss"})); + } + events.push(json!({"at_ms":252,"scenario":scenario.name,"run_id":run_id,"stage":"run","status":if relaxed {"failed"} else {"succeeded"},"message":"fixture"})); + fs::write( + case_dir.join("run-events.jsonl"), + events + .iter() + .map(Value::to_string) + .collect::>() + .join("\n"), + ) + .unwrap(); + write_json( + &case_dir, + "ack-calibration-mode.json", + &json!({ + "scenario":scenario.name,"run_id":run_id,"mode":mode,"observed_at_ms":2,"bucket_response":{"bucket":"bucket","mode":mode}, + "pods":(0..4).map(|index|json!({"name":format!("p{index}"),"uid":format!("u{index}"),"image":config.cluster.rustfs_image, + "image_id":format!("containerd://sha256:{}","a".repeat(64)),"container_id":format!("containerd://process-{index}"), + "process_mode":mode,"new_bucket_mode":mode})).collect::>() + }), + ); + if relaxed { + let mut failure = serde_json::to_value( + FailureSummary::from_checker( + &scenario.name, + "checker-verdict", + RecoveryStabilityClassification::CommittedVersionMissing, + "loss", + ) + .with_run_id(run_id) + .with_case_name(scenario.case_name), + ) + .unwrap(); + failure["observed_at_ms"] = json!(251); + failure["primary_evidence_refs"] = json!( + [ + "checker-report.json", + "fault-evidence.json", + "run-events.jsonl" + ] + .map(|name| case_dir + .join(name) + .strip_prefix(&suite_root) + .unwrap() + .display() + .to_string()) + ); + write_json(&case_dir, "failure-summary.json", &failure); + } + write_json(&suite_root, "suite-plan.json", &json!(expansion.plan)); + write_json( + &suite_root, + "suite-summary.json", + &json!({"runId":expansion.plan.run_id,"status":"succeeded","attempts":[{ + "runId":run_id,"scenario":scenario.name,"status":if relaxed {"expected-failure"} else {"succeeded"},"startedAtMs":1,"endedAtMs":253 + }]}), + ); + suite_root + } + + #[test] + fn native_ack_calibration_pair_revalidates_positive_and_negative_controls() { + let dir = tempfile::tempdir().unwrap(); + let strict = write_native_ack_calibration_control(&dir.path().join("strict"), false); + let relaxed = write_native_ack_calibration_control(&dir.path().join("relaxed"), true); + use crate::fault::ack_calibration::validate_ack_calibration_pair; + validate_ack_calibration_pair(&strict, &relaxed).unwrap(); + let relocated = dir.path().join("relocated-strict"); + fs::rename(&strict, &relocated).unwrap(); + validate_ack_calibration_pair(&relocated, &relaxed).unwrap(); + let plan: Value = read_json(&relaxed.join("suite-plan.json")).unwrap(); + let case = std::path::PathBuf::from( + plan["attempts"][0]["artifacts"]["caseDir"] + .as_str() + .unwrap(), + ); + let path = case.join("ack-calibration-mode.json"); + let original: Value = read_json(&path).unwrap(); + for (pointer, value) in [ + ("/observed_at_ms", json!(0)), + ("/mode", json!("strict")), + ("/pods/0/container_id", json!("containerd://restarted")), + ( + "/pods/0/image_id", + json!(format!("containerd://sha256:{}", "b".repeat(64))), + ), + ("/bucket_response/mode", json!("strict")), + ] { + let mut invalid = original.clone(); + *invalid.pointer_mut(pointer).unwrap() = value; + write_json(&case, "ack-calibration-mode.json", &invalid); + assert!( + validate_ack_calibration_pair(&relocated, &relaxed).is_err(), + "{pointer}" + ); + } + write_json(&case, "ack-calibration-mode.json", &original); + let summary: Value = read_json(&relaxed.join("suite-summary.json")).unwrap(); + let mut invalid = summary.clone(); + invalid["attempts"][0]["status"] = json!("succeeded"); + write_json(&relaxed, "suite-summary.json", &invalid); + assert!(validate_ack_calibration_pair(&relocated, &relaxed).is_err()); + write_json(&relaxed, "suite-summary.json", &summary); + let mut invalid = plan.clone(); + invalid["attempts"][0]["scenario"] = json!("dm-drop-writes-after-ack-overwrite"); + write_json(&relaxed, "suite-plan.json", &invalid); + assert!(validate_ack_calibration_pair(&relocated, &relaxed).is_err()); + let mut invalid = plan.clone(); + invalid["attempts"][0]["artifacts"]["caseDir"] = json!(format!( + "{}/../escape", + plan["artifactRoot"].as_str().unwrap() + )); + write_json(&relaxed, "suite-plan.json", &invalid); + assert!(validate_ack_calibration_pair(&relocated, &relaxed).is_err()); + write_json(&relaxed, "suite-plan.json", &plan); + #[cfg(unix)] + { + let escaped = dir.path().join("escaped-case"); + fs::rename(&case, &escaped).unwrap(); + std::os::unix::fs::symlink(&escaped, &case).unwrap(); + assert!(validate_ack_calibration_pair(&relocated, &relaxed).is_err()); + fs::remove_file(&case).unwrap(); + fs::rename(&escaped, &case).unwrap(); + } + let failure: Value = read_json(&case.join("failure-summary.json")).unwrap(); + let mut unrelated = failure.clone(); + unrelated["classification"] = json!("availability_failure"); + write_json(&case, "failure-summary.json", &unrelated); + assert!(validate_ack_calibration_pair(&relocated, &relaxed).is_err()); + write_json(&case, "failure-summary.json", &failure); + validate_ack_calibration_pair(&relocated, &relaxed).unwrap(); + } + #[test] fn ack_expected_failure_accepts_direct_loss_for_all_trigger_types() { for kind in [ diff --git a/src/fault/runner/ack.rs b/src/fault/runner/ack.rs index 52b0459..74b1044 100644 --- a/src/fault/runner/ack.rs +++ b/src/fault/runner/ack.rs @@ -85,7 +85,14 @@ impl FaultRun<'_> { if let Some(pods) = calibration_pods { ensure!( pods.len() == target.pods_before.len() - && pods.iter().all(|pod| target.pods_before.contains(pod)), + && pods + .iter() + .all(|pod| target.target_proof.resolved_pods.iter().any( + |proven| proven.name == pod.name + && proven.uid == pod.uid + && proven.rustfs_container_id.as_deref() + == Some(pod.container_id.as_str()) + )), "calibration Pod identities changed before target proof" ); } @@ -140,7 +147,7 @@ impl FaultRun<'_> { async fn observe_ack_calibration( &self, endpoint: &str, - ) -> Result> { + ) -> Result> { use crate::fault::acknowledged_mutation::{ ACK_CALIBRATION_ARTIFACT, AckBucketMode, AckCalibrationEvidence, calibration_pod, }; @@ -148,6 +155,9 @@ impl FaultRun<'_> { .config .ack_calibration .context("calibration mode missing")?; + crate::fault::acknowledged_mutation::require_calibration_image( + &self.config.cluster.rustfs_image, + )?; let kubectl = crate::framework::kubectl::Kubectl::new(&self.config.cluster) .namespaced(&self.config.cluster.test_namespace); let selector = format!("rustfs.tenant={}", self.config.cluster.tenant_name); @@ -211,14 +221,7 @@ impl FaultRun<'_> { ACK_CALIBRATION_ARTIFACT, &serde_json::to_string_pretty(&evidence)?, )?; - Ok(evidence - .pods - .into_iter() - .map(|pod| crate::fault::reporting::PodIdentity { - name: pod.name, - uid: pod.uid, - }) - .collect()) + Ok(evidence.pods) } async fn prepare_quiet_mutation( diff --git a/src/fault/suite_plan.rs b/src/fault/suite_plan.rs index 513738f..82e3e15 100644 --- a/src/fault/suite_plan.rs +++ b/src/fault/suite_plan.rs @@ -767,6 +767,9 @@ fn scenario_config( config.qualify_planned_admin = false; config.ack_calibration = scenario.ack_calibration; if let Some(mode) = config.ack_calibration { + crate::fault::acknowledged_mutation::require_calibration_image( + &config.cluster.rustfs_image, + )?; mode.configure(&mut config.cluster.rustfs_env)?; } config.scenario = scenario.name.clone(); From e1f7335845eba7a32647cb10720de2c44e85a9ee Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Tue, 22 Sep 2026 23:30:55 +0800 Subject: [PATCH 3/4] test: authenticate complete ACK calibration artifact pairs --- src/fault/artifact_validation.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/fault/artifact_validation.rs b/src/fault/artifact_validation.rs index 1218779..487d938 100644 --- a/src/fault/artifact_validation.rs +++ b/src/fault/artifact_validation.rs @@ -18159,7 +18159,7 @@ mod tests { preflight["phases"] .as_array_mut() .unwrap() - .push(json!({"name":"host-storage-mutation-proof","status":"passed","checks":[]})); + .push(json!({"name":"host-storage-mutation-proof","status":"passed","checks":[{"name":"host_storage_proof","status":"passed","message":"dedicated target proven","responsibilityDomain":"harness"}]})); write_json(&case_dir, "preflight-summary.json", &preflight); let host_proof = HostStorageMutationProof::prove_device_mapper( HostStorageMutationIntent { @@ -18361,6 +18361,15 @@ mod tests { } if relaxed { fixture.fail_data_version("trigger-version", true); + fixture.report.delete_marker_lineage_incomplete = vec![format!( + "{}: ListObjectVersions has no unique latest entry", + fixture.ack.trigger_key + )]; + fixture.report.list_warnings = vec![format!( + "LIST prefix fault-test/{run_id}/ did not include expected live key {}", + fixture.ack.trigger_key + )]; + fixture.report.final_list_warning_count = 1; } fixture.refresh_audit(); fixture.report.operation_cohorts.clear(); @@ -18421,7 +18430,11 @@ mod tests { (201, "checker-pre-recommit", "started"), (220, "checker-pre-recommit", "succeeded"), (230, "checker-final", "started"), - (250, "checker-final", "succeeded"), + ( + 250, + "checker-final", + if relaxed { "failed" } else { "succeeded" }, + ), ] { let mut event = json!({"at_ms":at,"scenario":scenario.name,"run_id":run_id,"stage":stage,"status":status,"message":"fixture"}); if stage == "recovery-health-baseline" { @@ -18464,6 +18477,8 @@ mod tests { .with_case_name(scenario.case_name), ) .unwrap(); + failure["final_list_warning_count"] = json!(fixture.report.final_list_warning_count); + failure["list_warnings"] = json!(fixture.report.list_warnings); failure["observed_at_ms"] = json!(251); failure["primary_evidence_refs"] = json!( [ From 77f319f74ba4b3d7a6c258a2dfd25d3991dc3734 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Tue, 22 Sep 2026 23:47:18 +0800 Subject: [PATCH 4/4] docs: describe executable storage recovery entrypoints --- docs/DURABILITY_QUALIFICATION.md | 53 +++++++++++++++++++------------- 1 file changed, 32 insertions(+), 21 deletions(-) diff --git a/docs/DURABILITY_QUALIFICATION.md b/docs/DURABILITY_QUALIFICATION.md index 0f8fee5..939a18a 100644 --- a/docs/DURABILITY_QUALIFICATION.md +++ b/docs/DURABILITY_QUALIFICATION.md @@ -1,9 +1,7 @@ # Durability live qualification A calibration result belongs to one detector, candidate image digest, workload, -and storage layout. Catalog `gate-candidate` is not evidence of calibration. -Fresh-volume replacement and bitrot remain Planned until their supported -variants have separate live evidence. +and storage layout. ## ACK positive and negative controls @@ -65,7 +63,7 @@ and seed, ACK timing, recovery policy, EC geometry, cluster/storage class, and pre-crash filesystem/mount options. Host kernel writeback settings and image provenance must additionally be preserved in the operator's lab report; the analyzer does not attest those external settings. Its output is evidence for -this pair, not automatic catalog promotion or proof of physical power loss. +this pair. Repeat with individually reviewed single-attempt suites for overwrite, delete-marker, zero-byte PUT and multipart completion. Choose the exact expected @@ -75,21 +73,40 @@ other mutation types. ## Fresh-volume and bitrot variants -Use the existing supervised qualification entrypoint and its exact target JSON, -helper image, dedicated Local-PV configuration and identity checks. See -`make fault-qualify-list` and the matching manifests under `fault/planned/`. -Preflight requires an explicit context, namespace and Tenant; it must succeed -before any host mutation. Run only one of these commands per fresh prepared -fixture, preserving and analyzing its evidence before the next: +[PR #99](https://github.com/rustfs/s3chaos/pull/99) enables ordinary execution +of fresh-volume replacement and bitrot, including their four recovery variants. +It also provides the corresponding single-attempt suites under `fault/examples/`: +`fresh-volume-replacement-automatic.yaml`, +`fresh-volume-replacement-admin-deep.yaml`, `on-disk-bitrot.yaml`, and +`on-disk-bitrot-admin-deep.yaml`. + +Prepare the exact target JSON, helper image, dedicated Local-PV configuration, +context, namespace and Tenant before running. Set +`RUSTFS_FAULT_TEST_DESTRUCTIVE=1` and select one explicit recovery case: ```bash -make fault-qualify QUALIFICATION_CASE=fresh-volume-replacement-automatic-replacement -make fault-qualify QUALIFICATION_CASE=fresh-volume-replacement-admin-deep -make fault-qualify QUALIFICATION_CASE=on-disk-bitrot-automatic-scanner -make fault-qualify QUALIFICATION_CASE=on-disk-bitrot-admin-deep +RUSTFS_FAULT_TEST_STORAGE_RECOVERY_CASE=fresh-volume-replacement-automatic-replacement \ + make fault-run SCENARIO=fresh-volume-replacement +RUSTFS_FAULT_TEST_STORAGE_RECOVERY_CASE=fresh-volume-replacement-admin-deep \ + make fault-run SCENARIO=fresh-volume-replacement +RUSTFS_FAULT_TEST_STORAGE_RECOVERY_CASE=on-disk-bitrot-automatic-scanner \ + make fault-run SCENARIO=on-disk-bitrot +RUSTFS_FAULT_TEST_STORAGE_RECOVERY_CASE=on-disk-bitrot-admin-deep \ + make fault-run SCENARIO=on-disk-bitrot ``` -These are four separate supervised runs, not a batch script. After each run: +Run one command per prepared fixture. The existing +`make fault-qualify QUALIFICATION_CASE=` entrypoint remains available for +each of these four case names; `make fault-qualify-list` describes its contract. + +Validate an ordinary run using its exact emitted artifact root and scenario: + +```bash +cargo run --quiet --bin s3chaos -- fault-validate-artifacts \ + '' '' +``` + +For a run started through `fault-qualify`, use its qualification run root: ```bash make fault-qualify-analyze RUN_ROOT='' @@ -102,9 +119,3 @@ Bitrot also needs the actual mutation receipt and matching checksum detection within the corruption window. Missing detection, background-repair races, unsupported diagnostics and incomplete restore are unqualified or failed, never PASS. Stop on an unexpected failure and preserve live evidence before cleanup. - -The earlier external replacement versioned-marker PASS recorded in backlog -#2347 does not qualify its unversioned branch or both current K8s variants. Each -current variant requires its own complete run on the selected image. Promote a -Planned scenario only after all advertised variants have valid live receipts, -static gates and independent review; retain unresolved product defects.