Merge remote-tracking branch 'origin/main' into CBG-5499

Author: torcolvin

auth/auth_test.go

@@ -2838,17 +2838,12 @@ func TestInvalidateRoles(t *testing.T) {
 	err = auth.InvalidateRoles("user", 10)
 	assert.NoError(t, err)
 
-	// Ensure the inval seq was set to 5 (raw get to avoid rebuild)
+	// Ensure the inval seq was set to 10 (raw get to avoid rebuild)
 	var userOut userImpl
 	_, err = leakyDataStore.Get(ctx, auth.DocIDForUser("user"), &userOut)
 	assert.NoError(t, err)
 
-	var expectedValue uint64
-	if leakyBucket.IsSupported(sgbucket.BucketStoreFeatureSubdocOperations) {
-		expectedValue = 10
-	} else {
-		expectedValue = 5
-	}
+	expectedValue := uint64(10)
 
 	assert.Equal(t, expectedValue, userOut.GetRoleInvalSeq())
 
@@ -2905,24 +2900,9 @@ func TestInvalidateChannels(t *testing.T) {
 			assert.NoError(t, err)
 
 			// Invalidate channels at invalSeq but cause cas retry by setting to 5
-			enableRetry := false
-			// If subdoc operations are supported, perform an initial invalidation at seq 5
-			if leakyBucket.IsSupported(sgbucket.BucketStoreFeatureSubdocOperations) {
-				err := auth.InvalidateDefaultChannels(testCase.name, testCase.isUser, 5)
-				assert.NoError(t, err)
-
-			} else {
-				// If subdoc ops aren't supported, use leakyBucket to invalidate
-				leakyDataStore.SetUpdateCallback(func(key string) {
-					if enableRetry {
-						enableRetry = false
-						err = auth.InvalidateDefaultChannels(testCase.name, testCase.isUser, 5)
-						assert.NoError(t, err)
-					}
-				})
-			}
+			err = auth.InvalidateDefaultChannels(testCase.name, testCase.isUser, 5)
+			assert.NoError(t, err)
 
-			enableRetry = true
 			err = auth.InvalidateDefaultChannels(testCase.name, testCase.isUser, 10)
 			assert.NoError(t, err)
 

base/bootstrap.go

@@ -52,6 +52,8 @@ type BootstrapConnection interface {
 	// for the given bucket; subsequent bootstrap reads for that bucket stop falling back to
 	// _default._default. Per-bucket so completing one bucket never disables another's fallback.
 	SetMigrationComplete(bucketName string)
+	// IsMigrationComplete reports whether this process has marked bootstrap-metadata migration complete for the given bucket (local cache).
+	IsMigrationComplete(bucketName string) bool
 	// GetMetadataMigrationStatus reads the bucket-level metadata-migration status doc directly from
 	// _system._mobile (never via the dual-collection wrapper). Returns ErrNotFound when the doc has
 	// not yet been stamped on this bucket.
@@ -108,7 +110,6 @@ type CouchbaseClusterSpec struct {
 type CouchbaseCluster struct {
 	server                      string
 	clusterOptions              gocb.ClusterOptions
-	forcePerBucketAuth          bool // Forces perBucketAuth authenticators to be used to connect to the bucket
 	perBucketAuth               map[string]*gocb.Authenticator
 	bucketConnectionMode        BucketConnectionMode    // Whether to cache cluster connections
 	cachedClusterConnection     *gocb.Cluster           // Cached cluster connection, should only be used by GetConfigBuckets
@@ -234,7 +235,7 @@ var _ BootstrapConnection = &CouchbaseCluster{}
 // _system._mobile is the source of truth and reads transparently fall back to _default._default
 // for any keys that have not yet been migrated; when false, _default._default is used unconditionally.
 func NewCouchbaseCluster(ctx context.Context, clusterSpec CouchbaseClusterSpec,
-	forcePerBucketAuth bool, perBucketCreds PerBucketCredentialsConfig,
+	perBucketCreds PerBucketCredentialsConfig,
 	useXattrConfig bool, useSystemMetadataCollection bool, bucketMode BucketConnectionMode) (*CouchbaseCluster, error) {
 	securityConfig, err := GoCBv2SecurityConfig(ctx, Ptr(clusterSpec.TLSSkipVerify), clusterSpec.CACertpath)
 	if err != nil {
@@ -271,7 +272,6 @@ func NewCouchbaseCluster(ctx context.Context, clusterSpec CouchbaseClusterSpec,
 
 	cbCluster := &CouchbaseCluster{
 		server:                      clusterSpec.Server,
-		forcePerBucketAuth:          forcePerBucketAuth,
 		perBucketAuth:               perBucketAuth,
 		clusterOptions:              clusterOptions,
 		bucketConnectionMode:        bucketMode,
@@ -511,6 +511,10 @@ func (cc *CouchbaseCluster) SetMigrationComplete(bucketName string) {
 	cc.bucketsBootstrapMigrationComplete.Store(bucketName, true)
 }
 
+func (cc *CouchbaseCluster) IsMigrationComplete(bucketName string) bool {
+	return cc.bucketBootstrapMigrationComplete(bucketName)
+}
+
 // bucketBootstrapMigrationComplete reports whether bootstrap-metadata migration has been marked
 // complete for the given bucket. Absence of an entry means not-complete, so reads keep the
 // _default._default fallback.
@@ -1109,8 +1113,6 @@ func defaultCollectionExists(b *gocb.Bucket) (bool, error) {
 func (cc *CouchbaseCluster) GetClusterConnectionForBucket(ctx context.Context, bucketName string) (connection *gocb.Cluster, teardownFn func(), err error) {
 	if bucketAuth, set := cc.perBucketAuth[bucketName]; set {
 		connection, err = cc.connect(bucketAuth)
-	} else if cc.forcePerBucketAuth {
-		return nil, nil, fmt.Errorf("unable to get bucket %q since credentials are not defined in bucket_credentials", MD(bucketName).Redact())
 	} else {
 		connection, err = cc.connect(nil)
 	}

base/bootstrap_test.go

@@ -53,8 +53,7 @@ func TestBootstrapRefCounting(t *testing.T) {
 	}, 2*time.Minute, 5*time.Millisecond) // Wait for bucket pool to be initialized, since GetConfigBuckets requires equal buckets to TestBucketPool.numBuckets
 
 	var perBucketCredentialsConfig map[string]*CredentialsConfig
-	forcePerBucketAuth := false
-	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), forcePerBucketAuth, perBucketCredentialsConfig, TestUseXattrs(), false, CachedClusterConnections)
+	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), perBucketCredentialsConfig, TestUseXattrs(), false, CachedClusterConnections)
 	require.NoError(t, err)
 	defer cluster.Close()
 	require.NotNil(t, cluster)
@@ -146,7 +145,7 @@ func newTestBootstrapConnection(t *testing.T) BootstrapConnection {
 		t.Cleanup(cluster.Close)
 		return cluster
 	}
-	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), false, nil, TestUseXattrs(), false, CachedClusterConnections)
+	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), nil, TestUseXattrs(), false, CachedClusterConnections)
 	require.NoError(t, err)
 	t.Cleanup(cluster.Close)
 	return cluster
@@ -360,7 +359,7 @@ func newCouchbaseBootstrapDualFixture(t *testing.T, useXattrs bool) bootstrapDua
 		assert.Equal(c, int32(GTestBucketPool.numBuckets), GTestBucketPool.stats.TotalBucketInitCount.Load())
 	}, 2*time.Minute, 5*time.Millisecond)
 
-	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), false, nil, useXattrs, true, CachedClusterConnections)
+	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), nil, useXattrs, true, CachedClusterConnections)
 	require.NoError(t, err)
 	t.Cleanup(cluster.Close)
 
@@ -469,7 +468,7 @@ func newCouchbaseBootstrapTwoBucketDualFixture(t *testing.T, useXattrs bool) boo
 		assert.Equal(c, int32(GTestBucketPool.numBuckets), GTestBucketPool.stats.TotalBucketInitCount.Load())
 	}, 2*time.Minute, 5*time.Millisecond)
 
-	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), false, nil, useXattrs, true, CachedClusterConnections)
+	cluster, err := NewCouchbaseCluster(ctx, TestClusterSpec(t), nil, useXattrs, true, CachedClusterConnections)
 	require.NoError(t, err)
 	t.Cleanup(cluster.Close)
 

base/bucket.go

@@ -116,8 +116,6 @@ type BucketSpec struct {
 	Auth                          AuthHandler
 	Certpath, Keypath, CACertPath string         // X.509 auth parameters
 	TLSSkipVerify                 bool           // Use insecureSkipVerify when secure scheme (couchbases) is used and cacertpath is undefined
-	KvTLSPort                     int            // Port to use for memcached over TLS.  Required for cbdatasource auth when using TLS
-	UseXattrs                     bool           // Whether to use xattrs to store _sync metadata.  Used during view initialization
 	ViewQueryTimeoutSecs          *uint32        // the view query timeout in seconds (default: 75 seconds)
 	MaxConcurrentQueryOps         *int           // maximum number of concurrent query operations (default: DefaultMaxConcurrentQueryOps)
 	BucketOpTimeout               *time.Duration // How long bucket ops should block returning "operation timed out". If nil, uses GoCB default.  GoCB buckets only.
@@ -302,16 +300,6 @@ func GetBucket(ctx context.Context, spec BucketSpec) (bucket Bucket, err error)
 		if err != nil {
 			return nil, err
 		}
-
-		// If XATTRS are enabled via enable_shared_bucket_access config flag, assert that Couchbase Server is 5.0
-		// or later, otherwise refuse to connect to the bucket since pre 5.0 versions don't support XATTRs
-		if spec.UseXattrs {
-			if !bucket.IsSupported(sgbucket.BucketStoreFeatureXattrs) {
-				WarnfCtx(ctx, "If using XATTRS, Couchbase Server version must be >= 5.0.")
-				return nil, ErrFatalBucketConnection
-			}
-		}
-
 	}
 
 	// TODO: CBG-2529 - LoggingBucket has been removed - pending a new approach to logging all bucket operations
@@ -469,10 +457,10 @@ func AsViewStore(ds DataStore) (sgbucket.ViewStore, bool) {
 	return viewStore, ok
 }
 
-// AsSubdocStore returns a SubdocStore if the underlying dataStore implements and supports subdoc operations.
+// AsSubdocStore returns a SubdocStore if the underlying dataStore implements subdoc operations.
 func AsSubdocStore(ds DataStore) (sgbucket.SubdocStore, bool) {
 	subdocStore, ok := ds.(sgbucket.SubdocStore)
-	return subdocStore, ok && ds.IsSupported(sgbucket.BucketStoreFeatureSubdocOperations)
+	return subdocStore, ok
 }
 
 // AsRangeScanStore returns a RangeScanStore if the dataStore implements and supports range scan operations.

base/bucket_gocb_test.go

@@ -725,27 +725,13 @@ func TestMultiXattrRoundtrip(t *testing.T) {
 		xattrKeys[2]: []byte(`{"key3": "value3"}`),
 	}
 	_, err := dataStore.WriteWithXattrs(ctx, docID, 0, 0, []byte(`{"key": "value"}`), inputXattrs, nil, nil)
-	if dataStore.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		require.NoError(t, err)
-	} else {
-		require.ErrorContains(t, err, "invalid xattr key combination")
-		inputXattrs = map[string][]byte{
-			xattrKeys[0]: inputXattrs[xattrKeys[0]],
-		}
-		// write a document an xattrs, because subsequent GetXattrs needs a document to produce an error without multi-xattr support
-		_, err := dataStore.WriteWithXattrs(ctx, docID, 0, 0, []byte(`{"key": "value"}`), inputXattrs, nil, nil)
-		require.NoError(t, err)
-	}
+	require.NoError(t, err)
 
 	xattrs, _, err := dataStore.GetXattrs(ctx, docID, xattrKeys)
-	if dataStore.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		require.NoError(t, err)
-		for _, key := range xattrKeys {
-			require.Contains(t, maps.Keys(xattrs), key)
-			require.JSONEq(t, string(inputXattrs[key]), string(xattrs[key]))
-		}
-	} else {
-		require.ErrorContains(t, err, "not supported")
+	require.NoError(t, err)
+	for _, key := range xattrKeys {
+		require.Contains(t, maps.Keys(xattrs), key)
+		require.JSONEq(t, string(inputXattrs[key]), string(xattrs[key]))
 	}
 }
 
@@ -2325,15 +2311,8 @@ func TestUserXattrGetWithXattr(t *testing.T) {
 		"_sync": MustJSONMarshal(t, syncXattrVal),
 		"test":  MustJSONMarshal(t, userXattrVal),
 	}
-	if bucket.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		_, err = dataStore.UpdateXattrs(ctx, docKey, 0, cas, xattrs, nil)
-		require.NoError(t, err)
-	} else {
-		for k, v := range xattrs {
-			cas, err = dataStore.UpdateXattrs(ctx, docKey, 0, cas, map[string][]byte{k: v}, nil)
-			require.NoError(t, err)
-		}
-	}
+	_, err = dataStore.UpdateXattrs(ctx, docKey, 0, cas, xattrs, nil)
+	require.NoError(t, err)
 	var docValRet map[string]any
 	docRaw, xattrsResult, _, err := dataStore.GetWithXattrs(ctx, docKey, []string{SyncXattrName, "test"})
 	require.NoError(t, JSONUnmarshal(docRaw, &docValRet))
@@ -2549,9 +2528,6 @@ func TestUpsertOptionPreserveExpiry(t *testing.T) {
 	bucket := GetTestBucket(t)
 	defer bucket.Close(ctx)
 	dataStore := bucket.GetSingleDataStore()
-	if !bucket.IsSupported(sgbucket.BucketStoreFeaturePreserveExpiry) {
-		t.Skip("Preserve expiry is not supported with this CBS version. Skipping test...")
-	}
 
 	testCases := []struct {
 		name          string
@@ -2676,11 +2652,6 @@ func TestWriteWithXattrsBodyUpdateXattrDelete(t *testing.T) {
 	xattrBody := []byte(`{"some": "val"}`)
 
 	cas, err := collection.WriteWithXattrs(ctx, docID, 0, 0, []byte(`{"foo": "bar"}`), map[string][]byte{xattr1Name: xattrBody, xattr2Name: xattrBody}, nil, nil)
-	// in Couchbase Server < 7.6, this will return an error because this is writing two xattrs
-	if !collection.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		require.ErrorContains(t, err, "SUBDOC_XATTR_INVALID_KEY_COMBO")
-		return
-	}
 	require.NoError(t, err)
 
 	_, err = collection.WriteWithXattrs(ctx, docID, 0, cas, []byte(`{"foo": "baz"}`), nil, []string{xattr1Name}, nil)
@@ -2706,10 +2677,6 @@ func TestWriteWithXattrsXattrWriteXattrDelete(t *testing.T) {
 	xattrBody := []byte(`{"some": "val"}`)
 
 	cas, err := collection.WriteWithXattrs(ctx, docID, 0, 0, []byte(`{"foo": "bar"}`), map[string][]byte{xattr1Name: xattrBody, xattr2Name: xattrBody}, nil, nil)
-	if !collection.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		require.ErrorContains(t, err, "SUBDOC_XATTR_INVALID_KEY_COMBO")
-		return
-	}
 	require.NoError(t, err)
 
 	newXattrBody := []byte(`{"another" : "val"}`)
@@ -2746,9 +2713,6 @@ func TestWriteUpdateWithXattrsDocumentTombstone(t *testing.T) {
 
 	bucket := GetTestBucket(t)
 	defer bucket.Close(ctx)
-	if !bucket.IsSupported(sgbucket.BucketStoreFeatureMultiXattrSubdocOperations) {
-		t.Skip("this test writes multiple xattrs")
-	}
 	dataStore := bucket.GetSingleDataStore()
 	key := t.Name()
 

base/collection.go

@@ -645,9 +645,6 @@ func (b *GocbV2Bucket) releaseQueryOp() {
 }
 
 func (b *GocbV2Bucket) ListDataStores(_ context.Context) ([]sgbucket.DataStoreName, error) {
-	if !b.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		return []sgbucket.DataStoreName{ScopeAndCollectionName{Scope: DefaultScope, Collection: DefaultCollection}}, nil
-	}
 	scopes, err := b.bucket.Collections().GetAllScopes(nil)
 	if err != nil {
 		return nil, err

base/collection_gocb.go

@@ -422,11 +422,9 @@ func (c *Collection) GetExpiry(ctx context.Context, k string) (expiry uint32, ge
 	}
 
 	getMetaOptions := gocbcore.GetMetaOptions{
-		Key:      []byte(k),
-		Deadline: c.Bucket.getBucketOpDeadline(),
-	}
-	if c.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		getMetaOptions.CollectionID = c.GetCollectionID()
+		Key:          []byte(k),
+		Deadline:     c.Bucket.getBucketOpDeadline(),
+		CollectionID: c.GetCollectionID(),
 	}
 
 	wg := sync.WaitGroup{}
@@ -564,10 +562,6 @@ func (c *Collection) GetCollectionID() uint32 {
 
 // setCollectionID sets private property of kv CollectionID.
 func (c *Collection) setCollectionID() error {
-	if !c.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		c.kvCollectionID = DefaultCollectionID
-		return nil
-	}
 	// default collection has a known ID
 	if c.IsDefaultScopeCollection() {
 		c.kvCollectionID = DefaultCollectionID

base/collection_n1ql.go

@@ -43,9 +43,6 @@ func IsMobileSystemCollection(dsn sgbucket.DataStoreName) bool {
 
 // EscapedKeyspace returns the escaped fully-qualified identifier for the keyspace (e.g. `bucket`.`scope`.`collection`)
 func (c *Collection) EscapedKeyspace() string {
-	if !c.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		return fmt.Sprintf("`%s`", c.BucketName())
-	}
 	return fmt.Sprintf("`%s`.`%s`.`%s`", c.BucketName(), c.ScopeName(), c.CollectionName())
 }
 
@@ -70,17 +67,12 @@ func (c *Collection) BucketName() string {
 }
 
 func (c *Collection) indexManager() *indexManager {
-	m := &indexManager{
+	return &indexManager{
 		bucketName:     c.BucketName(),
 		collectionName: c.CollectionName(),
 		scopeName:      c.ScopeName(),
+		collection:     c.Collection.QueryIndexes(),
 	}
-	if !c.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		m.cluster = c.Bucket.cluster.QueryIndexes()
-	} else {
-		m.collection = c.Collection.QueryIndexes()
-	}
-	return m
 }
 
 // IndexMetaKeyspaceID returns the value of keyspace_id for the system:indexes table for the collection.
@@ -181,13 +173,7 @@ func (b *GocbV2Bucket) runQuery(scopeName string, statement string, n1qlOptions
 		n1qlOptions = &gocb.QueryOptions{}
 	}
 
-	var queryResults *gocb.QueryResult
-	var err error
-	if b.IsSupported(sgbucket.BucketStoreFeatureCollections) {
-		queryResults, err = b.bucket.Scope(scopeName).Query(statement, n1qlOptions)
-	} else {
-		queryResults, err = b.cluster.Query(statement, n1qlOptions)
-	}
+	queryResults, err := b.bucket.Scope(scopeName).Query(statement, n1qlOptions)
 	// In the event that we get an error during query we should release a view op as Close() will not be called.
 	if err != nil {
 		b.releaseQueryOp()