tweaks

internal/retry/retry_info.go

@@ -1,6 +1,7 @@
 package retry
 
 import (
+	"fmt"
 	"slices"
 	"time"
 
@@ -89,11 +90,11 @@ func (fi *FuncInfo) GetDurationSoFar() time.Duration {
 //
 // Call this after every successful command in a multi-command callback.
 // (It’s useless--but harmless--in a single-command callback.)
-func (i *FuncInfo) NoteSuccess(description string) {
+func (i *FuncInfo) NoteSuccess(description string, descArgs ...any) {
 	totalResets := i.lastReset.Load().resetsSoFar
 
 	i.lastReset.Store(lastResetInfo{
-		description: option.Some(description),
+		description: option.Some(fmt.Sprintf(description, descArgs...)),
 		time:        time.Now(),
 		resetsSoFar: 1 + totalResets,
 	})

internal/verifier/change_stream.go

@@ -282,7 +282,11 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		}
 
 		if changeEventBatch == nil {
-			changeEventBatch = make([]ParsedEvent, cs.RemainingBatchLength()+1)
+			batchSize := cs.RemainingBatchLength() + 1
+
+			ri.NoteSuccess("received a batch of %d change event(s)", batchSize)
+
+			changeEventBatch = make([]ParsedEvent, batchSize)
 		}
 
 		if err := cs.Decode(&changeEventBatch[eventsRead]); err != nil {
@@ -303,9 +307,9 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		eventsRead++
 	}
 
-	ri.NoteSuccess("received a batch of change events")
-
 	if eventsRead == 0 {
+		ri.NoteSuccess("received an empty change stream response")
+
 		return nil
 	}
 
@@ -432,7 +436,9 @@ func (csr *ChangeStreamReader) iterateChangeStream(
 		infoLog = infoLog.Interface("lastEventTime", *csr.lastChangeEventTime)
 	}
 
-	infoLog.Msg("Change stream is done.")
+	infoLog.
+		Stringer("reader", csr).
+		Msg("Change stream reader is done.")
 
 	return nil
 }