chore: merge repaired main for drive diagnostics acceptance

* fix: repair s3s footprint ratchet violation, clippy warnings, and stack overflow

- Replace s3_error! macro with S3Error::with_message in ecfs.rs to keep
  s3s footprint ratchet at baseline (1586) after retention validation refactor
- Remove redundant clone on Copy type in list_objects.rs (clippy::redundant_clone)
- Remove clone_on_copy calls in access.rs test (clippy::clone_on_copy)
- Use manual thread with 8MB stack for heal_concurrent_delete_test to fix
  stack overflow on macOS (consistent with sibling tests in same file)

* test: align CI assertions with version authorization rules

crates/e2e_test/src/sts_query_compat_test.rs

@@ -740,7 +740,7 @@ async fn test_bucket_tags_opa_contract() -> TestResult {
         .version_id(version)
         .send()
         .await?;
-    opa.expect_bucket_tags("s3:GetObject", source, Some("finance")).await?;
+    opa.expect_bucket_tags("s3:GetObjectVersion", source, Some("finance")).await?;
     client.list_objects_v2().bucket(source).send().await?;
     opa.expect_bucket_tags("s3:ListBucket", source, Some("finance")).await?;
     client.list_object_versions().bucket(source).send().await?;

crates/ecstore/src/store/list_objects.rs

@@ -9221,7 +9221,7 @@ mod test {
         };
         let snapshot = PersistentListMetadataObject::from_object_info(&object);
         assert!(snapshot.to_object_info("bucket").etag.is_none());
-        let mut old_snapshot = snapshot.clone();
+        let mut old_snapshot = snapshot;
         old_snapshot.etag = object.etag.clone();
         let old_contents = format!(
             "# rustfs-listobjects-key-only-v2\n{}\n",

crates/heal/tests/heal_concurrent_delete_test.rs

@@ -563,9 +563,31 @@ fn admin_root_completes_after_unversioned_delete() {
     run_delete_case(DeleteCase::Unversioned);
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[test]
 #[serial]
-async fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+fn admin_root_rejects_recreated_bucket_before_successor_repair() {
+    // Incarnation-bound repairs use a spawned owner; retain the server's stack
+    // budget for the whole scenario. block_on alone would poll that future on
+    // libtest's smaller calling-thread stack.
+    let stack_size = 8 * 1024 * 1024;
+    std::thread::Builder::new()
+        .name("heal-concurrent-delete".to_owned())
+        .stack_size(stack_size)
+        .spawn(move || {
+            let runtime = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(4)
+                .thread_stack_size(stack_size)
+                .enable_all()
+                .build()
+                .expect("concurrent delete runtime");
+            runtime.block_on(recreated_bucket_repair());
+        })
+        .expect("concurrent delete scenario thread starts")
+        .join()
+        .expect("concurrent delete scenario must not panic");
+}
+
+async fn recreated_bucket_repair() {
     let directory = tempfile::tempdir().expect("recreated bucket fixture");
     let env = TestECStoreEnv::builder().base_dir(directory.path()).build().await;
     env.make_bucket(BUCKET, false).await;

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -964,7 +964,11 @@ async fn copy_object_allows_new_version_for_locked_destination_but_blocks_explic
         .await
         .expect_err("explicit CopyObject overwrite of a locked version should be blocked");
 
-    assert_eq!(err.code(), &s3s::S3ErrorCode::AccessDenied);
+    assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidArgument);
+    assert_eq!(
+        err.message(),
+        Some("Invalid argument: A destination versionId requires replication authorization")
+    );
     assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), dst_object).await, source_payload);
     assert_eq!(live_object_version_count(&ecstore, bucket.as_str(), dst_object).await, 2);
 }

rustfs/src/connect/diagnostics/perf_drive.rs

@@ -27,6 +27,8 @@ use std::io::{Cursor, Write as _};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+#[cfg(test)]
+use std::{cell::RefCell, mem};
 
 use base64_simd::URL_SAFE_NO_PAD;
 use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
@@ -70,6 +72,140 @@ const SCRATCH_PREFIX: &str = ".rustfs-connect-drive-";
 
 static DRIVE_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false);
 
+#[cfg(test)]
+#[derive(Clone, Copy, Debug)]
+enum DriveStage {
+    ScratchCreate,
+    Open,
+    Write,
+    Sync,
+    ReadbackVerify,
+    Cleanup,
+}
+
+#[cfg(test)]
+const MAX_DRIVE_STAGE_TIMINGS: usize = 6;
+
+#[cfg(test)]
+#[derive(Debug)]
+pub struct DriveStageTiming {
+    phase: DriveStage,
+    elapsed_micros: u64,
+    succeeded: bool,
+}
+
+#[cfg(test)]
+// Keep parallel test workers isolated; one measurement emits at most six fixed phases.
+thread_local! {
+    static DRIVE_STAGE_TIMINGS: RefCell<Vec<DriveStageTiming>> = const { RefCell::new(Vec::new()) };
+}
+
+#[cfg(test)]
+pub fn clear_drive_stage_timings() {
+    DRIVE_STAGE_TIMINGS.with(|timings| timings.borrow_mut().clear());
+}
+
+#[cfg(test)]
+pub fn take_drive_stage_timings() -> Vec<DriveStageTiming> {
+    DRIVE_STAGE_TIMINGS.with(|timings| mem::take(&mut *timings.borrow_mut()))
+}
+
+#[cfg(test)]
+struct DriveStageTimer {
+    phase: DriveStage,
+    started: Instant,
+    succeeded: bool,
+}
+
+#[cfg(test)]
+impl DriveStageTimer {
+    fn new(phase: DriveStage) -> Self {
+        Self {
+            phase,
+            started: Instant::now(),
+            succeeded: false,
+        }
+    }
+
+    fn finish(mut self, succeeded: bool) {
+        self.succeeded = succeeded;
+    }
+}
+
+#[cfg(test)]
+impl Drop for DriveStageTimer {
+    fn drop(&mut self) {
+        let timing = DriveStageTiming {
+            phase: self.phase,
+            elapsed_micros: u64::try_from(self.started.elapsed().as_micros()).unwrap_or(u64::MAX),
+            succeeded: self.succeeded,
+        };
+        DRIVE_STAGE_TIMINGS.with(|timings| {
+            let mut timings = timings.borrow_mut();
+            if timings.len() < MAX_DRIVE_STAGE_TIMINGS {
+                timings.push(timing);
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod drive_stage_timing_tests {
+    use super::*;
+
+    #[test]
+    fn unfinished_stage_is_recorded_as_failed_and_take_clears() {
+        clear_drive_stage_timings();
+        drop(DriveStageTimer::new(DriveStage::ReadbackVerify));
+
+        let timings = take_drive_stage_timings();
+        assert_eq!(timings.len(), 1);
+        assert!(matches!(timings[0].phase, DriveStage::ReadbackVerify));
+        assert!(!timings[0].succeeded);
+        assert!(timings[0].elapsed_micros < u64::MAX);
+        assert!(take_drive_stage_timings().is_empty());
+    }
+
+    #[test]
+    fn stage_records_are_bounded_and_clear_discards_them() {
+        clear_drive_stage_timings();
+        for _ in 0..(MAX_DRIVE_STAGE_TIMINGS + 2) {
+            DriveStageTimer::new(DriveStage::Write).finish(true);
+        }
+
+        let timings = take_drive_stage_timings();
+        assert_eq!(timings.len(), MAX_DRIVE_STAGE_TIMINGS);
+        assert!(timings.iter().all(|timing| timing.succeeded));
+
+        DriveStageTimer::new(DriveStage::Cleanup).finish(true);
+        clear_drive_stage_timings();
+        assert!(take_drive_stage_timings().is_empty());
+    }
+
+    #[test]
+    fn stage_records_are_isolated_between_threads() {
+        clear_drive_stage_timings();
+        DriveStageTimer::new(DriveStage::Open).finish(true);
+
+        let child_timings = std::thread::spawn(|| {
+            assert!(take_drive_stage_timings().is_empty());
+            DriveStageTimer::new(DriveStage::Sync).finish(true);
+            take_drive_stage_timings()
+        })
+        .join()
+        .expect("stage timing thread");
+
+        assert_eq!(child_timings.len(), 1);
+        assert!(matches!(child_timings[0].phase, DriveStage::Sync));
+        assert!(child_timings[0].succeeded);
+
+        let parent_timings = take_drive_stage_timings();
+        assert_eq!(parent_timings.len(), 1);
+        assert!(matches!(parent_timings[0].phase, DriveStage::Open));
+        assert!(parent_timings[0].succeeded);
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum DriveOutcome {
@@ -389,13 +525,22 @@ pub async fn measure_drive(
     let _lease = CollectorLease::acquire()?;
     let started = Instant::now();
     let deadline = tokio::time::Instant::now() + request.duration;
-    let mut scratch = match ScratchGuard::create(request) {
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::ScratchCreate);
+    let created = ScratchGuard::create(request);
+    #[cfg(test)]
+    stage.finish(created.is_ok());
+    let mut scratch = match created {
         Ok(scratch) => scratch,
         Err(error) => return Ok(failed_measurement(request, started.elapsed(), classify_io(&error), 0, 0, 0)),
     };
 
     let measured = run_benchmark(request, &scratch, cancel, deadline).await;
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Cleanup);
     let cleanup = scratch.cleanup();
+    #[cfg(test)]
+    stage.finish(cleanup.is_ok());
     let elapsed = started.elapsed().min(request.duration);
     if cleanup.is_err() {
         let (read_bytes, write_bytes, io_count) = match &measured {
@@ -661,7 +806,11 @@ async fn run_benchmark(
     cancel: &CancellationToken,
     deadline: tokio::time::Instant,
 ) -> Result<BenchmarkSample, BenchmarkFailure> {
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Open);
     let std_file = scratch.open_file().map_err(BenchmarkFailure::Io)?;
+    #[cfg(test)]
+    stage.finish(true);
     let mut file = tokio::fs::File::from_std(std_file);
     let block_size = usize::try_from(request.block_bytes).map_err(|_| BenchmarkFailure::Verification)?;
     let write_buffer = vec![0xa5; block_size];
@@ -669,6 +818,8 @@ async fn run_benchmark(
     let mut write_bytes = 0_u64;
     let mut io_count = 0_u64;
     let write_started = Instant::now();
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Write);
     while remaining > 0 {
         let length = usize::try_from(remaining.min(request.block_bytes)).map_err(|_| BenchmarkFailure::Verification)?;
         cancellable(file.write_all(&write_buffer[..length]), cancel, deadline).await?;
@@ -677,8 +828,16 @@ async fn run_benchmark(
         remaining -= length as u64;
         tokio::task::yield_now().await;
     }
+    #[cfg(test)]
+    stage.finish(true);
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::Sync);
     cancellable(file.sync_all(), cancel, deadline).await?;
+    #[cfg(test)]
+    stage.finish(true);
     let write_latency = write_started.elapsed();
+    #[cfg(test)]
+    let stage = DriveStageTimer::new(DriveStage::ReadbackVerify);
     cancellable(file.seek(std::io::SeekFrom::Start(0)), cancel, deadline).await?;
 
     let mut read_buffer = vec![0_u8; block_size];
@@ -701,6 +860,8 @@ async fn run_benchmark(
     if read_bytes != request.scratch_bytes || write_bytes != request.scratch_bytes || io_count > MAX_OPERATIONS {
         return Err(BenchmarkFailure::Verification);
     }
+    #[cfg(test)]
+    stage.finish(true);
     Ok(BenchmarkSample {
         read_bytes,
         write_bytes,

rustfs/src/storage/access.rs

@@ -4343,15 +4343,8 @@ mod tests {
             (S3Action::DeleteObjectTaggingAction, S3Action::DeleteObjectVersionTaggingAction),
         ] {
             for version in [None, Some("null"), Some("0194e0f1-0000-7000-8000-000000000000")] {
-                let action = versioned_tagging_action(current.clone(), version);
-                assert_eq!(
-                    action,
-                    Action::S3Action(if version.is_some() {
-                        versioned.clone()
-                    } else {
-                        current.clone()
-                    })
-                );
+                let action = versioned_tagging_action(current, version);
+                assert_eq!(action, Action::S3Action(if version.is_some() { versioned } else { current }));
                 let args = Args {
                     account: "writer",
                     groups: &groups,

rustfs/src/storage/ecfs.rs

@@ -415,7 +415,12 @@ pub(crate) fn object_lock_retention_from_metadata(metadata: &HashMap<String, Str
             date.checked_to_offset(time::UtcOffset::UTC)
                 .filter(|date| (0..=9999).contains(&date.year()))
                 .map(Timestamp::from)
-                .ok_or_else(|| s3_error!(InternalError, "Stored retention date is outside the supported UTC range"))
+                .ok_or_else(|| {
+                    S3Error::with_message(
+                        S3ErrorCode::InternalError,
+                        "Stored retention date is outside the supported UTC range".to_string(),
+                    )
+                })
         })
         .transpose()?;
     Ok(ObjectLockRetention { mode, retain_until_date })

rustfs/tests/connect_perf_drive.rs

@@ -81,14 +81,16 @@ async fn real_local_write_read_is_measured_and_scratch_is_removed() {
     let _guard = TEST_LOCK.lock().await;
     let root = tempfile::tempdir().expect("scratch root");
     let request = request(root.path());
+    perf_drive::clear_drive_stage_timings();
     let measurement = measure_drive(&request, &CancellationToken::new())
         .await
         .expect("real drive benchmark");
+    let stage_timings = perf_drive::take_drive_stage_timings();
 
     assert_eq!(
         measurement.result.outcome(),
         DriveOutcome::Succeeded,
-        "drive measurement: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}",
+        "drive measurement: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}, stage_timings={stage_timings:?}",
         measurement.result.reason_code(),
         measurement.target.reason_code,
         measurement.target.duration_millis,
@@ -366,13 +368,15 @@ async fn successful_result_has_signed_bounded_private_offline_export() {
     let _guard = TEST_LOCK.lock().await;
     let root = tempfile::tempdir().expect("scratch root");
     let request = request(root.path());
+    perf_drive::clear_drive_stage_timings();
     let measurement = measure_drive(&request, &CancellationToken::new())
         .await
         .expect("drive measurement");
+    let stage_timings = perf_drive::take_drive_stage_timings();
     assert_eq!(
         measurement.result.outcome(),
         DriveOutcome::Succeeded,
-        "drive measurement before signed export: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}",
+        "drive measurement before signed export: result_reason={:?}, target_reason={:?}, duration_millis={}, read_bytes={}, write_bytes={}, io_count={}, stage_timings={stage_timings:?}",
         measurement.result.reason_code(),
         measurement.target.reason_code,
         measurement.target.duration_millis,