diff --git a/processor/filter.go b/processor/filter.go index 145be76..0a5118d 100644 --- a/processor/filter.go +++ b/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 diff --git a/processor/filter_test.go b/processor/filter_test.go index a2cffe5..7fbffdf 100644 --- a/processor/filter_test.go +++ b/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") + } + }) +}