CBG-4839 set Unknown+Source for import from before ECCV (#7776)

Co-authored-by: Ben Brooks <[email protected]>

Author: torcolvin

base/bucket.go

@@ -61,6 +61,7 @@ type CouchbaseBucketStore interface {
 	HttpClient(context.Context) *http.Client
 	GetSpec() BucketSpec
 	GetMaxVbno() (uint16, error)
+	GetCCVSettings(ctx context.Context) (ccvEnabled bool, maxCAS map[VBNo]uint64, err error)
 
 	// GetStatsVbSeqno retrieves the high sequence number for all vbuckets and returns
 	// a map of UUIDS and a map of high sequence numbers (map from vbno -> seq)
@@ -576,3 +577,5 @@ func GetNewDatabaseSleeperFunc() RetrySleeper {
 		5,  // InitialRetrySleepTimeMS
 	)
 }
+
+var _ CouchbaseBucketStore = &GocbV2Bucket{}

base/bucket_gocb_test.go

@@ -2651,3 +2651,29 @@ func TestVersionPruningWindow(t *testing.T) {
 	// it's assumed that anywhere this test is running has a default pruning window (30 days) - but this at least ensures we can retrieve it
 	assert.Equal(t, time.Hour*24*30, vpw)
 }
+
+func TestGetCCVStartingCAS(t *testing.T) {
+	if UnitTestUrlIsWalrus() {
+		t.Skip("This test only works against Couchbase Server")
+	}
+
+	ctx := TestCtx(t)
+	bucket := GetTestBucket(t)
+	defer bucket.Close(ctx)
+
+	cbStore, ok := AsCouchbaseBucketStore(bucket)
+	require.True(t, ok)
+
+	eccv, startingCAS, err := cbStore.GetCCVSettings(ctx)
+	require.NoError(t, err)
+	require.True(t, eccv)
+
+	numVBuckets, err := cbStore.GetMaxVbno()
+	require.NoError(t, err)
+	require.Len(t, startingCAS, int(numVBuckets))
+	for vbNo := range numVBuckets {
+		cas, ok := startingCAS[VBNo(vbNo)]
+		require.True(t, ok, "Expected starting CAS for vbucket %d", vbNo)
+		require.NotEqual(t, uint64(0), cas, "Expected non-zero starting CAS for vbucket %d", vbNo)
+	}
+}

base/collection.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -175,12 +176,6 @@ func GetGocbV2BucketFromCluster(ctx context.Context, cluster *gocb.Cluster, spec
 	}
 	gocbv2Bucket.kvOps = make(chan struct{}, MaxConcurrentSingleOps*nodeCount*(*numPools))
 
-	// Query to see if mobile XDCR bucket setting is set and store on bucket object
-	err = gocbv2Bucket.queryHLVBucketSetting(ctx)
-	if err != nil {
-		return nil, err
-	}
-
 	return gocbv2Bucket, nil
 }
 
@@ -265,29 +260,6 @@ func (b *GocbV2Bucket) IsSupported(feature sgbucket.BucketStoreFeature) bool {
 	}
 }
 
-// queryHLVBucketSetting sends request to server to check for enableCrossClusterVersioning bucket setting
-func (b *GocbV2Bucket) queryHLVBucketSetting(ctx context.Context) error {
-	url := fmt.Sprintf("/pools/default/buckets/%s", b.GetName())
-	output, statusCode, err := b.MgmtRequest(ctx, http.MethodGet, url, "application/x-www-form-urlencoded", nil)
-	if err != nil || statusCode != http.StatusOK {
-		return fmt.Errorf("error executing query for mobile XDCR bucket setting, status code: %d error: %v output: %s", statusCode, err, string(output))
-	}
-
-	type bucket struct {
-		SupportsHLV *bool `json:"enableCrossClusterVersioning,omitempty"`
-	}
-	var bucketSettings bucket
-	err = JSONUnmarshal(output, &bucketSettings)
-	if err != nil {
-		return err
-	}
-	// In Server < 7.6.1 this field will not be present, but if it is not configured on the bucket, it will return false
-	if bucketSettings.SupportsHLV != nil {
-		b.supportsHLV = *bucketSettings.SupportsHLV
-	}
-	return nil
-}
-
 func (b *GocbV2Bucket) StartDCPFeed(ctx context.Context, args sgbucket.FeedArguments, callback sgbucket.FeedEventCallbackFunc, dbStats *expvar.Map) error {
 	groupID := ""
 	return StartGocbDCPFeed(ctx, b, b.Spec.BucketName, args, callback, dbStats, DCPMetadataStoreInMemory, groupID)
@@ -352,6 +324,56 @@ func (b *GocbV2Bucket) GetMaxVbno() (uint16, error) {
 	return uint16(vbNo), nil
 }
 
+// GetCCVSettings returns the highest CAS value across all vBuckets for a bucket with CCV enabled.
+func (b *GocbV2Bucket) GetCCVSettings(ctx context.Context) (ccvEnabled bool, maxCAS map[VBNo]uint64, err error) {
+	uri := "/pools/default/buckets/" + b.GetName()
+	output, status, err := b.MgmtRequest(ctx, http.MethodGet, uri, "application/json", nil)
+	if err != nil {
+		return false, nil, RedactErrorf("unable to get CCV starting cas for bucket %q: %w", UD(b.GetName()), err)
+	}
+	if status != http.StatusOK {
+		return false, nil, RedactErrorf("unable to get CCV starting cas for bucket %q, status %d", UD(b.GetName()), status)
+	}
+
+	var response struct {
+		EnableCrossClusterVersioning *bool    `json:"enableCrossClusterVersioning"`
+		VBucketsMaxCas               []string `json:"vBucketsMaxCas"`
+	}
+	if err := JSONUnmarshal(output, &response); err != nil {
+		return false, nil, RedactErrorf("unable to parse bucket info JSON for %q: %w", UD(b.GetName()), err)
+	}
+
+	// In Server < 7.6.1 this field will not be present at all
+	if response.EnableCrossClusterVersioning == nil {
+		return false, nil, nil
+	}
+	// CCV supported but not enabled
+	if !*response.EnableCrossClusterVersioning {
+		InfofCtx(ctx, KeyAll, "Bucket %q does not have enableCrossClusterVersioning set", UD(b.GetName()))
+		return false, nil, nil
+	}
+
+	numVBuckets, err := b.GetMaxVbno()
+	if err != nil {
+		return false, nil, fmt.Errorf("error getting vbucket count: %v", err)
+	}
+	// we'd always expect a CAS value per vbucket if CCV is enabled and has propagated correctly - so fail if that's not the case
+	if len(response.VBucketsMaxCas) != int(numVBuckets) {
+		return false, nil, fmt.Errorf("error getting vbucket CAS, expected %d vbucket CAS values, got %q", numVBuckets, response.VBucketsMaxCas)
+	}
+
+	highCAS := make(map[VBNo]uint64, numVBuckets)
+	for i, casStr := range response.VBucketsMaxCas {
+		cas, err := strconv.ParseUint(casStr, 10, 64)
+		if err != nil {
+			return false, nil, fmt.Errorf("error parsing vbucket CAS value %q for vBucket %d: %v", casStr, i, err)
+		}
+		highCAS[VBNo(i)] = cas
+	}
+
+	return true, highCAS, nil
+}
+
 func (b *GocbV2Bucket) getConfigSnapshot() (*gocbcore.ConfigSnapshot, error) {
 	agent, err := b.GetGoCBAgent()
 	if err != nil {

base/vbucket.go

@@ -0,0 +1,47 @@
+// Copyright 2025-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 (
+	"context"
+	"sync"
+)
+
+// VBNo represents a vBucket number
+type VBNo uint16
+
+// VBucketCAS is a map of vbucket number to a CAS value. This can not be copied, it is an atomic data structure.
+type VBucketCAS struct {
+	m sync.Map
+}
+
+// Load returns the CAS value for the specified vbucket. This is a thread safe accessor optimized for reads. If no CAS value is found, it asserts in debug builds and returns 0.
+func (v *VBucketCAS) Load(ctx context.Context, vbNo VBNo) uint64 {
+	rawCas, ok := v.m.Load(vbNo)
+	if !ok {
+		AssertfCtx(ctx, "VBucketCAS: No CAS found for vbucket %d", vbNo)
+		return 0
+	}
+	cas, ok := rawCas.(uint64)
+	if !ok {
+		AssertfCtx(ctx, "VBucketCAS: Invalid CAS type %#v for vbucket %d", rawCas, vbNo)
+		return 0
+	}
+	return cas
+}
+
+// Store the CAS value for the specified vbucket. This is a thread safe method optimized for reads.
+func (v *VBucketCAS) Store(vbNo VBNo, cas uint64) {
+	v.m.Store(vbNo, cas)
+}
+
+// Clear removes all entries from the VBucketCAS. This is a thread safe method.
+func (v *VBucketCAS) Clear() {
+	v.m.Clear()
+}

db/crud.go

@@ -992,7 +992,16 @@ func (db *DatabaseCollectionWithUser) updateHLV(ctx context.Context, d *Document
 		if !hasHLV || (!cvCASMatch && !mouMatch) {
 			// Otherwise this is an SDK mutation made by the local cluster that should be added to HLV.
 			newVVEntry := Version{}
-			newVVEntry.SourceID = db.dbCtx.EncodedSourceID
+			sourceID := db.dbCtx.EncodedSourceID
+			// use unknown source ID when CCV is enabled but doc cas is less than CCV CAS
+			if db.dbCtx.CachedCCVEnabled.Load() {
+				vbNo := sgbucket.VBHash(d.ID, db.dbCtx.numVBuckets)
+				ccvStartingCas := db.dbCtx.CachedCCVStartingCas.Load(ctx, base.VBNo(vbNo))
+				if d.Cas <= ccvStartingCas {
+					sourceID = unknownSourceID
+				}
+			}
+			newVVEntry.SourceID = sourceID
 			newVVEntry.Value = d.Cas
 			err := d.HLV.AddVersion(newVVEntry)
 			if err != nil {

db/database.go

@@ -161,6 +161,9 @@ type DatabaseContext struct {
 	DatabaseStartupError         *DatabaseError                 // Error that occurred during database online processes startup
 	CachedPurgeInterval          atomic.Pointer[time.Duration]  // If set, the cached value of the purge interval to avoid repeated lookups
 	CachedVersionPruningWindow   atomic.Pointer[time.Duration]  // If set, the cached value of the version pruning window to avoid repeated lookups
+	CachedCCVStartingCas         *base.VBucketCAS               // If set, the cached value of the CCV starting CAS value to avoid repeated lookups
+	CachedCCVEnabled             atomic.Bool                    // If set, the cached value of the CCV Enabled flag (this is not expected to transition from true->false, but could go false->true)
+	numVBuckets                  uint16                         // Number of vbuckets in the bucket
 }
 
 type Scope struct {
@@ -416,19 +419,28 @@ func NewDatabaseContext(ctx context.Context, dbName string, bucket base.Bucket,
 	RegisterImportPindexImpl(ctx, options.GroupID)
 
 	dbContext := &DatabaseContext{
-		Name:                dbName,
-		UUID:                cbgt.NewUUID(),
-		MetadataStore:       metadataStore,
-		Bucket:              bucket,
-		BucketUUID:          bucketUUID,
-		EncodedSourceID:     sourceID,
-		StartTime:           time.Now(),
-		autoImport:          autoImport,
-		Options:             options,
-		DbStats:             dbStats,
-		CollectionByID:      make(map[uint32]*DatabaseCollection),
-		ServerUUID:          serverUUID,
-		UserFunctionTimeout: defaultUserFunctionTimeout,
+		Name:                 dbName,
+		UUID:                 cbgt.NewUUID(),
+		MetadataStore:        metadataStore,
+		Bucket:               bucket,
+		BucketUUID:           bucketUUID,
+		EncodedSourceID:      sourceID,
+		StartTime:            time.Now(),
+		autoImport:           autoImport,
+		Options:              options,
+		DbStats:              dbStats,
+		CollectionByID:       make(map[uint32]*DatabaseCollection),
+		ServerUUID:           serverUUID,
+		UserFunctionTimeout:  defaultUserFunctionTimeout,
+		CachedCCVStartingCas: &base.VBucketCAS{},
+	}
+	dbContext.numVBuckets, err = bucket.GetMaxVbno()
+	if err != nil {
+		return nil, err
+	}
+	err = dbContext.updateCCVSettings(ctx)
+	if err != nil {
+		return nil, err
 	}
 
 	// set up cancellable context based on the background context (context lifecycle for the database
@@ -1665,6 +1677,30 @@ func (db *DatabaseContext) GetVersionPruningWindow(ctx context.Context, forceRef
 	return vpw
 }
 
+// updateCCVSettings performs a management query to determine the latest crossClusterVersioning and max cas settings.
+func (db *DatabaseContext) updateCCVSettings(ctx context.Context) error {
+	cbStore, ok := base.AsCouchbaseBucketStore(db.Bucket)
+	if !ok {
+		db.CachedCCVEnabled.Store(false)
+		return nil
+	}
+
+	// Fetch from Couchbase Server
+	enabled, maxCAS, err := cbStore.GetCCVSettings(ctx)
+	if err != nil {
+		return fmt.Errorf("Unable to retrieve server's CCV Starting CAS: %w", err)
+	}
+
+	db.CachedCCVEnabled.Store(enabled)
+	if !enabled {
+		db.CachedCCVStartingCas.Clear()
+	}
+	for vbNo, cas := range maxCAS {
+		db.CachedCCVStartingCas.Store(vbNo, cas)
+	}
+	return nil
+}
+
 func (c *DatabaseCollection) updateAllPrincipalsSequences(ctx context.Context) error {
 	users, roles, err := c.allPrincipalIDs(ctx)
 	if err != nil {

db/hybrid_logical_vector.go

@@ -51,6 +51,10 @@ const (
 	minPVEntriesRetained         = 3 // minPVEntriesRetained defines the minimum number of PV entries that should be retained after compaction, to avoid removing all history for infrequently updated/replicated documents.
 )
 
+const (
+	unknownSourceID = "Unknown+Source" // used if the document was written before ECCV was enabled
+)
+
 type HLVVersions map[string]uint64 // map of source ID to version uint64 version value
 
 // sorted will iterate through the map returning entries in a stable sorted order. Used by testing to make it easier