job: Consume content from rbuffer before invoking the callback again

While a job callback is active, it may be invoked again.  Since the
data handled by the first invocation of the callback hasn't been marked
as consumed, the subsequent invocation will see the same data.

Reported-by: Daniel Hahler
Patch-by: oni-link
Closes #5889

src/nvim/eval.c

@@ -23183,11 +23183,10 @@ static void on_job_output(Stream *stream, TerminalJobData *data, RBuffer *buf,
     terminal_receive(data->term, ptr, count);
   }
 
+  rbuffer_consumed(buf, count);
   if (callback->type != kCallbackNone) {
     process_job_event(data, callback, type, ptr, count, 0);
   }
-
-  rbuffer_consumed(buf, count);
 }
 
 static void eval_job_process_exit_cb(Process *proc, int status, void *d)

test/functional/core/job_spec.lua

@@ -312,6 +312,24 @@ describe('jobs', function()
     end)
   end)
 
+  it('does not repeat output with slow output handlers', function()
+    source([[
+      let d = {'data': []}
+      function! d.on_stdout(job, data, event) dict
+        call add(self.data, a:data)
+        sleep 200m
+      endfunction
+      if has('win32')
+        let cmd = '1,2,3,4,5 | foreach-object -process {echo $_; sleep 0.1}'
+      else
+        let cmd = ['sh', '-c', 'for i in $(seq 1 5); do echo $i; sleep 0.1; done']
+      endif
+      call jobwait([jobstart(cmd, d)])
+      call rpcnotify(g:channel, 'data', d.data)
+    ]])
+    eq({'notification', 'data', {{{'1', ''}, {'2', ''}, {'3', ''}, {'4', ''}, {'5', ''}}}}, next_msg())
+  end)
+
   describe('jobwait', function()
     it('returns a list of status codes', function()
       source([[