new cluster time

Author: FGasper

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,10 +18,9 @@ import (
 
 const opTimeKeyInServerResponse = "operationTime"
 
-// 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(
+// GetClusterTime 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,
 	client *mongo.Client,
@@ -36,7 +36,7 @@ func GetClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			clusterTime, err = fetchClusterTime(ctx, client)
+			clusterTime, err = fetchNewClusterTime(ctx, client)
 			return err
 		},
 	)
@@ -68,15 +68,15 @@ func GetClusterTime(
 
 // Use this when we just need the correct cluster time without
 // actually changing any shards’ oplogs.
-func fetchClusterTime(
+func fetchNewClusterTime(
 	ctx context.Context,
 	client *mongo.Client,
 ) (primitive.Timestamp, error) {
 	cmd, rawResponse, err := runAppendOplogNote(
 		ctx,
 		client,
 		"expect StaleClusterTime error",
-		primitive.Timestamp{1, 0},
+		option.None[primitive.Timestamp](),
 	)
 
 	// We expect an error here; if we didn't get one then something is
@@ -104,7 +104,7 @@ func syncClusterTimeAcrossShards(
 		ctx,
 		client,
 		"syncing cluster time",
-		maxTime,
+		option.Some(maxTime),
 	)
 
 	// If any shard’s cluster time >= maxTime, the mongos will return a
@@ -124,16 +124,19 @@ func runAppendOplogNote(
 	ctx context.Context,
 	client *mongo.Client,
 	note string,
-	maxClusterTime primitive.Timestamp,
+	maxClusterTimeOpt option.Option[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

@@ -1,13 +1,32 @@
 package verifier
 
-import "context"
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
 
 func (suite *IntegrationTestSuite) TestGetClusterTime() {
 	ctx := context.Background()
 	logger, _ := getLoggerAndWriter("stdout")
 
-	ts, err := GetClusterTime(ctx, logger, suite.srcMongoClient)
+	sess, err := suite.srcMongoClient.StartSession()
+	suite.Require().NoError(err)
+
+	_, err = suite.srcMongoClient.
+		Database(suite.DBNameForTest()).
+		Collection("mycoll").
+		InsertOne(ctx, bson.D{})
+	clusterTimeVal, err := sess.ClusterTime().LookupErr("$clusterTime", "clusterTime")
+	suite.Require().NoError(err)
+
+	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}))
 }

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 := GetClusterTime(
+		finalTs, err := GetNewClusterTime(
 			ctx,
 			verifier.logger,
 			verifier.srcClient,