CBG-4839 allow vBucketsMaxCas to not be populated (#7782)

Author: torcolvin

base/collection.go

@@ -357,12 +357,19 @@ func (b *GocbV2Bucket) GetCCVSettings(ctx context.Context) (ccvEnabled bool, max
 	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
+
+	highCAS := make(map[VBNo]uint64, numVBuckets)
+	// we'd always expect a CAS value per vbucket if CCV is enabled and has propagated correctly
+	// except after a bucket flushed in Server < 7.6.8 see MB-64705
+	// Treating this as ECCV=true with startingCas=0, which will mean imports will all get tagged with bucket SourceID.
 	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)
+		InfofCtx(ctx, KeyBucket, "Bucket %q has enableCrossClusterVersioning=true but unexpected number of vbucket CAS values - expected %d, got %+v. Treating all imports as originating on this Couchbase Server cluster.", MD(b.GetName()), numVBuckets, response.VBucketsMaxCas)
+		for i := range numVBuckets {
+			highCAS[VBNo(i)] = 0
+		}
+		return true, highCAS, nil
 	}
 
-	highCAS := make(map[VBNo]uint64, numVBuckets)
 	for i, casStr := range response.VBucketsMaxCas {
 		cas, err := strconv.ParseUint(casStr, 10, 64)
 		if err != nil {

db/database.go

@@ -1681,7 +1681,11 @@ func (db *DatabaseContext) GetVersionPruningWindow(ctx context.Context, forceRef
 func (db *DatabaseContext) updateCCVSettings(ctx context.Context) error {
 	cbStore, ok := base.AsCouchbaseBucketStore(db.Bucket)
 	if !ok {
-		db.CachedCCVEnabled.Store(false)
+		// for rosmar, ECCV is always enabled, mark starting cas as 0
+		db.CachedCCVEnabled.Store(true)
+		for vbNo := range db.numVBuckets {
+			db.CachedCCVStartingCas.Store(base.VBNo(vbNo), 0)
+		}
 		return nil
 	}
 

db/import_test.go

@@ -119,9 +119,6 @@ func TestFeedImport(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			if base.UnitTestUrlIsWalrus() {
-				t.Skip("rosmar does not support vBuckets and unknownSource")
-			}
 			docID := SafeDocumentName(t, t.Name())
 			db.CachedCCVEnabled.Store(testCase.eccv)
 			for vBucket := range db.numVBuckets {
@@ -282,9 +279,6 @@ func TestOnDemandImport(t *testing.T) {
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			if base.UnitTestUrlIsWalrus() {
-				t.Skip("rosmar does not support vBuckets and unknownSource")
-			}
 			docID := SafeDocumentName(t, t.Name())
 			db.CachedCCVEnabled.Store(testCase.eccv)
 			for vBucket := range db.numVBuckets {