only log once

Author: FGasper

internal/verifier/change_stream.go

@@ -44,11 +44,10 @@ type DocKey struct {
 
 const (
 	minChangeStreamPersistInterval     = time.Second * 10
+	maxChangeStreamAwaitTime           = time.Second
 	metadataChangeStreamCollectionName = "changeStream"
 )
 
-var maxChangeStreamAwaitTime = time.Second
-
 type UnknownEventError struct {
 	Event *ParsedEvent
 }
@@ -269,6 +268,8 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 	eventsRead := 0
 	var changeEventBatch []ParsedEvent
 
+	latestEvent := option.None[ParsedEvent]()
+
 	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
 		gotEvent := cs.TryNext(ctx)
 
@@ -294,10 +295,9 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		if changeEventBatch[eventsRead].ClusterTime != nil &&
 			(csr.lastChangeEventTime == nil ||
 				csr.lastChangeEventTime.Before(*changeEventBatch[eventsRead].ClusterTime)) {
+
 			csr.lastChangeEventTime = changeEventBatch[eventsRead].ClusterTime
-			csr.logger.Trace().
-				Interface("event", changeEventBatch[eventsRead]).
-				Msg("Updated lastChangeEventTime.")
+			latestEvent = option.Some(changeEventBatch[eventsRead])
 		}
 
 		eventsRead++
@@ -309,6 +309,12 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		return nil
 	}
 
+	if event, has := latestEvent.Get(); has {
+		csr.logger.Trace().
+			Interface("event", event).
+			Msg("Updated lastChangeEventTime.")
+	}
+
 	var curTs primitive.Timestamp
 	curTs, err := extractTimestampFromResumeToken(cs.ResumeToken())
 	if err == nil {