Simplify bucket names and sourceID for rosmar (#7655)

Author: torcolvin

base/bucket.go

@@ -10,8 +10,10 @@ package base
 
 import (
 	"context"
+	"crypto/md5"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"io"
@@ -372,14 +374,25 @@ func getMaxTTL(ctx context.Context, store CouchbaseBucketStore) (int, error) {
 	return bucketResponseWithMaxTTL.MaxTTLSeconds, nil
 }
 
-// Get the Server UUID of the bucket, this is also known as the Cluster UUID
-func GetServerUUID(ctx context.Context, store CouchbaseBucketStore) (uuid string, err error) {
-	respBytes, _, err := store.MgmtRequest(ctx, http.MethodGet, "/pools", "application/json", nil)
+// GetServerUUID returns Couchbase Server Cluster UUID on a timeout. If running against rosmar, do return an empty string.
+func GetServerUUID(ctx context.Context, bucket Bucket) (string, error) {
+	gocbV2Bucket, err := AsGocbV2Bucket(bucket)
 	if err != nil {
-		return "", err
+		return "", nil
+	}
+	// start a retry loop to get server ID
+	worker := func() (bool, error, string) {
+		respBytes, _, err := gocbV2Bucket.MgmtRequest(ctx, http.MethodGet, "/pools", "application/json", nil)
+		if err != nil {
+			return true, err, ""
+		}
+
+		uuid, err := ParseClusterUUID(respBytes)
+		return false, err, uuid
 	}
 
-	return ParseClusterUUID(respBytes)
+	err, uuid := RetryLoop(ctx, "Getting ServerUUID", worker, GetNewDatabaseSleeperFunc())
+	return uuid, err
 }
 
 func ParseClusterUUID(respBytes []byte) (string, error) {
@@ -522,3 +535,43 @@ func RequireNoBucketTTL(ctx context.Context, b Bucket) error {
 
 	return nil
 }
+
+// GetSourceID returns the source ID for a bucket.
+func GetSourceID(ctx context.Context, bucket Bucket) (string, error) {
+	// for rosmar bucket and testing, use the bucket name as the source ID to make it easier to identify the source
+	gocbBucket, err := AsGocbV2Bucket(bucket)
+	if err != nil {
+		return bucket.GetName(), nil
+	}
+
+	// If not overwriting the source ID, for rosmar, serverUUID would be ""
+	serverUUID, err := GetServerUUID(ctx, gocbBucket)
+	if err != nil {
+		return "", err
+	}
+	bucketUUID, err := bucket.UUID()
+	if err != nil {
+		return "", err
+	}
+	return CreateEncodedSourceID(bucketUUID, serverUUID)
+}
+
+// CreateEncodedSourceID will hash the bucket UUID and cluster UUID using md5 hash function then will base64 encode it
+// This function is in sync with xdcr implementation of UUIDstoDocumentSource https://github.com/couchbase/goxdcr/blob/dfba7a5b4251d93db46e2b0b4b55ea014218931b/hlv/hlv.go#L51
+func CreateEncodedSourceID(bucketUUID, clusterUUID string) (string, error) {
+	md5Hash := md5.Sum([]byte(bucketUUID + clusterUUID))
+	hexStr := hex.EncodeToString(md5Hash[:])
+	source, err := HexToBase64(hexStr)
+	if err != nil {
+		return "", err
+	}
+	return string(source), nil
+}
+
+// GetNewDatabaseSleeperFunc returns a sleeper function during database connection
+func GetNewDatabaseSleeperFunc() RetrySleeper {
+	return CreateDoublingSleeperFunc(
+		13, // MaxNumRetries approx 40 seconds total retry duration
+		5,  // InitialRetrySleepTimeMS
+	)
+}

base/main_test_bucket_pool.go

@@ -32,6 +32,43 @@ import (
 // GTestBucketPool is a global instance of a TestBucketPool used to manage a pool of buckets for integration testing.
 var GTestBucketPool *TestBucketPool
 
+// rosmarTracker is simplified implemenation of a bucketReadierQueue for rosmar buckets. The only purpose of this
+// tracker is to be able to name in low sequential orders like Couchbase Server bucket pool: rosmar0, rosmar1, etc.
+type rosmarTracker struct {
+	lock          sync.Mutex
+	activeBuckets []bool
+}
+
+// newRosmarTracker initializes a new rosmarTracker with the specified number of buckets.
+func newRosmarTracker(numBuckets int) *rosmarTracker {
+	r := &rosmarTracker{
+		activeBuckets: make([]bool, numBuckets),
+	}
+	return r
+}
+
+// getNextBucketIdx returns the next available bucket index in a rosmarTracker. This will return the lowest available
+// number starting at 0. If all buckets are in use, it will return an error.
+func (r *rosmarTracker) GetNextBucketIdx() (int, error) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	// iterate over len of activeBuckets to find the first lexicographically in a map
+	for i, active := range r.activeBuckets {
+		if !active {
+			r.activeBuckets[i] = true
+			return i, nil
+		}
+	}
+	return 0, fmt.Errorf("no rosmar buckets available, all have been used")
+}
+
+// ReleaseBucketIdx releases a bucket index in a rosmarTracker. This should be called when a bucket is no longer in use.
+func (r *rosmarTracker) ReleaseBucketIdx(idx int) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.activeBuckets[idx] = false
+}
+
 // TestBucketPool is used to manage a pool of pre-prepared buckets for testing purposes.
 type TestBucketPool struct {
 	// integrationMode should be true if using Couchbase Server. If this is false, Walrus buckets are returned instead of pooled buckets.
@@ -60,6 +97,8 @@ type TestBucketPool struct {
 	unclosedBuckets     map[string]map[string]struct{}
 	unclosedBucketsLock sync.Mutex
 
+	rosmarBuckets rosmarTracker
+
 	// skipCollections may be true for older Couchbase Server versions that do not support collections.
 	skipCollections bool
 	// numCollectionsPerBucket is the number of collections to create in each bucket
@@ -154,6 +193,7 @@ func NewTestBucketPoolWithOptions(ctx context.Context, bucketReadierFunc TBPBuck
 		preserveBuckets:                preserveBuckets,
 		bucketInitFunc:                 bucketInitFunc,
 		unclosedBuckets:                make(map[string]map[string]struct{}),
+		rosmarBuckets:                  *newRosmarTracker(numBuckets),
 		useExistingBucket:              TestUseExistingBucket(),
 		useDefaultScope:                options.UseDefaultScope,
 		numCollectionsPerBucket:        numCollectionsPerBucket,
@@ -250,12 +290,13 @@ func (tbp *TestBucketPool) GetWalrusTestBucket(t testing.TB, url string) (b Buck
 		tbp.Fatalf(testCtx, "nil TestBucketPool, but not using a Walrus test URL")
 	}
 
-	id, err := GenerateRandomID()
-	require.NoError(t, err)
-
+	bucketIdx, err := tbp.rosmarBuckets.GetNextBucketIdx()
+	if err != nil {
+		tbp.Fatalf(testCtx, "Couldn't get next rosmar bucket index: %v", err)
+	}
 	var walrusBucket *rosmar.Bucket
 	const typeName = "rosmar"
-	bucketName := tbpBucketNamePrefix + "rosmar_" + id
+	bucketName := fmt.Sprintf("rosmar%d", bucketIdx)
 	if url == "walrus:" || url == rosmar.InMemoryURL {
 		walrusBucket, err = rosmar.OpenBucket(url, bucketName, rosmar.CreateOrOpen)
 	} else {
@@ -305,9 +346,10 @@ func (tbp *TestBucketPool) GetWalrusTestBucket(t testing.TB, url string) (b Buck
 		// Persisted buckets should call close and delete
 		closeErr := walrusBucket.CloseAndDelete(ctx)
 		if closeErr != nil {
-			tbp.Logf(ctx, "Unexpected error closing persistent %s bucket: %v", typeName, closeErr)
+			tbp.Fatalf(ctx, "Unexpected error closing persistent %s bucket: %v", typeName, closeErr)
+			return
 		}
-
+		tbp.rosmarBuckets.ReleaseBucketIdx(bucketIdx)
 	}
 }
 

db/database.go

@@ -298,14 +298,6 @@ func ValidateDatabaseName(dbName string) error {
 	return nil
 }
 
-// getNewDatabaseSleeperFunc returns a sleeper function during database connection
-func getNewDatabaseSleeperFunc() base.RetrySleeper {
-	return base.CreateDoublingSleeperFunc(
-		13, // MaxNumRetries approx 40 seconds total retry duration
-		5,  // InitialRetrySleepTimeMS
-	)
-}
-
 // connectToBucketErrorHandling takes the given spec and error and returns a formatted error, along with whether it was a fatal error.
 func connectToBucketErrorHandling(ctx context.Context, spec base.BucketSpec, gotErr error) (fatalError bool, err error) {
 	if gotErr != nil {
@@ -352,30 +344,14 @@ func ConnectToBucket(ctx context.Context, spec base.BucketSpec, failFast bool) (
 	}
 
 	description := fmt.Sprintf("Attempt to connect to bucket : %v", spec.BucketName)
-	err, ibucket := base.RetryLoop(ctx, description, worker, getNewDatabaseSleeperFunc())
+	err, ibucket := base.RetryLoop(ctx, description, worker, base.GetNewDatabaseSleeperFunc())
 	if err != nil {
 		return nil, err
 	}
 
 	return ibucket.(base.Bucket), nil
 }
 
-// GetServerUUID returns Couchbase Server Cluster UUID on a timeout. If running against rosmar, do return an empty string.
-func GetServerUUID(ctx context.Context, bucket base.Bucket) (string, error) {
-	gocbV2Bucket, err := base.AsGocbV2Bucket(bucket)
-	if err != nil {
-		return "", nil
-	}
-	// start a retry loop to get server ID
-	worker := func() (bool, error, interface{}) {
-		uuid, err := base.GetServerUUID(ctx, gocbV2Bucket)
-		return err != nil, err, uuid
-	}
-
-	err, uuid := base.RetryLoop(ctx, "Getting ServerUUID", worker, getNewDatabaseSleeperFunc())
-	return uuid.(string), err
-}
-
 // Creates a new DatabaseContext on a bucket. The bucket will be closed when this context closes.
 func NewDatabaseContext(ctx context.Context, dbName string, bucket base.Bucket, autoImport bool, options DatabaseContextOptions) (dbc *DatabaseContext, returnedError error) {
 	cleanupFunctions := make([]func(), 0)
@@ -400,7 +376,7 @@ func NewDatabaseContext(ctx context.Context, dbName string, bucket base.Bucket,
 		return nil, err
 	}
 
-	serverUUID, err := GetServerUUID(ctx, bucket)
+	serverUUID, err := base.GetServerUUID(ctx, bucket)
 	if err != nil {
 		return nil, err
 	}
@@ -422,7 +398,7 @@ func NewDatabaseContext(ctx context.Context, dbName string, bucket base.Bucket,
 	if err != nil {
 		return nil, err
 	}
-	sourceID, err := CreateEncodedSourceID(bucketUUID, serverUUID)
+	sourceID, err := base.GetSourceID(ctx, bucket)
 	if err != nil {
 		return nil, err
 	}

db/hybrid_logical_vector.go

@@ -9,9 +9,7 @@
 package db
 
 import (
-	"crypto/md5"
 	"encoding/base64"
-	"encoding/hex"
 	"fmt"
 	"maps"
 	"sort"
@@ -597,18 +595,6 @@ func EncodeValueStr(value string) (string, error) {
 	return base.StringDecimalToLittleEndianHex(strings.TrimSpace(value))
 }
 
-// CreateEncodedSourceID will hash the bucket UUID and cluster UUID using md5 hash function then will base64 encode it
-// This function is in sync with xdcr implementation of UUIDstoDocumentSource https://github.com/couchbase/goxdcr/blob/dfba7a5b4251d93db46e2b0b4b55ea014218931b/hlv/hlv.go#L51
-func CreateEncodedSourceID(bucketUUID, clusterUUID string) (string, error) {
-	md5Hash := md5.Sum([]byte(bucketUUID + clusterUUID))
-	hexStr := hex.EncodeToString(md5Hash[:])
-	source, err := base.HexToBase64(hexStr)
-	if err != nil {
-		return "", err
-	}
-	return string(source), nil
-}
-
 func (hlv HybridLogicalVector) MarshalJSON() ([]byte, error) {
 	type BucketVector struct {
 		CurrentVersionCAS string    `json:"cvCas,omitempty"`

topologytest/peer_test.go

@@ -237,7 +237,7 @@ func NewPeer(t *testing.T, name string, buckets map[PeerBucketID]*base.TestBucke
 	case PeerTypeCouchbaseServer:
 		bucket, ok := buckets[opts.BucketID]
 		require.True(t, ok, "bucket not found for bucket ID %d", opts.BucketID)
-		sourceID, err := xdcr.GetSourceID(base.TestCtx(t), bucket)
+		sourceID, err := base.GetSourceID(base.TestCtx(t), bucket)
 		require.NoError(t, err)
 		p := &CouchbaseServerPeer{
 			name:               name,

xdcr/replication.go

@@ -14,7 +14,6 @@ import (
 	"fmt"
 
 	"github.com/couchbase/sync_gateway/base"
-	"github.com/couchbase/sync_gateway/db"
 	"github.com/couchbaselabs/rosmar"
 )
 
@@ -79,16 +78,3 @@ func NewXDCR(ctx context.Context, fromBucket, toBucket base.Bucket, opts XDCROpt
 	}
 	return newCouchbaseServerManager(ctx, gocbFromBucket, gocbToBucket, opts)
 }
-
-// GetSourceID returns the source ID for a bucket.
-func GetSourceID(ctx context.Context, bucket base.Bucket) (string, error) {
-	serverUUID, err := db.GetServerUUID(ctx, bucket)
-	if err != nil {
-		return "", err
-	}
-	bucketUUID, err := bucket.UUID()
-	if err != nil {
-		return "", err
-	}
-	return db.CreateEncodedSourceID(bucketUUID, serverUUID)
-}

xdcr/rosmar_xdcr.go

@@ -67,7 +67,7 @@ func newRosmarManager(ctx context.Context, fromBucket, toBucket *rosmar.Bucket,
 	if opts.Mobile != MobileOn {
 		return nil, errors.New("Only sgbucket.XDCRMobileOn is supported in rosmar")
 	}
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	if err != nil {
 		return nil, fmt.Errorf("Could not get source ID for %s: %w", fromBucket.GetName(), err)
 	}

xdcr/xdcr_test.go

@@ -70,7 +70,7 @@ func TestMobileXDCRNoSyncDataCopied(t *testing.T) {
 		fromDs = fromBucket.DefaultDataStore()
 		toDs = toBucket.DefaultDataStore()
 	}
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 	docCas := make(map[string]uint64)
 	for _, doc := range []string{syncDoc, attachmentDoc, normalDoc} {
@@ -155,7 +155,7 @@ func getTwoBucketDataStores(t *testing.T) (base.Bucket, sgbucket.DataStore, base
 func TestReplicateVV(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	hlvAgent := db.NewHLVAgent(t, fromDs, "fakeHLVSourceID", base.VvXattrName)
@@ -276,7 +276,7 @@ func TestReplicateVV(t *testing.T) {
 func TestVVWriteTwice(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"
@@ -311,7 +311,7 @@ func TestVVObeyMou(t *testing.T) {
 	base.SetUpTestLogging(t, base.LevelDebug, base.KeySGTest)
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"
@@ -390,7 +390,7 @@ func TestVVMouImport(t *testing.T) {
 	base.SetUpTestLogging(t, base.LevelDebug, base.KeySGTest)
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"
@@ -500,7 +500,7 @@ func TestVVMouImport(t *testing.T) {
 func TestLWWAfterInitialReplication(t *testing.T) {
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
 
 	docID := "doc1"
@@ -631,9 +631,9 @@ func TestVVMultiActor(t *testing.T) {
 	}
 	fromBucket, fromDs, toBucket, toDs := getTwoBucketDataStores(t)
 	ctx := base.TestCtx(t)
-	fromBucketSourceID, err := GetSourceID(ctx, fromBucket)
+	fromBucketSourceID, err := base.GetSourceID(ctx, fromBucket)
 	require.NoError(t, err)
-	toBucketSourceID, err := GetSourceID(ctx, toBucket)
+	toBucketSourceID, err := base.GetSourceID(ctx, toBucket)
 	require.NoError(t, err)
 
 	// Create document on source