event reporter working

Author: FGasper

internal/verifier/check.go

@@ -157,6 +157,10 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		verifier.phase = Idle
 	}()
 
+	// A “throwaway” EventRecorder so that any change events that arrive
+	// right away will have somewhere to be recorded.
+	verifier.generationEventRecorder = NewEventRecorder()
+
 	verifier.mux.RLock()
 	csRunning := verifier.changeStreamRunning
 	verifier.mux.RUnlock()

internal/verifier/recheck.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/10gen/migration-verifier/internal/types"
+	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -34,7 +35,11 @@ type RecheckDoc struct {
 func (verifier *Verifier) InsertFailedCompareRecheckDocs(
 	namespace string, documentIDs []interface{}, dataSizes []int) error {
 	dbName, collName := SplitNamespace(namespace)
-	return verifier.insertRecheckDocs(context.Background(),
+
+	verifier.mux.Lock()
+	defer verifier.mux.Unlock()
+
+	return verifier.insertRecheckDocsUnderLock(context.Background(),
 		dbName, collName, documentIDs, dataSizes)
 }
 
@@ -48,15 +53,20 @@ func (verifier *Verifier) InsertChangeEventRecheckDoc(ctx context.Context, chang
 	// total data size for noninitial generations in the log.
 	dataSizes := []int{maxBSONObjSize}
 
-	return verifier.insertRecheckDocs(
+	verifier.mux.Lock()
+	defer verifier.mux.Unlock()
+
+	if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
+		return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
+	}
+
+	return verifier.insertRecheckDocsUnderLock(
 		ctx, changeEvent.Ns.DB, changeEvent.Ns.Coll, documentIDs, dataSizes)
 }
 
-func (verifier *Verifier) insertRecheckDocs(
+func (verifier *Verifier) insertRecheckDocsUnderLock(
 	ctx context.Context,
 	dbName, collName string, documentIDs []interface{}, dataSizes []int) error {
-	verifier.mux.Lock()
-	defer verifier.mux.Unlock()
 
 	generation, _ := verifier.getGenerationWhileLocked()
 

internal/verifier/recheck_test.go

@@ -95,7 +95,7 @@ func (suite *MultiMetaVersionTestSuite) TestLargeIDInsertions() {
 	id3 := strings.Repeat("c", overlyLarge)
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{overlyLarge, overlyLarge, overlyLarge}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{
@@ -156,7 +156,7 @@ func (suite *MultiMetaVersionTestSuite) TestLargeDataInsertions() {
 	id3 := "c"
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{400 * 1024, 700 * 1024, 1024}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 	d1 := RecheckDoc{
 		PrimaryKey: RecheckPrimaryKey{
@@ -217,13 +217,13 @@ func (suite *MultiMetaVersionTestSuite) TestMultipleNamespaces() {
 	id3 := "c"
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{1000, 1000, 1000}
-	err := verifier.insertRecheckDocs(ctx, "testDB1", "testColl1", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB1", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB1", "testColl2", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB1", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl1", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB2", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl2", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB2", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
@@ -267,17 +267,17 @@ func (suite *MultiMetaVersionTestSuite) TestGenerationalClear() {
 	id2 := "b"
 	ids := []interface{}{id1, id2}
 	dataSizes := []int{1000, 1000}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{

internal/verifier/summary.go

@@ -395,7 +395,7 @@ func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder) {
 	sort.Slice(
 		reverseSortedNamespaces,
 		func(i, j int) bool {
-			return reverseSortedNamespaces[i] > reverseSortedNamespaces[j]
+			return nsTotals[reverseSortedNamespaces[i]] > nsTotals[reverseSortedNamespaces[j]]
 		},
 	)