perf(storage): optimize small-object GET/PUT paths (#6770)

* perf(ecstore): optimize small-object GET paths

Co-Authored-By: heihutu <heihutu@gmail.com>

* perf(rustfs): optimize small-object request paths

Co-Authored-By: heihutu <heihutu@gmail.com>

* chore(deps): upgrade argon2 and convert_case

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): restore reader hotpath attribution

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): cover external mid-size fixtures

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

rustfs/src/app/object/get.rs

@@ -180,6 +180,14 @@ const GET_OBJECT_STAGE_PATH_S3_HANDLER: &str = "s3_handler";
 
 const GET_OBJECT_STAGE_REQUEST_INGRESS_TO_CONTEXT: &str = "request_ingress_to_context";
 
+const GET_OBJECT_STAGE_REQUEST_SHAPE: &str = "request_shape";
+
+const GET_OBJECT_STAGE_REQUEST_VALIDATION: &str = "request_validation";
+
+const GET_OBJECT_STAGE_BUCKET_VALIDATION: &str = "bucket_validation";
+
+const GET_OBJECT_STAGE_RESPONSE_FINALIZE: &str = "response_finalize";
+
 const GET_OBJECT_STAGE_OUTPUT_STRATEGY: &str = "output_strategy";
 
 const GET_OBJECT_STAGE_BODY_BUILD: &str = "body_build";
@@ -3817,6 +3825,8 @@ impl DefaultObjectUsecase {
             let _ = context.object_store();
         }
 
+        let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        let request_shape_start = stage_metrics_enabled.then(std::time::Instant::now);
         let inbound_request_context = req.extensions.get::<request_context::RequestContext>();
         let request_id = inbound_request_context
             .map(|ctx| ctx.request_id.clone())
@@ -3831,6 +3841,7 @@ impl DefaultObjectUsecase {
             );
         }
         let bootstrap = self.init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id)?;
+        record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_SHAPE, request_shape_start);
         let timeout_config = bootstrap.timeout_config;
         let wrapper = bootstrap.wrapper;
         let request_start = bootstrap.request_start;
@@ -3842,6 +3853,7 @@ impl DefaultObjectUsecase {
 
         // Cheap request-shape validations run first so invalid requests keep
         // their InvalidArgument precedence over bucket existence.
+        let request_validation_start = stage_metrics_enabled.then(std::time::Instant::now);
         let validated = match Self::validate_get_object_request(&req) {
             Ok(validated) => validated,
             Err(err) => {
@@ -3849,6 +3861,7 @@ impl DefaultObjectUsecase {
                 return Err(err);
             }
         };
+        record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_REQUEST_VALIDATION, request_validation_start);
 
         // SF05: Store lookup next (5s-TTL bucket-validation cache). Bucket
         // existence is established before any bucket-metadata work, so requests
@@ -3859,24 +3872,26 @@ impl DefaultObjectUsecase {
         let object_metadata_progress = object_traffic_health
             .as_deref()
             .and_then(ObjectTrafficHealth::track_read_metadata);
-        let store_lookup_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
+        let store_lookup_start = stage_metrics_enabled.then(std::time::Instant::now);
         let Some(store) = self.object_store() else {
             lifecycle.finish_err();
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
-        if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await {
-            lifecycle.finish_err();
-            return Err(err);
-        }
         if let Some(store_lookup_start) = store_lookup_start {
             rustfs_io_metrics::record_get_object_stage_duration(
-                "s3_handler",
+                GET_OBJECT_STAGE_PATH_S3_HANDLER,
                 "store_lookup",
                 store_lookup_start.elapsed().as_secs_f64(),
             );
         }
+        let bucket_validation_start = stage_metrics_enabled.then(std::time::Instant::now);
+        if let Err(err) = validate_bucket_exists(&store, &req.input.bucket).await {
+            lifecycle.finish_err();
+            return Err(err);
+        }
+        record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BUCKET_VALIDATION, bucket_validation_start);
 
-        let request_context_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
+        let request_context_start = stage_metrics_enabled.then(std::time::Instant::now);
         let request_context = match Self::prepare_get_object_request_context(validated, &req.headers).await {
             Ok(request_context) => request_context,
             Err(err) => {
@@ -4051,7 +4066,8 @@ impl DefaultObjectUsecase {
             optimal_buffer_size,
         );
 
-        Self::finalize_get_object_response(
+        let response_finalize_start = stage_metrics_enabled.then(std::time::Instant::now);
+        let response = Self::finalize_get_object_response(
             helper,
             &bucket,
             &req.method,
@@ -4061,7 +4077,9 @@ impl DefaultObjectUsecase {
             output,
             extra_checksum_headers,
         )
-        .await
+        .await;
+        record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_RESPONSE_FINALIZE, response_finalize_start);
+        response
     }
 
     pub async fn execute_get_object_attributes(

rustfs/src/app/object/put.rs

@@ -23,8 +23,18 @@ const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 102
 
 const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
 
+/// Maximum body size materialized by the ordinary eager PUT path.
+///
+/// Bodies above this boundary stay streaming so a 1 MiB request does not
+/// reserve a full request-sized buffer while the EC writer is consuming it.
+/// The environment override keeps the boundary reversible for workload A/B
+/// tests and for deployments whose measured workload favors eager ingestion.
+const ENV_SMALL_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_SMALL_EAGER_PUT_MAX_SIZE_BYTES";
+
 const DEFAULT_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: usize = 16 * 1024 * 1024;
 
+const DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES: usize = 512 * 1024;
+
 const PUT_EAGER_STATUS_ELIGIBLE: &str = "eligible";
 
 const PUT_EAGER_STATUS_EXTRACT: &str = "extract";
@@ -43,6 +53,8 @@ const PUT_EAGER_STATUS_AWS_CHUNKED_MISSING_DECODED_LENGTH: &str = "aws_chunked_m
 
 static CACHED_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
 
+static CACHED_SMALL_EAGER_PUT_MAX_SIZE_BYTES: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
+
 const EVENT_PUT_OBJECT_STORE_INFLIGHT_SLOW: &str = "put_object_store_inflight_slow";
 
 const EVENT_PUT_OBJECT_STORE_RETURNED: &str = "put_object_store_returned";
@@ -483,13 +495,29 @@ fn should_use_small_eager_put_path(
     should_compress: bool,
     is_extract: bool,
 ) -> bool {
-    const SMALL_EAGER_PUT_MAX_SIZE: i64 = 1024 * 1024;
+    should_use_small_eager_put_path_with_max_size(
+        size,
+        headers,
+        server_side_encryption_requested,
+        should_compress,
+        is_extract,
+        small_eager_put_max_size_bytes(),
+    )
+}
 
+fn should_use_small_eager_put_path_with_max_size(
+    size: i64,
+    headers: &HeaderMap,
+    server_side_encryption_requested: bool,
+    should_compress: bool,
+    is_extract: bool,
+    max_size: i64,
+) -> bool {
     if is_extract || should_compress || server_side_encryption_requested {
         return false;
     }
 
-    if size <= 0 || size > SMALL_EAGER_PUT_MAX_SIZE {
+    if size <= 0 || size > max_size {
         return false;
     }
 
@@ -504,6 +532,42 @@ fn should_use_small_eager_put_path(
     true
 }
 
+fn small_eager_put_max_size_bytes() -> i64 {
+    let configured = *CACHED_SMALL_EAGER_PUT_MAX_SIZE_BYTES
+        .get_or_init(|| rustfs_utils::get_env_usize(ENV_SMALL_EAGER_PUT_MAX_SIZE_BYTES, DEFAULT_SMALL_EAGER_PUT_MAX_SIZE_BYTES));
+    i64::try_from(configured).unwrap_or(i64::MAX)
+}
+
+fn select_put_path(
+    size: i64,
+    headers: &HeaderMap,
+    server_side_encryption_requested: bool,
+    should_compress: bool,
+    is_extract: bool,
+) -> (&'static str, &'static str, bool, bool) {
+    let use_empty_or_small_eager_put_path = size == 0
+        || should_use_small_eager_put_path(size, headers, server_side_encryption_requested, should_compress, is_extract);
+    let zero_copy_eager_put_path_status =
+        zero_copy_eager_put_path_status(size, headers, server_side_encryption_requested, should_compress, is_extract);
+    let use_zero_copy_eager_put_path = zero_copy_eager_put_path_status == PUT_EAGER_STATUS_ELIGIBLE;
+    let put_path = if should_compress {
+        "stream_compressed"
+    } else if use_zero_copy_eager_put_path {
+        "zero_copy_eager"
+    } else if use_empty_or_small_eager_put_path {
+        "small_eager"
+    } else {
+        "streaming"
+    };
+
+    (
+        put_path,
+        zero_copy_eager_put_path_status,
+        use_zero_copy_eager_put_path,
+        use_empty_or_small_eager_put_path,
+    )
+}
+
 /// Objects at or below this size bypass BytesPool and use direct allocation.
 /// This avoids Small-tier Mutex contention under high concurrency for tiny objects
 /// where the allocation cost is negligible (≤4KiB memcpy).
@@ -800,12 +864,26 @@ impl DefaultObjectUsecase {
 
     async fn execute_put_object_inner(&self, _fs: &FS, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
         let start_time = std::time::Instant::now();
+        let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled();
         let mut req = req;
 
+        let request_shape_stage_start = put_stage_metrics_enabled.then(Instant::now);
+
         if let Some(context) = &self.context {
             let _ = context.object_store();
         }
 
+        // Authentication and header parsing happen in the S3 middleware before
+        // this use case runs. Attribute that already-paid request prefix from
+        // the request context without adding per-request work when stage
+        // metrics are disabled.
+        if put_stage_metrics_enabled && let Some(context) = req.extensions.get::<request_context::RequestContext>() {
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "request_ingress_to_context",
+                context.start_time.elapsed().as_secs_f64() * 1000.0,
+            );
+        }
+
         let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
         let max_content_length = parse_presigned_put_max_content_length(
             &req.headers,
@@ -893,6 +971,7 @@ impl DefaultObjectUsecase {
             req.headers.get("content-type").and_then(|value| value.to_str().ok()),
             req.headers.get("content-encoding").and_then(|value| value.to_str().ok()),
         )?;
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_request_shape", request_shape_stage_start);
 
         let Some(body) = body else { return Err(s3_error!(IncompleteBody)) };
 
@@ -929,6 +1008,7 @@ impl DefaultObjectUsecase {
 
         // The app check preserves the existing S3 error contract; the storage
         // commit path reserves the exact net logical growth under its locks.
+        let quota_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let quota_check = self
             .check_bucket_quota(
                 &bucket,
@@ -936,6 +1016,7 @@ impl DefaultObjectUsecase {
                 u64::try_from(size).map_err(|_| S3Error::new(S3ErrorCode::UnexpectedContent))?,
             )
             .await?;
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_quota_check", quota_stage_start);
         let quota_enabled = quota_check.as_ref().is_some_and(|result| result.quota_limit.is_some());
         if quota_enabled && ciphertext_passthrough {
             return Err(S3Error::with_message(
@@ -944,7 +1025,6 @@ impl DefaultObjectUsecase {
             ));
         }
 
-        let put_stage_metrics_enabled = rustfs_io_metrics::put_stage_metrics_enabled();
         let ingress_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let should_compress =
             is_disk_compressible(&req.headers, &key) && size > MIN_DISK_COMPRESSIBLE_SIZE as i64 && !ciphertext_passthrough;
@@ -954,11 +1034,13 @@ impl DefaultObjectUsecase {
         // Resolve the store through the request-bound server context
         // (backlog#1052 S6), not the process-global handle, so an embedded
         // second server never writes into the first server's store.
+        let store_lookup_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let Some(store) = self.object_store() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
         let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
         validate_bucket_exists(&store, &bucket).await?;
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_store_lookup", store_lookup_stage_start);
         rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
 
         let put_admission = match get_concurrency_manager()
@@ -1006,24 +1088,12 @@ impl DefaultObjectUsecase {
             debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key);
         }
 
-        let use_empty_or_small_eager_put_path = size == 0
-            || should_use_small_eager_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false);
-        let zero_copy_eager_put_path_status =
-            zero_copy_eager_put_path_status(size, &req.headers, server_side_encryption_requested, should_compress, false);
-        let use_zero_copy_eager_put_path = zero_copy_eager_put_path_status == PUT_EAGER_STATUS_ELIGIBLE;
+        let (put_path, zero_copy_eager_put_path_status, use_zero_copy_eager_put_path, use_empty_or_small_eager_put_path) =
+            select_put_path(size, &req.headers, server_side_encryption_requested, should_compress, false);
         if use_zero_copy_eager_put_path {
             counter!(buffered_write::ATTEMPTS_TOTAL).increment(1);
             histogram!(buffered_write::ATTEMPT_SIZE_BYTES).record(size as f64);
         }
-        let put_path = if should_compress {
-            "stream_compressed"
-        } else if use_zero_copy_eager_put_path {
-            "zero_copy_eager"
-        } else if use_empty_or_small_eager_put_path {
-            "small_eager"
-        } else {
-            "streaming"
-        };
         rustfs_io_metrics::record_put_object_diagnostics(
             put_path,
             zero_copy_eager_put_path_status,
@@ -1518,6 +1588,7 @@ impl DefaultObjectUsecase {
                 Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned })
             }
         });
+        let commit_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let put_commit_result = if let Some(cancellation) = eager_put_commit_cancellation {
             EagerPutCommitOwner::new(put_commit, cancellation, EAGER_PUT_COMMIT_CANCELLATION_GRACE)
                 .join()
@@ -1525,6 +1596,7 @@ impl DefaultObjectUsecase {
         } else {
             put_commit.await
         };
+        rustfs_io_metrics::record_put_object_stage_duration_from("store_commit", commit_stage_start);
         let PutObjectCommitResult { obj_info, put_versioned } = match put_commit_result {
             Ok(Ok(result)) => result,
             Ok(Err(err)) => {
@@ -1589,10 +1661,12 @@ impl DefaultObjectUsecase {
         // For browser-based POST uploads (multipart/form-data), response status/body handling
         // is decided by s3s PostObject serializer (success_action_status / redirect semantics).
 
+        let response_build_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let mut response = S3Response::new(output);
         // Echo XXHash3/64/128 / SHA-512 checksums that s3s PutObjectOutput has no typed
         // field for (#1256).
         inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers);
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_response_build", response_build_stage_start);
         let result = Ok(response);
         let _ = helper.complete(&result);
 
@@ -2230,12 +2304,51 @@ mod tests {
     }
 
     #[test]
-    fn should_use_small_eager_put_path_allows_up_to_1mb() {
+    fn should_use_small_eager_put_path_keeps_small_objects_eager() {
         let headers = HeaderMap::new();
 
         assert!(should_use_small_eager_put_path(1024, &headers, false, false, false));
-        assert!(should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false));
-        assert!(!should_use_small_eager_put_path(1024 * 1024 + 1, &headers, false, false, false));
+        assert!(should_use_small_eager_put_path(128 * 1024, &headers, false, false, false));
+        assert!(should_use_small_eager_put_path(512 * 1024, &headers, false, false, false));
+        assert!(!should_use_small_eager_put_path(512 * 1024 + 1, &headers, false, false, false));
+        assert!(!should_use_small_eager_put_path(1024 * 1024, &headers, false, false, false));
+    }
+
+    #[test]
+    fn select_put_path_switches_at_small_eager_boundary() {
+        let headers = HeaderMap::new();
+
+        let (small_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024, &headers, false, false, false);
+        assert_eq!(small_path, "small_eager");
+        assert!(!use_zero_copy);
+        assert!(use_small_eager);
+
+        let (streaming_path, _, use_zero_copy, use_small_eager) = select_put_path(512 * 1024 + 1, &headers, false, false, false);
+        assert_eq!(streaming_path, "streaming");
+        assert!(!use_zero_copy);
+        assert!(!use_small_eager);
+    }
+
+    #[test]
+    fn should_use_small_eager_put_path_allows_a_b_override_at_1mb() {
+        let headers = HeaderMap::new();
+
+        assert!(should_use_small_eager_put_path_with_max_size(
+            1024 * 1024,
+            &headers,
+            false,
+            false,
+            false,
+            1024 * 1024,
+        ));
+        assert!(!should_use_small_eager_put_path_with_max_size(
+            1024 * 1024 + 1,
+            &headers,
+            false,
+            false,
+            false,
+            1024 * 1024,
+        ));
     }
 
     #[test]
@@ -2456,6 +2569,35 @@ mod tests {
         assert_eq!(extra.code(), &S3ErrorCode::UnexpectedContent);
     }
 
+    #[tokio::test]
+    async fn streaming_put_hash_reader_rejects_extra_byte_at_eager_boundary() {
+        use tokio::io::AsyncReadExt;
+
+        let declared_size = 512 * 1024;
+        let declared_size_i64 = i64::try_from(declared_size).expect("test size should fit i64");
+        let mut reader = HashReader::from_stream(
+            std::io::Cursor::new(vec![0x5a; declared_size + 1]),
+            declared_size_i64,
+            declared_size_i64,
+            None,
+            None,
+            false,
+        )
+        .expect("streaming PUT hash reader should be constructed");
+        let mut body = Vec::new();
+
+        let err = reader
+            .read_to_end(&mut body)
+            .await
+            .expect_err("streaming PUT must reject a body larger than Content-Length");
+
+        assert!(
+            err.to_string().contains("more bytes than specified"),
+            "unexpected extra-body error: {err}"
+        );
+        assert_eq!(body.len(), declared_size);
+    }
+
     #[tokio::test]
     async fn read_small_put_body_exact_direct_handles_empty_body_boundary() {
         let empty = read_small_put_body_exact_direct(std::io::Cursor::new(Vec::<u8>::new()), 0)