Prevent WritesOff from hanging when the change stream has failed.

Author: FGasper

internal/verifier/change_stream_test.go

@@ -331,9 +331,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/check_runner.go

@@ -2,6 +2,7 @@ package verifier
 
 import (
 	"context"
+	"fmt"
 	"testing"
 )
 
@@ -49,7 +50,9 @@ func (cr *CheckRunner) Await() error {
 			return err
 
 		case <-cr.generationDoneChan:
+			fmt.Printf("===== verifier finished a generation\n")
 		case cr.doNextGenerationChan <- struct{}{}:
+			fmt.Printf("===== starting next generation\n")
 		}
 	}
 }

internal/verifier/migration_verifier.go

@@ -259,7 +259,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()
 	}