replication: keep a local sink path the source recreated after a delete (#12581)

* replication: let the local sink check a delete against the live source

LocalSink removed the destination path for any delete it was handed. A
delete that reaches the sink after the source recreated the path (a peer
filer's event delivered late, or a historical event replayed) removed the
newer file, and no later event brings it back.

DeleteEntryVersion receives the deleted entry. Before removing an existing
destination, the local sink looks up the source path and keeps the
destination when the source holds an entry of the same kind committed
after the deleted version (commit_ts_ns, or mtime for unstamped entries).
A live entry that is the deleted version itself or older means the source
store lags the event, so the delete proceeds. A failed lookup returns the
error so the event is retried instead of deleting a file that may be live.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.sync, filer.replicate: hand the deleted version to the sink

Pure deletes, renames without a native move and moves out of the
watched directory now go through sink.DeleteEntryVersion with the old
entry, so a sink that checks the live source (the local sink) can keep a
path the source recreated after the deleted version. Other sinks fall
back to DeleteEntry unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* replication: order the local sink check by the delete's mutation time

A source filer can hold a version written concurrently with the delete:
newer than the deleted entry but older than the delete itself. The
source's own replay applies the delete over it, so keeping the backup
copy because it outranks the deleted entry left a file the source no
longer has.

The sink now gets the delete's mutation time (MutationTsNs, else TsNs)
and keeps the destination only when the live entry was committed after
it, the order Filer.Replay keeps. That also covers a stamped replacement
of an unstamped entry whose mtime did not move. filer.replicate events
carry no time, so they keep the deleted entry's stamp as the bound, and
mtime only when nothing is stamped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* replication: count an unstamped live entry from its mtime

A source filer that predates commit stamps stores entries with no
CommitTsNs, so comparing the live stamp with the delete's time never
kept a replacement there. An unstamped live entry now counts from its
mtime against the delete's time.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* replication: keep a stamped live entry when the deleted one is unstamped and the delete has no time

filer.replicate gets no delete time from the notification queue. When the
deleted entry has no commit stamp either, a stamped live entry was written
after it, so its mtime no longer decides.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/command/filer_sync.go

@@ -681,6 +681,10 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 		if message.NewEntry != nil {
 			sourceNewKey = util.FullPath(targetDir).Child(message.NewEntry.Name)
 		}
+		mutationTsNs := resp.MutationTsNs
+		if mutationTsNs == 0 {
+			mutationTsNs = resp.TsNs
+		}
 
 		if debug {
 			glog.V(0).Infof("received %s ts_ns=%d", resp.Directory, resp.TsNs)
@@ -733,7 +737,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 				return nil
 			}
 			key := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
-			return sink.DeleteEntry(dataSink, key, message.OldEntry.IsDirectory, message.DeleteChunks, filer.IsMetadataOnlyDelete(message.OldEntry), message.Signatures)
+			return sink.DeleteEntryVersion(dataSink, key, message.OldEntry, mutationTsNs, message.DeleteChunks, filer.IsMetadataOnlyDelete(message.OldEntry), message.Signatures)
 		}
 
 		// handle new entries
@@ -775,7 +779,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 						if err := dataSink.CreateEntry(newKey, message.NewEntry, message.Signatures); err != nil {
 							return fmt.Errorf("create entry2 : %w", err)
 						}
-						if err := dataSink.DeleteEntry(oldKey, message.OldEntry.IsDirectory, false, message.Signatures); err != nil {
+						if err := sink.DeleteEntryVersion(dataSink, oldKey, message.OldEntry, mutationTsNs, false, false, message.Signatures); err != nil {
 							return fmt.Errorf("delete old entry %v: %w", oldKey, err)
 						}
 						return nil
@@ -817,7 +821,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 				// new key is outside the watched directory
 				if doDeleteFiles {
 					key := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
-					return dataSink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks, message.Signatures)
+					return sink.DeleteEntryVersion(dataSink, key, message.OldEntry, mutationTsNs, message.DeleteChunks, false, message.Signatures)
 				}
 			}
 		} else {

weed/command/filer_sync_replayed_delete_test.go

@@ -0,0 +1,94 @@
+package command
+
+import (
+	"context"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/sink/localsink"
+	"github.com/seaweedfs/seaweedfs/weed/replication/source"
+)
+
+type liveEntriesFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	entries map[string]*filer_pb.Entry
+}
+
+func (s *liveEntriesFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if entry, ok := s.entries[req.Directory+"/"+req.Name]; ok {
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil
+	}
+	return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+}
+
+func TestGenProcessFunctionLocalSinkKeepsPathTheSourceRecreated(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(server, &liveEntriesFiler{entries: map[string]*filer_pb.Entry{
+		"/src/dir/recreated":  {Name: "recreated", CommitTsNs: 300, Attributes: &filer_pb.FuseAttributes{}},
+		"/src/dir/renamed":    {Name: "renamed", CommitTsNs: 300, Attributes: &filer_pb.FuseAttributes{}},
+		"/src/dir/peer":       {Name: "peer", CommitTsNs: 350, Attributes: &filer_pb.FuseAttributes{}},
+		"/src/dir/concurrent": {Name: "concurrent", CommitTsNs: 200, Attributes: &filer_pb.FuseAttributes{}},
+	}})
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+	filerSource := &source.FilerSource{}
+	if err := filerSource.DoInitialize(listener.Addr().String(), listener.Addr().String(), "/src", false); err != nil {
+		t.Fatal(err)
+	}
+	filerSource.SetGrpcDialOption(grpc.WithTransportCredentials(insecure.NewCredentials()))
+
+	target := filepath.Join(t.TempDir(), "backup")
+	dataSink := &localsink.LocalSink{Dir: target}
+	dataSink.SetSourceFiler(filerSource)
+	if err := os.MkdirAll(filepath.Join(target, "dir"), 0755); err != nil {
+		t.Fatal(err)
+	}
+	for _, name := range []string{"recreated", "renamed", "deleted", "peer", "concurrent"} {
+		if err := os.WriteFile(filepath.Join(target, "dir", name), []byte("newer"), 0644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	processFn := genProcessFunction("/src", target, nil, nil, nil, nil, dataSink, true, false)
+	stale := func(name string) *filer_pb.Entry {
+		return &filer_pb.Entry{Name: name, CommitTsNs: 100, Attributes: &filer_pb.FuseAttributes{FileMode: 0644}}
+	}
+
+	events := []*filer_pb.SubscribeMetadataResponse{
+		{TsNs: 200, EventNotification: &filer_pb.EventNotification{OldEntry: stale("recreated")}},
+		{TsNs: 200, EventNotification: &filer_pb.EventNotification{OldEntry: stale("deleted")}},
+		{TsNs: 200, EventNotification: &filer_pb.EventNotification{OldEntry: stale("renamed"), NewParentPath: "/src/dir", NewEntry: &filer_pb.Entry{
+			Name: "moved", Content: []byte("moved"), Attributes: &filer_pb.FuseAttributes{FileMode: 0644, FileSize: 5},
+		}}},
+		{TsNs: 400, MutationTsNs: 300, EventNotification: &filer_pb.EventNotification{OldEntry: stale("peer")}},
+		{TsNs: 300, EventNotification: &filer_pb.EventNotification{OldEntry: stale("concurrent")}},
+	}
+	for _, event := range events {
+		event.Directory = "/src/dir"
+		if err := processFn(event); err != nil {
+			t.Fatalf("process %s: %v", event.EventNotification.OldEntry.Name, err)
+		}
+	}
+
+	for _, name := range []string{"recreated", "renamed", "moved", "peer"} {
+		if _, err := os.Stat(filepath.Join(target, "dir", name)); err != nil {
+			t.Errorf("%s missing: %v", name, err)
+		}
+	}
+	for _, name := range []string{"deleted", "concurrent"} {
+		if _, err := os.Stat(filepath.Join(target, "dir", name)); !os.IsNotExist(err) {
+			t.Errorf("%s kept: %v", name, err)
+		}
+	}
+}

weed/replication/replicator.go

@@ -101,7 +101,7 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
 
 	if oldEntry != nil && newEntry == nil {
 		glog.V(4).Infof("deleting %v", oldSinkKey)
-		return sink.DeleteEntry(r.sink, oldSinkKey, oldEntry.IsDirectory, message.DeleteChunks, filer.IsMetadataOnlyDelete(oldEntry), message.Signatures)
+		return sink.DeleteEntryVersion(r.sink, oldSinkKey, oldEntry, 0, message.DeleteChunks, filer.IsMetadataOnlyDelete(oldEntry), message.Signatures)
 	}
 	if oldEntry == nil && newEntry != nil {
 		glog.V(4).Infof("creating %v", oldSinkKey)
@@ -125,7 +125,7 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
 		if err := r.sink.CreateEntry(newSinkKey, newEntry, message.Signatures); err != nil {
 			return fmt.Errorf("create renamed entry %v: %w", newSinkKey, err)
 		}
-		return r.sink.DeleteEntry(oldSinkKey, oldEntry.IsDirectory, false, message.Signatures)
+		return sink.DeleteEntryVersion(r.sink, oldSinkKey, oldEntry, 0, false, false, message.Signatures)
 	}
 
 	// oldSinkKey == newSinkKey: pure in-place update (same path, content/attrs changed).

weed/replication/replicator_test.go

@@ -444,3 +444,39 @@ func TestReplicateMetadataOnlyDeleteKeepsTheRemoteObject(t *testing.T) {
 		t.Fatalf("ordinary deletes = %+v, want /dest/dir/gone.bin", s.deleteCalls)
 	}
 }
+
+type versionedSink struct {
+	*recordingSink
+	versionCalls []*filer_pb.Entry
+}
+
+var _ sink.VersionedDeleter = (*versionedSink)(nil)
+
+func (s *versionedSink) DeleteEntryVersion(key string, oldEntry *filer_pb.Entry, tsNs int64, deleteIncludeChunks bool, signatures []int32) error {
+	s.versionCalls = append(s.versionCalls, oldEntry)
+	return s.DeleteEntry(key, oldEntry.IsDirectory, deleteIncludeChunks, signatures)
+}
+
+func TestReplicateDeletesHandTheDeletedVersionToTheSink(t *testing.T) {
+	s := &versionedSink{recordingSink: &recordingSink{name: "local", sinkToDirectory: "/dest"}}
+	r := &Replicator{sink: s, source: &source.FilerSource{Dir: "/source"}}
+	deleted := &filer_pb.Entry{Name: "gone.bin", CommitTsNs: 100, Attributes: &filer_pb.FuseAttributes{}}
+	if err := r.Replicate(context.Background(), "/source/dir/gone.bin", &filer_pb.EventNotification{OldEntry: deleted}); err != nil {
+		t.Fatal(err)
+	}
+	renamed := &filer_pb.Entry{Name: "old.bin", CommitTsNs: 200, Attributes: &filer_pb.FuseAttributes{}}
+	if err := r.Replicate(context.Background(), "/source/dir/old.bin", &filer_pb.EventNotification{
+		OldEntry:      renamed,
+		NewEntry:      &filer_pb.Entry{Name: "new.bin", Attributes: &filer_pb.FuseAttributes{}},
+		NewParentPath: "/source/dir",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	if len(s.versionCalls) != 2 || s.versionCalls[0] != deleted || s.versionCalls[1] != renamed {
+		t.Fatalf("versioned deletes = %v, want the deleted and the renamed entries", s.versionCalls)
+	}
+	if len(s.deleteCalls) != 2 || s.deleteCalls[0].key != "/dest/dir/gone.bin" || s.deleteCalls[1].key != "/dest/dir/old.bin" {
+		t.Fatalf("deletes = %+v", s.deleteCalls)
+	}
+}

weed/replication/sink/localsink/local_sink.go

@@ -2,11 +2,14 @@ package localsink
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io/fs"
 	"math/rand/v2"
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -18,6 +21,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
+var sourceLookupTimeout = 30 * time.Second
+
 type LocalSink struct {
 	Dir           string
 	filerSource   *source.FilerSource
@@ -94,7 +99,15 @@ func (localsink *LocalSink) IsIncremental() bool {
 }
 
 func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bool, signatures []int32) error {
-	key = sanitizeFsKey(key)
+	return localsink.deleteEntry(key, nil, 0)
+}
+
+func (localsink *LocalSink) DeleteEntryVersion(key string, oldEntry *filer_pb.Entry, tsNs int64, deleteIncludeChunks bool, signatures []int32) error {
+	return localsink.deleteEntry(key, oldEntry, tsNs)
+}
+
+func (localsink *LocalSink) deleteEntry(sourceKey string, oldEntry *filer_pb.Entry, tsNs int64) error {
+	key := sanitizeFsKey(sourceKey)
 	if localsink.isMultiPartEntry(key) {
 		return nil
 	}
@@ -104,16 +117,89 @@ func (localsink *LocalSink) DeleteEntry(key string, isDirectory, deleteIncludeCh
 	}
 	glog.V(4).Infof("Delete Entry key: %s", key)
 	root, err := os.OpenRoot(localsink.Dir)
-	if err == nil {
-		err = root.Remove(localsink.resolveInsideDir(rel))
-		root.Close()
-	}
 	if err != nil {
 		glog.V(0).Infof("remove entry key %s: %s", key, err)
+		return nil
+	}
+	defer root.Close()
+	rel = localsink.resolveInsideDir(rel)
+	if oldEntry != nil {
+		if _, statErr := root.Lstat(rel); errors.Is(statErr, fs.ErrNotExist) {
+			return nil
+		}
+		recreated, lookupErr := localsink.sourceRecreated(sourceKey, oldEntry, tsNs)
+		if lookupErr != nil {
+			return lookupErr
+		}
+		if recreated {
+			glog.V(1).Infof("keep %s: the source recreated it after the deleted version", key)
+			return nil
+		}
+	}
+	if err := root.Remove(rel); err != nil {
+		glog.V(0).Infof("remove entry key %s: %s", key, err)
 	}
 	return nil
 }
 
+// sourceRecreated reports whether the live source holds an entry of the same
+// kind at key's source path committed after the delete at tsNs, the order the
+// source's own replay keeps; an unstamped live entry, written by a filer that
+// predates commit stamps, counts from its mtime. Without tsNs the deleted
+// entry's commit stamp is the bound; when the deleted entry is unstamped too,
+// a stamped live entry was written after it, and mtimes decide otherwise. An
+// older live entry means the source store lags the delete event or will apply
+// it next.
+func (localsink *LocalSink) sourceRecreated(key string, oldEntry *filer_pb.Entry, tsNs int64) (bool, error) {
+	sourcePath, ok := localsink.sourcePath(key)
+	if !ok {
+		return false, nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout)
+	defer cancel()
+	live, _, _, err := filer_pb.GetEntry(ctx, localsink.filerSource, sourcePath)
+	if errors.Is(err, filer_pb.ErrNotFound) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("look up source %s before deleting %s: %w", sourcePath, key, err)
+	}
+	if live == nil || live.IsDirectory != oldEntry.IsDirectory {
+		return false, nil
+	}
+	if tsNs <= 0 {
+		tsNs = oldEntry.GetCommitTsNs()
+	}
+	liveTsNs := live.GetCommitTsNs()
+	if tsNs > 0 {
+		if liveTsNs == 0 {
+			liveTsNs = mtimeNs(live)
+		}
+		return liveTsNs > tsNs, nil
+	}
+	if liveTsNs > 0 {
+		return true, nil
+	}
+	return mtimeNs(live) > mtimeNs(oldEntry), nil
+}
+
+// sourcePath maps a sink key back to the source path it replicates. Incremental
+// keys carry a date directory and do not map back.
+func (localsink *LocalSink) sourcePath(key string) (util.FullPath, bool) {
+	if localsink.filerSource == nil || localsink.isIncremental {
+		return "", false
+	}
+	rel, err := filepath.Rel(filepath.Clean(localsink.Dir), filepath.Clean(key))
+	if err != nil || rel == "." || !filepath.IsLocal(rel) {
+		return "", false
+	}
+	return util.FullPath(localsink.filerSource.Dir).Child(filepath.ToSlash(rel)), true
+}
+
+func mtimeNs(entry *filer_pb.Entry) int64 {
+	return entry.GetAttributes().GetMtime()*int64(time.Second) + int64(entry.GetAttributes().GetMtimeNs())
+}
+
 func (localsink *LocalSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 	key = sanitizeFsKey(key)
 	if entry.IsDirectory || localsink.isMultiPartEntry(key) {

weed/replication/sink/localsink/local_sink_delete_test.go

@@ -0,0 +1,144 @@
+package localsink
+
+import (
+	"context"
+	"net"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/replication/source"
+)
+
+type liveSourceFiler struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	live    *filer_pb.Entry
+	err     error
+	lookups atomic.Int32
+	asked   atomic.Value
+}
+
+func (s *liveSourceFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	s.lookups.Add(1)
+	s.asked.Store(req.Directory + "/" + req.Name)
+	if s.err != nil {
+		return nil, s.err
+	}
+	if s.live == nil {
+		return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error())
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: s.live}, nil
+}
+
+func startLiveSourceFiler(t *testing.T, impl *liveSourceFiler) *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, impl)
+	go server.Serve(listener)
+	t.Cleanup(server.Stop)
+
+	address := listener.Addr().String()
+	filerSource := &source.FilerSource{}
+	if err := filerSource.DoInitialize(address, address, "/src", false); err != nil {
+		t.Fatalf("DoInitialize: %v", err)
+	}
+	filerSource.SetGrpcDialOption(grpc.WithTransportCredentials(insecure.NewCredentials()))
+	return filerSource
+}
+
+func stampedEntry(name string, commitTsNs, mtimeSec int64, isDirectory bool) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name:        name,
+		IsDirectory: isDirectory,
+		CommitTsNs:  commitTsNs,
+		Attributes:  &filer_pb.FuseAttributes{Mtime: mtimeSec, FileMode: 0644},
+	}
+}
+
+func TestDeleteEntryVersionAgainstLiveSource(t *testing.T) {
+	deleted := stampedEntry("x", 100, 10, false)
+	tests := []struct {
+		name     string
+		live     *filer_pb.Entry
+		lookup   error
+		oldEntry *filer_pb.Entry
+		tsNs     int64
+		wantKept bool
+		wantErr  bool
+	}{
+		{name: "source recreated the path", live: stampedEntry("x", 300, 5, false), oldEntry: deleted, tsNs: 200, wantKept: true},
+		{name: "source deleted the path", oldEntry: deleted, tsNs: 200},
+		{name: "source store lags the delete", live: stampedEntry("x", 100, 10, false), oldEntry: deleted, tsNs: 200},
+		{name: "source holds a version written before the delete", live: stampedEntry("x", 200, 20, false), oldEntry: deleted, tsNs: 300},
+		{name: "source recreated a directory", live: stampedEntry("x", 300, 30, true), oldEntry: deleted, tsNs: 200},
+		{name: "stamped replacement of an unstamped entry", live: stampedEntry("x", 300, 5, false), oldEntry: stampedEntry("x", 0, 10, false), tsNs: 200, wantKept: true},
+		{name: "no delete time, newer than the deleted stamp", live: stampedEntry("x", 300, 5, false), oldEntry: deleted, wantKept: true},
+		{name: "no delete time, the deleted version", live: stampedEntry("x", 100, 10, false), oldEntry: deleted},
+		{name: "no delete time, stamped replacement of an unstamped entry", live: stampedEntry("x", 300, 5, false), oldEntry: stampedEntry("x", 0, 10, false), wantKept: true},
+		{name: "no delete time, unstamped entry with an older mtime", live: stampedEntry("x", 0, 5, false), oldEntry: stampedEntry("x", 0, 10, false)},
+		{name: "unstamped source, replacement written after the delete", live: stampedEntry("x", 0, 30, false), oldEntry: stampedEntry("x", 0, 10, false), tsNs: 20e9, wantKept: true},
+		{name: "unstamped source, entry written before the delete", live: stampedEntry("x", 0, 15, false), oldEntry: stampedEntry("x", 0, 10, false), tsNs: 20e9},
+		{name: "nothing stamped, newer mtime", live: stampedEntry("x", 0, 20, false), oldEntry: stampedEntry("x", 0, 10, false), wantKept: true},
+		{name: "nothing stamped, same mtime", live: stampedEntry("x", 0, 10, false), oldEntry: stampedEntry("x", 0, 10, false)},
+		{name: "source lookup fails", lookup: status.Error(codes.Unavailable, "down"), oldEntry: deleted, tsNs: 200, wantKept: true, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			dir := t.TempDir()
+			sourceFiler := &liveSourceFiler{live: tt.live, err: tt.lookup}
+			localSink := &LocalSink{}
+			localSink.initialize(dir, false)
+			localSink.SetSourceFiler(startLiveSourceFiler(t, sourceFiler))
+
+			key := filepath.Join(dir, "sub", "x")
+			if err := os.MkdirAll(filepath.Dir(key), 0755); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.WriteFile(key, []byte("replacement"), 0644); err != nil {
+				t.Fatal(err)
+			}
+
+			err := localSink.DeleteEntryVersion(key, tt.oldEntry, tt.tsNs, true, nil)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("DeleteEntryVersion err = %v, want error %v", err, tt.wantErr)
+			}
+			if got := sourceFiler.asked.Load(); got != "/src/sub/x" {
+				t.Fatalf("looked up %v, want /src/sub/x", got)
+			}
+			data, readErr := os.ReadFile(key)
+			if tt.wantKept {
+				if readErr != nil || string(data) != "replacement" {
+					t.Fatalf("destination lost: %q %v", data, readErr)
+				}
+			} else if !os.IsNotExist(readErr) {
+				t.Fatalf("destination kept after an ordered delete: %q %v", data, readErr)
+			}
+		})
+	}
+}
+
+func TestDeleteEntryVersionSkipsLookupWhenDestinationIsGone(t *testing.T) {
+	dir := t.TempDir()
+	sourceFiler := &liveSourceFiler{err: status.Error(codes.Unavailable, "down")}
+	localSink := &LocalSink{}
+	localSink.initialize(dir, false)
+	localSink.SetSourceFiler(startLiveSourceFiler(t, sourceFiler))
+
+	if err := localSink.DeleteEntryVersion(filepath.Join(dir, "x"), stampedEntry("x", 100, 10, false), 200, true, nil); err != nil {
+		t.Fatalf("duplicate delete: %v", err)
+	}
+	if n := sourceFiler.lookups.Load(); n != 0 {
+		t.Fatalf("duplicate delete looked up the source %d times", n)
+	}
+}

weed/replication/sink/replication_sink.go

@@ -63,6 +63,25 @@ func DeleteEntry(s ReplicationSink, key string, isDirectory, deleteIncludeChunks
 	return s.DeleteEntry(key, isDirectory, deleteIncludeChunks, signatures)
 }
 
+// VersionedDeleter is an optional capability for sinks that check a delete
+// against the live source before removing the destination: a delete that
+// reaches the sink after the source recreated the path must not remove the
+// newer destination, since no later event brings it back. tsNs is the
+// delete's mutation time, or 0 when the event does not carry one.
+type VersionedDeleter interface {
+	DeleteEntryVersion(key string, oldEntry *filer_pb.Entry, tsNs int64, deleteIncludeChunks bool, signatures []int32) error
+}
+
+func DeleteEntryVersion(s ReplicationSink, key string, oldEntry *filer_pb.Entry, tsNs int64, deleteIncludeChunks, keepRemoteObject bool, signatures []int32) error {
+	if _, ok := s.(MetadataOnlyDeleter); ok && keepRemoteObject {
+		return DeleteEntry(s, key, oldEntry.IsDirectory, deleteIncludeChunks, keepRemoteObject, signatures)
+	}
+	if deleter, ok := s.(VersionedDeleter); ok {
+		return deleter.DeleteEntryVersion(key, oldEntry, tsNs, deleteIncludeChunks, signatures)
+	}
+	return DeleteEntry(s, key, oldEntry.IsDirectory, deleteIncludeChunks, keepRemoteObject, signatures)
+}
+
 var (
 	Sinks []ReplicationSink
 )