revert change_stream

Author: FGasper

internal/verifier/change_stream.go

@@ -25,7 +25,7 @@ type ParsedEvent struct {
 }
 
 func (pe *ParsedEvent) String() string {
-	return fmt.Sprintf("{OpType: %s, namespace: %s, docID: %v, clusterTime: %v}", pe.OpType, pe.Ns, pe.DocKey.ID, pe.ClusterTime)
+	return fmt.Sprintf("{ OpType: %s, namespace: %s, docID: %v}", pe.OpType, pe.Ns, pe.DocKey.ID)
 }
 
 // DocKey is a deserialized form for the ChangeEvent documentKey field. We currently only care about
@@ -44,7 +44,7 @@ type UnknownEventError struct {
 }
 
 func (uee UnknownEventError) Error() string {
-	return fmt.Sprintf("Received event with unknown optype: %+v", uee.Event)
+	return fmt.Sprintf("Unknown event type: %#q", uee.Event.OpType)
 }
 
 // HandleChangeStreamEvent performs the necessary work for change stream events that occur during
@@ -63,12 +63,6 @@ func (verifier *Verifier) HandleChangeStreamEvent(ctx context.Context, changeEve
 	case "replace":
 		fallthrough
 	case "update":
-		/*
-			if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
-				return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
-			}
-		*/
-
 		return verifier.InsertChangeEventRecheckDoc(ctx, changeEvent)
 	default:
 		return UnknownEventError{Event: changeEvent}
@@ -135,19 +129,24 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		// source writes are ended. This means we should exit rather than continue
 		// reading the change stream since there should be no more events.
 		case <-verifier.changeStreamEnderChan:
+			var gotEvent bool
+
 			changeStreamEnded = true
 
 			// Read all change events until the source reports no events.
 			// (i.e., the `getMore` call returns empty)
 			for {
-				var gotEvent bool
 				gotEvent, err = readOneChangeEvent()
 
 				if !gotEvent || err != nil {
 					break
 				}
 			}
 
+			if err != nil {
+				break
+			}
+
 		default:
 			_, err = readOneChangeEvent()
 		}
@@ -157,22 +156,7 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		}
 
 		if err != nil && !errors.Is(err, context.Canceled) {
-			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
-			}
+			verifier.changeStreamErrChan <- err
 		}
 
 		if changeStreamEnded {