wait for recheck docs creation

Author: tdq45gj

internal/verifier/change_stream.go

@@ -114,6 +114,7 @@ func (verifier *Verifier) StartChangeEventHandler(ctx context.Context, reader *C
 					// Change stream reader has closed the event batch channel because it has finished.
 					return nil
 				}
+				verifier.logger.Trace().Msgf("Verifier is handling a change event batch from %s: %v", reader, batch)
 				err := verifier.HandleChangeStreamEvents(ctx, batch, reader.readerType)
 				if err != nil {
 					reader.ChangeStreamErrChan <- err
@@ -277,6 +278,8 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 			return errors.Wrapf(err, "failed to decode change event to %T", changeEventBatch[eventsRead])
 		}
 
+		csr.logger.Trace().Msgf("%s received a change event: %v", csr, changeEventBatch[eventsRead])
+
 		if changeEventBatch[eventsRead].ClusterTime != nil &&
 			(csr.lastChangeEventTime == nil ||
 				csr.lastChangeEventTime.Before(*changeEventBatch[eventsRead].ClusterTime)) {
@@ -606,7 +609,7 @@ func (csr *ChangeStreamReader) persistChangeStreamResumeToken(ctx context.Contex
 				Msg("failed to extract resume token timestamp")
 		}
 
-		logEvent.Msg("Persisted change stream resume token.")
+		logEvent.Msgf("Persisted %s's resume token.", csr)
 
 		return nil
 	}

internal/verifier/migration_verifier_test.go

@@ -1553,8 +1553,22 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 	<-checkDoneChan
 }
 
+func (suite *IntegrationTestSuite) waitForRecheckDocs(verifier *Verifier) {
+	suite.Eventually(func() bool {
+		cursor, err := suite.metaMongoClient.Database(verifier.metaDBName).Collection(recheckQueue).Find(suite.Context(), bson.D{})
+		var docs []bson.D
+		suite.Require().NoError(err)
+		suite.Require().NoError(cursor.All(suite.Context(), &docs))
+		return len(docs) > 0
+	}, 1*time.Minute, 100*time.Millisecond)
+}
+
 func (suite *IntegrationTestSuite) TestChangesOnDstBeforeSrc() {
+	zerolog.SetGlobalLevel(zerolog.TraceLevel)
+	defer zerolog.SetGlobalLevel(zerolog.DebugLevel)
+
 	ctx := suite.Context()
+
 	collName := "mycoll"
 
 	srcDB := suite.srcMongoClient.Database(suite.DBNameForTest())
@@ -1574,7 +1588,9 @@ func (suite *IntegrationTestSuite) TestChangesOnDstBeforeSrc() {
 	suite.Require().NoError(err)
 	_, err = dstDB.Collection(collName).InsertOne(ctx, bson.D{{"_id", 2}})
 	suite.Require().NoError(err)
+
 	suite.Require().NoError(runner.AwaitGenerationEnd())
+	suite.waitForRecheckDocs(verifier)
 
 	// Run generation 2 and get verification status.
 	suite.Require().NoError(runner.StartNextGeneration())
@@ -1591,6 +1607,8 @@ func (suite *IntegrationTestSuite) TestChangesOnDstBeforeSrc() {
 	_, err = srcDB.Collection(collName).InsertOne(ctx, bson.D{{"_id", 1}})
 	suite.Require().NoError(err)
 	suite.Require().NoError(runner.AwaitGenerationEnd())
+	suite.waitForRecheckDocs(verifier)
+
 	status, err = verifier.GetVerificationStatus(ctx)
 	suite.Require().NoError(err)
 	suite.Assert().Equal(
@@ -1603,6 +1621,7 @@ func (suite *IntegrationTestSuite) TestChangesOnDstBeforeSrc() {
 	_, err = srcDB.Collection(collName).InsertOne(ctx, bson.D{{"_id", 2}})
 	suite.Require().NoError(err)
 	suite.Require().NoError(runner.AwaitGenerationEnd())
+	suite.waitForRecheckDocs(verifier)
 
 	// Everything should match by the end of it.
 	status, err = verifier.GetVerificationStatus(ctx)