Let filer.sync move past a chunk the source cluster no longer has (#11019)

* Name the failure when the source cluster cannot locate a chunk's volume

LookupFileId formatted a nil err into the message it returned, so the only
thing a caller could do with "no locations for this volume" was match on the
text. Return a typed error instead.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Fail a source chunk read on a failure status instead of copying the error page

ReadPart never looked at the response status, so a volume server answering 404
for a needle vacuum had removed came back as a successful read whose body was
the error page. The caller counted those bytes as file content and reported a
size mismatch — a corruption claim about data the source had simply lost — and
a 404 from one replica ended the search instead of trying the next.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Stop retrying a chunk the source cluster can no longer produce

A chunk whose volume vacuum has removed fails the same way on every attempt, but
the retry loop had no way to say so and kept going forever. The sync job holding
it never finished, so it pinned the offset watermark at the event ahead of it and
filer.sync never checkpointed again — alive, quiet, and permanently behind.

Wait the source out for a grace period long enough to cover a volume server
restart or a master failover, then give up and mark the failure permanent.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Let replication continue past an entry whose source data is gone

An entry the source can no longer read holds the sync offset forever: the event
fails on every replay, so the checkpoint never moves past it and every later
event stays uncheckpointed, however long the sync keeps running. Nothing brings
those bytes back, so skip the entry with an error naming it and carry on.

Skip only while the source is demonstrably still serving other chunks. A volume
with no locations reads the same whether it was vacuumed away or every replica is
down, and during a cluster-wide outage that answer comes back for every chunk —
skipping then would drop live files wholesale.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Propagate a missing source chunk instead of waiting when supersession is unverifiable

An incremental sink's dated target keys cannot be mapped back to a source path,
so nothing here can tell a chunk the source lost from one a later version already
replaced. Waiting out the grace period would stall every vacuumed needle for half
an hour; hand the failure to the caller, which has the event's real source key.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Wait out a gone volume once, not once per file it held

A volume vacuum removed took every file it held with it, and each chunk was
timing its own grace period. With a bounded chunk executor those waits serialize,
so one gone volume holding many files stalls the sync for far longer than the
grace period — the wedge again, only slower.

Track the wait per source volume on the sink instead: the first chunk to find it
unlocatable starts the clock, every later chunk inherits it and gives up as soon
as it has run out, and a chunk the source does serve clears it.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

* Probe the source with a read, not a lookup, before writing an entry off

A lookup only proves the source master still has the topology. If every volume
server is unreachable while the master still lists them, the probe passed and the
sink wrote off an entry whose data was merely out of reach. Read the probe chunk
instead, and say in the log that the entry stays unreplicated.

Claude-Session: https://claude.ai/code/session_01SRPEP4jRu29FbLSN6bjLaK

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

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")
+	}
+}

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)
 }
 

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 {

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")
+	}
+}