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.