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: honour is_last in the tail sender instead of rescanning the whole volume (#11273)

Browse Source 
* rust volume: honour is_last in the tail sender instead of rescanning

volume_tail_sender discarded the is_last flag from
binary_search_by_append_at_ns:

    Ok((offset, _is_last)) => {
        if offset.is_zero() { Ok(sb_size) } ...

is_last means the caller is already caught up. Go answers that with a
heartbeat and does not scan at all (volume_grpc_tail.go, `if isLastOne`).
Dropping it is expensive rather than untidy, because the branches interact:
when the search reports caught-up it returns Offset::default(), which is
zero, so the start offset falls back to sb_size -- the beginning of the
data -- and scan_raw_needles_from materialises every needle from there to
EOF into a Vec. The timestamp filter discards all of it, the loop sleeps
2s, and it happens again.

A volume being moved is marked read-only before the copy, so it is ALWAYS
caught up during the tail phase. Measured on one volume.move of a 2.15 GB
volume, sampling the source's cgroup anon every 2s against the move's own
phase output:

  copying   16 -> 37 MB          CopyFile streams correctly, stays bounded
  tailing   904 -> 2166 -> 629 -> 2166 -> 342 -> 2173 -> 2179 MB
  deleting  46 MB

Six full-volume allocate/free cycles in 35s, peak 2179 MB against a volume
of 2147 MiB. The destination never exceeded 35 MB, so this is entirely
source-side. Under a per-process memory cap it OOM-kills the source
whenever the volume exceeds the cap.

The ordering here is the whole fix and is easy to get wrong: resolve the
start offset and is_last under a brief lock, return the heartbeat
immediately when caught up, and only then reach the scan. An earlier cut
set the flag correctly but placed the early return after the block that
performs the scan -- the heartbeat fired and the destination received
nothing, yet every iteration still read the whole volume and discarded it.
Production showed no improvement (1770 MB across five cycles), which is
what caught it. The binary search is over the .idx and costs nothing; the
scan is the expensive part and must not run speculatively.

Three tests, and the last two matter as much as the first: a fix that
always reported "caught up" would make tailing silently lose needles, a
worse bug than the one being fixed. One asserts is_last for a caller at or
beyond the newest append_at_ns; one asserts NOT is_last for a caller that
is behind, so real tail data is still scanned and shipped; one asserts NOT
is_last when the only newer record is a delete, and that scanning from the
returned offset ships exactly that tombstone.

Left deliberately unfixed, and worth separate changes: the scan still
collects into a Vec rather than streaming through a visitor as Go's
ScanVolumeFileFrom does, and it runs while holding store.read(), the same
lock-across-a-large-read shape as #11235. Both are latent once the rescan
is gone, since remaining scans are bounded by genuinely new data.

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

* rust volume: resolve and scan the tail under one store guard

The tail sender took store.read() once for the binary search and again
for the scan. A vacuum commit takes the store write lock and swaps
.dat/.idx, so it could land between the two: the offset resolved against
the old files would then be applied to the new ones and start the scan
inside an unrelated record. The code before the is_last fix held a
single guard for both. Restore that, and scan only when the caller is not
caught up, so the caught-up heartbeat still skips the scan and is sent
outside the lock.

Also pin the compacted-volume boundary raised in review. Compaction
writes .idx in needle-id order in both Go and Rust, so the search can
report caught-up while an earlier row is newer; such a caller's since_ns
is the last row's timestamp, so those rows were in the files it copied.
A write made afterwards is appended as the final row, which the search
cannot step past. The new test asserts it still reaches the scan.

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

* rust volume: make the compaction tail test a genuine overwrite

The compaction regression test's second id=1 write reused the first
write's data, so write_needle's dedup short-circuit (is_file_unchanged)
returned without appending or updating append_at_ns. Compaction then
kept key 1's original (older) timestamp, so the test passed without
exercising the overwrite it describes -- key 2 was the final row only
because key 1 was never actually newer.

Give the overwrite distinct data so it appends a new record, and assert
key1_ns > key2_ns up front so a future dedup regression fails the test
instead of silently hollowing it out. Trim the verbose comments on the
tail sender and the binary-search tests to their essentials.

---------

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-11 09:42:20 -07:00
committed by GitHub
co-authored by Claude Opus 5 Chris Lu
parent 0de9c1f231
commit 3ae9e332ec
2 changed files with 245 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+69 -34
View File
@@ -2714,56 +2714,91 @@ impl VolumeServer for VolumeGrpcService {
let mut draining_seconds = idle_timeout as i64;
loop {
// Use binary search to find starting offset, then scan from there
let scan_result = {
// 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();
if let Some((_, vol)) = store.find_volume(vid) {
let start_offset = if last_timestamp_ns > 0 {
match vol.binary_search_by_append_at_ns(last_timestamp_ns) {
Ok((offset, _is_last)) => {
if offset.is_zero() {
Ok(sb_size)
} else {
Ok(offset.to_actual_offset() as u64)
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
))
}
}
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))
}
}
} else {
Ok(sb_size)
};
match start_offset {
Ok(off) => Ok(vol.scan_raw_needles_from(off)),
Err(msg) => Err(msg),
}))
}
} else {
break;
None => None,
}
};
let scan_inner = match scan_result {
Ok(r) => r,
Err(msg) => {
let scan_result = match resolved {
None => break,
Some(Err(msg)) => {
let _ = tx.send(Err(Status::internal(msg))).await;
return;
}
Some(Ok(scan_result)) => scan_result,
};
let entries = match scan_inner {
// 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() {
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;
seaweed-volume/src/storage/volume.rs
+176
View File
@@ -4681,6 +4681,182 @@ mod tests {
assert!(Path::new(&v.file_name(".idx")).exists());
}
#[test]
fn binary_search_reports_is_last_when_caller_is_caught_up() {
// is_last drives the caught-up heartbeat; if it stops reporting true
// for an up-to-date caller, the tail sender re-reads the whole volume.
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let mut v = make_test_volume(dir);
let mut newest_ns = 0u64;
for id in 1..=3u64 {
let mut n = Needle {
id: NeedleId(id),
cookie: Cookie(0x12345678),
data: vec![b'x'; 64],
data_size: 64,
flags: 0,
..Needle::default()
};
v.write_needle(&mut n, true, false).unwrap();
newest_ns = n.append_at_ns;
}
assert!(newest_ns > 0, "needles must carry an append timestamp");
// Caught up: asking from the newest timestamp has nothing newer.
let (_offset, is_last) = v.binary_search_by_append_at_ns(newest_ns).unwrap();
assert!(
is_last,
"a caller at the newest append_at_ns must be reported as caught up"
);
// And from beyond the newest, likewise.
let (_offset, is_last_future) = v
.binary_search_by_append_at_ns(newest_ns + 1_000_000_000)
.unwrap();
assert!(
is_last_future,
"a caller ahead of the newest needle must be reported as caught up"
);
}
#[test]
fn binary_search_does_not_report_is_last_when_data_is_newer() {
// Complement: a behind caller must NOT be caught up, or tailing loses data.
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let mut v = make_test_volume(dir);
let mut first_ns = 0u64;
for id in 1..=3u64 {
let mut n = Needle {
id: NeedleId(id),
cookie: Cookie(0x12345678),
data: vec![b'y'; 64],
data_size: 64,
flags: 0,
..Needle::default()
};
v.write_needle(&mut n, true, false).unwrap();
if id == 1 {
first_ns = n.append_at_ns;
}
}
let (_offset, is_last) = v.binary_search_by_append_at_ns(first_ns).unwrap();
assert!(
!is_last,
"a caller behind the newest needle must NOT be reported as caught up"
);
}
#[test]
fn binary_search_does_not_report_is_last_when_only_a_delete_is_newer() {
// A delete is data the tail must ship. A caught-up caller before the
// delete must be sent to scan from the tombstone, not handed a heartbeat.
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let mut v = make_test_volume(dir);
let mut newest_ns = 0u64;
for id in 1..=3u64 {
let mut n = Needle {
id: NeedleId(id),
cookie: Cookie(0x12345678),
data: vec![b'z'; 64],
data_size: 64,
flags: 0,
..Needle::default()
};
v.write_needle(&mut n, true, false).unwrap();
newest_ns = n.append_at_ns;
}
let (_offset, is_last) = v.binary_search_by_append_at_ns(newest_ns).unwrap();
assert!(is_last, "precondition: the caller starts caught up");
v.delete_needle(&mut Needle {
id: NeedleId(2),
cookie: Cookie(0x12345678),
..Needle::default()
})
.unwrap();
let (offset, is_last) = v.binary_search_by_append_at_ns(newest_ns).unwrap();
assert!(
!is_last,
"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();
assert_eq!(shipped.len(), 1, "the tail must ship the tombstone");
}
#[test]
fn binary_search_on_compacted_volume_still_reports_a_later_write() {
// Compaction rewrites .idx in needle-id order, so append_at_ns is no
// longer monotonic by row: an overwritten key can sit before an older
// one. The caller's since_ns is the last row's timestamp, so such rows
// were already in the files it copied. A write made afterwards is
// appended as the final row, which the search cannot step past.
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
let mut v = make_test_volume(dir);
let write = |v: &mut Volume, id: u64, byte: u8| {
let mut n = Needle {
id: NeedleId(id),
cookie: Cookie(0x12345678),
data: vec![byte; 64],
data_size: 64,
flags: 0,
..Needle::default()
};
v.write_needle(&mut n, true, false).unwrap();
n.append_at_ns
};
write(&mut v, 1, b'c');
let key2_ns = write(&mut v, 2, b'c');
let key1_ns = write(&mut v, 1, b'd'); // genuine overwrite: different data
assert!(
key1_ns > key2_ns,
"precondition: the overwrite must append a newer record, not dedup"
);
v.compact_by_index(0, 0, |_| true).unwrap();
v.commit_compact().unwrap();
let (_offset, is_last) = v.binary_search_by_append_at_ns(key2_ns).unwrap();
assert!(
is_last,
"precondition: compaction ordered .idx by key, so key 2 is the final row"
);
let key3_ns = write(&mut v, 3, b'c');
let (offset, is_last) = v.binary_search_by_append_at_ns(key2_ns).unwrap();
assert!(
!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();
assert!(
shipped.contains(&key3_ns),
"the tail must ship the write made after compaction"
);
}
#[test]
fn test_volume_write_read() {
let tmp = TempDir::new().unwrap();