From 3128803754820b0c7764952ffeb906036de3dac6 Mon Sep 17 00:00:00 2001 From: Danny Xu Date: Mon, 2 Nov 2026 12:15:20 +0000 Subject: [PATCH 1/2] Do not deadlock when instant DDL completes the migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A migration that `--attempt-instant-ddl` completes never returns. Three goroutines form a cycle: - the changelog listener parks on the unbuffered `GhostTableMigrated` send, which the instant-DDL path returns without ever reading; - `notifyListeners` holds `listenersMutex` across that synchronous call, so `shouldDecodeRowsEvent` — called on the binlog parser's goroutine since #1699 — stalls on the same mutex; - `BinlogSyncer.Close`, which `finalCleanup` calls, waits for that parser goroutine. `Migrate()` therefore blocks until the migration context is cancelled. `notifyListeners` now collects the matching listeners under the mutex and invokes them after releasing it, and `ghostTableMigrated` gets a one-slot buffer, since the changelog carries exactly one such state per migration. `--attempt-instant-ddl` defaults to off and the short-circuit had no test, which is why this went unnoticed; it was found by a downstream consumer that enables the flag for every migration. Co-Authored-By: Claude Opus 5 --- go/logic/migrator.go | 14 +++++++------ go/logic/migrator_test.go | 41 +++++++++++++++++++++++++++++++++++++++ go/logic/streamer.go | 14 ++++++++++--- 3 files changed, 60 insertions(+), 9 deletions(-) diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..8f55639cd 100644 --- a/go/logic/migrator.go +++ b/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 diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index ad068691c..c9d1cfe33 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -1294,6 +1294,47 @@ 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): + 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() diff --git a/go/logic/streamer.go b/go/logic/streamer.go index 7d3d00120..ae1bffd99 100644 --- a/go/logic/streamer.go +++ b/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) From 46c90f4fc91aaa0e1509b456fbaf631528beddb3 Mon Sep 17 00:00:00 2001 From: Danny Xu Date: Mon, 2 Nov 2026 12:47:15 +0000 Subject: [PATCH 2/2] Stop the streamer and isolate the test on the instant-DDL path Two follow-ups from review. `canStopStreaming` reads `CutOverCompleteFlag`, which the instant-DDL path left unset. `StreamEvents` therefore treated the close that `finalCleanup` performs as an unexpected error and reconnected after its sleep, leaking a replication connection and goroutine past the point where `Migrate()` had already returned nil. Set the flag once the instant DDL succeeds, before `finalCleanup`, as the revert path already does. `TestMigrateInstantDDLReturns` now cancels the migration context and waits for the worker before failing. `Require().Fail` unwinds only the test goroutine, so on a regression `Migrate()` would stay blocked and hold its connections and locks on the test table for the rest of the suite. Co-Authored-By: Claude Opus 5 --- go/logic/migrator.go | 3 +++ go/logic/migrator_test.go | 8 ++++++++ 2 files changed, 11 insertions(+) diff --git a/go/logic/migrator.go b/go/logic/migrator.go index 8f55639cd..a13c4fb47 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -539,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 } diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index c9d1cfe33..9701b86d4 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -1326,6 +1326,14 @@ func (suite *MigratorTestSuite) TestMigrateInstantDDLReturns() { 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") }