revert non-test changes to ensure that existing tests verify

Author: FGasper

internal/verifier/change_stream.go

@@ -12,6 +12,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"golang.org/x/exp/constraints"
 )
 
 const fauxDocSizeForDeleteEvents = 1024
@@ -104,11 +105,6 @@ func (verifier *Verifier) HandleChangeStreamEvents(ctx context.Context, batch []
 // GetChangeStreamFilter returns an aggregation pipeline that filters
 // namespaces as per configuration.
 //
-// Note that this omits verifier.globalFilter because we still need to
-// recheck any out-filter documents that may have changed in order to
-// account for filter traversals (i.e., updates that change whether a
-// document matches the filter).
-//
 // NB: Ideally we could make the change stream give $bsonSize(fullDocument)
 // and omit fullDocument, but $bsonSize was new in MongoDB 4.4, and we still
 // want to verify migrations from 4.2. fullDocument is unlikely to be a
@@ -126,60 +122,55 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 	return []bson.D{stage}
 }
 
-func (verifier *Verifier) readAndHandleOneChangeEventBatch(ctx context.Context, cs *mongo.ChangeStream) error {
-	eventsRead := 0
-	var changeEventBatch []ParsedEvent
-
-	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
-		gotEvent := cs.TryNext(ctx)
-
-		if cs.Err() != nil {
-			return errors.Wrap(cs.Err(), "change stream iteration failed")
-		}
+func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
+	defer cs.Close(ctx)
 
-		if !gotEvent {
-			break
-		}
+	var lastPersistedTime time.Time
 
-		if changeEventBatch == nil {
-			changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
+	persistResumeTokenIfNeeded := func() error {
+		if time.Since(lastPersistedTime) <= minChangeStreamPersistInterval {
+			return nil
 		}
 
-		if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
-			return errors.Wrap(err, "failed to decode change event")
+		err := verifier.persistChangeStreamResumeToken(ctx, cs)
+		if err == nil {
+			lastPersistedTime = time.Now()
 		}
 
-		eventsRead++
+		return err
 	}
 
-	if eventsRead == 0 {
-		return nil
-	}
+	readAndHandleOneChangeEventBatch := func() (bool, error) {
+		eventsRead := 0
+		var changeEventBatch []ParsedEvent
 
-	err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
-	if err != nil {
-		return errors.Wrap(err, "failed to handle change events")
-	}
+		for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
+			gotEvent := cs.TryNext(ctx)
 
-	return nil
-}
+			if !gotEvent || cs.Err() != nil {
+				break
+			}
 
-func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
-	defer cs.Close(ctx)
+			if changeEventBatch == nil {
+				changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
+			}
 
-	var lastPersistedTime time.Time
+			if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
+				return false, errors.Wrap(err, "failed to decode change event")
+			}
 
-	persistResumeTokenIfNeeded := func() error {
-		if time.Since(lastPersistedTime) <= minChangeStreamPersistInterval {
-			return nil
+			eventsRead++
 		}
 
-		err := verifier.persistChangeStreamResumeToken(ctx, cs)
-		if err == nil {
-			lastPersistedTime = time.Now()
+		if eventsRead > 0 {
+			verifier.logger.Debug().Int("eventsCount", eventsRead).Msgf("Received a batch of events.")
+			err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
+			if err != nil {
+				return false, errors.Wrap(err, "failed to handle change events")
+			}
 		}
 
-		return err
+		return eventsRead > 0, errors.Wrap(cs.Err(), "change stream iteration failed")
 	}
 
 	for {
@@ -198,45 +189,29 @@ 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 <-verifier.changeStreamEnderChan:
 			verifier.logger.Debug().
-				Interface("finalTimestamp", finalTs).
-				Msg("Change stream thread received final timestamp. Finalizing change stream.")
+				Msg("Change stream thread received shutdown request.")
 
 			changeStreamEnded = true
 
 			// Read all change events until the source reports no events.
 			// (i.e., the `getMore` call returns empty)
 			for {
-				var curTs primitive.Timestamp
-				curTs, err = extractTimestampFromResumeToken(cs.ResumeToken())
-				if err != nil {
-					err = errors.Wrap(err, "failed to extract timestamp from change stream's resume token")
-					break
-				}
-
-				if curTs == finalTs || curTs.After(finalTs) {
-					verifier.logger.Debug().
-						Interface("currentTimestamp", curTs).
-						Interface("finalTimestamp", finalTs).
-						Msg("Change stream has reached the final timestamp. Shutting down.")
+				var gotEvent bool
+				gotEvent, err = readAndHandleOneChangeEventBatch()
 
-					break
-				}
-
-				err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
-
-				if err != nil {
+				if !gotEvent || err != nil {
 					break
 				}
 			}
 
 		default:
-			err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
+			_, err = readAndHandleOneChangeEventBatch()
+		}
 
-			if err == nil {
-				err = persistResumeTokenIfNeeded()
-			}
+		if err == nil {
+			err = persistResumeTokenIfNeeded()
 		}
 
 		if err != nil && !errors.Is(err, context.Canceled) {
@@ -267,9 +242,9 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 
 	infoLog := verifier.logger.Info()
 	if verifier.lastChangeEventTime == nil {
-		infoLog = infoLog.Str("lastEventTime", "none")
+		infoLog = infoLog.Str("changeStreamStopTime", "none")
 	} else {
-		infoLog = infoLog.Interface("lastEventTime", *verifier.lastChangeEventTime)
+		infoLog = infoLog.Interface("changeStreamStopTime", *verifier.lastChangeEventTime)
 	}
 
 	infoLog.Msg("Change stream is done.")
@@ -295,7 +270,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 
 		ts, err := extractTimestampFromResumeToken(savedResumeToken)
 		if err == nil {
-			logEvent = addTimestampToLogEvent(ts, logEvent)
+			logEvent = addUnixTimeToLogEvent(ts.T, logEvent)
 		} else {
 			verifier.logger.Warn().
 				Err(err).
@@ -348,10 +323,8 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 	return nil
 }
 
-func addTimestampToLogEvent(ts primitive.Timestamp, event *zerolog.Event) *zerolog.Event {
-	return event.
-		Interface("timestamp", ts).
-		Time("time", time.Unix(int64(ts.T), int64(0)))
+func addUnixTimeToLogEvent[T constraints.Integer](unixTime T, event *zerolog.Event) *zerolog.Event {
+	return event.Time("timestampTime", time.Unix(int64(unixTime), int64(0)))
 }
 
 func (v *Verifier) getChangeStreamMetadataCollection() *mongo.Collection {
@@ -390,7 +363,7 @@ func (verifier *Verifier) persistChangeStreamResumeToken(ctx context.Context, cs
 		logEvent := verifier.logger.Debug()
 
 		if err == nil {
-			logEvent = addTimestampToLogEvent(ts, logEvent)
+			logEvent = addUnixTimeToLogEvent(ts.T, logEvent)
 		} else {
 			verifier.logger.Warn().Err(err).
 				Msg("failed to extract resume token timestamp")

internal/verifier/change_stream_test.go

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

internal/verifier/check.go

@@ -42,24 +42,13 @@ 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() error {
 	verifier.mux.RLock()
 	csRunning := verifier.changeStreamRunning
 	verifier.mux.RUnlock()
 	if csRunning {
 		verifier.logger.Debug().Msg("Changestream still running, signalling that writes are done and waiting for change stream to exit")
-
-		finalTs, err := GetNewClusterTime(
-			ctx,
-			verifier.logger,
-			verifier.srcClient,
-		)
-
-		if err != nil {
-			return errors.Wrapf(err, "failed to fetch source's cluster time")
-		}
-
-		verifier.changeStreamFinalTsChan <- finalTs
+		verifier.changeStreamEnderChan <- struct{}{}
 		select {
 		case err := <-verifier.changeStreamErrChan:
 			verifier.logger.Warn().Err(err).
@@ -77,7 +66,6 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context) error {
 func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 	verifier.logger.Debug().Msgf("Starting %d verification workers", verifier.numWorkers)
 	ctx, cancel := context.WithCancel(ctx)
-
 	wg := sync.WaitGroup{}
 	for i := 0; i < verifier.numWorkers; i++ {
 		wg.Add(1)
@@ -186,9 +174,7 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	verifier.mux.RLock()
 	csRunning := verifier.changeStreamRunning
 	verifier.mux.RUnlock()
-	if csRunning {
-		verifier.logger.Debug().Msg("Check: Change stream already running.")
-	} else {
+	if !csRunning {
 		verifier.logger.Debug().Msg("Change stream not running; starting change stream")
 
 		err = verifier.StartChangeStream(ctx)
@@ -256,7 +242,7 @@ 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)
+			err := verifier.waitForChangeStream()
 			if err != nil {
 				return err
 			}
@@ -402,15 +388,12 @@ func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.Wait
 			if errors.Is(err, mongo.ErrNoDocuments) {
 				duration := verifier.workerSleepDelayMillis * time.Millisecond
 
-				if duration > 0 {
-					verifier.logger.Debug().
-						Int("workerNum", workerNum).
-						Stringer("duration", duration).
-						Msg("No tasks found. Sleeping.")
-
-					time.Sleep(duration)
-				}
+				verifier.logger.Debug().
+					Int("workerNum", workerNum).
+					Stringer("duration", duration).
+					Msg("No tasks found. Sleeping.")
 
+				time.Sleep(duration)
 				continue
 			} else if err != nil {
 				panic(err)

internal/verifier/migration_verifier.go

@@ -24,6 +24,7 @@ import (
 	"github.com/10gen/migration-verifier/internal/uuidutil"
 	"github.com/olekukonko/tablewriter"
 	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -48,6 +49,7 @@ const (
 	SrcNamespaceField = "query_filter.namespace"
 	DstNamespaceField = "query_filter.to"
 	NumWorkers        = 10
+	refetch           = "TODO_CHANGE_ME_REFETCH"
 	Idle              = "idle"
 	Check             = "check"
 	Recheck           = "recheck"
@@ -121,12 +123,12 @@ 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
+	mux                   sync.RWMutex
+	changeStreamRunning   bool
+	changeStreamEnderChan chan struct{}
+	changeStreamErrChan   chan error
+	changeStreamDoneChan  chan struct{}
+	lastChangeEventTime   *primitive.Timestamp
 
 	readConcernSetting ReadConcernSetting
 
@@ -186,15 +188,15 @@ 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,
+		changeStreamEnderChan: make(chan struct{}),
+		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.
@@ -313,7 +315,15 @@ func (verifier *Verifier) SetPartitionSizeMB(partitionSizeMB uint32) {
 }
 
 func (verifier *Verifier) SetLogger(logPath string) {
-	verifier.logger, verifier.writer = getLoggerAndWriter(logPath)
+	writer := getLogWriter(logPath)
+	verifier.writer = writer
+
+	consoleWriter := zerolog.ConsoleWriter{
+		Out:        writer,
+		TimeFormat: timeFormat,
+	}
+	l := zerolog.New(consoleWriter).With().Timestamp().Logger()
+	verifier.logger = logger.NewLogger(&l, writer)
 }
 
 func (verifier *Verifier) SetSrcNamespaces(arg []string) {
@@ -1162,6 +1172,10 @@ func (verifier *Verifier) verificationTaskCollection() *mongo.Collection {
 	return verifier.verificationDatabase().Collection(verificationTasksCollection)
 }
 
+func (verifier *Verifier) refetchCollection() *mongo.Collection {
+	return verifier.verificationDatabase().Collection(refetch)
+}
+
 func (verifier *Verifier) srcClientDatabase(dbName string) *mongo.Database {
 	db := verifier.srcClient.Database(dbName)
 	// No need to check the write concern because we do not write to the source database.