From 32a1b81e4c253736d042dda0fb21e306e3400e9e Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 2 Sep 2026 00:46:01 +0800 Subject: [PATCH] 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 --- cmd/admin-bucket-handlers.go | 83 +++++++++++++++++++++++++++++++- cmd/bucket-metadata-lock_test.go | 19 ++++++++ cmd/bucket-metadata-sys.go | 71 ++++++++++++++------------- cmd/bucket-metadata.go | 15 ++++++ 4 files changed, 153 insertions(+), 35 deletions(-) diff --git a/cmd/admin-bucket-handlers.go b/cmd/admin-bucket-handlers.go index 4ea93878f..d3173f3dd 100644 --- a/cmd/admin-bucket-handlers.go +++ b/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, diff --git a/cmd/bucket-metadata-lock_test.go b/cmd/bucket-metadata-lock_test.go index b5528e930..cdcacefaa 100644 --- a/cmd/bucket-metadata-lock_test.go +++ b/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(`existingtag`) + 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, ) { diff --git a/cmd/bucket-metadata-sys.go b/cmd/bucket-metadata-sys.go index 8cb8e9e17..7432eb038 100644 --- a/cmd/bucket-metadata-sys.go +++ b/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. diff --git a/cmd/bucket-metadata.go b/cmd/bucket-metadata.go index 9c066ba32..53099cf62 100644 --- a/cmd/bucket-metadata.go +++ b/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)