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.

4 files changed, 141 insertions, 94 deletions

diff --git a/src/nvim/event/loop.c b/src/nvim/event/loop.c
index 15d993cc62..f8b149ec78 100644
--- a/src/nvim/event/loop.c
+++ b/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);
 }
 
diff --git a/src/nvim/event/multiqueue.c b/src/nvim/event/multiqueue.c
index 8646173776..5072bd774d 100644
--- a/src/nvim/event/multiqueue.c
+++ b/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);
diff --git a/src/nvim/terminal.c b/src/nvim/terminal.c
index 8300b536e2..c0f938adda 100644
--- a/src/nvim/terminal.c
+++ b/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)
diff --git a/test/functional/terminal/buffer_spec.lua b/test/functional/terminal/buffer_spec.lua
index b134aa0225..4fd348af41 100644
--- a/test/functional/terminal/buffer_spec.lua
+++ b/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()
-
-    -- 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)
-
-  it('emits TermRequest 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)
-
-  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\\')
+  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')
+
+      -- 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 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)
+
+    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)
-
-  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')
+        })
+        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)
 
-    -- 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 --}                                    |
-    ]])
+    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')
 
-    -- 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'))
+      -- 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'))
+    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()