mount/shell: bucket allow-empty-folders toggle, mount keeps explicit false (#11370)

* mount: keep a deliberate bucket allow-empty-folders setting

* shell: s3.bucket.allowEmptyFolders toggles the empty folder cleaner

* shell: guard allow-empty-folders toggle with expected extended attrs

* filer: drop cached empty-folder policy on bucket entry update

* mount: guard allow-empty-folders write with expected extended attrs

* filer: skip caching a stale cleanup policy read across an update

* mount, shell: snapshot the full extended attributes for update preconditions

* filer: fail closed and invalidate on all bucket entry events for cleanup policy

* filer: key the cleanup policy generation by bucket

* filer: requeue cleanup when the bucket policy cannot be loaded

* filer: expire idle cleanup policy generations

* filer: skip requeueing cleanup after the cleaner stops

* filer: bound cleanup retries on repeated policy failures

* filer: cover cleanup requeue on repeated policy failures

* filer: keep cleanup policy generations while reads are in flight

* filer: exercise the cleanup queue lifecycle in the retry-cap test

weed/command/mount_common.go

@@ -17,7 +17,9 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/reflection"
+	"google.golang.org/grpc/status"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mount"
@@ -67,36 +69,46 @@ func ensureBucketAllowEmptyFolders(ctx context.Context, filerClient filer_pb.Fil
 		return nil
 	}
 
-	entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath))
-	if err != nil {
-		return err
-	}
-	if entry == nil {
-		return fmt.Errorf("bucket %s not found", bucketPath)
-	}
-
-	if entry.Extended == nil {
-		entry.Extended = make(map[string][]byte)
-	}
-	if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") {
-		return nil
-	}
-
-	entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true")
-
 	bucketFullPath := util.FullPath(bucketPath)
 	parent, _ := bucketFullPath.DirAndName()
-	if err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		return filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{
-			Directory: parent,
-			Entry:     entry,
+
+	for attempt := 0; attempt < 5; attempt++ {
+		entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(bucketPath))
+		if err != nil {
+			return err
+		}
+		if entry == nil {
+			return fmt.Errorf("bucket %s not found", bucketPath)
+		}
+
+		if value := strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])); value != "" {
+			return nil
+		}
+
+		expected := filer_pb.SnapshotExtended(entry.Extended, s3_constants.ExtAllowEmptyFolders)
+		if entry.Extended == nil {
+			entry.Extended = make(map[string][]byte)
+		}
+		entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true")
+
+		err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+			_, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+				Directory:        parent,
+				Entry:            entry,
+				ExpectedExtended: expected,
+			})
+			return err
 		})
-	}); err != nil {
-		return err
+		if err == nil {
+			glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders)
+			return nil
+		}
+		if status.Code(err) != codes.FailedPrecondition {
+			return err
+		}
 	}
 
-	glog.V(3).Infof("RunMount: set bucket %s %s=true", bucketPath, s3_constants.ExtAllowEmptyFolders)
-	return nil
+	return fmt.Errorf("bucket %s attributes keep changing", bucketPath)
 }
 
 func bucketPathForMountRoot(mountRoot, bucketRootPath string) (string, bool) {

weed/command/mount_common_test.go

@@ -2,7 +2,18 @@
 
 package command
 
-import "testing"
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+)
 
 func Test_volumeName(t *testing.T) {
 	tests := []struct {
@@ -102,3 +113,103 @@ func Test_volumeName(t *testing.T) {
 		})
 	}
 }
+
+type allowEmptyFoldersInnerClient struct {
+	filer_pb.SeaweedFilerClient
+	entry          *filer_pb.Entry
+	beforeUpdate   func()
+	lookups        int
+	updateAttempts int
+	updates        []*filer_pb.Entry
+}
+
+func (c *allowEmptyFoldersInnerClient) LookupDirectoryEntry(_ context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.lookups++
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(c.entry).(*filer_pb.Entry)}, nil
+}
+
+func (c *allowEmptyFoldersInnerClient) UpdateEntry(_ context.Context, in *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
+	c.updateAttempts++
+	if c.beforeUpdate != nil {
+		c.beforeUpdate()
+	}
+	for key, expected := range in.ExpectedExtended {
+		actual, found := c.entry.Extended[key]
+		if found && !bytes.Equal(actual, expected) || !found && len(expected) > 0 {
+			return nil, status.Error(codes.FailedPrecondition, "extended attribute changed")
+		}
+	}
+	c.entry = proto.Clone(in.Entry).(*filer_pb.Entry)
+	c.updates = append(c.updates, in.Entry)
+	return &filer_pb.UpdateEntryResponse{}, nil
+}
+
+type allowEmptyFoldersFilerClient struct {
+	inner *allowEmptyFoldersInnerClient
+}
+
+func (c *allowEmptyFoldersFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c.inner)
+}
+func (c *allowEmptyFoldersFilerClient) AdjustedUrl(_ *filer_pb.Location) string { return "" }
+func (c *allowEmptyFoldersFilerClient) GetDataCenter() string                   { return "" }
+
+func Test_ensureBucketAllowEmptyFolders(t *testing.T) {
+	bucketEntry := func(attrValue string) *filer_pb.Entry {
+		entry := &filer_pb.Entry{Name: "b1", IsDirectory: true, Extended: map[string][]byte{}}
+		if attrValue != "" {
+			entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte(attrValue)
+		}
+		return entry
+	}
+
+	tests := []struct {
+		name          string
+		mountRoot     string
+		attrValue     string
+		beforeUpdate  func(inner *allowEmptyFoldersInnerClient)
+		wantLookups   int
+		wantUpdates   int
+		wantAttrValue string
+	}{
+		{name: "no policy allows empty folders", mountRoot: "/buckets/b1", wantLookups: 1, wantUpdates: 1, wantAttrValue: "true"},
+		{name: "explicit true is kept", mountRoot: "/buckets/b1", attrValue: "true", wantLookups: 1, wantUpdates: 0},
+		{name: "explicit false is kept", mountRoot: "/buckets/b1", attrValue: "false", wantLookups: 1, wantUpdates: 0},
+		{name: "subdirectory mount leaves the bucket alone", mountRoot: "/buckets/b1/sub", wantLookups: 0, wantUpdates: 0},
+		{
+			name:      "a concurrent false wins over the mount default",
+			mountRoot: "/buckets/b1",
+			beforeUpdate: func(inner *allowEmptyFoldersInnerClient) {
+				inner.entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("false")
+			},
+			wantLookups:   2,
+			wantUpdates:   0,
+			wantAttrValue: "false",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			inner := &allowEmptyFoldersInnerClient{entry: bucketEntry(tt.attrValue)}
+			if tt.beforeUpdate != nil {
+				inner.beforeUpdate = func() { tt.beforeUpdate(inner) }
+			}
+			filerClient := &allowEmptyFoldersFilerClient{inner: inner}
+
+			if err := ensureBucketAllowEmptyFolders(context.Background(), filerClient, tt.mountRoot, "/buckets"); err != nil {
+				t.Fatalf("ensureBucketAllowEmptyFolders: %v", err)
+			}
+			if inner.lookups != tt.wantLookups {
+				t.Errorf("lookups = %d, want %d", inner.lookups, tt.wantLookups)
+			}
+			if len(inner.updates) != tt.wantUpdates {
+				t.Fatalf("updates = %d, want %d", len(inner.updates), tt.wantUpdates)
+			}
+			if tt.wantAttrValue != "" {
+				got := string(inner.entry.Extended[s3_constants.ExtAllowEmptyFolders])
+				if got != tt.wantAttrValue {
+					t.Errorf("%s = %q, want %q", s3_constants.ExtAllowEmptyFolders, got, tt.wantAttrValue)
+				}
+			}
+		})
+	}
+}

weed/filer/empty_folder_cleanup/empty_folder_cleaner.go

@@ -2,6 +2,7 @@ package empty_folder_cleanup
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"sort"
 	"strings"
@@ -17,12 +18,13 @@ import (
 )
 
 const (
-	DefaultMaxCountCheck  = 1000
-	DefaultCacheExpiry    = 5 * time.Minute
-	DefaultQueueMaxSize   = 1000
-	DefaultQueueMaxAge    = 2 * time.Minute
-	DefaultProcessorSleep = 30 * time.Second // How often to check queue
-	DefaultMaxDeletedKept = 10000            // Deleted folders remembered for the restore check
+	DefaultMaxCountCheck     = 1000
+	DefaultCacheExpiry       = 5 * time.Minute
+	DefaultQueueMaxSize      = 1000
+	DefaultQueueMaxAge       = 2 * time.Minute
+	DefaultProcessorSleep    = 30 * time.Second // How often to check queue
+	DefaultMaxDeletedKept    = 10000            // Deleted folders remembered for the restore check
+	DefaultMaxPolicyFailures = 3                // Consecutive policy-load failures before a folder is dropped
 	// 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.
@@ -59,10 +61,17 @@ type FilerOperations interface {
 
 // folderState tracks the state of a folder for empty folder cleanup
 type folderState struct {
-	roughCount  int       // Cached rough count (up to maxCountCheck)
-	lastAddTime time.Time // Last time an item was added
-	lastDelTime time.Time // Last time an item was deleted
-	lastCheck   time.Time // Last time we checked the actual count
+	roughCount     int       // Cached rough count (up to maxCountCheck)
+	lastAddTime    time.Time // Last time an item was added
+	lastDelTime    time.Time // Last time an item was deleted
+	lastCheck      time.Time // Last time we checked the actual count
+	policyFailures int       // Consecutive bucket policy load failures
+}
+
+type bucketPolicyGen struct {
+	gen      uint64
+	lastBump time.Time
+	inflight int
 }
 
 type bucketCleanupPolicyState struct {
@@ -82,6 +91,7 @@ type EmptyFolderCleaner struct {
 	mu                    sync.RWMutex
 	folderCounts          map[string]*folderState              // Rough count cache
 	bucketCleanupPolicies map[string]*bucketCleanupPolicyState // bucket path -> cleanup policy cache
+	policyGen             map[string]*bucketPolicyGen          // bucket path -> generation, bumped on bucket entry updates to invalidate in-flight policy loads
 
 	// Folders deleted recently, kept so that a create event arriving for one of them
 	// can put it back
@@ -115,6 +125,7 @@ func NewEmptyFolderCleaner(filer FilerOperations, lockRing *lock_manager.LockRin
 		host:                  host,
 		folderCounts:          make(map[string]*folderState),
 		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
+		policyGen:             make(map[string]*bucketPolicyGen),
 		deleted:               make(map[string]*deletedFolder),
 		cleanupQueue:          NewCleanupQueue(DefaultQueueMaxSize, cleanupDelay),
 		maxCountCheck:         DefaultMaxCountCheck,
@@ -251,6 +262,33 @@ func (efc *EmptyFolderCleaner) OnCreateEvent(directory string, entryName string,
 	}
 }
 
+// InvalidateBucketPolicy drops the cached cleanup policy when a bucket entry changes
+func (efc *EmptyFolderCleaner) InvalidateBucketPolicy(directory string, entryName string, isDirectory bool) {
+	if !isDirectory || directory != efc.bucketPath {
+		return
+	}
+
+	efc.mu.Lock()
+	defer efc.mu.Unlock()
+
+	if !efc.enabled {
+		return
+	}
+
+	if efc.policyGen == nil {
+		efc.policyGen = make(map[string]*bucketPolicyGen)
+	}
+	path := string(util.NewFullPath(directory, entryName))
+	gen := efc.policyGen[path]
+	if gen == nil {
+		gen = &bucketPolicyGen{}
+		efc.policyGen[path] = gen
+	}
+	gen.gen++
+	gen.lastBump = time.Now()
+	delete(efc.bucketCleanupPolicies, path)
+}
+
 // cleanupProcessor runs in background and processes the cleanup queue
 func (efc *EmptyFolderCleaner) cleanupProcessor() {
 	ticker := time.NewTicker(efc.processorSleep)
@@ -491,9 +529,28 @@ func (efc *EmptyFolderCleaner) executeCleanup(folder string, triggeredBy string)
 			return
 		}
 		glog.V(2).Infof("EmptyFolderCleaner: failed to load bucket cleanup policy for folder %s (triggered by %s): %v", folder, triggeredBy, err)
+		efc.mu.Lock()
+		if efc.enabled {
+			state, exists := efc.folderCounts[folder]
+			if !exists {
+				state = &folderState{}
+				efc.folderCounts[folder] = state
+			}
+			state.policyFailures++
+			if state.policyFailures <= DefaultMaxPolicyFailures {
+				efc.cleanupQueue.Add(folder, triggeredBy, time.Now())
+			}
+		}
+		efc.mu.Unlock()
 		return
 	}
 
+	efc.mu.Lock()
+	if state, exists := efc.folderCounts[folder]; exists {
+		state.policyFailures = 0
+	}
+	efc.mu.Unlock()
+
 	if !autoRemove {
 		glog.V(3).Infof("EmptyFolderCleaner: skipping folder %s (triggered by %s), bucket %s auto-remove-empty-folders disabled (source=%s attr=%s)",
 			folder, triggeredBy, bucketPath, source, attrValue)
@@ -617,25 +674,47 @@ func (efc *EmptyFolderCleaner) getBucketCleanupPolicy(ctx context.Context, folde
 	}
 	efc.mu.RUnlock()
 
-	attrs, err := efc.filer.GetEntryAttributes(ctx, util.FullPath(bucketPath))
-	if err != nil {
-		return "", true, "", "", err
+	for attempt := 0; attempt < 3; attempt++ {
+		efc.mu.Lock()
+		if efc.policyGen == nil {
+			efc.policyGen = make(map[string]*bucketPolicyGen)
+		}
+		state := efc.policyGen[bucketPath]
+		if state == nil {
+			state = &bucketPolicyGen{}
+			efc.policyGen[bucketPath] = state
+		}
+		state.inflight++
+		gen := state.gen
+		efc.mu.Unlock()
+
+		attrs, err := efc.filer.GetEntryAttributes(ctx, util.FullPath(bucketPath))
+
+		efc.mu.Lock()
+		state.inflight--
+		if err != nil {
+			efc.mu.Unlock()
+			return "", true, "", "", err
+		}
+		autoRemove, attrValue = autoRemoveEmptyFoldersEnabled(attrs)
+		if gen != state.gen {
+			efc.mu.Unlock()
+			continue
+		}
+		if efc.bucketCleanupPolicies == nil {
+			efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState)
+		}
+		efc.bucketCleanupPolicies[bucketPath] = &bucketCleanupPolicyState{
+			autoRemove: autoRemove,
+			attrValue:  attrValue,
+			lastCheck:  now,
+		}
+		efc.mu.Unlock()
+
+		return bucketPath, autoRemove, "filer", attrValue, nil
 	}
 
-	autoRemove, attrValue = autoRemoveEmptyFoldersEnabled(attrs)
-
-	efc.mu.Lock()
-	if efc.bucketCleanupPolicies == nil {
-		efc.bucketCleanupPolicies = make(map[string]*bucketCleanupPolicyState)
-	}
-	efc.bucketCleanupPolicies[bucketPath] = &bucketCleanupPolicyState{
-		autoRemove: autoRemove,
-		attrValue:  attrValue,
-		lastCheck:  now,
-	}
-	efc.mu.Unlock()
-
-	return bucketPath, autoRemove, "filer", attrValue, nil
+	return "", false, "", "", fmt.Errorf("bucket cleanup policy changed during read: %s", bucketPath)
 }
 
 // isCatalogEntry reports whether the directory is an s3tables catalog record.
@@ -775,6 +854,14 @@ func (efc *EmptyFolderCleaner) evictStaleCacheEntries() {
 		}
 	}
 
+	// A generation stays while a read that captured it can still return; an idle
+	// one drops after the expiry so bucket churn cannot grow the map forever.
+	for bucketPath, gen := range efc.policyGen {
+		if gen.inflight == 0 && now.Sub(gen.lastBump) > efc.cacheExpiry {
+			delete(efc.policyGen, bucketPath)
+		}
+	}
+
 	if expiredCount > 0 {
 		glog.V(3).Infof("EmptyFolderCleaner: evicted %d stale cache entries", expiredCount)
 	}
@@ -791,6 +878,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.policyGen = make(map[string]*bucketPolicyGen)
 	efc.deleted, efc.deletedDropped = make(map[string]*deletedFolder), 0
 }
 

weed/filer/empty_folder_cleanup/empty_folder_cleaner_test.go

@@ -1074,6 +1074,156 @@ func TestEmptyFolderCleaner_executeCleanup_bucketPolicyDisabledSkips(t *testing.
 	}
 }
 
+func TestEmptyFolderCleaner_InvalidateBucketPolicy(t *testing.T) {
+	cleaner := &EmptyFolderCleaner{
+		bucketPath: "/buckets",
+		enabled:    true,
+		bucketCleanupPolicies: map[string]*bucketCleanupPolicyState{
+			"/buckets/test": {},
+		},
+	}
+
+	cleaner.InvalidateBucketPolicy("/buckets", "test", true)
+	if _, found := cleaner.bucketCleanupPolicies["/buckets/test"]; found {
+		t.Fatal("expected cached bucket policy to be evicted")
+	}
+
+	cleaner.bucketCleanupPolicies["/buckets/test"] = &bucketCleanupPolicyState{}
+	cleaner.InvalidateBucketPolicy("/buckets/test", "dir", true)
+	cleaner.InvalidateBucketPolicy("/buckets", "test", false)
+	if _, found := cleaner.bucketCleanupPolicies["/buckets/test"]; !found {
+		t.Fatal("expected unrelated updates to keep the cached policy")
+	}
+}
+
+func TestEmptyFolderCleaner_getBucketCleanupPolicy_concurrentUpdate(t *testing.T) {
+	var cleaner *EmptyFolderCleaner
+	calls := 0
+	mock := &mockFilerOps{
+		attrsFn: func(_ util.FullPath) (map[string][]byte, error) {
+			calls++
+			if calls == 1 {
+				// a toggle lands while the stale read is in flight
+				cleaner.InvalidateBucketPolicy("/buckets", "test", true)
+				return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("true")}, nil
+			}
+			return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("false")}, nil
+		},
+	}
+
+	cleaner = &EmptyFolderCleaner{
+		filer:                 mock,
+		bucketPath:            "/buckets",
+		enabled:               true,
+		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
+	}
+
+	_, autoRemove, _, attrValue, err := cleaner.getBucketCleanupPolicy(context.Background(), "/buckets/test/folder")
+	if err != nil {
+		t.Fatalf("getBucketCleanupPolicy: %v", err)
+	}
+	if !autoRemove || attrValue != "false" {
+		t.Fatalf("expected the post-update policy autoRemove=true attr=false, got autoRemove=%v attr=%q", autoRemove, attrValue)
+	}
+	if calls < 2 {
+		t.Fatalf("expected the stale load to be retried, got %d attribute reads", calls)
+	}
+	if cached := cleaner.bucketCleanupPolicies["/buckets/test"]; cached == nil || cached.attrValue != "false" {
+		t.Fatalf("expected the fresh policy to be cached, got %+v", cached)
+	}
+}
+
+func TestEmptyFolderCleaner_getBucketCleanupPolicy_evictionKeepsInflightRead(t *testing.T) {
+	var cleaner *EmptyFolderCleaner
+	started := make(chan struct{})
+	release := make(chan struct{})
+	calls := 0
+	mock := &mockFilerOps{
+		attrsFn: func(_ util.FullPath) (map[string][]byte, error) {
+			calls++
+			if calls == 1 {
+				close(started)
+				<-release
+				return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("true")}, nil
+			}
+			return map[string][]byte{s3_constants.ExtAllowEmptyFolders: []byte("false")}, nil
+		},
+	}
+
+	cleaner = &EmptyFolderCleaner{
+		filer:                 mock,
+		bucketPath:            "/buckets",
+		enabled:               true,
+		cacheExpiry:           time.Minute,
+		folderCounts:          make(map[string]*folderState),
+		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
+		policyGen:             make(map[string]*bucketPolicyGen),
+		cleanupQueue:          NewCleanupQueue(1000, time.Minute),
+	}
+
+	go func() {
+		<-started
+		cleaner.InvalidateBucketPolicy("/buckets", "test", true)
+		cleaner.mu.Lock()
+		cleaner.policyGen["/buckets/test"].lastBump = time.Now().Add(-time.Hour)
+		cleaner.mu.Unlock()
+		cleaner.evictStaleCacheEntries()
+		close(release)
+	}()
+
+	_, autoRemove, _, attrValue, err := cleaner.getBucketCleanupPolicy(context.Background(), "/buckets/test/folder")
+	if err != nil {
+		t.Fatalf("getBucketCleanupPolicy: %v", err)
+	}
+	if !autoRemove || attrValue != "false" {
+		t.Fatalf("expected the stalled read to be discarded, got autoRemove=%v attr=%q", autoRemove, attrValue)
+	}
+	if calls < 2 {
+		t.Fatalf("expected the stale load to be retried, got %d attribute reads", calls)
+	}
+}
+
+func TestEmptyFolderCleaner_executeCleanup_policyFailureRequeues(t *testing.T) {
+	lockRing := lock_manager.NewLockRing(5 * time.Second)
+	lockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0)
+
+	mock := &mockFilerOps{
+		attrsFn: func(_ util.FullPath) (map[string][]byte, error) {
+			return nil, errors.New("attrs unavailable")
+		},
+	}
+
+	cleaner := &EmptyFolderCleaner{
+		filer:                 mock,
+		lockRing:              lockRing,
+		host:                  "filer1:8888",
+		bucketPath:            "/buckets",
+		enabled:               true,
+		folderCounts:          make(map[string]*folderState),
+		bucketCleanupPolicies: make(map[string]*bucketCleanupPolicyState),
+		cleanupQueue:          NewCleanupQueue(1000, time.Millisecond),
+		stopCh:                make(chan struct{}),
+	}
+
+	folder := "/buckets/test/folder"
+	for i := 0; i < DefaultMaxPolicyFailures; i++ {
+		cleaner.executeCleanup(folder, "triggered_item")
+		popped, _, ok := cleaner.cleanupQueue.Pop()
+		if !ok || popped != folder {
+			t.Fatalf("failure %d of %d should requeue the folder", i+1, DefaultMaxPolicyFailures)
+		}
+	}
+
+	cleaner.executeCleanup(folder, "triggered_item")
+
+	if got := cleaner.folderCounts[folder].policyFailures; got != DefaultMaxPolicyFailures+1 {
+		t.Fatalf("policyFailures = %d, want %d", got, DefaultMaxPolicyFailures+1)
+	}
+	if got := cleaner.cleanupQueue.Len(); got != 0 {
+		t.Fatalf("expected retries to stop after %d failures, got %d queued", DefaultMaxPolicyFailures, got)
+	}
+}
+
 func TestEmptyFolderCleaner_executeCleanup_directoryMarker(t *testing.T) {
 	testCases := []struct {
 		name           string

weed/filer/filer_on_meta_event.go

@@ -55,11 +55,18 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp
 	// Handle delete events - trigger folder cleanup check
 	if filer_pb.IsDelete(event) && message.OldEntry != nil {
 		f.EmptyFolderCleaner.OnDeleteEvent(directory, message.OldEntry.Name, message.OldEntry.IsDirectory, eventTime)
+		f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.OldEntry.Name, message.OldEntry.IsDirectory)
 	}
 
 	// Handle create events - cancel pending cleanup for the folder
 	if filer_pb.IsCreate(event) && message.NewEntry != nil {
 		f.EmptyFolderCleaner.OnCreateEvent(directory, message.NewEntry.Name, message.NewEntry.IsDirectory)
+		f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.NewEntry.Name, message.NewEntry.IsDirectory)
+	}
+
+	// Handle update events - drop the cached bucket cleanup policy
+	if filer_pb.IsUpdate(event) && message.NewEntry != nil {
+		f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.NewEntry.Name, message.NewEntry.IsDirectory)
 	}
 
 	// Handle rename/move events
@@ -67,6 +74,7 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp
 		// Treat the old location as a delete
 		if message.OldEntry != nil {
 			f.EmptyFolderCleaner.OnDeleteEvent(directory, message.OldEntry.Name, message.OldEntry.IsDirectory, eventTime)
+			f.EmptyFolderCleaner.InvalidateBucketPolicy(directory, message.OldEntry.Name, message.OldEntry.IsDirectory)
 		}
 		// Treat the new location as a create
 		if message.NewEntry != nil {
@@ -75,6 +83,7 @@ func (f *Filer) onEmptyFolderCleanupEvents(event *filer_pb.SubscribeMetadataResp
 				newDir = directory
 			}
 			f.EmptyFolderCleaner.OnCreateEvent(newDir, message.NewEntry.Name, message.NewEntry.IsDirectory)
+			f.EmptyFolderCleaner.InvalidateBucketPolicy(newDir, message.NewEntry.Name, message.NewEntry.IsDirectory)
 		}
 	}
 }

weed/pb/filer_pb/filer_pb_helper.go

@@ -177,6 +177,22 @@ func CreateEntryWithResponse(ctx context.Context, client SeaweedFilerClient, req
 	return resp, nil
 }
 
+// SnapshotExtended clones extended attributes for use as an UpdateEntry
+// ExpectedExtended precondition. Listed keys are asserted even when absent,
+// so a concurrent add still fails the precondition.
+func SnapshotExtended(extended map[string][]byte, keys ...string) map[string][]byte {
+	expected := make(map[string][]byte, len(extended)+len(keys))
+	for k, v := range extended {
+		expected[k] = v
+	}
+	for _, k := range keys {
+		if _, ok := expected[k]; !ok {
+			expected[k] = nil
+		}
+	}
+	return expected
+}
+
 func UpdateEntry(ctx context.Context, client SeaweedFilerClient, request *UpdateEntryRequest) error {
 	_, err := UpdateEntryWithResponse(ctx, client, request)
 	return err

weed/shell/command_s3_bucket_allow_empty_folders.go

@@ -0,0 +1,130 @@
+package shell
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3bucket"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3BucketAllowEmptyFolders{})
+}
+
+type commandS3BucketAllowEmptyFolders struct {
+}
+
+func (c *commandS3BucketAllowEmptyFolders) Name() string {
+	return "s3.bucket.allowEmptyFolders"
+}
+
+func (c *commandS3BucketAllowEmptyFolders) Help() string {
+	return `view or toggle whether a bucket keeps empty folders
+
+	When enabled, the empty folder cleaner skips this bucket, so empty
+	directories persist (POSIX semantics). weed mount sets this when
+	mounting a bucket root.
+
+	When disabled, folders emptied by object deletes are removed
+	asynchronously and stop appearing as CommonPrefix in S3 listings.
+	The explicit "false" marker is kept by weed mount.
+
+	Example:
+		# Show the current setting
+		s3.bucket.allowEmptyFolders -name <bucket_name>
+
+		# Keep empty folders (skip the cleaner for this bucket)
+		s3.bucket.allowEmptyFolders -name <bucket_name> -enable
+
+		# Let the cleaner remove empty folders
+		s3.bucket.allowEmptyFolders -name <bucket_name> -disable
+`
+}
+
+func (c *commandS3BucketAllowEmptyFolders) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandS3BucketAllowEmptyFolders) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	bucketCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	bucketName := bucketCommand.String("name", "", "bucket name")
+	enable := bucketCommand.Bool("enable", false, "keep empty folders in the bucket")
+	disable := bucketCommand.Bool("disable", false, "remove empty folders asynchronously")
+	if err = bucketCommand.Parse(args); err != nil {
+		return err
+	}
+
+	if *bucketName == "" {
+		return fmt.Errorf("empty bucket name")
+	}
+	if err := s3bucket.VerifyS3BucketName(*bucketName); err != nil {
+		return fmt.Errorf("invalid bucket name %q: %w", *bucketName, err)
+	}
+	if *enable && *disable {
+		return fmt.Errorf("only one of -enable or -disable can be set")
+	}
+
+	filerBucketsPath, err := readFilerBucketsPath(commandEnv)
+	if err != nil {
+		return fmt.Errorf("read buckets: %w", err)
+	}
+
+	return commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		for attempt := 0; attempt < 5; attempt++ {
+			lookupResp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+				Directory: filerBucketsPath,
+				Name:      *bucketName,
+			})
+			if err != nil {
+				return fmt.Errorf("lookup bucket %s: %w", *bucketName, err)
+			}
+			entry := lookupResp.Entry
+
+			if !*enable && !*disable {
+				state := "disabled"
+				if strings.EqualFold(strings.TrimSpace(string(entry.Extended[s3_constants.ExtAllowEmptyFolders])), "true") {
+					state = "enabled"
+				}
+				fmt.Fprintf(writer, "Bucket: %s\n", *bucketName)
+				fmt.Fprintf(writer, "Allow empty folders: %s\n", state)
+				return nil
+			}
+
+			expected := filer_pb.SnapshotExtended(entry.Extended, s3_constants.ExtAllowEmptyFolders)
+
+			if entry.Extended == nil {
+				entry.Extended = make(map[string][]byte)
+			}
+			state := "disabled"
+			if *enable {
+				entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("true")
+				state = "enabled"
+			} else {
+				entry.Extended[s3_constants.ExtAllowEmptyFolders] = []byte("false")
+			}
+
+			if _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
+				Directory:        filerBucketsPath,
+				Entry:            entry,
+				ExpectedExtended: expected,
+			}); err != nil {
+				if status.Code(err) == codes.FailedPrecondition {
+					continue
+				}
+				return fmt.Errorf("failed to update bucket: %w", err)
+			}
+
+			fmt.Fprintf(writer, "Bucket %s allow empty folders %s\n", *bucketName, state)
+			return nil
+		}
+		return fmt.Errorf("bucket %s changed concurrently; please retry", *bucketName)
+	})
+}