fix: abort import of source on batch processing errors

2 files changed, 18 insertions, 6 deletions

diff --git a/internal/importer/importer.go b/internal/importer/importer.go
index 53a87c9..6803f00 100644
--- a/internal/importer/importer.go
+++ b/internal/importer/importer.go
@@ -21,7 +21,7 @@ func process(
 	indexer *index.WriteIndex,
 	processor Processor,
 	logger *slog.Logger,
-) bool {
+) (bool, error) {
 	wg := sync.WaitGroup{}
 
 	wg.Add(1)
@@ -30,7 +30,8 @@ func process(
 	wg.Add(1)
 	iErrs := indexer.Import(ctx, objects)
 
-	var hadErrors bool
+	var hadObjectErrors bool
+	var criticalError error
 	go func() {
 		for {
 			select {
@@ -42,7 +43,13 @@ func process(
 
 					continue
 				}
-				hadErrors = true
+				be, isBatchError := err.(*index.BatchError)
+				if isBatchError {
+					criticalError = be
+
+					break
+				}
+				hadObjectErrors = true
 				logger.Warn("error ingesting object", "error", err)
 			case err, running := <-pErrs:
 				if !running {
@@ -52,7 +59,7 @@ func process(
 
 					continue
 				}
-				hadErrors = true
+				hadObjectErrors = true
 				logger.Warn("error processing object", "error", err)
 			}
 		}
@@ -60,5 +67,5 @@ func process(
 
 	wg.Wait()
 
-	return hadErrors
+	return hadObjectErrors, criticalError
 }
diff --git a/internal/importer/main.go b/internal/importer/main.go
index 663e8eb..6f462c3 100644
--- a/internal/importer/main.go
+++ b/internal/importer/main.go
@@ -124,7 +124,12 @@ func Start(
 				continue
 			}
 
-			hadWarnings := process(ctx, indexer, processor, logger)
+			hadWarnings, err := process(ctx, indexer, processor, logger)
+			if err != nil {
+				logger.Error("failed to process source", "error", err)
+
+				continue
+			}
 
 			if hadWarnings {
 				logger.Warn("importer succeeded, but with warnings/errors")