From 5c43c03b76c0c319b1d5f33ffa2985df9e727906 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 17 Aug 2026 00:04:07 -0700 Subject: [PATCH] filer: restore a folder that received an entry while it was deleted (#10783) * filer: restore a folder that received an entry while it was deleted The empty-folder cleaner checks that a folder is empty and then deletes it, and those two steps are not atomic. An entry created in between survives the delete but loses the directory holding it: still readable by its own path, yet absent from every listing until a later write happens to recreate the parent. Record the folders deleted in each pass and re-check them on the next one, putting back any that turned out to hold entries. The check waits a pass on purpose - a writer looks up the parent before inserting the child, so checking straight after the delete can still run ahead of the insert and see nothing. Restoring a directory that holds entries is always correct, and restoring one whose entry went away again just leaves an empty folder for a later pass to collect, so the repair needs no locking or coordination. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * filer: keep failed restores queued and inherit the ancestor's ownership Two gaps in the restore pass. A folder whose count or restore hit a transient store error was dropped from the tracking list and never looked at again, leaving its entries out of listings until some later write recreated the folder - the very thing the pass exists to avoid. Put those back for the next pass, still under the cap. A restored folder was minted with a fixed mode and no owner, so a directory that had been private came back world-readable and owned by root. Take the mode and ownership from the nearest ancestor still present instead. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * filer: let the redis stores keep a directory listing that still has entries On the redis stores the listing is not derived from the entries, it is the only record that they sit under that directory. DeleteEntry opened by dropping it outright, so an entry that arrived after the caller judged the directory empty lost its membership and became unreachable: readable by exact path, absent from every listing, and invisible to any later check, since counting the directory reads the listing that was just destroyed. Nothing could detect or repair it. Drop the listing in DeleteFolderChildren instead, alongside the children it describes, and leave it alone in DeleteEntry. redis3 needs it explicitly, since removeChildren clears the skip list nodes but not the list itself, and the plain redis store was leaking the key entirely. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * filer: restore folders with their own attributes, and observe them for a window Five gaps in the restore pass. The restored directory was reconstructed from whatever ancestor happened to still be present, and the mode was ORed with 0111 on the way. A private directory under a world-traversable parent came back granting traversal it had denied. Read the folder's own attributes before deleting it and put exactly those back. That also removes the ancestor walk, which treated a transient store error as "not found" and silently fell through to a broader ancestor. A single check a pass later was not a delay at all. Ticker sends coalesce, so when a pass runs long the next one starts immediately, and a writer already past its parent lookup can insert after the check has read zero - after which the folder was discarded for good. Keep each folder under observation for a bounded wall-clock window and re-check it on every pass until it expires. This narrows the exposure rather than closing it; only making the emptiness check and the delete atomic would do that. A delete that returned an error was never observed at all, though the redis stores drop the folder before its parent-list member, so a failure return is not proof the folder survived. Record the folder before the delete instead. Restores now run shallowest first, so a folder taken by the parent cascade is rebuilt with its own attributes before anything below it needs it as a parent. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * filer: recover a deleted folder from the create event for the entry that raced it Checking each deleted folder on a timer was the wrong instrument. It cost a listing per folder per pass, and it could only ever be a guess about when the racing write would land. The metadata stream already carries the answer. A folder is recorded before it is deleted, so any entry that can be orphaned is created after that record and its create event names that exact directory. Match the event against the recently deleted folders and the folder is known to need putting back, rather than inferred to. The window stops being a guess at the race and becomes what it should be: how far behind the event stream is allowed to run before a folder stops being watched. Listing is now done once, for a folder an event has already named, to skip the restore when the entry has since gone away again. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * filer: bound how long a folder is watched, and rebuild ancestors from themselves Four gaps found reviewing the restore pass. A folder whose restore kept failing was never let go: the written-to check ran before the age check, so it was picked up, retried, put back, and counted again on every pass for the life of the process. Apply the window first, whatever state the folder is in. At the cap, the folder being recorded was the one turned away, though it is the one whose race is still live - the older entries are already close to ageing out. Give up one of those instead, picked as the oldest of a small sample so the cost stays flat under heavy deletion rates. An ancestor taken by the same cascade was left to the descendant's restore to recreate, which minted it from the descendant's attributes and handed back access the ancestor never granted. Rebuild those from what they were, ahead of anything below them. Reading a directory's attributes assumed an entry came back. Some stores return nothing with no error, so treat that as not found. The mode is also taken whole rather than through Perm(), which was dropping setgid, setuid and sticky. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r * redis3: take a directory listing left behind by a failed delete Removing the last name deletes the list, and if that delete fails the header survives pointing at a name that is gone. The retry finds nothing to remove, reports no changes, and returns before reaching the delete, so the key stays for good. Take it on that path too. Claude-Session: https://claude.ai/code/session_01HdLXMUopwgofPb1ZEmiE6r --- .../empty_folder_cleaner.go | 185 +++++++++++++- .../empty_folder_cleaner_test.go | 233 ++++++++++++++++++ weed/filer/filer.go | 59 +++++ weed/filer/leveldb2/empty_folder_race_test.go | 118 +++++++++ weed/filer/redis/universal_redis_store.go | 7 +- .../filer/redis/universal_redis_store_test.go | 26 ++ weed/filer/redis2/universal_redis_store.go | 16 +- .../redis2/universal_redis_store_test.go | 30 +++ weed/filer/redis3/ItemList.go | 4 + weed/filer/redis3/kv_directory_children.go | 12 + weed/filer/redis3/universal_redis_store.go | 22 +- .../redis3/universal_redis_store_test.go | 129 ++++++++++ 12 files changed, 826 insertions(+), 15 deletions(-) create mode 100644 weed/filer/redis3/universal_redis_store_test.go diff --git a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go index 82213fce8..9bcd2a3cc 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go @@ -2,6 +2,8 @@ package empty_folder_cleanup import ( "context" + "os" + "sort" "strings" "sync" "time" @@ -20,14 +22,39 @@ const ( DefaultQueueMaxSize = 1000 DefaultQueueMaxAge = 2 * time.Minute DefaultProcessorSleep = 30 * time.Second // How often to check queue + DefaultMaxDeletedKept = 10000 // Deleted folders remembered for the restore check + // How long a deleted folder is kept so that a create event arriving for it can + // still put it back. It bounds how far behind the event stream may run, not how + // long the race window is. + DefaultObservationWindow = 2 * time.Minute ) +// DirectoryAttributes is what a restored directory needs to come back as it was. +type DirectoryAttributes struct { + Mode os.FileMode + Uid uint32 + Gid uint32 + UserName string + GroupNames []string +} + +// deletedFolder is a folder under observation for entries that landed while it was +// being deleted. writtenTo is set by the create event for such an entry. +type deletedFolder struct { + path string + attrs DirectoryAttributes + deletedAt time.Time + writtenTo bool +} + // FilerOperations defines the filer operations needed by EmptyFolderCleaner type FilerOperations interface { CountDirectoryEntries(ctx context.Context, dirPath util.FullPath, limit int) (count int, err error) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isFromOtherCluster bool, signatures []int32, ifNotModifiedAfter int64) error GetEntryAttributes(ctx context.Context, p util.FullPath) (attributes map[string][]byte, err error) IsDirectoryKeyObject(ctx context.Context, p util.FullPath) (bool, error) + DirectoryAttributes(ctx context.Context, p util.FullPath) (DirectoryAttributes, error) + EnsureDirectoryEntry(ctx context.Context, p util.FullPath, attrs DirectoryAttributes) error } // folderState tracks the state of a folder for empty folder cleanup @@ -56,6 +83,11 @@ type EmptyFolderCleaner struct { folderCounts map[string]*folderState // Rough count cache bucketCleanupPolicies map[string]*bucketCleanupPolicyState // bucket path -> cleanup policy cache + // Folders deleted recently, kept so that a create event arriving for one of them + // can put it back + deleted map[string]*deletedFolder + deletedDropped int + // Cleanup queue (thread-safe, has its own lock) cleanupQueue *CleanupQueue @@ -83,6 +115,7 @@ func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRin host: host, folderCounts: make(map[string]*folderState), bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + deleted: make(map[string]*deletedFolder), cleanupQueue: NewCleanupQueue(DefaultQueueMaxSize, cleanupDelay), maxCountCheck: DefaultMaxCountCheck, cacheExpiry: DefaultCacheExpiry, @@ -194,6 +227,14 @@ func (efc *EmptyFolderCleaner) OnCreateEvent(directory string, entryName string, state.lastAddTime = time.Now() } + // An entry landing in a folder we just deleted is the race this cleaner cannot + // exclude: the folder was empty when checked and is gone now, so the entry has + // nothing holding it. The event says so outright, which beats going back to look. + if folder, found := efc.deleted[directory]; found { + folder.writtenTo = true + glog.V(2).Infof("EmptyFolderCleaner: %s was written to while being deleted, restoring it", directory) + } + // Remove from cleanup queue (cancel pending cleanup) if efc.cleanupQueue.Remove(directory) { glog.V(3).Infof("EmptyFolderCleaner: cancelled cleanup for %s due to new entry", directory) @@ -217,6 +258,8 @@ func (efc *EmptyFolderCleaner) cleanupProcessor() { // processCleanupQueue processes items from the cleanup queue func (efc *EmptyFolderCleaner) processCleanupQueue() { + efc.restoreFoldersWrittenDuringDelete() + if efc.cleanupQueue.Len() == 0 { return } @@ -242,6 +285,127 @@ func (efc *EmptyFolderCleaner) processCleanupQueue() { } } +// restoreFoldersWrittenDuringDelete puts back folders that received an entry between +// the emptiness check and the delete, which leaves that entry with no directory +// holding it: reachable by its own path, but absent from any listing. +// +// A folder stays under observation for DefaultRestoreCheckWindow rather than being +// checked once. A writer looks up the parent before inserting the child, so a check +// can land in that gap and see nothing; ticks also coalesce when a pass runs long, so +// "next pass" is not a delay at all. Re-checking for a bounded wall-clock window +// covers both. This narrows the exposure rather than closing it - only making the +// emptiness check and the delete atomic would do that. +func (efc *EmptyFolderCleaner) restoreFoldersWrittenDuringDelete() { + efc.mu.Lock() + dropped := efc.deletedDropped + efc.deletedDropped = 0 + var restore []*deletedFolder + for path, folder := range efc.deleted { + // The window applies whatever the folder's state is. Checking writtenTo first + // would keep a folder whose restore keeps failing forever, re-counting it on + // every pass. + if time.Since(folder.deletedAt) >= DefaultObservationWindow { + delete(efc.deleted, path) + continue + } + if folder.writtenTo { + restore = append(restore, folder) + delete(efc.deleted, path) + } + } + // A cascade takes ancestors along with the folder. Rebuild those from what they + // were too: leaving them to the descendant's restore would mint them from the + // descendant's own attributes, handing back access the ancestor did not grant. + for i := 0; i < len(restore); i++ { + ancestor, _ := util.FullPath(restore[i].path).DirAndName() + for ancestor != "" && ancestor != "/" { + if folder, found := efc.deleted[ancestor]; found { + restore = append(restore, folder) + delete(efc.deleted, ancestor) + } + ancestor, _ = util.FullPath(ancestor).DirAndName() + } + } + efc.mu.Unlock() + + if dropped > 0 { + glog.V(1).Infof("EmptyFolderCleaner: %d deleted folders left unobserved, past the %d kept", dropped, DefaultMaxDeletedKept) + } + if len(restore) == 0 { + return + } + + // Restore shallowest first, so a folder taken by the parent cascade is rebuilt + // with its own attributes before anything below it needs it as a parent. + sort.Slice(restore, func(i, j int) bool { + return strings.Count(restore[i].path, "/") < strings.Count(restore[j].path, "/") + }) + + ctx := context.Background() + var retry []*deletedFolder + for i, folder := range restore { + if !efc.IsEnabled() { + retry = append(retry, restore[i:]...) + break + } + // An event named this folder, but the entry may have been removed again since, + // in which case there is nothing to hold and it can stay gone. Ancestors pulled + // in above carry no event and are rebuilt regardless, since the folder below + // them needs them. + if folder.writtenTo { + count, err := efc.countItems(ctx, folder.path) + if err != nil { + glog.V(2).Infof("EmptyFolderCleaner: cannot count %s before restoring it: %v", folder.path, err) + retry = append(retry, folder) + continue + } + if count == 0 { + continue + } + } + glog.V(1).Infof("EmptyFolderCleaner: restoring %s, written to while it was being deleted", folder.path) + if err := efc.filer.EnsureDirectoryEntry(ctx, util.FullPath(folder.path), folder.attrs); err != nil { + glog.V(2).Infof("EmptyFolderCleaner: failed to restore %s: %v", folder.path, err) + retry = append(retry, folder) + } + } + + if len(retry) == 0 { + return + } + efc.mu.Lock() + for _, folder := range retry { + if _, found := efc.deleted[folder.path]; !found { + efc.makeRoomForDeletedLocked() + } + efc.deleted[folder.path] = folder + } + efc.mu.Unlock() +} + +// makeRoomForDeletedLocked drops the oldest of a small sample when the set is full. +// The newest folders are the ones whose race is still live, so they must not be the +// ones given up; sampling keeps this cheap under heavy deletion rates. +func (efc *EmptyFolderCleaner) makeRoomForDeletedLocked() { + if len(efc.deleted) < DefaultMaxDeletedKept { + return + } + const sampleSize = 32 + oldestPath, seen := "", 0 + for path, folder := range efc.deleted { + if oldestPath == "" || folder.deletedAt.Before(efc.deleted[oldestPath].deletedAt) { + oldestPath = path + } + if seen++; seen >= sampleSize { + break + } + } + if oldestPath != "" { + delete(efc.deleted, oldestPath) + efc.deletedDropped++ + } +} + // executeCleanup performs the actual cleanup of an empty folder func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string) { // The bucket-shared .uploads staging tree holds in-progress multipart uploads. @@ -324,7 +488,25 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string) return } - // Delete the empty folder + // Read what it would take to put this folder back before removing it; without + // that a restore would have to invent attributes for it. + attrs, err := efc.filer.DirectoryAttributes(ctx, util.FullPath(folder)) + if err != nil { + glog.V(2).Infof("EmptyFolderCleaner: cannot read %s before deleting it: %v", folder, err) + return + } + + // Observe it before the delete rather than after. A delete can fail partway and + // still leave the folder gone - the redis stores remove the folder before their + // parent-list member - so a failure return is not proof that it is still there. + efc.mu.Lock() + if efc.deleted == nil { + efc.deleted = make(map[string]*deletedFolder) + } + efc.makeRoomForDeletedLocked() + efc.deleted[folder] = &deletedFolder{path: folder, attrs: attrs, deletedAt: time.Now()} + efc.mu.Unlock() + glog.Infof("EmptyFolderCleaner: deleting empty folder %s (triggered by %s)", folder, triggeredBy) if err := efc.deleteFolder(ctx, folder); err != nil { glog.V(2).Infof("EmptyFolderCleaner: failed to delete empty folder %s (triggered by %s): %v", folder, triggeredBy, err) @@ -539,6 +721,7 @@ func (efc *EmptyFolderCleaner) Stop() { efc.cleanupQueue.Clear() efc.folderCounts = make(map[string]*folderState) // Clear cache on stop efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState) + efc.deleted, efc.deletedDropped = make(map[string]*deletedFolder), 0 } // GetPendingCleanupCount returns the number of pending cleanup tasks (for testing) diff --git a/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go b/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go index 58e6d50e7..798b4ca66 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go @@ -2,6 +2,7 @@ package empty_folder_cleanup import ( "context" + "errors" "testing" "time" @@ -16,6 +17,22 @@ type mockFilerOps struct { deleteFn func(path util.FullPath) error attrsFn func(path util.FullPath) (map[string][]byte, error) isDirKeyObjFn func(path util.FullPath) (bool, error) + ensureDirFn func(path util.FullPath, attrs DirectoryAttributes) error + dirAttrsFn func(path util.FullPath) (DirectoryAttributes, error) +} + +func (m *mockFilerOps) EnsureDirectoryEntry(_ context.Context, p util.FullPath, attrs DirectoryAttributes) error { + if m.ensureDirFn == nil { + return nil + } + return m.ensureDirFn(p, attrs) +} + +func (m *mockFilerOps) DirectoryAttributes(_ context.Context, p util.FullPath) (DirectoryAttributes, error) { + if m.dirAttrsFn == nil { + return DirectoryAttributes{Mode: 0755}, nil + } + return m.dirAttrsFn(p) } func (m *mockFilerOps) CountDirectoryEntries(_ context.Context, dirPath util.FullPath, _ int) (int, error) { @@ -200,6 +217,222 @@ func TestEmptyFolderCleaner_executeCleanup_skipsMultipartUploads(t *testing.T) { } } +func TestEmptyFolderCleaner_restoreFoldersWrittenDuringDelete(t *testing.T) { + lockRing := lock_manager.NewLockRing(5 * time.Second) + lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0) + + newCleaner := func(mock *mockFilerOps) *EmptyFolderCleaner { + return &EmptyFolderCleaner{ + filer: mock, + lockRing: lockRing, + host: "filer1:8888", + bucketPath: "/buckets", + enabled: true, + maxCountCheck: DefaultMaxCountCheck, + cacheExpiry: DefaultCacheExpiry, + folderCounts: make(map[string]*folderState), + bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState), + deleted: make(map[string]*deletedFolder), + cleanupQueue: NewCleanupQueue(1000, 10*time.Minute), + stopCh: make(chan struct{}), + } + } + + const folder = "/buckets/mybucket/folder" + + t.Run("a create event for the deleted folder puts it back", func(t *testing.T) { + var restored []string + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + + // nothing has been written to it, so it is only observed + cleaner.processCleanupQueue() + if len(restored) != 0 { + t.Fatalf("a folder nothing was written to should not be restored, got %v", restored) + } + if len(cleaner.deleted) != 1 { + t.Fatalf("folder should still be observed, got %d", len(cleaner.deleted)) + } + + // the write that raced the delete arrives as a create event + mock.countFn = func(util.FullPath) (int, error) { return 1, nil } + cleaner.OnCreateEvent(folder, "obj", false) + cleaner.processCleanupQueue() + if len(restored) != 1 || restored[0] != folder { + t.Fatalf("a folder written to while being deleted should be restored, got %v", restored) + } + if len(cleaner.deleted) != 0 { + t.Fatalf("a restored folder should stop being observed, got %d", len(cleaner.deleted)) + } + }) + + t.Run("a create event elsewhere is ignored", func(t *testing.T) { + var restored []string + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + cleaner.OnCreateEvent("/buckets/mybucket/other", "obj", false) + cleaner.processCleanupQueue() + if len(restored) != 0 { + t.Fatalf("only the folder written to should be restored, got %v", restored) + } + }) + + t.Run("an entry that went away again is not restored", func(t *testing.T) { + var restored []string + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + cleaner.OnCreateEvent(folder, "obj", false) + cleaner.processCleanupQueue() + if len(restored) != 0 { + t.Fatalf("an empty folder should not be restored, got %v", restored) + } + }) + + t.Run("observation expires", func(t *testing.T) { + mock := &mockFilerOps{countFn: func(util.FullPath) (int, error) { return 0, nil }} + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + cleaner.processCleanupQueue() + if len(cleaner.deleted) != 1 { + t.Fatalf("folder should still be observed inside the window, got %d", len(cleaner.deleted)) + } + + cleaner.deleted[folder].deletedAt = time.Now().Add(-DefaultObservationWindow - time.Second) + cleaner.processCleanupQueue() + if len(cleaner.deleted) != 0 { + t.Fatalf("folder should stop being observed once the window has passed, got %d", len(cleaner.deleted)) + } + }) + + t.Run("a failed restore is retried", func(t *testing.T) { + attempts := 0 + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 1, nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { + attempts++ + if attempts == 1 { + return errors.New("store unavailable") + } + return nil + }, + } + + cleaner := newCleaner(mock) + // deleted while empty, then written to + mock.countFn = func(util.FullPath) (int, error) { return 0, nil } + cleaner.executeCleanup(folder, "file.txt") + mock.countFn = func(util.FullPath) (int, error) { return 1, nil } + cleaner.OnCreateEvent(folder, "obj", false) + + cleaner.processCleanupQueue() + if len(cleaner.deleted) != 1 { + t.Fatalf("a folder whose restore failed should be kept, got %d", len(cleaner.deleted)) + } + + cleaner.processCleanupQueue() + if attempts != 2 { + t.Fatalf("a failed restore should be retried, got %d attempts", attempts) + } + if len(cleaner.deleted) != 0 { + t.Fatalf("a restored folder should stop being observed, got %d", len(cleaner.deleted)) + } + }) + + t.Run("a folder whose restore keeps failing stops being observed", func(t *testing.T) { + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + ensureDirFn: func(util.FullPath, DirectoryAttributes) error { return errors.New("store unavailable") }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + mock.countFn = func(util.FullPath) (int, error) { return 1, nil } + cleaner.OnCreateEvent(folder, "obj", false) + + cleaner.processCleanupQueue() + if len(cleaner.deleted) != 1 { + t.Fatalf("a folder whose restore failed should be kept, got %d", len(cleaner.deleted)) + } + + // otherwise it would be retried on every pass for the rest of the process + cleaner.deleted[folder].deletedAt = time.Now().Add(-DefaultObservationWindow - time.Second) + cleaner.processCleanupQueue() + if len(cleaner.deleted) != 0 { + t.Fatalf("the window should apply to a failing folder too, got %d", len(cleaner.deleted)) + } + }) + + t.Run("an ancestor taken by the cascade is restored with its own attributes", func(t *testing.T) { + const ancestor = "/buckets/mybucket/data" + const nested = ancestor + "/abc" + + restored := map[string]DirectoryAttributes{} + var order []string + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + dirAttrsFn: func(p util.FullPath) (DirectoryAttributes, error) { + if string(p) == ancestor { + return DirectoryAttributes{Mode: 0700, Uid: 4242}, nil + } + return DirectoryAttributes{Mode: 0755, Uid: 7}, nil + }, + ensureDirFn: func(p util.FullPath, attrs DirectoryAttributes) error { + restored[string(p)] = attrs + order = append(order, string(p)) + return nil + }, + } + + cleaner := newCleaner(mock) + // the cascade takes the nested folder and then its parent + cleaner.executeCleanup(nested, "file.txt") + cleaner.executeCleanup(ancestor, "abc") + + // only the nested folder is written to + mock.countFn = func(util.FullPath) (int, error) { return 1, nil } + cleaner.OnCreateEvent(nested, "obj", false) + cleaner.processCleanupQueue() + + if len(order) != 2 || order[0] != ancestor { + t.Fatalf("the ancestor should be rebuilt first, got %v", order) + } + if got := restored[ancestor]; got.Mode != 0700 || got.Uid != 4242 { + t.Errorf("ancestor should keep its own attributes, got mode %o uid %d", got.Mode, got.Uid) + } + }) + + t.Run("a delete that reports failure is still observed", func(t *testing.T) { + // the redis stores drop the folder before its parent-list member, so a failure + // return is not proof that the folder survived + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + deleteFn: func(util.FullPath) error { return errors.New("partial delete") }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "file.txt") + if len(cleaner.deleted) != 1 { + t.Fatalf("a failed delete should still leave the folder observed, got %d", len(cleaner.deleted)) + } + }) +} + func Test_autoRemoveEmptyFoldersEnabled(t *testing.T) { tests := []struct { name string diff --git a/weed/filer/filer.go b/weed/filer/filer.go index bd349af67..6730ae763 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -373,6 +373,65 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di return nil } +// DirectoryAttributes reads what a directory would need to be recreated as it is. +func (f *Filer) DirectoryAttributes(ctx context.Context, dirPath util.FullPath) (attrs empty_folder_cleanup.DirectoryAttributes, err error) { + entry, err := f.FindEntry(ctx, dirPath) + if err != nil { + return attrs, err + } + if entry == nil { + return attrs, filer_pb.ErrNotFound + } + return empty_folder_cleanup.DirectoryAttributes{ + // everything but the type bits: Perm() alone would drop setgid, setuid and + // sticky, quietly changing group inheritance and delete semantics + Mode: entry.Mode & (os.ModePerm | os.ModeSetuid | os.ModeSetgid | os.ModeSticky), + Uid: entry.Uid, + Gid: entry.Gid, + UserName: entry.UserName, + GroupNames: entry.GroupNames, + }, nil +} + +// EnsureDirectoryEntry recreates dirPath, and any missing ancestor, for entries that +// outlived the directory holding them. dirPath comes back with the attributes it was +// deleted with, so a restore cannot hand back a directory more permissive than the one +// it replaces. It is a no-op when dirPath is already there. +func (f *Filer) EnsureDirectoryEntry(ctx context.Context, dirPath util.FullPath, attrs empty_folder_cleanup.DirectoryAttributes) error { + existing, err := f.FindEntry(ctx, dirPath) + if err != nil && !errors.Is(err, filer_pb.ErrNotFound) { + return err + } + if existing != nil { + return nil + } + + holder := &Entry{FullPath: dirPath, Attr: Attr{ + Mode: attrs.Mode, Uid: attrs.Uid, Gid: attrs.Gid, + UserName: attrs.UserName, GroupNames: attrs.GroupNames, + }} + dirParts := strings.Split(string(dirPath), "/") + if err := f.ensureParentDirectoryEntry(ctx, holder, dirParts, len(dirParts)-1, false); err != nil { + return err + } + + now := time.Now() + dirEntry := &Entry{FullPath: dirPath, Attr: Attr{ + Mtime: now, Crtime: now, + Mode: os.ModeDir | attrs.Mode, + Uid: attrs.Uid, + Gid: attrs.Gid, + UserName: attrs.UserName, GroupNames: attrs.GroupNames, + }} + f.ensureEntryInode(dirEntry) + if err := f.Store.InsertEntry(ctx, dirEntry); err != nil { + return fmt.Errorf("restore directory %s: %v", dirPath, err) + } + f.NotifyUpdateEvent(ctx, nil, dirEntry, false, false, nil) + + return nil +} + func (f *Filer) UpdateEntry(ctx context.Context, oldEntry, entry *Entry) (err error) { if oldEntry != nil { entry.Attr.Crtime = oldEntry.Attr.Crtime diff --git a/weed/filer/leveldb2/empty_folder_race_test.go b/weed/filer/leveldb2/empty_folder_race_test.go index 431b0c908..5fa799643 100644 --- a/weed/filer/leveldb2/empty_folder_race_test.go +++ b/weed/filer/leveldb2/empty_folder_race_test.go @@ -76,3 +76,121 @@ func TestNonRecursiveFolderDeleteKeepsRacingChild(t *testing.T) { t.Errorf("folder entry should still be removed, got %v", err) } } + +// TestEnsureDirectoryEntryRestoresRacingParent covers the leftover of the same +// race: the entry survives the folder delete but its directory does not, so the +// entry drops out of listings until the directory is put back. +func TestEnsureDirectoryEntryRestoresRacingParent(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) + + ctx := filer.WithSuppressedMetadataEvents(context.Background()) + parent := util.FullPath("/buckets/testbucket/data") + dir := parent.Child("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) + } + + // the cleaner reads these before deleting, so a restore does not have to guess + attrs, err := testFiler.DirectoryAttributes(ctx, dir) + if err != nil { + t.Fatalf("read folder attributes: %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 names := listNames(ctx, t, testFiler, parent); len(names) != 0 { + t.Fatalf("folder should be gone from its parent before the restore, got %v", names) + } + + if err := testFiler.EnsureDirectoryEntry(ctx, dir, attrs); err != nil { + t.Fatalf("restore folder: %v", err) + } + + restored, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("find restored folder: %v", err) + } + if !restored.IsDirectory() { + t.Errorf("restored %s is not a directory", dir) + } + if names := listNames(ctx, t, testFiler, parent); len(names) != 1 || names[0] != "abc" { + t.Errorf("restored folder should be listed under its parent, got %v", names) + } +} + +// TestEnsureDirectoryEntryRestoresExactAttributes checks that a restored directory +// comes back exactly as it was. Guessing from an ancestor, or forcing traversal bits, +// would hand back a directory that grants access the original one denied. +func TestEnsureDirectoryEntryRestoresExactAttributes(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) + } + testFiler.SetStore(store) + + ctx := filer.WithSuppressedMetadataEvents(context.Background()) + // a private directory under a world-traversable parent + dir := util.FullPath("/buckets/testbucket/data").Child("private") + dirEntry := &filer.Entry{FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0700, Uid: 4242, Gid: 4343}} + if err := testFiler.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil { + t.Fatalf("create folder: %v", err) + } + + attrs, err := testFiler.DirectoryAttributes(ctx, dir) + if err != nil { + t.Fatalf("read folder attributes: %v", err) + } + if err := testFiler.DeleteEntryMetaAndData(ctx, dir, false, false, false, false, nil, 0); err != nil { + t.Fatalf("delete folder: %v", err) + } + + if err := testFiler.EnsureDirectoryEntry(ctx, dir, attrs); err != nil { + t.Fatalf("restore folder: %v", err) + } + + restored, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("find restored folder: %v", err) + } + if !restored.IsDirectory() { + t.Errorf("restored %s is not a directory", dir) + } + if restored.Mode.Perm() != 0700 { + t.Errorf("restored folder should keep mode 0700, got %o", restored.Mode.Perm()) + } + if restored.Uid != 4242 || restored.Gid != 4343 { + t.Errorf("restored folder should keep its owner, got uid=%d gid=%d", restored.Uid, restored.Gid) + } +} + +func listNames(ctx context.Context, t *testing.T, f *filer.Filer, dir util.FullPath) []string { + t.Helper() + entries, _, err := f.ListDirectoryEntries(ctx, dir, "", false, 100, "", "", "") + if err != nil { + t.Fatalf("list %s: %v", dir, err) + } + var names []string + for _, entry := range entries { + names = append(names, entry.Name()) + } + return names +} diff --git a/weed/filer/redis/universal_redis_store.go b/weed/filer/redis/universal_redis_store.go index 7ed5d3877..bec4294af 100644 --- a/weed/filer/redis/universal_redis_store.go +++ b/weed/filer/redis/universal_redis_store.go @@ -116,7 +116,8 @@ func (store *UniversalRedisStore) DeleteEntry(ctx context.Context, fullpath util func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) { - members, err := store.Client.SMembers(ctx, genDirectoryListKey(string(fullpath))).Result() + dirListKey := genDirectoryListKey(string(fullpath)) + members, err := store.Client.SMembers(ctx, dirListKey).Result() if err != nil { return fmt.Errorf("delete folder %s : %v", fullpath, err) } @@ -131,6 +132,10 @@ func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, full store.Client.Del(ctx, genDirectoryListKey(string(path))) } + if _, err = store.Client.Del(ctx, dirListKey).Result(); err != nil { + return fmt.Errorf("delete folder %s list: %v", fullpath, err) + } + return nil } diff --git a/weed/filer/redis/universal_redis_store_test.go b/weed/filer/redis/universal_redis_store_test.go index 99ed8e579..7f5f7d98d 100644 --- a/weed/filer/redis/universal_redis_store_test.go +++ b/weed/filer/redis/universal_redis_store_test.go @@ -153,3 +153,29 @@ func TestRemoveOrphanedDirectoryListMemberKeepsDirectoryWithChildren(t *testing. t.Fatalf("child value key exists=%d err=%v, want it kept", exists, err) } } + +func TestDeleteFolderChildrenRemovesTheListing(t *testing.T) { + store, dir := newTestStore(t) + ctx := context.Background() + + insertTestEntry(t, store, dir.Child("obj")) + + // The directory entry going away must not take the listing with it; nothing + // else records that the child is under this directory. + if err := store.DeleteEntry(ctx, dir); err != nil { + t.Fatalf("DeleteEntry %s: %v", dir, err) + } + if names := listNames(t, store, dir); len(names) != 1 || names[0] != "obj" { + t.Errorf("child should still be listed after the directory entry is deleted, got %v", names) + } + + // Deleting the children takes the listing with them, rather than leaving it behind. + if err := store.DeleteFolderChildren(ctx, dir); err != nil { + t.Fatalf("DeleteFolderChildren %s: %v", dir, err) + } + if n, err := store.Client.Exists(ctx, genDirectoryListKey(string(dir))).Result(); err != nil { + t.Fatalf("exists %s: %v", dir, err) + } else if n != 0 { + t.Errorf("listing key should be removed with the children, still present") + } +} diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go index e74d5fc4c..132049558 100644 --- a/weed/filer/redis2/universal_redis_store.go +++ b/weed/filer/redis2/universal_redis_store.go @@ -118,11 +118,10 @@ func (store *UniversalRedis2Store) FindEntry(ctx context.Context, fullpath util. func (store *UniversalRedis2Store) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) { - _, err = store.Client.Del(ctx, store.getKey(genDirectoryListKey(string(fullpath)))).Result() - if err != nil { - return fmt.Errorf("delete dir list %s : %v", fullpath, err) - } - + // The child listing is dropped by DeleteFolderChildren, together with the + // children it describes. Dropping it here would also discard an entry that + // arrived after the caller judged this directory empty, and nothing else + // records that the entry is there. _, err = store.Client.Del(ctx, store.getKey(string(fullpath))).Result() if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) @@ -148,7 +147,8 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful return nil } - members, err := store.Client.ZRangeByLex(ctx, store.getKey(genDirectoryListKey(string(fullpath))), &redis.ZRangeBy{ + dirListKey := store.getKey(genDirectoryListKey(string(fullpath))) + members, err := store.Client.ZRangeByLex(ctx, dirListKey, &redis.ZRangeBy{ Min: "-", Max: "+", }).Result() @@ -166,6 +166,10 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful store.Client.Del(ctx, store.getKey(genDirectoryListKey(string(path)))) } + if _, err = store.Client.Del(ctx, dirListKey).Result(); err != nil { + return fmt.Errorf("DeleteFolderChildren %s list: %v", fullpath, err) + } + return nil } diff --git a/weed/filer/redis2/universal_redis_store_test.go b/weed/filer/redis2/universal_redis_store_test.go index c2dfc47d8..1558538da 100644 --- a/weed/filer/redis2/universal_redis_store_test.go +++ b/weed/filer/redis2/universal_redis_store_test.go @@ -299,3 +299,33 @@ func TestDeleteIfUnchangedScriptOnlyDeletesSameBytes(t *testing.T) { t.Fatalf("deleted=%d err=%v, want -1 on a missing key", deleted, err) } } + +func TestDeleteEntryKeepsChildListing(t *testing.T) { + store, dir := newTestStore(t, "") + ctx := context.Background() + + insertTestEntry(t, store, dir.Child("obj"), 0) + + // The listing is the only record that the child sits under this directory, so + // removing the directory entry must leave it alone. An entry that arrived after + // the caller judged the directory empty would otherwise become unreachable. + if err := store.DeleteEntry(ctx, dir); err != nil { + t.Fatalf("DeleteEntry %s: %v", dir, err) + } + if names := listNames(t, store, dir); len(names) != 1 || names[0] != "obj" { + t.Errorf("child should still be listed after the directory entry is deleted, got %v", names) + } + + // DeleteFolderChildren owns that cleanup, and takes the listing with it + if err := store.DeleteFolderChildren(ctx, dir); err != nil { + t.Fatalf("DeleteFolderChildren %s: %v", dir, err) + } + if names := listNames(t, store, dir); len(names) != 0 { + t.Errorf("listing should be empty once the children are deleted, got %v", names) + } + if n, err := store.Client.Exists(ctx, store.getKey(genDirectoryListKey(string(dir)))).Result(); err != nil { + t.Fatalf("exists %s: %v", dir, err) + } else if n != 0 { + t.Errorf("listing key should be removed with the children, still present") + } +} diff --git a/weed/filer/redis3/ItemList.go b/weed/filer/redis3/ItemList.go index 7a700f435..5da735096 100644 --- a/weed/filer/redis3/ItemList.go +++ b/weed/filer/redis3/ItemList.go @@ -342,6 +342,10 @@ func (nl *ItemList) ListNames(startFrom string, visitNamesFn func(name string) b return nil } +func (nl *ItemList) IsEmpty() bool { + return nl.skipList.StartLevels[0] == nil +} + func (nl *ItemList) RemoteAllListElement() error { t := nl.skipList diff --git a/weed/filer/redis3/kv_directory_children.go b/weed/filer/redis3/kv_directory_children.go index 281b01195..e1b080b41 100644 --- a/weed/filer/redis3/kv_directory_children.go +++ b/weed/filer/redis3/kv_directory_children.go @@ -70,9 +70,21 @@ func removeChild(ctx context.Context, redisStore *UniversalRedis3Store, key stri return err } if !nameList.HasChanges() { + // Nothing to remove. If the list is empty anyway, its header outlived the + // removal of the last name - a delete that failed here before - so take it + // now rather than leaving the key behind for good. + if nameList.IsEmpty() { + return client.Del(ctx, key).Err() + } return nil } + // An emptied list reads the same as no list at all, and keeping the header would + // leave a key behind for every directory that is emptied. + if nameList.IsEmpty() { + return client.Del(ctx, key).Err() + } + if err := client.Set(ctx, key, nameList.ToBytes(), 0).Err(); err != nil { return err } diff --git a/weed/filer/redis3/universal_redis_store.go b/weed/filer/redis3/universal_redis_store.go index 84cd42908..abaf7e698 100644 --- a/weed/filer/redis3/universal_redis_store.go +++ b/weed/filer/redis3/universal_redis_store.go @@ -95,11 +95,10 @@ func (store *UniversalRedis3Store) FindEntry(ctx context.Context, fullpath util. func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) { - _, err = store.Client.Del(ctx, genDirectoryListKey(string(fullpath))).Result() - if err != nil { - return fmt.Errorf("delete dir list %s : %v", fullpath, err) - } - + // The child listing is dropped by DeleteFolderChildren, together with the + // children it describes. Dropping it here would also discard an entry that + // arrived after the caller judged this directory empty, and nothing else + // records that the entry is there. _, err = store.Client.Del(ctx, string(fullpath)).Result() if err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) @@ -118,7 +117,8 @@ func (store *UniversalRedis3Store) DeleteEntry(ctx context.Context, fullpath uti func (store *UniversalRedis3Store) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) { - return removeChildren(ctx, store, genDirectoryListKey(string(fullpath)), func(name string) error { + dirListKey := genDirectoryListKey(string(fullpath)) + if err = removeChildren(ctx, store, dirListKey, func(name string) error { path := util.NewFullPath(string(fullpath), name) _, err = store.Client.Del(ctx, string(path)).Result() if err != nil { @@ -127,8 +127,16 @@ func (store *UniversalRedis3Store) DeleteFolderChildren(ctx context.Context, ful // not efficient, but need to remove if it is a directory store.Client.Del(ctx, genDirectoryListKey(string(path))) return nil - }) + }); err != nil { + return err + } + // removeChildren clears the skip list nodes but leaves the list itself behind + if _, err = store.Client.Del(ctx, dirListKey).Result(); err != nil { + return fmt.Errorf("DeleteFolderChildren %s list: %v", fullpath, err) + } + + return nil } func (store *UniversalRedis3Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { diff --git a/weed/filer/redis3/universal_redis_store_test.go b/weed/filer/redis3/universal_redis_store_test.go new file mode 100644 index 000000000..895fc2609 --- /dev/null +++ b/weed/filer/redis3/universal_redis_store_test.go @@ -0,0 +1,129 @@ +package redis3 + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/go-redsync/redsync/v4" + goredis "github.com/go-redsync/redsync/v4/redis/goredis/v9" + "github.com/redis/go-redis/v9" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func newTestStore(t *testing.T) (*UniversalRedis3Store, util.FullPath) { + t.Helper() + + if os.Getenv("RUN_REDIS_TESTS") != "1" { + t.Skip("redis3 tests are disabled. Start a redis-server and set RUN_REDIS_TESTS=1 to enable, REDIS_ADDR defaults to 127.0.0.1:6379.") + } + + addr := os.Getenv("REDIS_ADDR") + if addr == "" { + addr = "127.0.0.1:6379" + } + + ctx := context.Background() + client := redis.NewClient(&redis.Options{Addr: addr}) + t.Cleanup(func() { + if err := client.Close(); err != nil { + t.Errorf("close redis client: %v", err) + } + }) + if err := client.Ping(ctx).Err(); err != nil { + t.Fatalf("connect to redis at %s: %v", addr, err) + } + + store := &UniversalRedis3Store{Client: client, redsync: redsync.New(goredis.NewPool(client))} + + dir := util.FullPath(fmt.Sprintf("/redis3_test_%d", time.Now().UnixNano())) + t.Cleanup(func() { + if err := store.DeleteFolderChildren(ctx, dir); err != nil { + t.Errorf("cleanup %s children: %v", dir, err) + } + if err := store.DeleteEntry(ctx, dir); err != nil { + t.Errorf("cleanup %s: %v", dir, err) + } + }) + + return store, dir +} + +func listNames(t *testing.T, store *UniversalRedis3Store, dir util.FullPath) []string { + t.Helper() + + names := []string{} + if _, err := store.ListDirectoryEntries(context.Background(), dir, "", true, 100, func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }); err != nil { + t.Fatalf("list %s: %v", dir, err) + } + return names +} + +func TestDeleteEntryKeepsChildListing(t *testing.T) { + store, dir := newTestStore(t) + ctx := context.Background() + + now := time.Now() + child := dir.Child("obj") + if err := store.InsertEntry(ctx, &filer.Entry{ + FullPath: child, + Attr: filer.Attr{Crtime: now, Mtime: now, Mode: 0644}, + }); err != nil { + t.Fatalf("InsertEntry %s: %v", child, err) + } + + // The listing is the only record that the child sits under this directory, so + // removing the directory entry must leave it alone. An entry that arrived after + // the caller judged the directory empty would otherwise become unreachable. + if err := store.DeleteEntry(ctx, dir); err != nil { + t.Fatalf("DeleteEntry %s: %v", dir, err) + } + if names := listNames(t, store, dir); len(names) != 1 || names[0] != "obj" { + t.Errorf("child should still be listed after the directory entry is deleted, got %v", names) + } + + // DeleteFolderChildren owns that cleanup, and takes the listing with it + if err := store.DeleteFolderChildren(ctx, dir); err != nil { + t.Fatalf("DeleteFolderChildren %s: %v", dir, err) + } + if names := listNames(t, store, dir); len(names) != 0 { + t.Errorf("listing should be empty once the children are deleted, got %v", names) + } + if n, err := store.Client.Exists(ctx, genDirectoryListKey(string(dir))).Result(); err != nil { + t.Fatalf("exists %s: %v", dir, err) + } else if n != 0 { + t.Errorf("listing key should be removed with the children, still present") + } +} + +func TestRemovingTheLastChildDropsTheListing(t *testing.T) { + store, dir := newTestStore(t) + ctx := context.Background() + + now := time.Now() + child := dir.Child("obj") + if err := store.InsertEntry(ctx, &filer.Entry{ + FullPath: child, + Attr: filer.Attr{Crtime: now, Mtime: now, Mode: 0644}, + }); err != nil { + t.Fatalf("InsertEntry %s: %v", child, err) + } + + // An emptied listing must not leave its header behind: cleanup deletes such a + // folder without touching the listing, so the key would never be collected. + if err := store.DeleteEntry(ctx, child); err != nil { + t.Fatalf("DeleteEntry %s: %v", child, err) + } + if n, err := store.Client.Exists(ctx, genDirectoryListKey(string(dir))).Result(); err != nil { + t.Fatalf("exists %s: %v", dir, err) + } else if n != 0 { + t.Errorf("listing key should be gone once the last child is removed, still present") + } +}