feat(get): SF02 - inline data fast path

Add fast path for small inline objects that bypasses duplex pipe,
tokio::spawn, and bitrot reader creation when data is already in memory.

Changes:
- Add inline data detection before codec streaming gate
- Direct in-memory erasure decode for inline objects <= 128KB
- Add GET_OBJECT_PATH_INLINE_DIRECT metric path
- Skip duplex pipe and background task for inline data

Conditions for fast path:
- Single part object
- Inline data available
- Size <= 128KB
- Not encrypted/compressed/remote
- No range request

Expected impact: 2-3x improvement for small file GET latency.

Closes rustfs/backlog#767

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

Cargo.lock

@@ -9042,6 +9042,7 @@ dependencies = [
  "metrics",
  "mimalloc",
  "mime_guess",
+ "moka",
  "opentelemetry",
  "opentelemetry_sdk",
  "percent-encoding",

crates/ecstore/src/diagnostics/get.rs

@@ -20,6 +20,7 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming";
 pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
 pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine";
 pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
+pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct";
 pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
 pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
 pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use";

crates/ecstore/src/set_disk/mod.rs

@@ -25,9 +25,9 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
 use crate::cluster::rpc::heal_bucket_local_on_disks;
 use crate::diagnostics::get::{
     GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE,
-    GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX,
-    GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error,
-    record_get_object_pipeline_failure,
+    GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT,
+    GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA,
+    classify_storage_error, record_get_object_pipeline_failure,
 };
 use crate::disk::error_reduce::{
     BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,
@@ -1557,6 +1557,75 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             return Ok(reader);
         }
 
+        // Inline data fast path: skip duplex pipe for small inline objects
+        const INLINE_FAST_PATH_MAX_SIZE: i64 = 128 * 1024; // 128KB
+        if fi.inline_data()
+            && fi.data.is_some()
+            && fi.parts.len() == 1
+            && object_info.size <= INLINE_FAST_PATH_MAX_SIZE
+            && !object_info.is_encrypted()
+            && !object_info.is_compressed()
+            && !object_info.is_remote()
+            && range.is_none()
+        {
+            let data_shards = fi.erasure.data_blocks;
+
+            // Check if we have enough inline data shards
+            let inline_count = files.iter()
+                .take(data_shards)
+                .filter(|f| f.data.as_ref().is_some_and(|d| !d.is_empty()))
+                .count();
+
+            if inline_count >= data_shards {
+                // All data shards are inline - decode in memory
+                let erasure = coding::Erasure::new(
+                    fi.erasure.data_blocks,
+                    fi.erasure.parity_blocks,
+                    fi.erasure.block_size,
+                );
+
+                // Build bitrot readers from inline data
+                let checksum_info = fi.erasure.get_checksum_info(fi.parts[0].number);
+                let checksum_algo = checksum_info.algorithm;
+                let mut readers: Vec<Option<coding::BitrotReader<Box<dyn tokio::io::AsyncRead + Send + Sync + Unpin>>>> = Vec::new();
+                for file in files.iter().take(data_shards + fi.erasure.parity_blocks) {
+                    if let Some(data) = &file.data {
+                        let cursor: Box<dyn tokio::io::AsyncRead + Send + Sync + Unpin> = Box::new(Cursor::new(data.to_vec()));
+                        let reader = coding::BitrotReader::new(
+                            cursor,
+                            fi.erasure.block_size,
+                            checksum_algo.clone(),
+                            false,
+                        );
+                        readers.push(Some(reader));
+                    } else {
+                        readers.push(None);
+                    }
+                }
+
+                // Decode directly
+                let mut output = Cursor::new(Vec::with_capacity(fi.size as usize));
+                let (written, err) = erasure.decode(
+                    &mut output,
+                    readers,
+                    0,
+                    fi.size as usize,
+                    fi.size as usize,
+                ).await;
+
+                if let Some(e) = err {
+                    return Err(to_object_err(e.into(), vec![bucket, object]));
+                }
+
+                rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_INLINE_DIRECT);
+                let reader = GetObjectReader {
+                    stream: Box::new(Cursor::new(output.into_inner())),
+                    object_info,
+                };
+                return Ok(reader);
+            }
+        }
+
         let codec_streaming_gate = get_codec_streaming_reader_gate(&range, &object_info, &fi, lock_optimization_enabled);
 
         if object_info.is_remote() {