Merge pull request #726 from peco/fix-select1-streaming

fix: --select-1 triggers on interactive queries and respects context for streaming sources

action.go

@@ -41,7 +41,7 @@ var defaultKeyBinding map[string]Action
 // execQueryAndDraw runs ExecQuery and, if the query was non-empty
 // (ExecQuery returns false), sends a draw-prompt message.
 func execQueryAndDraw(ctx context.Context, state *Peco) {
-	if state.ExecQuery(ctx, nil) {
+	if state.ExecQuery(ctx, state.selectOneCallback()) {
 		return
 	}
 	state.Hub().SendDrawPrompt(ctx)
@@ -228,7 +228,7 @@ func doAcceptChar(ctx context.Context, state *Peco, e Event) {
 	h := state.Hub()
 	h.SendDrawPrompt(ctx) // Update prompt before running query
 
-	state.ExecQuery(ctx, nil)
+	state.ExecQuery(ctx, state.selectOneCallback())
 }
 
 func doRotateFilter(ctx context.Context, state *Peco, _ Event) {
@@ -682,7 +682,7 @@ func doKillEndOfLine(ctx context.Context, state *Peco, _ Event) {
 
 func doDeleteAll(ctx context.Context, state *Peco, _ Event) {
 	state.Query().Reset()
-	state.ExecQuery(ctx, nil)
+	state.ExecQuery(ctx, state.selectOneCallback())
 }
 
 func doDeleteForwardChar(ctx context.Context, state *Peco, _ Event) {

peco.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 	"unicode/utf8"
 
@@ -78,6 +79,7 @@ type Peco struct {
 	selectionRangeStart RangeStart
 	exitZeroAndExit     bool // True if --exit-0 is enabled
 	selectOneAndExit    bool // True if --select-1 is enabled
+	selectOneTriggered  atomic.Bool
 	selectAllAndExit    bool // True if --select-all is enabled
 	singleKeyJump       SingleKeyJumpState
 	heightSpec          *HeightSpec
@@ -411,8 +413,14 @@ func (p *Peco) Setup() (err error) {
 
 func (p *Peco) selectOneAndExitIfPossible() {
 	// If we have only one line, we just want to bail out
-	// printing that one line as the result
+	// printing that one line as the result.
+	// CAS guard: multiple goroutines may call this concurrently
+	// (startEarlyExitHandlers, ExecQuery callback, waitAndCall).
+	// Only the first to succeed after Size()==1 proceeds with exit.
 	if b := p.CurrentLineBuffer(); b.Size() == 1 {
+		if !p.selectOneTriggered.CompareAndSwap(false, true) {
+			return
+		}
 		if l, err := b.LineAt(0); err == nil {
 			ch := make(chan line.Line)
 			p.SetResultCh(ch)
@@ -423,6 +431,13 @@ func (p *Peco) selectOneAndExitIfPossible() {
 	}
 }
 
+func (p *Peco) selectOneCallback() func() {
+	if p.selectOneAndExit {
+		return p.selectOneAndExitIfPossible
+	}
+	return nil
+}
+
 func (p *Peco) exitZeroIfPossible() {
 	if p.CurrentLineBuffer().Size() == 0 {
 		p.Exit(setExitStatus(makeIgnorable(errors.New("no input, exiting")), 1))
@@ -885,11 +900,12 @@ func (p *Peco) sendQuery(ctx context.Context, q string, nextFunc func()) {
 
 	if p.source.IsInfinite() {
 		// If the source is a stream, we can't do batch mode, and hence
-		// we can't guarantee proper timing. But... okay, we simulate
-		// something like it
+		// we can't guarantee proper timing. Poll until select-1 is
+		// provably impossible (Size > 1), or fire the callback after
+		// a timeout as a best-effort check.
 		p.Hub().SendQuery(ctx, q)
 		if nextFunc != nil {
-			time.AfterFunc(time.Second, nextFunc)
+			go p.waitAndCall(ctx, nextFunc)
 		}
 	} else {
 		// No delay, execute immediately
@@ -902,6 +918,30 @@ func (p *Peco) sendQuery(ctx context.Context, q string, nextFunc func()) {
 	}
 }
 
+// waitAndCall is used for streaming/infinite sources where we can't wait
+// for the filter pipeline to complete. It fires fn after a timeout as a
+// best-effort check, respecting context cancellation. The ticker keeps
+// the loop iterating so the timer channel is reliably drained even if a
+// select iteration races with ctx.Done.
+func (p *Peco) waitAndCall(ctx context.Context, fn func()) {
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	timer := time.NewTimer(2 * time.Second)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			// periodic wake-up to re-enter select
+		case <-timer.C:
+			fn()
+			return
+		}
+	}
+}
+
 // ExecQuery executes the query, taking in consideration things like the
 // exec-delay, and user's multiple successive inputs in a very short span
 //

peco_test.go

@@ -1007,3 +1007,83 @@ func TestCancelFuncDataRace(t *testing.T) {
 		t.Fatal("timeout waiting for Run to return")
 	}
 }
+
+func TestSelect1WithQuery(t *testing.T) {
+	// --select-1 --query should auto-select when query narrows to exactly 1 match
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	p := newPeco()
+	p.Argv = []string{"--select-1", "--query", "bar"}
+	p.Stdin = bytes.NewBufferString("foo\nbar\nbaz\n")
+	var out bytes.Buffer
+	p.Stdout = &out
+
+	resultCh := make(chan error)
+	go func() {
+		defer close(resultCh)
+		select {
+		case <-ctx.Done():
+			return
+		case resultCh <- p.Run(ctx):
+			return
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("timeout: --select-1 --query bar should have auto-selected")
+	case err := <-resultCh:
+		require.True(t, util.IsCollectResultsError(err), "expected collectResultsError")
+		p.PrintResults()
+	}
+
+	require.Equal(t, "bar\n", out.String(), "output should be the single matching line")
+}
+
+func TestWaitAndCall(t *testing.T) {
+	t.Run("fires callback after timeout", func(t *testing.T) {
+		p := newPeco()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+
+		called := make(chan struct{})
+		start := time.Now()
+		go func() {
+			p.waitAndCall(ctx, func() { close(called) })
+		}()
+
+		select {
+		case <-called:
+			elapsed := time.Since(start)
+			require.True(t, elapsed >= 2*time.Second, "should wait at least 2s (got %v)", elapsed)
+		case <-time.After(5 * time.Second):
+			t.Fatal("callback was not fired within 5s")
+		}
+	})
+
+	t.Run("respects context cancellation", func(t *testing.T) {
+		p := newPeco()
+
+		ctx, cancel := context.WithCancel(context.Background())
+
+		called := false
+		done := make(chan struct{})
+		go func() {
+			p.waitAndCall(ctx, func() { called = true })
+			close(done)
+		}()
+
+		// Cancel quickly — before the 2s timer fires
+		time.Sleep(200 * time.Millisecond)
+		cancel()
+
+		select {
+		case <-done:
+			require.False(t, called, "callback should NOT fire after context cancellation")
+		case <-time.After(5 * time.Second):
+			t.Fatal("waitAndCall did not return after context cancellation")
+		}
+	})
+}