fix(replication): collect site status without holding topology locks

Snapshot peers before concurrent status collection and release the state lock before local metadata and IAM reads. Keep result indexing, peer credentials and errors tied to that snapshot; return a copied current peer state with cached IAM data.

Cover storage, remote requests, configuration changes, cancellation and cached peer state. Confirm the healing entry and all five helpers with deterministic writer probes under the race detector. Refs #341.

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/iam-revision.go

@@ -724,7 +724,7 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) {
 		}
 	}
 	var progressMu sync.Mutex
-	cerr := c.concDo(nil, func(id string, p madmin.PeerInfo) error {
+	cerr := c.state.concDo(nil, func(id string, p madmin.PeerInfo) error {
 		// Bound each pass, but retain acknowledgements independently of the
 		// pass deadline or unrelated changes at either site.
 		peerCtx, cancel := context.WithTimeout(ctx, defaultContextTimeout)

cmd/site-replication-status_test.go

@@ -0,0 +1,198 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+)
+
+// Pause before the real IAM loader, which can itself read replication state.
+// TryLock below proves that a configuration writer can proceed at this point
+// without leaving deadlocked goroutines behind when this test regresses.
+type siteStatusPausedStore struct {
+	IAMStorageAPI
+	once             sync.Once
+	loading, proceed chan struct{}
+}
+
+func (s *siteStatusPausedStore) loadUsers(ctx context.Context, typ IAMUserType, users map[string]UserIdentity) error {
+	s.once.Do(func() { close(s.loading); <-s.proceed })
+	return s.IAMStorageAPI.loadUsers(ctx, typ, users)
+}
+
+func TestSiteReplicationStatusReadsStorageUnlocked(t *testing.T) {
+	for _, entry := range []string{"metadata", "status", "heal"} {
+		for _, change := range []string{"disable", "topology", "cancel"} {
+			t.Run(entry+"/"+change, func(t *testing.T) {
+				site := newSvcAccSite(t)
+				c := &globalSiteReplicationSys
+				paused := &siteStatusPausedStore{IAMStorageAPI: site.sys.store.IAMStorageAPI, loading: make(chan struct{}), proceed: make(chan struct{})}
+				site.sys.store.IAMStorageAPI = paused
+				release := sync.OnceFunc(func() { close(paused.proceed) })
+				defer release()
+				c.iamMetaCache.Lock()
+				c.iamMetaCache.lastUpdate = time.Time{}
+				c.iamMetaCache.Unlock()
+				c.RLock()
+				peers := maps.Clone(c.state.Peers)
+				c.RUnlock()
+				ctx, cancel := context.WithCancel(site.ctx)
+				defer cancel()
+				var meta madmin.SRInfo
+				var status srStatusInfo
+				done := make(chan error, 1)
+				go func() {
+					var err error
+					opts := madmin.SRStatusOptions{Users: true, Groups: true, Policies: true, PeerState: true}
+					switch entry {
+					case "metadata":
+						meta, err = c.SiteReplicationMetaInfo(ctx, site.obj, opts)
+					case "status":
+						status, err = c.siteReplicationStatus(ctx, site.obj, opts)
+					case "heal":
+						err = c.healIAMSystem(ctx, site.obj)
+					}
+					done <- err
+				}()
+				select {
+				case <-paused.loading:
+				case err := <-done:
+					t.Fatalf("did not read IAM storage: %v", err)
+				case <-time.After(10 * time.Second):
+					t.Fatal("did not reach IAM loader")
+				}
+				if c.TryLock() {
+					switch change {
+					case "disable":
+						c.enabled = false
+						c.state = srState{}
+					case "topology":
+						delete(c.state.Peers, "remote")
+						c.state.UpdatedAt = UTCNow()
+					}
+					c.Unlock()
+				} else {
+					t.Error("replication configuration is locked during IAM storage read")
+				}
+				if change == "cancel" {
+					cancel()
+				}
+				release()
+				select {
+				case err := <-done:
+					if change != "cancel" {
+						mustIAM(t, err)
+					}
+				case <-time.After(10 * time.Second):
+					t.Fatal("operation did not finish")
+				}
+				if change != "cancel" {
+					if entry == "metadata" && !reflect.DeepEqual(meta.State.Peers, peers) || entry == "status" && !reflect.DeepEqual(status.Sites, peers) {
+						t.Fatal("result topology changed after its snapshot")
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestSiteReplicationStatusRemoteUnlocked(t *testing.T) {
+	for _, fail := range []bool{false, true} {
+		t.Run(map[bool]string{false: "success", true: "peer-error"}[fail], func(t *testing.T) {
+			site := newSvcAccSite(t)
+			entered, proceed := make(chan struct{}), make(chan struct{})
+			var once sync.Once
+			peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				once.Do(func() { close(entered) })
+				<-proceed
+				w.Header().Set("Content-Type", "application/json")
+				if fail {
+					w.WriteHeader(http.StatusForbidden)
+					_ = json.NewEncoder(w).Encode(madmin.ErrorResponse{Code: "AccessDenied", Message: "denied"})
+					return
+				}
+				_ = json.NewEncoder(w).Encode(madmin.SRInfo{DeploymentID: "remote"})
+			}))
+			defer peer.Close()
+			release := sync.OnceFunc(func() { close(proceed) })
+			defer release()
+			c := &globalSiteReplicationSys
+			c.Lock()
+			c.state.Peers = site.peers(peer.URL)
+			c.state.UpdatedAt = UTCNow()
+			updated := c.state.UpdatedAt
+			c.Unlock()
+			var status srStatusInfo
+			done := make(chan error, 1)
+			go func() {
+				var err error
+				status, err = c.siteReplicationStatus(site.ctx, site.obj, madmin.SRStatusOptions{})
+				done <- err
+			}()
+			select {
+			case <-entered:
+			case err := <-done:
+				t.Fatalf("did not request remote status: %v", err)
+			case <-time.After(10 * time.Second):
+				t.Fatal("did not reach peer")
+			}
+			if c.TryLock() {
+				delete(c.state.Peers, "remote")
+				c.state.UpdatedAt = updated.Add(time.Second)
+				c.Unlock()
+			} else {
+				t.Error("replication configuration is locked during remote request")
+			}
+			release()
+			select {
+			case err := <-done:
+				if fail {
+					if err == nil || !strings.Contains(err.Error(), "site remote (remote)") {
+						t.Fatalf("lost original peer name in error: %v", err)
+					}
+				} else {
+					mustIAM(t, err)
+					if len(status.Sites) != 2 || status.Sites["remote"].Endpoint != peer.URL || status.UpdatedAt != updated {
+						t.Fatal("status mixed different topology revisions")
+					}
+				}
+			case <-time.After(10 * time.Second):
+				t.Fatal("status did not finish")
+			}
+		})
+	}
+}
+
+func TestSiteReplicationMetaInfoCachedPeerState(t *testing.T) {
+	site := newSvcAccSite(t)
+	c := &globalSiteReplicationSys
+	c.srCacheIAMInfo(madmin.SRInfo{DeploymentID: globalDeploymentID(), State: madmin.SRStateInfo{Name: "obsolete"}})
+	opts := madmin.SRStatusOptions{Users: true, Groups: true, Policies: true, PeerState: true}
+	got, err := c.SiteReplicationMetaInfo(site.ctx, site.obj, opts)
+	mustIAM(t, err)
+	if !reflect.DeepEqual(got.State.Peers, c.state.Peers) || got.State.Name != c.state.Name {
+		t.Fatal("cached IAM response returned stale peer state")
+	}
+	delete(got.State.Peers, "remote")
+	if len(c.state.Peers) != 2 {
+		t.Fatal("returned peer map aliases the replication configuration")
+	}
+	opts.PeerState = false
+	got, err = c.SiteReplicationMetaInfo(site.ctx, site.obj, opts)
+	mustIAM(t, err)
+	if got.State.Name != "" || got.State.Peers != nil {
+		t.Fatal("returned cached peer state that was not requested")
+	}
+}

cmd/site-replication.go

@@ -831,7 +831,7 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string,
 	opts.CreatedAt = createdAt
 
 	// Create bucket and enable versioning on all peers.
-	makeBucketConcErr := c.concDo(
+	makeBucketConcErr := c.state.concDo(
 		func() error {
 			return c.annotateErr(makeBucketWithVersion, c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts))
 		},
@@ -847,7 +847,7 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string,
 	)
 
 	// Create bucket remotes and add replication rules for the bucket on self and peers.
-	makeRemotesConcErr := c.concDo(
+	makeRemotesConcErr := c.state.concDo(
 		func() error {
 			return c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket))
 		},
@@ -886,7 +886,7 @@ func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string
 	}
 
 	// Send bucket delete to other clusters.
-	cerr := c.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error {
+	cerr := c.state.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error {
 		admClient, err := c.getAdminClient(ctx, deploymentID)
 		if err != nil {
 			return wrapSRErr(err)
@@ -982,7 +982,7 @@ func (c *SiteReplicationSys) PeerBucketMakeWithVersioningHandler(ctx context.Con
 // PeerBucketConfigureReplHandler - configures replication remote and
 // replication rules to all other peers for the local bucket.
 func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, bucket string) error {
-	creds, err := c.getPeerCreds()
+	creds, err := c.state.getPeerCreds()
 	if err != nil {
 		return wrapSRErr(err)
 	}
@@ -1199,7 +1199,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context,
 		}
 		errMap[d] = configurePeerFn(d, peer)
 	}
-	return c.toErrorFromErrMap(errMap, configureReplication)
+	return c.state.toErrorFromErrMap(errMap, configureReplication)
 }
 
 // PeerBucketDeleteHandler - deletes bucket on local in response to a delete
@@ -1291,7 +1291,7 @@ func (c *SiteReplicationSys) IAMChangeHook(ctx context.Context, item madmin.SRIA
 	if !c.enabled {
 		return nil
 	}
-	cerr := c.concDo(nil, func(d string, p madmin.PeerInfo) error {
+	cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error {
 		admClient, err := c.getAdminClient(ctx, d)
 		if err != nil {
 			return wrapSRErr(err)
@@ -1608,7 +1608,7 @@ func (c *SiteReplicationSys) BucketMetaHook(ctx context.Context, item madmin.SRB
 		return nil
 	}
 
-	cerr := c.concDo(nil, func(d string, p madmin.PeerInfo) error {
+	cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error {
 		admClient, err := c.getAdminClient(ctx, d)
 		if err != nil {
 			return wrapSRErr(err)
@@ -2110,7 +2110,7 @@ func (c *SiteReplicationSys) PeerBucketLCConfigHandler(ctx context.Context, buck
 // getAdminClient - NOTE: ensure to take at least a read lock on SiteReplicationSys
 // before calling this.
 func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID string) (*madmin.AdminClient, error) {
-	creds, err := c.getPeerCreds()
+	creds, err := c.state.getPeerCreds()
 	if err != nil {
 		return nil, err
 	}
@@ -2126,7 +2126,7 @@ func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID st
 // getAdminClientWithEndpoint - NOTE: ensure to take at least a read lock on SiteReplicationSys
 // before calling this.
 func (c *SiteReplicationSys) getAdminClientWithEndpoint(ctx context.Context, deploymentID, endpoint string) (*madmin.AdminClient, error) {
-	creds, err := c.getPeerCreds()
+	creds, err := c.state.getPeerCreds()
 	if err != nil {
 		return nil, err
 	}
@@ -2137,8 +2137,8 @@ func (c *SiteReplicationSys) getAdminClientWithEndpoint(ctx context.Context, dep
 	return getAdminClient(endpoint, creds.AccessKey, creds.SecretKey)
 }
 
-func (c *SiteReplicationSys) getPeerCreds() (*auth.Credentials, error) {
+func (s srState) getPeerCreds() (*auth.Credentials, error) {
-	u, ok := globalIAMSys.store.GetUser(c.state.ServiceAccountAccessKey)
+	u, ok := globalIAMSys.store.GetUser(s.ServiceAccountAccessKey)
 	if !ok {
 		return nil, errors.New("site replication service account not found")
 	}
@@ -2460,7 +2460,7 @@ func (c concErr) Unwrap() error {
 	return c.summaryErr
 }
 
-func (c *SiteReplicationSys) toErrorFromErrMap(errMap map[string]error, actionName string) error {
+func (s srState) toErrorFromErrMap(errMap map[string]error, actionName string) error {
 	if len(errMap) == 0 {
 		return nil
 	}
@@ -2475,7 +2475,7 @@ func (c *SiteReplicationSys) toErrorFromErrMap(errMap map[string]error, actionNa
 	var success int
 	msgs := []string{}
 	for _, d := range keys {
-		name := c.state.Peers[d].Name
+		name := s.Peers[d].Name
 		err := errMap[d]
 		if err == nil {
 			msgs = append(msgs, fmt.Sprintf("'%s' on site %s (%s): succeeded", actionName, name, d))
@@ -2490,21 +2490,22 @@ func (c *SiteReplicationSys) toErrorFromErrMap(errMap map[string]error, actionNa
 	return fmt.Errorf("Site replication error(s): \n%s", strings.Join(msgs, "\n"))
 }
 
-func (c *SiteReplicationSys) newConcErr(errMap map[string]error, actionName string) error {
+func (s srState) newConcErr(errMap map[string]error, actionName string) error {
 	return concErr{
 		errMap:     errMap,
-		summaryErr: c.toErrorFromErrMap(errMap, actionName),
+		summaryErr: s.toErrorFromErrMap(errMap, actionName),
 	}
 }
 
 // concDo calls actions concurrently. selfActionFn is run for the current
-// cluster and peerActionFn is run for each peer replication cluster.
+// cluster and peerActionFn is run for each peer replication cluster. The state
-func (c *SiteReplicationSys) concDo(selfActionFn func() error, peerActionFn func(deploymentID string, p madmin.PeerInfo) error, actionName string) error {
+// must be a snapshot or remain protected by the caller throughout the call.
-	depIDs := make([]string, 0, len(c.state.Peers))
+func (s srState) concDo(selfActionFn func() error, peerActionFn func(deploymentID string, p madmin.PeerInfo) error, actionName string) error {
-	for d := range c.state.Peers {
+	depIDs := make([]string, 0, len(s.Peers))
+	for d := range s.Peers {
 		depIDs = append(depIDs, d)
 	}
-	errs := make([]error, len(c.state.Peers))
+	errs := make([]error, len(s.Peers))
 	var wg sync.WaitGroup
 	wg.Add(len(depIDs))
 	for i := range depIDs {
@@ -2515,23 +2516,23 @@ func (c *SiteReplicationSys) concDo(selfActionFn func() error, peerActionFn func
 					errs[i] = selfActionFn()
 				}
 			} else {
-				errs[i] = peerActionFn(depIDs[i], c.state.Peers[depIDs[i]])
+				errs[i] = peerActionFn(depIDs[i], s.Peers[depIDs[i]])
 			}
 		}(i)
 	}
 	wg.Wait()
-	errMap := make(map[string]error, len(c.state.Peers))
+	errMap := make(map[string]error, len(s.Peers))
 	for i, depID := range depIDs {
 		errMap[depID] = errs[i]
 		if errs[i] != nil && minio.IsNetworkOrHostDown(errs[i], true) {
-			ep := c.state.Peers[depID].Endpoint
+			ep := s.Peers[depID].Endpoint
 			epURL, _ := url.Parse(ep)
 			if !globalBucketTargetSys.isOffline(epURL) {
 				globalBucketTargetSys.markOffline(epURL)
 			}
 		}
 	}
-	return c.newConcErr(errMap, actionName)
+	return s.newConcErr(errMap, actionName)
 }
 
 func (c *SiteReplicationSys) annotateErr(annotation string, err error) error {
@@ -3004,20 +3005,26 @@ const (
 // siteReplicationStatus returns the site replication status across clusters participating in site replication.
 func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI ObjectLayer, opts madmin.SRStatusOptions) (info srStatusInfo, err error) {
 	c.RLock()
-	defer c.RUnlock()
 	if !c.enabled {
-		return info, err
+		c.RUnlock()
+		return info, nil
 	}
+	state := c.state
+	state.Peers = maps.Clone(state.Peers)
+	c.RUnlock()
+	// Local IAM reads can re-enter the replication lock. Remote work must
+	// also leave topology changes free to proceed. Use one peer snapshot for
+	// dispatch, result indexing and error reporting throughout this request.
 
-	sris := make([]madmin.SRInfo, len(c.state.Peers))
+	sris := make([]madmin.SRInfo, len(state.Peers))
-	depIdx := make(map[string]int, len(c.state.Peers))
+	depIdx := make(map[string]int, len(state.Peers))
 	i := 0
-	for d := range c.state.Peers {
+	for d := range state.Peers {
 		depIdx[d] = i
 		i++
 	}
 
-	metaInfoConcErr := c.concDo(
+	metaInfoConcErr := state.concDo(
 		func() error {
 			srInfo, err := c.SiteReplicationMetaInfo(ctx, objAPI, opts)
 			if err != nil {
@@ -3027,7 +3034,11 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 			return nil
 		},
 		func(deploymentID string, p madmin.PeerInfo) error {
-			admClient, err := c.getAdminClient(ctx, deploymentID)
+			creds, err := state.getPeerCreds()
+			if err != nil {
+				return err
+			}
+			admClient, err := getAdminClient(p.Endpoint, creds.AccessKey, creds.SecretKey)
 			if err != nil {
 				switch err.(type) {
 				case RemoteTargetConnectionErr:
@@ -3051,9 +3062,8 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 	}
 
 	info.Enabled = true
-	info.Sites = make(map[string]madmin.PeerInfo, len(c.state.Peers))
+	info.Sites = state.Peers
-	maps.Copy(info.Sites, c.state.Peers)
+	info.UpdatedAt = state.UpdatedAt
-	info.UpdatedAt = c.state.UpdatedAt
 
 	var maxBuckets int
 	for _, sri := range sris {
@@ -3162,7 +3172,7 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 		}
 	}
 
-	info.StatsSummary = make(map[string]madmin.SRSiteSummary, len(c.state.Peers))
+	info.StatsSummary = make(map[string]madmin.SRSiteSummary, len(state.Peers))
 	info.BucketStats = make(map[string]map[string]srBucketStatsSummary)
 	info.PolicyStats = make(map[string]map[string]srPolicyStatsSummary)
 	info.UserStats = make(map[string]map[string]srUserStatsSummary)
@@ -3919,10 +3929,18 @@ func (c *SiteReplicationSys) SiteReplicationMetaInfo(ctx context.Context, objAPI
 		return info, errSRObjectLayerNotReady
 	}
 	c.RLock()
-	defer c.RUnlock()
 	if !c.enabled {
+		c.RUnlock()
 		return info, nil
 	}
+	if opts.PeerState {
+		info.State = madmin.SRStateInfo{
+			Name:      c.state.Name,
+			Peers:     maps.Clone(c.state.Peers),
+			UpdatedAt: c.state.UpdatedAt,
+		}
+	}
+	c.RUnlock()
 	info.DeploymentID = globalDeploymentID()
 	if opts.Buckets || opts.Entity == madmin.SRBucketEntity {
 		var (
@@ -4045,6 +4063,8 @@ func (c *SiteReplicationSys) SiteReplicationMetaInfo(ctx context.Context, objAPI
 		// serialize SiteReplicationMetaInfo calls - if data in cache is within
 		// healing interval, avoid fetching IAM data again from disk.
 		if metaInfo, ok := c.getSRCachedIAMInfo(); ok {
+			// PeerState belongs to this request, not to the cached IAM data.
+			metaInfo.State = info.State
 			return metaInfo, nil
 		}
 	}
@@ -4115,13 +4135,6 @@ func (c *SiteReplicationSys) SiteReplicationMetaInfo(ctx context.Context, objAPI
 			maps.Copy(info.ILMExpiryRules, allRules)
 		}
 	}
-	if opts.PeerState {
-		info.State = madmin.SRStateInfo{
-			Name:      c.state.Name,
-			Peers:     c.state.Peers,
-			UpdatedAt: c.state.UpdatedAt,
-		}
-	}
 
 	if opts.Users || opts.Entity == madmin.SRUserEntity {
 		// Replicate policy mappings on local to all peers.