remove test

Author: FGasper

internal/verifier/change_stream.go

@@ -213,7 +213,7 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		// If the changeStreamEnderChan has a message, the user has indicated that
 		// source writes are ended. This means we should exit rather than continue
 		// reading the change stream since there should be no more events.
-		case finalTs := <-verifier.changeStreamFinalTsChan:
+		case finalTs := <-verifier.changeStreamWritesOffTsChan:
 			verifier.logger.Debug().
 				Interface("finalTimestamp", finalTs).
 				Msg("Change stream thread received final timestamp. Finalizing change stream.")

internal/verifier/change_stream_test.go

@@ -142,7 +142,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 	err = verifier.StartChangeStream(ctx)
 	suite.Require().NoError(err)
 	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
-	verifier.changeStreamFinalTsChan <- *origStartTs
+	verifier.changeStreamWritesOffTsChan <- *origStartTs
 	<-verifier.changeStreamDoneChan
 	suite.Require().Equal(verifier.srcStartAtTs, origStartTs)
 }
@@ -184,7 +184,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 		"session time after events should exceed the original",
 	)
 
-	verifier.changeStreamFinalTsChan <- *postEventsSessionTime
+	verifier.changeStreamWritesOffTsChan <- *postEventsSessionTime
 	<-verifier.changeStreamDoneChan
 
 	suite.Assert().Equal(

internal/verifier/check.go

@@ -25,8 +25,6 @@ var failedStatus = mapset.NewSet(
 	verificationTaskMetadataMismatch,
 )
 
-var verificationStatusCheckInterval time.Duration = 15 * time.Second
-
 // Check is the asynchronous entry point to Check, should only be called by the web server. Use
 // CheckDriver directly for synchronous run.
 // testChan is a pair of channels for coordinating generations in tests.
@@ -110,7 +108,7 @@ func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 		//wait for task to be created, if none of the tasks found.
 		if verificationStatus.AddedTasks > 0 || verificationStatus.ProcessingTasks > 0 || verificationStatus.RecheckTasks > 0 {
 			waitForTaskCreation++
-			time.Sleep(verificationStatusCheckInterval)
+			time.Sleep(verifier.verificationStatusCheckInterval)
 		} else {
 			verifier.PrintVerificationSummary(ctx, GenerationComplete)
 			verifier.logger.Debug().Msg("Verification tasks complete")

internal/verifier/integration_test_suite.go

@@ -3,6 +3,7 @@ package verifier
 import (
 	"context"
 	"strings"
+	"time"
 
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/pkg/errors"
@@ -166,6 +167,8 @@ func (suite *IntegrationTestSuite) BuildVerifier() *Verifier {
 	verifier.SetGenerationPauseDelayMillis(0)
 	verifier.SetWorkerSleepDelayMillis(0)
 
+	verifier.verificationStatusCheckInterval = 10 * time.Millisecond
+
 	ctx := suite.Context()
 
 	suite.Require().NoError(

internal/verifier/migration_verifier.go

@@ -121,13 +121,13 @@ type Verifier struct {
 	metaDBName    string
 	srcStartAtTs  *primitive.Timestamp
 
-	mux                     sync.RWMutex
-	changeStreamRunning     bool
-	changeStreamFinalTsChan chan primitive.Timestamp
-	changeStreamErrChan     chan error
-	changeStreamDoneChan    chan struct{}
-	lastChangeEventTime     *primitive.Timestamp
-	writesOffTimestamp      *primitive.Timestamp
+	mux                         sync.RWMutex
+	changeStreamRunning         bool
+	changeStreamWritesOffTsChan chan primitive.Timestamp
+	changeStreamErrChan         chan error
+	changeStreamDoneChan        chan struct{}
+	lastChangeEventTime         *primitive.Timestamp
+	writesOffTimestamp          *primitive.Timestamp
 
 	readConcernSetting ReadConcernSetting
 
@@ -137,6 +137,8 @@ type Verifier struct {
 	globalFilter map[string]any
 
 	pprofInterval time.Duration
+
+	verificationStatusCheckInterval time.Duration
 }
 
 // VerificationStatus holds the Verification Status
@@ -187,19 +189,21 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 	}
 
 	return &Verifier{
-		phase:                   Idle,
-		numWorkers:              NumWorkers,
-		readPreference:          readpref.Primary(),
-		partitionSizeInBytes:    400 * 1024 * 1024,
-		failureDisplaySize:      DefaultFailureDisplaySize,
-		changeStreamFinalTsChan: make(chan primitive.Timestamp),
-		changeStreamErrChan:     make(chan error),
-		changeStreamDoneChan:    make(chan struct{}),
-		readConcernSetting:      readConcern,
+		phase:                       Idle,
+		numWorkers:                  NumWorkers,
+		readPreference:              readpref.Primary(),
+		partitionSizeInBytes:        400 * 1024 * 1024,
+		failureDisplaySize:          DefaultFailureDisplaySize,
+		changeStreamWritesOffTsChan: make(chan primitive.Timestamp),
+		changeStreamErrChan:         make(chan error),
+		changeStreamDoneChan:        make(chan struct{}),
+		readConcernSetting:          readConcern,
 
 		// This will get recreated once gen0 starts, but we want it
 		// here in case the change streams gets an event before then.
 		eventRecorder: NewEventRecorder(),
+
+		verificationStatusCheckInterval: 15 * time.Second,
 	}
 }
 
@@ -250,7 +254,7 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 
 		verifier.writesOffTimestamp = &finalTs
 
-		verifier.changeStreamFinalTsChan <- finalTs
+		verifier.changeStreamWritesOffTsChan <- finalTs
 	}
 
 	return nil

internal/verifier/migration_verifier_test.go

@@ -1378,26 +1378,7 @@ func (suite *IntegrationTestSuite) TestGenerationalRechecking() {
 	status = waitForTasks()
 
 	// there should be no failures now, since they are are equivalent at this point in time
-	suite.Require().Equal(VerificationStatus{TotalTasks: 1, CompletedTasks: 1}, *status)
-
-	// turn writes off
-	suite.Require().NoError(verifier.WritesOff(ctx))
-
-	// now write to the source, this should not be seen by the change stream which should have ended
-	// because of the calls to WritesOff
-	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 1019, "x": 1019})
-	suite.Require().NoError(err)
-	checkContinueChan <- struct{}{}
-	<-checkDoneChan
-
-	status, err = verifier.GetVerificationStatus()
-	suite.Require().NoError(err)
-
-	// there should be a no more tasks
-	suite.Assert().Equal(VerificationStatus{}, *status)
-
-	checkContinueChan <- struct{}{}
-	require.NoError(suite.T(), errGroup.Wait())
+	suite.Assert().Equal(VerificationStatus{TotalTasks: 1, CompletedTasks: 1}, *status)
 }
 
 func (suite *IntegrationTestSuite) TestVerifierWithFilter() {