Merge pull request #682 from peco/fix-filter-apply-error-handling

Report filter Apply errors to status bar

filter.go

@@ -15,16 +15,17 @@ import (
 	"github.com/peco/peco/pipeline"
 )
 
-func newFilterProcessor(f filter.Filter, q string, bufSize int) *filterProcessor {
+func newFilterProcessor(f filter.Filter, q string, bufSize int, onError func(error)) *filterProcessor {
 	return &filterProcessor{
 		filter:  f,
 		query:   q,
 		bufSize: bufSize,
+		onError: onError,
 	}
 }
 
 func (fp *filterProcessor) Accept(ctx context.Context, in <-chan line.Line, out pipeline.ChanOutput) {
-	acceptAndFilter(ctx, fp.filter, fp.bufSize, in, out)
+	acceptAndFilter(ctx, fp.filter, fp.bufSize, fp.onError, in, out)
 }
 
 // orderedChunk is a batch of lines tagged with a sequence number
@@ -40,9 +41,18 @@ type orderedResult struct {
 	matched []line.Line
 }
 
+// reportFilterError calls onError with a non-context-cancellation error.
+// If the context is already cancelled or onError is nil, it does nothing.
+func reportFilterError(ctx context.Context, err error, onError func(error)) {
+	if err == nil || ctx.Err() != nil || onError == nil {
+		return
+	}
+	onError(err)
+}
+
 // flusher is the single-threaded fallback used when the filter does not
 // support parallel execution (e.g. Fuzzy with sortLongest).
-func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, done chan struct{}, out pipeline.ChanOutput) {
+func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, done chan struct{}, out pipeline.ChanOutput, onError func(error)) {
 	if pdebug.Enabled {
 		g := pdebug.Marker("flusher goroutine")
 		defer g.End()
@@ -62,7 +72,9 @@ func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, do
 			if pdebug.Enabled {
 				pdebug.Printf("flusher: %#v", buf)
 			}
-			_ = f.Apply(ctx, buf, out)
+			if err := f.Apply(ctx, buf, out); err != nil {
+				reportFilterError(ctx, err, onError)
+			}
 			buffer.ReleaseLineListBuf(buf)
 		}
 	}
@@ -70,7 +82,7 @@ func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, do
 
 // parallelFlusher distributes filter work across multiple goroutines
 // and merges the results back in sequence order.
-func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan orderedChunk, done chan struct{}, out pipeline.ChanOutput) {
+func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan orderedChunk, done chan struct{}, out pipeline.ChanOutput, onError func(error)) {
 	if pdebug.Enabled {
 		g := pdebug.Marker("parallelFlusher goroutine")
 		defer g.End()
@@ -107,13 +119,19 @@ func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan ordered
 				var matched []line.Line
 				if canCollect {
 					// Fast path: collect results directly into a slice
-					matched, _ = collector.ApplyCollect(ctx, chunk.lines)
+					var err error
+					matched, err = collector.ApplyCollect(ctx, chunk.lines)
+					if err != nil {
+						reportFilterError(ctx, err, onError)
+					}
 				} else {
 					// Fallback: use channel-based Apply for filters that
 					// don't implement Collector (e.g. ExternalCmd)
 					collectCh := make(chan line.Line, len(chunk.lines))
 					go func(chunk orderedChunk) {
-						_ = f.Apply(ctx, chunk.lines, pipeline.ChanOutput(collectCh))
+						if err := f.Apply(ctx, chunk.lines, pipeline.ChanOutput(collectCh)); err != nil {
+							reportFilterError(ctx, err, onError)
+						}
 						close(collectCh)
 					}(chunk)
 					matched = make([]line.Line, 0, len(chunk.lines)/2)
@@ -197,10 +215,10 @@ func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan ordered
 // It batches incoming lines and dispatches them to the filter, using parallel
 // workers when the filter supports it.
 func AcceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in <-chan line.Line, out pipeline.ChanOutput) {
-	acceptAndFilter(ctx, f, configBufSize, in, out)
+	acceptAndFilter(ctx, f, configBufSize, nil, in, out)
 }
 
-func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in <-chan line.Line, out pipeline.ChanOutput) {
+func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, onError func(error), in <-chan line.Line, out pipeline.ChanOutput) {
 	useParallel := f.SupportsParallel() && runtime.GOMAXPROCS(0) > 1
 
 	buf := buffer.GetLineListBuf()
@@ -214,26 +232,26 @@ func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in
 	}
 
 	if useParallel {
-		acceptAndFilterParallel(ctx, f, bufsiz, buf, in, out)
+		acceptAndFilterParallel(ctx, f, bufsiz, buf, onError, in, out)
 	} else {
-		acceptAndFilterSerial(ctx, f, bufsiz, buf, in, out)
+		acceptAndFilterSerial(ctx, f, bufsiz, buf, onError, in, out)
 	}
 }
 
-func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in <-chan line.Line, out pipeline.ChanOutput) {
+func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, onError func(error), in <-chan line.Line, out pipeline.ChanOutput) {
 	flush := make(chan []line.Line)
 	flushDone := make(chan struct{})
-	go flusher(ctx, f, flush, flushDone, out)
+	go flusher(ctx, f, flush, flushDone, out, onError)
 	defer func() { <-flushDone }()
 	defer close(flush)
 
 	batchAndFlush(ctx, bufsiz, buf, in, flush, func(b []line.Line) []line.Line { return b })
 }
 
-func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in <-chan line.Line, out pipeline.ChanOutput) {
+func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, onError func(error), in <-chan line.Line, out pipeline.ChanOutput) {
 	flush := make(chan orderedChunk)
 	flushDone := make(chan struct{})
-	go parallelFlusher(ctx, f, flush, flushDone, out)
+	go parallelFlusher(ctx, f, flush, flushDone, out, onError)
 	defer func() { <-flushDone }()
 	defer close(flush)
 
@@ -401,7 +419,12 @@ func (f *Filter) Work(ctx context.Context, q *hub.Payload[string]) {
 	p.SetSource(src)
 
 	ctx = selectedFilter.NewContext(ctx, query)
-	p.Add(newFilterProcessor(selectedFilter, query, state.config.FilterBufSize))
+	// Report non-cancellation filter errors (e.g. regex compilation failures)
+	// to the status bar so the user can see why results are missing.
+	onFilterError := func(err error) {
+		state.Hub().SendStatusMsg(ctx, err.Error(), 5*time.Second)
+	}
+	p.Add(newFilterProcessor(selectedFilter, query, state.config.FilterBufSize, onFilterError))
 
 	buf := NewMemoryBuffer(srcSize / 4)
 	p.SetDestination(buf)

filter_incremental_test.go

@@ -2,8 +2,10 @@ package peco
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"runtime"
+	"sync"
 	"testing"
 
 	"github.com/peco/peco/filter"
@@ -264,3 +266,52 @@ func TestIncrementalFiltering(t *testing.T) {
 	require.Contains(t, displayStrings, "foobaz entry")
 	require.Contains(t, displayStrings, "the foobird flies")
 }
+
+// errorFilter is a mock filter that always returns an error from Apply.
+type errorFilter struct {
+	err error
+}
+
+func (f *errorFilter) Apply(_ context.Context, _ []line.Line, _ pipeline.ChanOutput) error {
+	return f.err
+}
+
+func (f *errorFilter) BufSize() int                                             { return 0 }
+func (f *errorFilter) NewContext(ctx context.Context, _ string) context.Context { return ctx }
+func (f *errorFilter) String() string                                           { return "error-filter" }
+func (f *errorFilter) SupportsParallel() bool                                   { return false }
+
+// TestFilterApplyErrorReporting verifies that when a filter's Apply method
+// returns an error, the error is propagated to the onError callback (which in
+// production sends a status bar message to the user).
+func TestFilterApplyErrorReporting(t *testing.T) {
+	simulatedErr := errors.New("simulated filter error")
+	ef := &errorFilter{err: simulatedErr}
+
+	inputLines := []line.Line{
+		line.NewRaw(0, "alpha", false, false),
+		line.NewRaw(1, "bravo", false, false),
+	}
+
+	in := make(chan line.Line, len(inputLines))
+	for _, l := range inputLines {
+		in <- l
+	}
+	close(in)
+
+	var mu sync.Mutex
+	var reported []error
+	onError := func(err error) {
+		mu.Lock()
+		defer mu.Unlock()
+		reported = append(reported, err)
+	}
+
+	out := make(chan line.Line, len(inputLines))
+	acceptAndFilter(context.Background(), ef, 0, onError, in, pipeline.ChanOutput(out))
+
+	mu.Lock()
+	defer mu.Unlock()
+	require.Len(t, reported, 1, "onError should have been called once")
+	require.Equal(t, simulatedErr, reported[0])
+}

interface.go

@@ -525,4 +525,5 @@ type filterProcessor struct {
 	filter  filter.Filter
 	query   string
 	bufSize int
+	onError func(error)
 }

issues_test.go

@@ -135,7 +135,7 @@ func TestIssue557_FilterBufSize(t *testing.T) {
 	// Use a configBufSize large enough to hold all lines in a single batch.
 	// This ensures the fuzzy filter sees all lines at once and can sort globally.
 	configBufSize := totalLines + 100
-	fp := newFilterProcessor(f, query, configBufSize)
+	fp := newFilterProcessor(f, query, configBufSize, nil)
 
 	// Set up pipeline: source -> filterProcessor -> destination
 	p := pipeline.New()