detect hang & fail

Author: FGasper

internal/verifier/change_stream.go

@@ -155,7 +155,18 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		}
 
 		if err != nil && !errors.Is(err, context.Canceled) {
-			verifier.changeStreamErrChan <- err
+			timeout := time.Minute
+			timer := time.NewTimer(timeout)
+			defer timer.Stop()
+
+			select {
+			case <-timer.C:
+				verifier.logger.Fatal().
+					Err(err).
+					Stringer("timeout", timeout).
+					Msg("Failed to send change stream err within timeout.")
+			case verifier.changeStreamErrChan <- err:
+			}
 
 			if !changeStreamEnded {
 				return