mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 06:54:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

rust volume: stream the tail scan and release the store lock (#11275)

Browse Source 
* rust volume: add a .dat scan plan that runs without the store lock

DatScanPlan captures a fresh .dat handle, the version, the start offset
and an end bound while the caller holds a store guard, then visits one
record at a time with positional reads that never touch the Volume, the
way Go's ScanVolumeFileFrom feeds a scanner.

The handle pins the inode the offset was resolved against: a vacuum
commit renames .cpd over .dat and destroy unlinks it, and neither
rewrites the pinned bytes. The end bound is read while no writer can
hold store.write(), so the scan never meets a partial append. It is a
fresh open, not try_clone, because on Windows read_exact_at uses
seek_read, which moves a cursor a clone shares with the writer.

A header whose size is negative, or does not fit before the end bound,
ends the pass before the body length is computed or anything is
allocated. In today's scan a negative size reaches needle_body_length
and either overflows the buffer size or walks the scan from a wrong
offset. A size near i32::MAX overflows padding_length's i32 arithmetic,
which panics in debug builds.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018VF7E9SHPihG1jC1grU9H3

* rust volume: stream the tail scan with the store lock released

volume_tail_sender read every needle from the start offset to EOF into a
Vec while holding store.read(). volume.merge tails from zero, so that was
the whole volume in memory. And because needle writes and the heartbeat
take store.write() on a lock that prefers writers, the whole node stopped
serving until the scan finished: the failure #11235 fixed for EC scrub.

Each pass now runs on a blocking thread. Under one store guard it
resolves the start offset and captures a DatScanPlan, then drops the
guard and sends each needle as it is read, as Go's
VolumeFileScanner4Tailing does. This replaces the one-guard-across-
search-and-scan rule from the previous commit with a stronger invariant:
the offset, the handle and the end bound come from the same guard, and
the handle pins the inode, so a vacuum commit mid-scan cannot point the
offset into the compacted file.

A scan error now ends the stream with Status::internal instead of a
clean EOF, as Go's `streamFollow: %w` does. Once needles stream, a clean
EOF after a partial pass would let volume.move treat a truncated tail as
complete. A panic in the pass is reported the same way. A receiver that
hangs up is also noticed between skipped needles, not only on a send.

Unchanged: the append_at_ns filter, the header on every 2MB chunk, the
caught-up heartbeat without a scan, and the draining countdown.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018VF7E9SHPihG1jC1grU9H3

* rust volume: fail the tail pass on a short read below the snapshot end

DatScanPlan::scan treated an UnexpectedEof on the header or body read as
the end of the data and returned Ok. Every byte below the captured end
existed when the plan was taken, so a short read there can only mean the
inode was truncated under the plan: an unmount followed by a VolumeCopy
of the same volume id reopens .dat with truncate(true). The pass then
reported Scanned, the next pass found the volume gone, and the stream
ended cleanly after a prefix of the planned records, which volume.move
would take as a complete tail.

Both short-read arms now fail the scan with an I/O error that names the
offset and the snapshot end, so tail_pass reports Status::internal as it
does for every other read failure. The break arms were carried over from
scan_raw_needles_from, where the whole scan ran under the store guard and
nothing could truncate the file.

Found by the Devin and Greptile reviews on #11275.

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

* rust volume: sum the needle padding in i64 so a corrupt size cannot overflow

padding_length added the header, checksum and timestamp widths to the
needle size in i32. A size read from a corrupt header can sit near
i32::MAX, and that sum then overflows: a panic with overflow checks, a
wrapped padding without. DatScanPlan::scan bounds the size against the
bytes left before computing the body length, but that only keeps such a
size out of the arithmetic while under 2 GiB of the file remains, so on a
large volume the scan could still reach the overflow and, in release,
size a buffer from garbage.

Sum in i64 in both version branches. The result is at most
NEEDLE_PADDING_SIZE, so it still fits Size. The scan comment no longer
claims the bound check prevents the overflow.

Found by the CodeRabbit review on #11275.

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

* rust volume: propagate dat scan parse failures

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
This commit is contained in:
Eliah Rusin
2026-09-15 14:40:06 -07:00
committed by GitHub
co-authored by Claude Fable 5.1 Chris Lu
parent 55367afded
commit 563c729e70
3 changed files with 731 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+254 -108
View File
@@ -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<Result<volume_server_pb::VolumeTailSenderResponse, Status>>,
) -> 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<u8> = Vec::new();
let mut starting = true;
for m in &chunks {
if starting {
current_header = m.needle_header.clone();
shipped.extend_from_slice(&current_header);
ids.push(Needle::parse_header(&current_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() {
seaweed-volume/src/storage/needle/needle.rs
+26 -15
View File
@@ -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();
seaweed-volume/src/storage/volume.rs
+451 -71
View File
@@ -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<u8>, Vec<u8>, 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<Vec<RawNeedleEntry>, 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<DatScanPlan, VolumeError> {
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<u8>, Vec<u8>, 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<u64> = 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<u64> =
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<u64> = 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<u64> =
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<u64> = (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<NeedleId> = 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<u64> = 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<u64> = 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();