fix: preserve errored items through Filter to prevent silent error loss

processor/filter.go

@@ -12,6 +12,9 @@ type FilterFunc[T any] func(item *batch.Item[T]) bool
 
 // Filter is a processor that filters items based on a predicate function.
 // It can be used to remove items from the pipeline that don't meet certain criteria.
+//
+// Items that already carry an Error are always passed through unchanged so their
+// error is not lost; the predicate is only applied to error-free items.
 type Filter[T any] struct {
 	// Predicate is a function that returns true for items that should be kept
 	// and false for items that should be filtered out.
@@ -27,6 +30,11 @@ type Filter[T any] struct {
 // Process implements the Processor interface by filtering items according to the predicate.
 // Items that don't pass the filter are simply not included in the returned slice.
 // This does not set any errors on items, it just excludes them from further processing.
+//
+// Items that already carry an Error are always passed through unchanged so their
+// error is not lost; the predicate is only applied to error-free items. Dropping an
+// errored item here would silently discard its error, since the batch engine only
+// reports errors for items still present at the end of the processor chain.
 func (p *Filter[T]) Process(_ context.Context, items []*batch.Item[T]) ([]*batch.Item[T], error) {
 	if len(items) == 0 || p.Predicate == nil {
 		return items, nil
@@ -36,6 +44,14 @@ func (p *Filter[T]) Process(_ context.Context, items []*batch.Item[T]) ([]*batch
 	result := make([]*batch.Item[T], 0, len(items))
 
 	for _, item := range items {
+		// Items that already carry an error must pass through unchanged so their
+		// error survives to the engine's error-reporting step. The predicate is
+		// only applied to error-free items.
+		if item.Error != nil {
+			result = append(result, item)
+			continue
+		}
+
 		shouldKeep := p.Predicate(item)
 
 		// Invert logic if needed

processor/filter_test.go

@@ -2,6 +2,7 @@ package processor
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"github.com/MasterOfBinary/gobatch/batch"
@@ -173,3 +174,110 @@ func TestFilter_Process(t *testing.T) {
 		}
 	})
 }
+
+// containsID reports whether the result slice contains an item with the given ID.
+func containsID(result []*batch.Item[any], id uint64) bool {
+	for _, item := range result {
+		if item.ID == id {
+			return true
+		}
+	}
+	return false
+}
+
+func TestFilter_Process_PreservesErroredItems(t *testing.T) {
+	errBoom := errors.New("boom")
+
+	t.Run("errored item passes through even when predicate rejects it", func(t *testing.T) {
+		// The predicate would filter the errored item out, which used to silently
+		// drop its error before the batch engine could report it.
+		processor := &Filter[any]{
+			Predicate: func(item *batch.Item[any]) bool {
+				return item.Error == nil // would exclude any errored item
+			},
+		}
+
+		items := []*batch.Item[any]{
+			{ID: 1, Data: "ok"},
+			{ID: 2, Data: "failed", Error: errBoom},
+		}
+
+		ctx := context.Background()
+		result, err := processor.Process(ctx, items)
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+
+		if !containsID(result, 2) {
+			t.Fatalf("errored item (ID 2) was dropped; its error would be silently lost")
+		}
+
+		// Confirm the error is preserved unchanged on the passed-through item.
+		for _, item := range result {
+			if item.ID == 2 && !errors.Is(item.Error, errBoom) {
+				t.Errorf("expected errored item to retain its error, got %v", item.Error)
+			}
+		}
+
+		// The error-free item that the predicate keeps must still be present.
+		if !containsID(result, 1) {
+			t.Errorf("error-free matching item (ID 1) should be kept")
+		}
+	})
+
+	t.Run("predicate still filters error-free items", func(t *testing.T) {
+		// With one errored item passing through, the predicate must continue to
+		// apply normally to the error-free items.
+		processor := &Filter[any]{
+			Predicate: func(item *batch.Item[any]) bool {
+				return item.ID%2 == 0 // keep even IDs
+			},
+		}
+
+		items := []*batch.Item[any]{
+			{ID: 1, Data: "odd"},                     // filtered out
+			{ID: 2, Data: "even"},                    // kept
+			{ID: 3, Data: "errored", Error: errBoom}, // passes through despite odd ID
+			{ID: 4, Data: "even"},                    // kept
+		}
+
+		ctx := context.Background()
+		result, _ := processor.Process(ctx, items)
+
+		if len(result) != 3 {
+			t.Fatalf("expected 3 items (2 even + 1 errored), got %d", len(result))
+		}
+		if containsID(result, 1) {
+			t.Errorf("odd error-free item (ID 1) should have been filtered out")
+		}
+		if !containsID(result, 3) {
+			t.Errorf("errored item (ID 3) should pass through")
+		}
+	})
+
+	t.Run("errored item passes through even with InvertMatch", func(t *testing.T) {
+		// With InvertMatch the predicate keeps non-matching items. The errored item
+		// must bypass the predicate entirely regardless of inversion.
+		processor := &Filter[any]{
+			Predicate: func(item *batch.Item[any]) bool {
+				return true // inverted => would reject everything
+			},
+			InvertMatch: true,
+		}
+
+		items := []*batch.Item[any]{
+			{ID: 1, Data: "ok"},                     // inverted predicate => dropped
+			{ID: 2, Data: "failed", Error: errBoom}, // must pass through
+		}
+
+		ctx := context.Background()
+		result, _ := processor.Process(ctx, items)
+
+		if !containsID(result, 2) {
+			t.Fatalf("errored item (ID 2) was dropped under InvertMatch; error would be lost")
+		}
+		if containsID(result, 1) {
+			t.Errorf("error-free item (ID 1) should be dropped by inverted predicate")
+		}
+	})
+}