fix(ecstore): clear clippy warnings

.github/workflows/e2e-s3tests.yml

@@ -21,9 +21,6 @@
 #     suite and reports promotion candidates. Regressions, unclassified tests,
 #     incomplete execution, and infrastructure errors fail the job; classified
 #     failures for not-yet-implemented features remain informational.
-#   - Non-blocking upstream HEAD canary: collects current upstream node IDs and
-#     reports new, removed, duplicate, or overlapping classifications without
-#     making upstream drift a release gate.
 #   - Manual runs (workflow_dispatch): same, with configurable mode/scope.
 #
 # All test execution is delegated to scripts/s3-tests/run.sh (single source of
@@ -357,85 +354,6 @@ jobs:
           name: s3tests-${{ env.TEST_MODE }}-shard-${{ matrix.shard-index }}
           path: artifacts/**
 
-  upstream-head-canary:
-    name: Upstream HEAD classification canary
-    if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
-    continue-on-error: true
-    runs-on: ubuntu-latest
-    timeout-minutes: 20
-    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Set up Python
-        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
-        with:
-          python-version: "3.12"
-
-      - name: Install collection tool
-        run: |
-          python3 -m pip install --user "tox==4.60.0"
-          python3 - <<'PY'
-          from importlib.metadata import version
-
-          assert version("tox") == "4.60.0"
-          PY
-          echo "$HOME/.local/bin" >> "$GITHUB_PATH"
-
-      - name: Compare upstream HEAD classifications
-        id: upstream-compare
-        run: |
-          ARTIFACT_DIR="artifacts/s3tests-upstream-head"
-          UPSTREAM_DIR="${RUNNER_TEMP}/s3-tests-upstream"
-          mkdir -p "${ARTIFACT_DIR}"
-          git clone --depth 1 https://github.com/ceph/s3-tests.git "${UPSTREAM_DIR}"
-          git -C "${UPSTREAM_DIR}" rev-parse HEAD > "${ARTIFACT_DIR}/upstream-sha.txt"
-          cp "${UPSTREAM_DIR}/s3tests.conf.SAMPLE" "${UPSTREAM_DIR}/s3tests.conf"
-          (
-            cd "${UPSTREAM_DIR}"
-            S3TEST_CONF="${UPSTREAM_DIR}/s3tests.conf" tox -- \
-              -q --collect-only s3tests/functional/test_s3.py \
-              -m "not rustfs_never_marker"
-          ) 2>&1 | tee "${ARTIFACT_DIR}/collect.log"
-          grep -E '^s3tests/functional/test_s3\.py::' \
-            "${ARTIFACT_DIR}/collect.log" > "${ARTIFACT_DIR}/collected-nodeids.txt"
-          python3 scripts/s3-tests/report_compat.py \
-            --lists-dir scripts/s3-tests \
-            --collected-nodeids "${ARTIFACT_DIR}/collected-nodeids.txt" \
-            --check-classifications-only 2>&1 | tee "${ARTIFACT_DIR}/classification-drift.txt"
-
-      - name: Publish canary report
-        if: always()
-        env:
-          CANARY_OUTCOME: ${{ steps.upstream-compare.outcome }}
-        run: |
-          {
-            echo "## ceph/s3-tests upstream HEAD canary"
-            echo
-            if [ -f artifacts/s3tests-upstream-head/upstream-sha.txt ]; then
-              echo "Upstream HEAD: $(cat artifacts/s3tests-upstream-head/upstream-sha.txt)"
-            fi
-            echo
-            echo '```text'
-            if [ -s artifacts/s3tests-upstream-head/classification-drift.txt ]; then
-              cat artifacts/s3tests-upstream-head/classification-drift.txt
-            elif [ "${CANARY_OUTCOME}" != "success" ]; then
-              echo "Canary did not complete; inspect the collection log artifact."
-            else
-              echo "No classification drift detected."
-            fi
-            echo '```'
-          } >> "$GITHUB_STEP_SUMMARY"
-
-      - name: Upload canary artifacts
-        if: always() && env.ACT != 'true'
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: s3tests-upstream-head
-          path: artifacts/s3tests-upstream-head/**
-          retention-days: 14
-
   alert-on-failure:
     name: Alert on scheduled failure
     needs: [s3tests]

crates/ecstore/src/core/pools.rs

@@ -1455,6 +1455,36 @@ where
     Ok(())
 }
 
+async fn run_decommission_phases<F>(
+    rx: CancellationToken,
+    regular_buckets: Vec<DecomBucketInfo>,
+    meta_buckets: Vec<DecomBucketInfo>,
+    bucket_concurrency: usize,
+    mut start_bucket: F,
+) -> Result<()>
+where
+    F: FnMut(DecomBucketInfo, CancellationToken) -> BoxFuture<'static, Result<()>>,
+{
+    decommission_cancel_signal_result(rx.is_cancelled())?;
+
+    for bucket in meta_buckets {
+        decommission_cancel_signal_result(rx.is_cancelled())?;
+        start_bucket(bucket, rx.clone()).await?;
+    }
+
+    decommission_cancel_signal_result(rx.is_cancelled())?;
+
+    if bucket_concurrency <= 1 {
+        for bucket in regular_buckets {
+            decommission_cancel_signal_result(rx.is_cancelled())?;
+            start_bucket(bucket, rx.clone()).await?;
+        }
+        return Ok(());
+    }
+
+    run_decommission_buckets_bounded(rx, regular_buckets, bucket_concurrency, start_bucket).await
+}
+
 #[cfg(test)]
 async fn wait_decommission_worker_drain(workers: &Semaphore, limit: usize) -> Result<()> {
     let permits = u32::try_from(limit)
@@ -4902,25 +4932,6 @@ impl ECStore {
         Ok(())
     }
 
-    async fn decommission_buckets_concurrently(
-        self: &Arc<Self>,
-        rx: CancellationToken,
-        idx: usize,
-        pool: Arc<Sets>,
-        buckets: Vec<DecomBucketInfo>,
-        limit: usize,
-        entry_budget: Arc<Semaphore>,
-    ) -> Result<()> {
-        let store = Arc::clone(self);
-        run_decommission_buckets_bounded(rx, buckets, limit, move |bucket, rx| {
-            let store = Arc::clone(&store);
-            let pool = pool.clone();
-            let entry_budget = entry_budget.clone();
-            Box::pin(async move { store.decommission_pending_bucket(rx, idx, pool, bucket, entry_budget).await })
-        })
-        .await
-    }
-
     #[tracing::instrument(skip(self, rx))]
     async fn decommission_in_background(
         self: &Arc<Self>,
@@ -4935,31 +4946,15 @@ impl ECStore {
             pool_meta.pending_buckets(idx)
         };
         let bucket_concurrency = decommission_bucket_concurrency_limit();
-        if bucket_concurrency <= 1 {
-            for bucket in pending {
-                self.decommission_pending_bucket(rx.clone(), idx, pool.clone(), bucket, entry_budget.clone())
-                    .await?;
-            }
-            return Ok(());
-        }
-
         let (regular_buckets, meta_buckets) = split_decommission_buckets(pending);
-        self.decommission_buckets_concurrently(
-            rx.clone(),
-            idx,
-            pool.clone(),
-            regular_buckets,
-            bucket_concurrency,
-            entry_budget.clone(),
-        )
-        .await?;
-
-        for bucket in meta_buckets {
-            self.decommission_pending_bucket(rx.clone(), idx, pool.clone(), bucket, entry_budget.clone())
-                .await?;
-        }
-
-        Ok(())
+        let store = Arc::clone(self);
+        run_decommission_phases(rx, regular_buckets, meta_buckets, bucket_concurrency, move |bucket, rx| {
+            let store = Arc::clone(&store);
+            let pool = pool.clone();
+            let entry_budget = entry_budget.clone();
+            Box::pin(async move { store.decommission_pending_bucket(rx, idx, pool, bucket, entry_budget).await })
+        })
+        .await
     }
 
     #[tracing::instrument(skip(self))]
@@ -6377,8 +6372,8 @@ mod pools_tests {
         resolve_decommission_terminal_mark_after_error_result, resolve_decommission_terminal_mark_result,
         resolve_decommission_update_after_result, resolve_start_decommission_pool_meta_reload_result,
         rollback_start_decommission_pool_meta, run_decommission_buckets_bounded, run_decommission_listing_with_retry,
-        run_decommission_listing_with_retry_and_drain, run_decommission_side_effect, should_cleanup_decommission_source_entry,
-        should_continue_decommission_queue, should_count_decommission_version_complete,
+        run_decommission_listing_with_retry_and_drain, run_decommission_phases, run_decommission_side_effect,
+        should_cleanup_decommission_source_entry, should_continue_decommission_queue, should_count_decommission_version_complete,
         should_preserve_decommission_canceled_state, should_reject_decommission_cancel_as_terminal,
         should_retry_decommission_cancel_reload, should_retry_decommission_listing, should_skip_canceled_decommission_routine,
         spawn_decommission_index_cancelers, split_decommission_buckets, take_and_cancel_decommission_canceler,
@@ -6397,7 +6392,7 @@ mod pools_tests {
     use rustfs_filemeta::{MetaCacheEntries, MetadataResolutionParams};
     use rustfs_rio::Index;
     use std::sync::{
-        Arc,
+        Arc, Mutex as StdMutex,
         atomic::{AtomicBool, AtomicUsize, Ordering},
     };
     use std::time::Duration as StdDuration;
@@ -6690,6 +6685,66 @@ mod pools_tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_decommission_metadata_phase_precedes_regular_failure() {
+        let events = Arc::new(StdMutex::new(Vec::new()));
+        let err = run_decommission_phases(
+            CancellationToken::new(),
+            vec![
+                DecomBucketInfo {
+                    name: "regular-fails".to_string(),
+                    ..Default::default()
+                },
+                DecomBucketInfo {
+                    name: "regular-not-started".to_string(),
+                    ..Default::default()
+                },
+            ],
+            vec![
+                DecomBucketInfo {
+                    name: crate::disk::RUSTFS_META_BUCKET.to_string(),
+                    prefix: crate::config::com::CONFIG_PREFIX.to_string(),
+                },
+                DecomBucketInfo {
+                    name: crate::disk::RUSTFS_META_BUCKET.to_string(),
+                    prefix: crate::disk::BUCKET_META_PREFIX.to_string(),
+                },
+            ],
+            1,
+            {
+                let events = Arc::clone(&events);
+                move |bucket, _rx| {
+                    let events = Arc::clone(&events);
+                    Box::pin(async move {
+                        let event = if bucket.name == crate::disk::RUSTFS_META_BUCKET {
+                            format!("meta:{}", bucket.prefix)
+                        } else {
+                            format!("regular:{}", bucket.name)
+                        };
+                        events.lock().expect("phase event lock should not be poisoned").push(event);
+                        if bucket.name == "regular-fails" {
+                            Err(Error::SlowDown)
+                        } else {
+                            Ok(())
+                        }
+                    })
+                }
+            },
+        )
+        .await
+        .expect_err("regular failure should remain fatal after metadata completes");
+
+        assert!(matches!(err, Error::SlowDown));
+        assert_eq!(
+            *events.lock().expect("phase event lock should not be poisoned"),
+            vec![
+                format!("meta:{}", crate::config::com::CONFIG_PREFIX),
+                format!("meta:{}", crate::disk::BUCKET_META_PREFIX),
+                "regular:regular-fails".to_string(),
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn test_run_decommission_buckets_bounded_respects_limit() {
         let rx = CancellationToken::new();

crates/ecstore/src/store/object.rs

@@ -3194,7 +3194,7 @@ impl ECStore {
 
         // Default return value
         let mut del_objects = vec![DeletedObject::default(); objects.len()];
-        let mut accounting = vec![None; objects.len()];
+        let accounting = vec![None; objects.len()];
 
         let mut del_errs = Vec::with_capacity(objects.len());
         for _ in 0..objects.len() {

crates/ecstore/src/store/rebalance/support.rs

@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
             .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
             .collect::<Vec<_>>();
 
-        latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
+        latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
 
         let Some(winner) = latest_candidates.first() else {
             return Err(Error::ErasureReadQuorum);