fix: rebase imports and migrations under metadata.lock

Apply only validated import fields to a fresh locked record, block ForceCreate after real read errors, and route legacy or target-config migration saves through the shared lock. Compute lifecycle deletion state from the locked record.\n\nRefs: #102

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/admin-bucket-handlers.go

@@ -589,6 +589,43 @@ type importMetaReport struct {
 	madmin.BucketMetaImportErrs
 }
 
+type importMetadataFields map[string]struct{}
+
+func (f importMetadataFields) add(configFile string) {
+	f[configFile] = struct{}{}
+}
+
+func applyImportedBucketMetadata(dst *BucketMetadata, src BucketMetadata, fields importMetadataFields) {
+	for configFile := range fields {
+		switch configFile {
+		case bucketPolicyConfig:
+			dst.PolicyConfigJSON = bytes.Clone(src.PolicyConfigJSON)
+			dst.PolicyConfigUpdatedAt = src.PolicyConfigUpdatedAt
+		case bucketNotificationConfig:
+			dst.NotificationConfigXML = bytes.Clone(src.NotificationConfigXML)
+			dst.NotificationConfigUpdatedAt = src.NotificationConfigUpdatedAt
+		case bucketLifecycleConfig:
+			dst.LifecycleConfigXML = bytes.Clone(src.LifecycleConfigXML)
+			dst.LifecycleConfigUpdatedAt = src.LifecycleConfigUpdatedAt
+		case bucketSSEConfig:
+			dst.EncryptionConfigXML = bytes.Clone(src.EncryptionConfigXML)
+			dst.EncryptionConfigUpdatedAt = src.EncryptionConfigUpdatedAt
+		case bucketTaggingConfig:
+			dst.TaggingConfigXML = bytes.Clone(src.TaggingConfigXML)
+			dst.TaggingConfigUpdatedAt = src.TaggingConfigUpdatedAt
+		case bucketQuotaConfigFile:
+			dst.QuotaConfigJSON = bytes.Clone(src.QuotaConfigJSON)
+			dst.QuotaConfigUpdatedAt = src.QuotaConfigUpdatedAt
+		case objectLockConfig:
+			dst.ObjectLockConfigXML = bytes.Clone(src.ObjectLockConfigXML)
+			dst.ObjectLockConfigUpdatedAt = src.ObjectLockConfigUpdatedAt
+		case bucketVersioningConfig:
+			dst.VersioningConfigXML = bytes.Clone(src.VersioningConfigXML)
+			dst.VersioningConfigUpdatedAt = src.VersioningConfigUpdatedAt
+		}
+	}
+}
+
 func (i *importMetaReport) SetStatus(bucket, fname string, err error) {
 	st := i.Buckets[bucket]
 	var errMsg string
@@ -649,6 +686,16 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 	}
 
 	bucketMap := make(map[string]*BucketMetadata, len(zr.File))
+	importedFields := make(map[string]importMetadataFields, len(zr.File))
+	blockedBuckets := make(map[string]struct{})
+	markImported := func(bucket, configFile string) {
+		fields := importedFields[bucket]
+		if fields == nil {
+			fields = make(importMetadataFields)
+			importedFields[bucket] = fields
+		}
+		fields.add(configFile)
+	}
 
 	updatedAt := UTCNow()
 
@@ -664,6 +711,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			bucketMap[bucket] = &meta
 		} else if err != errConfigNotFound {
 			rpt.SetStatus(bucket, "", err)
+			blockedBuckets[bucket] = struct{}{}
 		}
 	}
 
@@ -675,6 +723,9 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			continue
 		}
 		bucket, fileName := slc[0], slc[1]
+		if _, blocked := blockedBuckets[bucket]; blocked {
+			continue
+		}
 		if fileName == objectLockConfig {
 			reader, err := file.Open()
 			if err != nil {
@@ -708,6 +759,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].ObjectLockConfigXML = configData
 			bucketMap[bucket].ObjectLockConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		}
 	}
@@ -720,6 +772,9 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			continue
 		}
 		bucket, fileName := slc[0], slc[1]
+		if _, blocked := blockedBuckets[bucket]; blocked {
+			continue
+		}
 		if fileName == bucketVersioningConfig {
 			reader, err := file.Open()
 			if err != nil {
@@ -764,6 +819,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].VersioningConfigXML = configData
 			bucketMap[bucket].VersioningConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		}
 	}
@@ -781,6 +837,9 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 			continue
 		}
 		bucket, fileName := slc[0], slc[1]
+		if _, blocked := blockedBuckets[bucket]; blocked {
+			continue
+		}
 
 		// create bucket if it does not exist yet.
 		if _, ok := bucketMap[bucket]; !ok {
@@ -813,6 +872,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].NotificationConfigXML = configData
 			bucketMap[bucket].NotificationConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketPolicyConfig:
 			// Error out if Content-Length is beyond allowed size.
@@ -847,6 +907,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].PolicyConfigJSON = configData
 			bucketMap[bucket].PolicyConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketLifecycleConfig:
 			bucketLifecycle, err := lifecycle.ParseLifecycleConfig(io.LimitReader(reader, sz))
@@ -879,6 +940,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].LifecycleConfigXML = configData
 			bucketMap[bucket].LifecycleConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketSSEConfig:
 			// Parse bucket encryption xml
@@ -917,6 +979,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].EncryptionConfigXML = configData
 			bucketMap[bucket].EncryptionConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketTaggingConfig:
 			tags, err := tags.ParseBucketXML(io.LimitReader(reader, sz))
@@ -933,6 +996,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].TaggingConfigXML = configData
 			bucketMap[bucket].TaggingConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		case bucketQuotaConfigFile:
 			data, err := io.ReadAll(reader)
@@ -949,6 +1013,7 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 
 			bucketMap[bucket].QuotaConfigJSON = data
 			bucketMap[bucket].QuotaConfigUpdatedAt = updatedAt
+			markImported(bucket, fileName)
 			rpt.SetStatus(bucket, fileName, nil)
 		}
 	}
@@ -962,11 +1027,27 @@ func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *
 	}
 
 	for bucket, meta := range bucketMap {
-		err := globalBucketMetadataSys.save(ctx, *meta)
+		fields := importedFields[bucket]
+		if len(fields) == 0 {
+			continue
+		}
+		lockCtx, unlock, err := lockBucketMetadata(ctx, objectAPI, bucket)
 		if err != nil {
 			rpt.SetStatus(bucket, "", err)
 			continue
 		}
+		merged, err := loadBucketMetadataParse(lockCtx, objectAPI, bucket, true)
+		if err == nil {
+			applyImportedBucketMetadata(&merged, *meta, fields)
+			err = globalBucketMetadataSys.saveMetadata(lockCtx, objectAPI, merged)
+		}
+		unlock()
+		if err != nil {
+			rpt.SetStatus(bucket, "", err)
+			continue
+		}
+		*meta = merged
+		globalNotificationSys.LoadBucketMetadata(bgContext(ctx), bucket)
 		// Call site replication hook.
 		if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
 			Bucket:           bucket,

cmd/bucket-metadata-lock_test.go

@@ -156,6 +156,25 @@ func testMakeBucketForceCreatePreservesMetadata(obj ObjectLayer, instanceType, b
 	}
 }
 
+func TestApplyImportedBucketMetadataPreservesUnspecifiedFields(t *testing.T) {
+	policyJSON := []byte(`{"Version":"2012-10-17","Statement":[]}`)
+	tagXML := []byte(`<Tagging><TagSet><Tag><Key>existing</Key><Value>tag</Value></Tag></TagSet></Tagging>`)
+	src := newBucketMetadata("bucket")
+	src.PolicyConfigJSON = policyJSON
+	src.PolicyConfigUpdatedAt = UTCNow()
+	dst := newBucketMetadata("bucket")
+	dst.TaggingConfigXML = bytes.Clone(tagXML)
+
+	applyImportedBucketMetadata(&dst, src, importMetadataFields{bucketPolicyConfig: {}})
+	if !bytes.Equal(dst.PolicyConfigJSON, policyJSON) || !bytes.Equal(dst.TaggingConfigXML, tagXML) {
+		t.Fatalf("import patch overwrote unspecified metadata: %+v", dst)
+	}
+	src.PolicyConfigJSON[0] = '!'
+	if dst.PolicyConfigJSON[0] == '!' {
+		t.Fatal("import patch retained the source byte slice")
+	}
+}
+
 func testBucketMetadataLockPreservesTaggingAndSSE(obj ObjectLayer, instanceType, bucket string,
 	_ http.Handler, _ auth.Credentials, t *testing.T,
 ) {

cmd/bucket-metadata-sys.go

@@ -114,7 +114,7 @@ func (sys *BucketMetadataSys) Set(bucket string, meta BucketMetadata) {
 	}
 }
 
-func (sys *BucketMetadataSys) updateAndParse(ctx context.Context, bucket string, configFile string, configData []byte, parse bool) (updatedAt time.Time, err error) {
+func (sys *BucketMetadataSys) updateAndParse(ctx context.Context, bucket string, configFile string, configData []byte, parse, lifecycleDelete bool) (updatedAt time.Time, err error) {
 	objAPI := newObjectLayerFn()
 	if objAPI == nil {
 		return updatedAt, errServerNotInitialized
@@ -139,6 +139,12 @@ func (sys *BucketMetadataSys) updateAndParse(ctx context.Context, bucket string,
 				return err
 			}
 		}
+		if lifecycleDelete {
+			configData, err = lifecycleDeleteConfig(meta.LifecycleConfigXML)
+			if err != nil {
+				return err
+			}
+		}
 		updatedAt = UTCNow()
 		switch configFile {
 		case bucketPolicyConfig:
@@ -227,53 +233,50 @@ func lockBucketMetadata(ctx context.Context, objectAPI ObjectLayer, bucket strin
 	if err != nil {
 		return nil, nil, err
 	}
-	return lkctx.Context(), func() { lock.Unlock(lkctx) }, nil
+	ctx = context.WithValue(lkctx.Context(), bucketMetadataLockContextKey{}, bucket)
+	return ctx, func() { lock.Unlock(lkctx) }, nil
+}
+
+type bucketMetadataLockContextKey struct{}
+
+func bucketMetadataLockHeld(ctx context.Context, bucket string) bool {
+	lockedBucket, _ := ctx.Value(bucketMetadataLockContextKey{}).(string)
+	return lockedBucket == bucket
 }
 
 // Delete delete the bucket metadata for the specified bucket.
 // must be used by all callers instead of using Update() with nil configData.
 func (sys *BucketMetadataSys) Delete(ctx context.Context, bucket string, configFile string) (updatedAt time.Time, err error) {
-	if configFile == bucketLifecycleConfig {
-		// Get bucket config from current site
-		meta, e := globalBucketMetadataSys.GetConfigFromDisk(ctx, bucket)
-		if e != nil && !errors.Is(e, errConfigNotFound) {
-			return updatedAt, e
-		}
-		var expiryRuleRemoved bool
-		if len(meta.LifecycleConfigXML) > 0 {
-			var lcCfg lifecycle.Lifecycle
-			if err := xml.Unmarshal(meta.LifecycleConfigXML, &lcCfg); err != nil {
-				return updatedAt, err
-			}
-			// find a single expiry rule set the flag
-			for _, rl := range lcCfg.Rules {
-				if !rl.Expiration.IsNull() || !rl.NoncurrentVersionExpiration.IsNull() {
-					expiryRuleRemoved = true
-					break
-				}
-			}
-		}
+	return sys.updateAndParse(ctx, bucket, configFile, nil, false, configFile == bucketLifecycleConfig)
+}
 
-		// Form empty ILM details with `ExpiryUpdatedAt` field and save
-		var cfgData []byte
-		if expiryRuleRemoved {
-			var lcCfg lifecycle.Lifecycle
-			currtime := time.Now()
-			lcCfg.ExpiryUpdatedAt = &currtime
-			cfgData, err = xml.Marshal(lcCfg)
-			if err != nil {
-				return updatedAt, err
+func lifecycleDeleteConfig(current []byte) ([]byte, error) {
+	var expiryRuleRemoved bool
+	if len(current) > 0 {
+		var lcCfg lifecycle.Lifecycle
+		if err := xml.Unmarshal(current, &lcCfg); err != nil {
+			return nil, err
+		}
+		for _, rl := range lcCfg.Rules {
+			if !rl.Expiration.IsNull() || !rl.NoncurrentVersionExpiration.IsNull() {
+				expiryRuleRemoved = true
+				break
 			}
 		}
-		return sys.updateAndParse(ctx, bucket, configFile, cfgData, false)
 	}
-	return sys.updateAndParse(ctx, bucket, configFile, nil, false)
+	if !expiryRuleRemoved {
+		return nil, nil
+	}
+	var lcCfg lifecycle.Lifecycle
+	currtime := time.Now()
+	lcCfg.ExpiryUpdatedAt = &currtime
+	return xml.Marshal(lcCfg)
 }
 
 // Update update bucket metadata for the specified bucket.
 // The configData data should not be modified after being sent here.
 func (sys *BucketMetadataSys) Update(ctx context.Context, bucket string, configFile string, configData []byte) (updatedAt time.Time, err error) {
-	return sys.updateAndParse(ctx, bucket, configFile, configData, true)
+	return sys.updateAndParse(ctx, bucket, configFile, configData, true, false)
 }
 
 // Get metadata for a bucket.

cmd/bucket-metadata.go

@@ -249,6 +249,9 @@ func loadBucketMetadataParse(ctx context.Context, objectAPI ObjectLayer, bucket
 		}
 
 		if len(configs) > 0 {
+			if !bucketMetadataLockHeld(ctx, bucket) {
+				return loadBucketMetadataParseUnderLock(ctx, objectAPI, bucket, parse)
+			}
 			// Old bucket without bucket metadata. Hence we migrate existing settings.
 			if err = b.convertLegacyConfigs(ctx, objectAPI, configs); err != nil {
 				return b, err
@@ -270,6 +273,9 @@ func loadBucketMetadataParse(ctx context.Context, objectAPI ObjectLayer, bucket
 	}
 
 	// migrate unencrypted remote targets
+	if len(b.BucketTargetsConfigJSON) != 0 && GlobalKMS != nil && len(b.BucketTargetsConfigMetaJSON) == 0 && !bucketMetadataLockHeld(ctx, bucket) {
+		return loadBucketMetadataParseUnderLock(ctx, objectAPI, bucket, parse)
+	}
 	if err = b.migrateTargetConfig(ctx, objectAPI); err != nil {
 		return b, err
 	}
@@ -277,6 +283,15 @@ func loadBucketMetadataParse(ctx context.Context, objectAPI ObjectLayer, bucket
 	return b, nil
 }
 
+func loadBucketMetadataParseUnderLock(ctx context.Context, objectAPI ObjectLayer, bucket string, parse bool) (BucketMetadata, error) {
+	ctx, unlock, err := lockBucketMetadata(ctx, objectAPI, bucket)
+	if err != nil {
+		return newBucketMetadata(bucket), err
+	}
+	defer unlock()
+	return loadBucketMetadataParse(ctx, objectAPI, bucket, parse)
+}
+
 // loadBucketMetadata loads and migrates to bucket metadata.
 func loadBucketMetadata(ctx context.Context, objectAPI ObjectLayer, bucket string) (BucketMetadata, error) {
 	return loadBucketMetadataParse(ctx, objectAPI, bucket, true)