From d485d2c9cda86a15fc4723d20200551ff01f8d39 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 10 Nov 2026 09:56:17 -0800 Subject: [PATCH] filer.sync: shorten retries for confirmed missing chunks (#12546) * filer.sync: shorten retries for confirmed missing chunks * filer.sync: clarify missing chunk grace period names --- .../replication/sink/filersink/fetch_write.go | 32 +++-- .../sink/filersink/fetch_write_test.go | 118 +++++++++++++++--- weed/replication/source/filer_source.go | 17 +++ weed/replication/source/filer_source_test.go | 78 +++++++++++- 4 files changed, 218 insertions(+), 27 deletions(-) diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 1dbc12b4a5..5c9b5f2cea 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -29,11 +29,16 @@ import ( util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) -// missingSourceChunkGrace is how long a chunk the source cannot produce keeps -// being retried before it counts as gone for good. Long enough to outlast a -// volume server restart or a master failover, after which every volume that -// still exists is registered again. -var missingSourceChunkGrace = 30 * time.Minute +// sourceRecoveryGrace allows missing volume locations and ambiguous/proxied +// 404s to recover from a volume server restart or master failover before the +// chunk counts as permanently missing. +var sourceRecoveryGrace = 30 * time.Minute + +// confirmedChunkMissingGrace bounds retries when every located volume replica +// answers 404. Allow brief read/write races without stalling each vacuumed +// needle for the volume recovery window. Keep the longer grace for missing +// volume locations and ambiguous/proxied 404s. +var confirmedChunkMissingGrace = 30 * time.Second // sourceLookupTimeout bounds a best-effort source entry probe: a source that // accepts the call but whose metadata store never answers must stall @@ -58,7 +63,7 @@ var errSourceChunkMissing = errors.New("source chunk missing") // does. Both shapes also cover a volume that is merely offline, which is why the // gate below waits them out instead of trusting a single answer. func isSourceChunkMissing(err error) bool { - return errors.Is(err, source.ErrVolumeNotFound) || errors.Is(err, util_http.ErrNotFound) + return errors.Is(err, source.ErrVolumeNotFound) || errors.Is(err, source.ErrChunkNotFound) || errors.Is(err, util_http.ErrNotFound) } // missingSourceChunkGate decides when to stop retrying a chunk the source cannot @@ -80,18 +85,25 @@ func (g *missingSourceChunkGate) isPermanent(err error) bool { switch { case errors.Is(err, source.ErrVolumeNotFound): since = g.sink.sourceVolumeMissingSince(g.volumeId) - case errors.Is(err, util_http.ErrNotFound): + case errors.Is(err, source.ErrChunkNotFound), errors.Is(err, util_http.ErrNotFound): since = g.sink.sourceNeedleMissingSince(g.fileId) default: return false } - if time.Since(since) < missingSourceChunkGrace { + if time.Since(since) < g.gracePeriod(err) { return false } g.gaveUp = true return true } +func (g *missingSourceChunkGate) gracePeriod(err error) time.Duration { + if errors.Is(err, source.ErrChunkNotFound) { + return confirmedChunkMissingGrace + } + return sourceRecoveryGrace +} + // wrap marks err permanent once the gate gave up, so the sink can tell a source // that lost the data from one that is only unreachable right now. func (g *missingSourceChunkGate) wrap(err error) error { @@ -292,7 +304,7 @@ func (fs *FilerSink) manifestResolveRetryGate(path string, replayed *filer_pb.En } if missingGate.isPermanent(resolveErr) { glog.Errorf("source has not had manifest %s for %s in %v, giving up on it: %v", - chunkName, path, missingSourceChunkGrace, resolveErr) + chunkName, path, missingGate.gracePeriod(resolveErr), resolveErr) return false } if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName, replayed) { @@ -530,7 +542,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, } if missingGate.isPermanent(retryErr) { glog.Errorf("source has not had %s for %s in %v, giving up on it: %v", - sourceChunk.GetFileIdString(), path, missingSourceChunkGrace, retryErr) + sourceChunk.GetFileIdString(), path, missingGate.gracePeriod(retryErr), retryErr) return false } if isRetryableNetworkError(retryErr) { diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index f9e4f11d6a..18eb454418 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -594,11 +594,11 @@ func startRecordingFiler(t *testing.T, existing *filer_pb.Entry) (*recordingFile // completed, so it held its slot and pinned the offset watermark at the event // ahead of it, and filer.sync never checkpointed again. func TestFetchAndWriteStopsOnMissingSourceChunk(t *testing.T) { - prevRetryWaitTime, prevGrace := util.RetryWaitTime, missingSourceChunkGrace + prevRetryWaitTime, prevGrace := util.RetryWaitTime, sourceRecoveryGrace util.RetryWaitTime = 100 * time.Millisecond - missingSourceChunkGrace = 500 * time.Millisecond + sourceRecoveryGrace = 500 * time.Millisecond t.Cleanup(func() { - util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace + util.RetryWaitTime, sourceRecoveryGrace = prevRetryWaitTime, prevGrace }) live := &filer_pb.Entry{ @@ -661,7 +661,7 @@ func TestMissingSourceChunkGate(t *testing.T) { } // a second file in the same volume inherits that wait instead of restarting it - fs.missingVolumes.Store("5617", time.Now().Add(-2*missingSourceChunkGrace)) + fs.missingVolumes.Store("5617", time.Now().Add(-2*sourceRecoveryGrace)) second := fs.newMissingSourceChunkGate("5617,02def") if !second.isPermanent(volumeGone) { t.Fatal("a later file must inherit the wait its volume already served") @@ -682,7 +682,7 @@ func TestMissingSourceChunkGate(t *testing.T) { // 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)) + fs.missingNeedles.Store("5619,01abc", time.Now().Add(-2*sourceRecoveryGrace)) needleGate := fs.newMissingSourceChunkGate("5619,01abc") fs.sourceServed("5619,02def") if !needleGate.isPermanent(needleGone) { @@ -702,6 +702,96 @@ func TestMissingSourceChunkGate(t *testing.T) { } } +func TestMissingSourceChunkGateConfirmedNotFound(t *testing.T) { + confirmed := source.ErrChunkNotFound + fs := &FilerSink{} + gate := fs.newMissingSourceChunkGate("5617,01abc") + if gate.isPermanent(confirmed) { + t.Fatal("a first 404 must allow a brief recovery window") + } + // Only the short needle grace has elapsed, not the volume recovery grace. + fs.missingNeedles.Store("5617,01abc", time.Now().Add(-2*confirmedChunkMissingGrace)) + if gate.isPermanent(util_http.ErrNotFound) { + t.Fatal("an ambiguous 404 must retain the volume recovery grace") + } + if gate.isPermanent(source.ErrVolumeNotFound) { + t.Fatal("an unlocatable volume must retain its recovery grace") + } + fs.sourceServed("5617,02def") + if !gate.isPermanent(confirmed) { + t.Fatal("confirmed missing needle must expire without the 30-minute wait") + } + if !errors.Is(gate.wrap(confirmed), errSourceChunkMissing) { + t.Fatal("confirmed missing needle must be marked permanent") + } + if fs.newMissingSourceChunkGate("5617,03abc").isPermanent(confirmed) { + t.Fatal("a different needle must get its own recovery window") + } + fs.sourceServed("5617,01abc") + if fs.newMissingSourceChunkGate("5617,01abc").isPermanent(confirmed) { + t.Fatal("serving the missing needle must reset its recovery window") + } +} + +// Unchanged source metadata can still reference vacuumed chunks. Each missing +// file must finish after the short 404 grace so later events can advance, even +// when the sink has not successfully read any other chunk yet. +func TestCreateEntryConfirmedMissingChunksUseShortGrace(t *testing.T) { + prevGrace := confirmedChunkMissingGrace + confirmedChunkMissingGrace = 20 * time.Millisecond + t.Cleanup(func() { confirmedChunkMissingGrace = prevGrace }) + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Not Found", http.StatusNotFound) + })) + defer gone.Close() + entries := map[string]*filer_pb.Entry{} + for i, name := range []string{"x.bin", "y.bin"} { + entries[name] = &filer_pb.Entry{ + Name: name, + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: fmt.Sprintf("5617,0%dabc", i+1), Size: 10}}, + } + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{ + volumeUrl: strings.TrimPrefix(gone.URL, "http://"), + resolvable: []string{"5617"}, + lookupFn: func(dir, name string) (*filer_pb.Entry, error) { + return entries[name], nil + }, + }) + dst, dstAddr := startRecordingFiler(t, nil) + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + done := make(chan error, 1) + go func() { + for _, name := range []string{"x.bin", "y.bin"} { + if err := fs.CreateEntry("/dst/"+name, entries[name], nil); err != nil { + done <- err + return + } + } + done <- nil + }() + select { + case err := <-done: + if err != nil { + t.Fatalf("confirmed missing files should be skipped: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("confirmed 404s waited for the volume recovery grace") + } + if dst.creates.Load() != 0 { + t.Fatal("missing chunk data must not be committed to the destination") + } +} + // Once the source has lost a chunk for good, holding the sync offset for its // entry only stops every later event from ever being checkpointed: the bytes are // not coming back. Skip it loudly instead — but only while the source is @@ -1024,11 +1114,11 @@ func TestFetchAndWriteStopsWhenOlderLiveEntryDroppedChunk(t *testing.T) { // 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 + prevRetryWaitTime, prevGrace := util.RetryWaitTime, confirmedChunkMissingGrace util.RetryWaitTime = 100 * time.Millisecond - missingSourceChunkGrace = 500 * time.Millisecond + confirmedChunkMissingGrace = 500 * time.Millisecond t.Cleanup(func() { - util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace + util.RetryWaitTime, confirmedChunkMissingGrace = prevRetryWaitTime, prevGrace }) gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -1184,11 +1274,11 @@ func committedOverWithOlderMtime() (live, replayed *filer_pb.Entry) { // listed the replay's chunk. That lagging entry cannot prove the chunk dropped: // the 404 must be waited out like any other, not cut short. func TestFetchAndWriteLaggingLookupKeepsRetrying(t *testing.T) { - prevRetryWaitTime, prevGrace := util.RetryWaitTime, missingSourceChunkGrace + prevRetryWaitTime, prevGrace := util.RetryWaitTime, confirmedChunkMissingGrace util.RetryWaitTime = 100 * time.Millisecond - missingSourceChunkGrace = 500 * time.Millisecond + confirmedChunkMissingGrace = 500 * time.Millisecond t.Cleanup(func() { - util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace + util.RetryWaitTime, confirmedChunkMissingGrace = prevRetryWaitTime, prevGrace }) gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -1287,9 +1377,9 @@ func TestCreateEntryReplicatesCurrentVersion(t *testing.T) { // The source has no volume to serve the chunk, so replication fails — the // point is that it was attempted, not skipped as stale. A short grace lets // the give-up path return instead of holding the test for the full wait. - prevGrace := missingSourceChunkGrace - missingSourceChunkGrace = 200 * time.Millisecond - t.Cleanup(func() { missingSourceChunkGrace = prevGrace }) + prevGrace := sourceRecoveryGrace + sourceRecoveryGrace = 200 * time.Millisecond + t.Cleanup(func() { sourceRecoveryGrace = prevGrace }) if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err == nil { t.Fatal("replication of a chunk the source cannot serve must fail") diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index cb8f4fcd0c..256ab443d0 100644 --- a/weed/replication/source/filer_source.go +++ b/weed/replication/source/filer_source.go @@ -25,6 +25,10 @@ import ( // still get past. var ErrVolumeNotFound = errors.New("volume not found") +// ErrChunkNotFound reports that every located volume replica answered 404 for +// the chunk. A failed request or a filer-proxied 404 cannot establish this. +var ErrChunkNotFound = errors.New("chunk not found on any volume replica") + type FilerSource struct { grpcAddress string grpcDialOption grpc.DialOption @@ -165,6 +169,7 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h return "", nil, nil, err } + var replicaErr error for _, fileUrl := range fileUrls { filename, header, resp, err = downloadFn(fileUrl, "", offset) if err == nil { @@ -172,12 +177,24 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h } if err != nil { resp = nil + if !errors.Is(err, util_http.ErrNotFound) { + replicaErr = err + } glog.V(0).Infof("fail to read part %s from %s offset %d: %v", fileId, fileUrl, offset, err) continue } glog.V(4).Infof("read part %s from %s offset %d content-length:%s", fileId, fileUrl, offset, header.Get("Content-Length")) break } + if err != nil { + if replicaErr != nil { + // An unreachable replica may still hold the chunk. Its failure must + // not be hidden by another replica answering 404 later in the list. + err = replicaErr + } else if errors.Is(err, util_http.ErrNotFound) { + err = fmt.Errorf("%w: %w", ErrChunkNotFound, err) + } + } return filename, header, resp, err } diff --git a/weed/replication/source/filer_source_test.go b/weed/replication/source/filer_source_test.go index dd5d2215b3..34a653b163 100644 --- a/weed/replication/source/filer_source_test.go +++ b/weed/replication/source/filer_source_test.go @@ -378,18 +378,90 @@ func TestReadPartSkipsNotFoundReplica(t *testing.T) { // Every replica gone: the caller must see a not-found error it can act on, and no // response left open behind it. func TestReadPartAllReplicasNotFound(t *testing.T) { - gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var hits atomic.Int32 + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) http.Error(w, "Not Found", http.StatusNotFound) - })) + }) + gone := httptest.NewServer(handler) defer gone.Close() + second := httptest.NewServer(handler) + defer second.Close() - filerSource := startLookupFiler(t, strings.TrimPrefix(gone.URL, "http://")) + filerSource := startLookupFiler(t, strings.TrimPrefix(gone.URL, "http://"), strings.TrimPrefix(second.URL, "http://")) _, _, resp, err := filerSource.ReadPart("5617,01abc", 0) if !errors.Is(err, util_http.ErrNotFound) { t.Fatalf("expected ErrNotFound, got %v", err) } + if !errors.Is(err, ErrChunkNotFound) { + t.Fatalf("all replicas answered 404, expected ErrChunkNotFound, got %v", err) + } + if hits.Load() != 2 { + t.Fatalf("expected every replica to be tried, got %d requests", hits.Load()) + } if resp != nil { t.Fatal("expected no response alongside the error") } } + +// A 404 from one replica cannot rule out a chunk held by another replica that +// failed to answer. Preserve the other failure regardless of replica order. +func TestReadPartPreservesReplicaFailure(t *testing.T) { + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Not Found", http.StatusNotFound) + })) + defer gone.Close() + for _, tc := range []struct { + name string + status int + unreachable bool + }{ + {http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable, false}, + {http.StatusText(http.StatusUnauthorized), http.StatusUnauthorized, false}, + {"connection refused", http.StatusServiceUnavailable, true}, + } { + t.Run(tc.name, func(t *testing.T) { + failed := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tc.status) + })) + defer failed.Close() + if tc.unreachable { + failed.Close() + } + locations := []string{strings.TrimPrefix(failed.URL, "http://"), strings.TrimPrefix(gone.URL, "http://")} + for _, reverse := range []bool{false, true} { + if reverse { + locations[0], locations[1] = locations[1], locations[0] + } + filerSource := startLookupFiler(t, locations...) + _, _, resp, err := filerSource.ReadPart("5617,01abc", 0) + if err == nil || !strings.Contains(err.Error(), tc.name) { + t.Fatalf("expected replica failure %s, got %v", tc.name, err) + } + if errors.Is(err, ErrChunkNotFound) || errors.Is(err, util_http.ErrNotFound) { + t.Fatalf("a failed replica must not count as a missing chunk: %v", err) + } + if resp != nil { + t.Fatal("expected no response alongside the error") + } + } + }) + } +} + +func TestReadPartProxyNotFoundIsUnconfirmed(t *testing.T) { + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Not Found", http.StatusNotFound) + })) + defer gone.Close() + address := strings.TrimPrefix(gone.URL, "http://") + filerSource := &FilerSource{} + if err := filerSource.DoInitialize(address, address, "/", true); err != nil { + t.Fatal(err) + } + _, _, _, err := filerSource.ReadPart("5617,01abc", 0) + if !errors.Is(err, util_http.ErrNotFound) || errors.Is(err, ErrChunkNotFound) { + t.Fatalf("a proxied 404 must remain ambiguous: %v", err) + } +}