change to private fields

Author: tdq45gj

internal/verifier/change_stream.go

@@ -65,10 +65,10 @@ type ChangeStreamReader struct {
 	clusterInfo   util.ClusterInfo
 
 	changeStreamRunning  bool
-	ChangeEventBatchChan chan []ParsedEvent
-	WritesOffTsChan      chan primitive.Timestamp
-	ErrChan              chan error
-	DoneChan             chan struct{}
+	changeEventBatchChan chan []ParsedEvent
+	writesOffTsChan      chan primitive.Timestamp
+	errChan              chan error
+	doneChan             chan struct{}
 
 	startAtTs *primitive.Timestamp
 }
@@ -82,10 +82,10 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		watcherClient:        verifier.srcClient,
 		clusterInfo:          *verifier.srcClusterInfo,
 		changeStreamRunning:  false,
-		ChangeEventBatchChan: make(chan []ParsedEvent),
-		WritesOffTsChan:      make(chan primitive.Timestamp),
-		ErrChan:              make(chan error),
-		DoneChan:             make(chan struct{}),
+		changeEventBatchChan: make(chan []ParsedEvent),
+		writesOffTsChan:      make(chan primitive.Timestamp),
+		errChan:              make(chan error),
+		doneChan:             make(chan struct{}),
 	}
 	verifier.dstChangeStreamReader = &ChangeStreamReader{
 		readerType:           dst,
@@ -95,10 +95,10 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		watcherClient:        verifier.dstClient,
 		clusterInfo:          *verifier.dstClusterInfo,
 		changeStreamRunning:  false,
-		ChangeEventBatchChan: make(chan []ParsedEvent),
-		WritesOffTsChan:      make(chan primitive.Timestamp),
-		ErrChan:              make(chan error),
-		DoneChan:             make(chan struct{}),
+		changeEventBatchChan: make(chan []ParsedEvent),
+		writesOffTsChan:      make(chan primitive.Timestamp),
+		errChan:              make(chan error),
+		doneChan:             make(chan struct{}),
 	}
 }
 
@@ -109,15 +109,15 @@ func (verifier *Verifier) StartChangeEventHandler(ctx context.Context, reader *C
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case batch, more := <-reader.ChangeEventBatchChan:
+		case batch, more := <-reader.changeEventBatchChan:
 			if !more {
 				verifier.logger.Debug().Msgf("Change Event Batch Channel has been closed by %s, returning...", reader)
 				return nil
 			}
 			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
+				reader.errChan <- err
 				return err
 			}
 		}
@@ -295,7 +295,7 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		return nil
 	}
 
-	csr.ChangeEventBatchChan <- changeEventBatch
+	csr.changeEventBatchChan <- changeEventBatch
 	return nil
 }
 
@@ -333,7 +333,7 @@ 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 := <-csr.writesOffTsChan:
 			csr.logger.Debug().
 				Interface("writesOffTimestamp", writesOffTs).
 				Msgf("%s thread received writesOff timestamp. Finalizing change stream.", csr)
@@ -386,7 +386,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{}{}
+			csr.doneChan <- struct{}{}
 			break
 		}
 	}
@@ -486,9 +486,9 @@ func (csr *ChangeStreamReader) StartChangeStream(ctx context.Context) error {
 	initialCreateResultChan := make(chan mo.Result[primitive.Timestamp])
 
 	go func() {
-		// Closing ChangeEventBatchChan at the end of change stream goroutine
+		// Closing changeEventBatchChan at the end of change stream goroutine
 		// notifies the verifier's change event handler to exit.
-		defer close(csr.ChangeEventBatchChan)
+		defer close(csr.changeEventBatchChan)
 
 		retryer := retry.New(retry.DefaultDurationLimit)
 		retryer = retryer.WithErrorCodes(util.CursorKilled)
@@ -524,8 +524,8 @@ func (csr *ChangeStreamReader) StartChangeStream(ctx context.Context) error {
 		if err != nil {
 			// NB: This failure always happens after the initial change stream
 			// creation.
-			csr.ErrChan <- err
-			close(csr.ErrChan)
+			csr.errChan <- err
+			close(csr.errChan)
 		}
 	}()
 

internal/verifier/change_stream_test.go

@@ -165,8 +165,8 @@ 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.DoneChan
+	verifier.srcChangeStreamReader.writesOffTsChan <- *origStartTs
+	<-verifier.srcChangeStreamReader.doneChan
 	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origStartTs)
 }
 
@@ -211,8 +211,8 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 		"session time after events should exceed the original",
 	)
 
-	verifier.srcChangeStreamReader.WritesOffTsChan <- *postEventsSessionTime
-	<-verifier.srcChangeStreamReader.DoneChan
+	verifier.srcChangeStreamReader.writesOffTsChan <- *postEventsSessionTime
+	<-verifier.srcChangeStreamReader.doneChan
 
 	suite.Assert().Equal(
 		*postEventsSessionTime,

internal/verifier/check.go

@@ -44,11 +44,11 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context, csr *ChangeSt
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case err := <-csr.ErrChan:
+	case err := <-csr.errChan:
 		verifier.logger.Warn().Err(err).
 			Msgf("Received error from %s.", csr)
 		return err
-	case <-csr.DoneChan:
+	case <-csr.doneChan:
 		verifier.logger.Debug().
 			Msgf("Received completion signal from %s.", csr)
 		break
@@ -82,9 +82,9 @@ func (verifier *Verifier) CheckWorker(ctxIn context.Context) error {
 	// If the change stream fails, everything should stop.
 	eg.Go(func() error {
 		select {
-		case err := <-verifier.srcChangeStreamReader.ErrChan:
+		case err := <-verifier.srcChangeStreamReader.errChan:
 			return errors.Wrapf(err, "%s failed", verifier.srcChangeStreamReader)
-		case err := <-verifier.dstChangeStreamReader.ErrChan:
+		case err := <-verifier.dstChangeStreamReader.errChan:
 			return errors.Wrapf(err, "%s failed", verifier.dstChangeStreamReader)
 		case <-ctx.Done():
 			return nil

internal/verifier/migration_verifier.go

@@ -255,7 +255,6 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 		return errors.New("writesOff already set")
 	}
 	verifier.writesOff = true
-	verifier.mux.Unlock()
 
 	verifier.logger.Debug().Msg("Signalling that writes are done.")
 
@@ -266,6 +265,7 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	)
 
 	if err != nil {
+		verifier.mux.Unlock()
 		return errors.Wrapf(err, "failed to fetch source's cluster time")
 	}
 
@@ -276,21 +276,23 @@ func (verifier *Verifier) WritesOff(ctx context.Context) error {
 	)
 
 	if err != nil {
+		verifier.mux.Unlock()
 		return errors.Wrapf(err, "failed to fetch destination's cluster time")
 	}
+	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:
+	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)
 	}
 
 	select {
-	case verifier.dstChangeStreamReader.WritesOffTsChan <- dstFinalTs:
-	case err := <-verifier.dstChangeStreamReader.ErrChan:
+	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)
 	}