Fix deadlocks on the GhostTableMigrated changelog signal

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

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")

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")