REP-5329 Improve retryer & apply it more consistently. (#62)

This changeset adds additional error codes to the retryer’s list of transient errors.

It also applies the retryer in several new places:
- metadata prep at the beginning of check
- generation of recheck tasks
- clearing of the recheck queue
- fetching failed & incomplete tasks
- find the next worker task
- get verification status
- inserting a partition verification task
- inserting a recheck task
- updating a verification task
- creating primary task
- setting the primary task to complete

A few functions are renamed for clarity as well, and some function calls are rewritten to read (hopefully) a bit more clearly.

Author: FGasper

internal/retry/retry.go

@@ -15,6 +15,17 @@ import (
 
 type RetryCallback = func(context.Context, *FuncInfo) error
 
+// Retry is a convenience that creates a retryer and executes it.
+// See RunForTransientErrorsOnly for argument details.
+func Retry(
+	ctx context.Context,
+	logger *logger.Logger,
+	callbacks ...RetryCallback,
+) error {
+	retryer := New(DefaultDurationLimit)
+	return retryer.Run(ctx, logger, callbacks...)
+}
+
 // Run() runs each given callback in parallel. If none of them fail,
 // then no error is returned.
 //

internal/verifier/change_stream_test.go

@@ -349,6 +349,7 @@ func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
 
 	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
 		ctx,
+		verifier.logger,
 		verifier.verificationTaskCollection(),
 		verificationTaskVerifyDocuments,
 		verifier.generation,
@@ -398,6 +399,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"
@@ -34,7 +36,10 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	go func() {
 		err := verifier.CheckDriver(ctx, filter)
 		if err != nil {
-			verifier.logger.Fatal().Err(err).Msgf("Fatal error in generation %d", verifier.generation)
+			verifier.logger.Fatal().
+				Int("generation", verifier.generation).
+				Err(err).
+				Msg("Fatal error.")
 		}
 	}()
 	verifier.MaybeStartPeriodicHeapProfileCollection(ctx)
@@ -185,19 +190,31 @@ 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(
+	err = retry.Retry(
 		ctx,
-		func(ctx context.Context, sCtx mongo.SessionContext) error {
-			return verifier.ResetInProgressTasks(sCtx)
+		verifier.logger,
+		func(ctx context.Context, _ *retry.FuncInfo) error {
+			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)
+				},
+			)
+			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")
@@ -304,13 +321,23 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		}
 		verifier.generation++
 		verifier.phase = Recheck
-		err = verifier.GenerateRecheckTasks(ctx)
+
+		// Generation of recheck tasks can partial-fail. The following will
+		// cause a full redo in that case, which is inefficient but simple.
+		// Such failures seem unlikely anyhow.
+		err = retry.Retry(
+			ctx,
+			verifier.logger,
+			func(ctx context.Context, _ *retry.FuncInfo) error {
+				return verifier.GenerateRecheckTasksWhileLocked(ctx)
+			},
+		)
 		if err != nil {
 			verifier.mux.Unlock()
 			return err
 		}
 
-		err = verifier.ClearRecheckDocs(ctx)
+		err = verifier.ClearRecheckDocsWhileLocked(ctx)
 		if err != nil {
 			verifier.logger.Warn().
 				Err(err).
@@ -367,7 +394,7 @@ func (verifier *Verifier) CreateInitialTasks(ctx context.Context) error {
 			)
 		}
 	}
-	isPrimary, err := verifier.CheckIsPrimary(ctx)
+	isPrimary, err := verifier.CreatePrimaryTaskIfNeeded(ctx)
 	if err != nil {
 		return err
 	}
@@ -401,30 +428,44 @@ 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(ctx context.Context, _ *retry.FuncInfo) 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 failedStatuses.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 failedStatuses.Contains(t.Status) {
+					FailedTasks = append(FailedTasks, t)
+				} else if t.Status != verificationTaskCompleted {
+					IncompleteTasks = append(IncompleteTasks, t)
+				}
+			}
 
-	return FailedTasks, IncompleteTasks, nil
+			return nil
+		},
+	)
+
+	return FailedTasks, IncompleteTasks, err
 }
 
 // work is the logic for an individual worker thread.
@@ -438,27 +479,30 @@ func (verifier *Verifier) work(ctx context.Context, workerNum int) error {
 		Msg("Worker finished.")
 
 	for {
-		task, err := verifier.FindNextVerifyTaskAndUpdate(ctx)
-		if errors.Is(err, mongo.ErrNoDocuments) {
+		taskOpt, err := verifier.FindNextVerifyTaskAndUpdate(ctx)
+		if err != nil {
+			return errors.Wrap(
+				err,
+				"failed to seek next task",
+			)
+		}
+
+		task, gotTask := taskOpt.Get()
+		if !gotTask {
 			duration := verifier.workerSleepDelayMillis * time.Millisecond
 
 			if duration > 0 {
 				time.Sleep(duration)
 			}
 
 			continue
-		} else if err != nil {
-			return errors.Wrap(
-				err,
-				"failed to seek next task",
-			)
 		}
 
-		verifier.workerTracker.Set(workerNum, *task)
+		verifier.workerTracker.Set(workerNum, task)
 
 		switch task.Type {
 		case verificationTaskVerifyCollection:
-			err := verifier.ProcessCollectionVerificationTask(ctx, workerNum, task)
+			err := verifier.ProcessCollectionVerificationTask(ctx, workerNum, &task)
 			verifier.workerTracker.Unset(workerNum)
 
 			if err != nil {
@@ -471,7 +515,7 @@ func (verifier *Verifier) work(ctx context.Context, workerNum int) error {
 				}
 			}
 		case verificationTaskVerifyDocuments:
-			err := verifier.ProcessVerifyTask(ctx, workerNum, task)
+			err := verifier.ProcessVerifyTask(ctx, workerNum, &task)
 			verifier.workerTracker.Unset(workerNum)
 
 			if err != nil {

internal/verifier/migration_verifier.go

@@ -1232,26 +1232,36 @@ func (verifier *Verifier) GetVerificationStatus(ctx context.Context) (*Verificat
 	taskCollection := verifier.verificationTaskCollection()
 	generation, _ := verifier.getGeneration()
 
-	aggregation := []bson.M{
-		{
-			"$match": bson.M{
-				"type":       bson.M{"$ne": "primary"},
-				"generation": generation,
-			},
-		},
-		{
-			"$group": bson.M{
-				"_id":   "$status",
-				"count": bson.M{"$sum": 1},
-			},
-		},
-	}
-	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(ctx context.Context, _ *retry.FuncInfo) error {
+			cursor, err := taskCollection.Aggregate(
+				ctx,
+				[]bson.M{
+					{
+						"$match": bson.M{
+							"type":       bson.M{"$ne": "primary"},
+							"generation": generation,
+						},
+					},
+					{
+						"$group": bson.M{
+							"_id":   "$status",
+							"count": bson.M{"$sum": 1},
+						},
+					},
+				},
+			)
+			if err != nil {
+				return err
+			}
+
+			return cursor.All(ctx, &results)
+		},
+	)
 	if err != nil {
 		return nil, err
 	}

internal/verifier/migration_verifier_test.go

@@ -179,7 +179,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Recheck() {
 	func() {
 		verifier.mux.Lock()
 		defer func() { verifier.mux.Unlock() }()
-		suite.Require().NoError(verifier.GenerateRecheckTasks(ctx))
+		suite.Require().NoError(verifier.GenerateRecheckTasksWhileLocked(ctx))
 	}()
 
 	stats, err := verifier.GetNamespaceStatistics(ctx)
@@ -436,7 +436,7 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 		verifier.mux.Lock()
 		defer verifier.mux.Unlock()
 
-		err = verifier.GenerateRecheckTasks(ctx)
+		err = verifier.GenerateRecheckTasksWhileLocked(ctx)
 		suite.Require().NoError(err)
 	}()