chore: merge repaired main for drive diagnostics acceptance

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/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,