fix

Author: FGasper

internal/verifier/change_stream.go

@@ -223,7 +223,9 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 					return errors.Wrap(err, "failed to extract timestamp from change stream's resume token")
 				}
 
-				if !curTs.Before(writesOffTs) {
+				// writesOffTs refers to a real event (i.e., not appendOplogNote),
+				// so we need to TryNext() if curTs == writesOffTs.
+				if curTs.After(writesOffTs) {
 					verifier.logger.Debug().
 						Interface("currentTimestamp", curTs).
 						Interface("writesOffTimestamp", writesOffTs).

internal/verifier/clustertime.go

@@ -7,7 +7,6 @@ import (
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/10gen/migration-verifier/mbson"
-	"github.com/10gen/migration-verifier/option"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -18,9 +17,10 @@ import (
 
 const opTimeKeyInServerResponse = "operationTime"
 
-// GetNewClusterTime advances the cluster time and returns that time.
-// All shards’ cluster times will meet or exceed the returned time.
-func GetNewClusterTime(
+// GetClusterTime get the current cluster time and returns that time.
+// All shards’ cluster times will also be updated to meet or exceed the
+// returned time.
+func GetClusterTime(
 	ctx context.Context,
 	logger *logger.Logger,
 	client *mongo.Client,
@@ -36,7 +36,7 @@ func GetNewClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			clusterTime, err = fetchNewClusterTime(ctx, client)
+			clusterTime, err = fetchClusterTime(ctx, client)
 			return err
 		},
 	)
@@ -68,22 +68,27 @@ func GetNewClusterTime(
 
 // Use this when we just need the correct cluster time without
 // actually changing any shards’ oplogs.
-func fetchNewClusterTime(
+func fetchClusterTime(
 	ctx context.Context,
 	client *mongo.Client,
 ) (primitive.Timestamp, error) {
 	cmd, rawResponse, err := runAppendOplogNote(
 		ctx,
 		client,
 		"expect StaleClusterTime error",
-		option.None[primitive.Timestamp](),
+		primitive.Timestamp{1, 0},
 	)
 
-	if err != nil {
-		return primitive.Timestamp{}, errors.Wrapf(
+	// We expect an error here; if we didn't get one then something is
+	// amiss on the server.
+	if err == nil {
+		return primitive.Timestamp{}, errors.Errorf("server request unexpectedly succeeded: %v", cmd)
+	}
+
+	if !util.IsStaleClusterTimeError(err) {
+		return primitive.Timestamp{}, errors.Wrap(
 			err,
-			"unexpected error (expected StaleClusterTime) from request (%v)",
-			cmd,
+			"unexpected error (expected StaleClusterTime) from request",
 		)
 	}
 
@@ -99,7 +104,7 @@ func syncClusterTimeAcrossShards(
 		ctx,
 		client,
 		"syncing cluster time",
-		option.Some(maxTime),
+		maxTime,
 	)
 
 	// If any shard’s cluster time >= maxTime, the mongos will return a
@@ -119,19 +124,16 @@ func runAppendOplogNote(
 	ctx context.Context,
 	client *mongo.Client,
 	note string,
-	maxClusterTimeOpt option.Option[primitive.Timestamp],
+	maxClusterTime primitive.Timestamp,
 ) (bson.D, bson.Raw, error) {
 	cmd := bson.D{
 		{"appendOplogNote", 1},
+		{"maxClusterTime", maxClusterTime},
 		{"data", bson.D{
 			{"migration-verifier", note},
 		}},
 	}
 
-	if maxClusterTime, has := maxClusterTimeOpt.Get(); has {
-		cmd = append(cmd, bson.E{"maxClusterTime", maxClusterTime})
-	}
-
 	resp := client.
 		Database(
 			"admin",

internal/verifier/clustertime_test.go

@@ -6,7 +6,7 @@ func (suite *IntegrationTestSuite) TestGetClusterTime() {
 	ctx := context.Background()
 	logger, _ := getLoggerAndWriter("stdout")
 
-	ts, err := GetNewClusterTime(ctx, logger, suite.srcMongoClient)
+	ts, err := GetClusterTime(ctx, logger, suite.srcMongoClient)
 	suite.Require().NoError(err)
 
 	suite.Assert().NotZero(ts, "timestamp should be nonzero")

internal/verifier/migration_verifier.go

@@ -241,7 +241,7 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	if verifier.writesOffTimestamp == nil {
 		verifier.logger.Debug().Msg("Change stream still running. Signalling that writes are done.")
 
-		finalTs, err := GetNewClusterTime(
+		finalTs, err := GetClusterTime(
 			ctx,
 			verifier.logger,
 			verifier.srcClient,