test(e2e): prove operations overlap data movement

crates/e2e_test/src/distributed/concurrent_data_movement_test.rs

@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use super::harness::{
-    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, payload_for, put_inventory_retrying,
-    retrying_get_equals, retrying_put, start_decommission, unique_bucket, wait_for_decommission_active,
-    wait_for_decommission_complete,
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, decommission_running_with_progress,
+    decommission_status_json, payload_for, put_inventory_retrying, retrying_get_equals, retrying_put, start_decommission,
+    unique_bucket, wait_for_decommission_complete, wait_for_decommission_running_with_progress,
 };
 use crate::common::init_logging;
 use std::sync::Arc;
@@ -33,10 +33,9 @@ async fn concurrent_puts_during_decommission_do_not_lose_baseline_or_new_objects
     dist.expand_to_four_pools().await?;
 
     start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
-    wait_for_decommission_active(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
 
     let clients = Arc::new(dist.clients()?);
-    let barrier = Arc::new(Barrier::new(16));
+    let barrier = Arc::new(Barrier::new(17));
     let mut handles = Vec::new();
     for idx in 0..16 {
         let clients = clients.clone();
@@ -52,10 +51,17 @@ async fn concurrent_puts_during_decommission_do_not_lose_baseline_or_new_objects
         }));
     }
 
+    wait_for_decommission_running_with_progress(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
+    barrier.wait().await;
+
     let mut live_objects = Vec::new();
     for handle in handles {
         live_objects.push(handle.await??);
     }
+    let status = decommission_status_json(&dist.cluster).await?;
+    if !decommission_running_with_progress(&status, DECOMMISSION_POOL_ID)? {
+        return Err(format!("decommission did not remain active across concurrent PUTs: {status}").into());
+    }
 
     wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(180)).await?;
 

crates/e2e_test/src/distributed/harness.rs

@@ -723,6 +723,21 @@ pub(crate) fn decommission_active(status: &serde_json::Value, pool_id: usize) ->
     Ok(queued || status_text.eq_ignore_ascii_case("running") || pool_status.eq_ignore_ascii_case("decommissioning"))
 }
 
+pub(crate) fn decommission_running_with_progress(status: &serde_json::Value, pool_id: usize) -> TestResult<bool> {
+    let pool = pool_entry(status, pool_id).ok_or_else(|| format!("pool {pool_id} missing from decommission status: {status}"))?;
+    if let Some(reason) = decommission_failure(pool) {
+        return Err(format!("{reason}: {pool}").into());
+    }
+    let info = pool
+        .get("decommissionInfo")
+        .ok_or_else(|| format!("pool {pool_id} has no decommissionInfo: {pool}"))?;
+    let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
+    let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
+    let running = status_text.eq_ignore_ascii_case("running") || pool_status.eq_ignore_ascii_case("decommissioning");
+    let progressed = nonzero_u64(info.get("objectsDecommissioned")) || nonzero_u64(info.get("bytesDecommissioned"));
+    Ok(running && progressed)
+}
+
 pub(crate) fn decommission_complete(status: &serde_json::Value, pool_id: usize) -> TestResult<bool> {
     let pool = pool_entry(status, pool_id).ok_or_else(|| format!("pool {pool_id} missing from decommission status: {status}"))?;
     if let Some(reason) = decommission_failure(pool) {
@@ -735,7 +750,7 @@ pub(crate) fn decommission_complete(status: &serde_json::Value, pool_id: usize)
     let status_text = pool.get("status").and_then(serde_json::Value::as_str).unwrap_or("");
     let pool_status = pool.get("poolStatus").and_then(serde_json::Value::as_str).unwrap_or("");
     let terminal = status_text.eq_ignore_ascii_case("complete") && pool_status.eq_ignore_ascii_case("decommissioned");
-    let moved_data = nonzero_u64(info.get("objectsDecommissioned")) || nonzero_u64(info.get("bytesDecommissioned"));
+    let moved_data = nonzero_u64(info.get("objectsDecommissioned")) && nonzero_u64(info.get("bytesDecommissioned"));
     Ok(complete && terminal && moved_data)
 }
 
@@ -747,6 +762,27 @@ pub(crate) async fn wait_for_decommission_active(
     wait_for_decommission_state(cluster, pool_id, timeout, "active", decommission_active).await
 }
 
+pub(crate) async fn wait_for_decommission_running_with_progress(
+    cluster: &RustFSTestClusterEnvironment,
+    pool_id: usize,
+    timeout: Duration,
+) -> TestResult {
+    let deadline = Instant::now() + timeout;
+    loop {
+        let status = decommission_status_json(cluster).await?;
+        if decommission_running_with_progress(&status, pool_id)? {
+            return Ok(());
+        }
+        if Instant::now() >= deadline {
+            return Err(format!(
+                "decommission did not become active with non-zero progress within {timeout:?}; last status: {status}"
+            )
+            .into());
+        }
+        sleep(Duration::from_millis(100)).await;
+    }
+}
+
 pub(crate) async fn wait_for_decommission_complete(
     cluster: &RustFSTestClusterEnvironment,
     pool_id: usize,
@@ -866,6 +902,20 @@ pub(crate) fn rebalance_active(status: &serde_json::Value, expected_id: &str) ->
     }))
 }
 
+pub(crate) fn rebalance_running_with_progress(status: &serde_json::Value, expected_id: &str) -> TestResult<bool> {
+    Ok(validate_rebalance_status(status, expected_id)?.iter().any(|pool| {
+        let started = pool
+            .get("status")
+            .and_then(serde_json::Value::as_str)
+            .is_some_and(|value| value.eq_ignore_ascii_case("started"));
+        let progress = pool.get("progress");
+        started
+            && (nonzero_u64(progress.and_then(|value| value.get("objects")))
+                || nonzero_u64(progress.and_then(|value| value.get("versions")))
+                || nonzero_u64(progress.and_then(|value| value.get("bytes"))))
+    }))
+}
+
 pub(crate) fn rebalance_complete(status: &serde_json::Value, expected_id: &str) -> TestResult<bool> {
     let pools = validate_rebalance_status(status, expected_id)?;
     let completed: Vec<&serde_json::Value> = pools
@@ -908,6 +958,27 @@ pub(crate) async fn wait_for_rebalance_active(
     }
 }
 
+pub(crate) async fn wait_for_rebalance_running_with_progress(
+    cluster: &RustFSTestClusterEnvironment,
+    expected_id: &str,
+    timeout: Duration,
+) -> TestResult {
+    let deadline = Instant::now() + timeout;
+    loop {
+        let status = rebalance_status_json(cluster).await?;
+        if rebalance_running_with_progress(&status, expected_id)? {
+            return Ok(());
+        }
+        if Instant::now() >= deadline {
+            return Err(format!(
+                "rebalance did not become active with non-zero progress within {timeout:?}; last status: {status}"
+            )
+            .into());
+        }
+        sleep(Duration::from_millis(100)).await;
+    }
+}
+
 pub(crate) async fn wait_for_rebalance_complete(
     cluster: &RustFSTestClusterEnvironment,
     expected_id: &str,
@@ -1072,8 +1143,45 @@ fn decommission_complete_requires_terminal_status_and_clean_counters() {
             { "id": 1, "status": "none", "poolStatus": "active" }
         ]
     });
-    assert!(decommission_complete(&status, 0).unwrap());
+    assert!(decommission_complete(&status, 0).expect("complete fixture should be accepted"));
     assert!(decommission_complete(&status, 1).is_err());
+
+    for missing_counter in ["objectsDecommissioned", "bytesDecommissioned"] {
+        let mut one_sided = status.clone();
+        one_sided["pools"][0]["decommissionInfo"][missing_counter] = serde_json::json!(0);
+        assert!(
+            !decommission_complete(&one_sided, 0).expect("one-sided progress fixture should be readable"),
+            "completion must require both movement counters; zeroed {missing_counter}"
+        );
+    }
+}
+
+#[test]
+fn decommission_overlap_requires_running_state_and_progress() {
+    let mut status = serde_json::json!({
+        "pools": [{
+            "id": 0,
+            "status": "queued",
+            "poolStatus": "active",
+            "decommissionInfo": {
+                "queued": true,
+                "objectsDecommissioned": 1,
+                "bytesDecommissioned": 1024
+            }
+        }]
+    });
+    assert!(
+        !decommission_running_with_progress(&status, 0).expect("queued fixture should be readable"),
+        "queued work is not temporal overlap"
+    );
+    status["pools"][0]["status"] = serde_json::json!("running");
+    assert!(decommission_running_with_progress(&status, 0).expect("running fixture should be readable"));
+    status["pools"][0]["decommissionInfo"]["objectsDecommissioned"] = serde_json::json!(0);
+    status["pools"][0]["decommissionInfo"]["bytesDecommissioned"] = serde_json::json!(0);
+    assert!(
+        !decommission_running_with_progress(&status, 0).expect("zero-progress fixture should be readable"),
+        "running state alone does not prove movement started"
+    );
 }
 
 #[test]
@@ -1084,6 +1192,25 @@ fn rebalance_active_treats_started_as_in_progress() {
     assert!(!rebalance_active(&done, "run-1").unwrap());
 }
 
+#[test]
+fn rebalance_overlap_requires_started_state_and_progress() {
+    let mut status = serde_json::json!({
+        "id": "run-1",
+        "pools": [{ "id": 0, "status": "Started", "stopping": false, "progress": { "objects": 0, "bytes": 0 } }]
+    });
+    assert!(
+        !rebalance_running_with_progress(&status, "run-1").expect("zero-progress fixture should be readable"),
+        "started state alone does not prove movement"
+    );
+    status["pools"][0]["progress"]["objects"] = serde_json::json!(1);
+    assert!(rebalance_running_with_progress(&status, "run-1").expect("progress fixture should be readable"));
+    status["pools"][0]["status"] = serde_json::json!("Completed");
+    assert!(
+        !rebalance_running_with_progress(&status, "run-1").expect("completed fixture should be readable"),
+        "completed movement is not temporal overlap"
+    );
+}
+
 #[test]
 fn rebalance_complete_accepts_non_participating_pools_but_requires_progress() {
     let completed = serde_json::json!({

crates/e2e_test/src/distributed/s3_during_data_movement_test.rs

@@ -13,9 +13,10 @@
 // limitations under the License.
 
 use super::harness::{
-    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, put_inventory_retrying, retrying_get_equals,
-    retrying_put, start_decommission, start_rebalance, unique_bucket, wait_for_decommission_active,
-    wait_for_decommission_complete, wait_for_rebalance_active, wait_for_rebalance_complete,
+    DECOMMISSION_POOL_ID, DistCluster, DistLayout, TestResult, assert_inventory, decommission_running_with_progress,
+    decommission_status_json, put_inventory_retrying, rebalance_running_with_progress, rebalance_status_json,
+    retrying_get_equals, retrying_put, start_decommission, start_rebalance, unique_bucket, wait_for_decommission_complete,
+    wait_for_decommission_running_with_progress, wait_for_rebalance_complete, wait_for_rebalance_running_with_progress,
 };
 use crate::common::init_logging;
 use std::time::Duration;
@@ -31,7 +32,7 @@ async fn s3_put_get_list_succeed_during_decommission_and_rebalance() -> TestResu
     dist.expand_to_four_pools().await?;
 
     start_decommission(&dist.cluster, DECOMMISSION_POOL_ID).await?;
-    wait_for_decommission_active(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
+    wait_for_decommission_running_with_progress(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(30)).await?;
     let live = dist.client(2)?;
     retrying_put(
         &live,
@@ -57,12 +58,16 @@ async fn s3_put_get_list_succeed_during_decommission_and_rebalance() -> TestResu
             .any(|object| object.key() == Some("during-decommission.bin")),
         "list during decommission missed the newly written key"
     );
+    let status = decommission_status_json(&dist.cluster).await?;
+    if !decommission_running_with_progress(&status, DECOMMISSION_POOL_ID)? {
+        return Err(format!("decommission did not remain active across the S3 operations: {status}").into());
+    }
 
     wait_for_decommission_complete(&dist.cluster, DECOMMISSION_POOL_ID, Duration::from_secs(180)).await?;
     assert_inventory(&live, &bucket, &inventory).await?;
 
     let rebalance_id = start_rebalance(&dist.cluster).await?;
-    wait_for_rebalance_active(&dist.cluster, &rebalance_id, Duration::from_secs(30)).await?;
+    wait_for_rebalance_running_with_progress(&dist.cluster, &rebalance_id, Duration::from_secs(30)).await?;
     retrying_put(
         &live,
         &bucket,
@@ -79,6 +84,10 @@ async fn s3_put_get_list_succeed_during_decommission_and_rebalance() -> TestResu
         Duration::from_secs(30),
     )
     .await?;
+    let status = rebalance_status_json(&dist.cluster).await?;
+    if !rebalance_running_with_progress(&status, &rebalance_id)? {
+        return Err(format!("rebalance did not remain active across the S3 operations: {status}").into());
+    }
     wait_for_rebalance_complete(&dist.cluster, &rebalance_id, Duration::from_secs(180)).await?;
     assert_inventory(&dist.client(1)?, &bucket, &inventory).await?;
     Ok(())