diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..c82360948 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -267,6 +267,22 @@ func (mgtr *Migrator) consumeRowCopyComplete() { }() } +// waitForGhostTableMigrated blocks until the ghost table has been migrated, or +// until the migration context is cancelled by an abort. The only sender on +// ghostTableMigrated publishes via base.SendWithContext, which stops sending +// once the context is cancelled, so waiting on the channel alone would block +// forever after an abort. +func (mgtr *Migrator) waitForGhostTableMigrated() error { + select { + case <-mgtr.ghostTableMigrated: + mgtr.migrationContext.Log.Debugf("ghost table migrated") + return nil + case <-mgtr.migrationContext.GetContext().Done(): + // Abort cancelled the context + return mgtr.checkAbort() + } +} + func (mgtr *Migrator) canStopStreaming() bool { return atomic.LoadInt64(&mgtr.migrationContext.CutOverCompleteFlag) != 0 } @@ -537,6 +553,17 @@ 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 { + // initiateApplier emits the GhostTableMigrated signal whenever + // !Revert && !Resume, regardless of whether instant DDL succeeds. + // The publisher (onChangelogStateEvent) sends it synchronously while + // holding EventsStreamer.listenersMutex, so it must be drained here + // or the send blocks forever, and finalCleanup then deadlocks closing + // the binlog reader, which needs the same mutex. + if !mgtr.migrationContext.Resume { + if err := mgtr.waitForGhostTableMigrated(); err != nil { + return err + } + } if err := mgtr.finalCleanup(); err != nil { return nil } @@ -554,8 +581,9 @@ func (mgtr *Migrator) Migrate() (err error) { initialLag, _ := mgtr.inspector.getReplicationLag() if !mgtr.migrationContext.Resume { mgtr.migrationContext.Log.Infof("Waiting for ghost table to be migrated. Current lag is %+v", initialLag) - <-mgtr.ghostTableMigrated - mgtr.migrationContext.Log.Debugf("ghost table migrated") + if err := mgtr.waitForGhostTableMigrated(); err != nil { + return err + } } // Yay! We now know the Ghost and Changelog tables are good to examine! // When running on replica, this means the replica has those tables. When running diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index ad068691c..6c03b2521 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -1566,6 +1566,63 @@ func TestAbort_DuringInspection(t *testing.T) { } } +func TestAbort_DuringGhostTableWait(t *testing.T) { + migrationContext := base.NewMigrationContext() + migrator := NewMigrator(migrationContext, "1.0.0") + + // Start listenOnPanicAbort + go migrator.listenOnPanicAbort() + + // Give listenOnPanicAbort time to start + time.Sleep(20 * time.Millisecond) + + // Simulate an abort raised while Migrate() waits for the ghost table + testErr := errors.New("ghost table wait aborted") + go func() { + time.Sleep(10 * time.Millisecond) + select { + case migrationContext.PanicAbort <- testErr: + case <-migrationContext.GetContext().Done(): + } + }() + + // Nothing sends on ghostTableMigrated, mirroring an abort that cancels the + // context before the changelog event arrives: the real sender publishes via + // base.SendWithContext, which stops sending once the context is cancelled. + // Waiting on the channel alone would block here forever. + done := make(chan error, 1) + go func() { + done <- migrator.waitForGhostTableMigrated() + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("Expected an error once the abort cancelled the context") + } + if err.Error() != "ghost table wait aborted" { + t.Errorf("Expected 'ghost table wait aborted', got %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Expected waitForGhostTableMigrated to return after the abort cancelled the context") + } +} + +func TestWaitForGhostTableMigrated(t *testing.T) { + migrationContext := base.NewMigrationContext() + migrator := NewMigrator(migrationContext, "1.0.0") + + // ghostTableMigrated is unbuffered, so the send must be async + go func() { + time.Sleep(10 * time.Millisecond) + migrator.ghostTableMigrated <- true + }() + + if err := migrator.waitForGhostTableMigrated(); err != nil { + t.Fatalf("Expected no error, got %v", err) + } +} + func TestAbort_DuringStreaming(t *testing.T) { migrationContext := base.NewMigrationContext() migrator := NewMigrator(migrationContext, "1.0.0") diff --git a/go/logic/streamer_test.go b/go/logic/streamer_test.go index 5b28c47c2..c04a4eb8a 100644 --- a/go/logic/streamer_test.go +++ b/go/logic/streamer_test.go @@ -7,7 +7,9 @@ import ( "testing" "time" + "github.com/github/gh-ost/go/base" "github.com/github/gh-ost/go/binlog" + "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/mysql" @@ -287,6 +289,70 @@ func TestEventsStreamerShouldDecodeRowsEvent(t *testing.T) { } } +// TestEventsStreamerInstantDDLDeadlockIsResolvedByDraining reproduces the +// deadlock that occurs when the GhostTableMigrated signal is never received on +// the instant-DDL success path: notifyListeners invokes the changelog listener +// synchronously while holding listenersMutex, the listener blocks on an +// unbuffered send until something receives, and shouldDecodeRowsEvent needs the +// same mutex to run. Without a receiver, both stay blocked forever. It proves +// that receiving the signal (what Migrator.waitForGhostTableMigrated does on +// the instant-DDL success path) resolves it. +func TestEventsStreamerInstantDDLDeadlockIsResolvedByDraining(t *testing.T) { + migrationContext := newTestMigrationContext() + streamer := NewEventsStreamer(migrationContext) + + ghostTableMigrated := make(chan bool) // unbuffered, mirrors Migrator.ghostTableMigrated + + err := streamer.AddListener(false, testMysqlDatabase, testMysqlTableName, func(event *binlog.BinlogEntry) error { + return base.SendWithContext(migrationContext.GetContext(), ghostTableMigrated, true) + }) + require.NoError(t, err) + + entry := &binlog.BinlogEntry{ + DmlEvent: binlog.NewBinlogDMLEvent(testMysqlDatabase, testMysqlTableName, binlog.InsertDML), + } + + notifyReturned := make(chan struct{}) + go func() { + streamer.notifyListeners(entry) // holds listenersMutex, blocks on the listener's send + close(notifyReturned) + }() + + decodeReturned := make(chan bool, 1) + go func() { + decodeReturned <- streamer.shouldDecodeRowsEvent(testMysqlDatabase, testMysqlTableName) + }() + + // Both goroutines are blocked and cannot progress until the signal is received: + // notifyListeners on the send, shouldDecodeRowsEvent on the mutex. + select { + case <-notifyReturned: + t.Fatal("notifyListeners returned before receiving; the test no longer reproduces the deadlock") + case <-time.After(200 * time.Millisecond): + } + + // The fix: the instant-DDL path waits for the signal before finalCleanup. + select { + case <-ghostTableMigrated: + case <-time.After(2 * time.Second): + t.Fatal("GhostTableMigrated signal was never published") + } + + // Receiving releases the listener, so notifyListeners returns and frees the + // mutex, which unblocks the decode path. + select { + case <-notifyReturned: + case <-time.After(2 * time.Second): + t.Fatal("notifyListeners still blocked after receive: deadlock not resolved") + } + select { + case decoded := <-decodeReturned: + require.True(t, decoded, "registered table should be decoded") + case <-time.After(2 * time.Second): + t.Fatal("shouldDecodeRowsEvent still blocked after receive: mutex was not released") + } +} + func TestEventsStreamer(t *testing.T) { if testing.Short() { t.Skip("skipping events streamer test suite in short mode")