diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index ae432ac538..caaaddc502 100644 --- a/weed/command/filer_sync.go +++ b/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 { diff --git a/weed/command/filer_sync_replayed_delete_test.go b/weed/command/filer_sync_replayed_delete_test.go new file mode 100644 index 0000000000..eb6cfe6ee3 --- /dev/null +++ b/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) + } + } +} diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go index c3ad29fb7d..fbdc320a8d 100644 --- a/weed/replication/replicator.go +++ b/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). diff --git a/weed/replication/replicator_test.go b/weed/replication/replicator_test.go index bdfcb843ec..557662f8ec 100644 --- a/weed/replication/replicator_test.go +++ b/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) + } +} diff --git a/weed/replication/sink/localsink/local_sink.go b/weed/replication/sink/localsink/local_sink.go index b66e8abff4..0937788685 100644 --- a/weed/replication/sink/localsink/local_sink.go +++ b/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) { diff --git a/weed/replication/sink/localsink/local_sink_delete_test.go b/weed/replication/sink/localsink/local_sink_delete_test.go new file mode 100644 index 0000000000..8aed5f6432 --- /dev/null +++ b/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) + } +} diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index dd4c7bac07..8e3a51c2cc 100644 --- a/weed/replication/sink/replication_sink.go +++ b/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 )