filer: close the empty-folder race by checking after each mutation (#10799)

* filer: re-list a folder after deleting it, and put it back if it is not empty

The emptiness check inside the delete and the removal of the folder entry are
not atomic, so an entry can land between them and be left reachable by its own
path but out of every listing. Looking again after the delete catches the ones
whose create event has not arrived yet, and does not depend on the event stream
or on the observation window holding.

* filer: create the directories holding an entry after the entry

A parent checked before the insert can be taken by the empty-folder cleaner
before the entry lands, which leaves the entry reachable by its own path but out
of every listing. Creating the parents afterwards cannot be undone by a delete
that was authorised before the insert, and pairs with the cleaner re-listing
after its own delete: whichever of the two acts second sees what the other did.

Going second means the entry is already stored when the parent fails, so it is
taken back out and the caller still sees the error it used to get.

* filer: narrow a directory that came back wider than the one it replaced

A writer recreating its own missing parent has only the entry it is inserting to
go on, so the directory it mints can grant access the deleted one denied - a
0700 folder comes back 0751. The cleaner read the real attributes before
deleting, so its restore now puts the original mode back instead of leaving the
inferred one in place. It only ever narrows, so a directory deliberately
tightened since is left as it is.

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

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")
+		}
+	})
+}

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
 	}
 

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, "", "", "")