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

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

@@ -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 {
@@ -657,6 +669,11 @@ pub struct DataUsageCacheInfo {
     /// structural plan remains reusable across ordinary bucket writes.
     #[serde(default)]
     pub scan_execution_digest: Option<DataUsageScanPlanDigest>,
+    /// Process-epoch and generation window that produced a complete set cache
+    /// with all known segment invalidation producers wired. This proof is
+    /// additive compatibility metadata; absence keeps segment reuse disabled.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub segment_invalidation_proof: Option<DataUsageSegmentInvalidationProof>,
     /// Durable bucket incarnations captured for a complete set aggregate.
     /// Missing or nil entries are legacy/unproven and cannot authorize
     /// skipping an unselected bucket in a later scoped set scan.
@@ -686,6 +703,7 @@ impl Serialize for DataUsageCacheInfo {
             + usize::from(self.lkg_leader_epoch.is_some())
             + usize::from(self.lkg_scan_plan_digest.is_some())
             + usize::from(self.scan_execution_digest.is_some())
+            + usize::from(self.segment_invalidation_proof.is_some())
             + usize::from(!self.scan_bucket_incarnations.is_empty());
         let mut state = serializer.serialize_map(Some(field_count))?;
         state.serialize_entry("name", &self.name)?;
@@ -746,6 +764,9 @@ impl Serialize for DataUsageCacheInfo {
         if let Some(scan_execution_digest) = self.scan_execution_digest {
             state.serialize_entry("scan_execution_digest", &scan_execution_digest)?;
         }
+        if let Some(proof) = &self.segment_invalidation_proof {
+            state.serialize_entry("segment_invalidation_proof", proof)?;
+        }
         if !self.scan_bucket_incarnations.is_empty() {
             state.serialize_entry("scan_bucket_incarnations", &self.scan_bucket_incarnations)?;
         }

crates/scanner/src/data_usage_define/tests.rs

@@ -1095,6 +1095,7 @@ fn test_data_usage_cache_info_deserialize_defaults_scan_resume_after() {
     assert!(!decoded.snapshot_complete);
     assert!(decoded.scan_plan_digest.is_none());
     assert!(decoded.scan_execution_digest.is_none());
+    assert!(decoded.segment_invalidation_proof.is_none());
     assert_eq!(decoded.cache_key_format, 0);
 }
 
@@ -1183,6 +1184,12 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
             snapshot_complete: true,
             scan_plan_digest: Some(TEST_PLAN_DIGEST),
             scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])),
+            segment_invalidation_proof: Some(DataUsageSegmentInvalidationProof {
+                process_epoch: "scanner-process".to_string(),
+                generation_start: 7,
+                generation_end: 9,
+                producer_identity_coverage_complete: true,
+            }),
             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
             ..Default::default()
         },
@@ -1212,6 +1219,15 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
     assert!(current.info.snapshot_complete);
     assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
     assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32])));
+    assert_eq!(
+        current.info.segment_invalidation_proof,
+        Some(DataUsageSegmentInvalidationProof {
+            process_epoch: "scanner-process".to_string(),
+            generation_start: 7,
+            generation_end: 9,
+            producer_identity_coverage_complete: true,
+        })
+    );
     assert_eq!(current.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT);
     assert_eq!(current.find("bucket").map(|entry| entry.objects), Some(3));
 

crates/scanner/src/scanner_io.rs

@@ -603,6 +603,7 @@ pub struct ScannerBucketScanPlan {
     pending_maintenance_work: Arc<AtomicBool>,
     cache_cycle_floor: Arc<AtomicU64>,
     cold_zero_walk_reuse_observed: Arc<AtomicBool>,
+    segment_invalidation_proof: Option<crate::DataUsageSegmentInvalidationProof>,
 }
 
 #[derive(Clone, Default)]

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -73,6 +73,21 @@ pub(super) struct DirtyUsageProducerEvidence {
     pub(super) durable_producer_identity: bool,
     pub(super) restart_gap_absent: bool,
     pub(super) generation_window_bound: bool,
+    pub(super) generation_start: u64,
+    pub(super) generation_end: u64,
+}
+
+impl DirtyUsageProducerEvidence {
+    pub(super) fn segment_invalidation_proof(self) -> Option<crate::DataUsageSegmentInvalidationProof> {
+        (self.generation_window_bound && self.producer_identity_coverage_complete).then(|| {
+            crate::DataUsageSegmentInvalidationProof {
+                process_epoch: scanner_activity_epoch().to_string(),
+                generation_start: self.generation_start,
+                generation_end: self.generation_end,
+                producer_identity_coverage_complete: true,
+            }
+        })
+    }
 }
 
 /// A point-in-time view of the local dirty bucket generations.
@@ -727,6 +742,8 @@ pub(super) fn dirty_usage_producer_evidence(snapshot: &DirtyUsageSnapshot) -> Di
         durable_producer_identity: false,
         restart_gap_absent: false,
         generation_window_bound,
+        generation_start: snapshot.generation,
+        generation_end: snapshot.generation,
     }
 }
 

crates/scanner/src/scanner_io/io_cache.rs

@@ -173,6 +173,7 @@ impl ScannerIOCache for SetDisks {
             pending_maintenance_work,
             cache_cycle_floor,
             cold_zero_walk_reuse_observed,
+            segment_invalidation_proof,
         } = scan_plan;
         let scan_plan_digest = scanner_bucket_work_digest(scan_plan_digest, scan_mode, requires_full_scan);
         let bucket_work_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan);
@@ -243,6 +244,7 @@ impl ScannerIOCache for SetDisks {
                             scan_plan_digest: Some(scan_plan_digest),
                             scan_coverage_digest: Some(bucket_coverage_digest),
                             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                            segment_invalidation_proof: segment_invalidation_proof.clone(),
                             scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                             ..Default::default()
                         },
@@ -258,6 +260,7 @@ impl ScannerIOCache for SetDisks {
             cache.info.last_update = Some(now);
             cache.info.snapshot_complete = true;
             cache.info.scan_execution_digest = Some(execution_digest);
+            cache.info.segment_invalidation_proof = segment_invalidation_proof.clone();
             cache.info.lkg_snapshot_complete = false;
             cache.info.lkg_next_cycle = None;
             cache.info.lkg_last_update = None;
@@ -539,6 +542,7 @@ impl ScannerIOCache for SetDisks {
                     lkg_last_update: old_cache.info.lkg_last_update,
                     lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
                     lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
+                    segment_invalidation_proof: None,
                     scan_bucket_incarnations: current_bucket_incarnations.clone().unwrap_or_default(),
                     ..Default::default()
                 },
@@ -1465,6 +1469,7 @@ impl ScannerIOCache for SetDisks {
                 cache.info.last_update.get_or_insert_with(SystemTime::now);
                 cache.info.snapshot_complete = true;
                 cache.info.scan_execution_digest = Some(execution_digest);
+                cache.info.segment_invalidation_proof = segment_invalidation_proof.clone();
                 cache.info.lkg_snapshot_complete = false;
                 cache.info.lkg_next_cycle = None;
                 cache.info.lkg_last_update = None;
@@ -1493,6 +1498,7 @@ impl ScannerIOCache for SetDisks {
             incomplete_scope.info.tier_registry_generation = Some(tier_registry_generation);
             incomplete_scope.info.source = Some(source);
             incomplete_scope.info.snapshot_complete = false;
+            incomplete_scope.info.segment_invalidation_proof = None;
             incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
             incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
             if let Err(e) = updates.send(incomplete_scope).await {

crates/scanner/src/scanner_io/io_cycle.rs

@@ -411,6 +411,7 @@ where
     let remote_dirty_usage_acknowledgements = scope_resolution.remote_dirty_usage_acknowledgements;
     let distributed_segment_invalidation_evidence = scope_resolution.distributed_segment_invalidation_evidence;
     let scan_scope = scope_resolution.scope;
+    let segment_invalidation_proof = dirty_usage_producer_evidence(&dirty_usage_snapshot).segment_invalidation_proof();
     #[cfg(test)]
     if let Some(observer) = resolved_scope_observer {
         let _ = observer.send(scan_scope.clone());
@@ -600,6 +601,7 @@ where
             pending_maintenance_work: pending_maintenance_work.clone(),
             cache_cycle_floor: cache_cycle_floor.clone(),
             cold_zero_walk_reuse_observed: cold_zero_walk_reuse_observed.clone(),
+            segment_invalidation_proof: segment_invalidation_proof.clone(),
         };
         // Spawn task to run the scanner
         let scanner_fut = tokio::spawn(async move {

crates/scanner/src/scanner_io/tests.rs

@@ -270,6 +270,8 @@ fn complete_process_local_producer_evidence() -> DirtyUsageProducerEvidence {
         durable_producer_identity: false,
         restart_gap_absent: false,
         generation_window_bound: true,
+        generation_start: 7,
+        generation_end: 7,
     }
 }
 
@@ -1557,6 +1559,12 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
 
     let ctx = CancellationToken::new();
     let empty_execution = DataUsageScanPlanDigest([5; 32]);
+    let segment_invalidation_proof = crate::DataUsageSegmentInvalidationProof {
+        process_epoch: scanner_activity_epoch().to_string(),
+        generation_start: 8,
+        generation_end: 8,
+        producer_identity_coverage_complete: true,
+    };
     set.nsscanner_cache(
         ctx.clone(),
         ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()),
@@ -1577,6 +1585,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
             pending_maintenance_work: Arc::new(AtomicBool::new(false)),
             cache_cycle_floor: Arc::new(AtomicU64::new(8)),
             cold_zero_walk_reuse_observed: Arc::new(AtomicBool::new(false)),
+            segment_invalidation_proof: Some(segment_invalidation_proof.clone()),
         },
         tx,
         8,
@@ -1586,6 +1595,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers
     .expect("empty set scope should replace its prior nonempty cache");
     let empty = rx.try_recv().expect("empty set snapshot should be published");
     assert_eq!(empty.info.scan_execution_digest, Some(empty_execution));
+    assert_eq!(empty.info.segment_invalidation_proof, Some(segment_invalidation_proof));
     assert!(empty.info.snapshot_complete);
     let root = empty.checked_flatten(DATA_USAGE_ROOT).expect("complete empty root");
     assert_eq!((root.size, root.objects), (0, 0));