fix: abort import of source on batch processing errors
1 file changed, 12 insertions(+), 5 deletions(-)
changed files
M internal/importer/importer.go → internal/importer/importer.go
@@ -21,7 +21,7 @@ ctx context.Context, indexer *index.WriteIndex, processor Processor, logger *slog.Logger, -) bool { +) (bool, error) { wg := sync.WaitGroup{} wg.Add(1)@@ -30,7 +30,8 @@ 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 @@ logger.Debug("ingest completed") 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 @@ logger.Debug("processing completed") continue } - hadErrors = true + hadObjectErrors = true logger.Warn("error processing object", "error", err) } }@@ -60,5 +67,5 @@ }() wg.Wait() - return hadErrors + return hadObjectErrors, criticalError }