fix(terminal): don't poll for output during scrollback refresh (#38365)

Problem:
If buffer update callbacks poll for uv events during terminal scrollback
refresh, new output from PTY process may lead to incorrect scrollback.

Solution:
Don't poll for output to the same terminal as the one being refreshed.

src/nvim/api/vim.c

@@ -1153,6 +1153,7 @@ Integer nvim_open_term(Buffer buffer, Dict(open_term) *opts, Error *err)
     // displaying the buffer
     .width = (uint16_t)MAX(curwin->w_view_width - win_col_off(curwin), 0),
     .height = (uint16_t)curwin->w_view_height,
+    .read_pause_cb = term_read_pause,
     .write_cb = term_write,
     .resize_cb = term_resize,
     .resume_cb = term_resume,
@@ -1185,6 +1186,11 @@ Integer nvim_open_term(Buffer buffer, Dict(open_term) *opts, Error *err)
   return (Integer)chan->id;
 }
 
+static void term_read_pause(bool pause, void *data)
+{
+  // Not currently needed as sending to channel isn't allowed during buffer updates.
+}
+
 static void term_write(const char *buf, size_t size, void *data)
 {
   Channel *chan = data;

src/nvim/channel.c

@@ -847,6 +847,7 @@ void channel_terminal_alloc(buf_T *buf, Channel *chan)
     .data = chan,
     .width = chan->stream.pty.width,
     .height = chan->stream.pty.height,
+    .read_pause_cb = term_read_pause,
     .write_cb = term_write,
     .resize_cb = term_resize,
     .resume_cb = term_resume,
@@ -858,6 +859,19 @@ void channel_terminal_alloc(buf_T *buf, Channel *chan)
   chan->term = terminal_alloc(buf, topts);
 }
 
+static void term_read_pause(bool pause, void *data)
+{
+  Channel *chan = data;
+  if (chan->stream.proc.out.s.closed) {
+    return;
+  }
+  if (pause) {
+    rstream_stop_inner(&chan->stream.proc.out);
+  } else {
+    rstream_start_inner(&chan->stream.proc.out);
+  }
+}
+
 static void term_write(const char *buf, size_t size, void *data)
 {
   Channel *chan = data;

src/nvim/terminal.c

@@ -2530,6 +2530,10 @@ static void adjust_scrollback(Terminal *term, buf_T *buf)
 // Refresh the scrollback of an invalidated terminal.
 static void refresh_scrollback(Terminal *term, buf_T *buf)
 {
+  // Buffer update callbacks may poll for uv events.
+  // Avoid polling for output to the same terminal as the one being refreshed.
+  term->opts.read_pause_cb(true, term->opts.data);
+
   linenr_T deleted = (linenr_T)(term->sb_deleted - term->old_sb_deleted);
   deleted = MIN(deleted, buf->b_ml.ml_line_count);
   mark_adjust_buf(buf, 1, deleted, MAXLNUM, -deleted, true, kMarkAdjustTerm, kExtmarkUndo);
@@ -2569,6 +2573,8 @@ static void refresh_scrollback(Terminal *term, buf_T *buf)
   }
 
   adjust_scrollback(term, buf);
+
+  term->opts.read_pause_cb(false, term->opts.data);
 }
 
 // Refresh the screen (visible part of the buffer when the terminal is

src/nvim/terminal.h

@@ -7,6 +7,7 @@
 #include "nvim/api/private/defs.h"  // IWYU pragma: keep
 #include "nvim/types_defs.h"  // IWYU pragma: keep
 
+typedef void (*terminal_read_pause_cb)(bool pause, void *data);
 typedef void (*terminal_write_cb)(const char *buffer, size_t size, void *data);
 typedef void (*terminal_resize_cb)(uint16_t width, uint16_t height, void *data);
 typedef void (*terminal_resume_cb)(void *data);
@@ -15,6 +16,7 @@ typedef void (*terminal_close_cb)(void *data);
 typedef struct {
   void *data;  // PTY process channel
   uint16_t width, height;
+  terminal_read_pause_cb read_pause_cb;
   terminal_write_cb write_cb;
   terminal_resize_cb resize_cb;
   terminal_resume_cb resume_cb;

test/functional/terminal/buffer_spec.lua

@@ -992,8 +992,9 @@ describe(':terminal buffer', function()
   end)
 
   --- @param subcmd 'REP'|'REPFAST'
-  local function check_term_rep(subcmd, count)
+  local function check_term_rep(subcmd, count, scrollback)
     local screen = Screen.new(50, 7)
+    api.nvim_set_option_value('scrollback', scrollback, {})
     api.nvim_create_autocmd('TermClose', { command = 'let g:did_termclose = 1' })
     fn.jobstart({ testprg('shell-test'), subcmd, count, 'TEST' }, { term = true })
     retry(nil, nil, function()
@@ -1010,23 +1011,24 @@ describe(':terminal buffer', function()
       {5:-- TERMINAL --}                                    |
     ]]):format(count - 5, count - 4, count - 3, count - 2, count - 1))
     local lines = api.nvim_buf_get_lines(0, 0, -1, true)
-    for i = 1, count do
-      eq(('%d: TEST'):format(i - 1), lines[i])
+    local start = math.max(count + 1 - scrollback - 6, 0)
+    for i = start, count - 1 do
+      eq(('%d: TEST'):format(i), lines[i - start + 1])
     end
+    eq('', lines[#lines])
+    eq(count - start + 1, #lines)
   end
 
   it('does not drop data when job exits immediately after output #3030', function()
-    api.nvim_set_option_value('scrollback', 30000, {})
-    check_term_rep('REPFAST', 20000)
+    check_term_rep('REPFAST', 20000, 30000)
   end)
 
   it('does not drop data when autocommands poll for events #37559', function()
-    api.nvim_set_option_value('scrollback', 30000, {})
     api.nvim_create_autocmd('BufFilePre', { command = 'sleep 50m', nested = true })
     api.nvim_create_autocmd('BufFilePost', { command = 'sleep 50m', nested = true })
     api.nvim_create_autocmd('TermOpen', { command = 'sleep 50m', nested = true })
     -- REP pauses 1 ms every 100 lines, so each autocommand processes some output.
-    check_term_rep('REP', 20000)
+    check_term_rep('REP', 20000, 30000)
   end)
 
   describe('scrollback is correct if all output is drained by', function()
@@ -1037,13 +1039,44 @@ describe(':terminal buffer', function()
           it(('%.1f * terminal refresh delay'):format(delay / 10), function()
             local cmd = ('sleep %dm'):format(delay)
             api.nvim_create_autocmd(event, { command = cmd, nested = true })
-            check_term_rep('REPFAST', 200)
+            check_term_rep('REPFAST', 200, 10000)
           end)
         end
       end)
     end
   end)
 
+  it('scrollback is correct if buffer update callbacks poll for uv events', function()
+    -- Use vim.regex:match_str(), which may poll for uv events.
+    exec_lua(function()
+      local regex = vim.regex([[^\d\+: TEST]])
+      _G.matched_lines = {} --- @type table<string,boolean>
+      vim.api.nvim_create_autocmd('TermOpen', {
+        callback = function(ev)
+          vim.api.nvim_buf_attach(ev.buf, false, {
+            on_lines = function(_, buf, _, first, _, last, _)
+              local lines = vim.api.nvim_buf_get_lines(buf, first, last, true)
+              for _, line in ipairs(lines) do
+                if regex:match_str(line) then
+                  _G.matched_lines[vim.trim(line)] = true
+                end
+              end
+            end,
+          })
+        end,
+      })
+    end)
+    -- Use a 'scrollback' value smaller than the number of printed lines.
+    -- REP pauses 1 ms every 100 lines, so this can take at least 20 refresh cycles.
+    check_term_rep('REP', 20000, 10000)
+    -- Check that buffer update callbacks have seen all output lines.
+    local matched_lines = exec_lua('return _G.matched_lines')
+    for i = 0, 19999 do
+      local line = ('%d: TEST'):format(i)
+      eq(true, matched_lines[line], line)
+    end
+  end)
+
   it('handles unprintable chars', function()
     local screen = Screen.new(50, 7)
     feed 'i'