@@ -4,7 +4,9 @@ import (
 	"context"
 	"sync"
 
-	"alin.ovh/searchix/internal/index"
+	"github.com/Southclaws/fault"
+	"github.com/Southclaws/fault/fmsg"
+
 	"alin.ovh/searchix/internal/nix"
 )
 
@@ -19,47 +21,27 @@ ) (hadObjectErrors bool, criticalError error) {
 	wg := sync.WaitGroup{}
 
 	objects := make(chan nix.Importable, 1)
-	pErrs := make(chan error)
+	errs := make(chan error)
 	wg.Go(func() {
-		processor.Process(ctx, objects, pErrs)
+		processor.Process(ctx, objects, errs)
+		close(objects)
 	})
 
-	iErrs := make(chan error)
 	wg.Go(func() {
-		imp.options.WriteIndex.Import(ctx, objects, iErrs)
+		err := imp.options.WriteIndex.Import(ctx, objects, errs)
+		if err != nil {
+			criticalError = fault.Wrap(err, fmsg.With("error writing batch"))
+		}
+		close(errs)
 	})
 
-	go func() {
-		for {
-			select {
-			case err, running := <-iErrs:
-				if !running {
-					iErrs = nil
-					imp.options.Logger.Debug("ingest completed")
-
-					continue
-				}
-				be, isBatchError := err.(*index.BatchError)
-				if isBatchError {
-					criticalError = be
-
-					break
-				}
-				hadObjectErrors = true
-				imp.options.Logger.Warn("error ingesting object", "error", err)
-			case err, running := <-pErrs:
-				if !running {
-					pErrs = nil
-
-					continue
-				}
-				hadObjectErrors = true
-				imp.options.Logger.Warn("error processing object", "error", err)
-			}
-		}
-	}()
+	for err := range errs {
+		hadObjectErrors = true
+		imp.options.Logger.Warn("error processing object", "error", err)
+	}
 
 	wg.Wait()
+	imp.options.Logger.Debug("ingest completed")
 
 	return hadObjectErrors, criticalError
 }

@@ -104,8 +104,6 @@ results chan<- nix.Importable,
 	errs chan<- error,
 ) {
 	defer i.infile.Close()
-	defer close(results)
-	defer close(errs)
 
 outer:
 	for mv := range i.dec.Stream() {

@@ -134,8 +134,6 @@ }
 	}
 
 	defer i.infile.Close()
-	defer close(results)
-	defer close(errs)
 
 	if i.programs != nil {
 		defer i.programs.Close()

@@ -319,10 +319,10 @@ func (i *WriteIndex) Import(
 	ctx context.Context,
 	objects <-chan nix.Importable,
 	errs chan<- error,
-) {
+) error {
 	indexMapping := i.index.Mapping()
 
-	i.WithBatchObjects(ctx, objects, errs, func(batch *bleve.Batch, obj nix.Importable) error {
+	return i.WithBatchObjects(ctx, objects, errs, func(batch *bleve.Batch, obj nix.Importable) error {
 		doc := document.NewDocument(nix.GetKey(obj))
 		if err := indexMapping.MapDocument(doc, obj); err != nil {
 			return fault.Wrap(err, fmsg.Withf("could not map document for object: %s", obj.GetName()))
@@ -372,20 +372,16 @@ ctx context.Context,
 	objects <-chan nix.Importable,
 	errs chan<- error,
 	processor func(batch *bleve.Batch, obj nix.Importable) error,
-) {
-	var err error
-
-	defer close(errs)
+) error {
 	k := 0
 	batch := i.index.NewBatch()
 
-outer:
 	for obj := range objects {
 		select {
 		case <-ctx.Done():
 			i.log.Warn("batch process aborted")
 
-			break outer
+			return ctx.Err()
 		default:
 		}
 
@@ -396,19 +392,19 @@ continue
 		}
 
 		if k++; k%i.batchSize == 0 {
-			err = i.Flush(batch)
+			err := i.Flush(batch)
 			if err != nil {
-				errs <- err
-
-				return
+				return err
 			}
 		}
 	}
 
-	err = i.Flush(batch)
+	err := i.Flush(batch)
 	if err != nil {
-		errs <- err
+		return err
 	}
+
+	return nil
 }
 
 func (i *WriteIndex) Flush(batch *bleve.Batch) error {