From 62cf066ff529c7d281703daa365f555cebba717a Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 12 Sep 2026 13:50:49 +0800 Subject: [PATCH] fix(replication): recover physical creation time and align policy status An independent adversarial review of the bucket metadata convergence work found three defects it had introduced. GetBucketInfo overwrote the physical creation probe with cached metadata, which a bucket that never held a configuration legitimately lacks. The new creation-time requirement then failed every policy, tag, SSE, quota, versioning and Object Lock write on such a bucket, with no operator recovery path, and initial synchronization skipped it silently. Return the physical result unchanged when metadata is not requested, as ListBuckets already does, recover the time during initial synchronization, and pass it to MakeBucketHook so peers adopt the same bucket generation. Replication status compared parsed policies statement by statement while heal compares the canonical key. An upgraded peer that stored an equivalent statement order was therefore reported as mismatched forever, and heal never had anything to write. Compare the key heal compares; per-site presence counting is unchanged. Heal diagnostics shared one log key across four conditions, so a real peer RPC failure could be deduplicated away by an earlier message, and they were logged at error level for the normal transient of a peer that does not have the bucket yet. Give each reason its own key at warning level, report only a field state that exists and still cannot be ordered, and diagnose nothing when no site holds a state to propagate. The recovery test now runs against the real ObjectLayer; the stub it replaced returned the expected time and hid the defect. The policy status test uses a statement order the canonical encoder reorders, and adoption coverage is extended past a real field time. Signed-off-by: Feng Ruohang --- cmd/bucket-metadata-replication.go | 14 +++- cmd/bucket-metadata-sys.go | 8 +- cmd/erasure-server-pool.go | 5 ++ cmd/site-replication-metadata-gate_test.go | 88 ++++++++++++++++------ cmd/site-replication-metadata.go | 36 +++++---- cmd/site-replication-metadata_test.go | 64 +++++++++++++++- cmd/site-replication.go | 24 ++++-- docs/site-replication/README.md | 25 ++++-- 8 files changed, 206 insertions(+), 58 deletions(-) diff --git a/cmd/bucket-metadata-replication.go b/cmd/bucket-metadata-replication.go index d461f0bbb..fdb181b46 100644 --- a/cmd/bucket-metadata-replication.go +++ b/cmd/bucket-metadata-replication.go @@ -33,7 +33,8 @@ import ( ) // Only these fields share the site-replication source-time ordering contract. -// Object Lock is applied before Versioning, whose effective document depends on it. +// Bulk apply/import process Object Lock before Versioning, whose effective +// document depends on it. Periodic heal retains its existing type order. var replicatedBucketConfigs = [...]string{ objectLockConfig, bucketVersioningConfig, bucketPolicyConfig, bucketTaggingConfig, bucketSSEConfig, bucketQuotaConfigFile, @@ -57,6 +58,17 @@ func replicatedBucketConfig(meta *BucketMetadata, file string) (*[]byte, *time.T return nil, nil } +// Callers that only need to know whether a file is under the contract must not +// probe replicatedBucketConfig with a throwaway BucketMetadata. +func isReplicatedBucketConfig(file string) bool { + for _, replicated := range replicatedBucketConfigs { + if replicated == file { + return true + } + } + return false +} + func bucketConfigUpdateOnly(file string) bool { return file == bucketVersioningConfig || file == objectLockConfig } diff --git a/cmd/bucket-metadata-sys.go b/cmd/bucket-metadata-sys.go index 3af457faa..5346d0fe8 100644 --- a/cmd/bucket-metadata-sys.go +++ b/cmd/bucket-metadata-sys.go @@ -126,6 +126,10 @@ func (sys *BucketMetadataSys) Set(bucket string, meta BucketMetadata) { } } +// bucketMetadataUpdate returns the committed snapshot to the caller. meta and +// updatedAt hold the saved state only when changed is true. Local writes always +// change state, because localBucketConfigUpdatedAt is strictly greater than the +// current field time, so their handlers can broadcast meta without rechecking. type bucketMetadataUpdate struct { meta BucketMetadata updatedAt time.Time @@ -148,7 +152,7 @@ func (sys *BucketMetadataSys) updateAndParseMetadata(ctx context.Context, bucket } // Load deletions without parsed caches (notably quota), and compare the // six replicated fields against the raw document under the same lock. - if data, _ := replicatedBucketConfig(&result.meta, configFile); data != nil { + if isReplicatedBucketConfig(configFile) { parse = false if bucketConfigUpdateOnly(configFile) && len(configData) == 0 { return result, nil @@ -178,7 +182,7 @@ func (sys *BucketMetadataSys) updateAndParseMetadata(ctx context.Context, bucket } } updatedAt := UTCNow() - if data, _ := replicatedBucketConfig(&meta, configFile); data != nil { + if isReplicatedBucketConfig(configFile) { if err := ensureBucketMetadataCreated(ctx, objAPI, &meta); err != nil { logBucketConfigReplication(ctx, bucket, configFile, "indeterminate", time.Time{}, meta.Created, err.Error()) return err diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index 3755f324c..e94ae0c10 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -2276,6 +2276,11 @@ func (z *erasureServerPools) GetBucketInfo(ctx context.Context, bucket string, o if err != nil { return bucketInfo, toObjectErr(err, bucket) } + // Physical existence/creation probes must not be overwritten by cached + // metadata, which can legitimately lack Created on an unmigrated bucket. + if opts.NoMetadata { + return bucketInfo, nil + } meta, err := globalBucketMetadataSys.Get(bucket) if err == nil { diff --git a/cmd/site-replication-metadata-gate_test.go b/cmd/site-replication-metadata-gate_test.go index 6a0f2acfa..8a843f61b 100644 --- a/cmd/site-replication-metadata-gate_test.go +++ b/cmd/site-replication-metadata-gate_test.go @@ -18,9 +18,10 @@ package cmd import ( - "bytes" "context" "net/http" + "os" + "strconv" "testing" "time" @@ -144,7 +145,6 @@ func TestPeerBucketMetadataLegacyAndGeneration(t *testing.T) { type bucketMetadataCreatedObjectLayer struct { ObjectLayer - created time.Time missing bool } @@ -153,7 +153,8 @@ func (o bucketMetadataCreatedObjectLayer) GetBucketInfo(ctx context.Context, buc if o.missing { return BucketInfo{}, BucketNotFound{Bucket: bucket} } - return BucketInfo{Name: bucket, Created: o.created}, nil + // A physical bucket that reports no creation time either. + return BucketInfo{Name: bucket}, nil } return o.ObjectLayer.GetBucketInfo(ctx, bucket, opts) } @@ -163,37 +164,74 @@ func TestPeerBucketMetadataUnknownCreated(t *testing.T) { t.Run(backend, func(t *testing.T) { defer setObjectLayer(obj) data := bucketConfigTestData(bucket)[bucketTaggingConfig] - for _, mode := range []string{"unknown", "missing", "physical-created"} { + for _, mode := range []string{"unknown", "missing"} { t.Run(mode, func(t *testing.T) { - meta := newBucketMetadata(bucket) setObjectLayer(obj) - if err := globalBucketMetadataSys.save(t.Context(), meta); err != nil { + if err := globalBucketMetadataSys.save(t.Context(), newBucketMetadata(bucket)); err != nil { t.Fatal(err) } - created := UTCNow().Add(-time.Hour) - physical := bucketMetadataCreatedObjectLayer{ObjectLayer: obj, missing: mode == "missing"} - if mode == "physical-created" { - physical.created = created - } - counter := &bucketConfigWriteCounter{ObjectLayer: physical} + counter := &bucketConfigWriteCounter{ObjectLayer: bucketMetadataCreatedObjectLayer{ObjectLayer: obj, missing: mode == "missing"}} setObjectLayer(counter) - stamp := created.Add(time.Minute) + stamp := UTCNow() _, err := globalBucketMetadataSys.updateAndParseMetadata(t.Context(), bucket, bucketTaggingConfig, data, false, false, &stamp) - if mode != "physical-created" { - if err == nil || counter.writes.Load() != 0 { - t.Fatalf("unknown generation was invented: %v writes=%d", err, counter.writes.Load()) - } - } else { - if err != nil { - t.Fatal(err) - } - got, err := readBucketMetadata(t.Context(), obj, bucket) - if err != nil || !got.Created.Equal(created) || !got.TaggingConfigUpdatedAt.Equal(stamp) || !bytes.Equal(got.TaggingConfigXML, data) { - t.Fatalf("physical creation recovery: %v", err) - } + if err == nil || counter.writes.Load() != 0 { + t.Fatalf("unknown generation was invented: %v writes=%d", err, counter.writes.Load()) } }) } }) }}) } + +// setPhysicalBucketCreated stamps the bucket directory on every local drive, +// which is what StatVol reports as the physical creation time. +func setPhysicalBucketCreated(t *testing.T, bucket string, at time.Time) { + t.Helper() + globalLocalDrivesMu.RLock() + drives := cloneDrives(globalLocalDrivesMap) + globalLocalDrivesMu.RUnlock() + if len(drives) == 0 { + t.Fatal("no local drives registered") + } + for _, drive := range drives { + if err := os.Chtimes(pathJoin(drive.Endpoint().Path, bucket), at, at); err != nil { + t.Fatal(err) + } + } +} + +// Recovery has to run against the real ObjectLayer: a stub GetBucketInfo +// returning the expected time would hide the cached zero creation time +// overwriting it, which is what a bucket that never held a configuration has. +func TestBucketMetadataPhysicalCreatedRecovery(t *testing.T) { + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { + // One known time on every drive, so the recovered value can be neither + // confused with UTCNow() nor dependent on which drive answers first. + physical := UTCNow().Add(-3 * time.Hour).Truncate(time.Second) + for _, missing := range []bool{false, true} { + for _, file := range replicatedBucketConfigs { + t.Run(backend+"/"+file+"/missing="+strconv.FormatBool(missing), func(t *testing.T) { + ctx := t.Context() + setPhysicalBucketCreated(t, bucket, physical) + if err := globalBucketMetadataSys.save(ctx, newBucketMetadata(bucket)); err != nil { + t.Fatal(err) + } + if missing { + if err := deleteConfig(ctx, obj, pathJoin(bucketMetaPrefix, bucket, bucketMetadataFile)); err != nil { + t.Fatal(err) + } + } + data := bucketConfigTestData(bucket)[file] + at, err := globalBucketMetadataSys.Update(ctx, bucket, file, data) + if err != nil { + t.Fatalf("bucket without a recorded creation time cannot update %s: %v", file, err) + } + got, err := readBucketMetadata(ctx, obj, bucket) + if err != nil || !got.Created.Equal(physical) || !at.After(physical) { + t.Fatalf("physical creation not persisted: created=%v physical=%v updated=%v err=%v", got.Created, physical, at, err) + } + }) + } + } + }}) +} diff --git a/cmd/site-replication-metadata.go b/cmd/site-replication-metadata.go index 8ae37def4..ca2168c07 100644 --- a/cmd/site-replication-metadata.go +++ b/cmd/site-replication-metadata.go @@ -39,7 +39,7 @@ func logBucketConfigReplication(ctx context.Context, bucket, file, reason string req.AppendTags("created", created.UTC().Format(time.RFC3339Nano)) req.AppendTags("detail", detail) replLogOnceIf(logger.SetReqInfo(ctx, req), errors.New("bucket metadata replication: "+reason), - "bucket-metadata/"+bucket+"/"+file+"/"+reason) + "bucket-metadata/"+bucket+"/"+file+"/"+reason, logger.WarningKind) } func initialBucketConfigReplicationEvent(meta BucketMetadata, file string) (madmin.SRBucketMeta, bool, error) { @@ -131,27 +131,32 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file if !c.enabled { return nil } - for id := range info.Sites { - if _, present := info.BucketStats[bucket][id]; !present { - logBucketConfigReplication(ctx, bucket, file, "indeterminate", time.Time{}, time.Time{}, "missing peer "+id) - } - } - for id, status := range info.BucketStats[bucket] { - state, err := bucketConfigStateFromInfo(bucket, file, status.meta.SRBucketInfo) - _, known := info.Sites[id] - if !known || id == "" || err != nil || !state.valid { - logBucketConfigReplication(ctx, bucket, file, "indeterminate", state.at, status.meta.CreatedAt, "unusable peer "+id) - } - } latest, found := latestBucketConfig(bucket, file, info) if !found { + // No site holds a state worth propagating for this field, so a peer + // that did not report or cannot be ordered is not actionable either. return nil } + // Every reason keeps its own log key, so a site that did not report cannot + // deduplicate away an unusable peer state or a real heal RPC failure for + // the same bucket and field. + for id := range info.Sites { + if _, present := info.BucketStats[bucket][id]; !present { + logBucketConfigReplication(ctx, bucket, file, "unreachable", latest.at, time.Time{}, "peer "+id+" did not report") + } + } for id, status := range info.BucketStats[bucket] { if _, known := info.Sites[id]; !known || id == "" { continue } target := status.meta.SRBucketInfo + current, currentErr := bucketConfigStateFromInfo(bucket, file, target) + // A peer without the bucket reports neither a field nor a creation + // time; bucket healing covers that normal transient. Report only a + // state that exists and still cannot be ordered. + if currentErr != nil || (!current.valid && (len(current.data) != 0 || !current.at.IsZero())) { + logBucketConfigReplication(ctx, bucket, file, "indeterminate", current.at, target.CreatedAt, "unusable peer "+id) + } if target.CreatedAt.IsZero() { continue } @@ -166,8 +171,7 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file if err != nil || !incoming.candidate() { continue } - current, err := bucketConfigStateFromInfo(bucket, file, target) - if err == nil && compareBucketConfigStates(incoming, current) <= 0 { + if currentErr == nil && compareBucketConfigStates(incoming, current) <= 0 { continue } if id == globalDeploymentID() { @@ -182,7 +186,7 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file if err != nil { // A missing credential or unreachable peer must not abandon the other // targets simply because it happened to be visited first in this map. - logBucketConfigReplication(ctx, bucket, file, "indeterminate", latest.at, target.CreatedAt, "peer "+id+": "+err.Error()) + logBucketConfigReplication(ctx, bucket, file, "peer-error", latest.at, target.CreatedAt, "peer "+id+": "+err.Error()) } } return nil diff --git a/cmd/site-replication-metadata_test.go b/cmd/site-replication-metadata_test.go index fb2ce57e2..c2ffc6f10 100644 --- a/cmd/site-replication-metadata_test.go +++ b/cmd/site-replication-metadata_test.go @@ -33,6 +33,7 @@ import ( "github.com/minio/madmin-go/v3" "github.com/minio/minio/internal/auth" + "github.com/pgsty/silo-pkg/v3/policy" ) func TestPeerBucketMetadataSourceTimeAndDeletion(t *testing.T) { @@ -311,6 +312,56 @@ func TestBucketPolicyReplicationKey(t *testing.T) { } } +func TestBucketPolicyReplicationStatusLegacyOrder(t *testing.T) { + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) { + t.Run(backend, func(t *testing.T) { + ctx := t.Context() + // Statement order here is the reverse of the canonical encoder's, + // which is what an upgraded peer stores: the permutation must not + // be reported as a permanent mismatch. + legacy := []byte(fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Sid":"allow","Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::%s/*"},{"Sid":"deny","Effect":"Deny","Principal":"*","Action":"s3:DeleteObject","Resource":"arn:aws:s3:::%s/*"}]}`, bucket, bucket)) + source, err := policy.ParseBucketPolicyConfig(bytes.NewReader(legacy), bucket) + if err != nil { + t.Fatal(err) + } + meta := newBucketMetadata(bucket) + meta.Created = UTCNow().Add(-time.Hour) + meta.PolicyConfigJSON = legacy + meta.PolicyConfigUpdatedAt = meta.Created.Add(time.Minute) + event, send, err := initialBucketConfigReplicationEvent(meta, bucketPolicyConfig) + if err != nil || !send { + t.Fatalf("legacy initial event: %v send=%v", err, send) + } + target := newBucketMetadata(bucket) + target.Created = meta.Created + if err := globalBucketMetadataSys.save(ctx, target); err != nil { + t.Fatal(err) + } + if rec := applySRBucketMetaViaAdmin(t, cred, event); rec.Code != http.StatusOK { + t.Fatalf("peer apply: %d %s", rec.Code, rec.Body.String()) + } + received, _, err := globalBucketMetadataSys.GetPolicyConfig(bucket) + if err != nil { + t.Fatal(err) + } + if !isBktPolicyReplicated(2, []*policy.BucketPolicy{source, received}) { + t.Fatal("equivalent legacy and received policy reported as permanently mismatched") + } + changed, err := policy.ParseBucketPolicyConfig(bytes.NewReader(legacy), bucket) + if err != nil { + t.Fatal(err) + } + changed.Statements[0].SID = "different" + if isBktPolicyReplicated(2, []*policy.BucketPolicy{source, changed}) { + t.Fatal("distinct policy state reported as replicated") + } + if isBktPolicyReplicated(2, []*policy.BucketPolicy{source, nil}) || !isBktPolicyReplicated(2, []*policy.BucketPolicy{nil, nil}) { + t.Fatal("per-site presence accounting changed") + } + }) + }}) +} + func TestPeerBucketMetadataWireAtomicity(t *testing.T) { ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, cred auth.Credentials, t *testing.T) { t.Run(backend, func(t *testing.T) { @@ -393,7 +444,7 @@ func TestPeerBucketMetadataWireAtomicity(t *testing.T) { func TestPeerBucketAdoptionRebasesOnlyDefaults(t *testing.T) { ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { - for _, shift := range []time.Duration{-time.Hour, 0, time.Hour} { + for _, shift := range []time.Duration{-time.Hour, 0, time.Hour, 3 * time.Hour} { t.Run(fmt.Sprintf("%s/%s", backend, shift), func(t *testing.T) { created := UTCNow().Add(-3 * time.Hour) meta := newBucketMetadata(bucket) @@ -419,6 +470,17 @@ func TestPeerBucketAdoptionRebasesOnlyDefaults(t *testing.T) { if !got.TaggingConfigUpdatedAt.Equal(meta.TaggingConfigUpdatedAt) || !got.EncryptionConfigUpdatedAt.Equal(meta.EncryptionConfigUpdatedAt) || !bytes.Equal(got.EncryptionConfigXML, meta.EncryptionConfigXML) { t.Fatal("actual state changed during adoption") } + if shift > 2*time.Hour { + // Preserve history, but do not promote state from an earlier + // bucket generation to a new valid source by retimestamping it. + for _, file := range []string{bucketTaggingConfig, bucketSSEConfig} { + data, at := replicatedBucketConfig(&got, file) + state, err := newBucketConfigState(bucket, file, *data, *at, got.Created, false) + if err != nil || state.candidate() { + t.Fatalf("pre-generation history became a source: %s %v", file, err) + } + } + } }) } }}) diff --git a/cmd/site-replication.go b/cmd/site-replication.go index 4330d56ff..8f46a32d0 100644 --- a/cmd/site-replication.go +++ b/cmd/site-replication.go @@ -819,6 +819,9 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, optsMap["forceCreate"] = "true" } createdAt, _ := globalBucketMetadataSys.CreatedAt(bucket) + if createdAt.IsZero() { + createdAt = opts.CreatedAt + } optsMap["createdAt"] = createdAt.UTC().Format(time.RFC3339Nano) opts.CreatedAt = createdAt @@ -2139,10 +2142,14 @@ func (c *SiteReplicationSys) syncToAllPeers(ctx context.Context, addOpts madmin. if err != nil && !errors.Is(err, errConfigNotFound) { return errSRBackendIssue(err) } + if err := ensureBucketMetadataCreated(ctx, objAPI, &meta); err != nil { + logBucketConfigReplication(ctx, bucket, "initial-sync", "indeterminate", time.Time{}, meta.Created, err.Error()) + return errSRBackendIssue(err) + } opts := MakeBucketOptions{ LockEnabled: meta.ObjectLocking(), - CreatedAt: bucketInfo.Created.UTC(), + CreatedAt: meta.Created.UTC(), } // Now call the MakeBucketHook on existing bucket - this will @@ -3784,18 +3791,19 @@ func isBktPolicyReplicated(total int, policies []*policy.BucketPolicy) bool { return false } // check if policies match between sites - var prev *policy.BucketPolicy - for i, p := range policies { + var prev []byte + first := true + for _, p := range policies { if p == nil { continue } - if i == 0 { - prev = p - continue - } - if !prev.Equals(*p) { + // Heal treats statement/set permutations as the same effective state. + // Status must agree even when an upgraded peer retains legacy bytes. + key, err := canonicalBucketPolicy(p) + if err != nil || !first && !bytes.Equal(prev, key) { return false } + prev, first = key, false } return true } diff --git a/docs/site-replication/README.md b/docs/site-replication/README.md index d37d106b0..cff38bd99 100644 --- a/docs/site-replication/README.md +++ b/docs/site-replication/README.md @@ -103,12 +103,20 @@ bucket identity conflicts first, then resubmit the intended configuration or delete at the authoritative site. A local write advances beyond an existing future field timestamp. Source times before the target bucket's creation are ignored; an unknown creation time is recovered from the physical bucket, or the -operation fails without writing. +operation fails without writing. That recovery happens on the write path. While +a bucket's stored metadata still carries no creation time, site status reports +it that way and periodic healing skips that bucket in both directions; the first +configuration write on it, local or replicated, records the physical time and +returns the bucket to the normal path. -The server emits bounded diagnostics for `legacy-zero`, `before-created` and -`indeterminate`. Keys and error messages remain stable for each bucket/field/ -reason; timestamps and peer details are log attributes. Existing hourly logger -cleanup applies. Normal duplicates, older events and resolved ties are quiet. +The server emits bounded warnings for `legacy-zero`, `before-created`, +`indeterminate`, `unreachable` and `peer-error`. Each reason keeps its own log +key, so a peer that did not report cannot hide an unusable peer state or a real +heal RPC failure for the same bucket and field. A peer that simply does not +have the bucket yet is a normal transient and is not reported here. Keys and +error messages remain stable for each bucket/field/reason; timestamps and peer +details are log attributes. Existing hourly logger cleanup applies. Normal +duplicates, older events and resolved ties are quiet. A local PUT of a policy whose parsed statements are empty now consistently means deletion: PUT succeeds and GET returns the existing NotFound response. @@ -117,3 +125,10 @@ JSON (`{}`, `null`, or a valid zero quota document) remains a live document; it is not silently sent as a deletion. Bulk omission preserves a field, whereas an explicit Policy JSON `null` deletes it. These rules use the existing wire fields and on-disk metadata format. + +Policy GET and admin export use the same validated encoder as replication. +Statement and set arrays may appear in a different order from older output; +policy evaluation is unchanged. This also makes policies using the parser's +existing NotAction/NotResource alternatives writable and readable. Public +replication status compares the same stable policy key as heal, so a peer's +equivalent legacy statement order does not remain a false mismatch.