From 2f228284f31ded34ff34442b239a7e1589caa778 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 11 Nov 2026 14:43:42 +0800 Subject: [PATCH] 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 --- cmd/iam-revision.go | 2 +- cmd/site-replication-status_test.go | 198 ++++++++++++++++++++++++++++ cmd/site-replication.go | 99 ++++++++------ 3 files changed, 255 insertions(+), 44 deletions(-) create mode 100644 cmd/site-replication-status_test.go diff --git a/cmd/iam-revision.go b/cmd/iam-revision.go index 09a8ae6b5..2eb215c64 100644 --- a/cmd/iam-revision.go +++ b/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) diff --git a/cmd/site-replication-status_test.go b/cmd/site-replication-status_test.go new file mode 100644 index 000000000..d5170b943 --- /dev/null +++ b/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") + } +} diff --git a/cmd/site-replication.go b/cmd/site-replication.go index 95bdcd379..4fb0cd38b 100644 --- a/cmd/site-replication.go +++ b/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) { - u, ok := globalIAMSys.store.GetUser(c.state.ServiceAccountAccessKey) +func (s srState) getPeerCreds() (*auth.Credentials, error) { + 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. -func (c *SiteReplicationSys) concDo(selfActionFn func() error, peerActionFn func(deploymentID string, p madmin.PeerInfo) error, actionName string) error { - depIDs := make([]string, 0, len(c.state.Peers)) - for d := range c.state.Peers { +// cluster and peerActionFn is run for each peer replication cluster. The state +// must be a snapshot or remain protected by the caller throughout the call. +func (s srState) concDo(selfActionFn func() error, peerActionFn func(deploymentID string, p madmin.PeerInfo) error, actionName string) error { + 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)) - depIdx := make(map[string]int, len(c.state.Peers)) + sris := make([]madmin.SRInfo, len(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)) - maps.Copy(info.Sites, c.state.Peers) - info.UpdatedAt = c.state.UpdatedAt + info.Sites = state.Peers + info.UpdatedAt = 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.