[Bug]: SyncWrapper reaps queue tasks awaited by live request workers

Start with SyncWrapper._target() and _task_is_orphan() in salt/utils/asynchronous.py, then inspect the existing ownership regression tests. Run the reported minimal reproduction and relevant tests for wait_for, gather, and unowned worker cleanup. Done means retained workers and tasks they await survive across calls, unowned background tasks are reaped, and close() drains owned tasks.

What happened?

On 3008.x, SyncWrapper._target() can cancel a queue task awaited by a live, explicitly retained request worker after a successful wrapped call. The worker then exits, leaving subsequent requests without a consumer.

This was found while validating DeltaProxy over ZeroMQ for #70711 / PR #70739. With real daemons and preaccepted controller/child keys, the controller blocks during initial module synchronization:

saltutil.sync_all -> HighState.get_top -> RemoteClient.envs
  -> SyncWrapper._wrap -> thread.join

Request tracing shows the first request completing and the next request finding RequestClient._send_recv already finished. The controller cannot reach readiness. This is separate from the transport-topic routing failure in #70711.

Minimal reproduction without Salt daemons

Run the following using a Python environment containing the 3008.x checkout:

import asyncio
from salt.utils.asynchronous import SyncWrapper

class Worker:
    __slots__ = ("queue", "task")
    async_methods = ["send"]

    def __init__(self):
        self.queue = asyncio.Queue()
        self.task = None

    async def consume(self):
        while True:
            reply, value = await asyncio.wait_for(self.queue.get(), timeout=60)
            reply.set_result(value)

    async def send(self, value):
        if self.task is None:
            self.task = asyncio.create_task(self.consume())
        reply = asyncio.get_running_loop().create_future()
        self.queue.put_nowait((reply, value))
        return await asyncio.wait_for(reply, timeout=1)

async def main():
    wrapper = SyncWrapper(Worker)
    try:
        print("first reply:", wrapper.send("first"))
        print("retained worker finished:", wrapper.obj.task.done())
        assert not wrapper.obj.task.done(), "cleanup stopped the owned worker"
        assert wrapper.send("second") == "second"
    finally:
        wrapper.close()

asyncio.run(main())

Actual on baseline 987eefd0f3 (upstream 3008.x): the first reply is first, but the retained worker is already finished and the assertion fails. Removing that assertion makes the second request time out. No master, external extension, outage, or runtime workaround is needed for this reproduction.

Expected: an explicitly retained worker and the tasks it awaits survive between dispatches; repeated requests succeed. Unowned background tasks must still be reaped, and close() must drain owned tasks normally.

Diagnosis and regression coverage

_task_is_orphan() checks direct referrers. A queue coroutine wrapped by asyncio.wait_for() or asyncio.gather() becomes a separate task without a direct user-object reference. The per-dispatch reaper cancels that child even though a retained worker is waiting for it. Cancellation propagates into the worker.

The expanded ownership regressions on the unmodified baseline produce 12 failed and 5 passed cases, covering wait_for, gather, custom task factories, and explicitly materialized instance dictionaries. Controls cover direct/shielded queue awaits and continued cleanup of an unowned worker.

Related history: #70169 introduced per-call cleanup to address retained task leaks. This report concerns cleanup cancelling tasks still needed by a live worker, rather than a task leak or an outage-triggered context-finalizer deadlock.

PR #70741 provides the focused startup fix independently against 3008.x. Its exact commit (0d5e723ddc) is also included in the stacked transport PR #70739 for combined verification. Review and merge #70741 first. The transport-topic changes are not required to reproduce this issue.

Type of salt install

pip (git), testing the upstream 3008.x source at 987eefd0f3. The archive-based test environment reports 3008.0; the source revision above identifies the tested code.

Major version

3008.x

What supported OS are you seeing the problem on?

Debian 13.7 (trixie), Linux aarch64; Python 3.11.18.

salt --versions-report output

Salt Version:
                  Salt: 3008.0
 
Python Version:
                Python: 3.11.18 (main, Oct 28 2026, 03:03:40) [GCC 14.2.0]
 
Dependency Versions:
                  cffi: 2.1.1
              cherrypy: 18.10.0
          cryptography: 50.0.2
              dateutil: 4.0.0
             docker-py: 7.2.0
                 gitdb: 4.0.12
             gitpython: Not Installed
                Jinja2: 3.1.6
               libgit2: Not Installed
          looseversion: 1.3.0
              M2Crypto: Not Installed
                  Mako: Not Installed
               msgpack: 1.2.4
          msgpack-pure: Not Installed
          mysql-python: Not Installed
             packaging: 26.3
             pycparser: 3.12
              pycrypto: Not Installed
          pycryptodome: 3.24.2
                pygit2: Not Installed
          python-gnupg: 0.5.7
                PyYAML: 6.0.3
                 PyZMQ: 27.3.0
                relenv: Not Installed
                 smmap: 5.0.3
               timelib: 0.3.0
               Tornado: 6.6
                   ZMQ: 4.3.5
 
Salt Extensions:
 pytest-salt-factories: 1.0.5
 
Salt Package Information:
          Package Type: pip
 
System Versions:
                  dist: debian 13.7 trixie
                locale: utf-8
               machine: aarch64
               release: 7.0.12-linuxkit
                system: Linux
               version: Debian GNU/Linux 13.7 trixie