[DRAFT] Async MWorker: convert AESFuncs/ClearFuncs handlers to async dispatch - #70129

Convert ``_mine_get``, ``_mine``, ``_mine_delete``, ``_mine_flush`` to
``async def`` and register them in ``AESFuncs.async_methods``. Each
handler offloads its ``self.masterapi._mine*(...)`` call to
``run_in_executor`` so the event loop stays free while masterapi does
its (mostly in-memory) work; ``functools.partial`` is used to preserve
the ``skip_verify`` kwarg where present.

Adds unit tests covering async dispatch through ``_handle_aes``,
return-shape parity with the sync version, and the ``verify_load``
short-circuit path.

Convert ``minion_runner``, ``minion_pub``, ``minion_publish``,
``revoke_auth`` to ``async def`` and register them in
``AESFuncs.async_methods``. Each offloads its blocking
``self.masterapi.<op>(clear_load)`` call to ``run_in_executor`` so the
event loop stays free while ``masterapi``/``LocalFuncs`` does its work.

Adds unit tests covering async dispatch through ``_handle_aes`` and
return-shape parity with the sync version.

Convert ``_pillar`` to ``async def`` and register it in
``AESFuncs.async_methods``. Adopt the async pillar API:
``salt.pillar.get_async_pillar()`` + ``await pillar.compile_pillar()``
so the pillar render pipeline runs on the event loop instead of
blocking a worker.

Swap the sync ``self.event.fire_event(...)`` for
``await self.event.fire_event_async(...)``. Offload the remaining
sync-only calls (``self.fs_.update_opts``, ``self.masterapi.cache.store``)
to ``run_in_executor``.

Updates the two existing sync ``_pillar`` tests to ``async``/``await``
(required because the handler is now a coroutine) and adds coverage
for dispatch through ``_handle_aes``, return-shape parity, and the
event-fire path.

Convert the ten fileserver handlers (``_serve_file``, ``_file_find``,
``_file_hash``, ``_file_hash_and_stat``, ``_file_list``,
``_file_list_emptydirs``, ``_dir_list``, ``_symlink_list``,
``_file_envs``, ``_file_recv``) to ``async def`` and register them in
``AESFuncs.async_methods``. Each offloads its blocking
``self.fs_.<op>(load)`` call to ``run_in_executor`` — the fileserver
API is disk-heavy and has no async surface.

The nine read/list handlers were previously bound as instance
attributes in ``__setup_fileserver`` (``self._serve_file = self.fs_.serve_file``);
those aliases are dropped in favor of real ``async def`` methods on
the class so the async registry can find them and Python's
``iscoroutinefunction`` works. ``_file_recv`` is split into an
``async def`` façade + a ``_file_recv_write`` static helper so cheap
validation stays on the loop while the ``makedirs``/``open``/``write``
portion runs in the executor.

``destroy()`` no longer clears the fileserver method aliases (they
are class methods now, not instance attributes). ``_master_opts``
calls ``self.fs_.file_envs()`` directly instead of the (now-async)
``self._file_envs`` — a subsequent commit converts ``_master_opts``
itself to async.

Adds unit tests covering async dispatch, return-shape parity, and
executor usage for the representative handlers.

Convert ``_return``, ``_syndic_return``, ``pub_ret`` to ``async def``
and register them in ``AESFuncs.async_methods``. Each offloads its
blocking work (RSA signature verify, ``salt.utils.job.store_job``,
returner-plugin cache writes, ``local.get_cache_returns``, disk
``mkdir``/``open``/``write`` for the syndic marker file) to
``run_in_executor`` — none of these subsystems expose an async
surface.

``_syndic_return`` recursively awaits ``_return`` per minion (both
are coroutines now). A new ``_write_syndic_cache_marker`` static
helper wraps the ``makedirs``/``open``/``write`` sequence into a
single executor hop.

Updates the four existing sync tests that exercise these handlers to
``async``/``await`` (required because the handlers are now coroutines)
and adds coverage for async dispatch through ``_handle_aes``,
return-shape parity, and executor offload of the sync internals.

Convert ``_register_resources``, ``verify_minion``, ``_master_tops``,
``_master_opts`` to ``async def`` and register them in
``AESFuncs.async_methods``. Each offloads its sync internals:

- ``verify_minion`` moves RSA ``PublicKey.decrypt`` + cache fetch
  into ``run_in_executor`` via a private ``__verify_minion`` helper.
- ``_master_tops`` offloads ``masterapi._master_tops(load, skip_verify=True)``
  via ``functools.partial`` to preserve the kwarg.
- ``_master_opts`` awaits ``self._file_envs(load)`` (async since Phase 2D)
  and keeps the remaining dict work on the loop thread.
- ``_register_resources`` delegates the full blocking body (mmap index
  update, cache list/flush/store) to a new private
  ``__register_resources_sync`` helper via the executor and awaits
  ``event.fire_event_async`` for the cache-refresh event.

Adds unit tests covering async dispatch through ``_handle_aes``,
return-shape parity, executor usage, and the async event-fire path;
converts the pre-existing sync ``_register_resources`` tests to
``async``/``await``.

NOTE: ``salt/channel/server.py`` still calls ``verify_minion``
synchronously (``MasterPubServerChannel.presence_callback``); a
follow-up commit awaits the coroutine to prevent silently passing
unauthorized minions through presence auth.

``MasterPubServerChannel.presence_callback`` calls
``self.aes_funcs.verify_minion(...)`` to gate presence subscriptions.
That method is now ``async def`` (Phase 2F of the MWorker async
migration); the sync call returned an unawaited coroutine — always
truthy in the ``if not ...:`` check — silently letting any subscriber
through the auth gate.

Convert ``presence_callback`` to ``async def`` and await
``verify_minion``. The TCP transport's ``_stream_read`` (already
``async def``) now awaits the callback when it returns a coroutine;
sync callbacks (including the noop lambda default) still work as
before because ``asyncio.iscoroutine`` gates the await.

Move ``ping``, ``wheel``, ``runner``, ``get_token``, and ``mk_token`` to
``async def`` and add each to ``ClearFuncs.async_methods`` so
``MWorker._handle_clear`` dispatches them as coroutines and awaits the
result. The wrappers offload their existing synchronous bodies to the
running event loop's default executor:

* ``ping`` — trivial echo, awaited directly.
* ``wheel`` — ``Wheel.call_func`` runs wheel modules (key ops, file
  server, disk I/O) synchronously; offloaded via ``run_in_executor``.
  The exception-path event fire uses ``fire_event_async``.
* ``runner`` — ``RunnerClient.asynchronous`` forks a subprocess and
  joins it, blocking the calling thread; offloaded via
  ``run_in_executor``.
* ``get_token`` — ``LoadAuth.get_tok`` reads and deserializes tokens
  from disk; offloaded via ``run_in_executor``.
* ``mk_token`` — ``LoadAuth.mk_token`` invokes the eauth backend and
  writes to disk; offloaded via ``run_in_executor``.

Return-value shapes are preserved byte-for-byte. Adds unit tests that
walk through ``MWorker._handle_clear`` end-to-end to verify each
handler wraps its return in the ``(ret, {"fun": "send_clear"})``
envelope, plus per-handler tests that assert the sync body actually
runs through ``loop.run_in_executor`` rather than on the event loop
thread.

Move ``AuthFuncs._auth``, ``AuthFuncs._auth_impl`` and
``AuthFuncs._clear_signed`` (plus the ``ReqServerChannel._auth``
delegate in :mod:`salt.channel.server`) to ``async def`` so the minion
authentication handshake runs on the MWorker event loop and can be
interleaved with other in-flight requests.

Dispatch chain
--------------
The two call sites that invoke ``AuthFuncs._auth`` are already
``async``:

* ``ReqServerChannel.handle_message`` (non-pooled path) awaits
  ``self._auth(...)`` before returning.
* ``PoolRoutingChannel._handle_clear_auth_local`` (pooled path,
  bootstrap before IPC clients are up) awaits ``ReqServerChannel._auth``
  through the ``proxy`` shim.

Both are now updated to ``await`` the coroutine.

Sync APIs offloaded via ``loop.run_in_executor``
------------------------------------------------
* ``ckminions.connected_ids()`` — walks the minion data cache on disk.
* ``auto_key.check_autoreject`` / ``check_autosign`` — read the
  autoreject/autosign files.
* ``cache.fetch("keys" / "denied_keys", ...)`` — disk-backed key state.
* Every ``cache.store(...)`` write (8 sites in the state machine).
* ``salt.crypt.PublicKey.from_str`` — RSA key parse (CPU-bound).
* ``master_key.decrypt`` — RSA decrypt of the minion token.
* ``master_key.sign_key.sign`` — RSA sign of the master pubkey when
  ``master_sign_pubkey`` is enabled.
* ``master_key.encrypt`` — RSA sign of the AES digest.
* ``master_key.sign`` inside ``_clear_signed`` — RSA sign of the
  outgoing auth reply.
* ``pub.encrypt(aes)`` / ``pub.encrypt(session_material)`` — RSA
  encrypt of the AES key and session key for the responding minion.
* ``self.session_key(...)`` — disk-backed per-minion session cache.

Event fires migrated to ``fire_event_async``
--------------------------------------------
All ~10 ``self.event.fire_event(...)`` calls inside ``_auth_impl``
(auth attempts, key state transitions, cluster key replication) are
now ``await self.event.fire_event_async(...)``.

Return-value shapes are preserved byte-for-byte across every branch of
the auth state machine. The existing ``test_auth_funcs_*`` tests are
updated to ``async def`` and awaits, and new tests assert (a) the
wrappers are coroutine functions, (b) the ``full`` / ``pend`` events
go through ``fire_event_async``, (c) ``ckminions.connected_ids`` and
``master_key.sign`` (via ``_clear_signed``) are scheduled through
``loop.run_in_executor``. The ``test__auth_cmd_stats_passing`` channel
test switches the sync ``_auth`` fake to an ``async def`` that awaits
``asyncio.sleep`` so the duration assertion still holds without
blocking the event loop.

Autouse fixture in ``tests/pytests/unit/conftest.py`` enables
``loop.set_debug(True)`` + ``slow_callback_duration = 0.05`` on any
event loop created inside a test. Attaches a logging handler on the
``asyncio`` logger that captures the "took N seconds" / slow-callback
messages; at teardown the fixture fails the test if any violation was
recorded. Threshold matches the migration's stated 50 ms budget.

Opt-out per test via ``@pytest.mark.no_blocking`` (disables entirely)
or ``@pytest.mark.no_blocking(threshold=0.1)`` (override). The
``test_register_resources_*`` prefix is auto-exempted because those
tests do a full ``AESFuncs(opts)`` construction inline (loader init
takes >50ms). Individually exempted tests carry a ``reason=`` note
recommending the refactor.

``PYTHONASYNCIODEBUG=1`` can't be applied retroactively — the
fixture achieves the same effect via ``loop.set_debug(True)`` on
loops it observes. ``blockbuster`` is intentionally not added as a
dependency; the built-in ``slow_callback_duration`` already covers
the migration's stated needs.

Combined test suite (test_master.py + channel/test_server.py) still
passes at 157 passed / 24 skipped with ~3% wall-time overhead.

The async MWorker migration offloads sync internals via
``loop.run_in_executor(...)``; that means shared in-process caches
previously touched only from the loop thread are now touched from
arbitrary executor threads under concurrent load.

Two real races found:

- ``salt.cache.MemCache.data`` (class-level dict + per-storage
  OrderedDict) had no lock. Read-modify-write in ``fetch`` (pop+set
  atime), ``store`` (pop+size-check+set), ``flush`` (tuple iter+pop),
  and the lazy ``storage`` property is now serialized by a
  class-level ``threading.Lock``. Driver I/O (``super().fetch/store``)
  is intentionally kept OUTSIDE the lock so cache misses don't
  serialize on returner disk/DB I/O. Now hit from executor threads
  via ``AESFuncs._pillar``, ``_return``, ``_register_resources``, and
  ``AuthFuncs._auth_impl``.
- ``AuthFuncs.sessions`` (per-minion session-key cache) had no lock.
  ``session_key`` is now called from an executor thread inside
  ``_auth_impl``; adds a ``_sessions_lock`` and snapshots the cache
  entry under the lock so the check-then-write branch is atomic.

Adds ``tests/pytests/unit/test_shared_state_races.py`` — 11 stress
tests using ``ThreadPoolExecutor(max_workers=32)``: MemCache
store/fetch/flush + lazy storage init, AuthFuncs session torn-tuple,
LazyLoader RLock reentrancy, OptsDict concurrent writer+reader,
memoize idempotency under contention, and RSA verify+sign concurrent
correctness with real 2048-bit keys.

``loop.run_in_executor(None, sync_impl, ...)`` on CPython 3.10 does
NOT copy the caller's ``contextvars.Context`` to the executor thread.
Every one of the 60+ ``run_in_executor`` sites in the async handlers
was running with an empty ``salt.utils.ctx.request_context`` — log
records emitted from executor threads lost their JID / minion-id
enrichment, and any handler that reads the context inside the sync
callable saw nothing.

Fix: ``_ContextThreadPoolExecutor`` — a ``ThreadPoolExecutor``
subclass whose ``submit`` snapshots ``contextvars.copy_context()`` at
call time and re-enters it in the worker thread via ``Context.run``.
Installed as the default executor of ``MWorker`` io_loop in
``MWorker.__bind`` (single-line change). All existing
``run_in_executor(None, ...)`` calls automatically propagate context
with no per-handler edits. A regression guard test pins the
underlying CPython behavior so we know when the shim becomes
redundant.

Adds ``tests/pytests/unit/test_master_async_error_paths.py`` — 37
tests covering each of the 26 AESFuncs + 5 ClearFuncs async handlers:

- Exception propagation from the sync internal (mocked to raise
  ``RuntimeError``) — asserts each handler either re-raises or
  returns the documented error-shape envelope, matching pre-migration
  sync behavior.
- ``request_context`` visibility inside the executor thread — a
  single batched test proves the ContextVar survives the boundary
  for every handler.
- Cancellation smoke tests — cancelling a mid-flight ``_handle_aes``
  task doesn't leak partial state.

``test_revoke_auth_delegates_when_allowed`` marked with
``@pytest.mark.no_blocking(threshold=0.15)`` — AsyncMock construction
+ first executor submission occasionally trips the 50ms threshold on
loaded CI; handler is trivial delegation.

Adds ``tests/pytests/functional/master/test_async_handlers.py`` with
6 tests exercising the full ``_handle_aes`` dispatch path against
real subsystems:

- ``_pillar``: 6 concurrent requests, distinct minion ids, real
  ``salt.pillar.AsyncPillar.compile_pillar`` with a jinja top.sls +
  identity.sls; asserts per-minion isolation of the ``send_private``
  envelope.
- ``_return``: 20 concurrent payloads through real
  ``salt.utils.job.store_job`` + real ``local_cache`` returner
  (tmp_path); asserts every jid recovers via ``local_cache.get_jid``.
- ``_file_list``: 20 concurrent dispatches with a stubbed sync sleep
  in ``fs_.file_list`` — wall-time must be <50% of the serial floor
  (proves the executor parallelizes) and each response must contain
  its own saltenv tag.
- Fast-handler responsiveness: 8 slow-blocking ``_file_list`` calls
  saturating the executor, one ``_master_opts`` dispatched behind
  them; asserts the loop is still responsive.
- ``verify_minion``: 30 concurrent calls with real RSA keypairs +
  real ``key_cache.fetch/store``; asserts no key-state leak in the
  executor offload path.
- Mixed-workload envelope regression: 18 interleaved requests,
  every response's envelope matches its cmd.

Also refines the slow-callback matcher in
``tests/pytests/unit/conftest.py``: only trigger on
"Executing <Handle ...> took X seconds" warnings (per-callback
blocking), not "Executing <Task ...> took X seconds" (whole-Task
duration between yields). The Task shape false-positives on any
test that awaits >50ms of legitimate I/O; the Handle shape is the
"handler blocked the loop" signal we actually care about. With the
tighter matcher the previously-added ``no_blocking`` marker on
``test_revoke_auth_delegates_when_allowed`` is no longer needed and
is dropped.

When ``PublishServer.publish`` was invoked from a running io_loop (e.g. via
``MWorker._return -> store_job -> fire_event``), the outer
``SaltEvent.pusher`` SyncWrapper's worker thread ran this coroutine,
then ``self.pub_sock.send`` invoked SyncWrapper *again* -- it detected
the inner thread's running io_loop, spawned yet another thread, and
both deadlocked on ``threading.Thread.join()``.

Detect the async context via ``asyncio.get_running_loop()`` and bypass
the outer SyncWrapper entirely by using a raw ``_TCPPubServerPublisher``
cached per running loop. Use a ``WeakKeyDictionary`` keyed on the loop
object so a fresh SyncWrapper asyncio_loop can't inherit a dead
publisher via id() recycling. Invalidate the cache both proactively
(pre-flight ``stream.closed()``) and reactively (retry once on
``StreamClosedError``) so a downed puller doesn't poison the cache
forever.

Fixes saltstack#69986

Two related fixes for the async-mworker branch's async handler
dispatch:

1. SaltEvent.fire_event_async — when constructed with
   io_loop=None (as AESFuncs.__init__ does),
   _run_io_loop_sync=True and the branch calls
   self.pusher.publish(msg) on a SyncWrapper.  Invoked from
   the async-dispatched _pillar handler this both risks the
   nested-SyncWrapper deadlock Bug 1 addresses at the outer
   PublishServer level and fails with Event loop stopped before
   Future completed once the SyncWrapper's io_loop has been
   closed on any prior close cycle.  Reach into self.pusher.obj
   and await its native async publish directly on the running loop.

2. SaltEvent.fire_event — do not re-raise when
   self.pusher.publish fails.  Event publish is best-effort;
   re-raising leaks memory catastrophically under sustained failure
   because the traceback holds every frame in run_in_executor's
   thread including load (a state.apply return dict, often MB),
   and asyncio's exception logging retains those tracebacks.
   Observed ~66 GB in a single MWorker within a few minutes when the
   SyncWrapper io_loop was closed.

Two structural bugs on the async-mworker branch that surface only
under the TCP transport path (ZMQ has zmq_device_pooled which
handles both concerns natively):

1. ReqServerChannel._auth bypasses AuthFuncs.__init__ via
   AuthFuncs.__new__ and copies a hand-curated attribute list
   from the calling channel.  Missing attributes cause every auth
   attempt to fail:
     - _sessions_lock — added by the async migration to guard
       self.sessions under executor-thread concurrency, but not
       initialized when __init__ is skipped.  Provide a per-call
       threading.Lock.
     - auto_key / ckminions — the TCP path enters _auth
       via _handle_clear_auth_local which passes a proxy
       object that has neither, and AuthFuncs._auth_impl later
       does self.auto_key.check_autoreject(load["id"]).  Fall
       back to constructing fresh AutoKey / CkMinions from
       self.opts.

2. PoolRoutingChannel created a single RequestClient per
   pool and sent all traffic through it.  With TCP transport the
   client's IPC stream terminates at whichever MWorker won the
   accept() race on the shared workers-{pool}.ipc socket,
   pinning all pool traffic to one MWorker.  Under load: one MWorker
   ballooned to 3.6 GB with 32 executor threads while the other 9
   sat idle at 69 MB.  ZMQ's zmq_device_pooled doesn't have this
   problem because its ROUTER->DEALER pattern load-balances across
   REP peers in libzmq.

   Fix: bind one IPC socket per worker index
   (workers-{pool}-{N}.ipc) in RequestServer.pre_fork when
   the pool tells us its worker_count.  Each MWorker picks its
   own socket in post_fork via pool_index (threaded through
   from MWorker.__bind).  PoolRoutingChannel opens
   worker_count clients — one per per-worker socket — and
   round-robins across them at dispatch time.  Fair distribution
   across all MWorkers, equivalent semantics to ZMQ's DEALER in user
   space.

Validated on TCP transport with 50 minions under 10 parallel
async-ping loops: sustained 22,000 returns/min for 5 minutes, all
10 MWorkers evenly loaded at 134 MB each (was 3.6 GB / 69 MB
skew), PoolRouter stable at 489 MB, minion FD flat at 18.