feat: enhance error handling with JobError struct and update extractor logic

cmd/go_migrate/process.go

@@ -24,18 +24,31 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	logColumnTypes(sourceColTypes, "Source col types")
 	logColumnTypes(targetColTypes, "Target col types")
 
-	mssqlCtx := context.Background()
-	batches, err := batchGeneratorMssql(mssqlCtx, sourceDb, job)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	batches, err := batchGeneratorMssql(ctx, sourceDb, job)
 	if err != nil {
 		log.Error("Unexpected error calculating batch ranges: ", err)
 	}
 
-	chJobErrors := make(chan error)
+	chJobErrors := make(chan JobError)
 	defer close(chJobErrors)
 
+	go func() {
+		if err := jobErrorHandler(chJobErrors); err != nil {
+			cancel()
+		}
+	}()
+
 	chBatches := make(chan Batch, len(batches))
-	chChunks := make(chan []UnknownRowValues, QueueSize)
 	chExtractorErrors := make(chan ExtractorError, len(batches))
+
+	go func() {
+		extractorErrorHandler(chExtractorErrors, chBatches, chJobErrors)
+	}()
+
+	chChunks := make(chan []UnknownRowValues, QueueSize)
 	maxExtractors := min(NumExtractors, len(batches))
 	var wgMssqlExtractors sync.WaitGroup
 
@@ -43,7 +56,7 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	extractStartTime := time.Now()
 	for range maxExtractors {
 		wgMssqlExtractors.Go(func() {
-			extractFromMssql(mssqlCtx, sourceDb, job, sourceColTypes, ChunkSize, chBatches, chChunks, chExtractorErrors)
+			extractFromMssql(ctx, sourceDb, job, sourceColTypes, ChunkSize, chBatches, chChunks, chExtractorErrors, chJobErrors)
 		})
 	}
 
@@ -55,10 +68,6 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 		close(chExtractorErrors)
 	}()
 
-	go func() {
-		extractorErrorHandler(chExtractorErrors, chBatches, chJobErrors)
-	}()
-
 	go func() {
 		wgMssqlExtractors.Wait()
 		close(chChunks)
@@ -83,14 +92,13 @@ func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job Migration
 	}()
 
 	var wgPostgresLoaders sync.WaitGroup
-	postgresLoaderCtx := context.Background()
 
 	log.Infof("Starting %d PostgreSQL loader(s)...", NumLoaders)
 	loaderStartTime := time.Now()
 
 	for range NumLoaders {
 		wgPostgresLoaders.Go(func() {
-			if err := loadRowsPostgres(postgresLoaderCtx, job, targetColTypes, targetDb, chRowsTransform); err != nil {
+			if err := loadRowsPostgres(ctx, job, targetColTypes, targetDb, chRowsTransform); err != nil {
 				log.Error("Unexpected error loading data into postgres: ", err)
 			}
 		})