Merge branch 'main' into REP-5317-add-dst-change-stream

Author: tdq45gj

internal/util/collections.go

@@ -0,0 +1,84 @@
+package util
+
+import (
+	"context"
+
+	"github.com/10gen/migration-verifier/option"
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+// CollectionSpec is like mongo.CollectionSpecification except:
+// - IDIndex is a bson.Raw rather than mongo.IndexSpecification.
+// - It can detect unexpected fields.
+type CollectionSpec struct {
+	Name    string
+	Type    string
+	Options bson.Raw
+	Info    struct {
+		ReadOnly bool `bson:"readOnly"`
+		UUID     primitive.Binary
+
+		Extra map[string]any
+	}
+	IDIndex bson.Raw `bson:"idIndex"`
+
+	Extra map[string]any
+}
+
+// FullName returns the collection's full namespace.
+func FullName(collection *mongo.Collection) string {
+	return collection.Database().Name() + "." + collection.Name()
+}
+
+// GetCollectionSpecIfExists returns the given collection’s specification,
+// or empty if the collection doesn’t exist. If any unexpected properties
+// exist in the collection specification then an error is returned.
+func GetCollectionSpecIfExists(
+	ctx context.Context,
+	coll *mongo.Collection,
+) (option.Option[CollectionSpec], error) {
+	cursor, err := coll.Database().ListCollections(ctx, bson.M{"name": coll.Name()})
+	if err != nil {
+		return option.None[CollectionSpec](), errors.Wrapf(
+			err,
+			"failed to fetch %#q's specification",
+			FullName(coll),
+		)
+	}
+
+	var specs []CollectionSpec
+	err = cursor.All(ctx, &specs)
+	if err != nil {
+		return option.None[CollectionSpec](), errors.Wrapf(
+			err,
+			"failed to parse %#q's specification",
+			FullName(coll),
+		)
+	}
+
+	switch len(specs) {
+	case 0:
+		return option.None[CollectionSpec](), nil
+	case 1:
+		if len(specs[0].Extra) > 0 || len(specs[0].Info.Extra) > 0 {
+			return option.None[CollectionSpec](), errors.Wrapf(
+				err,
+				"%#q's specification (%v) contains unrecognized fields",
+				FullName(coll),
+				specs[0],
+			)
+		}
+
+		return option.Some(specs[0]), nil
+	}
+
+	return option.None[CollectionSpec](), errors.Wrapf(
+		err,
+		"received multiple results (%v) when fetching %#q's specification",
+		specs,
+		FullName(coll),
+	)
+}

internal/verifier/change_stream.go

@@ -319,7 +319,7 @@ func (csr *ChangeStreamReader) iterateChangeStream(ctx context.Context, cs *mong
 					break
 				}
 
-				if curTs == writesOffTs || curTs.After(writesOffTs) {
+				if curTs.After(writesOffTs) {
 					csr.logger.Debug().
 						Interface("currentTimestamp", curTs).
 						Interface("writesOffTimestamp", writesOffTs).

internal/verifier/change_stream_test.go

@@ -163,7 +163,13 @@ func (suite *IntegrationTestSuite) TestStartAtTimeWithChanges() {
 	suite.Require().NotNil(origSessionTime)
 	err = verifier.srcChangeStreamReader.StartChangeStream(ctx)
 	suite.Require().NoError(err)
-	suite.Require().Equal(verifier.srcChangeStreamReader.startAtTs, origSessionTime)
+
+	// srcStartAtTs derives from the change stream’s resume token, which can
+	// postdate our session time but should not precede it.
+	suite.Require().False(
+		verifier.srcChangeStreamReader.startAtTs.Before(*origSessionTime),
+		"srcStartAtTs should be >= the insert’s optime",
+	)
 
 	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl").InsertOne(
 		sctx, bson.D{{"_id", 1}})
@@ -242,7 +248,15 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 	)
 }
 
-func (suite *IntegrationTestSuite) TestEventBeforeWritesOff() {
+func (suite *IntegrationTestSuite) TestManyInsertsBeforeWritesOff() {
+	suite.testInsertsBeforeWritesOff(10_000)
+}
+
+func (suite *IntegrationTestSuite) TestOneInsertBeforeWritesOff() {
+	suite.testInsertsBeforeWritesOff(1)
+}
+
+func (suite *IntegrationTestSuite) testInsertsBeforeWritesOff(docsCount int) {
 	ctx := suite.Context()
 
 	verifier := suite.BuildVerifier()
@@ -259,7 +273,6 @@ func (suite *IntegrationTestSuite) TestEventBeforeWritesOff() {
 	// wait for generation 0 to end
 	verifierRunner.AwaitGenerationEnd()
 
-	docsCount := 10_000
 	docs := lo.RepeatBy(docsCount, func(_ int) bson.D { return bson.D{} })
 	_, err := coll.InsertMany(
 		ctx,