REP-5324 Normalize error handling. (#59)

In certain cases migration-verifier would log on fatal errors rather than propagating them back up the call stack. This has led to failures to compare documents that were effectively ignored save for being noted in the log. This changeset fixes that so that such exceptional cases will halt the verification.

Broadly speaking, this changeset:
a) brings idiomatic Context propagation in places where context.Background() was used
b) converts Error() logs and “failed” task statuses to returned errors

As of this changeset, any task in “failed” indicates a mismatch rather than a failure to complete the check.

A separate changeset will introduce a retryer to FetchAndCompareDocuments.

Other, specific changes:
- CheckWorker() is rewritten to use an errgroup, which automatically cancels context if a thread fails.
- CheckDriver() now returns errors rather than Error() logging them.
- A few places’ direct checks for ctx.Done() are removed. (They’re superfluous.)
- TestGenerationalRechecking and TestPartitionWithFilter no longer use constant DB names.
- RunForUUIDErrorOnly() now takes a context. (NB: This method is currently unused.)
- Most uses of context.Background() in tests are now suite.Context().
- The unused RecheckTasks is removed from VerificationStatus.

Author: FGasper

internal/retry/retry.go

@@ -59,12 +59,12 @@ func (r *Retryer) RunForUUIDAndTransientErrors(
 //
 // RunForUUIDErrorOnly returns the collection's current name in all cases.
 func (r *Retryer) RunForUUIDErrorOnly(
-	logger *logger.Logger, expectedCollName string, f func(*Info, string) error,
+	ctx context.Context, logger *logger.Logger, expectedCollName string, f func(*Info, string) error,
 ) (string, error) {
 	// Since we're not actually sleeping when checking for UUID/name mismatch
 	// errors, we don't need to provide a real context to handle
 	// cancellations.
-	return r.runRetryLoop(context.Background(), logger, expectedCollName, f, false, true)
+	return r.runRetryLoop(ctx, logger, expectedCollName, f, false, true)
 }
 
 // RunForTransientErrorsOnly retries f() for transient errors only, and

internal/retry/retryer_test.go

@@ -26,7 +26,7 @@ func (suite *UnitTestSuite) TestRetryer() {
 		suite.NoError(err)
 		suite.Equal(0, attemptNumber)
 
-		_, err = retryer.RunForUUIDErrorOnly(logger, "foo", f)
+		_, err = retryer.RunForUUIDErrorOnly(suite.Context(), logger, "foo", f)
 		suite.NoError(err)
 		suite.Equal(0, attemptNumber)
 
@@ -92,7 +92,7 @@ func (suite *UnitTestSuite) TestRetryer() {
 			}
 			return nil
 		}
-		_, err := retryer.RunForUUIDErrorOnly(logger, "bar", f)
+		_, err := retryer.RunForUUIDErrorOnly(suite.Context(), logger, "bar", f)
 		suite.NoError(err)
 		suite.Equal(attemptLimit/2, attemptNumber)
 	})
@@ -109,7 +109,7 @@ func (suite *UnitTestSuite) TestRetryer() {
 				Raw:  bson.Raw(raw),
 			}
 		}
-		_, err := retryer.RunForUUIDErrorOnly(logger, "bar", f)
+		_, err := retryer.RunForUUIDErrorOnly(suite.Context(), logger, "bar", f)
 		suite.NoError(err)
 		// We only did one retry because the actual collection name matched the
 		// previous attempt.
@@ -130,7 +130,7 @@ func (suite *UnitTestSuite) TestRetryerDurationLimitIsZero() {
 		return cmdErr
 	}
 
-	_, err := retryer.RunForUUIDErrorOnly(suite.Logger(), "bar", f)
+	_, err := retryer.RunForUUIDErrorOnly(suite.Context(), suite.Logger(), "bar", f)
 	suite.Equal(cmdErr, err)
 	suite.Equal(0, attemptNumber)
 }
@@ -280,7 +280,7 @@ func (suite *UnitTestSuite) TestRetryerWithEmptyCollectionName() {
 		return nil
 	}
 
-	name, err := retryer.RunForUUIDErrorOnly(suite.Logger(), "", f)
+	name, err := retryer.RunForUUIDErrorOnly(suite.Context(), suite.Logger(), "", f)
 	suite.NoError(err)
 	suite.Equal("", name)
 }

internal/verifier/change_stream_test.go

@@ -55,7 +55,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 
 	func() {
 		verifier1 := suite.BuildVerifier()
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx, cancel := context.WithCancel(suite.Context())
 		defer cancel()
 		err := verifier1.StartChangeStream(ctx)
 		suite.Require().NoError(err)
@@ -147,8 +147,7 @@ func (suite *IntegrationTestSuite) fetchVerifierRechecks(ctx context.Context, ve
 
 func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 	verifier := suite.BuildVerifier()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := suite.Context()
 	sess, err := suite.srcMongoClient.StartSession()
 	suite.Require().NoError(err)
 	sctx := mongo.NewSessionContext(ctx, sess)
@@ -167,8 +166,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 
 func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 	verifier := suite.BuildVerifier()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := suite.Context()
 	sess, err := suite.srcMongoClient.StartSession()
 	suite.Require().NoError(err)
 	sctx := mongo.NewSessionContext(ctx, sess)
@@ -220,8 +218,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 
 func (suite *IntegrationTestSuite) TestNoStartAtTime() {
 	verifier := suite.BuildVerifier()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := suite.Context()
 	sess, err := suite.srcMongoClient.StartSession()
 	suite.Require().NoError(err)
 	sctx := mongo.NewSessionContext(ctx, sess)