call IterationSuccess()

Author: FGasper

internal/verifier/change_stream.go

@@ -140,6 +140,7 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 // shouldn’t really happen anyway by definition.
 func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 	sctx mongo.SessionContext,
+	ri *retry.Info,
 	cs *mongo.ChangeStream,
 ) error {
 	eventsRead := 0
@@ -167,6 +168,8 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 		eventsRead++
 	}
 
+	ri.IterationSuccess()
+
 	if eventsRead == 0 {
 		return nil
 	}
@@ -187,7 +190,11 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 	return nil
 }
 
-func (verifier *Verifier) iterateChangeStream(sctx mongo.SessionContext, cs *mongo.ChangeStream) error {
+func (verifier *Verifier) iterateChangeStream(
+	sctx mongo.SessionContext,
+	ri *retry.Info,
+	cs *mongo.ChangeStream,
+) error {
 	var lastPersistedTime time.Time
 
 	persistResumeTokenIfNeeded := func() error {
@@ -243,15 +250,15 @@ func (verifier *Verifier) iterateChangeStream(sctx mongo.SessionContext, cs *mon
 					break
 				}
 
-				err = verifier.readAndHandleOneChangeEventBatch(sctx, cs)
+				err = verifier.readAndHandleOneChangeEventBatch(sctx, ri, cs)
 
 				if err != nil {
 					return err
 				}
 			}
 
 		default:
-			err = verifier.readAndHandleOneChangeEventBatch(sctx, cs)
+			err = verifier.readAndHandleOneChangeEventBatch(sctx, ri, cs)
 
 			if err == nil {
 				err = persistResumeTokenIfNeeded()
@@ -372,7 +379,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 			RunForTransientErrorsOnly(
 				ctx,
 				verifier.logger,
-				func(i *retry.Info) error {
+				func(ri *retry.Info) error {
 					sess, err := verifier.srcClient.StartSession()
 					if err != nil {
 						return errors.Wrap(err, "failed to start change stream session")
@@ -393,7 +400,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 						parentThreadWaiting = false
 					}
 
-					return verifier.iterateChangeStream(sctx, srcChangeStream)
+					return verifier.iterateChangeStream(sctx, ri, srcChangeStream)
 				},
 			)