debug assistance

internal/verifier/change_stream.go

@@ -294,7 +294,12 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		}
 
 		// This only logs in tests.
-		csr.logger.Trace().Interface("event", changeEventBatch[eventsRead]).Msgf("%s received a change event", csr)
+		csr.logger.Debug().
+			Stringer("changeStream", csr).
+			Interface("event", changeEventBatch[eventsRead]).
+			Int("eventsPreviouslyReadInBatch", eventsRead).
+			Int("batchSize", len(changeEventBatch)).
+			Msg("Received a change event.")
 
 		if changeEventBatch[eventsRead].ClusterTime != nil &&
 			(csr.lastChangeEventTime == nil ||
@@ -315,10 +320,13 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 
 	if event, has := latestEvent.Get(); has {
 		csr.logger.Trace().
+			Stringer("changeStream", csr).
 			Interface("event", event).
 			Msg("Updated lastChangeEventTime.")
 	}
 
+	ri.NoteSuccess("parsed %d-event batch", len(changeEventBatch))
+
 	var tokenTs primitive.Timestamp
 	tokenTs, err := extractTimestampFromResumeToken(cs.ResumeToken())
 	if err == nil {
@@ -336,6 +344,8 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 	case csr.changeEventBatchChan <- changeEventBatch:
 	}
 
+	ri.NoteSuccess("sent %d-event batch to handler", len(changeEventBatch))
+
 	return nil
 }