fix(terminal): avoid rescheduling events onto the same queue (#32755)

Problem:

When a function like vim.wait() is used, we continuously drain the main
event queue until it is empty, never stopping for user input. This means
the libuv timer never runs and the terminal never gets refreshed, so
emit_termrequest continously reschedules itself onto the same event
queue, causing an infinite loop.

Solution:

Use a separate "pending" event queue, where events that require a
terminal refresh are temporarily placed. Drain this queue after a
terminal refresh and events are copied back onto the main queue. This
prevents infinite loops since the main event queue will always be able
to properly drain.

src/nvim/event/loop.c

@@ -212,10 +212,7 @@ static void async_cb(uv_async_t *handle)
   Loop *l = handle->loop->data;
   uv_mutex_lock(&l->mutex);
   // Flush thread_events to fast_events for processing on main loop.
-  while (!multiqueue_empty(l->thread_events)) {
-    Event ev = multiqueue_get(l->thread_events);
-    multiqueue_put_event(l->fast_events, ev);
-  }
+  multiqueue_move_events(l->fast_events, l->thread_events);
   uv_mutex_unlock(&l->mutex);
 }
 

src/nvim/event/multiqueue.c

@@ -140,6 +140,16 @@ void multiqueue_put_event(MultiQueue *self, Event event)
   }
 }
 
+/// Move events from src to dest.
+void multiqueue_move_events(MultiQueue *dest, MultiQueue *src)
+  FUNC_ATTR_NONNULL_ALL
+{
+  while (!multiqueue_empty(src)) {
+    Event event = multiqueue_get(src);
+    multiqueue_put_event(dest, event);
+  }
+}
+
 void multiqueue_process_events(MultiQueue *self)
 {
   assert(self);

src/nvim/terminal.c

@@ -178,6 +178,7 @@ struct terminal {
     bool resize;          ///< pending width/height
     bool cursor;          ///< pending cursor shape or blink change
     StringBuilder *send;  ///< When there is a pending TermRequest autocommand, block and store input.
+    MultiQueue *events;   ///< Events waiting for refresh.
   } pending;
 
   bool theme_updates;  ///< Send a theme update notification when 'bg' changes
@@ -222,9 +223,10 @@ static void emit_termrequest(void **argv)
 
   if (term->sb_pending > 0) {
     // Don't emit the event while there is pending scrollback because we need
-    // the buffer contents to be fully updated. If this is the case, re-schedule
-    // the event.
-    multiqueue_put(main_loop.events, emit_termrequest, term, sequence, (void *)sequence_length,
+    // the buffer contents to be fully updated. If this is the case, schedule
+    // the event onto the pending queue where it will be executed after the
+    // terminal is refreshed and the pending scrollback is cleared.
+    multiqueue_put(term->pending.events, emit_termrequest, term, sequence, (void *)sequence_length,
                    pending_send, (void *)(intptr_t)row, (void *)(intptr_t)col);
     return;
   }
@@ -490,6 +492,13 @@ void terminal_open(Terminal **termpp, buf_T *buf, TerminalOptions opts)
   term->invalid_start = 0;
   term->invalid_end = opts.height;
 
+  // Create a separate queue for events which need to wait for a terminal
+  // refresh. We cannot reschedule events back onto the main queue because this
+  // can create an infinite loop (#32753).
+  // This queue is never processed directly: when the terminal is refreshed, all
+  // events from this queue are copied back onto the main event queue.
+  term->pending.events = multiqueue_new_parent(NULL, NULL);
+
   aco_save_T aco;
   aucmd_prepbuf(&aco, buf);
 
@@ -972,6 +981,7 @@ void terminal_destroy(Terminal **termpp)
     kv_destroy(term->selection);
     kv_destroy(term->termrequest_buffer);
     vterm_free(term->vt);
+    multiqueue_free(term->pending.events);
     xfree(term);
     *termpp = NULL;  // coverity[dead-store]
   }
@@ -2029,6 +2039,9 @@ static void refresh_terminal(Terminal *term)
 
   int ml_added = buf->b_ml.ml_line_count - ml_before;
   adjust_topline(term, buf, ml_added);
+
+  // Copy pending events back to the main event queue
+  multiqueue_move_events(main_loop.events, term->pending.events);
 }
 
 static void refresh_cursor(Terminal *term, bool *cursor_visible)

test/functional/terminal/buffer_spec.lua

@@ -333,96 +333,123 @@ describe(':terminal buffer', function()
     command('bdelete!')
   end)
 
-  it('emits TermRequest events #26972', function()
-    local term = api.nvim_open_term(0, {})
-    local termbuf = api.nvim_get_current_buf()
+  describe('TermRequest', function()
+    it('emits events #26972', function()
+      local term = api.nvim_open_term(0, {})
+      local termbuf = api.nvim_get_current_buf()
 
-    -- Test that <abuf> is the terminal buffer, not the current buffer
-    command('au TermRequest * let g:termbuf = +expand("<abuf>")')
-    command('wincmd p')
+      -- Test that <abuf> is the terminal buffer, not the current buffer
+      command('au TermRequest * let g:termbuf = +expand("<abuf>")')
+      command('wincmd p')
 
-    -- cwd will be inserted in a file URI, which cannot contain backs
-    local cwd = t.fix_slashes(fn.getcwd())
-    local parent = cwd:match('^(.+/)')
-    local expected = '\027]7;file://host' .. parent
-    api.nvim_chan_send(term, string.format('%s\027\\', expected))
-    eq(expected, eval('v:termrequest'))
-    eq(termbuf, eval('g:termbuf'))
-  end)
+      -- cwd will be inserted in a file URI, which cannot contain backs
+      local cwd = t.fix_slashes(fn.getcwd())
+      local parent = cwd:match('^(.+/)')
+      local expected = '\027]7;file://host' .. parent
+      api.nvim_chan_send(term, string.format('%s\027\\', expected))
+      eq(expected, eval('v:termrequest'))
+      eq(termbuf, eval('g:termbuf'))
+    end)
 
-  it('emits TermRequest events for APC', function()
-    local term = api.nvim_open_term(0, {})
+    it('emits events for APC', function()
+      local term = api.nvim_open_term(0, {})
 
-    -- cwd will be inserted in a file URI, which cannot contain backs
-    local cwd = t.fix_slashes(fn.getcwd())
-    local parent = cwd:match('^(.+/)')
-    local expected = '\027_Gfile://host' .. parent
-    api.nvim_chan_send(term, string.format('%s\027\\', expected))
-    eq(expected, eval('v:termrequest'))
-  end)
+      -- cwd will be inserted in a file URI, which cannot contain backs
+      local cwd = t.fix_slashes(fn.getcwd())
+      local parent = cwd:match('^(.+/)')
+      local expected = '\027_Gfile://host' .. parent
+      api.nvim_chan_send(term, string.format('%s\027\\', expected))
+      eq(expected, eval('v:termrequest'))
+    end)
 
-  it('TermRequest synchronization #27572', function()
-    command('autocmd! nvim.terminal TermRequest')
-    local term = exec_lua([[
-      _G.input = {}
-      local term = vim.api.nvim_open_term(0, {
-        on_input = function(_, _, _, data)
-          table.insert(_G.input, data)
-        end,
-        force_crlf = false,
-      })
-      vim.api.nvim_create_autocmd('TermRequest', {
-        callback = function(args)
-          if args.data.sequence == '\027]11;?' then
-            table.insert(_G.input, '\027]11;rgb:0000/0000/0000\027\\')
+    it('synchronization #27572', function()
+      command('autocmd! nvim.terminal TermRequest')
+      local term = exec_lua([[
+        _G.input = {}
+        local term = vim.api.nvim_open_term(0, {
+          on_input = function(_, _, _, data)
+            table.insert(_G.input, data)
+          end,
+          force_crlf = false,
+        })
+        vim.api.nvim_create_autocmd('TermRequest', {
+          callback = function(args)
+            if args.data.sequence == '\027]11;?' then
+              table.insert(_G.input, '\027]11;rgb:0000/0000/0000\027\\')
+            end
           end
-        end
-      })
-      return term
-    ]])
-    api.nvim_chan_send(term, '\027]11;?\007\027[5n\027]11;?\007\027[5n')
-    eq({
-      '\027]11;rgb:0000/0000/0000\027\\',
-      '\027[0n',
-      '\027]11;rgb:0000/0000/0000\027\\',
-      '\027[0n',
-    }, exec_lua('return _G.input'))
-  end)
+        })
+        return term
+      ]])
+      api.nvim_chan_send(term, '\027]11;?\007\027[5n\027]11;?\007\027[5n')
+      eq({
+        '\027]11;rgb:0000/0000/0000\027\\',
+        '\027[0n',
+        '\027]11;rgb:0000/0000/0000\027\\',
+        '\027[0n',
+      }, exec_lua('return _G.input'))
+    end)
 
-  it('TermRequest includes cursor position #31609', function()
-    command('autocmd! nvim.terminal TermRequest')
-    local screen = Screen.new(50, 10)
-    local term = exec_lua([[
-      _G.cursor = {}
-      local term = vim.api.nvim_open_term(0, {})
-      vim.api.nvim_create_autocmd('TermRequest', {
-        callback = function(args)
-          _G.cursor = args.data.cursor
-        end
-      })
-      return term
-    ]])
-    -- Enter terminal mode so that the cursor follows the output
-    feed('a')
+    it('includes cursor position #31609', function()
+      command('autocmd! nvim.terminal TermRequest')
+      local screen = Screen.new(50, 10)
+      local term = exec_lua([[
+        _G.cursor = {}
+        local term = vim.api.nvim_open_term(0, {})
+        vim.api.nvim_create_autocmd('TermRequest', {
+          callback = function(args)
+            _G.cursor = args.data.cursor
+          end
+        })
+        return term
+      ]])
+      -- Enter terminal mode so that the cursor follows the output
+      feed('a')
 
-    -- Put some lines into the scrollback. This tests the conversion from terminal line to buffer
-    -- line.
-    api.nvim_chan_send(term, string.rep('>\n', 20))
-    screen:expect([[
-      >                                                 |*8
-      ^                                                  |
-      {5:-- TERMINAL --}                                    |
-    ]])
+      -- Put some lines into the scrollback. This tests the conversion from terminal line to buffer
+      -- line.
+      api.nvim_chan_send(term, string.rep('>\n', 20))
+      screen:expect([[
+        >                                                 |*8
+        ^                                                  |
+        {5:-- TERMINAL --}                                    |
+      ]])
 
-    -- Emit an OSC escape sequence
-    api.nvim_chan_send(term, 'Hello\nworld!\027]133;D\027\\')
-    screen:expect([[
-      >                                                 |*7
-      Hello                                             |
-      world!^                                            |
-      {5:-- TERMINAL --}                                    |
-    ]])
-    eq({ 22, 6 }, exec_lua('return _G.cursor'))
+      -- Emit an OSC escape sequence
+      api.nvim_chan_send(term, 'Hello\nworld!\027]133;D\027\\')
+      screen:expect([[
+        >                                                 |*7
+        Hello                                             |
+        world!^                                            |
+        {5:-- TERMINAL --}                                    |
+      ]])
+      eq({ 22, 6 }, exec_lua('return _G.cursor'))
+    end)
+
+    it('does not cause hang in vim.wait() #32753', function()
+      local screen = Screen.new(50, 10)
+
+      exec_lua(function()
+        local term = vim.api.nvim_open_term(0, {})
+
+        -- Write OSC sequence with pending scrollback. TermRequest will
+        -- reschedule itself onto an event queue until the pending scrollback is
+        -- processed (i.e. the terminal is refreshed).
+        vim.api.nvim_chan_send(term, string.format('%s\027]133;;\007', string.rep('a\n', 100)))
+
+        -- vim.wait() drains the event queue. The terminal won't be refreshed
+        -- until the event queue is empty. This test ensures that TermRequest
+        -- does not continuously reschedule itself onto the same event queue,
+        -- causing an infinite loop.
+        vim.wait(100)
+      end)
+
+      screen:expect([[
+        ^a                                                 |
+        a                                                 |*8
+                                                          |
+      ]])
+    end)
   end)
 
   it('no heap-buffer-overflow when using jobstart("echo",{term=true}) #3161', function()