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/index/indexer.go | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) (limited to 'internal/index/indexer.go') 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