From fcbb93e8957275bfa7ad4e6154e93f4d7b0a7025 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 12 Sep 2026 17:04:15 +0800 Subject: [PATCH] fix(replication): diagnose unusable metadata without a heal source Signed-off-by: Feng Ruohang --- cmd/bucket-cors-site-replication_test.go | 19 --- cmd/bucket-metadata-sys.go | 6 +- cmd/site-replication-metadata-gate_test.go | 98 ++++++++++++++++ cmd/site-replication-metadata-heal_test.go | 128 +++++++++++++++++++++ cmd/site-replication-metadata.go | 17 ++- cmd/site-replication-metadata_test.go | 9 ++ cmd/site-replication.go | 11 -- cmd/site-replication_test.go | 25 ---- docs/site-replication/README.md | 14 ++- 9 files changed, 258 insertions(+), 69 deletions(-) diff --git a/cmd/bucket-cors-site-replication_test.go b/cmd/bucket-cors-site-replication_test.go index 64c37330b..275d0fa01 100644 --- a/cmd/bucket-cors-site-replication_test.go +++ b/cmd/bucket-cors-site-replication_test.go @@ -467,25 +467,6 @@ func testAdversarialHealCorsPropagatesNewerEqualValueTimestamp(obj ObjectLayer, } } -func TestAdversarialBucketMetadataComparisonIsBase64CaseSensitive(t *testing.T) { - upper := "QQ==" - lower := "qQ==" - upperBytes, err := base64.StdEncoding.Strict().DecodeString(upper) - if err != nil { - t.Fatal(err) - } - lowerBytes, err := base64.StdEncoding.Strict().DecodeString(lower) - if err != nil { - t.Fatal(err) - } - if string(upperBytes) == string(lowerBytes) { - t.Fatal("test inputs unexpectedly decode to the same bytes") - } - if isBucketMetadataEqual(&upper, &lower) { - t.Fatal("different decoded payloads were treated as equal") - } -} - func TestSiteReplicationStatusDetectsCorsTimestampMismatch(t *testing.T) { ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{ t: t, diff --git a/cmd/bucket-metadata-sys.go b/cmd/bucket-metadata-sys.go index 5346d0fe8..afe09e6aa 100644 --- a/cmd/bucket-metadata-sys.go +++ b/cmd/bucket-metadata-sys.go @@ -184,7 +184,11 @@ func (sys *BucketMetadataSys) updateAndParseMetadata(ctx context.Context, bucket updatedAt := UTCNow() if isReplicatedBucketConfig(configFile) { if err := ensureBucketMetadataCreated(ctx, objAPI, &meta); err != nil { - logBucketConfigReplication(ctx, bucket, configFile, "indeterminate", time.Time{}, meta.Created, err.Error()) + var at time.Time + if sourceTime != nil { + at = *sourceTime + } + logBucketConfigReplication(ctx, bucket, configFile, "indeterminate", at, meta.Created, err.Error()) return err } if sourceTime == nil || sourceTime.IsZero() { diff --git a/cmd/site-replication-metadata-gate_test.go b/cmd/site-replication-metadata-gate_test.go index 8a843f61b..49818b548 100644 --- a/cmd/site-replication-metadata-gate_test.go +++ b/cmd/site-replication-metadata-gate_test.go @@ -18,10 +18,16 @@ package cmd import ( + "bytes" "context" + "encoding/base64" + "encoding/json" "net/http" + "net/http/httptest" "os" "strconv" + "strings" + "sync" "testing" "time" @@ -235,3 +241,95 @@ func TestBucketMetadataPhysicalCreatedRecovery(t *testing.T) { } }}) } + +func TestBucketMetadataInitialSyncPhysicalCreated(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() + serviceCred, _, err := globalIAMSys.NewServiceAccount(ctx, cred.AccessKey, nil, newServiceAccountOpts{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { globalIAMSys.DeleteServiceAccount(context.Background(), serviceCred.AccessKey, false) }) + physical := UTCNow().Add(-3 * time.Hour).Truncate(time.Second) + setPhysicalBucketCreated(t, bucket, physical) + meta := newBucketMetadata(bucket) + meta.TaggingConfigXML = bucketConfigTestData(bucket)[bucketTaggingConfig] + if err := globalBucketMetadataSys.save(ctx, meta); err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var createdAt string + var events []madmin.SRBucketMeta + peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + if r.URL.Query().Get("operation") == string(madmin.MakeWithVersioningBktOp) { + createdAt = r.URL.Query().Get("createdAt") + } + if strings.HasSuffix(r.URL.Path, "/bucket-meta") { + var event madmin.SRBucketMeta + if err := json.NewDecoder(r.Body).Decode(&event); err != nil { + t.Error(err) + } + events = append(events, event) + } + w.WriteHeader(http.StatusOK) + })) + defer peer.Close() + // Exercise the complete outgoing sync sequence with real source + // storage. This peer acknowledges RPCs; it is not a second ObjectLayer. + c := &SiteReplicationSys{enabled: true, state: srState{ServiceAccountAccessKey: serviceCred.AccessKey, + Peers: map[string]madmin.PeerInfo{"initial-peer": {DeploymentID: "initial-peer", Endpoint: peer.URL}}}} + if err := c.syncToAllPeers(ctx, madmin.SRAddOptions{}); err != nil { + t.Fatal(err) + } + mu.Lock() + defer mu.Unlock() + if createdAt != physical.Format(time.RFC3339Nano) { + t.Fatalf("peer creation time %q, want physical time %s", createdAt, physical) + } + for _, event := range events { + if event.Bucket == bucket && event.Type == madmin.SRBucketMetaTypeTags { + if event.Tags == nil || *event.Tags != base64.StdEncoding.EncodeToString(meta.TaggingConfigXML) || !event.UpdatedAt.Equal(physical) { + t.Fatalf("historical tags lost baseline: %+v", event) + } + return + } + } + t.Fatal("initial sync silently skipped historical tags") + }) + }}) +} + +func TestPeerBucketMetadataPhysicalCreatedBoundary(t *testing.T) { + ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{t: t, objAPITest: func(obj ObjectLayer, backend, bucket string, _ http.Handler, _ auth.Credentials, t *testing.T) { + t.Run(backend, func(t *testing.T) { + ctx := t.Context() + physical := UTCNow().Truncate(time.Second) + setPhysicalBucketCreated(t, bucket, physical) + if err := globalBucketMetadataSys.save(ctx, newBucketMetadata(bucket)); err != nil { + t.Fatal(err) + } + counter := &bucketConfigWriteCounter{ObjectLayer: obj} + setObjectLayer(counter) + defer setObjectLayer(obj) + data := bucketConfigTestData(bucket)[bucketTaggingConfig] + at := physical.Add(-time.Hour) + _, err := globalBucketMetadataSys.updateAndParseMetadata(ctx, bucket, bucketTaggingConfig, data, false, false, &at) + if err != nil || counter.writes.Load() != 0 { + t.Fatalf("event before recovered creation must be skipped: err=%v writes=%d", err, counter.writes.Load()) + } + // Physical mtime is only an approximation. A local correction can + // establish it; an earlier peer event cannot lower the bucket identity. + at, err = globalBucketMetadataSys.Update(ctx, bucket, bucketTaggingConfig, data) + if err != nil { + t.Fatal(err) + } + got, err := readBucketMetadata(ctx, obj, bucket) + if err != nil || !got.Created.Equal(physical) || !at.After(physical) || !bytes.Equal(got.TaggingConfigXML, data) { + t.Fatalf("local correction did not establish physical creation: %+v %v", got, err) + } + }) + }}) +} diff --git a/cmd/site-replication-metadata-heal_test.go b/cmd/site-replication-metadata-heal_test.go index abf53a6b5..bedb3ea5e 100644 --- a/cmd/site-replication-metadata-heal_test.go +++ b/cmd/site-replication-metadata-heal_test.go @@ -22,11 +22,15 @@ import ( "encoding/base64" "fmt" "net/http" + "strings" + "sync" "testing" "time" "github.com/minio/madmin-go/v3" "github.com/minio/minio/internal/auth" + "github.com/minio/minio/internal/logger" + "github.com/minio/minio/internal/logger/target/testlogger" ) func bucketConfigTestData(bucket string) map[string][]byte { @@ -40,6 +44,130 @@ func bucketConfigTestData(bucket string) map[string][]byte { } } +type bucketConfigLogCapture struct { + testing.TB + mu sync.Mutex + lines []string +} + +func (c *bucketConfigLogCapture) Logf(format string, args ...any) { + c.mu.Lock() + defer c.mu.Unlock() + c.lines = append(c.lines, fmt.Sprintf(format, args...)) +} + +func TestHealBucketConfigDiagnostics(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) { + recordBucketConfigPeer(t, cred) + capture := &bucketConfigLogCapture{TB: t} + defer testlogger.T.SetLogTB(capture)() + disabled := logger.DisableLog + logger.DisableLog = false + defer func() { logger.DisableLog = disabled }() + created := UTCNow().Add(-time.Hour) + local := globalDeploymentID() + bs := map[string]srBucketStatsSummary{ + local: bucketConfigTestInfo(bucket, bucketTaggingConfig, nil, created, created), + "missing-bucket": {}, + "broken-rpc": bucketConfigTestInfo(bucket, bucketTaggingConfig, nil, created, created), + } + info := srStatusInfo{Sites: map[string]madmin.PeerInfo{local: {}, "missing-bucket": {}, "unreachable": {}, "broken-rpc": {}}, BucketStats: map[string]map[string]srBucketStatsSummary{bucket: bs}} + if err := globalSiteReplicationSys.healBucketConfig(t.Context(), bucket, bucketTaggingConfig, info); err != nil { + t.Fatal(err) + } + if len(capture.lines) != 0 { + t.Fatal("empty baselines produced diagnostics") + } + bs[local] = bucketConfigTestInfo(bucket, bucketTaggingConfig, bucketConfigTestData(bucket)[bucketTaggingConfig], created.Add(time.Minute), created) + for range 2 { + if err := globalSiteReplicationSys.healBucketConfig(t.Context(), bucket, bucketTaggingConfig, info); err != nil { + t.Fatal(err) + } + } + capture.mu.Lock() + defer capture.mu.Unlock() + var unreachable, peerError int + for _, line := range capture.lines { + if strings.Contains(line, "bucket metadata replication: unreachable") { + unreachable++ + } else if strings.Contains(line, "bucket metadata replication: peer-error") { + peerError++ + } else { + t.Fatalf("unexpected diagnostic: %s", line) + } + if !strings.HasPrefix(line, "WARNING:") { + t.Fatalf("diagnostic is not a warning: %s", line) + } + } + if unreachable != 1 || peerError != 1 { + t.Fatalf("distinct reasons were lost or not deduplicated: unreachable=%d peer-error=%d", unreachable, peerError) + } + }) + }}) +} + +func TestHealBucketConfigWithoutSourceDiagnostics(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) { + events := recordBucketConfigPeer(t, cred) + disabled := logger.DisableLog + logger.DisableLog = false + defer func() { logger.DisableLog = disabled }() + created := UTCNow().Add(-time.Hour) + data := bucketConfigTestData(bucket)[bucketTaggingConfig] + for _, tc := range []struct { + name string + data []byte + at time.Time + created time.Time + wantLog bool + }{ + {"unknown-created", data, created.Add(time.Minute), time.Time{}, true}, + {"malformed", []byte(""), created.Add(time.Minute), created, true}, + {"before-created", data, created.Add(-time.Minute), created, true}, + {"empty-baseline", nil, created, created, false}, + } { + t.Run(tc.name, func(t *testing.T) { + capture := &bucketConfigLogCapture{TB: t} + defer testlogger.T.SetLogTB(capture)() + // Each case has its own diagnostic key; no source can be selected + // and neither local storage nor the recording peer may be written. + name := bucket + "-" + tc.name + local := globalDeploymentID() + info := srStatusInfo{Sites: map[string]madmin.PeerInfo{local: {}, "metadata-peer": {}, "unreachable": {}}, + BucketStats: map[string]map[string]srBucketStatsSummary{name: { + local: bucketConfigTestInfo(name, bucketTaggingConfig, tc.data, tc.at, tc.created), + "metadata-peer": {}, + }}} + for range 2 { + if err := globalSiteReplicationSys.healBucketConfig(t.Context(), name, bucketTaggingConfig, info); err != nil { + t.Fatal(err) + } + } + capture.mu.Lock() + defer capture.mu.Unlock() + want := 0 + if tc.wantLog { + want = 1 + } + if len(capture.lines) != want { + t.Fatalf("got %d diagnostics, want %d: %v", len(capture.lines), want, capture.lines) + } + for _, line := range capture.lines { + if !strings.HasPrefix(line, "WARNING:") || !strings.Contains(line, "bucket metadata replication: indeterminate") { + t.Fatalf("unexpected diagnostic: %s", line) + } + } + if len(events()) != 0 { + t.Fatal("healing without a source sent a metadata RPC") + } + }) + } + }) + }}) +} + // Construct independent wire fixtures, including timestamps hidden by legacy // exporters, without going through the production event or state helpers. func bucketConfigTestInfo(bucket, file string, data []byte, at, created time.Time) srBucketStatsSummary { diff --git a/cmd/site-replication-metadata.go b/cmd/site-replication-metadata.go index ca2168c07..356fb4e4b 100644 --- a/cmd/site-replication-metadata.go +++ b/cmd/site-replication-metadata.go @@ -132,17 +132,14 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file return nil } 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") + if found { + 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] { @@ -157,7 +154,9 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file 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() { + // Invalid existing state still needs a diagnosis when no source can + // be selected. Only propagation depends on having a valid source. + if !found || target.CreatedAt.IsZero() { continue } if latest.at.Before(target.CreatedAt) { diff --git a/cmd/site-replication-metadata_test.go b/cmd/site-replication-metadata_test.go index c2ffc6f10..de69b2b50 100644 --- a/cmd/site-replication-metadata_test.go +++ b/cmd/site-replication-metadata_test.go @@ -479,6 +479,15 @@ func TestPeerBucketAdoptionRebasesOnlyDefaults(t *testing.T) { if err != nil || state.candidate() { t.Fatalf("pre-generation history became a source: %s %v", file, err) } + // As a target, that invalid history must yield to a valid + // state from the adopted generation, including a live tag + // replacing the preserved earlier-generation deletion. + incoming := bucketConfigTestData(bucket)[file] + incomingAt := got.Created.Add(time.Minute) + changed, err := applyBucketConfig(&got, file, incoming, incomingAt) + if err != nil || !changed || !at.Equal(incomingAt) || !bytes.Equal(*data, incoming) { + t.Fatalf("adopted generation did not replace invalid target: %s %v", file, err) + } } } }) diff --git a/cmd/site-replication.go b/cmd/site-replication.go index 8f46a32d0..41dce704e 100644 --- a/cmd/site-replication.go +++ b/cmd/site-replication.go @@ -5169,17 +5169,6 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob return nil } -func isBucketMetadataEqual(one, two *string) bool { - switch { - case one == nil && two == nil: - return true - case one == nil || two == nil: - return false - default: - return *one == *two - } -} - func (c *SiteReplicationSys) healIAMSystem(ctx context.Context, objAPI ObjectLayer) error { info, err := c.siteReplicationStatus(ctx, objAPI, madmin.SRStatusOptions{ Users: true, diff --git a/cmd/site-replication_test.go b/cmd/site-replication_test.go index 6f4064b04..60f05504d 100644 --- a/cmd/site-replication_test.go +++ b/cmd/site-replication_test.go @@ -133,28 +133,3 @@ func TestSRBucketMetaCorsRoundTrip(t *testing.T) { t.Fatalf("expected nil Cors for deletion, got %q", *gotDel.Cors) } } - -// TestIsBucketMetadataEqualCors covers the pointer-comparison helper used by -// the CORS heal path to decide whether a peer already holds the latest config. -func TestIsBucketMetadataEqualCors(t *testing.T) { - a := base64.StdEncoding.EncodeToString([]byte("config-a")) - b := base64.StdEncoding.EncodeToString([]byte("config-b")) - - cases := []struct { - name string - one *string - two *string - want bool - }{ - {"both nil", nil, nil, true}, - {"one nil", &a, nil, false}, - {"other nil", nil, &b, false}, - {"equal", &a, &a, true}, - {"different", &a, &b, false}, - } - for _, tc := range cases { - if got := isBucketMetadataEqual(tc.one, tc.two); got != tc.want { - t.Errorf("%s: got %v want %v", tc.name, got, tc.want) - } - } -} diff --git a/docs/site-replication/README.md b/docs/site-replication/README.md index cff38bd99..84a6ecc89 100644 --- a/docs/site-replication/README.md +++ b/docs/site-replication/README.md @@ -103,11 +103,15 @@ 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. That recovery happens on the write path. While +operation fails without writing. Recovery uses a physical approximation (the +bucket directory's modification time), which can differ across drives and be +later than the real creation time. A source event older than this recovered +value is still ignored; its timestamp is not used to invent an earlier bucket +identity. Recovery happens on the write path and during initial site sync. 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. +successful configuration write on it, local or replicated, or initial site sync +records the physical time and returns the bucket to the normal path. The server emits bounded warnings for `legacy-zero`, `before-created`, `indeterminate`, `unreachable` and `peer-error`. Each reason keeps its own log @@ -116,7 +120,9 @@ 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. +duplicates, older events and resolved ties are quiet. Invalid existing field +states emit `indeterminate` even when no valid source can be selected; empty +baselines alone remain quiet and do not cause a heal RPC. A local PUT of a policy whose parsed statements are empty now consistently means deletion: PUT succeeds and GET returns the existing NotFound response.