Leave reads to the end to the task that replaces the one reading

Some actions read the content of the main view to the end before they
act on it: the jump-to-file menu, going to the bottom, searching, and
jumping to a file from the diffstat. If a re-render replaces the task
while it reads, these actions are told that the content has been read
although only a part of it has, so the menu lists only the files in
that part. A background fetch re-renders the main view once a minute,
so now and then the menu is incomplete. The same happens if a read is
asked for after a render has been asked for but before its task has
started.

These actions act on the content of the task asked for last, so leave
the reads waiting for content to that task. Hold them from the moment
the task is asked for, hand over the ones a replaced task hasn't
answered, including the one it was reading for, and answer them once
the task asked for last has given the view all of its content. The
replacing task still reads its own first lines first, so that the reads
it takes over don't delay its first paint.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

pkg/tasks/read_request_queue.go

@@ -15,9 +15,12 @@ import "sync"
 // 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.
+// A caller acts on the content the view shows once it is answered, so a request
+// is held for the task asked for last, from the moment it is asked for. When a
+// task is replaced, the requests it hasn't answered stay for the task replacing it.
+// Only once the task asked for last has given the view all of its content does the
+// queue stop holding requests, and hand back the ones it still holds to answer. A
+// request made after that 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
@@ -30,54 +33,82 @@ type readRequestQueue struct {
 	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
+	// Whether requests are held for a task to serve: the one reading the content, or
+	// one that has been asked for and is going to replace it. False before the first
+	// task is asked for, and once the task asked for last has read all of its
+	// content.
+	holding 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{} {
+// hold says that a task has been asked for, so that requests are held for it until
+// it serves them.
+func (self *readRequestQueue) hold() {
 	self.mutex.Lock()
 	defer self.mutex.Unlock()
 
-	self.serving = true
+	self.holding = true
+}
+
+// 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. The task's own first
+// request is served first, ahead of those it takes over from the task it replaces,
+// so that it shows its content as soon as it would have without them.
+func (self *readRequestQueue) beginServing(firstRequest LinesToRead) <-chan struct{} {
+	self.mutex.Lock()
+	self.holding = true
+	self.requests = append([]LinesToRead{firstRequest}, self.requests...)
+	self.mutex.Unlock()
+
+	self.ringDoorbell()
 	return self.doorbell
 }
 
-// stopServing takes the task away and hands back the requests it never answered,
-// for the caller to answer in its place.
+// stopServing stops holding requests and hands back the ones it still holds, for
+// the caller to answer in place of a task.
 func (self *readRequestQueue) stopServing() []LinesToRead {
 	self.mutex.Lock()
 	defer self.mutex.Unlock()
 
-	self.serving = false
+	self.holding = 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.
+// enqueue gives a request to the task that is going to serve it, and reports
+// whether there is one. When there isn't, the request is the caller's to answer.
 func (self *readRequestQueue) enqueue(request LinesToRead) bool {
 	self.mutex.Lock()
-	if !self.serving {
+	if !self.holding {
 		self.mutex.Unlock()
 		return false
 	}
 	self.requests = append(self.requests, request)
 	self.mutex.Unlock()
 
+	self.ringDoorbell()
+	return true
+}
+
+// requeue puts back a request that a task was serving when it was stopped, ahead of
+// the ones made after it, for the task replacing it to serve.
+func (self *readRequestQueue) requeue(request LinesToRead) {
+	self.mutex.Lock()
+	self.requests = append([]LinesToRead{request}, self.requests...)
+	self.mutex.Unlock()
+
+	self.ringDoorbell()
+}
+
+func (self *readRequestQueue) ringDoorbell() {
 	select {
 	case self.doorbell <- struct{}{}:
 	default:
 	}
-	return true
 }
 
 // dequeue takes the oldest request, reporting false when there are none.

pkg/tasks/read_request_queue_test.go

@@ -9,41 +9,60 @@ import (
 func TestReadRequestQueueHandsBackWhatNoTaskWillServe(t *testing.T) {
 	queue := newReadRequestQueue()
 
-	// Nothing has begun serving, so the request comes straight back to its caller.
+	// No task has been asked for, 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}))
+	// Once one has, requests are held for it, and the task serves its own first
+	// request ahead of them.
+	queue.hold()
 	assert.True(t, queue.enqueue(LinesToRead{Total: 2}))
+	queue.beginServing(LinesToRead{Total: 1})
+	assert.True(t, queue.enqueue(LinesToRead{Total: 3}))
 
 	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.
+	// A request that a task was serving when it was stopped goes back ahead of the
+	// ones made after it.
+	queue.requeue(request)
+	request, _ = queue.dequeue()
+	assert.Equal(t, 1, request.Total)
+	request, _ = queue.dequeue()
+	assert.Equal(t, 2, request.Total)
+
+	// What no task got to comes back when the queue stops holding requests, 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.Equal(t, 3, unanswered[0].Total)
 
-	assert.False(t, queue.enqueue(LinesToRead{Total: 3}))
+	assert.False(t, queue.enqueue(LinesToRead{Total: 4}))
 	_, ok = queue.dequeue()
 	assert.False(t, ok)
 }
 
 func TestReadRequestQueueRingsTheDoorbell(t *testing.T) {
 	queue := newReadRequestQueue()
-	doorbell := queue.beginServing()
+	doorbell := queue.beginServing(LinesToRead{Total: 1})
+
+	// The task's first request is there to be served right away.
+	select {
+	case <-doorbell:
+	default:
+		assert.Fail(t, "the doorbell didn't ring for the first request")
+	}
+	_, _ = queue.dequeue()
 
 	select {
 	case <-doorbell:
-		assert.Fail(t, "the doorbell rang before anything was queued")
+		assert.Fail(t, "the doorbell rang with nothing queued")
 	default:
 	}
 
 	// A burst leaves one token, which stands for everything queued.
-	queue.enqueue(LinesToRead{Total: 1})
 	queue.enqueue(LinesToRead{Total: 2})
+	queue.enqueue(LinesToRead{Total: 3})
 
 	select {
 	case <-doorbell:

pkg/tasks/tasks.go

@@ -435,11 +435,30 @@ func (self *ViewBufferManager) readHoldingATask(totalLines int, then func()) {
 	}
 }
 
-// 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.
+// stopServingReadRequests stops holding read requests for a task and answers the
+// ones still waiting, so that nobody is left waiting for a callback that isn't
+// coming.
 func (self *ViewBufferManager) stopServingReadRequests() {
-	for _, request := range self.readRequests.stopServing() {
+	answerReadRequests(self.readRequests.stopServing())
+}
+
+// answerReadRequestsIfLatest answers the read requests still waiting once the given
+// task has given the view all of its content, unless another task has been asked for
+// since. The requests are waiting for that task's content then, so they are left for
+// it to serve.
+func (self *ViewBufferManager) answerReadRequestsIfLatest(taskID int) {
+	self.taskIDMutex.Lock()
+	var unanswered []LinesToRead
+	if taskID == self.newTaskID {
+		unanswered = self.readRequests.stopServing()
+	}
+	self.taskIDMutex.Unlock()
+
+	answerReadRequests(unanswered)
+}
+
+func answerReadRequests(requests []LinesToRead) {
+	for _, request := range requests {
 		if request.Then != nil {
 			request.Then()
 		}
@@ -545,9 +564,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 
 		loadingMutex := deadlock.Mutex{}
 
-		// 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()
+		readRequests := self.readRequests.beginServing(initialRequest)
 
 		scanner := bufio.NewScanner(r)
 		scanner.Split(utils.ScanLinesAndTruncateWhenLongerThanBuffer(bufio.MaxScanTokenSize))
@@ -705,6 +722,15 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 							linesToRead.Then()
 						}
 					}
+					// A task that is stopped is being replaced by one that renders the
+					// content the caller is going to act on, so the request is left for
+					// that one to serve. A request without a Then has nobody waiting on
+					// it, and the task replacing this one reads what it needs by itself.
+					handOver := func() {
+						if linesToRead.Then != nil {
+							self.readRequests.requeue(linesToRead)
+						}
+					}
 					// A restore that hasn't painted yet keeps us reading past the lines
 					// asked for, all the way to the end of the input if need be. What it
 					// is looking for may be anywhere in the new content, and a rendering
@@ -713,14 +739,14 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 					// and the view somewhere the user didn't put it.
 					for linesToRead.Total == -1 || linesRead < linesToRead.Total || (restore != nil && !painted) {
 						if stopped() {
-							callThen()
+							handOver()
 							break outer
 						}
 						var ok bool
 						var line []byte
 						select {
 						case <-opts.Stop:
-							callThen()
+							handOver()
 							break outer
 						case line, ok = <-lineChan:
 							// process line below
@@ -752,7 +778,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 							// case above.
 							select {
 							case <-opts.Stop:
-								callThen()
+								handOver()
 								break outer
 							default:
 							}
@@ -805,10 +831,10 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 			}
 
 			// 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()
+			// content it asked for has been read. At end of input it has all been read
+			// already. A task that was stopped leaves the requests to the one replacing
+			// it.
+			self.answerReadRequestsIfLatest(opts.taskID)
 
 			refreshViewIfStale()
 
@@ -832,8 +858,6 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 			close(lineWrittenChan)
 		})
 
-		self.readRequests.enqueue(initialRequest)
-
 		<-done
 
 		return nil
@@ -842,6 +866,10 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix
 
 // Close closes the task manager, killing whatever task may currently be running
 func (self *ViewBufferManager) Close() {
+	// No task is going to read anything after this, so nobody may be left waiting
+	// for content to be read.
+	defer self.stopServingReadRequests()
+
 	// stopCurrentTask is written by NewTask's goroutine under waitingMutex (and
 	// so is the sync.Once it closes over), so read it under the lock and call
 	// the captured value; a task starting on shutdown must not race us here.
@@ -900,6 +928,8 @@ func (self *ViewBufferManager) ReserveTask() TaskReservation {
 	defer self.taskIDMutex.Unlock()
 
 	self.newTaskID++
+	// Whoever asks for content to be read from now on is asking for this task's.
+	self.readRequests.hold()
 	return TaskReservation{taskID: self.newTaskID}
 }
 
@@ -986,9 +1016,6 @@ func (self *ViewBufferManager) NewReservedTask(reservation TaskReservation, f fu
 			self.stopCurrentTask()
 		}
 
-		// Nothing serves read requests between one task and the next.
-		self.stopServingReadRequests()
-
 		stop := make(chan struct{})
 		notifyStopped := make(chan struct{})
 
@@ -1006,6 +1033,10 @@ func (self *ViewBufferManager) NewReservedTask(reservation TaskReservation, f fu
 			self.Log.Error(err) // might need an onError callback
 		}
 
+		// A task that reads no command's output, such as one showing a message, has
+		// given the view all of its content once it returns.
+		self.answerReadRequestsIfLatest(taskID)
+
 		close(notifyStopped)
 	})
 

pkg/tasks/tasks_test.go

@@ -842,45 +842,6 @@ func TestNewCmdTaskRefresh(t *testing.T) {
 	}
 }
 
-// A read request is answered by the task that serves it calling the request's Then.
-// This checks that requests still waiting when the task is stopped are answered too,
-// which is what happens when a re-render replaces the task.
-func TestQueuedReadRequestsAreAnsweredWhenTheTaskStops(t *testing.T) {
-	noop := func() {}
-	task := gocui.NewFakeTask()
-
-	// A pipe the task blocks on, so that the requests are still waiting when it stops.
-	pipeReader, pipeWriter := io.Pipe()
-	defer pipeWriter.Close()
-
-	manager := NewViewBufferManager(
-		utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop,
-		func() gocui.Task { return task },
-		func(f func()) error { f(); return nil },
-	)
-
-	stop := make(chan struct{})
-	fn := manager.NewCmdTask(
-		func() (Cmd, io.Reader) { return ExecCmd{Cmd: exec.Command("true")}, pipeReader },
-		nil, LinesToRead{Total: 1, InitialRefreshAfter: -1}, noop)
-	go func() { _, _ = pipeWriter.Write([]byte("first line\n")) }()
-	go func() { _ = fn(TaskOpts{Stop: stop, InitialContentLoaded: noop}) }()
-	// Let the task start and read the line it was asked for, so that the requests
-	// below are handed to a task that is waiting for them.
-	time.Sleep(50 * time.Millisecond)
-
-	answered := atomic.Int32{}
-	manager.ReadToEnd(func() { answered.Add(1) })
-	manager.ReadToEnd(func() { answered.Add(1) })
-
-	// Let the first request be picked up and block on the pipe, then stop the task.
-	time.Sleep(50 * time.Millisecond)
-	close(stop)
-	time.Sleep(50 * time.Millisecond)
-
-	assert.EqualValues(t, 2, answered.Load())
-}
-
 // A caller of ReadToEnd acts on the whole of what the view shows once it is answered,
 // such as listing the files of the diff in a menu. If a re-render replaces the task
 // that is reading the content, the content is the re-render's, so whoever is waiting
@@ -942,10 +903,107 @@ func TestReadToEndWaitingWhenARerenderReplacesTheTask(t *testing.T) {
 	startTask(&BlankLineReader{totalLinesToYield: 50})
 	answered.Wait()
 
-	/* EXPECTED:
 	assert.Equal(t, []int{50, 50}, linesWrittenWhenAnswered)
-	ACTUAL: */
-	assert.Equal(t, []int{3, 3}, linesWrittenWhenAnswered)
+}
+
+// A task that reads no command's output, such as one showing a message, has given the
+// view all of its content once it returns. So when it replaces a task that was reading
+// for someone, it answers the requests that task leaves it, after showing its content.
+func TestReadRequestsLeftToAMessageAreAnsweredOnceItIsShown(t *testing.T) {
+	noop := func() {}
+	manager := NewViewBufferManager(
+		utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop,
+		func() gocui.Task { return gocui.NewFakeTask() },
+		// no UI thread in the test; run the view mutations inline
+		func(f func()) error { f(); return nil },
+	)
+	defer manager.Close()
+
+	stalled := BlockingLineReader{linesToYield: 1, blocked: make(chan struct{}), unblock: make(chan struct{})}
+	defer close(stalled.unblock)
+	start := func() (Cmd, io.Reader) {
+		// not actually starting this because it's not necessary
+		return ExecCmd{Cmd: exec.Command("blah")}, &stalled
+	}
+	_ = manager.NewTask(manager.NewCmdTask(start, nil, LinesToRead{Total: 1, InitialRefreshAfter: 1}, nil), "cmd")
+	<-stalled.blocked
+
+	var messageShown atomic.Bool
+	answeredAfterTheMessage := make(chan bool, 2)
+	for range 2 {
+		manager.ReadToEnd(func() { answeredAfterTheMessage <- messageShown.Load() })
+	}
+
+	_ = manager.NewTask(func(TaskOpts) error {
+		messageShown.Store(true)
+		return nil
+	}, "message")
+
+	assert.True(t, <-answeredAfterTheMessage)
+	assert.True(t, <-answeredAfterTheMessage)
+}
+
+// A render whose task is created after the layout is the one whose content a request
+// made in the meantime is about, so the request waits for that task to read it, even
+// though the task before it had read all of its own.
+func TestReadToEndWaitsForATaskAskedForButNotCreatedYet(t *testing.T) {
+	writer := bytes.NewBuffer(nil)
+	linesWritten := func() int { return strings.Count(writer.String(), "\n") }
+	noop := func() {}
+	manager := NewViewBufferManager(
+		utils.NewDummyLog(), writer, noop, noop, noop, noop, writer.Reset, noop,
+		func() gocui.Task { return gocui.NewFakeTask() },
+		// no UI thread in the test; run the view mutations inline
+		func(f func()) error { f(); return nil },
+	)
+	defer manager.Close()
+
+	newCmdTask := func(lines int, onDone func()) func(TaskOpts) error {
+		start := func() (Cmd, io.Reader) {
+			// not actually starting this because it's not necessary
+			return ExecCmd{Cmd: exec.Command("blah")}, &BlankLineReader{totalLinesToYield: lines}
+		}
+		return manager.NewCmdTask(start, nil, LinesToRead{Total: 100, InitialRefreshAfter: 1}, onDone)
+	}
+
+	done := make(chan struct{})
+	_ = manager.NewTask(newCmdTask(3, func() { close(done) }), "cmd1")
+	<-done
+
+	reservation := manager.ReserveTask()
+	linesWrittenWhenAnswered := make(chan int, 1)
+	manager.ReadToEnd(func() { linesWrittenWhenAnswered <- linesWritten() })
+
+	_ = manager.NewReservedTask(reservation, newCmdTask(50, nil), "cmd2")
+	assert.Equal(t, 50, <-linesWrittenWhenAnswered)
+}
+
+// Once the view's tasks are closed, no task is going to read anything, so whoever is
+// waiting for content to be read is answered.
+func TestCloseAnswersWaitingReadRequests(t *testing.T) {
+	noop := func() {}
+	manager := NewViewBufferManager(
+		utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop,
+		func() gocui.Task { return gocui.NewFakeTask() },
+		// no UI thread in the test; run the view mutations inline
+		func(f func()) error { f(); return nil },
+	)
+
+	stalled := BlockingLineReader{linesToYield: 1, blocked: make(chan struct{}), unblock: make(chan struct{})}
+	defer close(stalled.unblock)
+	start := func() (Cmd, io.Reader) {
+		// not actually starting this because it's not necessary
+		return ExecCmd{Cmd: exec.Command("blah")}, &stalled
+	}
+	_ = manager.NewTask(manager.NewCmdTask(start, nil, LinesToRead{Total: 1, InitialRefreshAfter: 1}, nil), "cmd")
+	<-stalled.blocked
+
+	answered := atomic.Int32{}
+	manager.ReadToEnd(func() { answered.Add(1) })
+	manager.ReadToEnd(func() { answered.Add(1) })
+
+	manager.Close()
+	assert.EqualValues(t, 2, answered.Load())
 }
 
 // The callers of ReadToEnd hand their follow-up work to the UI thread from then, and