REP-5317 Add destination change stream

internal/verifier/change_stream_test.go

@@ -23,15 +23,16 @@ import (
 func TestChangeStreamFilter(t *testing.T) {
 	verifier := Verifier{}
 	verifier.SetMetaDBName("metadb")
+	verifier.initializeChangeStreamReaders()
 	assert.Contains(t,
-		verifier.GetChangeStreamFilter(),
+		verifier.srcChangeStreamReader.GetChangeStreamFilter(),
 		bson.D{
 			{"$match", bson.D{{"ns.db", bson.D{{"$ne", "metadb"}}}}},
 		},
 	)
-	verifier.srcNamespaces = []string{"foo.bar", "foo.baz", "test.car", "test.chaz"}
+	verifier.srcChangeStreamReader.namespaces = []string{"foo.bar", "foo.baz", "test.car", "test.chaz"}
 	assert.Contains(t,
-		verifier.GetChangeStreamFilter(),
+		verifier.srcChangeStreamReader.GetChangeStreamFilter(),
 		bson.D{{"$match", bson.D{
 			{"$or", []bson.D{
 				{{"ns", bson.D{{"db", "foo"}, {"coll", "bar"}}}},
@@ -57,7 +58,8 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 		verifier1 := suite.BuildVerifier()
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
-		err := verifier1.StartChangeStream(ctx)
+		verifier1.StartChangeEventHandler(ctx, verifier1.srcChangeStreamReader)
+		err := verifier1.srcChangeStreamReader.StartChangeStream(ctx)
 		suite.Require().NoError(err)
 	}()
 
@@ -82,13 +84,14 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 
 	newTime := suite.getClusterTime(ctx, suite.srcMongoClient)
 
-	err = verifier2.StartChangeStream(ctx)
+	verifier2.StartChangeEventHandler(ctx, verifier2.srcChangeStreamReader)
+	err = verifier2.srcChangeStreamReader.StartChangeStream(ctx)
 	suite.Require().NoError(err)
 
-	suite.Require().NotNil(verifier2.srcStartAtTs)
+	suite.Require().NotNil(verifier2.srcChangeStreamReader.startAtTs)
 
 	suite.Assert().False(
-		verifier2.srcStartAtTs.After(newTime),
+		verifier2.srcChangeStreamReader.startAtTs.After(newTime),
 		"verifier2's change stream should be no later than this new session",
 	)
 
@@ -157,12 +160,13 @@ func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 	suite.Require().NoError(err)
 	origStartTs := sess.OperationTime()
 	suite.Require().NotNil(origStartTs)
-	err = verifier.StartChangeStream(ctx)
+	verifier.StartChangeEventHandler(ctx, verifier.srcChangeStreamReader)
+	err = verifier.srcChangeStreamReader.StartChangeStream(ctx)
 	suite.Require().NoError(err)
-	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
-	verifier.changeStreamWritesOffTsChan <- *origStartTs
-	<-verifier.changeStreamDoneChan
-	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
+	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
+	verifier.srcChangeStreamReader.ChangeStreamWritesOffTsChan <- *origStartTs
+	<-verifier.srcChangeStreamReader.ChangeStreamDoneChan
+	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
 }
 
 func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
@@ -178,13 +182,14 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 
 	origSessionTime := sess.OperationTime()
 	suite.Require().NotNil(origSessionTime)
-	err = verifier.StartChangeStream(ctx)
+	verifier.StartChangeEventHandler(ctx, verifier.srcChangeStreamReader)
+	err = verifier.srcChangeStreamReader.StartChangeStream(ctx)
 	suite.Require().NoError(err)
 
 	// srcStartAtTs derives from the change stream’s resume token, which can
 	// postdate our session time but should not precede it.
 	suite.Require().False(
-		verifier.srcStartAtTs.Before(*origSessionTime),
+		verifier.srcChangeStreamReader.startAtTs.Before(*origSessionTime),
 		"srcStartAtTs should be >= the insert’s optime",
 	)
 
@@ -208,12 +213,12 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 		"session time after events should exceed the original",
 	)
 
-	verifier.changeStreamWritesOffTsChan <- *postEventsSessionTime
-	<-verifier.changeStreamDoneChan
+	verifier.srcChangeStreamReader.ChangeStreamWritesOffTsChan <- *postEventsSessionTime
+	<-verifier.srcChangeStreamReader.ChangeStreamDoneChan
 
 	suite.Assert().Equal(
 		*postEventsSessionTime,
-		*verifier.srcStartAtTs,
+		*verifier.srcChangeStreamReader.startAtTs,
 		"verifier.srcStartAtTs should now be our session timestamp",
 	)
 }
@@ -230,10 +235,11 @@ func (suite *IntegrationTestSuite) TestNoStartAtTime() {
 	suite.Require().NoError(err)
 	origStartTs := sess.OperationTime()
 	suite.Require().NotNil(origStartTs)
-	err = verifier.StartChangeStream(ctx)
+	verifier.StartChangeEventHandler(ctx, verifier.srcChangeStreamReader)
+	err = verifier.srcChangeStreamReader.StartChangeStream(ctx)
 	suite.Require().NoError(err)
-	suite.Require().NotNil(verifier.srcStartAtTs)
-	suite.Require().LessOrEqual(origStartTs.Compare(*verifier.srcStartAtTs), 0)
+	suite.Require().NotNil(verifier.srcChangeStreamReader.startAtTs)
+	suite.Require().LessOrEqual(origStartTs.Compare(*verifier.srcChangeStreamReader.startAtTs), 0)
 }
 
 func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
@@ -249,7 +255,8 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 
 	verifier := suite.BuildVerifier()
 
-	suite.Require().NoError(verifier.StartChangeStream(ctx))
+	verifier.StartChangeEventHandler(ctx, verifier.srcChangeStreamReader)
+	suite.Require().NoError(verifier.srcChangeStreamReader.StartChangeStream(ctx))
 
 	_, err := coll1.InsertOne(ctx, bson.D{{"_id", 1}})
 	suite.Require().NoError(err)
@@ -270,7 +277,6 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 		500*time.Millisecond,
 		"the verifier should flush a recheck doc after a batch",
 	)
-
 }
 
 func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
@@ -454,6 +460,68 @@ func (suite *IntegrationTestSuite) TestCreateForbidden() {
 	suite.Assert().Equal("create", eventErr.Event.OpType)
 }
 
+func (suite *IntegrationTestSuite) TestRecheckDocsWithDstChangeEvents() {
+	ctx := suite.Context()
+
+	db := suite.dstMongoClient.Database("dstDB")
+	coll1 := db.Collection("dstColl1")
+	coll2 := db.Collection("dstColl2")
+
+	for _, coll := range mslices.Of(coll1, coll2) {
+		suite.Require().NoError(db.CreateCollection(ctx, coll.Name()))
+	}
+
+	verifier := suite.BuildVerifier()
+	verifier.SetSrcNamespaces([]string{"srcDB.srcColl1", "srcDB.srcColl2"})
+	verifier.SetDstNamespaces([]string{"dstDB.dstColl1", "dstDB.dstColl2"})
+	verifier.SetNamespaceMap()
+
+	verifier.StartChangeEventHandler(ctx, verifier.dstChangeStreamReader)
+	suite.Require().NoError(verifier.dstChangeStreamReader.StartChangeStream(ctx))
+
+	_, err := coll1.InsertOne(ctx, bson.D{{"_id", 1}})
+	suite.Require().NoError(err)
+	_, err = coll1.InsertOne(ctx, bson.D{{"_id", 2}})
+	suite.Require().NoError(err)
+
+	_, err = coll2.InsertOne(ctx, bson.D{{"_id", 1}})
+	suite.Require().NoError(err)
+
+	var rechecks []RecheckDoc
+	require.Eventually(
+		suite.T(),
+		func() bool {
+			recheckColl := verifier.verificationDatabase().Collection(recheckQueue)
+			cursor, err := recheckColl.Find(ctx, bson.D{})
+			if errors.Is(err, mongo.ErrNoDocuments) {
+				return false
+			}
+
+			suite.Require().NoError(err)
+			suite.Require().NoError(cursor.All(ctx, &rechecks))
+			return len(rechecks) == 3
+		},
+		time.Minute,
+		500*time.Millisecond,
+		"the verifier should flush a recheck doc after a batch",
+	)
+
+	coll1RecheckCount, coll2RecheckCount := 0, 0
+	for _, recheck := range rechecks {
+		suite.Require().Equal("srcDB", recheck.PrimaryKey.DatabaseName)
+		switch recheck.PrimaryKey.CollectionName {
+		case "srcColl1":
+			coll1RecheckCount++
+		case "srcColl2":
+			coll2RecheckCount++
+		default:
+			suite.T().Fatalf("unknown collection name: %v", recheck.PrimaryKey.CollectionName)
+		}
+	}
+	suite.Require().Equal(2, coll1RecheckCount)
+	suite.Require().Equal(1, coll2RecheckCount)
+}
+
 func (suite *IntegrationTestSuite) TestLargeEvents() {
 	ctx := suite.Context()
 

internal/verifier/check.go

@@ -40,17 +40,17 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	verifier.MaybeStartPeriodicHeapProfileCollection(ctx)
 }
 
-func (verifier *Verifier) waitForChangeStream(ctx context.Context) error {
+func (verifier *Verifier) waitForChangeStream(ctx context.Context, csr *ChangeStreamReader) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case err := <-verifier.changeStreamErrChan:
+	case err := <-csr.ChangeStreamErrChan:
 		verifier.logger.Warn().Err(err).
-			Msg("Received error from change stream.")
+			Msgf("Received error from %s.", csr)
 		return err
-	case <-verifier.changeStreamDoneChan:
+	case <-csr.ChangeStreamDoneChan:
 		verifier.logger.Debug().
-			Msg("Received completion signal from change stream.")
+			Msgf("Received completion signal from %s.", csr)
 		break
 	}
 
@@ -85,9 +85,12 @@ func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 
 	for {
 		select {
-		case err := <-verifier.changeStreamErrChan:
+		case err := <-verifier.srcChangeStreamReader.ChangeStreamErrChan:
 			cancel()
-			return errors.Wrap(err, "change stream failed")
+			return errors.Wrapf(err, "got an error from %s", verifier.srcChangeStreamReader)
+		case err := <-verifier.dstChangeStreamReader.ChangeStreamErrChan:
+			cancel()
+			return errors.Wrapf(err, "got an error from %s", verifier.dstChangeStreamReader)
 		case <-ctx.Done():
 			cancel()
 			return nil
@@ -130,6 +133,7 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	}
 	verifier.running = true
 	verifier.globalFilter = filter
+	verifier.initializeChangeStreamReaders()
 	verifier.mux.Unlock()
 	defer func() {
 		verifier.mux.Lock()
@@ -166,19 +170,30 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		verifier.phase = Idle
 	}()
 
-	verifier.mux.RLock()
-	csRunning := verifier.changeStreamRunning
-	verifier.mux.RUnlock()
-	if csRunning {
-		verifier.logger.Debug().Msg("Check: Change stream already running.")
+	if verifier.srcChangeStreamReader.changeStreamRunning {
+		verifier.logger.Debug().Msg("Check: Source change stream already running.")
 	} else {
-		verifier.logger.Debug().Msg("Change stream not running; starting change stream")
+		verifier.logger.Debug().Msg("Source change stream not running; starting change stream")
 
-		err = verifier.StartChangeStream(ctx)
+		err = verifier.srcChangeStreamReader.StartChangeStream(ctx)
 		if err != nil {
 			return errors.Wrap(err, "failed to start change stream on source")
 		}
+		verifier.StartChangeEventHandler(ctx, verifier.srcChangeStreamReader)
+	}
+
+	if verifier.dstChangeStreamReader.changeStreamRunning {
+		verifier.logger.Debug().Msg("Check: Destination change stream already running.")
+	} else {
+		verifier.logger.Debug().Msg("Destination change stream not running; starting change stream")
+
+		err = verifier.dstChangeStreamReader.StartChangeStream(ctx)
+		if err != nil {
+			return errors.Wrap(err, "failed to start change stream on destination")
+		}
+		verifier.StartChangeEventHandler(ctx, verifier.dstChangeStreamReader)
 	}
+
 	// Log out the verification status when initially booting up so it's easy to see the current state
 	verificationStatus, err := verifier.GetVerificationStatus()
 	if err != nil {
@@ -242,8 +257,10 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 			// It's necessary to wait for the change stream to finish before incrementing the
 			// generation number, or the last changes will not be checked.
 			verifier.mux.Unlock()
-			err := verifier.waitForChangeStream(ctx)
-			if err != nil {
+			if err = verifier.waitForChangeStream(ctx, verifier.srcChangeStreamReader); err != nil {
+				return err
+			}
+			if err = verifier.waitForChangeStream(ctx, verifier.dstChangeStreamReader); err != nil {
 				return err
 			}
 			verifier.mux.Lock()

internal/verifier/compare.go

@@ -267,7 +267,7 @@ func (verifier *Verifier) getFetcherChannels(
 			ctx,
 			verifier.srcClientCollection(task),
 			verifier.srcBuildInfo,
-			verifier.srcStartAtTs,
+			verifier.srcChangeStreamReader.startAtTs,
 			task,
 		)
 
@@ -291,7 +291,7 @@ func (verifier *Verifier) getFetcherChannels(
 			ctx,
 			verifier.dstClientCollection(task),
 			verifier.dstBuildInfo,
-			nil, //startAtTs
+			verifier.dstChangeStreamReader.startAtTs,
 			task,
 		)
 

internal/verifier/integration_test_suite.go

@@ -123,7 +123,6 @@ func (suite *IntegrationTestSuite) TearDownTest() {
 
 	suite.contextCanceller(errors.Errorf("tearing down test %#q", suite.T().Name()))
 	suite.testContext, suite.contextCanceller = nil, nil
-
 	ctx := context.Background()
 	for _, client := range []*mongo.Client{suite.srcMongoClient, suite.dstMongoClient} {
 		dbNames, err := client.ListDatabaseNames(ctx, bson.D{})
@@ -185,6 +184,7 @@ func (suite *IntegrationTestSuite) BuildVerifier() *Verifier {
 	)
 	verifier.SetLogger("stderr")
 	verifier.SetMetaDBName(metaDBName)
+	verifier.initializeChangeStreamReaders()
 
 	suite.Require().NoError(verifier.srcClientCollection(&task).Drop(ctx))
 	suite.Require().NoError(verifier.dstClientCollection(&task).Drop(ctx))