From a5243038ec3c8d51da92c3414b886424f28af56c Mon Sep 17 00:00:00 2001 From: Gabriel Gilder Date: Thu, 20 Aug 2026 13:00:57 -0700 Subject: [PATCH 1/2] Handle context cancellation while waiting for ghost table migration Migrate() blocks on an unbuffered receive from ghostTableMigrated while waiting for the ghost table to be created. The only sender is onChangelogStateEvent(), which publishes via base.SendWithContext(). If the migration aborts during this window, abort() cancels the migration context, so SendWithContext() takes its ctx.Done() branch and returns without ever sending. Nothing else writes to the channel, so Migrate() blocks forever: it never returns, its deferred teardown() never runs, finishedMigrating is never set, and the status and throttler tickers -- which exit on finishedMigrating rather than on the context -- keep looping. The process stays alive indefinitely, logging a frozen status line, until it is killed externally. Extract the wait into waitForGhostTableMigrated() and select on the migration context alongside the channel, returning checkAbort() so the original abort error is surfaced rather than a bare context error. The extraction mirrors consumeRowCopyComplete() and makes the behaviour testable without a database. This is the same deadlock, and the same fix, as #1677 applied to consumeRowCopyComplete; ghostTableMigrated is its remaining sibling. TestAbort_DuringGhostTableWait follows the existing TestAbort_* pattern and fails (blocking until its timeout) without this change. Co-Authored-By: Claude Opus 5 --- go/logic/migrator.go | 21 +++++++++++++-- go/logic/migrator_test.go | 57 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..977f00cea 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 } @@ -554,8 +570,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") From fdf29f8916b786d157e46a6ea7d5a814caff5fd5 Mon Sep 17 00:00:00 2001 From: Gabriel Gilder Date: Tue, 22 Sep 2026 09:00:00 -0700 Subject: [PATCH 2/2] Drain GhostTableMigrated on the instant-DDL success path initiateApplier emits the GhostTableMigrated changelog signal whenever !Revert && !Resume, regardless of whether instant DDL succeeds. The instant-DDL success path returned early without ever receiving it, so the publisher (onChangelogStateEvent) blocked forever holding EventsStreamer.listenersMutex, and finalCleanup then deadlocked closing the binlog reader, which needs the same mutex (#1735, #1736). Reuse waitForGhostTableMigrated() here instead of a bare receive, so this doesn't trade the instant-DDL deadlock for the abort-path deadlock on the same channel. Co-Authored-By: Claude Sonnet 5 --- go/logic/migrator.go | 11 +++++++ go/logic/streamer_test.go | 66 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+) diff --git a/go/logic/migrator.go b/go/logic/migrator.go index 977f00cea..c82360948 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -553,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 } 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")