CBG-5413: supportability enhancements around metadata migration (#8320)

Author: gregns1

base/bootstrap.go

@@ -86,6 +86,9 @@ type BootstrapConnection interface {
 	// information about whether a new DB on this bucket has opted in (e.g., PUT /<db>/ with
 	// use_system_metadata_collection: true) so the very first bootstrap doc lands correctly.
 	SetBucketBootstrapTargetHint(ctx context.Context, bucket string, optInHint bool) error
+	// CachedBootstrapTargets returns the cached bootstrap doc target for each bucket, for observability purposes. Values are "system_mobile", "default".
+	// The snapshot is not guaranteed to be consistent across concurrent updates.
+	CachedBootstrapTargets() map[string]string
 	// Close releases any long-lived connections
 	Close()
 }
@@ -116,9 +119,9 @@ type CouchbaseCluster struct {
 	useSystemMetadataCollection bool                    // When true, bootstrap metadata is stored in _system._mobile, with read-fallback to _default._default during migration
 	migrationComplete           atomic.Bool             // When set, fallback reads are skipped even if useSystemMetadataCollection is true
 	// bucketBootstrapTargets caches the resolved bootstrap-doc location for each bucket.
-	// Resolved lazily on first interaction (or eagerly via SetBucketBootstrapTargetHint). Values
-	// are bucketBootstrapTarget; absence of an entry means "fall back to the connection-wide flag."
-	bucketBootstrapTargets sync.Map
+	// Resolved lazily on first interaction (or eagerly via SetBucketBootstrapTargetHint).
+	// Absence of an entry means "fall back to the connection-wide flag."
+	bucketBootstrapTargets SyncMap[string, bucketBootstrapTarget]
 	useGOCBFastFailRetry   bool // When true, readiness checks fail fast instead of using the best-effort retry strategy
 }
 
@@ -131,6 +134,16 @@ const (
 	bucketTargetSystemMobile                              // _system._mobile primary; _default._default fallback until migration complete
 )
 
+func (t bucketBootstrapTarget) String() string {
+	switch t {
+	case bucketTargetSystemMobile:
+		return "_system._mobile"
+	case bucketTargetDefault:
+		return "_default._default"
+	} // exhaustive:enforce
+	return ""
+}
+
 type BucketConnectionMode int
 
 const (
@@ -372,7 +385,7 @@ func (cc *CouchbaseCluster) metadataCollections(b *gocb.Bucket) (primary, fallba
 	cached, hasCachedTarget := cc.bucketBootstrapTargets.Load(b.Name())
 	useSystemMobile := cc.useSystemMetadataCollection
 	if hasCachedTarget {
-		useSystemMobile = cached.(bucketBootstrapTarget) == bucketTargetSystemMobile
+		useSystemMobile = cached == bucketTargetSystemMobile
 	}
 	if !useSystemMobile {
 		// When this bucket has any opt-in indication — either cached as bucketTargetDefault
@@ -399,7 +412,7 @@ func (cc *CouchbaseCluster) metadataCollections(b *gocb.Bucket) (primary, fallba
 // the bucket-level migration). A no-op when the cache already says _system._mobile, since the
 // systemMobile→default fallback direction is the legitimate in-progress-migration legacy path.
 func (cc *CouchbaseCluster) noteBucketFallbackHit(bucketName string) {
-	if cached, ok := cc.bucketBootstrapTargets.Load(bucketName); ok && cached.(bucketBootstrapTarget) == bucketTargetSystemMobile {
+	if cached, ok := cc.bucketBootstrapTargets.Load(bucketName); ok && cached == bucketTargetSystemMobile {
 		return
 	}
 	cc.bucketBootstrapTargets.Store(bucketName, bucketTargetSystemMobile)
@@ -436,6 +449,20 @@ func (cc *CouchbaseCluster) SetBucketBootstrapTargetHint(ctx context.Context, bu
 	return nil
 }
 
+// CachedBootstrapTargets returns the cached bootstrap doc target for each bucket, for observability purposes. Values are "system_mobile", "default".
+// The snapshot is not guaranteed to be consistent across concurrent updates.
+func (cc *CouchbaseCluster) CachedBootstrapTargets() map[string]string {
+	targets := make(map[string]string)
+	for key, value := range cc.bucketBootstrapTargets.Range {
+		if s := value.String(); s != "" {
+			targets[key] = s
+		} else {
+			targets[key] = "unknown"
+		}
+	}
+	return targets
+}
+
 // probeRegistryLocation checks both collections for an existing _sync:registry doc. Returns
 // (target, found): when found, target identifies the collection; when not found, target is
 // undefined and the caller picks based on its own policy (cluster flag or per-DB hint).

base/dual_metadata_store.go

@@ -62,6 +62,32 @@ func (ms *MetadataStore) MigrationComplete() bool {
 	return ms.migrationComplete.Load()
 }
 
+// MetadataStoreMode classifies a database's metadata store for support / observability.
+type MetadataStoreMode string
+
+const (
+	// MetadataStoreModeFallbackActive means the dual store is wrapping primary+fallback and
+	// fallback reads are still in play (migration not yet complete).
+	MetadataStoreModeFallbackActive MetadataStoreMode = "fallback_active"
+	// MetadataStoreModeFallbackInactive means the dual store is wrapping primary+fallback but
+	// migration has completed, so reads no longer fall back.
+	MetadataStoreModeFallbackInactive MetadataStoreMode = "fallback_inactive"
+)
+
+// GetMetadataStoreMode classifies the given datastore. Returns an empty string when the
+// datastore isn't a *MetadataStore (single-store databases); callers should use JSON
+// omitempty so the field is dropped in that case rather than emitted as "".
+func GetMetadataStoreMode(ds DataStore) MetadataStoreMode {
+	ms, ok := ds.(*MetadataStore)
+	if !ok {
+		return ""
+	}
+	if ms.MigrationComplete() {
+		return MetadataStoreModeFallbackInactive
+	}
+	return MetadataStoreModeFallbackActive
+}
+
 // readFromFallback returns true when err indicates the key was not found in primary and metadata migration has
 // not yet complete, meaning the operation should be retried against the fallback DataStore.
 func (ms *MetadataStore) readFromFallback(ctx context.Context, err error) bool {

base/rosmar_cluster.go

@@ -16,7 +16,6 @@ import (
 	"os"
 	"runtime"
 	"strings"
-	"sync"
 	"sync/atomic"
 
 	sgbucket "github.com/couchbase/sg-bucket"
@@ -33,7 +32,7 @@ type RosmarCluster struct {
 	migrationComplete           atomic.Bool // When set, fallback reads are skipped even if useSystemMetadataCollection is true
 	// bucketBootstrapTargets caches the resolved bootstrap-doc target for each bucket.
 	// Mirrors CouchbaseCluster.bucketBootstrapTargets — see that comment for the decision tree.
-	bucketBootstrapTargets sync.Map
+	bucketBootstrapTargets SyncMap[string, bucketBootstrapTarget]
 }
 
 // NewRosmarCluster creates a from a given URL. useSystemMetadataCollection mirrors
@@ -162,7 +161,7 @@ func (c *RosmarCluster) metadataDataStores(ctx context.Context, bucketName strin
 	cached, hasCachedTarget := c.bucketBootstrapTargets.Load(bucketName)
 	useSystemMobile := c.useSystemMetadataCollection
 	if hasCachedTarget {
-		useSystemMobile = cached.(bucketBootstrapTarget) == bucketTargetSystemMobile
+		useSystemMobile = cached == bucketTargetSystemMobile
 	}
 	if !useSystemMobile {
 		// Reads/writes route to default first. When this bucket has any opt-in indication —
@@ -197,7 +196,7 @@ func (c *RosmarCluster) metadataDataStores(ctx context.Context, bucketName strin
 // the bucket-level migration). A no-op when the cache already says _system._mobile, since the
 // systemMobile→default fallback direction is the legitimate in-progress-migration legacy path.
 func (c *RosmarCluster) noteBucketFallbackHit(bucketName string) {
-	if cached, ok := c.bucketBootstrapTargets.Load(bucketName); ok && cached.(bucketBootstrapTarget) == bucketTargetSystemMobile {
+	if cached, ok := c.bucketBootstrapTargets.Load(bucketName); ok && cached == bucketTargetSystemMobile {
 		return
 	}
 	c.bucketBootstrapTargets.Store(bucketName, bucketTargetSystemMobile)
@@ -672,6 +671,20 @@ func (c *RosmarCluster) MigrateBootstrapDocs(ctx context.Context, bucket string,
 	return nil
 }
 
+// CachedBootstrapTargets returns the cached bootstrap doc target for each bucket, for observability purposes. Values are "system_mobile", "default".
+// The snapshot is not guaranteed to be consistent across concurrent updates.
+func (c *RosmarCluster) CachedBootstrapTargets() map[string]string {
+	targets := make(map[string]string)
+	for key, value := range c.bucketBootstrapTargets.Range {
+		if s := value.String(); s != "" {
+			targets[key] = s
+		} else {
+			targets[key] = "unknown"
+		}
+	}
+	return targets
+}
+
 // Close calls teardown for any cached buckets and removes from cachedBucketConnections
 func (c *RosmarCluster) Close() {
 }

base/stats.go

@@ -928,6 +928,13 @@ type MigrationStats struct {
 	SeqPoisonPillApplied *SgwIntStat `json:"seq_poison_pill_applied"`
 	// Cumulative count of MigrateMetadata pass invocations.
 	Passes *SgwIntStat `json:"passes"`
+	// AbandonedRuns counts metadata migration runs that the orchestrator gave up on after the
+	// bounded pass loop exhausted itself without a clean pass (zero unknown-prefix remaining
+	// AND zero per-doc errors on the same pass). Runs that exit early via a hard error from
+	// MigrateMetadata, or are stopped cooperatively via the terminator, are not counted here —
+	// this stat is specifically the "we hit the retry ceiling" failure mode, useful for
+	// distinguishing transient/abortive errors from buckets that need operator intervention.
+	AbandonedRuns *SgwIntStat `json:"abandoned_runs"`
 }
 
 type SgwStatWrapper interface {
@@ -2577,6 +2584,10 @@ func (d *DbStats) InitMigrationStats() error {
 	if err != nil {
 		return err
 	}
+	resUtil.AbandonedRuns, err = NewIntStat(SubsystemMetadataMigration, "abandoned_runs", StatUnitNoUnits, MetadataMigrationAbandonedRunsDesc, StatAddedVersion4dot1dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	if err != nil {
+		return err
+	}
 	d.MigrationStats = resUtil
 	return nil
 }
@@ -2589,6 +2600,7 @@ func (d *DbStats) unregisterMigrationStats() {
 	prometheus.Unregister(d.MigrationStats.Errors)
 	prometheus.Unregister(d.MigrationStats.SeqPoisonPillApplied)
 	prometheus.Unregister(d.MigrationStats.Passes)
+	prometheus.Unregister(d.MigrationStats.AbandonedRuns)
 }
 
 func (d *DbStats) MetadataMigration() *MigrationStats {

base/stats_descriptions.go

@@ -426,6 +426,13 @@ const (
 	MetadataMigrationSeqPoisonPillAppliedDesc = "The total number of times this node applied the seq-counter poison pill to initiate fallback→primary sequence handoff. Typically 0 or 1 per migration run."
 
 	MetadataMigrationPassesDesc = "The total number of MigrateMetadata range-scan passes executed for this database."
+
+	MetadataMigrationAbandonedRunsDesc = "The total number of metadata migration runs that the " +
+		"orchestrator gave up on after the bounded pass loop exhausted itself without a clean pass. " +
+		"Runs that exit early via a hard error from MigrateMetadata, or are stopped cooperatively via " +
+		"the terminator, are not counted here. This stat specifically captures the \"hit the retry " +
+		"ceiling\" failure mode, useful for distinguishing transient/abortive errors from buckets that " +
+		"need operator intervention."
 )
 
 // DB Replicators stats descriptions (ISGR Specific)

base/sync_map.go

@@ -0,0 +1,50 @@
+/*
+Copyright 2026-Present Couchbase, Inc.
+
+Use of this software is governed by the Business Source License included in
+the file licenses/BSL-Couchbase.txt.  As of the Change Date specified in that
+file, in accordance with the Business Source License, use of this software will
+be governed by the Apache License, Version 2.0, included in the file
+licenses/APL2.txt.
+*/
+
+package base
+
+import (
+	"iter"
+	"sync"
+)
+
+// SyncMap is a type-safe wrapper around sync.Map that eliminates runtime type assertions.
+type SyncMap[K comparable, V any] struct {
+	m sync.Map
+}
+
+func (s *SyncMap[K, V]) Load(key K) (V, bool) {
+	v, ok := s.m.Load(key)
+	if !ok {
+		var zero V
+		return zero, false
+	}
+	return v.(V), true
+}
+
+func (s *SyncMap[K, V]) Store(key K, value V) {
+	s.m.Store(key, value)
+}
+
+func (s *SyncMap[K, V]) LoadOrStore(key K, value V) (V, bool) {
+	actual, loaded := s.m.LoadOrStore(key, value)
+	return actual.(V), loaded
+}
+
+func (s *SyncMap[K, V]) Range(yield func(K, V) bool) {
+	s.m.Range(func(key, value any) bool {
+		return yield(key.(K), value.(V))
+	})
+}
+
+// All returns an iterator over all key-value pairs.
+func (s *SyncMap[K, V]) All() iter.Seq2[K, V] {
+	return s.Range
+}

db/background_mgr_metadata_migration.go

@@ -242,6 +242,9 @@ func (m *MetadataMigrationManager) Run(ctx context.Context, options map[string]a
 
 			if pass+1 >= maxPasses {
 				base.WarnfCtx(ctx, "[%s] gave up after %d passes with %d unknown-prefix doc(s) and %d per-doc error(s) on the last pass", metadataMigrationLoggingID, maxPasses, remaining, passErrors)
+				if promStats != nil {
+					promStats.AbandonedRuns.Add(1)
+				}
 				return fmt.Errorf("%s still not clear of metadata after %d passes: %d unknown-prefix doc(s), %d per-doc error(s) remain", ms.Fallback().GetName(), maxPasses, remaining, passErrors)
 			}
 		}

db/util_testing.go

@@ -1191,3 +1191,10 @@ func (db *DatabaseContext) RestartChangeListener(t testing.TB, flushCache bool)
 func (db *DatabaseContext) FlushChannelCache(t testing.TB) {
 	db.RestartChangeListener(t, true)
 }
+
+// MigrateSeqCounterForTest exposes the unexported migrateSeqCounter for cross-package tests.
+func MigrateSeqCounterForTest(t testing.TB, ctx context.Context, ms *base.MetadataStore, seqKey string) {
+	t.Helper()
+	stats := &MigrationStats{}
+	require.NoError(t, migrateSeqCounter(ctx, ms, seqKey, stats))
+}