Prevent WritesOff from hanging when the change stream has failed.

Author: FGasper

internal/verifier/change_stream_test.go

@@ -344,9 +344,13 @@ func (suite *IntegrationTestSuite) TestCreateForbidden() {
 		db.CreateCollection(ctx, coll.Name()),
 	)
 
-	suite.Require().NoError(verifier.WritesOff(ctx))
+	// The error from the create event will come either at WritesOff
+	// or when we finalize the change stream.
+	err = verifier.WritesOff(ctx)
+	if err == nil {
+		err = verifierRunner.Await()
+	}
 
-	err = verifierRunner.Await()
 	suite.Require().Error(err, "should detect forbidden create event")
 
 	eventErr := UnknownEventError{}

internal/verifier/check.go

@@ -236,6 +236,9 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		// paying attention. Also, this should not matter too much because any failures will be
 		// caught again on the next iteration.
 		if verifier.writesOff {
+			verifier.logger.Debug().
+				Msg("Waiting for change stream to end.")
+
 			// It's necessary to wait for the change stream to finish before incrementing the
 			// generation number, or the last changes will not be checked.
 			verifier.mux.Unlock()

internal/verifier/migration_verifier.go

@@ -261,7 +261,11 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 		// This has to happen under the lock because the change stream
 		// might be inserting docs into the recheck queue, which happens
 		// under the lock.
-		verifier.changeStreamWritesOffTsChan <- finalTs
+		select {
+		case verifier.changeStreamWritesOffTsChan <- finalTs:
+		case err := <-verifier.changeStreamErrChan:
+			return errors.Wrap(err, "tried to send writes-off timestamp to change stream, but change stream already failed")
+		}
 	} else {
 		verifier.mux.Unlock()
 	}