Fix deadlock on PanicAbort sends

Author: ggilder

go/base/context.go

@@ -324,7 +324,7 @@ func NewMigrationContext() *MigrationContext {
 		pointOfInterestTimeMutex:            &sync.Mutex{},
 		lastHeartbeatOnChangelogMutex:       &sync.Mutex{},
 		ColumnRenameMap:                     make(map[string]string),
-		PanicAbort:                          make(chan error),
+		PanicAbort:                          make(chan error, 1),
 		ctx:                                 ctx,
 		cancelFunc:                          cancelFunc,
 		abortMutex:                          &sync.Mutex{},

go/logic/migrator.go

@@ -204,6 +204,8 @@ func (this *Migrator) consumeRowCopyComplete() {
 	if err := <-this.rowCopyComplete; err != nil {
 		// Use helper to prevent deadlock if listenOnPanicAbort already exited
 		_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
+		// Don't mark row copy as complete if there was an error
+		return
 	}
 	atomic.StoreInt64(&this.rowCopyCompleteFlag, 1)
 	this.migrationContext.MarkRowCopyEndTime()
@@ -212,6 +214,7 @@ func (this *Migrator) consumeRowCopyComplete() {
 			if err != nil {
 				// Use helper to prevent deadlock if listenOnPanicAbort already exited
 				_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
+				return
 			}
 		}
 	}()

go/logic/throttler.go

@@ -364,6 +364,7 @@ func (this *Throttler) collectGeneralThrottleMetrics() error {
 		if base.FileExists(this.migrationContext.PanicFlagFile) {
 			// Use helper to prevent deadlock if listenOnPanicAbort already exited
 			_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, fmt.Errorf("Found panic-file %s. Aborting without cleanup", this.migrationContext.PanicFlagFile))
+			return nil
 		}
 	}
 
@@ -388,6 +389,7 @@ func (this *Throttler) collectGeneralThrottleMetrics() error {
 	if criticalLoadMet && this.migrationContext.CriticalLoadIntervalMilliseconds == 0 {
 		// Use helper to prevent deadlock if listenOnPanicAbort already exited
 		_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, fmt.Errorf("critical-load met: %s=%d, >=%d", variableName, value, threshold))
+		return nil
 	}
 	if criticalLoadMet && this.migrationContext.CriticalLoadIntervalMilliseconds > 0 {
 		this.migrationContext.Log.Errorf("critical-load met once: %s=%d, >=%d. Will check again in %d millis", variableName, value, threshold, this.migrationContext.CriticalLoadIntervalMilliseconds)