filer.sync: skip superseded replays before fetching any chunks (#12212)

* filer.sync: skip superseded replays before fetching any chunks

On a directory under heavy overwrite, most entries the sync replays are
already superseded on the source. The sink only learned that after chunk
reads failed — or worse, after fetching and uploading bytes a later event
replaces — so the backlog drained slowly and the log filled with 404s.

CreateEntry and UpdateEntry now check sourceMovedOn before touching a
chunk: one entry lookup skips a stale replay entirely. A mid-replication
supersession is still caught by the existing retry gates.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: bound the source liveness probes with a timeout

The supersession checks added a source entry lookup to the replication
path, and every lookup ran on context.Background() with no deadline. A
source that accepts the call but whose metadata store stalls would hold
the sync forever — before a single chunk fetch, even when every chunk is
still readable. Bound the probe at 30s; an unanswered lookup proves
nothing, so replication proceeds and the existing retry gates decide.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: sync the live version a superseded replay points at

Skipping a stale replay assumed a later event always re-carries the
current content. It does not when the superseding write was itself
replicated in from another cluster: those events are dropped before they
reach a filer sink, so the destination could be left missing the file.
When the source reports a live entry newer than the replay, commit that
version through the normal path instead — equivalent to applying the
later event, with no dependence on its delivery. A gone source entry
still skips, since the destination already matches the source.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: reconcile moved entry content after a native rename

AtomicRenameEntry relocates whatever version the destination held. If the
preceding update replay was skipped because the source had already moved
to the new name, the moved entry keeps the old bytes and — with no
further writes — stays stale forever. MoveEntry now runs the same
reconciliation as an in-place update against the rename's snapshot, so
the moved entry picks up pending content changes.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: repair from the live snapshot a superseded replay resolves to

An incremental diff against the replayed oldEntry can resolve manifests the
overwrite already vacuumed, which fails every retry of the event.

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/command/filer_sync.go

@@ -768,7 +768,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 						oldKey := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
 						if mover, ok := dataSink.(sink.EntryMover); ok {
 							// native atomic move: no re-copy, no descendant gap, no chunk leak.
-							return mover.MoveEntry(oldKey, newKey, message.NewEntry, message.Signatures)
+							return mover.MoveEntry(oldKey, newKey, message.OldEntry, message.NewEntry, message.Signatures)
 						}
 						// no native move: create the new entry first, then delete the
 						// old, so a crash between the two leaves the entry visible.

weed/command/filer_sync_process_test.go

@@ -91,7 +91,7 @@ type movingSyncSink struct {
 
 var _ sink.EntryMover = (*movingSyncSink)(nil)
 
-func (s *movingSyncSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+func (s *movingSyncSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error {
 	s.moveOldKeys = append(s.moveOldKeys, oldKey)
 	s.moveNewKeys = append(s.moveNewKeys, newKey)
 	s.ordered = append(s.ordered, "move")

weed/replication/replicator.go

@@ -116,7 +116,7 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
 		// A real move: the path changed. UpdateEntry cannot move an entry.
 		if mover, ok := r.sink.(sink.EntryMover); ok {
 			glog.V(4).Infof("moving %v => %v", oldSinkKey, newSinkKey)
-			return mover.MoveEntry(oldSinkKey, newSinkKey, newEntry, message.Signatures)
+			return mover.MoveEntry(oldSinkKey, newSinkKey, oldEntry, newEntry, message.Signatures)
 		}
 		// Sinks without a native move: create at the new key first, then delete the
 		// old, so a crash between the two leaves the entry visible under both names

weed/replication/replicator_test.go

@@ -96,7 +96,7 @@ type movingSink struct {
 
 var _ sink.EntryMover = (*movingSink)(nil)
 
-func (s *movingSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+func (s *movingSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error {
 	s.moveCalls = append(s.moveCalls, moveCall{oldKey: oldKey, newKey: newKey})
 	s.ordered = append(s.ordered, "move")
 	return nil

weed/replication/sink/filersink/fetch_write.go

@@ -35,6 +35,20 @@ import (
 // still exists is registered again.
 var missingSourceChunkGrace = 30 * time.Minute
 
+// sourceLookupTimeout bounds a best-effort source entry probe: a source that
+// accepts the call but whose metadata store never answers must stall
+// replication only this long, not forever. A variable so tests can shorten it.
+var sourceLookupTimeout = 30 * time.Second
+
+// sourceEntry returns the source's live entry at sourcePath within
+// sourceLookupTimeout.
+func (fs *FilerSink) sourceEntry(sourcePath util.FullPath) (*filer_pb.Entry, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout)
+	defer cancel()
+	entry, _, _, err := filer_pb.GetEntry(ctx, fs.filerSource, sourcePath)
+	return entry, err
+}
+
 // 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")
@@ -648,7 +662,9 @@ func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int6
 	if !ok {
 		return false
 	}
-	return SourceSupersedes(context.Background(), fs.filerSource, sourcePath, sourceMtimeNs)
+	ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout)
+	defer cancel()
+	return SourceSupersedes(ctx, fs.filerSource, sourcePath, sourceMtimeNs)
 }
 
 // sourceDroppedChunk reports whether the live source entry still exists but no
@@ -661,7 +677,7 @@ func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, source
 	if !ok {
 		return false
 	}
-	live, _, _, err := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath)
+	live, err := fs.sourceEntry(sourcePath)
 	if err != nil || live == nil || sourceMtimeNs <= 0 || getEntryMtimeNs(live) < sourceMtimeNs {
 		return false
 	}

weed/replication/sink/filersink/fetch_write_test.go

@@ -438,13 +438,19 @@ func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) {
 // that has vacuumed the volume away — or lost every replica of it.
 type sourceFilerServer struct {
 	filer_pb.UnimplementedSeaweedFilerServer
-	mtime      int64
-	volumeUrl  string
-	resolvable []string
-	entry      *filer_pb.Entry
+	mtime         int64
+	volumeUrl     string
+	resolvable    []string
+	entry         *filer_pb.Entry
+	lookupDelay   time.Duration
+	lookupFn      func(dir, name string) (*filer_pb.Entry, error)
+	volumeLookups *atomic.Int32
 }
 
 func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
+	if s.volumeLookups != nil {
+		s.volumeLookups.Add(int32(len(req.VolumeIds)))
+	}
 	locationsMap := make(map[string]*filer_pb.Locations)
 	for _, vid := range req.VolumeIds {
 		if slices.Contains(s.resolvable, vid) {
@@ -457,6 +463,17 @@ 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.lookupDelay > 0 {
+		select {
+		case <-time.After(s.lookupDelay):
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+	if s.lookupFn != nil {
+		entry, err := s.lookupFn(req.Directory, req.Name)
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, err
+	}
 	if s.entry != nil {
 		return &filer_pb.LookupDirectoryEntryResponse{Entry: s.entry}, nil
 	}
@@ -498,13 +515,18 @@ func startSourceFiler(t *testing.T, mtime int64, volumeUrl string, resolvable ..
 }
 
 func startSourceFilerWithEntry(t *testing.T, entry *filer_pb.Entry, volumeUrl string, resolvable ...string) *source.FilerSource {
+	t.Helper()
+	return startSourceFilerServer(t, &sourceFilerServer{volumeUrl: volumeUrl, resolvable: resolvable, entry: entry})
+}
+
+func startSourceFilerServer(t *testing.T, impl *sourceFilerServer) *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})
+	filer_pb.RegisterSeaweedFilerServer(server, impl)
 	go server.Serve(listener)
 	t.Cleanup(server.Stop)
 
@@ -517,6 +539,56 @@ func startSourceFilerWithEntry(t *testing.T, entry *filer_pb.Entry, volumeUrl st
 	return filerSrc
 }
 
+// recordingFilerServer is a destination filer that counts CreateEntry and
+// UpdateEntry calls and answers entry lookups from a fixed entry.
+type recordingFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	existing   *filer_pb.Entry
+	creates    atomic.Int32
+	updates    atomic.Int32
+	renames    atomic.Int32
+	lastCreate atomic.Pointer[filer_pb.Entry]
+	lastUpdate atomic.Pointer[filer_pb.Entry]
+}
+
+func (s *recordingFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if s.existing == nil {
+		return &filer_pb.LookupDirectoryEntryResponse{}, filer_pb.ErrNotFound
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: s.existing}, nil
+}
+
+func (s *recordingFilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
+	s.creates.Add(1)
+	s.lastCreate.Store(req.Entry)
+	return &filer_pb.CreateEntryResponse{}, nil
+}
+
+func (s *recordingFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
+	s.updates.Add(1)
+	s.lastUpdate.Store(req.Entry)
+	return &filer_pb.UpdateEntryResponse{}, nil
+}
+
+func (s *recordingFilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) {
+	s.renames.Add(1)
+	return &filer_pb.AtomicRenameEntryResponse{}, nil
+}
+
+func startRecordingFiler(t *testing.T, existing *filer_pb.Entry) (*recordingFilerServer, string) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	impl := &recordingFilerServer{existing: existing}
+	server := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(server, impl)
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+	return impl, listener.Addr().String()
+}
+
 // 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
@@ -793,6 +865,30 @@ func TestSourceMovedOn(t *testing.T) {
 	}
 }
 
+// A source that accepts the lookup but never answers must not hang the
+// supersession probe: the bounded context gives up and replication continues.
+func TestSourceMovedOnStalledLookupContinues(t *testing.T) {
+	prev := sourceLookupTimeout
+	sourceLookupTimeout = 200 * time.Millisecond
+	t.Cleanup(func() { sourceLookupTimeout = prev })
+
+	replayed := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	fs := &FilerSink{
+		filerSource: startSourceFilerServer(t, &sourceFilerServer{lookupDelay: 30 * time.Second}),
+		dir:         "/dst",
+	}
+	start := time.Now()
+	if fs.sourceMovedOn("/dst/x.bin", replayed) {
+		t.Fatal("an unanswered lookup cannot prove supersession")
+	}
+	if time.Since(start) > 10*time.Second {
+		t.Fatal("the supersession probe must give up at sourceLookupTimeout")
+	}
+}
+
 // 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.
@@ -886,3 +982,279 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) {
 		t.Fatal("a still-listed manifest's missing child must not retry past the gate")
 	}
 }
+
+// A replayed version whose source entry is gone must not cost a single chunk
+// read or destination write: on a busy directory every skipped fetch is one
+// that would have 404'd or, worse, uploaded bytes a later event replaces.
+func TestCreateEntrySkipsSupersededVersion(t *testing.T) {
+	var volumeLookups atomic.Int32
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{
+		lookupFn: func(dir, name string) (*filer_pb.Entry, error) {
+			return nil, filer_pb.ErrNotFound
+		},
+		volumeLookups: &volumeLookups,
+	})
+	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))
+
+	replayed := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err != nil {
+		t.Fatalf("superseded replay should be skipped, not fail: %v", err)
+	}
+	if dst.creates.Load() != 0 {
+		t.Fatal("a superseded version must not be committed to the destination")
+	}
+	if volumeLookups.Load() != 0 {
+		t.Fatal("a superseded version must not fetch any chunks")
+	}
+}
+
+// The same skip must not fire while the live source still matches the replayed
+// version: replication proceeds and the chunk fetch is attempted.
+func TestCreateEntryReplicatesCurrentVersion(t *testing.T) {
+	var volumeLookups atomic.Int32
+	replayed := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: replayed, volumeLookups: &volumeLookups})
+	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))
+
+	// 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 })
+
+	if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err == nil {
+		t.Fatal("replication of a chunk the source cannot serve must fail")
+	}
+	if volumeLookups.Load() == 0 {
+		t.Fatal("a version the source still serves must be replicated, not skipped")
+	}
+	if dst.creates.Load() != 0 {
+		t.Fatal("no destination write is expected while chunks cannot be fetched")
+	}
+}
+
+// UpdateEntry takes the same pre-flight skip: a replayed version whose source
+// entry is gone must not re-replicate or update the destination.
+func TestUpdateEntrySkipsSupersededVersion(t *testing.T) {
+	var volumeLookups atomic.Int32
+	existing := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}},
+	}
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{
+		lookupFn: func(dir, name string) (*filer_pb.Entry, error) {
+			return nil, filer_pb.ErrNotFound
+		},
+		volumeLookups: &volumeLookups,
+	})
+	dst, dstAddr := startRecordingFiler(t, existing)
+
+	fs := &FilerSink{
+		filerSource:    filerSrc,
+		dir:            "/dst",
+		grpcAddress:    dstAddr,
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		executor:       util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	replayed := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	found, err := fs.UpdateEntry("/dst/x.bin", existing, "", replayed, true, nil)
+	if err != nil {
+		t.Fatalf("superseded replay should be skipped, not fail: %v", err)
+	}
+	if !found {
+		t.Fatal("the existing entry was looked up; foundExistingEntry must report it")
+	}
+	if dst.updates.Load() != 0 {
+		t.Fatal("a superseded version must not be committed to the destination")
+	}
+	if volumeLookups.Load() != 0 {
+		t.Fatal("a superseded version must not fetch any chunks")
+	}
+}
+
+// A superseded replay is replaced by the live version the source reports: the
+// superseding event may never be delivered to this sink, so the live content
+// must be committed now — still without fetching the replayed chunks.
+func TestCreateEntrySyncsLiveVersion(t *testing.T) {
+	var volumeLookups atomic.Int32
+	live := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Content:    []byte("live"),
+	}
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live, volumeLookups: &volumeLookups})
+	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))
+
+	replayed := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err != nil {
+		t.Fatalf("syncing the live version should succeed: %v", err)
+	}
+	if dst.creates.Load() != 1 {
+		t.Fatal("the live version must be committed to the destination")
+	}
+	created := dst.lastCreate.Load()
+	if created == nil || string(created.GetContent()) != "live" || created.GetAttributes().GetMtime() != 6 {
+		t.Fatalf("the destination must get the live version, got %+v", created)
+	}
+	if volumeLookups.Load() != 0 {
+		t.Fatal("the superseded version's chunks must not be fetched")
+	}
+}
+
+// UpdateEntry applies the same substitution: a superseded replay commits the
+// live source version through the normal update path.
+func TestUpdateEntrySyncsLiveVersion(t *testing.T) {
+	existing := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "4,00aaa", SourceFileId: "5617,00aaa", Size: 10}},
+	}
+	oldEntry := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}},
+	}
+	live := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Content:    []byte("live"),
+	}
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live})
+	dst, dstAddr := startRecordingFiler(t, existing)
+
+	fs := &FilerSink{
+		filerSource:    filerSrc,
+		dir:            "/dst",
+		grpcAddress:    dstAddr,
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		executor:       util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	replayed := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+	}
+	found, err := fs.UpdateEntry("/dst/x.bin", oldEntry, "", replayed, true, nil)
+	if err != nil {
+		t.Fatalf("syncing the live version should succeed: %v", err)
+	}
+	if !found {
+		t.Fatal("the existing entry was looked up; foundExistingEntry must report it")
+	}
+	if dst.updates.Load() != 1 {
+		t.Fatal("the live version must be committed to the destination")
+	}
+	updated := dst.lastUpdate.Load()
+	if updated == nil || string(updated.GetContent()) != "live" || updated.GetAttributes().GetMtime() != 6 {
+		t.Fatalf("the destination must get the live version, got %+v", updated)
+	}
+}
+
+// A rename carries whatever version the destination happened to hold. When the
+// preceding content update was skipped — the source had already moved to the
+// new name — the moved entry keeps the old bytes. MoveEntry must reconcile the
+// moved entry with the event's snapshot, like an in-place update.
+func TestMoveEntryReconcilesMovedContent(t *testing.T) {
+	existing := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "4,0v1d", SourceFileId: "5617,0v1", Size: 10}},
+	}
+	dst, dstAddr := startRecordingFiler(t, existing)
+
+	live := &filer_pb.Entry{
+		Name:       "y.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Content:    []byte("v2"),
+	}
+	filerSrc := startSourceFilerServer(t, &sourceFilerServer{
+		lookupFn: func(dir, name string) (*filer_pb.Entry, error) {
+			if name == "y.bin" {
+				return live, nil
+			}
+			return nil, filer_pb.ErrNotFound
+		},
+	})
+
+	fs := &FilerSink{
+		filerSource:    filerSrc,
+		dir:            "/dst",
+		grpcAddress:    dstAddr,
+		grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+		executor:       util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	oldEntry := &filer_pb.Entry{
+		Name:       "x.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Content:    []byte("v2"),
+	}
+	newEntry := &filer_pb.Entry{
+		Name:       "y.bin",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Content:    []byte("v2"),
+	}
+	if err := fs.MoveEntry("/dst/x.bin", "/dst/y.bin", oldEntry, newEntry, nil); err != nil {
+		t.Fatalf("move with reconcile should succeed: %v", err)
+	}
+	if dst.renames.Load() != 1 {
+		t.Fatal("the entry must be relocated with the native rename")
+	}
+	if dst.updates.Load() != 1 {
+		t.Fatal("the moved entry must be reconciled with the event's snapshot")
+	}
+	updated := dst.lastUpdate.Load()
+	if updated == nil || string(updated.GetContent()) != "v2" || updated.GetAttributes().GetMtime() != 6 {
+		t.Fatalf("the moved entry must carry the rename's content, got %+v", updated)
+	}
+}

weed/replication/sink/filersink/filer_sink.go

@@ -223,7 +223,7 @@ var _ sink.EntryMover = (*FilerSink)(nil)
 // is re-checked with a direct lookup rather than inferred from the rename error,
 // so a rolled-back move that left the old entry intact propagates for retry
 // instead of being mistaken for "gone".
-func (fs *FilerSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+func (fs *FilerSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error {
 	oldDir, oldName := util.FullPath(oldKey).DirAndName()
 	newDir, newName := util.FullPath(newKey).DirAndName()
 
@@ -238,7 +238,15 @@ func (fs *FilerSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry,
 		return err
 	})
 	if err == nil {
-		return nil
+		// The rename relocates whatever version the destination held, which can
+		// predate the event's snapshot — e.g. a queued update skipped after the
+		// source already moved to the new name. Reconcile content the same way
+		// an in-place update does.
+		if oldEntry == nil {
+			return nil
+		}
+		_, uerr := fs.UpdateEntry(newKey, oldEntry, newDir, newEntry, true, signatures)
+		return uerr
 	}
 	if missing, lookupErr := fs.entryMissing(oldKey); lookupErr == nil && missing {
 		glog.V(2).Infof("move %s => %s: old path gone, creating %s", oldKey, newKey, newKey)
@@ -294,6 +302,20 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 				key, len(resp.Entry.GetChunks()), len(entry.GetChunks()))
 		}
 
+		// A version the source already moved past is superseded by a later
+		// event, so fetching its chunks only feeds a write that is obsolete on
+		// arrival — skip before any read. When the source holds a newer live
+		// version, replicate it directly: the superseding event may never be
+		// delivered to this sink.
+		if live, movedOn := fs.sourceVersion(key, entry); movedOn {
+			if live == nil {
+				glog.Warningf("skip stale entry %s, source superseded before replicate", key)
+				return nil
+			}
+			glog.Warningf("entry %s superseded before replicate; syncing live version", key)
+			entry = live
+		}
+
 		replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "")
 
 		if err != nil {
@@ -366,7 +388,25 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 	// now, which for a rename is newParentPath/newEntry.Name, not the old key.
 	targetKey := updatedEntryKey(key, newParentPath, newEntry)
 
-	switch chooseUpdateAction(existingEntry, oldEntry, newEntry) {
+	action := chooseUpdateAction(existingEntry, oldEntry, newEntry)
+	if action != updateSkip {
+		if live, movedOn := fs.sourceVersion(targetKey, newEntry); movedOn {
+			if live == nil {
+				glog.Warningf("skip stale entry %s, source superseded before replicate", targetKey)
+				return true, nil
+			}
+			glog.Warningf("entry %s superseded before replicate; syncing live version", targetKey)
+			newEntry = live
+			// The replayed oldEntry predates the live version, so an
+			// incremental diff could resolve manifests the overwrite already
+			// vacuumed; repair from the live snapshot instead.
+			if action = chooseUpdateAction(existingEntry, oldEntry, newEntry); action == updateNormal {
+				action = updateRepair
+			}
+		}
+	}
+
+	switch action {
 	case updateSkip:
 		// a newer, complete version already landed (out-of-order); leave it
 		glog.V(2).Infof("late updates %s", key)
@@ -506,34 +546,43 @@ func (fs *FilerSink) onReplicateChunkError(key string, entry *filer_pb.Entry, er
 	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 {
+// sourceVersion returns the source's live entry for targetPath and whether it
+// moved past the replayed version: gone (live == nil), 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, and neither does a failed lookup.
+func (fs *FilerSink) sourceVersion(targetPath string, entry *filer_pb.Entry) (live *filer_pb.Entry, movedOn bool) {
 	sourcePath, ok := fs.targetPathToSourcePath(targetPath)
 	if !ok {
-		return false
+		return nil, false
 	}
 	replayedMtimeNs := getEntryMtimeNs(entry)
-	live, _, _, lookupErr := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath)
+	live, lookupErr := fs.sourceEntry(sourcePath)
 	if sourceSupersedes(sourcePath, live, lookupErr, replayedMtimeNs) {
-		return true
+		return live, true
 	}
 	if lookupErr != nil || live == nil || replayedMtimeNs <= 0 || getEntryMtimeNs(live) < replayedMtimeNs {
-		return false
+		return nil, false
 	}
 	liveIds := chunkFileIds(live.GetChunks())
 	replayedIds := chunkFileIds(entry.GetChunks())
 	if len(liveIds) != len(replayedIds) {
-		return true
+		return live, true
 	}
 	for id := range replayedIds {
 		if !liveIds[id] {
-			return true
+			return live, true
 		}
 	}
-	return false
+	return nil, false
+}
+
+// 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.
+func (fs *FilerSink) sourceMovedOn(targetPath string, entry *filer_pb.Entry) bool {
+	_, movedOn := fs.sourceVersion(targetPath, entry)
+	return movedOn
 }
 
 // updateAction decides how UpdateEntry reconciles an incoming source version with

weed/replication/sink/replication_sink.go

@@ -29,7 +29,7 @@ type ReplicationSink interface {
 // committed destination, a directory move never deletes descendants before they
 // are recreated, and the entry's chunks are neither re-copied nor leaked.
 type EntryMover interface {
-	MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error
+	MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error
 }
 
 // MetadataOnlyDeleter is an optional capability for sinks whose own delete