Make pending unpostpone observable and cancellable

doc/interactive-commands.md

@@ -43,6 +43,8 @@ Both interfaces may serve at the same time. Both respond to simple text command,
 - `no-throttle`: cancel forced suspension (though other throttling reasons may still apply)
 - `postpone-cut-over-flag-file=<path>`: Postpone the [cut-over](cut-over.md) phase, writing a cut over flag file to the given path
 - `unpostpone`: at a time where `gh-ost` is postponing the [cut-over](cut-over.md) phase, instruct `gh-ost` to stop postponing and proceed immediately to cut-over.
+  - Cut-over still waits until the heartbeat lag is below both `--max-lag-millis` and `--cut-over-lock-timeout-seconds`. Until it starts, `status` reports the request, e.g. `State: postponing cut-over, unpostpone requested: waiting for heartbeat lag (4.20s) to drop below 1.50s`.
+- `cancel-unpostpone`: withdraw a pending `unpostpone` that has not started cut-over yet, so `gh-ost` keeps postponing as long as the postpone flag file exists. Replies `There is no pending unpostpone to cancel` if no request is pending or cut-over has already started. Useful for tooling that issues `unpostpone` and wants to give up if the lag does not recover in time.
 - `panic`: immediately panic and abort operation
 
 ### Querying for data

go/base/context.go

@@ -238,6 +238,10 @@ type MigrationContext struct {
 	AbortError error
 	abortMutex *sync.Mutex
 
+	// postponeMutex serializes unpostpone requests, cancellations and their
+	// consumption by the cut-over loop with the end of postponement.
+	postponeMutex sync.Mutex
+
 	Metrics metrics.Emitter
 
 	OriginalTableColumnsOnApplier    *sql.ColumnList
@@ -342,6 +346,48 @@ func NewMigrationContext() *MigrationContext {
 	}
 }
 
+// RequestUnpostpone records an 'unpostpone' request. It returns false, and records
+// nothing, unless cut-over is currently being postponed.
+func (mctx *MigrationContext) RequestUnpostpone() bool {
+	mctx.postponeMutex.Lock()
+	defer mctx.postponeMutex.Unlock()
+	if atomic.LoadInt64(&mctx.IsPostponingCutOver) == 0 {
+		return false
+	}
+	atomic.StoreInt64(&mctx.UserCommandedUnpostponeFlag, 1)
+	return true
+}
+
+// CancelUnpostpone withdraws a pending 'unpostpone' request. It returns false if
+// no request is pending, including when cut-over has already consumed it.
+func (mctx *MigrationContext) CancelUnpostpone() bool {
+	mctx.postponeMutex.Lock()
+	defer mctx.postponeMutex.Unlock()
+	return atomic.CompareAndSwapInt64(&mctx.UserCommandedUnpostponeFlag, 1, 0)
+}
+
+// ConsumeUnpostpone is called by the cut-over loop. If an 'unpostpone' request is
+// pending it consumes it and ends postponement in the same critical section, so a
+// repeated request or a cancellation cannot slip in after cut-over has begun.
+func (mctx *MigrationContext) ConsumeUnpostpone() bool {
+	mctx.postponeMutex.Lock()
+	defer mctx.postponeMutex.Unlock()
+	if !atomic.CompareAndSwapInt64(&mctx.UserCommandedUnpostponeFlag, 1, 0) {
+		return false
+	}
+	atomic.StoreInt64(&mctx.IsPostponingCutOver, 0)
+	return true
+}
+
+// EndPostponingCutOver marks postponement as over and drops any request that
+// arrived after the cut-over loop decided to proceed.
+func (mctx *MigrationContext) EndPostponingCutOver() {
+	mctx.postponeMutex.Lock()
+	defer mctx.postponeMutex.Unlock()
+	atomic.StoreInt64(&mctx.IsPostponingCutOver, 0)
+	atomic.StoreInt64(&mctx.UserCommandedUnpostponeFlag, 0)
+}
+
 func (mctx *MigrationContext) SetConnectionConfig(storageEngine string) error {
 	var transactionIsolation string
 	switch storageEngine {

go/logic/migrator.go

@@ -870,6 +870,51 @@ func (mgtr *Migrator) analyzeGhostTableBeforeCutOver(analyze func() error) error
 	return analyze()
 }
 
+// cutOverHeartbeatLag returns the current changelog heartbeat lag and the limit it
+// must be under before cut-over may begin: the lower of --max-lag-millis and
+// --cut-over-lock-timeout-seconds. tooHigh reports whether the lag exceeds it.
+func (mgtr *Migrator) cutOverHeartbeatLag() (lag, limit time.Duration, tooHigh bool) {
+	lag = mgtr.migrationContext.TimeSinceLastHeartbeatOnChangelog()
+	maxLagMillisecondsThrottle := time.Duration(atomic.LoadInt64(&mgtr.migrationContext.MaxLagMillisecondsThrottleThreshold)) * time.Millisecond
+	cutOverLockTimeout := time.Duration(mgtr.migrationContext.CutOverLockTimeoutSeconds) * time.Second
+	limit = min(maxLagMillisecondsThrottle, cutOverLockTimeout)
+	return lag, limit, lag > limit
+}
+
+// shouldPostponeCutOver is polled once per second before cut-over and returns true
+// while cut-over should keep waiting: either the heartbeat lag is too high, or the
+// postpone flag file exists and no 'unpostpone' command is pending.
+//
+// A pending 'unpostpone' is only consumed once the heartbeat lag allows cut-over to
+// start. Until then it stays visible in the status output and can be withdrawn with
+// the 'cancel-unpostpone' command. Requests, cancellations and consumption are
+// serialized with the end of postponement, so a request is either consumed here or
+// cancelled by the user, never both, and none is accepted once cut-over has begun.
+func (mgtr *Migrator) shouldPostponeCutOver() (bool, error) {
+	if lag, limit, tooHigh := mgtr.cutOverHeartbeatLag(); tooHigh {
+		mgtr.migrationContext.Log.Debugf("current HeartbeatLag (%.2fs) is too high, it needs to be less than both --max-lag-millis and --cut-over-lock-timeout-seconds (%.2fs) to continue", lag.Seconds(), limit.Seconds())
+		return true, nil
+	}
+	if mgtr.migrationContext.PostponeCutOverFlagFile == "" {
+		return false, nil
+	}
+	if mgtr.migrationContext.ConsumeUnpostpone() {
+		return false, nil
+	}
+	if base.FileExists(mgtr.migrationContext.PostponeCutOverFlagFile) {
+		// Postpone file defined and exists!
+		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
+	}
+	mgtr.migrationContext.EndPostponingCutOver()
+	return false, nil
+}
+
 // cutOver performs the final step of migration, based on migration
 // type (on replica? atomic? safe?)
 func (mgtr *Migrator) cutOver() (err error) {
@@ -884,39 +929,10 @@ func (mgtr *Migrator) cutOver() (err error) {
 
 	mgtr.migrationContext.MarkPointOfInterest()
 	mgtr.migrationContext.Log.Debugf("checking for cut-over postpone")
-	if err := mgtr.sleepWhileTrue(
-		"cut_over_postpone",
-		func() (bool, error) {
-			heartbeatLag := mgtr.migrationContext.TimeSinceLastHeartbeatOnChangelog()
-			maxLagMillisecondsThrottle := time.Duration(atomic.LoadInt64(&mgtr.migrationContext.MaxLagMillisecondsThrottleThreshold)) * time.Millisecond
-			cutOverLockTimeout := time.Duration(mgtr.migrationContext.CutOverLockTimeoutSeconds) * time.Second
-			if heartbeatLag > maxLagMillisecondsThrottle || heartbeatLag > cutOverLockTimeout {
-				mgtr.migrationContext.Log.Debugf("current HeartbeatLag (%.2fs) is too high, it needs to be less than both --max-lag-millis (%.2fs) and --cut-over-lock-timeout-seconds (%.2fs) to continue", heartbeatLag.Seconds(), maxLagMillisecondsThrottle.Seconds(), cutOverLockTimeout.Seconds())
-				return true, nil
-			}
-			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) {
-				// Postpone file defined and exists!
-				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 {
+	if err := mgtr.sleepWhileTrue("cut_over_postpone", mgtr.shouldPostponeCutOver); err != nil {
 		return err
 	}
-	atomic.StoreInt64(&mgtr.migrationContext.IsPostponingCutOver, 0)
+	mgtr.migrationContext.EndPostponingCutOver()
 	mgtr.migrationContext.MarkPointOfInterest()
 	mgtr.migrationContext.Log.Debugf("checking for cut-over postpone: complete")
 
@@ -1434,6 +1450,13 @@ func (mgtr *Migrator) getMigrationStateAndETA(rowsEstimate int64) (state, eta st
 	} else if atomic.LoadInt64(&mgtr.migrationContext.IsPostponingCutOver) > 0 {
 		eta = "due"
 		state = "postponing cut-over"
+		if atomic.LoadInt64(&mgtr.migrationContext.UserCommandedUnpostponeFlag) > 0 {
+			if lag, limit, tooHigh := mgtr.cutOverHeartbeatLag(); tooHigh {
+				state = fmt.Sprintf("postponing cut-over, unpostpone requested: waiting for heartbeat lag (%.2fs) to drop below %.2fs", lag.Seconds(), limit.Seconds())
+			} else {
+				state = "postponing cut-over, unpostpone requested"
+			}
+		}
 	} else if isThrottled, throttleReason, _ := mgtr.migrationContext.IsThrottled(); isThrottled {
 		state = fmt.Sprintf("throttled, %s", throttleReason)
 	}