eventual feels better

Author: FGasper

internal/replay/replay.go

@@ -0,0 +1,51 @@
+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.
+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/replay/replay_test.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/10gen/migration-verifier/internal/keystring"
 	"github.com/10gen/migration-verifier/internal/logger"
-	"github.com/10gen/migration-verifier/internal/replay"
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/10gen/migration-verifier/msync"
@@ -69,8 +68,7 @@ type ChangeStreamReader struct {
 
 	changeStreamRunning  bool
 	changeEventBatchChan chan []ParsedEvent
-	writesOffTsChan      <-chan primitive.Timestamp
-	writesOffTsWriteFunc func(primitive.Timestamp)
+	writesOffTs          *util.Eventual[primitive.Timestamp]
 	errChan              chan error
 	doneChan             chan struct{}
 
@@ -80,7 +78,6 @@ type ChangeStreamReader struct {
 }
 
 func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
-	srcWritesOffReaderChan, srcWritesOffWriter := replay.CreateChannel[primitive.Timestamp](ctx)
 
 	verifier.srcChangeStreamReader = &ChangeStreamReader{
 		readerType:           src,
@@ -91,15 +88,12 @@ func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
 		clusterInfo:          *verifier.srcClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      srcWritesOffReaderChan,
-		writesOffTsWriteFunc: srcWritesOffWriter,
+		writesOffTs:          util.NewEventual[primitive.Timestamp](),
 		errChan:              make(chan error),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
 	}
 
-	dstWritesOffReaderChan, dstWritesOffWriter := replay.CreateChannel[primitive.Timestamp](ctx)
-
 	verifier.dstChangeStreamReader = &ChangeStreamReader{
 		readerType:           dst,
 		logger:               verifier.logger,
@@ -109,8 +103,7 @@ func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
 		clusterInfo:          *verifier.dstClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      dstWritesOffReaderChan,
-		writesOffTsWriteFunc: dstWritesOffWriter,
+		writesOffTs:          util.NewEventual[primitive.Timestamp](),
 		errChan:              make(chan error),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
@@ -363,10 +356,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, more := <-csr.writesOffTsChan:
-			if !more {
-				panic(csr.String() + ": attempted read from already-closed 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)

internal/util/eventual.go

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

internal/verifier/change_stream.go

@@ -284,14 +284,14 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	case err := <-verifier.srcChangeStreamReader.errChan:
 		return errors.Wrapf(err, "tried to send writes-off timestamp to %s, but change stream already failed", verifier.srcChangeStreamReader)
 	default:
-		verifier.srcChangeStreamReader.writesOffTsWriteFunc(srcFinalTs)
+		verifier.srcChangeStreamReader.writesOffTs.Set(srcFinalTs)
 	}
 
 	select {
 	case err := <-verifier.dstChangeStreamReader.errChan:
 		return errors.Wrapf(err, "tried to send writes-off timestamp to %s, but change stream already failed", verifier.dstChangeStreamReader)
 	default:
-		verifier.dstChangeStreamReader.writesOffTsWriteFunc(dstFinalTs)
+		verifier.dstChangeStreamReader.writesOffTs.Set(dstFinalTs)
 	}
 
 	return nil