comments

Author: FGasper

internal/verifier/change_stream.go

@@ -223,9 +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")
 				}
 
-				// writesOffTs refers to a real event (i.e., not appendOplogNote),
-				// so we need to TryNext() if curTs == writesOffTs.
-				if curTs.After(writesOffTs) {
+				// writesOffTs never refers to a real event,
+				// so we can stop once curTs >= writesOffTs.
+				if !curTs.Before(writesOffTs) {
 					verifier.logger.Debug().
 						Interface("currentTimestamp", curTs).
 						Interface("writesOffTimestamp", writesOffTs).
@@ -351,6 +351,7 @@ func (verifier *Verifier) createChangeStream(
 
 // StartChangeStream starts the change stream.
 func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
+	// Result seems a bit simpler than messing with 2 separate channels.
 	resultChan := make(chan mo.Result[primitive.Timestamp])
 
 	go func() {