eventual for change stream error too

Author: FGasper

internal/util/eventual.go

@@ -6,8 +6,12 @@ import (
 	"github.com/10gen/migration-verifier/option"
 )
 
-// Eventual represents a value that isn’t available when
-// this struct is created but can be awaited via a channel.
+// Eventual represents a value that isn’t available when this struct is created
+// but can be awaited via a channel.
+//
+// This is much like how context.Context’s Done() and Err() methods work.
+// It’s useful to await a value’s readiness via channel but then read it
+// multiple times.
 type Eventual[T any] struct {
 	ready chan struct{}
 	val   option.Option[T]

internal/verifier/change_stream.go

@@ -69,7 +69,7 @@ type ChangeStreamReader struct {
 	changeStreamRunning  bool
 	changeEventBatchChan chan []ParsedEvent
 	writesOffTs          *util.Eventual[primitive.Timestamp]
-	errChan              chan error
+	error                *util.Eventual[error]
 	doneChan             chan struct{}
 
 	startAtTs *primitive.Timestamp
@@ -89,7 +89,7 @@ func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
 		writesOffTs:          util.NewEventual[primitive.Timestamp](),
-		errChan:              make(chan error),
+		error:                util.NewEventual[error](),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
 	}
@@ -104,7 +104,7 @@ func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
 		writesOffTs:          util.NewEventual[primitive.Timestamp](),
-		errChan:              make(chan error),
+		error:                util.NewEventual[error](),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
 	}
@@ -548,8 +548,7 @@ func (csr *ChangeStreamReader) StartChangeStream(ctx context.Context) error {
 		if err != nil {
 			// NB: This failure always happens after the initial change stream
 			// creation.
-			csr.errChan <- err
-			close(csr.errChan)
+			csr.error.Set(err)
 		}
 	}()
 

internal/verifier/check.go

@@ -49,7 +49,9 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context, csr *ChangeSt
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case err := <-csr.errChan:
+	case <-csr.error.Ready():
+		err := csr.error.Get().MustGet()
+
 		verifier.logger.Warn().Err(err).
 			Msgf("Received error from %s.", csr)
 		return err
@@ -84,12 +86,14 @@ func (verifier *Verifier) CheckWorker(ctxIn context.Context) error {
 	cancelableCtx, canceler := context.WithCancelCause(ctxIn)
 	eg, ctx := errgroup.WithContext(cancelableCtx)
 
-	// If the change stream fails, everything should stop.
+	// If a change stream fails, everything should stop.
 	eg.Go(func() error {
 		select {
-		case err := <-verifier.srcChangeStreamReader.errChan:
+		case <-verifier.srcChangeStreamReader.error.Ready():
+			err := verifier.srcChangeStreamReader.error.Get().MustGet()
 			return errors.Wrapf(err, "%s failed", verifier.srcChangeStreamReader)
-		case err := <-verifier.dstChangeStreamReader.errChan:
+		case <-verifier.dstChangeStreamReader.error.Ready():
+			err := verifier.dstChangeStreamReader.error.Get().MustGet()
 			return errors.Wrapf(err, "%s failed", verifier.dstChangeStreamReader)
 		case <-ctx.Done():
 			return nil

internal/verifier/migration_verifier.go

@@ -281,14 +281,16 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	// might be inserting docs into the recheck queue, which happens
 	// under the lock.
 	select {
-	case err := <-verifier.srcChangeStreamReader.errChan:
+	case <-verifier.srcChangeStreamReader.error.Ready():
+		err := verifier.srcChangeStreamReader.error.Get().MustGet()
 		return errors.Wrapf(err, "tried to send writes-off timestamp to %s, but change stream already failed", verifier.srcChangeStreamReader)
 	default:
 		verifier.srcChangeStreamReader.writesOffTs.Set(srcFinalTs)
 	}
 
 	select {
-	case err := <-verifier.dstChangeStreamReader.errChan:
+	case <-verifier.dstChangeStreamReader.error.Ready():
+		err := verifier.dstChangeStreamReader.error.Get().MustGet()
 		return errors.Wrapf(err, "tried to send writes-off timestamp to %s, but change stream already failed", verifier.dstChangeStreamReader)
 	default:
 		verifier.dstChangeStreamReader.writesOffTs.Set(dstFinalTs)