From 3428890b0ee92b9dd4426b7645a6ab8610f3f3d5 Mon Sep 17 00:00:00 2001 From: twangboy Date: Mon, 3 Aug 2026 16:31:11 -0600 Subject: [PATCH 1/2] Reset event-bus pusher on failed fire_event send (#69914) SaltEvent.fire_event() re-raised send failures without resetting self.pusher/self.cpush, so once an MWorker's IPC pusher stream broke (e.g. a stale epoll fd after EventPublisher restarts), every subsequent job return on that worker hit the same exception forever, silently dropping the return before it reached the job cache and burning memory/CPU on repeated thread+IOLoop churn. Close the pusher on failure, mirroring the existing reconnect pattern on the subscribe side, so the next fire_event() call reconnects instead. --- changelog/69914.fixed.md | 1 + salt/utils/event.py | 1 + tests/pytests/unit/utils/event/test_event.py | 16 ++++++++++++++++ 3 files changed, 18 insertions(+) create mode 100644 changelog/69914.fixed.md diff --git a/changelog/69914.fixed.md b/changelog/69914.fixed.md new file mode 100644 index 000000000000..8d6d76334e45 --- /dev/null +++ b/changelog/69914.fixed.md @@ -0,0 +1 @@ +Fixed the master event bus keeping a broken pusher connection after a failed send, which caused every subsequent job return on that worker to fail and silently drop the job return instead of reconnecting. diff --git a/salt/utils/event.py b/salt/utils/event.py index 93a3d0518ea0..956caacbfcf7 100644 --- a/salt/utils/event.py +++ b/salt/utils/event.py @@ -887,6 +887,7 @@ def fire_event(self, data, tag, timeout=1000): exc, exc_info_on_loglevel=logging.DEBUG, ) + self.close_pull() raise else: self.io_loop.spawn_callback(self.pusher.send, msg) diff --git a/tests/pytests/unit/utils/event/test_event.py b/tests/pytests/unit/utils/event/test_event.py index 3b3c2944bde6..e7f2ce97a887 100644 --- a/tests/pytests/unit/utils/event/test_event.py +++ b/tests/pytests/unit/utils/event/test_event.py @@ -333,6 +333,22 @@ def test_connect_pull_should_error_log_on_other_errors(error): ) +def test_fire_event_closes_pusher_on_send_failure(): + """ + A failed pusher.send() must drop the broken pusher (close_pull) so the + next fire_event() reconnects instead of hammering the same dead stream + forever. See https://github.com/saltstack/salt/issues/69914 + """ + event = SaltEvent(node=None) + with patch.object(event, "pusher") as mock_pusher: + event.cpush = True + mock_pusher.send.side_effect = FileNotFoundError(2, "No such file or directory") + with pytest.raises(FileNotFoundError): + event.fire_event({"data": "foo1"}, "evt1") + assert event.cpush is False + assert event.pusher is None + + @pytest.mark.slow_test def test_master_pub_permissions(sock_dir): with eventpublisher_process(str(sock_dir)): From 6ecb95230cebabb1011f3f4a2f3aa9cef2209513 Mon Sep 17 00:00:00 2001 From: twangboy Date: Mon, 3 Aug 2026 21:46:22 -0600 Subject: [PATCH 2/2] Add functional test for event-bus pusher recovery (#69914) dwoz requested a functional/integration test on PR #69937 since the existing unit test only exercised a mocked pusher. Add a test that spins up a real EventPublisher and a real SaltEvent pusher, fakes a send() failure at the IPCMessageClient boundary to reproduce the reported FileNotFoundError deterministically, and asserts the pusher is dropped and a subsequent fire_event() reconnects and actually delivers the event to a live listener. --- .../functional/master/test_event_publisher.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/tests/pytests/functional/master/test_event_publisher.py b/tests/pytests/functional/master/test_event_publisher.py index ba3f30a4d7ac..e27f01ca5de3 100644 --- a/tests/pytests/functional/master/test_event_publisher.py +++ b/tests/pytests/functional/master/test_event_publisher.py @@ -8,6 +8,7 @@ import salt.config import salt.utils.event +from tests.support.mock import patch log = logging.getLogger() # __name__) @@ -154,6 +155,57 @@ def listeners(opts, stop_event): thread.join() +def test_fire_event_recovers_after_pusher_send_failure(publisher, opts): + """ + Regression test for https://github.com/saltstack/salt/issues/69914 + + A failed ``pusher.send()`` (e.g. the ``FileNotFoundError`` raised by a + stale IPC stream after ``EventPublisher`` is restarted) must not + permanently wedge the event bus for the lifetime of the ``SaltEvent`` + instance. Before the fix, every subsequent ``fire_event()`` call kept + hitting the same dead pusher forever; after the fix, the broken pusher + is dropped and the next call transparently reconnects and delivers. + + This exercises the real ``SyncWrapper`` + threaded ``IOLoop`` + + ``IPCMessageClient`` stack used by master worker processes (only the + innermost ``send()`` call is faked, to deterministically reproduce the + failure without racing the actual epoll bug), against a real, running + ``EventPublisher`` process. + """ + event = salt.utils.event.get_event("master", opts=opts, listen=False) + try: + # Establish a real, connected pusher against the live EventPublisher. + assert event.fire_event({"data": "foo1"}, "evt1") is True + assert event.cpush is True + + # Simulate the real-world failure: the underlying IPCMessageClient's + # send() raises inside the SyncWrapper's worker thread. + with patch.object( + event.pusher.obj, + "send", + side_effect=FileNotFoundError(2, "No such file or directory"), + ): + with pytest.raises(FileNotFoundError): + event.fire_event({"data": "foo2"}, "evt2") + + # The broken pusher must be dropped, not reused. + assert event.cpush is False + assert event.pusher is None + + # The next fire_event() call must reconnect and actually deliver, + # instead of raising the same exception forever. + listener = salt.utils.event.get_event("master", opts=opts, listen=True) + try: + assert event.fire_event({"data": "foo3"}, "evt3") is True + evt = listener.get_event(tag="evt3", wait=10, match_type="startswith") + assert evt is not None + assert evt["data"] == "foo3" + finally: + listener.destroy() + finally: + event.destroy() + + def test_publisher_mem(publisher, publish, listeners, stop_event): """ Test event publisher memory consumption.