save

Author: FGasper

internal/verifier/change_stream.go

@@ -66,7 +66,7 @@ func (verifier *Verifier) HandleChangeStreamEvents(ctx context.Context, batch []
 	for i, changeEvent := range batch {
 		if changeEvent.ClusterTime != nil &&
 			(verifier.lastChangeEventTime == nil ||
-				verifier.lastChangeEventTime.Compare(*changeEvent.ClusterTime) < 0) {
+				verifier.lastChangeEventTime.Before(*changeEvent.ClusterTime)) {
 			verifier.lastChangeEventTime = changeEvent.ClusterTime
 		}
 		switch changeEvent.OpType {
@@ -223,11 +223,11 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 					return errors.Wrap(err, "failed to extract timestamp from change stream's resume token")
 				}
 
-				if curTs == writesOffTs || curTs.After(writesOffTs) {
+				if !curTs.Before(writesOffTs) {
 					verifier.logger.Debug().
 						Interface("currentTimestamp", curTs).
 						Interface("writesOffTimestamp", writesOffTs).
-						Msg("Change stream has reached the writesOff timestamp. Shutting down.")
+						Msg("Change stream has passed the writesOff timestamp. Shutting down.")
 
 					break
 				}

internal/verifier/change_stream_test.go

@@ -11,6 +11,8 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
 )
 
 func TestChangeStreamFilter(t *testing.T) {
@@ -241,6 +243,106 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 	)
 }
 
+func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
+	ctx := suite.Context()
+
+	verifier := suite.BuildVerifier()
+
+	db := suite.srcMongoClient.Database(suite.DBNameForTest())
+	coll := db.Collection("mycoll")
+	suite.Require().NoError(
+		db.CreateCollection(ctx, coll.Name()),
+	)
+
+	// start verifier
+	verifierRunner := RunVerifierCheck(suite.Context(), suite.T(), verifier)
+
+	// wait for generation 0 to end
+	verifierRunner.AwaitGenerationEnd()
+
+	const mvName = "Migration Verifier"
+
+	// Kill verifier’s change stream.
+	cursor, err := suite.srcMongoClient.Database(
+		"admin",
+		options.Database().SetReadConcern(readconcern.Local()),
+	).Aggregate(
+		ctx,
+		mongo.Pipeline{
+			{
+				{"$currentOp", bson.D{
+					{"idleCursors", true},
+				}},
+			},
+			{
+				{"$match", bson.D{
+					{"clientMetadata.application.name", mvName},
+					{"command.collection", "$cmd.aggregate"},
+					{"cursor.originatingCommand.pipeline.0.$_internalChangeStreamOplogMatch",
+						bson.D{{"$type", "object"}},
+					},
+				}},
+			},
+		},
+	)
+	suite.Require().NoError(err)
+
+	var ops []bson.Raw
+	suite.Require().NoError(cursor.All(ctx, &ops))
+
+	/*
+		_, err = coll.InsertOne(
+			ctx,
+			bson.D{{"_id", "before kill"}},
+		)
+		suite.Require().NoError(err)
+	*/
+
+	for _, cursorRaw := range ops {
+		opId, err := cursorRaw.LookupErr("opid")
+		suite.Require().NoError(err, "should get opid from op")
+
+		suite.T().Logf("Killing change stream op %+v", opId)
+
+		suite.Require().NoError(
+			suite.srcMongoClient.Database("admin").RunCommand(
+				ctx,
+				bson.D{
+					{"killOp", 1},
+					{"op", opId},
+				},
+			).Err(),
+		)
+	}
+
+	sess, err := suite.srcMongoClient.StartSession()
+	suite.Require().NoError(err)
+	_, err = coll.InsertOne(
+		mongo.NewSessionContext(ctx, sess),
+		bson.D{{"_id", "after kill"}},
+	)
+	suite.Require().NoError(err)
+
+	suite.T().Logf("cluster time after insert: %+v", sess.ClusterTime())
+
+	suite.Require().NoError(verifier.WritesOff(ctx))
+
+	suite.Require().NoError(verifierRunner.Await())
+
+	failedTasks, incompleteTasks, err := FetchFailedAndIncompleteTasks(
+		ctx,
+		verifier.verificationTaskCollection(),
+		verificationTaskVerifyDocuments,
+		verifier.generation,
+	)
+	suite.Require().NoError(err)
+
+	suite.Assert().Zero(incompleteTasks, "no incomplete tasks")
+	suite.Require().Len(failedTasks, 1, "expect one failed task")
+
+	suite.T().Logf("task: %+v", failedTasks[0])
+}
+
 func (suite *IntegrationTestSuite) TestEventBeforeWritesOff() {
 	ctx := suite.Context()
 

internal/verifier/clustertime.go

@@ -35,7 +35,7 @@ func GetNewClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			clusterTime, err = fetchClusterTime(ctx, client)
+			clusterTime, err = fetchNewClusterTime(ctx, client)
 			return err
 		},
 	)
@@ -67,27 +67,22 @@ func GetNewClusterTime(
 
 // Use this when we just need the correct cluster time without
 // actually changing any shards’ oplogs.
-func fetchClusterTime(
+func fetchNewClusterTime(
 	ctx context.Context,
 	client *mongo.Client,
 ) (primitive.Timestamp, error) {
 	cmd, rawResponse, err := runAppendOplogNote(
 		ctx,
 		client,
 		"expect StaleClusterTime error",
-		primitive.Timestamp{1, 0},
+		option.None[primitive.Timestamp](),
 	)
 
-	// We expect an error here; if we didn't get one then something is
-	// amiss on the server.
-	if err == nil {
-		return primitive.Timestamp{}, errors.Errorf("server request unexpectedly succeeded: %v", cmd)
-	}
-
-	if !util.IsStaleClusterTimeError(err) {
-		return primitive.Timestamp{}, errors.Wrap(
+	if err != nil {
+		return primitive.Timestamp{}, errors.Wrapf(
 			err,
-			"unexpected error (expected StaleClusterTime) from request",
+			"unexpected error (expected StaleClusterTime) from request (%v)",
+			cmd,
 		)
 	}
 
@@ -123,16 +118,19 @@ func runAppendOplogNote(
 	ctx context.Context,
 	client *mongo.Client,
 	note string,
-	maxClusterTime primitive.Timestamp,
+	maxClusterTimeOpt option.Option[primitive.Timestamp],
 ) (bson.D, bson.Raw, error) {
 	cmd := bson.D{
 		{"appendOplogNote", 1},
-		{"maxClusterTime", maxClusterTime},
 		{"data", bson.D{
 			{"migration-verifier", note},
 		}},
 	}
 
+	if maxClusterTime, has := maxClusterTimeOpt.Get(); has {
+		cmd = append(cmd, bson.E{"maxClusterTime", maxClusterTime})
+	}
+
 	resp := client.
 		Database(
 			"admin",