Guard PublishClient.recv against torn-down stream socket (#66435)

changelog/66435.fixed.md

@@ -0,0 +1 @@
+Fixed ``TypeError: argument must be an int, or have a fileno() method.`` (or, after the fd>1023 cleanup in #68136, a ``ValueError`` from the selectors backend) raised from ``salt.transport.tcp.PublishClient.recv`` when the IOStream's underlying socket was torn down between the stream-not-None check and the non-blocking read. The ``recv(timeout=0)`` path now detects the dead stream, drops it, invokes the disconnect callback and reconnects instead of crashing or spinning forever.

salt/transport/tcp.py

@@ -512,6 +512,24 @@ async def recv(self, timeout=None):
             # misses data that ``read_bytes`` would return immediately.
             if self._stream is None:
                 return None
+            # Tornado's IOStream keeps a reference to itself but sets
+            # ``socket`` to ``None`` once closed.  If we let a read task
+            # start against that stream ``read_bytes`` blows up with
+            # ``TypeError: argument must be an int, or have a fileno()
+            # method.`` (or a ``ValueError`` from the selectors backend
+            # after the fd>1023 cleanup in #68136), which escaped all the
+            # way out to the salt CLI.  Drop the dead stream and trigger
+            # the normal reconnect path so a caller looping on
+            # ``recv(timeout=0)`` doesn't spin returning ``None`` forever.
+            # See issue #66435.
+            if self._stream.socket is None:
+                stream = self._stream
+                self._stream = None
+                stream.close()
+                if self.disconnect_callback:
+                    await self.disconnect_callback()
+                await self.connect()
+                return None
             task = self._ensure_read_task()
             if task is None:
                 return None

tests/pytests/unit/transport/test_publish_client.py

@@ -17,7 +17,7 @@
 import salt.transport.ws
 import salt.transport.zeromq
 import salt.utils.stringutils
-from tests.support.mock import MagicMock, patch
+from tests.support.mock import AsyncMock, MagicMock, patch
 
 log = logging.getLogger(__name__)
 
@@ -339,6 +339,54 @@ async def test_recv_timeout_zero():
             pass
 
 
+async def test_recv_timeout_zero_stream_socket_none():
+    """
+    Regression test for #66435.
+
+    If a stream's underlying socket has been torn down concurrently (the
+    Tornado ``IOStream`` keeps a reference to itself but its ``socket``
+    attribute becomes ``None`` once closed), ``recv(timeout=0)`` used to
+    let ``_read_into_unpacker`` call ``read_bytes`` on the dead stream
+    and crash with::
+
+        TypeError: argument must be an int, or have a fileno() method.
+
+    (or, after the fd>1023 cleanup in #68136, a ``ValueError`` from the
+    selectors backend) escaping all the way out to the salt CLI.
+
+    The non-blocking peek must drop the dead stream and trigger the
+    normal reconnect path so a caller looping on ``recv(timeout=0)``
+    doesn't spin returning ``None`` forever.
+    """
+    host = "127.0.0.1"
+    port = 11122
+    ioloop = asyncio.get_running_loop()
+    mock_stream = MagicMock()
+    mock_stream.socket = None
+    mock_unpacker = MagicMock()
+    mock_unpacker.__iter__.return_value = []
+    disconnect_callback = AsyncMock()
+
+    async def fake_connect(*args, **kwargs):
+        return None
+
+    with patch("salt.utils.msgpack.Unpacker", return_value=mock_unpacker):
+        client = salt.transport.tcp.PublishClient(
+            {}, ioloop, host=host, port=port, disconnect_callback=disconnect_callback
+        )
+        client._stream = mock_stream
+        with patch.object(client, "connect", side_effect=fake_connect) as mock_connect:
+            # Must not raise.
+            result = await client.recv(timeout=0)
+
+    assert result is None
+    # Dead stream is dropped and reconnect path is triggered.
+    assert client._stream is None
+    mock_stream.close.assert_called_once()
+    disconnect_callback.assert_awaited_once()
+    mock_connect.assert_called_once()
+
+
 def test_close_does_not_leak_pending_read_task(caplog):
     """
     Regression test for #68998.