more diag

Author: FGasper

internal/verifier/change_stream.go

@@ -63,7 +63,7 @@ func (verifier *Verifier) HandleChangeStreamEvent(ctx context.Context, changeEve
 	case "replace":
 		fallthrough
 	case "update":
-		if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
+		if err := verifier.eventRecorder.AddEvent(changeEvent); err != nil {
 			return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
 		}
 
@@ -171,12 +171,12 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 				verifier.logger.Fatal().
 					Err(err).
 					Stringer("timeout", timeout).
-					Msg("Failed to send change stream err within timeout.")
+					Msg("Failed to send change stream error within timeout.")
 			case verifier.changeStreamErrChan <- err:
 			}
 
 			if !changeStreamEnded {
-				return
+				break
 			}
 		}
 
@@ -191,8 +191,8 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 			// finished the change stream changes so that Recheck can continue.
 			verifier.changeStreamDoneChan <- struct{}{}
 			// since the changeStream is exhausted, we now return
-			verifier.logger.Debug().Msg("Change stream is done")
-			return
+			verifier.logger.Debug().Msg("Change stream is done.")
+			break
 		}
 	}
 }

internal/verifier/check.go

@@ -51,8 +51,10 @@ func (verifier *Verifier) waitForChangeStream() error {
 		verifier.changeStreamEnderChan <- struct{}{}
 		select {
 		case err := <-verifier.changeStreamErrChan:
+			verifier.logger.Warn().Err(err).Msg("Received error from change stream.")
 			return err
 		case <-verifier.changeStreamDoneChan:
+			verifier.logger.Debug().Msg("Received completion signal from change stream.")
 			break
 		}
 	}
@@ -182,16 +184,23 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	// Now enter the multi-generational steady check state
 	for {
 		verifier.generationStartTime = time.Now()
+		verifier.eventRecorder.Reset()
 
 		err := verifier.CheckWorker(ctx)
 		if err != nil {
 			return err
 		}
 		// we will only coordinate when the number of channels is exactly 2.
-		// * Channel 0 signals a generation is done
+		// * Channel 0 signals a generation boundary
 		// * Channel 1 signals to check to continue the next generation
 		if len(testChan) == 2 {
+
+			verifier.logger.Debug().
+				Msg("Telling test about generation boundary.")
 			testChan[0] <- struct{}{}
+
+			verifier.logger.Debug().
+				Msg("Awaiting test's signal to continue.")
 			<-testChan[1]
 		}
 		time.Sleep(verifier.generationPauseDelayMillis * time.Millisecond)

internal/verifier/migration_verifier.go

@@ -95,7 +95,7 @@ type Verifier struct {
 	numWorkers         int
 	failureDisplaySize int64
 
-	generationEventRecorder *EventRecorder
+	eventRecorder *EventRecorder
 
 	// Used only with generation 0 to defer the first
 	// progress report until after we’ve finished partitioning
@@ -202,7 +202,7 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 
 		// This will get recreated once gen0 starts, but we want it
 		// here in case the change streams gets an event before then.
-		generationEventRecorder: NewEventRecorder(),
+		eventRecorder: NewEventRecorder(),
 	}
 }
 

internal/verifier/migration_verifier_test.go

@@ -1461,7 +1461,6 @@ func (suite *MultiDataVersionTestSuite) TestGenerationalRechecking() {
 	suite.Require().Equal(VerificationStatus{TotalTasks: 1, FailedTasks: 1}, *status)
 }
 
-/*
 func (suite *MultiDataVersionTestSuite) TestVerifierWithFilter() {
 	zerolog.SetGlobalLevel(zerolog.DebugLevel)
 
@@ -1571,8 +1570,11 @@ func (suite *MultiDataVersionTestSuite) TestVerifierWithFilter() {
 
 	// Turn writes off.
 	verifier.WritesOff(ctx)
+
+	// Tell CheckDriver to do one more pass. This should terminate the change stream.
+	checkContinueChan <- struct{}{}
+	<-checkDoneChan
 }
-*/
 
 func (suite *MultiDataVersionTestSuite) TestPartitionWithFilter() {
 	zerolog.SetGlobalLevel(zerolog.DebugLevel)

internal/verifier/recheck.go

@@ -56,7 +56,7 @@ func (verifier *Verifier) InsertChangeEventRecheckDoc(ctx context.Context, chang
 	verifier.mux.Lock()
 	defer verifier.mux.Unlock()
 
-	if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
+	if err := verifier.eventRecorder.AddEvent(changeEvent); err != nil {
 		return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
 	}
 

internal/verifier/summary.go

@@ -369,7 +369,7 @@ func (verifier *Verifier) printMismatchInvestigationNotes(strBuilder *strings.Bu
 }
 
 func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder) {
-	nsStats := verifier.generationEventRecorder.Read()
+	nsStats := verifier.eventRecorder.Read()
 
 	activeNamespacesCount := len(nsStats)