backup: resume incremental copies by index position instead of appendAtNs (#12228)

* volume server: add last-entry cursor fields to VolumeIncrementalCopy

* backup: resume incremental copies by index position instead of appendAtNs

appendAtNs is stamped by whichever replica received the write, so the
same needle carries a different timestamp on every replica. A backup
that resumes from a different replica than last time can append the
last needle again or skip needles written inside the clock gap.

Send the local copy's index entry count and last entry identity (key,
size, cookie) with VolumeIncrementalCopy. The replica verifies the same
record sits at that position and streams from the next index entry, so
the resume no longer depends on any replica's clock. When the boundary
record does not match, or the local copy is ahead, the replica returns
FailedPrecondition and the backup rebuilds the local copy from scratch.

Old servers ignore the new fields and keep the timestamp behavior.

* rust volume server: mirror positional resume in VolumeIncrementalCopy

* volume server: add entries_hash prefix checksum to VolumeIncrementalCopy

* backup: verify the whole index prefix, not just the boundary entry

* rust volume server: verify the index prefix hash in resume_by_last_entry

* backup: keep the source's compaction revision on rebuilt copies

* backup: diverge on a zero-offset boundary entry

A remote-tier tombstone lives in the .idx with no .dat record, so
readNeedleAt(0) parses the superblock and its error bypassed the
ErrIncrementalCopyDiverged mapping; the backup got a generic failure
instead of FailedPrecondition and never rebuilt.

* rust volume server: mirror zero-offset boundary divergence

* rust volume server: gate the empty-dat early return on entry_count

With a positional resume the Go handler still reaches ResumeByLastEntry,
so an empty replica reports divergence; the Rust early return answered a
clean empty stream and weed backup kept a stale copy.

* backup: hash the index entry's size field, not trailing offset bytes

In a 5BytesOffset build the offset occupies bytes 8-12 and the size
starts at byte 13, so the hash folded in one byte of the offset.
Matching records at different offsets then hashed differently and a
valid backup was reported diverged and rebuilt.

* backup: read the index in whole-entry chunks for the prefix hash

When entries are 17 bytes a 1 MiB read buffer does not end on an entry
boundary: the partial tail entry was skipped by the loop but still
counted as consumed, so every later chunk was hashed at a misaligned
position.

* rust volume server: hash the index entry's size field after the offset

Mirrors the Go hashIndexEntries change: with the 5bytes feature the
offset occupies bytes 8-12, so starting the size slice at byte 12
folded one offset byte into the hash and matching records at different
offsets compared diverged.

* rust volume server: read the index in whole-entry chunks for the prefix hash

Mirrors the Go change. With 17-byte entries a 1 MiB buffer does not
end on an entry boundary, so slicing the partial tail entry panicked
once a backup held 61681 entries.

* rust volume server: default the new incremental copy request fields in tests

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;

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

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<Offset, VolumeError> {
+    /// 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<Offset, VolumeError> {
+        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<u64, VolumeError> {
+        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();

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
 }

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;

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" +

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)
 

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
 }
 

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