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 }