backup: carry index-only tombstones through incremental copies (#12231)

* volume server: add idx_content to VolumeIncrementalCopyResponse

A remote-tier delete is an index-only tombstone row with no .dat record,
so it cannot reach an incremental copy through the .dat stream. Carrying
the .idx tail verbatim lets the backup reproduce those rows at their
exact positions.

* backup: carry index-only tombstones through incremental copies

A remote-tier delete leaves an offset-0 tombstone in .idx with no .dat
record. The copy only streamed .dat bytes and regenerated .idx from
them, so the tombstone never reached the backup: the deleted needle
stayed live and every later incremental diverged and rebuilt again.

backupCursor and ResumeByLastEntry now accept an index-only tombstone
as a boundary, matched by key and size since there is no record to read
a cookie from. VolumeIncrementalCopy appends the .idx tail after the
.dat tail, and IncrementalBackup writes it at the index boundary,
truncates, and loads the new rows — so tombstones land at the exact
positions the replica has them.

* seaweed-volume: mirror index-only tombstone handling in incremental copy

resume_by_last_entry accepts an offset-0 tombstone boundary matched on
key and size, skips its missing cookie check, and resumes .dat streaming
at the next entry that has a record — index-only rows ride the new
idx_content stream instead of reporting divergence. The server captures
the .idx bound before the .dat one (a row is published only after its
bytes) and streams the tail after the data.

* backup: keep incremental copy bounds and streams consistent

Review fixes:

* Bound capture moved after the resume reads and reads .dat/.idx sizes
  together (Go FileStat, Rust file_stat under one lock), so every row
  resume saw is inside the index bound and each row's record is always
  inside the data bound.
* A compaction commit changes CompactionRevision: the server now rejects
  the copy when the revision moved between resume, file opens, or the
  end of the stream, so a backup never mixes revisions. The .idx is
  opened up front to pin the inode.
* An index tail that ends before the captured bound is an error, not a
  silent success.
* IncrementalBackup rolls the .idx tail and .dat bytes back to the
  pre-copy boundary when the stream fails, so a partial row cannot wedge
  backupCursor and unapplied rows cannot pass for current ones.
* The index tail is replayed through a small indexTailReceiver interface,
  covering the leveldb needle maps too, not just the in-memory map.

* backup: bound the resume scan to the captured index size

The resume and the stream bounds are now captured in one order:
revision, then the .dat/.idx pair, then the scan. ResumeByLastEntry and
resume_by_last_entry take the index bound and scan no further, so a row
appended while the copy is being set up can never enter the index
stream without its .dat bytes.

The index-tail replay stays on the in-memory map only: the leveldb
map's DoOffsetLoading writes to the .cpldb compaction database, not the
live one, so it cannot serve a live tail replay; other maps keep
regenerating the index from .dat as before.

* seaweed-volume: guard incremental copy total against a located offset past .dat

A corrupt index can place the resume offset beyond the captured .dat
size; has_data already refuses the reader, but total still evaluates
dat_size - s and underflows.

* backup: bound the incremental .dat tail by index-covered records

A record staged but not yet indexed can sit inside the captured .dat
bound while its row stays outside the index bound — the fsync gap
between staging and publishing is not covered by one stat. Streaming
those bytes leaves them unindexed on the backup, and the next run
appends the record again. Bound the .dat tail by the last record the
streamed rows cover on both producers, and cover the bound walk with
tests.

* backup: cap the .dat stream at the captured bound and roll back replay failures

sendFileContent could read past stopOffset on its last chunk, so a
needle appended after FileStat arrived without its index row and the
next copy appended at the wrong position. Each read is now limited to
the remaining captured bytes.

A failed truncateIndex or DoOffsetLoading after a successful stream
left the index tail on disk unapplied, which a later backupCursor could
mistake for applied rows; both now roll the tail back to the pre-copy
boundary like the stream-error path.

* backup: bound the incremental .dat tail by the maximum covered offset

A key-sorted .idx ends on the highest-key row, which need not be the
record physically last in .dat, so the last-row bound could cut off
live records. Take the end of the highest-offset row inside the bound
instead, and cover the case with a sorted-idx test in Go and Rust.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* backup: restore .dat and the needle map when index replay fails

A failed DoOffsetLoading left the received .dat bytes in place and could
leave partial mappings for rows the index rollback just removed, so the
next copy appended the same records again or read stale entries. Truncate
.dat back to the copy boundary and restore the touched mappings and
metrics from a pre-replay snapshot.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* backup: cover no index rows with the superblock bound

When no row inside the index bound names a record — an empty index or a
bound of index-only tombstones — the covered end is the superblock, not
the captured .dat size, so staged bytes are never streamed without a
covering row.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* seaweed-volume: apply the covered bound even when the index streams no rows

The bound was gated on idx_end > idx_start, so an empty captured index
skipped it entirely and a staged .dat tail still shipped. Gate on the
stream-index predicate instead, matching Go's streamIndex check.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* backup: propagate index coverage read errors

* backup: rebuild the needle map from the truncated index on replay failure

* seaweed-volume: propagate index coverage read errors

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/proto/volume_server.proto

@@ -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 {

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<u64> = 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<volume_server_pb::VolumeIncrementalCopyResponse, Status>,
         >(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) =

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<u64, VolumeError> {
+        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<u8> = 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

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 {

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

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

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

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 {

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 {

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 {

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)

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)

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

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