change stream test

Author: FGasper

internal/testutil/testutil.go

@@ -1,8 +1,14 @@
 package testutil
 
 import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
 )
 
 // Marshal wraps `bsonMarshal` with a panic on failure.
@@ -34,3 +40,64 @@ func convertDocsToAnys(docs []bson.D) []any {
 
 	return anys
 }
+
+func KillApplicationChangeStreams(
+	ctx context.Context,
+	t *testing.T,
+	client *mongo.Client,
+	appName string,
+) error {
+	// Kill verifier’s change stream.
+	cursor, err := client.Database(
+		"admin",
+		options.Database().SetReadConcern(readconcern.Local()),
+	).Aggregate(
+		ctx,
+		mongo.Pipeline{
+			{
+				{"$currentOp", bson.D{
+					{"idleCursors", true},
+				}},
+			},
+			{
+				{"$match", bson.D{
+					{"clientMetadata.application.name", appName},
+					{"command.collection", "$cmd.aggregate"},
+					{"cursor.originatingCommand.pipeline.0.$_internalChangeStreamOplogMatch",
+						bson.D{{"$type", "object"}},
+					},
+				}},
+			},
+		},
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to list %#q's change streams", appName)
+	}
+
+	ops := []struct {
+		Opid any
+	}{}
+	err = cursor.All(ctx, &ops)
+	if err != nil {
+		return errors.Wrapf(err, "failed to decode %#q's change streams", appName)
+	}
+
+	for _, op := range ops {
+		t.Logf("Killing change stream op %+v", op.Opid)
+
+		err :=
+			client.Database("admin").RunCommand(
+				ctx,
+				bson.D{
+					{"killOp", 1},
+					{"op", op.Opid},
+				},
+			).Err()
+
+		if err != nil {
+			return errors.Wrapf(err, "failed to kill change stream with opId %#q", op.Opid)
+		}
+	}
+
+	return nil
+}

internal/verifier/change_stream_test.go

@@ -15,8 +15,6 @@ 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 (suite *IntegrationTestSuite) TestChangeStreamFilter_NoNamespaces() {
@@ -428,6 +426,52 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 	)
 }
 
+func (suite *IntegrationTestSuite) TestWritesOffCursorKilledResilience() {
+	ctx := suite.Context()
+
+	coll := suite.srcMongoClient.
+		Database(suite.DBNameForTest()).
+		Collection("mycoll")
+
+	suite.Require().NoError(
+		coll.Database().CreateCollection(
+			ctx,
+			coll.Name(),
+		),
+	)
+
+	suite.Require().NoError(
+		suite.dstMongoClient.
+			Database(coll.Database().Name()).
+			CreateCollection(
+				ctx,
+				coll.Name(),
+			),
+	)
+
+	for range 100 {
+		verifier := suite.BuildVerifier()
+
+		docs := lo.RepeatBy(1_000, func(_ int) bson.D { return bson.D{} })
+		_, err := coll.InsertMany(
+			ctx,
+			lo.ToAnySlice(docs),
+		)
+		suite.Require().NoError(err)
+
+		suite.Require().NoError(verifier.WritesOff(ctx))
+
+		suite.Require().NoError(
+			testutil.KillApplicationChangeStreams(
+				suite.Context(),
+				suite.T(),
+				suite.srcMongoClient,
+				clientAppName,
+			),
+		)
+	}
+}
+
 func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
 	ctx := suite.Context()
 
@@ -445,54 +489,16 @@ func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
 	// wait for generation 0 to end
 	suite.Require().NoError(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(
+		testutil.KillApplicationChangeStreams(
+			suite.Context(),
+			suite.T(),
+			suite.srcMongoClient,
+			clientAppName,
+		),
 	)
-	suite.Require().NoError(err)
-
-	var ops []bson.Raw
-	suite.Require().NoError(cursor.All(ctx, &ops))
-
-	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(),
-		)
-	}
 
-	_, err = coll.InsertOne(
+	_, err := coll.InsertOne(
 		ctx,
 		bson.D{{"_id", "after kill"}},
 	)

internal/verifier/migration_verifier.go

@@ -68,6 +68,8 @@ const (
 	okSymbol    = "\u2705" // white heavy check mark
 	infoSymbol  = "\u24d8" // circled Latin small letter I
 	notOkSymbol = "\u2757" // heavy exclamation mark symbol
+
+	clientAppName = "Migration Verifier"
 )
 
 type whichCluster string
@@ -221,7 +223,7 @@ func (verifier *Verifier) ConfigureReadConcern(setting ReadConcernSetting) {
 }
 
 func (verifier *Verifier) getClientOpts(uri string) *options.ClientOptions {
-	appName := "Migration Verifier"
+	appName := clientAppName
 	opts := &options.ClientOptions{
 		AppName: &appName,
 	}