diff --git a/changelog/66435.fixed.md b/changelog/66435.fixed.md new file mode 100644 index 000000000000..6f6e7b6a2ef0 --- /dev/null +++ b/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. diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py index 642804aa6ecc..b6e0565dbffc 100644 --- a/salt/transport/tcp.py +++ b/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 diff --git a/tests/pytests/unit/transport/test_publish_client.py b/tests/pytests/unit/transport/test_publish_client.py index 39f541700e4d..a8cfbec11e82 100644 --- a/tests/pytests/unit/transport/test_publish_client.py +++ b/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.