perf(ecstore): fold metadata read open+fstat+read into a single spawn_blocking (HP-12 item 1) (#4554)

Sub-change A only: the two local-disk metadata read paths
(`read_metadata_with_dmtime`, `read_all_data_with_dmtime` in
crates/ecstore/src/disk/local.rs) previously dispatched open, fstat and the
xl.meta read as three separate async fs hops (three spawn_blocking round-trips
under tokio::fs). Each is now folded into a single tokio::task::spawn_blocking
closure using std::fs, cutting per-metadata-read dispatch from 3 to 1.

This does NOT touch sub-change B (removing the entry-point access() call).

Correctness first: this is the hottest metadata read path, so the refactor
preserves byte-for-byte-equivalent Results. Every error mapping is kept
identical:
  - open failure     -> to_file_error
  - is_dir           -> Error::FileNotFound (not to_file_error(EISDIR))
  - metadata failure -> to_file_error
  - xl.meta parse    -> propagated verbatim from the parser
  - try_reserve      -> Error::other
  - read_to_end      -> to_file_error
For read_all_data_with_dmtime the async NotFound -> access(volume_dir) ->
VolumeNotFound fallback (and its warn! event) is preserved on the async side:
the closure returns the raw open error unmapped; only open() can yield ENOENT
once the fd is valid, so gating the fallback on the open error is equivalent to
the original open-arm-only fallback.

To run the parser inside a blocking closure, filemeta gains a synchronous twin
`read_xl_meta_no_data_sync` (+ `read_more_sync`) in
crates/filemeta/src/filemeta/version.rs that line-for-line mirrors the async
version, differing only in std vs tokio read_exact. A new equivalence test
(`read_xl_meta_sync_equivalence_tests`) feeds identical buffers to both the
async and sync readers and asserts equal Ok bytes / equal Err variants across:
v1.0; v1.1/v1.2/v1.3; large meta triggering read_more; header truncation ->
UnexpectedEof; CRC-trailer truncation -> FileCorrupt; unknown major/minor ->
InvalidData; and want boundaries (exact fit, inline-data drop, read_more EOF).

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

crates/ecstore/src/disk/local.rs

@@ -39,7 +39,7 @@ use metrics::counter;
 use parking_lot::{Mutex as ParkingLotMutex, RwLock as ParkingLotRwLock};
 use rustfs_filemeta::{
     Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn,
-    get_file_info, read_xl_meta_no_data,
+    get_file_info, read_xl_meta_no_data_sync,
 };
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::os::get_info;
@@ -62,7 +62,7 @@ use std::{
 };
 use time::OffsetDateTime;
 use tokio::fs::{self, File};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf};
+use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf};
 use tokio::sync::{Notify, RwLock};
 use tokio::time::{Instant, Sleep, interval_at, timeout};
 use tracing::{debug, error, info, warn};
@@ -3140,25 +3140,41 @@ impl LocalDisk {
     async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
 
-        let mut f = super::fs::open_file(file_path.as_ref(), O_RDONLY)
-            .await
-            .map_err(to_file_error)?;
+        // HP-12 item 1 (sub-change A): fold the open + fstat + bounded xl.meta
+        // read into a single spawn_blocking dispatch instead of three separate
+        // async fs hops. The closure mirrors the previous async body one-to-one,
+        // including every error mapping, so the returned Result stays
+        // byte-for-byte equivalent (see read_xl_meta_no_data_sync equivalence
+        // tests in rustfs-filemeta):
+        //  - open failure     -> to_file_error (was fs::open_file(..).map_err(to_file_error))
+        //  - is_dir           -> Error::FileNotFound (NOT to_file_error(EISDIR))
+        //  - metadata failure -> to_file_error
+        //  - parse failure    -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
+        let path = file_path.as_ref().to_path_buf();
+        let (data, modtime) = tokio::task::spawn_blocking(move || -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
+            // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
+            let mut f = std::fs::File::open(&path).map_err(to_file_error)?;
 
-        let meta = f.metadata().await.map_err(to_file_error)?;
+            let meta = f.metadata().map_err(to_file_error)?;
 
-        if meta.is_dir() {
-            // fix use io::Error
-            return Err(Error::FileNotFound);
-        }
+            if meta.is_dir() {
+                // fix use io::Error
+                return Err(Error::FileNotFound);
+            }
 
-        let size = meta.len() as usize;
+            let size = meta.len() as usize;
 
-        let data = read_xl_meta_no_data(&mut f, size).await?;
+            let data = read_xl_meta_no_data_sync(&mut f, size)?;
 
-        let modtime = match meta.modified() {
-            Ok(md) => Some(OffsetDateTime::from(md)),
-            Err(_) => None,
-        };
+            let modtime = match meta.modified() {
+                Ok(md) => Some(OffsetDateTime::from(md)),
+                Err(_) => None,
+            };
+
+            Ok((data, modtime))
+        })
+        .await
+        .map_err(DiskError::from)??;
 
         Ok((data, modtime))
     }
@@ -3177,9 +3193,60 @@ impl LocalDisk {
         volume_dir: impl AsRef<Path>,
         file_path: impl AsRef<Path>,
     ) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
-        let mut f = match super::fs::open_file(file_path.as_ref(), O_RDONLY).await {
-            Ok(f) => f,
-            Err(e) => {
+        // HP-12 item 1 (sub-change A): fold open + fstat + is_dir + try_reserve +
+        // read_to_end into a single spawn_blocking dispatch. The closure mirrors
+        // the previous async body one-to-one so the returned Result stays
+        // byte-for-byte equivalent. Post-open errors are mapped to their final
+        // DiskError inside the closure (metadata/read -> to_file_error; is_dir ->
+        // FileNotFound; try_reserve -> Error::other) exactly as before. The raw
+        // open error is carried out unmapped so the async side can preserve the
+        // original NotFound -> access(volume_dir) -> VolumeNotFound fallback.
+        //
+        // Only the open() call can yield ErrorKind::NotFound here: once open
+        // succeeds the fd is valid, so fstat/read never return ENOENT. Hence
+        // gating the volume fallback on the open error alone is equivalent to
+        // the original code, where the fallback lived solely in the open match arm.
+        enum ReadAllError {
+            /// Raw, unmapped open() error; the async side decides the fallback.
+            Open(std::io::Error),
+            /// Already-mapped post-open error.
+            Disk(DiskError),
+        }
+
+        let path = file_path.as_ref().to_path_buf();
+        let res =
+            tokio::task::spawn_blocking(move || -> core::result::Result<(Vec<u8>, Option<OffsetDateTime>), ReadAllError> {
+                // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
+                let mut f = std::fs::File::open(&path).map_err(ReadAllError::Open)?;
+
+                let meta = f.metadata().map_err(|e| ReadAllError::Disk(to_file_error(e).into()))?;
+
+                if meta.is_dir() {
+                    return Err(ReadAllError::Disk(DiskError::FileNotFound));
+                }
+
+                let size = meta.len() as usize;
+                let mut bytes = Vec::new();
+                bytes
+                    .try_reserve_exact(size)
+                    .map_err(|e| ReadAllError::Disk(Error::other(e)))?;
+
+                std::io::Read::read_to_end(&mut f, &mut bytes).map_err(|e| ReadAllError::Disk(to_file_error(e).into()))?;
+
+                let modtime = match meta.modified() {
+                    Ok(md) => Some(OffsetDateTime::from(md)),
+                    Err(_) => None,
+                };
+
+                Ok((bytes, modtime))
+            })
+            .await
+            .map_err(DiskError::from)?;
+
+        let (bytes, modtime) = match res {
+            Ok(v) => v,
+            Err(ReadAllError::Disk(e)) => return Err(e),
+            Err(ReadAllError::Open(e)) => {
                 if e.kind() == ErrorKind::NotFound
                     && !skip_access_checks(volume)
                     && let Err(er) = access(volume_dir.as_ref()).await
@@ -3200,23 +3267,6 @@ impl LocalDisk {
             }
         };
 
-        let meta = f.metadata().await.map_err(to_file_error)?;
-
-        if meta.is_dir() {
-            return Err(DiskError::FileNotFound);
-        }
-
-        let size = meta.len() as usize;
-        let mut bytes = Vec::new();
-        bytes.try_reserve_exact(size).map_err(Error::other)?;
-
-        f.read_to_end(&mut bytes).await.map_err(to_file_error)?;
-
-        let modtime = match meta.modified() {
-            Ok(md) => Some(OffsetDateTime::from(md)),
-            Err(_) => None,
-        };
-
         Ok((bytes, modtime))
     }
 

crates/filemeta/src/filemeta/version.rs

@@ -3198,6 +3198,113 @@ pub async fn read_xl_meta_no_data<R: AsyncRead + Unpin>(reader: &mut R, size: us
     }
 }
 
+/// Synchronous twin of [`read_more`].
+///
+/// Line-for-line mirror of the async version; the only difference is that
+/// `reader.read_exact(&mut buf[has..]).await?` becomes the blocking
+/// `std::io::Read::read_exact`. std's `read_exact` reports the same
+/// `ErrorKind::UnexpectedEof` on a short read as tokio's, so the `?`
+/// conversion into [`Error`] is identical. Kept in lockstep with `read_more`
+/// so the two paths stay byte-for-byte equivalent (see equivalence tests).
+fn read_more_sync<R: std::io::Read>(
+    reader: &mut R,
+    buf: &mut Vec<u8>,
+    total_size: usize,
+    read_size: usize,
+    has_full: bool,
+) -> Result<()> {
+    let has = buf.len();
+
+    if has >= read_size {
+        return Ok(());
+    }
+
+    if has_full || read_size > total_size {
+        return Err(Error::other(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "Unexpected EOF")));
+    }
+
+    let extra = read_size - has;
+    if buf.capacity() >= read_size {
+        // Extend the buffer if we have enough space.
+        buf.resize(read_size, 0);
+    } else {
+        buf.extend(vec![0u8; extra]);
+    }
+
+    reader.read_exact(&mut buf[has..])?;
+    Ok(())
+}
+
+/// Synchronous twin of [`read_xl_meta_no_data`].
+///
+/// Byte-for-byte equivalent to the async version for any input: the parsing
+/// logic (`check_xl2_v1` / `read_bytes_header` / major-minor branches / `want`
+/// computation / 5-byte CRC trailer handling / `truncate` / `FileCorrupt` /
+/// `InvalidData` / `UnexpectedEof`) is copied verbatim; only the reads switch
+/// from async `read_exact(...).await` to blocking `std::io::Read::read_exact`.
+/// This lets a caller fold open+fstat+read into a single `spawn_blocking`
+/// closure without changing any observable result.
+pub fn read_xl_meta_no_data_sync<R: std::io::Read>(reader: &mut R, size: usize) -> Result<Vec<u8>> {
+    let mut initial = size;
+    let mut has_full = true;
+
+    if initial > META_DATA_READ_DEFAULT {
+        initial = META_DATA_READ_DEFAULT;
+        has_full = false;
+    }
+
+    let mut buf = vec![0u8; initial];
+    reader.read_exact(&mut buf)?;
+
+    let (tmp_buf, major, minor) = FileMeta::check_xl2_v1(&buf)?;
+
+    match major {
+        1 => match minor {
+            0 => {
+                read_more_sync(reader, &mut buf, size, size, has_full)?;
+                Ok(buf)
+            }
+            1..=3 => {
+                let (sz, tmp_buf) = FileMeta::read_bytes_header(tmp_buf)?;
+                let mut want = sz as usize + (buf.len() - tmp_buf.len());
+
+                if minor < 2 {
+                    read_more_sync(reader, &mut buf, size, want, has_full)?;
+                    buf.truncate(want);
+                    return Ok(buf);
+                }
+
+                let want_max = usize::min(want + MSGP_UINT32_SIZE, size);
+                read_more_sync(reader, &mut buf, size, want_max, has_full)?;
+
+                if buf.len() < want {
+                    return Err(Error::FileCorrupt);
+                }
+
+                // The metadata block is followed by a 5-byte msgp uint32 CRC trailer;
+                // a file truncated inside the trailer is corrupt, not a shorter meta.
+                let crc_size = 5;
+                if buf.len() - want < crc_size {
+                    return Err(Error::FileCorrupt);
+                }
+
+                want += crc_size;
+
+                buf.truncate(want);
+                Ok(buf)
+            }
+            _ => Err(Error::other(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "Unknown minor metadata version",
+            ))),
+        },
+        _ => Err(Error::other(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "Unknown major metadata version",
+        ))),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -4380,3 +4487,183 @@ mod tests {
         assert!(decoded.mod_time.is_none(), "absent mod_time must not become Some(UNIX_EPOCH)");
     }
 }
+
+/// Equivalence tests proving `read_xl_meta_no_data_sync` returns byte-for-byte
+/// identical `Result`s to the async `read_xl_meta_no_data` for every code path.
+///
+/// This is the critical guard for HP-12 item 1: the ecstore metadata read path
+/// folds open+fstat+read into one `spawn_blocking` closure that calls the sync
+/// twin, and it MUST NOT drift from the async version in either the Ok bytes or
+/// the Err variant.
+#[cfg(test)]
+mod read_xl_meta_sync_equivalence_tests {
+    use super::*;
+    use std::io::Cursor as StdCursor;
+    // tokio implements `AsyncRead` for `std::io::Cursor`, so the same type
+    // drives both the async and the sync readers.
+    use std::io::Cursor as TokioCursor;
+
+    /// 8-byte xl.meta header: "XL2 " magic + LE major + LE minor.
+    fn header(major: u16, minor: u16) -> Vec<u8> {
+        let mut v = Vec::with_capacity(8);
+        v.extend_from_slice(&XL_FILE_HEADER);
+        v.extend_from_slice(&major.to_le_bytes());
+        v.extend_from_slice(&minor.to_le_bytes());
+        v
+    }
+
+    /// 5-byte msgpack `bin32` length prefix (marker 0xc6 + 4-byte BE length),
+    /// matching what `read_bytes_header` decodes via `read_bin_len`.
+    fn bin32_prefix(len: u32) -> [u8; 5] {
+        let mut p = [0u8; 5];
+        p[0] = 0xc6;
+        p[1..].copy_from_slice(&len.to_be_bytes());
+        p
+    }
+
+    /// Build a v1.x xl.meta buffer: header + bin32(meta_len) + meta payload +
+    /// `crc_bytes` trailer bytes + `inline` trailing inline-data bytes. The meta
+    /// payload content is irrelevant: `read_xl_meta_no_data` never parses it.
+    fn build_v1x(minor: u16, meta_len: usize, crc_bytes: usize, inline: usize) -> Vec<u8> {
+        let mut v = header(1, minor);
+        v.extend_from_slice(&bin32_prefix(meta_len as u32));
+        v.extend(std::iter::repeat_n(0xABu8, meta_len));
+        v.extend(std::iter::repeat_n(0xC1u8, crc_bytes));
+        v.extend(std::iter::repeat_n(0xCDu8, inline));
+        v
+    }
+
+    async fn run_async(buf: &[u8], size: usize) -> Result<Vec<u8>> {
+        let mut r = TokioCursor::new(buf.to_vec());
+        read_xl_meta_no_data(&mut r, size).await
+    }
+
+    fn run_sync(buf: &[u8], size: usize) -> Result<Vec<u8>> {
+        let mut r = StdCursor::new(buf.to_vec());
+        read_xl_meta_no_data_sync(&mut r, size)
+    }
+
+    /// Drive both implementations with the same input and assert the results
+    /// are equivalent (Ok bytes equal, or Err variant equal per `Error`'s
+    /// `PartialEq`, which compares io kind + message for `Error::Io`).
+    async fn assert_equivalent(label: &str, buf: &[u8], size: usize) -> Result<Vec<u8>> {
+        let a = run_async(buf, size).await;
+        let s = run_sync(buf, size);
+        match (&a, &s) {
+            (Ok(ab), Ok(sb)) => assert_eq!(ab, sb, "[{label}] Ok bytes must match"),
+            (Err(ae), Err(se)) => assert_eq!(ae, se, "[{label}] Err variant must match"),
+            _ => panic!("[{label}] async/sync disagree on Ok vs Err: async={a:?} sync={s:?}"),
+        }
+        a
+    }
+
+    #[tokio::test]
+    async fn equivalence_across_all_paths() {
+        // (a) v1.0: whole-file read path.
+        {
+            let mut buf = header(1, 0);
+            buf.extend(std::iter::repeat_n(0x11u8, 24));
+            let len = buf.len();
+            let out = assert_equivalent("v1.0", &buf, len).await.unwrap();
+            assert_eq!(out, buf, "v1.0 returns the whole file");
+        }
+
+        // (b) v1.1 / v1.2 / v1.3, each a clean well-formed buffer.
+        // v1.1: minor < 2, no CRC trailer; returns header+prefix+meta.
+        {
+            let buf = build_v1x(1, 10, 0, 0);
+            let len = buf.len();
+            let out = assert_equivalent("v1.1", &buf, len).await.unwrap();
+            assert_eq!(out.len(), 8 + 5 + 10, "v1.1 -> header+prefix+meta");
+        }
+        // v1.2: minor >= 2, 5-byte CRC trailer retained.
+        {
+            let buf = build_v1x(2, 15, 5, 0);
+            let len = buf.len();
+            let out = assert_equivalent("v1.2", &buf, len).await.unwrap();
+            assert_eq!(out.len(), 8 + 5 + 15 + 5, "v1.2 -> header+prefix+meta+crc");
+        }
+        // v1.3: minor >= 2, 5-byte CRC trailer retained.
+        {
+            let buf = build_v1x(3, 20, 5, 0);
+            let len = buf.len();
+            let out = assert_equivalent("v1.3", &buf, len).await.unwrap();
+            assert_eq!(out.len(), 8 + 5 + 20 + 5, "v1.3 -> header+prefix+meta+crc");
+        }
+
+        // (c) size > META_DATA_READ_DEFAULT: initial 4KiB read then read_more.
+        {
+            let meta_len = 5000; // total = 8 + 5 + 5000 + 5 = 5018 > 4096
+            let buf = build_v1x(3, meta_len, 5, 0);
+            assert!(buf.len() > META_DATA_READ_DEFAULT);
+            let len = buf.len();
+            let out = assert_equivalent("v1.3-large", &buf, len).await.unwrap();
+            assert_eq!(out.len(), 8 + 5 + meta_len + 5);
+        }
+
+        // (d) truncated inside the header -> UnexpectedEof (initial read_exact
+        // fails because the reader holds fewer bytes than the claimed size).
+        {
+            let buf = XL_FILE_HEADER.to_vec(); // only 4 bytes present
+            assert_equivalent("truncated-header", &buf, 8).await.unwrap_err();
+        }
+
+        // (e) truncated inside the CRC trailer -> FileCorrupt (v1.3 with only
+        // 2 of the 5 trailer bytes present; size matches the buffer).
+        {
+            let buf = build_v1x(3, 10, 2, 0); // want=23, buf.len()=25, 25-23=2 < 5
+            let len = buf.len();
+            let e = assert_equivalent("truncated-crc", &buf, len).await.unwrap_err();
+            assert_eq!(e, Error::FileCorrupt);
+        }
+
+        // (f) unknown major / unknown minor -> InvalidData.
+        // Unknown major: major=0 passes check_xl2_v1 (0 <= MAJOR) but misses the
+        // `1 =>` arm.
+        {
+            let buf = header(0, 0);
+            let e = assert_equivalent("unknown-major", &buf, buf.len()).await.unwrap_err();
+            assert_eq!(
+                e,
+                Error::other(std::io::Error::new(std::io::ErrorKind::InvalidData, "Unknown major metadata version"))
+            );
+        }
+        // Unknown minor: major=1, minor=4 misses the `0` and `1..=3` arms.
+        {
+            let buf = header(1, 4);
+            let e = assert_equivalent("unknown-minor", &buf, buf.len()).await.unwrap_err();
+            assert_eq!(
+                e,
+                Error::other(std::io::Error::new(std::io::ErrorKind::InvalidData, "Unknown minor metadata version"))
+            );
+        }
+
+        // (g) want boundaries.
+        // Exact fit: size == want + crc, no inline data.
+        {
+            let buf = build_v1x(3, 20, 5, 0); // size == want+5 exactly
+            let len = buf.len();
+            let out = assert_equivalent("want-exact-fit", &buf, len).await.unwrap();
+            assert_eq!(out, buf);
+        }
+        // Inline data past the trailer: truncate drops it, want_max clamps below size.
+        {
+            let buf = build_v1x(3, 20, 5, 100); // size = want+5+100
+            let len = buf.len();
+            let out = assert_equivalent("want-with-inline", &buf, len).await.unwrap();
+            assert_eq!(out.len(), 8 + 5 + 20 + 5, "inline data is dropped by truncate");
+        }
+
+        // (h) read_more's own read_exact hits EOF (has_full=false, read_size
+        // within claimed size but the reader is physically shorter).
+        {
+            // prefix claims a 5000-byte meta so size=5018, but only 4100 bytes exist.
+            let mut buf = header(1, 3);
+            buf.extend_from_slice(&bin32_prefix(5000));
+            buf.extend(std::iter::repeat_n(0xABu8, 4100 - buf.len()));
+            assert_eq!(buf.len(), 4100);
+            let e = assert_equivalent("read_more-eof", &buf, 5018).await.unwrap_err();
+            assert_eq!(e, Error::Unexpected, "short read surfaces as Unexpected");
+        }
+    }
+}