Add SendWithContext helper to avoid deadlocks

Author: ggilder

.github/CONTRIBUTING.md

@@ -19,6 +19,24 @@ Here are a few things you can do that will increase the likelihood of your pull
 - Keep your change as focused as possible. If there are multiple changes you would like to make that are not dependent upon each other, consider submitting them as separate pull requests.
 - Write a [good commit message](http://tbaggery.com/2008/04/19/a-note-about-git-commit-messages.html).
 
+## Development Guidelines
+
+### Channel Safety
+
+When working with channels in goroutines, it's critical to prevent deadlocks that can occur when a channel receiver exits due to an error while senders are still trying to send values. Always use `base.SendWithContext` for channel sends to avoid deadlocks:
+
+```go
+// ✅ CORRECT - Uses helper to prevent deadlock
+if err := base.SendWithContext(ctx, ch, value); err != nil {
+    return err  // context was cancelled
+}
+
+// ❌ WRONG - Can deadlock if receiver exits
+ch <- value
+```
+
+Even if the destination channel is buffered, deadlocks could still occur if the buffer fills up and the receiver exits, so it's important to use `SendWithContext` in those cases as well.
+
 ## Resources
 
 - [Contributing to Open Source on GitHub](https://guides.github.com/activities/contributing-to-open-source/)

go/base/context.go

@@ -1027,3 +1027,24 @@ func (this *MigrationContext) CancelContext() {
 		this.cancelFunc()
 	}
 }
+
+// SendWithContext attempts to send a value to a channel, but returns early
+// if the context is cancelled. This prevents goroutine deadlocks when the
+// channel receiver has exited due to an error.
+//
+// Use this instead of bare channel sends (ch <- val) in goroutines to ensure
+// proper cleanup when the migration is aborted.
+//
+// Example:
+//
+//	if err := base.SendWithContext(ctx, ch, value); err != nil {
+//	    return err  // context was cancelled
+//	}
+func SendWithContext[T any](ctx context.Context, ch chan<- T, val T) error {
+	select {
+	case ch <- val:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}

go/logic/applier.go

@@ -716,12 +716,8 @@ func (this *Applier) InitiateHeartbeat() {
 			continue
 		}
 		if err := injectHeartbeat(); err != nil {
-			select {
-			case this.migrationContext.PanicAbort <- fmt.Errorf("injectHeartbeat writing failed %d times, last error: %w", numSuccessiveFailures, err):
-				// Error sent successfully
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, someone else already reported an error
-			}
+			// Use helper to prevent deadlock if listenOnPanicAbort already exited
+			_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, fmt.Errorf("injectHeartbeat writing failed %d times, last error: %w", numSuccessiveFailures, err))
 			return
 		}
 	}

go/logic/migrator.go

@@ -163,12 +163,8 @@ func (this *Migrator) retryOperation(operation func() error, notFatalHint ...boo
 		// there's an error. Let's try again.
 	}
 	if len(notFatalHint) == 0 {
-		select {
-		case this.migrationContext.PanicAbort <- err:
-			// Error sent successfully
-		case <-this.migrationContext.GetContext().Done():
-			// Context cancelled, someone else already reported an error
-		}
+		// Use helper to prevent deadlock if listenOnPanicAbort already exited
+		_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 	}
 	return err
 }
@@ -196,12 +192,8 @@ func (this *Migrator) retryOperationWithExponentialBackoff(operation func() erro
 		}
 	}
 	if len(notFatalHint) == 0 {
-		select {
-		case this.migrationContext.PanicAbort <- err:
-			// Error sent successfully
-		case <-this.migrationContext.GetContext().Done():
-			// Context cancelled, someone else already reported an error
-		}
+		// Use helper to prevent deadlock if listenOnPanicAbort already exited
+		_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 	}
 	return err
 }
@@ -210,24 +202,16 @@ func (this *Migrator) retryOperationWithExponentialBackoff(operation func() erro
 // consumes and drops any further incoming events that may be left hanging.
 func (this *Migrator) consumeRowCopyComplete() {
 	if err := <-this.rowCopyComplete; err != nil {
-		select {
-		case this.migrationContext.PanicAbort <- err:
-			// Error sent successfully
-		case <-this.migrationContext.GetContext().Done():
-			// Context cancelled, someone else already reported an error
-		}
+		// Use helper to prevent deadlock if listenOnPanicAbort already exited
+		_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 	}
 	atomic.StoreInt64(&this.rowCopyCompleteFlag, 1)
 	this.migrationContext.MarkRowCopyEndTime()
 	go func() {
 		for err := range this.rowCopyComplete {
 			if err != nil {
-				select {
-				case this.migrationContext.PanicAbort <- err:
-					// Error sent successfully
-				case <-this.migrationContext.GetContext().Done():
-					// Context cancelled, someone else already reported an error
-				}
+				// Use helper to prevent deadlock if listenOnPanicAbort already exited
+				_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 			}
 		}
 	}()
@@ -258,32 +242,23 @@ func (this *Migrator) onChangelogStateEvent(dmlEntry *binlog.BinlogEntry) (err e
 	case Migrated, ReadMigrationRangeValues:
 		// no-op event
 	case GhostTableMigrated:
-		select {
-		case this.ghostTableMigrated <- true:
-			// Successfully sent
-		case <-this.migrationContext.GetContext().Done():
-			// Context cancelled, migration is aborting
-		}
+		// Use helper to prevent deadlock if migration aborts before receiver is ready
+		_ = base.SendWithContext(this.migrationContext.GetContext(), this.ghostTableMigrated, true)
 	case AllEventsUpToLockProcessed:
 		var applyEventFunc tableWriteFunc = func() error {
-			this.allEventsUpToLockProcessed <- &lockProcessedStruct{
+			return base.SendWithContext(this.migrationContext.GetContext(), this.allEventsUpToLockProcessed, &lockProcessedStruct{
 				state:  changelogStateString,
 				coords: dmlEntry.Coordinates.Clone(),
-			}
-			return nil
+			})
 		}
 		// at this point we know all events up to lock have been read from the streamer,
 		// because the streamer works sequentially. So those events are either already handled,
 		// or have event functions in applyEventsQueue.
 		// So as not to create a potential deadlock, we write this func to applyEventsQueue
 		// asynchronously, understanding it doesn't really matter.
 		go func() {
-			select {
-			case this.applyEventsQueue <- newApplyEventStructByFunc(&applyEventFunc):
-				// Successfully enqueued
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, migration is aborting
-			}
+			// Use helper to prevent deadlock if buffer fills and executeWriteFuncs exits
+			_ = base.SendWithContext(this.migrationContext.GetContext(), this.applyEventsQueue, newApplyEventStructByFunc(&applyEventFunc))
 		}()
 	default:
 		return fmt.Errorf("Unknown changelog state: %+v", changelogState)
@@ -1387,12 +1362,8 @@ func (this *Migrator) initiateStreaming() error {
 		this.migrationContext.Log.Debugf("Beginning streaming")
 		err := this.eventsStreamer.StreamEvents(this.canStopStreaming)
 		if err != nil {
-			select {
-			case this.migrationContext.PanicAbort <- err:
-				// Error sent successfully
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, someone else already reported an error
-			}
+			// Use helper to prevent deadlock if listenOnPanicAbort already exited
+			_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 		}
 		this.migrationContext.Log.Debugf("Done streaming")
 	}()
@@ -1418,14 +1389,9 @@ func (this *Migrator) addDMLEventsListener() error {
 		this.migrationContext.DatabaseName,
 		this.migrationContext.OriginalTableName,
 		func(dmlEntry *binlog.BinlogEntry) error {
-			select {
-			case this.applyEventsQueue <- newApplyEventStructByDML(dmlEntry):
-				// Successfully enqueued
-				return nil
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, stop processing events
-				return this.migrationContext.GetContext().Err()
-			}
+			// Use helper to prevent deadlock if buffer fills and executeWriteFuncs exits
+			// This is critical because this callback blocks the event streamer
+			return base.SendWithContext(this.migrationContext.GetContext(), this.applyEventsQueue, newApplyEventStructByDML(dmlEntry))
 		},
 	)
 	return err
@@ -1503,7 +1469,7 @@ func (this *Migrator) initiateApplier() error {
 // a chunk of rows onto the ghost table.
 func (this *Migrator) iterateChunks() error {
 	terminateRowIteration := func(err error) error {
-		this.rowCopyComplete <- err
+		_ = base.SendWithContext(this.migrationContext.GetContext(), this.rowCopyComplete, err)
 		return this.migrationContext.Log.Errore(err)
 	}
 	if this.migrationContext.Noop {
@@ -1590,15 +1556,13 @@ func (this *Migrator) iterateChunks() error {
 			return nil
 		}
 		// Enqueue copy operation; to be executed by executeWriteFuncs()
-		select {
-		case this.copyRowsQueue <- copyRowsFunc:
-			// Successfully enqueued
-		case <-this.migrationContext.GetContext().Done():
+		// Use helper to prevent deadlock if executeWriteFuncs exits
+		if err := base.SendWithContext(this.migrationContext.GetContext(), this.copyRowsQueue, copyRowsFunc); err != nil {
 			// Context cancelled, check for abort and exit
-			if err := this.checkAbort(); err != nil {
-				return terminateRowIteration(err)
+			if abortErr := this.checkAbort(); abortErr != nil {
+				return terminateRowIteration(abortErr)
 			}
-			return terminateRowIteration(this.migrationContext.GetContext().Err())
+			return terminateRowIteration(err)
 		}
 	}
 }

go/logic/server.go

@@ -450,12 +450,8 @@ help                                 # This message
 				return NoPrintStatusRule, err
 			}
 			err := fmt.Errorf("User commanded 'panic'. The migration will be aborted without cleanup. Please drop the gh-ost tables before trying again.")
-			select {
-			case this.migrationContext.PanicAbort <- err:
-				// Error sent successfully
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, someone else already reported an error
-			}
+			// Use helper to prevent deadlock if listenOnPanicAbort already exited
+			_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, err)
 			return NoPrintStatusRule, err
 		}
 	default:

go/logic/throttler.go

@@ -362,12 +362,8 @@ func (this *Throttler) collectGeneralThrottleMetrics() error {
 	// Regardless of throttle, we take opportunity to check for panic-abort
 	if this.migrationContext.PanicFlagFile != "" {
 		if base.FileExists(this.migrationContext.PanicFlagFile) {
-			select {
-			case this.migrationContext.PanicAbort <- fmt.Errorf("Found panic-file %s. Aborting without cleanup", this.migrationContext.PanicFlagFile):
-				// Error sent successfully
-			case <-this.migrationContext.GetContext().Done():
-				// Context cancelled, someone else already reported an error
-			}
+			// 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))
 		}
 	}
 
@@ -390,25 +386,17 @@ func (this *Throttler) collectGeneralThrottleMetrics() error {
 	}
 
 	if criticalLoadMet && this.migrationContext.CriticalLoadIntervalMilliseconds == 0 {
-		select {
-		case this.migrationContext.PanicAbort <- fmt.Errorf("critical-load met: %s=%d, >=%d", variableName, value, threshold):
-			// Error sent successfully
-		case <-this.migrationContext.GetContext().Done():
-			// Context cancelled, someone else already reported an error
-		}
+		// 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))
 	}
 	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)
 		go func() {
 			timer := time.NewTimer(time.Millisecond * time.Duration(this.migrationContext.CriticalLoadIntervalMilliseconds))
 			<-timer.C
 			if criticalLoadMetAgain, variableName, value, threshold, _ := this.criticalLoadIsMet(); criticalLoadMetAgain {
-				select {
-				case this.migrationContext.PanicAbort <- fmt.Errorf("critical-load met again after %d millis: %s=%d, >=%d", this.migrationContext.CriticalLoadIntervalMilliseconds, variableName, value, threshold):
-					// Error sent successfully
-				case <-this.migrationContext.GetContext().Done():
-					// Context cancelled, someone else already reported an error
-				}
+				// Use helper to prevent deadlock if listenOnPanicAbort already exited
+				_ = base.SendWithContext(this.migrationContext.GetContext(), this.migrationContext.PanicAbort, fmt.Errorf("critical-load met again after %d millis: %s=%d, >=%d", this.migrationContext.CriticalLoadIntervalMilliseconds, variableName, value, threshold))
 			}
 		}()
 	}