diff --git a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go index 9bcd2a3cc..22aad0b4b 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner.go @@ -383,6 +383,39 @@ func (efc *EmptyFolderCleaner) restoreFoldersWrittenDuringDelete() { efc.mu.Unlock() } +// restoreIfWrittenTo puts folder back when an entry landed in it while it was being +// deleted. True means the parent is no longer empty, so the caller must not cascade. +func (efc *EmptyFolderCleaner) restoreIfWrittenTo(ctx context.Context, folder string, attrs DirectoryAttributes) bool { + count, err := efc.countItems(ctx, folder) + if err != nil || count > 0 { + // Ask for the retry here rather than leaving it to a create event that may + // already have gone by: this pass can be the only sight of the entry. + efc.mu.Lock() + if observed, found := efc.deleted[folder]; found { + observed.writtenTo = true + } + efc.mu.Unlock() + } + if err != nil { + glog.V(2).Infof("EmptyFolderCleaner: cannot re-check %s after deleting it: %v", folder, err) + return false + } + if count == 0 { + return false + } + + glog.V(1).Infof("EmptyFolderCleaner: restoring %s, written to while it was being deleted", folder) + if err := efc.filer.EnsureDirectoryEntry(ctx, util.FullPath(folder), attrs); err != nil { + glog.V(2).Infof("EmptyFolderCleaner: failed to restore %s: %v", folder, err) + return false + } + + efc.mu.Lock() + delete(efc.deleted, folder) + efc.mu.Unlock() + return true +} + // 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. @@ -518,6 +551,12 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string) delete(efc.folderCounts, folder) efc.mu.Unlock() + // The delete's own emptiness check and the entry removal are not atomic either. + // Paired with parents being created after the insert, whoever acts second sees it. + if efc.restoreIfWrittenTo(ctx, folder, attrs) { + return + } + // After deleting this folder, immediately try to clean the parent. // Relying solely on cascading metadata events would re-enter the full // delay queue for each ancestor level, causing multi-minute cascading 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 798b4ca66..d37305b5e 100644 --- a/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go +++ b/weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go @@ -1140,3 +1140,128 @@ func TestEmptyFolderCleaner_executeCleanup_directoryMarker(t *testing.T) { }) } } + +func TestEmptyFolderCleaner_restoreIfWrittenTo(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/a/b" + const parent = "/buckets/mybucket/a" + + t.Run("an entry that landed during the delete is restored without waiting for its event", func(t *testing.T) { + counts := map[string]int{} + var deleted, restored []string + mock := &mockFilerOps{ + countFn: func(p util.FullPath) (int, error) { return counts[string(p)], nil }, + deleteFn: func(p util.FullPath) error { + deleted = append(deleted, string(p)) + if string(p) == folder { + // the racing write lands between the delete's own emptiness + // check and the removal of the folder entry + counts[folder] = 1 + } + return nil + }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "obj") + + if len(restored) != 1 || restored[0] != folder { + t.Fatalf("folder written to during the delete should be restored, got %v", restored) + } + if len(deleted) != 1 || deleted[0] != folder { + t.Fatalf("a restored folder leaves the parent non-empty, so it must not cascade, got %v", deleted) + } + if len(cleaner.deleted) != 0 { + t.Fatalf("a restored folder should stop being observed, got %d", len(cleaner.deleted)) + } + }) + + t.Run("a folder still empty after the delete is not restored and the parent is cascaded", func(t *testing.T) { + var deleted, restored []string + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { return 0, nil }, + deleteFn: func(p util.FullPath) error { deleted = append(deleted, string(p)); return nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "obj") + + if len(restored) != 0 { + t.Fatalf("an empty folder should not be restored, got %v", restored) + } + if len(deleted) != 2 || deleted[0] != folder || deleted[1] != parent { + t.Fatalf("the parent should still be cascaded, got %v", deleted) + } + }) + + t.Run("a restore that fails is left marked for the deferred pass", func(t *testing.T) { + counts := map[string]int{} + mock := &mockFilerOps{ + countFn: func(p util.FullPath) (int, error) { return counts[string(p)], nil }, + deleteFn: func(p util.FullPath) error { + if string(p) == folder { + counts[folder] = 1 + } + return nil + }, + ensureDirFn: func(util.FullPath, DirectoryAttributes) error { return errors.New("store unavailable") }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "obj") + + observed, found := cleaner.deleted[folder] + if !found { + t.Fatal("a folder whose restore failed must stay under observation") + } + if !observed.writtenTo { + t.Fatal("a folder seen to hold an entry must be marked, or the deferred pass skips it") + } + }) + + t.Run("a re-check that fails leaves the folder to the event path", func(t *testing.T) { + var restored []string + deletedFolder := false + mock := &mockFilerOps{ + countFn: func(util.FullPath) (int, error) { + if deletedFolder { + return 0, errors.New("store unavailable") + } + return 0, nil + }, + deleteFn: func(p util.FullPath) error { deletedFolder = string(p) == folder; return nil }, + ensureDirFn: func(p util.FullPath, _ DirectoryAttributes) error { restored = append(restored, string(p)); return nil }, + } + + cleaner := newCleaner(mock) + cleaner.executeCleanup(folder, "obj") + + if len(restored) != 0 { + t.Fatalf("a failed re-check should not restore, got %v", restored) + } + if _, found := cleaner.deleted[folder]; !found { + t.Fatal("a folder whose re-check failed must stay under observation") + } + }) +} diff --git a/weed/filer/filer.go b/weed/filer/filer.go index 6730ae763..069db1970 100644 --- a/weed/filer/filer.go +++ b/weed/filer/filer.go @@ -253,18 +253,28 @@ func (f *Filer) CreateEntry(ctx context.Context, entry *Entry, existing *Entry, if oldEntry == nil { f.ensureEntryInode(entry) - if !skipCreateParentDir { - dirParts := strings.Split(string(entry.FullPath), "/") - if err := f.ensureParentDirectoryEntry(ctx, entry, dirParts, len(dirParts)-1, isFromOtherCluster); err != nil { - return err - } - } - glog.V(4).InfofCtx(ctx, "InsertEntry %s: new entry: %v", entry.FullPath, entry.Name()) if err := f.Store.InsertEntry(ctx, entry); err != nil { glog.ErrorfCtx(ctx, "insert entry %s: %v", entry.FullPath, err) return fmt.Errorf("insert entry %s: %v", entry.FullPath, err) } + + // Parents go after the entry: one checked first can be taken by the + // empty-folder cleaner before the entry lands, and nothing would look again. + if !skipCreateParentDir { + dirParts := strings.Split(string(entry.FullPath), "/") + if err := f.ensureParentDirectoryEntry(ctx, entry, dirParts, len(dirParts)-1, isFromOtherCluster); err != nil { + // The entry stays: deleting by path would destroy a concurrent create + // that already succeeded through the update branch, and the update keeps + // the inode and crtime while mtime only survives the store to the second. + // It is not announced - the aggregator replicates creates into peer + // stores, so a failed write would land on all of them rather than on the + // one filer whose next write into that folder repairs it. + glog.ErrorfCtx(ctx, "create parent directories of %s: %v", entry.FullPath, err) + return err + } + } + if !entry.IsDirectory() { stats.FilerObjectSizeBytesHistogram.Observe(float64(entry.Size())) } @@ -373,6 +383,10 @@ func (f *Filer) ensureParentDirectoryEntry(ctx context.Context, entry *Entry, di return nil } +// restorableModeBits is everything but the type bits: ModePerm alone would drop +// setgid, setuid and sticky, quietly changing group inheritance and delete semantics. +const restorableModeBits = os.ModePerm | os.ModeSetuid | os.ModeSetgid | os.ModeSticky + // 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) @@ -383,9 +397,7 @@ func (f *Filer) DirectoryAttributes(ctx context.Context, dirPath util.FullPath) 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), + Mode: entry.Mode & restorableModeBits, Uid: entry.Uid, Gid: entry.Gid, UserName: entry.UserName, @@ -396,13 +408,27 @@ func (f *Filer) DirectoryAttributes(ctx context.Context, dirPath util.FullPath) // 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. +// it replaces - including when someone else already put it back with wider ones. 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 { + // A writer recreating its own missing parent infers the mode from the entry it + // is inserting, so it can come back wider. Intersect rather than replace, or a + // mode holding a bit the saved one lacks would be granted the ones it lacks. + kept := existing.Mode & attrs.Mode & restorableModeBits + if existing.Mode&restorableModeBits == kept { + return nil + } + narrowed := existing.ShallowClone() + narrowed.Mode = existing.Mode&^restorableModeBits | kept + glog.V(1).InfofCtx(ctx, "restore directory %s: narrowing %v to %v", dirPath, existing.Mode, narrowed.Mode) + if err := f.UpdateEntry(ctx, existing, narrowed); err != nil { + return err + } + f.NotifyUpdateEvent(ctx, existing, narrowed, false, false, nil) return nil } diff --git a/weed/filer/leveldb2/empty_folder_race_test.go b/weed/filer/leveldb2/empty_folder_race_test.go index 5fa799643..a87267603 100644 --- a/weed/filer/leveldb2/empty_folder_race_test.go +++ b/weed/filer/leveldb2/empty_folder_race_test.go @@ -7,29 +7,40 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/filer/empty_folder_cleanup" "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 { +// hookStore runs a hook once, right after the store call it is attached to returns, +// so a test can act inside a window that is otherwise not reachable. +type hookStore struct { filer.FilerStore - hook func() - fired bool + afterList func() + listFired bool + afterInsert func(entry *filer.Entry) + insertFired bool } -func (s *listHookStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (string, error) { +func (s *hookStore) 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() + if s.afterList != nil && !s.listFired { + s.listFired = true + s.afterList() } return lastFileName, err } +func (s *hookStore) InsertEntry(ctx context.Context, entry *filer.Entry) error { + err := s.FilerStore.InsertEntry(ctx, entry) + if s.afterInsert != nil && !s.insertFired { + s.insertFired = true + s.afterInsert(entry) + } + return 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 @@ -40,7 +51,7 @@ func TestNonRecursiveFolderDeleteKeepsRacingChild(t *testing.T) { if err := store.initialize(t.TempDir(), 2); err != nil { t.Fatal(err) } - hooked := &listHookStore{FilerStore: store} + hooked := &hookStore{FilerStore: store} testFiler.SetStore(hooked) // the test has no metadata log consumer @@ -53,7 +64,7 @@ func TestNonRecursiveFolderDeleteKeepsRacingChild(t *testing.T) { t.Fatalf("create folder: %v", err) } - hooked.hook = func() { + hooked.afterList = 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) @@ -86,7 +97,7 @@ func TestEnsureDirectoryEntryRestoresRacingParent(t *testing.T) { if err := store.initialize(t.TempDir(), 2); err != nil { t.Fatal(err) } - hooked := &listHookStore{FilerStore: store} + hooked := &hookStore{FilerStore: store} testFiler.SetStore(hooked) ctx := filer.WithSuppressedMetadataEvents(context.Background()) @@ -105,7 +116,7 @@ func TestEnsureDirectoryEntryRestoresRacingParent(t *testing.T) { t.Fatalf("read folder attributes: %v", err) } - hooked.hook = func() { + hooked.afterList = 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) @@ -182,6 +193,223 @@ func TestEnsureDirectoryEntryRestoresExactAttributes(t *testing.T) { } } +// TestCreateEntryCreatesParentAfterTheEntry covers the writer's half of the pair: the +// folder is taken while the entry is landing, which used to orphan it. +func TestCreateEntryCreatesParentAfterTheEntry(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 := &hookStore{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 found the folder empty a moment ago and removes it now + hooked.afterInsert = func(entry *filer.Entry) { + if entry.FullPath != child { + return + } + if err := store.DeleteEntry(ctx, dir); err != nil { + t.Errorf("delete folder racing the insert: %v", err) + } + } + + 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.Fatalf("create entry racing the folder delete: %v", err) + } + + if _, err := testFiler.FindEntry(ctx, child); err != nil { + t.Fatalf("entry created during the folder delete is gone: %v", err) + } + restored, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("parent taken during the insert should be created after it: %v", err) + } + if !restored.IsDirectory() { + t.Errorf("recreated %s is not a directory", dir) + } + if names := listNames(ctx, t, testFiler, parent); len(names) != 1 || names[0] != "abc" { + t.Errorf("entry should be reachable by listing again, got %v", names) + } +} + +// TestCreateEntryKeepsTheEntryWhenTheParentCannotBeCreated pins the cost of going +// second: the entry is already stored when the parent fails, and it is left there, +// since nothing can tell it apart from a concurrent create that has succeeded. +func TestCreateEntryKeepsTheEntryWhenTheParentCannotBeCreated(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()) + // an invalid bucket name is rejected while the parents are being created + child := util.FullPath("/buckets/Not_A_Valid_Bucket/obj") + + 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.Fatal("an invalid bucket name should still fail the create") + } + + if _, err := testFiler.FindEntry(ctx, child); err != nil { + t.Errorf("the entry is kept rather than risking a concurrent create: %v", err) + } +} + +// TestCreateEntryDoesNotAnnounceAnEntryWhoseParentFailed pins the other half of keeping +// the entry: it stays off the log. The aggregator replicates creates into peer stores, +// so announcing a create that failed would put the parentless row on every filer. +func TestCreateEntryDoesNotAnnounceAnEntryWhoseParentFailed(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, sink := filer.WithMetadataEventSink(context.Background()) + child := util.FullPath("/buckets/Not_A_Valid_Bucket/obj") + + 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.Fatal("an invalid bucket name should still fail the create") + } + + if event := sink.Last(); event != nil { + t.Errorf("a create that failed must not be announced, got %+v", event.EventNotification) + } + if _, err := testFiler.FindEntry(ctx, child); err != nil { + t.Errorf("the entry itself is still kept: %v", err) + } +} + +// TestEnsureDirectoryEntryNarrowsAWiderRestore covers a writer getting to the missing +// parent first and minting it wider than the one that was deleted. +func TestEnsureDirectoryEntryNarrowsAWiderRestore(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()) + 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) + } + + // the writer's own entry is 0640, so the parent it mints comes back 0751 + entry := &filer.Entry{FullPath: dir.Child("obj"), Attr: filer.Attr{Mode: 0640}} + if err := testFiler.CreateEntry(ctx, entry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil { + t.Fatalf("create entry: %v", err) + } + widened, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("find minted folder: %v", err) + } + if widened.Mode.Perm() == 0700 { + t.Skip("the writer no longer widens the parent, nothing left to narrow") + } + + 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("restore should hand back mode 0700, got %o", restored.Mode.Perm()) + } + if _, err := testFiler.FindEntry(ctx, entry.FullPath); err != nil { + t.Errorf("narrowing the folder should not disturb what it holds: %v", err) + } +} + +// TestEnsureDirectoryEntryKeepsATightenedDirectory checks the other direction: a +// restore only ever narrows, so a deliberate tightening is left alone. +func TestEnsureDirectoryEntryKeepsATightenedDirectory(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()) + dir := util.FullPath("/buckets/testbucket/data").Child("tightened") + dirEntry := &filer.Entry{FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0700}} + if err := testFiler.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil { + t.Fatalf("create folder: %v", err) + } + + if err := testFiler.EnsureDirectoryEntry(ctx, dir, empty_folder_cleanup.DirectoryAttributes{Mode: 0755}); err != nil { + t.Fatalf("restore folder: %v", err) + } + kept, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("find folder: %v", err) + } + if kept.Mode.Perm() != 0700 { + t.Errorf("a directory tightened since should keep mode 0700, got %o", kept.Mode.Perm()) + } +} + +// TestEnsureDirectoryEntryIntersectsMixedModes covers modes that are not nested, where +// replacing rather than intersecting would grant a bit the directory denied. +func TestEnsureDirectoryEntryIntersectsMixedModes(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()) + dir := util.FullPath("/buckets/testbucket/data").Child("mixed") + dirEntry := &filer.Entry{FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0705}} + if err := testFiler.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, testFiler.MaxFilenameLength); err != nil { + t.Fatalf("create folder: %v", err) + } + + // 0750 grants group access that 0705 denies, and denies the other access it grants + if err := testFiler.EnsureDirectoryEntry(ctx, dir, empty_folder_cleanup.DirectoryAttributes{Mode: 0750}); err != nil { + t.Fatalf("restore folder: %v", err) + } + restored, err := testFiler.FindEntry(ctx, dir) + if err != nil { + t.Fatalf("find folder: %v", err) + } + if restored.Mode.Perm() != 0700 { + t.Errorf("restore should grant only what both modes allow, got %o", restored.Mode.Perm()) + } +} + 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, "", "", "")