writesoffts is now rereadable

Author: FGasper

internal/replay/replay.go

@@ -0,0 +1,42 @@
+package replay
+
+import "context"
+
+func CreateChannel[T any](
+	ctx context.Context,
+) (<-chan T, func(T)) {
+	retChan := make(chan T)
+
+	writer := make(chan T)
+
+	go func() {
+		defer close(retChan)
+
+		var val T
+		select {
+		case <-ctx.Done():
+			return
+		case val = <-writer:
+		}
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case retChan <- val:
+			}
+		}
+	}()
+
+	writerCalled := false
+
+	return retChan, func(val T) {
+		if writerCalled {
+			panic("replay-channel writer called twice")
+		}
+		writerCalled = true
+
+		writer <- val
+		close(writer)
+	}
+}

internal/replay/replay_test.go

@@ -0,0 +1,46 @@
+package replay
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/10gen/migration-verifier/internal/logger"
+	"github.com/stretchr/testify/suite"
+)
+
+type UnitTestSuite struct {
+	suite.Suite
+	logger *logger.Logger
+}
+
+func TestUnitTestSuite(t *testing.T) {
+	ts := new(UnitTestSuite)
+	suite.Run(t, ts)
+}
+
+func (s *UnitTestSuite) TestReplay() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	theChan, theWriter := CreateChannel[int](ctx)
+
+	select {
+	case <-theChan:
+		s.Require().Fail("should not be readable")
+	case <-time.NewTimer(time.Second).C:
+	}
+
+	theWriter(123)
+
+	s.Assert().Panics(func() { theWriter(234) })
+
+	for range 5 {
+		select {
+		case val := <-theChan:
+			s.Require().Equal(val, 123)
+		case <-time.NewTimer(time.Second).C:
+			s.Require().Fail("should be readable")
+		}
+	}
+}

internal/verifier/change_stream.go

@@ -7,6 +7,7 @@ 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"
@@ -68,7 +69,8 @@ type ChangeStreamReader struct {
 
 	changeStreamRunning  bool
 	changeEventBatchChan chan []ParsedEvent
-	writesOffTsChan      chan primitive.Timestamp
+	writesOffTsChan      <-chan primitive.Timestamp
+	writesOffTsWriteFunc func(primitive.Timestamp)
 	errChan              chan error
 	doneChan             chan struct{}
 
@@ -77,7 +79,9 @@ type ChangeStreamReader struct {
 	lag *msync.TypedAtomic[option.Option[time.Duration]]
 }
 
-func (verifier *Verifier) initializeChangeStreamReaders() {
+func (verifier *Verifier) initializeChangeStreamReaders(ctx context.Context) {
+	srcWritesOffReaderChan, srcWritesOffWriter := replay.CreateChannel[primitive.Timestamp](ctx)
+
 	verifier.srcChangeStreamReader = &ChangeStreamReader{
 		readerType:           src,
 		logger:               verifier.logger,
@@ -87,11 +91,15 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		clusterInfo:          *verifier.srcClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      make(chan primitive.Timestamp),
+		writesOffTsChan:      srcWritesOffReaderChan,
+		writesOffTsWriteFunc: srcWritesOffWriter,
 		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,
@@ -101,7 +109,8 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		clusterInfo:          *verifier.dstClusterInfo,
 		changeStreamRunning:  false,
 		changeEventBatchChan: make(chan []ParsedEvent),
-		writesOffTsChan:      make(chan primitive.Timestamp),
+		writesOffTsChan:      dstWritesOffReaderChan,
+		writesOffTsWriteFunc: dstWritesOffWriter,
 		errChan:              make(chan error),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
@@ -123,7 +132,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 +363,10 @@ 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 writesOffTs, more := <-csr.writesOffTsChan:
+			if !more {
+				panic(csr.String() + ": attempted read from already-closed writesOffTsChan")
+			}
 			csr.logger.Debug().
 				Interface("writesOffTimestamp", writesOffTs).
 				Msgf("%s thread received writesOff timestamp. Finalizing change stream.", csr)
@@ -408,7 +419,8 @@ 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
 		}
 	}

internal/verifier/change_stream_test.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.writesOffTsChan <- *origStartTs
+	verifier.srcChangeStreamReader.writesOffTsWriteFunc(*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.writesOffTsChan <- *postEventsSessionTime
+	verifier.srcChangeStreamReader.writesOffTsWriteFunc(*postEventsSessionTime)
 	<-verifier.srcChangeStreamReader.doneChan
 
 	suite.Assert().Equal(

internal/verifier/check.go

@@ -179,7 +179,7 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	}
 	verifier.running = true
 	verifier.globalFilter = filter
-	verifier.initializeChangeStreamReaders()
+	verifier.initializeChangeStreamReaders(ctx)
 	verifier.mux.Unlock()
 	defer func() {
 		verifier.mux.Lock()

internal/verifier/integration_test_suite.go

@@ -176,7 +176,7 @@ func (suite *IntegrationTestSuite) BuildVerifier() *Verifier {
 		"should set metadata connection string",
 	)
 	verifier.SetMetaDBName(metaDBName)
-	verifier.initializeChangeStreamReaders()
+	verifier.initializeChangeStreamReaders(ctx)
 
 	suite.Require().NoError(verifier.srcClientCollection(&task).Drop(ctx))
 	suite.Require().NoError(verifier.dstClientCollection(&task).Drop(ctx))

internal/verifier/migration_verifier.go

@@ -248,7 +248,6 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 		verifier.mux.Unlock()
 		return errors.New("writesOff already set")
 	}
-	verifier.writesOff = true
 
 	verifier.logger.Debug().Msg("Signalling that writes are done.")
 
@@ -273,21 +272,26 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 		verifier.mux.Unlock()
 		return errors.Wrapf(err, "failed to fetch destination's cluster time")
 	}
+
+	verifier.writesOff = true
+
 	verifier.mux.Unlock()
 
 	// This has to happen outside the lock because the change streams
 	// might be inserting docs into the recheck queue, which happens
 	// under the lock.
 	select {
-	case verifier.srcChangeStreamReader.writesOffTsChan <- srcFinalTs:
 	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)
 	}
 
 	select {
-	case verifier.dstChangeStreamReader.writesOffTsChan <- dstFinalTs:
 	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)
 	}
 
 	return nil