Guard PublishClient.recv against torn-down stream socket (#66435) - #69479
Guard PublishClient.recv against torn-down stream socket (#66435)#69479dwoz wants to merge 1 commit into
Conversation
Per review feedback on saltstack#69479: if ``_stream.socket`` is ``None`` because Tornado tore the IOStream down concurrently, simply returning ``None`` left ``_stream`` set and a caller looping on ``recv(timeout=0)`` would spin returning ``None`` forever without ever reconnecting. Mirror the ``StreamClosedError`` branch: drop the dead stream, fire the disconnect callback, and call ``connect()`` so the next ``recv`` enters a healthy stream.
PublishClient.recv(timeout=0) was passing self._stream.socket straight
to selectors.DefaultSelector().register() without checking whether the
IOStream's underlying socket had been concurrently torn down. Tornado
sets IOStream.socket to None once the stream is closed, and the
non-blocking peek would then raise
TypeError: argument must be an int, or have a fileno() method.
(or, after the fd>1023 cleanup in saltstack#68136, a ValueError from the
selectors backend) escaping all the way out to the salt CLI, breaking
every salt and salt-master invocation on hosts where the
publisher-side stream closed underneath the client.
Treat a missing socket as "no events pending" and return None so the
caller re-enters its connect/reconnect loop instead of crashing.
Fixes saltstack#66435
dwoz
left a comment
dwoz
left a comment
There was a problem hiding this comment.
Regression-suspect. Two HIGH blockers (own-PR — cannot REQUEST_CHANGES, flagged as COMMENT).
The guard swaps a crash-on-dead-socket for an unbounded block-on-dead-socket in the exact recv(timeout=0) non-blocking path used by SaltEvent.get_event_noblock / _get_event(no_block=True) (salt CLI tight-poll). Details inline.
Cross-branch: PR body claims "3008.x and master are unaffected: that recv path has already been rewritten to use asyncio.ensure_future(self._read_into_unpacker())". Verified against origin/3008.x and origin/master — PublishClient.recv(timeout=0) is byte-identical to origin/3007.x. PR body also cites selectors.DefaultSelector as the crash site, but commit 591bfe0b51c removed that from origin/3007.x tip; the crash reproduces on released v3007.14, not on tip. Correct the PR body; if the guard is desired on 3008.x/master, plan sibling PRs or merge-forward rather than assuming bots will do the right thing.
LOW (cross-transport): ZMQ PublishClient.recv (salt/transport/zeromq.py:360) polls self._socket without a symmetric None check; ws.PublishClient.recv (salt/transport/ws.py:200) has its own while self._ws is None retry. This PR only patches TCP. Matches the reported TCP-only crash, but the ZMQ path has a comparable close() vs recv() window (raises zmq.error.ZMQError, existing consumers catch it). Not a blocker — but document the intent.
LOW (nit): comment "so a caller looping on recv(timeout=0) doesn't spin returning None forever" is misleading; the pre-existing if self._stream is None: return None at line 513 already returns None for the same state. The guard doesn't stop the spin; the await connect() does — which is exactly the source of HIGH 1.
| stream.close() | ||
| if self.disconnect_callback: | ||
| await self.disconnect_callback() | ||
| await self.connect() |
There was a problem hiding this comment.
HIGH (regression / hangs-forever) — await self.connect() inside PublishClient.recv(timeout=0). recv(timeout=0) is a non-blocking peek used by SaltEvent.get_event_noblock and _get_event(no_block=True) (the salt CLI's tight polling loop). connect() delegates to _connect() → getstream(), whose retry loop only exits on _closed/_closing or after getstream's per-attempt timeout (default 5 s) plus a 1 s backoff, looping forever if the master is unreachable. Non-blocking callers will now block seconds → minutes instead of returning None.
- Suggested fix: drop the dead stream, clear
_read_task, fire the disconnect callback, and returnNone— let the existingwhile self._stream is Noneloop in the blocking branch (or theon_recv_handlersleep-loop) do the reconnect. Do notawait connect()in thetimeout==0branch.
| self._stream = None | ||
| stream.close() | ||
| if self.disconnect_callback: | ||
| await self.disconnect_callback() |
There was a problem hiding this comment.
HIGH (masks-real-error / latent TypeError) — await self.disconnect_callback(). On the minion, _MinionPubChannel.disconnect_callback (salt/channel/client.py:587) is a synchronous def returning None. await None raises TypeError: object NoneType can't be used in 'await' expression. The test hides this by injecting an AsyncMock. Same latent bug pre-exists at _read_into_unpacker:485, but this PR adds a second site with the same shape and its own test that hides it.
- Suggested fix:
if inspect.iscoroutinefunction(self.disconnect_callback): await self.disconnect_callback()else call it; or make_MinionPubChannel.disconnect_callbackasync. And dropAsyncMockfrom the test — use a plainMagicMockmatching real-world semantics.
| pass | ||
|
|
||
|
|
||
| async def test_recv_timeout_zero_stream_socket_none(): |
There was a problem hiding this comment.
MED (test gap) — Test replaces client.connect with an AsyncMock that instantly returns. Does not verify: (a) caller doesn't block when real connect() retries fail (the actual regression above), (b) sync-disconnect_callback TypeError path, (c) _read_task state coherence after the guard runs, (d) any behavior on the actual production crash reproduction (.socket=None combined with a real Tornado IOStream under read_bytes).
- Suggested fix: add a test using a
disconnect_callbackthat's a plaindef(matches production), and a test assertingrecv(timeout=0)returns in <100 ms even whenconnect()is slow.
3 participants
What does this PR do?
Adds a small
None-check insalt.transport.tcp.PublishClient.recv(timeout=0)so that a stream whose underlying socket has been concurrently torn down no longer crashes the caller. The non-blocking peek now treats a missing socket as "no events pending" and returnsNone, letting the existing reconnect loop take over.What issues does this PR fix or reference?
Fixes #66435
Related (different bugs in the same module, do not fix this one):
EventListenercallback typefd > 1023select.select()failure (already merged; replacedselect.select()withselectors.DefaultSelector()but did not add theNone-socket guard)Previous Behavior
Under load — and reliably on FreeBSD 14 with the 3007.x packages and on RHEL 9.2 / Debian 12 with
ipc_mode: ipc— everysaltandsalt-masterinvocation crashed out ofPublishClient.recvwith one of:(in 3007.0, from
select.select([self._stream.socket], [], [], 0)) or, after #68136 swapped inselectors.DefaultSelector,from
selectors._fileobj_to_fd. In both cases the root cause is the same: between thewhile self._stream is None: await self.connect()check at the top ofrecv()and the selector peek a few lines later, the TornadoIOStreamfor the publish IPC socket can be closed by another task. Tornado setsIOStream.sockettoNoneon close, so the peek tries to registerNonewith the selector and dies with an unhandled exception. The error escaped all the way throughsalt.utils.asynchronous.SyncWrapperto the salt CLI, breaking every command.New Behavior
recv(timeout=0)snapshotsself._stream.socketonce. If it'sNone, the method returnsNoneimmediately — the same return value the caller already handles when no events are pending — and the existing reconnect path takes over without crashing.A regression test (
tests/pytests/unit/transport/test_publish_client.py::test_recv_timeout_zero_stream_socket_none) constructs aPublishClient, sets its_streamto a mock whose.socketisNone, and assertsrecv(timeout=0)returnsNonewithout raising. It fails on unmodified3007.xwithValueError: Invalid file object: Noneand passes with the fix.3008.x and master are unaffected: that recv path has already been rewritten to use
asyncio.ensure_future(self._read_into_unpacker())instead of a kernel-level socket peek, so no merge-forward port is needed beyond what the merge bots will do.Merge requirements satisfied?
changelog/66435.fixed.md)tests/pytests/unit/transport/test_publish_client.py)Commits signed with GPG?
No (matches surrounding non-merge commits on this branch).