Merge remote-tracking branch 'origin/main' into codex/issue-77-status-accounting

cmd/site-replication.go

@@ -46,6 +46,7 @@ import (
 	"github.com/minio/minio/internal/bucket/cors"
 	"github.com/minio/minio/internal/bucket/lifecycle"
 	sreplication "github.com/minio/minio/internal/bucket/replication"
+	"github.com/minio/minio/internal/bucket/versioning"
 	"github.com/minio/minio/internal/logger"
 	xldap "github.com/minio/pkg/v3/ldap"
 	"github.com/minio/pkg/v3/policy"
@@ -888,6 +889,32 @@ func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string
 	return errors.Unwrap(cerr)
 }
 
+func enablePeerBucketVersioning(meta *BucketMetadata) error {
+	if len(meta.VersioningConfigXML) == 0 {
+		meta.VersioningConfigXML = enabledBucketVersioningConfig
+		if meta.VersioningConfigUpdatedAt.IsZero() {
+			meta.VersioningConfigUpdatedAt = meta.Created
+		}
+		return nil
+	}
+	config, err := versioning.ParseConfig(bytes.NewReader(meta.VersioningConfigXML))
+	if err != nil {
+		meta.VersioningConfigXML = enabledBucketVersioningConfig
+		meta.VersioningConfigUpdatedAt = UTCNow()
+		return nil
+	}
+	if config.Enabled() {
+		return nil
+	}
+	config.Status = versioning.Enabled
+	meta.VersioningConfigXML, err = xml.Marshal(config)
+	if err != nil {
+		return err
+	}
+	meta.VersioningConfigUpdatedAt = UTCNow()
+	return nil
+}
+
 // PeerBucketMakeWithVersioningHandler - creates bucket and enables versioning.
 func (c *SiteReplicationSys) PeerBucketMakeWithVersioningHandler(ctx context.Context, bucket string, opts MakeBucketOptions) error {
 	objAPI := newObjectLayerFn()
@@ -916,9 +943,14 @@ func (c *SiteReplicationSys) PeerBucketMakeWithVersioningHandler(ctx context.Con
 
 	meta.SetCreatedAt(opts.CreatedAt)
 
-	meta.VersioningConfigXML = enabledBucketVersioningConfig
-	if opts.LockEnabled {
+	if err := enablePeerBucketVersioning(&meta); err != nil {
+		return wrapSRErr(err)
+	}
+	if opts.LockEnabled && len(meta.ObjectLockConfigXML) == 0 {
 		meta.ObjectLockConfigXML = enabledBucketObjectLockConfig
+		if meta.ObjectLockConfigUpdatedAt.IsZero() {
+			meta.ObjectLockConfigUpdatedAt = meta.Created
+		}
 	}
 
 	if err := meta.Save(context.Background(), objAPI); err != nil {
@@ -1727,6 +1759,26 @@ func (c *SiteReplicationSys) PeerBucketTaggingHandler(ctx context.Context, bucke
 	return nil
 }
 
+func newSRBucketObjectLockMeta(bucket string, config *string, updatedAt time.Time) madmin.SRBucketMeta {
+	return madmin.SRBucketMeta{
+		Type:             madmin.SRBucketMetaTypeObjectLockConfig,
+		Bucket:           bucket,
+		ObjectLockConfig: config,
+		UpdatedAt:        updatedAt,
+	}
+}
+
+func srObjectLockPayload(item madmin.SRBucketMeta) *string {
+	if item.ObjectLockConfig != nil {
+		return item.ObjectLockConfig
+	}
+	return item.Tags
+}
+
+func (c *SiteReplicationSys) peerBucketObjectLockConfigItem(ctx context.Context, item madmin.SRBucketMeta) error {
+	return c.PeerBucketObjectLockConfigHandler(ctx, item.Bucket, srObjectLockPayload(item), item.UpdatedAt)
+}
+
 // PeerBucketObjectLockConfigHandler - sets object lock on local bucket.
 func (c *SiteReplicationSys) PeerBucketObjectLockConfigHandler(ctx context.Context, bucket string, objectLockData *string, updatedAt time.Time) error {
 	if objectLockData != nil {
@@ -2176,12 +2228,7 @@ func (c *SiteReplicationSys) syncToAllPeers(ctx context.Context, addOpts madmin.
 		objLockCfgData, tm := meta.ObjectLockConfigXML, meta.ObjectLockConfigUpdatedAt
 		if len(objLockCfgData) > 0 {
 			objLockStr := base64.StdEncoding.EncodeToString(objLockCfgData)
-			err = c.BucketMetaHook(ctx, madmin.SRBucketMeta{
-				Type:      madmin.SRBucketMetaTypeObjectLockConfig,
-				Bucket:    bucket,
-				Tags:      &objLockStr,
-				UpdatedAt: tm,
-			})
+			err = c.BucketMetaHook(ctx, newSRBucketObjectLockMeta(bucket, &objLockStr, tm))
 			if err != nil {
 				return errSRBucketMetaError(err)
 			}
@@ -5364,12 +5411,7 @@ func (c *SiteReplicationSys) healOLockConfigMetadata(ctx context.Context, objAPI
 			return wrapSRErr(err)
 		}
 		peerName := info.Sites[dID].Name
-		err = admClient.SRPeerReplicateBucketMeta(ctx, madmin.SRBucketMeta{
-			Type:      madmin.SRBucketMetaTypeObjectLockConfig,
-			Bucket:    bucket,
-			Tags:      latestObjLockConfig,
-			UpdatedAt: lastUpdate,
-		})
+		err = admClient.SRPeerReplicateBucketMeta(ctx, newSRBucketObjectLockMeta(bucket, latestObjLockConfig, lastUpdate))
 		if err != nil {
 			replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
 				fmt.Errorf("Unable to heal object lock config metadata for peer %s from peer %s : %w",