generational rechecking

Author: FGasper

internal/verifier/compare.go

@@ -64,8 +64,8 @@ func (verifier *Verifier) compareDocsFromChannels(
 	error,
 ) {
 	results := []VerificationResult{}
-	var docCount types.DocumentCount
-	var byteCount types.ByteCount
+	var srcDocCount types.DocumentCount
+	var srcByteCount types.ByteCount
 
 	mapKeyFieldNames := make([]string, 1+len(task.QueryFilter.ShardKeys))
 	mapKeyFieldNames[0] = "_id"
@@ -89,6 +89,9 @@ func (verifier *Verifier) compareDocsFromChannels(
 		if isSrc {
 			ourMap = srcCache
 			theirMap = dstCache
+
+			srcDocCount++
+			srcByteCount += types.ByteCount(len(doc))
 		} else {
 			ourMap = dstCache
 			theirMap = srcCache
@@ -148,8 +151,11 @@ func (verifier *Verifier) compareDocsFromChannels(
 	for !srcClosed || !dstClosed {
 		simpleTimerReset(readTimer, readTimeout)
 
+		var srcDoc, dstDoc bson.Raw
+
 		eg, egCtx := errgroup.WithContext(ctx)
 		eg.Go(func() error {
+			var alive bool
 			select {
 			case <-egCtx.Done():
 				return egCtx.Err()
@@ -158,30 +164,19 @@ func (verifier *Verifier) compareDocsFromChannels(
 					"failed to read from source after %s",
 					readTimeout,
 				)
-			case doc, alive := <-srcChannel:
+			case srcDoc, alive = <-srcChannel:
 				if !alive {
 					srcClosed = true
 					break
 				}
-				docCount++
-				byteCount += types.ByteCount(len(doc))
-
-				err := handleNewDoc(doc, true)
-
-				if err != nil {
-					return errors.Wrapf(
-						err,
-						"comparer thread failed to handle source doc with ID %v",
-						doc.Lookup("_id"),
-					)
-				}
 			}
 
 			return nil
 		})
 
 		if !dstClosed {
 			eg.Go(func() error {
+				var alive bool
 				select {
 				case <-egCtx.Done():
 					return egCtx.Err()
@@ -190,21 +185,11 @@ func (verifier *Verifier) compareDocsFromChannels(
 						"failed to read from destination after %s",
 						readTimeout,
 					)
-				case doc, alive := <-dstChannel:
+				case dstDoc, alive = <-dstChannel:
 					if !alive {
 						dstClosed = true
 						break
 					}
-
-					err := handleNewDoc(doc, false)
-
-					if err != nil {
-						return errors.Wrapf(
-							err,
-							"comparer thread failed to handle destination doc with ID %v",
-							doc.Lookup("_id"),
-						)
-					}
 				}
 
 				return nil
@@ -214,9 +199,35 @@ func (verifier *Verifier) compareDocsFromChannels(
 		if err := eg.Wait(); err != nil {
 			return nil, 0, 0, errors.Wrap(
 				err,
-				"failed to compare documents",
+				"failed to read documents",
 			)
 		}
+
+		if srcDoc != nil {
+			err := handleNewDoc(srcDoc, true)
+
+			if err != nil {
+				return nil, 0, 0, errors.Wrapf(
+					err,
+					"comparer thread failed to handle task %s's source doc with ID %v",
+					task.PrimaryKey,
+					srcDoc.Lookup("_id"),
+				)
+			}
+		}
+
+		if dstDoc != nil {
+			err := handleNewDoc(dstDoc, false)
+
+			if err != nil {
+				return nil, 0, 0, errors.Wrapf(
+					err,
+					"comparer thread failed to handle task %s's destination doc with ID %v",
+					task.PrimaryKey,
+					dstDoc.Lookup("_id"),
+				)
+			}
+		}
 	}
 
 	// We got here because both srcChannel and dstChannel are closed,
@@ -255,7 +266,7 @@ func (verifier *Verifier) compareDocsFromChannels(
 		)
 	}
 
-	return results, docCount, byteCount, nil
+	return results, srcDocCount, srcByteCount, nil
 }
 
 func simpleTimerReset(t *time.Timer, dur time.Duration) {

internal/verifier/migration_verifier_test.go

@@ -1281,16 +1281,19 @@ func (suite *IntegrationTestSuite) TestMetadataMismatchAndPartitioning() {
 }
 
 func (suite *IntegrationTestSuite) TestGenerationalRechecking() {
+	dbname1 := suite.DBNameForTest("1")
+	dbname2 := suite.DBNameForTest("2")
+
 	zerolog.SetGlobalLevel(zerolog.DebugLevel)
 	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()
 
 	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")
 	_, err := srcColl.InsertOne(ctx, bson.M{"_id": 1, "x": 42})
 	suite.Require().NoError(err)
 	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 2, "x": 43})