perf(cutover): drop two 1s sleeps from the locked cut-over window

Both sleeps happened while the original table was write-locked, so they
were pure table downtime:

- executeWriteFuncs slept 1s whenever both queues were empty, which is the
  steady state once row copy is done. The cut-over sentinel arrives on
  applyEventsQueue and had to wait that sleep out. It now blocks on both
  queues with a 1s timeout instead.

- waitForRename watched for the blocking RENAME through retryOperation,
  which backs off a flat 1s. The first check usually runs before the RENAME
  shows up. It now checks every 10ms, via a new retryOperationWithInterval
  (retryOperation with the attempt count and wait made explicit).

"Lock & rename duration" over the 73 localtests that reach cut-over. Before,
it was ~1s on essentially every cut-over:

  mariadb:11.8   p50  9ms   p90 21ms   p99 90ms   max 90ms
  mysql:8.4.3    p50 23ms   p90 31ms   p99 96ms   max 96ms

The tail is not this code path. It is waitForEventsUpToLock: on both
flavours the same four cases, which happen to see no DML during the
migration, spend 54-77ms waiting for the sentinel to come back through an
otherwise idle binlog stream. No case with concurrent DML exceeds 50ms.
That delay reproduces with a plain binlog reader, so it is in binlog
delivery, not in gh-ost.

TestCutOverLossDataCaseLockGhostBeforeRename now locks the ghost table
before un-postponing, instead of relying on cut-over being slow.

Fixes #1630

Author: conf

go/logic/migrator.go

@@ -165,13 +165,18 @@ func (mgtr *Migrator) retryBatchCopyWithHooks(operation func() error, notFatalHi
 // retryOperation attempts up to `count` attempts at running given function,
 // exiting as soon as it returns with non-error.
 func (mgtr *Migrator) retryOperation(operation func() error, notFatalHint ...bool) (err error) {
-	maxRetries := int(mgtr.migrationContext.MaxRetries())
-	for i := 0; i < maxRetries; i++ {
+	return mgtr.retryOperationWithInterval(operation, int(mgtr.migrationContext.MaxRetries()), time.Second, notFatalHint...)
+}
+
+// retryOperationWithInterval is `retryOperation` with an explicit attempt count and
+// wait between attempts. Callers that run while tables are locked use a sub-second
+// interval, where the default 1s wait would be pure table downtime.
+func (mgtr *Migrator) retryOperationWithInterval(operation func() error, attempts int, interval time.Duration, notFatalHint ...bool) (err error) {
+	for i := 0; i < attempts; i++ {
 		if i != 0 {
 			// sleep after previous iteration
-			sleepDuration := 1 * time.Second
-			metrics.RecordSleep(mgtr.migrationContext.Metrics, "retry_backoff", sleepDuration)
-			RetrySleepFn(sleepDuration)
+			metrics.RecordSleep(mgtr.migrationContext.Metrics, "retry_backoff", interval)
+			RetrySleepFn(interval)
 		}
 		// Check for abort/context cancellation before each retry
 		if abortErr := mgtr.checkAbort(); abortErr != nil {
@@ -1119,8 +1124,21 @@ func (mgtr *Migrator) atomicCutOver() (err error) {
 		}
 		return mgtr.applier.ExpectProcess(renameSessionId, "metadata lock", "rename")
 	}
-	// Wait for the RENAME to appear in PROCESSLIST
-	if err := mgtr.retryOperation(waitForRename, true); err != nil {
+	// Wait for the RENAME to appear in PROCESSLIST. The first poll usually loses the
+	// race against the RENAME registering its metadata-lock wait, and this runs with
+	// the original table write-locked -- so poll fast rather than paying
+	// retryOperation's flat 1s backoff in table downtime. What we wait on is a
+	// statement starting on an already-open connection: a round-trip, not seconds.
+	//
+	// The RENAME runs with lock_wait_timeout=CutOverLockTimeoutSeconds (see
+	// Applier.AtomicCutoverRename), so past that it has errored out and set
+	// tableRenameKnownToHaveFailed -- at which point waitForRename returns
+	// immediately. Budget twice that, so the flag always wins and running out of
+	// attempts is unreachable in practice.
+	const renamePollInterval = 10 * time.Millisecond
+	renameWaitTimeout := 2 * time.Duration(mgtr.migrationContext.CutOverLockTimeoutSeconds) * time.Second
+	renamePollAttempts := int(renameWaitTimeout / renamePollInterval)
+	if err := mgtr.retryOperationWithInterval(waitForRename, renamePollAttempts, renamePollInterval, true); err != nil {
 		metrics.RecordCutOverPhase(mgtr.migrationContext.Metrics, metrics.CutOverPhaseMagicRename, time.Since(phaseStartTime), err)
 		// Abort! Release the lock
 		okToUnlockTable <- true
@@ -1936,7 +1954,18 @@ func (mgtr *Migrator) executeWriteFuncs() error {
 			}
 		default:
 			{
+				// Nothing was immediately available on the events queue. Block until one
+				// of the queues has work instead of sleeping a fixed second: during
+				// cut-over the AllEventsUpToLockProcessed sentinel arrives on
+				// applyEventsQueue while the tables are locked, and an unconditional
+				// sleep adds up to a full second of lock time (issue #1630).
 				select {
+				case eventStruct := <-mgtr.applyEventsQueue:
+					{
+						if err := mgtr.onApplyEventStruct(eventStruct); err != nil {
+							return err
+						}
+					}
 				case copyRowsFunc := <-mgtr.copyRowsQueue:
 					{
 						copyRowsStartTime := time.Now()
@@ -1956,12 +1985,11 @@ func (mgtr *Migrator) executeWriteFuncs() error {
 							}
 						}
 					}
-				default:
+				case <-time.After(time.Second):
 					{
-						// Hmmmmm... nothing in the queue; no events, but also no row copy.
-						// This is possible upon load. Let's just sleep it over.
-						mgtr.migrationContext.Log.Debugf("Getting nothing in the write queue. Sleeping...")
-						time.Sleep(time.Second)
+						// Nothing in the queue; no events, but also no row copy.
+						// Loop around to re-check abort/throttle state.
+						mgtr.migrationContext.Log.Debugf("Getting nothing in the write queue. Waiting...")
 					}
 				}
 			}

go/logic/migrator_test.go

@@ -1250,19 +1250,23 @@ func (suite *MigratorTestSuite) TestCutOverLossDataCaseLockGhostBeforeRename() {
 	}()
 
 	time.Sleep(2 * time.Second)
-	//nolint:dogsled
-	_, filename, _, _ := runtime.Caller(0)
-	err = os.Remove(filepath.Join(filepath.Dir(filename), "../../tmp/ghost.postpone.flag"))
-	if err != nil {
-		suite.Require().NoError(err)
-	}
-	time.Sleep(1 * time.Second)
+
+	// Hold a read lock on the ghost table *before* un-postponing: cut-over completes
+	// in milliseconds, so grabbing the lock after the flag removal is a race.
 	go func() {
 		holdConn, err := suite.db.Conn(ctx)
 		suite.Require().NoError(err)
 		_, err = holdConn.ExecContext(ctx, "SELECT *, sleep(2) FROM test._testing_gho WHERE id = 1")
 		suite.Require().NoError(err)
 	}()
+	time.Sleep(200 * time.Millisecond)
+
+	//nolint:dogsled
+	_, filename, _, _ := runtime.Caller(0)
+	err = os.Remove(filepath.Join(filepath.Dir(filename), "../../tmp/ghost.postpone.flag"))
+	if err != nil {
+		suite.Require().NoError(err)
+	}
 
 	dmlConn, err := suite.db.Conn(ctx)
 	suite.Require().NoError(err)