REP-5358 Fix hangs when reading change stream channels.

internal/testutil/testutil.go

@@ -1,8 +1,14 @@
 package testutil
 
 import (
+	"context"
+	"testing"
+
+	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
 )
 
 // Marshal wraps `bsonMarshal` with a panic on failure.
@@ -34,3 +40,64 @@ func convertDocsToAnys(docs []bson.D) []any {
 
 	return anys
 }
+
+func KillApplicationChangeStreams(
+	ctx context.Context,
+	t *testing.T,
+	client *mongo.Client,
+	appName string,
+) error {
+	// Kill verifier’s change stream.
+	cursor, err := client.Database(
+		"admin",
+		options.Database().SetReadConcern(readconcern.Local()),
+	).Aggregate(
+		ctx,
+		mongo.Pipeline{
+			{
+				{"$currentOp", bson.D{
+					{"idleCursors", true},
+				}},
+			},
+			{
+				{"$match", bson.D{
+					{"clientMetadata.application.name", appName},
+					{"command.collection", "$cmd.aggregate"},
+					{"cursor.originatingCommand.pipeline.0.$_internalChangeStreamOplogMatch",
+						bson.D{{"$type", "object"}},
+					},
+				}},
+			},
+		},
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to list %#q's change streams", appName)
+	}
+
+	ops := []struct {
+		Opid any
+	}{}
+	err = cursor.All(ctx, &ops)
+	if err != nil {
+		return errors.Wrapf(err, "failed to decode %#q's change streams", appName)
+	}
+
+	for _, op := range ops {
+		t.Logf("Killing change stream op %+v", op.Opid)
+
+		err :=
+			client.Database("admin").RunCommand(
+				ctx,
+				bson.D{
+					{"killOp", 1},
+					{"op", op.Opid},
+				},
+			).Err()
+
+		if err != nil {
+			return errors.Wrapf(err, "failed to kill change stream with opId %#q", op.Opid)
+		}
+	}
+
+	return nil
+}

internal/util/eventual.go

@@ -0,0 +1,58 @@
+package util
+
+import (
+	"sync"
+
+	"github.com/10gen/migration-verifier/option"
+)
+
+// Eventual solves the “one writer, many readers” problem: a value gets
+// written once, then the readers will see that the value is `Ready()` and
+// can then `Get()` it.
+//
+// It’s like how `context.Context`’s `Done()` and `Err()` methods work, but
+// generalized to any data type.
+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 sets the Eventual’s value. It may be called only once;
+// if called again it will panic.
+func (e *Eventual[T]) Set(val T) {
+	e.mux.Lock()
+	defer e.mux.Unlock()
+
+	if e.val.IsSome() {
+		panic("Tried to set an eventual twice!")
+	}
+
+	// NB: This *must* happen before the close(), or else a fast reader may
+	// not see this value.
+	e.val = option.Some(val)
+
+	close(e.ready)
+}

internal/util/eventual_test.go

@@ -0,0 +1,37 @@
+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",
+	)
+
+	select {
+	case <-eventual.Ready():
+		s.Require().Fail("should not be ready")
+	case <-time.NewTimer(time.Second).C:
+	}
+
+	eventual.Set(123)
+
+	select {
+	case <-eventual.Ready():
+	case <-time.NewTimer(time.Second).C:
+		s.Require().Fail("should be ready")
+	}
+
+	s.Assert().Equal(
+		option.Some(123),
+		eventual.Get(),
+		"Get() should return the value",
+	)
+}

internal/verifier/change_stream.go

@@ -68,8 +68,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
@@ -87,8 +87,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{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
 	}
@@ -101,8 +101,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{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
 	}
@@ -123,7 +123,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
 			}
 		}
@@ -355,7 +354,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)
@@ -408,7 +409,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
 		}
 	}
@@ -542,10 +543,7 @@ func (csr *ChangeStreamReader) StartChangeStream(ctx context.Context) error {
 		).Run(ctx, csr.logger)
 
 		if err != nil {
-			// NB: This failure always happens after the initial change stream
-			// creation.
-			csr.errChan <- err
-			close(csr.errChan)
+			csr.error.Set(err)
 		}
 	}()