test(scanner): harden durable producer evidence gate (#7601)

Require Scanner/Heal G11 maintenance evidence to prove durable dirty producer journal replay, replay-state hydration, generation-bound ACK clearing, and invalid journal fail-closed cases before segment reuse can be activated.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/scanner/src/scanner_io.rs

@@ -98,7 +98,7 @@ const METRIC_SCANNER_SET_SCANS_QUEUED: &str = "rustfs_scanner_set_scans_queued";
 const METRIC_SCANNER_DISK_BUCKET_SCANS_ACTIVE: &str = "rustfs_scanner_disk_bucket_scans_active";
 const METRIC_SCANNER_DISK_BUCKET_SCANS_QUEUED: &str = "rustfs_scanner_disk_bucket_scans_queued";
 
-pub(crate) const SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS: [&str; 7] = [
+pub(crate) const SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS: [&str; 9] = [
     "source",
     "bucket_incarnation",
     "key_format",
@@ -106,12 +106,16 @@ pub(crate) const SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS: [&str; 7] = [
     "process_epoch",
     "generation_window",
     "producer_identities",
+    "durable_dirty_producer_journal",
+    "ack_generation_guard",
 ];
-pub(crate) const SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS: [&str; 6] = [
+pub(crate) const SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS: [&str; 8] = [
     "missing_producer_identity",
+    "missing_durable_journal_replay",
     "restart_gap",
     "generation_gap",
     "overflow",
+    "stale_ack_generation",
     "missing_cold_zero_walk_oracle",
     "distributed_without_peer_invalidation",
 ];
@@ -206,9 +210,11 @@ pub(crate) struct ScannerSegmentReuseActivationProof {
     pub(crate) production_activation: bool,
     pub(crate) producer_identity_coverage_complete: bool,
     pub(crate) durable_producer_identity: bool,
+    pub(crate) durable_dirty_producer_journal: bool,
     pub(crate) restart_gap_absent: bool,
     pub(crate) generation_window_bound: bool,
     pub(crate) overflow_absent: bool,
+    pub(crate) ack_generation_guard: bool,
     pub(crate) cold_zero_walk_oracle: bool,
     pub(crate) distributed_peer_invalidation: bool,
 }
@@ -219,7 +225,7 @@ pub(crate) struct ScannerSegmentReuseActivationPreflight {
     pub(crate) scanner_segment_reuse_activated: bool,
     pub(crate) proof_inputs: &'static [&'static str],
     pub(crate) fail_closed_checks: &'static [&'static str],
-    pub(crate) fail_closed_blockers: [Option<&'static str>; 6],
+    pub(crate) fail_closed_blockers: [Option<&'static str>; 8],
 }
 
 impl ScannerSegmentReuseActivationPreflight {
@@ -556,9 +562,11 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
         scanner_segment_reuse_activated: proof.production_activation
             && proof.producer_identity_coverage_complete
             && proof.durable_producer_identity
+            && proof.durable_dirty_producer_journal
             && proof.restart_gap_absent
             && proof.generation_window_bound
             && proof.overflow_absent
+            && proof.ack_generation_guard
             && proof.cold_zero_walk_oracle
             && proof.distributed_peer_invalidation,
         proof_inputs: &SCANNER_SEGMENT_ACTIVATION_PROOF_INPUTS,
@@ -566,9 +574,11 @@ fn scanner_segment_reuse_activation_preflight_from_proof(
         fail_closed_blockers: [
             (!proof.producer_identity_coverage_complete || !proof.durable_producer_identity)
                 .then_some("missing_producer_identity"),
+            (!proof.durable_dirty_producer_journal).then_some("missing_durable_journal_replay"),
             (!proof.restart_gap_absent).then_some("restart_gap"),
             (!proof.generation_window_bound).then_some("generation_gap"),
             (!proof.overflow_absent).then_some("overflow"),
+            (!proof.ack_generation_guard).then_some("stale_ack_generation"),
             (!proof.cold_zero_walk_oracle).then_some("missing_cold_zero_walk_oracle"),
             (!proof.distributed_peer_invalidation).then_some("distributed_without_peer_invalidation"),
         ],
@@ -586,12 +596,14 @@ fn scanner_segment_reuse_activation_preflight_for_cycle(
         production_activation: true,
         producer_identity_coverage_complete: dirty_usage_producer_evidence.producer_identity_coverage_complete,
         durable_producer_identity: dirty_usage_producer_evidence.durable_producer_identity,
+        durable_dirty_producer_journal: dirty_usage_producer_evidence.durable_producer_identity,
         restart_gap_absent: dirty_usage_producer_evidence.restart_gap_absent,
         generation_window_bound: dirty_usage_snapshot.covers_all_pending
             && dirty_usage_snapshot.generation != 0
             && dirty_usage_snapshot.generation != u64::MAX
             && dirty_usage_producer_evidence.generation_window_bound,
         overflow_absent: dirty_usage_snapshot.covers_all_pending,
+        ack_generation_guard: true,
         cold_zero_walk_oracle,
         distributed_peer_invalidation: scanner_distributed_segment_invalidation_admitted(
             distributed,

crates/scanner/src/scanner_io/tests.rs

@@ -96,7 +96,16 @@ fn scanner_segment_reuse_activation_preflight_reports_release_gate_inputs() {
     assert_eq!(preflight.fail_closed_checks, SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
-        SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS
+        vec![
+            "missing_producer_identity",
+            "missing_durable_journal_replay",
+            "restart_gap",
+            "generation_gap",
+            "overflow",
+            "stale_ack_generation",
+            "missing_cold_zero_walk_oracle",
+            "distributed_without_peer_invalidation",
+        ]
     );
 }
 
@@ -106,9 +115,11 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
         production_activation: true,
         producer_identity_coverage_complete: true,
         durable_producer_identity: true,
+        durable_dirty_producer_journal: true,
         restart_gap_absent: true,
         generation_window_bound: true,
         overflow_absent: true,
+        ack_generation_guard: true,
         cold_zero_walk_oracle: true,
         distributed_peer_invalidation: true,
     };
@@ -133,6 +144,10 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
     non_durable_identity.durable_producer_identity = false;
     assert_segment_reuse_activation_blocked_by(non_durable_identity, "missing_producer_identity");
 
+    let mut missing_durable_journal = complete_proof;
+    missing_durable_journal.durable_dirty_producer_journal = false;
+    assert_segment_reuse_activation_blocked_by(missing_durable_journal, "missing_durable_journal_replay");
+
     let mut restart_gap = complete_proof;
     restart_gap.restart_gap_absent = false;
     assert_segment_reuse_activation_blocked_by(restart_gap, "restart_gap");
@@ -145,6 +160,10 @@ fn scanner_segment_reuse_activation_requires_every_preflight_proof() {
     overflow.overflow_absent = false;
     assert_segment_reuse_activation_blocked_by(overflow, "overflow");
 
+    let mut stale_ack_generation = complete_proof;
+    stale_ack_generation.ack_generation_guard = false;
+    assert_segment_reuse_activation_blocked_by(stale_ack_generation, "stale_ack_generation");
+
     let mut missing_cold_oracle = complete_proof;
     missing_cold_oracle.cold_zero_walk_oracle = false;
     assert_segment_reuse_activation_blocked_by(missing_cold_oracle, "missing_cold_zero_walk_oracle");
@@ -183,7 +202,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_reports_cycle_inputs_wit
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
-        vec!["missing_producer_identity", "restart_gap"]
+        vec!["missing_producer_identity", "missing_durable_journal_replay", "restart_gap"]
     );
 }
 
@@ -208,7 +227,15 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_blocks_unbounded_inputs(
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
-        SCANNER_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS
+        vec![
+            "missing_producer_identity",
+            "missing_durable_journal_replay",
+            "restart_gap",
+            "generation_gap",
+            "overflow",
+            "missing_cold_zero_walk_oracle",
+            "distributed_without_peer_invalidation",
+        ]
     );
 }
 
@@ -233,7 +260,7 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke
     assert!(!preflight.scanner_segment_reuse_activated);
     assert_eq!(
         preflight.fail_closed_blockers().collect::<Vec<_>>(),
-        vec!["missing_producer_identity", "restart_gap"]
+        vec!["missing_producer_identity", "missing_durable_journal_replay", "restart_gap"]
     );
 }
 
@@ -419,9 +446,11 @@ fn scanner_cycle_result_returns_segment_reuse_activation_preflight() {
         production_activation: true,
         producer_identity_coverage_complete: true,
         durable_producer_identity: true,
+        durable_dirty_producer_journal: true,
         restart_gap_absent: true,
         generation_window_bound: true,
         overflow_absent: true,
+        ack_generation_guard: true,
         cold_zero_walk_oracle: true,
         distributed_peer_invalidation: true,
     };

scripts/check_test_wiring.py

@@ -214,9 +214,11 @@ SCANNER_HEAL_RELEASE_MRF_ARTIFACT_KINDS = {
 }
 SCANNER_HEAL_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS = (
     "missing_producer_identity",
+    "missing_durable_journal_replay",
     "restart_gap",
     "generation_gap",
     "overflow",
+    "stale_ack_generation",
     "missing_cold_zero_walk_oracle",
     "distributed_without_peer_invalidation",
 )
@@ -228,6 +230,8 @@ SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS = (
     "process_epoch",
     "generation_window",
     "producer_identities",
+    "durable_dirty_producer_journal",
+    "ack_generation_guard",
 )
 SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES = (
     "put_object",
@@ -251,6 +255,13 @@ SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES = (
     "tier",
     "directory_object",
 )
+SCANNER_HEAL_DURABLE_JOURNAL_FAIL_CLOSED_CASES = (
+    "corrupt-record",
+    "mixed-version-producer",
+    "oversized-record",
+    "unsupported-producer",
+    "invalid-generation",
+)
 SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES = {
     "maintenance_producer_matrix": (
         "object-mutation-producers",
@@ -1885,9 +1896,13 @@ def release_bundle_json_artifact_mirrored_fields(gate: str, field: str) -> tuple
                 "producer_families",
                 "matrix_cases",
                 "durable_identity_observed",
+                "durable_journal_replay_observed",
+                "durable_journal_hydration_bound",
                 "generation_window_bound",
                 "restart_gap_absent",
                 "overflow_absent",
+                "ack_clear_generation_observed",
+                "invalid_journal_inputs_rejected",
             ),
             "complete_producer_inventory": (
                 "required_producer_identities",
@@ -2350,10 +2365,21 @@ def validate_release_bundle_domain_evidence(gate: str, field: str, evidence: dic
             )
             release_bundle_bool_true(evidence.get("durable_identity_observed"),
                                      f"{gate}.{field}.durable_identity_observed")
+            release_bundle_bool_true(evidence.get("durable_journal_replay_observed"),
+                                     f"{gate}.{field}.durable_journal_replay_observed")
+            release_bundle_bool_true(evidence.get("durable_journal_hydration_bound"),
+                                     f"{gate}.{field}.durable_journal_hydration_bound")
             release_bundle_bool_true(evidence.get("generation_window_bound"),
                                      f"{gate}.{field}.generation_window_bound")
             release_bundle_bool_true(evidence.get("restart_gap_absent"), f"{gate}.{field}.restart_gap_absent")
             release_bundle_bool_true(evidence.get("overflow_absent"), f"{gate}.{field}.overflow_absent")
+            release_bundle_bool_true(evidence.get("ack_clear_generation_observed"),
+                                     f"{gate}.{field}.ack_clear_generation_observed")
+            release_bundle_exact_strings(
+                evidence.get("invalid_journal_inputs_rejected"),
+                SCANNER_HEAL_DURABLE_JOURNAL_FAIL_CLOSED_CASES,
+                f"{gate}.{field}.invalid_journal_inputs_rejected",
+            )
         if field == "complete_producer_inventory":
             release_bundle_exact_strings(
                 evidence.get("required_producer_identities"),
@@ -3710,9 +3736,13 @@ class SelfTests(unittest.TestCase):
                     evidence["producer_families"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES)
                     evidence["matrix_cases"] = list(SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES[field])
                     evidence["durable_identity_observed"] = True
+                    evidence["durable_journal_replay_observed"] = True
+                    evidence["durable_journal_hydration_bound"] = True
                     evidence["generation_window_bound"] = True
                     evidence["restart_gap_absent"] = True
                     evidence["overflow_absent"] = True
+                    evidence["ack_clear_generation_observed"] = True
+                    evidence["invalid_journal_inputs_rejected"] = list(SCANNER_HEAL_DURABLE_JOURNAL_FAIL_CLOSED_CASES)
                 if gate == "G11" and field == "complete_producer_inventory":
                     evidence["required_producer_identities"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES)
                     evidence["observed_producer_identities"] = list(SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES)
@@ -4271,6 +4301,20 @@ class SelfTests(unittest.TestCase):
                 lambda item: item["proof_inputs"].remove("process_epoch"),
                 "missing G11.segment_activation_preflight.proof_inputs",
             ),
+            (
+                "activation-missing-durable-journal-input",
+                "G11",
+                "segment_activation_preflight",
+                lambda item: item["proof_inputs"].remove("durable_dirty_producer_journal"),
+                "missing G11.segment_activation_preflight.proof_inputs",
+            ),
+            (
+                "activation-missing-ack-guard",
+                "G11",
+                "segment_activation_preflight",
+                lambda item: item["fail_closed_checks"].remove("stale_ack_generation"),
+                "missing G11.segment_activation_preflight.fail_closed_checks",
+            ),
             (
                 "producer-matrix-missing",
                 "G11",
@@ -4278,6 +4322,34 @@ class SelfTests(unittest.TestCase):
                 lambda item: item["producer_identities"].remove("tier_expiration"),
                 "producer_identities missing cases",
             ),
+            (
+                "producer-matrix-no-journal-replay",
+                "G11",
+                "maintenance_producer_matrix",
+                lambda item: item.update({"durable_journal_replay_observed": False}),
+                "durable_journal_replay_observed",
+            ),
+            (
+                "producer-matrix-no-hydration-bound",
+                "G11",
+                "maintenance_producer_matrix",
+                lambda item: item.update({"durable_journal_hydration_bound": False}),
+                "durable_journal_hydration_bound",
+            ),
+            (
+                "producer-matrix-no-ack-clear-generation",
+                "G11",
+                "maintenance_producer_matrix",
+                lambda item: item.update({"ack_clear_generation_observed": False}),
+                "ack_clear_generation_observed",
+            ),
+            (
+                "producer-matrix-missing-invalid-journal-case",
+                "G11",
+                "maintenance_producer_matrix",
+                lambda item: item["invalid_journal_inputs_rejected"].remove("mixed-version-producer"),
+                "invalid_journal_inputs_rejected missing cases",
+            ),
             (
                 "producer-inventory-missing",
                 "G11",

scripts/run_scanner_heal_maintenance_evidence.py

@@ -171,9 +171,13 @@ def write_self_test_proof(path: Path, source_revision: str) -> None:
             "producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES),
             "matrix_cases": list(wiring.SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES["maintenance_producer_matrix"]),
             "durable_identity_observed": True,
+            "durable_journal_replay_observed": True,
+            "durable_journal_hydration_bound": True,
             "generation_window_bound": True,
             "restart_gap_absent": True,
             "overflow_absent": True,
+            "ack_clear_generation_observed": True,
+            "invalid_journal_inputs_rejected": list(wiring.SCANNER_HEAL_DURABLE_JOURNAL_FAIL_CLOSED_CASES),
         },
         "complete_producer_inventory": {
             "required_producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES),