diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index b8532f091..b05a926b6 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -29,6 +29,55 @@ 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 + +// errSourceChunkMissing is a permanent replication failure: the source cluster +// no longer holds the chunk's data, so no later attempt can fetch it. +var errSourceChunkMissing = errors.New("source chunk missing") + +// isSourceChunkMissing reports whether the source answered that it does not have +// the chunk — no location for its volume, or a 404 from the volume server that +// 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) +} + +// 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. +type missingSourceChunkGate struct { + sink *FilerSink + volumeId string + gaveUp bool +} + +func (g *missingSourceChunkGate) isPermanent(err error) bool { + if !isSourceChunkMissing(err) { + return false + } + if time.Since(g.sink.sourceVolumeMissingSince(g.volumeId)) < missingSourceChunkGrace { + return false + } + g.gaveUp = true + return true +} + +// 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 { + if err == nil || !g.gaveUp { + return err + } + 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) { if len(sourceChunks) == 0 { return @@ -144,6 +193,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk // supersession or, when that cannot be checked, after a few attempts. var resolvedChunks []*filer_pb.FileChunk resolveName := fmt.Sprintf("resolve manifest %s", sourceChunk.GetFileIdString()) + missingGate := fs.newMissingSourceChunkGate(sourceChunk.GetFileIdString()) err := util.RetryUntil(resolveName, func() error { rc, e := filer.ResolveOneChunkManifest(ctx, fs.filerSource.LookupFileId, sourceChunk) if e != nil { @@ -151,9 +201,9 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk } resolvedChunks = rc return nil - }, fs.manifestResolveRetryGate(path, sourceMtimeNs, sourceChunk.GetFileIdString())) + }, fs.manifestResolveRetryGate(path, sourceMtimeNs, sourceChunk.GetFileIdString(), missingGate)) if err != nil { - return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), err) + return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), missingGate.wrap(err)) } replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, sourceMtimeNs) @@ -196,13 +246,14 @@ 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 +// skips it as lossless), stop once the source has been unable to produce the +// manifest for the whole grace period, 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 { +func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName string, missingGate *missingSourceChunkGate) func(error) bool { _, canCheckSupersession := fs.targetPathToSourcePath(path) attempts := 0 return func(resolveErr error) (shouldContinue bool) { @@ -210,6 +261,11 @@ func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", chunkName, path, resolveErr) return false } + 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) + return false + } if !isTransientResolveError(resolveErr) { glog.V(0).Infof("resolve manifest %s for %s: non-transient error, propagating: %v", chunkName, path, resolveErr) return false @@ -316,6 +372,8 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, defer fs.activeTransfers.Delete(sourceChunk.GetFileIdString()) transientBackoff := time.Duration(0) + _, canCheckSupersession := fs.targetPathToSourcePath(path) + missingGate := fs.newMissingSourceChunkGate(sourceChunk.GetFileIdString()) var partialData []byte var savedFilename string var savedHeader http.Header @@ -359,6 +417,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, if err := validateReplicatedReadSize(sourceChunk, len(fullData)); err != nil { return err } + fs.sourceServed(sourceChunk.GetFileIdString()) transferStatus.mu.Lock() transferStatus.BytesReceived = int64(len(fullData)) @@ -417,6 +476,16 @@ 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 missingGate.isPermanent(retryErr) { + glog.Errorf("source has not had %s for %s in %v, giving up on it: %v", + sourceChunk.GetFileIdString(), path, missingSourceChunkGrace, retryErr) + return false + } if isRetryableNetworkError(retryErr) { transientBackoff = nextTransientBackoff(transientBackoff) transferStatus.mu.Lock() @@ -435,7 +504,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, return true }) if err != nil { - return "", err + return "", missingGate.wrap(err) } return fileId, nil @@ -486,6 +555,48 @@ func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) e return nil } +func (fs *FilerSink) newMissingSourceChunkGate(fileId string) *missingSourceChunkGate { + return &missingSourceChunkGate{sink: fs, volumeId: filer.VolumeId(fileId)} +} + +// sourceVolumeMissingSince returns when the sink first found volumeId +// unlocatable, recording now on the first sighting. +func (fs *FilerSink) sourceVolumeMissingSince(volumeId string) time.Time { + since, _ := fs.missingVolumes.LoadOrStore(volumeId, 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. +func (fs *FilerSink) sourceServed(fileId string) { + fs.lastServedFileId.Store(&fileId) + fs.missingVolumes.Delete(filer.VolumeId(fileId)) +} + +// sourceStillServesChunks reports whether the source cluster can still produce the +// last chunk it served. A volume with no locations reads the same whether it was +// vacuumed away or every replica is down, so before an entry is written off the +// sink re-reads a file id the source did serve: if that one cannot be produced +// either the source is in an outage, and skipping would drop live files wholesale. +// It is a read and not a lookup because a lookup only proves the master still has +// the topology, not that a volume server answers. +func (fs *FilerSink) sourceStillServesChunks() bool { + if fs.filerSource == nil { + return false + } + probe := fs.lastServedFileId.Load() + if probe == nil { + return false + } + _, _, resp, err := fs.filerSource.ReadPart(*probe, 0) + if err != nil { + glog.V(0).Infof("source cannot serve %s either, so it is not only the one chunk: %v", *probe, err) + return false + } + util_http.CloseResponse(resp) + return true +} + // hasSourceNewerVersion reports whether the source's current entry for targetPath // has moved past sourceMtimeNs — gone, or a strictly-newer mtime — meaning the // version being replayed is stale. The lookup runs regardless of sourceMtimeNs so diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index b4b357d44..72e50d0c9 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -1,17 +1,23 @@ package filersink import ( + "context" "errors" "fmt" "io" + "net" "net/http" "net/http/httptest" "os" + "slices" "strings" "sync/atomic" "testing" "time" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/replication/source" @@ -390,7 +396,7 @@ func TestSourceSupersedesEpochMtime(t *testing.T) { // 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") + gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) resolveErr := errors.New("LookupFileId volume id 3: not found") for i := 1; i < maxUnverifiableResolveAttempts; i++ { if !gate(resolveErr) { @@ -408,7 +414,7 @@ func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) { // 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") + gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc")) permanentErrs := []error{ errors.New("fail to unmarshal manifest 3,01abc: proto: cannot parse invalid wire-format data"), errors.New("invalid fileId abc"), @@ -425,3 +431,245 @@ func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) { t.Error("isTransientResolveError(nil) must be false") } } + +// sourceFilerServer answers as a source filer that still holds the entry, +// unchanged, while its master locates the chunk's volume only for the volume ids +// in resolvable, which it serves from volumeUrl. With none listed it is a cluster +// that has vacuumed the volume away — or lost every replica of it. +type sourceFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + mtime int64 + volumeUrl string + resolvable []string +} + +func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + locationsMap := make(map[string]*filer_pb.Locations) + for _, vid := range req.VolumeIds { + if slices.Contains(s.resolvable, vid) { + locationsMap[vid] = &filer_pb.Locations{ + Locations: []*filer_pb.Location{{Url: s.volumeUrl}}, + } + } + } + return &filer_pb.LookupVolumeResponse{LocationsMap: locationsMap}, nil +} + +func (s *sourceFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{Entry: &filer_pb.Entry{ + Name: req.Name, + Attributes: &filer_pb.FuseAttributes{Mtime: s.mtime}, + }}, nil +} + +// liveVolume serves a chunk read the way a healthy volume server would, so the +// sink's probe finds the source still producing data. +func liveVolume(t *testing.T) string { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte("chunk bytes")) + })) + t.Cleanup(server.Close) + return strings.TrimPrefix(server.URL, "http://") +} + +func startSourceFiler(t *testing.T, mtime int64, 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{mtime: mtime, volumeUrl: volumeUrl, resolvable: resolvable}) + 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 +// ahead of it, and filer.sync never checkpointed again. +func TestFetchAndWriteStopsOnMissingSourceChunk(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 + }) + + filerSrc := startSourceFiler(t, 5, "") + + 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, errSourceChunkMissing) { + t.Fatalf("expected errSourceChunkMissing, got %v", fetchErr) + } + if !errors.Is(fetchErr, source.ErrVolumeNotFound) { + t.Fatalf("expected the underlying lookup failure to survive, got %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("fetchAndWrite is still retrying a chunk the source cannot produce") + } +} + +// The gate waits out the shapes a restarting volume server produces, and only +// gives up once the source has been saying "gone" for the whole grace period. +// The wait belongs to the volume: a vacuumed one takes every file it held with +// it, and waiting it out per file would stall the sync for as long as it holds +// files. +func TestMissingSourceChunkGate(t *testing.T) { + volumeGone := fmt.Errorf("read part 5617,01abc: %w", source.ErrVolumeNotFound) + needleGone := fmt.Errorf("read part 5617,01abc: 404 Not Found: %w", util_http.ErrNotFound) + + fs := &FilerSink{} + gate := fs.newMissingSourceChunkGate("5617,01abc") + for _, err := range []error{volumeGone, needleGone} { + if gate.isPermanent(err) { + t.Fatalf("must keep retrying inside the grace period: %v", err) + } + } + if got := gate.wrap(volumeGone); errors.Is(got, errSourceChunkMissing) { + t.Fatalf("must not mark permanent while still retrying: %v", got) + } + + // an unrelated failure is not the source answering "gone": it must not start a wait + other := fs.newMissingSourceChunkGate("5618,01abc") + other.isPermanent(errors.New("connection reset by peer")) + if _, found := fs.missingVolumes.Load("5618"); found { + t.Fatal("a non-missing error must not start the volume's wait") + } + + // a second file in the same volume inherits that wait instead of restarting it + fs.missingVolumes.Store("5617", time.Now().Add(-2*missingSourceChunkGrace)) + second := fs.newMissingSourceChunkGate("5617,02def") + if !second.isPermanent(volumeGone) { + t.Fatal("a later file must inherit the wait its volume already served") + } + wrapped := second.wrap(volumeGone) + if !errors.Is(wrapped, errSourceChunkMissing) || !errors.Is(wrapped, source.ErrVolumeNotFound) { + t.Fatalf("wrapped error lost a sentinel: %v", wrapped) + } + if second.wrap(nil) != nil { + t.Fatal("a successful retry must not be turned into an error") + } + + // the volume answering again clears the wait it had served + fs.sourceServed("5617,09fff") + if _, found := fs.missingVolumes.Load("5617"); found { + t.Fatal("a served chunk must clear its volume's wait") + } +} + +// 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 +// demonstrably still serving other chunks, since a cluster that cannot locate +// anything is having an outage and skipping would drop live files wholesale. +func TestOnReplicateChunkErrorMissingSourceChunk(t *testing.T) { + const probe = "5616,01abc" + liveEntry := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{Mtime: 5}} + missing := fmt.Errorf("copy 5617,02def: %w: %w", errSourceChunkMissing, source.ErrVolumeNotFound) + + t.Run("source still serving", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFiler(t, 5, liveVolume(t), "5616"), dir: "/dst"} + served := probe + fs.lastServedFileId.Store(&served) + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, missing); err != nil { + t.Fatalf("expected the entry to be skipped, got %v", err) + } + }) + + t.Run("source locating nothing", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFiler(t, 5, ""), dir: "/dst"} + served := probe + fs.lastServedFileId.Store(&served) + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, missing); !errors.Is(err, errSourceChunkMissing) { + t.Fatalf("expected the error to be held for a retry, got %v", err) + } + }) + + t.Run("probe locates but cannot be read", func(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() + fs := &FilerSink{ + filerSource: startSourceFiler(t, 5, strings.TrimPrefix(gone.URL, "http://"), "5616"), + dir: "/dst", + } + served := probe + fs.lastServedFileId.Store(&served) + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, missing); !errors.Is(err, errSourceChunkMissing) { + t.Fatalf("a volume the master lists but no server answers must not license a skip, got %v", err) + } + }) + + t.Run("nothing served yet", func(t *testing.T) { + fs := &FilerSink{filerSource: startSourceFiler(t, 5, liveVolume(t), "5616"), dir: "/dst"} + + if err := fs.onReplicateChunkError("/dst/x.bin", liveEntry, missing); !errors.Is(err, errSourceChunkMissing) { + t.Fatalf("expected the error to be held with no probe to check, got %v", err) + } + }) +} + +// An incremental sink's dated target keys cannot be mapped back to a source path, +// so nothing here can tell a vacuumed chunk from a superseded one. Waiting out the +// grace period would stall every such entry for half an hour; propagate instead +// and let filer.backup decide with the event's real source key. +func TestFetchAndWriteMissingSourceChunkUnverifiableSupersession(t *testing.T) { + prevRetryWaitTime := util.RetryWaitTime + util.RetryWaitTime = 100 * time.Millisecond + t.Cleanup(func() { util.RetryWaitTime = prevRetryWaitTime }) + + fs := &FilerSink{ + filerSource: startSourceFiler(t, 5, ""), + dir: "/backup", + isIncremental: true, + 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}, + "/backup/2026-07-10/x.bin", 5*int64(time.Second)) + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if !errors.Is(fetchErr, source.ErrVolumeNotFound) { + t.Fatalf("expected the lookup failure to propagate, got %v", fetchErr) + } + if errors.Is(fetchErr, errSourceChunkMissing) { + t.Fatalf("must not write the chunk off without checking supersession: %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("fetchAndWrite waited out the grace period with supersession unverifiable") + } +} diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 0eaca800e..4552312be 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -8,6 +8,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -61,6 +62,10 @@ type FilerSink struct { signature int32 activeTransfers sync.Map // chunkFileId -> *ChunkTransferStatus uploader *operation.Uploader + // lastServedFileId is the most recent chunk the source did serve, the probe + // sourceStillServesChunks re-checks before writing an entry off. + lastServedFileId atomic.Pointer[string] + missingVolumes sync.Map // source volume id -> time.Time first found unlocatable } func init() { @@ -447,15 +452,21 @@ func (fs *FilerSink) onCorruptChunk(key string, entry *filer_pb.Entry, err error } // 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. +// Skip (return nil) 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, or when the source has lost the chunk for good: it can never +// serve those bytes to anyone, and holding the offset for it only stops every +// 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)) { glog.Warningf("skip stale entry %s, source superseded during replicate: %v", key, err) return nil } + if errors.Is(err, errSourceChunkMissing) && 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) } diff --git a/weed/replication/source/filer_source.go b/weed/replication/source/filer_source.go index babbb6c71..a40e2a462 100644 --- a/weed/replication/source/filer_source.go +++ b/weed/replication/source/filer_source.go @@ -2,6 +2,7 @@ package source import ( "context" + "errors" "fmt" "net/http" "strings" @@ -18,6 +19,12 @@ import ( util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client" ) +// ErrVolumeNotFound reports that the source cluster has no location for a +// chunk's volume: vacuumed away, deleted, or every replica offline. Callers +// need it apart from a lookup that simply failed, which a later attempt can +// still get past. +var ErrVolumeNotFound = errors.New("volume not found") + type FilerSource struct { grpcAddress string grpcDialOption grpc.DialOption @@ -88,8 +95,8 @@ func (fs *FilerSource) LookupFileId(ctx context.Context, part string) (fileUrls locations := vid2Locations[vid] if locations == nil || len(locations.Locations) == 0 { - glog.V(1).InfofCtx(ctx, "LookupFileId locate volume id %s: %v", vid, err) - return nil, fmt.Errorf("LookupFileId locate volume id %s: %v", vid, err) + glog.V(1).InfofCtx(ctx, "LookupFileId locate volume id %s: %v", vid, ErrVolumeNotFound) + return nil, fmt.Errorf("LookupFileId locate volume id %s: %w", vid, ErrVolumeNotFound) } if !fs.proxyByFiler { @@ -118,12 +125,16 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h } if fs.proxyByFiler { - filename, header, resp, err = downloadFn("http://"+fs.address+"/?proxyChunkId="+fileId, "", offset) + fileUrl := "http://" + fs.address + "/?proxyChunkId=" + fileId + filename, header, resp, err = downloadFn(fileUrl, "", offset) + if err == nil { + err = readPartStatusError(fileUrl, resp) + } if err != nil { glog.V(0).Infof("read part %s via filer proxy %s offset %d: %v", fileId, fs.address, offset, err) - } else { - glog.V(4).Infof("read part %s via filer proxy %s offset %d content-length:%s", fileId, fs.address, offset, header.Get("Content-Length")) + return "", nil, nil, err } + glog.V(4).Infof("read part %s via filer proxy %s offset %d content-length:%s", fileId, fs.address, offset, header.Get("Content-Length")) return } @@ -134,17 +145,36 @@ func (fs *FilerSource) ReadPart(fileId string, offset int64) (filename string, h for _, fileUrl := range fileUrls { filename, header, resp, err = downloadFn(fileUrl, "", offset) - if err != nil { - glog.V(0).Infof("fail to read part %s from %s offset %d: %v", fileId, fileUrl, offset, err) - } else { - 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 { + err = readPartStatusError(fileUrl, resp) } + if err != nil { + resp = nil + 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 } return filename, header, resp, err } +// readPartStatusError turns a failure status into an error and closes the +// response. Otherwise the caller copies the error page as chunk content and +// reports it as a short read; a needle vacuum has removed answers 404, which +// is the source having lost the data rather than corruption. +func readPartStatusError(fileUrl string, resp *http.Response) error { + if resp == nil || resp.StatusCode < http.StatusBadRequest { + return nil + } + defer util_http.CloseResponse(resp) + if resp.StatusCode == http.StatusNotFound { + return fmt.Errorf("%s: %s: %w", fileUrl, resp.Status, util_http.ErrNotFound) + } + return fmt.Errorf("%s: %s", fileUrl, resp.Status) +} + var _ = filer_pb.FilerClient(&FilerSource{}) func (fs *FilerSource) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { diff --git a/weed/replication/source/filer_source_test.go b/weed/replication/source/filer_source_test.go index 72ac122aa..dd5d2215b 100644 --- a/weed/replication/source/filer_source_test.go +++ b/weed/replication/source/filer_source_test.go @@ -3,14 +3,22 @@ package source import ( "bytes" "compress/gzip" + "context" + "errors" "fmt" "io" + "net" "net/http" "net/http/httptest" "os" + "strings" "sync/atomic" "testing" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) @@ -282,3 +290,106 @@ func TestDownloadFile_GzipPartialReadThenResume(t *testing.T) { t.Fatalf("combined data mismatch: got %d bytes, want %d", len(fullData), len(testData)) } } + +// lookupFilerServer answers volume lookups with a fixed set of locations, so a +// test can hand ReadPart several replicas, or none at all. +type lookupFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + locations []string +} + +func (s *lookupFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + locationsMap := make(map[string]*filer_pb.Locations) + for _, vid := range req.VolumeIds { + if len(s.locations) == 0 { + continue + } + locations := &filer_pb.Locations{} + for _, url := range s.locations { + locations.Locations = append(locations.Locations, &filer_pb.Location{Url: url}) + } + locationsMap[vid] = locations + } + return &filer_pb.LookupVolumeResponse{LocationsMap: locationsMap}, nil +} + +func startLookupFiler(t *testing.T, locations ...string) *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, &lookupFilerServer{locations: locations}) + go server.Serve(listener) + t.Cleanup(server.Stop) + + address := listener.Addr().String() + filerSource := &FilerSource{} + if err := filerSource.DoInitialize(address, address, "/", false); err != nil { + t.Fatalf("DoInitialize: %v", err) + } + filerSource.SetGrpcDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())) + return filerSource +} + +// A volume the source cluster no longer has must be reported as ErrVolumeNotFound, +// not as an untyped message the caller can only string-match. +func TestLookupFileIdVolumeNotFound(t *testing.T) { + filerSource := startLookupFiler(t) + + _, err := filerSource.LookupFileId(context.Background(), "5617,01abc") + if !errors.Is(err, ErrVolumeNotFound) { + t.Fatalf("expected ErrVolumeNotFound, got %v", err) + } +} + +// A replica answering 404 is a failed read, not an empty file: ReadPart must move +// on to the next replica instead of handing the error page back as content. +func TestReadPartSkipsNotFoundReplica(t *testing.T) { + const body = "chunk bytes" + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Not Found", http.StatusNotFound) + })) + defer gone.Close() + live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(body)) + })) + defer live.Close() + + filerSource := startLookupFiler(t, + strings.TrimPrefix(gone.URL, "http://"), strings.TrimPrefix(live.URL, "http://")) + + _, _, resp, err := filerSource.ReadPart("5617,01abc", 0) + if err != nil { + t.Fatalf("ReadPart: %v", err) + } + defer util_http.CloseResponse(resp) + data, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if string(data) != body { + t.Fatalf("got %q, want %q", data, body) + } +} + +// 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) { + http.Error(w, "Not Found", http.StatusNotFound) + })) + defer gone.Close() + + filerSource := startLookupFiler(t, strings.TrimPrefix(gone.URL, "http://")) + + _, _, resp, err := filerSource.ReadPart("5617,01abc", 0) + if !errors.Is(err, util_http.ErrNotFound) { + t.Fatalf("expected ErrNotFound, got %v", err) + } + if resp != nil { + t.Fatal("expected no response alongside the error") + } +}