From 985706cc32b27b04e7cce1e6fcd76d6944c1bc4e Mon Sep 17 00:00:00 2001 From: Justin Cichra <1342149+jrcichra@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:53:07 -0500 Subject: [PATCH] volume: check a needle's CRC the first time a range of it is read (#12467) * volume: check a needle's CRC the first time a range of it is read A ranged read never checked the CRC, so a replica with a damaged needle served the damaged bytes with 206 and the reader never tried another copy. The first ranged read of a needle now streams the whole needle through the existing CRC check, with the usual pooled page buffers, before writing any byte. A mismatch fails the read, so the reader falls back to another replica. Needles that pass go into a bounded LRU, keyed by volume, compaction revision, id, append time and checksum, so later ranges of the same needle stream as before. Fixes #5959 for ranged reads. Co-Authored-By: Claude Opus 5.5 * volume: key verified ranges by volume load, share concurrent checks From review: the cache key read v.SuperBlock.CompactionRevision without the data file lock, racing a compaction commit that reloads the super block, and a remounted copy of a volume inherited the old copy's verified entries. Key entries by a load generation instead, an atomic counter that changes on every load, including the reload in CommitCompact, and cache a result only if no reload happened during the check. Concurrent first ranges of a needle now share one check through singleflight instead of each reading the whole needle. Co-Authored-By: Claude Opus 5.5 * volume: keep hot needles verified, leave the io error streak to the range read From review: Contains does not refresh an entry's place in the LRU, so even a constantly read needle was evicted after 64k other needles were verified; look it up with Get instead. The verification pass also cleared the io error streak before the range read ran, so a range read failing with EIO after a passing check never built up a streak; the pass into io.Discard now leaves the streak to the range read, like the meta read before a streamed GET. Co-Authored-By: Claude Opus 5.5 * volume (Rust): check a needle's CRC the first time a range of it is read Mirror of the Go change: a range served from the data file now checks the whole needle's CRC before any byte of it is sent, so a damaged replica answers 500 and the reader tries another copy. Needles that pass go into a bounded LRU keyed by the volume's load generation, needle id, append time and checksum; concurrent first ranges of a needle share one check, and a check that passes leaves the io error streak to the range read. Co-Authored-By: Claude Opus 5.5 * volume (Rust): a range over the whole needle checks its CRC Go reads a range that covers the whole needle through the checked full-read path. The Rust volume server read it from the data file unchecked, so "bytes=0-" served a damaged needle with 206. Check the slice's CRC before answering, as the full read does. Co-Authored-By: Claude Opus 5.5 * volume: check the covered blocks of every range against CRCs recorded at the first check The first ranged read of a needle now records the CRC of each block (64 KiB to 1 MiB, at most 64 blocks below 64 MiB) while it checks the whole needle. Later ranges read the block-aligned span around them and write a block's bytes only once it matches, so damage that appears after the first check is not served. Co-Authored-By: Claude Opus 5.5 * volume (Rust): check the covered blocks of every range against CRCs recorded at the first check verify_checksum now records a CRC per block (same sizes as Go's rangeCheckBlockSize) and returns them from the cache; read_checked_range reads the block-aligned span of a range and answers 500 if a block no longer matches, so damage after the first check is not served. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Chris Lu --- seaweed-volume/Cargo.lock | 1 + seaweed-volume/Cargo.toml | 1 + seaweed-volume/src/server/handlers.rs | 188 ++++++++++++++++++++- seaweed-volume/src/storage/volume.rs | 232 ++++++++++++++++++++++++++ weed/storage/volume.go | 3 +- weed/storage/volume_loading.go | 1 + weed/storage/volume_read.go | 189 ++++++++++++++++++++- weed/storage/volume_read_test.go | 185 ++++++++++++++++++++ 8 files changed, 788 insertions(+), 12 deletions(-) diff --git a/seaweed-volume/Cargo.lock b/seaweed-volume/Cargo.lock index 8a21b5077a..6112f8540f 100644 --- a/seaweed-volume/Cargo.lock +++ b/seaweed-volume/Cargo.lock @@ -4608,6 +4608,7 @@ dependencies = [ "jsonwebtoken", "kamadak-exif", "libc", + "lru", "md-5", "memmap2", "mime_guess", diff --git a/seaweed-volume/Cargo.toml b/seaweed-volume/Cargo.toml index 960b256e1d..aaa4dd837a 100644 --- a/seaweed-volume/Cargo.toml +++ b/seaweed-volume/Cargo.toml @@ -130,6 +130,7 @@ chrono = "0.4" hex = "0.4" parking_lot = "0.12" dashmap = "6" +lru = "0.16" thiserror = "1" anyhow = "1" async-trait = "0.1" diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index df5599a970..883b296b8f 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -2008,8 +2008,9 @@ async fn range_from_source_response( ) -> Response { let range_str = range_str.to_string(); let tracking = track_download.then(|| state.clone()); + let chunk_size = streaming_chunk_size(state.read_buffer_size_bytes, info.data_size as usize); tokio::task::spawn_blocking(move || { - handle_range_request_from_source(&range_str, info, response_headers, tracking) + handle_range_request_from_source(&range_str, info, response_headers, tracking, chunk_size) }) .await .unwrap_or_else(|e| { @@ -2353,6 +2354,7 @@ fn handle_range_request_from_source( info: crate::storage::volume::NeedleStreamInfo, mut headers: HeaderMap, state: Option>, + chunk_size: usize, ) -> Response { let total = info.data_size as i64; let ranges = match parse_range_header(range_str, total) { @@ -2376,21 +2378,57 @@ fn handle_range_request_from_source( return (StatusCode::OK, headers).into_response(); } + let block_crcs = if ranges.iter().any(|r| r.start != 0 || r.length != total) { + match info.verify_checksum(chunk_size) { + Ok(crcs) => Some(crcs), + Err(err) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("range read error: {}", err), + ) + .into_response(); + } + } + } else { + None + }; + let read_slice = |start: i64, length: i64| -> Result, std::io::Error> { if length <= 0 { return Ok(Vec::new()); } - let mut buf = vec![0u8; length as usize]; - match info - .source - .read_exact_at(&mut buf, info.data_file_offset + start as u64) - { + let read = match &block_crcs { + Some(crcs) if length != total => { + info.read_checked_range(crcs, start as u64, length as u64) + } + _ => { + let mut buf = vec![0u8; length as usize]; + info.source + .read_exact_at(&mut buf, info.data_file_offset + start as u64) + .map(|()| buf) + } + }; + match read { + Ok(buf) if length == total && !info.checksum_matches(&buf) => { + metrics::HANDLER_COUNTER + .with_label_values(&[metrics::ERROR_CRC]) + .inc(); + Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "needle data checksum mismatch for needle {}", + info.needle_id + ), + )) + } // A finished range read clears the streak a failed meta read // counted, like readNeedleDataInto's terminal report. - Ok(()) => { + Ok(buf) => { info.io_errors.check_read_write_error(None); Ok(buf) } + // A block that fails its CRC leaves the streak alone, as Go's. + Err(e) if e.kind() == std::io::ErrorKind::InvalidData => Err(e), Err(e) => { info.io_errors.check_read_write_error(Some(&e)); Err(e) @@ -7013,13 +7051,145 @@ mod tests { io_unavailable: Arc::default(), io_errors: io_errors.clone(), needle_id: NeedleId(1), - checksum: 0, + checksum: CRC::new(&data).0, + append_at_ns: 0, + load_generation: u64::MAX, }; - let resp = handle_range_request_from_source("bytes=10-19", info, HeaderMap::new(), None); + // The checksum pass leaves the streak to the range read. + info.verify_checksum(64).unwrap(); + assert_eq!(io_errors.get_io_error_state().1, 2); + let resp = + handle_range_request_from_source("bytes=10-19", info, HeaderMap::new(), None, 64); assert_eq!(resp.status(), StatusCode::PARTIAL_CONTENT); assert_eq!(io_errors.get_io_error_state().1, 0); } + /// A range carries no checksum of its own, so the first ranged read of a + /// needle checks the whole needle's CRC and sends none of the range on a + /// mismatch; the reader can then fall back to another replica. A verified + /// needle is checked again once the volume reloads. + #[tokio::test] + async fn test_range_read_checks_the_needle_checksum() { + let tmp = tempfile::TempDir::new().unwrap(); + let state = volume_test_state(&tmp); + let payload = + |seed: u32| -> Vec { (0..3000u32).map(|i| ((i * 7 + seed) % 251) as u8).collect() }; + let damage = |data: &[u8]| { + let dat = tmp.path().join("1.dat"); + let mut bytes = std::fs::read(&dat).unwrap(); + let at = bytes.windows(data.len()).position(|w| w == data).unwrap(); + bytes[at + 10] ^= 0xff; + std::fs::write(&dat, bytes).unwrap(); + }; + + let intact = payload(1); + let path = put_test_needle(&state, 0x6e7a_0c01, &intact); + let (status, _, body) = + send_read(&state, Method::GET, &path, Some(b"bytes=1000-1499")).await; + assert_eq!(status, StatusCode::PARTIAL_CONTENT); + assert_eq!(body, &intact[1000..1500]); + + // Damage a byte outside the range: the range itself is intact, but + // the needle is not, so this replica must not serve it. + let damaged = payload(2); + let path = put_test_needle(&state, 0x6e7a_0c02, &damaged); + damage(&damaged); + for range in [&b"bytes=1000-1499"[..], b"bytes=0-", b"bytes=0-1,1000-1001"] { + let (status, _, body) = send_read(&state, Method::GET, &path, Some(range)).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{range:?}"); + assert!( + !body.windows(2).any(|w| w == &damaged[1000..1002]), + "{range:?}" + ); + } + + // Concurrent first ranges of a needle all see the outcome of the check. + for (seed, damage_it) in [(3, false), (4, true)] { + let data = payload(seed); + let path = put_test_needle(&state, 0x6e7a_0c00 + seed as u64, &data); + if damage_it { + damage(&data); + } + let reads = (0..8).map(|_| send_read(&state, Method::GET, &path, Some(b"bytes=10-19"))); + for (status, _, _) in futures::future::join_all(reads).await { + assert_eq!(status == StatusCode::PARTIAL_CONTENT, !damage_it, "{seed}"); + } + } + + // A verified needle is checked again after a reload, as a compaction + // commit does; until then later ranges skip the check. + let data = payload(5); + let path = put_test_needle(&state, 0x6e7a_0c05, &data); + let (status, _, _) = send_read(&state, Method::GET, &path, Some(b"bytes=10-19")).await; + assert_eq!(status, StatusCode::PARTIAL_CONTENT); + let (_, v) = state + .store + .read() + .unwrap() + .find_volume(VolumeId(1)) + .unwrap(); + v.compact_by_index(0, 0, |_| true).unwrap(); + v.commit_compact().unwrap(); + damage(&data); + let (status, _, _) = send_read(&state, Method::GET, &path, Some(b"bytes=10-19")).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + + // Damage that appears after a needle passed its check is caught by + // the block it falls in, while ranges over intact blocks still read. + let block = 64 * 1024; + let data: Vec = (0..5 * block - 100) + .map(|i| (i * 7 + i / 251) as u8) + .collect(); + let path = put_test_needle(&state, 0x6e7a_0c06, &data); + let range = |from: usize, to: usize| format!("bytes={}-{}", from, to - 1).into_bytes(); + let damaged_ranges = [ + (3 * block + 10, 3 * block + 110), + (block - 50, 4 * block - 50), + ]; + for (from, to) in damaged_ranges { + let (status, _, body) = + send_read(&state, Method::GET, &path, Some(&range(from, to))).await; + assert_eq!(status, StatusCode::PARTIAL_CONTENT); + assert_eq!(body, &data[from..to]); + } + let dat = tmp.path().join("1.dat"); + let mut bytes = std::fs::read(&dat).unwrap(); + let at = bytes.windows(64).position(|w| w == &data[..64]).unwrap(); + bytes[at + 3 * block + 500] ^= 0xff; + std::fs::write(&dat, bytes).unwrap(); + for (from, to) in damaged_ranges { + let (status, _, _) = + send_read(&state, Method::GET, &path, Some(&range(from, to))).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{from}-{to}"); + } + for (from, to) in [ + (10, 110), + (2 * block + 1, 3 * block), + (4 * block + 5, data.len()), + ] { + let (status, _, body) = + send_read(&state, Method::GET, &path, Some(&range(from, to))).await; + assert_eq!(status, StatusCode::PARTIAL_CONTENT, "{from}-{to}"); + assert_eq!(body, &data[from..to]); + } + } + + #[test] + fn test_range_check_block_size() { + use crate::storage::volume::range_check_block_size; + let block = 64 * 1024; + let mib = 1024 * 1024; + for (size, want) in [ + (0, block), + (64 * block, block), + (64 * block + 1, 2 * block), + (64 * mib, mib), + (1 << 34, mib), + ] { + assert_eq!(range_check_block_size(size), want, "{size}"); + } + } + /// Range is read as Go reads it: a byte its parseRange cannot parse is a /// 416, Unicode whitespace around a range is trimmed, an empty value is no /// range, and HEAD ignores it. diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index b351a246e2..7ffb4c00aa 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -24,6 +24,7 @@ use crate::storage::disk_location::is_read_only_mount; use crate::storage::idx; use crate::storage::io::read_exact_at; use crate::storage::io_error::{IoErrorTracker, StreakMark}; +use crate::storage::needle::crc::CRC; 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}; @@ -845,6 +846,7 @@ pub(crate) struct NeedleReadPlan { data_file_access_control: Arc, io_errors: Arc, io_unavailable: Arc, + load_generation: u64, } impl NeedleReadPlan { @@ -924,6 +926,8 @@ impl NeedleReadPlan { io_errors: self.io_errors, needle_id: self.needle_id, checksum: n.checksum.0, + append_at_ns: n.append_at_ns, + load_generation: self.load_generation, } } } @@ -1205,6 +1209,224 @@ pub struct NeedleStreamInfo { /// Checksum stored in the needle tail, verified once the last chunk has /// been read — before that frame is emitted. pub checksum: u32, + pub(crate) append_at_ns: u64, + pub(crate) load_generation: u64, +} + +/// A range carries no checksum of its own, so a ranged read checks the whole +/// needle's CRC the first time any range of it is read, recording the CRC of +/// each block on the way. `VERIFIED_NEEDLES` keeps those block CRCs, and later +/// ranges check just the blocks they cover, so damage that appears after the +/// first check is still caught. Entries are per volume load, so a remounted +/// copy or a compaction is checked again, and concurrent first reads of a +/// needle share one check. +static VERIFIED_NEEDLES: LazyLock>>> = + LazyLock::new(|| { + Mutex::new(lru::LruCache::new( + std::num::NonZeroUsize::new(64 * 1024).unwrap(), + )) + }); +static VERIFYING_NEEDLES: LazyLock>>> = + LazyLock::new(Default::default); +static VOLUME_LOADS: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +struct VerifiedNeedle { + load_generation: u64, + id: NeedleId, + append_at_ns: u64, + checksum: u32, +} + +const RANGE_CHECK_BLOCKS: u64 = 64; +const MIN_RANGE_CHECK_BLOCK_SIZE: u64 = 64 * 1024; +const MAX_RANGE_CHECK_BLOCK_SIZE: u64 = 1024 * 1024; + +/// Keeps a needle to 64 blocks, each 64 KiB to 1 MiB, as Go's +/// rangeCheckBlockSize. +pub(crate) fn range_check_block_size(data_size: u64) -> u64 { + let mut block_size = MIN_RANGE_CHECK_BLOCK_SIZE; + while block_size < MAX_RANGE_CHECK_BLOCK_SIZE && data_size > RANGE_CHECK_BLOCKS * block_size { + block_size *= 2; + } + block_size +} + +#[derive(Default)] +struct VerifyingNeedle { + outcome: Mutex, (io::ErrorKind, String)>>>, + done: Condvar, +} + +/// Publishes the check's outcome to the readers waiting on it, also when the +/// check unwinds. +struct VerifyingNeedleGuard { + key: VerifiedNeedle, + call: Arc, +} + +impl Drop for VerifyingNeedleGuard { + fn drop(&mut self) { + let mut outcome = self.call.outcome.lock().unwrap_or_else(|e| e.into_inner()); + outcome.get_or_insert_with(|| { + Err(( + io::ErrorKind::Other, + "needle checksum check did not finish".into(), + )) + }); + drop(outcome); + self.call.done.notify_all(); + VERIFYING_NEEDLES + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(&self.key); + } +} + +impl NeedleStreamInfo { + /// Whether `data`, the whole payload, matches the checksum in the tail. + pub(crate) fn checksum_matches(&self, data: &[u8]) -> bool { + let crc = CRC::new(data); + self.checksum == crc.0 || self.checksum == crc.legacy_value() + } + + /// Check the whole needle's CRC before any byte of a range is sent, so a + /// damaged replica fails the read and the reader tries another copy, and + /// return the CRC of each block for `read_checked_range`. + pub(crate) fn verify_checksum(&self, chunk_size: usize) -> io::Result> { + let key = VerifiedNeedle { + load_generation: self.load_generation, + id: self.needle_id, + append_at_ns: self.append_at_ns, + checksum: self.checksum, + }; + if let Some(crcs) = VERIFIED_NEEDLES + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&key) + { + return Ok(crcs.clone()); + } + let (call, leader) = { + let mut calls = VERIFYING_NEEDLES.lock().unwrap_or_else(|e| e.into_inner()); + match calls.get(&key) { + Some(call) => (call.clone(), false), + None => { + let call = Arc::new(VerifyingNeedle::default()); + calls.insert(key, call.clone()); + (call, true) + } + } + }; + if !leader { + let mut outcome = call.outcome.lock().unwrap_or_else(|e| e.into_inner()); + while outcome.is_none() { + outcome = call.done.wait(outcome).unwrap_or_else(|e| e.into_inner()); + } + return outcome + .clone() + .unwrap() + .map_err(|(kind, msg)| io::Error::new(kind, msg)); + } + let guard = VerifyingNeedleGuard { key, call }; + let result = self.read_checksum(chunk_size); + if let Ok(crcs) = &result { + VERIFIED_NEEDLES + .lock() + .unwrap_or_else(|e| e.into_inner()) + .put(key, crcs.clone()); + } + *guard.call.outcome.lock().unwrap_or_else(|e| e.into_inner()) = Some( + result + .as_ref() + .map(Arc::clone) + .map_err(|e| (e.kind(), e.to_string())), + ); + result + } + + /// Read the whole payload in chunks and compare its CRC, recording the CRC + /// of each block. A read failure counts toward the io error streak; a pass + /// leaves the streak to the range read that follows. + fn read_checksum(&self, chunk_size: usize) -> io::Result> { + let total = self.data_size as usize; + let block_size = range_check_block_size(self.data_size as u64) as usize; + let mut buf = vec![0u8; chunk_size.clamp(1, total.max(1))]; + let mut crc = CRC(0); + let mut block_crc = CRC(0); + let mut crcs = Vec::with_capacity(total.div_ceil(block_size)); + let mut pos = 0; + while pos < total { + let len = buf + .len() + .min(total - pos) + .min(block_size - pos % block_size); + if let Err(e) = self + .source + .read_exact_at(&mut buf[..len], self.data_file_offset + pos as u64) + { + self.io_errors.check_read_write_error(Some(&e)); + return Err(e); + } + crc = crc.update(&buf[..len]); + block_crc = block_crc.update(&buf[..len]); + pos += len; + if pos % block_size == 0 || pos == total { + crcs.push(block_crc.0); + block_crc = CRC(0); + } + } + if self.checksum != crc.0 && self.checksum != crc.legacy_value() { + crate::metrics::HANDLER_COUNTER + .with_label_values(&[crate::metrics::ERROR_CRC]) + .inc(); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "needle data checksum {} expected {} for needle {}", + crc.0, self.checksum, self.needle_id + ), + )); + } + Ok(crcs.into()) + } + + /// Read whole blocks around `[start, start + length)`, check each against + /// the CRC `verify_checksum` recorded, and return the range's bytes, so + /// damage that appears after the needle passed its check is not served. + pub(crate) fn read_checked_range( + &self, + crcs: &[u32], + start: u64, + length: u64, + ) -> io::Result> { + let block_size = range_check_block_size(self.data_size as u64); + let span_start = start / block_size * block_size; + let span_end = (start + length).div_ceil(block_size) * block_size; + let span_end = span_end.min(self.data_size as u64); + let mut buf = vec![0u8; (span_end - span_start) as usize]; + self.source + .read_exact_at(&mut buf, self.data_file_offset + span_start)?; + for (i, block) in buf.chunks(block_size as usize).enumerate() { + let index = (span_start / block_size) as usize + i; + if crcs.get(index) != Some(&CRC::new(block).0) { + crate::metrics::HANDLER_COUNTER + .with_label_values(&[crate::metrics::ERROR_CRC]) + .inc(); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "checksum mismatch in block {} of needle {}", + index, self.needle_id + ), + )); + } + } + let from = (start - span_start) as usize; + buf.truncate(from + length as usize); + buf.drain(..from); + Ok(buf) + } } #[derive(Clone)] @@ -1442,6 +1664,9 @@ pub struct Volume { /// for volume health monitoring. io_errors: Arc, + /// Changes on every load, including a compaction commit. + load_generation: AtomicU64, + /// Protobuf VolumeInfo for tiered storage (.vif file), Go's /// volumeInfoRWLock domain. /// @@ -1574,6 +1799,7 @@ impl Volume { closed: AtomicBool::new(false), compaction_byte_per_second: AtomicI64::new(0), io_errors: Arc::default(), + load_generation: AtomicU64::new(0), volume_info: RwLock::new(PbVolumeInfo::default()), }; @@ -1628,6 +1854,7 @@ impl Volume { closed: AtomicBool::new(false), compaction_byte_per_second: AtomicI64::new(0), io_errors: Arc::default(), + load_generation: AtomicU64::new(0), volume_info: RwLock::new(PbVolumeInfo::default()), } } @@ -1752,6 +1979,10 @@ impl Volume { preallocate: u64, version: Version, ) -> Result<(), VolumeError> { + self.load_generation.store( + VOLUME_LOADS.fetch_add(1, Ordering::Relaxed) + 1, + Ordering::Relaxed, + ); let dat_path = self.file_name(".dat"); let mut already_has_super_block = false; @@ -2758,6 +2989,7 @@ impl Volume { data_file_access_control: self.data_file_access_control.clone(), io_errors: self.io_errors.clone(), io_unavailable: self.io_unavailable.clone(), + load_generation: self.load_generation.load(Ordering::Relaxed), }) } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index d2b03fcf30..bc70947aab 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -73,7 +73,8 @@ type Volume struct { ldbTimeout int64 isCompactionInProgress atomic.Bool - lastDiskCheckNs atomic.Int64 // unix time in nanoseconds for phantom volume detection + loadGeneration atomic.Uint64 // changes on every load, including a compaction commit + lastDiskCheckNs atomic.Int64 // unix time in nanoseconds for phantom volume detection volumeInfoRWLock sync.RWMutex volumeInfo *volume_server_pb.VolumeInfo diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index 6e464c0845..c9dcea91c4 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -141,6 +141,7 @@ func (v *Volume) reopenIdxForWriteLocked() error { } func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind NeedleMapKind, preallocate int64, ver needle.Version) (err error) { + v.loadGeneration.Store(volumeLoads.Add(1)) alreadyHasSuperBlock := false hasLoadedVolume := false diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index 2a00cbe1c5..6570dc0a7b 100644 --- a/weed/storage/volume_read.go +++ b/weed/storage/volume_read.go @@ -3,8 +3,12 @@ package storage import ( "fmt" "io" + "sync/atomic" "time" + lru "github.com/hashicorp/golang-lru/v2" + "golang.org/x/sync/singleflight" + "github.com/seaweedfs/seaweedfs/weed/util/mem" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -18,6 +22,40 @@ import ( const PagedReadLimit = 1024 * 1024 +// A range carries no checksum of its own, so a ranged read checks the whole +// needle's CRC the first time any range of it is read, recording the CRC of +// each block on the way. verifiedNeedles keeps those block CRCs, and later +// ranges check just the blocks they cover, so damage that appears after the +// first check is still caught. Entries are per volume load, so a remounted +// copy or a compaction is checked again, and concurrent first reads of a +// needle share one check. +var ( + verifiedNeedles, _ = lru.New[verifiedNeedle, []needle.CRC](64 * 1024) + verifyingNeedles singleflight.Group + volumeLoads atomic.Uint64 +) + +const ( + rangeCheckBlocks = 64 + minRangeCheckBlockSize = 64 * 1024 +) + +// rangeCheckBlockSize keeps a needle to 64 blocks, each 64 KiB to 1 MiB. +func rangeCheckBlockSize(dataSize int64) int64 { + blockSize := int64(minRangeCheckBlockSize) + for blockSize < PagedReadLimit && dataSize > rangeCheckBlocks*blockSize { + blockSize *= 2 + } + return blockSize +} + +type verifiedNeedle struct { + loadGeneration uint64 + id NeedleId + appendAtNs uint64 + checksum needle.CRC +} + // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedle(n *needle.Needle, readOption *ReadOption, onReadSizeFn func(size Size)) (count int, err error) { v.dataFileAccessLock.RLock() @@ -157,7 +195,150 @@ func (v *Volume) verifyMovedRecord(n *needle.Needle, moved *needle_map.NeedleVal // read fills in Needle content by looking up n.Id from NeedleMapper func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, writer io.Writer, offset int64, size int64) (err error) { + dataSize := int64(n.DataSize) + if size <= 0 || (offset == 0 && size == dataSize) { + return v.streamNeedleData(n, readOption, writer, offset, size) + } + // Check the CRC before writing any byte of the range, so a damaged + // replica fails the read and the reader tries another copy. + crcs, err := v.verifiedNeedleBlocks(n, readOption) + if err != nil { + return err + } + blockSize := rangeCheckBlockSize(dataSize) + start := offset / blockSize * blockSize + end := (offset + size + blockSize - 1) / blockSize * blockSize + if end > dataSize { + end = dataSize + } + checker := &rangeBlockChecker{writer: writer, blockSize: blockSize, crcs: crcs, pos: start, from: offset, to: offset + size, end: end} + defer checker.free() + if err = v.streamNeedleData(n, readOption, checker, start, end-start); err != nil { + return err + } + if checker.pos != end { + return fmt.Errorf("ReadNeedleData: needle %v,%v ended at %d of %d", v.Id, n.Id, checker.pos, end) + } + return nil +} +func (v *Volume) verifiedNeedleBlocks(n *needle.Needle, readOption *ReadOption) ([]needle.CRC, error) { + key := verifiedNeedle{v.loadGeneration.Load(), n.Id, n.AppendAtNs, n.Checksum} + if crcs, ok := verifiedNeedles.Get(key); ok { + return crcs, nil + } + crcs, err, _ := verifyingNeedles.Do(fmt.Sprint(key), func() (any, error) { + dataSize := int64(n.DataSize) + blocks := &needleBlockCRCs{blockSize: rangeCheckBlockSize(dataSize)} + if err := v.streamNeedleData(n, readOption, blocks, 0, dataSize); err != nil { + return nil, err + } + if blocks.pos != dataSize { + return nil, fmt.Errorf("ReadNeedleData: needle %v,%v ended at %d of %d", v.Id, n.Id, blocks.pos, dataSize) + } + if blocks.pos%blocks.blockSize != 0 { + blocks.crcs = append(blocks.crcs, blocks.crc) + } + // A reload during the check may have swapped the data file. + if v.loadGeneration.Load() == key.loadGeneration { + verifiedNeedles.Add(key, blocks.crcs) + } + return blocks.crcs, nil + }) + if err != nil { + return nil, err + } + return crcs.([]needle.CRC), nil +} + +// needleBlockCRCs records the CRC of each block the verification pass reads. +type needleBlockCRCs struct { + blockSize int64 + pos int64 + crc needle.CRC + crcs []needle.CRC +} + +func (b *needleBlockCRCs) Write(p []byte) (int, error) { + written := len(p) + for len(p) > 0 { + k := len(p) + if left := b.blockSize - b.pos%b.blockSize; int64(k) > left { + k = int(left) + } + b.crc = b.crc.Update(p[:k]) + b.pos += int64(k) + p = p[k:] + if b.pos%b.blockSize == 0 { + b.crcs = append(b.crcs, b.crc) + b.crc = 0 + } + } + return written, nil +} + +// rangeBlockChecker reads whole blocks around [from, to) and writes the bytes +// of the range once their block matches the CRC the verification recorded. +type rangeBlockChecker struct { + writer io.Writer + blockSize int64 + crcs []needle.CRC + pos, from, to int64 + end int64 + crc needle.CRC + held []byte +} + +func (c *rangeBlockChecker) Write(p []byte) (int, error) { + written := len(p) + for len(p) > 0 { + k := len(p) + if left := c.blockSize - c.pos%c.blockSize; int64(k) > left { + k = int(left) + } + if left := c.end - c.pos; int64(k) > left { + k = int(left) + } + if k <= 0 { + return 0, fmt.Errorf("ReadNeedleData: %d bytes past the range end %d", len(p), c.end) + } + c.crc = c.crc.Update(p[:k]) + if lo, hi := max(c.pos, c.from), c.pos+int64(k); lo < c.to && hi > c.from { + if hi > c.to { + hi = c.to + } + if c.held == nil { + c.held = mem.Allocate(int(c.blockSize))[:0] + } + c.held = append(c.held, p[lo-c.pos:hi-c.pos]...) + } + c.pos += int64(k) + p = p[k:] + if c.pos%c.blockSize == 0 || c.pos == c.end { + block := (c.pos - 1) / c.blockSize + if block >= int64(len(c.crcs)) || c.crc != c.crcs[block] { + stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorCRC).Inc() + return 0, fmt.Errorf("checksum mismatch in block %d of the needle", block) + } + c.crc = 0 + if len(c.held) > 0 { + if _, err := c.writer.Write(c.held); err != nil { + return 0, err + } + c.held = c.held[:0] + } + } + } + return written, nil +} + +func (c *rangeBlockChecker) free() { + if c.held != nil { + mem.Free(c.held[:cap(c.held)]) + } +} + +func (v *Volume) streamNeedleData(n *needle.Needle, readOption *ReadOption, writer io.Writer, offset int64, size int64) (err error) { if !readOption.HasSlowRead { v.dataFileAccessLock.RLock() defer v.dataFileAccessLock.RUnlock() @@ -367,8 +548,12 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr return fmt.Errorf("ReadNeedleData checksum %v expected %v for Needle: %v,%v", crc, n.Checksum, v.Id, n) } // The read succeeded only once the checksum passes; a failed needle - // leaves the io error streak untouched, matching the streamed path. - v.checkReadWriteError(nil) + // leaves the io error streak untouched, matching the streamed path. A + // ranged read's verification pass leaves it alone too: the range read + // that follows reports the outcome. + if _, verifying := writer.(*needleBlockCRCs); !verifying { + v.checkReadWriteError(nil) + } if pending != nil { if _, err = writer.Write(pending); err != nil { return fmt.Errorf("ReadNeedleData write: %w", err) diff --git a/weed/storage/volume_read_test.go b/weed/storage/volume_read_test.go index 3955c192f1..4a149d0c63 100644 --- a/weed/storage/volume_read_test.go +++ b/weed/storage/volume_read_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "reflect" "strings" + "sync" "syscall" "testing" "time" @@ -324,6 +325,190 @@ func TestReadNeedleDataIntoChecksumMismatchHoldsLastPage(t *testing.T) { } } +// A range carries no checksum of its own, so the first ranged read of a +// needle checks the whole needle's CRC and writes nothing on a mismatch. The +// reader can then fall back to another replica. +func TestReadNeedleDataIntoRangeChecksumMismatch(t *testing.T) { + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + data := bytes.Repeat([]byte("0123456789"), 300) + // One buffer holds the whole needle, or the check streams it in pages. + for i, page := range []int{4096, 1024} { + write := func(id uint64) (types.Offset, int64) { + n := &needle.Needle{Id: types.Uint64ToNeedleId(id), Cookie: 0x1234, Data: append([]byte(nil), data...)} + n.Checksum = needle.NewCRC(n.Data) + offset, _, _, err := v.writeNeedle2(n, true, false, false) + if err != nil { + t.Fatalf("write needle: %v", err) + } + nv, _ := v.nm.Get(n.Id) + return nv.Offset, int64(offset) + } + read := func(id uint64, at types.Offset) (bytes.Buffer, error) { + meta := &needle.Needle{Id: types.Uint64ToNeedleId(id)} + nv, _ := v.nm.Get(meta.Id) + if err := v.readNeedleMetaAt(meta, at.ToActualOffset(), int32(nv.Size)); err != nil { + t.Fatalf("read meta: %v", err) + } + var buf bytes.Buffer + err := v.readNeedleDataInto(meta, &ReadOption{ReadBufferSize: page}, &buf, 1000, 500) + return buf, err + } + + intact := uint64(10*i + 1) + at, _ := write(intact) + if got, err := read(intact, at); err != nil || !bytes.Equal(got.Bytes(), data[1000:1500]) { + t.Fatalf("page %d: intact range read: %d bytes, %v", page, got.Len(), err) + } + + // Damage a byte outside the range: the range itself is intact, but + // the needle is not, so this replica must not serve it. + damaged := uint64(10*i + 2) + at, raw := write(damaged) + if _, err := v.DataBackend.WriteAt([]byte{0xff}, raw+types.NeedleHeaderSize+types.DataSizeSize+10); err != nil { + t.Fatalf("damage needle: %v", err) + } + got, err := read(damaged, at) + if err == nil || !strings.Contains(err.Error(), "checksum") || got.Len() != 0 { + t.Fatalf("page %d: damaged range read wrote %d bytes, err %v; want a checksum error and nothing written", page, got.Len(), err) + } + + // Concurrent first ranges of a needle all see the outcome of the check. + for _, tc := range []struct { + id uint64 + damage bool + }{{uint64(10*i + 3), false}, {uint64(10*i + 4), true}} { + at, raw := write(tc.id) + if tc.damage { + if _, err := v.DataBackend.WriteAt([]byte{0xff}, raw+types.NeedleHeaderSize+types.DataSizeSize+10); err != nil { + t.Fatalf("damage needle: %v", err) + } + } + var wg sync.WaitGroup + errs := make([]error, 8) + for g := range errs { + wg.Add(1) + go func(g int) { + defer wg.Done() + _, errs[g] = read(tc.id, at) + }(g) + } + wg.Wait() + for _, err := range errs { + if (err != nil) != tc.damage { + t.Fatalf("page %d: concurrent range read of needle %d (damaged %v): %v", page, tc.id, tc.damage, err) + } + } + } + + // A verified needle is checked again after a reload, as a remounted + // copy or a compaction commit gets. + reloaded := uint64(10*i + 5) + at, raw = write(reloaded) + if _, err := read(reloaded, at); err != nil { + t.Fatalf("page %d: first range read: %v", page, err) + } + if _, err := v.DataBackend.WriteAt([]byte{0xff}, raw+types.NeedleHeaderSize+types.DataSizeSize+10); err != nil { + t.Fatalf("damage needle: %v", err) + } + v.loadGeneration.Store(volumeLoads.Add(1)) + if _, err := read(reloaded, at); err == nil || !strings.Contains(err.Error(), "checksum") { + t.Fatalf("page %d: range read after reload: %v; want a checksum error", page, err) + } + } + + // The verification pass leaves the io error streak to the range read. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + meta := &needle.Needle{Id: types.Uint64ToNeedleId(1)} + nv, _ := v.nm.Get(meta.Id) + if err := v.readNeedleMetaAt(meta, nv.Offset.ToActualOffset(), int32(nv.Size)); err != nil { + t.Fatalf("read meta: %v", err) + } + v.loadGeneration.Store(volumeLoads.Add(1)) + if _, err := v.verifiedNeedleBlocks(meta, &ReadOption{ReadBufferSize: 4096}); err != nil { + t.Fatalf("verification pass: %v", err) + } + if _, count, _ := v.getIoErrorState(); count != 1 { + t.Errorf("io error streak %d after a verification pass, want 1", count) + } +} + +// Damage that appears after a needle passed its check is caught too: a later +// range checks the blocks it covers against the CRCs the check recorded, so it +// never serves damaged bytes, while ranges over intact blocks still read. +func TestReadNeedleDataIntoRangeDamagedAfterCheck(t *testing.T) { + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + const block = minRangeCheckBlockSize + data := make([]byte, 5*block-100) + for i := range data { + data[i] = byte(i*7 + i/251) + } + for i, page := range []int{4096, PagedReadLimit} { + n := &needle.Needle{Id: types.Uint64ToNeedleId(uint64(i + 1)), Cookie: 0x1234, Data: append([]byte(nil), data...)} + n.Checksum = needle.NewCRC(n.Data) + raw, _, _, err := v.writeNeedle2(n, true, false, false) + if err != nil { + t.Fatalf("write needle: %v", err) + } + read := func(offset, size int64) ([]byte, error) { + meta := &needle.Needle{Id: n.Id} + nv, _ := v.nm.Get(meta.Id) + if err := v.readNeedleMetaAt(meta, nv.Offset.ToActualOffset(), int32(nv.Size)); err != nil { + t.Fatalf("read meta: %v", err) + } + var buf bytes.Buffer + err := v.readNeedleDataInto(meta, &ReadOption{ReadBufferSize: page}, &buf, offset, size) + return buf.Bytes(), err + } + + ranges := [][2]int64{{3*block + 10, 100}, {block - 50, 3 * block}, {4*block + 5, int64(len(data)) - 4*block - 5}} + for _, r := range ranges { + if got, err := read(r[0], r[1]); err != nil || !bytes.Equal(got, data[r[0]:r[0]+r[1]]) { + t.Fatalf("page %d: range %v before damage: %d bytes, %v", page, r, len(got), err) + } + } + + damagedAt := int64(3*block + 500) + if _, err := v.DataBackend.WriteAt([]byte{^data[damagedAt]}, int64(raw)+types.NeedleHeaderSize+types.DataSizeSize+damagedAt); err != nil { + t.Fatalf("damage needle: %v", err) + } + for _, r := range ranges[:2] { + got, err := read(r[0], r[1]) + if err == nil || !strings.Contains(err.Error(), "checksum") { + t.Fatalf("page %d: range %v over a block damaged after the check: %v; want a checksum error", page, r, err) + } + // Only bytes of blocks that passed go out, and they are intact. + if int64(len(got)) > max(0, 3*block-r[0]) || !bytes.Equal(got, data[r[0]:r[0]+int64(len(got))]) { + t.Fatalf("page %d: range %v wrote %d bytes reaching into the damaged block", page, r, len(got)) + } + } + for _, r := range [][2]int64{ranges[2], {10, 100}, {2*block + 1, block - 1}} { + if got, err := read(r[0], r[1]); err != nil || !bytes.Equal(got, data[r[0]:r[0]+r[1]]) { + t.Fatalf("page %d: range %v over intact blocks: %d bytes, %v", page, r, len(got), err) + } + } + } + + for _, tc := range []struct{ size, block int64 }{ + {0, block}, {64 * block, block}, {64*block + 1, 2 * block}, {64 * PagedReadLimit, PagedReadLimit}, {1 << 34, PagedReadLimit}, + } { + if got := rangeCheckBlockSize(tc.size); got != tc.block { + t.Errorf("rangeCheckBlockSize(%d) = %d, want %d", tc.size, got, tc.block) + } + } +} + // With MustMetaOnly set, the payload is skipped whatever the needle's size. // This is the option a delete uses. A chunk manifest is the exception: its // data is still read, because a delete needs it to find the chunks.