From ad8de095b5e9d68bfd6816bcaa7b968ffb4e59cb Mon Sep 17 00:00:00 2001 From: Vaughn Friesen Date: Fri, 29 May 2026 19:54:02 +0800 Subject: [PATCH] fix(processor): preserve errored items through Filter Filter.Process applied the predicate to every item, including items whose Error field was already set by an earlier processor. When the predicate excluded such an item it was dropped from the output slice, so the batch engine (which only surfaces item.Error for items present at end-of-chain) silently lost the error. Items carrying an error now pass through the filter unconditionally; the predicate (and InvertMatch) is applied only to error-free items. No exported API changed. Godoc updated to document the passthrough guarantee. Co-Authored-By: Claude Opus 4.8 (1M context) --- processor/filter.go | 16 ++++++ processor/filter_test.go | 108 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+) 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") + } + }) +}