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/cluster_time.go

@@ -0,0 +1,24 @@
+package util
+
+import (
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+func GetClusterTimeFromSession(sess mongo.Session) (primitive.Timestamp, error) {
+	ctStruct := struct {
+		ClusterTime struct {
+			ClusterTime primitive.Timestamp `bson:"clusterTime"`
+		} `bson:"$clusterTime"`
+	}{}
+
+	clusterTimeRaw := sess.ClusterTime()
+	err := bson.Unmarshal(sess.ClusterTime(), &ctStruct)
+	if err != nil {
+		return primitive.Timestamp{}, errors.Wrapf(err, "failed to find clusterTime in session cluster time document (%v)", clusterTimeRaw)
+	}
+
+	return ctStruct.ClusterTime.ClusterTime, 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

@@ -44,6 +44,7 @@ type DocKey struct {
 
 const (
 	minChangeStreamPersistInterval     = time.Second * 10
+	maxChangeStreamAwaitTime           = time.Second
 	metadataChangeStreamCollectionName = "changeStream"
 )
 
@@ -68,8 +69,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 +88,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 +102,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 +124,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
 			}
 		}
@@ -268,6 +268,8 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 	eventsRead := 0
 	var changeEventBatch []ParsedEvent
 
+	latestEvent := option.None[ParsedEvent]()
+
 	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
 		gotEvent := cs.TryNext(ctx)
 
@@ -293,7 +295,9 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		if changeEventBatch[eventsRead].ClusterTime != nil &&
 			(csr.lastChangeEventTime == nil ||
 				csr.lastChangeEventTime.Before(*changeEventBatch[eventsRead].ClusterTime)) {
+
 			csr.lastChangeEventTime = changeEventBatch[eventsRead].ClusterTime
+			latestEvent = option.Some(changeEventBatch[eventsRead])
 		}
 
 		eventsRead++
@@ -305,6 +309,12 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		return nil
 	}
 
+	if event, has := latestEvent.Get(); has {
+		csr.logger.Trace().
+			Interface("event", event).
+			Msg("Updated lastChangeEventTime.")
+	}
+
 	var curTs primitive.Timestamp
 	curTs, err := extractTimestampFromResumeToken(cs.ResumeToken())
 	if err == nil {
@@ -355,7 +365,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 +420,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
 		}
 	}
@@ -430,7 +442,7 @@ func (csr *ChangeStreamReader) createChangeStream(
 ) (*mongo.ChangeStream, mongo.Session, primitive.Timestamp, error) {
 	pipeline := csr.GetChangeStreamFilter()
 	opts := options.ChangeStream().
-		SetMaxAwaitTime(1 * time.Second).
+		SetMaxAwaitTime(maxChangeStreamAwaitTime).
 		SetFullDocument(options.UpdateLookup)
 
 	if csr.clusterInfo.VersionArray[0] >= 6 {
@@ -487,11 +499,17 @@ func (csr *ChangeStreamReader) createChangeStream(
 	// With sharded clusters the resume token might lead the cluster time
 	// by 1 increment. In that case we need the actual cluster time;
 	// otherwise we will get errors.
-	clusterTime, err := getClusterTimeFromSession(sess)
+	clusterTime, err := util.GetClusterTimeFromSession(sess)
 	if err != nil {
 		return nil, nil, primitive.Timestamp{}, errors.Wrap(err, "failed to read cluster time from session")
 	}
 
+	csr.logger.Debug().
+		Interface("resumeTokenTimestamp", startTs).
+		Interface("clusterTime", clusterTime).
+		Stringer("changeStreamReader", csr).
+		Msg("Using earlier time as start timestamp.")
+
 	if startTs.After(clusterTime) {
 		startTs = clusterTime
 	}
@@ -542,10 +560,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)
 		}
 	}()
 
@@ -661,19 +676,3 @@ func extractTimestampFromResumeToken(resumeToken bson.Raw) (primitive.Timestamp,
 
 	return resumeTokenTime, nil
 }
-
-func getClusterTimeFromSession(sess mongo.Session) (primitive.Timestamp, error) {
-	ctStruct := struct {
-		ClusterTime struct {
-			ClusterTime primitive.Timestamp `bson:"clusterTime"`
-		} `bson:"$clusterTime"`
-	}{}
-
-	clusterTimeRaw := sess.ClusterTime()
-	err := bson.Unmarshal(sess.ClusterTime(), &ctStruct)
-	if err != nil {
-		return primitive.Timestamp{}, errors.Wrapf(err, "failed to find clusterTime in session cluster time document (%v)", clusterTimeRaw)
-	}
-
-	return ctStruct.ClusterTime.ClusterTime, nil
-}