From 3053e41b1528ef898cccd44e056e4d167619af6b Mon Sep 17 00:00:00 2001
From: Alan Pearce
Date: Thu, 23 May 2024 11:45:38 +0200
Subject: fix: abort import of source on batch processing errors

---
 internal/importer/importer.go | 17 ++++++++++++-----
 internal/importer/main.go     |  7 ++++++-
 internal/index/indexer.go     | 19 ++++++++++++++++++-
 3 files changed, 36 insertions(+), 7 deletions(-)

(limited to 'internal')

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")
diff --git a/internal/index/indexer.go b/internal/index/indexer.go
index 3a146c3..1f93c06 100644
--- a/internal/index/indexer.go
+++ b/internal/index/indexer.go
@@ -32,6 +32,14 @@ type WriteIndex struct {
 	meta  *Meta
 }
 
+type BatchError struct {
+	error
+}
+
+func (e *BatchError) Error() string {
+	return e.error.Error()
+}
+
 const batchSize = 10_000
 
 func createIndexMapping() (mapping.IndexMapping, error) {
@@ -292,6 +300,8 @@ func (i *WriteIndex) Import(
 				err = i.Flush(batch)
 				if err != nil {
 					errs <- err
+
+					return
 				}
 			}
 		}
@@ -307,11 +317,18 @@ func (i *WriteIndex) Import(
 
 func (i *WriteIndex) Flush(batch *bleve.Batch) error {
 	size := batch.Size()
+	if size == 0 {
+		return &BatchError{
+			error: errors.New("no documents to flush"),
+		}
+	}
 	slog.Debug("flushing batch", "size", size)
 
 	err := i.index.Batch(batch)
 	if err != nil {
-		return errors.WithMessagef(err, "could not flush batch")
+		return &BatchError{
+			error: errors.WithMessagef(err, "could not flush batch"),
+		}
 	}
 
 	batch.Reset()
-- 
cgit 1.4.1