fix(strategies): avoid concurrent writes

See #488

Author: rcarriga

lua/neotest/client/strategies/integrated/init.lua

@@ -29,7 +29,7 @@ return function(spec)
   assert(not open_err, open_err)
 
   output_accum:subscribe(function(data)
-    vim.loop.fs_write(output_fd, data, nil, function(write_err)
+    vim.uv.fs_write(output_fd, data, nil, function(write_err)
       assert(not write_err, write_err)
     end)
   end)
@@ -45,7 +45,7 @@ return function(spec)
         output_finish_future.set()
         return
       end
-      output_accum:push(table.concat(data, "\n"))
+      nio.run(function() output_accum:push(table.concat(data, "\n")) end)
     end,
     on_exit = function(_, code)
       result_code = code

lua/neotest/types/fanout_accum.lua

@@ -1,3 +1,4 @@
+local nio = require("nio")
 local logger = require("neotest.logging")
 local neotest = {}
 
@@ -7,6 +8,7 @@ local neotest = {}
 ---@field consumers fun(data: T)[]
 ---@field data? T
 ---@field accum fun(prev: T, new: any): T A function to combine previous data and new data
+---@field semaphore nio.control.Semaphore
 neotest.FanoutAccum = {}
 
 ---@generic T
@@ -19,6 +21,7 @@ function neotest.FanoutAccum:new(accum, init)
     data = init,
     accum = accum,
     consumers = {},
+    semaphore = nio.control.semaphore(1)
   }, self)
 end
 
@@ -40,12 +43,15 @@ function neotest.FanoutAccum:subscribe(cb)
   end
 end
 
+---@async
 ---@param data T
 function neotest.FanoutAccum:push(data)
-  self.data = self.accum(self.data, data)
-  for _, cb in ipairs(self.consumers) do
-    cb(data)
-  end
+  self.semaphore.with(function()
+    self.data = self.accum(self.data, data)
+    for _, cb in ipairs(self.consumers) do
+      cb(data)
+    end
+  end)
 end
 
 ---@private