REP-6088 Tolerate high numbers of mismatches (#117)

Previously all document mismatches were recorded directly in the verification task. This meant, though, that if a task encompassed a large number of mismatched or missing documents, the verifier could fail to persist all of the mismatches, which caused a crash.

(The usual cause of excess mismatched/missing documents is starting migration-verifier before initial sync finishes, but it can also reasonably happen without REP-6129’s fix for queries against pre-v5 servers. See HELP-75910.)

This changeset makes the verifier save mismatches to a dedicated collection instead, one document per mismatch.

This change upends some familiar workflows for investigating mismatches: it’s no longer sufficient just to query the `verification_tasks` collection for mismatch information since the actual mismatches are recorded in a separate collection. To address this, the documentation now gives an aggregation pipeline that yields a similarly-useful result.

This entails a metadata version change. Because that’s happening, this also changes the task type `verify` to `verifyDocuments`. (That required some sorting workarounds in tests, which were tight-coupled to the task type strings.)

Author: FGasper

README.md

@@ -142,6 +142,31 @@ The verifier will now check to completion to make sure that there are no inconsi
 | `--ignoreReadConcern`                   | Use connection-default read concerns rather than setting majority read concern. This option may degrade consistency, so only enable it if majority read concern (the default) doesn’t work. |
 | `--help`, `-h`                          | show help                                                                                                                                                                                   |
 
+# Investigation of Mismatches
+
+The verifier records any mismatches it finds in its metadata’s `mismatches`
+collection. Mismatches are indexed by verification task ID. To find a given
+generation’s mismatches, aggregate like this on the metadata cluster:
+
+    // Change this as needed if you specify a custom metadata database:
+    use migration_verification_metadata
+
+    db.verification_tasks.aggregate(
+        { $match: {
+            generation: <whichever generation>,
+            status: "failed",
+        } },
+        { $lookup: {
+            from: "mismatches",
+            localField: "_id",
+            foreignField: "task",
+            as: "mismatch",
+        }},
+        { $unwind: "$mismatch" },
+    )
+
+Note that each mismatch includes timestamps. You can cross-reference
+these with the clusters’ oplogs to diagnose problems.
 
 # Benchmarking Results
 

chanutil/io.go

@@ -0,0 +1,39 @@
+package chanutil
+
+import (
+	"context"
+
+	"github.com/10gen/migration-verifier/internal/util"
+	"github.com/10gen/migration-verifier/option"
+)
+
+// ReadWithDoneCheck takes a context and a channel to read from. It will read
+// from `ctx.Done()` and the given channel in a select. If it reads an error
+// from the `ctx.Done()` channel, it returns the value of `Err(ctx)`, which
+// includes the cancellation cause. If it reads a value from that channel, it
+// returns the value. If the channel was closed, the return value is None.
+func ReadWithDoneCheck[T any](ctx context.Context, ch <-chan T) (option.Option[T], error) {
+	select {
+	case <-ctx.Done():
+		return option.None[T](), util.WrapCtxErrWithCause(ctx)
+	case val, ok := <-ch:
+		if ok {
+			return option.Some(val), nil
+		}
+
+		return option.None[T](), nil
+	}
+}
+
+// WriteWithDoneCheck takes a context, a channel to write to, and a value to write to that
+// channel. It will read from `ctx.Done()` and write to the given channel in a select. If it reads
+// an error from the `ctx.Done()` channel, it returns the value of `Err(ctx)`, which includes the
+// cancellation cause. If it writes a value to that channel, it returns `nil`.
+func WriteWithDoneCheck[T any](ctx context.Context, ch chan<- T, val T) error {
+	select {
+	case <-ctx.Done():
+		return util.WrapCtxErrWithCause(ctx)
+	case ch <- val:
+		return nil
+	}
+}

go.mod

@@ -1,6 +1,6 @@
 module github.com/10gen/migration-verifier
 
-go 1.22
+go 1.24
 
 require (
 	github.com/cespare/permute/v2 v2.0.0-beta2

internal/testutil/testutil.go

@@ -2,15 +2,63 @@ package testutil
 
 import (
 	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"log"
 	"testing"
 
 	"github.com/pkg/errors"
+	"github.com/samber/lo"
 	"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"
 )
 
+func generateRandomFieldName(baseName string) string {
+	b := make([]byte, 4) // 8 hex chars
+	_, err := rand.Read(b)
+	if err != nil {
+		log.Fatal(err)
+	}
+	return baseName + "_" + hex.EncodeToString(b)
+}
+
+// Returns an agg pipeline that sorts results according to a static list
+// of values. Any result that doesn’t match a list value goes at the end.
+func SortByListAgg[T any](
+	fieldName string,
+	values []T,
+) []bson.D {
+	fieldRef := "$" + fieldName
+
+	sortField := generateRandomFieldName("sortOrder")
+
+	branches := lo.Map(
+		values,
+		func(v T, i int) bson.D {
+			return bson.D{
+				{"case", bson.D{{"$eq", bson.A{
+					fieldRef,
+					bson.D{{"$literal", v}},
+				}}}},
+				{"then", i},
+			}
+		},
+	)
+
+	return mongo.Pipeline{
+		{{"$addFields", bson.D{
+			{sortField, bson.D{{"$switch", bson.D{
+				{"branches", branches},
+				{"default", len(values)},
+			}}}},
+		}}},
+		{{"$sort", bson.D{{sortField, 1}}}},
+		{{"$project", bson.D{{sortField, 0}}}},
+	}
+}
+
 // Marshal wraps `bsonMarshal` with a panic on failure.
 func MustMarshal(doc any) bson.Raw {
 	raw, err := bson.Marshal(doc)