Merge branch 'main' into REP-5317-add-dst-change-stream

Author: tdq45gj

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

@@ -56,7 +56,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 
 	func() {
 		verifier1 := suite.BuildVerifier()
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx, cancel := context.WithCancel(suite.Context())
 		defer cancel()
 		verifier1.StartChangeEventHandler(ctx, verifier1.srcChangeStreamReader)
 		err := verifier1.srcChangeStreamReader.StartChangeStream(ctx)
@@ -150,8 +150,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)
@@ -171,8 +170,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)
@@ -225,8 +223,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)