squashed

Author: FGasper

go.mod

@@ -1,6 +1,6 @@
 module github.com/10gen/migration-verifier
 
-go 1.20
+go 1.22
 
 require (
 	github.com/cespare/permute/v2 v2.0.0-beta2

internal/retry/retry.go

@@ -3,6 +3,7 @@ package retry
 import (
 	"context"
 	"math/rand"
+	"slices"
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/logger"
@@ -11,6 +12,8 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 )
 
+var CustomTransientErr = errors.New("possibly-transient failure detected")
+
 // RunForUUIDAndTransientErrors retries f() for the CollectionUUIDMismatch error and for transient errors.
 // This should be used to run a driver operation that optionally specifies the `collectionUUID` parameter
 // for a collection that may have been:
@@ -273,25 +276,42 @@ func (r *Retryer) shouldRetryWithSleep(
 		return false
 	}
 
+	var whyTransient string
+
 	errCode := util.GetErrorCode(err)
-	if util.IsTransientError(err) {
-		logger.Warn().Int("error code", errCode).Err(err).Msgf(
-			"Waiting %s seconds to retry operation after transient error.", sleepTime)
-		return true
+
+	if util.IsTransientError(err) || slices.Contains(r.additionalErrorCodes, errCode) {
+		whyTransient = "Error code suggests a transient error."
+	} else if errors.Is(err, CustomTransientErr) {
+		whyTransient = "Error may be transient."
 	}
 
-	for _, code := range r.additionalErrorCodes {
-		if code == errCode {
-			logger.Warn().Int("error code", errCode).Err(err).Msgf(
-				"Waiting %s seconds to retry operation after an error because it is in our additional codes list.", sleepTime)
-			return true
+	if whyTransient == "" {
+		event := logger.Debug().Err(err)
+
+		if errCode != 0 {
+			event = event.Int("error code", errCode)
 		}
+
+		event.Msg("Not retrying on error because it appears not to be transient.")
+
+		return false
+	}
+
+	retryLogEvent := logger.Warn().
+		Err(err).
+		Stringer("duration", sleepTime)
+
+	if errCode != 0 {
+		retryLogEvent = retryLogEvent.
+			Int("error code", errCode)
 	}
 
-	logger.Debug().Err(err).Int("error code", errCode).
-		Msg("Not retrying on error because it is not transient nor is it in our additional codes list.")
+	retryLogEvent.
+		Str("reason", whyTransient).
+		Msg("Pausing then will retry operation.")
 
-	return false
+	return true
 }
 
 // Use this method for aggregates which should take a UUID in new versions but not old ones.

internal/verifier/change_stream.go

@@ -12,7 +12,6 @@ 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
@@ -138,12 +137,14 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		return err
 	}
 
-	readAndHandleOneChangeEventBatch := func() (bool, error) {
+	readAndHandleOneChangeEventBatch := func() error {
 		eventsRead := 0
 		var changeEventBatch []ParsedEvent
 
 		for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
+			verifier.logger.Info().Msg("reading event")
 			gotEvent := cs.TryNext(ctx)
+			verifier.logger.Info().Err(cs.Err()).Msgf("got event? %v", gotEvent)
 
 			if !gotEvent || cs.Err() != nil {
 				break
@@ -154,21 +155,29 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 			}
 
 			if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
-				return false, errors.Wrap(err, "failed to decode change event")
+				return errors.Wrap(err, "failed to decode change event")
 			}
 
+			verifier.logger.Debug().Interface("event", changeEventBatch[eventsRead]).Msg("Got event.")
+
 			eventsRead++
 		}
 
-		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")
-			}
+		if cs.Err() != nil {
+			return errors.Wrap(cs.Err(), "change stream iteration failed")
+		}
+
+		if eventsRead == 0 {
+			return nil
 		}
 
-		return eventsRead > 0, errors.Wrap(cs.Err(), "change stream iteration failed")
+		verifier.logger.Debug().Int("eventsCount", eventsRead).Msgf("Received a batch of events.")
+		err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
+		if err != nil {
+			return errors.Wrap(err, "failed to handle change events")
+		}
+
+		return nil
 	}
 
 	for {
@@ -184,29 +193,45 @@ 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 <-verifier.changeStreamEnderChan:
+		case finalTs := <-verifier.changeStreamFinalTsChan:
 			verifier.logger.Debug().
-				Msg("Change stream thread received shutdown request.")
+				Interface("finalTimestamp", finalTs).
+				Msg("Change stream thread received final timestamp. Finalizing change stream.")
 
 			changeStreamEnded = true
 
 			// Read all change events until the source reports no events.
 			// (i.e., the `getMore` call returns empty)
 			for {
-				var gotEvent bool
-				gotEvent, err = readAndHandleOneChangeEventBatch()
+				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.Compare(finalTs) >= 0 {
+					verifier.logger.Debug().
+						Interface("currentTimestamp", curTs).
+						Interface("finalTimestamp", finalTs).
+						Msg("Change stream has reached the final timestamp. Shutting down.")
 
-				if !gotEvent || err != nil {
+					break
+				}
+
+				err = readAndHandleOneChangeEventBatch()
+
+				if err != nil {
 					break
 				}
 			}
 
 		default:
-			_, err = readAndHandleOneChangeEventBatch()
-		}
+			err = readAndHandleOneChangeEventBatch()
 
-		if err == nil {
-			err = persistResumeTokenIfNeeded()
+			if err == nil {
+				err = persistResumeTokenIfNeeded()
+			}
 		}
 
 		if err != nil && !errors.Is(err, context.Canceled) {
@@ -237,9 +262,9 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 
 	infoLog := verifier.logger.Info()
 	if verifier.lastChangeEventTime == nil {
-		infoLog = infoLog.Str("changeStreamStopTime", "none")
+		infoLog = infoLog.Str("lastEventTime", "none")
 	} else {
-		infoLog = infoLog.Interface("changeStreamStopTime", *verifier.lastChangeEventTime)
+		infoLog = infoLog.Interface("lastEventTime", *verifier.lastChangeEventTime)
 	}
 
 	infoLog.Msg("Change stream is done.")
@@ -265,7 +290,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 
 		ts, err := extractTimestampFromResumeToken(savedResumeToken)
 		if err == nil {
-			logEvent = addUnixTimeToLogEvent(ts.T, logEvent)
+			logEvent = addTimestampToLogEvent(ts, logEvent)
 		} else {
 			verifier.logger.Warn().
 				Err(err).
@@ -284,6 +309,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 		return errors.Wrap(err, "failed to start session")
 	}
 	sctx := mongo.NewSessionContext(ctx, sess)
+	verifier.logger.Debug().Interface("pipeline", pipeline).Msg("Opened change stream.")
 	srcChangeStream, err := verifier.srcClient.Watch(sctx, pipeline, opts)
 	if err != nil {
 		return errors.Wrap(err, "failed to open change stream")
@@ -318,8 +344,10 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 	return nil
 }
 
-func addUnixTimeToLogEvent[T constraints.Integer](unixTime T, event *zerolog.Event) *zerolog.Event {
-	return event.Time("clockTime", time.Unix(int64(unixTime), int64(0)))
+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 (v *Verifier) getChangeStreamMetadataCollection() *mongo.Collection {
@@ -358,7 +386,7 @@ func (verifier *Verifier) persistChangeStreamResumeToken(ctx context.Context, cs
 		logEvent := verifier.logger.Debug()
 
 		if err == nil {
-			logEvent = addUnixTimeToLogEvent(ts.T, logEvent)
+			logEvent = addTimestampToLogEvent(ts, logEvent)
 		} else {
 			verifier.logger.Warn().Err(err).
 				Msg("failed to extract resume token timestamp")

internal/verifier/check.go

@@ -42,13 +42,24 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	verifier.MaybeStartPeriodicHeapProfileCollection(ctx)
 }
 
-func (verifier *Verifier) waitForChangeStream() error {
+func (verifier *Verifier) waitForChangeStream(ctx context.Context) 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.changeStreamEnderChan <- struct{}{}
+
+		finalTs, err := GetClusterTime(
+			ctx,
+			verifier.logger,
+			verifier.srcClient,
+		)
+
+		if err != nil {
+			return errors.Wrapf(err, "failed to fetch source's cluster time")
+		}
+
+		verifier.changeStreamFinalTsChan <- finalTs
 		select {
 		case err := <-verifier.changeStreamErrChan:
 			verifier.logger.Warn().Err(err).
@@ -242,7 +253,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()
+			err := verifier.waitForChangeStream(ctx)
 			if err != nil {
 				return err
 			}