From bef1b13b3d1a8f1255789345e9785a49026636c2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 09:23:54 -0800 Subject: [PATCH] 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 * 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 * 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 * 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 * 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 --------- Co-authored-by: Claude Opus 5.5 --- weed/command/filer_sync.go | 10 +- .../filer_sync_replayed_delete_test.go | 94 ++++++++++++ weed/replication/replicator.go | 4 +- weed/replication/replicator_test.go | 36 +++++ weed/replication/sink/localsink/local_sink.go | 96 +++++++++++- .../sink/localsink/local_sink_delete_test.go | 144 ++++++++++++++++++ weed/replication/sink/replication_sink.go | 19 +++ 7 files changed, 393 insertions(+), 10 deletions(-) create mode 100644 weed/command/filer_sync_replayed_delete_test.go create mode 100644 weed/replication/sink/localsink/local_sink_delete_test.go 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 )