fix(filer.backup): prevent silent backup data loss on transient not-found (#10295)

* fix(filer.backup): stop silently dropping events on transient not-found Under a write burst, filer.backup could consume a metadata event (advancing the persisted offset) without replicating the file, with no error logged: 1. filersink CreateEntry/UpdateEntry swallowed replicateChunks errors (glog.Warningf + return nil), so the offset advanced past entries that were never written. 2. The manifest-chunk branch of replicateChunks resolved via LookupFileId with no retry, unlike the data-chunk branch — transient lookup races dropped exactly the large manifest-backed files while small inline-content siblings landed. 3. isIgnorable404 matched "LookupFileId" / "volume id ... not found", misclassifying those races as genuine source 404s at the backup layer. Fix: on a replicateChunks failure the filer sink now skips only when the live source has moved past the replayed version (deleted or strictly-newer mtime) — lossless, a later event carries the current content — and propagates otherwise so the event is retried. The manifest resolve retries transient errors like the data-chunk path. isIgnorable404 is narrowed to genuine 404s; non-filer sinks and the initial-snapshot walk, which relied on the broad match as their only lossless-skip valve, now make the same live-source decision (filersink.SourceSupersedes) instead of retrying forever on a permanently gone volume. Tests cover propagation of unconfirmed lookup failures, the narrowed 404 classification, and the supersession guards. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(filer.backup): derive supersession path with directory fallback eventSourceSuperseded built the source path from NewParentPath alone. Legacy metadata events (persisted by older filers) carry an empty NewParentPath, so the probe looked up "/<name>", read the miss as "source gone", and skipped a live file on a transient lookup error — the silent drop this change is meant to eliminate. Derive the path via MetadataEventTargetFullPath (the same directory fallback genProcessFunction uses) and cover both event shapes with TestEventSupersessionProbe_PathDerivation. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(filersink): retry manifest resolve only for transient errors, bounded when unverifiable The manifest-resolve retry stopped only when hasSourceNewerVersion proved the source moved past the replayed version, which wedged the sink in two cases: incremental sinks use dated target keys that cannot map back to a source path (supersession never provable), and permanent resolve errors (corrupt manifest data, bad file ids) fail forever while the source entry stays live. Gate the retry instead: keep retrying only transient errors (volume-lookup races, network interruptions), stop after a few attempts when supersession cannot be checked, and propagate everything else immediately so the configured metadata error policy applies (-disableErrorRetry included). Propagation is lossless: filer.backup's fallback decides with the event's real source key, and both filer.backup and filer.sync re-deliver the event (RetryForeverOnError) without advancing the offset. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(filer.backup): make error classifiers nil-safe isIgnorable404, isSourceLookupError, and isTransientResolveError called err.Error() without a nil guard. All current call sites pass a non-nil error, but the guard is free and matches isRetryableNetworkError. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-20 06:52:24 +00:00 · 2026-07-10 17:47:46 +09:00
parent db42bb4975
commit c17aebeecf
6 changed files with 353 additions and 24 deletions
@@ -17,6 +17,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication/repl_util"
 	"github.com/seaweedfs/seaweedfs/weed/replication/sink"
+	"github.com/seaweedfs/seaweedfs/weed/replication/sink/filersink"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -207,6 +208,10 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 				glog.V(0).Infof("got 404 error for %s, ignore it: %s", getSourceKey(resp), err.Error())
 				return nil
 			}
+			if isSourceLookupError(err) && eventSourceSuperseded(filerSource, resp) {
+				glog.V(0).Infof("source superseded %s during lookup failure, skip it: %s", getSourceKey(resp), err.Error())
+				return nil
+			}
 			return err
 		}
 	} else {
@@ -271,22 +276,57 @@ func getSourceKey(resp *filer_pb.SubscribeMetadataResponse) string {
 	return ""
 }

-// isIgnorable404 returns true if the error represents a 404/not-found condition
-// that should be silently ignored during backup. This covers:
-//   - errors wrapping http.ErrNotFound (direct volume server 404 via non-S3 sinks)
-//   - errors containing the "404 Not Found: not found" status string (S3 sink path
-//     where AWS SDK breaks the errors.Is unwrap chain)
-//   - LookupFileId or volume-id-not-found errors from the volume id map
+// isIgnorable404 returns true only for a genuine source-side 404, where skipping
+// the event is lossless: errors wrapping http.ErrNotFound, or carrying the S3
+// "404 Not Found: not found" status string (the AWS SDK breaks the unwrap chain).
+// It deliberately does not match "LookupFileId" / "volume id ... not found" —
+// usually transient lookup races whose skip drops live files; those are resolved
+// against the live source instead (isSourceLookupError + eventSourceSuperseded).
 func isIgnorable404(err error) bool {
+	if err == nil {
+		return false
+	}
 	if errors.Is(err, http.ErrNotFound) {
 		return true
 	}
+	return strings.Contains(err.Error(), ignorable404ErrString)
+}
+
+// isSourceLookupError reports whether err is a source volume-lookup failure.
+// The same strings cover a transient race and a permanently gone volume, so
+// the caller must consult the live source to pick skip or retry.
+func isSourceLookupError(err error) bool {
+	if err == nil {
+		return false
+	}
 	errStr := err.Error()
-	return strings.Contains(errStr, ignorable404ErrString) ||
-		strings.Contains(errStr, "LookupFileId") ||
+	return strings.Contains(errStr, "LookupFileId") ||
 		(strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found"))
 }

+// eventSourceSuperseded reports whether the live source has moved past the
+// version carried by this event (entry gone or strictly newer), making a skip
+// lossless. Without a NewEntry it reports false so the error propagates.
+func eventSourceSuperseded(filerSource *source.FilerSource, resp *filer_pb.SubscribeMetadataResponse) bool {
+	sourcePath, mtimeNs, ok := eventSupersessionProbe(resp)
+	if !ok {
+		return false
+	}
+	return filersink.SourceSupersedes(context.Background(), filerSource, sourcePath, mtimeNs)
+}
+
+// eventSupersessionProbe derives the source path and replayed mtime of the
+// event's landing entry. Legacy events can carry an empty NewParentPath, so the
+// path comes from MetadataEventTargetFullPath (falls back to resp.Directory) —
+// a wrong path here would read as "gone" and skip a live file.
+func eventSupersessionProbe(resp *filer_pb.SubscribeMetadataResponse) (util.FullPath, int64, bool) {
+	if resp == nil || resp.EventNotification == nil || resp.EventNotification.NewEntry == nil {
+		return "", 0, false
+	}
+	return util.FullPath(filer_pb.MetadataEventTargetFullPath(resp)),
+		filersink.EntryMtimeNs(resp.EventNotification.NewEntry), true
+}
+
 // persistSnapshotOffset writes the snapshot high-water mark to the source
 // filer's KV, retrying a few times with exponential backoff on transient
 // errors. Losing this write forces a full re-walk on the next retry loop
@@ -386,9 +426,17 @@ func runInitialSnapshot(
 			// -ignore404Error is set; apply the same policy here so the walk
 			// doesn't abort (and trigger a full re-walk on retry) just because
 			// a single entry disappeared mid-snapshot.
-			if ignore404Error && isIgnorable404(err) {
-				glog.V(0).Infof("initialSnapshot: source entry %s disappeared, ignore it: %s", sourceKey, err.Error())
-				return nil
+			if ignore404Error {
+				if isIgnorable404(err) {
+					glog.V(0).Infof("initialSnapshot: source entry %s disappeared, ignore it: %s", sourceKey, err.Error())
+					return nil
+				}
+				// Same decision as the follow phase; lossless because the
+				// post-walk subscription replays the newer change.
+				if isSourceLookupError(err) && filersink.SourceSupersedes(ctx, filerSource, sourceKey, filersink.EntryMtimeNs(entry)) {
+					glog.V(0).Infof("initialSnapshot: source superseded %s mid-walk, skip it: %s", sourceKey, err.Error())
+					return nil
+				}
 			}
 			return fmt.Errorf("seed %s: %w", targetKey, err)
 		}
@@ -77,6 +77,141 @@ func TestIsIgnorable404_NonIgnorableError(t *testing.T) {
 	}
 }

+// Regression for the partial-landing swallow: transient volume-lookup races
+// ("LookupFileId ... failed", "volume id N not found") raised while replicating
+// a checkpoint write burst must NOT be classified as a genuine source 404.
+// They previously matched isIgnorable404 by substring, so the event was treated
+// as a deletion — the subscription offset advanced and the file (typically a
+// large manifest-backed .pt) was never replicated, with no error logged. They
+// are now propagated so the offset stays put and the event is reprocessed; only
+// a genuinely-gone source (verified live by the filer sink) is ever skipped.
+func TestIsIgnorable404_TransientLookupNotSwallowed(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{
+			"lookup file id race",
+			fmt.Errorf("create entry1 : %w",
+				fmt.Errorf("replicate manifest data chunks 3,01abc: LookupFileId 3,01abc failed, err: context deadline exceeded")),
+		},
+		{
+			"volume id not found race",
+			fmt.Errorf("create entry1 : %w",
+				fmt.Errorf("replicate entry chunks /buckets/x/model.pt: copy 7,02def: read part 7,02def: volume id 7 not found")),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if isIgnorable404(tc.err) {
+				t.Errorf("transient lookup race must not be ignorable (would swallow a live file): %v", tc.err)
+			}
+			if !isSourceLookupError(tc.err) {
+				t.Errorf("lookup race must classify as a source lookup error (resolved via the live source): %v", tc.err)
+			}
+		})
+	}
+}
+
+func TestErrorClassifiersNilSafe(t *testing.T) {
+	if isIgnorable404(nil) {
+		t.Error("isIgnorable404(nil) must be false")
+	}
+	if isSourceLookupError(nil) {
+		t.Error("isSourceLookupError(nil) must be false")
+	}
+}
+
+func TestIsSourceLookupError_NonLookupErrors(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{"genuine S3 404", fmt.Errorf("upload part: 404 Not Found: not found")},
+		{"network error", fmt.Errorf("dial tcp 10.0.0.1:8080: connection refused")},
+		{"plain not found without volume id", fmt.Errorf("entry not found")},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if isSourceLookupError(tc.err) {
+				t.Errorf("must not classify as a source lookup error: %v", tc.err)
+			}
+		})
+	}
+}
+
+// Legacy events carry an empty NewParentPath; the probe must fall back to
+// resp.Directory instead of building "/<name>", which would read as "gone"
+// and skip a live file.
+func TestEventSupersessionProbe_PathDerivation(t *testing.T) {
+	entry := &filer_pb.Entry{
+		Name:       "f",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 123},
+	}
+	cases := []struct {
+		name string
+		resp *filer_pb.SubscribeMetadataResponse
+		want string
+	}{
+		{
+			"legacy event without NewParentPath",
+			&filer_pb.SubscribeMetadataResponse{
+				Directory:         "/buckets/x",
+				EventNotification: &filer_pb.EventNotification{NewEntry: entry},
+			},
+			"/buckets/x/f",
+		},
+		{
+			"rename event with NewParentPath",
+			&filer_pb.SubscribeMetadataResponse{
+				Directory: "/buckets/x",
+				EventNotification: &filer_pb.EventNotification{
+					NewParentPath: "/buckets/y",
+					NewEntry:      entry,
+				},
+			},
+			"/buckets/y/f",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			path, mtimeNs, ok := eventSupersessionProbe(tc.resp)
+			if !ok {
+				t.Fatal("probe must succeed when NewEntry is present")
+			}
+			if string(path) != tc.want {
+				t.Errorf("path = %q, want %q", path, tc.want)
+			}
+			if mtimeNs != 123*int64(1e9) {
+				t.Errorf("mtimeNs = %d, want %d", mtimeNs, 123*int64(1e9))
+			}
+		})
+	}
+}
+
+// When supersession cannot be proven, never skip — that would drop a live file.
+func TestEventSourceSuperseded_Guards(t *testing.T) {
+	if eventSourceSuperseded(nil, nil) {
+		t.Error("nil response must not be skippable")
+	}
+	if eventSourceSuperseded(nil, &filer_pb.SubscribeMetadataResponse{
+		EventNotification: &filer_pb.EventNotification{},
+	}) {
+		t.Error("event without NewEntry must not be skippable")
+	}
+	if eventSourceSuperseded(nil, &filer_pb.SubscribeMetadataResponse{
+		EventNotification: &filer_pb.EventNotification{
+			NewParentPath: "/buckets/x",
+			NewEntry: &filer_pb.Entry{
+				Name:       "model.pt",
+				Attributes: &filer_pb.FuseAttributes{Mtime: 1234567890},
+			},
+		},
+	}) {
+		t.Error("nil filerSource must not be skippable")
+	}
+}
+
 // stubSink is a minimal ReplicationSink used to exercise initialSnapshotTargetKey
 // without standing up a real sink. Only the two methods read by the key builder
 // (GetName, IsIncremental) need meaningful behavior; the rest satisfy the interface.
@@ -26,6 +26,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )

@@ -139,7 +140,19 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path str
 }

 func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64) (*filer_pb.FileChunk, error) {
-	resolvedChunks, err := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk)
+	// LookupFileId can transiently fail during a write burst (volume not yet
+	// registered). Retry like the data-chunk path does; the gate stops on
+	// supersession or, when that cannot be checked, after a few attempts.
+	var resolvedChunks []*filer_pb.FileChunk
+	resolveName := fmt.Sprintf("resolve manifest %s", sourceChunk.GetFileIdString())
+	err := util.RetryUntil(resolveName, func() error {
+		rc, e := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk)
+		if e != nil {
+			return e
+		}
+		resolvedChunks = rc
+		return nil
+	}, fs.manifestResolveRetryGate(path, sourceMtimeNs, sourceChunk.GetFileIdString()))
 	if err != nil {
 		return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), err)
 	}
@@ -178,6 +191,56 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 	}, nil
 }

+// maxUnverifiableResolveAttempts bounds manifest-resolve retries when
+// supersession cannot be checked for the target path.
+const maxUnverifiableResolveAttempts = 3
+
+// manifestResolveRetryGate decides whether a failing manifest resolve keeps
+// retrying: stop when the source superseded the replayed version (the caller
+// skips it as lossless), stop immediately on non-transient errors (e.g. corrupt
+// manifest data) so the configured metadata error policy applies, and stop
+// after a few attempts when supersession cannot be checked at all (incremental
+// dated target keys don't map back to a source path) — propagating lets
+// filer.backup decide with the event's real source key instead of spinning
+// here forever.
+func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName string) func(error) bool {
+	_, canCheckSupersession := fs.targetPathToSourcePath(path)
+	attempts := 0
+	return func(resolveErr error) (shouldContinue bool) {
+		if fs.hasSourceNewerVersion(path, sourceMtimeNs) {
+			glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", chunkName, path, resolveErr)
+			return false
+		}
+		if !isTransientResolveError(resolveErr) {
+			glog.V(0).Infof("resolve manifest %s for %s: non-transient error, propagating: %v", chunkName, path, resolveErr)
+			return false
+		}
+		attempts++
+		if !canCheckSupersession && attempts >= maxUnverifiableResolveAttempts {
+			glog.V(0).Infof("resolve manifest %s for %s: supersession unverifiable, propagating after %d attempts: %v", chunkName, path, attempts, resolveErr)
+			return false
+		}
+		glog.V(0).Infof("resolve manifest %s for %s: %v", chunkName, path, resolveErr)
+		return true
+	}
+}
+
+// isTransientResolveError reports whether a manifest resolve failure belongs to
+// the classes that clear on their own during a write burst — volume-lookup
+// races and network interruptions. Anything else (corrupt manifest data, bad
+// file ids) is permanent and must propagate rather than retry forever.
+func isTransientResolveError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if isRetryableNetworkError(err) {
+		return true
+	}
+	errStr := err.Error()
+	return strings.Contains(errStr, "LookupFileId") ||
+		(strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found"))
+}
+
 func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourceFileId string, manifestData []byte) (fileId string, err error) {
 	uploader, err := fs.getUploader()
 	if err != nil {
@@ -440,17 +503,22 @@ func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) e
 // version being replayed is stale. The lookup runs regardless of sourceMtimeNs so
 // a deleted source is detected even when the replayed mtime is epoch/unset.
 func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int64) bool {
-	if fs.filerSource == nil {
-		return false
-	}
-
 	sourcePath, ok := fs.targetPathToSourcePath(targetPath)
 	if !ok {
 		return false
 	}
+	return SourceSupersedes(context.Background(), fs.filerSource, sourcePath, sourceMtimeNs)
+}

-	sourceEntry, err := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath)
-	return sourceSupersedes(sourcePath, sourceEntry, err, sourceMtimeNs)
+// SourceSupersedes reports whether the live source's entry at sourcePath has
+// moved past the replayed version (mtimeNs) — deleted or strictly newer — so
+// skipping it is lossless. Exported for filer.backup's non-filer-sink decision.
+func SourceSupersedes(ctx context.Context, filerSource *source.FilerSource, sourcePath util.FullPath, mtimeNs int64) bool {
+	if filerSource == nil {
+		return false
+	}
+	sourceEntry, err := filer_pb.GetEntry(ctx, filerSource, sourcePath)
+	return sourceSupersedes(sourcePath, sourceEntry, err, mtimeNs)
 }

 // sourceSupersedes reports whether the replayed version (sourceMtimeNs) is stale:
@@ -384,3 +384,44 @@ func TestSourceSupersedesEpochMtime(t *testing.T) {
 		t.Fatal("epoch-mtime live source must not be reported superseded")
 	}
 }
+
+// An incremental sink's dated target keys cannot be mapped back to a source
+// path, so supersession is unverifiable: the gate must stop after the bounded
+// attempts and propagate instead of spinning forever.
+func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) {
+	fs := &FilerSink{isIncremental: true, dir: "/backup"}
+	gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", 123, "3,01abc")
+	resolveErr := errors.New("LookupFileId volume id 3: not found")
+	for i := 1; i < maxUnverifiableResolveAttempts; i++ {
+		if !gate(resolveErr) {
+			t.Fatalf("attempt %d: gate must keep retrying before the bound", i)
+		}
+	}
+	if gate(resolveErr) {
+		t.Error("gate must propagate once the bound is reached with supersession unverifiable")
+	}
+}
+
+// Non-transient resolve errors (corrupt manifest data, bad file ids) must
+// propagate immediately — before any attempt counting or supersession
+// mapping — so the configured metadata error policy applies, instead of
+// retrying until the source is superseded.
+func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) {
+	fs := &FilerSink{dir: "/backup"}
+	gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", 123, "3,01abc")
+	permanentErrs := []error{
+		errors.New("fail to unmarshal manifest 3,01abc: proto: cannot parse invalid wire-format data"),
+		errors.New("invalid fileId abc"),
+	}
+	for _, err := range permanentErrs {
+		if gate(err) {
+			t.Errorf("non-transient error must propagate immediately: %v", err)
+		}
+	}
+	if !gate(errors.New("LookupFileId volume id 3: not found")) {
+		t.Error("transient lookup race must keep retrying on the first attempt")
+	}
+	if isTransientResolveError(nil) {
+		t.Error("isTransientResolveError(nil) must be false")
+	}
+}
@@ -260,8 +260,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 			if errors.Is(err, errChunkSizeMismatch) {
 				return fs.onCorruptChunk(key, entry, err)
 			}
-			glog.Warningf("replicate entry chunks %s: %v", key, err)
-			return nil
+			return fs.onReplicateChunkError(key, entry, err)
 		}

 		// glog.V(4).Infof("replicated %s %+v ===> %+v", key, entry.GetChunks(), replicatedChunks)
@@ -339,8 +338,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 			if errors.Is(err, errChunkSizeMismatch) {
 				return true, fs.onCorruptChunk(targetKey, newEntry, err)
 			}
-			glog.Warningf("replicate entry chunks %s: %v", key, err)
-			return true, nil
+			return true, fs.onReplicateChunkError(targetKey, newEntry, err)
 		}
 		existingEntry.Chunks = replicatedChunks
 		existingEntry.Attributes = newEntry.Attributes
@@ -368,8 +366,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 			if errors.Is(err, errChunkSizeMismatch) {
 				return true, fs.onCorruptChunk(targetKey, newEntry, err)
 			}
-			glog.Warningf("replicate entry chunks %s: %v", key, err)
-			return true, nil
+			return true, fs.onReplicateChunkError(targetKey, newEntry, err)
 		}
 		existingEntry.Chunks = append(existingEntry.GetChunks(), replicatedChunks...)
 		existingEntry.Attributes = newEntry.Attributes
@@ -426,6 +423,11 @@ func getEntryMtimeNs(entry *filer_pb.Entry) int64 {
 	return entry.Attributes.Mtime*int64(1e9) + int64(entry.Attributes.MtimeNs)
 }

+// EntryMtimeNs exposes getEntryMtimeNs for filer.backup's supersession check.
+func EntryMtimeNs(entry *filer_pb.Entry) int64 {
+	return getEntryMtimeNs(entry)
+}
+
 // onCorruptChunk handles a permanent chunk size-mismatch (fetched source bytes
 // disagree with source metadata). If the source already holds a newer version the
 // event is a stale replay whose chunk was overwritten and GC'd — skip it (lossless:
@@ -440,6 +442,19 @@ func (fs *FilerSink) onCorruptChunk(key string, entry *filer_pb.Entry, err error
 	return err
 }

+// onReplicateChunkError handles a non-size-mismatch replicateChunks failure.
+// Skip (return nil) only when the live source has moved past this version —
+// deleted or strictly-newer mtime — so a later event carries the current
+// content and the skip is lossless. Otherwise propagate so the offset stays
+// put and the event is retried; returning nil would silently drop the file.
+func (fs *FilerSink) onReplicateChunkError(key string, entry *filer_pb.Entry, err error) error {
+	if fs.hasSourceNewerVersion(key, getEntryMtimeNs(entry)) {
+		glog.Warningf("skip stale entry %s, source superseded during replicate: %v", key, err)
+		return nil
+	}
+	return fmt.Errorf("replicate entry chunks %s: %w", key, err)
+}
+
 // updateAction decides how UpdateEntry reconciles an incoming source version with
 // the destination's current entry.
 type updateAction int
@@ -109,6 +109,28 @@ func TestOnCorruptChunkRefusesWhenSupersessionUnconfirmed(t *testing.T) {
 	}
 }

+// onReplicateChunkError must NOT swallow a transient/partial replicate failure
+// (a volume-id lookup race or unretried manifest-chunk resolve during a write
+// burst) when it cannot confirm the source has moved on. Here filerSource is
+// nil, so hasSourceNewerVersion is always false — the "supersession unconfirmed"
+// case. Returning nil would let the metadata subscription advance its offset
+// past an event whose file was never written, silently dropping it (the
+// filer.backup partial-landing bug). It must propagate so the offset stays put
+// and the event is reprocessed. Genuine source-404 (gone/superseded) skipping is
+// covered by TestSourceSupersedes, which gates the return-nil branch.
+func TestOnReplicateChunkErrorPropagatesWhenSupersessionUnconfirmed(t *testing.T) {
+	fs := &FilerSink{} // filerSource nil => hasSourceNewerVersion always false
+	sentinel := errors.New("replicate manifest data chunks 7,02def: volume id 7 not found")
+
+	got := fs.onReplicateChunkError("/buckets/x/model.pt",
+		&filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: 5, MtimeNs: 200}},
+		sentinel)
+
+	if !errors.Is(got, sentinel) {
+		t.Fatalf("expected the transient replicate error to be propagated (not swallowed), got %v", got)
+	}
+}
+
 // chooseUpdateAction decides skip vs repair vs normal. A destination left
 // truncated by an earlier failed replication can carry a newer mtime (the
 // source preserved an old mtime while the partial copy was written recently),