From f470fe4a40584197d79ffb531ae460e06e02ad3b Mon Sep 17 00:00:00 2001 From: Mohith26 <109003762+Mohith26@users.noreply.github.com> Date: Wed, 4 Nov 2026 08:03:37 -0500 Subject: [PATCH 1/2] Make pending unpostpone observable and cancellable After `unpostpone`, cut-over still waits for the changelog heartbeat lag to drop below --max-lag-millis and --cut-over-lock-timeout-seconds. Until now nothing in `status` showed that a request was pending, and there was no way to withdraw it, so tooling driving cut-over could not tell "waiting on lag" apart from "attempted, failed and back to postponing", and could not give up. - `status`/`sup` now report `postponing cut-over, unpostpone requested`, plus the current heartbeat lag and limit when that is what cut-over is waiting on. The state is unchanged when no request is pending. - New `cancel-unpostpone` interactive command withdraws a pending request. The cut-over loop and the command both use compare-and-swap on UserCommandedUnpostponeFlag, so a request is either consumed by cut-over or cancelled, never both. - The postpone check is extracted into shouldPostponeCutOver and cutOverHeartbeatLag so it can be unit tested; behavior is otherwise the same. Closes #1685 --- doc/interactive-commands.md | 2 + go/logic/migrator.go | 81 ++++++++++------- go/logic/migrator_test.go | 174 ++++++++++++++++++++++++++++++++++++ go/logic/server.go | 17 ++++ go/logic/server_test.go | 72 +++++++++++++++ 5 files changed, 316 insertions(+), 30 deletions(-) diff --git a/doc/interactive-commands.md b/doc/interactive-commands.md index af37935d1..5490b20d8 100644 --- a/doc/interactive-commands.md +++ b/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=`: 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 diff --git a/go/logic/migrator.go b/go/logic/migrator.go index f2f6b3f20..238d34dc3 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -870,6 +870,49 @@ 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. Both sides use compare-and-swap, so a request is +// either consumed here or cancelled by the user, never both. +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 atomic.CompareAndSwapInt64(&mgtr.migrationContext.UserCommandedUnpostponeFlag, 1, 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 +} + // cutOver performs the final step of migration, based on migration // type (on replica? atomic? safe?) func (mgtr *Migrator) cutOver() (err error) { @@ -884,36 +927,7 @@ 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) @@ -1434,6 +1448,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) } diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index ad068691c..06f76782c 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -6,6 +6,7 @@ package logic import ( + "bufio" "bytes" "context" gosql "database/sql" @@ -1814,3 +1815,176 @@ func (suite *MigratorTestSuite) TestPanicOnWarningsDuplicateDuringCutoverWithHig suite.Require().NoError(err) suite.Require().Equal(2, duplicateCount, "Should have 2 duplicate email entries") } + +// newPostponeTestMigrator returns a migrator whose cut-over is postponed by an +// existing flag file, with a fresh heartbeat, and a server sharing its context. +func newPostponeTestMigrator(t *testing.T) (*Migrator, *Server) { + t.Helper() + migrationContext := base.NewMigrationContext() + migrationContext.OriginalTableName = "users" + migrationContext.PostponeCutOverFlagFile = filepath.Join(t.TempDir(), "postpone.flag") + require.NoError(t, base.TouchFile(migrationContext.PostponeCutOverFlagFile)) + atomic.StoreInt64(&migrationContext.MaxLagMillisecondsThrottleThreshold, 1500) + migrationContext.CutOverLockTimeoutSeconds = 3 + migrationContext.SetLastHeartbeatOnChangelogTime(time.Now()) + migrator := NewMigrator(migrationContext, "1.2.3") + migrator.hooksExecutor = CompositeHooks{} + return migrator, NewServer(migrationContext, CompositeHooks{}, nil) +} + +func applyTestServerCommand(t *testing.T, srv *Server, command string) string { + t.Helper() + var buf bytes.Buffer + writer := bufio.NewWriter(&buf) + _, err := srv.applyServerCommand(command, writer) + require.NoError(t, err) + require.NoError(t, writer.Flush()) + return buf.String() +} + +func TestMigratorCutOverHeartbeatLag(t *testing.T) { + migrator, _ := newPostponeTestMigrator(t) + _, limit, tooHigh := migrator.cutOverHeartbeatLag() + require.False(t, tooHigh) + require.Equal(t, 1500*time.Millisecond, limit) + + // The limit is the lower of --max-lag-millis and --cut-over-lock-timeout-seconds. + atomic.StoreInt64(&migrator.migrationContext.MaxLagMillisecondsThrottleThreshold, 10000) + _, limit, _ = migrator.cutOverHeartbeatLag() + require.Equal(t, 3*time.Second, limit) + + migrator.migrationContext.SetLastHeartbeatOnChangelogTime(time.Now().Add(-5 * time.Second)) + lag, _, tooHigh := migrator.cutOverHeartbeatLag() + require.True(t, tooHigh) + require.GreaterOrEqual(t, lag, 5*time.Second) +} + +func TestMigratorShouldPostponeCutOver(t *testing.T) { + t.Run("postpones while the flag file exists", func(t *testing.T) { + migrator, _ := newPostponeTestMigrator(t) + postpone, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, postpone) + require.Equal(t, int64(1), atomic.LoadInt64(&migrator.migrationContext.IsPostponingCutOver)) + }) + + t.Run("pending unpostpone is kept while heartbeat lag is too high", func(t *testing.T) { + migrator, srv := newPostponeTestMigrator(t) + postpone, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, postpone) + + require.Equal(t, "Unpostponed\n", applyTestServerCommand(t, srv, "unpostpone")) + migrator.migrationContext.SetLastHeartbeatOnChangelogTime(time.Now().Add(-time.Hour)) + + for range 3 { + postpone, err = migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, postpone) + require.Equal(t, int64(1), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) + } + state, _, _ := migrator.getMigrationStateAndETA(0) + require.Regexp(t, `^postponing cut-over, unpostpone requested: waiting for heartbeat lag \(\d+\.\d{2}s\) to drop below 1\.50s$`, state) + + // Once the lag recovers, the request is consumed and cut-over may begin. + migrator.migrationContext.SetLastHeartbeatOnChangelogTime(time.Now()) + postpone, err = migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.False(t, postpone) + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) + }) + + t.Run("cancelled unpostpone keeps postponing", func(t *testing.T) { + migrator, srv := newPostponeTestMigrator(t) + _, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + applyTestServerCommand(t, srv, "unpostpone") + state, _, _ := migrator.getMigrationStateAndETA(0) + require.Equal(t, "postponing cut-over, unpostpone requested", state) + + require.Contains(t, applyTestServerCommand(t, srv, "cancel-unpostpone"), "Unpostpone cancelled") + state, _, _ = migrator.getMigrationStateAndETA(0) + require.Equal(t, "postponing cut-over", state) + + postpone, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, postpone) + }) +} + +// TestMigratorUnpostponeCancelRace checks that a pending unpostpone is either +// consumed by the cut-over loop or withdrawn by cancel-unpostpone, never both. +func TestMigratorUnpostponeCancelRace(t *testing.T) { + migrator, srv := newPostponeTestMigrator(t) + _, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + + for range 500 { + atomic.StoreInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag, 1) + var consumed, cancelled bool + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + postpone, err := migrator.shouldPostponeCutOver() + assert.NoError(t, err) + consumed = !postpone + }() + go func() { + defer wg.Done() + var buf bytes.Buffer + writer := bufio.NewWriter(&buf) + _, err := srv.applyServerCommand("cancel-unpostpone", writer) + assert.NoError(t, err) + assert.NoError(t, writer.Flush()) + cancelled = strings.Contains(buf.String(), "Unpostpone cancelled") + }() + wg.Wait() + require.True(t, consumed != cancelled, "consumed=%v cancelled=%v", consumed, cancelled) + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) + } +} + +// TestMigratorCutOverPostponeLoop drives the real sleepWhileTrue loop through +// unpostpone, cancel-unpostpone and unpostpone again under high heartbeat lag. +func TestMigratorCutOverPostponeLoop(t *testing.T) { + migrator, srv := newPostponeTestMigrator(t) + // Generous limits so a fresh heartbeat stays under them for the whole test. + atomic.StoreInt64(&migrator.migrationContext.MaxLagMillisecondsThrottleThreshold, 60000) + migrator.migrationContext.CutOverLockTimeoutSeconds = 60 + done := make(chan error, 1) + go func() { + done <- migrator.sleepWhileTrue("cut_over_postpone", migrator.shouldPostponeCutOver) + }() + require.Eventually(t, func() bool { + return atomic.LoadInt64(&migrator.migrationContext.IsPostponingCutOver) == 1 + }, 5*time.Second, 10*time.Millisecond) + + migrator.migrationContext.SetLastHeartbeatOnChangelogTime(time.Now().Add(-time.Hour)) + applyTestServerCommand(t, srv, "unpostpone") + time.Sleep(1500 * time.Millisecond) // at least one more loop iteration + select { + case err := <-done: + t.Fatalf("cut-over started while heartbeat lag was too high: %v", err) + default: + } + require.Contains(t, applyTestServerCommand(t, srv, "cancel-unpostpone"), "Unpostpone cancelled") + + // Lag recovers, but the request was withdrawn: still postponing. + migrator.migrationContext.SetLastHeartbeatOnChangelogTime(time.Now()) + time.Sleep(1500 * time.Millisecond) + select { + case err := <-done: + t.Fatalf("cut-over started after unpostpone was cancelled: %v", err) + default: + } + + applyTestServerCommand(t, srv, "unpostpone") + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("cut-over did not proceed after unpostpone") + } + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) +} diff --git a/go/logic/server.go b/go/logic/server.go index 4705ba9b9..d43fde0c4 100644 --- a/go/logic/server.go +++ b/go/logic/server.go @@ -234,6 +234,7 @@ throttle # Force throttling no-throttle # End forced throttling (other throttling may still apply) postpone-cut-over-flag-file= # Postpone the cut-over phase, writing a cut over flag file to the given path unpostpone # Bail out a cut-over postpone; proceed to cut-over +cancel-unpostpone # Withdraw a pending 'unpostpone' that has not started cut-over yet; keep postponing panic # panic and quit without cleanup help # This message - use '?' (question mark) as argument to get info rather than set. e.g. "max-load=?" will just print out current max-load. @@ -438,6 +439,22 @@ help # This message fmt.Fprintf(writer, "You may only invoke this when gh-ost is actively postponing migration. At this time it is not\n") return NoPrintStatusRule, nil } + case "cancel-unpostpone": + { + if arg != "" && arg != srv.migrationContext.OriginalTableName { + // User explicitly provided table name. This is a courtesy protection mechanism + err := fmt.Errorf("user commanded 'cancel-unpostpone' on %s, but migrated table is %s; ignoring request", arg, srv.migrationContext.OriginalTableName) + return NoPrintStatusRule, err + } + // The migrator consumes the request with the same compare-and-swap once it + // starts cut-over, so exactly one side wins. + if atomic.CompareAndSwapInt64(&srv.migrationContext.UserCommandedUnpostponeFlag, 1, 0) { + fmt.Fprintf(writer, "Unpostpone cancelled; cut-over stays postponed while the postpone flag file exists\n") + return ForcePrintStatusAndHintRule, nil + } + fmt.Fprintf(writer, "There is no pending unpostpone to cancel. Either none was requested, or cut-over has already started\n") + return NoPrintStatusRule, nil + } case "panic": { if arg == "" && srv.migrationContext.ForceNamedPanicCommand { diff --git a/go/logic/server_test.go b/go/logic/server_test.go index 52624a27d..d7f89b892 100644 --- a/go/logic/server_test.go +++ b/go/logic/server_test.go @@ -1,8 +1,11 @@ package logic import ( + "bufio" + "bytes" "os" "path" + "sync/atomic" "testing" "time" @@ -103,3 +106,72 @@ func TestServerCreatePostponeCutOverFlagFile(t *testing.T) { require.FileExists(t, filePath) }) } + +func TestServerUnpostponeCommands(t *testing.T) { + t.Parallel() + + newServer := func() *Server { + migrationContext := base.NewMigrationContext() + migrationContext.OriginalTableName = "users" + return NewServer(migrationContext, CompositeHooks{}, nil) + } + run := func(t *testing.T, s *Server, command string) (string, PrintStatusRule, error) { + var buf bytes.Buffer + writer := bufio.NewWriter(&buf) + rule, err := s.applyServerCommand(command, writer) + require.NoError(t, writer.Flush()) + return buf.String(), rule, err + } + + t.Run("unpostpone is rejected while not postponing", func(t *testing.T) { + s := newServer() + out, rule, err := run(t, s, "unpostpone") + require.NoError(t, err) + require.Equal(t, NoPrintStatusRule, rule) + require.Contains(t, out, "only invoke this when gh-ost is actively postponing") + require.Equal(t, int64(0), atomic.LoadInt64(&s.migrationContext.UserCommandedUnpostponeFlag)) + }) + + t.Run("cancel-unpostpone withdraws a pending unpostpone", func(t *testing.T) { + s := newServer() + atomic.StoreInt64(&s.migrationContext.IsPostponingCutOver, 1) + + out, _, err := run(t, s, "unpostpone") + require.NoError(t, err) + require.Equal(t, "Unpostponed\n", out) + require.Equal(t, int64(1), atomic.LoadInt64(&s.migrationContext.UserCommandedUnpostponeFlag)) + + out, rule, err := run(t, s, "cancel-unpostpone") + require.NoError(t, err) + require.Equal(t, PrintStatusRule(ForcePrintStatusAndHintRule), rule) + require.Contains(t, out, "Unpostpone cancelled") + require.Equal(t, int64(0), atomic.LoadInt64(&s.migrationContext.UserCommandedUnpostponeFlag)) + require.Equal(t, int64(1), atomic.LoadInt64(&s.migrationContext.IsPostponingCutOver)) + + // A second cancel has nothing to withdraw. + out, rule, err = run(t, s, "cancel-unpostpone") + require.NoError(t, err) + require.Equal(t, NoPrintStatusRule, rule) + require.Contains(t, out, "no pending unpostpone") + }) + + t.Run("cancel-unpostpone with table name", func(t *testing.T) { + s := newServer() + atomic.StoreInt64(&s.migrationContext.UserCommandedUnpostponeFlag, 1) + + _, _, err := run(t, s, "cancel-unpostpone=orders") + require.Error(t, err) + require.Equal(t, int64(1), atomic.LoadInt64(&s.migrationContext.UserCommandedUnpostponeFlag)) + + out, _, err := run(t, s, "cancel-unpostpone=users") + require.NoError(t, err) + require.Contains(t, out, "Unpostpone cancelled") + require.Equal(t, int64(0), atomic.LoadInt64(&s.migrationContext.UserCommandedUnpostponeFlag)) + }) + + t.Run("help lists cancel-unpostpone", func(t *testing.T) { + out, _, err := run(t, newServer(), "help") + require.NoError(t, err) + require.Contains(t, out, "cancel-unpostpone") + }) +} From b37800e4e9fe11c5a548f9f75fadb218d30b6a68 Mon Sep 17 00:00:00 2001 From: Mohith26 <109003762+Mohith26@users.noreply.github.com> Date: Wed, 4 Nov 2026 08:46:32 -0500 Subject: [PATCH 2/2] Serialize unpostpone requests with the end of postponement A repeated 'unpostpone' could pass the IsPostponingCutOver check, then re-set the pending flag after the cut-over loop had consumed the first request. 'cancel-unpostpone' would then report success while cut-over was already proceeding. Requests, cancellations and consumption now go through MigrationContext helpers guarded by one mutex. Consuming a request ends postponement in the same critical section, and leaving the postpone loop clears any late request, so a repeated 'unpostpone' is refused once cut-over has begun and a later cancel reports nothing pending. --- go/base/context.go | 46 ++++++++++++++++++++++++++++++++ go/logic/migrator.go | 10 ++++--- go/logic/migrator_test.go | 56 +++++++++++++++++++++++++++++++++++++++ go/logic/server.go | 8 +++--- 4 files changed, 111 insertions(+), 9 deletions(-) diff --git a/go/base/context.go b/go/base/context.go index b2ec8ff1e..de239a7f4 100644 --- a/go/base/context.go +++ b/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 { diff --git a/go/logic/migrator.go b/go/logic/migrator.go index 238d34dc3..6784b1329 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -887,8 +887,9 @@ func (mgtr *Migrator) cutOverHeartbeatLag() (lag, limit time.Duration, tooHigh b // // 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. Both sides use compare-and-swap, so a request is -// either consumed here or cancelled by the user, never both. +// 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()) @@ -897,7 +898,7 @@ func (mgtr *Migrator) shouldPostponeCutOver() (bool, error) { if mgtr.migrationContext.PostponeCutOverFlagFile == "" { return false, nil } - if atomic.CompareAndSwapInt64(&mgtr.migrationContext.UserCommandedUnpostponeFlag, 1, 0) { + if mgtr.migrationContext.ConsumeUnpostpone() { return false, nil } if base.FileExists(mgtr.migrationContext.PostponeCutOverFlagFile) { @@ -910,6 +911,7 @@ func (mgtr *Migrator) shouldPostponeCutOver() (bool, error) { atomic.StoreInt64(&mgtr.migrationContext.IsPostponingCutOver, 1) return true, nil } + mgtr.migrationContext.EndPostponingCutOver() return false, nil } @@ -930,7 +932,7 @@ func (mgtr *Migrator) cutOver() (err error) { 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") diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index 06f76782c..493ea61f1 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -1920,6 +1920,7 @@ func TestMigratorUnpostponeCancelRace(t *testing.T) { require.NoError(t, err) for range 500 { + atomic.StoreInt64(&migrator.migrationContext.IsPostponingCutOver, 1) atomic.StoreInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag, 1) var consumed, cancelled bool var wg sync.WaitGroup @@ -1945,6 +1946,61 @@ func TestMigratorUnpostponeCancelRace(t *testing.T) { } } +// TestMigratorRepeatedUnpostponeAfterConsume checks that once the cut-over loop has +// consumed a request, a repeated 'unpostpone' is refused and 'cancel-unpostpone' +// does not report success while cut-over is proceeding. +func TestMigratorRepeatedUnpostponeAfterConsume(t *testing.T) { + migrator, srv := newPostponeTestMigrator(t) + postpone, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, postpone) + + require.Equal(t, "Unpostponed\n", applyTestServerCommand(t, srv, "unpostpone")) + postpone, err = migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.False(t, postpone) + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.IsPostponingCutOver)) + + require.Contains(t, applyTestServerCommand(t, srv, "unpostpone"), "You may only invoke this when gh-ost is actively postponing") + require.Contains(t, applyTestServerCommand(t, srv, "cancel-unpostpone"), "There is no pending unpostpone to cancel") + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) +} + +// TestMigratorRepeatedUnpostponeRace interleaves repeated 'unpostpone' commands with +// the cut-over loop consuming the first one. Whatever the timing, no request may be +// left pending after cut-over has begun, so a later cancel never claims success. +func TestMigratorRepeatedUnpostponeRace(t *testing.T) { + for range 300 { + migrator, srv := newPostponeTestMigrator(t) + _, err := migrator.shouldPostponeCutOver() + require.NoError(t, err) + require.True(t, migrator.migrationContext.RequestUnpostpone()) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + postpone, err := migrator.shouldPostponeCutOver() + assert.NoError(t, err) + assert.False(t, postpone) + }() + go func() { + defer wg.Done() + var buf bytes.Buffer + writer := bufio.NewWriter(&buf) + _, err := srv.applyServerCommand("unpostpone", writer) + assert.NoError(t, err) + assert.NoError(t, writer.Flush()) + }() + wg.Wait() + + // The cut-over loop has exited; cut-over is proceeding. + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.IsPostponingCutOver)) + require.Equal(t, int64(0), atomic.LoadInt64(&migrator.migrationContext.UserCommandedUnpostponeFlag)) + require.Contains(t, applyTestServerCommand(t, srv, "cancel-unpostpone"), "There is no pending unpostpone to cancel") + } +} + // TestMigratorCutOverPostponeLoop drives the real sleepWhileTrue loop through // unpostpone, cancel-unpostpone and unpostpone again under high heartbeat lag. func TestMigratorCutOverPostponeLoop(t *testing.T) { diff --git a/go/logic/server.go b/go/logic/server.go index d43fde0c4..60a288ee1 100644 --- a/go/logic/server.go +++ b/go/logic/server.go @@ -431,8 +431,7 @@ help # This message err := fmt.Errorf("user commanded 'unpostpone' on %s, but migrated table is %s; ignoring request", arg, srv.migrationContext.OriginalTableName) return NoPrintStatusRule, err } - if atomic.LoadInt64(&srv.migrationContext.IsPostponingCutOver) > 0 { - atomic.StoreInt64(&srv.migrationContext.UserCommandedUnpostponeFlag, 1) + if srv.migrationContext.RequestUnpostpone() { fmt.Fprintf(writer, "Unpostponed\n") return ForcePrintStatusAndHintRule, nil } @@ -446,9 +445,8 @@ help # This message err := fmt.Errorf("user commanded 'cancel-unpostpone' on %s, but migrated table is %s; ignoring request", arg, srv.migrationContext.OriginalTableName) return NoPrintStatusRule, err } - // The migrator consumes the request with the same compare-and-swap once it - // starts cut-over, so exactly one side wins. - if atomic.CompareAndSwapInt64(&srv.migrationContext.UserCommandedUnpostponeFlag, 1, 0) { + // Serialized with the migrator consuming the request, so exactly one side wins. + if srv.migrationContext.CancelUnpostpone() { fmt.Fprintf(writer, "Unpostpone cancelled; cut-over stays postponed while the postpone flag file exists\n") return ForcePrintStatusAndHintRule, nil }