Convert change stream err and writesOffTs channels to Eventuals.

Author: FGasper

internal/util/eventual.go

@@ -0,0 +1,55 @@
+package util
+
+import (
+	"sync"
+
+	"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.
+//
+// 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]
+	mux   sync.RWMutex
+}
+
+// NewEventual creates an Eventual and returns a pointer
+// to it.
+func NewEventual[T any]() *Eventual[T] {
+	return &Eventual[T]{
+		ready: make(chan struct{}),
+	}
+}
+
+// Ready returns a channel that closes once the Eventual’s value is ready.
+func (e *Eventual[T]) Ready() <-chan struct{} {
+	return e.ready
+}
+
+// Get returns an option that contains the Eventual’s value, or
+// empty if the value isn’t ready yet.
+func (e *Eventual[T]) Get() option.Option[T] {
+	e.mux.RLock()
+	defer e.mux.RUnlock()
+
+	return e.val
+}
+
+// Set
+func (e *Eventual[T]) Set(val T) {
+	e.mux.Lock()
+	defer e.mux.Unlock()
+
+	if e.val.IsSome() {
+		panic("Double set on eventual!")
+	}
+
+	e.val = option.Some(val)
+
+	close(e.ready)
+}

internal/verifier/change_stream.go

@@ -66,8 +66,8 @@ type ChangeStreamReader struct {
 
 	changeStreamRunning  bool
 	changeEventBatchChan chan []ParsedEvent
-	writesOffTsChan      chan primitive.Timestamp
-	errChan              chan error
+	writesOffTs          *util.Eventual[primitive.Timestamp]
+	error                *util.Eventual[error]
 	doneChan             chan struct{}
 
 	startAtTs *primitive.Timestamp
@@ -83,8 +83,8 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		clusterInfo:          *verifier.srcClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      make(chan primitive.Timestamp),
-		errChan:              make(chan error),
+		writesOffTs:          util.NewEventual[primitive.Timestamp](),
+		error:                util.NewEventual[error](),
 		doneChan:             make(chan struct{}),
 	}
 	verifier.dstChangeStreamReader = &ChangeStreamReader{
@@ -96,8 +96,8 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		clusterInfo:          *verifier.dstClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      make(chan primitive.Timestamp),
-		errChan:              make(chan error),
+		writesOffTs:          util.NewEventual[primitive.Timestamp](),
+		error:                util.NewEventual[error](),
 		doneChan:             make(chan struct{}),
 	}
 }
@@ -117,7 +117,6 @@ func (verifier *Verifier) StartChangeEventHandler(ctx context.Context, reader *C
 			verifier.logger.Trace().Msgf("Verifier is handling a change event batch from %s: %v", reader, batch)
 			err := verifier.HandleChangeStreamEvents(ctx, batch, reader.readerType)
 			if err != nil {
-				reader.errChan <- err
 				return err
 			}
 		}
@@ -336,7 +335,9 @@ func (csr *ChangeStreamReader) iterateChangeStream(
 		// source writes are ended and the migration tool is finished / committed.
 		// This means we should exit rather than continue reading the change stream
 		// since there should be no more events.
-		case writesOffTs := <-csr.writesOffTsChan:
+		case <-csr.writesOffTs.Ready():
+			writesOffTs := csr.writesOffTs.Get().MustGet()
+
 			csr.logger.Debug().
 				Interface("writesOffTimestamp", writesOffTs).
 				Msgf("%s thread received writesOff timestamp. Finalizing change stream.", csr)
@@ -389,7 +390,7 @@ func (csr *ChangeStreamReader) iterateChangeStream(
 			}
 			// since we have started Recheck, we must signal that we have
 			// finished the change stream changes so that Recheck can continue.
-			csr.doneChan <- struct{}{}
+			close(csr.doneChan)
 			break
 		}
 	}
@@ -525,10 +526,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/change_stream_test.go

@@ -319,7 +319,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeNoChanges() {
 	suite.Require().NotNil(origStartTs)
 	suite.startSrcChangeStreamReaderAndHandler(ctx, verifier)
 	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
-	verifier.srcChangeStreamReader.writesOffTsChan <- *origStartTs
+	verifier.srcChangeStreamReader.writesOffTs.Set(*origStartTs)
 	<-verifier.srcChangeStreamReader.doneChan
 	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
 }
@@ -365,7 +365,7 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 		"session time after events should exceed the original",
 	)
 
-	verifier.srcChangeStreamReader.writesOffTsChan <- *postEventsSessionTime
+	verifier.srcChangeStreamReader.writesOffTs.Set(*postEventsSessionTime)
 	<-verifier.srcChangeStreamReader.doneChan
 
 	suite.Assert().Equal(

internal/verifier/check.go

@@ -49,7 +49,8 @@ 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
@@ -87,9 +88,11 @@ func (verifier *Verifier) CheckWorker(ctxIn context.Context) error {
 	// If the 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

@@ -279,15 +279,19 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	// might be inserting docs into the recheck queue, which happens
 	// under the lock.
 	select {
-	case verifier.srcChangeStreamReader.writesOffTsChan <- srcFinalTs:
-	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 verifier.dstChangeStreamReader.writesOffTsChan <- dstFinalTs:
-	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)
 	}
 
 	return nil