diff --git a/weed/filer/filer_delete_entry.go b/weed/filer/filer_delete_entry.go index 646caf5c9..207fdc7e6 100644 --- a/weed/filer/filer_delete_entry.go +++ b/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) diff --git a/weed/filer/leveldb2/empty_folder_race_test.go b/weed/filer/leveldb2/empty_folder_race_test.go new file mode 100644 index 000000000..431b0c908 --- /dev/null +++ b/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) + } +}