From 0fb6c0e77dd5b5b895992421d6698c98f004620e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 19 Oct 2026 18:11:51 -0700 Subject: [PATCH] filer.sync: unstick sync on chunks the source no longer has (#11845) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer.sync: keep the needle-404 clock from resetting on live volumes The give-up clock for a source chunk that cannot be produced is kept per volume so a vacuumed volume's files all inherit one wait. But a needle 404 and a volume lookup failure shared that clock, and sourceServed cleared it whenever any needle on the same volume read fine, so on a busy cluster the grace never elapsed for a single deleted needle on a healthy volume and filer.sync retried it forever (#11841). Give needle not-found answers their own per-volume clock that serving siblings does not reset. * filer.sync: an answered 404 needs no liveness probe before skipping After the missing-chunk gate gives up, onReplicateChunkError only skips the entry when a re-read of the most recently served chunk succeeds. That probe can never have been armed on a fresh sync, and its single stored file id can itself be vacuumed since — either way the entry stays pinned and the offset never advances. A needle 404 was answered by a live volume server, which is itself proof the source is serving requests, so skip on it directly. Keep the probe for the lookup-failure flavor, where no server answered and the silence can still mean a source outage. * filer.sync: skip a replayed version the source rewrote in place Supersession was decided by entry mtime alone, so a source entry whose chunks were swapped without an mtime bump — remote cache eviction and re-cache, an in-place repair — never read as superseded, and its vacuumed chunks were retried indefinitely. When a chunk is missing, compare the replayed chunk's file id against the live entry's resolved chunk set: absent means the replayed version is obsolete and its event is skipped on the first failure instead of after the give-up window. The same live-entry comparison now also decides the entry-level skip in onCorruptChunk/onReplicateChunkError. * filer.sync: give each missing needle its own grace clock A 404 is one needle's answer, but the clock was kept per volume: once the first needle's wait expired, every other needle on that volume inherited an already-elapsed clock and could be written off on its first sighting. Track the wait per file id and clear it only when that needle itself serves again, so a live volume's healthy needles neither shorten nor restart it. * filer.sync: keep the probe for filer-proxied 404s ReadPart proxied through the source filer collapses "no volume locations" into the same 404 a missing needle returns, so that answer cannot stand in for the liveness probe the way a direct volume read's can. Only treat 404 as an answered needle when the sink reads volume servers directly. * filer.sync: only a not-older live entry can supersede a replay Chunk-list comparison must respect the same ordering the mtime check does: a live entry older than the replayed version means the source was rolled back, so its differing chunks say nothing about whether the replayed data is still needed. Gate the content check on live mtime >= replayed mtime in both sourceMovedOn and sourceDroppedChunk. * filer.sync: compare chunk ids without resolving manifests resolvedChunkFileIds downloaded manifest files to expand chunk lists, so a manifest sitting on a dead volume made the supersession check itself fail and left the offset pinned behind a replica the source cannot read. File ids are immutable needle identities — the unresolved sets already compare chunk content exactly, and the check no longer performs I/O. * filer.sync: judge nested chunks by their top-level chunk A manifest child is not in the entry's chunk list even when the manifest still references it, so checking the live entry for the child id reported it dropped and returned the raw 404 — unwrapped, so the offset still pinned. Thread the entry-level chunk id down the manifest recursion and consult it for the dropped-chunk check instead. * filer.sync: let the manifest-child test exit through the gate Run the worker with a shortened grace period and assert the returned error is marked errSourceChunkMissing, instead of racing the worker still running against a restored RetryWaitTime. --- .../replication/sink/filersink/fetch_write.go | 109 +++++++-- .../sink/filersink/fetch_write_test.go | 223 +++++++++++++++++- weed/replication/sink/filersink/filer_sink.go | 51 +++- weed/replication/source/filer_source.go | 7 + 4 files changed, 359 insertions(+), 31 deletions(-) diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 1c165c7d1f..bffce98b9e 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -49,20 +49,29 @@ func isSourceChunkMissing(err error) bool { // missingSourceChunkGate decides when to stop retrying a chunk the source cannot // produce, so a lookup race or a restarting volume server is waited out while a -// vacuumed one is written off instead of retried forever. The wait is kept per -// volume on the sink: a volume that is gone took every file it held with it, and -// waiting it out once per file would stall the sync for as long as it holds files. +// vacuumed one is written off instead of retried forever. A volume that is gone +// took every file it held with it, so its wait is kept per volume on the sink; +// a 404 is one needle's answer, so its wait is kept per file id instead — a +// sibling needle succeeding must not restart it, and a different needle must get +// its own full wait. type missingSourceChunkGate struct { sink *FilerSink volumeId string + fileId string gaveUp bool } func (g *missingSourceChunkGate) isPermanent(err error) bool { - if !isSourceChunkMissing(err) { + var since time.Time + switch { + case errors.Is(err, source.ErrVolumeNotFound): + since = g.sink.sourceVolumeMissingSince(g.volumeId) + case errors.Is(err, util_http.ErrNotFound): + since = g.sink.sourceNeedleMissingSince(g.fileId) + default: return false } - if time.Since(g.sink.sourceVolumeMissingSince(g.volumeId)) < missingSourceChunkGrace { + if time.Since(since) < missingSourceChunkGrace { return false } g.gaveUp = true @@ -78,7 +87,10 @@ func (g *missingSourceChunkGate) wrap(err error) error { return fmt.Errorf("%w: %w", errSourceChunkMissing, err) } -func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, sourceMtimeNs int64) (replicatedChunks []*filer_pb.FileChunk, err error) { +// topLevelId anchors nested chunks to the entry-level chunk that contains them: +// supersession is judged by whether the live entry still lists that chunk, since +// a manifest's id is the identity of its contents. +func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (replicatedChunks []*filer_pb.FileChunk, err error) { if len(sourceChunks) == 0 { return } @@ -125,7 +137,11 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ } if sourceChunk.IsChunkManifest { - replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, sourceMtimeNs) + anchor := topLevelId + if anchor == "" { + anchor = sourceChunk.GetFileIdString() + } + replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, sourceMtimeNs, anchor) if replicateErr != nil { setError(replicateErr) break @@ -143,7 +159,7 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ defer wg.Done() var replicatedChunk *filer_pb.FileChunk retryErr := util.Retry("replicate chunks", func() error { - chunk, e := fs.replicateOneChunk(source, path, sourceMtimeNs) + chunk, e := fs.replicateOneChunk(source, path, sourceMtimeNs, topLevelId) if e != nil { return e } @@ -166,9 +182,9 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_ return } -func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64) (*filer_pb.FileChunk, error) { +func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) { - fileId, err := fs.fetchAndWrite(sourceChunk, path, sourceMtimeNs) + fileId, err := fs.fetchAndWrite(sourceChunk, path, sourceMtimeNs, topLevelId) if err != nil { return nil, fmt.Errorf("copy %s: %w", sourceChunk.GetFileIdString(), err) } @@ -187,7 +203,7 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path str }, nil } -func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64) (*filer_pb.FileChunk, error) { +func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) { // 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. @@ -201,12 +217,12 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk } resolvedChunks = rc return nil - }, fs.manifestResolveRetryGate(path, sourceMtimeNs, sourceChunk.GetFileIdString(), missingGate)) + }, fs.manifestResolveRetryGate(path, sourceMtimeNs, topLevelId, missingGate)) if err != nil { return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), missingGate.wrap(err)) } - replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, sourceMtimeNs) + replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, sourceMtimeNs, topLevelId) if err != nil { return nil, fmt.Errorf("replicate manifest data chunks %s: %w", sourceChunk.GetFileIdString(), err) } @@ -266,6 +282,11 @@ func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName, path, missingSourceChunkGrace, resolveErr) return false } + if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName, sourceMtimeNs) { + glog.V(0).Infof("source no longer lists manifest %s for %s, skipping the obsolete version: %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 @@ -355,7 +376,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc return fileId, nil } -func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64) (fileId string, err error) { +func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (fileId string, err error) { uploader, err := fs.getUploader() if err != nil { glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err) @@ -478,10 +499,21 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, transferStatus.mu.Lock() transferStatus.LastErr = retryErr.Error() transferStatus.mu.Unlock() - if isSourceChunkMissing(retryErr) && !canCheckSupersession { - glog.V(0).Infof("source does not have %s for %s, supersession unverifiable, propagating: %v", - sourceChunk.GetFileIdString(), path, retryErr) - return false + if isSourceChunkMissing(retryErr) { + if !canCheckSupersession { + glog.V(0).Infof("source does not have %s for %s, supersession unverifiable, propagating: %v", + sourceChunk.GetFileIdString(), path, retryErr) + return false + } + anchorId := topLevelId + if anchorId == "" { + anchorId = sourceChunk.GetFileIdString() + } + if fs.sourceDroppedChunk(path, anchorId, sourceMtimeNs) { + glog.V(0).Infof("source no longer lists %s for %s, skipping the obsolete version: %v", + anchorId, path, retryErr) + return false + } } if missingGate.isPermanent(retryErr) { glog.Errorf("source has not had %s for %s in %v, giving up on it: %v", @@ -558,7 +590,7 @@ func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) e } func (fs *FilerSink) newMissingSourceChunkGate(fileId string) *missingSourceChunkGate { - return &missingSourceChunkGate{sink: fs, volumeId: filer.VolumeId(fileId)} + return &missingSourceChunkGate{sink: fs, volumeId: filer.VolumeId(fileId), fileId: fileId} } // sourceVolumeMissingSince returns when the sink first found volumeId @@ -568,11 +600,20 @@ func (fs *FilerSink) sourceVolumeMissingSince(volumeId string) time.Time { return since.(time.Time) } +// sourceNeedleMissingSince returns when fileId first came back not-found from +// an answering volume, recording now on the first sighting. +func (fs *FilerSink) sourceNeedleMissingSince(fileId string) time.Time { + since, _ := fs.missingNeedles.LoadOrStore(fileId, time.Now()) + return since.(time.Time) +} + // sourceServed records a chunk the source did serve: the probe -// sourceStillServesChunks re-checks, and proof its volume is locatable again. +// sourceStillServesChunks re-checks, proof its volume is locatable again, and +// proof the needle itself is back if it had been reported missing. func (fs *FilerSink) sourceServed(fileId string) { fs.lastServedFileId.Store(&fileId) fs.missingVolumes.Delete(filer.VolumeId(fileId)) + fs.missingNeedles.Delete(fileId) } // sourceStillServesChunks reports whether the source cluster can still produce the @@ -611,6 +652,34 @@ func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int6 return SourceSupersedes(context.Background(), fs.filerSource, sourcePath, sourceMtimeNs) } +// sourceDroppedChunk reports whether the live source entry still exists but no +// longer references fileId — a version rewritten in place, which the mtime +// supersession check cannot see. Only a live entry at least as new as the +// replayed version can prove that; an older one means the source was rolled +// back and says nothing about whether the chunk is still needed. +func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, sourceMtimeNs int64) bool { + sourcePath, ok := fs.targetPathToSourcePath(targetPath) + if !ok { + return false + } + live, _, _, err := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath) + if err != nil || live == nil || sourceMtimeNs <= 0 || getEntryMtimeNs(live) < sourceMtimeNs { + return false + } + return !chunkFileIds(live.GetChunks())[fileId] +} + +// chunkFileIds indexes a chunk list by file id. The id is the immutable needle +// identity, so the sets compare chunk content exactly — manifest chunks need no +// resolution, and no volume reads are involved. +func chunkFileIds(chunks []*filer_pb.FileChunk) map[string]bool { + ids := make(map[string]bool, len(chunks)) + for _, c := range chunks { + ids[c.GetFileIdString()] = true + } + return ids +} + // 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. diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index 72e50d0c9e..05eb1855a0 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -233,7 +233,7 @@ func TestFetchAndWriteRejectsZeroByteSource(t *testing.T) { err error }, 1) go func() { - gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", 0) + gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", 0, "") done <- struct { fileId string err error @@ -331,7 +331,7 @@ func TestReplicateChunksPreservesSizeMismatchSentinel(t *testing.T) { sourceChunks := []*filer_pb.FileChunk{{FileId: fid, Size: expectedSize}} - _, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", 0) + _, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", 0, "") if err == nil { t.Fatal("expected error from replicateChunks, got nil") } @@ -441,6 +441,7 @@ type sourceFilerServer struct { mtime int64 volumeUrl string resolvable []string + entry *filer_pb.Entry } func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { @@ -456,6 +457,9 @@ func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.Look } func (s *sourceFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if s.entry != nil { + return &filer_pb.LookupDirectoryEntryResponse{Entry: s.entry}, nil + } return &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{ Name: req.Name, Attributes: &filer_pb.FuseAttributes{Mtime: s.mtime}, @@ -493,6 +497,26 @@ func startSourceFiler(t *testing.T, mtime int64, volumeUrl string, resolvable .. return filerSrc } +func startSourceFilerWithEntry(t *testing.T, entry *filer_pb.Entry, volumeUrl string, resolvable ...string) *source.FilerSource { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, &sourceFilerServer{volumeUrl: volumeUrl, resolvable: resolvable, entry: entry}) + go server.Serve(listener) + t.Cleanup(server.Stop) + + address := listener.Addr().String() + filerSrc := &source.FilerSource{} + if err := filerSrc.DoInitialize(address, address, "/src", false); err != nil { + t.Fatalf("filerSource.DoInitialize: %v", err) + } + filerSrc.SetGrpcDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())) + return filerSrc +} + // A chunk the source cluster cannot produce must stop being retried once the // grace period is up. Before, the retry loop ran forever: the sync job never // completed, so it held its slot and pinned the offset watermark at the event @@ -505,7 +529,11 @@ func TestFetchAndWriteStopsOnMissingSourceChunk(t *testing.T) { util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace }) - filerSrc := startSourceFiler(t, 5, "") + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + filerSrc := startSourceFilerWithEntry(t, live, "") fs := &FilerSink{ filerSource: filerSrc, @@ -516,7 +544,7 @@ func TestFetchAndWriteStopsOnMissingSourceChunk(t *testing.T) { done := make(chan error, 1) go func() { - _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, "/dst/x.bin", 5*int64(time.Second)) + _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, "/dst/x.bin", 5*int64(time.Second), "") done <- fetchErr }() @@ -579,6 +607,27 @@ func TestMissingSourceChunkGate(t *testing.T) { if _, found := fs.missingVolumes.Load("5617"); found { t.Fatal("a served chunk must clear its volume's wait") } + + // a needle not-found answer keeps its own clock: the volume staying alive + // for other needles must not restart the wait for the one that is gone + fs.missingNeedles.Store("5619,01abc", time.Now().Add(-2*missingSourceChunkGrace)) + needleGate := fs.newMissingSourceChunkGate("5619,01abc") + fs.sourceServed("5619,02def") + if !needleGate.isPermanent(needleGone) { + t.Fatal("a needle 404 must keep aging while its volume serves other needles") + } + + // one needle's elapsed wait must not cut short a different needle's + fresh := fs.newMissingSourceChunkGate("5619,03ghi") + if fresh.isPermanent(needleGone) { + t.Fatal("a different needle on the same volume must get its own wait") + } + + // the needle itself serving again is the only thing that restarts its wait + fs.sourceServed("5619,01abc") + if needleGate.isPermanent(needleGone) { + t.Fatal("a needle that serves again must restart its wait") + } } // Once the source has lost a chunk for good, holding the sync offset for its @@ -601,6 +650,28 @@ func TestOnReplicateChunkErrorMissingSourceChunk(t *testing.T) { } }) + t.Run("needle 404 needs no probe", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFiler(t, 5, ""), dir: "/dst"} + notFound := fmt.Errorf("copy 5617,02def: %w: %w", errSourceChunkMissing, util_http.ErrNotFound) + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, notFound); err != nil { + t.Fatalf("a 404 answered by the source is itself proof of liveness, got %v", err) + } + }) + + t.Run("proxied 404 still needs the probe", func(t *testing.T) { + proxied := &source.FilerSource{} + if err := proxied.DoInitialize("127.0.0.1:1", "127.0.0.1:1", "/src", true); err != nil { + t.Fatalf("DoInitialize: %v", err) + } + fs := &FilerSink{filerSource: proxied, dir: "/dst"} + notFound := fmt.Errorf("copy 5617,02def: %w: %w", errSourceChunkMissing, util_http.ErrNotFound) + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, notFound); !errors.Is(err, errSourceChunkMissing) { + t.Fatalf("a filer-proxied 404 also covers missing locations, so it must not skip without the probe, got %v", err) + } + }) + t.Run("source locating nothing", func(t *testing.T) { fs := &FilerSink{filerSource: startSourceFiler(t, 5, ""), dir: "/dst"} served := probe @@ -657,7 +728,7 @@ func TestFetchAndWriteMissingSourceChunkUnverifiableSupersession(t *testing.T) { done := make(chan error, 1) go func() { _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, - "/backup/2026-07-10/x.bin", 5*int64(time.Second)) + "/backup/2026-07-10/x.bin", 5*int64(time.Second), "") done <- fetchErr }() @@ -673,3 +744,145 @@ func TestFetchAndWriteMissingSourceChunkUnverifiableSupersession(t *testing.T) { t.Fatal("fetchAndWrite waited out the grace period with supersession unverifiable") } } + +func TestSourceDroppedChunk(t *testing.T) { + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, + } + fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} + + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", 5*int64(time.Second)) { + t.Fatal("live entry no longer referencing the chunk must count as dropped") + } + if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def", 5*int64(time.Second)) { + t.Fatal("a chunk the live entry still lists must not count as dropped") + } + live.Attributes.Mtime = 4 + if fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", 5*int64(time.Second)) { + t.Fatal("a live entry older than the replayed version cannot prove the chunk dropped") + } +} + +func TestSourceMovedOn(t *testing.T) { + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, + } + replayed := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + + fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} + if !fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("same-mtime entry with different chunks is moved on") + } + replayed.Chunks = live.Chunks + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("identical chunks at equal mtime is the same version") + } + live.Attributes.Mtime = 6 + if !fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("newer mtime is still moved on") + } + live.Attributes.Mtime = 4 + replayed.Chunks = []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}} + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("an older live entry cannot supersede the replayed version") + } +} + +// A replayed version the source rewrote in place must not wait out the +// missing-chunk grace period: the live entry no longer lists the chunk, so the +// loop stops on the first 404. +func TestFetchAndWriteStopsWhenSourceDroppedChunk(t *testing.T) { + prevRetryWaitTime := util.RetryWaitTime + util.RetryWaitTime = 100 * time.Millisecond + t.Cleanup(func() { util.RetryWaitTime = prevRetryWaitTime }) + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer gone.Close() + + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, + } + filerSrc := startSourceFilerWithEntry(t, live, strings.TrimPrefix(gone.URL, "http://"), "5617") + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + done := make(chan error, 1) + go func() { + _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, + "/dst/x.bin", 5*int64(time.Second), "") + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if !errors.Is(fetchErr, util_http.ErrNotFound) { + t.Fatalf("expected the 404 to propagate, got %v", fetchErr) + } + if errors.Is(fetchErr, errSourceChunkMissing) { + t.Fatalf("an obsolete version must not wait for the give-up gate: %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("fetchAndWrite kept retrying a chunk the live entry no longer lists") + } +} + +// A missing needle that lives inside a manifest is judged by whether the live +// entry still lists the manifest, not the child: the manifest id is the +// identity of its contents, so a live entry that kept it still needs the child. +// The child must keep retrying under the gate — its error must come back marked +// permanent by the clock, not returned raw by the dropped-chunk check. +func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) { + prevRetryWaitTime, prevGrace := util.RetryWaitTime, missingSourceChunkGrace + util.RetryWaitTime = 100 * time.Millisecond + missingSourceChunkGrace = 500 * time.Millisecond + t.Cleanup(func() { + util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace + }) + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer gone.Close() + + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,09fed", Size: 10}}, + } + filerSrc := startSourceFilerWithEntry(t, live, strings.TrimPrefix(gone.URL, "http://"), "5617") + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + done := make(chan error, 1) + go func() { + _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, + "/dst/x.bin", 5*int64(time.Second), "5617,09fed") + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if !errors.Is(fetchErr, errSourceChunkMissing) { + t.Fatalf("a child of a still-listed manifest must expire via the gate, got %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("a still-listed manifest's missing child must not retry past the gate") + } +} diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index a710478136..838cd16950 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -24,6 +24,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/replication/sink" "github.com/seaweedfs/seaweedfs/weed/replication/source" "github.com/seaweedfs/seaweedfs/weed/util" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) // ChunkTransferSnapshot is a lock-free, copyable view of a chunk transfer's @@ -67,6 +68,10 @@ type FilerSink struct { // sourceStillServesChunks re-checks before writing an entry off. lastServedFileId atomic.Pointer[string] missingVolumes sync.Map // source volume id -> time.Time first found unlocatable + // missingNeedles records when each file id first came back not-found from an + // answering volume: only that needle serving again resets it, so a live + // volume's other needles neither shorten nor restart the wait. + missingNeedles sync.Map // source file id -> time.Time first answered not-found } func init() { @@ -289,7 +294,7 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [ key, len(resp.Entry.GetChunks()), len(entry.GetChunks())) } - replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry)) + replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "") if err != nil { // don't swallow: committing mismatched bytes would propagate corruption @@ -369,7 +374,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent case updateRepair: 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)) + replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, getEntryMtimeNs(newEntry), "") if err != nil { if errors.Is(err, errChunkSizeMismatch) { return true, fs.onCorruptChunk(targetKey, newEntry, err) @@ -397,7 +402,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent } // replicate the chunks that are new in the source - replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, getEntryMtimeNs(newEntry)) + replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, getEntryMtimeNs(newEntry), "") if err != nil { if errors.Is(err, errChunkSizeMismatch) { return true, fs.onCorruptChunk(targetKey, newEntry, err) @@ -470,7 +475,7 @@ func EntryMtimeNs(entry *filer_pb.Entry) int64 { // meta events are full snapshots, so a later event re-carries the current chunks). // Otherwise the live version is corrupt, so return the error and halt loudly. func (fs *FilerSink) onCorruptChunk(key string, entry *filer_pb.Entry, err error) error { - if fs.hasSourceNewerVersion(key, getEntryMtimeNs(entry)) { + if fs.sourceMovedOn(key, entry) { glog.Warningf("skip stale entry %s with superseded corrupt chunk: %v", key, err) return nil } @@ -486,17 +491,51 @@ func (fs *FilerSink) onCorruptChunk(key string, entry *filer_pb.Entry, err error // later event from ever being checkpointed. 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)) { + if fs.sourceMovedOn(key, entry) { glog.Warningf("skip stale entry %s, source superseded during replicate: %v", key, err) return nil } - if errors.Is(err, errSourceChunkMissing) && fs.sourceStillServesChunks() { + // A volume server's own 404 names this needle, so a direct-read 404 needs no + // liveness probe; a filer-proxied 404 also covers "no volume locations" and + // could be an outage, so it still must pass the probe. + direct404 := errors.Is(err, util_http.ErrNotFound) && fs.filerSource != nil && !fs.filerSource.ProxyByFiler() + if errors.Is(err, errSourceChunkMissing) && (direct404 || fs.sourceStillServesChunks()) { glog.Errorf("skip %s: the source cluster cannot produce its data, so it stays unreplicated: %v", key, err) return nil } return fmt.Errorf("replicate entry chunks %s: %w", key, err) } +// sourceMovedOn reports whether the live source entry moved past the replayed +// version: gone, strictly newer, or rewritten in place so it no longer +// references the same chunks — the last is invisible to the mtime check. A live +// entry older than the replayed version proves nothing about it. +func (fs *FilerSink) sourceMovedOn(targetPath string, entry *filer_pb.Entry) bool { + sourcePath, ok := fs.targetPathToSourcePath(targetPath) + if !ok { + return false + } + replayedMtimeNs := getEntryMtimeNs(entry) + live, _, _, lookupErr := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath) + if sourceSupersedes(sourcePath, live, lookupErr, replayedMtimeNs) { + return true + } + if lookupErr != nil || live == nil || replayedMtimeNs <= 0 || getEntryMtimeNs(live) < replayedMtimeNs { + return false + } + liveIds := chunkFileIds(live.GetChunks()) + replayedIds := chunkFileIds(entry.GetChunks()) + if len(liveIds) != len(replayedIds) { + return true + } + for id := range replayedIds { + if !liveIds[id] { + return true + } + } + return false +} + // updateAction decides how UpdateEntry reconciles an incoming source version with // the destination's current entry. type updateAction int diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index 376a518046..cb8f4fcd0c 100644 --- a/weed/replication/source/filer_source.go +++ b/weed/replication/source/filer_source.go @@ -74,6 +74,13 @@ func (fs *FilerSource) SetFilerJwtProvider(provider security.FilerJwtProvider) { // FilerJwt returns the side-specific filer API credential, or nil when the // process-wide jwt.filer_signing configuration applies. +// ProxyByFiler reports whether chunk reads are proxied through the source +// filer. A proxied 404 can also mean the filer found no volume locations, so it +// does not isolate a missing needle the way a direct volume read's 404 does. +func (fs *FilerSource) ProxyByFiler() bool { + return fs.proxyByFiler +} + func (fs *FilerSource) FilerJwt() security.FilerJwtProvider { return fs.jwtForFiler }