update types etc.

Author: FGasper

.github/workflows/all.yml

@@ -60,8 +60,12 @@ jobs:
         with:
           go-version: stable
 
-      - name: Install m
-        run: npm install -g m mongosh
+      - name: Install m and mtools
+        run: |-
+            {
+                echo npm install -g m
+                echo pipx install 'mtools[all]'
+            } | parallel
 
       - name: Install MongoDB ${{ matrix.mongodb_versions[0] }} (source)
         run: yes | m ${{ matrix.mongodb_versions[0] }} && dirname $(readlink $(which mongod)) > .srcpath
@@ -72,18 +76,15 @@ jobs:
       - 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"
+                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

internal/verifier/change_stream.go

@@ -42,6 +42,7 @@ const (
 )
 
 type UnknownEventError struct {
+	Event    ParsedEvent
 	RawEvent bson.Raw
 }
 
@@ -96,7 +97,10 @@ func (verifier *Verifier) HandleChangeStreamEvents(ctx context.Context, batch []
 				dataSizes[i] = len(changeEvent.FullDocument)
 			}
 		default:
-			return UnknownEventError{RawEvent: rawChangeEvent}
+			return UnknownEventError{
+				Event:    changeEvent,
+				RawEvent: rawChangeEvent,
+			}
 		}
 	}
 
@@ -138,7 +142,7 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 	writesOffTs *primitive.Timestamp,
 ) error {
 	eventsRead := 0
-	var changeEventBatch []ParsedEvent
+	var changeEventBatch []bson.Raw
 
 	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
 		// Once the change stream reaches the writesOff timestamp we should stop reading.
@@ -163,12 +167,10 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 		}
 
 		if changeEventBatch == nil {
-			changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
+			changeEventBatch = make([]bson.Raw, cs.RemainingBatchLength()+1)
 		}
 
-		if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
-			return errors.Wrap(err, "failed to decode change event")
-		}
+		copy(changeEventBatch[eventsRead], cs.Current)
 
 		eventsRead++
 	}

internal/verifier/migration_verifier_test.go

@@ -145,28 +145,36 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Recheck() {
 
 	err := verifier.HandleChangeStreamEvents(
 		ctx,
-		[]ParsedEvent{{
-			OpType: "insert",
-			Ns:     &Namespace{DB: "mydb", Coll: "coll2"},
-			DocKey: DocKey{
-				ID: "heyhey",
-			},
-		}},
+		[]bson.Raw{
+			testutil.MustMarshal(
+				ParsedEvent{
+					OpType: "insert",
+					Ns:     &Namespace{DB: "mydb", Coll: "coll2"},
+					DocKey: DocKey{
+						ID: "heyhey",
+					},
+				},
+			),
+		},
 	)
 	suite.Require().NoError(err)
 
 	err = verifier.HandleChangeStreamEvents(
 		ctx,
-		[]ParsedEvent{{
-			ID: bson.M{
-				"docID": "ID/docID",
-			},
-			OpType: "insert",
-			Ns:     &Namespace{DB: "mydb", Coll: "coll1"},
-			DocKey: DocKey{
-				ID: "hoohoo",
-			},
-		}},
+		[]bson.Raw{
+			testutil.MustMarshal(
+				ParsedEvent{
+					ID: bson.M{
+						"docID": "ID/docID",
+					},
+					OpType: "insert",
+					Ns:     &Namespace{DB: "mydb", Coll: "coll1"},
+					DocKey: DocKey{
+						ID: "hoohoo",
+					},
+				},
+			),
+		},
 	)
 	suite.Require().NoError(err)
 
@@ -408,19 +416,19 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 		},
 	}
 
-	err = verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err = verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	suite.Require().NoError(err)
 	event.OpType = "insert"
-	err = verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err = verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	suite.Require().NoError(err)
 	event.OpType = "replace"
-	err = verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err = verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	suite.Require().NoError(err)
 	event.OpType = "update"
-	err = verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err = verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	suite.Require().NoError(err)
 	event.OpType = "flibbity"
-	err = verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err = verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	badEventErr := UnknownEventError{}
 	suite.Require().ErrorAs(err, &badEventErr)
 	suite.Assert().Equal("flibbity", badEventErr.Event.OpType)

internal/verifier/recheck_test.go

@@ -52,7 +52,7 @@ func (suite *IntegrationTestSuite) TestFailedCompareThenReplace() {
 		FullDocument: testutil.MustMarshal(bson.D{{"foo", 1}}),
 	}
 
-	err := verifier.HandleChangeStreamEvents(ctx, []ParsedEvent{event})
+	err := verifier.HandleChangeStreamEvents(ctx, []bson.Raw{testutil.MustMarshal(event)})
 	suite.Require().NoError(err)
 
 	recheckDocs = suite.fetchRecheckDocs(ctx, verifier)