Allow arbitrary DDL on the destination. (#125)

Some DDL operations are internal to a migration. We currently allow `modify` events for so that capped collections can have their limits set at the end of a migration. We also need to allow `createIndexes` to accommodate migrations where the destination’s indexes are created after initial sync.

Rather than allowing specific events, though, this changeset generalizes tolerance of all unknown events—which should be DDL—on the destination. Any time such an event appears in the destination’s change stream, we now log about it then discard it. We still fail if a DDL event happens on the source. This allows migrator tools flexibility in when they create collections or indexes while still making verification fail if an unsupported event happens on the source.

Author: FGasper

internal/verifier/change_stream.go

@@ -23,14 +23,12 @@ import (
 	"golang.org/x/exp/slices"
 )
 
-type modifyEventHandling string
+type ddlEventHandling string
 
 const (
 	fauxDocSizeForDeleteEvents = 1024
 
-	modifyEventType = "modify"
-
-	onModifyEventIgnore modifyEventHandling = "ignore"
+	onDDLEventAllow ddlEventHandling = "allow"
 )
 
 var supportedEventOpTypes = mapset.NewSet(
@@ -101,7 +99,7 @@ type ChangeStreamReader struct {
 
 	lag *msync.TypedAtomic[option.Option[time.Duration]]
 
-	onModifyEvent modifyEventHandling
+	onDDLEvent ddlEventHandling
 }
 
 func (verifier *Verifier) initializeChangeStreamReaders() {
@@ -134,7 +132,7 @@ func (verifier *Verifier) initializeChangeStreamReaders() {
 		handlerError:         util.NewEventual[error](),
 		doneChan:             make(chan struct{}),
 		lag:                  msync.NewTypedAtomic(option.None[time.Duration]()),
-		onModifyEvent:        onModifyEventIgnore,
+		onDDLEvent:           onDDLEventAllow,
 	}
 }
 
@@ -381,14 +379,16 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		opType := changeEvents[eventsRead].OpType
 		if !supportedEventOpTypes.Contains(opType) {
 
-			// We expect modify events on the destination as part of finalizing
+			// We expect certain DDL events on the destination as part of
 			// a migration. For example, mongosync enables indexes’ uniqueness
-			// constraints and sets capped collection sizes.
-			if opType == modifyEventType && csr.onModifyEvent == onModifyEventIgnore {
+			// constraints and sets capped collection sizes, and sometimes
+			// indexes are created after initial sync.
+
+			if csr.onDDLEvent == onDDLEventAllow {
 				csr.logger.Info().
 					Stringer("changeStream", csr).
 					Stringer("event", cs.Current).
-					Msg("This event is probably internal to the migration. Ignoring.")
+					Msg("Ignoring event with unrecognized type on destination. (It’s assumedly internal to the migration.)")
 
 				// Discard this event, then keep reading.
 				changeEvents = changeEvents[:len(changeEvents)-1]

internal/verifier/change_stream_test.go

@@ -784,6 +784,20 @@ func (suite *IntegrationTestSuite) TestTolerateDestinationCollMod() {
 		"should alter capped size",
 	)
 
+	_, err = suite.dstMongoClient.
+		Database(suite.DBNameForTest()).
+		Collection("mycoll").
+		Indexes().CreateOne(
+		ctx,
+		mongo.IndexModel{
+			Keys: bson.D{
+				{"foo", 1},
+				{"barbar", 1},
+			},
+		},
+	)
+	suite.Require().NoError(err, "should create index")
+
 	err = verifier.WritesOff(ctx)
 	if err == nil {
 		err = verifierRunner.Await()
@@ -796,6 +810,12 @@ func (suite *IntegrationTestSuite) TestTolerateDestinationCollMod() {
 		"cappedSize",
 		"modify event should be recorded in log",
 	)
+
+	suite.Assert().Contains(
+		logBuffer.String(),
+		"barbar_1",
+		"createIndexes event should be recorded in log",
+	)
 }
 
 func (suite *IntegrationTestSuite) TestRecheckDocsWithDstChangeEvents() {