Make Eventual Get() panic if the value is unset.

mongodb-labs · Dec 9, 2024 · 42376f5 · 42376f5
1 parent d11cf4a
commit 42376f5
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 17 deletions.
diff --git a/internal/util/eventual.go b/internal/util/eventual.go
@@ -31,13 +31,18 @@ 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] {
+// Get returns the Eventual’s value if it’s ready.
+// It panics otherwise.
+func (e *Eventual[T]) Get() T {
 	e.mux.RLock()
 	defer e.mux.RUnlock()
 
-	return e.val
+	val, has := e.val.Get()
+	if has {
+		return val
+	}
+
+	panic("Eventual's Get() called before value was ready.")
 }
 
 // Set sets the Eventual’s value. It may be called only once;

diff --git a/internal/util/eventual_test.go b/internal/util/eventual_test.go
@@ -2,17 +2,14 @@ package util
 
 import (
 	"time"
-
-	"github.com/10gen/migration-verifier/option"
 )
 
 func (s *UnitTestSuite) TestEventual() {
 	eventual := NewEventual[int]()
 
-	s.Assert().Equal(
-		option.None[int](),
-		eventual.Get(),
-		"Get() should return empty",
+	s.Assert().Panics(
+		func() { eventual.Get() },
+		"Get() should panic before the value is set",
 	)
 
 	select {
@@ -30,7 +27,7 @@ func (s *UnitTestSuite) TestEventual() {
 	}
 
 	s.Assert().Equal(
-		option.Some(123),
+		123,
 		eventual.Get(),
 		"Get() should return the value",
 	)

diff --git a/internal/verifier/change_stream.go b/internal/verifier/change_stream.go
@@ -366,7 +366,7 @@ func (csr *ChangeStreamReader) iterateChangeStream(
 		// This means we should exit rather than continue reading the change stream
 		// since there should be no more events.
 		case <-csr.writesOffTs.Ready():
-			writesOffTs := csr.writesOffTs.Get().MustGet()
+			writesOffTs := csr.writesOffTs.Get()
 
 			csr.logger.Debug().
 				Interface("writesOffTimestamp", writesOffTs).

diff --git a/internal/verifier/check.go b/internal/verifier/check.go
@@ -50,7 +50,7 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context, csr *ChangeSt
 	case <-ctx.Done():
 		return ctx.Err()
 	case <-csr.error.Ready():
-		err := csr.error.Get().MustGet()
+		err := csr.error.Get()
 		verifier.logger.Warn().Err(err).
 			Msgf("Received error from %s.", csr)
 		return err
@@ -89,10 +89,10 @@ func (verifier *Verifier) CheckWorker(ctxIn context.Context) error {
 	eg.Go(func() error {
 		select {
 		case <-verifier.srcChangeStreamReader.error.Ready():
-			err := verifier.srcChangeStreamReader.error.Get().MustGet()
+			err := verifier.srcChangeStreamReader.error.Get()
 			return errors.Wrapf(err, "%s failed", verifier.srcChangeStreamReader)
 		case <-verifier.dstChangeStreamReader.error.Ready():
-			err := verifier.dstChangeStreamReader.error.Get().MustGet()
+			err := verifier.dstChangeStreamReader.error.Get()
 			return errors.Wrapf(err, "%s failed", verifier.dstChangeStreamReader)
 		case <-ctx.Done():
 			return nil

diff --git a/internal/verifier/migration_verifier.go b/internal/verifier/migration_verifier.go
@@ -282,15 +282,15 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	// under the lock.
 	select {
 	case <-verifier.srcChangeStreamReader.error.Ready():
-		err := verifier.srcChangeStreamReader.error.Get().MustGet()
+		err := verifier.srcChangeStreamReader.error.Get()
 		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.error.Ready():
-		err := verifier.dstChangeStreamReader.error.Get().MustGet()
+		err := verifier.dstChangeStreamReader.error.Get()
 		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)