diff options
Diffstat (limited to 'internal/importer')
| -rw-r--r-- | internal/importer/importer.go | 17 | ||||
| -rw-r--r-- | internal/importer/main.go | 7 |
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") |