fix(rpc): assertion failure due to invalid msgpack input

Problem:
  rbuffer_consumed assertion fails if Unpacker fails to parse msgpack,
  because it doesn't consume bytes on errors

Solution:
  Call rbuffer_consumed_compact only if Unpacker isn't closed

2 files changed, 43 insertions, 3 deletions

diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 286bb8098a..0178ef622b 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -304,8 +304,11 @@ static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
   p->read_ptr = rbuffer_read_ptr(rbuf, &size);
   p->read_size = size;
   parse_msgpack(channel);
-  size_t consumed = size - p->read_size;
-  rbuffer_consumed_compact(rbuf, consumed);
+
+  if (!unpacker_closed(p)) {
+    size_t consumed = size - p->read_size;
+    rbuffer_consumed_compact(rbuf, consumed);
+  }
 
 end:
   channel_decref(channel);
diff --git a/test/functional/core/channels_spec.lua b/test/functional/core/channels_spec.lua
index ce13c4755d..856c34a60b 100644
--- a/test/functional/core/channels_spec.lua
+++ b/test/functional/core/channels_spec.lua
@@ -4,7 +4,7 @@ local clear, eq, eval, next_msg, ok, source =
 local command, fn, api = helpers.command, helpers.fn, helpers.api
 local sleep = vim.uv.sleep
 local spawn, nvim_argv = helpers.spawn, helpers.nvim_argv
-local set_session = helpers.set_session
+local get_session, set_session = helpers.get_session, helpers.set_session
 local nvim_prog = helpers.nvim_prog
 local is_os = helpers.is_os
 local retry = helpers.retry
@@ -59,6 +59,43 @@ describe('channels', function()
     eq({ 'notification', 'data', { id, { '' } } }, next_msg())
   end)
 
+  it('dont crash due to garbage in rpc #23781', function()
+    local client = get_session()
+    local server = spawn(nvim_argv, nil, nil, true)
+    set_session(server)
+    local address = funcs.serverlist()[1]
+    set_session(client)
+
+    meths.set_var('address', address)
+    command("let g:id = sockconnect('pipe', address, {'on_data':'OnEvent'})")
+    local id = eval('g:id')
+    ok(id > 0)
+
+    command("call chansend(g:id, 'F')")
+    eq({'notification', 'data', {id, {''}}}, next_msg())
+    set_session(server)
+    assert_alive()
+
+    set_session(client)
+    command('call chanclose(g:id)')
+    command("let g:id = sockconnect('pipe', address, {'on_data':'OnEvent'})")
+    id = eval('g:id')
+    ok(id > 0)
+
+    command("call chansend(g:id, msgpackdump([[2, 'redraw', 'F']], 'B')[:-4])")
+    set_session(server)
+    assert_alive()
+    set_session(client)
+    command("call chansend(g:id, 'F')")
+    eq({'notification', 'data', {id, {''}}}, next_msg())
+
+    set_session(server)
+    assert_alive()
+    set_session(client)
+    command('call chanclose(g:id)')
+    server:close()
+  end)
+
   it('can use stdio channel', function()
     source([[
       let g:job_opts = {