fix(s3api/iam): avoid transient AccessDenied from full reloads on single IAM file changes (#11271)

* fix(s3api/iam): fail config snapshot on empty or malformed IAM files

A full IAM reload reads every identity/policy/service-account/group file
from the filer. When an external secrets tool rewrites a file, a reload
that reads it mid-rewrite sees empty or partially-written content. The
identity, policy and service-account loaders silently skipped such files
(``continue``), so the snapshot was missing entries that still existed
on disk. The atomic swap then installed an incomplete identity set while
``isAuthEnabled`` stayed on, denying unrelated clients mid-reload
(#11259).

The group loader and the read-error paths already fail the snapshot in
this situation (a skipped entry reads as deleted). Apply the same
behavior to empty content and unmarshal failures across the identity,
policy, service-account and group loaders, so a transient mid-rewrite
fails the reload (preserving the last known-good state) instead of
silently dropping entries.

* fix(s3api/iam): coalesce burst IAM config reloads through the reload queue

onIamConfigChange did a full synchronous reload for every identity/policy
file change event. When several independently-refreshing credentials
rewrite their files within the same second, that produced a burst of
dozens of back-to-back full reloads, each reading the whole store and
widening the window where a mid-rewrite file is observed (#11259).

Route every IAM config change through the existing coalescing reload
queue (scheduleReload/reloadRetryLoop) instead. A burst of N events now
collapses into a single reload (plus one tail reload for events that
arrived while one was in flight). scheduleReload gains a reason argument
for the existing log line; the reloadRetryLoop already retries failed
reloads, so the per-event failure handoff is no longer needed.

Tests that asserted on the synchronous reload now wire up the queue
(centralized in newTestS3ApiServerWithMemoryIAM) and poll via
waitForIdentity/waitForIdentityGone. Adds TestOnIamConfigChangeCoalescesBurstReloads
showing 50 events coalesce into <=3 reloads.

* fix(s3api/iam): skip non-JSON auxiliary files before failing IAM snapshot

Per review: the multi-file loaders unmarshal every entry in an IAM
directory, so a non-JSON auxiliary file (README, .DS_Store, a migration
backup such as identity.json.old) would hit the new empty/malformed
errors and reject the whole snapshot, blocking all later IAM reloads.

Only *.json files are IAM objects (SeaweedFS writes identities,
policies, service accounts and groups as <name>.json, and other call
sites already gate on the .json suffix). Skip non-.json entries at the
top of each loader loop, before reading content, so auxiliary files are
ignored while empty/malformed .json files still fail the snapshot.

Adds TestLoadConfigurationIgnoresNonJsonAuxiliaryFiles.

* fix(s3api/iam): reject IAM files with empty identifiers and skip aux in listing

Per review:

- ListPolicyNames listed every regular entry in the policies directory as a
  policy name, including non-JSON auxiliary files, but GetPolicy cannot
  retrieve them. Apply the same .json suffix filter used by the loader so
  the list only exposes retrievable policies.

- json.Unmarshal accepts `{}` and unknown fields. The identity and group
  loaders merge by the decoded Name (not the file name), so a `{}` file
  could install an empty-key record and displace a real one; the
  service-account loader accepted an empty Id. Validate Identity.Name,
  Group.Name and ServiceAccount.Id (via validateServiceAccountId) after
  unmarshal and fail the snapshot on empty identifiers.

Adds TestFilerEtcStoreListPolicyNamesSkipsNonJsonAuxiliary and
empty-identifier regression tests for identity, group and service-account
files.

weed/credential/filer_etc/filer_etc_group.go

@@ -30,6 +30,9 @@ func (store *FilerEtcStore) loadGroupsFromMultiFile(ctx context.Context, s3cfg *
 			if entry.IsDirectory {
 				continue
 			}
+			if !strings.HasSuffix(entry.Name, ".json") {
+				continue
+			}
 
 			var content []byte
 			if len(entry.Content) > 0 {
@@ -42,24 +45,28 @@ func (store *FilerEtcStore) loadGroupsFromMultiFile(ctx context.Context, s3cfg *
 				content = c
 			}
 
-			if len(content) > 0 {
-				g := &iam_pb.Group{}
-				if err := json.Unmarshal(content, g); err != nil {
-					return fmt.Errorf("failed to unmarshal group %s: %w", entry.Name, err)
-				}
-				// Merge: overwrite existing group with same name or append
-				found := false
-				for i, existing := range s3cfg.Groups {
-					if existing.Name == g.Name {
-						s3cfg.Groups[i] = g
-						found = true
-						break
-					}
-				}
-				if !found {
-					s3cfg.Groups = append(s3cfg.Groups, g)
+			if len(content) == 0 {
+				return fmt.Errorf("group file %s is empty", entry.Name)
+			}
+			g := &iam_pb.Group{}
+			if err := json.Unmarshal(content, g); err != nil {
+				return fmt.Errorf("failed to unmarshal group %s: %w", entry.Name, err)
+			}
+			if g.Name == "" {
+				return fmt.Errorf("group file %s has empty name", entry.Name)
+			}
+			// Merge: overwrite existing group with same name or append
+			found := false
+			for i, existing := range s3cfg.Groups {
+				if existing.Name == g.Name {
+					s3cfg.Groups[i] = g
+					found = true
+					break
 				}
 			}
+			if !found {
+				s3cfg.Groups = append(s3cfg.Groups, g)
+			}
 		}
 		return nil
 	})

weed/credential/filer_etc/filer_etc_identity.go

@@ -92,6 +92,9 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb
 			if entry.IsDirectory {
 				continue
 			}
+			if !strings.HasSuffix(entry.Name, ".json") {
+				continue
+			}
 			hasIdentities = true
 
 			var content []byte
@@ -100,26 +103,28 @@ func (store *FilerEtcStore) loadFromMultiFile(ctx context.Context, s3cfg *iam_pb
 			} else {
 				c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name)
 				if err != nil {
-					// fail the snapshot: a skipped identity would read as deleted
 					return fmt.Errorf("failed to read identity file %s: %w", entry.Name, err)
 				}
 				content = c
 			}
 
-			if len(content) > 0 {
-				identity := &iam_pb.Identity{}
-				if err := json.Unmarshal(content, identity); err != nil {
-					glog.Warningf("Failed to unmarshal identity %s: %v", entry.Name, err)
-					continue
-				}
+			if len(content) == 0 {
+				return fmt.Errorf("identity file %s is empty", entry.Name)
+			}
+			identity := &iam_pb.Identity{}
+			if err := json.Unmarshal(content, identity); err != nil {
+				return fmt.Errorf("failed to unmarshal identity %s: %w", entry.Name, err)
+			}
+			if identity.Name == "" {
+				return fmt.Errorf("identity file %s has empty name", entry.Name)
+			}
 
-				// Merge logic: Overwrite existing or Append
-				idx := findIdentity(identity.Name)
-				if idx != -1 {
-					s3cfg.Identities[idx] = identity
-				} else {
-					s3cfg.Identities = append(s3cfg.Identities, identity)
-				}
+			// Merge logic: Overwrite existing or Append
+			idx := findIdentity(identity.Name)
+			if idx != -1 {
+				s3cfg.Identities[idx] = identity
+			} else {
+				s3cfg.Identities = append(s3cfg.Identities, identity)
 			}
 		}
 		return nil

weed/credential/filer_etc/filer_etc_identity_test.go

@@ -0,0 +1,144 @@
+package filer_etc
+
+import (
+	"context"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func putEntry(t *testing.T, server *policyTestFilerServer, dir, name string, content []byte) {
+	t.Helper()
+	_, err := server.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
+		Directory: dir,
+		Entry:     &filer_pb.Entry{Name: name, Content: content},
+	})
+	require.NoError(t, err)
+}
+
+// A mid-rewrite identity file (empty content, as seen while a secrets tool
+// truncates and rewrites the file) must fail the snapshot instead of being
+// silently dropped. Silently dropping it would make a full reload install an
+// incomplete identity set and deny unrelated clients mid-reload.
+func TestLoadConfigurationFailsOnEmptyIdentityFile(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`))
+	putEntry(t, server, identDir, "bob.json", []byte{}) // mid-rewrite: empty
+
+	_, err := store.LoadConfiguration(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "bob.json")
+}
+
+// A malformed identity file (partial JSON from a mid-rewrite) must also fail
+// the snapshot rather than being skipped.
+func TestLoadConfigurationFailsOnMalformedIdentityFile(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice"}`))
+	putEntry(t, server, identDir, "bob.json", []byte(`{"name":"bob"`)) // truncated
+
+	_, err := store.LoadConfiguration(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "bob.json")
+}
+
+// A mid-rewrite policy file must fail the snapshot rather than be dropped.
+func TestLoadManagedPoliciesFailsOnEmptyPolicyFile(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory
+	putEntry(t, server, polDir, "good.json", []byte(`{"Version":"2012-10-17","Statement":[]}`))
+	putEntry(t, server, polDir, "bad.json", []byte{}) // mid-rewrite: empty
+
+	_, err := store.LoadManagedPolicies(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "bad.json")
+}
+
+// A non-JSON auxiliary file (README, .DS_Store, a migration backup) in an IAM
+// directory must not fail the snapshot: only *.json files are IAM objects.
+func TestLoadConfigurationIgnoresNonJsonAuxiliaryFiles(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`))
+	putEntry(t, server, identDir, "README.txt", []byte(`not an identity`))
+	putEntry(t, server, identDir, "alice.json.old", []byte(`{"name":"alice"}`))
+
+	polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory
+	putEntry(t, server, polDir, "p.json", []byte(`{"Version":"2012-10-17","Statement":[]}`))
+	putEntry(t, server, polDir, "notes.md", []byte(`# policies`))
+
+	cfg, err := store.LoadConfiguration(ctx)
+	require.NoError(t, err)
+	require.Len(t, cfg.Identities, 1)
+	assert.Equal(t, "alice", cfg.Identities[0].Name)
+}
+
+// A valid JSON file with an empty name (e.g. `{}`) would unmarshal cleanly but
+// install a garbage empty-key record that can displace a real one. Reject it.
+func TestLoadConfigurationFailsOnEmptyIdentityName(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice"}`))
+	putEntry(t, server, identDir, "empty.json", []byte(`{}`))
+
+	_, err := store.LoadConfiguration(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "empty.json")
+}
+
+func TestLoadConfigurationFailsOnEmptyGroupName(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	groupDir := filer.IamConfigDirectory + "/" + IamGroupsDirectory
+	putEntry(t, server, groupDir, "empty.json", []byte(`{}`))
+
+	_, err := store.LoadConfiguration(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "empty.json")
+}
+
+func TestLoadConfigurationFailsOnEmptyServiceAccountId(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	saDir := filer.IamConfigDirectory + "/" + IamServiceAccountsDirectory
+	putEntry(t, server, saDir, "empty.json", []byte(`{}`))
+
+	_, err := store.LoadConfiguration(ctx)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "empty.json")
+}
+
+// A valid full snapshot loads without error (regression guard).
+func TestLoadConfigurationSucceedsOnValidFiles(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	identDir := filer.IamConfigDirectory + "/" + IamIdentitiesDirectory
+	putEntry(t, server, identDir, "alice.json", []byte(`{"name":"alice","credentials":[{"accessKey":"AK","secretKey":"SK"}]}`))
+	putEntry(t, server, identDir, "bob.json", []byte(`{"name":"bob","credentials":[{"accessKey":"BK","secretKey":"SK"}]}`))
+
+	cfg, err := store.LoadConfiguration(ctx)
+	require.NoError(t, err)
+	names := make([]string, 0, len(cfg.Identities))
+	for _, id := range cfg.Identities {
+		names = append(names, id.Name)
+	}
+	assert.ElementsMatch(t, []string{"alice", "bob"}, names)
+}

weed/credential/filer_etc/filer_etc_policy.go

@@ -181,6 +181,9 @@ func (store *FilerEtcStore) loadPoliciesFromMultiFile(ctx context.Context, polic
 			if entry.IsDirectory {
 				continue
 			}
+			if !strings.HasSuffix(entry.Name, ".json") {
+				continue
+			}
 
 			var content []byte
 			if len(entry.Content) > 0 {
@@ -188,26 +191,21 @@ func (store *FilerEtcStore) loadPoliciesFromMultiFile(ctx context.Context, polic
 			} else {
 				c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name)
 				if err != nil {
-					// fail the snapshot: a skipped policy would read as deleted
 					return fmt.Errorf("failed to read policy file %s: %w", entry.Name, err)
 				}
 				content = c
 			}
 
-			if len(content) > 0 {
-				var policy policy_engine.PolicyDocument
-				if err := json.Unmarshal(content, &policy); err != nil {
-					glog.Warningf("Failed to unmarshal policy %s: %v", entry.Name, err)
-					continue
-				}
-
-				// The file name is "policyName.json"
-				policyName := entry.Name
-				if strings.HasSuffix(policyName, ".json") {
-					policyName = policyName[:len(policyName)-5]
-					policies[policyName] = policy
-				}
+			if len(content) == 0 {
+				return fmt.Errorf("policy file %s is empty", entry.Name)
 			}
+			var policy policy_engine.PolicyDocument
+			if err := json.Unmarshal(content, &policy); err != nil {
+				return fmt.Errorf("failed to unmarshal policy %s: %w", entry.Name, err)
+			}
+
+			policyName := strings.TrimSuffix(entry.Name, ".json")
+			policies[policyName] = policy
 		}
 		return nil
 	})
@@ -498,10 +496,10 @@ func (store *FilerEtcStore) ListPolicyNames(ctx context.Context) ([]string, erro
 			if entry.IsDirectory {
 				continue
 			}
-			name := entry.Name
-			if strings.HasSuffix(name, ".json") {
-				name = name[:len(name)-5]
+			if !strings.HasSuffix(entry.Name, ".json") {
+				continue
 			}
+			name := entry.Name[:len(entry.Name)-5]
 			if _, found := seenNames[name]; found {
 				continue
 			}

weed/credential/filer_etc/filer_etc_policy_test.go

@@ -192,6 +192,22 @@ func TestFilerEtcStoreListPolicyNamesIncludesLegacyPolicies(t *testing.T) {
 	assert.ElementsMatch(t, []string{"legacy-only", "multi-file-only", "shared"}, names)
 }
 
+// A non-JSON auxiliary file in the policies directory must not be listed as a
+// policy name (it cannot be retrieved by GetPolicy either).
+func TestFilerEtcStoreListPolicyNamesSkipsNonJsonAuxiliary(t *testing.T) {
+	ctx := context.Background()
+	store, server := newPolicyTestStoreWithServer(t)
+
+	require.NoError(t, store.savePolicy(ctx, "real", testPolicyDocument("s3:GetObject", "arn:aws:s3:::real/*")))
+
+	polDir := filer.IamConfigDirectory + "/" + IamPoliciesDirectory
+	putEntry(t, server, polDir, "notes.md", []byte(`# not a policy`))
+
+	names, err := store.ListPolicyNames(ctx)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"real"}, names)
+}
+
 func TestFilerEtcStoreDeletePolicyRemovesLegacyManagedCopy(t *testing.T) {
 	ctx := context.Background()
 	store := newPolicyTestStore(t)

weed/credential/filer_etc/filer_etc_service_account.go

@@ -33,6 +33,9 @@ func (store *FilerEtcStore) loadServiceAccountsFromMultiFile(ctx context.Context
 			if entry.IsDirectory {
 				continue
 			}
+			if !strings.HasSuffix(entry.Name, ".json") {
+				continue
+			}
 
 			var content []byte
 			if len(entry.Content) > 0 {
@@ -40,20 +43,22 @@ func (store *FilerEtcStore) loadServiceAccountsFromMultiFile(ctx context.Context
 			} else {
 				c, err := filer.ReadInsideFiler(ctx, client, dir, entry.Name)
 				if err != nil {
-					glog.Warningf("Failed to read service account file %s: %v", entry.Name, err)
-					continue
+					return fmt.Errorf("failed to read service account file %s: %w", entry.Name, err)
 				}
 				content = c
 			}
 
-			if len(content) > 0 {
-				sa := &iam_pb.ServiceAccount{}
-				if err := json.Unmarshal(content, sa); err != nil {
-					glog.Warningf("Failed to unmarshal service account %s: %v", entry.Name, err)
-					continue
-				}
-				s3cfg.ServiceAccounts = append(s3cfg.ServiceAccounts, sa)
+			if len(content) == 0 {
+				return fmt.Errorf("service account file %s is empty", entry.Name)
 			}
+			sa := &iam_pb.ServiceAccount{}
+			if err := json.Unmarshal(content, sa); err != nil {
+				return fmt.Errorf("failed to unmarshal service account %s: %w", entry.Name, err)
+			}
+			if err := validateServiceAccountId(sa.Id); err != nil {
+				return fmt.Errorf("service account file %s: %w", entry.Name, err)
+			}
+			s3cfg.ServiceAccounts = append(s3cfg.ServiceAccounts, sa)
 		}
 		return nil
 	})

weed/s3api/auth_credentials.go

@@ -455,10 +455,11 @@ func (iam *IdentityAccessManagement) markStaticIdentities(config *iam_pb.S3ApiCo
 
 var iamReloadRetryInterval = 5 * time.Second
 
-// scheduleReload queues a full configuration reload that retries until it
-// succeeds. Signals coalesce, and the reload is state-based, so it is safe to
-// call for every failed event.
-func (iam *IdentityAccessManagement) scheduleReload() {
+// scheduleReload queues a coalesced full configuration reload that retries
+// until it succeeds. Safe to call for every IAM config change event: bursts
+// collapse into a single reload via the buffered reloadCh.
+func (iam *IdentityAccessManagement) scheduleReload(reason string) {
+	glog.V(1).Infof("IAM change detected in %s, scheduling reload", reason)
 	select {
 	case iam.reloadCh <- struct{}{}:
 	default:

weed/s3api/auth_credentials_static_config_test.go

@@ -38,9 +38,7 @@ func TestIamConfigWithoutIdentitiesIsNotStatic(t *testing.T) {
 	if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err != nil {
 		t.Fatalf("onIamConfigChange returned error: %v", err)
 	}
-	if !hasIdentity(s3a.iam, "alice") {
-		t.Fatalf("expected alice to load after filer change with -iam.config-only setup")
-	}
+	waitForIdentity(t, s3a.iam, "alice")
 }
 
 // A -config identity file protects its identities but must not block live
@@ -70,9 +68,7 @@ func TestConfigWithIdentitiesStillLiveReloadsDynamic(t *testing.T) {
 	if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err != nil {
 		t.Fatalf("onIamConfigChange returned error: %v", err)
 	}
-	if !hasIdentity(s3a.iam, "alice") {
-		t.Fatalf("expected alice to live-reload despite the static identity file")
-	}
+	waitForIdentity(t, s3a.iam, "alice")
 	if !hasIdentity(s3a.iam, "static-admin") {
 		t.Fatalf("static-admin must survive the dynamic reload")
 	}
@@ -84,9 +80,7 @@ func TestConfigWithIdentitiesStillLiveReloadsDynamic(t *testing.T) {
 	if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", &filer_pb.Entry{Name: "alice.json"}, nil); err != nil {
 		t.Fatalf("onIamConfigChange returned error: %v", err)
 	}
-	if hasIdentity(s3a.iam, "alice") {
-		t.Fatalf("expected alice to be removed after deletion on the filer")
-	}
+	waitForIdentityGone(t, s3a.iam, "alice")
 	if !hasIdentity(s3a.iam, "static-admin") {
 		t.Fatalf("static-admin must survive the deletion reload")
 	}
@@ -285,7 +279,8 @@ func (f *flakyStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfig
 	return f.CredentialStore.LoadConfiguration(ctx)
 }
 
-// A failed event-driven reload must keep retrying until the store recovers.
+// A failed reload queued from an IAM config change must keep retrying until
+// the store recovers.
 func TestFailedReloadRetriesUntilSuccess(t *testing.T) {
 	s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{})
 
@@ -293,26 +288,17 @@ func TestFailedReloadRetriesUntilSuccess(t *testing.T) {
 	iamReloadRetryInterval = 10 * time.Millisecond
 	t.Cleanup(func() { iamReloadRetryInterval = prev })
 
-	s3a.iam.reloadCh = make(chan struct{}, 1)
-	go s3a.iam.reloadRetryLoop()
-	t.Cleanup(s3a.iam.Shutdown)
-
 	if err := s3a.iam.credentialManager.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil {
 		t.Fatalf("failed to create alice: %v", err)
 	}
 	s3a.iam.credentialManager.Store = &flakyStore{CredentialStore: s3a.iam.credentialManager.Store, failures: 2}
 
-	// the event-driven reload fails and hands off to the retry loop
-	if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err == nil {
-		t.Fatalf("expected the event-driven reload to fail")
-	}
-	deadline := time.Now().Add(5 * time.Second)
-	for !hasIdentity(s3a.iam, "alice") {
-		if time.Now().After(deadline) {
-			t.Fatalf("expected alice to load once the store recovered")
-		}
-		time.Sleep(10 * time.Millisecond)
+	// onIamConfigChange only queues a coalesced reload; the retry loop does the
+	// work and retries through the transient store failures.
+	if err := s3a.onIamConfigChange(filer.IamConfigDirectory+"/identities", nil, &filer_pb.Entry{Name: "alice.json"}); err != nil {
+		t.Fatalf("onIamConfigChange returned error: %v", err)
 	}
+	waitForIdentity(t, s3a.iam, "alice")
 }
 
 func hasPolicy(iam *IdentityAccessManagement, name string) bool {

weed/s3api/auth_credentials_subscribe.go

@@ -84,44 +84,27 @@ func (s3a *S3ApiServer) onIamConfigChange(dir string, oldEntry *filer_pb.Entry,
 		return nil
 	}
 
-	reloadIamConfig := func(reason string) error {
-		glog.V(1).Infof("IAM change detected in %s, reloading configuration", reason)
-		if err := s3a.iam.LoadS3ApiConfigurationFromCredentialManager(); err != nil {
-			// the event stream moves on; retry state-based until a reload succeeds
-			glog.Errorf("failed to reload IAM configuration after change in %s: %v", reason, err)
-			s3a.iam.scheduleReload()
-			return err
-		}
-		return nil
-	}
+	// Coalesce bursts through the reload queue instead of one full synchronous
+	// reload per event: independently-refreshing credentials can rewrite several
+	// files within the same second.
 
 	// 1. Handle traditional single identity.json file
 	if dir == filer.IamConfigDirectory {
-		// Handle create/update/delete events on legacy identity.json.
-		// During migration this file is renamed, which emits a delete event.
-		// Always reload from the credential manager so we keep the migrated identities.
 		if (oldEntry != nil && oldEntry.Name == filer.IamIdentityFile) ||
 			(newEntry != nil && newEntry.Name == filer.IamIdentityFile) {
-			if err := reloadIamConfig(dir + "/" + filer.IamIdentityFile); err != nil {
-				return err
-			}
+			s3a.iam.scheduleReload(dir + "/" + filer.IamIdentityFile)
 		}
 		return nil
 	}
 
-	// 2. Handle multiple-file identities and policies
-	// Watch /etc/iam/{identities,policies,service_accounts}
+	// 2. Handle multiple-file identities, policies, service accounts and groups
 	isIdentityDir := dir == filer.IamConfigDirectory+"/identities" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/identities/")
 	isPolicyDir := dir == filer.IamConfigDirectory+"/policies" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/policies/")
 	isServiceAccountDir := dir == filer.IamConfigDirectory+"/service_accounts" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/service_accounts/")
 	isGroupDir := dir == filer.IamConfigDirectory+"/groups" || strings.HasPrefix(dir, filer.IamConfigDirectory+"/groups/")
 
 	if isIdentityDir || isPolicyDir || isServiceAccountDir || isGroupDir {
-		// For multiple-file mode, any change in these directories should trigger a full reload
-		// from the credential manager (which handles the details of loading from multiple files).
-		if err := reloadIamConfig(dir); err != nil {
-			return err
-		}
+		s3a.iam.scheduleReload(dir)
 	}
 
 	return nil

weed/s3api/auth_credentials_subscribe_test.go

@@ -2,8 +2,11 @@ package s3api
 
 import (
 	"context"
+	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/credential"
 	_ "github.com/seaweedfs/seaweedfs/weed/credential/memory"
@@ -53,18 +56,15 @@ func TestOnIamConfigChangeReloadsOnIamIdentityDirectoryChanges(t *testing.T) {
 		t.Fatalf("failed to create alice in memory credential manager: %v", err)
 	}
 
-	err := s3a.onIamConfigChange(
+	if err := s3a.onIamConfigChange(
 		filer.IamConfigDirectory+"/identities",
 		nil,
 		&filer_pb.Entry{Name: "alice.json"},
-	)
-	if err != nil {
+	); err != nil {
 		t.Fatalf("onIamConfigChange returned error for identities directory update: %v", err)
 	}
 
-	if !hasIdentity(s3a.iam, "alice") {
-		t.Fatalf("expected alice identity to be loaded after /etc/iam/identities update")
-	}
+	waitForIdentity(t, s3a.iam, "alice")
 }
 
 func newTestS3ApiServerWithMemoryIAM(t *testing.T, identities []*iam_pb.Identity) *S3ApiServer {
@@ -101,9 +101,12 @@ func newTestS3ApiServerWithMemoryIAM(t *testing.T, identities []*iam_pb.Identity
 		hashCounters:      make(map[string]*int32),
 		isAuthEnabled:     false,
 		stopChan:          make(chan struct{}),
+		reloadCh:          make(chan struct{}, 1),
 		useStaticConfig:   false,
 		credentialManager: cm,
 	}
+	go iam.reloadRetryLoop()
+	t.Cleanup(iam.Shutdown)
 
 	// Load test configuration
 	if err := iam.ReplaceS3ApiConfiguration(config); err != nil {
@@ -122,3 +125,78 @@ func hasIdentity(iam *IdentityAccessManagement, identityName string) bool {
 	_, ok := iam.nameToIdentity[identityName]
 	return ok
 }
+
+func waitForIdentity(t *testing.T, iam *IdentityAccessManagement, name string) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !hasIdentity(iam, name) {
+		if time.Now().After(deadline) {
+			t.Fatalf("expected identity %s to be loaded", name)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func waitForIdentityGone(t *testing.T, iam *IdentityAccessManagement, name string) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for hasIdentity(iam, name) {
+		if time.Now().After(deadline) {
+			t.Fatalf("expected identity %s to be gone", name)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+// countingStore wraps a store and counts LoadConfiguration calls.
+type countingStore struct {
+	credential.CredentialStore
+	loads int64
+}
+
+func (c *countingStore) LoadConfiguration(ctx context.Context) (*iam_pb.S3ApiConfiguration, error) {
+	atomic.AddInt64(&c.loads, 1)
+	return c.CredentialStore.LoadConfiguration(ctx)
+}
+
+// A burst of IAM config change events must coalesce into a handful of reloads,
+// not one full reload per event.
+func TestOnIamConfigChangeCoalescesBurstReloads(t *testing.T) {
+	s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{{Name: "anonymous"}})
+
+	counter := &countingStore{CredentialStore: s3a.iam.credentialManager.Store}
+	s3a.iam.credentialManager.Store = counter
+
+	const burst = 50
+	for i := 0; i < burst; i++ {
+		if err := s3a.onIamConfigChange(
+			filer.IamConfigDirectory+"/identities",
+			nil,
+			&filer_pb.Entry{Name: fmt.Sprintf("u%d.json", i)},
+		); err != nil {
+			t.Fatalf("onIamConfigChange returned error: %v", err)
+		}
+	}
+
+	// Wait for the queue to drain: no pending signal.
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		s3a.iam.reloadMu.Lock()
+		empty := len(s3a.iam.reloadCh) == 0
+		s3a.iam.reloadMu.Unlock()
+		if empty {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("reload queue did not drain")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	// Let any final coalesced reload finish.
+	time.Sleep(50 * time.Millisecond)
+
+	loads := atomic.LoadInt64(&counter.loads)
+	if loads > 3 {
+		t.Fatalf("expected a burst of %d events to coalesce into <=3 reloads, got %d", burst, loads)
+	}
+}