filer: sweep the local subtree on a replayed directory delete (#11785)

* filer: sweep the local subtree on a replayed directory delete A recursive delete emits only the directory's own key, so peers applied it with a single DeleteEntry. Children the peer held that the origin had not listed stayed behind unlisted but still readable, and their needles were never deleted. Sweep the local children on a replayed directory delete and drop their chunks when the origin deleted chunks. * filer: walk the subtree on a replayed directory delete DeleteFolderChildren removes only direct children in stores keyed by immediate parent, so the sweep now lists and deletes each descendant entry itself. Entries the local filer committed after the event survive — with the directories that hold them — and hard-linked files keep their chunks for the origin's own events to settle. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: key replayed subtree chunk deletes to each entry's path The sweep queued every descendant's chunks under the deleted directory's storage rule and skipped directory entries entirely. Chunks now collect keyed by their owning path, so per-path rules apply to each file — and a directory carrying its own chunks (a file promoted by an S3 child) frees them too. * filer: fence kept replayed dirs and flush collected chunks on error A directory kept because a newer descendant survived still needs its delete timestamp recorded, or a stale replayed create can resurrect another child under it. And chunks collected for deleted children must be flushed even when the sweep fails partway — the entries are gone and a retry cannot list them again. Collection now also runs only after the entry delete succeeds. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: flush swept chunk references in batches --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-17 19:48:35 +02:00 · 2026-10-16 19:47:20 -07:00
1 parent 5b4a707c43
commit 348116b2c4
@@ -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")
+	}
+}
@@ -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