test(scanner): wait for EC8+4 heal control readiness (#7540)

crates/data-usage/src/data_usage.rs

@@ -593,18 +593,6 @@ pub struct DataUsageSnapshotIdentity {
     pub scanner_epoch: Option<u64>,
 }
 
-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
-pub struct DataUsageSegmentInvalidationProof {
-    #[serde(default)]
-    pub process_epoch: String,
-    #[serde(default)]
-    pub generation_start: u64,
-    #[serde(default)]
-    pub generation_end: u64,
-    #[serde(default)]
-    pub producer_identity_coverage_complete: bool,
-}
-
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct DataUsageSnapshotSetState {
     pub pool_index: u64,
@@ -619,8 +607,6 @@ pub struct DataUsageSnapshotSetState {
     pub complete: bool,
     #[serde(default)]
     pub tombstone: bool,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub segment_invalidation_proof: Option<DataUsageSegmentInvalidationProof>,
 }
 
 impl DataUsageInfo {
@@ -3087,7 +3073,6 @@ mod tests {
             scan_plan_digest: Some([1; 32]),
             complete: false,
             tombstone: false,
-            segment_invalidation_proof: None,
         }];
         assert!(observed_data_usage_is_newer(&partial, &authoritative));
     }
@@ -3110,7 +3095,6 @@ mod tests {
                     scan_plan_digest: Some([1; 32]),
                     complete: true,
                     tombstone: false,
-                    segment_invalidation_proof: None,
                 },
                 DataUsageSnapshotSetState {
                     pool_index: 1,
@@ -3120,7 +3104,6 @@ mod tests {
                     scan_plan_digest: Some([2; 32]),
                     complete: false,
                     tombstone: false,
-                    segment_invalidation_proof: None,
                 },
             ],
             ..Default::default()
@@ -3130,41 +3113,6 @@ mod tests {
         assert!(partial.is_valid_partial_snapshot());
     }
 
-    #[test]
-    fn set_state_segment_invalidation_proof_is_additive() {
-        #[derive(Deserialize)]
-        struct LegacySetState {
-            pool_index: u64,
-            set_index: u64,
-            complete: bool,
-        }
-
-        let proof = DataUsageSegmentInvalidationProof {
-            process_epoch: "scanner-process".to_string(),
-            generation_start: 3,
-            generation_end: 5,
-            producer_identity_coverage_complete: true,
-        };
-        let state = DataUsageSnapshotSetState {
-            pool_index: 1,
-            set_index: 2,
-            scanner_cycle: Some(9),
-            scanner_epoch: Some(4),
-            scan_plan_digest: Some([7; 32]),
-            complete: true,
-            tombstone: false,
-            segment_invalidation_proof: Some(proof.clone()),
-        };
-        let encoded = rmp_serde::to_vec_named(&state).expect("set state should encode with additive proof");
-        let legacy: LegacySetState = rmp_serde::from_slice(&encoded).expect("legacy readers should ignore proof metadata");
-        assert_eq!(legacy.pool_index, 1);
-        assert_eq!(legacy.set_index, 2);
-        assert!(legacy.complete);
-
-        let decoded: DataUsageSnapshotSetState = rmp_serde::from_slice(&encoded).expect("new readers should restore proof");
-        assert_eq!(decoded.segment_invalidation_proof, Some(proof));
-    }
-
     #[test]
     fn completeness_marker_requires_a_snapshot_timestamp() {
         let untimestamped = DataUsageInfo {

crates/e2e_test/src/distributed/heal_test.rs

@@ -26,6 +26,7 @@ use std::collections::{BTreeMap, HashSet};
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::time::Duration;
+use tokio::time::{Instant, sleep};
 
 const EC84_NODE_COUNT: usize = 3;
 const EC84_DRIVES_PER_NODE: usize = 4;
@@ -33,6 +34,8 @@ const EC84_DATA_BLOCKS: usize = 8;
 const EC84_PARITY_BLOCKS: usize = 4;
 const EC84_TARGET_DRIVE_RESTART_CASE: &str = "ec84-target-drive-restart";
 const EC84_TARGET_DRIVE_RESTART_ORACLE: &str = "ec84-target-drive-restart.json";
+const EC84_HEAL_CONTROL_READY_TIMEOUT: Duration = Duration::from_secs(45);
+const EC84_HEAL_CONTROL_RETRY_DELAY: Duration = Duration::from_millis(250);
 
 #[derive(Clone)]
 struct ExpectedShard {
@@ -211,6 +214,29 @@ fn assert_replaced_drive_empty(drive: &Path, bucket: &str, keys: &[String]) -> T
     Ok(())
 }
 
+fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::Error + Send + Sync)) -> bool {
+    let message = error.to_string();
+    message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable")
+}
+
+async fn start_ec84_root_heal_when_control_ready(
+    heal_url: &str,
+    heal_body: &str,
+    access_key: &str,
+    secret_key: &str,
+) -> TestResult {
+    let deadline = Instant::now() + EC84_HEAL_CONTROL_READY_TIMEOUT;
+    loop {
+        match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await {
+            Ok(_) => return Ok(()),
+            Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && Instant::now() < deadline => {
+                sleep(EC84_HEAL_CONTROL_RETRY_DELAY).await;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
+
 async fn put_large_inventory(client: &Client, bucket: &str) -> TestResult<Vec<ExpectedShard>> {
     let mut expected = Vec::new();
     for index in 0..4 {
@@ -304,7 +330,7 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
     let heal_body =
         r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
     let heal_url = format!("{}/rustfs/admin/v3/heal/{bucket}?forceStart=true", dist.cluster.nodes[0].url);
-    signed_admin_post(&heal_url, Some(heal_body), &dist.cluster.access_key, &dist.cluster.secret_key).await?;
+    start_ec84_root_heal_when_control_ready(&heal_url, heal_body, &dist.cluster.access_key, &dist.cluster.secret_key).await?;
 
     wait_until(
         Duration::from_secs(120),
@@ -365,3 +391,23 @@ async fn three_node_four_drive_ec8_4_root_heal_rebuilds_replaced_drive_after_res
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn cluster_heal_coordination_retry_is_exact() {
+        let retryable: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 500 Internal Server Error cluster heal coordination unavailable".into();
+        assert!(is_cluster_heal_coordination_unavailable(retryable.as_ref()));
+
+        let other_internal: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 500 Internal Server Error unrelated".into();
+        assert!(!is_cluster_heal_coordination_unavailable(other_internal.as_ref()));
+
+        let wrong_status: Box<dyn std::error::Error + Send + Sync> =
+            "admin POST failed: 503 Service Unavailable cluster heal coordination unavailable".into();
+        assert!(!is_cluster_heal_coordination_unavailable(wrong_status.as_ref()));
+    }
+}

crates/e2e_test/src/on_demand_migration/interaction_test.rs

@@ -1137,7 +1137,12 @@ async fn test_odm_admin_config_is_redacted_and_status_counts_match_the_source()
         let miss = env.raw_get(bucket, miss_key).await?;
         assert_eq!(miss.status, 404, "{}", String::from_utf8_lossy(&miss.body));
     }
-    assert!(env.wait_local_listed(bucket, hit_key, SETTLE).await?);
+    let (listed, _, _) = tokio::try_join!(
+        env.wait_local_listed(bucket, hit_key, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_objects_total/inline", 1, SETTLE),
+        env.wait_for_status_counter(bucket, "/counters/pulled_bytes_total", body.len() as u64, SETTLE),
+    )?;
+    assert!(listed);
 
     let status = env.status_json(bucket).await?;
     assert_eq!(status.pointer("/configured").and_then(Value::as_bool), Some(true), "{status}");

crates/scanner/src/data_usage_define.rs

@@ -28,10 +28,10 @@ use metrics::{counter, describe_counter, describe_histogram, histogram};
 use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
-    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, DataUsageSegmentInvalidationProof, DataUsageSnapshotSetState,
-    LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary,
-    SizeReconciliationEntry, SizeReconciliationScope, SizeSummary, TierAccountingProof, TierStats, UNKNOWN_TIER,
-    UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP, UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP, UnknownTierStats, hash_path, prefix_usage_in_cache,
+    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, DataUsageSnapshotSetState, LEGACY_DATA_USAGE_OBJECT_NAME,
+    PrefixUsageEntry, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry,
+    SizeReconciliationScope, SizeSummary, TierAccountingProof, TierStats, UNKNOWN_TIER, UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP,
+    UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP, UnknownTierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_heal_contracts::heal_channel::HealScanMode;
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
@@ -564,6 +564,18 @@ impl DataUsageCacheSource {
 #[serde(transparent)]
 pub struct DataUsageScanPlanDigest(pub [u8; 32]);
 
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct DataUsageSegmentInvalidationProof {
+    #[serde(default)]
+    pub process_epoch: String,
+    #[serde(default)]
+    pub generation_start: u64,
+    #[serde(default)]
+    pub generation_end: u64,
+    #[serde(default)]
+    pub producer_identity_coverage_complete: bool,
+}
+
 #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum PendingScannerHealKind {

crates/scanner/src/scanner_io.rs

@@ -539,47 +539,6 @@ fn scanner_segment_reuse_activation_preflight_for_cycle(
     })
 }
 
-fn scanner_durable_segment_invalidation_evidence(
-    dirty_usage_snapshot: &DirtyUsageSnapshot,
-    results: &[DataUsageCache],
-    expected_sources: &HashSet<DataUsageCacheSource>,
-) -> DirtyUsageProducerEvidence {
-    let mut evidence = dirty_usage_producer_evidence(dirty_usage_snapshot);
-    if !evidence.generation_window_bound
-        || !evidence.producer_identity_coverage_complete
-        || !scanner_results_form_complete_snapshot(results, expected_sources)
-    {
-        return evidence;
-    }
-
-    let mut covered_sources = HashSet::with_capacity(expected_sources.len());
-    let all_sets_proved = results.iter().all(|result| {
-        let Some(source) = result.info.source else {
-            return false;
-        };
-        expected_sources.contains(&source)
-            && covered_sources.insert(source)
-            && scanner_segment_invalidation_proof_matches(result.info.segment_invalidation_proof.as_ref(), &evidence)
-    });
-    if all_sets_proved && covered_sources.len() == expected_sources.len() {
-        evidence.durable_producer_identity = true;
-        evidence.restart_gap_absent = true;
-    }
-    evidence
-}
-
-fn scanner_segment_invalidation_proof_matches(
-    proof: Option<&crate::DataUsageSegmentInvalidationProof>,
-    evidence: &DirtyUsageProducerEvidence,
-) -> bool {
-    proof.is_some_and(|proof| {
-        proof.process_epoch == scanner_activity_epoch()
-            && proof.generation_start == evidence.generation_start
-            && proof.generation_end == evidence.generation_end
-            && proof.producer_identity_coverage_complete
-    })
-}
-
 fn scanner_segment_reuse_activated() -> bool {
     scanner_segment_reuse_activation_preflight().scanner_segment_reuse_activated
 }

crates/scanner/src/scanner_io/cache.rs

@@ -464,7 +464,6 @@ pub(super) fn completed_usage_candidate(
                 scan_plan_digest: Some(result.info.scan_plan_digest?.0),
                 complete: true,
                 tombstone: false,
-                segment_invalidation_proof: result.info.segment_invalidation_proof.clone(),
             })
         })
         .collect::<Option<Vec<_>>>()?;
@@ -643,14 +642,13 @@ pub(super) fn observational_data_usage_info(
         let current_snapshot = current.is_some();
         let selected = current.or(lkg);
         if let Some(selected) = selected {
-            let (cycle, epoch, digest, last_update, complete, segment_invalidation_proof) = if current_snapshot {
+            let (cycle, epoch, digest, last_update, complete) = if current_snapshot {
                 (
                     Some(selected.info.next_cycle),
                     Some(selected.info.leader_epoch),
                     selected.info.scan_plan_digest.map(|digest| digest.0),
                     selected.info.last_update,
                     true,
-                    selected.info.segment_invalidation_proof.clone(),
                 )
             } else {
                 (
@@ -659,7 +657,6 @@ pub(super) fn observational_data_usage_info(
                     selected.info.lkg_scan_plan_digest.map(|digest| digest.0),
                     selected.info.lkg_last_update,
                     false,
-                    None,
                 )
             };
             set_states.push(DataUsageSnapshotSetState {
@@ -670,7 +667,6 @@ pub(super) fn observational_data_usage_info(
                 scan_plan_digest: digest,
                 complete,
                 tombstone: false,
-                segment_invalidation_proof,
             });
             usable.push((selected, last_update));
         } else {
@@ -682,7 +678,6 @@ pub(super) fn observational_data_usage_info(
                 scan_plan_digest: Some(expected_plan_digest.0),
                 complete: false,
                 tombstone: false,
-                segment_invalidation_proof: None,
             });
         }
     }

crates/scanner/src/scanner_io/io_cycle.rs

@@ -715,7 +715,7 @@ where
     );
     let segment_reuse_activation_preflight = scanner_segment_reuse_activation_preflight_for_cycle(
         &dirty_usage_snapshot,
-        scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources),
+        dirty_usage_producer_evidence(&dirty_usage_snapshot),
         distributed,
         distributed_segment_invalidation_evidence,
         cold_zero_walk_oracle,

crates/scanner/src/scanner_io/publish_gate_tests.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::*;
-use crate::data_usage_define::{DataUsageSegmentInvalidationProof, UNKNOWN_TIER, UnknownTierStats, hash_path};
+use crate::data_usage_define::{UNKNOWN_TIER, UnknownTierStats, hash_path};
 use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccountingProof};
 
 const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
@@ -176,26 +176,6 @@ fn completed_data_usage_info_rejects_duplicate_bucket_inventory() {
     assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
 }
 
-#[test]
-fn completed_data_usage_info_carries_segment_invalidation_proof_to_set_state() {
-    let source = DataUsageCacheSource::new(0, 0);
-    let proof = DataUsageSegmentInvalidationProof {
-        process_epoch: "scanner-process".to_string(),
-        generation_start: 5,
-        generation_end: 8,
-        producer_identity_coverage_complete: true,
-    };
-    let mut set = completed_root_cache("bucket", 2, 10, source);
-    set.info.segment_invalidation_proof = Some(proof.clone());
-
-    let (usage, _) =
-        completed_usage_for_scope(&[set], &HashSet::from([source]), &["bucket".to_string()], &[], true, false, false)
-            .expect("complete set should publish root usage");
-
-    assert_eq!(usage.usage_snapshot_set_states.len(), 1);
-    assert_eq!(usage.usage_snapshot_set_states[0].segment_invalidation_proof, Some(proof));
-}
-
 #[test]
 fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() {
     let buckets = vec!["bucket".to_string()];
@@ -370,7 +350,6 @@ fn set_membership_add_remove_uses_generation_and_tombstone() {
         scan_plan_digest: Some(TEST_PLAN_DIGEST.0),
         complete: false,
         tombstone: true,
-        segment_invalidation_proof: None,
     };
     let encoded = serde_json::to_vec(&state).expect("set state should serialize");
     let decoded: DataUsageSnapshotSetState = serde_json::from_slice(&encoded).expect("set state should deserialize");
@@ -392,7 +371,6 @@ fn set_membership_add_remove_uses_generation_and_tombstone() {
                 scan_plan_digest: Some(TEST_PLAN_DIGEST.0),
                 complete: true,
                 tombstone: false,
-                segment_invalidation_proof: None,
             },
             state,
         ],

crates/scanner/src/scanner_io/tests.rs

@@ -237,49 +237,6 @@ fn scanner_segment_reuse_activation_preflight_for_cycle_skips_distributed_blocke
     );
 }
 
-#[test]
-#[serial]
-fn scanner_durable_segment_invalidation_evidence_requires_matching_complete_set_proofs() {
-    use crate::segment_invalidation::SegmentInvalidationProducerIdentity;
-
-    clear_dirty_usage_buckets_for_tests();
-    record_dirty_usage_bucket_from_producers("photos", SegmentInvalidationProducerIdentity::REQUIRED_PRODUCTION);
-    let dirty_usage_snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation());
-    let process_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot)
-        .segment_invalidation_proof()
-        .expect("complete process-local producer coverage should produce proof metadata");
-    let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(0, 1)]);
-    let results = vec![
-        complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 0), process_proof.clone()),
-        complete_set_cache_with_segment_proof(DataUsageCacheSource::new(0, 1), process_proof.clone()),
-    ];
-
-    let durable_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources);
-
-    assert!(durable_evidence.producer_identity_coverage_complete);
-    assert!(durable_evidence.durable_producer_identity);
-    assert!(durable_evidence.restart_gap_absent);
-
-    let mut stale_epoch = results.clone();
-    stale_epoch[0]
-        .info
-        .segment_invalidation_proof
-        .as_mut()
-        .expect("proof fixture should exist")
-        .process_epoch = "stale-process".to_string();
-    let stale_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &stale_epoch, &expected_sources);
-    assert!(stale_evidence.producer_identity_coverage_complete);
-    assert!(!stale_evidence.durable_producer_identity);
-    assert!(!stale_evidence.restart_gap_absent);
-
-    record_dirty_usage_bucket("videos");
-    let changed_evidence = scanner_durable_segment_invalidation_evidence(&dirty_usage_snapshot, &results, &expected_sources);
-    assert!(!changed_evidence.producer_identity_coverage_complete);
-    assert!(!changed_evidence.durable_producer_identity);
-    assert!(!changed_evidence.restart_gap_absent);
-    clear_dirty_usage_buckets_for_tests();
-}
-
 #[test]
 fn scanner_cycle_result_returns_segment_reuse_activation_preflight() {
     let proof = ScannerSegmentReuseActivationProof {
@@ -318,26 +275,6 @@ fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence {
     }
 }
 
-fn complete_set_cache_with_segment_proof(
-    source: DataUsageCacheSource,
-    proof: crate::DataUsageSegmentInvalidationProof,
-) -> DataUsageCache {
-    DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: DATA_USAGE_ROOT.to_string(),
-            next_cycle: 7,
-            last_update: Some(SystemTime::UNIX_EPOCH),
-            leader_epoch: 11,
-            source: Some(source),
-            snapshot_complete: true,
-            scan_plan_digest: Some(DataUsageScanPlanDigest([3; 32])),
-            segment_invalidation_proof: Some(proof),
-            ..Default::default()
-        },
-        cache: HashMap::new(),
-    }
-}
-
 async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
     init_ecstore_config_for_scanner_tests();
     let temp_dir = tempfile::tempdir().expect("multi-pool scanner test directory should be created");
@@ -1686,7 +1623,6 @@ fn complete_usage_baseline(
             scan_plan_digest: Some(scan_plan_digest.0),
             complete: true,
             tombstone: false,
-            segment_invalidation_proof: None,
         }],
         ..Default::default()
     };