From cba409cae73574c3d20a3251bfbe6752a96b3e89 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 22:37:31 -0700 Subject: [PATCH] backup: carry index-only tombstones through incremental copies (#12231) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume server: add idx_content to VolumeIncrementalCopyResponse A remote-tier delete is an index-only tombstone row with no .dat record, so it cannot reach an incremental copy through the .dat stream. Carrying the .idx tail verbatim lets the backup reproduce those rows at their exact positions. * backup: carry index-only tombstones through incremental copies A remote-tier delete leaves an offset-0 tombstone in .idx with no .dat record. The copy only streamed .dat bytes and regenerated .idx from them, so the tombstone never reached the backup: the deleted needle stayed live and every later incremental diverged and rebuilt again. backupCursor and ResumeByLastEntry now accept an index-only tombstone as a boundary, matched by key and size since there is no record to read a cookie from. VolumeIncrementalCopy appends the .idx tail after the .dat tail, and IncrementalBackup writes it at the index boundary, truncates, and loads the new rows — so tombstones land at the exact positions the replica has them. * seaweed-volume: mirror index-only tombstone handling in incremental copy resume_by_last_entry accepts an offset-0 tombstone boundary matched on key and size, skips its missing cookie check, and resumes .dat streaming at the next entry that has a record — index-only rows ride the new idx_content stream instead of reporting divergence. The server captures the .idx bound before the .dat one (a row is published only after its bytes) and streams the tail after the data. * backup: keep incremental copy bounds and streams consistent Review fixes: * Bound capture moved after the resume reads and reads .dat/.idx sizes together (Go FileStat, Rust file_stat under one lock), so every row resume saw is inside the index bound and each row's record is always inside the data bound. * A compaction commit changes CompactionRevision: the server now rejects the copy when the revision moved between resume, file opens, or the end of the stream, so a backup never mixes revisions. The .idx is opened up front to pin the inode. * An index tail that ends before the captured bound is an error, not a silent success. * IncrementalBackup rolls the .idx tail and .dat bytes back to the pre-copy boundary when the stream fails, so a partial row cannot wedge backupCursor and unapplied rows cannot pass for current ones. * The index tail is replayed through a small indexTailReceiver interface, covering the leveldb needle maps too, not just the in-memory map. * backup: bound the resume scan to the captured index size The resume and the stream bounds are now captured in one order: revision, then the .dat/.idx pair, then the scan. ResumeByLastEntry and resume_by_last_entry take the index bound and scan no further, so a row appended while the copy is being set up can never enter the index stream without its .dat bytes. The index-tail replay stays on the in-memory map only: the leveldb map's DoOffsetLoading writes to the .cpldb compaction database, not the live one, so it cannot serve a live tail replay; other maps keep regenerating the index from .dat as before. * seaweed-volume: guard incremental copy total against a located offset past .dat A corrupt index can place the resume offset beyond the captured .dat size; has_data already refuses the reader, but total still evaluates dat_size - s and underflows. * backup: bound the incremental .dat tail by index-covered records A record staged but not yet indexed can sit inside the captured .dat bound while its row stays outside the index bound — the fsync gap between staging and publishing is not covered by one stat. Streaming those bytes leaves them unindexed on the backup, and the next run appends the record again. Bound the .dat tail by the last record the streamed rows cover on both producers, and cover the bound walk with tests. * backup: cap the .dat stream at the captured bound and roll back replay failures sendFileContent could read past stopOffset on its last chunk, so a needle appended after FileStat arrived without its index row and the next copy appended at the wrong position. Each read is now limited to the remaining captured bytes. A failed truncateIndex or DoOffsetLoading after a successful stream left the index tail on disk unapplied, which a later backupCursor could mistake for applied rows; both now roll the tail back to the pre-copy boundary like the stream-error path. * backup: bound the incremental .dat tail by the maximum covered offset A key-sorted .idx ends on the highest-key row, which need not be the record physically last in .dat, so the last-row bound could cut off live records. Take the end of the highest-offset row inside the bound instead, and cover the case with a sorted-idx test in Go and Rust. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * backup: restore .dat and the needle map when index replay fails A failed DoOffsetLoading left the received .dat bytes in place and could leave partial mappings for rows the index rollback just removed, so the next copy appended the same records again or read stale entries. Truncate .dat back to the copy boundary and restore the touched mappings and metrics from a pre-replay snapshot. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * backup: cover no index rows with the superblock bound When no row inside the index bound names a record — an empty index or a bound of index-only tombstones — the covered end is the superblock, not the captured .dat size, so staged bytes are never streamed without a covering row. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * seaweed-volume: apply the covered bound even when the index streams no rows The bound was gated on idx_end > idx_start, so an empty captured index skipped it entirely and a staged .dat tail still shipped. Gate on the stream-index predicate instead, matching Go's streamIndex check. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * backup: propagate index coverage read errors * backup: rebuild the needle map from the truncated index on replay failure * seaweed-volume: propagate index coverage read errors --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- seaweed-volume/proto/volume_server.proto | 4 + seaweed-volume/src/server/grpc_server.rs | 198 +++++++++++-- seaweed-volume/src/storage/volume.rs | 265 ++++++++++++++++-- weed/pb/volume_server.proto | 4 + weed/pb/volume_server_pb/volume_server.pb.go | 23 +- weed/server/volume_grpc_copy_incremental.go | 84 +++++- .../volume_grpc_copy_incremental_test.go | 124 ++++++++ weed/storage/needle_map.go | 17 +- weed/storage/needle_map_leveldb.go | 20 -- weed/storage/needle_map_memory.go | 13 +- weed/storage/needle_map_metric.go | 11 - weed/storage/needle_map_sorted_file.go | 8 - weed/storage/volume_backup.go | 151 ++++++++-- weed/storage/volume_backup_test.go | 142 +++++++++- 14 files changed, 925 insertions(+), 139 deletions(-) create mode 100644 weed/server/volume_grpc_copy_incremental_test.go diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 34d4375fff..b36932c9eb 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -243,6 +243,10 @@ message VolumeIncrementalCopyRequest { } message VolumeIncrementalCopyResponse { bytes file_content = 1; + // Index rows after the copy's entry_count, verbatim. A remote-tier + // tombstone is an index-only row with no .dat record, so it can only + // reach the backup through this stream. + bytes idx_content = 2; } message VolumeMountRequest { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 2f2a369a7a..3f861b0017 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1584,9 +1584,17 @@ impl VolumeServer for VolumeGrpcService { return Err(Status::unavailable(e.to_string())); } - let dat_size = v.dat_file_size().unwrap_or(0); let super_block_size = v.super_block().block_size() as u64; + // Capture the revision and the .dat/.idx bounds before the resume + // reads, so the resume scans only rows the streams can actually carry: + // a row inside the index bound always has its record inside the data + // bound (a row publishes only after its bytes). + let revision = v.super_block().compaction_revision; + let (dat_size, idx_size) = v + .file_stat() + .map_err(|e| Status::internal(format!("stat volume {}: {}", vid, e)))?; + // If since_ns is very large (after all data), return empty. A // positional resume must still reach resume_by_last_entry so an empty // replica reports divergence instead of a clean empty stream. @@ -1599,20 +1607,29 @@ impl VolumeServer for VolumeGrpcService { // Resume by index position when the backup identified its last entry; // append_at_ns stamps differ across replicas, so a since_ns resume can // duplicate or skip needles after the source replica changes. - let start_offset = if req.entry_count > 0 { + let start_offset: Option = if req.entry_count > 0 { match v.resume_by_last_entry( req.entry_count, NeedleId(req.last_needle_key), Cookie(req.last_needle_cookie), Size(req.last_needle_size), req.entries_hash, + idx_size, ) { Ok((_offset, true)) => { drop(store); let stream = tokio_stream::iter(Vec::new()); return Ok(Response::new(Box::pin(stream))); } - Ok((offset, false)) => offset.to_actual_offset() as u64, + Ok((offset, false)) => { + // index-only rows like remote-tier tombstones have no .dat + // record; they ride the index stream instead + if offset.is_zero() { + None + } else { + Some(offset.to_actual_offset() as u64) + } + } Err(crate::storage::volume::VolumeError::IncrementalCopyDiverged) => { return Err(Status::failed_precondition( crate::storage::volume::VolumeError::IncrementalCopyDiverged.to_string(), @@ -1626,7 +1643,7 @@ impl VolumeServer for VolumeGrpcService { } } } else if req.since_ns == 0 { - super_block_size + Some(super_block_size) } else { match v.binary_search_by_append_at_ns(req.since_ns) { Ok((_offset, true)) => { @@ -1637,11 +1654,11 @@ impl VolumeServer for VolumeGrpcService { } Ok((offset, false)) => { let actual = offset.to_actual_offset(); - if actual <= 0 { + Some(if actual <= 0 { super_block_size } else { actual as u64 - } + }) } Err(e) => { return Err(Status::internal(format!( @@ -1651,9 +1668,37 @@ impl VolumeServer for VolumeGrpcService { } } }; + let idx_start = req.entry_count * NEEDLE_MAP_ENTRY_SIZE as u64; + let idx_end = if req.entry_count > 0 || req.since_ns == 0 { + idx_size + } else { + idx_start + }; + // Bound the .dat tail by the records the streamed rows cover: a + // record staged but not yet indexed would reach the backup without + // its row and be appended again on the next run. + let dat_size = if req.entry_count > 0 || req.since_ns == 0 { + let covered = match v.dat_size_covered_by_index(idx_size) { + Ok(covered) => covered, + Err(e) => { + return Err(Status::internal(format!( + "fail to bound data by covered index: {}", + e + ))); + } + }; + if covered > 0 && covered < dat_size { + covered + } else { + dat_size + } + } else { + dat_size + }; // A corrupt index could place the located offset past the captured // `.dat` size; nothing to send, and it would underflow `total` below. - if start_offset >= dat_size { + let has_data = start_offset.is_some_and(|s| s < dat_size); + if !has_data && idx_end <= idx_start { drop(store); let stream = tokio_stream::iter(Vec::new()); return Ok(Response::new(Box::pin(stream))); @@ -1670,26 +1715,46 @@ impl VolumeServer for VolumeGrpcService { Local(std::fs::File), Remote(crate::storage::volume::RemoteDatFile), } - let reader = if v.has_remote_file() { - match v.remote_dat_file() { - Some(r) => DatReader::Remote(r), - None => { - return Err(Status::not_found(format!( - "remote dat not loaded for volume id {}", - vid - ))); + let reader = if has_data { + if v.has_remote_file() { + match v.remote_dat_file() { + Some(r) => Some(DatReader::Remote(r)), + None => { + return Err(Status::not_found(format!( + "remote dat not loaded for volume id {}", + vid + ))); + } } + } else { + let path = v.file_name(".dat"); + let file = std::fs::File::open(&path) + .map_err(|e| Status::internal(format!("open {}: {}", path, e)))?; + Some(DatReader::Local(file)) } } else { - let path = v.file_name(".dat"); + None + }; + let idx_file = if idx_end > idx_start { + let path = v.file_name(".idx"); let file = std::fs::File::open(&path) .map_err(|e| Status::internal(format!("open {}: {}", path, e)))?; - DatReader::Local(file) + Some(file) + } else { + None }; + // A compaction swap between the bound capture and these opens would + // mix a new file with bounds from the old revision; the backup must + // rebuild instead of copying a mismatched pair. + if v.super_block().compaction_revision != revision { + return Err(Status::failed_precondition( + crate::storage::volume::VolumeError::IncrementalCopyDiverged.to_string(), + )); + } let state = self.state.clone(); drop(store); - let total = dat_size - start_offset; + let total = start_offset.map_or(0, |s| dat_size.saturating_sub(s)); let (tx, rx) = tokio::sync::mpsc::channel::< Result, >(8); @@ -1710,10 +1775,10 @@ impl VolumeServer for VolumeGrpcService { } let buffer_size = 2 * 1024 * 1024u64; // 2MB chunks let mut bytes_to_read = total; - let mut offset = start_offset; + let mut offset = start_offset.unwrap_or(0); match reader { - DatReader::Local(file) => { + Some(DatReader::Local(file)) => { // Local volume: stream directly from the .dat, no lock held. use std::io::{Read, Seek, SeekFrom}; let mut reader = std::io::BufReader::new(file); @@ -1731,6 +1796,7 @@ impl VolumeServer for VolumeGrpcService { buf.truncate(n); let msg = volume_server_pb::VolumeIncrementalCopyResponse { file_content: buf, + ..Default::default() }; if tx.blocking_send(Ok(msg)).is_err() { return; // receiver dropped @@ -1744,7 +1810,7 @@ impl VolumeServer for VolumeGrpcService { } } } - DatReader::Remote(remote) => { + Some(DatReader::Remote(remote)) => { // Remote/tiered volume: read through the cloned backend // handle. No store lock is held while the (potentially slow) // S3 fetch runs, so it never blocks store writers. @@ -1757,6 +1823,7 @@ impl VolumeServer for VolumeGrpcService { let read_len = buf.len() as u64; let msg = volume_server_pb::VolumeIncrementalCopyResponse { file_content: buf, + ..Default::default() }; if tx.blocking_send(Ok(msg)).is_err() { return; // receiver dropped @@ -1771,6 +1838,50 @@ impl VolumeServer for VolumeGrpcService { } } } + None => {} + } + + // the .idx tail after the .dat: index-only rows like remote-tier + // tombstones have no .dat record and reach the copy only here + if let Some(file) = idx_file { + use std::io::{Read, Seek, SeekFrom}; + let mut reader = std::io::BufReader::new(file); + if let Err(e) = reader.seek(SeekFrom::Start(idx_start)) { + let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); + return; + } + let mut bytes_to_read = idx_end - idx_start; + while bytes_to_read > 0 { + bail_if_unavailable!(); + let chunk = std::cmp::min(bytes_to_read, buffer_size) as usize; + let mut buf = vec![0u8; chunk]; + match reader.read(&mut buf) { + Ok(0) => { + // a shortened .idx cannot deliver the captured + // tail; the copy must restart rather than install + // a partial index + let _ = tx.blocking_send(Err(Status::internal( + "index file ended before the captured tail", + ))); + return; + } + Ok(n) => { + buf.truncate(n); + let msg = volume_server_pb::VolumeIncrementalCopyResponse { + idx_content: buf, + ..Default::default() + }; + if tx.blocking_send(Ok(msg)).is_err() { + return; // receiver dropped + } + bytes_to_read -= n as u64; + } + Err(e) => { + let _ = tx.blocking_send(Err(Status::internal(e.to_string()))); + return; + } + } + } } }); @@ -8440,6 +8551,51 @@ mod tests { .remove("s3.incr_copy_test"); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_incremental_copy_streams_index_tail_with_tombstones() { + let (service, tmp, shutdown_tx, _dat_bytes, _super_block_size, _delete_count) = + make_remote_only_service("incr_copy_idx"); + + // a delete on a tiered volume leaves an index-only tombstone row + { + let store = service.state.store.read().unwrap(); + store + .delete_volume_needle( + VolumeId(1), + &mut Needle { + id: NeedleId(7), + ..Needle::default() + }, + ) + .unwrap(); + } + let idx_bytes = std::fs::read(format!("{}/1.idx", tmp.path().to_str().unwrap())).unwrap(); + + let response = service + .volume_incremental_copy(Request::new( + volume_server_pb::VolumeIncrementalCopyRequest { + volume_id: 1, + since_ns: 0, + ..Default::default() + }, + )) + .await + .unwrap(); + + let mut stream = response.into_inner(); + let mut copied_idx = Vec::new(); + while let Some(message) = stream.next().await { + copied_idx.extend_from_slice(&message.unwrap().idx_content); + } + assert_eq!(copied_idx, idx_bytes); + + let _ = shutdown_tx.send(()); + global_s3_tier_registry() + .write() + .unwrap() + .remove("s3.incr_copy_idx"); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_tier_move_from_remote_swaps_to_local_and_deletes_object() { let (service, tmp, shutdown_tx, dat_bytes, _super_block_size, delete_count) = diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index cfbffe1d0f..14a40e7dcf 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -4891,11 +4891,15 @@ impl Volume { /// Resume an incremental copy by index position instead of by append_at_ns, /// whose stamps differ across replicas for the same needle. The replica's /// entry at position entry_count-1 must match the backup's last entry in - /// key, size and cookie, and its first entry_count entries must hash the - /// same — a boundary match alone is not proof, since replicated writes can - /// land in a different order on this replica. The copy then continues from - /// the replica's entry at position entry_count, so nothing is skipped or - /// appended twice. + /// key and size — and cookie, unless it is an index-only tombstone + /// (a remote-tier delete), which has no .dat record to read a cookie + /// from — and its first entry_count entries must hash the same — a + /// boundary match alone is not proof, since replicated writes can land + /// in a different order on this replica. The copy then continues from the + /// replica's next entry with a .dat record; index-only rows ride the + /// index stream, so nothing is skipped or appended twice. index_bound + /// caps the scan to the index size the caller captured, so rows appended + /// after that point stay out of this copy. /// Mirrors Go's ResumeByLastEntry in weed/storage/volume_backup.go. /// Returns (offset, up_to_date). pub fn resume_by_last_entry( @@ -4905,6 +4909,7 @@ impl Volume { cookie: Cookie, size: Size, entries_hash: u64, + index_bound: u64, ) -> Result<(Offset, bool), VolumeError> { let io = self.io.read().unwrap(); let file_size = io.nm.as_ref().map_or(0, |nm| nm.index_file_size()) as i64; @@ -4924,15 +4929,13 @@ impl Volume { if last_key != key || last_size != size { return Err(VolumeError::IncrementalCopyDiverged); } - // a remote-tier tombstone sits in the index with no .dat record to verify - if last_offset.is_zero() { - return Err(VolumeError::IncrementalCopyDiverged); - } - let mut header_buf = [0u8; NEEDLE_HEADER_SIZE]; - self.read_exact_at_backend(&io, &mut header_buf, last_offset.to_actual_offset() as u64)?; - let (last_cookie, _id, _s) = Needle::parse_header(&header_buf); - if last_cookie != cookie { - return Err(VolumeError::IncrementalCopyDiverged); + if !last_offset.is_zero() { + let mut header_buf = [0u8; NEEDLE_HEADER_SIZE]; + self.read_exact_at_backend(&io, &mut header_buf, last_offset.to_actual_offset() as u64)?; + let (last_cookie, _id, _s) = Needle::parse_header(&header_buf); + if last_cookie != cookie { + return Err(VolumeError::IncrementalCopyDiverged); + } } if self.hash_index_entries(entry_count)? != entries_hash { @@ -4942,11 +4945,17 @@ impl Volume { if server_count == entry_count { return Ok((Offset::default(), true)); } - let (_k, offset, _s) = self.read_index_entry(entry_count as i64)?; - if offset.to_actual_offset() <= 0 { - return Err(VolumeError::IncrementalCopyDiverged); + let scan_end = std::cmp::min( + server_count as i64, + (index_bound / NEEDLE_MAP_ENTRY_SIZE as u64) as i64, + ); + for i in entry_count as i64..scan_end { + let (_k, offset, _s) = self.read_index_entry(i)?; + if !offset.is_zero() { + return Ok((offset, false)); + } } - Ok((offset, false)) + Ok((Offset::default(), false)) } /// Write a raw needle blob at a specific offset in the .dat file. @@ -5168,6 +5177,67 @@ impl Volume { self.current_dat_file_size() } + /// Size of the .idx file in bytes, from the needle map's append offset. + /// Mirrors Go's Volume.IndexFileSize. + pub fn index_file_size(&self) -> u64 { + self.io + .read() + .unwrap() + .nm + .as_ref() + .map_or(0, |nm| nm.index_file_size()) + } + + /// The .dat and .idx sizes read while excluding writes, so a row inside + /// the index bound always has its record inside the data bound. + /// Mirrors Go's Volume.FileStat. + pub(crate) fn file_stat(&self) -> io::Result<(u64, u64)> { + let _guard = self.data_file_access_control.read_lock(); + let io = self.io.read().unwrap(); + let dat_size = self.current_dat_file_size_io(&io)?; + let idx_size = io.nm.as_ref().map_or(0, |nm| nm.index_file_size()); + Ok((dat_size, idx_size)) + } + + /// .dat end of the last record the index rows before `index_size` + /// cover, or the superblock size when no row inside the bound names a + /// record. Rows are not always in .dat order — rebuilds can write the + /// .idx sorted by key — so the bound comes from the maximum offset: + /// records do not overlap, so the highest offset names the highest end. + /// Records appended after the bound — including staged writes whose rows + /// are not published yet — lie past it, so a copy must not stream them. + /// Mirrors Go's DatSizeCoveredByIndex in weed/storage/volume_backup.go. + pub(crate) fn dat_size_covered_by_index(&self, index_size: u64) -> Result { + let bound_rows = index_size / NEEDLE_MAP_ENTRY_SIZE as u64; + let idx_path = self.file_name(".idx"); + let idx_file = File::open(&idx_path)?; + let mut max_actual_offset: i64 = -1; + let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; + let mut reader = io::BufReader::new(idx_file); + for _ in 0..bound_rows { + match reader.read_exact(&mut buf) { + Ok(()) => {} + Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => break, + Err(e) => return Err(e.into()), + } + let (_key, offset, _size) = idx_entry_from_bytes(&buf); + if !offset.is_zero() { + let actual_offset = offset.to_actual_offset(); + if actual_offset > max_actual_offset { + max_actual_offset = actual_offset; + } + } + } + if max_actual_offset < 0 { + return Ok(self.super_block().block_size() as u64); + } + let io = self.io.read().unwrap(); + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + self.read_exact_at_backend(&io, &mut header, max_actual_offset as u64)?; + let (_cookie, _id, size) = Needle::parse_header(&header); + Ok(max_actual_offset as u64 + get_actual_size(size, self.version()) as u64) + } + /// Disk space held by the local .dat file, or 0 when it is unknown. It /// exceeds the file size when the file was preallocated with /// `FALLOC_FL_KEEP_SIZE`. @@ -6630,6 +6700,165 @@ mod tests { ); } + #[test] + fn resume_by_last_entry_carries_index_only_tombstones() { + // A remote-tier delete leaves an offset-0 tombstone in .idx with no + // .dat record: the row still names a position a copy can end on or + // resume past, so it is matched by key and size rather than diverging. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + for id in 1..=5u64 { + write_test_needle(&mut v, id, &[id as u8; 64]); + } + + let entry_at = |v: &Volume, m: i64| { + let (key, offset, size) = v.read_index_entry(m).unwrap(); + let io = v.io.read().unwrap(); + let mut header = [0u8; NEEDLE_HEADER_SIZE]; + v.read_exact_at_backend(&io, &mut header, offset.to_actual_offset() as u64) + .unwrap(); + let (cookie, _id, _s) = Needle::parse_header(&header); + (key, cookie, size) + }; + + // a copy whose tail is the replica's tail is already up to date + let (key, cookie, size) = entry_at(&v, 4); + let (_offset, up_to_date) = v + .resume_by_last_entry( + 5, + key, + cookie, + size, + v.hash_index_entries(5).unwrap(), + v.index_file_size(), + ) + .unwrap(); + assert!(up_to_date); + + // a remote-tier tombstone is a valid boundary matched on key and size + v.io.write() + .unwrap() + .nm + .as_mut() + .unwrap() + .delete(NeedleId(4), Offset::default()) + .unwrap(); + let (_offset, up_to_date) = v + .resume_by_last_entry( + 6, + NeedleId(4), + Cookie(0), + TOMBSTONE_FILE_SIZE, + v.hash_index_entries(6).unwrap(), + v.index_file_size(), + ) + .unwrap(); + assert!(up_to_date); + + // a resume past it keeps going to the next record with .dat bytes + write_test_needle(&mut v, 6, b"x"); + let (key, cookie, size) = entry_at(&v, 4); + let (offset, up_to_date) = v + .resume_by_last_entry( + 5, + key, + cookie, + size, + v.hash_index_entries(5).unwrap(), + v.index_file_size(), + ) + .unwrap(); + assert!(!up_to_date); + let (_k, want, _s) = v.read_index_entry(6).unwrap(); + assert_eq!(offset, want); + + // a copy ahead of only tombstones resumes with no .dat left to send + v.io.write() + .unwrap() + .nm + .as_mut() + .unwrap() + .delete(NeedleId(5), Offset::default()) + .unwrap(); + let (key, cookie, size) = entry_at(&v, 6); + let (offset, up_to_date) = v + .resume_by_last_entry( + 7, + key, + cookie, + size, + v.hash_index_entries(7).unwrap(), + v.index_file_size(), + ) + .unwrap(); + assert!(!up_to_date); + assert!(offset.is_zero()); + } + + #[test] + fn dat_size_covered_by_index_ends_at_last_record() { + // A copy must not stream past the last record the index rows cover: + // bytes appended after the bound have no row to carry them. + // Mirrors Go's TestDatSizeCoveredByIndex. + 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, &[id as u8; 64]); + } + let dat_size = v.dat_file_size().unwrap(); + let idx_size = v.index_file_size(); + assert_eq!(v.dat_size_covered_by_index(idx_size).unwrap(), dat_size); + + // a bound partway through ends where the next record begins + let (_k, next, _s) = v.read_index_entry(2).unwrap(); + assert_eq!( + v.dat_size_covered_by_index(2 * NEEDLE_MAP_ENTRY_SIZE as u64) + .unwrap(), + next.to_actual_offset() as u64 + ); + + // index-only tombstones past the bound still resolve to the last record + v.io.write() + .unwrap() + .nm + .as_mut() + .unwrap() + .delete(NeedleId(3), Offset::default()) + .unwrap(); + assert_eq!( + v.dat_size_covered_by_index(v.index_file_size()).unwrap(), + dat_size + ); + } + + #[test] + fn dat_size_covered_by_index_key_sorted() { + // A key-sorted .idx ends on the highest-key row, which need not be + // the record physically last in .dat — the bound must still cover + // the whole covered tail. Mirrors Go's + // TestDatSizeCoveredByIndexKeySorted. + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + for id in [2u64, 3, 1] { + write_test_needle(&mut v, id, &[id as u8; 64]); + } + let dat_size = v.dat_file_size().unwrap(); + let idx_size = v.index_file_size(); + + // rewrite the .idx sorted by key, as rebuilds do + let idx_path = v.file_name(".idx"); + let content = fs::read(&idx_path).unwrap(); + let mut rows: Vec<&[u8]> = content.chunks(NEEDLE_MAP_ENTRY_SIZE).collect(); + rows.sort_by_key(|row| idx_entry_from_bytes(row).0); + let sorted: Vec = rows.concat(); + fs::write(&idx_path, sorted).unwrap(); + + assert_eq!(v.dat_size_covered_by_index(idx_size).unwrap(), dat_size); + } + #[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 diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 3c96ce1401..905813e43b 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -243,6 +243,10 @@ message VolumeIncrementalCopyRequest { } message VolumeIncrementalCopyResponse { bytes file_content = 1; + // Index rows after the copy's entry_count, verbatim. A remote-tier + // tombstone is an index-only row with no .dat record, so it can only + // reach the backup through this stream. + bytes idx_content = 2; } message VolumeMountRequest { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index dc04bc08c0..4e131ce276 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v3.21.12 +// protoc v7.35.0 // source: volume_server.proto package volume_server_pb @@ -1223,8 +1223,12 @@ func (x *VolumeIncrementalCopyRequest) GetEntriesHash() uint64 { } type VolumeIncrementalCopyResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - FileContent []byte `protobuf:"bytes,1,opt,name=file_content,json=fileContent,proto3" json:"file_content,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + FileContent []byte `protobuf:"bytes,1,opt,name=file_content,json=fileContent,proto3" json:"file_content,omitempty"` + // Index rows after the copy's entry_count, verbatim. A remote-tier + // tombstone is an index-only row with no .dat record, so it can only + // reach the backup through this stream. + IdxContent []byte `protobuf:"bytes,2,opt,name=idx_content,json=idxContent,proto3" json:"idx_content,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1266,6 +1270,13 @@ func (x *VolumeIncrementalCopyResponse) GetFileContent() []byte { return nil } +func (x *VolumeIncrementalCopyResponse) GetIdxContent() []byte { + if x != nil { + return x.IdxContent + } + return nil +} + type VolumeMountRequest struct { state protoimpl.MessageState `protogen:"open.v1"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` @@ -7396,9 +7407,11 @@ const file_volume_server_proto_rawDesc = "" + "\x0flast_needle_key\x18\x04 \x01(\x04R\rlastNeedleKey\x12,\n" + "\x12last_needle_cookie\x18\x05 \x01(\rR\x10lastNeedleCookie\x12(\n" + "\x10last_needle_size\x18\x06 \x01(\x05R\x0elastNeedleSize\x12!\n" + - "\fentries_hash\x18\a \x01(\x04R\ventriesHash\"B\n" + + "\fentries_hash\x18\a \x01(\x04R\ventriesHash\"c\n" + "\x1dVolumeIncrementalCopyResponse\x12!\n" + - "\ffile_content\x18\x01 \x01(\fR\vfileContent\"e\n" + + "\ffile_content\x18\x01 \x01(\fR\vfileContent\x12\x1f\n" + + "\vidx_content\x18\x02 \x01(\fR\n" + + "idxContent\"e\n" + "\x12VolumeMountRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12#\n" + "\n" + diff --git a/weed/server/volume_grpc_copy_incremental.go b/weed/server/volume_grpc_copy_incremental.go index 49403fae2d..811d1767a4 100644 --- a/weed/server/volume_grpc_copy_incremental.go +++ b/weed/server/volume_grpc_copy_incremental.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "os" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -25,9 +26,29 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem return err } - stopOffset, _, _ := v.FileStat() + // capture the revision and the .dat/.idx bounds before the resume reads, + // so the resume scans only rows the streams below can actually carry + revision := v.SuperBlock.CompactionRevision + stopOffset, stopIndex, _ := v.FileStat() + + // a since_ns resume from an old client has no verified prefix, so the + // index rows are only safe to replay when the whole file is sent + streamIndex := req.EntryCount > 0 || req.SinceNs == 0 + if streamIndex { + // bound the .dat tail by the records the streamed rows cover: a + // record staged but not yet indexed would reach the backup without + // its row and be appended again on the next run + end, err := v.DatSizeCoveredByIndex(int64(stopIndex)) + if err != nil { + return fmt.Errorf("fail to bound data by covered index: %v", err) + } + if end > 0 && end < int64(stopOffset) { + stopOffset = uint64(end) + } + } var startOffset int64 + var hasData bool if req.EntryCount > 0 { foundOffset, upToDate, err := v.ResumeByLastEntry( req.EntryCount, @@ -35,6 +56,7 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem req.LastNeedleCookie, types.Size(req.LastNeedleSize), req.EntriesHash, + stopIndex, ) if err != nil { if errors.Is(err, storage.ErrIncrementalCopyDiverged) { @@ -45,6 +67,7 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem if upToDate { return nil } + hasData = !foundOffset.IsZero() startOffset = foundOffset.ToActualOffset() } else { foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs) @@ -54,11 +77,37 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem if isLastOne { return nil } + hasData = true startOffset = foundOffset.ToActualOffset() } + var indexFile *os.File + if streamIndex && int64(stopIndex) > int64(req.EntryCount)*types.NeedleMapEntrySize { + var err error + if indexFile, err = os.Open(v.FileName(".idx")); err != nil { + return fmt.Errorf("cannot read %s: %v", v.FileName(".idx"), err) + } + defer indexFile.Close() + if v.SuperBlock.CompactionRevision != revision { + return status.Error(codes.FailedPrecondition, storage.ErrIncrementalCopyDiverged.Error()) + } + } + buf := make([]byte, 1024*1024*2) - return sendFileContent(v, buf, startOffset, int64(stopOffset), stream) + if hasData { + if err := sendFileContent(v, buf, startOffset, int64(stopOffset), stream); err != nil { + return err + } + } + if indexFile != nil { + if err := sendIndexContent(v, indexFile, buf, int64(req.EntryCount)*types.NeedleMapEntrySize, int64(stopIndex), stream); err != nil { + return err + } + } + if v.SuperBlock.CompactionRevision != revision { + return status.Error(codes.FailedPrecondition, storage.ErrIncrementalCopyDiverged.Error()) + } + return nil } @@ -75,13 +124,42 @@ func (vs *VolumeServer) VolumeSyncStatus(ctx context.Context, req *volume_server } +func sendIndexContent(v *storage.Volume, indexFile *os.File, buf []byte, startOffset, stopOffset int64, stream volume_server_pb.VolumeServer_VolumeIncrementalCopyServer) error { + var blockSizeLimit = int64(len(buf)) + for i := int64(0); i < stopOffset-startOffset; i += blockSizeLimit { + if err := v.UnavailableError(); err != nil { + return err + } + read := stopOffset - startOffset - i + if read > blockSizeLimit { + read = blockSizeLimit + } + n, readErr := indexFile.ReadAt(buf[:read], startOffset+i) + if n > 0 { + resp := &volume_server_pb.VolumeIncrementalCopyResponse{} + resp.IdxContent = buf[:int64(n)] + if sendErr := stream.Send(resp); sendErr != nil { + return sendErr + } + } + if readErr != nil { + return readErr + } + } + return nil +} + func sendFileContent(v *storage.Volume, buf []byte, startOffset, stopOffset int64, stream volume_server_pb.VolumeServer_VolumeIncrementalCopyServer) error { var blockSizeLimit = int64(len(buf)) for i := int64(0); i < stopOffset-startOffset; i += blockSizeLimit { if err := v.UnavailableError(); err != nil { return err } - n, readErr := v.DataBackend.ReadAt(buf, startOffset+i) + read := stopOffset - startOffset - i + if read > blockSizeLimit { + read = blockSizeLimit + } + n, readErr := v.DataBackend.ReadAt(buf[:read], startOffset+i) if readErr == nil || readErr == io.EOF { resp := &volume_server_pb.VolumeIncrementalCopyResponse{} resp.FileContent = buf[:int64(n)] diff --git a/weed/server/volume_grpc_copy_incremental_test.go b/weed/server/volume_grpc_copy_incremental_test.go new file mode 100644 index 0000000000..7d5b0e65aa --- /dev/null +++ b/weed/server/volume_grpc_copy_incremental_test.go @@ -0,0 +1,124 @@ +package weed_server + +import ( + "bytes" + "os" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/idx" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// A delete on a tiered volume appends an offset-0 tombstone to .idx with no +// .dat record. The incremental copy has to carry that row: regenerating the +// backup's index from .dat alone resurrects the needle, and the row-count +// skew makes every later backup diverge and rebuild. +func TestIncrementalCopyCarriesRemoteTierTombstone(t *testing.T) { + store := tierUploadForDeletes(t, "000") + const vid = needle.VolumeId(1) + if _, err := store.DeleteVolumeNeedle(vid, &needle.Needle{Id: 2}); err != nil { + t.Fatalf("delete on tiered volume: %v", err) + } + source := store.GetVolume(vid) + wantIdxSize := source.IndexFileSize() + + port := serveGrpc(t, func(server *grpc.Server) { + volume_server_pb.RegisterVolumeServerServer(server, &VolumeServer{store: store}) + }) + serverAddress := pb.NewServerAddressWithGrpcPort("127.0.0.1:0", port) + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + + backupDir := t.TempDir() + backup, err := storage.NewVolume(backupDir, backupDir, "", vid, storage.NeedleMapInMemory, + &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, source.Version(), 0, 0) + if err != nil { + t.Fatalf("backup volume: %v", err) + } + + if err := backup.IncrementalBackup(serverAddress, dialOption); err != nil { + t.Fatalf("incremental backup: %v", err) + } + if got := backup.IndexFileSize(); got != wantIdxSize { + t.Fatalf("backup idx size %d, want %d", got, wantIdxSize) + } + sourceIdx, err := os.ReadFile(source.FileName(".idx")) + if err != nil { + t.Fatalf("read source idx: %v", err) + } + backupIdx, err := os.ReadFile(backup.FileName(".idx")) + if err != nil { + t.Fatalf("read backup idx: %v", err) + } + if !bytes.Equal(backupIdx, sourceIdx) { + t.Fatalf("backup idx differs from the replica's") + } + + idxFile, err := os.Open(backup.FileName(".idx")) + if err != nil { + t.Fatalf("open backup idx: %v", err) + } + seenDeleted := false + err = idx.WalkIndexFile(idxFile, 0, func(key types.NeedleId, offset types.Offset, size types.Size) error { + if key == 2 && (offset.IsZero() || size.IsDeleted()) { + seenDeleted = true + } + return nil + }) + idxFile.Close() + if err != nil { + t.Fatalf("walk backup idx: %v", err) + } + if !seenDeleted { + t.Fatalf("backup idx has no tombstone for the deleted needle") + } + backup.Close() + + checkStore := newVolumeCopyTestStore(t, backupDir) + checkVolume := checkStore.GetVolume(vid) + if checkVolume == nil { + t.Fatalf("reloaded backup volume not found") + } + if _, err := checkStore.ReadVolumeNeedle(vid, &needle.Needle{Id: 2}, nil, nil); err == nil { + t.Fatalf("the deleted needle reads on the backup") + } + if _, err := checkStore.ReadVolumeNeedle(vid, &needle.Needle{Id: 3}, nil, nil); err != nil { + t.Fatalf("a live needle does not read on the backup: %v", err) + } + + // the copy now ends on the tombstone row; the next run stays a no-op + // instead of diverging into a rebuild + if err := checkVolume.IncrementalBackup(serverAddress, dialOption); err != nil { + t.Fatalf("second incremental backup: %v", err) + } + + // a delete landing after the copy rides the index stream alone: the run + // appends just the tombstone row and the needle stays deleted + if _, err := store.DeleteVolumeNeedle(vid, &needle.Needle{Id: 3}); err != nil { + t.Fatalf("second delete on tiered volume: %v", err) + } + if err := checkVolume.IncrementalBackup(serverAddress, dialOption); err != nil { + t.Fatalf("incremental backup over a new tombstone: %v", err) + } + sourceIdx, err = os.ReadFile(source.FileName(".idx")) + if err != nil { + t.Fatalf("re-read source idx: %v", err) + } + backupIdx, err = os.ReadFile(checkVolume.FileName(".idx")) + if err != nil { + t.Fatalf("re-read backup idx: %v", err) + } + if !bytes.Equal(backupIdx, sourceIdx) { + t.Fatalf("backup idx differs from the replica's after a new tombstone") + } + if _, err := checkStore.ReadVolumeNeedle(vid, &needle.Needle{Id: 3}, nil, nil); err == nil { + t.Fatalf("the newly deleted needle still reads on the backup") + } +} diff --git a/weed/storage/needle_map.go b/weed/storage/needle_map.go index e13903e180..0f25c313f0 100644 --- a/weed/storage/needle_map.go +++ b/weed/storage/needle_map.go @@ -41,19 +41,6 @@ type batchIndexRollbacker interface { truncateIndex(offset int64) error } -// batchMapRollbacker restores the in-memory/durable needle mapping without -// touching the index file: a rolled-back batch rewrites the index wholesale -// via truncateIndex, so replay-correcting entries are not needed here. -type batchMapRollbacker interface { - removeMapping(key NeedleId) error - restoreMapping(key NeedleId, offset Offset, size Size) error -} - -type batchMetricRollbacker interface { - snapshotBatchMetrics() batchMapMetricSnapshot - restoreBatchMetrics(snapshot batchMapMetricSnapshot) -} - type baseNeedleMapper struct { mapMetric @@ -68,6 +55,10 @@ type TempNeedleMapper interface { UpdateNeedleMap(v *Volume, indexFile *os.File, opts *opt.Options, ldbTimeout int64) error } +func (nm *baseNeedleMapper) IndexFile() *os.File { + return nm.indexFile +} + func (nm *baseNeedleMapper) IndexFileSize() uint64 { stat, err := nm.indexFile.Stat() if err == nil { diff --git a/weed/storage/needle_map_leveldb.go b/weed/storage/needle_map_leveldb.go index 250e88e886..0080071e0f 100644 --- a/weed/storage/needle_map_leveldb.go +++ b/weed/storage/needle_map_leveldb.go @@ -197,26 +197,6 @@ func (m *LevelDbNeedleMap) truncateIndex(offset int64) error { return nil } -func (m *LevelDbNeedleMap) removeMapping(key NeedleId) error { - if m.ldbTimeout > 0 { - if err := m.ensureLdbLoaded(); err != nil { - return err - } - defer m.ldbAccessLock.RUnlock() - } - return levelDbDelete(m.db, key) -} - -func (m *LevelDbNeedleMap) restoreMapping(key NeedleId, offset Offset, size Size) error { - if m.ldbTimeout > 0 { - if err := m.ensureLdbLoaded(); err != nil { - return err - } - defer m.ldbAccessLock.RUnlock() - } - return levelDbWrite(m.db, key, offset, size, false, 0) -} - func getWatermark(db *leveldb.DB) uint64 { data, err := db.Get(watermarkKey, nil) if err != nil || len(data) != 8 { diff --git a/weed/storage/needle_map_memory.go b/weed/storage/needle_map_memory.go index 8ef9bf33b3..d80762c042 100644 --- a/weed/storage/needle_map_memory.go +++ b/weed/storage/needle_map_memory.go @@ -73,13 +73,12 @@ func (nm *NeedleMap) Delete(key NeedleId, offset Offset) error { nm.logDelete(deletedBytes) return nm.appendToIndexFile(key, offset, TombstoneFileSize) } -func (nm *NeedleMap) removeMapping(key NeedleId) error { - nm.m.Remove(NeedleId(key)) - return nil -} -func (nm *NeedleMap) restoreMapping(key NeedleId, offset Offset, size Size) error { - nm.m.Set(NeedleId(key), offset, size) - return nil + +// reset drops all mappings and counters, e.g. before the index is reloaded +// after a rolled-back batch +func (nm *NeedleMap) reset() { + nm.m = needle_map.NewCompactMap() + nm.restoreBatchMetrics(batchMapMetricSnapshot{}) } func (nm *NeedleMap) Close() { if nm.indexFile == nil { diff --git a/weed/storage/needle_map_metric.go b/weed/storage/needle_map_metric.go index 498ae49cc2..5a481c281c 100644 --- a/weed/storage/needle_map_metric.go +++ b/weed/storage/needle_map_metric.go @@ -36,17 +36,6 @@ type batchMapMetricSnapshot struct { maximumNeedleEnd int64 } -func (mm *mapMetric) snapshotBatchMetrics() batchMapMetricSnapshot { - return batchMapMetricSnapshot{ - deletionCounter: atomic.LoadUint32(&mm.DeletionCounter), - fileCounter: atomic.LoadUint32(&mm.FileCounter), - deletionByteCounter: atomic.LoadUint64(&mm.DeletionByteCounter), - fileByteCounter: atomic.LoadUint64(&mm.FileByteCounter), - maximumFileKey: atomic.LoadUint64(&mm.MaximumFileKey), - maximumNeedleEnd: atomic.LoadInt64(&mm.MaximumNeedleEnd), - } -} - func (mm *mapMetric) restoreBatchMetrics(snapshot batchMapMetricSnapshot) { atomic.StoreUint32(&mm.DeletionCounter, snapshot.deletionCounter) atomic.StoreUint32(&mm.FileCounter, snapshot.fileCounter) diff --git a/weed/storage/needle_map_sorted_file.go b/weed/storage/needle_map_sorted_file.go index e5de53925a..5a30525213 100644 --- a/weed/storage/needle_map_sorted_file.go +++ b/weed/storage/needle_map_sorted_file.go @@ -101,14 +101,6 @@ func (m *SortedFileNeedleMap) Put(key NeedleId, offset Offset, size Size) error return fmt.Errorf("needle map %s.sdx is read only: %w", m.baseFileName, os.ErrInvalid) } -func (m *SortedFileNeedleMap) removeMapping(key NeedleId) error { - return fmt.Errorf("needle map %s.sdx is read only: %w", m.baseFileName, os.ErrInvalid) -} - -func (m *SortedFileNeedleMap) restoreMapping(key NeedleId, offset Offset, size Size) error { - return fmt.Errorf("needle map %s.sdx is read only: %w", m.baseFileName, os.ErrInvalid) -} - func (m *SortedFileNeedleMap) Delete(key NeedleId, offset Offset) error { f, err := pooledIndexFiles.borrow(m.dbFileName, true) diff --git a/weed/storage/volume_backup.go b/weed/storage/volume_backup.go index 4843f32af6..96ec4b4a7f 100644 --- a/weed/storage/volume_backup.go +++ b/weed/storage/volume_backup.go @@ -2,6 +2,7 @@ package storage import ( "context" + "errors" "fmt" "hash/fnv" "io" @@ -80,6 +81,11 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption } writeOffset := int64(startFromOffset) + indexOffset := int64(cursor.entryCount) * NeedleMapEntrySize + // only the in-memory map can reload rows it just wrote to the index + // file; other maps fall back to regenerating the index from .dat + cnm, canReplayIndex := v.nm.(*NeedleMap) + receivedIndex := false err = operation.WithVolumeServerClient(false, volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { @@ -109,6 +115,18 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption if err := v.UnavailableError(); err != nil { return err } + if len(resp.IdxContent) > 0 { + receivedIndex = true + if !canReplayIndex { + continue + } + n, writeErr := cnm.IndexFile().WriteAt(resp.IdxContent, indexOffset) + if writeErr != nil { + return writeErr + } + indexOffset += int64(n) + continue + } n, writeErr := v.DataBackend.WriteAt(resp.FileContent, writeOffset) if writeErr != nil { return writeErr @@ -121,9 +139,39 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption }) if err != nil { + // Roll the files back to the pre-copy boundary: leftover bytes could + // split a row or be mistaken for applied ones on the next run. + if receivedIndex && canReplayIndex { + _ = cnm.truncateIndex(int64(cursor.entryCount) * NeedleMapEntrySize) + } + if writeOffset > int64(startFromOffset) { + _ = v.DataBackend.Truncate(int64(startFromOffset)) + } return err } + if receivedIndex && canReplayIndex { + baseOffset := int64(cursor.entryCount) * NeedleMapEntrySize + // a failed replay is rolled back to the pre-copy boundary and the + // map rebuilt from the truncated index, so nothing keyed by the + // half-replayed tail is left behind + rollback := func() { + _ = cnm.truncateIndex(baseOffset) + _ = v.DataBackend.Truncate(int64(startFromOffset)) + cnm.reset() + _ = cnm.DoOffsetLoading(v, cnm.IndexFile(), 0) + } + if err := cnm.truncateIndex(indexOffset); err != nil { + rollback() + return err + } + if err := cnm.DoOffsetLoading(v, cnm.IndexFile(), uint64(cursor.entryCount)); err != nil { + rollback() + return err + } + return nil + } + // add to needle map return ScanVolumeFileFrom(v.Version(), v.DataBackend, int64(startFromOffset), &VolumeFileScanner4GenIdx{v: v}) @@ -144,13 +192,17 @@ type backupCursor struct { func (v *Volume) backupCursor() (backupCursor, error) { key, offset, size, entryCount, err := v.locateLastAppendEntry() - if err != nil || offset.IsZero() { + if err != nil { return backupCursor{entryCount: entryCount}, err } entriesHash, err := v.hashIndexEntries(entryCount) if err != nil { return backupCursor{}, err } + if offset.IsZero() { + // an index-only tombstone has no needle to read; key and size carry it + return backupCursor{entryCount: entryCount, key: key, size: size, entriesHash: entriesHash}, nil + } n, err := v.readNeedleAt(offset) if err != nil { return backupCursor{}, err @@ -258,12 +310,15 @@ var ErrIncrementalCopyDiverged = fmt.Errorf("local copy diverged from the replic // ResumeByLastEntry resumes an incremental copy by index position instead of // by appendAtNs, whose stamps differ across replicas for the same needle. // The replica's entry at position entryCount-1 must match the backup's last -// entry in key, size and cookie, and its first entryCount entries must hash -// the same — a boundary match alone is not proof, since replicated writes can -// land in a different order on this replica. The copy then continues from the -// replica's entry at position entryCount, so nothing is skipped or appended -// twice. -func (v *Volume) ResumeByLastEntry(entryCount uint64, key NeedleId, cookie uint32, size Size, entriesHash uint64) (offset Offset, upToDate bool, err error) { +// entry in key and size — and cookie, unless it is an index-only tombstone +// (a remote-tier delete), which has no .dat record to read a cookie from — +// and its first entryCount entries must hash the same — a boundary match +// alone is not proof, since replicated writes can land in a different order +// on this replica. The copy then continues from the replica's next entry +// with a .dat record; index-only rows ride the index stream, so nothing is +// skipped or appended twice. indexBound caps the scan to the index size the +// caller captured, so rows appended after that point stay out of this copy. +func (v *Volume) ResumeByLastEntry(entryCount uint64, key NeedleId, cookie uint32, size Size, entriesHash uint64, indexBound uint64) (offset Offset, upToDate bool, err error) { fileSize := int64(v.IndexFileSize()) if fileSize%NeedleMapEntrySize != 0 { @@ -285,19 +340,17 @@ func (v *Volume) ResumeByLastEntry(entryCount uint64, key NeedleId, cookie uint3 err = ErrIncrementalCopyDiverged return } - // a remote-tier tombstone sits in the index with no .dat record to verify - if lastOffset.IsZero() { - err = ErrIncrementalCopyDiverged - return - } - n, e := v.readNeedleAt(lastOffset) - if e != nil { - err = e - return - } - if uint32(n.Cookie) != cookie { - err = ErrIncrementalCopyDiverged - return + if !lastOffset.IsZero() { + var n *needle.Needle + n, e = v.readNeedleAt(lastOffset) + if e != nil { + err = e + return + } + if uint32(n.Cookie) != cookie { + err = ErrIncrementalCopyDiverged + return + } } prefixHash, e := v.hashIndexEntries(entryCount) @@ -314,9 +367,17 @@ func (v *Volume) ResumeByLastEntry(entryCount uint64, key NeedleId, cookie uint3 upToDate = true return } - offset, err = v.readOffsetFromIndex(int64(entryCount)) - if err == nil && offset.ToActualOffset() <= 0 { - err = ErrIncrementalCopyDiverged + scanEnd := int64(serverCount) + if boundRows := int64(indexBound / uint64(NeedleMapEntrySize)); boundRows < scanEnd { + scanEnd = boundRows + } + for i := int64(entryCount); i < scanEnd; i++ { + if offset, err = v.readOffsetFromIndex(i); err != nil { + return + } + if !offset.IsZero() { + return + } } return } @@ -456,6 +517,50 @@ func (v *Volume) readOffsetFromIndex(m int64) (Offset, error) { return offset, err } +// DatSizeCoveredByIndex returns the .dat end of the last record the index +// rows before indexSize cover, or the superblock size when no row inside the +// bound names a record. Rows are not always in .dat order — rebuilds can +// write the .idx sorted by key — so the bound comes from the maximum offset: +// records do not overlap, so the highest offset names the highest end. +// Records appended after the index bound — including staged writes whose +// rows are not published yet — lie past it, so a copy must not stream them: +// they would reach the backup without their row and be appended again on the +// next run. +func (v *Volume) DatSizeCoveredByIndex(indexSize int64) (int64, error) { + indexFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) + if err != nil { + return 0, err + } + defer indexFile.Close() + + boundRows := indexSize / NeedleMapEntrySize + row := int64(0) + maxActualOffset := int64(-1) + err = idx.WalkIndexFile(indexFile, 0, func(_ NeedleId, offset Offset, _ Size) error { + if row >= boundRows { + return errStopIdxWalk + } + row++ + if !offset.IsZero() { + if actualOffset := offset.ToActualOffset(); actualOffset > maxActualOffset { + maxActualOffset = actualOffset + } + } + return nil + }) + if err != nil && !errors.Is(err, errStopIdxWalk) { + return 0, err + } + if maxActualOffset < 0 { + return int64(v.SuperBlock.BlockSize()), nil + } + _, _, bodyLength, err := needle.ReadNeedleHeader(v.DataBackend, v.SuperBlock.Version, maxActualOffset) + if err != nil { + return 0, err + } + return maxActualOffset + NeedleHeaderSize + bodyLength, nil +} + // generate the volume idx type VolumeFileScanner4GenIdx struct { v *Volume diff --git a/weed/storage/volume_backup_test.go b/weed/storage/volume_backup_test.go index 9f3fc12463..08ca468fbc 100644 --- a/weed/storage/volume_backup_test.go +++ b/weed/storage/volume_backup_test.go @@ -1,8 +1,11 @@ package storage import ( + "os" + "sort" "testing" + "github.com/seaweedfs/seaweedfs/weed/storage/idx" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -81,13 +84,13 @@ func TestResumeByLastEntry(t *testing.T) { // a copy whose tail is the replica's tail is already up to date key, cookie, size := entryAt(4) - if _, upToDate, err := v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5)); err != nil || !upToDate { + if _, upToDate, err := v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5), v.IndexFileSize()); err != nil || !upToDate { t.Fatalf("tail cursor: upToDate=%v err=%v", upToDate, err) } // a copy ending at an earlier record resumes from the next entry key, cookie, size = entryAt(2) - offset, upToDate, err := v.ResumeByLastEntry(3, key, cookie, size, prefixHash(3)) + offset, upToDate, err := v.ResumeByLastEntry(3, key, cookie, size, prefixHash(3), v.IndexFileSize()) if err != nil || upToDate { t.Fatalf("mid cursor: upToDate=%v err=%v", upToDate, err) } @@ -96,31 +99,150 @@ func TestResumeByLastEntry(t *testing.T) { } // a copy ahead of the replica diverged - if _, _, err = v.ResumeByLastEntry(6, key, cookie, size, prefixHash(3)); err != ErrIncrementalCopyDiverged { + if _, _, err = v.ResumeByLastEntry(6, key, cookie, size, prefixHash(3), v.IndexFileSize()); err != ErrIncrementalCopyDiverged { t.Fatalf("ahead cursor: %v", err) } // a copy whose tail record or earlier records differ diverged key, cookie, size = entryAt(4) - if _, _, err = v.ResumeByLastEntry(5, key+1, cookie, size, prefixHash(5)); err != ErrIncrementalCopyDiverged { + if _, _, err = v.ResumeByLastEntry(5, key+1, cookie, size, prefixHash(5), v.IndexFileSize()); err != ErrIncrementalCopyDiverged { t.Fatalf("key mismatch: %v", err) } - if _, _, err = v.ResumeByLastEntry(5, key, cookie+1, size, prefixHash(5)); err != ErrIncrementalCopyDiverged { + if _, _, err = v.ResumeByLastEntry(5, key, cookie+1, size, prefixHash(5), v.IndexFileSize()); err != ErrIncrementalCopyDiverged { t.Fatalf("cookie mismatch: %v", err) } - if _, _, err = v.ResumeByLastEntry(5, key, cookie, size+1, prefixHash(5)); err != ErrIncrementalCopyDiverged { + if _, _, err = v.ResumeByLastEntry(5, key, cookie, size+1, prefixHash(5), v.IndexFileSize()); err != ErrIncrementalCopyDiverged { t.Fatalf("size mismatch: %v", err) } - if _, _, err = v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5)+1); err != ErrIncrementalCopyDiverged { + if _, _, err = v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5)+1, v.IndexFileSize()); err != ErrIncrementalCopyDiverged { t.Fatalf("prefix hash mismatch: %v", err) } - // a remote-tier tombstone has no .dat record; a boundary at offset zero diverges + // a remote-tier tombstone has no .dat record; a copy ending on it is a + // valid boundary identified by key and size alone deletedKey, _, _ := entryAt(3) if err := v.nm.Delete(deletedKey, types.ToOffset(0)); err != nil { t.Fatalf("zero-offset tombstone: %v", err) } - if _, _, err = v.ResumeByLastEntry(6, deletedKey, 0, types.TombstoneFileSize, prefixHash(6)); err != ErrIncrementalCopyDiverged { - t.Fatalf("zero-offset boundary: %v", err) + if _, upToDate, err := v.ResumeByLastEntry(6, deletedKey, 0, types.TombstoneFileSize, prefixHash(6), v.IndexFileSize()); err != nil || !upToDate { + t.Fatalf("tombstone tail cursor: upToDate=%v err=%v", upToDate, err) + } + + // index-only rows are skipped when locating the .dat resume offset + if _, _, _, err := v.writeNeedle2(newRandomNeedle(6), true, false, false); err != nil { + t.Fatalf("write needle 6: %v", err) + } + key, cookie, size = entryAt(4) + offset, upToDate, err = v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5), v.IndexFileSize()) + if err != nil || upToDate { + t.Fatalf("cursor before tombstone: upToDate=%v err=%v", upToDate, err) + } + if want, _ := v.readOffsetFromIndex(6); offset != want { + t.Fatalf("resume offset %v, want %v", offset, want) + } + + // a copy ahead of only tombstones resumes with no .dat left to send + if err := v.nm.Delete(deletedKey+1, types.ToOffset(0)); err != nil { + t.Fatalf("second zero-offset tombstone: %v", err) + } + key, cookie, size = entryAt(6) + offset, upToDate, err = v.ResumeByLastEntry(7, key, cookie, size, prefixHash(7), v.IndexFileSize()) + if err != nil || upToDate { + t.Fatalf("cursor before trailing tombstone: upToDate=%v err=%v", upToDate, err) + } + if !offset.IsZero() { + t.Fatalf("resume offset %v, want zero", offset) + } +} + +func TestDatSizeCoveredByIndex(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + for i := 1; i <= 3; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false); err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + } + datSize, idxSize, _ := v.FileStat() + + covered, err := v.DatSizeCoveredByIndex(int64(idxSize)) + if err != nil || covered != int64(datSize) { + t.Fatalf("covered %d, want %d, err %v", covered, datSize, err) + } + + // a bound partway through covers only up to that row's record end, + // which is where the next record begins + next, err := v.readOffsetFromIndex(2) + if err != nil { + t.Fatalf("read row 2: %v", err) + } + covered, err = v.DatSizeCoveredByIndex(2 * types.NeedleMapEntrySize) + if err != nil || covered != next.ToActualOffset() { + t.Fatalf("mid bound covered %d, want %d, err %v", covered, next.ToActualOffset(), err) + } + + // index-only tombstones past the bound still resolve to the last record + key, _, _, _, err := v.locateLastAppendEntry() + if err != nil { + t.Fatalf("locate last entry: %v", err) + } + if err := v.nm.Delete(key, types.ToOffset(0)); err != nil { + t.Fatalf("zero-offset tombstone: %v", err) + } + covered, err = v.DatSizeCoveredByIndex(int64(v.IndexFileSize())) + if err != nil || covered != int64(datSize) { + t.Fatalf("covered %d after tombstone, want %d, err %v", covered, datSize, err) + } +} + +func TestDatSizeCoveredByIndexKeySorted(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + // write keys out of order so a key-sorted .idx ends on a mid-file row + for _, id := range []uint64{2, 3, 1} { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(id), true, false, false); err != nil { + t.Fatalf("write needle %d: %v", id, err) + } + } + datSize, idxSize, _ := v.FileStat() + + // rewrite the .idx sorted by key, as rebuilds do + indexFile := v.FileName(".idx") + content, err := os.ReadFile(indexFile) + if err != nil { + t.Fatalf("read idx: %v", err) + } + var rows [][]byte + for i := 0; i+types.NeedleMapEntrySize <= len(content); i += types.NeedleMapEntrySize { + rows = append(rows, content[i:i+types.NeedleMapEntrySize]) + } + sort.Slice(rows, func(i, j int) bool { + keyI, _, _ := idx.IdxFileEntry(rows[i]) + keyJ, _, _ := idx.IdxFileEntry(rows[j]) + return keyI < keyJ + }) + var sorted []byte + for _, row := range rows { + sorted = append(sorted, row...) + } + if err := os.WriteFile(indexFile, sorted, 0644); err != nil { + t.Fatalf("rewrite idx: %v", err) + } + + covered, err := v.DatSizeCoveredByIndex(int64(idxSize)) + if err != nil || covered != int64(datSize) { + t.Fatalf("key-sorted covered %d, want %d, err %v", covered, datSize, err) } }