preliminary

Author: FGasper

internal/util/value.go

@@ -0,0 +1,7 @@
+package util
+
+// ResetToZero is a convenience to set any variable to its
+// type's zero value without writing out the type.
+func ResetToZero[T any](ptr *T) {
+	*ptr = *new(T)
+}

internal/verifier/change_stream.go

@@ -160,15 +160,21 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 			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 false, errors.Wrap(cs.Err(), "change stream iteration failed")
+		}
+
+		if eventsRead == 0 {
+			return false, 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 false, errors.Wrap(err, "failed to handle change events")
+		}
+
+		return true, nil
 	}
 
 	for {
@@ -184,15 +190,31 @@ 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.")
 
 			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/check.go

@@ -9,6 +9,7 @@ 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"
 )
 
@@ -42,13 +43,13 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	verifier.MaybeStartPeriodicHeapProfileCollection(ctx)
 }
 
-func (verifier *Verifier) waitForChangeStream() error {
+func (verifier *Verifier) finalizeChangeStream(finalTs primitive.Timestamp) 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{}{}
+		verifier.changeStreamFinalTsChan <- finalTs
 		select {
 		case err := <-verifier.changeStreamErrChan:
 			verifier.logger.Warn().Err(err).
@@ -238,11 +239,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 verifier.writesOff {
+		if ts, writesOff := verifier.writesOffTs.Get(); 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.waitForChangeStream()
+			err := verifier.finalizeChangeStream(ts)
 			if err != nil {
 				return err
 			}

internal/verifier/clustertime.go

@@ -54,23 +54,13 @@ 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.
-	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
-		},
-	)
+	// Since this is just an optimization, failures here are nonfatal,
+	// and we might as well forgo the retryer.
+	_, err = runAppendOplogNote(ctx, client, false)
 	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 append oplog note; change stream may need extra time to finish.")
+			Msg("Failed to update oplog; change stream may need extra time to finish.")
 	}
 
 	return optime, nil

internal/verifier/migration_verifier.go

@@ -21,7 +21,9 @@ 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"
@@ -78,7 +80,7 @@ var timeFormat = time.RFC3339
 
 // Verifier is the main state for the migration verifier
 type Verifier struct {
-	writesOff          bool
+	writesOffTs        option.Option[primitive.Timestamp]
 	lastGeneration     bool
 	running            bool
 	generation         int
@@ -123,12 +125,12 @@ type Verifier struct {
 	metaDBName    string
 	srcStartAtTs  *primitive.Timestamp
 
-	mux                   sync.RWMutex
-	changeStreamRunning   bool
-	changeStreamEnderChan chan struct{}
-	changeStreamErrChan   chan error
-	changeStreamDoneChan  chan struct{}
-	lastChangeEventTime   *primitive.Timestamp
+	mux                     sync.RWMutex
+	changeStreamRunning     bool
+	changeStreamFinalTsChan chan primitive.Timestamp
+	changeStreamErrChan     chan error
+	changeStreamDoneChan    chan struct{}
+	lastChangeEventTime     *primitive.Timestamp
 
 	readConcernSetting ReadConcernSetting
 
@@ -188,15 +190,15 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 	}
 
 	return &Verifier{
-		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,
+		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,
 
 		// This will get recreated once gen0 starts, but we want it
 		// here in case the change streams gets an event before then.
@@ -228,18 +230,30 @@ func (verifier *Verifier) SetFailureDisplaySize(size int64) {
 	verifier.failureDisplaySize = size
 }
 
-func (verifier *Verifier) WritesOff(ctx context.Context) {
+func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	verifier.logger.Debug().
 		Msg("WritesOff called.")
 
 	verifier.mux.Lock()
-	verifier.writesOff = true
-	verifier.mux.Unlock()
+	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
 }
 
 func (verifier *Verifier) WritesOn(ctx context.Context) {
 	verifier.mux.Lock()
-	verifier.writesOff = false
+	util.ResetToZero(&verifier.writesOffTs)
 	verifier.mux.Unlock()
 }
 

internal/verifier/web_server.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)
+	WritesOff(ctx context.Context) error
 	WritesOn(ctx context.Context)
 	GetProgress(ctx context.Context) (Progress, error)
 }

internal/verifier/web_server_test.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) {}
-func (verifier *MockVerifier) WritesOn(ctx context.Context)  {}
+func (verifier *MockVerifier) WritesOff(ctx context.Context) error { return nil }
+func (verifier *MockVerifier) WritesOn(ctx context.Context)        {}
 func (verifier *MockVerifier) GetProgress(ctx context.Context) (Progress, error) {
 	return Progress{}, nil
 }

option/bson.go

@@ -0,0 +1,49 @@
+package option
+
+import (
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/bsontype"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+// MarshalBSONValue implements bson.ValueMarshaler.
+func (o Option[T]) MarshalBSONValue() (bsontype.Type, []byte, error) {
+	val, exists := o.Get()
+	if !exists {
+		return bson.MarshalValue(primitive.Null{})
+	}
+
+	return bson.MarshalValue(val)
+}
+
+// UnmarshalBSONValue implements bson.ValueUnmarshaler.
+func (o *Option[T]) UnmarshalBSONValue(bType bsontype.Type, raw []byte) error {
+
+	switch bType {
+	case bson.TypeNull:
+		o.val = nil
+
+	default:
+		valPtr := new(T)
+
+		err := bson.UnmarshalValue(bType, raw, &valPtr)
+		if err != nil {
+			return errors.Wrapf(err, "failed to unmarshal %T", *o)
+		}
+
+		// This may not even be possible, but we should still check.
+		if isNil(*valPtr) {
+			return errors.Wrapf(err, "refuse to unmarshal nil %T value", *o)
+		}
+
+		o.val = valPtr
+	}
+
+	return nil
+}
+
+// IsZero implements bsoncodec.Zeroer.
+func (o Option[T]) IsZero() bool {
+	return o.IsNone()
+}

option/json.go

@@ -0,0 +1,37 @@
+package option
+
+import (
+	"bytes"
+	"encoding/json"
+)
+
+var _ json.Marshaler = &Option[int]{}
+var _ json.Unmarshaler = &Option[int]{}
+
+// MarshalJSON encodes Option into json.
+func (o Option[T]) MarshalJSON() ([]byte, error) {
+	val, exists := o.Get()
+	if exists {
+		return json.Marshal(val)
+	}
+
+	return json.Marshal(nil)
+}
+
+// UnmarshalJSON decodes Option from json.
+func (o *Option[T]) UnmarshalJSON(b []byte) error {
+	if bytes.Equal(b, []byte("null")) {
+		o.val = nil
+	} else {
+		val := *new(T)
+
+		err := json.Unmarshal(b, &val)
+		if err != nil {
+			return err
+		}
+
+		o.val = &val
+	}
+
+	return nil
+}