From e7db38fa078a86acdee43b6b9d50b7e787c03588 Mon Sep 17 00:00:00 2001 From: Daisuke Maki Date: Wed, 18 Feb 2026 09:16:14 +0900 Subject: [PATCH 1/2] Report filter Apply errors to status bar --- filter.go | 55 +++++++++++++++++++++++++++----------- filter_incremental_test.go | 51 +++++++++++++++++++++++++++++++++++ interface.go | 1 + issues_test.go | 2 +- 4 files changed, 92 insertions(+), 17 deletions(-) diff --git a/filter.go b/filter.go index bb2ea20..0d3eea2 100644 --- a/filter.go +++ b/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) diff --git a/filter_incremental_test.go b/filter_incremental_test.go index 4717bc9..b70e128 100644 --- a/filter_incremental_test.go +++ b/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]) +} diff --git a/interface.go b/interface.go index 50a5a34..5869386 100644 --- a/interface.go +++ b/interface.go @@ -525,4 +525,5 @@ type filterProcessor struct { filter filter.Filter query string bufSize int + onError func(error) } diff --git a/issues_test.go b/issues_test.go index ca481b9..3e9f13f 100644 --- a/issues_test.go +++ b/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() From 73d5d370a4a7a2e0a1898c543262c0b0a47b055f Mon Sep 17 00:00:00 2001 From: Daisuke Maki Date: Wed, 18 Feb 2026 09:35:29 +0900 Subject: [PATCH 2/2] appease linter --- filter_incremental_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/filter_incremental_test.go b/filter_incremental_test.go index b70e128..24a9790 100644 --- a/filter_incremental_test.go +++ b/filter_incremental_test.go @@ -276,10 +276,10 @@ func (f *errorFilter) Apply(_ context.Context, _ []line.Line, _ pipeline.ChanOut return f.err } -func (f *errorFilter) BufSize() int { return 0 } +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 } +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