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) } }