Cache multiple change event batches at once (#156)

This lets the verifier read the change stream & persist rechecks concurrently.

This also fixes a few minor formatting issues (along with adding the buffer-saturation to the log) and reduces the lag threshold from 5min to 2min.

The updated log format is:
```
Source change events this generation: 312,966 total, across 1 namespace(s)
Source: 45,395.47 writes per second (lag: 8m 35s; buffer 0% full)
⚠️ Lag is excessive. Verification may fail. See documentation.

Destination change events this generation: 258,761 total, across 1 namespace(s)
Destination: 40,404.2 writes per second (lag: 7m 40s; buffer 97% full)
⚠️ Lag is excessive. Verification may fail. See documentation.
```

Author: FGasper

internal/verifier/change_stream.go

@@ -31,6 +31,9 @@ type ddlEventHandling string
 const (
 	fauxDocSizeForDeleteEvents = 1024
 
+	// The number of batches we’ll hold in memory at once.
+	batchChanBufferSize = 100
+
 	onDDLEventAllow ddlEventHandling = "allow"
 )
 
@@ -108,7 +111,7 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 	for _, csr := range mslices.Of(srcReader, dstReader) {
 		csr.logger = verifier.logger
 		csr.metaDB = verifier.metaClient.Database(verifier.metaDBName)
-		csr.changeEventBatchChan = make(chan changeEventBatch)
+		csr.changeEventBatchChan = make(chan changeEventBatch, batchChanBufferSize)
 		csr.writesOffTs = util.NewEventual[bson.Timestamp]()
 		csr.readerError = util.NewEventual[error]()
 		csr.handlerError = util.NewEventual[error]()
@@ -770,10 +773,22 @@ func (csr *ChangeStreamReader) StartChangeStream(ctx context.Context) error {
 	return nil
 }
 
+// GetLag returns the observed change stream lag (i.e., the delta between
+// cluster time and the most-recently-seen change event).
 func (csr *ChangeStreamReader) GetLag() option.Option[time.Duration] {
 	return csr.lag.Load()
 }
 
+// GetSaturation returns the reader’s internal buffer’s saturation level as
+// a fraction. If saturation rises, that means we’re reading events faster than
+// we can persist them.
+func (csr *ChangeStreamReader) GetSaturation() float64 {
+	return util.DivideToF64(len(csr.changeEventBatchChan), cap(csr.changeEventBatchChan))
+}
+
+// GetEventsPerSecond returns the number of change events per second we’ve been
+// seeing “recently”. (See implementation for the actual period over which we
+// compile this metric.)
 func (csr *ChangeStreamReader) GetEventsPerSecond() option.Option[float64] {
 	logs := csr.batchSizeHistory.Get()
 	lastLog, hasLogs := lo.Last(logs)

internal/verifier/summary.go

@@ -21,7 +21,11 @@ import (
 	"golang.org/x/exp/maps"
 )
 
-const changeEventsTableMaxSize = 10
+const (
+	changeEventsTableMaxSize = 10
+
+	lagWarnThreshold = 2 * time.Minute
+)
 
 // NOTE: Each of the following should print one trailing and one final
 // newline.
@@ -516,6 +520,8 @@ func (verifier *Verifier) printMismatchInvestigationNotes(strBuilder *strings.Bu
 func (verifier *Verifier) printChangeEventStatistics(builder io.Writer) {
 	var eventsTable *tablewriter.Table
 
+	fmt.Fprint(builder, "\n")
+
 	for _, cluster := range []struct {
 		title         string
 		eventRecorder *EventRecorder
@@ -544,35 +550,38 @@ func (verifier *Verifier) printChangeEventStatistics(builder io.Writer) {
 			)
 		}
 
-		fmt.Fprintf(builder, "\n%s change events this generation: %s\n", cluster.title, eventsDescr)
+		fmt.Fprintf(builder, "%s change events this generation: %s\n", cluster.title, eventsDescr)
 
 		if eventsPerSec, has := cluster.csReader.GetEventsPerSecond().Get(); has {
 			var lagNote string
 
 			lag, hasLag := cluster.csReader.GetLag().Get()
 
 			if hasLag {
-				lagNote = fmt.Sprintf(" (lag: %s)", reportutils.DurationToHMS(lag))
+				lagNote = fmt.Sprintf("lag: %s; ", reportutils.DurationToHMS(lag))
 			}
 
 			fmt.Fprintf(
 				builder,
-				"%s observed change rate: %s/sec%s",
+				"%s: %s writes per second (%sbuffer %s%% full)\n",
 				cluster.title,
 				reportutils.FmtReal(eventsPerSec),
 				lagNote,
+				reportutils.FmtReal(100*cluster.csReader.GetSaturation()),
 			)
 
-			const lagWarnThreshold = 5 * time.Minute
-
 			if hasLag && lag > lagWarnThreshold {
 				fmt.Fprintf(
 					builder,
-					"⚠️ Lag is excessive. Verification may fail. See documentation.",
+					"⚠️ Lag is excessive. Verification may fail. See documentation.\n",
 				)
 			}
 		}
 
+		if cluster.csReader == verifier.srcChangeStreamReader {
+			fmt.Fprint(builder, "\n")
+		}
+
 		// We only print event breakdowns for the source because we assume that
 		// events on the destination will largely mirror the source’s.
 		if totalEvents > 0 && cluster.csReader == verifier.srcChangeStreamReader {