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

2 files changed, 19 insertions, 2 deletions

diff --git a/src/nvim/eval.c b/src/nvim/eval.c
index cdf60d9765..6688405860 100644
--- a/src/nvim/eval.c
+++ b/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)
diff --git a/test/functional/core/job_spec.lua b/test/functional/core/job_spec.lua
index 75b50aad0a..48a4689545 100644
--- a/test/functional/core/job_spec.lua
+++ b/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([[