REP-6735 Allow rechecks to be enqueued during recheck task gen (#157)

Historically, the verifier blocked the enqueuing of rechecks while generating new recheck tasks. This slowed down the change stream because it couldn’t enqueue rechecks during that time.

There seems to be no reason why the change stream needs to be blocked while creating new tasks, though. This changeset removes the restriction. It also tweaks the generation-numbering so that the generation persisted with the recheck is the generation where the document will be rechecked rather than the generation where the recheck was enqueued.

The net effect here is that, with the change stream not blocked for several seconds between generations, higher throughput is possible.

Author: FGasper

internal/verifier/change_stream_test.go

@@ -359,7 +359,7 @@ func (suite *IntegrationTestSuite) TestChangeStream_Resume_NoSkip() {
 	assert.Eventually(
 		suite.T(),
 		func() bool {
-			rechecks := suite.fetchVerifierRechecks(ctx, verifier2)
+			rechecks := suite.fetchPendingVerifierRechecks(ctx, verifier2)
 
 			return lo.SomeBy(
 				rechecks,
@@ -386,7 +386,7 @@ func (suite *IntegrationTestSuite) TestChangeStream_Resume_NoSkip() {
 	sess := lo.Must(verifier2.verificationDatabase().Client().StartSession())
 	sctx := mongo.NewSessionContext(ctx, sess)
 
-	rechecks := suite.fetchVerifierRechecks(sctx, verifier2)
+	rechecks := suite.fetchPendingVerifierRechecks(sctx, verifier2)
 	if !assert.EqualValues(suite.T(), lastDocID, len(rechecks), "all source docs should be rechecked") {
 		for _, recheck := range rechecks {
 			suite.T().Logf("found recheck: %v", recheck)
@@ -426,7 +426,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 	verifier2 := suite.BuildVerifier()
 
 	suite.Require().Empty(
-		suite.fetchVerifierRechecks(ctx, verifier2),
+		suite.fetchPendingVerifierRechecks(ctx, verifier2),
 		"no rechecks should be enqueued before starting change stream",
 	)
 
@@ -446,7 +446,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 	require.Eventually(
 		suite.T(),
 		func() bool {
-			recheckDocs = suite.fetchVerifierRechecks(ctx, verifier2)
+			recheckDocs = suite.fetchPendingVerifierRechecks(ctx, verifier2)
 
 			return len(recheckDocs) > 0
 		},
@@ -490,10 +490,10 @@ func (suite *IntegrationTestSuite) getClusterTime(ctx context.Context, client *m
 	return newTime
 }
 
-func (suite *IntegrationTestSuite) fetchVerifierRechecks(ctx context.Context, verifier *Verifier) []bson.M {
+func (suite *IntegrationTestSuite) fetchPendingVerifierRechecks(ctx context.Context, verifier *Verifier) []bson.M {
 	recheckDocs := []bson.M{}
 
-	recheckColl := verifier.getRecheckQueueCollection(verifier.generation)
+	recheckColl := verifier.getRecheckQueueCollection(1 + verifier.generation)
 	cursor, err := recheckColl.Aggregate(
 		ctx,
 		mongo.Pipeline{
@@ -732,7 +732,7 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 	require.Eventually(
 		suite.T(),
 		func() bool {
-			rechecks = suite.fetchVerifierRechecks(ctx, verifier)
+			rechecks = suite.fetchPendingVerifierRechecks(ctx, verifier)
 			return len(rechecks) == 3
 		},
 		time.Minute,
@@ -1069,7 +1069,7 @@ func (suite *IntegrationTestSuite) TestRecheckDocsWithDstChangeEvents() {
 	require.Eventually(
 		suite.T(),
 		func() bool {
-			recheckColl := verifier.getRecheckQueueCollection(verifier.generation)
+			recheckColl := verifier.getRecheckQueueCollection(1 + verifier.generation)
 			cursor, err := recheckColl.Find(ctx, bson.D{})
 			if errors.Is(err, mongo.ErrNoDocuments) {
 				return false

internal/verifier/check.go

@@ -390,30 +390,35 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter bson.D, testCh
 			verifier.mux.Lock()
 			verifier.lastGeneration = true
 		}
+
+		// Increment the in-memory generation so that the change streams will
+		// mark rechecks for the next generation. For example, if we just
+		// finished generation 2, the change streams need to mark generation 3
+		// on enqueued rechecks. Meanwhile, generaiton 3’s recheck tasks will
+		// derive from rechecks enqueued during generation 2.
 		verifier.generation++
 		verifier.phase = Recheck
+		verifier.mux.Unlock()
 
 		// Generation of recheck tasks can partial-fail. The following will
 		// cause a full redo in that case, which is inefficient but simple.
 		// Such failures seem unlikely anyhow.
 		err = retry.New().WithCallback(
 			func(ctx context.Context, _ *retry.FuncInfo) error {
-				return verifier.GenerateRecheckTasksWhileLocked(ctx)
+				return verifier.GenerateRecheckTasks(ctx)
 			},
 			"generating recheck tasks",
 		).Run(ctx, verifier.logger)
 		if err != nil {
-			verifier.mux.Unlock()
 			return err
 		}
 
-		err = verifier.DropOldRecheckQueueWhileLocked(ctx)
+		err = verifier.DropCurrentGenRecheckQueue(ctx)
 		if err != nil {
 			verifier.logger.Warn().
 				Err(err).
 				Msg("Failed to clear out old recheck docs. (This is probably unimportant.)")
 		}
-		verifier.mux.Unlock()
 	}
 }
 

internal/verifier/metadata.go

@@ -3,5 +3,7 @@ package verifier
 // Metadata version history:
 // 1: Defined metadata version.
 // 2: Split failed-task discrepancies into separate collection.
+// 3: Enqueued rechecks now reference the generation in which they’ll be
+//    rechecked rather than the generation during which they were enqueued.
 
-const verifierMetadataVersion = 2
+const verifierMetadataVersion = 3

internal/verifier/migration_verifier_test.go

@@ -714,11 +714,7 @@ func (suite *IntegrationTestSuite) TestGetPersistedNamespaceStatistics_Recheck()
 
 	verifier.generation++
 
-	func() {
-		verifier.mux.Lock()
-		defer func() { verifier.mux.Unlock() }()
-		suite.Require().NoError(verifier.GenerateRecheckTasksWhileLocked(ctx))
-	}()
+	suite.Require().NoError(verifier.GenerateRecheckTasks(ctx))
 
 	stats, err := verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
@@ -958,6 +954,7 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 		[]int32{100},
 	)
 	suite.Require().NoError(err)
+
 	event := ParsedEvent{
 		DocID:  mbson.ToRawValue(int32(55)),
 		OpType: "delete",
@@ -976,6 +973,7 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 
 	err = verifier.HandleChangeStreamEvents(ctx, batch, src)
 	suite.Require().NoError(err)
+
 	event.OpType = "insert"
 	err = verifier.HandleChangeStreamEvents(ctx, batch, src)
 	suite.Require().NoError(err)
@@ -995,13 +993,9 @@ func (suite *IntegrationTestSuite) TestFailedVerificationTaskInsertions() {
 	)
 
 	verifier.generation++
-	func() {
-		verifier.mux.Lock()
-		defer verifier.mux.Unlock()
 
-		err = verifier.GenerateRecheckTasksWhileLocked(ctx)
-		suite.Require().NoError(err)
-	}()
+	err = verifier.GenerateRecheckTasks(ctx)
+	suite.Require().NoError(err)
 
 	var doc bson.M
 	cur, err := verifier.verificationTaskCollection().Find(ctx, bson.M{"generation": 1})
@@ -2323,10 +2317,12 @@ func (suite *IntegrationTestSuite) TestVerifierWithFilter() {
 }
 
 func (suite *IntegrationTestSuite) awaitEnqueueOfRechecks(verifier *Verifier, minDocs int) {
+	suite.T().Helper()
+
 	var lastNonzeroRechecksCount int
 
 	suite.Eventually(func() bool {
-		cursor, err := verifier.getRecheckQueueCollection(verifier.generation).
+		cursor, err := verifier.getRecheckQueueCollection(1+verifier.generation).
 			Find(suite.Context(), bson.D{})
 		var rechecks []bson.D
 		suite.Require().NoError(err)

internal/verifier/recheck.go

@@ -61,6 +61,9 @@ func (verifier *Verifier) insertRecheckDocs(
 
 	generation, _ := verifier.getGenerationWhileLocked()
 
+	// We enqueue for the generation after the current one.
+	generation++
+
 	eg, groupCtx := contextplus.ErrGroup(ctx)
 
 	// MongoDB’s Go driver starts failing requests if we try to exceed
@@ -233,66 +236,37 @@ func buildRequestBSON(collName string, rechecks []bson.Raw) bson.Raw {
 	return requestBSON
 }
 
-// DropOldRecheckQueueWhileLocked deletes the previous generation’s recheck
-// documents from the verifier’s metadata.
-//
-// The verifier **MUST** be locked when this function is called (or panic).
-func (verifier *Verifier) DropOldRecheckQueueWhileLocked(ctx context.Context) error {
-	prevGeneration := verifier.getPreviousGenerationWhileLocked()
+// DropCurrentGenRecheckQueue deletes the current generation’s recheck
+// documents from the verifier’s metadata. This should only be called
+// after new recheck tasks have been created from those rechecks.
+func (verifier *Verifier) DropCurrentGenRecheckQueue(ctx context.Context) error {
+	generation := verifier.generation
 
 	verifier.logger.Debug().
-		Int("previousGeneration", prevGeneration).
-		Msg("Deleting previous generation's enqueued rechecks.")
+		Int("generation", generation).
+		Msg("Deleting current generation's enqueued rechecks.")
 
-	genCollection := verifier.getRecheckQueueCollection(prevGeneration)
+	genCollection := verifier.getRecheckQueueCollection(generation)
 
 	return retry.New().WithCallback(
 		func(ctx context.Context, i *retry.FuncInfo) error {
 			return genCollection.Drop(ctx)
 		},
 		"deleting generation %d's enqueued rechecks",
-		prevGeneration,
+		generation,
 	).Run(ctx, verifier.logger)
 }
 
-func (verifier *Verifier) getPreviousGenerationWhileLocked() int {
-	generation, _ := verifier.getGenerationWhileLocked()
-	if generation < 1 {
-		panic("This function is forbidden before generation 1!")
-	}
-
-	return generation - 1
-}
-
-// GenerateRecheckTasksWhileLocked fetches the previous generation’s recheck
-// documents from the verifier’s metadata and creates current-generation
-// document-verification tasks from them.
+// GenerateRecheckTasks fetches the rechecks enqueued for the current generation
+// from the verifier’s metadata and creates document-verification tasks from
+// them.
 //
 // Note that this function DOES NOT retry on failure, so callers should wrap
 // calls to this function in a retryer.
-//
-// The verifier **MUST** be locked when this function is called (or panic).
-func (verifier *Verifier) GenerateRecheckTasksWhileLocked(ctx context.Context) error {
-	prevGeneration := verifier.getPreviousGenerationWhileLocked()
+func (verifier *Verifier) GenerateRecheckTasks(ctx context.Context) error {
+	generation, _ := verifier.getGeneration()
 
-	verifier.logger.Debug().
-		Int("priorGeneration", prevGeneration).
-		Msgf("Counting prior generation’s enqueued rechecks.")
-
-	recheckColl := verifier.getRecheckQueueCollection(prevGeneration)
-
-	rechecksCount, err := recheckColl.CountDocuments(ctx, bson.D{})
-	if err != nil {
-		return errors.Wrapf(err,
-			"failed to count generation %d’s rechecks",
-			prevGeneration,
-		)
-	}
-
-	verifier.logger.Debug().
-		Int("priorGeneration", prevGeneration).
-		Int("rechecksCount", int(rechecksCount)).
-		Msgf("Creating recheck tasks from prior generation’s enqueued rechecks.")
+	recheckColl := verifier.getRecheckQueueCollection(generation)
 
 	startTime := time.Now()
 
@@ -428,7 +402,7 @@ func (verifier *Verifier) GenerateRecheckTasksWhileLocked(ctx context.Context) e
 
 	if err == nil && totalDocs > 0 {
 		verifier.logger.Info().
-			Int("generation", 1+prevGeneration).
+			Int("generation", generation).
 			Int64("totalDocs", int64(totalDocs)).
 			Str("totalData", reportutils.FmtBytes(totalRecheckData)).
 			Stringer("timeElapsed", time.Since(startTime)).
@@ -439,6 +413,10 @@ func (verifier *Verifier) GenerateRecheckTasksWhileLocked(ctx context.Context) e
 }
 
 func (v *Verifier) getRecheckQueueCollection(generation int) *mongo.Collection {
+	if generation == 0 {
+		panic("Recheck for generation 0 is nonsense!")
+	}
+
 	return v.verificationDatabase().
 		Collection(fmt.Sprintf("%s_gen%d", recheckQueueCollectionNameBase, generation))
 }

internal/verifier/recheck_test.go

@@ -85,7 +85,7 @@ func (suite *IntegrationTestSuite) TestFailedCompareThenReplace() {
 }
 
 func (suite *IntegrationTestSuite) fetchRecheckDocs(ctx context.Context, verifier *Verifier) []recheck.Doc {
-	metaColl := verifier.getRecheckQueueCollection(verifier.generation)
+	metaColl := verifier.getRecheckQueueCollection(1 + verifier.generation)
 
 	cursor, err := metaColl.Aggregate(
 		ctx,
@@ -182,7 +182,7 @@ func (suite *IntegrationTestSuite) TestRecheckResumability_Mismatch() {
 		verificationStatus, err := verifier.GetVerificationStatus(ctx)
 		suite.Require().NoError(err)
 
-		recheckDocs := suite.fetchVerifierRechecks(ctx, verifier)
+		recheckDocs := suite.fetchPendingVerifierRechecks(ctx, verifier)
 
 		if verificationStatus.FailedTasks != 0 && len(recheckDocs) == 2 {
 			break
@@ -212,7 +212,7 @@ func (suite *IntegrationTestSuite) TestRecheckResumability_Mismatch() {
 		"restarted verifier should immediately see mismatches",
 	)
 
-	recheckDocs := suite.fetchVerifierRechecks(ctx, verifier2)
+	recheckDocs := suite.fetchPendingVerifierRechecks(ctx, verifier2)
 	suite.Require().Len(recheckDocs, 2, "expect # of rechecks: %+v", recheckDocs)
 }
 
@@ -268,13 +268,10 @@ func (suite *IntegrationTestSuite) TestManyManyRechecks() {
 	)
 	suite.Require().NoError(err)
 
-	verifier.mux.Lock()
-	defer verifier.mux.Unlock()
-
 	verifier.generation++
 
 	suite.T().Logf("Generating recheck tasks …")
-	err = verifier.GenerateRecheckTasksWhileLocked(ctx)
+	err = verifier.GenerateRecheckTasks(ctx)
 	suite.Require().NoError(err)
 }
 
@@ -307,8 +304,7 @@ func (suite *IntegrationTestSuite) TestLargeIDInsertions() {
 	suite.ElementsMatch([]any{d1, d2, d3}, results)
 
 	verifier.generation++
-	verifier.mux.Lock()
-	err = verifier.GenerateRecheckTasksWhileLocked(ctx)
+	err = verifier.GenerateRecheckTasks(ctx)
 	suite.Require().NoError(err)
 	taskColl := suite.metaMongoClient.Database(verifier.metaDBName).Collection(verificationTasksCollection)
 	cursor, err := taskColl.Find(ctx, bson.D{}, options.Find().SetProjection(bson.D{{"_id", 0}}))
@@ -367,8 +363,7 @@ func (suite *IntegrationTestSuite) TestLargeDataInsertions() {
 	suite.ElementsMatch([]any{d1, d2, d3}, results)
 
 	verifier.generation++
-	verifier.mux.Lock()
-	err = verifier.GenerateRecheckTasksWhileLocked(ctx)
+	err = verifier.GenerateRecheckTasks(ctx)
 	suite.Require().NoError(err)
 	taskColl := suite.metaMongoClient.Database(verifier.metaDBName).Collection(verificationTasksCollection)
 	cursor, err := taskColl.Find(ctx, bson.D{}, options.Find().SetProjection(bson.D{{"_id", 0}}))
@@ -417,8 +412,7 @@ func (suite *IntegrationTestSuite) TestMultipleNamespaces() {
 	suite.Require().NoError(err)
 
 	verifier.generation++
-	verifier.mux.Lock()
-	err = verifier.GenerateRecheckTasksWhileLocked(ctx)
+	err = verifier.GenerateRecheckTasks(ctx)
 	suite.Require().NoError(err)
 	taskColl := suite.metaMongoClient.Database(verifier.metaDBName).Collection(verificationTasksCollection)
 	cursor, err := taskColl.Find(ctx, bson.D{}, options.Find().SetProjection(bson.D{{"_id", 0}}))
@@ -477,7 +471,7 @@ func (suite *IntegrationTestSuite) TestGenerationalClear() {
 
 	verifier.generation++
 
-	err = verifier.DropOldRecheckQueueWhileLocked(ctx)
+	err = verifier.DropCurrentGenRecheckQueue(ctx)
 	suite.Require().NoError(err)
 
 	// This never happens in real life but is needed for this test.