Do not deadlock when instant DDL completes the migration

go/logic/migrator.go

@@ -110,12 +110,14 @@ func NewMigrator(context *base.MigrationContext, appVersion string) *Migrator {
 		hooks = NewHooksExecutor(context)
 	}
 	migrator := &Migrator{
-		appVersion:               appVersion,
-		hooksExecutor:            hooks,
-		migrationContext:         context,
-		statusWriter:             os.Stdout,
-		parser:                   sql.NewAlterTableParser(),
-		ghostTableMigrated:       make(chan bool),
+		appVersion:       appVersion,
+		hooksExecutor:    hooks,
+		migrationContext: context,
+		statusWriter:     os.Stdout,
+		parser:           sql.NewAlterTableParser(),
+		// Buffered: the changelog carries one GhostTableMigrated state per migration,
+		// and the --attempt-instant-ddl path returns without ever reading it.
+		ghostTableMigrated:       make(chan bool, 1),
 		firstThrottlingCollected: make(chan bool, 3),
 		rowCopyComplete:          make(chan error),
 		// Buffered with capacity 1; the send uses overwrite-oldest semantics
@@ -537,6 +539,9 @@ func (mgtr *Migrator) Migrate() (err error) {
 		} else {
 			mgtr.migrationContext.Log.Infof("Attempting to execute alter with ALGORITHM=INSTANT")
 			if err := mgtr.applier.AttemptInstantDDL(); err == nil {
+				// The migration is over: without this the streamer keeps reconnecting
+				// after finalCleanup closes it, since canStopStreaming reads this flag.
+				atomic.StoreInt64(&mgtr.migrationContext.CutOverCompleteFlag, 1)
 				if err := mgtr.finalCleanup(); err != nil {
 					return nil
 				}

go/logic/migrator_test.go

@@ -1294,6 +1294,55 @@ func (suite *MigratorTestSuite) TestCutOverLossDataCaseLockGhostBeforeRename() {
 	suite.Require().Equal("CREATE TABLE `testing` (\n  `id` int NOT NULL,\n  `name` varchar(64) DEFAULT NULL,\n  `foobar` varchar(255) DEFAULT NULL,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci", createTableSQL)
 }
 
+// TestMigrateInstantDDLReturns locks down the deadlock in which a migration that
+// --attempt-instant-ddl completes never returns: the changelog listener parks on the
+// GhostTableMigrated send that this path never reads, and because notifyListeners used
+// to hold listenersMutex across that call, the binlog parser's shouldDecodeRowsEvent
+// stalled behind it and BinlogSyncer.Close in finalCleanup waited on the parser.
+func (suite *MigratorTestSuite) TestMigrateInstantDDLReturns() {
+	ctx := context.Background()
+
+	_, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, name VARCHAR(64))", getTestTableName()))
+	suite.Require().NoError(err)
+
+	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
+	suite.Require().NoError(err)
+
+	migrationContext := newTestMigrationContext()
+	migrationContext.ApplierConnectionConfig = connectionConfig
+	migrationContext.InspectorConnectionConfig = connectionConfig
+	migrationContext.SetConnectionConfig("innodb")
+	migrationContext.DropServeSocket = true
+	migrationContext.InitiallyDropOldTable = true
+	migrationContext.AttemptInstantDDL = true
+	migrationContext.AlterStatementOptions = "ADD COLUMN author varchar(255)"
+
+	migrator := NewMigrator(migrationContext, "0.0.0")
+
+	migrateDone := make(chan error, 1)
+	go func() { migrateDone <- migrator.Migrate() }()
+
+	select {
+	case err := <-migrateDone:
+		suite.Require().NoError(err)
+	case <-time.After(60 * time.Second):
+		// Cancelling releases the parked changelog send, which unwinds the deadlock
+		// so the migration goroutine does not hold its connections and its locks on
+		// the test table for the rest of the suite.
+		migrationContext.CancelContext()
+		select {
+		case <-migrateDone:
+		case <-time.After(30 * time.Second):
+		}
+		suite.Require().Fail("Migrate did not return after the instant DDL succeeded")
+	}
+
+	var tableName, createTableSQL string
+	err = suite.db.QueryRow("SHOW CREATE TABLE "+getTestTableName()).Scan(&tableName, &createTableSQL)
+	suite.Require().NoError(err)
+	suite.Require().Contains(createTableSQL, "`author` varchar(255)")
+}
+
 func (suite *MigratorTestSuite) TestRevertEmpty() {
 	ctx := context.Background()
 

go/logic/streamer.go

@@ -101,18 +101,26 @@ func (es *EventsStreamer) shouldDecodeRowsEvent(databaseName, tableName string)
 
 // notifyListeners will notify relevant listeners with given DML event. Only
 // listeners registered for changes on the table on which the DML operates are notified.
+//
+// The matching listeners are collected under listenersMutex and invoked after it is
+// released. A synchronous listener can block indefinitely, and shouldDecodeRowsEvent
+// takes the same mutex on the binlog parser's goroutine: holding it across the call
+// stalls the parser, which in turn stalls BinlogSyncer.Close.
 func (es *EventsStreamer) notifyListeners(binlogEntry *binlog.BinlogEntry) {
 	es.listenersMutex.Lock()
-	defer es.listenersMutex.Unlock()
-
+	matched := make([]*BinlogEventListener, 0, len(es.listeners))
 	for _, listener := range es.listeners {
-		listener := listener
 		if !strings.EqualFold(listener.databaseName, binlogEntry.DmlEvent.DatabaseName) {
 			continue
 		}
 		if !strings.EqualFold(listener.tableName, binlogEntry.DmlEvent.TableName) {
 			continue
 		}
+		matched = append(matched, listener)
+	}
+	es.listenersMutex.Unlock()
+
+	for _, listener := range matched {
 		if listener.async {
 			go func() {
 				listener.onDmlEvent(binlogEntry)