many retries

Author: FGasper

internal/retry/retry.go

@@ -11,6 +11,17 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 )
 
+// Retry is a convenience that creates a retryer and executes it.
+// See RunForTransientErrorsOnly for argument details.
+func Retry(
+	ctx context.Context,
+	logger *logger.Logger,
+	f func(*Info) error,
+) error {
+	retryer := New(DefaultDurationLimit)
+	return retryer.RunForTransientErrorsOnly(ctx, logger, f)
+}
+
 // RunForUUIDAndTransientErrors retries f() for the CollectionUUIDMismatch error and for transient errors.
 // This should be used to run a driver operation that optionally specifies the `collectionUUID` parameter
 // for a collection that may have been:

internal/verifier/change_stream_test.go

@@ -346,6 +346,7 @@ func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
 
 	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
 		ctx,
+		verifier.logger,
 		verifier.verificationTaskCollection(),
 		verificationTaskVerifyDocuments,
 		verifier.generation,
@@ -395,6 +396,7 @@ func (suite *IntegrationTestSuite) testInsertsBeforeWritesOff(docsCount int) {
 	generation := verifier.generation
 	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
 		ctx,
+		verifier.logger,
 		verifier.verificationTaskCollection(),
 		verificationTaskVerifyDocuments,
 		generation,

internal/verifier/check.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/10gen/migration-verifier/internal/logger"
+	"github.com/10gen/migration-verifier/internal/retry"
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
@@ -178,19 +180,27 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 			return err
 		}
 	}
-	err = verifier.AddMetaIndexes(ctx)
-	if err != nil {
-		return err
-	}
 
-	err = verifier.doInMetaTransaction(
-		ctx,
-		func(ctx context.Context, sCtx mongo.SessionContext) error {
-			return verifier.ResetInProgressTasks(sCtx)
-		},
-	)
+	err = retry.Retry(ctx, verifier.logger, func(_ *retry.Info) error {
+		err := verifier.AddMetaIndexes(ctx)
+		if err != nil {
+			return errors.Wrap(err, "failed to create verifier metadata's indexes")
+		}
+
+		err = verifier.doInMetaTransaction(
+			ctx,
+			func(ctx context.Context, sCtx mongo.SessionContext) error {
+				return verifier.ResetInProgressTasks(sCtx)
+			},
+		)
+		if err != nil {
+			return errors.Wrap(err, "failed to reset any in-progress tasks")
+		}
+
+		return nil
+	})
 	if err != nil {
-		return errors.Wrap(err, "failed to reset any in-progress tasks")
+		return err
 	}
 
 	verifier.logger.Debug().Msg("Starting Check")
@@ -352,7 +362,7 @@ func (verifier *Verifier) CreateInitialTasks(ctx context.Context) error {
 			)
 		}
 	}
-	isPrimary, err := verifier.CheckIsPrimary(ctx)
+	isPrimary, err := verifier.CreatePrimaryTaskIfNeeded(ctx)
 	if err != nil {
 		return err
 	}
@@ -386,30 +396,40 @@ func (verifier *Verifier) CreateInitialTasks(ctx context.Context) error {
 	return nil
 }
 
-func FetchFailedAndIncompleteTasks(ctx context.Context, coll *mongo.Collection, taskType verificationTaskType, generation int) ([]VerificationTask, []VerificationTask, error) {
+func FetchFailedAndIncompleteTasks(
+	ctx context.Context,
+	logger *logger.Logger,
+	coll *mongo.Collection,
+	taskType verificationTaskType,
+	generation int,
+) ([]VerificationTask, []VerificationTask, error) {
 	var FailedTasks, allTasks, IncompleteTasks []VerificationTask
 
-	cur, err := coll.Find(ctx, bson.D{
-		bson.E{Key: "type", Value: taskType},
-		bson.E{Key: "generation", Value: generation},
-	})
-	if err != nil {
-		return FailedTasks, IncompleteTasks, err
-	}
+	err := retry.Retry(ctx, logger, func(_ *retry.Info) error {
+		cur, err := coll.Find(ctx, bson.D{
+			bson.E{Key: "type", Value: taskType},
+			bson.E{Key: "generation", Value: generation},
+		})
+		if err != nil {
+			return err
+		}
 
-	err = cur.All(ctx, &allTasks)
-	if err != nil {
-		return FailedTasks, IncompleteTasks, err
-	}
-	for _, t := range allTasks {
-		if failedStatus.Contains(t.Status) {
-			FailedTasks = append(FailedTasks, t)
-		} else if t.Status != verificationTaskCompleted {
-			IncompleteTasks = append(IncompleteTasks, t)
+		err = cur.All(ctx, &allTasks)
+		if err != nil {
+			return err
 		}
-	}
+		for _, t := range allTasks {
+			if failedStatus.Contains(t.Status) {
+				FailedTasks = append(FailedTasks, t)
+			} else if t.Status != verificationTaskCompleted {
+				IncompleteTasks = append(IncompleteTasks, t)
+			}
+		}
+
+		return nil
+	})
 
-	return FailedTasks, IncompleteTasks, nil
+	return FailedTasks, IncompleteTasks, err
 }
 
 // work is the logic for an individual worker thread.

internal/verifier/migration_verifier.go

@@ -1277,12 +1277,17 @@ func (verifier *Verifier) GetVerificationStatus(ctx context.Context) (*Verificat
 			},
 		},
 	}
-	cursor, err := taskCollection.Aggregate(ctx, aggregation)
-	if err != nil {
-		return nil, err
-	}
+
 	var results []bson.Raw
-	err = cursor.All(ctx, &results)
+
+	err := retry.Retry(ctx, verifier.logger, func(_ *retry.Info) error {
+		cursor, err := taskCollection.Aggregate(ctx, aggregation)
+		if err != nil {
+			return err
+		}
+
+		return cursor.All(ctx, &results)
+	})
 	if err != nil {
 		return nil, err
 	}

internal/verifier/reset_test.go

@@ -15,7 +15,7 @@ func (suite *IntegrationTestSuite) TestResetPrimaryTask() {
 
 	verifier := suite.BuildVerifier()
 
-	created, err := verifier.CheckIsPrimary(ctx)
+	created, err := verifier.CreatePrimaryTaskIfNeeded(ctx)
 	suite.Require().NoError(err)
 	suite.Require().True(created)
 
@@ -45,7 +45,7 @@ func (suite *IntegrationTestSuite) TestResetNonPrimaryTasks() {
 	verifier := suite.BuildVerifier()
 
 	// Create a primary task, and set it to complete.
-	created, err := verifier.CheckIsPrimary(ctx)
+	created, err := verifier.CreatePrimaryTaskIfNeeded(ctx)
 	suite.Require().NoError(err)
 	suite.Require().True(created)
 

internal/verifier/summary.go

@@ -29,8 +29,13 @@ const changeEventsTableMaxSize = 10
 func (verifier *Verifier) reportCollectionMetadataMismatches(ctx context.Context, strBuilder *strings.Builder) (bool, bool, error) {
 	generation, _ := verifier.getGeneration()
 
-	failedTasks, incompleteTasks, err :=
-		FetchFailedAndIncompleteTasks(ctx, verifier.verificationTaskCollection(), verificationTaskVerifyCollection, generation)
+	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
+		ctx,
+		verifier.logger,
+		verifier.verificationTaskCollection(),
+		verificationTaskVerifyCollection,
+		generation,
+	)
 	if err != nil {
 		return false, false, err
 	}
@@ -58,7 +63,13 @@ func (verifier *Verifier) reportCollectionMetadataMismatches(ctx context.Context
 func (verifier *Verifier) reportDocumentMismatches(ctx context.Context, strBuilder *strings.Builder) (bool, bool, error) {
 	generation, _ := verifier.getGeneration()
 
-	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(ctx, verifier.verificationTaskCollection(), verificationTaskVerifyDocuments, generation)
+	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
+		ctx,
+		verifier.logger,
+		verifier.verificationTaskCollection(),
+		verificationTaskVerifyDocuments,
+		generation,
+	)
 
 	if err != nil {
 		return false, false, err