REP-6875 Make ChangeReader distinguish last-handled & last operation times. (#194)

Previously the ChangeReader exposed only the reader lag, but external tooling needs to track not merely the lag but also the reader’s exact last-handled timestamp.

This changeset reworks the ChangeReader accordingly so that we can add that information to the /progress endpoint in a later PR.

(This is an internal-only change that will facilitate further PRs for this JIRA ticket.

Author: FGasper

internal/verifier/change_reader.go

@@ -31,14 +31,25 @@ const (
 	changeReaderCollectionName = "changeReader"
 )
 
+type readerCurrentTimes struct {
+	LastHandledTime   bson.Timestamp `json:"lastHandledTime"`
+	LastOperationTime bson.Timestamp `json:"lastOperationTime"`
+}
+
+func (rp readerCurrentTimes) Lag() time.Duration {
+	return time.Second * time.Duration(
+		int(rp.LastOperationTime.T)-int(rp.LastHandledTime.T),
+	)
+}
+
 type changeReader interface {
 	getWhichCluster() whichCluster
 	getReadChannel() <-chan changeEventBatch
 	getEventRecorder() *EventRecorder
 	getStartTimestamp() bson.Timestamp
 	getLastSeenClusterTime() option.Option[bson.Timestamp]
 	getEventsPerSecond() option.Option[float64]
-	getLag() option.Option[time.Duration]
+	getCurrentTimes() option.Option[readerCurrentTimes]
 	getBufferSaturation() float64
 	setWritesOff(bson.Timestamp)
 	start(context.Context, *errgroup.Group) error
@@ -67,9 +78,10 @@ type ChangeReaderCommon struct {
 
 	lastChangeEventTime *msync.TypedAtomic[option.Option[bson.Timestamp]]
 
+	currentTimes *msync.TypedAtomic[option.Option[readerCurrentTimes]]
+
 	startAtTs *bson.Timestamp
 
-	lag              *msync.TypedAtomic[option.Option[time.Duration]]
 	batchSizeHistory *history.History[int]
 
 	onDDLEvent ddlEventHandling
@@ -81,7 +93,7 @@ func newChangeReaderCommon(clusterName whichCluster) ChangeReaderCommon {
 		changeEventBatchChan: make(chan changeEventBatch, batchChanBufferSize),
 		eventRecorder:        NewEventRecorder(),
 		writesOffTs:          util.NewEventual[bson.Timestamp](),
-		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
+		currentTimes:         msync.NewTypedAtomic(option.None[readerCurrentTimes]()),
 		lastChangeEventTime:  msync.NewTypedAtomic(option.None[bson.Timestamp]()),
 		batchSizeHistory:     history.New[int](time.Minute),
 		onDDLEvent: lo.Ternary(
@@ -131,10 +143,8 @@ func (rc *ChangeReaderCommon) getBufferSaturation() float64 {
 	return util.DivideToF64(len(rc.changeEventBatchChan), cap(rc.changeEventBatchChan))
 }
 
-// getLag returns the observed change stream lag (i.e., the delta between
-// cluster time and the most-recently-seen change event).
-func (rc *ChangeReaderCommon) getLag() option.Option[time.Duration] {
-	return rc.lag.Load()
+func (rc *ChangeReaderCommon) getCurrentTimes() option.Option[readerCurrentTimes] {
+	return rc.currentTimes.Load()
 }
 
 // getEventsPerSecond returns the number of change events per second we’ve been
@@ -230,11 +240,19 @@ func (rc *ChangeReaderCommon) loadResumeToken(ctx context.Context) (option.Optio
 	return option.Some(token), nil
 }
 
-func (rc *ChangeReaderCommon) updateLag(sess *mongo.Session, token bson.Raw) {
+func (rc *ChangeReaderCommon) updateTimes(sess *mongo.Session, token bson.Raw) {
 	tokenTs, err := rc.resumeTokenTSExtractor(token)
 	if err == nil {
-		lagSecs := int64(sess.OperationTime().T) - int64(tokenTs.T)
-		rc.lag.Store(option.Some(time.Second * time.Duration(lagSecs)))
+		opTime := sess.OperationTime()
+
+		if opTime == nil {
+			panic("session operationTime is nil … did this get called prematurely?")
+		}
+
+		rc.currentTimes.Store(option.Some(readerCurrentTimes{
+			LastHandledTime:   tokenTs,
+			LastOperationTime: *opTime,
+		}))
 	} else {
 		rc.logger.Warn().
 			Err(err).

internal/verifier/change_stream.go

@@ -225,7 +225,7 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		eventsRead++
 	}
 
-	csr.updateLag(sess, cs.ResumeToken())
+	csr.updateTimes(sess, cs.ResumeToken())
 
 	if eventsRead == 0 {
 		ri.NoteSuccess("received an empty change stream response")

internal/verifier/change_stream_test.go

@@ -636,7 +636,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamLag() {
 				verifierRunner.AwaitGenerationEnd(),
 			)
 
-			return verifier.srcChangeReader.getLag().IsSome()
+			return verifier.srcChangeReader.getCurrentTimes().IsSome()
 		},
 		time.Minute,
 		100*time.Millisecond,
@@ -645,7 +645,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamLag() {
 	// NB: The lag will include whatever time elapsed above before
 	// verifier read the event, so it can be several seconds.
 	suite.Assert().Less(
-		verifier.srcChangeReader.getLag().MustGet(),
+		verifier.srcChangeReader.getCurrentTimes().MustGet().Lag(),
 		10*time.Minute,
 		"verifier lag is as expected",
 	)

internal/verifier/summary.go

@@ -586,10 +586,10 @@ func (verifier *Verifier) printChangeEventStatistics(builder io.Writer) {
 		if eventsPerSec, has := cluster.csReader.getEventsPerSecond().Get(); has {
 			var lagNote string
 
-			lag, hasLag := cluster.csReader.getLag().Get()
+			prog, hasProg := cluster.csReader.getCurrentTimes().Get()
 
-			if hasLag {
-				lagNote = fmt.Sprintf("lag: %s; ", reportutils.DurationToHMS(lag))
+			if hasProg {
+				lagNote = fmt.Sprintf("lag: %s; ", reportutils.DurationToHMS(prog.Lag()))
 			}
 
 			saturation := cluster.csReader.getBufferSaturation()
@@ -603,7 +603,7 @@ func (verifier *Verifier) printChangeEventStatistics(builder io.Writer) {
 				reportutils.FmtReal(100*saturation),
 			)
 
-			if hasLag && lag > lagWarnThreshold {
+			if hasProg && prog.Lag() > lagWarnThreshold {
 				fmt.Fprint(
 					builder,
 					"⚠️ Lag is excessive. Verification may fail. See documentation.\n",