Fix "Jump to File" menu showing only half of the entries, in some rare cases (#6136)

When opening the `<ctrl+g>` files menu exactly at the moment where a
background fetch finishes, it could sometimes only show some of its
entries. The same could happen when pressing `>` to jump to the end of a
long diff at that moment.

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,42 +842,167 @@ 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()
+// 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
+// is answered once that has been read to the end, whether the replaced task was
+// already reading to the end for them or hadn't got to their request yet.
+func TestReadToEndWaitingWhenARerenderReplacesTheTask(t *testing.T) {
+	writer := bytes.NewBuffer(nil)
+	linesWritten := func() int { return strings.Count(writer.String(), "\n") }
+	firstPaint := make(chan struct{}, 1)
 
+	manager := NewViewBufferManager(
+		utils.NewDummyLog(),
+		writer,
+		func() {}, // beforeStart
+		func() {}, // refreshView
+		func() {}, // onEndOfInput
+		func() {}, // resetOrigin
+		writer.Reset,
+		func() { firstPaint <- struct{}{} },
+		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()
+
+	startTask := func(reader io.Reader) {
+		start := func() (Cmd, io.Reader) {
+			// not actually starting this because it's not necessary
+			return ExecCmd{Cmd: exec.Command("blah")}, reader
+		}
+		linesToRead := LinesToRead{Total: 1, InitialRefreshAfter: 1}
+		_ = manager.NewTask(manager.NewCmdTask(start, nil, linesToRead, nil), "cmd")
+	}
+
+	var mutex sync.Mutex
+	linesWrittenWhenAnswered := []int{}
+	var answered sync.WaitGroup
+	readToEnd := func() {
+		answered.Add(1)
+		manager.ReadToEnd(func() {
+			mutex.Lock()
+			linesWrittenWhenAnswered = append(linesWrittenWhenAnswered, linesWritten())
+			mutex.Unlock()
+			answered.Done()
+		})
+	}
+
+	// The render reads the line it was asked for, and stalls while reading to the end
+	// for the first caller. The second one's request waits behind it.
+	stalled := BlockingLineReader{linesToYield: 3, blocked: make(chan struct{}), unblock: make(chan struct{})}
+	defer close(stalled.unblock)
+	startTask(&stalled)
+	<-firstPaint
+	readToEnd()
+	<-stalled.blocked
+	readToEnd()
+
+	// A re-render of the same content, as a refresh does.
+	startTask(&BlankLineReader{totalLinesToYield: 50})
+	answered.Wait()
+
+	assert.Equal(t, []int{50, 50}, 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 task },
+		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 },
 	)
 
-	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)
+	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) })
 
-	// 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)
-
+	manager.Close()
 	assert.EqualValues(t, 2, answered.Load())
 }