Merge branch 'main' into rep-5219-batch-attempt-2

Author: tdq45gj

internal/verifier/change_stream.go

@@ -25,7 +25,7 @@ type ParsedEvent struct {
 }
 
 func (pe *ParsedEvent) String() string {
-	return fmt.Sprintf("{ OpType: %s, namespace: %s, docID: %v}", pe.OpType, pe.Ns, pe.DocKey.ID)
+	return fmt.Sprintf("{OpType: %s, namespace: %s, docID: %v, clusterTime: %v}", pe.OpType, pe.Ns, pe.DocKey.ID, pe.ClusterTime)
 }
 
 // DocKey is a deserialized form for the ChangeEvent documentKey field. We currently only care about
@@ -47,7 +47,7 @@ type UnknownEventError struct {
 }
 
 func (uee UnknownEventError) Error() string {
-	return fmt.Sprintf("Unknown event type: %#q", uee.Event.OpType)
+	return fmt.Sprintf("Received event with unknown optype: %+v", uee.Event)
 }
 
 // HandleChangeStreamEvent performs the necessary work for change stream events that occur during
@@ -88,8 +88,6 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 }
 
 func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) {
-	var changeEvent ParsedEvent
-
 	var lastPersistedTime time.Time
 
 	persistResumeTokenIfNeeded := func() error {
@@ -109,10 +107,13 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		eventsRead := 0
 		for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
 			gotEvent := cs.TryNext(ctx)
+
 			if !gotEvent {
 				break
 			}
 
+			var changeEvent ParsedEvent
+
 			if err := cs.Decode(&changeEvent); err != nil {
 				return false, errors.Wrap(err, "failed to decode change event")
 			}
@@ -147,24 +148,22 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		// source writes are ended. This means we should exit rather than continue
 		// reading the change stream since there should be no more events.
 		case <-verifier.changeStreamEnderChan:
-			var gotEvent bool
+			verifier.logger.Debug().
+				Msg("Change stream thread received shutdown request.")
 
 			changeStreamEnded = true
 
 			// Read all change events until the source reports no events.
 			// (i.e., the `getMore` call returns empty)
 			for {
+				var gotEvent bool
 				gotEvent, err = readAndHandleOneChangeEventBatch()
 
 				if !gotEvent || err != nil {
 					break
 				}
 			}
 
-			if err != nil {
-				break
-			}
-
 		default:
 			_, err = readAndHandleOneChangeEventBatch()
 		}
@@ -174,7 +173,15 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		}
 
 		if err != nil && !errors.Is(err, context.Canceled) {
+			verifier.logger.Debug().
+				Err(err).
+				Msg("Sending change stream error.")
+
 			verifier.changeStreamErrChan <- err
+
+			if !changeStreamEnded {
+				break
+			}
 		}
 
 		if changeStreamEnded {
@@ -187,11 +194,18 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 			// since we have started Recheck, we must signal that we have
 			// finished the change stream changes so that Recheck can continue.
 			verifier.changeStreamDoneChan <- struct{}{}
-			// since the changeStream is exhausted, we now return
-			verifier.logger.Debug().Msg("Change stream is done")
-			return
+			break
 		}
 	}
+
+	infoLog := verifier.logger.Info()
+	if verifier.lastChangeEventTime == nil {
+		infoLog = infoLog.Str("changeStreamStopTime", "none")
+	} else {
+		infoLog = infoLog.Interface("changeStreamStopTime", *verifier.lastChangeEventTime)
+	}
+
+	infoLog.Msg("Change stream is done.")
 }
 
 // StartChangeStream starts the change stream.

internal/verifier/check.go

@@ -51,8 +51,12 @@ func (verifier *Verifier) waitForChangeStream() error {
 		verifier.changeStreamEnderChan <- struct{}{}
 		select {
 		case err := <-verifier.changeStreamErrChan:
+			verifier.logger.Warn().Err(err).
+				Msg("Received error from change stream.")
 			return err
 		case <-verifier.changeStreamDoneChan:
+			verifier.logger.Debug().
+				Msg("Received completion signal from change stream.")
 			break
 		}
 	}
@@ -149,6 +153,17 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	if err != nil {
 		return err
 	}
+
+	err = verifier.doInMetaTransaction(
+		ctx,
+		func(ctx context.Context, sCtx mongo.SessionContext) error {
+			return verifier.ResetInProgressTasks(sCtx)
+		},
+	)
+	if err != nil {
+		return errors.Wrap(err, "failed to reset any in-progress tasks")
+	}
+
 	verifier.logger.Debug().Msg("Starting Check")
 
 	verifier.phase = Check
@@ -188,16 +203,31 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 			return err
 		}
 		// we will only coordinate when the number of channels is exactly 2.
-		// * Channel 0 signals a generation is done
-		// * Channel 1 signals to check to continue the next generation
+		// * Channel 0 informs the test of a generation bounary.
+		// * Block until the test (via channel 1) tells us to do the
+		//   next generation.
 		if len(testChan) == 2 {
+
+			verifier.logger.Debug().
+				Msg("Telling test about generation boundary.")
 			testChan[0] <- struct{}{}
+
+			verifier.logger.Debug().
+				Msg("Awaiting test's signal to continue.")
 			<-testChan[1]
+
+			verifier.logger.Debug().
+				Msg("Received test's signal. Continuing.")
 		}
 		time.Sleep(verifier.generationPauseDelayMillis * time.Millisecond)
 		verifier.mux.Lock()
 		if verifier.lastGeneration {
 			verifier.mux.Unlock()
+
+			verifier.logger.Debug().
+				Int("generation", verifier.generation).
+				Msg("Final generation done.")
+
 			return nil
 		}
 		// TODO: wait here until writesOff is hit or enough time has passed, so we don't spin

internal/verifier/migration_verifier_test.go

@@ -1581,6 +1581,10 @@ func (suite *MultiDataVersionTestSuite) TestVerifierWithFilter() {
 
 	// Turn writes off.
 	verifier.WritesOff(ctx)
+
+	// Tell CheckDriver to do one more pass. This should terminate the change stream.
+	checkContinueChan <- struct{}{}
+	<-checkDoneChan
 }
 
 func (suite *MultiDataVersionTestSuite) TestPartitionWithFilter() {

internal/verifier/reset.go

@@ -0,0 +1,161 @@
+package verifier
+
+import (
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+var defaultTaskUpdate = bson.M{
+	"$set":   bson.M{"status": verificationTaskAdded},
+	"$unset": bson.M{"begin_time": 1},
+}
+
+func (verifier *Verifier) ResetInProgressTasks(ctx mongo.SessionContext) error {
+	didReset, err := verifier.handleIncompletePrimary(ctx)
+
+	if err == nil {
+		if didReset {
+			return nil
+		}
+
+		err = verifier.resetCollectionTasksIfNeeded(ctx)
+	}
+
+	if err == nil {
+		err = verifier.resetPartitionTasksIfNeeded(ctx)
+	}
+
+	return err
+}
+
+func (verifier *Verifier) handleIncompletePrimary(ctx mongo.SessionContext) (bool, error) {
+	taskColl := verifier.verificationTaskCollection()
+
+	cursor, err := taskColl.Find(
+		ctx,
+		bson.M{
+			"type":   verificationTaskPrimary,
+			"status": bson.M{"$ne": verificationTaskCompleted},
+		},
+	)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to fetch incomplete %#q task", verificationTaskPrimary)
+	}
+
+	var incompletePrimaries []VerificationTask
+	err = cursor.All(ctx, &incompletePrimaries)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to read incomplete %#q task", verificationTaskPrimary)
+	}
+
+	switch len(incompletePrimaries) {
+	case 0:
+		// Nothing to do.
+	case 1:
+		// Invariant: task status should be “added”.
+		if incompletePrimaries[0].Status != verificationTaskAdded {
+			verifier.logger.Panic().
+				Interface("task", incompletePrimaries[0]).
+				Msg("Primary task status has invalid state.")
+		}
+
+		verifier.logger.Info().
+			Msg("Previous verifier run left primary task incomplete. Deleting non-primary tasks.")
+
+		deleted, err := taskColl.DeleteMany(
+			ctx,
+			bson.M{
+				"type": bson.M{
+					"$ne": verificationTaskPrimary,
+				},
+			},
+		)
+		if err != nil {
+			return false, errors.Wrapf(err, "failed to delete non-%#q tasks", verificationTaskPrimary)
+		}
+
+		verifier.logger.Info().
+			Int64("deletedTasksCount", deleted.DeletedCount).
+			Msg("Found and deleted non-primary tasks.")
+
+		return true, nil
+	default:
+		verifier.logger.Panic().
+			Interface("tasks", incompletePrimaries).
+			Msg("Found multiple incomplete primary tasks; there should only be 1.")
+	}
+
+	return false, nil
+}
+
+func (verifier *Verifier) resetCollectionTasksIfNeeded(ctx mongo.SessionContext) error {
+	taskColl := verifier.verificationTaskCollection()
+
+	cursor, err := taskColl.Find(
+		ctx,
+		bson.M{
+			"type":   verificationTaskVerifyCollection,
+			"status": verificationTaskProcessing,
+		},
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to find incomplete %#q tasks", verificationTaskVerifyCollection)
+	}
+	var incompleteCollTasks []VerificationTask
+	err = cursor.All(ctx, &incompleteCollTasks)
+	if err != nil {
+		return errors.Wrapf(err, "failed to read incomplete %#q tasks", verificationTaskVerifyCollection)
+	}
+
+	if len(incompleteCollTasks) > 0 {
+		verifier.logger.Info().
+			Int("count", len(incompleteCollTasks)).
+			Msg("Previous verifier run left collection-level verification task(s) pending. Resetting.")
+	}
+
+	for _, task := range incompleteCollTasks {
+		_, err := taskColl.DeleteMany(
+			ctx,
+			bson.M{
+				"type":                   verificationTaskVerifyDocuments,
+				"query_filter.namespace": task.QueryFilter.Namespace,
+			},
+		)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete namespace %#q's %#q tasks", task.QueryFilter.Namespace, verificationTaskVerifyDocuments)
+		}
+
+		_, err = taskColl.UpdateOne(
+			ctx,
+			bson.M{
+				"type":                   verificationTaskVerifyCollection,
+				"query_filter.namespace": task.QueryFilter.Namespace,
+			},
+			defaultTaskUpdate,
+		)
+		if err != nil {
+			return errors.Wrapf(err, "failed to reset namespace %#q's %#q task", task.QueryFilter.Namespace, verificationTaskVerifyCollection)
+		}
+	}
+
+	return nil
+}
+
+func (verifier *Verifier) resetPartitionTasksIfNeeded(ctx mongo.SessionContext) error {
+	taskColl := verifier.verificationTaskCollection()
+
+	_, err := taskColl.UpdateMany(
+		ctx,
+		bson.M{
+			"type":   verificationTaskVerifyDocuments,
+			"status": verificationTaskProcessing,
+		},
+		defaultTaskUpdate,
+	)
+	if err != nil {
+		return errors.Wrapf(err, "failed to reset in-progress %#q tasks", verificationTaskVerifyDocuments)
+	}
+
+	return nil
+}