wip

Author: torcolvin

base/dcp_client_metadata.go

@@ -228,7 +228,6 @@ func (m *DCPMetadataCS) Persist(ctx context.Context, workerID int, vbIDs []uint1
 	for _, vbID := range vbIDs {
 		meta.DCPMeta[vbID] = m.metadata[vbID]
 	}
-	fmt.Printf("HONK m.setMetadataKey(workerID): %s\n", m.getMetadataKey(workerID))
 	err := m.dataStore.Set(m.getMetadataKey(workerID), 0, nil, meta)
 	if err != nil {
 		InfofCtx(ctx, KeyDCP, "Unable to persist DCP metadata: %v", err)
@@ -240,7 +239,6 @@ func (m *DCPMetadataCS) Persist(ctx context.Context, workerID int, vbIDs []uint1
 
 func (m *DCPMetadataCS) load(ctx context.Context, workerID int) {
 	var meta WorkerMetadata
-	fmt.Printf("HONK m.getMetadataKey(workerID): %s\n", m.getMetadataKey(workerID))
 	_, err := m.dataStore.Get(m.getMetadataKey(workerID), &meta)
 	if err != nil {
 		if IsDocNotFoundError(err) {

base/dcp_client_test.go

@@ -51,7 +51,7 @@ func TestOneShotDCP(t *testing.T) {
 	}
 
 	dcpOptions := DCPClientOptions{
-		ID: t.Name(),
+		ID: t.Name() + UUID(t),
 		CollectionNames: CollectionNames{
 			dataStore.ScopeName(): {dataStore.CollectionName()},
 		},
@@ -112,7 +112,7 @@ func TestTerminateDCPFeed(t *testing.T) {
 	}
 
 	dcpOptions := DCPClientOptions{
-		ID: t.Name(),
+		ID: t.Name() + UUID(t),
 		CollectionNames: map[string][]string{
 			dataStore.ScopeName(): {dataStore.CollectionName()},
 		},
@@ -207,8 +207,6 @@ func TestDCPClientMultiFeedConsistency(t *testing.T) {
 				return false
 			}
 
-			feedID := t.Name()
-
 			// Add documents
 			updatedBody := map[string]any{"foo": "bar"}
 			for i := range 10000 {
@@ -225,7 +223,7 @@ func TestDCPClientMultiFeedConsistency(t *testing.T) {
 				CollectionNames:  collectionNames,
 				CheckpointPrefix: DefaultMetadataKeys.DCPCheckpointPrefix(t.Name()),
 				Callback:         counterCallback,
-				ID:               feedID,
+				ID:               t.Name() + UUID(t),
 			}
 
 			dcpClient, err := NewDCPClient(ctx, bucket, dcpClientOpts)
@@ -256,7 +254,7 @@ func TestDCPClientMultiFeedConsistency(t *testing.T) {
 			uuidMismatchMetadata[0].SnapEndSeqNo = test.startSeqNo
 
 			dcpClientOpts = DCPClientOptions{
-				ID:               feedID,
+				ID:               t.Name() + UUID(t),
 				CollectionNames:  collectionNames,
 				InitialMetadata:  uuidMismatchMetadata,
 				FailOnRollback:   true,
@@ -276,7 +274,7 @@ func TestDCPClientMultiFeedConsistency(t *testing.T) {
 			// Perform a third DCP feed - mismatched VbUUID, failOnRollback=false
 			atomic.StoreUint64(&mutationCount, 0)
 			dcpClientOpts = DCPClientOptions{
-				ID:               feedID,
+				ID:               t.Name() + UUID(t),
 				InitialMetadata:  uuidMismatchMetadata,
 				FailOnRollback:   false,
 				OneShot:          true,
@@ -688,7 +686,7 @@ func TestDCPFeedEventTypes(t *testing.T) {
 		return true
 	}
 	clientOptions := DCPClientOptions{
-		ID:               t.Name(),
+		ID:               t.Name() + UUID(t),
 		CollectionNames:  collectionNames,
 		CheckpointPrefix: DefaultMetadataKeys.DCPCheckpointPrefix(t.Name()),
 		Callback:         callback,
@@ -777,7 +775,7 @@ func TestDCPClientAgentConfig(t *testing.T) {
 			defer func() { gocbv2Bucket.Spec.Server = oldBucketSpecServer }()
 			gocbv2Bucket.Spec.Server += tc.serverSuffix
 			opts := DCPClientOptions{
-				ID: "fakeFeedID",
+				ID: "fakeFeedID" + UUID(t),
 				CollectionNames: CollectionNames{
 					bucket.GetSingleDataStore().ScopeName(): {bucket.GetSingleDataStore().CollectionName()},
 				},

base/util_testing.go

@@ -36,6 +36,7 @@ import (
 	"github.com/couchbase/gocb/v2"
 	sgbucket "github.com/couchbase/sg-bucket"
 	"github.com/couchbaselabs/rosmar"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -1023,3 +1024,9 @@ func RequireXattrNotFound(t testing.TB, dataStore sgbucket.DataStore, docID stri
 func underGoTest() bool {
 	return testing.Testing()
 }
+
+func UUID(t testing.TB) string {
+	id, err := uuid.NewRandom()
+	require.NoError(t, err)
+	return id.String()
+}

db/background_mgr_attachment_migration.go

@@ -162,29 +162,24 @@ func (a *AttachmentMigrationManager) Run(ctx context.Context, options map[string
 	if err != nil {
 		return err
 	}
-	dcpFeedKey, err := GenerateAttachmentMigrationDCPStreamName(a.MigrationID)
+	dcpOptions, err := getMigrationDCPClientOptions(db, a.MigrationID, scopes, callback)
 	if err != nil {
 		return err
 	}
-	dcpPrefix := db.MetadataKeys.DCPCheckpointPrefix(db.Options.GroupID)
 
 	// check for mismatch in collection id's between current collections on the db and prev run
-	checkpointPrefix := fmt.Sprintf("%s:%v", dcpPrefix, dcpFeedKey)
-	err = a.resetDCPMetadataIfNeeded(ctx, db, checkpointPrefix, currCollectionIDs)
+	err = a.resetDCPMetadataIfNeeded(ctx, db, dcpOptions.CheckpointPrefix, currCollectionIDs)
 	if err != nil {
 		return err
 	}
 
 	a.SetCollectionIDs(currCollectionIDs)
-	dcpOptions := getMigrationDCPClientOptions(scopes, dcpPrefix)
-	dcpOptions.ID = dcpFeedKey
-	dcpOptions.Callback = callback
 	dcpClient, err := base.NewDCPClient(ctx, db.Bucket, dcpOptions)
 	if err != nil {
 		base.WarnfCtx(ctx, "[%s] Failed to create attachment migration DCP client: %v", migrationLoggingID, err)
 		return err
 	}
-	base.DebugfCtx(ctx, base.KeyAll, "[%s] Starting DCP feed %q for attachment migration", migrationLoggingID, dcpFeedKey)
+	base.DebugfCtx(ctx, base.KeyAll, "[%s] Starting DCP feed for attachment migration", migrationLoggingID)
 
 	doneChan, err := dcpClient.Start(ctx)
 	if err != nil {
@@ -283,14 +278,22 @@ func (a *AttachmentMigrationManager) GetProcessStatus(status BackgroundManagerSt
 	return statusJSON, metaJSON, err
 }
 
-func getMigrationDCPClientOptions(scopes base.CollectionNames, prefix string) base.DCPClientOptions {
+func getMigrationDCPClientOptions(db *DatabaseContext, migrationID string, scopes base.CollectionNames, callback sgbucket.FeedEventCallbackFunc) (base.DCPClientOptions, error) {
+	prefix := getAttachmentMigrationPrefix(migrationID)
+
+	feedID, err := base.GenerateDcpStreamName(prefix)
+	if err != nil {
+		return base.DCPClientOptions{}, err
+	}
 	return base.DCPClientOptions{
+		ID:                feedID,
 		OneShot:           true,
 		FailOnRollback:    false,
 		MetadataStoreType: base.DCPMetadataStoreCS,
 		CollectionNames:   scopes,
-		CheckpointPrefix:  prefix,
-	}
+		CheckpointPrefix:  fmt.Sprintf("%s:%v", db.MetadataKeys.DCPCheckpointPrefix(db.Options.GroupID), prefix),
+		Callback:          callback,
+	}, nil
 }
 
 type AttachmentMigrationManagerResponse struct {
@@ -310,12 +313,12 @@ type AttachmentMigrationManagerStatusDoc struct {
 	AttachmentMigrationMeta            `json:"meta"`
 }
 
-// GenerateAttachmentMigrationDCPStreamName returns the DCP stream name for attachment migration job.
-func GenerateAttachmentMigrationDCPStreamName(migrationID string) (string, error) {
-	return base.GenerateDcpStreamName(fmt.Sprintf(
+// getAttachmentMigrationPrefix returns a prefix for identifying attachment migration dcp feed and checkpoints.
+func getAttachmentMigrationPrefix(migrationID string) string {
+	return fmt.Sprintf(
 		"sg-%v:att_migration:%v",
 		base.ProductAPIVersion,
-		migrationID))
+		migrationID)
 }
 
 // resetDCPMetadataIfNeeded will check for mismatch between current collectionIDs and collectionIDs on previous run