diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 3729c9cff8..34d4375fff 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -230,6 +230,16 @@ message VolumeSyncStatusResponse { message VolumeIncrementalCopyRequest { uint32 volume_id = 1; uint64 since_ns = 2; + // The backup's index position and last entry. When set, the replica + // resumes by entry position instead of by its own append_at_ns stamp, + // which differs across replicas for the same needle. + uint64 entry_count = 3; + uint64 last_needle_key = 4; + uint32 last_needle_cookie = 5; + int32 last_needle_size = 6; + // FNV-1a checksum of the (key, size) of every local index entry, + // verifying the whole prefix since replica record order can differ. + uint64 entries_hash = 7; } message VolumeIncrementalCopyResponse { bytes file_content = 1; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index cc3446770b..2f2a369a7a 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1587,15 +1587,45 @@ impl VolumeServer for VolumeGrpcService { let dat_size = v.dat_file_size().unwrap_or(0); let super_block_size = v.super_block().block_size() as u64; - // If since_ns is very large (after all data), return empty - if req.since_ns == u64::MAX || dat_size <= super_block_size { + // 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. + if req.entry_count == 0 && (req.since_ns == u64::MAX || dat_size <= super_block_size) { drop(store); let stream = tokio_stream::iter(Vec::new()); return Ok(Response::new(Box::pin(stream))); } - // Use binary search to find the starting offset - let start_offset = if req.since_ns == 0 { + // 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 { + 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, + ) { + 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, + Err(crate::storage::volume::VolumeError::IncrementalCopyDiverged) => { + return Err(Status::failed_precondition( + crate::storage::volume::VolumeError::IncrementalCopyDiverged.to_string(), + )); + } + Err(e) => { + return Err(Status::internal(format!( + "fail to locate by entry {}: {}", + req.entry_count, e + ))); + } + } + } else if req.since_ns == 0 { super_block_size } else { match v.binary_search_by_append_at_ns(req.since_ns) { @@ -8389,6 +8419,7 @@ mod tests { volume_server_pb::VolumeIncrementalCopyRequest { volume_id: 1, since_ns: 0, + ..Default::default() }, )) .await @@ -10367,6 +10398,7 @@ mod tests { volume_server_pb::VolumeIncrementalCopyRequest { volume_id: 1, since_ns: 0, + ..Default::default() }, )) .await diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index c212d154c4..cfbffe1d0f 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -56,6 +56,9 @@ pub enum VolumeError { #[error("cookie mismatch: {0:#x}")] CookieMismatch(u32), + #[error("local copy diverged from the replica")] + IncrementalCopyDiverged, + #[error("volume not empty")] NotEmpty, @@ -4701,8 +4704,8 @@ impl Volume { // ---- Binary search for incremental copy ---- - /// Read a single index entry's offset from the .idx file by entry index. - fn read_offset_from_index(&self, m: i64) -> Result { + /// Read a single index entry from the .idx file by entry index. + fn read_index_entry(&self, m: i64) -> Result<(NeedleId, Offset, Size), VolumeError> { let idx_path = self.file_name(".idx"); let idx_file = File::open(&idx_path)?; let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE]; @@ -4718,7 +4721,12 @@ impl Volume { f.seek(SeekFrom::Start(file_offset))?; std::io::Read::read_exact(&mut f, &mut buf)?; } - let (_key, offset, _size) = idx_entry_from_bytes(&buf); + Ok(idx_entry_from_bytes(&buf)) + } + + /// Read a single index entry's offset from the .idx file by entry index. + fn read_offset_from_index(&self, m: i64) -> Result { + let (_key, offset, _size) = self.read_index_entry(m)?; Ok(offset) } @@ -4850,6 +4858,97 @@ impl Volume { Ok((offset, false)) } + /// FNV-1a 64 over the (key, size) of the first entry_count index entries. + /// The offset field is skipped because the same record sits at different + /// offsets on different replicas. Mirrors hashIndexEntries in + /// weed/storage/volume_backup.go. + fn hash_index_entries(&self, entry_count: u64) -> Result { + const FNV_OFFSET: u64 = 0xcbf29ce484222325; + const FNV_PRIME: u64 = 0x100000001b3; + + let idx_path = self.file_name(".idx"); + let mut idx_file = File::open(&idx_path)?; + let mut h: u64 = FNV_OFFSET; + let mut buf = vec![0u8; 1 << 20]; + let mut remaining = entry_count * NEEDLE_MAP_ENTRY_SIZE as u64; + while remaining > 0 { + let mut want = remaining.min(buf.len() as u64) as usize; + want -= want % NEEDLE_MAP_ENTRY_SIZE; + std::io::Read::read_exact(&mut idx_file, &mut buf[..want])?; + for i in (0..want).step_by(NEEDLE_MAP_ENTRY_SIZE) { + for &b in &buf[i..i + NEEDLE_ID_SIZE] { + h = (h ^ b as u64).wrapping_mul(FNV_PRIME); + } + for &b in &buf[i + NEEDLE_ID_SIZE + OFFSET_SIZE..i + NEEDLE_MAP_ENTRY_SIZE] { + h = (h ^ b as u64).wrapping_mul(FNV_PRIME); + } + } + remaining -= want as u64; + } + Ok(h) + } + + /// 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. + /// Mirrors Go's ResumeByLastEntry in weed/storage/volume_backup.go. + /// Returns (offset, up_to_date). + pub fn resume_by_last_entry( + &self, + entry_count: u64, + key: NeedleId, + cookie: Cookie, + size: Size, + entries_hash: 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; + if file_size % NEEDLE_MAP_ENTRY_SIZE as i64 != 0 { + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::InvalidData, + format!("unexpected idx file size: {}", file_size), + ))); + } + let server_count = (file_size / NEEDLE_MAP_ENTRY_SIZE as i64) as u64; + if server_count < entry_count { + return Err(VolumeError::IncrementalCopyDiverged); + } + + let (last_key, last_offset, last_size) = + self.read_index_entry(entry_count as i64 - 1)?; + 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 self.hash_index_entries(entry_count)? != entries_hash { + return Err(VolumeError::IncrementalCopyDiverged); + } + + 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); + } + Ok((offset, false)) + } + /// Write a raw needle blob at a specific offset in the .dat file. pub fn write_needle_blob(&self, offset: i64, needle_blob: &[u8]) -> Result<(), VolumeError> { let _sync_guard = self.data_file_sync_lock.lock().unwrap(); diff --git a/weed/command/backup.go b/weed/command/backup.go index cfeabcde28..ea6e083fe6 100644 --- a/weed/command/backup.go +++ b/weed/command/backup.go @@ -5,6 +5,8 @@ import ( "fmt" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -155,6 +157,7 @@ func backupFromLocation(volumeServer pb.ServerAddress, grpcDialOption grpc.DialO datSize, _, _ := v.FileStat() // If local volume is larger than remote, recreate it + rebuilt := false if datSize > stats.TailOffset { if err := v.Destroy(false, false, false); err != nil { v.Close() @@ -166,14 +169,39 @@ func backupFromLocation(volumeServer pb.ServerAddress, grpcDialOption grpc.DialO if err != nil { return fmt.Errorf("recreating volume: %w", err), false } + rebuilt = true } // Perform the incremental backup - if err := v.IncrementalBackup(volumeServer, grpcDialOption); err != nil { + err = v.IncrementalBackup(volumeServer, grpcDialOption) + if err != nil && status.Code(err) == codes.FailedPrecondition { + // The local copy is not a prefix of this replica's records; rebuild it. + if err := v.Destroy(false, false, false); err != nil { + v.Close() + return fmt.Errorf("destroying volume: %w", err), false + } + v, err = storage.NewVolume(*s.dir, *s.dir, *s.collection, vid, storage.NeedleMapInMemory, replication, ttl, 0, ver, 0, 0) + if err != nil { + return fmt.Errorf("recreating volume: %w", err), false + } + err = v.IncrementalBackup(volumeServer, grpcDialOption) + rebuilt = true + } + if err != nil { v.Close() return fmt.Errorf("incremental backup: %w", err), false } + // A rebuilt copy starts at compaction revision zero; adopt the source's + // revision or the next run compacts it in key order and diverges again. + if rebuilt { + v.SuperBlock.CompactionRevision = uint16(stats.CompactRevision) + if _, err = v.DataBackend.WriteAt(v.SuperBlock.Bytes(), 0); err != nil { + v.Close() + return fmt.Errorf("writing superblock: %w", err), false + } + } + v.Close() return nil, false } diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 3a66ffff77..3c96ce1401 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -230,6 +230,16 @@ message VolumeSyncStatusResponse { message VolumeIncrementalCopyRequest { uint32 volume_id = 1; uint64 since_ns = 2; + // The backup's index position and last entry. When set, the replica + // resumes by entry position instead of by its own append_at_ns stamp, + // which differs across replicas for the same needle. + uint64 entry_count = 3; + uint64 last_needle_key = 4; + uint32 last_needle_cookie = 5; + int32 last_needle_size = 6; + // FNV-1a checksum of the (key, size) of every local index entry, + // verifying the whole prefix since replica record order can differ. + uint64 entries_hash = 7; } message VolumeIncrementalCopyResponse { bytes file_content = 1; diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index ce6c0c6c71..dc04bc08c0 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -1126,9 +1126,19 @@ func (x *VolumeSyncStatusResponse) GetVersion() uint32 { } type VolumeIncrementalCopyRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` - SinceNs uint64 `protobuf:"varint,2,opt,name=since_ns,json=sinceNs,proto3" json:"since_ns,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId,proto3" json:"volume_id,omitempty"` + SinceNs uint64 `protobuf:"varint,2,opt,name=since_ns,json=sinceNs,proto3" json:"since_ns,omitempty"` + // The backup's index position and last entry. When set, the replica + // resumes by entry position instead of by its own append_at_ns stamp, + // which differs across replicas for the same needle. + EntryCount uint64 `protobuf:"varint,3,opt,name=entry_count,json=entryCount,proto3" json:"entry_count,omitempty"` + LastNeedleKey uint64 `protobuf:"varint,4,opt,name=last_needle_key,json=lastNeedleKey,proto3" json:"last_needle_key,omitempty"` + LastNeedleCookie uint32 `protobuf:"varint,5,opt,name=last_needle_cookie,json=lastNeedleCookie,proto3" json:"last_needle_cookie,omitempty"` + LastNeedleSize int32 `protobuf:"varint,6,opt,name=last_needle_size,json=lastNeedleSize,proto3" json:"last_needle_size,omitempty"` + // FNV-1a checksum of the (key, size) of every local index entry, + // verifying the whole prefix since replica record order can differ. + EntriesHash uint64 `protobuf:"varint,7,opt,name=entries_hash,json=entriesHash,proto3" json:"entries_hash,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1177,6 +1187,41 @@ func (x *VolumeIncrementalCopyRequest) GetSinceNs() uint64 { return 0 } +func (x *VolumeIncrementalCopyRequest) GetEntryCount() uint64 { + if x != nil { + return x.EntryCount + } + return 0 +} + +func (x *VolumeIncrementalCopyRequest) GetLastNeedleKey() uint64 { + if x != nil { + return x.LastNeedleKey + } + return 0 +} + +func (x *VolumeIncrementalCopyRequest) GetLastNeedleCookie() uint32 { + if x != nil { + return x.LastNeedleCookie + } + return 0 +} + +func (x *VolumeIncrementalCopyRequest) GetLastNeedleSize() int32 { + if x != nil { + return x.LastNeedleSize + } + return 0 +} + +func (x *VolumeIncrementalCopyRequest) GetEntriesHash() uint64 { + if x != nil { + return x.EntriesHash + } + return 0 +} + 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"` @@ -7342,10 +7387,16 @@ const file_volume_server_proto_rawDesc = "" + "tailOffset\x12)\n" + "\x10compact_revision\x18\a \x01(\rR\x0fcompactRevision\x12\"\n" + "\ridx_file_size\x18\b \x01(\x04R\vidxFileSize\x12\x18\n" + - "\aversion\x18\t \x01(\rR\aversion\"V\n" + + "\aversion\x18\t \x01(\rR\aversion\"\x9a\x02\n" + "\x1cVolumeIncrementalCopyRequest\x12\x1b\n" + "\tvolume_id\x18\x01 \x01(\rR\bvolumeId\x12\x19\n" + - "\bsince_ns\x18\x02 \x01(\x04R\asinceNs\"B\n" + + "\bsince_ns\x18\x02 \x01(\x04R\asinceNs\x12\x1f\n" + + "\ventry_count\x18\x03 \x01(\x04R\n" + + "entryCount\x12&\n" + + "\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" + "\x1dVolumeIncrementalCopyResponse\x12!\n" + "\ffile_content\x18\x01 \x01(\fR\vfileContent\"e\n" + "\x12VolumeMountRequest\x12\x1b\n" + diff --git a/weed/server/volume_grpc_copy_incremental.go b/weed/server/volume_grpc_copy_incremental.go index 335b5c63ef..49403fae2d 100644 --- a/weed/server/volume_grpc_copy_incremental.go +++ b/weed/server/volume_grpc_copy_incremental.go @@ -2,12 +2,17 @@ package weed_server import ( "context" + "errors" "fmt" "io" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" ) func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrementalCopyRequest, stream volume_server_pb.VolumeServer_VolumeIncrementalCopyServer) error { @@ -21,17 +26,37 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem } stopOffset, _, _ := v.FileStat() - foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs) - if err != nil { - return fmt.Errorf("fail to locate by appendAtNs %d: %s", req.SinceNs, err) - } - if isLastOne { - return nil + var startOffset int64 + if req.EntryCount > 0 { + foundOffset, upToDate, err := v.ResumeByLastEntry( + req.EntryCount, + types.NeedleId(req.LastNeedleKey), + req.LastNeedleCookie, + types.Size(req.LastNeedleSize), + req.EntriesHash, + ) + if err != nil { + if errors.Is(err, storage.ErrIncrementalCopyDiverged) { + return status.Error(codes.FailedPrecondition, err.Error()) + } + return fmt.Errorf("fail to locate by entry %d: %s", req.EntryCount, err) + } + if upToDate { + return nil + } + startOffset = foundOffset.ToActualOffset() + } else { + foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs) + if err != nil { + return fmt.Errorf("fail to locate by appendAtNs %d: %s", req.SinceNs, err) + } + if isLastOne { + return nil + } + startOffset = foundOffset.ToActualOffset() } - startOffset := foundOffset.ToActualOffset() - buf := make([]byte, 1024*1024*2) return sendFileContent(v, buf, startOffset, int64(stopOffset), stream) diff --git a/weed/storage/volume_backup.go b/weed/storage/volume_backup.go index 80377b34c9..4843f32af6 100644 --- a/weed/storage/volume_backup.go +++ b/weed/storage/volume_backup.go @@ -3,6 +3,7 @@ package storage import ( "context" "fmt" + "hash/fnv" "io" "os" @@ -73,7 +74,7 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption } startFromOffset, _, _ := v.FileStat() - appendAtNs, err := v.findLastAppendAtNs() + cursor, err := v.backupCursor() if err != nil { return err } @@ -83,8 +84,13 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption err = operation.WithVolumeServerClient(false, volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { stream, err := client.VolumeIncrementalCopy(context.Background(), &volume_server_pb.VolumeIncrementalCopyRequest{ - VolumeId: uint32(v.Id), - SinceNs: appendAtNs, + VolumeId: uint32(v.Id), + SinceNs: cursor.appendAtNs, + EntryCount: cursor.entryCount, + LastNeedleKey: uint64(cursor.key), + LastNeedleCookie: cursor.cookie, + LastNeedleSize: int32(cursor.size), + EntriesHash: cursor.entriesHash, }) if err != nil { return err @@ -123,58 +129,196 @@ func (v *Volume) IncrementalBackup(volumeServer pb.ServerAddress, grpcDialOption } -func (v *Volume) findLastAppendAtNs() (uint64, error) { - offset, err := v.locateLastAppendEntry() - if err != nil { - return 0, err - } - if offset.IsZero() { - return 0, nil - } - return v.readAppendAtNs(offset) +// backupCursor identifies the last record of the local copy: its index +// position and the needle itself. A replica can verify the same record at the +// same position and resume from there, which stays correct when a different +// replica answers, because appendAtNs stamps differ across replicas. +type backupCursor struct { + entryCount uint64 + key NeedleId + size Size + cookie uint32 + appendAtNs uint64 + entriesHash uint64 } -func (v *Volume) locateLastAppendEntry() (Offset, error) { - indexFile, e := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) - if e != nil { - return Offset{}, fmt.Errorf("cannot read %s: %v", v.FileName(".idx"), e) +func (v *Volume) backupCursor() (backupCursor, error) { + key, offset, size, entryCount, err := v.locateLastAppendEntry() + if err != nil || offset.IsZero() { + return backupCursor{entryCount: entryCount}, err + } + entriesHash, err := v.hashIndexEntries(entryCount) + if err != nil { + return backupCursor{}, err + } + n, err := v.readNeedleAt(offset) + if err != nil { + return backupCursor{}, err + } + return backupCursor{ + entryCount: entryCount, + key: key, + size: size, + cookie: uint32(n.Cookie), + appendAtNs: n.AppendAtNs, + entriesHash: entriesHash, + }, nil +} + +// hashIndexEntries folds the (key, size) of the first entryCount index entries +// into one FNV-1a checksum. The offset field is skipped because the same +// record sits at different offsets on different replicas. +func (v *Volume) hashIndexEntries(entryCount uint64) (uint64, error) { + indexFile, err := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) + if err != nil { + return 0, fmt.Errorf("cannot read %s: %v", v.FileName(".idx"), err) } defer indexFile.Close() - fi, err := indexFile.Stat() - if err != nil { - return Offset{}, fmt.Errorf("file %s stat error: %v", indexFile.Name(), err) + h := fnv.New64a() + buf := make([]byte, 1<<20) + remaining := int64(entryCount) * NeedleMapEntrySize + for remaining > 0 { + want := int64(len(buf)) + if remaining < want { + want = remaining + } + want -= want % int64(NeedleMapEntrySize) + read, err := io.ReadFull(indexFile, buf[:want]) + if err != nil { + return 0, fmt.Errorf("file %s read error: %v", indexFile.Name(), err) + } + for i := 0; i+NeedleMapEntrySize <= read; i += NeedleMapEntrySize { + h.Write(buf[i : i+NeedleIdSize]) + h.Write(buf[i+NeedleIdSize+OffsetSize : i+NeedleMapEntrySize]) + } + remaining -= int64(read) + } + return h.Sum64(), nil +} + +func (v *Volume) locateLastAppendEntry() (key NeedleId, offset Offset, size Size, entryCount uint64, err error) { + indexFile, e := os.OpenFile(v.FileName(".idx"), os.O_RDONLY, 0644) + if e != nil { + err = fmt.Errorf("cannot read %s: %v", v.FileName(".idx"), e) + return + } + defer indexFile.Close() + + fi, e := indexFile.Stat() + if e != nil { + err = fmt.Errorf("file %s stat error: %v", indexFile.Name(), e) + return } fileSize := fi.Size() if fileSize%NeedleMapEntrySize != 0 { - return Offset{}, fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) + err = fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) + return } + entryCount = uint64(fileSize / NeedleMapEntrySize) if fileSize == 0 { - return Offset{}, nil + return } bytes := make([]byte, NeedleMapEntrySize) n, e := indexFile.ReadAt(bytes, fileSize-NeedleMapEntrySize) if n != NeedleMapEntrySize { - return Offset{}, fmt.Errorf("file %s read error: %v", indexFile.Name(), e) + err = fmt.Errorf("file %s read error: %v", indexFile.Name(), e) + return } - _, offset, _ := idx.IdxFileEntry(bytes) - - return offset, nil + key, offset, size = idx.IdxFileEntry(bytes) + return } -func (v *Volume) readAppendAtNs(offset Offset) (uint64, error) { - +func (v *Volume) readNeedleAt(offset Offset) (*needle.Needle, error) { n, _, bodyLength, err := needle.ReadNeedleHeader(v.DataBackend, v.SuperBlock.Version, offset.ToActualOffset()) if err != nil { - return 0, fmt.Errorf("ReadNeedleHeader %s [%d,%d): %v", v.DataBackend.Name(), offset.ToActualOffset(), offset.ToActualOffset()+NeedleHeaderSize, err) + return nil, fmt.Errorf("ReadNeedleHeader %s [%d,%d): %v", v.DataBackend.Name(), offset.ToActualOffset(), offset.ToActualOffset()+NeedleHeaderSize, err) } _, err = n.ReadNeedleBody(v.DataBackend, v.SuperBlock.Version, offset.ToActualOffset()+NeedleHeaderSize, bodyLength) if err != nil { - return 0, fmt.Errorf("ReadNeedleBody offset %d, bodyLength %d: %v", offset.ToActualOffset(), bodyLength, err) + return nil, fmt.Errorf("ReadNeedleBody offset %d, bodyLength %d: %v", offset.ToActualOffset(), bodyLength, err) + } + return n, nil +} + +func (v *Volume) readAppendAtNs(offset Offset) (uint64, error) { + n, err := v.readNeedleAt(offset) + if err != nil { + return 0, err } return n.AppendAtNs, nil +} +// ErrIncrementalCopyDiverged means the local copy is not a prefix of this +// replica's records, so an incremental copy cannot resume safely and the +// backup should be rebuilt from scratch. +var ErrIncrementalCopyDiverged = fmt.Errorf("local copy diverged from the replica") + +// 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) { + + fileSize := int64(v.IndexFileSize()) + if fileSize%NeedleMapEntrySize != 0 { + err = fmt.Errorf("unexpected file %s.idx size: %d", v.IndexFileName(), fileSize) + return + } + serverCount := uint64(fileSize / NeedleMapEntrySize) + if serverCount < entryCount { + err = ErrIncrementalCopyDiverged + return + } + + lastKey, lastOffset, lastSize, e := v.readIndexEntry(int64(entryCount) - 1) + if e != nil { + err = e + return + } + if lastKey != key || lastSize != size { + 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 + } + + prefixHash, e := v.hashIndexEntries(entryCount) + if e != nil { + err = e + return + } + if prefixHash != entriesHash { + err = ErrIncrementalCopyDiverged + return + } + + if serverCount == entryCount { + upToDate = true + return + } + offset, err = v.readOffsetFromIndex(int64(entryCount)) + if err == nil && offset.ToActualOffset() <= 0 { + err = ErrIncrementalCopyDiverged + } + return } // on server side @@ -297,14 +441,18 @@ func (v *Volume) readLeftNs(m int64) (index int64, offset Offset, ts uint64, err return } -// bytes is of size NeedleMapEntrySize -func (v *Volume) readOffsetFromIndex(m int64) (Offset, error) { +func (v *Volume) readIndexEntry(m int64) (key NeedleId, offset Offset, size Size, err error) { v.dataFileAccessLock.RLock() defer v.dataFileAccessLock.RUnlock() if v.nm == nil { - return Offset{}, io.EOF + return 0, Offset{}, 0, io.EOF } - _, offset, _, err := v.nm.ReadIndexEntry(m) + return v.nm.ReadIndexEntry(m) +} + +// bytes is of size NeedleMapEntrySize +func (v *Volume) readOffsetFromIndex(m int64) (Offset, error) { + _, offset, _, err := v.readIndexEntry(m) return offset, err } diff --git a/weed/storage/volume_backup_test.go b/weed/storage/volume_backup_test.go index 3291d203a4..9f3fc12463 100644 --- a/weed/storage/volume_backup_test.go +++ b/weed/storage/volume_backup_test.go @@ -1,6 +1,12 @@ package storage -import "testing" +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) func TestBinarySearch(t *testing.T) { var testInput []int @@ -37,3 +43,84 @@ func binarySearchForLargerThanTarget(nums []int, target int) int { } return l } + +func TestResumeByLastEntry(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 <= 5; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false); err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + } + + entryAt := func(m int64) (key types.NeedleId, cookie uint32, size types.Size) { + key, offset, size, err := v.readIndexEntry(m) + if err != nil { + t.Fatalf("readIndexEntry %d: %v", m, err) + } + n, err := v.readNeedleAt(offset) + if err != nil { + t.Fatalf("readNeedleAt %d: %v", m, err) + } + return key, uint32(n.Cookie), size + } + + prefixHash := func(count uint64) uint64 { + h, err := v.hashIndexEntries(count) + if err != nil { + t.Fatalf("hashIndexEntries %d: %v", count, err) + } + return h + } + + // 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 { + 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)) + if err != nil || upToDate { + t.Fatalf("mid cursor: upToDate=%v err=%v", upToDate, err) + } + if want, _ := v.readOffsetFromIndex(3); offset != want { + t.Fatalf("resume offset %v, want %v", offset, want) + } + + // a copy ahead of the replica diverged + if _, _, err = v.ResumeByLastEntry(6, key, cookie, size, prefixHash(3)); 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 { + t.Fatalf("key mismatch: %v", err) + } + if _, _, err = v.ResumeByLastEntry(5, key, cookie+1, size, prefixHash(5)); err != ErrIncrementalCopyDiverged { + t.Fatalf("cookie mismatch: %v", err) + } + if _, _, err = v.ResumeByLastEntry(5, key, cookie, size+1, prefixHash(5)); err != ErrIncrementalCopyDiverged { + t.Fatalf("size mismatch: %v", err) + } + if _, _, err = v.ResumeByLastEntry(5, key, cookie, size, prefixHash(5)+1); err != ErrIncrementalCopyDiverged { + t.Fatalf("prefix hash mismatch: %v", err) + } + + // a remote-tier tombstone has no .dat record; a boundary at offset zero diverges + 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) + } +}