[Memory Leak] Reset event-bus pusher on failed fire_event send (#69914)

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.

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)

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.

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