feat: add context support to error handlers for improved cancellation and error management

cmd/go_migrate/extractor-error-handler.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/google/uuid"
@@ -19,29 +20,57 @@ func (e *ExtractorError) Error() string {
 
 const maxRetryAttempts = 3
 
-func extractorErrorHandler(chErrorsIn <-chan ExtractorError, chBatchesOut chan<- Batch, chJobErrorsOut chan<- JobError) {
+func extractorErrorHandler(
-	for err := range chErrorsIn {
+	ctx context.Context,
-		if err.RetryCounter >= maxRetryAttempts {
+	chErrorsIn <-chan ExtractorError,
-			jobError := JobError{
+	chBatchesOut chan<- Batch,
-				ShouldCancelJob: false,
+	chJobErrorsOut chan<- JobError,
-				Msg:             fmt.Sprintf("batch %v reached max retries (%d)", err.Id, maxRetryAttempts),
+) {
-				Prev:            &err,
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+
+		case err, ok := <-chErrorsIn:
+			if !ok {
+				return
+			}
+
+			if err.RetryCounter >= maxRetryAttempts {
+				jobError := JobError{
+					ShouldCancelJob: false,
+					Msg:             fmt.Sprintf("batch %v reached max retries (%d)", err.Id, maxRetryAttempts),
+					Prev:            &err,
+				}
+
+				select {
+				case chJobErrorsOut <- jobError:
+				case <-ctx.Done():
+					return
+				}
+				continue
+			}
+
+			newBatch := err.Batch
+			newBatch.RetryCounter++
+
+			if err.HasLastId {
+				newBatch.ParentId = err.Id
+				newBatch.Id = uuid.New()
+				newBatch.LowerLimit = err.LastId
+				newBatch.IsLowerLimitInclusive = false
+			}
+
+			select {
+			case chBatchesOut <- newBatch:
+			case <-ctx.Done():
+				return
 			}
-			chJobErrorsOut <- jobError
-			continue
 		}
-
-		newBatch := err.Batch
-		newBatch.RetryCounter++
-
-		if err.HasLastId {
-			newBatch.ParentId = err.Id
-			newBatch.Id = uuid.New()
-			newBatch.LowerLimit = err.LastId
-			newBatch.IsLowerLimitInclusive = false
-		}
-
-		chBatchesOut <- newBatch
 	}
 }
 

cmd/go_migrate/job-error-handler.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 
 	log "github.com/sirupsen/logrus"
@@ -20,14 +21,26 @@ func (e *JobError) Error() string {
 	return e.Msg
 }
 
-func jobErrorHandler(chErrorsIn <-chan JobError) error {
+func jobErrorHandler(ctx context.Context, chErrorsIn <-chan JobError) error {
-	for err := range chErrorsIn {
+	for {
-		if err.ShouldCancelJob {
+		if ctx.Err() != nil {
-			return &err
+			return nil
 		}
 
-		log.Error(err)
+		select {
-	}
+		case <-ctx.Done():
+			return nil
 
-	return nil
+		case err, ok := <-chErrorsIn:
+			if !ok {
+				return nil
+			}
+
+			if err.ShouldCancelJob {
+				return &err
+			}
+
+			log.Error(err)
+		}
+	}
 }

cmd/go_migrate/process.go

@@ -36,8 +36,10 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	defer close(chJobErrors)
 
 	go func() {
-		if err := jobErrorHandler(chJobErrors); err != nil {
+		if err := jobErrorHandler(ctx, chJobErrors); err != nil {
-			cancel()
+			if ctx.Err() == nil {
+				cancel()
+			}
 		}
 	}()
 
@@ -45,7 +47,7 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	chExtractorErrors := make(chan ExtractorError, len(batches))
 
 	go func() {
-		extractorErrorHandler(chExtractorErrors, chBatches, chJobErrors)
+		extractorErrorHandler(ctx, chExtractorErrors, chBatches, chJobErrors)
 	}()
 
 	chChunks := make(chan Chunk, QueueSize)