fix(replication): diagnose unusable metadata without a heal source

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

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,

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() {

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)
+			}
+		})
+	}})
+}

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("<Tagging>"), 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 {

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) {

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)
+						}
 					}
 				}
 			})

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,

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)
-		}
-	}
-}