rust volume: one positional read helper; never seek a dup'd handle on Windows (#11342)

* rust volume: one positional read helper; never seek a dup'd handle on Windows

Positional read-exact was hand-rolled four times: the complete
cross-platform version in needle_map/sorted_file.rs, a Windows-only half
in volume.rs whose unix half was inlined as a
cfg(unix)/cfg(windows)/compile_error! triple at three call sites, a
byte-identical Windows-only copy in ec_volume.rs, and read_full_at in
ec_bitrot.rs. Three more sites -- EcVolumeShard::read_at,
EcLocalShard::read_at and ec_encoder::read_at_most -- hand-rolled the
short-read-permitted variant with a cfg(not(unix)) arm that
try_clone()s the handle and seeks it.

That last arm is wrong. A duplicated descriptor shares one kernel file
offset with the original, so seek-then-read is two syscalls against
state another thread can move in between: a concurrent reader or an
append repositions the offset and the read returns bytes from somewhere
else entirely. EcLocalShard::read_at documents that it must never seek,
one line above the seek. Windows seek_read carries its own offset in a
single call, so that window does not exist.

All seven now go through storage::io::{read_exact_at, read_at}, whose
module doc records why duplicating a handle is not a way to get a
private file position -- opening the file again is, as
Volume::dat_scan_plan already does. read_at_most keeps its own
fill-until-EOF loop; only the per-iteration positional read changes.

Behaviour on unix is unchanged: every unix arm was already
FileExt::read_exact_at or FileExt::read_at. The one exception is
ec_bitrot::verify_shard_blocks, which now retries on EINTR (std's
read_exact_at does; the loop it replaces did not) and, on unix, reports
the standard "failed to fill whole buffer" text instead of "short read
on shard block". The Windows arm still says "unexpected EOF in
seek_read"; both carry ErrorKind::UnexpectedEof, as before.

NeedleStreamSource::read_exact_at and Volume::read_exact_at_backend keep
their signatures; only their bodies shrink.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* rust volume: retry Interrupted in Windows read_exact_at

Unix std's FileExt::read_exact_at ignores ErrorKind::Interrupted and
retries, but the Windows seek_read loop propagated it, so the shared
exact-read contract differed by platform. seek_read can surface
ERROR_OPERATION_ABORTED, which std maps to Interrupted.

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs

@@ -30,6 +30,7 @@ use crate::pb::volume_server_pb::{
     ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums, EcShardConfig,
 };
 use crate::storage::erasure_coding::ec_shard::MAX_SHARD_COUNT;
+use crate::storage::io::read_exact_at;
 use crate::storage::needle::crc::CRC;
 
 /// Canonical extension for the checksum sidecar. Generation 0 (legacy/fresh
@@ -537,7 +538,7 @@ pub fn verify_shard_blocks(
             break;
         }
         let to_read = to_read as usize;
-        read_full_at(f, &mut buf[..to_read], offset as u64)?;
+        read_exact_at(f, &mut buf[..to_read], offset as u64)?;
         if CRC::new(&buf[..to_read]).0 != *want_crc {
             mismatched.push(i);
         }
@@ -546,33 +547,6 @@ pub fn verify_shard_blocks(
     Ok(mismatched)
 }
 
-/// Reads exactly `buf.len()` bytes from `f` at `offset`, erroring on early EOF.
-fn read_full_at(f: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
-    let mut total = 0usize;
-    while total < buf.len() {
-        #[cfg(unix)]
-        let n = {
-            use std::os::unix::fs::FileExt;
-            f.read_at(&mut buf[total..], offset + total as u64)?
-        };
-        #[cfg(not(unix))]
-        let n = {
-            use std::io::{Read, Seek, SeekFrom};
-            let mut fc = f.try_clone()?;
-            fc.seek(SeekFrom::Start(offset + total as u64))?;
-            fc.read(&mut buf[total..])?
-        };
-        if n == 0 {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                "short read on shard block",
-            ));
-        }
-        total += n;
-    }
-    Ok(())
-}
-
 /// Builds the `EcShardConfig` proto for the given layout. The bitrot sidecar
 /// carries its own top-level encode_uuid, so the nested config leaves it empty.
 pub fn ec_shard_config(data_shards: u32, parity_shards: u32, block_size: i64) -> EcShardConfig {

seaweed-volume/src/storage/erasure_coding/ec_encoder.rs

@@ -5,8 +5,6 @@
 
 use std::fs::File;
 use std::io;
-#[cfg(not(unix))]
-use std::io::{Read, Seek, SeekFrom};
 
 use reed_solomon_erasure::galois_8::ReedSolomon;
 
@@ -840,17 +838,7 @@ impl EncodeRun<'_> {
 fn read_at_most(dat_file: &File, buf: &mut [u8], offset: u64) -> io::Result<usize> {
     let mut n = 0;
     while n < buf.len() {
-        #[cfg(unix)]
-        let r = {
-            use std::os::unix::fs::FileExt;
-            dat_file.read_at(&mut buf[n..], offset + n as u64)?
-        };
-        #[cfg(not(unix))]
-        let r = {
-            let mut f = dat_file.try_clone()?;
-            f.seek(SeekFrom::Start(offset + n as u64))?;
-            f.read(&mut buf[n..])?
-        };
+        let r = crate::storage::io::read_at(dat_file, &mut buf[n..], offset + n as u64)?;
         if r == 0 {
             break;
         }

seaweed-volume/src/storage/erasure_coding/ec_shard.rs

@@ -94,21 +94,7 @@ impl EcVolumeShard {
             .as_ref()
             .ok_or_else(|| io::Error::other("shard file not open"))?;
 
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::FileExt;
-            file.read_at(buf, offset)
-        }
-
-        #[cfg(not(unix))]
-        {
-            use std::io::{Read, Seek, SeekFrom};
-            // File::read_at is unix-only; fall back to seek + read.
-            // We need a mutable reference for seek/read, so clone the handle.
-            let mut f = file.try_clone()?;
-            f.seek(SeekFrom::Start(offset))?;
-            f.read(buf)
-        }
+        crate::storage::io::read_at(file, buf, offset)
     }
 
     /// Write data to the shard file (appends).

seaweed-volume/src/storage/erasure_coding/ec_volume.rs

@@ -12,6 +12,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use crate::pb::master_pb;
 use crate::storage::erasure_coding::ec_locate;
 use crate::storage::erasure_coding::ec_shard::*;
+use crate::storage::io::read_exact_at;
 use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size};
 use crate::storage::types::*;
 use crate::storage::volume_open::open_volume_file;
@@ -1068,28 +1069,12 @@ impl EcVolume {
             let mid = lo + (hi - lo) / 2;
             let file_offset = (mid * NEEDLE_MAP_ENTRY_SIZE) as u64;
 
-            #[cfg(unix)]
-            {
-                use std::os::unix::fs::FileExt;
-                if let Err(e) = ecx_file.read_exact_at(&mut entry_buf, file_offset) {
-                    self.check_read_write_error(Some(&e));
-                    return Err(e);
-                }
-            }
-            #[cfg(windows)]
-            {
-                // Positional read so concurrent find_needle_from_ecx calls on
-                // the shared .ecx handle don't interleave seek/read and corrupt
-                // each other's binary search. Mirrors the read_exact_at helper
-                // in storage::volume.
-                if let Err(e) = read_exact_at(ecx_file, &mut entry_buf, file_offset) {
-                    self.check_read_write_error(Some(&e));
-                    return Err(e);
-                }
-            }
-            #[cfg(not(any(unix, windows)))]
-            {
-                compile_error!("Platform not supported: only unix and windows are supported");
+            // Positional read so concurrent find_needle_from_ecx calls on the
+            // shared .ecx handle don't interleave a seek and a read and corrupt
+            // each other's binary search.
+            if let Err(e) = read_exact_at(ecx_file, &mut entry_buf, file_offset) {
+                self.check_read_write_error(Some(&e));
+                return Err(e);
             }
 
             let (key, offset, size) = idx_entry_from_bytes(&entry_buf);
@@ -4312,18 +4297,7 @@ impl EcLocalShard {
             .file
             .as_ref()
             .map_err(|e| io::Error::new(e.kind(), e.to_string()))?;
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::FileExt;
-            file.read_at(buf, offset)
-        }
-        #[cfg(not(unix))]
-        {
-            use std::io::{Read, Seek, SeekFrom};
-            let mut f = file.try_clone()?;
-            f.seek(SeekFrom::Start(offset))?;
-            f.read(buf)
-        }
+        crate::storage::io::read_at(file, buf, offset)
     }
 }
 
@@ -4619,27 +4593,3 @@ impl EcLocalScrubPlan {
         (count, broken, errs)
     }
 }
-
-/// Windows helper: loop `seek_read` until the buffer is fully filled.
-///
-/// `seek_read` is positional (it passes the offset through `OVERLAPPED` and
-/// never touches the shared file cursor), so concurrent callers reading the
-/// same `&File` — as `find_needle_from_ecx` does on the cached `.ecx` handle —
-/// can't interleave their reads. Mirrors the helper in `storage::volume`.
-#[cfg(windows)]
-fn read_exact_at(file: &File, buf: &mut [u8], mut offset: u64) -> io::Result<()> {
-    use std::os::windows::fs::FileExt;
-    let mut filled = 0;
-    while filled < buf.len() {
-        let n = file.seek_read(&mut buf[filled..], offset)?;
-        if n == 0 {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                "unexpected EOF in seek_read",
-            ));
-        }
-        filled += n;
-        offset += n as u64;
-    }
-    Ok(())
-}

seaweed-volume/src/storage/io.rs

@@ -0,0 +1,142 @@
+//! Positional file reads.
+//!
+//! Every read here is "these bytes at this offset", never "the next bytes".
+//! The handles are shared — `.dat` and `.idx` descriptors are borrowed from
+//! [`file_pool`](super::needle_map::file_pool), a mounted EC shard's handle is
+//! duplicated into a scrub plan — so no caller may rely on a file position.
+//!
+//! On unix that is `pread(2)` through `std::os::unix::fs::FileExt`. On Windows
+//! it is `seek_read`, which passes the offset through `OVERLAPPED`, so the read
+//! itself is independent of the current cursor.
+//!
+//! What these helpers replace is `try_clone()` + `seek()` + `read()`. A
+//! duplicated handle shares one kernel file offset with the original, so that
+//! sequence is two syscalls against state another thread can move in between:
+//! the seek positions the offset, a concurrent reader or an append moves it,
+//! and the read returns bytes from somewhere else entirely. `seek_read` carries
+//! its own offset in a single call, so there is no window.
+//!
+//! `seek_read` does still advance the cursor as a side effect — Windows updates
+//! the file pointer even for an `OVERLAPPED` read — which nothing here relies
+//! on. A caller that genuinely needs a private position must open the file
+//! again rather than duplicate a handle; see `Volume::dat_scan_plan` in
+//! [`storage::volume`](super::volume).
+
+use std::fs::File;
+use std::io;
+
+/// Reads exactly `buf.len()` bytes from `file` starting at `offset`.
+///
+/// Fails with [`io::ErrorKind::UnexpectedEof`] if the file ends first.
+pub(crate) fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::FileExt;
+        file.read_exact_at(buf, offset)?;
+    }
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::FileExt;
+        let mut filled = 0;
+        let mut at = offset;
+        while filled < buf.len() {
+            let n = match file.seek_read(&mut buf[filled..], at) {
+                Ok(n) => n,
+                Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
+                Err(err) => return Err(err),
+            };
+            if n == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "unexpected EOF in seek_read",
+                ));
+            }
+            filled += n;
+            at += n as u64;
+        }
+    }
+    #[cfg(not(any(unix, windows)))]
+    {
+        compile_error!("Platform not supported: only unix and windows are supported");
+    }
+    Ok(())
+}
+
+/// Reads up to `buf.len()` bytes from `file` starting at `offset`, returning
+/// how many were read.
+///
+/// A short read — including `0` at or past end of file — is not an error; use
+/// [`read_exact_at`] when the whole buffer must be filled.
+pub(crate) fn read_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<usize> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::FileExt;
+        file.read_at(buf, offset)
+    }
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::FileExt;
+        file.seek_read(buf, offset)
+    }
+    #[cfg(not(any(unix, windows)))]
+    {
+        compile_error!("Platform not supported: only unix and windows are supported");
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{read_at, read_exact_at};
+    use std::io::{ErrorKind, Write};
+
+    fn temp_file(bytes: &[u8]) -> tempfile::NamedTempFile {
+        let mut f = tempfile::NamedTempFile::new().expect("temp file");
+        f.write_all(bytes).expect("write");
+        f.flush().expect("flush");
+        f
+    }
+
+    #[test]
+    fn read_exact_at_fills_the_whole_buffer() {
+        let f = temp_file(b"0123456789");
+        let mut buf = [0u8; 10];
+        read_exact_at(f.as_file(), &mut buf, 0).expect("read");
+        assert_eq!(&buf, b"0123456789");
+    }
+
+    #[test]
+    fn read_exact_at_reads_from_the_offset() {
+        let f = temp_file(b"0123456789");
+        let mut buf = [0u8; 4];
+        read_exact_at(f.as_file(), &mut buf, 3).expect("read");
+        assert_eq!(&buf, b"3456");
+
+        // The helper is positional: a second read at a lower offset sees the
+        // bytes at that offset, not wherever the first read left a cursor.
+        let mut again = [0u8; 4];
+        read_exact_at(f.as_file(), &mut again, 1).expect("read");
+        assert_eq!(&again, b"1234");
+    }
+
+    #[test]
+    fn read_exact_at_short_file_is_unexpected_eof() {
+        let f = temp_file(b"0123");
+        let mut buf = [0u8; 8];
+        let err = read_exact_at(f.as_file(), &mut buf, 0).expect_err("short file");
+        assert_eq!(err.kind(), ErrorKind::UnexpectedEof);
+    }
+
+    #[test]
+    fn read_at_allows_a_short_read_at_eof() {
+        let f = temp_file(b"0123456789");
+        let mut buf = [0u8; 8];
+
+        let n = read_at(f.as_file(), &mut buf, 6).expect("read");
+        assert_eq!(n, 4);
+        assert_eq!(&buf[..n], b"6789");
+
+        // Entirely past the end is zero bytes, not an error.
+        let n = read_at(f.as_file(), &mut buf, 10).expect("read");
+        assert_eq!(n, 0);
+    }
+}

seaweed-volume/src/storage/mod.rs

@@ -1,6 +1,7 @@
 pub mod disk_location;
 pub mod erasure_coding;
 pub mod idx;
+pub(crate) mod io;
 pub mod needle;
 pub mod needle_map;
 pub mod store;

seaweed-volume/src/storage/needle_map/sorted_file.rs

@@ -18,6 +18,7 @@ use std::sync::{Mutex, RwLock};
 
 use super::file_pool::pooled_index_files;
 use crate::storage::idx;
+use crate::storage::io::read_exact_at;
 use crate::storage::needle_map::{CompactNeedleMap, NeedleMapMetric, NeedleValue};
 use crate::storage::types::*;
 
@@ -520,32 +521,6 @@ fn search_sorted_index(
     Ok(None)
 }
 
-fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
-    #[cfg(unix)]
-    {
-        use std::os::unix::fs::FileExt;
-        file.read_exact_at(buf, offset)
-    }
-    #[cfg(windows)]
-    {
-        use std::os::windows::fs::FileExt;
-        let mut filled = 0;
-        let mut at = offset;
-        while filled < buf.len() {
-            let n = file.seek_read(&mut buf[filled..], at)?;
-            if n == 0 {
-                return Err(io::Error::new(
-                    io::ErrorKind::UnexpectedEof,
-                    "unexpected EOF in seek_read",
-                ));
-            }
-            filled += n;
-            at += n as u64;
-        }
-        Ok(())
-    }
-}
-
 fn write_at(file: &File, buf: &[u8], offset: u64) -> io::Result<()> {
     #[cfg(unix)]
     {

seaweed-volume/src/storage/volume.rs

@@ -21,6 +21,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use tracing::{error, info, warn};
 
 use crate::storage::idx;
+use crate::storage::io::read_exact_at;
 use crate::storage::needle::needle::{self, Needle, NeedleError, get_actual_size};
 use crate::storage::needle_map::sorted_file::SortedFileNeedleMap;
 use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap};
@@ -455,22 +456,7 @@ impl NeedleStreamSource {
 
     pub(crate) fn read_exact_at(&self, buf: &mut [u8], offset: u64) -> io::Result<()> {
         match self {
-            NeedleStreamSource::Local(file) => {
-                #[cfg(unix)]
-                {
-                    use std::os::unix::fs::FileExt;
-                    file.read_exact_at(buf, offset)?;
-                }
-                #[cfg(windows)]
-                {
-                    read_exact_at(file, buf, offset)?;
-                }
-                #[cfg(not(any(unix, windows)))]
-                {
-                    compile_error!("Platform not supported: only unix and windows are supported");
-                }
-                Ok(())
-            }
+            NeedleStreamSource::Local(file) => read_exact_at(file, buf, offset),
             NeedleStreamSource::Remote(remote) => remote.read_exact_at(buf, offset),
         }
     }
@@ -722,25 +708,6 @@ pub struct Volume {
     pub has_remote_file: bool,
 }
 
-/// Windows helper: loop seek_read until buffer is fully filled.
-#[cfg(windows)]
-fn read_exact_at(file: &File, buf: &mut [u8], mut offset: u64) -> io::Result<()> {
-    use std::os::windows::fs::FileExt;
-    let mut filled = 0;
-    while filled < buf.len() {
-        let n = file.seek_read(&mut buf[filled..], offset)?;
-        if n == 0 {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                "unexpected EOF in seek_read",
-            ));
-        }
-        filled += n;
-        offset += n as u64;
-    }
-    Ok(())
-}
-
 /// What a volume is created with beyond its id, directories and index kind:
 /// the tail of Go's `NewVolume` argument list. The default is an empty
 /// collection with no replication, no TTL and no preallocation at the
@@ -1349,19 +1316,7 @@ impl Volume {
         offset: u64,
     ) -> Result<(), VolumeError> {
         if let Some(dat_file) = self.dat_file.as_ref() {
-            #[cfg(unix)]
-            {
-                use std::os::unix::fs::FileExt;
-                dat_file.read_exact_at(buf, offset)?;
-            }
-            #[cfg(windows)]
-            {
-                read_exact_at(dat_file, buf, offset)?;
-            }
-            #[cfg(not(any(unix, windows)))]
-            {
-                compile_error!("Platform not supported: only unix and windows are supported");
-            }
+            read_exact_at(dat_file, buf, offset)?;
             Ok(())
         } else if let Some(remote_dat_file) = self.remote_dat_file.as_ref() {
             remote_dat_file.read_exact_at(buf, offset)?;
@@ -4142,20 +4097,11 @@ impl Volume {
                 let actual_size = crate::storage::needle::needle::get_actual_size(size, version);
                 let mut blob = vec![0u8; actual_size as usize];
 
-                #[cfg(unix)]
-                {
-                    use std::os::unix::fs::FileExt;
-                    old_dat_file
-                        .read_exact_at(&mut blob, needle_offset.to_actual_offset() as u64)?;
-                }
-                #[cfg(windows)]
-                {
-                    crate::storage::volume::read_exact_at(
-                        &old_dat_file,
-                        &mut blob,
-                        needle_offset.to_actual_offset() as u64,
-                    )?;
-                }
+                read_exact_at(
+                    &old_dat_file,
+                    &mut blob,
+                    needle_offset.to_actual_offset() as u64,
+                )?;
 
                 dst_dat.write_all(&blob)?;