diff --git a/weed/filer/meta_aggregator_replay_test.go b/weed/filer/meta_aggregator_replay_test.go index 10522f7f1..e41c222c1 100644 --- a/weed/filer/meta_aggregator_replay_test.go +++ b/weed/filer/meta_aggregator_replay_test.go @@ -3,6 +3,7 @@ package filer import ( "context" "errors" + "os" "sync" "testing" "time" @@ -708,3 +709,172 @@ func TestPendingDeleteRecordKeepsItsTime(t *testing.T) { t.Fatalf("MpDl = %v, %v, %v; want 100", ts, found, err) } } + +// A replayed directory delete must remove the local subtree too: the origin +// only emits the directory's own key, and children the peer holds that the +// origin never listed would orphan under the missing parent. +func TestReplayDirectoryDeleteSweepsLocalChildren(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + + put := func(path string, isDir bool) { + e := &Entry{FullPath: util.FullPath(path)} + if isDir { + e.Attr.Mode = os.ModeDir + } + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatalf("insert %s: %v", path, err) + } + } + put("/d", true) + put("/d/a.txt", false) + put("/d/sub", true) + put("/d/sub/nested.txt", false) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + for _, path := range []string{"/d", "/d/a.txt", "/d/sub", "/d/sub/nested.txt"} { + if _, err := store.FindEntry(ctx, util.FullPath(path)); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("expected %s gone after replayed directory delete, got %v", path, err) + } + } +} + +func TestReplayFileDeleteKeepsSiblingEntries(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + + dir := &Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir}} + if err := store.InsertEntry(ctx, dir); err != nil { + t.Fatal(err) + } + for _, name := range []string{"gone.txt", "kept.txt"} { + if err := store.InsertEntry(ctx, &Entry{FullPath: util.NewFullPath("/d", name)}); err != nil { + t.Fatal(err) + } + } + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/d", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "gone.txt"}, + }, + TsNs: time.Now().UnixNano(), + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + if _, err := store.FindEntry(ctx, "/d/gone.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("expected deleted file gone, got %v", err) + } + if _, err := store.FindEntry(ctx, "/d/kept.txt"); err != nil { + t.Fatalf("expected sibling kept, got %v", err) + } +} + +func TestReplayDirectoryDeleteKeepsNewerChildren(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + + eventTs := time.Now().UnixNano() + put := func(path string, isDir bool, mtime time.Time) { + e := &Entry{FullPath: util.FullPath(path), Attr: Attr{Mtime: mtime}} + if isDir { + e.Attr.Mode = os.ModeDir + } + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatalf("insert %s: %v", path, err) + } + } + old := time.Unix(1600000000, 0) + newer := time.Unix(0, eventTs).Add(time.Second) + put("/d", true, old) + put("/d/old.txt", false, old) + put("/d/sub", true, old) + put("/d/sub/newer.txt", false, newer) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + DeleteChunks: false, + }, + TsNs: eventTs, + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + + if _, err := store.FindEntry(ctx, "/d/old.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("older child should be swept, got %v", err) + } + for _, path := range []string{"/d", "/d/sub", "/d/sub/newer.txt"} { + if _, err := store.FindEntry(ctx, util.FullPath(path)); err != nil { + t.Fatalf("newer child and its ancestors should survive: %s: %v", path, err) + } + } +} + +// A directory kept for newer descendants still records the delete: a stale +// create replayed later for another child must be fenced by it. +func TestReplayKeptDirectoryDeleteFencesStaleCreates(t *testing.T) { + store := newStubFilerStore() + f := newReplayTestFiler(store) + ctx := context.Background() + + eventTs := time.Now().UnixNano() + old := time.Unix(1600000000, 0) + newer := time.Unix(0, eventTs).Add(time.Second) + put := func(path string, isDir bool, mtime time.Time) { + e := &Entry{FullPath: util.FullPath(path), Attr: Attr{Mtime: mtime}} + if isDir { + e.Attr.Mode = os.ModeDir + } + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatalf("insert %s: %v", path, err) + } + } + put("/d", true, old) + put("/d/newer.txt", false, newer) + + event := &filer_pb.SubscribeMetadataResponse{ + Directory: "/", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true}, + }, + TsNs: eventTs, + } + if err := f.Replay(event); err != nil { + t.Fatalf("replay: %v", err) + } + if _, err := store.FindEntry(ctx, "/d"); err != nil { + t.Fatalf("newer child should keep the directory: %v", err) + } + + stale := &filer_pb.SubscribeMetadataResponse{ + Directory: "/d", + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "late.txt"}, + }, + TsNs: eventTs - int64(time.Minute), + } + if err := f.Replay(stale); err != nil { + t.Fatalf("replay stale create: %v", err) + } + if _, err := store.FindEntry(ctx, "/d/late.txt"); !errors.Is(err, filer_pb.ErrNotFound) { + t.Fatal("stale create resurrected a child under a directory the origin deleted") + } +} diff --git a/weed/filer/meta_replay.go b/weed/filer/meta_replay.go index 6895490b8..cad4f41cf 100644 --- a/weed/filer/meta_replay.go +++ b/weed/filer/meta_replay.go @@ -2,6 +2,7 @@ package filer import ( "context" + "fmt" "slices" "strings" @@ -78,10 +79,26 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { return err } if !newer { - glog.V(4).Infof("deleting %v", oldPath) - if err := f.Store.DeleteEntry(ctx, oldPath); err != nil { - return err + // A directory delete on the origin removed the whole subtree, but + // only the directory's own key replicates. Sweep local children the + // origin never listed, or they orphan under a missing parent. + keepDir := false + if message.OldEntry.IsDirectory { + var err error + keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message.DeleteChunks, resp.TsNs) + if err != nil { + return err + } } + if !keepDir { + glog.V(4).Infof("deleting %v", oldPath) + if err := f.Store.DeleteEntry(ctx, oldPath); err != nil { + return err + } + } + // The origin deleted the directory even when a newer descendant + // keeps its entry here — recording the delete fences stale child + // creates that would otherwise resurrect under it. f.notePathChange(oldPath, resp.TsNs, true) } } @@ -104,6 +121,99 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error { return nil } +// deleteReplayedFolderChildren removes local entries under dir, and queues +// the chunks of removed files for deletion. Children this filer committed +// after the event stay — a newer local write wins over a delayed delete — +// along with the directories that hold them. +func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64) (bool, error) { + sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)} + kept, err := f.sweepReplayedFolder(ctx, dir, deleteChunks, tsNs, sweep) + // Flush what was collected even on error: the entries already deleted + // cannot be listed again on retry, so their chunks must be queued now. + sweep.flush(ctx, f) + if err != nil { + return false, err + } + return kept, nil +} + +// folderSweep accumulates the chunks of removed children for deletion. It +// flushes in batches so a large subtree cannot hold every chunk reference in +// memory at once, and so chunk queueing overlaps the listing work. +type folderSweep struct { + chunksByPath map[util.FullPath][]*filer_pb.FileChunk + count int +} + +const folderSweepChunkBatch = 4096 + +func (s *folderSweep) add(path util.FullPath, chunks []*filer_pb.FileChunk) { + s.chunksByPath[path] = append(s.chunksByPath[path], chunks...) + s.count += len(chunks) +} + +func (s *folderSweep) flush(ctx context.Context, f *Filer) { + for path, chunks := range s.chunksByPath { + f.DeleteChunks(ctx, path, chunks) + } + s.chunksByPath = make(map[util.FullPath][]*filer_pb.FileChunk) + s.count = 0 +} + +// sweepReplayedFolder deletes every child of dir whose local commit is not +// newer than the event, working subtree-first so nested entries go too. +// It reports whether a newer descendant keeps dir alive — the caller then +// keeps the dir entry itself. +func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64, sweep *folderSweep) (kept bool, err error) { + lastFileName := "" + for { + entries, _, lerr := f.ListDirectoryEntries(ctx, dir, lastFileName, false, PaginationSize, "", "", "") + if lerr != nil { + return false, fmt.Errorf("list folder %s: %w", dir, lerr) + } + for _, sub := range entries { + lastFileName = sub.Name() + childKept := false + if tsNs > 0 { + localTsNs, tsErr := f.localCommitTsNs(ctx, sub.FullPath) + if tsErr != nil { + return false, tsErr + } + childKept = localTsNs > tsNs + } + if !childKept && sub.IsDirectory() { + childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, deleteChunks, tsNs, sweep) + if err != nil { + return false, err + } + } + if childKept { + kept = true + continue + } + if derr := f.Store.DeleteEntry(ctx, sub.FullPath); derr != nil { + return false, fmt.Errorf("delete %s: %w", sub.FullPath, derr) + } + f.notePathChange(sub.FullPath, tsNs, true) + // hard-linked chunks are owned by the shared inode record; the + // origin's own events decide their deletion. A directory entry + // can still carry chunks of its own — an S3 child promotes an + // existing file to a directory without dropping them — so the + // entry's chunks are keyed to its own path whatever it is. + if deleteChunks && len(sub.HardLinkId) == 0 && len(sub.GetChunks()) > 0 { + sweep.add(sub.FullPath, sub.GetChunks()) + if sweep.count >= folderSweepChunkBatch { + sweep.flush(ctx, f) + } + } + } + if len(entries) < PaginationSize { + break + } + } + return kept, nil +} + // replayBlockedByLocal reports whether the local store already holds a // version of path committed after the event's timestamp. Lookup failures are // propagated: a replayed change must not be applied on the guess that the