From c17aebeecf1cf47f8ff19a5007db42b7754c4772 Mon Sep 17 00:00:00 2001 From: Jaehoon Kim Date: Fri, 10 Jul 2026 17:47:46 +0900 Subject: [PATCH] fix(filer.backup): prevent silent backup data loss on transient not-found (#10295) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 "/", 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 * 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 * 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 --------- Co-authored-by: Claude Fable 5 --- weed/command/filer_backup.go | 70 +++++++-- weed/command/filer_backup_test.go | 135 ++++++++++++++++++ .../replication/sink/filersink/fetch_write.go | 82 ++++++++++- .../sink/filersink/fetch_write_test.go | 41 ++++++ weed/replication/sink/filersink/filer_sink.go | 27 +++- .../sink/filersink/filer_sink_test.go | 22 +++ 6 files changed, 353 insertions(+), 24 deletions(-) diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index 0f87e6309..c5adeed08 100644 --- a/weed/command/filer_backup.go +++ b/weed/command/filer_backup.go @@ -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) } diff --git a/weed/command/filer_backup_test.go b/weed/command/filer_backup_test.go index 52031f7d0..addd06276 100644 --- a/weed/command/filer_backup_test.go +++ b/weed/command/filer_backup_test.go @@ -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 "/", 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. diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 4bfa354d3..d9d553de8 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -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: diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index 0cdd2c7f5..b4b357d44 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -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") + } +} diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 6f232f123..a76923c41 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -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 diff --git a/weed/replication/sink/filersink/filer_sink_test.go b/weed/replication/sink/filersink/filer_sink_test.go index 224bb6c23..eeab374cc 100644 --- a/weed/replication/sink/filersink/filer_sink_test.go +++ b/weed/replication/sink/filersink/filer_sink_test.go @@ -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),