From ef6efd3120737506f9dd706e2c8564647b2c5685 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 11 Nov 2026 15:43:41 +0800 Subject: [PATCH] fix(site-replication): snapshot state before bucket healing and peer hooks Snapshot site-replication topology and release the state read lock before bucket healing, metadata healing, IAM deletion healing, and replication hooks perform storage I/O or peer RPCs. Guard remaining peer-state reads and cloning edits so concurrent configuration updates neither deadlock waiting writers nor race on state maps. Signed-off-by: Feng Ruohang --- cmd/iam-revision.go | 11 +- cmd/site-replication-metadata.go | 7 +- cmd/site-replication-status_test.go | 321 +++++++++++++++++++++++++++ cmd/site-replication.go | 322 ++++++++++++++++------------ 4 files changed, 510 insertions(+), 151 deletions(-) diff --git a/cmd/iam-revision.go b/cmd/iam-revision.go index 2eb215c64..96b1e9136 100644 --- a/cmd/iam-revision.go +++ b/cmd/iam-revision.go @@ -676,9 +676,8 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) { }() c.iamHealMu.Lock() defer c.iamHealMu.Unlock() - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } snapshot := globalIAMSys.store.revisionIndex().snapshot() @@ -719,17 +718,17 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) { c.iamRevisionProgress = make(map[string]iamRevisionProgress) } for id := range c.iamRevisionProgress { - if _, present := c.state.Peers[id]; !present { + if _, present := state.Peers[id]; !present { delete(c.iamRevisionProgress, id) } } var progressMu sync.Mutex - cerr := c.state.concDo(nil, func(id string, p madmin.PeerInfo) error { + cerr := 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) defer cancel() - client, err := c.getAdminClient(peerCtx, id) + client, err := state.getAdminClient(id) if err != nil { return err } diff --git a/cmd/site-replication-metadata.go b/cmd/site-replication-metadata.go index 356fb4e4b..3dee6398b 100644 --- a/cmd/site-replication-metadata.go +++ b/cmd/site-replication-metadata.go @@ -126,9 +126,8 @@ func latestBucketConfig(bucket, file string, info srStatusInfo) (bucketConfigSta } func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file string, info srStatusInfo) error { - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } latest, found := latestBucketConfig(bucket, file, info) @@ -177,7 +176,7 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file _, err = globalBucketMetadataSys.updateAndParseMetadata(ctx, bucket, file, latest.data, false, false, &latest.at) } else { var client *madmin.AdminClient - client, err = c.getAdminClient(ctx, id) + client, err = state.getAdminClient(id) if err == nil { err = client.SRPeerReplicateBucketMeta(ctx, newBucketConfigReplicationEvent(bucket, file, incoming)) } diff --git a/cmd/site-replication-status_test.go b/cmd/site-replication-status_test.go index d5170b943..9a0e56eab 100644 --- a/cmd/site-replication-status_test.go +++ b/cmd/site-replication-status_test.go @@ -5,6 +5,7 @@ package cmd import ( "context" + "encoding/base64" "encoding/json" "maps" "net/http" @@ -196,3 +197,323 @@ func TestSiteReplicationMetaInfoCachedPeerState(t *testing.T) { t.Fatal("returned cached peer state that was not requested") } } + +type siteBucketDeletePausedLayer struct { + ObjectLayer + once sync.Once + deleting, resume chan struct{} +} + +func (l *siteBucketDeletePausedLayer) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error { + l.once.Do(func() { close(l.deleting); <-l.resume }) + return l.ObjectLayer.DeleteBucket(ctx, bucket, opts) +} + +func TestSiteReplicationHealAndHooksReleaseLock(t *testing.T) { + cases := []string{ + "make-bucket-hook", + "delete-bucket-hook", + "bucket-meta-hook", + "iam-change-hook", + "configure-repl", + "peer-bucket-delete", + "heal-bucket-make", + "heal-bucket-delete", + "heal-bucket-repl-config", + "heal-bucket-config", + "heal-cors", + "heal-ilm-expiry-config", + "heal-bucket-ilm-expiry", + } + for _, tc := range cases { + t.Run(tc, func(t *testing.T) { + site := newSvcAccSite(t) + c := &globalSiteReplicationSys + const bucket = "sr-lock-test-bucket" + mustIAM(t, site.obj.MakeBucket(site.ctx, bucket, MakeBucketOptions{VersioningEnabled: true})) + _, err := globalBucketMetadataSys.Update(site.ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig) + mustIAM(t, err) + + entered, proceed := make(chan struct{}), make(chan struct{}) + var once sync.Once + pause := func() { + once.Do(func() { close(entered) }) + <-proceed + } + release := sync.OnceFunc(func() { close(proceed) }) + defer release() + + peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + pause() + if strings.HasSuffix(r.URL.Path, iamRevisionPeerPath) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(iamRevisionResponse{ + iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: "node", Instance: "boot", Digest: "fixture"}, + }) + return + } + if r.Method == http.MethodGet { + if _, ok := r.URL.Query()["versioning"]; ok { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(`Enabled`)) + return + } + if _, ok := r.URL.Query()["location"]; ok { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(``)) + return + } + } + w.WriteHeader(http.StatusOK) + })) + defer peer.Close() + + pausedObj := &siteBucketDeletePausedLayer{ + ObjectLayer: site.obj, + deleting: entered, + resume: proceed, + } + if tc == "peer-bucket-delete" || tc == "heal-bucket-delete" { + setObjectLayer(pausedObj) + defer setObjectLayer(site.obj) + } + + local := globalDeploymentID() + now := UTCNow().Truncate(time.Second) + peers := site.peers(peer.URL) + for id, p := range peers { + p.ReplicateILMExpiry = true + peers[id] = p + } + c.Lock() + c.enabled = true + c.state = srState{ + Name: "local", + ServiceAccountAccessKey: site.parent.AccessKey, + Peers: peers, + UpdatedAt: now, + } + c.Unlock() + + done := make(chan error, 1) + go func() { + switch tc { + case "make-bucket-hook": + done <- c.MakeBucketHook(site.ctx, bucket, MakeBucketOptions{CreatedAt: now}) + case "delete-bucket-hook": + done <- c.DeleteBucketHook(site.ctx, bucket, true) + case "bucket-meta-hook": + done <- c.BucketMetaHook(site.ctx, madmin.SRBucketMeta{ + Type: madmin.SRBucketMetaTypeTags, + Bucket: bucket, + UpdatedAt: now, + }) + case "iam-change-hook": + site.create(t, "sr-lock-hook-sa") + done <- c.IAMChangeHook(site.ctx, madmin.SRIAMItem{ + Type: madmin.SRIAMItemSvcAcc, + SvcAccChange: &madmin.SRSvcAccChange{Create: &madmin.SRSvcAccCreate{AccessKey: "sr-lock-hook-sa"}}, + UpdatedAt: now, + }) + case "configure-repl": + done <- c.PeerBucketConfigureReplHandler(site.ctx, bucket) + case "peer-bucket-delete": + done <- c.PeerBucketDeleteHandler(site.ctx, bucket, DeleteBucketOptions{Force: true}) + case "heal-bucket-make": + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now}}}, + "remote": {HasBucket: false}, + }, + }, + } + done <- c.healBucket(site.ctx, site.obj, bucket, info) + case "heal-bucket-delete": + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), DeletedAt: now}}}, + "remote": {HasBucket: false, BucketMarkedDeleted: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{DeletedAt: now.Add(-2 * time.Minute)}}}, + }, + }, + } + done <- c.healBucket(site.ctx, pausedObj, bucket, info) + case "heal-bucket-repl-config": + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, ReplicationCfgMismatch: true}, + }, + }, + } + done <- c.healBucketReplicationConfig(site.ctx, site.obj, bucket, info, &validateReplicationDestinationOptions{CheckReady: true}) + case "heal-bucket-config": + tagPayload := base64.StdEncoding.EncodeToString([]byte(`kv`)) + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), Tags: &tagPayload, TagConfigUpdatedAt: now}}}, + "remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}}, + }, + }, + } + done <- c.healBucketConfig(site.ctx, bucket, bucketTaggingConfig, info) + case "heal-cors": + corsPayload := base64.StdEncoding.EncodeToString([]byte(`*GET`)) + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), CorsConfig: &corsPayload, CorsConfigUpdatedAt: now}}}, + "remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}}, + }, + }, + } + done <- c.healCORSMetadata(site.ctx, site.obj, bucket, info) + case "heal-ilm-expiry-config": + stalePeers := site.peers(peer.URL) + for id, p := range stalePeers { + p.ReplicateILMExpiry = false + stalePeers[id] = p + } + info := srStatusInfo{ + Sites: site.peers(peer.URL), + PeerStates: map[string]madmin.SRStateInfo{ + local: {Name: "local", Peers: peers, UpdatedAt: now}, + "remote": {Name: "remote", Peers: stalePeers, UpdatedAt: now.Add(-time.Minute)}, + }, + } + done <- c.healILMExpiryConfig(site.ctx, site.obj, info) + case "heal-bucket-ilm-expiry": + lcXML := `expEnabled1` + lcPayload := base64.StdEncoding.EncodeToString([]byte(lcXML)) + info := srStatusInfo{ + Sites: site.peers(peer.URL), + BucketStats: map[string]map[string]srBucketStatsSummary{ + bucket: { + local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), ExpiryLCConfig: &lcPayload, ExpiryLCConfigUpdatedAt: now}}}, + "remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}}, + }, + }, + } + done <- c.healBucketILMExpiry(site.ctx, site.obj, bucket, info) + } + }() + + select { + case <-entered: + case err := <-done: + t.Fatalf("operation returned before reaching storage/peer pause point: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("timed out waiting for storage/peer pause point") + } + + if c.TryLock() { + c.state.Name = "mutated-while-in-flight" + delete(c.state.Peers, "remote") + c.state.UpdatedAt = now.Add(time.Second) + c.Unlock() + } else { + t.Error("SiteReplicationSys lock held across storage I/O or peer RPC") + } + + release() + select { + case err := <-done: + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("operation deadlocked after writer mutation") + } + }) + } +} + +func TestSiteReplicationConcurrentStateUpdates(t *testing.T) { + site := newSvcAccSite(t) + c := &globalSiteReplicationSys + const bucket = "sr-concurrent-state-bucket" + mustIAM(t, site.obj.MakeBucket(site.ctx, bucket, MakeBucketOptions{VersioningEnabled: true})) + _, err := globalBucketMetadataSys.Update(site.ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig) + mustIAM(t, err) + + peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "metainfo") { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(madmin.SRInfo{DeploymentID: "remote"}) + return + } + if r.Method == http.MethodGet { + if _, ok := r.URL.Query()["versioning"]; ok { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(`Enabled`)) + return + } + if _, ok := r.URL.Query()["location"]; ok { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(``)) + return + } + } + w.WriteHeader(http.StatusOK) + })) + defer peer.Close() + + basePeers := site.peers(peer.URL) + now := UTCNow().Truncate(time.Second) + c.Lock() + c.enabled = true + c.state = srState{ + Name: "local", + ServiceAccountAccessKey: site.parent.AccessKey, + Peers: maps.Clone(basePeers), + UpdatedAt: now, + } + c.Unlock() + + var wg sync.WaitGroup + wg.Add(4) + go func() { + defer wg.Done() + for i := range 10 { + _ = c.PeerEditReq(site.ctx, madmin.PeerInfo{ + DeploymentID: "remote", + Name: "remote", + Endpoint: peer.URL, + ReplicateILMExpiry: i%2 == 0, + }) + } + }() + go func() { + defer wg.Done() + for i := range 10 { + peers := site.peers(peer.URL) + p := peers["remote"] + p.ReplicateILMExpiry = i%2 == 1 + peers["remote"] = p + _ = c.PeerStateEditReq(site.ctx, madmin.SRStateEditReq{ + Peers: peers, + UpdatedAt: now.Add(time.Duration(i+1) * time.Second), + }) + } + }() + go func() { + defer wg.Done() + for range 10 { + _ = c.PeerBucketConfigureReplHandler(site.ctx, bucket) + } + }() + go func() { + defer wg.Done() + for range 5 { + _ = c.healBuckets(site.ctx, site.obj) + } + }() + wg.Wait() +} diff --git a/cmd/site-replication.go b/cmd/site-replication.go index 4fb0cd38b..41471fd63 100644 --- a/cmd/site-replication.go +++ b/cmd/site-replication.go @@ -516,12 +516,10 @@ func (c *SiteReplicationSys) AddPeerClusters(ctx context.Context, psites []madmi } // check if few peers exist already and ILM expiry replcation is set to true replicateILMExpirySet := false - if c.state.Peers != nil { - for _, pi := range c.state.Peers { - if pi.ReplicateILMExpiry { - replicateILMExpirySet = true - break - } + for _, pi := range c.cloneState().Peers { + if pi.ReplicateILMExpiry { + replicateILMExpirySet = true + break } } for _, v := range sites { @@ -644,15 +642,14 @@ func (c *SiteReplicationSys) PeerJoinReq(ctx context.Context, arg madmin.SRPeerJ return errSRServiceAccount(fmt.Errorf("unable to create service account on %s: %v", ourName, err)) } + prevPeers := c.cloneState().Peers peers := make(map[string]madmin.PeerInfo, len(arg.Peers)) for dID, pi := range arg.Peers { - if c.state.Peers != nil { - if existingPeer, ok := c.state.Peers[dID]; ok { - // retain existing ReplicateILMExpiry of peer if its already set - // and incoming arg has it false. it could be default false - if !pi.ReplicateILMExpiry && existingPeer.ReplicateILMExpiry { - pi.ReplicateILMExpiry = existingPeer.ReplicateILMExpiry - } + if existingPeer, ok := prevPeers[dID]; ok { + // retain existing ReplicateILMExpiry of peer if its already set + // and incoming arg has it false. it could be default false + if !pi.ReplicateILMExpiry && existingPeer.ReplicateILMExpiry { + pi.ReplicateILMExpiry = existingPeer.ReplicateILMExpiry } } peers[dID] = pi @@ -806,9 +803,8 @@ const ( func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, opts MakeBucketOptions) error { // At this point, the local bucket is created. - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } @@ -831,33 +827,33 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, opts.CreatedAt = createdAt // Create bucket and enable versioning on all peers. - makeBucketConcErr := c.state.concDo( + makeBucketConcErr := state.concDo( func() error { - return c.annotateErr(makeBucketWithVersion, c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)) + return state.annotateErr(makeBucketWithVersion, c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)) }, func(deploymentID string, p madmin.PeerInfo) error { - admClient, err := c.getAdminClient(ctx, deploymentID) + admClient, err := state.getAdminClient(deploymentID) if err != nil { return err } - return c.annotatePeerErr(p.Name, makeBucketWithVersion, admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap)) + return state.annotatePeerErr(p.Name, makeBucketWithVersion, admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap)) }, makeBucketWithVersion, ) // Create bucket remotes and add replication rules for the bucket on self and peers. - makeRemotesConcErr := c.state.concDo( + makeRemotesConcErr := state.concDo( func() error { - return c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)) + return state.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)) }, func(deploymentID string, p madmin.PeerInfo) error { - admClient, err := c.getAdminClient(ctx, deploymentID) + admClient, err := state.getAdminClient(deploymentID) if err != nil { return err } - return c.annotatePeerErr(p.Name, configureReplication, admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil)) + return state.annotatePeerErr(p.Name, configureReplication, admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil)) }, configureReplication, ) @@ -874,9 +870,8 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string, forceDelete bool) error { // At this point, the local bucket is deleted. - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } @@ -886,13 +881,13 @@ func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string } // Send bucket delete to other clusters. - cerr := c.state.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error { - admClient, err := c.getAdminClient(ctx, deploymentID) + cerr := state.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error { + admClient, err := state.getAdminClient(deploymentID) if err != nil { return wrapSRErr(err) } - return c.annotatePeerErr(p.Name, deleteBucket, admClient.SRPeerBucketOps(ctx, bucket, op, nil)) + return state.annotatePeerErr(p.Name, deleteBucket, admClient.SRPeerBucketOps(ctx, bucket, op, nil)) }, deleteBucket, ) @@ -982,7 +977,8 @@ 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.state.getPeerCreds() + state := c.cloneState() + creds, err := state.getPeerCreds() if err != nil { return wrapSRErr(err) } @@ -1068,7 +1064,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, } err := globalBucketTargetSys.SetTarget(ctx, bucket, &targetToUpdate, true) if err != nil { - return c.annotatePeerErr(peer.Name, "Bucket target update error", err) + return state.annotatePeerErr(peer.Name, "Bucket target update error", err) } targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket) if err != nil { @@ -1105,7 +1101,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, if !exists { // persist newly generated ARN to targets and metadata on disk err := globalBucketTargetSys.SetTarget(ctx, bucket, &bucketTarget, false) if err != nil { - return c.annotatePeerErr(peer.Name, "Bucket target creation error", err) + return state.annotatePeerErr(peer.Name, "Bucket target creation error", err) } targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket) if err != nil { @@ -1159,7 +1155,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, err = replicationConfig.AddRule(opts) } if err != nil { - return c.annotatePeerErr(peer.Name, "Error adding bucket replication rule", err) + return state.annotatePeerErr(peer.Name, "Error adding bucket replication rule", err) } // Now convert the configuration back to server's type so we can @@ -1187,27 +1183,23 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, } _, err = globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, replCfgData) - return c.annotatePeerErr(peer.Name, "Error updating replication configuration", err) + return state.annotatePeerErr(peer.Name, "Error updating replication configuration", err) } - c.RLock() - defer c.RUnlock() - errMap := make(map[string]error, len(c.state.Peers)) - for d, peer := range c.state.Peers { + errMap := make(map[string]error, len(state.Peers)) + for d, peer := range state.Peers { if d == globalDeploymentID() { continue } errMap[d] = configurePeerFn(d, peer) } - return c.state.toErrorFromErrMap(errMap, configureReplication) + return state.toErrorFromErrMap(errMap, configureReplication) } // PeerBucketDeleteHandler - deletes bucket on local in response to a delete // bucket request from a peer. func (c *SiteReplicationSys) PeerBucketDeleteHandler(ctx context.Context, bucket string, opts DeleteBucketOptions) error { - c.RLock() - defer c.RUnlock() - if !c.enabled { + if !c.isEnabled() { return errSRNotEnabled } @@ -1286,19 +1278,18 @@ func (c *SiteReplicationSys) IAMChangeHook(ctx context.Context, item madmin.SRIA return err } - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } - cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error { - admClient, err := c.getAdminClient(ctx, d) + cerr := state.concDo(nil, func(d string, p madmin.PeerInfo) error { + admClient, err := state.getAdminClient(d) if err != nil { return wrapSRErr(err) } _, err = executeIAMRevisionRequest(ctx, admClient, http.MethodPut, &iamRevisionBatch{Version: iamRevisionProtocol, Items: []iamReplicationItem{versioned}}) - return c.annotatePeerErr(p.Name, replicateIAMItem, err) + return state.annotatePeerErr(p.Name, replicateIAMItem, err) }, replicateIAMItem, ) @@ -1602,19 +1593,18 @@ func (c *SiteReplicationSys) BucketMetaHook(ctx context.Context, item madmin.SRB // The change has already been applied to the local cluster at this // point, and only needs to be updated on all remote peer clusters. - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } - cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error { - admClient, err := c.getAdminClient(ctx, d) + cerr := state.concDo(nil, func(d string, p madmin.PeerInfo) error { + admClient, err := state.getAdminClient(d) if err != nil { return wrapSRErr(err) } - return c.annotatePeerErr(p.Name, replicateBucketMetadata, admClient.SRPeerReplicateBucketMeta(ctx, item)) + return state.annotatePeerErr(p.Name, replicateBucketMetadata, admClient.SRPeerReplicateBucketMeta(ctx, item)) }, replicateBucketMetadata, ) @@ -2107,15 +2097,40 @@ func (c *SiteReplicationSys) PeerBucketLCConfigHandler(ctx context.Context, buck return nil } -// getAdminClient - NOTE: ensure to take at least a read lock on SiteReplicationSys -// before calling this. +func (c *SiteReplicationSys) cloneState() srState { + c.RLock() + defer c.RUnlock() + state := c.state + state.Peers = maps.Clone(c.state.Peers) + return state +} + +func (c *SiteReplicationSys) stateSnapshot() (srState, bool) { + c.RLock() + defer c.RUnlock() + if !c.enabled { + return srState{}, false + } + state := c.state + state.Peers = maps.Clone(c.state.Peers) + return state, true +} + func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID string) (*madmin.AdminClient, error) { - creds, err := c.state.getPeerCreds() + return c.cloneState().getAdminClient(deploymentID) +} + +func (c *SiteReplicationSys) getAdminClientWithEndpoint(ctx context.Context, deploymentID, endpoint string) (*madmin.AdminClient, error) { + return c.cloneState().getAdminClientWithEndpoint(deploymentID, endpoint) +} + +func (s srState) getAdminClient(deploymentID string) (*madmin.AdminClient, error) { + creds, err := s.getPeerCreds() if err != nil { return nil, err } - peer, ok := c.state.Peers[deploymentID] + peer, ok := s.Peers[deploymentID] if !ok { return nil, errSRPeerNotFound } @@ -2123,15 +2138,13 @@ func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID st return getAdminClient(peer.Endpoint, creds.AccessKey, creds.SecretKey) } -// 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.state.getPeerCreds() +func (s srState) getAdminClientWithEndpoint(deploymentID, endpoint string) (*madmin.AdminClient, error) { + creds, err := s.getPeerCreds() if err != nil { return nil, err } - if _, ok := c.state.Peers[deploymentID]; !ok { + if _, ok := s.Peers[deploymentID]; !ok { return nil, errSRPeerNotFound } return getAdminClient(endpoint, creds.AccessKey, creds.SecretKey) @@ -2535,18 +2548,38 @@ func (s srState) concDo(selfActionFn func() error, peerActionFn func(deploymentI return s.newConcErr(errMap, actionName) } +func (s srState) annotateErr(annotation string, err error) error { + if err == nil { + return nil + } + return fmt.Errorf("%s: %s: %w", s.Name, annotation, err) +} + +func (s srState) annotatePeerErr(dstPeer string, annotation string, err error) error { + if err == nil { + return nil + } + return fmt.Errorf("%s->%s: %s: %w", s.Name, dstPeer, annotation, err) +} + +func (c *SiteReplicationSys) siteName() string { + c.RLock() + defer c.RUnlock() + return c.state.Name +} + func (c *SiteReplicationSys) annotateErr(annotation string, err error) error { if err == nil { return nil } - return fmt.Errorf("%s: %s: %w", c.state.Name, annotation, err) + return fmt.Errorf("%s: %s: %w", c.siteName(), annotation, err) } func (c *SiteReplicationSys) annotatePeerErr(dstPeer string, annotation string, err error) error { if err == nil { return nil } - return fmt.Errorf("%s->%s: %s: %w", c.state.Name, dstPeer, annotation, err) + return fmt.Errorf("%s->%s: %s: %w", c.siteName(), dstPeer, annotation, err) } // isEnabled returns true if site replication is enabled @@ -2568,12 +2601,14 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob return st, errSRBackendIssue(err) } peerMap := make(map[string]madmin.PeerInfo) + peerByID := make(map[string]madmin.PeerInfo, len(info.Sites)) var rmvEndpoints []string siteNames := rreq.SiteNames updatedPeers := make(map[string]madmin.PeerInfo) for _, pi := range info.Sites { updatedPeers[pi.DeploymentID] = pi + peerByID[pi.DeploymentID] = pi peerMap[pi.Name] = pi if rreq.RemoveAll { siteNames = append(siteNames, pi.Name) @@ -2588,7 +2623,8 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob delete(updatedPeers, pinfo.DeploymentID) } var wg sync.WaitGroup - errs := make(map[string]error, len(c.state.Peers)) + var errsMu sync.Mutex + errs := make(map[string]error, len(info.Sites)) for _, v := range info.Sites { wg.Add(1) @@ -2596,7 +2632,9 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob go func() { defer wg.Done() err := c.RemoveRemoteTargetsForEndpoint(ctx, objectAPI, rmvEndpoints, false) + errsMu.Lock() errs[globalDeploymentID()] = err + errsMu.Unlock() }() continue } @@ -2604,17 +2642,22 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob defer wg.Done() admClient, err := c.getAdminClient(ctx, pi.DeploymentID) if err != nil { + errsMu.Lock() errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err)) + errsMu.Unlock() return } // set the requesting site's deploymentID for verification of peer request - rreq.RequestingDepID = globalDeploymentID() - if _, err = admClient.SRPeerRemove(ctx, rreq); err != nil { + req := rreq + req.RequestingDepID = globalDeploymentID() + if _, err = admClient.SRPeerRemove(ctx, req); err != nil { if errors.Is(err, errMissingSRConfig) { // ignore if peer is already removed. return } + errsMu.Lock() errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err)) + errsMu.Unlock() return } }(v) @@ -2630,7 +2673,7 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob return madmin.ReplicateRemoveStatus{ ErrDetail: err.Error(), Status: madmin.ReplicateRemoveStatusPartial, - }, errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", c.state.Peers[dID].Name, err)) + }, errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", peerByID[dID].Name, err)) } errdID = dID } @@ -2686,13 +2729,14 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob // InternalRemoveReq - sends an unlink request to peer cluster to remove one or more sites // from the site replication configuration. func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI ObjectLayer, rreq madmin.SRRemoveReq) error { - if !c.isEnabled() { + snapshot, ok := c.stateSnapshot() + if !ok { return errSRNotEnabled } if rreq.RequestingDepID != "" { // validate if requesting site is still part of site replication var foundRequestor bool - for _, p := range c.state.Peers { + for _, p := range snapshot.Peers { if p.DeploymentID == rreq.RequestingDepID { foundRequestor = true break @@ -2708,7 +2752,7 @@ func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI Ob updatedPeers := make(map[string]madmin.PeerInfo) siteNames := rreq.SiteNames - for _, p := range c.state.Peers { + for _, p := range snapshot.Peers { peerMap[p.Name] = p if p.DeploymentID == globalDeploymentID() { ourName = p.Name @@ -2739,9 +2783,9 @@ func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI Ob var state srState if !unlinkSelf { state = srState{ - Name: c.state.Name, + Name: snapshot.Name, Peers: updatedPeers, - ServiceAccountAccessKey: c.state.ServiceAccountAccessKey, + ServiceAccountAccessKey: snapshot.ServiceAccountAccessKey, } } @@ -3004,14 +3048,10 @@ 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() - if !c.enabled { - c.RUnlock() + state, ok := c.stateSnapshot() + if !ok { 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. @@ -4324,10 +4364,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe return madmin.ReplicateEditStatus{}, errSRInvalidRequest(fmt.Errorf("%s not found in existing replicated sites", peer.DeploymentID)) } successMsg := "Cluster replication configuration updated successfully with:" - var state srState - c.RLock() - state = c.state - c.RUnlock() + state := c.cloneState() // in case of --disable-ilm-expiry-replication and --enable-ilm-expiry-replication // --deployment-id is not passed @@ -4335,7 +4372,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe prevPeerInfo, pi madmin.PeerInfo ) if peer.DeploymentID != "" { - pi = c.state.Peers[peer.DeploymentID] + pi = state.Peers[peer.DeploymentID] prevPeerInfo = pi if !peer.SyncState.Empty() { // update replication to peer to be sync/async pi.SyncState = peer.SyncState @@ -4387,6 +4424,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe state.UpdatedAt = time.Now() errs := make(map[string]error, len(state.Peers)) + var errsMu sync.Mutex var wg sync.WaitGroup for dID, v := range state.Peers { @@ -4405,16 +4443,20 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe wg.Add(1) go func(pi madmin.PeerInfo, dID string) { defer wg.Done() - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if dID == peer.DeploymentID { - admClient, err = c.getAdminClientWithEndpoint(ctx, dID, pi.Endpoint) + admClient, err = state.getAdminClientWithEndpoint(dID, pi.Endpoint) } if err != nil { + errsMu.Lock() errs[dID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err)) + errsMu.Unlock() return } if err = admClient.SRPeerEdit(ctx, pi); err != nil { + errsMu.Lock() errs[dID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err)) + errsMu.Unlock() return } }(p, dID) @@ -4515,32 +4557,33 @@ func (c *SiteReplicationSys) updateTargetEndpoints(ctx context.Context, prevInfo // PeerEditReq - internal API handler to respond to a peer cluster's request // to edit endpoint. func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInfo) error { + state := c.cloneState() ourName := "" // Set ReplicateILMExpiry for all peers currTime := time.Now() - for i := range c.state.Peers { - p := c.state.Peers[i] + for i := range state.Peers { + p := state.Peers[i] if p.ReplicateILMExpiry == arg.ReplicateILMExpiry { // its already set due to previous edit req break } p.ReplicateILMExpiry = arg.ReplicateILMExpiry - c.state.UpdatedAt = currTime - c.state.Peers[i] = p + state.UpdatedAt = currTime + state.Peers[i] = p } - for i := range c.state.Peers { - p := c.state.Peers[i] + for i := range state.Peers { + p := state.Peers[i] if p.DeploymentID == arg.DeploymentID { p.Endpoint = arg.Endpoint - c.state.Peers[arg.DeploymentID] = p + state.Peers[arg.DeploymentID] = p } if p.DeploymentID == globalDeploymentID() { ourName = p.Name } } - if err := c.saveToDisk(ctx, c.state); err != nil { + if err := c.saveToDisk(ctx, state); err != nil { return errSRBackendIssue(fmt.Errorf("unable to save cluster-replication state to drive on %s: %v", ourName, err)) } return nil @@ -4549,11 +4592,11 @@ func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInf // PeerStateEditReq - internal API handler to respond to a peer cluster's request // to edit state. func (c *SiteReplicationSys) PeerStateEditReq(ctx context.Context, arg madmin.SRStateEditReq) error { - if arg.UpdatedAt.After(c.state.UpdatedAt) { - state := c.state + state := c.cloneState() + if arg.UpdatedAt.After(state.UpdatedAt) { // update only the ReplicateILMExpiry flag for the peers from incoming request for _, peer := range arg.Peers { - currPeer := c.state.Peers[peer.DeploymentID] + currPeer := state.Peers[peer.DeploymentID] currPeer.ReplicateILMExpiry = peer.ReplicateILMExpiry state.Peers[peer.DeploymentID] = currPeer } @@ -4705,9 +4748,8 @@ func getSRBucketDeleteOp(isSiteReplicated bool) SRBucketDeleteOp { } func (c *SiteReplicationSys) healILMExpiryConfig(ctx context.Context, objAPI ObjectLayer, info srStatusInfo) error { - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } var ( @@ -4749,12 +4791,12 @@ func (c *SiteReplicationSys) healILMExpiryConfig(ctx context.Context, objAPI Obj } // Send details to other sites for healing - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { return wrapSRErr(err) } if err = admClient.SRStateEdit(ctx, madmin.SRStateEditReq{Peers: latestPeers, UpdatedAt: lastUpdate}); err != nil { - replLogIf(ctx, c.annotatePeerErr(ps.Name, siteReplicationEdit, + replLogIf(ctx, state.annotatePeerErr(ps.Name, siteReplicationEdit, fmt.Errorf("Unable to heal site replication state for peer %s from peer %s : %w", ps.Name, latestPeerName, err))) } @@ -4797,7 +4839,7 @@ func (c *SiteReplicationSys) healBuckets(ctx context.Context, objAPI ObjectLayer c.healILMExpiryConfig(ctx, objAPI, info) ilmExpiryCfgHealed = true } - if ilmExpiryReplicationEnabled(c.state.Peers) { + if state, ok := c.stateSnapshot(); ok && ilmExpiryReplicationEnabled(state.Peers) { c.healBucketILMExpiry(ctx, objAPI, bucket, info) } } @@ -4809,9 +4851,8 @@ func (c *SiteReplicationSys) healBuckets(ctx context.Context, objAPI ObjectLayer func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error { bs := info.BucketStats[bucket] - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } @@ -4863,7 +4904,7 @@ func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI Obj continue } - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { return wrapSRErr(err) } @@ -4874,7 +4915,7 @@ func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI Obj ExpiryLCConfig: latestExpLCConfig, UpdatedAt: lastUpdate, }); err != nil { - replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata, + replLogIf(ctx, state.annotatePeerErr(peerName, replicateBucketMetadata, fmt.Errorf("Unable to heal bucket ILM expiry data for peer %s from peer %s : %w", peerName, latestPeerName, err))) } @@ -4919,9 +4960,8 @@ func latestCORSConfig(bs map[string]srBucketStatsSummary) (latestID string, late } func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error { - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } @@ -4946,7 +4986,7 @@ func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI Object continue } - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { return wrapSRErr(err) } @@ -4958,7 +4998,7 @@ func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI Object UpdatedAt: latestState.updatedAt, }) if err != nil { - replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata, + replLogIf(ctx, state.annotatePeerErr(peerName, replicateBucketMetadata, fmt.Errorf("Unable to heal CORS config metadata for peer %s from peer %s : %w", peerName, latestPeerName, err))) } @@ -4981,12 +5021,11 @@ func (c *SiteReplicationSys) purgeDeletedBucket(ctx context.Context, objAPI Obje // healBucket creates/deletes the bucket according to latest state across clusters participating in site replication. func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error { bs := info.BucketStats[bucket] - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } - numSites := len(c.state.Peers) + numSites := len(state.Peers) mostRecent := func(d1, d2 time.Time) time.Time { if d1.IsZero() { return d2 @@ -5068,19 +5107,19 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, if dID == globalDeploymentID() { err := c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts) if err != nil { - return c.annotateErr(makeBucketWithVersion, fmt.Errorf("error healing bucket for site replication %w from %s -> %s", + return state.annotateErr(makeBucketWithVersion, fmt.Errorf("error healing bucket for site replication %w from %s -> %s", err, latestPeerName, peerName)) } } else { - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { - return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) + return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) } if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap); err != nil { - return c.annotatePeerErr(peerName, makeBucketWithVersion, err) + return state.annotatePeerErr(peerName, makeBucketWithVersion, err) } if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil); err != nil { - return c.annotatePeerErr(peerName, configureReplication, err) + return state.annotatePeerErr(peerName, configureReplication, err) } } } @@ -5088,7 +5127,7 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, // configure replication from current cluster to other clusters err := c.PeerBucketConfigureReplHandler(ctx, bucket) if err != nil { - return c.annotateErr(configureReplication, err) + return state.annotateErr(configureReplication, err) } } return nil @@ -5101,12 +5140,12 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, if dID == globalDeploymentID() { c.purgeDeletedBucket(ctx, objAPI, bucket) } else { - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { - return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) + return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) } if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.PurgeDeletedBucketOp, nil); err != nil { - return c.annotatePeerErr(peerName, deleteBucket, err) + return state.annotatePeerErr(peerName, deleteBucket, err) } } } @@ -5120,16 +5159,16 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, Force: true, }) if err != nil { - return c.annotateErr(deleteBucket, fmt.Errorf("error healing bucket for site replication %w from %s -> %s", + return state.annotateErr(deleteBucket, fmt.Errorf("error healing bucket for site replication %w from %s -> %s", err, latestPeerName, peerName)) } } else { - admClient, err := c.getAdminClient(ctx, dID) + admClient, err := state.getAdminClient(dID) if err != nil { - return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) + return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err)) } if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ForceDeleteBucketBktOp, nil); err != nil { - return c.annotatePeerErr(peerName, deleteBucket, err) + return state.annotatePeerErr(peerName, deleteBucket, err) } } } @@ -5141,9 +5180,8 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo, opts *validateReplicationDestinationOptions) error { bs := info.BucketStats[bucket] - c.RLock() - defer c.RUnlock() - if !c.enabled { + state, ok := c.stateSnapshot() + if !ok { return nil } @@ -5172,7 +5210,7 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob arnTgtMap[t.Arn] = t } } - for _, p := range c.state.Peers { + for _, p := range state.Peers { epDeplIDMap[p.Endpoint] = p.DeploymentID } // fix stale ARN's in replication config and endpoint mismatch between site config and @@ -5199,7 +5237,7 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob } if replMismatch { - replLogOnceIf(ctx, c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)), "heal-bucket-relication-config") + replLogOnceIf(ctx, state.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)), "heal-bucket-relication-config") } return nil } @@ -5678,7 +5716,8 @@ func (c *SiteReplicationSys) getPeerForUpload(deplID string) (pi srPeerInfo, loc func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) { c.resyncMu.Lock() defer c.resyncMu.Unlock() - if !c.isEnabled() { + state, ok := c.stateSnapshot() + if !ok { return res, errSRNotEnabled } if objAPI == nil { @@ -5688,7 +5727,7 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer if peer.DeploymentID == globalDeploymentID() { return res, errSRResyncToSelf } - if _, ok := c.state.Peers[peer.DeploymentID]; !ok { + if _, ok := state.Peers[peer.DeploymentID]; !ok { return res, errSRPeerNotFound } rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID) @@ -5799,7 +5838,8 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) { c.resyncMu.Lock() defer c.resyncMu.Unlock() - if !c.isEnabled() { + state, ok := c.stateSnapshot() + if !ok { return res, errSRNotEnabled } if objAPI == nil { @@ -5808,7 +5848,7 @@ func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLaye if peer.DeploymentID == globalDeploymentID() { return res, errSRResyncToSelf } - if _, ok := c.state.Peers[peer.DeploymentID]; !ok { + if _, ok := state.Peers[peer.DeploymentID]; !ok { return res, errSRPeerNotFound } rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID)