Fix race between throttler changelog reads and changelog table drop

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

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()
+}

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