diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..ae78cd536 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -1998,6 +1998,17 @@ func (mgtr *Migrator) executeDMLWriteFuncs() error { func (mgtr *Migrator) finalCleanup() error { atomic.StoreInt64(&mgtr.migrationContext.CleanupImminentFlag, 1) + // The throttler polls the changelog table (`_ghc`) from background + // goroutines. Setting CleanupImminentFlag above stops any *new* polls + // from starting, but one may already be in flight (possibly against a + // lagging replica); wait for it to finish before we drop the table below, + // or it can spuriously fail with "table doesn't exist". The throttler may + // not have been initiated yet (e.g. finalCleanup is reached via the + // instant-DDL path before initiateThrottler runs). + if mgtr.throttler != nil { + mgtr.throttler.WaitForPendingChangelogReads() + } + mgtr.migrationContext.Log.Infof("Writing changelog state: %+v", Migrated) if _, err := mgtr.applier.WriteChangelogState(string(Migrated)); err != nil { return err diff --git a/go/logic/throttler.go b/go/logic/throttler.go index ee6e3d132..100ed0787 100644 --- a/go/logic/throttler.go +++ b/go/logic/throttler.go @@ -10,6 +10,7 @@ import ( "fmt" "net/http" "strings" + "sync" "sync/atomic" "time" @@ -55,6 +56,14 @@ type Throttler struct { inspector *Inspector finishedMigrating int64 + // pendingChangelogReads tracks throttler goroutines that are reading from + // the changelog table (`_ghc`), so that finalCleanup can wait for them to + // finish before dropping that table. Without this, a read that was + // already in flight when cleanup began can run into "table doesn't + // exist" once the drop lands (or, for reads issued against a replica, + // once the drop has replicated there). + pendingChangelogReads sync.WaitGroup + throttleStartedAt time.Time throttleStartedReason string throttleActiveEmitted time.Time @@ -180,7 +189,16 @@ func (thlr *Throttler) collectReplicationLag(firstThrottlingCollected chan<- boo if atomic.LoadInt64(&thlr.finishedMigrating) > 0 { return } - go collectFunc() + if atomic.LoadInt64(&thlr.migrationContext.CleanupImminentFlag) > 0 { + // Cleanup (which drops the changelog table) is about to start or + // already in progress; don't kick off any more reads against it. + return + } + thlr.pendingChangelogReads.Add(1) + go func() { + defer thlr.pendingChangelogReads.Done() + collectFunc() + }() } } @@ -263,6 +281,11 @@ func (thlr *Throttler) collectControlReplicasLag() { if atomic.LoadInt64(&thlr.finishedMigrating) > 0 { return } + if atomic.LoadInt64(&thlr.migrationContext.CleanupImminentFlag) > 0 { + // Cleanup (which drops the changelog table) is about to start or + // already in progress; don't kick off any more reads against it. + return + } if counter%relaxedFactor == 0 { // we only check if we wish to be aggressive once per second. The parameters for being aggressive // do not typically change at all throughout the migration, but nonetheless we check them. @@ -271,8 +294,13 @@ func (thlr *Throttler) collectControlReplicasLag() { shouldReadLagAggressively = (maxLagMillisecondsThrottleThreshold < 1000) } if counter == 0 || shouldReadLagAggressively { - // We check replication lag every so often, or if we wish to be aggressive + // We check replication lag every so often, or if we wish to be aggressive. + // checkControlReplicasLag blocks until all its replica reads complete, so + // track it as pending to let finalCleanup wait it out before dropping the + // changelog table. + thlr.pendingChangelogReads.Add(1) checkControlReplicasLag() + thlr.pendingChangelogReads.Done() } counter++ } @@ -563,3 +591,11 @@ func (thlr *Throttler) Teardown() { thlr.migrationContext.Log.Debugf("Tearing down...") atomic.StoreInt64(&thlr.finishedMigrating, 1) } + +// WaitForPendingChangelogReads blocks until any throttler goroutines that were +// already reading from the changelog table (`_ghc`) when cleanup began have +// finished. Callers must set CleanupImminentFlag first, so that no further +// reads get started; this only needs to wait out ones already in flight. +func (thlr *Throttler) WaitForPendingChangelogReads() { + thlr.pendingChangelogReads.Wait() +} diff --git a/go/logic/throttler_test.go b/go/logic/throttler_test.go index 01104805b..2a7b853d2 100644 --- a/go/logic/throttler_test.go +++ b/go/logic/throttler_test.go @@ -159,3 +159,95 @@ func TestRecordThrottleMetricsEmitsOneIntervalMetricOnThrottleExit(t *testing.T) assert.Equal(t, []string{"reason:commanded by user"}, spy.tags[3]) assert.Equal(t, []string{"reason:commanded by user"}, spy.tags[4]) } + +// Regression tests for https://github.com/github/gh-ost/issues/1622: a +// changelog-table (`_ghc`) read that was already in flight when cleanup +// began must be waited out before the table is dropped, or it can fail with +// "table doesn't exist". + +func TestWaitForPendingChangelogReadsReturnsImmediatelyWhenIdle(t *testing.T) { + thlr := newTestThrottler() + + done := make(chan struct{}) + go func() { + thlr.WaitForPendingChangelogReads() + close(done) + }() + + select { + case <-done: + case <-time.After(200 * time.Millisecond): + t.Fatal("WaitForPendingChangelogReads blocked with nothing pending") + } +} + +func TestWaitForPendingChangelogReadsBlocksUntilInFlightReadCompletes(t *testing.T) { + thlr := newTestThrottler() + thlr.pendingChangelogReads.Add(1) + + readDone := make(chan struct{}) + go func() { + time.Sleep(150 * time.Millisecond) + close(readDone) + thlr.pendingChangelogReads.Done() + }() + + waitReturned := make(chan struct{}) + go func() { + thlr.WaitForPendingChangelogReads() + close(waitReturned) + }() + + select { + case <-waitReturned: + t.Fatal("WaitForPendingChangelogReads returned before the in-flight read finished") + case <-time.After(50 * time.Millisecond): + } + + select { + case <-waitReturned: + case <-time.After(1 * time.Second): + t.Fatal("WaitForPendingChangelogReads did not return after the in-flight read finished") + } + <-readDone // sanity: the simulated read did actually complete first +} + +func TestCollectReplicationLagStopsWhenCleanupImminent(t *testing.T) { + thlr := newTestThrottler() + thlr.migrationContext.SetHeartbeatIntervalMilliseconds(5) + // Simulate finalCleanup having already flagged that cleanup (and the + // `_ghc` drop) is imminent, before the collection loop starts ticking. + atomic.StoreInt64(&thlr.migrationContext.CleanupImminentFlag, 1) + + firstCollected := make(chan bool, 1) + loopReturned := make(chan struct{}) + go func() { + thlr.collectReplicationLag(firstCollected) + close(loopReturned) + }() + + select { + case <-firstCollected: + case <-time.After(1 * time.Second): + t.Fatal("collectReplicationLag never signaled its first collection") + } + + select { + case <-loopReturned: + case <-time.After(1 * time.Second): + t.Fatal("collectReplicationLag did not stop once CleanupImminentFlag was set") + } + + // No reads should have been spawned once CleanupImminentFlag was set, so + // waiting for pending reads must return immediately. + done := make(chan struct{}) + go func() { + thlr.WaitForPendingChangelogReads() + close(done) + }() + select { + case <-done: + case <-time.After(200 * time.Millisecond): + t.Fatal("WaitForPendingChangelogReads blocked though no reads should have been in flight") + } +}