refactor: turn StdioSink into a proper class

32 files changed, 217 insertions, 151 deletions

diff --git a/lua/mason-core/installer/InstallHandle.lua b/lua/mason-core/installer/InstallHandle.lua
index f5a42c53..22c654a3 100644
--- a/lua/mason-core/installer/InstallHandle.lua
+++ b/lua/mason-core/installer/InstallHandle.lua
@@ -45,7 +45,7 @@ end
 ---@class InstallHandle : EventEmitter
 ---@field package AbstractPackage
 ---@field state InstallHandleState
----@field stdio { buffers: { stdout: string[], stderr: string[] }, sink: StdioSink }
+---@field stdio_sink BufferedSink
 ---@field is_terminated boolean
 ---@field location InstallLocation
 ---@field private spawn_handles InstallHandleSpawnHandle[]
@@ -53,33 +53,17 @@ local InstallHandle = {}
 InstallHandle.__index = InstallHandle
 setmetatable(InstallHandle, { __index = EventEmitter })
 
----@param handle InstallHandle
-local function new_sink(handle)
-    local stdout, stderr = {}, {}
-    return {
-        buffers = { stdout = stdout, stderr = stderr },
-        sink = {
-            stdout = function(chunk)
-                stdout[#stdout + 1] = chunk
-                handle:emit("stdout", chunk)
-            end,
-            stderr = function(chunk)
-                stderr[#stderr + 1] = chunk
-                handle:emit("stderr", chunk)
-            end,
-        },
-    }
-end
-
 ---@param pkg AbstractPackage
 ---@param location InstallLocation
 function InstallHandle:new(pkg, location)
     ---@type InstallHandle
     local instance = EventEmitter.new(self)
+    local sink = process.BufferedSink:new()
+    sink:connect_events(instance)
     instance.state = "IDLE"
     instance.package = pkg
     instance.spawn_handles = {}
-    instance.stdio = new_sink(instance)
+    instance.stdio_sink = sink
     instance.is_terminated = false
     instance.location = location
     return instance
diff --git a/lua/mason-core/installer/InstallRunner.lua b/lua/mason-core/installer/InstallRunner.lua
index fa2b3fcf..342dc443 100644
--- a/lua/mason-core/installer/InstallRunner.lua
+++ b/lua/mason-core/installer/InstallRunner.lua
@@ -55,13 +55,13 @@ function InstallRunner:execute(opts, callback)
     ---@async
     local function finalize_logs(success, result)
         if not success then
-            context.stdio_sink.stderr(tostring(result))
-            context.stdio_sink.stderr "\n"
+            context.stdio_sink:stderr(tostring(result))
+            context.stdio_sink:stderr "\n"
         end
 
         if opts.debug then
             context.fs:write_file("mason-debug.log", table.concat(tailed_output, ""))
-            context.stdio_sink.stdout(("[debug] Installation directory retained at %q.\n"):format(context.cwd:get()))
+            context.stdio_sink:stdout(("[debug] Installation directory retained at %q.\n"):format(context.cwd:get()))
         end
     end
 
diff --git a/lua/mason-core/installer/compiler/schemas.lua b/lua/mason-core/installer/compiler/schemas.lua
index 5e578dbd..889a2ad9 100644
--- a/lua/mason-core/installer/compiler/schemas.lua
+++ b/lua/mason-core/installer/compiler/schemas.lua
@@ -20,7 +20,7 @@ local function download_lsp_schema(ctx, url)
 
         if is_vscode_schema then
             local url = unpack(_.match("^vscode:(.+)$", url))
-            ctx.stdio_sink.stdout(("Downloading LSP configuration schema from %q…\n"):format(url))
+            ctx.stdio_sink:stdout(("Downloading LSP configuration schema from %q…\n"):format(url))
             local json = try(fetch(url))
 
             ---@type { contributes?: { configuration?: table } }
@@ -34,7 +34,7 @@ local function download_lsp_schema(ctx, url)
                 return Result.failure "Unable to find LSP entry in VSCode schema."
             end
         else
-            ctx.stdio_sink.stdout(("Downloading LSP configuration schema from %q…\n"):format(url))
+            ctx.stdio_sink:stdout(("Downloading LSP configuration schema from %q…\n"):format(url))
             try(std.download_file(url, out_file))
             ctx.links.share[share_file] = out_file
         end
diff --git a/lua/mason-core/installer/compiler/util.lua b/lua/mason-core/installer/compiler/util.lua
index b3735c9c..c244cca8 100644
--- a/lua/mason-core/installer/compiler/util.lua
+++ b/lua/mason-core/installer/compiler/util.lua
@@ -44,7 +44,7 @@ function M.ensure_valid_version(versions_thunk)
     local version = ctx.opts.version
 
     if version and not ctx.opts.force then
-        ctx.stdio_sink.stdout "Fetching available versions…\n"
+        ctx.stdio_sink:stdout "Fetching available versions…\n"
         local all_versions = versions_thunk()
         if all_versions:is_failure() then
             log.warn("Failed to fetch versions for package", ctx.package)
@@ -54,10 +54,10 @@ function M.ensure_valid_version(versions_thunk)
         all_versions = all_versions:get_or_else {}
 
         if not _.any(_.equals(version), all_versions) then
-            ctx.stdio_sink.stderr(("Tried to install invalid version %q. Available versions:\n"):format(version))
-            ctx.stdio_sink.stderr(_.compose(_.join "\n", _.map(_.join ", "), _.split_every(15))(all_versions))
-            ctx.stdio_sink.stderr "\n\n"
-            ctx.stdio_sink.stderr(
+            ctx.stdio_sink:stderr(("Tried to install invalid version %q. Available versions:\n"):format(version))
+            ctx.stdio_sink:stderr(_.compose(_.join "\n", _.map(_.join ", "), _.split_every(15))(all_versions))
+            ctx.stdio_sink:stderr "\n\n"
+            ctx.stdio_sink:stderr(
                 ("Run with --force flag to bypass version validation:\n  :MasonInstall --force %s@%s\n\n"):format(
                     ctx.package.name,
                     version
diff --git a/lua/mason-core/installer/context/InstallContextSpawn.lua b/lua/mason-core/installer/context/InstallContextSpawn.lua
index f2ce8df2..29e62101 100644
--- a/lua/mason-core/installer/context/InstallContextSpawn.lua
+++ b/lua/mason-core/installer/context/InstallContextSpawn.lua
@@ -25,7 +25,7 @@ function InstallContextSpawn:__index(cmd)
     ---@param args JobSpawnOpts
     return function(args)
         args.cwd = args.cwd or self.cwd:get()
-        args.stdio_sink = args.stdio_sink or self.handle.stdio.sink
+        args.stdio_sink = args.stdio_sink or self.handle.stdio_sink
         local on_spawn = args.on_spawn
         local captured_handle
         args.on_spawn = function(handle, stdio, pid, ...)
diff --git a/lua/mason-core/installer/context/init.lua b/lua/mason-core/installer/context/init.lua
index 6f0d4c57..44490782 100644
--- a/lua/mason-core/installer/context/init.lua
+++ b/lua/mason-core/installer/context/init.lua
@@ -2,9 +2,9 @@ local InstallContextCwd = require "mason-core.installer.context.InstallContextCw
 local InstallContextFs = require "mason-core.installer.context.InstallContextFs"
 local InstallContextSpawn = require "mason-core.installer.context.InstallContextSpawn"
 local Result = require "mason-core.result"
-local fetch = require "mason-core.fetch"
 local _ = require "mason-core.functional"
 local a = require "mason-core.async"
+local fetch = require "mason-core.fetch"
 local fs = require "mason-core.fs"
 local log = require "mason-core.log"
 local path = require "mason-core.path"
@@ -39,7 +39,7 @@ function InstallContext:new(handle, opts)
         package = handle.package, -- for convenience
         fs = fs,
         receipt = receipt.InstallReceiptBuilder:new(),
-        stdio_sink = handle.stdio.sink,
+        stdio_sink = handle.stdio_sink,
         links = {
             bin = {},
             share = {},
diff --git a/lua/mason-core/installer/managers/cargo.lua b/lua/mason-core/installer/managers/cargo.lua
index 8a3c35cf..22ec9ed6 100644
--- a/lua/mason-core/installer/managers/cargo.lua
+++ b/lua/mason-core/installer/managers/cargo.lua
@@ -15,7 +15,7 @@ function M.install(crate, version, opts)
     opts = opts or {}
     log.fmt_debug("cargo: install %s %s %s", crate, version, opts)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing crate %s@%s…\n"):format(crate, version))
+    ctx.stdio_sink:stdout(("Installing crate %s@%s…\n"):format(crate, version))
     return ctx.spawn.cargo {
         "install",
         "--root",
diff --git a/lua/mason-core/installer/managers/composer.lua b/lua/mason-core/installer/managers/composer.lua
index a4a94270..3afd3ff8 100644
--- a/lua/mason-core/installer/managers/composer.lua
+++ b/lua/mason-core/installer/managers/composer.lua
@@ -14,7 +14,7 @@ local M = {}
 function M.install(package, version)
     log.fmt_debug("composer: install %s %s", package, version)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing composer package %s@%s…\n"):format(package, version))
+    ctx.stdio_sink:stdout(("Installing composer package %s@%s…\n"):format(package, version))
     return Result.try(function(try)
         try(ctx.spawn.composer {
             "init",
diff --git a/lua/mason-core/installer/managers/gem.lua b/lua/mason-core/installer/managers/gem.lua
index e8723d7e..30bff29d 100644
--- a/lua/mason-core/installer/managers/gem.lua
+++ b/lua/mason-core/installer/managers/gem.lua
@@ -15,7 +15,7 @@ function M.install(pkg, version, opts)
     opts = opts or {}
     log.fmt_debug("gem: install %s %s %s", pkg, version, opts)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing gem %s@%s…\n"):format(pkg, version))
+    ctx.stdio_sink:stdout(("Installing gem %s@%s…\n"):format(pkg, version))
     return ctx.spawn.gem {
         "install",
         "--no-user-install",
@@ -50,22 +50,16 @@ function M.create_bin_wrapper(bin)
         return Result.failure(("Cannot link Gem executable %q because it doesn't exist."):format(bin))
     end
 
-    return Result.pcall(
-        ctx.write_shell_exec_wrapper,
-        ctx,
-        bin,
-        path.concat { ctx:get_install_path(), bin_path },
-        {
-            GEM_PATH = platform.when {
-                unix = function()
-                    return ("%s:$GEM_PATH"):format(ctx:get_install_path())
-                end,
-                win = function()
-                    return ("%s;%%GEM_PATH%%"):format(ctx:get_install_path())
-                end,
-            },
-        }
-    )
+    return Result.pcall(ctx.write_shell_exec_wrapper, ctx, bin, path.concat { ctx:get_install_path(), bin_path }, {
+        GEM_PATH = platform.when {
+            unix = function()
+                return ("%s:$GEM_PATH"):format(ctx:get_install_path())
+            end,
+            win = function()
+                return ("%s;%%GEM_PATH%%"):format(ctx:get_install_path())
+            end,
+        },
+    })
 end
 
 return M
diff --git a/lua/mason-core/installer/managers/golang.lua b/lua/mason-core/installer/managers/golang.lua
index 2d7b9b0b..04b24741 100644
--- a/lua/mason-core/installer/managers/golang.lua
+++ b/lua/mason-core/installer/managers/golang.lua
@@ -15,7 +15,7 @@ function M.install(pkg, version, opts)
         opts = opts or {}
         log.fmt_debug("golang: install %s %s %s", pkg, version, opts)
         local ctx = installer.context()
-        ctx.stdio_sink.stdout(("Installing go package %s@%s…\n"):format(pkg, version))
+        ctx.stdio_sink:stdout(("Installing go package %s@%s…\n"):format(pkg, version))
         local env = {
             GOBIN = ctx.cwd:get(),
         }
diff --git a/lua/mason-core/installer/managers/luarocks.lua b/lua/mason-core/installer/managers/luarocks.lua
index 7a2e2b45..f40124cd 100644
--- a/lua/mason-core/installer/managers/luarocks.lua
+++ b/lua/mason-core/installer/managers/luarocks.lua
@@ -15,7 +15,7 @@ function M.install(pkg, version, opts)
     opts = opts or {}
     log.fmt_debug("luarocks: install %s %s %s", pkg, version, opts)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing luarocks package %s@%s…\n"):format(pkg, version))
+    ctx.stdio_sink:stdout(("Installing luarocks package %s@%s…\n"):format(pkg, version))
     ctx:promote_cwd() -- luarocks encodes absolute paths during installation
     return ctx.spawn.luarocks {
         "install",
diff --git a/lua/mason-core/installer/managers/npm.lua b/lua/mason-core/installer/managers/npm.lua
index df8ece35..d31fe768 100644
--- a/lua/mason-core/installer/managers/npm.lua
+++ b/lua/mason-core/installer/managers/npm.lua
@@ -50,7 +50,7 @@ function M.init()
             end
         end))
 
-        ctx.stdio_sink.stdout "Initialized npm root.\n"
+        ctx.stdio_sink:stdout "Initialized npm root.\n"
     end)
 end
 
@@ -62,7 +62,7 @@ function M.install(pkg, version, opts)
     opts = opts or {}
     log.fmt_debug("npm: install %s %s %s", pkg, version, opts)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing npm package %s@%s…\n"):format(pkg, version))
+    ctx.stdio_sink:stdout(("Installing npm package %s@%s…\n"):format(pkg, version))
     return ctx.spawn.npm {
         "install",
         ("%s@%s"):format(pkg, version),
@@ -74,7 +74,7 @@ end
 ---@param pkg string
 function M.uninstall(pkg)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Uninstalling npm package %s…\n"):format(pkg))
+    ctx.stdio_sink:stdout(("Uninstalling npm package %s…\n"):format(pkg))
     return ctx.spawn.npm { "uninstall", pkg }
 end
 
diff --git a/lua/mason-core/installer/managers/nuget.lua b/lua/mason-core/installer/managers/nuget.lua
index 9f1badc7..5a4021d0 100644
--- a/lua/mason-core/installer/managers/nuget.lua
+++ b/lua/mason-core/installer/managers/nuget.lua
@@ -12,7 +12,7 @@ local M = {}
 function M.install(package, version)
     log.fmt_debug("nuget: install %s %s", package, version)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing nuget package %s@%s…\n"):format(package, version))
+    ctx.stdio_sink:stdout(("Installing nuget package %s@%s…\n"):format(package, version))
     return ctx.spawn.dotnet {
         "tool",
         "update",
diff --git a/lua/mason-core/installer/managers/opam.lua b/lua/mason-core/installer/managers/opam.lua
index 875ee12b..20990953 100644
--- a/lua/mason-core/installer/managers/opam.lua
+++ b/lua/mason-core/installer/managers/opam.lua
@@ -14,7 +14,7 @@ local M = {}
 function M.install(package, version)
     log.fmt_debug("opam: install %s %s", package, version)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing opam package %s@%s…\n"):format(package, version))
+    ctx.stdio_sink:stdout(("Installing opam package %s@%s…\n"):format(package, version))
     return ctx.spawn.opam {
         "install",
         "--destdir=.",
diff --git a/lua/mason-core/installer/managers/pypi.lua b/lua/mason-core/installer/managers/pypi.lua
index 85fadc9f..e12b4561 100644
--- a/lua/mason-core/installer/managers/pypi.lua
+++ b/lua/mason-core/installer/managers/pypi.lua
@@ -116,14 +116,14 @@ local function create_venv(pkg)
         and not pep440_check_version(tostring(target.version), supported_python_versions)
     then
         if ctx.opts.force then
-            ctx.stdio_sink.stderr(
+            ctx.stdio_sink:stderr(
                 ("Warning: The resolved python3 version %s is not compatible with the required Python versions: %s.\n"):format(
                     target.version,
                     supported_python_versions
                 )
             )
         else
-            ctx.stdio_sink.stderr "Run with :MasonInstall --force to bypass this version validation.\n"
+            ctx.stdio_sink:stderr "Run with :MasonInstall --force to bypass this version validation.\n"
             return Result.failure(
                 ("Failed to find a python3 installation in PATH that meets the required versions (%s). Found version: %s."):format(
                     supported_python_versions,
@@ -134,7 +134,7 @@ local function create_venv(pkg)
     end
 
     log.fmt_debug("Found python3 installation version=%s, executable=%s", target.version, target.executable)
-    ctx.stdio_sink.stdout "Creating virtual environment…\n"
+    ctx.stdio_sink:stdout "Creating virtual environment…\n"
     return ctx.spawn[target.executable] { "-m", "venv", "--system-site-packages", VENV_DIR }
 end
 
@@ -193,7 +193,7 @@ function M.init(opts)
         try(create_venv(opts.package))
 
         if opts.upgrade_pip then
-            ctx.stdio_sink.stdout "Upgrading pip inside the virtual environment…\n"
+            ctx.stdio_sink:stdout "Upgrading pip inside the virtual environment…\n"
             try(pip_install({ "pip" }, opts.install_extra_args))
         end
     end)
@@ -207,7 +207,7 @@ function M.install(pkg, version, opts)
     opts = opts or {}
     log.fmt_debug("pypi: install %s %s %s", pkg, version, opts or "")
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Installing pip package %s@%s…\n"):format(pkg, version))
+    ctx.stdio_sink:stdout(("Installing pip package %s@%s…\n"):format(pkg, version))
     return pip_install({
         opts.extra and ("%s[%s]==%s"):format(pkg, opts.extra, version) or ("%s==%s"):format(pkg, version),
         opts.extra_packages or vim.NIL,
diff --git a/lua/mason-core/installer/managers/std.lua b/lua/mason-core/installer/managers/std.lua
index b4eb11ab..701bb6c9 100644
--- a/lua/mason-core/installer/managers/std.lua
+++ b/lua/mason-core/installer/managers/std.lua
@@ -95,7 +95,7 @@ end
 function M.download_file(url, out_file)
     log.fmt_debug("std: downloading file %s", url, out_file)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout(("Downloading file %q…\n"):format(url))
+    ctx.stdio_sink:stdout(("Downloading file %q…\n"):format(url))
     return fetch(url, {
         out_file = path.concat { ctx.cwd:get(), out_file },
     }):map_err(function(err)
@@ -234,7 +234,7 @@ local unpack_by_filename = _.cond {
 function M.unpack(rel_path)
     log.fmt_debug("std: unpack %s", rel_path)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout((("Unpacking %q…\n"):format(rel_path)))
+    ctx.stdio_sink:stdout((("Unpacking %q…\n"):format(rel_path)))
     return unpack_by_filename(rel_path)
 end
 
@@ -246,7 +246,7 @@ function M.clone(git_url, opts)
     opts = opts or {}
     log.fmt_debug("std: clone %s %s", git_url, opts)
     local ctx = installer.context()
-    ctx.stdio_sink.stdout((("Cloning git repository %q…\n"):format(git_url)))
+    ctx.stdio_sink:stdout((("Cloning git repository %q…\n"):format(git_url)))
     return Result.try(function(try)
         try(ctx.spawn.git {
             "clone",
diff --git a/lua/mason-core/process.lua b/lua/mason-core/process.lua
index 1690c116..22610ef1 100644
--- a/lua/mason-core/process.lua
+++ b/lua/mason-core/process.lua
@@ -6,9 +6,82 @@ local uv = vim.loop
 ---@alias luv_pipe any
 ---@alias luv_handle any
 
----@class StdioSink
----@field stdout fun(chunk: string)
----@field stderr fun(chunk: string)
+---@class IStdioSink
+local IStdioSink = {}
+---@param chunk string
+function IStdioSink:stdout(chunk) end
+---@param chunk string
+function IStdioSink:stderr(chunk) end
+
+---@class StdioSink : IStdioSink
+---@field stdout_sink? fun(chunk: string)
+---@field stderr_sink? fun(chunk: string)
+local StdioSink = {}
+StdioSink.__index = StdioSink
+
+---@param opts { stdout?: fun(chunk: string), stderr?: fun(chunk: string) }
+function StdioSink:new(opts)
+    ---@type StdioSink
+    local instance = {}
+    setmetatable(instance, self)
+    instance.stdout_sink = opts.stdout
+    instance.stderr_sink = opts.stderr
+    return instance
+end
+
+---@param chunk string
+function StdioSink:stdout(chunk)
+    if self.stdout_sink then
+        self.stdout_sink(chunk)
+    end
+end
+
+---@param chunk string
+function StdioSink:stderr(chunk)
+    if self.stderr_sink then
+        self.stderr_sink(chunk)
+    end
+end
+
+---@class BufferedSink : IStdioSink
+---@field buffers { stdout: string[], stderr: string[] }
+---@field events? EventEmitter
+local BufferedSink = {}
+BufferedSink.__index = BufferedSink
+
+function BufferedSink:new()
+    ---@type BufferedSink
+    local instance = {}
+    setmetatable(instance, self)
+    instance.buffers = {
+        stdout = {},
+        stderr = {},
+    }
+    return instance
+end
+
+---@param events EventEmitter
+function BufferedSink:connect_events(events)
+    self.events = events
+end
+
+---@param chunk string
+function BufferedSink:stdout(chunk)
+    local stdout = self.buffers.stdout
+    stdout[#stdout + 1] = chunk
+    if self.events then
+        self.events:emit("stdout", chunk)
+    end
+end
+
+---@param chunk string
+function BufferedSink:stderr(chunk)
+    local stderr = self.buffers.stderr
+    stderr[#stderr + 1] = chunk
+    if self.events then
+        self.events:emit("stderr", chunk)
+    end
+end
 
 local M = {}
 
@@ -91,7 +164,7 @@ end
 ---@field env string[]? List of "key=value" string.
 ---@field args string[]
 ---@field cwd string
----@field stdio_sink StdioSink
+---@field stdio_sink IStdioSink
 
 ---@param cmd string The command/executable.
 ---@param opts JobSpawnOpts
@@ -151,9 +224,9 @@ function M.spawn(cmd, opts, callback)
     if handle == nil then
         log.fmt_error("Failed to spawn process. cmd=%s, err=%s", cmd, pid_or_err)
         if type(pid_or_err) == "string" and pid_or_err:find "ENOENT" == 1 then
-            opts.stdio_sink.stderr(("Could not find executable %q in path.\n"):format(cmd))
+            opts.stdio_sink:stderr(("Could not find executable %q in path.\n"):format(cmd))
         else
-            opts.stdio_sink.stderr(("Failed to spawn process cmd=%s err=%s\n"):format(cmd, pid_or_err))
+            opts.stdio_sink:stderr(("Failed to spawn process cmd=%s err=%s\n"):format(cmd, pid_or_err))
         end
         callback(false)
         return nil, nil, nil
@@ -161,42 +234,16 @@ function M.spawn(cmd, opts, callback)
 
     log.debug("Spawned with pid", pid_or_err)
 
-    stdout:read_start(connect_sink(stdout, opts.stdio_sink.stdout))
-    stderr:read_start(connect_sink(stderr, opts.stdio_sink.stderr))
+    stdout:read_start(connect_sink(stdout, function(...)
+        opts.stdio_sink:stdout(...)
+    end))
+    stderr:read_start(connect_sink(stderr, function(...)
+        opts.stdio_sink:stderr(...)
+    end))
 
     return handle, stdio, pid_or_err
 end
 
-function M.empty_sink()
-    local function noop() end
-    return {
-        stdout = noop,
-        stderr = noop,
-    }
-end
-
-function M.simple_sink()
-    return {
-        stdout = vim.schedule_wrap(vim.api.nvim_out_write),
-        stderr = vim.schedule_wrap(vim.api.nvim_err_write),
-    }
-end
-
-function M.in_memory_sink()
-    local stdout, stderr = {}, {}
-    return {
-        buffers = { stdout = stdout, stderr = stderr },
-        sink = {
-            stdout = function(chunk)
-                stdout[#stdout + 1] = chunk
-            end,
-            stderr = function(chunk)
-                stderr[#stderr + 1] = chunk
-            end,
-        },
-    }
-end
-
 ---@param luv_handle luv_handle
 ---@param signal integer
 function M.kill(luv_handle, signal)
@@ -211,4 +258,7 @@ function M.kill(luv_handle, signal)
     uv.process_kill(luv_handle, signal)
 end
 
+M.StdioSink = StdioSink
+M.BufferedSink = BufferedSink
+
 return M
diff --git a/lua/mason-core/spawn.lua b/lua/mason-core/spawn.lua
index 33af9ea4..26434bd9 100644
--- a/lua/mason-core/spawn.lua
+++ b/lua/mason-core/spawn.lua
@@ -70,10 +70,8 @@ setmetatable(spawn, {
                 args = cmd_args,
             }
 
-            local stdio
             if not spawn_args.stdio_sink then
-                stdio = process.in_memory_sink()
-                spawn_args.stdio_sink = stdio.sink
+                spawn_args.stdio_sink = process.BufferedSink:new()
             end
 
             local cmd = self._aliases[normalized_cmd] or normalized_cmd
@@ -93,17 +91,30 @@ setmetatable(spawn, {
             end)
 
             if exit_code == 0 and signal == 0 then
-                return Result.success {
-                    stdout = stdio and table.concat(stdio.buffers.stdout, "") or nil,
-                    stderr = stdio and table.concat(stdio.buffers.stderr, "") or nil,
-                }
+                if getmetatable(spawn_args.stdio_sink) == process.BufferedSink then
+                    local sink = spawn_args.stdio_sink --[[@as BufferedSink]]
+                    return Result.success {
+                        stdout = table.concat(sink.buffers.stdout, "") or nil,
+                        stderr = table.concat(sink.buffers.stderr, "") or nil,
+                    }
+                else
+                    return Result.success()
+                end
             else
-                return Failure({
-                    exit_code = exit_code,
-                    signal = signal,
-                    stdout = stdio and table.concat(stdio.buffers.stdout, "") or nil,
-                    stderr = stdio and table.concat(stdio.buffers.stderr, "") or nil,
-                }, cmd)
+                if getmetatable(spawn_args.stdio_sink) == process.BufferedSink then
+                    local sink = spawn_args.stdio_sink --[[@as BufferedSink]]
+                    return Failure({
+                        exit_code = exit_code,
+                        signal = signal,
+                        stdout = table.concat(sink.buffers.stdout, "") or nil,
+                        stderr = table.concat(sink.buffers.stderr, "") or nil,
+                    }, cmd)
+                else
+                    return Failure({
+                        exit_code = exit_code,
+                        signal = signal,
+                    }, cmd)
+                end
             end
         end
     end,
diff --git a/lua/mason-registry/sources/file.lua b/lua/mason-registry/sources/file.lua
index 628b1253..62e7d7a4 100644
--- a/lua/mason-registry/sources/file.lua
+++ b/lua/mason-registry/sources/file.lua
@@ -7,6 +7,7 @@ local async_uv = require "mason-core.async.uv"
 local fs = require "mason-core.fs"
 local log = require "mason-core.log"
 local path = require "mason-core.path"
+local process = require "mason-core.process"
 local spawn = require "mason-core.spawn"
 local util = require "mason-registry.sources.util"
 
@@ -131,7 +132,7 @@ function FileRegistrySource:install()
             [yq]({
                 "-I0", -- output one document per line
                 { "-o", "json" },
-                stdio_sink = {
+                stdio_sink = process.StdioSink:new {
                     stdout = function(chunk)
                         local raw_spec = streaming_parser(chunk)
                         if raw_spec then
diff --git a/tests/mason-core/async/async_spec.lua b/tests/mason-core/async/async_spec.lua
index 79f74462..49a92563 100644
--- a/tests/mason-core/async/async_spec.lua
+++ b/tests/mason-core/async/async_spec.lua
@@ -42,11 +42,11 @@ describe("async", function()
 
     it("should wrap callback-style async functions via promisify", function()
         local async_spawn = _.compose(_.table_pack, a.promisify(process.spawn))
-        local stdio = process.in_memory_sink()
+        local stdio = process.BufferedSink:new()
         local success, exit_code = unpack(a.run_blocking(async_spawn, "env", {
             args = {},
             env = { "FOO=BAR", "BAR=BAZ" },
-            stdio_sink = stdio.sink,
+            stdio_sink = stdio,
         }))
         assert.is_true(success)
         assert.equals(0, exit_code)
diff --git a/tests/mason-core/installer/InstallRunner_spec.lua b/tests/mason-core/installer/InstallRunner_spec.lua
index 696f7b34..0ff7a40f 100644
--- a/tests/mason-core/installer/InstallRunner_spec.lua
+++ b/tests/mason-core/installer/InstallRunner_spec.lua
@@ -212,8 +212,8 @@ describe("InstallRunner ::", function()
         local runner = InstallRunner:new(handle, Semaphore:new(1))
 
         stub(dummy.spec.source, "install", function(ctx)
-            ctx.stdio_sink.stdout "Hello "
-            ctx.stdio_sink.stderr "world!"
+            ctx.stdio_sink:stdout "Hello "
+            ctx.stdio_sink:stderr "world!"
         end)
 
         local callback = test_helpers.sync_runner_execute(runner, { debug = true })
@@ -229,7 +229,7 @@ describe("InstallRunner ::", function()
         local runner = InstallRunner:new(handle, Semaphore:new(1))
 
         stub(dummy.spec.source, "install", function(ctx)
-            ctx.stdio_sink.stderr "Something will go terribly wrong.\n"
+            ctx.stdio_sink:stderr "Something will go terribly wrong.\n"
             error("This went terribly wrong.", 0)
         end)
 
@@ -246,7 +246,7 @@ describe("InstallRunner ::", function()
         local runner = InstallRunner:new(handle, Semaphore:new(1))
 
         stub(dummy.spec.source, "install", function(ctx)
-            ctx.stdio_sink.stderr "Something will go terribly wrong.\n"
+            ctx.stdio_sink:stderr "Something will go terribly wrong.\n"
             error("This went terribly wrong.", 0)
         end)
 
diff --git a/tests/mason-core/installer/managers/cargo_spec.lua b/tests/mason-core/installer/managers/cargo_spec.lua
index bc5c5f21..66f89ca2 100644
--- a/tests/mason-core/installer/managers/cargo_spec.lua
+++ b/tests/mason-core/installer/managers/cargo_spec.lua
@@ -1,4 +1,5 @@
 local cargo = require "mason-core.installer.managers.cargo"
+local match = require "luassert.match"
 local spy = require "luassert.spy"
 local test_helpers = require "mason-test.helpers"
 
@@ -30,7 +31,9 @@ describe("cargo manager", function()
             cargo.install("my-crate", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing crate my-crate@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing crate my-crate@1.0.0…\n")
     end)
 
     it("should install locked", function()
diff --git a/tests/mason-core/installer/managers/composer_spec.lua b/tests/mason-core/installer/managers/composer_spec.lua
index f3887c68..8559e353 100644
--- a/tests/mason-core/installer/managers/composer_spec.lua
+++ b/tests/mason-core/installer/managers/composer_spec.lua
@@ -1,4 +1,5 @@
 local composer = require "mason-core.installer.managers.composer"
+local match = require "luassert.match"
 local spy = require "luassert.spy"
 local test_helpers = require "mason-test.helpers"
 
@@ -29,6 +30,8 @@ describe("composer manager", function()
             composer.install("my-package", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing composer package my-package@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing composer package my-package@1.0.0…\n")
     end)
 end)
diff --git a/tests/mason-core/installer/managers/gem_spec.lua b/tests/mason-core/installer/managers/gem_spec.lua
index 3a72521a..0345d799 100644
--- a/tests/mason-core/installer/managers/gem_spec.lua
+++ b/tests/mason-core/installer/managers/gem_spec.lua
@@ -1,4 +1,5 @@
 local gem = require "mason-core.installer.managers.gem"
+local match = require "luassert.match"
 local spy = require "luassert.spy"
 local test_helper = require "mason-test.helpers"
 
@@ -34,7 +35,9 @@ describe("gem manager", function()
             gem.install("my-gem", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing gem my-gem@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing gem my-gem@1.0.0…\n")
     end)
 
     it("should install extra packages", function()
diff --git a/tests/mason-core/installer/managers/golang_spec.lua b/tests/mason-core/installer/managers/golang_spec.lua
index e1a99cbd..146fea02 100644
--- a/tests/mason-core/installer/managers/golang_spec.lua
+++ b/tests/mason-core/installer/managers/golang_spec.lua
@@ -1,4 +1,5 @@
 local golang = require "mason-core.installer.managers.golang"
+local match = require "luassert.match"
 local spy = require "luassert.spy"
 local test_helpers = require "mason-test.helpers"
 
@@ -29,7 +30,9 @@ describe("golang manager", function()
             golang.install("my-golang", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing go package my-golang@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing go package my-golang@1.0.0…\n")
     end)
 
     it("should install extra packages", function()
diff --git a/tests/mason-core/installer/managers/luarocks_spec.lua b/tests/mason-core/installer/managers/luarocks_spec.lua
index 406c5c51..31dd3dc0 100644
--- a/tests/mason-core/installer/managers/luarocks_spec.lua
+++ b/tests/mason-core/installer/managers/luarocks_spec.lua
@@ -1,4 +1,5 @@
 local luarocks = require "mason-core.installer.managers.luarocks"
+local match = require "luassert.match"
 local spy = require "luassert.spy"
 local stub = require "luassert.stub"
 local test_helpers = require "mason-test.helpers"
@@ -80,6 +81,8 @@ describe("luarocks manager", function()
             luarocks.install("my-rock", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing luarocks package my-rock@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing luarocks package my-rock@1.0.0…\n")
     end)
 end)
diff --git a/tests/mason-core/installer/managers/npm_spec.lua b/tests/mason-core/installer/managers/npm_spec.lua
index b2fabc80..e3a2bc76 100644
--- a/tests/mason-core/installer/managers/npm_spec.lua
+++ b/tests/mason-core/installer/managers/npm_spec.lua
@@ -98,6 +98,8 @@ describe("npm manager", function()
             npm.install("my-package", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing npm package my-package@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing npm package my-package@1.0.0…\n")
     end)
 end)
diff --git a/tests/mason-core/installer/managers/nuget_spec.lua b/tests/mason-core/installer/managers/nuget_spec.lua
index fdfbdc82..2f16a652 100644
--- a/tests/mason-core/installer/managers/nuget_spec.lua
+++ b/tests/mason-core/installer/managers/nuget_spec.lua
@@ -1,3 +1,4 @@
+local match = require "luassert.match"
 local nuget = require "mason-core.installer.managers.nuget"
 local spy = require "luassert.spy"
 local test_helpers = require "mason-test.helpers"
@@ -28,6 +29,8 @@ describe("nuget manager", function()
             nuget.install("nuget-package", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing nuget package nuget-package@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing nuget package nuget-package@1.0.0…\n")
     end)
 end)
diff --git a/tests/mason-core/installer/managers/opam_spec.lua b/tests/mason-core/installer/managers/opam_spec.lua
index 51f116e8..9ff53b98 100644
--- a/tests/mason-core/installer/managers/opam_spec.lua
+++ b/tests/mason-core/installer/managers/opam_spec.lua
@@ -1,3 +1,4 @@
+local match = require "luassert.match"
 local opam = require "mason-core.installer.managers.opam"
 local spy = require "luassert.spy"
 local test_helpers = require "mason-test.helpers"
@@ -28,6 +29,8 @@ describe("opam manager", function()
             opam.install("opam-package", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing opam package opam-package@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing opam package opam-package@1.0.0…\n")
     end)
 end)
diff --git a/tests/mason-core/installer/managers/pypi_spec.lua b/tests/mason-core/installer/managers/pypi_spec.lua
index ea3da250..7ae28563 100644
--- a/tests/mason-core/installer/managers/pypi_spec.lua
+++ b/tests/mason-core/installer/managers/pypi_spec.lua
@@ -143,7 +143,7 @@ describe("pypi manager", function()
         )
         assert
             .spy(ctx.stdio_sink.stderr)
-            .was_called_with "Run with :MasonInstall --force to bypass this version validation.\n"
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Run with :MasonInstall --force to bypass this version validation.\n")
     end)
 
     it(
@@ -179,9 +179,10 @@ describe("pypi manager", function()
                 "--system-site-packages",
                 "venv",
             }
-            assert
-                .spy(ctx.stdio_sink.stderr)
-                .was_called_with "Warning: The resolved python3 version 3.5.0 is not compatible with the required Python versions: >=3.8.\n"
+            assert.spy(ctx.stdio_sink.stderr).was_called_with(
+                match.is_ref(ctx.stdio_sink),
+                "Warning: The resolved python3 version 3.5.0 is not compatible with the required Python versions: >=3.8.\n"
+            )
         end
     )
 
@@ -249,7 +250,9 @@ describe("pypi manager", function()
             pypi.install("pypi-package", "1.0.0")
         end)
 
-        assert.spy(ctx.stdio_sink.stdout).was_called_with "Installing pip package pypi-package@1.0.0…\n"
+        assert
+            .spy(ctx.stdio_sink.stdout)
+            .was_called_with(match.is_ref(ctx.stdio_sink), "Installing pip package pypi-package@1.0.0…\n")
     end)
 
     it("should install extra specifier", function()
diff --git a/tests/mason-core/process_spec.lua b/tests/mason-core/process_spec.lua
index 06330cdd..480b047b 100644
--- a/tests/mason-core/process_spec.lua
+++ b/tests/mason-core/process_spec.lua
@@ -5,7 +5,7 @@ local spy = require "luassert.spy"
 describe("process.spawn", function()
     -- Unix only
     it("should spawn command and feed output to sink", function()
-        local stdio = process.in_memory_sink()
+        local stdio = process.BufferedSink:new()
         local callback = spy.new()
         process.spawn("env", {
             args = {},
@@ -13,7 +13,7 @@ describe("process.spawn", function()
                 "HELLO=world",
                 "MY_ENV=var",
             },
-            stdio_sink = stdio.sink,
+            stdio_sink = stdio,
         }, callback)
 
         assert.wait(function()
diff --git a/tests/mason-core/spawn_spec.lua b/tests/mason-core/spawn_spec.lua
index a1432294..9fc91200 100644
--- a/tests/mason-core/spawn_spec.lua
+++ b/tests/mason-core/spawn_spec.lua
@@ -26,16 +26,19 @@ describe("async spawn", function()
     end)
 
     it("should use provided stdio_sink", function()
-        local stdio = process.in_memory_sink()
+        local stdout = spy.new()
+        local stdio = process.StdioSink:new {
+            stdout = stdout,
+        }
         local result = a.run_blocking(spawn.env, {
             env_raw = { "FOO=bar" },
-            stdio_sink = stdio.sink,
+            stdio_sink = stdio,
         })
         assert.is_true(result:is_success())
-        assert.equals(nil, result:get_or_nil().stdout)
-        assert.equals(nil, result:get_or_nil().stderr)
-        assert.equals("FOO=bar\n", table.concat(stdio.buffers.stdout, ""))
-        assert.equals("", table.concat(stdio.buffers.stderr, ""))
+        assert.equals(nil, result:get_or_nil())
+        -- Not 100 %guaranteed it's only called once because output is always buffered, but it's extremely likely
+        assert.spy(stdout).was_called(1)
+        assert.spy(stdout).was_called_with "FOO=bar\n"
     end)
 
     it("should pass command arguments", function()
@@ -68,10 +71,7 @@ describe("async spawn", function()
         assert.spy(process.spawn).was_called_with(
             "bash",
             match.tbl_containing {
-                stdio_sink = match.tbl_containing {
-                    stdout = match.is_function(),
-                    stderr = match.is_function(),
-                },
+                stdio_sink = match.instanceof(process.BufferedSink),
                 env = match.list_containing "VAR=world",
                 args = match.tbl_containing {
                     "-c",
@@ -134,7 +134,7 @@ describe("async spawn", function()
 
     it("should format failure message", function()
         stub(process, "spawn", function(cmd, opts, callback)
-            opts.stdio_sink.stderr(("This is an error message for %s!"):format(cmd))
+            opts.stdio_sink:stderr(("This is an error message for %s!"):format(cmd))
             callback(false, 127)
         end)