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/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. 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)):