Unskip tests

Author: torcolvin

base/util_testing.go

@@ -758,24 +758,10 @@ func TestRequiresCollections(t testing.TB) {
 	}
 }
 
-// TestRequiresOneShotDCPClient will skip the current test until rosmar supports one-shot DCP.
-func TestRequiresOneShotDCPClient(t testing.TB) {
-	if UnitTestUrlIsWalrus() {
-		t.Skip("rosmar doesn't have an abstracted one shot DCP client CBG-4246")
-	}
-}
-
-// TestRequiresDCPResync will skip the current test DCP sync is not supported.
-func TestRequiresDCPResync(t testing.TB) {
-	if UnitTestUrlIsWalrus() {
-		t.Skip("Walrus doesn't support DCP resync CBG-2661/CBG-4246")
-	}
-}
-
 // TestRequiresGocbDCPClient will skip the current test if using rosmar.
 func TestRequiresGocbDCPClient(t testing.TB) {
 	if UnitTestUrlIsWalrus() {
-		t.Skip("rosmar doesn't support base.DCPClient")
+		t.Skip("rosmar doesn't support GocbDCPClient")
 	}
 }
 

db/attachment_compaction.go

@@ -29,7 +29,7 @@ const (
 	CleanupPhase    = "cleanup"
 )
 
-func attachmentCompactMarkPhase(ctx context.Context, dataStore base.DataStore, collectionID uint32, db *Database, compactionID string, terminator *base.SafeTerminator, markedAttachmentCount *base.AtomicInt) (count int64, vbUUIDs []uint64, checkpointPrefix string, err error) {
+func attachmentCompactMarkPhase(ctx context.Context, dataStore base.DataStore, collectionID uint32, db *Database, compactionID string, terminator *base.SafeTerminator, markedAttachmentCount *base.AtomicInt) (count int64, vbUUIDs []uint64, checkpointPrefix string, feedPrefix string, err error) {
 	base.InfofCtx(ctx, base.KeyAll, "Starting first phase of attachment compaction (mark phase) with compactionID: %q", compactionID)
 	compactionLoggingID := "Compaction Mark: " + compactionID
 
@@ -138,43 +138,43 @@ func attachmentCompactMarkPhase(ctx context.Context, dataStore base.DataStore, c
 	dcpClient, err := base.NewDCPClient(ctx, db.Bucket, clientOptions)
 	if err != nil {
 		base.WarnfCtx(ctx, "[%s] Failed to create attachment compaction DCP client! %v", compactionLoggingID, err)
-		return 0, nil, "", err
+		return 0, nil, "", "", err
 	}
 	metadataKeyPrefix := dcpClient.GetMetadataKeyPrefix()
 
 	doneChan, err := dcpClient.Start(ctx)
 	if err != nil {
 		base.WarnfCtx(ctx, "[%s] Failed to start attachment compaction DCP feed! %v", compactionLoggingID, err)
 		dcpClient.Close()
-		return 0, nil, metadataKeyPrefix, err
+		return 0, nil, metadataKeyPrefix, clientOptions.FeedPrefix, err
 	}
 	base.DebugfCtx(ctx, base.KeyAll, "[%s] DCP feed started.", compactionLoggingID)
 
 	select {
 	case <-doneChan:
 		base.InfofCtx(ctx, base.KeyAll, "[%s] Mark phase of attachment compaction completed. Marked %d attachments", compactionLoggingID, markedAttachmentCount.Value())
 		if markProcessFailureErr != nil {
-			return markedAttachmentCount.Value(), nil, metadataKeyPrefix, markProcessFailureErr
+			return markedAttachmentCount.Value(), nil, metadataKeyPrefix, clientOptions.FeedPrefix, markProcessFailureErr
 		}
 	case <-terminator.Done():
 		base.DebugfCtx(ctx, base.KeyAll, "[%s] Terminator closed. Stopping mark phase.", compactionLoggingID)
 		dcpClient.Close()
 		if markProcessFailureErr != nil {
-			return markedAttachmentCount.Value(), nil, metadataKeyPrefix, markProcessFailureErr
+			return markedAttachmentCount.Value(), nil, metadataKeyPrefix, clientOptions.FeedPrefix, markProcessFailureErr
 		}
 		if err != nil {
-			return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, err
+			return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, clientOptions.FeedPrefix, err
 		}
 
 		err = <-doneChan
 		if err != nil {
-			return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, err
+			return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, clientOptions.FeedPrefix, err
 		}
 
 		base.InfofCtx(ctx, base.KeyAll, "[%s] Mark phase of attachment compaction was terminated. Marked %d attachments", compactionLoggingID, markedAttachmentCount.Value())
 	}
 
-	return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, err
+	return markedAttachmentCount.Value(), base.GetVBUUIDs(dcpClient.GetMetadata()), metadataKeyPrefix, clientOptions.FeedPrefix, err
 }
 
 // AttachmentsMetaMap struct is a very minimal struct to unmarshal into when getting attachments from bodies
@@ -406,7 +406,7 @@ func attachmentCompactSweepPhase(ctx context.Context, dataStore base.DataStore,
 	return purgedAttachmentCount.Value(), err
 }
 
-func attachmentCompactCleanupPhase(ctx context.Context, dataStore base.DataStore, collectionID uint32, db *Database, compactionID string, vbUUIDs []uint64, terminator *base.SafeTerminator) (string, error) {
+func attachmentCompactCleanupPhase(ctx context.Context, dataStore base.DataStore, collectionID uint32, db *Database, compactionID string, vbUUIDs []uint64, terminator *base.SafeTerminator) (checkpointPrefix string, feedPrefix string, err error) {
 	base.InfofCtx(ctx, base.KeyAll, "Starting third phase of attachment compaction (cleanup phase) with compactionID: %q", compactionID)
 	compactionLoggingID := "Compaction Cleanup: " + compactionID
 
@@ -495,23 +495,18 @@ func attachmentCompactCleanupPhase(ctx context.Context, dataStore base.DataStore
 
 	base.InfofCtx(ctx, base.KeyAll, "[%s] Starting DCP feed for cleanup phase of attachment compaction", compactionLoggingID)
 
-	bucket, err := base.AsGocbV2Bucket(db.Bucket)
-	if err != nil {
-		return "", err
-	}
-
-	dcpClient, err := base.NewDCPClient(ctx, bucket, clientOptions)
+	dcpClient, err := base.NewDCPClient(ctx, db.Bucket, clientOptions)
 	if err != nil {
 		base.WarnfCtx(ctx, "[%s] Failed to create attachment compaction DCP client! %v", compactionLoggingID, err)
-		return "", err
+		return "", "", err
 	}
 	metadataKeyPrefix := dcpClient.GetMetadataKeyPrefix()
 
 	doneChan, err := dcpClient.Start(ctx)
 	if err != nil {
 		base.WarnfCtx(ctx, "[%s] Failed to start attachment compaction DCP feed! %v", compactionLoggingID, err)
 		dcpClient.Close()
-		return metadataKeyPrefix, err
+		return metadataKeyPrefix, clientOptions.FeedPrefix, err
 	}
 
 	select {
@@ -521,13 +516,13 @@ func attachmentCompactCleanupPhase(ctx context.Context, dataStore base.DataStore
 		dcpClient.Close()
 		err = <-doneChan
 		if err != nil {
-			return metadataKeyPrefix, err
+			return metadataKeyPrefix, clientOptions.FeedPrefix, err
 		}
 
 		base.InfofCtx(ctx, base.KeyAll, "[%s] Cleanup phase of attachment compaction was terminated", compactionLoggingID)
 	}
 
-	return metadataKeyPrefix, err
+	return metadataKeyPrefix, clientOptions.FeedPrefix, err
 }
 
 // getCompactionIDSubDocPath is just a tiny helper func that just concatenates the subdoc path we're using to store

db/attachment_compaction_test.go

@@ -62,7 +62,7 @@ func TestAttachmentMark(t *testing.T) {
 	attKeys = append(attKeys, createDocWithInBodyAttachment(t, ctx, "inBodyDoc", []byte(`{}`), "attForInBodyRef", []byte(`{"val": "inBodyAtt"}`), databaseCollection))
 
 	terminator := base.NewSafeTerminator()
-	attachmentsMarked, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, t.Name(), terminator, &base.AtomicInt{})
+	attachmentsMarked, _, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, t.Name(), terminator, &base.AtomicInt{})
 	assert.NoError(t, err)
 	assert.Equal(t, int64(13), attachmentsMarked)
 
@@ -202,7 +202,7 @@ func TestAttachmentCleanup(t *testing.T) {
 	}
 
 	terminator := base.NewSafeTerminator()
-	_, err := attachmentCompactCleanupPhase(ctx, dataStore, collectionID, testDb, t.Name(), nil, terminator)
+	_, _, err := attachmentCompactCleanupPhase(ctx, dataStore, collectionID, testDb, t.Name(), nil, terminator)
 	assert.NoError(t, err)
 
 	for _, docID := range singleMarkedAttIDs {
@@ -357,7 +357,7 @@ func TestAttachmentMarkAndSweepAndCleanup(t *testing.T) {
 	}
 
 	terminator := base.NewSafeTerminator()
-	attachmentsMarked, vbUUIDS, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, t.Name(), terminator, &base.AtomicInt{})
+	attachmentsMarked, vbUUIDS, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, t.Name(), terminator, &base.AtomicInt{})
 	assert.NoError(t, err)
 	assert.Equal(t, int64(10), attachmentsMarked)
 
@@ -382,7 +382,7 @@ func TestAttachmentMarkAndSweepAndCleanup(t *testing.T) {
 		}
 	}
 
-	_, err = attachmentCompactCleanupPhase(ctx, dataStore, collectionID, testDb, t.Name(), vbUUIDS, terminator)
+	_, _, err = attachmentCompactCleanupPhase(ctx, dataStore, collectionID, testDb, t.Name(), vbUUIDS, terminator)
 	assert.NoError(t, err)
 
 	for _, attDocKey := range attKeys {
@@ -696,7 +696,7 @@ func TestAttachmentDifferentVBUUIDsBetweenPhases(t *testing.T) {
 
 	// Run mark phase as usual
 	terminator := base.NewSafeTerminator()
-	_, vbUUIDs, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDB, t.Name(), terminator, &base.AtomicInt{})
+	_, vbUUIDs, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDB, t.Name(), terminator, &base.AtomicInt{})
 	assert.NoError(t, err)
 
 	// Manually modify a vbUUID and ensure the Sweep phase errors
@@ -975,7 +975,7 @@ func TestAttachmentCompactIncorrectStat(t *testing.T) {
 	stat := &base.AtomicInt{}
 	count := int64(0)
 	go func() {
-		attachmentCount, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, "mark", terminator, stat)
+		attachmentCount, _, _, _, err := attachmentCompactMarkPhase(ctx, dataStore, collectionID, testDb, "mark", terminator, stat)
 		atomic.StoreInt64(&count, attachmentCount)
 		require.NoError(t, err)
 	}()

db/background_mgr_attachment_compaction.go

@@ -133,9 +133,10 @@ func (a *AttachmentCompactionManager) Run(ctx context.Context, options map[strin
 		a.SetPhase("mark")
 		worker := func() (shouldRetry bool, err error, value any) {
 			persistClusterStatus()
-			_, a.VBUUIDs, metadataKeyPrefix, err = attachmentCompactMarkPhase(ctx, dataStore, collectionID, database, a.CompactID, terminator, &a.MarkedAttachments)
+			var feedPrefix string
+			_, a.VBUUIDs, metadataKeyPrefix, feedPrefix, err = attachmentCompactMarkPhase(ctx, dataStore, collectionID, database, a.CompactID, terminator, &a.MarkedAttachments)
 			if err != nil {
-				shouldRetry, err = a.handleAttachmentCompactionRollbackError(ctx, options, dataStore, database, err, MarkPhase, metadataKeyPrefix)
+				shouldRetry, err = a.handleAttachmentCompactionRollbackError(ctx, options, dataStore, database, err, MarkPhase, metadataKeyPrefix, feedPrefix)
 			}
 			return shouldRetry, err, nil
 		}
@@ -161,9 +162,10 @@ func (a *AttachmentCompactionManager) Run(ctx context.Context, options map[strin
 		a.SetPhase("cleanup")
 		worker := func() (shouldRetry bool, err error, value any) {
 			persistClusterStatus()
-			metadataKeyPrefix, err = attachmentCompactCleanupPhase(ctx, dataStore, collectionID, database, a.CompactID, a.VBUUIDs, terminator)
+			var feedPrefix string
+			metadataKeyPrefix, feedPrefix, err = attachmentCompactCleanupPhase(ctx, dataStore, collectionID, database, a.CompactID, a.VBUUIDs, terminator)
 			if err != nil {
-				shouldRetry, err = a.handleAttachmentCompactionRollbackError(ctx, options, dataStore, database, err, CleanupPhase, metadataKeyPrefix)
+				shouldRetry, err = a.handleAttachmentCompactionRollbackError(ctx, options, dataStore, database, err, CleanupPhase, metadataKeyPrefix, feedPrefix)
 			}
 			return shouldRetry, err, nil
 		}
@@ -182,13 +184,13 @@ func (a *AttachmentCompactionManager) Run(ctx context.Context, options map[strin
 	return nil
 }
 
-func (a *AttachmentCompactionManager) handleAttachmentCompactionRollbackError(ctx context.Context, options map[string]any, dataStore base.DataStore, database *Database, err error, phase, keyPrefix string) (bool, error) {
+func (a *AttachmentCompactionManager) handleAttachmentCompactionRollbackError(ctx context.Context, options map[string]any, dataStore base.DataStore, database *Database, err error, phase, checkpointPrefix, feedPrefix string) (bool, error) {
 	var rollbackErr gocbcore.DCPRollbackError
 	if errors.As(err, &rollbackErr) || errors.Is(err, base.ErrVbUUIDMismatch) {
 		base.InfofCtx(ctx, base.KeyDCP, "rollback indicated on %s phase of attachment compaction, resetting the task", phase)
 		// to rollback any phase for attachment compaction we need to purge all persisted dcp metadata
 		base.InfofCtx(ctx, base.KeyDCP, "Purging invalid checkpoints for background task run %s", a.CompactID)
-		err = PurgeDCPCheckpoints(ctx, database.DatabaseContext, keyPrefix, a.CompactID)
+		err = PurgeDCPCheckpoints(ctx, database.DatabaseContext, checkpointPrefix, feedPrefix)
 		if err != nil {
 			base.WarnfCtx(ctx, "error occurred during purging of dcp metadata: %s", err)
 			return false, err

db/background_mgr_attachment_migration.go

@@ -165,7 +165,7 @@ func (a *AttachmentMigrationManager) Run(ctx context.Context, options map[string
 	dcpOptions := getMigrationDCPClientOptions(db, a.MigrationID, scopes, callback)
 
 	// check for mismatch in collection id's between current collections on the db and prev run
-	err = a.resetDCPMetadataIfNeeded(ctx, db, dcpOptions.CheckpointPrefix, currCollectionIDs)
+	err = a.resetDCPMetadataIfNeeded(ctx, db, dcpOptions.CheckpointPrefix, dcpOptions.FeedPrefix, currCollectionIDs)
 	if err != nil {
 		return err
 	}
@@ -315,15 +315,15 @@ func getAttachmentMigrationPrefix(migrationID string) string {
 }
 
 // resetDCPMetadataIfNeeded will check for mismatch between current collectionIDs and collectionIDs on previous run
-func (a *AttachmentMigrationManager) resetDCPMetadataIfNeeded(ctx context.Context, database *DatabaseContext, metadataKeyPrefix string, collectionIDs []uint32) error {
+func (a *AttachmentMigrationManager) resetDCPMetadataIfNeeded(ctx context.Context, database *DatabaseContext, checkpointPrefix string, feedPrefix string, collectionIDs []uint32) error {
 	// if we are on our first run, no collections will be defined on the manager yet
 	if len(a.CollectionIDs) == 0 {
 		return nil
 	}
 	if len(a.CollectionIDs) != len(collectionIDs) {
 		base.InfofCtx(ctx, base.KeyDCP, "Purging invalid checkpoints for background task run %s", a.MigrationID)
-		err := PurgeDCPCheckpoints(ctx, database, metadataKeyPrefix, a.MigrationID)
-		if err != nil {
+		err := PurgeDCPCheckpoints(ctx, database, checkpointPrefix, feedPrefix)
+		if err != nil && !base.IsDocNotFoundError(err) {
 			return err
 		}
 		return nil
@@ -333,7 +333,7 @@ func (a *AttachmentMigrationManager) resetDCPMetadataIfNeeded(ctx context.Contex
 	purgeNeeded := slices.Compare(collectionIDs, a.CollectionIDs)
 	if purgeNeeded != 0 {
 		base.InfofCtx(ctx, base.KeyDCP, "Purging invalid checkpoints for background task run %s", a.MigrationID)
-		err := PurgeDCPCheckpoints(ctx, database, metadataKeyPrefix, a.MigrationID)
+		err := PurgeDCPCheckpoints(ctx, database, checkpointPrefix, feedPrefix)
 		if err != nil {
 			return err
 		}

db/database.go

@@ -2466,7 +2466,7 @@ func (db *DatabaseContext) StartOnlineProcesses(ctx context.Context) (returnedEr
 
 	db.AttachmentMigrationManager = NewAttachmentMigrationManager(db)
 	// if we have collections requiring migration, run the job
-	if len(db.RequireAttachmentMigration) > 0 && !db.BucketSpec.IsWalrusBucket() {
+	if len(db.RequireAttachmentMigration) > 0 {
 		err := db.AttachmentMigrationManager.Start(ctx, nil)
 		if err != nil {
 			base.WarnfCtx(ctx, "Error trying to migrate attachments for %s with error: %v", db.Name, err)
@@ -2572,11 +2572,21 @@ func (db *DatabaseContext) GetCollectionIDs() []uint32 {
 }
 
 // PurgeDCPCheckpoints will purge all DCP metadata from previous run in the bucket, used to reset dcp client to 0
-func PurgeDCPCheckpoints(ctx context.Context, database *DatabaseContext, checkpointPrefix string, taskID string) error {
+func PurgeDCPCheckpoints(ctx context.Context, database *DatabaseContext, checkpointPrefix string, feedPrefix string) error {
 
 	bucket, err := base.AsGocbV2Bucket(database.Bucket)
 	if err != nil {
-		return err
+		checkpoint := checkpointPrefix + ":" + feedPrefix
+		fmt.Printf("Deleting DCP checkpoint %q\n", checkpoint)
+		err := database.MetadataStore.Delete(checkpoint)
+		if err != nil && !base.IsDocNotFoundError(err) {
+			return err
+		}
+		if base.IsDocNotFoundError(err) {
+			fmt.Printf("No DCP checkpoint found %q\n", checkpoint)
+			return nil
+		}
+		return nil
 	}
 	numVbuckets, err := bucket.GetMaxVbno()
 	if err != nil {

rest/access_test.go

@@ -760,7 +760,6 @@ func TestAllDocsAccessControl(t *testing.T) {
 }
 
 func TestChannelAccessChanges(t *testing.T) {
-	base.TestRequiresDCPResync(t)
 	base.SetUpTestLogging(t, base.LevelDebug, base.KeyCache, base.KeyChanges, base.KeyCRUD)
 	rtConfig := RestTesterConfig{SyncFn: `function(doc) {access(doc.owner, doc._id);channel(doc.channel)}`, PersistentConfig: true}
 	rt := NewRestTester(t, &rtConfig)

rest/adminapitest/admin_api_test.go

@@ -620,7 +620,6 @@ func TestDBGetConfigCustomLogging(t *testing.T) {
 }
 
 func TestDBOfflineSingleResyncUsingDCPStream(t *testing.T) {
-	base.TestRequiresDCPResync(t)
 	syncFn := `
 	function(doc) {
 		channel("x")
@@ -647,7 +646,6 @@ func TestDBOfflineSingleResyncUsingDCPStream(t *testing.T) {
 }
 
 func TestDCPResyncCollectionsStatus(t *testing.T) {
-	base.TestRequiresDCPResync(t)
 	base.TestRequiresCollections(t)
 
 	testCases := []struct {
@@ -704,8 +702,6 @@ func TestDCPResyncCollectionsStatus(t *testing.T) {
 }
 
 func TestResyncUsingDCPStream(t *testing.T) {
-	base.TestRequiresDCPResync(t)
-
 	testCases := []struct {
 		docsCreated int
 	}{
@@ -770,8 +766,6 @@ func TestResyncUsingDCPStream(t *testing.T) {
 }
 
 func TestResyncUsingDCPStreamReset(t *testing.T) {
-	base.TestRequiresDCPResync(t)
-
 	syncFn := `
 	function(doc) {
 		channel("x")
@@ -832,7 +826,6 @@ func TestResyncUsingDCPStreamReset(t *testing.T) {
 }
 
 func TestResyncUsingDCPStreamForNamedCollection(t *testing.T) {
-	base.TestRequiresDCPResync(t)
 	base.TestRequiresCollections(t)
 
 	numCollections := 2
@@ -894,8 +887,6 @@ func TestResyncUsingDCPStreamForNamedCollection(t *testing.T) {
 }
 
 func TestResyncErrorScenariosUsingDCPStream(t *testing.T) {
-	base.TestRequiresDCPResync(t)
-
 	syncFn := `
 	function(doc) {
 		channel("x")
@@ -1420,8 +1411,6 @@ func TestConfigPollingRemoveDatabase(t *testing.T) {
 }
 
 func TestResyncStopUsingDCPStream(t *testing.T) {
-	base.TestRequiresDCPResync(t)
-
 	syncFn := `
 	function(doc) {
 		channel("x")

rest/adminapitest/resync_test.go

@@ -24,9 +24,7 @@ import (
 
 // TestResyncRollback ensures that we allow rollback of
 func TestResyncRollback(t *testing.T) {
-	if base.UnitTestUrlIsWalrus() {
-		t.Skip("This test doesn't works with walrus")
-	}
+	base.TestRequiresGocbDCPClient(t)
 	rt := rest.NewRestTester(t, &rest.RestTesterConfig{
 		SyncFn: `function(doc) { channel("x") }`, // use custom sync function to increment sync function counter
 	})
@@ -128,7 +126,6 @@ func TestResyncRegenerateSequencesCorruptDocumentSequence(t *testing.T) {
 }
 
 func TestResyncRegenerateSequencesPrincipals(t *testing.T) {
-	base.TestRequiresDCPResync(t)
 	if !base.TestsUseNamedCollections() {
 		t.Skip("Test requires named collections, performs default collection handling independently")
 	}