REP-6804 Simplify change handling logic (#173)

Previously the code maintained:
- an Eventual for the error in persisting rechecks
- an Eventual for the error in reading changes
- an empty channel to indicate when the change reader is done

This is overly complex.

This changeset simplifies the above. Now there is a single Eventual that holds the final status of both readers & both persistors. Together the reader & persistor threads are called “change handling”.

This also fixes a small bug in Eventual: previously it was impossible to store a nil value because Eventual was implemented via Option. Now it’s possible to store nil.

Author: FGasper

internal/util/eventual.go

@@ -2,8 +2,6 @@ package util
 
 import (
 	"sync"
-
-	"github.com/10gen/migration-verifier/option"
 )
 
 // Eventual solves the “one writer, many readers” problem: a value gets
@@ -14,7 +12,7 @@ import (
 // generalized to any data type.
 type Eventual[T any] struct {
 	ready chan struct{}
-	val   option.Option[T]
+	val   T
 	mux   sync.RWMutex
 }
 
@@ -37,12 +35,14 @@ func (e *Eventual[T]) Get() T {
 	e.mux.RLock()
 	defer e.mux.RUnlock()
 
-	val, has := e.val.Get()
-	if has {
-		return val
+	// If the ready channel is still open then there’s no value yet,
+	// which means this method should not have been called.
+	select {
+	case <-e.ready:
+		return e.val
+	default:
+		panic("Eventual's Get() called before value was ready.")
 	}
-
-	panic("Eventual's Get() called before value was ready.")
 }
 
 // Set sets the Eventual’s value. It may be called only once;
@@ -51,13 +51,16 @@ func (e *Eventual[T]) Set(val T) {
 	e.mux.Lock()
 	defer e.mux.Unlock()
 
-	if e.val.IsSome() {
+	select {
+	case <-e.ready:
 		panic("Tried to set an eventual twice!")
+	default:
 	}
 
 	// NB: This *must* happen before the close(), or else a fast reader may
 	// not see this value.
-	e.val = option.Some(val)
+	e.val = val
 
+	// This allows Get() to work:
 	close(e.ready)
 }

internal/util/eventual_test.go

@@ -15,14 +15,14 @@ func (s *UnitTestSuite) TestEventual() {
 	select {
 	case <-eventual.Ready():
 		s.Require().Fail("should not be ready")
-	case <-time.NewTimer(time.Second).C:
+	case <-time.NewTimer(time.Millisecond).C:
 	}
 
 	eventual.Set(123)
 
 	select {
 	case <-eventual.Ready():
-	case <-time.NewTimer(time.Second).C:
+	case <-time.NewTimer(time.Millisecond).C:
 		s.Require().Fail("should be ready")
 	}
 
@@ -31,4 +31,30 @@ func (s *UnitTestSuite) TestEventual() {
 		eventual.Get(),
 		"Get() should return the value",
 	)
+
+	s.Assert().Equal(
+		123,
+		eventual.Get(),
+		"Get() should return the value a 2nd time",
+	)
+}
+
+func (s *UnitTestSuite) TestEventualNil() {
+	eventual := NewEventual[error]()
+
+	select {
+	case <-eventual.Ready():
+		s.Require().Fail("should not be ready")
+	case <-time.NewTimer(time.Millisecond).C:
+	}
+
+	eventual.Set(nil)
+
+	select {
+	case <-eventual.Ready():
+	case <-time.NewTimer(time.Millisecond).C:
+		s.Require().Fail("should be ready")
+	}
+
+	s.Assert().Nil(eventual.Get())
 }

internal/verifier/change_reader.go

@@ -15,6 +15,7 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/mongo/options"
+	"golang.org/x/sync/errgroup"
 )
 
 type ddlEventHandling string
@@ -33,15 +34,12 @@ const (
 type changeReader interface {
 	getWhichCluster() whichCluster
 	getReadChannel() <-chan changeEventBatch
-	getError() *util.Eventual[error]
 	getStartTimestamp() option.Option[bson.Timestamp]
 	getEventsPerSecond() option.Option[float64]
 	getLag() option.Option[time.Duration]
 	getBufferSaturation() float64
 	setWritesOff(bson.Timestamp)
-	setPersistorError(error)
-	start(context.Context) error
-	done() <-chan struct{}
+	start(context.Context, *errgroup.Group) error
 	persistResumeToken(context.Context, bson.Raw) error
 	isRunning() bool
 	String() string
@@ -63,9 +61,6 @@ type ChangeReaderCommon struct {
 	running              bool
 	changeEventBatchChan chan changeEventBatch
 	writesOffTs          *util.Eventual[bson.Timestamp]
-	readerError          *util.Eventual[error]
-	persistorError       *util.Eventual[error]
-	doneChan             chan struct{}
 
 	startAtTs *bson.Timestamp
 
@@ -79,14 +74,6 @@ func (rc *ChangeReaderCommon) getWhichCluster() whichCluster {
 	return rc.readerType
 }
 
-func (rc *ChangeReaderCommon) setPersistorError(err error) {
-	rc.persistorError.Set(err)
-}
-
-func (rc *ChangeReaderCommon) getError() *util.Eventual[error] {
-	return rc.readerError
-}
-
 func (rc *ChangeReaderCommon) getStartTimestamp() option.Option[bson.Timestamp] {
 	return option.FromPointer(rc.startAtTs)
 }
@@ -103,10 +90,6 @@ func (rc *ChangeReaderCommon) getReadChannel() <-chan changeEventBatch {
 	return rc.changeEventBatchChan
 }
 
-func (rc *ChangeReaderCommon) done() <-chan struct{} {
-	return rc.doneChan
-}
-
 // getBufferSaturation returns the reader’s internal buffer’s saturation level
 // as a fraction. If saturation rises, that means we’re reading events faster
 // than we can persist them.
@@ -224,13 +207,6 @@ func (rc *ChangeReaderCommon) logIgnoredDDL(rawEvent bson.Raw) {
 		Msg("Ignoring event with unrecognized type on destination. (It’s assumedly internal to the migration.)")
 }
 
-func (rc *ChangeReaderCommon) wrapPersistorErrorForReader() error {
-	return errors.Wrap(
-		rc.persistorError.Get(),
-		"event persistor failed, so no more events can be processed",
-	)
-}
-
 func addTimestampToLogEvent(ts bson.Timestamp, event *zerolog.Event) *zerolog.Event {
 	return event.
 		Any("timestamp", ts).