reshuffle a bit for clarity

Author: FGasper

internal/verifier/change_stream.go

@@ -366,56 +366,60 @@ func (verifier *Verifier) createChangeStream(
 
 // StartChangeStream starts the change stream.
 func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
-	// Rather than Result we could have separate Timestamp and error channels,
-	// but the single channel seems simpler/cleaner.
-	resultChan := make(chan mo.Result[primitive.Timestamp])
+	// This channel holds the first change stream creation's result, whether
+	// success or failure. Rather than using a Result we could make separate
+	// Timestamp and error channels, but the single channel is cleaner since
+	// there's no chance of "nonsense" like both channels returning a payload.
+	initialCreateResultChan := make(chan mo.Result[primitive.Timestamp])
 
 	go func() {
 		retryer := retry.New(retry.DefaultDurationLimit)
 		retryer = retryer.WithErrorCodes(util.CursorKilled)
 
 		parentThreadWaiting := true
 
-		err := retryer.
-			RunForTransientErrorsOnly(
-				ctx,
-				verifier.logger,
-				func(ri *retry.Info) error {
-					sess, err := verifier.srcClient.StartSession()
-					if err != nil {
-						return errors.Wrap(err, "failed to start change stream session")
-					}
+		err := retryer.RunForTransientErrorsOnly(
+			ctx,
+			verifier.logger,
+			func(ri *retry.Info) error {
+				sess, err := verifier.srcClient.StartSession()
+				if err != nil {
+					return errors.Wrap(err, "failed to start change stream session")
+				}
 
-					sctx := mongo.NewSessionContext(ctx, sess)
+				sctx := mongo.NewSessionContext(ctx, sess)
 
-					srcChangeStream, startTs, err := verifier.createChangeStream(sctx)
-					if err != nil {
-						return err
+				srcChangeStream, startTs, err := verifier.createChangeStream(sctx)
+				if err != nil {
+					if parentThreadWaiting {
+						initialCreateResultChan <- mo.Err[primitive.Timestamp](err)
+						return nil
 					}
 
-					defer srcChangeStream.Close(ctx)
+					return err
+				}
 
-					if parentThreadWaiting {
-						resultChan <- mo.Ok(startTs)
-						close(resultChan)
-						parentThreadWaiting = false
-					}
+				defer srcChangeStream.Close(ctx)
+
+				if parentThreadWaiting {
+					initialCreateResultChan <- mo.Ok(startTs)
+					close(initialCreateResultChan)
+					parentThreadWaiting = false
+				}
 
-					return verifier.iterateChangeStream(sctx, ri, srcChangeStream)
-				},
-			)
+				return verifier.iterateChangeStream(sctx, ri, srcChangeStream)
+			},
+		)
 
 		if err != nil {
-			if parentThreadWaiting {
-				resultChan <- mo.Err[primitive.Timestamp](err)
-			} else {
-				verifier.changeStreamErrChan <- err
-				close(verifier.changeStreamErrChan)
-			}
+			// NB: This failure always happens after the initial change stream
+			// creation.
+			verifier.changeStreamErrChan <- err
+			close(verifier.changeStreamErrChan)
 		}
 	}()
 
-	result := <-resultChan
+	result := <-initialCreateResultChan
 
 	startTs, err := result.Get()
 	if err != nil {