mirror of
https://github.com/pgsty/minio.git
synced 2026-11-12 10:35:33 +03:00
Feng Ruohang
ef6efd3120
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 <rh@vonng.com>
520 lines
17 KiB
Go
520 lines
17 KiB
Go
// Copyright (c) 2026 Feng Ruohang
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
package cmd
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"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")
|
|
}
|
|
}
|
|
|
|
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(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Enabled</Status></VersioningConfiguration>`))
|
|
return
|
|
}
|
|
if _, ok := r.URL.Query()["location"]; ok {
|
|
w.Header().Set("Content-Type", "application/xml")
|
|
_, _ = w.Write([]byte(`<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`))
|
|
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(`<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>`))
|
|
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(`<CORSConfiguration><CORSRule><AllowedOrigin>*</AllowedOrigin><AllowedMethod>GET</AllowedMethod></CORSRule></CORSConfiguration>`))
|
|
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 := `<LifecycleConfiguration><Rule><ID>exp</ID><Status>Enabled</Status><Expiration><Days>1</Days></Expiration></Rule></LifecycleConfiguration>`
|
|
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(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Enabled</Status></VersioningConfiguration>`))
|
|
return
|
|
}
|
|
if _, ok := r.URL.Query()["location"]; ok {
|
|
w.Header().Set("Content-Type", "application/xml")
|
|
_, _ = w.Write([]byte(`<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`))
|
|
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()
|
|
}
|