fix(filer.backup): repair chunk-incomplete and stale destination entries (#10082)

* fix(filer.backup): repair chunk-incomplete and stale destination entries

filer.backup left destinations diverged while metadata advanced — chunk-incomplete
(missing/gapped ranges at full attr.file_size) or holding a chunk superseded by a
missed overwrite. The skip/repair decision keyed on filer.FileSize (the attr),
which a truncated entry keeps full, so it never repaired.

Decide from actual chunk state instead:
- coversReference: range-by-range containment (scalar byte totals and attr
  FileSize/Md5 cannot see chunk-level gaps).
- hasStaleBackupChunk: a backup-written chunk (SourceFileId) the source no longer
  lists; ignores out-of-band (rsync/direct) chunks.
- destinationMatchesReference: allocation-free positional fast path gating the
  above so they run only on divergence (the in-sync path stays cheap).
- A strictly-newer destination is never repaired, so an older out-of-order replay
  cannot roll it back. The stale signal is deferred at equal mtime (same-second
  versions cannot be ordered; reliable S3 sub-second ordering is a separate fix).

Tests in filer_sink_test.go.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* filer.backup: verify chunk range in destinationMatchesReference fast path

The allocation-free fast path matched a destination chunk to its reference
by SourceFileId alone. That is correct today only because replicateOneChunk
copies the source chunk's Offset/Size verbatim, so SourceFileId identity
implies an identical range — an invariant that lives in another file with no
guard linking the two. If replication ever re-chunks (split/coalesce), a
chunk with the right SourceFileId but a different range would fast-path as a
full match and skip a needed repair (a false positive in the very class this
change otherwise prevents).

Compare Offset/Size alongside SourceFileId so the fast path is self-contained
and can only be more conservative (a range mismatch falls through to the
precise coversReference/hasStaleBackupChunk checks). Add tests for a shifted
offset and a larger size at matching identity.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

weed/replication/sink/filersink/filer_sink.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"sort"
 	"strings"
 	"sync"
 
@@ -244,18 +245,12 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 		}
 		// glog.V(1).Infof("lookup: %v", lookupRequest)
 		if resp, err := filer_pb.LookupEntry(context.Background(), client, lookupRequest); err == nil {
-			if filer.ETag(resp.Entry) == filer.ETag(entry) {
-				glog.V(3).Infof("already replicated %s", key)
+			if destinationSatisfiesSource(resp.Entry, entry) {
+				glog.V(3).Infof("skip overwriting %s", key)
 				return nil
 			}
-			if getEntryMtimeNs(resp.Entry) >= getEntryMtimeNs(entry) {
-				if filer.FileSize(resp.Entry) >= filer.FileSize(entry) {
-					glog.V(3).Infof("skip overwriting %s", key)
-					return nil
-				}
-				glog.Warningf("repair truncated %s: destination %d bytes < source %d bytes but has a newer mtime; overwriting from source",
-					key, filer.FileSize(resp.Entry), filer.FileSize(entry))
-			}
+			glog.Warningf("re-replicating %s: destination not current (dst %d chunks, src %d chunks)",
+				key, len(resp.Entry.GetChunks()), len(entry.GetChunks()))
 		}
 
 		replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry))
@@ -331,14 +326,14 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 	// now, which for a rename is newParentPath/newEntry.Name, not the old key.
 	targetKey := updatedEntryKey(key, newParentPath, newEntry)
 
-	switch chooseUpdateAction(existingEntry, newEntry) {
+	switch chooseUpdateAction(existingEntry, oldEntry, newEntry) {
 	case updateSkip:
 		// a newer, complete version already landed (out-of-order); leave it
 		glog.V(2).Infof("late updates %s", key)
 		return true, nil
 	case updateRepair:
-		glog.Warningf("repair truncated %s: destination %d bytes < source %d bytes but has a newer mtime; re-replicating full source content",
-			key, filer.FileSize(existingEntry), filer.FileSize(newEntry))
+		glog.Warningf("repair %s: destination diverged from source (dst %d chunks, src %d chunks); re-replicating full source content",
+			key, len(existingEntry.GetChunks()), len(newEntry.GetChunks()))
 		replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, getEntryMtimeNs(newEntry))
 		if err != nil {
 			if errors.Is(err, errChunkSizeMismatch) {
@@ -470,12 +465,174 @@ func updatedEntryKey(key, newParentPath string, newEntry *filer_pb.Entry) string
 	return string(util.NewFullPath(targetDir, newEntry.Name))
 }
 
-func chooseUpdateAction(existing, incoming *filer_pb.Entry) updateAction {
-	if getEntryMtimeNs(existing) <= getEntryMtimeNs(incoming) {
-		return updateNormal
+func chooseUpdateAction(existing, oldEntry, incoming *filer_pb.Entry) updateAction {
+	exNs, inNs := getEntryMtimeNs(existing), getEntryMtimeNs(incoming)
+	if exNs > inNs {
+		// Destination is strictly newer: incoming is a stale/out-of-order replay of
+		// an older source version, so never roll the destination back to it — even
+		// when the newer destination is smaller (a legitimate truncate). A newer copy
+		// that is itself incomplete is reconciled by -initialSnapshot, not by an
+		// older event.
+		return updateSkip
 	}
-	if filer.FileSize(existing) < filer.FileSize(incoming) {
-		return updateRepair
+	// Destination at or behind incoming. updateNormal applies the incremental
+	// oldEntry->incoming diff, valid only when the destination already holds
+	// oldEntry's chunks. A cheap, allocation-free positional identity match proves
+	// that for the common (in-sync) path; only when it fails — divergence, chunk
+	// reordering, or an out-of-band destination — do we run the precise O(n log n)
+	// checks.
+	if existing != nil && oldEntry != nil && !destinationMatchesReference(existing, oldEntry) {
+		// Truncated/gapped: the destination does not cover every byte range oldEntry
+		// holds. Range containment (not a scalar byte count, attr FileSize, or ETag)
+		// catches it for any destination, and is safe at equal mtime since a complete
+		// destination always covers oldEntry.
+		if !coversReference(existing, oldEntry) {
+			return updateRepair
+		}
+		// Stale superseded chunk at full coverage: a chunk this backup wrote whose
+		// source oldEntry no longer references. Restricted to backup-written chunks
+		// (out-of-band rsync/direct chunks are ignored) AND to a strictly-older
+		// destination — at equal mtime same-second versions cannot be ordered, so
+		// defer rather than risk a rollback (reliable ordering is a separate fix).
+		if exNs < inNs && hasStaleBackupChunk(existing, oldEntry) {
+			return updateRepair
+		}
 	}
-	return updateSkip
+	return updateNormal
+}
+
+// destinationSatisfiesSource reports whether an existing destination entry already
+// satisfies the incoming source entry, so CreateEntry can skip the write. Coverage
+// is checked first: a chunk-truncated destination (covering fewer bytes than the
+// source) must be re-replicated even when its attr.Md5-backed ETag still matches
+// the source, so ETag equality must never bypass the coverage check. A destination
+// that is at least as complete is skippable when it has identical content (ETag) or
+// is a newer-or-equal version.
+func destinationSatisfiesSource(existing, incoming *filer_pb.Entry) bool {
+	if existing == nil {
+		return false
+	}
+	exNs, inNs := getEntryMtimeNs(existing), getEntryMtimeNs(incoming)
+	if exNs > inNs {
+		return true // destination strictly newer: incoming is older → keep destination
+	}
+	if !destinationMatchesReference(existing, incoming) {
+		if !coversReference(existing, incoming) {
+			return false // missing a byte range the source covers (truncated/gapped)
+		}
+		if exNs < inNs && hasStaleBackupChunk(existing, incoming) {
+			return false // a provably-older destination holds a superseded backup chunk
+		}
+	}
+	if filer.ETag(existing) == filer.ETag(incoming) {
+		return true // identical content
+	}
+	return exNs >= inNs // equal mtime → keep (deferred); strictly older → re-replicate
+}
+
+// destinationMatchesReference is a cheap, allocation-free sufficient check that the
+// destination already holds exactly the reference's chunks: equal count and, in
+// order, each destination chunk was replicated from the corresponding reference
+// chunk (SourceFileId == reference FileId) at the same byte range (Offset/Size).
+// True ⇒ full coverage and no stale chunk, so updateNormal/skip is safe and the
+// O(n log n) range and identity checks can be skipped — the common in-sync path.
+// False is never a false positive: a reordered, diverged, or out-of-band
+// (rsync/direct) destination just falls back to the precise checks.
+func destinationMatchesReference(existing, reference *filer_pb.Entry) bool {
+	ex, ref := existing.GetChunks(), reference.GetChunks()
+	if len(ex) != len(ref) {
+		return false
+	}
+	for i := range ref {
+		if ex[i].SourceFileId == "" || ex[i].SourceFileId != ref[i].GetFileIdString() {
+			return false
+		}
+		// Require the same byte range too. replicateOneChunk copies the source
+		// chunk's Offset/Size verbatim, so SourceFileId identity already implies an
+		// identical range today; verifying it here keeps this fast path sound even
+		// if replication ever re-chunks (split/coalesce), instead of silently
+		// depending on that invariant from another file.
+		if ex[i].Offset != ref[i].Offset || ex[i].Size != ref[i].Size {
+			return false
+		}
+	}
+	return true
+}
+
+// hasStaleBackupChunk reports whether existing holds a chunk that this backup
+// wrote (SourceFileId set) whose source chunk `reference` no longer references.
+// That means an overwrite/deletion event was missed: the destination still carries
+// the superseded source chunk's bytes even though total coverage looks full, so the
+// incremental diff cannot fix it. Chunks without SourceFileId (seeded out-of-band,
+// e.g. rsync or a direct write) are ignored, so such destinations keep their normal
+// path and are never re-copied on this signal.
+func hasStaleBackupChunk(existing, reference *filer_pb.Entry) bool {
+	if existing == nil || reference == nil {
+		return false
+	}
+	refFids := make(map[string]bool, len(reference.GetChunks()))
+	for _, c := range reference.GetChunks() {
+		refFids[c.GetFileIdString()] = true
+	}
+	for _, c := range existing.GetChunks() {
+		if c.SourceFileId == "" {
+			continue // not backup-written → do not judge staleness
+		}
+		if !refFids[c.SourceFileId] {
+			return true
+		}
+	}
+	return false
+}
+
+// mergedIntervals returns the entry's chunk byte ranges, sorted and merged
+// (overlapping or touching ranges combined). Chunk manifests are counted by their
+// own offset/size span, which already covers the range they wrap, so no manifest
+// resolution is needed.
+func mergedIntervals(entry *filer_pb.Entry) [][2]int64 {
+	if entry == nil {
+		return nil
+	}
+	chunks := entry.GetChunks()
+	if len(chunks) == 0 {
+		return nil
+	}
+	ivs := make([][2]int64, 0, len(chunks))
+	for _, c := range chunks {
+		ivs = append(ivs, [2]int64{int64(c.Offset), int64(c.Offset) + int64(c.Size)})
+	}
+	sort.Slice(ivs, func(i, j int) bool { return ivs[i][0] < ivs[j][0] })
+	merged := make([][2]int64, 0, len(ivs))
+	for _, iv := range ivs {
+		if n := len(merged); n > 0 && iv[0] <= merged[n-1][1] {
+			if iv[1] > merged[n-1][1] {
+				merged[n-1][1] = iv[1]
+			}
+			continue
+		}
+		merged = append(merged, iv)
+	}
+	return merged
+}
+
+// coversReference reports whether existing's chunks cover every byte range that
+// reference's chunks cover. Equal total coverage is not sufficient proof — extra
+// chunks at other offsets can hide a missing range — so containment is verified
+// range by range against the merged interval sets.
+func coversReference(existing, reference *filer_pb.Entry) bool {
+	ex := mergedIntervals(existing)
+	i := 0
+	for _, r := range mergedIntervals(reference) {
+		lo, hi := r[0], r[1]
+		for lo < hi {
+			for i < len(ex) && ex[i][1] <= lo {
+				i++
+			}
+			if i >= len(ex) || ex[i][0] > lo {
+				return false // a byte in reference is not covered by existing
+			}
+			lo = ex[i][1]
+		}
+	}
+	return true
 }

weed/replication/sink/filersink/filer_sink_test.go

@@ -18,6 +18,59 @@ func entryNs(mtime int64, ns int32, size uint64) *filer_pb.Entry {
 	return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime, MtimeNs: ns, FileSize: size}}
 }
 
+// entryChunks builds an entry with nChunks contiguous 1-byte chunks at a given
+// mtime, so coveredBytes == nChunks. Different counts give different coverage,
+// which is what the truncation-repair routing keys on.
+func entryChunks(mtime int64, nChunks int) *filer_pb.Entry {
+	e := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime, FileSize: uint64(nChunks)}}
+	for i := range nChunks {
+		e.Chunks = append(e.Chunks, &filer_pb.FileChunk{
+			FileId: "3," + string(rune('a'+i)), Offset: int64(i), Size: 1,
+		})
+	}
+	return e
+}
+
+// entryMd5 is like entryChunks but also sets attr.Md5, so filer.ETag returns the
+// stored Md5 (the S3 PutObject case) instead of a chunk-derived ETag. Used to show
+// that coverage — not the Md5-backed ETag — drives the truncation decision.
+func entryMd5(mtime int64, nChunks int, md5 []byte) *filer_pb.Entry {
+	e := entryChunks(mtime, nChunks)
+	e.Attributes.Md5 = md5
+	return e
+}
+
+// srcEntry builds a source-side entry at mtime whose chunks carry the given
+// FileIds — the identities a backup destination's SourceFileId is matched against.
+func srcEntry(mtime int64, fids ...string) *filer_pb.Entry {
+	e := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime}}
+	for i, f := range fids {
+		e.Chunks = append(e.Chunks, &filer_pb.FileChunk{FileId: f, Offset: int64(i), Size: 1})
+	}
+	return e
+}
+
+// dstEntry builds a destination entry at mtime. An empty src marks a chunk
+// written out-of-band (no SourceFileId, e.g. rsync/direct); otherwise it is a
+// backup chunk copied from that source FileId.
+func dstEntry(mtime int64, srcFids ...string) *filer_pb.Entry {
+	e := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime}}
+	for i, s := range srcFids {
+		e.Chunks = append(e.Chunks, &filer_pb.FileChunk{FileId: "dst", SourceFileId: s, Offset: int64(i), Size: 1})
+	}
+	return e
+}
+
+// ivEntry builds an entry at mtime with chunks at the given [offset,end) byte
+// ranges — for coverage/containment tests where exact ranges matter.
+func ivEntry(mtime int64, ivs ...[2]int64) *filer_pb.Entry {
+	e := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: mtime}}
+	for _, iv := range ivs {
+		e.Chunks = append(e.Chunks, &filer_pb.FileChunk{Offset: iv[0], Size: uint64(iv[1] - iv[0])})
+	}
+	return e
+}
+
 // getEntryMtimeNs must order versions written within the same second, which
 // plain second-grained mtime cannot. It also must be nil-safe.
 func TestGetEntryMtimeNs(t *testing.T) {
@@ -65,28 +118,181 @@ func TestChooseUpdateAction(t *testing.T) {
 	tests := []struct {
 		name     string
 		existing *filer_pb.Entry
+		oldEntry *filer_pb.Entry
 		incoming *filer_pb.Entry
 		want     updateAction
 	}{
-		{"nil existing entry", nil, entry(200, 100), updateNormal},
-		{"destination older, catching up", entry(100, 50), entry(200, 100), updateNormal},
-		{"same mtime", entry(200, 50), entry(200, 100), updateNormal},
-		{"destination newer and complete", entry(300, 100), entry(200, 100), updateSkip},
-		{"destination newer and larger", entry(300, 200), entry(200, 100), updateSkip},
-		{"destination newer but truncated", entry(300, 90), entry(200, 100), updateRepair},
+		{"nil existing entry", nil, entry(100, 100), entry(200, 100), updateNormal},
+		{"destination older, catching up", entry(100, 50), entry(100, 50), entry(200, 100), updateNormal},
+		{"same mtime", entry(200, 50), entry(200, 50), entry(200, 100), updateNormal},
+		{"destination newer and complete", entry(300, 100), entry(100, 100), entry(200, 100), updateSkip},
+		{"destination newer and larger", entry(300, 200), entry(100, 100), entry(200, 100), updateSkip},
+		// out-of-order: a newer (smaller) destination must NOT be rolled back by an
+		// older, larger replayed event — even though it does not cover oldEntry.
+		{"newer smaller dest, older larger event -> skip", ivEntry(300, [2]int64{0, 5}), ivEntry(100, [2]int64{0, 10}), ivEntry(100, [2]int64{0, 10}), updateSkip},
 		// same second: sub-second ordering must still pick the winner
-		{"same second, destination newer and complete", entryNs(5, 200, 100), entryNs(5, 100, 100), updateSkip},
-		{"same second, destination older", entryNs(5, 100, 100), entryNs(5, 200, 100), updateNormal},
+		{"same second, destination newer and complete", entryNs(5, 200, 100), entryNs(5, 100, 100), entryNs(5, 100, 100), updateSkip},
+		{"same second, destination older", entryNs(5, 100, 100), entryNs(5, 100, 100), entryNs(5, 200, 100), updateNormal},
+		// chunk-incompleteness (the truncation bug): destination is at/behind the
+		// incoming mtime but covers fewer bytes than oldEntry, so the incremental
+		// diff cannot heal it — must repair in full, not append. entryChunks(n) has
+		// n contiguous 1-byte chunks → coverage n.
+		{"behind, covers less than oldEntry -> repair", entryChunks(200, 2), entryChunks(200, 4), entryChunks(200, 4), updateRepair},
+		{"behind, matches oldEntry coverage -> normal", entryChunks(200, 3), entryChunks(200, 3), entryChunks(200, 5), updateNormal},
+		{"behind, exceeds oldEntry coverage -> normal", entryChunks(200, 5), entryChunks(200, 3), entryChunks(200, 5), updateNormal},
+		// range containment, not scalar coverage: equal total bytes but a missing
+		// range (the destination has an extra range at another offset) → repair.
+		{"behind, equal total but range gap -> repair", ivEntry(200, [2]int64{0, 10}, [2]int64{30, 40}), ivEntry(200, [2]int64{0, 20}), ivEntry(200, [2]int64{0, 20}), updateRepair},
+		// same-range stale: full coverage but a backup-written chunk whose source
+		// oldEntry no longer references → repair only when the destination is
+		// provably older. Equal-mtime is deferred (cannot order same-second
+		// versions). Out-of-band (rsync) chunks are never stale-checked.
+		{"older, stale backup chunk -> repair", dstEntry(100, "A", "X"), srcEntry(100, "A", "B"), srcEntry(200, "A", "B"), updateRepair},
+		{"equal-mtime stale -> deferred (normal)", dstEntry(200, "A", "X"), srcEntry(200, "A", "B"), srcEntry(200, "A", "B"), updateNormal},
+		{"older, rsync chunks -> normal", dstEntry(100, "", ""), srcEntry(100, "A", "B"), srcEntry(200, "A", "B"), updateNormal},
+		// in-sync fast path: destination holds exactly oldEntry's chunks → normal
+		{"in-sync (fast path) -> normal", dstEntry(100, "A", "B"), srcEntry(100, "A", "B"), srcEntry(200, "A", "B"), updateNormal},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			if got := chooseUpdateAction(tc.existing, tc.incoming); got != tc.want {
+			if got := chooseUpdateAction(tc.existing, tc.oldEntry, tc.incoming); got != tc.want {
 				t.Fatalf("chooseUpdateAction = %v, want %v", got, tc.want)
 			}
 		})
 	}
 }
 
+// coversReference must verify range containment, not just equal total bytes: a
+// destination with the same byte count but a missing range (extra chunk elsewhere)
+// does not cover the reference.
+func TestCoversReference(t *testing.T) {
+	cases := []struct {
+		name                string
+		existing, reference *filer_pb.Entry
+		want                bool
+	}{
+		{"empty reference", ivEntry(0), ivEntry(0), true},
+		{"reference no chunks", ivEntry(0, [2]int64{0, 10}), ivEntry(0), true},
+		{"exact", ivEntry(0, [2]int64{0, 10}), ivEntry(0, [2]int64{0, 10}), true},
+		{"superset", ivEntry(0, [2]int64{0, 20}), ivEntry(0, [2]int64{0, 10}), true},
+		{"truncated tail", ivEntry(0, [2]int64{0, 5}), ivEntry(0, [2]int64{0, 10}), false},
+		{"equal total, range gap", ivEntry(0, [2]int64{0, 10}, [2]int64{30, 40}), ivEntry(0, [2]int64{0, 20}), false},
+		{"contiguous parts cover", ivEntry(0, [2]int64{0, 10}, [2]int64{10, 20}), ivEntry(0, [2]int64{0, 20}), true},
+		{"overlapping parts cover", ivEntry(0, [2]int64{0, 12}, [2]int64{8, 20}), ivEntry(0, [2]int64{0, 20}), true},
+		{"existing empty, reference has bytes", ivEntry(0), ivEntry(0, [2]int64{0, 1}), false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := coversReference(tc.existing, tc.reference); got != tc.want {
+				t.Fatalf("coversReference = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// destinationSatisfiesSource gates CreateEntry's skip. Coverage/identity must be
+// checked before ETag: a truncated or stale destination whose attr.Md5 still
+// matches the source (so filer.ETag is equal) must NOT be skipped, or
+// -initialSnapshot would leave it permanently diverged.
+func TestDestinationSatisfiesSource(t *testing.T) {
+	md5A := []byte("0123456789abcdef")
+	md5B := []byte("fedcba9876543210")
+	cases := []struct {
+		name               string
+		existing, incoming *filer_pb.Entry
+		want               bool
+	}{
+		{"nil existing", nil, entryChunks(200, 4), false},
+		{"identical complete -> skip", entryChunks(200, 4), entryChunks(200, 4), true},
+		{"truncated (no md5) -> replicate", entryChunks(200, 2), entryChunks(200, 4), false},
+		// regression: same attr.Md5 (equal filer.ETag) but fewer chunks must repair
+		{"md5-backed truncated -> replicate", entryMd5(200, 2, md5A), entryMd5(200, 4, md5A), false},
+		{"md5 equal and complete -> skip", entryMd5(200, 4, md5A), entryMd5(200, 4, md5A), true},
+		// complete but different content: keep newer, replace older
+		{"complete, dest newer -> skip", entryMd5(300, 4, md5A), entryMd5(200, 4, md5B), true},
+		{"complete, dest older -> replicate", entryMd5(200, 4, md5A), entryMd5(300, 4, md5B), false},
+		// same-range stale backup chunk: replicate only when provably older; an
+		// equal-mtime stale dest is deferred (skip), and rsync chunks are not
+		// stale-checked.
+		{"older, stale backup chunk -> replicate", dstEntry(100, "A", "X"), srcEntry(200, "A", "B"), false},
+		{"equal-mtime stale -> skip (deferred)", dstEntry(200, "A", "X"), srcEntry(200, "A", "B"), true},
+		{"rsync complete -> skip", dstEntry(200, "", ""), srcEntry(200, "A", "B"), true},
+		{"in-sync (fast path) -> skip", dstEntry(200, "A", "B"), srcEntry(200, "A", "B"), true},
+		// out-of-order: newer (smaller) destination must not be rolled back by an
+		// older, larger source replay.
+		{"newer smaller dest, older larger source -> skip", ivEntry(300, [2]int64{0, 5}), ivEntry(100, [2]int64{0, 10}), true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := destinationSatisfiesSource(tc.existing, tc.incoming); got != tc.want {
+				t.Fatalf("destinationSatisfiesSource = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// destinationMatchesReference is the allocation-free fast path: true only when the
+// destination holds exactly the reference's chunks in order (by SourceFileId). It
+// must be conservative — never a false positive — so reorders, count mismatches,
+// and out-of-band (no SourceFileId) chunks all return false.
+func TestDestinationMatchesReference(t *testing.T) {
+	cases := []struct {
+		name                string
+		existing, reference *filer_pb.Entry
+		want                bool
+	}{
+		{"both empty", dstEntry(0), srcEntry(0), true},
+		{"exact in order", dstEntry(0, "A", "B"), srcEntry(0, "A", "B"), true},
+		{"count mismatch", dstEntry(0, "A"), srcEntry(0, "A", "B"), false},
+		{"reordered", dstEntry(0, "B", "A"), srcEntry(0, "A", "B"), false},
+		{"stale identity", dstEntry(0, "A", "X"), srcEntry(0, "A", "B"), false},
+		{"out-of-band chunk", dstEntry(0, "A", ""), srcEntry(0, "A", "B"), false},
+		// Same identity but a different byte range must not fast-path as a match,
+		// so the precise checks still run if replication ever re-chunks.
+		{"shifted offset, same identity",
+			&filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: 0}, Chunks: []*filer_pb.FileChunk{
+				{FileId: "dst", SourceFileId: "A", Offset: 0, Size: 1},
+				{FileId: "dst", SourceFileId: "B", Offset: 5, Size: 1},
+			}}, srcEntry(0, "A", "B"), false},
+		{"larger size, same identity",
+			&filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: 0}, Chunks: []*filer_pb.FileChunk{
+				{FileId: "dst", SourceFileId: "A", Offset: 0, Size: 2},
+			}}, srcEntry(0, "A"), false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := destinationMatchesReference(tc.existing, tc.reference); got != tc.want {
+				t.Fatalf("destinationMatchesReference = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// hasStaleBackupChunk flags a destination that still holds a backup-written chunk
+// (SourceFileId set) the source reference no longer lists — a missed
+// overwrite/deletion. Out-of-band chunks (no SourceFileId, e.g. rsync) are ignored.
+func TestHasStaleBackupChunk(t *testing.T) {
+	cases := []struct {
+		name                string
+		existing, reference *filer_pb.Entry
+		want                bool
+	}{
+		{"nil existing", nil, srcEntry(200, "A"), false},
+		{"in sync", dstEntry(200, "A", "B"), srcEntry(200, "A", "B"), false},
+		{"stale backup chunk", dstEntry(200, "A", "X"), srcEntry(200, "A", "B"), true},
+		{"rsync chunks ignored", dstEntry(200, "", ""), srcEntry(200, "A", "B"), false},
+		{"synced backup + rsync", dstEntry(200, "A", ""), srcEntry(200, "A", "B"), false},
+		{"stale backup among rsync", dstEntry(200, "X", ""), srcEntry(200, "A", "B"), true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := hasStaleBackupChunk(tc.existing, tc.reference); got != tc.want {
+				t.Fatalf("hasStaleBackupChunk = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
 // updatedEntryKey must resolve the incoming entry's new path. For a rename the
 // supersession check has to target newParentPath/newEntry.Name, not the old key,
 // or the renamed-away old path looks deleted on the source and skips a live event.