fix test

Author: FGasper

internal/verifier/change_stream.go

@@ -104,6 +104,11 @@ func (verifier *Verifier) HandleChangeStreamEvents(ctx context.Context, batch []
 // GetChangeStreamFilter returns an aggregation pipeline that filters
 // namespaces as per configuration.
 //
+// Note that this omits verifier.globalFilter because we still need to
+// recheck any out-filter documents that may have changed in order to
+// account for filter traversals (i.e., updates that change whether a
+// document matches the filter).
+//
 // NB: Ideally we could make the change stream give $bsonSize(fullDocument)
 // and omit fullDocument, but $bsonSize was new in MongoDB 4.4, and we still
 // want to verify migrations from 4.2. fullDocument is unlikely to be a
@@ -144,14 +149,20 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		var changeEventBatch []ParsedEvent
 
 		for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
-			verifier.logger.Info().Msg("reading event")
 			gotEvent := cs.TryNext(ctx)
-			verifier.logger.Info().Err(cs.Err()).Msgf("got event? %v", gotEvent)
 
-			if !gotEvent || cs.Err() != nil {
+			if cs.Err() != nil {
+				return errors.Wrap(cs.Err(), "change stream iteration failed")
+			}
+
+			if !gotEvent {
 				break
 			}
 
+			verifier.logger.Debug().
+				Stringer("resumeToken", cs.ResumeToken()).
+				Msg("TryNext got an event!")
+
 			if changeEventBatch == nil {
 				changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
 			}
@@ -160,20 +171,13 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 				return errors.Wrap(err, "failed to decode change event")
 			}
 
-			verifier.logger.Debug().Interface("event", changeEventBatch[eventsRead]).Msg("Got event.")
-
 			eventsRead++
 		}
 
-		if cs.Err() != nil {
-			return errors.Wrap(cs.Err(), "change stream iteration failed")
-		}
-
 		if eventsRead == 0 {
 			return nil
 		}
 
-		verifier.logger.Debug().Int("eventsCount", eventsRead).Msgf("Received a batch of events.")
 		err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
 		if err != nil {
 			return errors.Wrap(err, "failed to handle change events")
@@ -302,7 +306,9 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 				Msg("Failed to extract timestamp from persisted resume token.")
 		}
 
-		logEvent.Msg("Starting change stream from persisted resume token.")
+		logEvent.
+			Interface("pipeline", pipeline).
+			Msg("Starting change stream from persisted resume token.")
 
 		opts = opts.SetStartAfter(savedResumeToken)
 	} else {

internal/verifier/check.go

@@ -77,6 +77,12 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context) error {
 func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 	verifier.logger.Debug().Msgf("Starting %d verification workers", verifier.numWorkers)
 	ctx, cancel := context.WithCancel(ctx)
+
+	if verifier.workerSleepDelayMillis == 0 {
+		verifier.logger.Info().
+			Msg("Worker sleep delay is zero. Only tests should do this.")
+	}
+
 	wg := sync.WaitGroup{}
 	for i := 0; i < verifier.numWorkers; i++ {
 		wg.Add(1)
@@ -185,7 +191,9 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	verifier.mux.RLock()
 	csRunning := verifier.changeStreamRunning
 	verifier.mux.RUnlock()
-	if !csRunning {
+	if csRunning {
+		verifier.logger.Debug().Msg("Check: Change stream already running.")
+	} else {
 		verifier.logger.Debug().Msg("Change stream not running; starting change stream")
 
 		err = verifier.StartChangeStream(ctx)
@@ -399,12 +407,15 @@ func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.Wait
 			if errors.Is(err, mongo.ErrNoDocuments) {
 				duration := verifier.workerSleepDelayMillis * time.Millisecond
 
-				verifier.logger.Debug().
-					Int("workerNum", workerNum).
-					Stringer("duration", duration).
-					Msg("No tasks found. Sleeping.")
+				if duration > 0 {
+					verifier.logger.Debug().
+						Int("workerNum", workerNum).
+						Stringer("duration", duration).
+						Msg("No tasks found. Sleeping.")
+
+					time.Sleep(duration)
+				}
 
-				time.Sleep(duration)
 				continue
 			} else if err != nil {
 				panic(err)

internal/verifier/integration_test_suite.go

@@ -94,6 +94,19 @@ func (suite *IntegrationTestSuite) SetupTest() {
 		dbname,
 	)
 
+	for _, client := range []*mongo.Client{suite.srcMongoClient, suite.dstMongoClient} {
+		dbNames, err := client.ListDatabaseNames(ctx, bson.D{})
+		suite.Require().NoError(err, "should list database names")
+		for _, dbName := range dbNames {
+			if strings.Index(dbName, suite.DBNameForTest()) == 0 {
+				suite.T().Logf("Dropping database %#q because it seems to be left over from an earlier run of this test.", dbName)
+				suite.Require().NoError(client.Database(dbName).Drop(ctx))
+			}
+
+			suite.initialDbNames.Add(dbName)
+		}
+	}
+
 	suite.testContext, suite.contextCanceller = ctx, canceller
 }
 

internal/verifier/migration_verifier_test.go

@@ -14,6 +14,7 @@ import (
 	"regexp"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/10gen/migration-verifier/internal/partitions"
 	"github.com/10gen/migration-verifier/internal/testutil"
@@ -1236,10 +1237,13 @@ func (suite *IntegrationTestSuite) TestVerifierNamespaceList() {
 	suite.ElementsMatch([]string{"testDb1.testColl1", "testDb1.testColl2", "testDb1.testView1"}, verifier.dstNamespaces)
 
 	// Collections in admin, config, and local should not be found
-	err = suite.srcMongoClient.Database("local").CreateCollection(ctx, "islocalSrc")
-	suite.Require().NoError(err)
-	err = suite.dstMongoClient.Database("local").CreateCollection(ctx, "islocalDest")
-	suite.Require().NoError(err)
+	if suite.GetSrcTopology() != TopologySharded {
+		err = suite.srcMongoClient.Database("local").CreateCollection(ctx, "islocalSrc")
+		suite.Require().NoError(err)
+		err = suite.dstMongoClient.Database("local").CreateCollection(ctx, "islocalDest")
+		suite.Require().NoError(err)
+	}
+
 	err = suite.srcMongoClient.Database("admin").CreateCollection(ctx, "isAdminSrc")
 	suite.Require().NoError(err)
 	err = suite.dstMongoClient.Database("admin").CreateCollection(ctx, "isAdminDest")
@@ -1388,19 +1392,22 @@ func (suite *IntegrationTestSuite) TestGenerationalRechecking() {
 func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 	zerolog.SetGlobalLevel(zerolog.DebugLevel)
 
-	filter := map[string]any{"inFilter": map[string]any{"$ne": false}}
+	dbname1 := suite.DBNameForTest("1")
+	dbname2 := suite.DBNameForTest("2")
+
+	filter := bson.M{"inFilter": bson.M{"$ne": false}}
 	verifier := suite.BuildVerifier()
-	verifier.SetSrcNamespaces([]string{"testDb1.testColl1"})
-	verifier.SetDstNamespaces([]string{"testDb2.testColl3"})
+	verifier.SetSrcNamespaces([]string{dbname1 + ".testColl1"})
+	verifier.SetDstNamespaces([]string{dbname2 + ".testColl3"})
 	verifier.SetNamespaceMap()
 	verifier.SetIgnoreBSONFieldOrder(true)
 	// Set this value low to test the verifier with multiple partitions.
 	verifier.partitionSizeInBytes = 50
 
 	ctx := suite.Context()
 
-	srcColl := suite.srcMongoClient.Database("testDb1").Collection("testColl1")
-	dstColl := suite.dstMongoClient.Database("testDb2").Collection("testColl3")
+	srcColl := suite.srcMongoClient.Database(dbname1).Collection("testColl1")
+	dstColl := suite.dstMongoClient.Database(dbname2).Collection("testColl3")
 
 	// Documents with _id in [0, 100) should match.
 	var docs []interface{}
@@ -1432,7 +1439,11 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 		suite.Require().NoError(err)
 
 		for status.TotalTasks == 0 && verifier.generation < 10 {
-			suite.T().Logf("TotalTasks is 0 (generation=%d); waiting another generation …", verifier.generation)
+			delay := time.Second
+
+			suite.T().Logf("TotalTasks is 0 (generation=%d); waiting %s then will run another generation …", verifier.generation, delay)
+
+			time.Sleep(delay)
 			checkContinueChan <- struct{}{}
 			<-checkDoneChan
 			status, err = verifier.GetVerificationStatus()
@@ -1449,6 +1460,7 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 	suite.Require().Equal(status.FailedTasks, 0)
 
 	// Insert another document that is not in the filter.
+	// This should trigger a recheck despite being outside the filter.
 	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 200, "x": 200, "inFilter": false})
 	suite.Require().NoError(err)
 
@@ -1457,11 +1469,13 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 
 	// Wait for generation to finish.
 	<-checkDoneChan
+
 	status = waitForTasks()
+
 	// There should be no failures, since the inserted document is not in the filter.
 	suite.Require().Equal(VerificationStatus{TotalTasks: 1, CompletedTasks: 1}, *status)
 
-	// Now insert in the source, this should come up next generation.
+	// Now insert in the source. This should come up next generation.
 	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 201, "x": 201, "inFilter": true})
 	suite.Require().NoError(err)