Merge branch 'REP-5306-tests_to_integration' into REP-5299-writesoff-timestamp

Author: FGasper

.github/workflows/all.yml

@@ -11,21 +11,43 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os:
-          - runsOn: macos-latest
-            mongodb_distro: mongodb-macos-arm64
+        mongodb_versions:
+          - [ '4.2', '4.2' ]
+          - [ '4.2', '4.4' ]
+          - [ '4.2', '5.0' ]
+          - [ '4.2', '6.0' ]
 
-          - runsOn: ubuntu-latest
-            mongodb_distro: mongodb-linux-x86_64-ubuntu1804
+          - [ '4.4', '4.4' ]
+          - [ '4.4', '5.0' ]
+          - [ '4.4', '6.0' ]
 
-        go_version:
+          - [ '5.0', '5.0' ]
+          - [ '5.0', '6.0' ]
+          - [ '5.0', '7.0' ]
 
-          # This is hard-coded by design in order to catch inadvertent changes
-          # to the minimum-required Go version to build migration-verifier.
-          - '1.20'
-          - stable
+          - [ '6.0', '6.0' ]
+          - [ '6.0', '7.0' ]
+          - [ '6.0', '8.0' ]
 
-    runs-on: ${{matrix.os.runsOn}}
+          - [ '7.0', '7.0' ]
+          - [ '7.0', '8.0' ]
+
+          - [ '8.0', '8.0' ]
+
+        topology:
+          - name: replset
+            srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
+            dstConnStr: mongodb://localhost:27030,localhost:27031,localhost:27032
+
+          - name: sharded
+            args: --sharded 2
+            srcConnStr: mongodb://localhost:27020
+            dstConnStr: mongodb://localhost:27030
+
+    # There seems no good reason to test on other OSes … ?
+    runs-on: ubuntu-latest
+
+    name: ${{ matrix.mongodb_versions[0] }} to ${{ matrix.mongodb_versions[1] }}, ${{ matrix.topology.name }}
 
     steps:
       - name: Check out repository
@@ -34,12 +56,37 @@ jobs:
       - name: Fetch Go ${{ matrix.go_version }}
         uses: actions/setup-go@v5
         with:
-          go-version: ${{ matrix.go_version }}
+          go-version: stable
+
+      - name: Install m
+        run: npm install -g m mongosh
+
+      - name: Install MongoDB ${{ matrix.mongodb_versions[0] }} (source)
+        run: yes | m ${{ matrix.mongodb_versions[0] }} && dirname $(readlink $(which mongod)) > .srcpath
+
+      - name: Install MongoDB ${{ matrix.mongodb_versions[1] }} (destination)
+        run: yes | m ${{ matrix.mongodb_versions[1] }} && dirname $(readlink $(which mongod)) > .dstpath
+
+      - name: Install latest stable MongoDB (metadata)
+        run: yes | m stable && dirname $(readlink $(which mongod)) > .metapath
+
+      - name: Install mtools
+        run: pipx install 'mtools[all]'
 
       - name: Build
         run: go build main/migration_verifier.go
 
+      - name: Start clusters
+        run: |-
+            {
+                echo "mlaunch init --binarypath $(cat .srcpath) --port 27020 --dir src --replicaset ${{ matrix.topology.args }}"
+                echo "mlaunch init --binarypath $(cat .dstpath) --port 27030 --dir dst --replicaset ${{ matrix.topology.args }}"
+                echo "mlaunch init --binarypath $(cat .metapath) --port 27040 --dir meta --replicaset --nodes 1"
+            } | parallel
+
       - name: Test
-        run: go test -v ./...
+        run: go test -v ./... -race
         env:
-          MONGODB_DISTRO: ${{matrix.os.mongodb_distro}}
+          MVTEST_SRC: ${{matrix.topology.srcConnStr}}
+          MVTEST_DST: ${{matrix.topology.dstConnStr}}
+          MVTEST_META: mongodb://localhost:27040

README.md

@@ -147,6 +147,17 @@ On default settings it used about **200GB of RAM on m6id.metal machine when usin
 
 **This means it does about 1TB/20min but it is HIGHLY dependent on the source and dest machines**
 
+# Tests
+
+This project’s tests run as normal Go tests, to, with `go test`.
+
+`IntegrationTestSuite`'s tests require external clusters. You must provision these yourself.
+(See the project’s GitHub CI setup for one way to simplify it.) Once provisioned, set the relevant
+connection strings in the following environment variables:
+
+- MVTEST_SRC
+- MVTEST_DST
+- MVTEST_META
 
 # How the Verifier Works
 

internal/verifier/change_stream.go

@@ -122,6 +122,8 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 }
 
 func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
+	defer cs.Close(ctx)
+
 	var lastPersistedTime time.Time
 
 	persistResumeTokenIfNeeded := func() error {
@@ -188,6 +190,9 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 
 		// If the context is canceled, return immmediately.
 		case <-ctx.Done():
+			verifier.logger.Debug().
+				Err(ctx.Err()).
+				Msg("Change stream quitting.")
 			return
 
 		// If the changeStreamEnderChan has a message, the user has indicated that

internal/verifier/change_stream_test.go

@@ -33,9 +33,9 @@ func TestChangeStreamFilter(t *testing.T) {
 // TestChangeStreamResumability creates a verifier, starts its change stream,
 // terminates that verifier, updates the source cluster, starts a new
 // verifier with change stream, and confirms that things look as they should.
-func (suite *MultiSourceVersionTestSuite) TestChangeStreamResumability() {
+func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 	func() {
-		verifier1 := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+		verifier1 := suite.BuildVerifier()
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 		err := verifier1.StartChangeStream(ctx)
@@ -46,15 +46,15 @@ func (suite *MultiSourceVersionTestSuite) TestChangeStreamResumability() {
 	defer cancel()
 
 	_, err := suite.srcMongoClient.
-		Database("testDb").
+		Database(suite.DBNameForTest()).
 		Collection("testColl").
 		InsertOne(
 			ctx,
 			bson.D{{"_id", "heyhey"}},
 		)
 	suite.Require().NoError(err)
 
-	verifier2 := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+	verifier2 := suite.BuildVerifier()
 
 	suite.Require().Empty(
 		suite.fetchVerifierRechecks(ctx, verifier2),
@@ -89,7 +89,7 @@ func (suite *MultiSourceVersionTestSuite) TestChangeStreamResumability() {
 
 	suite.Assert().Equal(
 		bson.M{
-			"db":         "testDb",
+			"db":         suite.DBNameForTest(),
 			"coll":       "testColl",
 			"generation": int32(0),
 			"docID":      "heyhey",
@@ -99,7 +99,7 @@ func (suite *MultiSourceVersionTestSuite) TestChangeStreamResumability() {
 	)
 }
 
-func (suite *MultiSourceVersionTestSuite) getClusterTime(ctx context.Context, client *mongo.Client) primitive.Timestamp {
+func (suite *IntegrationTestSuite) getClusterTime(ctx context.Context, client *mongo.Client) primitive.Timestamp {
 	sess, err := client.StartSession()
 	suite.Require().NoError(err, "should start session")
 
@@ -112,7 +112,7 @@ func (suite *MultiSourceVersionTestSuite) getClusterTime(ctx context.Context, cl
 	return newTime
 }
 
-func (suite *MultiSourceVersionTestSuite) fetchVerifierRechecks(ctx context.Context, verifier *Verifier) []bson.M {
+func (suite *IntegrationTestSuite) fetchVerifierRechecks(ctx context.Context, verifier *Verifier) []bson.M {
 	recheckDocs := []bson.M{}
 
 	recheckColl := verifier.verificationDatabase().Collection(recheckQueue)
@@ -126,8 +126,8 @@ func (suite *MultiSourceVersionTestSuite) fetchVerifierRechecks(ctx context.Cont
 	return recheckDocs
 }
 
-func (suite *MultiSourceVersionTestSuite) TestStartAtTimeNoChanges() {
-	verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
+	verifier := suite.BuildVerifier()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	sess, err := suite.srcMongoClient.StartSession()
@@ -146,8 +146,12 @@ func (suite *MultiSourceVersionTestSuite) TestStartAtTimeNoChanges() {
 	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
 }
 
-func (suite *MultiSourceVersionTestSuite) TestStartAtTimeWithChanges() {
-	verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
+	if suite.GetSrcTopology() == TopologySharded {
+		suite.T().Skip("Skipping pending REP-5299.")
+	}
+
+	verifier := suite.BuildVerifier()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	sess, err := suite.srcMongoClient.StartSession()
@@ -156,11 +160,13 @@ func (suite *MultiSourceVersionTestSuite) TestStartAtTimeWithChanges() {
 	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne(
 		sctx, bson.D{{"_id", 0}})
 	suite.Require().NoError(err)
-	origStartTs := sess.OperationTime()
-	suite.Require().NotNil(origStartTs)
+
+	origSessionTime := sess.OperationTime()
+	suite.Require().NotNil(origSessionTime)
 	err = verifier.StartChangeStream(ctx)
 	suite.Require().NoError(err)
-	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
+	suite.Require().Equal(verifier.srcStartAtTs, origSessionTime)
+
 	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne(
 		sctx, bson.D{{"_id", 1}})
 	suite.Require().NoError(err)
@@ -173,16 +179,26 @@ func (suite *MultiSourceVersionTestSuite) TestStartAtTimeWithChanges() {
 	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").DeleteOne(
 		sctx, bson.D{{"_id", 1}})
 	suite.Require().NoError(err)
-	newStartTs := sess.OperationTime()
-	suite.Require().NotNil(newStartTs)
-	suite.Require().Negative(origStartTs.Compare(*newStartTs))
+
+	postEventsSessionTime := sess.OperationTime()
+	suite.Require().NotNil(postEventsSessionTime)
+	suite.Require().Negative(
+		origSessionTime.Compare(*postEventsSessionTime),
+		"session time after events should exceed the original",
+	)
+
 	verifier.changeStreamEnderChan <- struct{}{}
 	<-verifier.changeStreamDoneChan
-	suite.Require().Equal(verifier.srcStartAtTs, newStartTs)
+
+	suite.Assert().GreaterOrEqual(
+		verifier.srcStartAtTs.Compare(*postEventsSessionTime),
+		0,
+		"verifier.srcStartAtTs should now meet or exceed our session timestamp",
+	)
 }
 
-func (suite *MultiSourceVersionTestSuite) TestNoStartAtTime() {
-	verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+func (suite *IntegrationTestSuite) TestNoStartAtTime() {
+	verifier := suite.BuildVerifier()
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	sess, err := suite.srcMongoClient.StartSession()
@@ -199,8 +215,9 @@ func (suite *MultiSourceVersionTestSuite) TestNoStartAtTime() {
 	suite.Require().LessOrEqual(origStartTs.Compare(*verifier.srcStartAtTs), 0)
 }
 
-func (suite *MultiSourceVersionTestSuite) TestWithChangeEventsBatching() {
-	verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
+	verifier := suite.BuildVerifier()
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()