feat(move-tables): implement T0-T6 cooperative cutover protocol (#8209)

Postpone gate (flag file + socket, no heartbeat-lag check), RENAME on
the privileged source connection with no retry, drain-GTID capture on
the same connection, 100ms GTID-containment drain poll with timeout,
CutOverCompleteFlag set before return, on-success hook.

Drain GTID persistence is #8210. Hook env vars are #8211.

Ref: database-infrastructure#8209

Author: womoruyi

go/logic/migrator.go

@@ -20,13 +20,26 @@ import (
 	"github.com/github/gh-ost/go/binlog"
 	"github.com/github/gh-ost/go/mysql"
 	"github.com/github/gh-ost/go/sql"
+
+	"github.com/openark/golib/sqlutils"
 )
 
 var (
 	ErrMigratorUnsupportedRenameAlter = errors.New("alter statement seems to RENAME the table. This is not supported, and you should run your RENAME outside gh-ost")
 	ErrMigrationNotAllowedOnMaster    = errors.New("it seems like this migration attempt to run directly on master. Preferably it would be executed on a replica (this reduces load from the master). To proceed please provide --allow-on-master")
 	RetrySleepFn                      = time.Sleep
 	checkpointTimeout                 = 2 * time.Second
+
+	// moveTablesCutOverDrainTimeout caps how long T3 of the move-tables
+	// cooperative cutover (#8209) waits for the applier to catch up to the
+	// drain GTID captured at T2. 60s is a starting default chosen over
+	// CutOverLockTimeoutSeconds (which is capped at 1-10s by
+	// SetCutOverLockTimeoutSeconds and would be too short for a real drain
+	// under load). Up for review with Daniel/Eric in the PR.
+	moveTablesCutOverDrainTimeout = 60 * time.Second
+	// moveTablesCutOverDrainPollInterval is the per-iteration sleep in T3's
+	// drain poll. 100ms per move_table_mode.md §1.5.
+	moveTablesCutOverDrainPollInterval = 100 * time.Millisecond
 )
 
 type ChangelogState string
@@ -893,9 +906,6 @@ func (mgtr *Migrator) MoveTables() (err error) {
 	if err := mgtr.finalCleanup(); err != nil {
 		return nil
 	}
-	if err := mgtr.hooksExecutor.OnSuccess(false); err != nil {
-		return err
-	}
 	mgtr.migrationContext.Log.Infof("Done moving tables %v from %s to %s (%s)",
 		mgtr.migrationContext.MoveTables.TableNames, sql.EscapeName(mgtr.migrationContext.DatabaseName),
 		sql.EscapeName(mgtr.migrationContext.MoveTables.TargetDatabase), mgtr.migrationContext.MoveTables.TargetHost)
@@ -915,69 +925,137 @@ func (mgtr *Migrator) MoveTables() (err error) {
 // single-server assumption that no longer holds when the applier writes target
 // and the streamer reads source. Each is replaced or dropped here.
 //
-// This commit (1 of 3) lays out the protocol as labeled comment blocks only.
-// The protocol logic (RENAME, GTID capture, drain poll, flag set, hooks) lands
-// in commit 2. Tests land in commit 3.
+// Crash safety (persisting the drain GTID before T3) is #8210. Enriched hook
+// env vars (GH_OST_DRAIN_GTID, GH_OST_TARGET_*) are #8211. Target-side
+// throttling is #8212. None of those are wired here.
 func (mgtr *Migrator) moveTablesCutOver() (err error) {
 	if mgtr.migrationContext.Noop {
 		mgtr.migrationContext.Log.Debugf("Noop operation; not really moving tables")
 		return nil
 	}
 
 	// ----- Postpone gate (precedes T0) -----
-	// Reuses sleepWhileTrue with a predicate that KEEPS the postpone-flag-file
-	// branch from standard cutOver() (migrator.go:965-984) and DROPS the
-	// heartbeat-lag branch (lines 958-964). The heartbeat-lag check is dropped
-	// because move-tables mode disables _ghc writes (#8206), so
-	// TimeSinceLastHeartbeatOnChangelog() returns ~58 years (time.Since(zero))
-	// and would deadlock the gate forever.
-	// TODO(#8209 commit 2): implement postpone gate (flag-file + unpostpone
-	// socket command, with OnBeginPostponed firing once).
+	// Mirrors standard cutOver()'s sleepWhileTrue postpone structure but DROPS the
+	// heartbeat-lag branch: move-tables mode disables _ghc heartbeat writes (#8206),
+	// so TimeSinceLastHeartbeatOnChangelog() returns ~58 years (time.Since(zero))
+	// and would deadlock the gate forever. KEEPS the postpone-flag-file +
+	// unpostpone-socket gate because per coop_cutover.md §1.1 P4, operator-removes-
+	// postpone is the trigger for the entire cutover phase.
+	mgtr.migrationContext.Log.Debugf("checking for cut-over postpone")
+	if err := mgtr.sleepWhileTrue(func() (bool, error) {
+		if mgtr.migrationContext.PostponeCutOverFlagFile == "" {
+			return false, nil
+		}
+		if atomic.LoadInt64(&mgtr.migrationContext.UserCommandedUnpostponeFlag) > 0 {
+			atomic.StoreInt64(&mgtr.migrationContext.UserCommandedUnpostponeFlag, 0)
+			return false, nil
+		}
+		if base.FileExists(mgtr.migrationContext.PostponeCutOverFlagFile) {
+			if atomic.LoadInt64(&mgtr.migrationContext.IsPostponingCutOver) == 0 {
+				if err := mgtr.hooksExecutor.OnBeginPostponed(); err != nil {
+					return true, err
+				}
+			}
+			atomic.StoreInt64(&mgtr.migrationContext.IsPostponingCutOver, 1)
+			return true, nil
+		}
+		return false, nil
+	}); err != nil {
+		return err
+	}
+	atomic.StoreInt64(&mgtr.migrationContext.IsPostponingCutOver, 0)
+	mgtr.migrationContext.Log.Debugf("checking for cut-over postpone: complete")
 
 	// ----- T0: on-before-cut-over hook -----
-	// Fires gh-ost-on-before-cut-over* via mgtr.hooksExecutor.OnBeforeCutOver().
-	// Non-zero hook exit aborts cutover BEFORE any source DDL.
-	// TODO(#8209 commit 2): implement T0 hook call with error propagation.
+	// Non-zero hook exit aborts cutover BEFORE any source DDL fires.
+	if err := mgtr.hooksExecutor.OnBeforeCutOver(); err != nil {
+		return fmt.Errorf("on-before-cut-over hook failed: %w", err)
+	}
 
 	// ----- T1: RENAME source table -----
-	// Connection: mgtr.inspector.db (privileged source connection per
-	// move_table_mode.md §1.5; same connection as T2 so the RENAME's own GTID
-	// is visible in @@gtid_executed when T2 reads it).
-	// Exec: sqlutils.ExecNoPrepare with no retry — RENAME is not idempotent;
-	// a partial success leaves the table already renamed and a retry would
-	// fail. On error: return wrapped err; operator re-runs the whole hook.
-	// SQL shape: RENAME TABLE `<source_db>`.`<table>` TO `<source_db>`.`_<table>_del`
-	// TODO(#8209 commit 2): implement RENAME via inspector.db.
-
-	// ----- T2: capture @@gtid_executed -----
-	// Same connection as T1. Parse the result into a *mysql.GTIDBinlogCoordinates
-	// (the drain GTID — a closed superset of every committed source write to
-	// the migrated table, per coop_cutover.md §3.2 steps 1-3).
-	// TODO(#8209 commit 2): implement GTID capture on the inspector connection.
+	// Issued on mgtr.inspector.db (the privileged source connection per
+	// move_table_mode.md §1.5). No retry: RENAME is not idempotent — a partial
+	// success leaves the table already renamed and a retry would fail. The
+	// operator re-runs the whole hook chain on failure.
+	sourceDB := mgtr.migrationContext.DatabaseName
+	sourceTable := mgtr.migrationContext.OriginalTableName
+	delTable := mgtr.migrationContext.GetOldTableName()
+	renameQuery := fmt.Sprintf("RENAME TABLE %s.%s TO %s.%s",
+		sql.EscapeName(sourceDB), sql.EscapeName(sourceTable),
+		sql.EscapeName(sourceDB), sql.EscapeName(delTable))
+	mgtr.migrationContext.Log.Infof("T1: renaming source table: %s", renameQuery)
+	if _, err := sqlutils.ExecNoPrepare(mgtr.inspector.db, renameQuery); err != nil {
+		return fmt.Errorf("RENAME failed: %w", err)
+	}
+
+	// ----- T2: capture @@gtid_executed on the SAME *sql.DB handle as T1 -----
+	// Design doc specifies @@gtid_executed; using @@global.gtid_executed because
+	// the unqualified form resolves to session scope (only this session's GTIDs),
+	// while drain requires the global GTID set (all committed transactions).
+	// Design: https://github.com/github/gh-ost-tablemove-poc/blob/9dc6df75c4c88ff473906a497836c7518f5614ec/design/coop_cutover.md#32-correctness-verification-for-p4
+	var drainGTIDStr string
+	if err := mgtr.inspector.db.QueryRow("select @@global.gtid_executed").Scan(&drainGTIDStr); err != nil {
+		return fmt.Errorf("drain GTID capture failed: %w", err)
+	}
+	drainGTID, err := mysql.NewGTIDBinlogCoordinates(drainGTIDStr)
+	if err != nil {
+		return fmt.Errorf("drain GTID parse failed: %w", err)
+	}
+	mgtr.migrationContext.Log.Infof("T2: captured drain GTID: %s", drainGTID.DisplayString())
 
 	// ----- T3: drain poll -----
-	// Loop ~100ms ticks (per move_table_mode.md §1.5) until
-	// applier.CurrentCoordinates contains the drain GTID, comparing via
-	// GTIDBinlogCoordinates.SmallerThan (go/mysql/binlog_gtid.go). Reads of
-	// CurrentCoordinates MUST hold CurrentCoordinatesMutex (applier.go:75).
-	// Timeout after CutOverLockTimeoutSeconds; on timeout return an error
-	// naming the drain.
-	// TODO(#8209 commit 2): implement drain poll with mutex-protected reads.
+	// Wait until applier.CurrentCoordinates catches up to drainGTID. The drain
+	// is complete when the applier's coords are not strictly smaller than the
+	// drain target (i.e. the applier contains every GTID in drainGTID). Reads
+	// of CurrentCoordinates hold the mutex per applier.go:75. Per-iteration
+	// logging is Debug only to avoid spamming Info on a hot loop.
+	//
+	// Timeout is a named constant (moveTablesCutOverDrainTimeout = 60s). NOTE:
+	// the internalization doc suggested reusing CutOverLockTimeoutSeconds, but
+	// that's capped at 1-10s (context.go:SetCutOverLockTimeoutSeconds) — too
+	// short for a real drain under load. 60s default is up for review with
+	// Daniel/Eric; flagged in the PR description.
+	mgtr.migrationContext.Log.Infof("T3: draining applier to drain GTID (timeout %s, poll %s)",
+		moveTablesCutOverDrainTimeout, moveTablesCutOverDrainPollInterval)
+	drainCtx, cancel := context.WithTimeout(context.Background(), moveTablesCutOverDrainTimeout)
+	defer cancel()
+	ticker := time.NewTicker(moveTablesCutOverDrainPollInterval)
+	defer ticker.Stop()
+	for {
+		mgtr.applier.CurrentCoordinatesMutex.Lock()
+		applierCoords := mgtr.applier.CurrentCoordinates
+		mgtr.applier.CurrentCoordinatesMutex.Unlock()
+		if applierCoords != nil && !applierCoords.IsEmpty() && !applierCoords.SmallerThan(drainGTID) {
+			mgtr.migrationContext.Log.Infof("T3: drain complete; applier caught up to drain GTID")
+			break
+		}
+		mgtr.migrationContext.Log.Debugf("T3: applier still behind drain GTID, polling")
+		select {
+		case <-drainCtx.Done():
+			return fmt.Errorf("drain poll timed out after %s: applier did not catch up to drain GTID", moveTablesCutOverDrainTimeout)
+		case <-ticker.C:
+			// next iteration
+		}
+	}
 
 	// ----- T4: set CutOverCompleteFlag -----
-	// atomic.StoreInt64(&mgtr.migrationContext.CutOverCompleteFlag, 1).
-	// MUST be set BEFORE T5 so the streamer's StreamEvents(canStopStreaming)
-	// loop can wind down in parallel with the (potentially slow) on-success
-	// hook. Forgetting this flag has no visible failure mode at the call site
-	// — the run silently hangs after cutover.
-	// TODO(#8209 commit 2): implement T4 flag set.
-
-	// ----- T5: fire on-success hook -----
-	// mgtr.hooksExecutor.OnSuccess(false). Hook unlocks user_rw@target via
-	// db-user-management and flips the write_cutover? feature flag.
-	// Standard env vars only — GH_OST_DRAIN_GTID + GH_OST_TARGET_* are #8211
-	// (1.7), not this PR.
-	// TODO(#8209 commit 2): implement T5 hook call with error propagation.
+	// MUST be set before T5 so the streamer's canStopStreaming loop (migrator.go:256)
+	// can wind down in parallel with the (potentially slow) on-success hook.
+	// Forgetting this has no visible failure at the call site — the run silently
+	// hangs after cutover because eventsStreamer.StreamEvents() never returns.
+	atomic.StoreInt64(&mgtr.migrationContext.CutOverCompleteFlag, 1)
+	mgtr.migrationContext.Log.Debugf("T4: CutOverCompleteFlag set")
+
+	// ----- T5: on-success hook -----
+	// Hook unlocks user_rw@target via db-user-management and flips the
+	// write_cutover? feature flag. Standard env vars only — GH_OST_DRAIN_GTID +
+	// GH_OST_TARGET_* are #8211 (1.7), not this PR. The pre-protocol placeholder
+	// OnSuccess call that used to live in MoveTables() (after finalCleanup) has
+	// been removed so the hook fires in the order coop_cutover.md §3.2 step 6
+	// requires (T5 between T4 and T6, BEFORE finalCleanup).
+	if err := mgtr.hooksExecutor.OnSuccess(false); err != nil {
+		return fmt.Errorf("on-success hook failed: %w", err)
+	}
 
 	// ----- T6: return nil -----
 	return nil