diff --git a/crates/e2e_test/src/distributed/concurrent_data_movement_test.rs b/crates/e2e_test/src/distributed/concurrent_data_movement_test.rs index 730d84d07..66b30f5de 100644 --- a/crates/e2e_test/src/distributed/concurrent_data_movement_test.rs +++ b/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?; diff --git a/crates/e2e_test/src/distributed/harness.rs b/crates/e2e_test/src/distributed/harness.rs index f5cf72cce..f610beaa4 100644 --- a/crates/e2e_test/src/distributed/harness.rs +++ b/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 { + 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 { 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 { + 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 { 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!({ diff --git a/crates/e2e_test/src/distributed/s3_during_data_movement_test.rs b/crates/e2e_test/src/distributed/s3_during_data_movement_test.rs index 896b25924..66c835bbd 100644 --- a/crates/e2e_test/src/distributed/s3_during_data_movement_test.rs +++ b/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(())