Revert "preliminary"

This reverts commit e7d8a4b.

Author: FGasper

internal/util/value.go

@@ -160,21 +160,15 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 			eventsRead++
 		}
 
-		if cs.Err() != nil {
-			return false, errors.Wrap(cs.Err(), "change stream iteration failed")
-		}
-
-		if eventsRead == 0 {
-			return false, nil
-		}
-
-		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")
+		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 true, nil
+		return eventsRead > 0, errors.Wrap(cs.Err(), "change stream iteration failed")
 	}
 
 	for {
@@ -190,31 +184,15 @@ 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.")
+				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 {
-				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.Compare(finalTs) >= 0 {
-					verifier.logger.Debug().
-						Interface("currentTimestamp", curTs).
-						Interface("finalTimestamp", finalTs).
-						Msg("Change stream has reached the final timestamp. Shutting down.")
-
-					break
-				}
-
 				var gotEvent bool
 				gotEvent, err = readAndHandleOneChangeEventBatch()
 

internal/verifier/change_stream.go

@@ -9,7 +9,6 @@ import (
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
 )
 
@@ -43,13 +42,13 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	verifier.MaybeStartPeriodicHeapProfileCollection(ctx)
 }
 
-func (verifier *Verifier) finalizeChangeStream(finalTs primitive.Timestamp) 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")
-		verifier.changeStreamFinalTsChan <- finalTs
+		verifier.changeStreamEnderChan <- struct{}{}
 		select {
 		case err := <-verifier.changeStreamErrChan:
 			verifier.logger.Warn().Err(err).
@@ -239,11 +238,11 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		// will result in an extra iteration. The odds of this are lower and the user should be
 		// paying attention. Also, this should not matter too much because any failures will be
 		// caught again on the next iteration.
-		if ts, writesOff := verifier.writesOffTs.Get(); writesOff {
+		if verifier.writesOff {
 			// 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.finalizeChangeStream(ts)
+			err := verifier.waitForChangeStream()
 			if err != nil {
 				return err
 			}

internal/verifier/check.go

@@ -54,13 +54,23 @@ func GetClusterTime(
 	// OPTIMIZATION: We now append an oplog entry--this time for real!--to
 	// cause any lagging shards to output their events.
 	//
-	// Since this is just an optimization, failures here are nonfatal,
-	// and we might as well forgo the retryer.
-	_, err = runAppendOplogNote(ctx, client, false)
+	// Since this is just an optimization, failures here are nonfatal.
+	err = retryer.RunForTransientErrorsOnly(
+		ctx,
+		logger,
+		func(_ *retry.Info) error {
+			var err error
+			optime, err = runAppendOplogNote(ctx, client, false)
+			if err != nil {
+				err = fmt.Errorf("%w: %w", retry.CustomTransientErr, err)
+			}
+			return err
+		},
+	)
 	if err != nil {
 		// This isn't serious enough even to warn on, so leave it at info-level.
 		logger.Info().Err(err).
-			Msg("Failed to update oplog; change stream may need extra time to finish.")
+			Msg("Failed to append oplog note; change stream may need extra time to finish.")
 	}
 
 	return optime, nil

internal/verifier/clustertime.go

@@ -21,9 +21,7 @@ import (
 	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/types"
-	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/10gen/migration-verifier/internal/uuidutil"
-	"github.com/10gen/migration-verifier/option"
 	"github.com/olekukonko/tablewriter"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -80,7 +78,7 @@ var timeFormat = time.RFC3339
 
 // Verifier is the main state for the migration verifier
 type Verifier struct {
-	writesOffTs        option.Option[primitive.Timestamp]
+	writesOff          bool
 	lastGeneration     bool
 	running            bool
 	generation         int
@@ -125,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
 
@@ -190,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.
@@ -230,30 +228,18 @@ func (verifier *Verifier) SetFailureDisplaySize(size int64) {
 	verifier.failureDisplaySize = size
 }
 
-func (verifier *Verifier) WritesOff(ctx context.Context) error {
+func (verifier *Verifier) WritesOff(ctx context.Context) {
 	verifier.logger.Debug().
 		Msg("WritesOff called.")
 
 	verifier.mux.Lock()
-	defer verifier.mux.Unlock()
-	writesOffTs, err := GetClusterTime(
-		ctx,
-		verifier.logger,
-		verifier.srcClient,
-	)
-
-	if err != nil {
-		return errors.Wrapf(err, "failed to fetch source's cluster time")
-	}
-
-	verifier.writesOffTs = option.Some(writesOffTs)
-
-	return nil
+	verifier.writesOff = true
+	verifier.mux.Unlock()
 }
 
 func (verifier *Verifier) WritesOn(ctx context.Context) {
 	verifier.mux.Lock()
-	util.ResetToZero(&verifier.writesOffTs)
+	verifier.writesOff = false
 	verifier.mux.Unlock()
 }
 

internal/verifier/migration_verifier.go

@@ -22,7 +22,7 @@ const RequestInProgressErrorDescription = "Another request is currently in progr
 // MigrationVerifierAPI represents the interaction webserver with mongosync
 type MigrationVerifierAPI interface {
 	Check(ctx context.Context, filter map[string]any)
-	WritesOff(ctx context.Context) error
+	WritesOff(ctx context.Context)
 	WritesOn(ctx context.Context)
 	GetProgress(ctx context.Context) (Progress, error)
 }

internal/verifier/web_server.go

@@ -30,8 +30,8 @@ func NewMockVerifier() *MockVerifier {
 func (verifier *MockVerifier) Check(ctx context.Context, filter map[string]any) {
 	verifier.filter = filter
 }
-func (verifier *MockVerifier) WritesOff(ctx context.Context) error { return nil }
-func (verifier *MockVerifier) WritesOn(ctx context.Context)        {}
+func (verifier *MockVerifier) WritesOff(ctx context.Context) {}
+func (verifier *MockVerifier) WritesOn(ctx context.Context)  {}
 func (verifier *MockVerifier) GetProgress(ctx context.Context) (Progress, error) {
 	return Progress{}, nil
 }