Answer every read request, whether or not a task is still serving

A caller of ReadLines or ReadToEnd is told that the content it asked for
has been read by the request's Then being called. A task that reaches the
end of its input answers the requests still queued behind the one it was
serving, but a task that is stopped drops them, and their callers wait
for a callback that never comes. Pressing "/" in the focused main view
opens the search prompt from such a callback, so if a re-render replaces
the task at that moment the prompt never opens.

Answering them as the read loop ends would leave a request handed over
after that point unanswered, and there is a window for one. A caller
reads the channel to send on, and can reach the send itself only once the
loop has gone. So hand requests over through a queue instead. Asking
whether a task is there and giving it the request are one step, as are
taking the task away and handing back what it never answered; a request
made in between goes back to the caller to answer.

The queue is unbounded rather than a fixed-size channel, for the reasons
gocui's userEventQueue is. Requests are handed over from the UI thread,
where a blocking send would deadlock against the task waiting to be let
go, and a fixed channel that fills up leaves only blocking, dropping,
reordering or panicking to choose between.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

pkg/tasks/read_request_queue.go

@@ -0,0 +1,101 @@
+package tasks
+
+import "sync"
+
+// readRequestQueue is an unbounded, order-preserving FIFO of the read requests a
+// view's running command task serves (see LinesToRead), with a reader that comes
+// and goes.
+//
+// It's unbounded, rather than a fixed-size channel, for the same reasons as the
+// user-event queue in gocui. Requests are handed over from the UI thread, where a
+// blocking send would deadlock against the task that is waiting to be let go, and
+// a fixed channel that fills up leaves only bad choices: blocking, dropping,
+// reordering, or panicking on overflow. Appending to a slice does none of those.
+//
+// The reader coming and going is the other half of what it's for. A request is
+// how a caller asks for content to be read and hears, through the request's Then,
+// that it has been; a request nobody answers leaves that caller waiting for good.
+// So asking whether a task is there and handing it the request are one step, and
+// so are taking the task away and handing back what it never answered. A request
+// made in between finds no task and goes back to its caller to answer.
+//
+// enqueue appends under the mutex and rings the doorbell; the task selects on the
+// doorbell to wake, then takes requests until there are none left. The doorbell is
+// buffered(1) and rung with a non-blocking send, so it's a coalescing "work
+// pending" flag rather than a per-request signal: a burst of appends leaves at
+// most one token, and the task takes everything the token stands for on a single
+// wake. A token left over after the queue empties causes one harmless empty wake.
+type readRequestQueue struct {
+	mutex    sync.Mutex
+	requests []LinesToRead
+	doorbell chan struct{}
+
+	// Whether a task is there to serve the requests. False before the first task
+	// starts, and between one task ending and the next starting.
+	serving bool
+}
+
+func newReadRequestQueue() *readRequestQueue {
+	return &readRequestQueue{doorbell: make(chan struct{}, 1)}
+}
+
+// beginServing says that a task is now there to serve the queue, and returns the
+// doorbell that tells it when there is something to serve.
+func (self *readRequestQueue) beginServing() <-chan struct{} {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+
+	self.serving = true
+	return self.doorbell
+}
+
+// stopServing takes the task away and hands back the requests it never answered,
+// for the caller to answer in its place.
+func (self *readRequestQueue) stopServing() []LinesToRead {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+
+	self.serving = false
+	unanswered := self.requests
+	self.requests = nil
+	return unanswered
+}
+
+// enqueue gives a request to the task serving the queue, and reports whether
+// there was one to give it to. When there wasn't, the request is the caller's to
+// answer.
+func (self *readRequestQueue) enqueue(request LinesToRead) bool {
+	self.mutex.Lock()
+	if !self.serving {
+		self.mutex.Unlock()
+		return false
+	}
+	self.requests = append(self.requests, request)
+	self.mutex.Unlock()
+
+	select {
+	case self.doorbell <- struct{}{}:
+	default:
+	}
+	return true
+}
+
+// dequeue takes the oldest request, reporting false when there are none.
+func (self *readRequestQueue) dequeue() (LinesToRead, bool) {
+	self.mutex.Lock()
+	defer self.mutex.Unlock()
+
+	if len(self.requests) == 0 {
+		return LinesToRead{}, false
+	}
+	request := self.requests[0]
+	if len(self.requests) == 1 {
+		// Release the backing array whenever the queue drains, so a one-off burst
+		// doesn't pin its peak size for the rest of the session.
+		self.requests = nil
+	} else {
+		self.requests[0] = LinesToRead{}
+		self.requests = self.requests[1:]
+	}
+	return request, true
+}

pkg/tasks/read_request_queue_test.go

@@ -0,0 +1,58 @@
+package tasks
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReadRequestQueueHandsBackWhatNoTaskWillServe(t *testing.T) {
+	queue := newReadRequestQueue()
+
+	// Nothing has begun serving, so the request comes straight back to its caller.
+	assert.False(t, queue.enqueue(LinesToRead{Total: 1}))
+
+	queue.beginServing()
+	assert.True(t, queue.enqueue(LinesToRead{Total: 1}))
+	assert.True(t, queue.enqueue(LinesToRead{Total: 2}))
+
+	request, ok := queue.dequeue()
+	assert.True(t, ok)
+	assert.Equal(t, 1, request.Total)
+
+	// What the task never got to comes back when it stops, and nothing is taken
+	// from a caller after that.
+	unanswered := queue.stopServing()
+	assert.Len(t, unanswered, 1)
+	assert.Equal(t, 2, unanswered[0].Total)
+
+	assert.False(t, queue.enqueue(LinesToRead{Total: 3}))
+	_, ok = queue.dequeue()
+	assert.False(t, ok)
+}
+
+func TestReadRequestQueueRingsTheDoorbell(t *testing.T) {
+	queue := newReadRequestQueue()
+	doorbell := queue.beginServing()
+
+	select {
+	case <-doorbell:
+		assert.Fail(t, "the doorbell rang before anything was queued")
+	default:
+	}
+
+	// A burst leaves one token, which stands for everything queued.
+	queue.enqueue(LinesToRead{Total: 1})
+	queue.enqueue(LinesToRead{Total: 2})
+
+	select {
+	case <-doorbell:
+	default:
+		assert.Fail(t, "the doorbell didn't ring")
+	}
+	select {
+	case <-doorbell:
+		assert.Fail(t, "the doorbell rang twice for one wake")
+	default:
+	}
+}

pkg/tasks/tasks.go

@@ -70,12 +70,11 @@ type ViewBufferManager struct {
 	taskIDMutex deadlock.Mutex
 	Log         *logrus.Entry
 	newTaskID   int
-	// The channel by which the currently-running task is told to read more
-	// lines (e.g. as the user scrolls). Held in an atomic because it's swapped
-	// out as tasks come and go while ReadLines/ReadToEnd read it from the UI
-	// thread; nil when no task is running.
-	readLines atomic.Pointer[chan LinesToRead]
-	taskKey   string
+	// The requests by which the currently-running task is told to read more lines
+	// (e.g. as the user scrolls), and which it answers once it has. The task
+	// serving them comes and goes; see readRequestQueue.
+	readRequests *readRequestQueue
+	taskKey      string
 
 	// Resets the view's scroll position to the top. A render whose content is
 	// different from what the view last showed (a different command key) calls
@@ -173,6 +172,7 @@ func NewViewBufferManager(
 	onUIThread func(f func()) error,
 ) *ViewBufferManager {
 	return &ViewBufferManager{
+		readRequests: newReadRequestQueue(),
 		Log:          log,
 		writer:       writer,
 		beforeStart:  beforeStart,
@@ -191,12 +191,9 @@ func NewViewBufferManager(
 // (e.g. as the user scrolls down, back up, and down again) don't re-read lines
 // that have already been read: the task only ever reads the shortfall.
 func (self *ViewBufferManager) ReadLines(totalLines int) {
-	if ch := self.readLines.Load(); ch != nil {
-		readLines := *ch
-		go utils.Safe(func() {
-			readLines <- LinesToRead{Total: totalLines, InitialRefreshAfter: -1}
-		})
-	}
+	// A request with no Then needs no answer, so there is nothing to do when no
+	// task is there to take it.
+	self.readRequests.enqueue(LinesToRead{Total: totalLines, InitialRefreshAfter: -1})
 }
 
 // IsLoading reports whether a command task is currently reading content into the
@@ -215,16 +212,24 @@ func (self *ViewBufferManager) StartLoading() {
 }
 
 func (self *ViewBufferManager) ReadToEnd(then func()) {
-	if ch := self.readLines.Load(); ch != nil {
-		readLines := *ch
-		go utils.Safe(func() {
-			readLines <- LinesToRead{Total: -1, InitialRefreshAfter: -1, Then: then}
-		})
-	} else if then != nil {
+	request := LinesToRead{Total: -1, InitialRefreshAfter: -1, Then: then}
+	if !self.readRequests.enqueue(request) && then != nil {
+		// With no task reading, everything there is to read has been read.
 		then()
 	}
 }
 
+// stopServingReadRequests takes the task away from the read-request queue and
+// answers whatever it never got to, so that nobody is left waiting for a callback
+// that isn't coming.
+func (self *ViewBufferManager) stopServingReadRequests() {
+	for _, request := range self.readRequests.stopServing() {
+		if request.Then != nil {
+			request.Then()
+		}
+	}
+}
+
 func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix string, linesToRead LinesToRead, onDoneFn func()) func(TaskOpts) error {
 	return func(opts TaskOpts) error {
 		var onDoneOnce sync.Once
@@ -289,8 +294,9 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 
 		loadingMutex := deadlock.Mutex{}
 
-		readLines := make(chan LinesToRead, 1024)
-		self.readLines.Store(&readLines)
+		// Begin serving before any goroutine starts, so that the first request below
+		// can't arrive before there is a task to take it.
+		readRequests := self.readRequests.beginServing()
 
 		scanner := bufio.NewScanner(r)
 		scanner.Split(utils.ScanLinesAndTruncateWhenLongerThanBuffer(bufio.MaxScanTokenSize))
@@ -423,10 +429,17 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 				if stopped() {
 					break outer
 				}
-				select {
-				case <-opts.Stop:
-					break outer
-				case linesToRead := <-readLines:
+				linesToRead, ok := self.readRequests.dequeue()
+				if !ok {
+					// Nothing to read yet: wait to be told there is, or to be stopped.
+					select {
+					case <-opts.Stop:
+						break outer
+					case <-readRequests:
+					}
+					continue
+				}
+				{
 					callThen := func() {
 						if linesToRead.Then != nil {
 							linesToRead.Then()
@@ -493,21 +506,6 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 							// means a newer task is taking over and is still loading.
 							self.loading.Store(false)
 							callThen()
-							// Any read requests that were queued while we were reading are
-							// now trivially satisfied, since we've read everything. Fire
-							// their callbacks instead of dropping them when we break out of
-							// the loop below (and nil out readLines).
-						drain:
-							for {
-								select {
-								case queued := <-readLines:
-									if queued.Then != nil {
-										queued.Then()
-									}
-								default:
-									break drain
-								}
-							}
 							break outer
 						}
 						writeToView(append(line, '\n'))
@@ -532,7 +530,11 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 				}
 			}
 
-			self.readLines.Store(nil)
+			// Whoever made a request the loop never got to is waiting to hear that the
+			// content it asked for has been read, and there is nothing here to read it
+			// any more: at end of input it has all been read already, and a task that
+			// was stopped is handing the view over to the one replacing it.
+			self.stopServingReadRequests()
 
 			refreshViewIfStale()
 
@@ -556,7 +558,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 			close(lineWrittenChan)
 		})
 
-		readLines <- linesToRead
+		self.readRequests.enqueue(linesToRead)
 
 		<-done
 
@@ -670,7 +672,8 @@ func (self *ViewBufferManager) NewTask(f func(TaskOpts) error, key string) error
 			self.stopCurrentTask()
 		}
 
-		self.readLines.Store(nil)
+		// Nothing serves read requests between one task and the next.
+		self.stopServingReadRequests()
 
 		stop := make(chan struct{})
 		notifyStopped := make(chan struct{})

pkg/tasks/tasks_test.go

@@ -253,12 +253,9 @@ func TestNewCmdTaskQueuedReadAtEndOfInput(t *testing.T) {
 	})
 
 	<-reader.blocked
+	// The request is queued by the time this returns, so it is outstanding when we
+	// let the task reach EOF below.
 	manager.ReadToEnd(func() { thenCalled = true })
-	// ReadToEnd queues its request from a goroutine; wait for it to land so that
-	// it is definitely outstanding by the time we let the task reach EOF.
-	for len(*manager.readLines.Load()) == 0 {
-		time.Sleep(time.Millisecond)
-	}
 	close(reader.unblock)
 
 	wg.Wait()
@@ -523,8 +520,5 @@ func TestQueuedReadRequestsAreAnsweredWhenTheTaskStops(t *testing.T) {
 	close(stop)
 	time.Sleep(50 * time.Millisecond)
 
-	/* EXPECTED:
 	assert.EqualValues(t, 2, answered.Load())
-	ACTUAL: */
-	assert.EqualValues(t, 1, answered.Load())
 }