refactored appendOplogNote (& al.)

Author: FGasper

internal/verifier/change_stream.go

@@ -126,60 +126,60 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 	return []bson.D{stage}
 }
 
-func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
-	defer cs.Close(ctx)
+func (verifier *Verifier) readAndHandleOneChangeEventBatch(ctx context.Context, cs *mongo.ChangeStream) error {
+	eventsRead := 0
+	var changeEventBatch []ParsedEvent
 
-	var lastPersistedTime time.Time
+	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
+		gotEvent := cs.TryNext(ctx)
 
-	persistResumeTokenIfNeeded := func() error {
-		if time.Since(lastPersistedTime) <= minChangeStreamPersistInterval {
-			return nil
+		if cs.Err() != nil {
+			return errors.Wrap(cs.Err(), "change stream iteration failed")
 		}
 
-		err := verifier.persistChangeStreamResumeToken(ctx, cs)
-		if err == nil {
-			lastPersistedTime = time.Now()
+		if !gotEvent {
+			break
 		}
 
-		return err
-	}
+		if changeEventBatch == nil {
+			changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
+		}
 
-	readAndHandleOneChangeEventBatch := func() error {
-		eventsRead := 0
-		var changeEventBatch []ParsedEvent
+		if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
+			return errors.Wrap(err, "failed to decode change event")
+		}
 
-		for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
-			gotEvent := cs.TryNext(ctx)
+		eventsRead++
+	}
 
-			if cs.Err() != nil {
-				return errors.Wrap(cs.Err(), "change stream iteration failed")
-			}
+	if eventsRead == 0 {
+		return nil
+	}
 
-			if !gotEvent {
-				break
-			}
+	err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
+	if err != nil {
+		return errors.Wrap(err, "failed to handle change events")
+	}
 
-			if changeEventBatch == nil {
-				changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
-			}
+	return nil
+}
 
-			if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
-				return errors.Wrap(err, "failed to decode change event")
-			}
+func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
+	defer cs.Close(ctx)
 
-			eventsRead++
-		}
+	var lastPersistedTime time.Time
 
-		if eventsRead == 0 {
+	persistResumeTokenIfNeeded := func() error {
+		if time.Since(lastPersistedTime) <= minChangeStreamPersistInterval {
 			return nil
 		}
 
-		err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
-		if err != nil {
-			return errors.Wrap(err, "failed to handle change events")
+		err := verifier.persistChangeStreamResumeToken(ctx, cs)
+		if err == nil {
+			lastPersistedTime = time.Now()
 		}
 
-		return nil
+		return err
 	}
 
 	for {
@@ -224,15 +224,15 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 					break
 				}
 
-				err = readAndHandleOneChangeEventBatch()
+				err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
 
 				if err != nil {
 					break
 				}
 			}
 
 		default:
-			err = readAndHandleOneChangeEventBatch()
+			err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
 
 			if err == nil {
 				err = persistResumeTokenIfNeeded()

internal/verifier/clustertime.go

@@ -23,7 +23,7 @@ const opTimeKeyInServerResponse = "operationTime"
 func GetClusterTime(
 	ctx context.Context,
 	logger *logger.Logger,
-	client *mongo.Client, // could change to type=any if we need
+	client *mongo.Client,
 ) (primitive.Timestamp, error) {
 	retryer := retry.New(retry.DefaultDurationLimit)
 
@@ -39,7 +39,7 @@ func GetClusterTime(
 		logger,
 		func(_ *retry.Info) error {
 			var err error
-			optime, err = runAppendOplogNote(ctx, client, true)
+			optime, err = syncClusterTimeAcrossShards(ctx, client)
 			if err != nil {
 				err = fmt.Errorf("%w: %w", retry.CustomTransientErr, err)
 			}
@@ -51,16 +51,17 @@ func GetClusterTime(
 		return primitive.Timestamp{}, err
 	}
 
-	// OPTIMIZATION: We now append an oplog entry--this time for real!--to
-	// cause any lagging shards to output their events.
+	// OPTIMIZATION FOR SHARDED CLUSTERS: We append another oplog entry to
+	// bring all shards to a cluster time that’s *after* the optime that we’ll
+	// return. That way any events *at*
 	//
 	// 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)
+			optime, err = syncClusterTimeAcrossShards(ctx, client)
 			if err != nil {
 				err = fmt.Errorf("%w: %w", retry.CustomTransientErr, err)
 			}
@@ -76,55 +77,75 @@ func GetClusterTime(
 	return optime, nil
 }
 
-func runAppendOplogNote(
+// Use this when we just need the correct cluster time without
+// actually changing any shards’ oplogs.
+func fetchClusterTime(
 	ctx context.Context,
 	client *mongo.Client,
-	doomed bool,
 ) (primitive.Timestamp, error) {
-	// We don’t need to write to any shards’ oplogs; we just
-	// need to fetch
-	cmd := bson.D{
-		{"appendOplogNote", 1},
-		{"data", bson.D{
-			{"migration-verifier", "expect StaleClusterTime error"},
-		}},
-	}
+	cmd, rawResponse, err := runAppendOplogNote(
+		ctx,
+		client,
+		"expect StaleClusterTime error",
+		bson.E{"maxClusterTime", primitive.Timestamp{1, 0}},
+	)
 
-	if doomed {
-		cmd = append(cmd, bson.E{"maxClusterTime", 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)
 	}
 
-	resp := client.Database("admin").RunCommand(ctx, cmd)
-
-	err := resp.Err()
-
-	if doomed {
-		// 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.Errorf(
-				"unexpected error (expected StaleClusterTime) from request (%v): %v",
-				err,
-				cmd,
-			)
-		}
-	} else if err != nil {
-		return primitive.Timestamp{}, errors.Wrapf(
+	if !util.IsStaleClusterTimeError(err) {
+		return primitive.Timestamp{}, errors.Wrap(
 			err,
-			"command (%v) failed unexpectedly",
-			cmd,
+			"unexpected error (expected StaleClusterTime) from request",
 		)
 	}
 
-	rawResponse, _ := resp.Raw()
+	return getOpTimeFromRawResponse(rawResponse)
+}
+
+func syncClusterTimeAcrossShards(
+	ctx context.Context,
+	client *mongo.Client,
+) (primitive.Timestamp, error) {
+	_, rawResponse, err := runAppendOplogNote(ctx, client, "syncing cluster time")
+
+	if err != nil {
+		return primitive.Timestamp{}, err
+	}
 
 	return getOpTimeFromRawResponse(rawResponse)
 }
 
+func runAppendOplogNote(
+	ctx context.Context,
+	client *mongo.Client,
+	note string,
+	extraPieces ...bson.E,
+) (bson.D, bson.Raw, error) {
+	cmd := append(
+		bson.D{
+			{"appendOplogNote", 1},
+			{"data", bson.D{
+				{"migration-verifier", note},
+			}},
+		},
+		extraPieces...,
+	)
+
+	resp := client.Database("admin").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