diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 214462314..6a0532f9c 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -4,6 +4,7 @@ //! 48 RPCs: core volume operations are fully implemented, streaming and //! EC operations are stubbed with appropriate error messages. +use std::ops::ControlFlow; use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::Ordering; @@ -2754,7 +2755,6 @@ impl VolumeServer for VolumeGrpcService { let state = self.state.clone(); let (tx, rx) = tokio::sync::mpsc::channel(32); - const BUFFER_SIZE_LIMIT: usize = 2 * 1024 * 1024; tokio::spawn(async move { let since_ns = req.since_ns; @@ -2763,121 +2763,51 @@ impl VolumeServer for VolumeGrpcService { let mut draining_seconds = idle_timeout as i64; loop { - // Resolve the start offset and the caught-up flag under one - // store read guard. is_last means the caller is caught up: send - // a heartbeat without scanning, as Go does. Dropping that flag - // re-reads the whole volume every iteration (a moved volume is - // read-only, so it is always caught up). The single guard - // spans both the search and the scan: a vacuum commit takes - // the store write lock and rewrites .dat/.idx, so an offset - // resolved under one guard would point into a different file - // under the next. - let resolved = { - let store = state.store.read().unwrap(); - match store.find_volume(vid) { - Some((_, vol)) => { - let start = if last_timestamp_ns > 0 { - match vol.binary_search_by_append_at_ns(last_timestamp_ns) { - Ok((offset, is_last)) => { - let off = if offset.is_zero() { - sb_size - } else { - offset.to_actual_offset() as u64 - }; - Ok((off, is_last)) - } - Err(e) => { - tracing::warn!( - "fail to locate by appendAtNs {}: {}", - last_timestamp_ns, - e - ); - Err(format!( - "fail to locate by appendAtNs {}: {}", - last_timestamp_ns, e - )) - } - } - } else { - // No timestamp yet: the caller wants everything. - Ok((sb_size, false)) - }; - Some(start.map(|(off, is_last)| { - if is_last { - None - } else { - Some(vol.scan_raw_needles_from(off)) - } - })) - } - None => None, - } - }; + // The search and the scan are blocking file (or S3) I/O, so + // each pass runs on a blocking thread and takes and drops the + // store guard itself (see tail_pass). The heartbeat, the sleep + // and the draining countdown stay here, so no thread is parked + // while the stream is idle. + let pass_state = state.clone(); + let pass_tx = tx.clone(); + let pass = tokio::task::spawn_blocking(move || { + tail_pass( + &pass_state, + vid, + sb_size, + version, + last_timestamp_ns, + &pass_tx, + ) + }) + .await; - let scan_result = match resolved { - None => break, - Some(Err(msg)) => { + let (last_processed_ns, sent_any) = match pass { + Ok(TailPass::Scanned { + last_processed_ns, + sent_any, + }) => (last_processed_ns, sent_any), + // Caught up: heartbeat WITHOUT scanning, as Go does. + Ok(TailPass::CaughtUp) => (last_timestamp_ns, false), + Ok(TailPass::VolumeGone) => break, + Ok(TailPass::ClientGone) => return, + Ok(TailPass::Failed(msg)) => { let _ = tx.send(Err(Status::internal(msg))).await; return; } - Some(Ok(scan_result)) => scan_result, - }; - - // Caught up: heartbeat WITHOUT scanning, as Go does. - let Some(scan_result) = scan_result else { - let msg = volume_server_pb::VolumeTailSenderResponse { - is_last_chunk: true, - version, - ..Default::default() - }; - if tx.send(Ok(msg)).await.is_err() { + // A panic on the blocking thread must not look like a + // clean end of stream to the receiver. + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!( + "streamFollow: tail pass for volume {} failed: {}", + vid, e + )))) + .await; return; } - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - if idle_timeout == 0 { - continue; - } - draining_seconds -= 1; - if draining_seconds <= 0 { - return; // EOF - } - continue; }; - let entries = match scan_result { - Ok(e) => e, - Err(_) => break, - }; - // Filter entries since last_timestamp_ns - let mut last_processed_ns = last_timestamp_ns; - let mut sent_any = false; - for (header, body, append_at_ns) in &entries { - if *append_at_ns <= last_timestamp_ns && last_timestamp_ns > 0 { - continue; - } - sent_any = true; - // Send body in chunks of BUFFER_SIZE_LIMIT - // Go sends needle_header on every chunk - let mut i = 0; - while i < body.len() { - let end = std::cmp::min(i + BUFFER_SIZE_LIMIT, body.len()); - let is_last_chunk = end >= body.len(); - let msg = volume_server_pb::VolumeTailSenderResponse { - needle_header: header.clone(), - needle_body: body[i..end].to_vec(), - is_last_chunk, - version, - }; - if tx.send(Ok(msg)).await.is_err() { - return; - } - i = end; - } - if *append_at_ns > last_processed_ns { - last_processed_ns = *append_at_ns; - } - } - if !sent_any { // Send heartbeat let msg = volume_server_pb::VolumeTailSenderResponse { @@ -5763,6 +5693,152 @@ fn find_last_append_at_ns(idx_path: &str, dat_path: &str, version: u32) -> Optio if ts > 0 { Some(ts) } else { None } } +/// What one `volume_tail_sender` pass did. +enum TailPass { + /// The volume is no longer mounted: the stream ends. + VolumeGone, + /// The receiver hung up: the stream ends without a status. + ClientGone, + /// The pass failed: the stream ends with this internal error. + Failed(String), + /// Nothing is newer than the caller's timestamp, and nothing was scanned. + CaughtUp, + /// A scan ran. `last_processed_ns` is the newest timestamp shipped, or the + /// caller's own when nothing was. + Scanned { + last_processed_ns: u64, + sent_any: bool, + }, +} + +/// One pass of `volume_tail_sender`, on a blocking thread: resolve where to +/// start under one store guard, then stream every needle newer than +/// `last_timestamp_ns` with the guard released. +fn tail_pass( + state: &VolumeServerState, + vid: VolumeId, + sb_size: u64, + version: u32, + last_timestamp_ns: u64, + tx: &tokio::sync::mpsc::Sender>, +) -> TailPass { + const BUFFER_SIZE_LIMIT: usize = 2 * 1024 * 1024; + + // `is_last` means the client is already caught up: nothing in the + // volume is newer than last_timestamp_ns. Go answers that with a + // heartbeat and does NOT scan (volume_grpc_tail.go, `if isLastOne`). + // Dropping that flag here is expensive, not just untidy: the scan + // below reads every needle from its start offset to the end bound, and + // when the search yields offset 0 the start offset falls back to + // sb_size -- the whole volume. A volume being moved is marked + // read-only first, so it is ALWAYS caught up, and the tail loop + // would re-read and discard the entire volume every 2s until the + // idle timeout expired. Measured at ~2.17 GB re-read six times in + // 35s for a 2.15 GB volume, which OOM-kills the source under a + // per-process memory cap. + // + // The start offset, the .dat handle and the end bound are captured under + // ONE store read guard, and the guard is dropped before any needle is + // read. The handle pins the inode the offset was resolved against: a + // vacuum commit replaces .dat by rename, and unmount or destroy unlink it, + // so the offset stays meaningful after either (see DatScanPlan). Reading + // needles under the guard stalled the node: needle writes and the + // heartbeat take store.write(), and the lock prefers writers, so every + // reader queued behind them until the scan finished. + let (plan, from) = { + let store = state.store.read().unwrap(); + let Some((_, vol)) = store.find_volume(vid) else { + return TailPass::VolumeGone; + }; + let from = if last_timestamp_ns > 0 { + match vol.binary_search_by_append_at_ns(last_timestamp_ns) { + Ok((_, true)) => return TailPass::CaughtUp, + Ok((offset, false)) => { + if offset.is_zero() { + sb_size + } else { + offset.to_actual_offset() as u64 + } + } + Err(e) => { + tracing::warn!("fail to locate by appendAtNs {}: {}", last_timestamp_ns, e); + return TailPass::Failed(format!( + "fail to locate by appendAtNs {}: {}", + last_timestamp_ns, e + )); + } + } + } else { + // No timestamp yet: the caller wants everything. + sb_size + }; + match vol.dat_scan_plan(from) { + Ok(plan) => (plan, from), + Err(e) => { + return TailPass::Failed(format!( + "streamFollow: scan volume {} from offset {}: {}", + vid, from, e + )); + } + } + }; + + let mut last_processed_ns = last_timestamp_ns; + let mut sent_any = false; + let mut client_gone = false; + let scanned = plan.scan(|needle| { + // Notice a receiver that hung up between sends too, so a pass over + // needles it already has does not read on for nobody. + if tx.is_closed() { + client_gone = true; + return ControlFlow::Break(()); + } + if needle.append_at_ns <= last_timestamp_ns && last_timestamp_ns > 0 { + return ControlFlow::Continue(()); + } + sent_any = true; + // Send body in chunks of BUFFER_SIZE_LIMIT + // Go sends needle_header on every chunk + let mut i = 0; + while i < needle.body.len() { + let end = std::cmp::min(i + BUFFER_SIZE_LIMIT, needle.body.len()); + let msg = volume_server_pb::VolumeTailSenderResponse { + needle_header: needle.header.to_vec(), + needle_body: needle.body[i..end].to_vec(), + is_last_chunk: end >= needle.body.len(), + version, + }; + if tx.blocking_send(Ok(msg)).is_err() { + client_gone = true; + return ControlFlow::Break(()); + } + i = end; + } + if needle.append_at_ns > last_processed_ns { + last_processed_ns = needle.append_at_ns; + } + ControlFlow::Continue(()) + }); + + if client_gone { + return TailPass::ClientGone; + } + match scanned { + Ok(()) => TailPass::Scanned { + last_processed_ns, + sent_any, + }, + // Streaming makes a failed pass partial: some needles may already be + // on the wire. Ending the stream cleanly would let the receiver + // (volume.move) treat a truncated tail as complete, so report it, as + // Go does (`streamFollow: %w`). + Err(e) => TailPass::Failed(format!( + "streamFollow: scan volume {} from offset {}: {}", + vid, from, e + )), + } +} + /// Get disk usage (total, free) in bytes for the given path. fn get_disk_usage(path: &str) -> (u64, u64) { use sysinfo::Disks; @@ -6875,6 +6951,76 @@ mod tests { assert_eq!(copied, dat_bytes); } + // volume_tail_sender streams each needle in 2MB chunks with the header on + // every chunk, heartbeats once the caller is caught up, and ends when the + // idle timeout runs out. Reassembling the stream must reproduce the .dat + // after the superblock byte for byte, so a scan that skipped, reordered or + // re-sent a record fails here. The fixture holds a small needle and a 5MB + // one, covering the single- and multi-chunk paths. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_tail_sender_streams_chunks_then_heartbeats() { + let (service, _tmp, dat_bytes) = make_local_service_with_large_volume(); + let sb_size = { + let store = service.state.store.read().unwrap(); + let (_, v) = store.find_volume(VolumeId(1)).unwrap(); + v.super_block.block_size() + }; + + let response = service + .volume_tail_sender(Request::new(volume_server_pb::VolumeTailSenderRequest { + volume_id: 1, + since_ns: 0, + idle_timeout_seconds: 1, + })) + .await + .unwrap(); + let mut stream = response.into_inner(); + let messages = tokio::time::timeout(std::time::Duration::from_secs(30), async { + let mut messages = Vec::new(); + while let Some(m) = stream.next().await { + messages.push(m.unwrap()); + } + messages + }) + .await + .expect("the tail must end once its idle timeout runs out"); + + let (heartbeats, chunks): (Vec<_>, Vec<_>) = + messages.iter().partition(|m| m.needle_header.is_empty()); + assert_eq!(heartbeats.len(), 1, "one caught-up pass before the timeout"); + assert!(heartbeats[0].is_last_chunk && heartbeats[0].needle_body.is_empty()); + assert!( + messages.last().unwrap().needle_header.is_empty(), + "the heartbeat follows the data" + ); + + let mut shipped = Vec::new(); + let mut ids = Vec::new(); + let mut chunks_per_needle = Vec::new(); + let mut current_header: Vec = Vec::new(); + let mut starting = true; + for m in &chunks { + if starting { + current_header = m.needle_header.clone(); + shipped.extend_from_slice(¤t_header); + ids.push(Needle::parse_header(¤t_header).1); + chunks_per_needle.push(0); + } else { + assert_eq!(m.needle_header, current_header, "header on every chunk"); + } + shipped.extend_from_slice(&m.needle_body); + *chunks_per_needle.last_mut().unwrap() += 1; + starting = m.is_last_chunk; + } + assert!(starting, "the last needle must end on is_last_chunk"); + assert_eq!(ids, vec![NeedleId(11), NeedleId(99)]); + assert_eq!(chunks_per_needle, vec![1, 3], "a 5MB body in 2MB chunks"); + assert!( + shipped == dat_bytes[sb_size..], + "the stream must reproduce the .dat after the superblock" + ); + } + // copy_file must stop exactly at stop_offset, never streaming past it. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_copy_file_respects_stop_offset() { diff --git a/seaweed-volume/src/storage/needle/needle.rs b/seaweed-volume/src/storage/needle/needle.rs index df4205ad1..278c944f5 100644 --- a/seaweed-volume/src/storage/needle/needle.rs +++ b/seaweed-volume/src/storage/needle/needle.rs @@ -581,23 +581,19 @@ impl Needle { // ============================================================================ /// Compute padding to align needle to NEEDLE_PADDING_SIZE (8 bytes). +/// +/// The sum is formed in i64: a size read from a corrupt header can sit near +/// `i32::MAX`, and adding the header, checksum and timestamp widths to it in +/// i32 would overflow (a panic with overflow checks, a wrapped padding +/// without). The result is at most NEEDLE_PADDING_SIZE, so it fits `Size`. pub fn padding_length(needle_size: Size, version: Version) -> Size { - if version == VERSION_3 { - Size( - NEEDLE_PADDING_SIZE as i32 - - ((NEEDLE_HEADER_SIZE as i32 - + needle_size.0 - + NEEDLE_CHECKSUM_SIZE as i32 - + TIMESTAMP_SIZE as i32) - % NEEDLE_PADDING_SIZE as i32), - ) + let fixed = if version == VERSION_3 { + NEEDLE_HEADER_SIZE + NEEDLE_CHECKSUM_SIZE + TIMESTAMP_SIZE } else { - Size( - NEEDLE_PADDING_SIZE as i32 - - ((NEEDLE_HEADER_SIZE as i32 + needle_size.0 + NEEDLE_CHECKSUM_SIZE as i32) - % NEEDLE_PADDING_SIZE as i32), - ) - } + NEEDLE_HEADER_SIZE + NEEDLE_CHECKSUM_SIZE + }; + let unpadded = fixed as i64 + needle_size.0 as i64; + Size((NEEDLE_PADDING_SIZE as i64 - unpadded % NEEDLE_PADDING_SIZE as i64) as i32) } /// Body length = Size + Checksum + [Timestamp] + Padding. @@ -925,6 +921,21 @@ mod tests { } } + #[test] + fn padding_length_does_not_overflow_on_a_corrupt_size() { + // A header read from a corrupt or truncated file can carry any i32 + // size. The scanners bound it against the bytes left before sizing a + // buffer, but on a volume with more than 2 GiB left a size near + // i32::MAX passes that bound, so the padding arithmetic itself must + // not overflow. Overflow checks are on in test builds, so an i32 sum + // here would panic rather than wrap. + for version in [VERSION_2, VERSION_3] { + let padding = padding_length(Size(i32::MAX), version).0 as i64; + assert!((1..=NEEDLE_PADDING_SIZE as i64).contains(&padding)); + assert_eq!(get_actual_size(Size(i32::MAX), version) % 8, 0); + } + } + #[test] fn test_file_id_parse() { let fid = FileId::parse("3,01637037d6").unwrap(); diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index dadb10e32..e6f000c69 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -11,6 +11,7 @@ use std::fs::{self, File, OpenOptions}; use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::ops::ControlFlow; use std::path::Path; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -475,6 +476,131 @@ impl NeedleStreamSource { } } +/// A `.dat` scan that runs with the store lock released, built by +/// `Volume::dat_scan_plan` while the caller holds a store guard. +/// +/// The start offset, the handle and the end bound are captured together +/// while the guard excludes every writer. The handle pins the inode the +/// offset was resolved against: a vacuum commit replaces `.dat` by rename, +/// and unmount or destroy close and unlink it, none of which rewrites the +/// pinned bytes. `.dat` is otherwise append-only and a failed append +/// truncates only its own bytes, so `[from, end)` holds complete records +/// that nothing rewrites while the scan runs. A short read below `end` can +/// therefore only mean the inode was truncated under the plan (an unmount +/// followed by a `VolumeCopy` of the same id reopens `.dat` with +/// `truncate(true)`), and the scan fails rather than ending as if the +/// snapshot had been read. The fields are private so a plan can only come +/// from `dat_scan_plan`. +pub(crate) struct DatScanPlan { + source: NeedleStreamSource, + version: Version, + from: u64, + end: u64, +} + +/// One record visited by `DatScanPlan::scan`: the raw on-disk bytes and the +/// parsed append timestamp. +pub(crate) struct RawNeedle<'a> { + pub header: &'a [u8], + pub body: &'a [u8], + pub append_at_ns: u64, +} + +impl DatScanPlan { + /// Visit the records in `[from, end)` in file order, the way Go's + /// `ScanVolumeFileFrom` feeds a `VolumeFileScanner`: only the record being + /// visited is in memory. Reads are positional and never touch the + /// `Volume`. The pass ends `Ok` at the end of the data, at a record that + /// does not fit before `end`, at a corrupt header, or when `visit` breaks. + /// It fails on any read error, including a short read below `end`: every + /// byte below `end` existed when the plan was taken, so a short read means + /// the file was truncated under the plan and the pass is incomplete. + pub(crate) fn scan( + &self, + mut visit: impl FnMut(RawNeedle<'_>) -> ControlFlow<()>, + ) -> Result<(), VolumeError> { + let mut offset = self.from; + while offset + NEEDLE_HEADER_SIZE as u64 <= self.end { + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + self.source + .read_exact_at(&mut header, offset) + .map_err(|e| self.read_error(e, offset))?; + + let (_cookie, id, size) = Needle::parse_header(&header); + if size.0 == 0 && id.is_empty() { + break; + } + // A negative size is a corrupt header: body_length would size the + // buffer from a negative length or walk the scan from a wrong + // offset. Go's scanners stop here by returning io.EOF. + if size.0 < 0 { + break; + } + + // Nothing past `end` was a complete record when the plan was + // taken. The size is checked against the bytes left, and the + // whole record after it. Both run before allocating, so a corrupt + // size cannot size the buffer. (The size check alone does not + // keep a corrupt size out of the padding arithmetic on a volume + // with more than 2 GiB left, which is why `padding_length` sums + // in i64.) + if size.0 as u64 > self.end - offset - NEEDLE_HEADER_SIZE as u64 { + break; + } + let body_length = needle::needle_body_length(size, self.version) as u64; + let total_size = NEEDLE_HEADER_SIZE as u64 + body_length; + if offset + total_size > self.end { + break; + } + + // Match Go's ScanVolumeFileFrom: visit ALL needles including deleted ones. + // This is critical for incremental copy where tombstones must be propagated. + let mut record = vec![0u8; total_size as usize]; + record[..NEEDLE_HEADER_SIZE].copy_from_slice(&header); + self.source + .read_exact_at( + &mut record[NEEDLE_HEADER_SIZE..], + offset + NEEDLE_HEADER_SIZE as u64, + ) + .map_err(|e| self.read_error(e, offset + NEEDLE_HEADER_SIZE as u64))?; + + let append_at_ns = { + let mut n = Needle::default(); + n.read_bytes(&record, offset as i64, size, self.version)?; + n.append_at_ns + }; + let needle = RawNeedle { + header: &record[..NEEDLE_HEADER_SIZE], + body: &record[NEEDLE_HEADER_SIZE..], + append_at_ns, + }; + if visit(needle).is_break() { + break; + } + offset += total_size; + } + Ok(()) + } + + /// Name the snapshot in a read failure. A short read is the truncation + /// case described on the type; every other error is passed through with + /// the offset. + fn read_error(&self, e: io::Error, offset: u64) -> VolumeError { + let what = if e.kind() == io::ErrorKind::UnexpectedEof { + "short read below the snapshot end, the dat file was truncated under the scan" + } else { + "read failed" + }; + VolumeError::Io(io::Error::new( + e.kind(), + format!( + "{} (offset {}, snapshot end {}): {}", + what, offset, self.end, e + ), + )) + } +} + pub struct NeedleStreamInfo { /// Stream source for the dat file, local or remote. pub(crate) source: NeedleStreamSource, @@ -541,10 +667,6 @@ impl RemoteDatFile { // Volume // ============================================================================ -/// One raw needle as `scan_raw_needles_from` yields it: the header bytes, -/// the body bytes, and the needle's `append_at_ns`. -pub type RawNeedleEntry = (Vec, Vec, u64); - pub struct Volume { pub id: VolumeId, dir: String, @@ -2787,58 +2909,31 @@ impl Volume { Ok((count, broken)) } - /// Scan raw needle entries from the .dat file starting at `from_offset`. - /// Returns a [`RawNeedleEntry`] for each needle. - /// Used by VolumeTailSender to stream raw bytes. - pub fn scan_raw_needles_from( - &self, - from_offset: u64, - ) -> Result, VolumeError> { - let version = self.version(); - let dat_size = self.current_dat_file_size()?; - let mut entries = Vec::new(); - let mut offset = from_offset; - - while offset < dat_size { - // Read needle header (16 bytes) - let mut header = [0u8; NEEDLE_HEADER_SIZE]; - match self.read_exact_at_backend(&mut header, offset) { - Ok(()) => {} - Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => break, - Err(e) => return Err(e), - } - - let (_cookie, _id, size) = Needle::parse_header(&header); - if size.0 == 0 && _id.is_empty() { - break; - } - - let body_length = needle::needle_body_length(size, version); - let total_size = NEEDLE_HEADER_SIZE as u64 + body_length as u64; - - // Match Go's ScanVolumeFileFrom: visit ALL needles including deleted ones. - // This is critical for incremental copy where tombstones must be propagated. - - // Read body bytes - let mut body = vec![0u8; body_length as usize]; - match self.read_exact_at_backend(&mut body, offset + NEEDLE_HEADER_SIZE as u64) { - Ok(()) => {} - Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => break, - Err(e) => return Err(e), - } - - // Parse the needle to get append_at_ns - let mut full = vec![0u8; total_size as usize]; - full[..NEEDLE_HEADER_SIZE].copy_from_slice(&header); - full[NEEDLE_HEADER_SIZE..].copy_from_slice(&body); - let mut n = Needle::default(); - let _ = n.read_bytes(&full, offset as i64, size, version); - - entries.push((header.to_vec(), body, n.append_at_ns)); - offset += total_size; - } - - Ok(entries) + /// Capture a `.dat` scan from `from_offset` that the caller can run after + /// dropping its store guard. See `DatScanPlan` for why the offset, the + /// handle and the end bound must come from the same guard. + pub(crate) fn dat_scan_plan(&self, from_offset: u64) -> Result { + let source = if self.dat_file.is_some() { + // A fresh open, not `try_clone`: a duplicated handle shares the + // file position, and on Windows `read_exact_at` goes through + // `seek_read`, which moves it under a concurrent append. Opened + // while the caller's guard excludes a vacuum swap, the path names + // the inode `dat_file` holds. + NeedleStreamSource::Local(open_volume_file( + OpenOptions::new().read(true), + self.file_name(".dat"), + )?) + } else if let Some(remote) = self.remote_dat_file() { + NeedleStreamSource::Remote(remote) + } else { + return Err(VolumeError::Io(io::Error::other("dat file not open"))); + }; + Ok(DatScanPlan { + source, + version: self.version(), + from: from_offset, + end: self.current_dat_file_size()?, + }) } /// Insert or update a needle index entry (for low-level blob writes). @@ -4670,6 +4765,34 @@ mod tests { .unwrap() } + fn write_test_needle(v: &mut Volume, id: u64, data: &[u8]) -> u64 { + let mut n = Needle { + id: NeedleId(id), + cookie: Cookie(0x12345678), + data: data.to_vec(), + data_size: data.len() as u32, + flags: 0, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + n.append_at_ns + } + + /// Run a plan to completion, keeping owned copies of every record. + fn scan_all(plan: &DatScanPlan) -> Vec<(Vec, Vec, u64)> { + let mut records = Vec::new(); + plan.scan(|needle| { + records.push(( + needle.header.to_vec(), + needle.body.to_vec(), + needle.append_at_ns, + )); + ControlFlow::Continue(()) + }) + .unwrap(); + records + } + #[test] fn test_data_file_access_control_blocks_writer_until_reader_releases() { let control = Arc::new(DataFileAccessControl::default()); @@ -4823,14 +4946,14 @@ mod tests { "a caller older than a trailing delete must NOT be reported as caught up" ); - // Scan and filter as volume_tail_sender does: only the tombstone is newer. - let shipped: Vec = v - .scan_raw_needles_from(offset.to_actual_offset() as u64) - .unwrap() - .into_iter() - .map(|(_, _, append_at_ns)| append_at_ns) - .filter(|&append_at_ns| append_at_ns > newest_ns) - .collect(); + // Scan and filter exactly as volume_tail_sender does: the tombstone, + // and only the tombstone, is newer than the caller. + let shipped: Vec = + scan_all(&v.dat_scan_plan(offset.to_actual_offset() as u64).unwrap()) + .into_iter() + .map(|(_, _, append_at_ns)| append_at_ns) + .filter(|&append_at_ns| append_at_ns > newest_ns) + .collect(); assert_eq!(shipped.len(), 1, "the tail must ship the tombstone"); } @@ -4879,19 +5002,276 @@ mod tests { !is_last, "a write after compaction is the final row and must not be hidden" ); - let shipped: Vec = v - .scan_raw_needles_from(offset.to_actual_offset() as u64) - .unwrap() - .into_iter() - .map(|(_, _, append_at_ns)| append_at_ns) - .filter(|&append_at_ns| append_at_ns > key2_ns) - .collect(); + let shipped: Vec = + scan_all(&v.dat_scan_plan(offset.to_actual_offset() as u64).unwrap()) + .into_iter() + .map(|(_, _, append_at_ns)| append_at_ns) + .filter(|&append_at_ns| append_at_ns > key2_ns) + .collect(); assert!( shipped.contains(&key3_ns), "the tail must ship the write made after compaction" ); } + #[test] + fn dat_scan_plan_visits_needles_and_tombstones_in_file_order() { + // The tail ships exactly what the scan visits, so the scan must cover + // the file record by record: the on-disk bytes, in file order, + // tombstones included. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + let written: Vec = (1..=3u64) + .map(|id| write_test_needle(&mut v, id, b"scan")) + .collect(); + v.delete_needle(&mut Needle { + id: NeedleId(2), + cookie: Cookie(0x12345678), + ..Needle::default() + }) + .unwrap(); + + let sb_size = v.super_block.block_size() as u64; + let records = scan_all(&v.dat_scan_plan(sb_size).unwrap()); + let dat = std::fs::read(v.file_name(".dat")).unwrap(); + + assert_eq!(records.len(), 4, "three writes and one tombstone"); + let mut offset = sb_size as usize; + for (header, body, _) in &records { + assert_eq!(&dat[offset..offset + header.len()], header.as_slice()); + offset += header.len(); + assert_eq!(&dat[offset..offset + body.len()], body.as_slice()); + offset += body.len(); + } + assert_eq!(offset, dat.len(), "the scan must reach the end of the file"); + + let ids: Vec = records + .iter() + .map(|(header, _, _)| Needle::parse_header(header).1) + .collect(); + assert_eq!( + ids, + vec![NeedleId(1), NeedleId(2), NeedleId(3), NeedleId(2)] + ); + let stamps: Vec = records.iter().map(|r| r.2).collect(); + assert_eq!(&stamps[..3], written.as_slice()); + assert!(stamps[3] > stamps[2], "the tombstone is the newest record"); + } + + #[cfg(unix)] + #[test] + fn dat_scan_plan_outlives_a_vacuum_commit() { + // The tail sender drops the store guard before scanning, so a vacuum + // commit can land mid-scan. The commit renames .cpd over .dat; a plan + // taken before it must keep reading the file its offset was resolved + // against, not the compacted one. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + write_test_needle(&mut v, 1, b"first"); + write_test_needle(&mut v, 2, b"second"); + write_test_needle(&mut v, 1, b"first, overwritten"); + + let sb_size = v.super_block.block_size() as u64; + let before = scan_all(&v.dat_scan_plan(sb_size).unwrap()); + let plan = v.dat_scan_plan(sb_size).unwrap(); + let old_len = std::fs::metadata(v.file_name(".dat")).unwrap().len(); + + v.compact_by_index(0, 0, |_| true).unwrap(); + v.commit_compact().unwrap(); + let new_len = std::fs::metadata(v.file_name(".dat")).unwrap().len(); + assert!( + new_len < old_len, + "precondition: the commit swapped in a smaller .dat" + ); + + assert_eq!(scan_all(&plan), before); + } + + #[cfg(unix)] + #[test] + fn dat_scan_plan_outlives_volume_destroy() { + // Unmount or delete can land mid-scan too. The plan holds a handle, + // not a path, so it finishes reading what was there. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + write_test_needle(&mut v, 1, b"going"); + write_test_needle(&mut v, 2, b"gone"); + + let sb_size = v.super_block.block_size() as u64; + let before = scan_all(&v.dat_scan_plan(sb_size).unwrap()); + let plan = v.dat_scan_plan(sb_size).unwrap(); + let dat_path = v.file_name(".dat"); + + v.destroy(false, false).unwrap(); + assert!( + !Path::new(&dat_path).exists(), + "precondition: destroy removed .dat" + ); + + assert_eq!(scan_all(&plan), before); + } + + #[test] + fn dat_scan_plan_fails_when_the_snapshot_is_truncated() { + // Unmount followed by a VolumeCopy of the same id reopens .dat with + // truncate(true) on the inode a running plan holds. Every byte below + // the captured end existed when the plan was taken, so a short read + // there is an incomplete pass, not the end of the data: the stream + // must fail rather than let volume.move take a partial tail as + // complete. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + write_test_needle(&mut v, 1, b"kept"); + write_test_needle(&mut v, 2, b"cut off"); + + let sb_size = v.super_block.block_size() as u64; + let first_record_len = { + let records = scan_all(&v.dat_scan_plan(sb_size).unwrap()); + assert_eq!(records.len(), 2, "precondition: two records on disk"); + (records[0].0.len() + records[0].1.len()) as u64 + }; + let plan = v.dat_scan_plan(sb_size).unwrap(); + let end = plan.end; + + // Truncate through the header of the second record. + let second = sb_size + first_record_len; + OpenOptions::new() + .write(true) + .open(v.file_name(".dat")) + .unwrap() + .set_len(second + NEEDLE_HEADER_SIZE as u64 / 2) + .unwrap(); + + let mut visited = Vec::new(); + let err = plan + .scan(|n| { + visited.push(n.append_at_ns); + ControlFlow::Continue(()) + }) + .unwrap_err(); + assert_eq!(visited.len(), 1, "the intact first record is still visited"); + match err { + VolumeError::Io(e) => { + assert_eq!(e.kind(), io::ErrorKind::UnexpectedEof); + let msg = e.to_string(); + assert!(msg.contains("truncated"), "{msg}"); + assert!(msg.contains(&format!("snapshot end {end}")), "{msg}"); + } + other => panic!("expected an I/O error, got {other:?}"), + } + } + + #[test] + fn dat_scan_plan_stops_at_the_snapshot_end() { + // The end bound is read under the guard, while no append can be in + // flight. A record appended after the plan belongs to the next pass. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + let first_ns = write_test_needle(&mut v, 1, b"before"); + + let plan = v.dat_scan_plan(v.super_block.block_size() as u64).unwrap(); + write_test_needle(&mut v, 2, b"after"); + + let stamps: Vec = scan_all(&plan).into_iter().map(|r| r.2).collect(); + assert_eq!(stamps, vec![first_ns]); + } + + #[test] + fn dat_scan_plan_stops_when_the_visitor_breaks() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + for id in 1..=3u64 { + write_test_needle(&mut v, id, b"brk"); + } + + let mut visited = 0; + v.dat_scan_plan(v.super_block.block_size() as u64) + .unwrap() + .scan(|_| { + visited += 1; + ControlFlow::Break(()) + }) + .unwrap(); + assert_eq!(visited, 1); + } + + #[test] + fn dat_scan_plan_ends_the_pass_at_a_corrupt_header() { + // A negative size is a corrupt header: sizing a buffer from it + // overflows, and a small one walks the scan from a wrong offset. A + // size running past the end bound cannot be a complete record either, + // and one near i32::MAX overflows the padding arithmetic. Both end + // the pass after the records before them, without reading or + // allocating the bogus body. + for bad_size in [-1000, i32::MAX] { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + write_test_needle(&mut v, 1, b"good"); + + let sb_size = v.super_block.block_size() as u64; + let dat_path = v.file_name(".dat"); + let start = sb_size as usize; + let mut bad = + std::fs::read(&dat_path).unwrap()[start..start + NEEDLE_HEADER_SIZE].to_vec(); + Size(bad_size).to_bytes(&mut bad[COOKIE_SIZE + NEEDLE_ID_SIZE..NEEDLE_HEADER_SIZE]); + // Bytes after the header, so the size check stops the walk, not EOF. + bad.extend_from_slice(&[0u8; 64]); + OpenOptions::new() + .append(true) + .open(&dat_path) + .unwrap() + .write_all(&bad) + .unwrap(); + + let records = scan_all(&v.dat_scan_plan(sb_size).unwrap()); + assert_eq!(records.len(), 1, "size {}: only the good record", bad_size); + } + } + + #[test] + fn dat_scan_plan_fails_when_the_record_body_is_corrupt() { + // A record can fit inside the captured snapshot and still fail needle + // parsing. The tail must surface that failure instead of emitting a + // raw record with a zero append timestamp and reporting a clean pass. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + write_test_needle(&mut v, 1, b"good"); + + let sb_size = v.super_block.block_size() as u64; + let plan = v.dat_scan_plan(sb_size).unwrap(); + + let mut dat = OpenOptions::new() + .write(true) + .open(v.file_name(".dat")) + .unwrap(); + dat.seek(SeekFrom::Start(sb_size + NEEDLE_HEADER_SIZE as u64)) + .unwrap(); + dat.write_all(&u32::MAX.to_be_bytes()).unwrap(); + drop(dat); + + let mut visited = 0; + let err = plan + .scan(|_| { + visited += 1; + ControlFlow::Continue(()) + }) + .unwrap_err(); + + assert_eq!(visited, 0, "the corrupt record must not be emitted"); + assert!( + matches!(err, VolumeError::Needle(_)), + "expected a needle parse error, got {err:?}" + ); + } + #[test] fn test_volume_write_read() { let tmp = TempDir::new().unwrap();