filer: do not sweep children when deleting a folder non-recursively (#10782)

* filer: do not sweep children when deleting a folder non-recursively

doBatchDeleteFolderMetaAndData lists a folder and bails out if it has any
children, then calls Store.DeleteFolderChildren unconditionally. On the
non-recursive path that bulk sweep has nothing legitimate to remove: it only
runs once the listing came back empty, so the sole rows it can delete are
ones inserted after the check.

The S3 empty-folder cleaner deletes through this path, so a PUT landing
between the listing and the sweep loses its entry after the write was already
acknowledged. Neither side sees an error - the client has its 200 and the
cleaner logs an ordinary empty-folder deletion - and the chunks leak, since
the cleaner passes shouldDeleteChunks=false and nothing was enumerated to
collect. Workloads that scatter objects over many shallow prefixes empty and
refill those folders constantly, which is what makes the window reachable.

Sweep only when the delete is recursive, or when the whole-bucket shortcut
skipped the listing and depends on it.

Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r

* filer: pin the folder entry removal left by the racing-child test

The surviving entry is reachable by path but drops out of listings until the
folder comes back, and nothing in the test said so. Assert it, so the exposure
that remains after this change is visible rather than implied.

Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r

weed/filer/filer_delete_entry.go

@@ -78,7 +78,8 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 	var chunksToDelete []*filer_pb.FileChunk
 	lastFileName := ""
 	includeLastFile := false
-	if !isDeletingBucket || !f.Store.CanDropWholeBucket() {
+	listedChildren := !isDeletingBucket || !f.Store.CanDropWholeBucket()
+	if listedChildren {
 		for {
 			entries, _, err := f.ListDirectoryEntries(ctx, entry.FullPath, lastFileName, includeLastFile, PaginationSize, "", "", "")
 			if err != nil {
@@ -131,8 +132,12 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
 
 	glog.V(3).InfofCtx(ctx, "deleting directory %v delete chunks: %v", entry.FullPath, shouldDeleteChunks)
 
-	if storeDeletionErr := f.Store.DeleteFolderChildren(ctx, entry.FullPath); storeDeletionErr != nil {
-		return fmt.Errorf("filer store delete: %w", storeDeletionErr)
+	// a non-recursive delete already proved the folder empty above, so sweeping the
+	// children now can only remove entries that raced in after that listing
+	if isRecursive || !listedChildren {
+		if storeDeletionErr := f.Store.DeleteFolderChildren(ctx, entry.FullPath); storeDeletionErr != nil {
+			return fmt.Errorf("filer store delete: %w", storeDeletionErr)
+		}
 	}
 
 	f.NotifyUpdateEvent(ctx, entry, nil, shouldDeleteChunks, isFromOtherCluster, signatures)

weed/filer/leveldb2/empty_folder_race_test.go

@@ -0,0 +1,78 @@
+package leveldb
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// listHookStore runs a hook once, right after a directory listing returns, so a
+// test can act inside the window between a delete's emptiness check and the
+// removal of the folder.
+type listHookStore struct {
+	filer.FilerStore
+	hook  func()
+	fired bool
+}
+
+func (s *listHookStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (string, error) {
+	lastFileName, err := s.FilerStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, eachEntryFunc)
+	if s.hook != nil && !s.fired {
+		s.fired = true
+		s.hook()
+	}
+	return lastFileName, err
+}
+
+// TestNonRecursiveFolderDeleteKeepsRacingChild covers the S3 empty-folder
+// cleanup path: the delete lists a folder, finds it empty, and must not then
+// bulk-delete its children, because an object written in between would be
+// destroyed after the write was already acknowledged.
+func TestNonRecursiveFolderDeleteKeepsRacingChild(t *testing.T) {
+	testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil)
+	store := &LevelDB2Store{}
+	if err := store.initialize(t.TempDir(), 2); err != nil {
+		t.Fatal(err)
+	}
+	hooked := &listHookStore{FilerStore: store}
+	testFiler.SetStore(hooked)
+
+	// the test has no metadata log consumer
+	ctx := filer.WithSuppressedMetadataEvents(context.Background())
+	dir := util.FullPath("/buckets/testbucket/data/abc")
+	child := dir.Child("obj")
+
+	dirEntry := &filer.Entry{FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0755}}
+	if err := testFiler.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil {
+		t.Fatalf("create folder: %v", err)
+	}
+
+	hooked.hook = func() {
+		entry := &filer.Entry{FullPath: child, Attr: filer.Attr{Mode: 0640}}
+		if err := testFiler.CreateEntry(ctx, entry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil {
+			t.Errorf("create entry racing the folder delete: %v", err)
+		}
+	}
+
+	if err := testFiler.DeleteEntryMetaAndData(ctx, dir, false, false, false, false, nil, 0); err != nil {
+		t.Fatalf("delete empty folder: %v", err)
+	}
+
+	if _, err := testFiler.FindEntry(ctx, child); err != nil {
+		t.Errorf("entry created during the folder delete was removed: %v", err)
+	}
+
+	// The folder entry itself still goes, so the surviving entry is reachable by
+	// path but absent from listings until the folder comes back. Pinning that here
+	// keeps the remaining exposure visible; tighten it if the two steps ever become
+	// atomic.
+	if _, err := testFiler.FindEntry(ctx, dir); !errors.Is(err, filer_pb.ErrNotFound) {
+		t.Errorf("folder entry should still be removed, got %v", err)
+	}
+}