Fix flapping TestStartAtTimeNoChanges.

Author: FGasper

internal/verifier/change_stream.go

@@ -47,6 +47,8 @@ const (
 	metadataChangeStreamCollectionName = "changeStream"
 )
 
+var maxChangeStreamAwaitTime = time.Second
+
 type UnknownEventError struct {
 	Event *ParsedEvent
 }
@@ -293,6 +295,9 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 			(csr.lastChangeEventTime == nil ||
 				csr.lastChangeEventTime.Before(*changeEventBatch[eventsRead].ClusterTime)) {
 			csr.lastChangeEventTime = changeEventBatch[eventsRead].ClusterTime
+			csr.logger.Trace().
+				Interface("event", changeEventBatch[eventsRead]).
+				Msg("Updated lastChangeEventTime.")
 		}
 
 		eventsRead++
@@ -431,7 +436,7 @@ func (csr *ChangeStreamReader) createChangeStream(
 ) (*mongo.ChangeStream, mongo.Session, primitive.Timestamp, error) {
 	pipeline := csr.GetChangeStreamFilter()
 	opts := options.ChangeStream().
-		SetMaxAwaitTime(1 * time.Second).
+		SetMaxAwaitTime(maxChangeStreamAwaitTime).
 		SetFullDocument(options.UpdateLookup)
 
 	if csr.clusterInfo.VersionArray[0] >= 6 {
@@ -488,11 +493,17 @@ func (csr *ChangeStreamReader) createChangeStream(
 	// With sharded clusters the resume token might lead the cluster time
 	// by 1 increment. In that case we need the actual cluster time;
 	// otherwise we will get errors.
-	clusterTime, err := getClusterTimeFromSession(sess)
+	clusterTime, err := util.GetClusterTimeFromSession(sess)
 	if err != nil {
 		return nil, nil, primitive.Timestamp{}, errors.Wrap(err, "failed to read cluster time from session")
 	}
 
+	csr.logger.Debug().
+		Interface("resumeTokenTimestamp", startTs).
+		Interface("clusterTime", clusterTime).
+		Stringer("changeStreamReader", csr).
+		Msg("Using earlier time as start timestamp.")
+
 	if startTs.After(clusterTime) {
 		startTs = clusterTime
 	}
@@ -659,19 +670,3 @@ func extractTimestampFromResumeToken(resumeToken bson.Raw) (primitive.Timestamp,
 
 	return resumeTokenTime, nil
 }
-
-func getClusterTimeFromSession(sess mongo.Session) (primitive.Timestamp, error) {
-	ctStruct := struct {
-		ClusterTime struct {
-			ClusterTime primitive.Timestamp `bson:"clusterTime"`
-		} `bson:"$clusterTime"`
-	}{}
-
-	clusterTimeRaw := sess.ClusterTime()
-	err := bson.Unmarshal(sess.ClusterTime(), &ctStruct)
-	if err != nil {
-		return primitive.Timestamp{}, errors.Wrapf(err, "failed to find clusterTime in session cluster time document (%v)", clusterTimeRaw)
-	}
-
-	return ctStruct.ClusterTime.ClusterTime, nil
-}

internal/verifier/change_stream_test.go

@@ -283,7 +283,7 @@ func (suite *IntegrationTestSuite) getClusterTime(ctx context.Context, client *m
 	sctx := mongo.NewSessionContext(ctx, sess)
 	suite.Require().NoError(sess.Client().Ping(sctx, nil))
 
-	newTime, err := getClusterTimeFromSession(sess)
+	newTime, err := util.GetClusterTimeFromSession(sess)
 	suite.Require().NoError(err, "should fetch cluster time")
 
 	return newTime
@@ -306,21 +306,39 @@ func (suite *IntegrationTestSuite) fetchVerifierRechecks(ctx context.Context, ve
 func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 	zerolog.SetGlobalLevel(zerolog.TraceLevel)
 
-	verifier := suite.BuildVerifier()
-	ctx := suite.Context()
-	sess, err := suite.srcMongoClient.StartSession()
-	suite.Require().NoError(err)
-	sctx := mongo.NewSessionContext(ctx, sess)
-	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne(
-		sctx, bson.D{{"_id", 0}})
-	suite.Require().NoError(err)
-	origStartTs := sess.OperationTime()
-	suite.Require().NotNil(origStartTs)
-	suite.startSrcChangeStreamReaderAndHandler(ctx, verifier)
-	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
-	verifier.srcChangeStreamReader.writesOffTs.Set(*origStartTs)
-	<-verifier.srcChangeStreamReader.doneChan
-	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
+	// Each of these takes ~1s, so don’t do too many of them.
+	for range 5 {
+		verifier := suite.BuildVerifier()
+		ctx := suite.Context()
+		sess, err := suite.srcMongoClient.StartSession()
+		suite.Require().NoError(err)
+		sctx := mongo.NewSessionContext(ctx, sess)
+		_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne(
+			sctx, bson.D{})
+		suite.Require().NoError(err, "should insert doc")
+
+		insertTs, err := util.GetClusterTimeFromSession(sess)
+		suite.Require().NoError(err, "should get cluster time")
+
+		suite.startSrcChangeStreamReaderAndHandler(ctx, verifier)
+
+		startAtTs := verifier.srcChangeStreamReader.startAtTs
+		suite.Require().NotNil(startAtTs)
+
+		suite.Require().False(
+			startAtTs.After(insertTs),
+			"change stream should start no later than the last operation",
+		)
+
+		verifier.srcChangeStreamReader.writesOffTs.Set(insertTs)
+
+		<-verifier.srcChangeStreamReader.doneChan
+
+		suite.Require().False(
+			verifier.srcChangeStreamReader.startAtTs.Before(*startAtTs),
+			"new startAtTs should be no earlier than last one",
+		)
+	}
 }
 
 func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {