test(e2e): accept retried rebalance deferrals; retry busy decom preflight (#8813)

test(e2e): accept retried rebalance deferrals and retry busy decommission preflight

.config/e2e-distributed-selection.txt

@@ -1,2 +1,2 @@
-sha256-linux=15fc5a29a558764622b02b1e7bc33d4b7aeb7de1a1a440dc110e207d56bebe49
-sha256-darwin=15fc5a29a558764622b02b1e7bc33d4b7aeb7de1a1a440dc110e207d56bebe49
+sha256-linux=e31182a075f38b8a6680f4563ed674f0cb92d7ad9e764d3da667c99510656325
+sha256-darwin=e31182a075f38b8a6680f4563ed674f0cb92d7ad9e764d3da667c99510656325

crates/e2e_test/src/distributed/harness.rs

@@ -976,6 +976,21 @@ pub(crate) async fn rebalance_status_json(cluster: &RustFSTestClusterEnvironment
     Ok(serde_json::from_str(&body)?)
 }
 
+/// Mirrors `REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX` in `crates/ecstore/src/services/rebalance`.
+const REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX: &str = "deferred transient rebalance entry failure:";
+
+/// A transient entry failure (for example a write quorum miss on a busy
+/// target) is recorded as `lastError` and retried at the end of the run; the
+/// worker clears it when the pool completes. It is only acceptable while the
+/// pool is still started: a terminal pool keeping it is a real failure.
+fn is_retrying_deferred_rebalance_error(pool: &serde_json::Value, error: &str) -> bool {
+    error.starts_with(REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX)
+        && pool
+            .get("status")
+            .and_then(serde_json::Value::as_str)
+            .is_some_and(|status| status.eq_ignore_ascii_case("started"))
+}
+
 fn validate_rebalance_status<'a>(status: &'a serde_json::Value, expected_id: &str) -> TestResult<&'a [serde_json::Value]> {
     let id = status
         .get("id")
@@ -1003,7 +1018,7 @@ fn validate_rebalance_status<'a>(status: &'a serde_json::Value, expected_id: &st
         if pool
             .get("lastError")
             .and_then(serde_json::Value::as_str)
-            .is_some_and(|error| !error.is_empty())
+            .is_some_and(|error| !error.is_empty() && !is_retrying_deferred_rebalance_error(pool, error))
         {
             return Err(format!("rebalance reported lastError: {status}").into());
         }
@@ -1526,3 +1541,23 @@ fn rebalance_complete_accepts_non_participating_pools_but_requires_progress() {
     });
     assert!(!rebalance_complete(&no_movement, "run-1").unwrap());
 }
+
+#[test]
+fn rebalance_status_tolerates_deferred_entry_error_only_while_started() {
+    let deferred = "deferred transient rebalance entry failure: Storage resources are insufficient for the write operation: b/k";
+    let mut status = serde_json::json!({
+        "id": "run-1",
+        "pools": [{ "id": 0, "status": "Started", "stopping": false, "lastError": deferred, "progress": { "objects": 1, "bytes": 1 } }]
+    });
+    assert!(rebalance_active(&status, "run-1").expect("a retried deferral keeps the run active"));
+
+    status["pools"][0]["status"] = serde_json::json!("Completed");
+    assert!(
+        rebalance_complete(&status, "run-1").is_err(),
+        "a completed pool must have cleared its deferred entry error"
+    );
+
+    status["pools"][0]["status"] = serde_json::json!("Started");
+    status["pools"][0]["lastError"] = serde_json::json!("rebalance entry failed: Storage resources are insufficient");
+    assert!(rebalance_active(&status, "run-1").is_err(), "a non-deferred lastError stays fatal");
+}

crates/ecstore/src/core/pools.rs

@@ -3486,7 +3486,13 @@ fn resolve_decommission_progress_save_result(result: Result<()>) -> Option<Error
 }
 
 fn resolve_decommission_preflight_heal_result<T>(bucket: &str, result: Result<T>) -> Result<T> {
-    result.map_err(|err| Error::other(format!("decommission preflight heal failed for bucket {bucket}: {err}")))
+    result.map_err(|err| match err {
+        // A peer whose movement gate has a queued writer refuses the bucket
+        // heal instead of waiting; the refusal crosses the disk layer as
+        // `TooManyOpenFiles`. Keep it retryable so the start answers 503.
+        Error::SlowDown | Error::TooManyOpenFiles => Error::SlowDown,
+        err => Error::other(format!("decommission preflight heal failed for bucket {bucket}: {err}")),
+    })
 }
 
 fn resolve_decommission_optional_bucket_config_result<T>(bucket: &str, stage: &str, result: Result<T>) -> Result<Option<T>> {
@@ -33548,7 +33554,7 @@ mod pools_tests {
 
     #[test]
     fn test_resolve_decommission_preflight_heal_result_wraps_error_context() {
-        let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(Error::SlowDown))
+        let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(Error::FaultyDisk))
             .expect_err("heal failure should carry preflight context");
         assert!(
             err.to_string()
@@ -33556,6 +33562,15 @@ mod pools_tests {
         );
     }
 
+    #[test]
+    fn test_resolve_decommission_preflight_heal_result_keeps_busy_gate_retryable() {
+        for busy in [Error::SlowDown, Error::TooManyOpenFiles] {
+            let err = resolve_decommission_preflight_heal_result::<()>("bucket-a", Err(busy))
+                .expect_err("a refused heal must fail the start");
+            assert_eq!(err, Error::SlowDown);
+        }
+    }
+
     #[test]
     fn test_resolve_decommission_optional_bucket_config_result_passthrough() {
         let result = resolve_decommission_optional_bucket_config_result("bucket-a", "replication", Ok(42_u8))

docs/testing/scanner-heal-topology-gates.md

@@ -11,7 +11,7 @@ Gate names come from the Scanner/Heal program (rustfs/backlog#2872); the quick m
 |---|---|---|
 | G-EXP | `distributed::scanner_heal_topology_test::scanner_heal_topology_expand_one_to_four_pools_preserves_versions_and_placement` | 1→2→4 pools with scanner and heal enabled: every recorded version, listing, marker and lock survives on every node after each step and after a full restart; seed-pool shards and erasure placement are byte-identical; new writes land in exactly one pool; the pool map reports four pools after restart |
 | G-EXP (existing baseline) | `distributed::expand_decommission_rebalance_test::four_node_pool_expand_preserves_objects_then_rebalance` | 64 unversioned objects survive expansion and a completed rebalance |
-| G-REB | `distributed::scanner_heal_topology_test::scanner_heal_topology_rebalance_survives_live_writes_stop_and_restart` | rebalance with concurrent PUT/GET/DELETE and the scanner running: no content error, acknowledged writes and deletes hold on every node, a stop leaves every version readable and held by at least one pool (a committed target beside its retained source is allowed), writes issued while the seed pool rebalances avoid it, every node serves after a restart, and a second run completes |
+| G-REB | `distributed::scanner_heal_topology_test::scanner_heal_topology_rebalance_survives_live_writes_stop_and_restart` | rebalance with concurrent PUT/GET/DELETE and the scanner running: no content error, acknowledged writes and deletes hold on every node, a stop leaves every version readable and held by at least one pool (a committed target beside its retained source is allowed), writes issued while the seed pool rebalances avoid it, every node serves after a restart, and a second run completes. A `lastError` is fatal except a `deferred transient rebalance entry failure:` while the pool is still started, which the worker retries and clears on completion |
 | G-REB fault | `distributed::scanner_heal_topology_test::scanner_heal_topology_rebalance_kill_before_source_cleanup_resumes_without_loss` | SIGKILL of the rebalancing node at `before_source_cleanup` (target committed, source retained): the resumed run completes without loss or duplication |
 | G-DEC | `distributed::scanner_heal_topology_test::scanner_heal_topology_decommission_queue_cancel_resume_and_last_pool_guard` | cancel of an active decommission keeps every version in exactly one pool; a two-pool queue retires pools 0 and 1 in order with pool 1 reported `queued`; versions, markers, listings, legal holds and compliance retention are intact on the surviving pools and after a restart that removes the retired pools from the command line; retiring the last active pools is rejected with a client error |
 | G-DEC (existing baseline) | `distributed::expand_decommission_rebalance_test::four_pool_decommission_moves_objects_without_loss` | 96 unversioned objects survive retiring the seed pool |
@@ -61,7 +61,7 @@ The barrier directory is set for every node of a run; only the process whose ope
 | movement barrier reach | 240 s (the movement budget) | a window can sit deep in the object order; on Linux CI a busy target capacity gate (160 inline retries, entries deferred to round end) delayed the first fault object by about 2 minutes. The wait fails at once if the operation completes, fails or is canceled without entering the window. Each status probe is bounded to 5 s and a slow probe counts as "still running", because the admin status request can wait on the held mover until the barrier itself times out. The server releases a held barrier on its own after 120 s |
 | pre-ACK window reach | 60 s | the decommission start normally reaches its durable save in about a second (POST to response ~0.8 s locally); exceeding 60 s is reported as a stalled start path, not retried with a longer budget |
 | cancel, stop and terminal transitions | 90 s | stop and cancel are acknowledged before every in-flight entry finishes its cleanup, so terminal state and placement are polled within this budget |
-| decommission start retry | 503 within the movement budget; fleet-proof 500 for 45 s | a start that meets a busy movement gate answers a retryable 503 instead of waiting; any other status is fatal |
+| decommission start retry | 503 within the movement budget; fleet-proof 500 for 45 s | a start that meets a busy movement gate answers a retryable 503 instead of waiting; any other status is fatal . The start-time bucket heal that finds a peer's movement gate busy reports the same 503 |
 | per test (lane) | 6 × 120 s | `slow-timeout` of `[profile.e2e-distributed]`, which also serializes the lane |
 
 ## Run

rustfs/src/admin/handlers/pools.rs

@@ -337,7 +337,9 @@ fn decommission_start_api_error(err: crate::storage_api::error::StorageError) ->
             source: Some(Box::new(err)),
         };
     }
-    if is_lock_timeout(&err) {
+    // `SlowDown` is a peer refusing the start-time bucket heal while its
+    // movement gate has a queued writer: the same transient contention.
+    if is_lock_timeout(&err) || matches!(err, crate::storage_api::error::StorageError::SlowDown) {
         return ApiError {
             code: S3ErrorCode::ServiceUnavailable,
             message: err.to_string(),
@@ -1312,6 +1314,15 @@ mod pools_handler_tests {
         );
     }
 
+    #[test]
+    fn test_decommission_start_api_error_reports_a_refused_preflight_heal_as_retryable() {
+        let err = decommission_start_api_error(crate::storage_api::error::StorageError::SlowDown);
+
+        assert_eq!(err.code, s3s::S3ErrorCode::ServiceUnavailable);
+        let err = s3s::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
+        assert_eq!(err.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
+    }
+
     #[test]
     fn test_decommission_start_rebalance_refresh_error_keeps_busy_gates_retryable() {
         let busy = decommission_start_rebalance_refresh_error(