REP-5318 Make GetNewClusterTime return a new time.

Previously this function actually returned the most recent cluster
time rather than creating a new cluster time. This changeset makes it
now create a new time instead. This simplifies the change stream’s
handling of the writesOff timestamp because the writesOff timestamp is
separate from any change event.

Author: FGasper

internal/verifier/change_stream.go

@@ -226,7 +226,7 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 					break
 				}
 
-				if curTs.After(writesOffTs) {
+				if !curTs.Before(writesOffTs) {
 					verifier.logger.Debug().
 						Interface("currentTimestamp", curTs).
 						Interface("writesOffTimestamp", writesOffTs).

internal/verifier/clustertime.go

@@ -7,6 +7,7 @@ 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"
@@ -17,8 +18,8 @@ import (
 
 const opTimeKeyInServerResponse = "operationTime"
 
-// GetNewClusterTime advances the cluster time and returns that time.
-// All shards’ cluster times will meet or exceed the returned time.
+// GetNewClusterTime creates a new cluster time, updates all shards’
+// cluster times to meet or exceed that time, then returns it.
 func GetNewClusterTime(
 	ctx context.Context,
 	logger *logger.Logger,
@@ -35,7 +36,12 @@ func GetNewClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			clusterTime, err = fetchClusterTime(ctx, client)
+			clusterTime, err = runAppendOplogNote(
+				ctx,
+				client,
+				"new ts",
+				option.None[primitive.Timestamp](),
+			)
 			return err
 		},
 	)
@@ -52,7 +58,12 @@ func GetNewClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			_, err = syncClusterTimeAcrossShards(ctx, client, clusterTime)
+			_, err = runAppendOplogNote(
+				ctx,
+				client,
+				"sync ts",
+				option.Some(clusterTime),
+			)
 			return err
 		},
 	)
@@ -65,46 +76,31 @@ func GetNewClusterTime(
 	return clusterTime, nil
 }
 
-// Use this when we just need the correct cluster time without
-// actually changing any shards’ oplogs.
-func fetchClusterTime(
+func runAppendOplogNote(
 	ctx context.Context,
 	client *mongo.Client,
+	note string,
+	maxClusterTimeOpt option.Option[primitive.Timestamp],
 ) (primitive.Timestamp, error) {
-	cmd, rawResponse, err := runAppendOplogNote(
-		ctx,
-		client,
-		"expect StaleClusterTime error",
-		primitive.Timestamp{1, 0},
-	)
-
-	// 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)
+	cmd := bson.D{
+		{"appendOplogNote", 1},
+		{"data", bson.D{
+			{"migration-verifier", note},
+		}},
 	}
 
-	if !util.IsStaleClusterTimeError(err) {
-		return primitive.Timestamp{}, errors.Wrap(
-			err,
-			"unexpected error (expected StaleClusterTime) from request",
-		)
+	if maxClusterTime, has := maxClusterTimeOpt.Get(); has {
+		cmd = append(cmd, bson.E{"maxClusterTime", maxClusterTime})
 	}
 
-	return getOpTimeFromRawResponse(rawResponse)
-}
+	resp := client.
+		Database(
+			"admin",
+			options.Database().SetWriteConcern(writeconcern.Majority()),
+		).
+		RunCommand(ctx, cmd)
 
-func syncClusterTimeAcrossShards(
-	ctx context.Context,
-	client *mongo.Client,
-	maxTime primitive.Timestamp,
-) (primitive.Timestamp, error) {
-	_, rawResponse, err := runAppendOplogNote(
-		ctx,
-		client,
-		"syncing cluster time",
-		maxTime,
-	)
+	rawResponse, err := resp.Raw()
 
 	// If any shard’s cluster time >= maxTime, the mongos will return a
 	// StaleClusterTime error. This particular error doesn’t indicate a
@@ -119,36 +115,6 @@ func syncClusterTimeAcrossShards(
 	return getOpTimeFromRawResponse(rawResponse)
 }
 
-func runAppendOplogNote(
-	ctx context.Context,
-	client *mongo.Client,
-	note string,
-	maxClusterTime primitive.Timestamp,
-) (bson.D, bson.Raw, error) {
-	cmd := bson.D{
-		{"appendOplogNote", 1},
-		{"maxClusterTime", maxClusterTime},
-		{"data", bson.D{
-			{"migration-verifier", note},
-		}},
-	}
-
-	resp := client.
-		Database(
-			"admin",
-			options.Database().SetWriteConcern(writeconcern.Majority()),
-		).
-		RunCommand(ctx, cmd)
-
-	raw, err := resp.Raw()
-
-	return cmd, raw, errors.Wrapf(
-		err,
-		"command (%v) failed unexpectedly",
-		cmd,
-	)
-}
-
 func getOpTimeFromRawResponse(rawResponse bson.Raw) (primitive.Timestamp, error) {
 	// Get the `operationTime` from the response and return it.
 	var optime primitive.Timestamp

internal/verifier/clustertime_test.go

@@ -1,13 +1,35 @@
 package verifier
 
-import "context"
+import (
+	"context"
 
-func (suite *IntegrationTestSuite) TestGetClusterTime() {
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+func (suite *IntegrationTestSuite) TestGetNewClusterTime() {
 	ctx := context.Background()
 	logger, _ := getLoggerAndWriter("stdout")
 
+	sess, err := suite.srcMongoClient.StartSession()
+	suite.Require().NoError(err)
+
+	_, err = suite.srcMongoClient.
+		Database(suite.DBNameForTest()).
+		Collection("mycoll").
+		InsertOne(mongo.NewSessionContext(ctx, sess), bson.D{})
+	suite.Require().NoError(err)
+
+	clusterTimeVal, err := sess.ClusterTime().LookupErr("$clusterTime", "clusterTime")
+	suite.Require().NoError(err, "should extract cluster time from %+v", sess.ClusterTime())
+
+	clusterT, clusterI, ok := clusterTimeVal.TimestampOK()
+	suite.Require().True(ok, "session cluster time (%s: %v) must be a timestamp", clusterTimeVal.Type, clusterTimeVal)
+
 	ts, err := GetNewClusterTime(ctx, logger, suite.srcMongoClient)
 	suite.Require().NoError(err)
 
-	suite.Assert().NotZero(ts, "timestamp should be nonzero")
+	suite.Require().NotZero(ts, "timestamp should be nonzero")
+	suite.Assert().True(ts.After(primitive.Timestamp{T: clusterT, I: clusterI}))
 }

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
+}