feat(worker): deterministic async task worker module

Author: asmodeus812

README.md

@@ -93,6 +93,7 @@ with ease, as the picker will only ever render and decorate a small subset of th
     - [Using vim-plug](#using-vim-plug)
 - [Configuration](#configuration)
 - [Tracing](#tracing)
+- [Workers](#workers)
 - [Pool](#pool)
     - [Lifecycle](#lifecycle)
     - [Normalization](#normalization)
@@ -245,6 +246,22 @@ Supported modules:
 - `async.trace`: Async creation and completion events.
 - `registry.trace`: Registry registration, touch, removal, and prune events.
 
+## Workers
+
+Workers are internal coordination helpers that sit on top of the async scheduler. They are not user-configurable, but they define how
+different subsystems sequence work so ordering is deterministic even under heavy async load.
+
+There are two worker modes used in the codebase:
+
+- **Coalescer**: Keeps only the latest request while work is in flight. This is used for UI rendering in `Select`, where multiple list updates
+  can arrive rapidly. The coalescer guarantees that only the most recent render request runs, so stale intermediate renders are dropped.
+
+- **Queue**: Runs every request in strict FIFO order. This is used for stream processing, where chunk order is semantically important. The
+  queue executes tasks one after the other, and each task can yield without allowing another queued task to start early.
+
+Workers are cooperative: each task runs inside an `Async` coroutine and may call `Async.yield()`. Yielding allows other unrelated async tasks
+to run, but does not violate the ordering guarantees within a worker.
+
 ## Pool
 
 The pool is a table-reuse subsystem that keeps allocation pressure low for hot paths (streaming, matching, selection). It is a single global

lua/fuzzy/async.lua

@@ -113,7 +113,7 @@ function Async:abort()
     self:_done(nil, "abort")
 end
 
---- Registers a callback for when the async completes, or calls it immediately if already done (unless aborted).
+--- Registers a callback for when the async completes, or calls it immediately if already done (unless aborted). The callbacks are executed in order of registration
 --- @param callback function Function to call on completion
 function Async:await(callback)
     assert(type(callback) == "function")

lua/fuzzy/select.lua

@@ -12,7 +12,7 @@ local spacing = { ",", "SelectHeaderDelimiter" }
 
 local utils = require("fuzzy.utils")
 local Async = require("fuzzy.async")
-local Scheduler = require("fuzzy.scheduler")
+local Worker = require("fuzzy.worker")
 
 --- @class Select
 --- @field private source_window integer|nil The window ID of the source window where the selection interface was opened from.
@@ -1688,7 +1688,7 @@ end
 
 --- Internal: schedules a UI render of list+related features.
 function Select:_render_list()
-    local executor = Async.wrap(function()
+    local render = function()
         local entries = self._state.entries or {}
         cursor_clamp(self._state.cursor, #entries)
 
@@ -1710,19 +1710,8 @@ function Select:_render_list()
                 flush = true,
             })
         end
-    end)
-
-    local renderer = function()
-        self._state.renderer = executor()
-        Scheduler.add(self._state.renderer)
-    end
-
-    if self:_is_rendering() then
-        self._state.renderer:await(renderer)
-        self:_stop_rendering()
-    else
-        renderer()
     end
+    self._state.renderer.enqueue(render)
 end
 
 --- Resets the state variables for a Select instance.
@@ -1859,9 +1848,8 @@ function Select:_clear_view(force)
         self:_reset_state()
     end
 
-    if self:_is_rendering() then
-        self._state.renderer:await(clearer)
-        self:_stop_rendering()
+    if self._state.renderer then
+        self._state.renderer.cancel(clearer)
         self._state.renderer = nil
     else
         clearer()
@@ -1897,9 +1885,8 @@ function Select:_close_view(force)
         end
     end
 
-    if self:_is_rendering() then
-        self._state.renderer:await(closer)
-        self:_stop_rendering()
+    if self._state.renderer then
+        self._state.renderer.cancel(closer)
         self._state.renderer = nil
     else
         closer()
@@ -1910,15 +1897,15 @@ end
 --- @return boolean
 function Select:_is_rendering()
     if self._state.renderer then
-        return self._state.renderer:is_running()
+        return self._state.renderer.is_running()
     end
     return false
 end
 
 --- Cancels and clears the running renderer coroutine.
 function Select:_stop_rendering()
     if self._state.renderer then
-        self._state.renderer:cancel()
+        self._state.renderer.cancel()
         self._state.renderer = nil
     end
 end
@@ -2585,6 +2572,9 @@ function Select:open()
         self:_init_preview()
     end
 
+    if not self._state.renderer then
+        self._state.renderer = Worker.coalesce()
+    end
     self.source_window = vim.api.nvim_get_current_win()
     local factor = (opts.prompt_list and opts.preview) and 2.0 or 1.0
     local size = compute_height(opts.window_ratio, factor)

lua/fuzzy/stream.lua

@@ -1,6 +1,6 @@
-local Scheduler = require("fuzzy.scheduler")
 local Async = require("fuzzy.async")
 local Pool = require("fuzzy.pool")
+local Worker = require("fuzzy.worker")
 local utils = require("fuzzy.utils")
 
 --- @class Stream
@@ -50,6 +50,7 @@ function Stream:_close_stream()
         close_handle(self._state.handle)
         self._state.handle = nil
     end
+    self:_stop_streaming()
 
     if self._state.buffer then
         -- clean up the string references to avoid any memory leaks, the buffer could still be reused in future calls to start
@@ -83,8 +84,6 @@ function Stream:_close_stream()
 
     self.callback = nil
     self.transform = nil
-
-    self:_stop_streaming()
 end
 
 function Stream:_make_stream()
@@ -166,6 +165,7 @@ function Stream:_handle_data(data, size)
         self._state.buffer[self._state.size + 1] = self:_transform_data(data)
         self._state.size = self._state.size + 1
     end
+    return self._state.size
 end
 
 function Stream:_handle_out(code, chunk, kind)
@@ -264,8 +264,11 @@ function Stream:_handle_in(err, chunk, kind)
                 end
 
                 local line = chunk:sub(start, pos - 1)
-                self:_handle_data(line, size)
+                size = self:_handle_data(line, size)
 
+                if size % 512 == 0 then
+                    Async.yield()
+                end
                 start = next + 1
                 count = count + 1
             end
@@ -285,48 +288,21 @@ function Stream:_handle_in(err, chunk, kind)
 end
 
 function Stream:_handle_stdout(err, chunk)
-    local executor = Async.wrap(Stream._handle_in)
-
-    local streamer = function()
-        self._state.streamer = executor(self, err, chunk, 1)
-        Scheduler.add(self._state.streamer)
-    end
-
-    if self:_is_streaming() then
-        self._state.streamer:await(streamer)
-    else
-        streamer()
-    end
+    self._state.streamer.enqueue(function()
+        self:_handle_in(err, chunk, 1)
+    end)
 end
 
 function Stream:_handle_stderr(err, chunk)
-    local executor = Async.wrap(Stream._handle_in)
-
-    local streamer = function()
-        self._state.streamer = executor(self, err, chunk, 2)
-        Scheduler.add(self._state.streamer)
-    end
-
-    if self:_is_streaming() then
-        self._state.streamer:await(streamer)
-    else
-        streamer()
-    end
+    self._state.streamer.enqueue(function()
+        self:_handle_in(err, chunk, 2)
+    end)
 end
 
 function Stream:_handle_exit(e, c)
-    local executor = Async.wrap(Stream._handle_out)
-
-    local streamer = function()
-        self._state.streamer = executor(self, e, c, 3)
-        Scheduler.add(self._state.streamer)
-    end
-
-    if self:_is_streaming() then
-        self._state.streamer:await(streamer)
-    else
-        streamer()
-    end
+    self._state.streamer.enqueue(function()
+        self:_handle_out(e, c, 3)
+    end)
 end
 
 function Stream:_is_streaming()
@@ -335,7 +311,7 @@ end
 
 function Stream:_stop_streaming()
     if self._state.streamer then
-        self._state.streamer:abort()
+        self._state.streamer.cancel()
         self._state.streamer = nil
     end
 end
@@ -443,6 +419,10 @@ function Stream:start(cmd, opts)
         self._state.accum = Pool.obtain(size)
     end
 
+    if not self._state.streamer then
+        self._state.streamer = Worker.queue()
+    end
+
     if type(cmd) == "function" then
         local finalized = false
         local worker = self:_bind_guarded(function(stream, data)
@@ -459,13 +439,13 @@ function Stream:start(cmd, opts)
                 finalized = true
             end
         end, token)
-        local executor = Async.wrap(function(stream)
+        self._state.streamer.enqueue(function()
             local ok, err = utils.safe_call(
                 cmd, worker, opts.args, opts.cwd, opts.env
             )
             local code = not ok and 1 or 0
             if finalized == false then
-                local finalizer = stream:_bind_guarded(function(s, c, e)
+                local finalizer = self:_bind_guarded(function(s, c, e)
                     s._state.stdouteof = true
                     s._state.stderreof = true
                     s:_handle_out(c, e, 3)
@@ -474,8 +454,6 @@ function Stream:start(cmd, opts)
                 finalized = true
             end
         end)
-        self._state.streamer = executor(self)
-        Scheduler.add(self._state.streamer)
     else
         local stdio = self:_make_stream()
         assert(vim.fn.executable(cmd) == 1)