fix(rpc): do not crash when no input is consumed

fixes #23781

Co-authored-by: glacambre <code@lacamb.re>

src/nvim/rbuffer.c

@@ -123,7 +123,10 @@ char *rbuffer_read_ptr(RBuffer *buf, size_t *read_count) FUNC_ATTR_NONNULL_ALL
 void rbuffer_consumed(RBuffer *buf, size_t count)
   FUNC_ATTR_NONNULL_ALL
 {
-  assert(count && count <= buf->size);
+  if (count == 0) {
+    return;
+  }
+  assert(count <= buf->size);
 
   buf->read_ptr += count;
   if (buf->read_ptr >= buf->end_ptr) {

test/functional/api/server_requests_spec.lua

@@ -363,6 +363,24 @@ describe('server -> client', function()
       server:close()
       client:close()
     end)
+
+    it('via stdio, with many small flushes does not crash #23781', function()
+      source([[
+      let chan = jobstart([v:progpath, '--embed', '--headless', '-n', '-u', 'NONE', '-i', 'NONE'], { 'rpc':v:false })
+      call chansend(chan, 0Z94)
+      sleep 50m
+      call chansend(chan, 0Z00)
+      call chansend(chan, 0Z01)
+      call chansend(chan, 0ZAC)
+      call chansend(chan, 0Z6E76696D5F636F6D6D616E64)
+      call chansend(chan, 0Z91)
+      call chansend(chan, 0ZA5)
+      call chansend(chan, 0Z71616C6C21)
+      let g:statuses = jobwait([chan])
+      ]])
+      eq(eval('g:statuses'), { 0 })
+      assert_alive()
+    end)
   end)
 
   describe('connecting to its own pipe address', function()