From 799c4952261812d965b317c3bfb57addb63c94ed Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Thu, 17 Sep 2026 21:47:38 +0300 Subject: [PATCH] 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 * 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 --- .../src/storage/erasure_coding/ec_bitrot.rs | 30 +--- .../src/storage/erasure_coding/ec_encoder.rs | 14 +- .../src/storage/erasure_coding/ec_shard.rs | 16 +- .../src/storage/erasure_coding/ec_volume.rs | 66 +------- seaweed-volume/src/storage/io.rs | 142 ++++++++++++++++++ seaweed-volume/src/storage/mod.rs | 1 + .../src/storage/needle_map/sorted_file.rs | 27 +--- seaweed-volume/src/storage/volume.rs | 70 +-------- 8 files changed, 164 insertions(+), 202 deletions(-) create mode 100644 seaweed-volume/src/storage/io.rs diff --git a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs index af44045ed..d6afaa5e7 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs +++ b/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 { diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index db132ae5e..ded50f044 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/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 { 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; } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs b/seaweed-volume/src/storage/erasure_coding/ec_shard.rs index 538fba8e8..14195a7da 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs +++ b/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). diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index dd3b0148a..4e37772d1 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/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(()) -} diff --git a/seaweed-volume/src/storage/io.rs b/seaweed-volume/src/storage/io.rs new file mode 100644 index 000000000..93a29a326 --- /dev/null +++ b/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 { + #[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); + } +} diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index 32d1274be..2a62e55f4 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/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; diff --git a/seaweed-volume/src/storage/needle_map/sorted_file.rs b/seaweed-volume/src/storage/needle_map/sorted_file.rs index 6f1f9b4a6..5b2695bd4 100644 --- a/seaweed-volume/src/storage/needle_map/sorted_file.rs +++ b/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)] { diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index a2668604b..00f8f3cfe 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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)?;