From 3ae9e332ec2d45ce5c4c9802d41f9191174a1d03 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Fri, 11 Sep 2026 19:42:20 +0300 Subject: [PATCH] rust volume: honour is_last in the tail sender instead of rescanning the whole volume (#11273) * 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) 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) 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) Co-authored-by: Chris Lu --- seaweed-volume/src/server/grpc_server.rs | 103 ++++++++----- seaweed-volume/src/storage/volume.rs | 176 +++++++++++++++++++++++ 2 files changed, 245 insertions(+), 34 deletions(-) diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 059aedbe3..bf201098e 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -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; diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 229315986..e04707eba 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -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 = 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 = 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();