From f71f7cc0d915fdb66b7b40bcaf672dbcd3b4b547 Mon Sep 17 00:00:00 2001 From: "Gary T. Giesen" Date: Sat, 4 Jul 2026 16:35:39 -0400 Subject: [PATCH 1/2] Fix event delivery to multiple websocket clients on /events EventListener._handle_event_socket_recv iterated over the futures list while removing delivered futures from that same list, which skips every other future. With multiple websocket clients registered under the same (tag, matcher) key, each event only reached some of the waiting clients. Iterate over a snapshot of the list so every matching future is resolved. Fixes #35798 --- changelog/35798.fixed.md | 1 + salt/netapi/rest_tornado/saltnado.py | 6 ++- .../netapi/saltnado/test_event_listener.py | 46 +++++++++++++++++++ 3 files changed, 52 insertions(+), 1 deletion(-) create mode 100644 changelog/35798.fixed.md create mode 100644 tests/pytests/unit/netapi/saltnado/test_event_listener.py diff --git a/changelog/35798.fixed.md b/changelog/35798.fixed.md new file mode 100644 index 000000000000..2d35485c14cb --- /dev/null +++ b/changelog/35798.fixed.md @@ -0,0 +1 @@ +Fixed a race in the rest_tornado event listener so a single event is delivered to every websocket client waiting on a matching tag instead of only some of them diff --git a/salt/netapi/rest_tornado/saltnado.py b/salt/netapi/rest_tornado/saltnado.py index 99971b778b9d..1b9c0c7962a7 100644 --- a/salt/netapi/rest_tornado/saltnado.py +++ b/salt/netapi/rest_tornado/saltnado.py @@ -385,7 +385,11 @@ def _handle_event_socket_recv(self, raw): if not is_matched: continue - for future in futures: + # Iterate over a snapshot of the futures list. We remove delivered + # futures from the underlying list below, and mutating the list + # while iterating it would skip futures, causing some waiting + # clients to miss the event (see #35798). + for future in list(futures): if future.done(): continue future.set_result({"data": data, "tag": mtag}) diff --git a/tests/pytests/unit/netapi/saltnado/test_event_listener.py b/tests/pytests/unit/netapi/saltnado/test_event_listener.py new file mode 100644 index 000000000000..19cde5286e0c --- /dev/null +++ b/tests/pytests/unit/netapi/saltnado/test_event_listener.py @@ -0,0 +1,46 @@ +from collections import defaultdict + +import salt.netapi.rest_tornado.saltnado as saltnado_app +from salt.ext.tornado.concurrent import Future +from tests.support.mock import MagicMock + + +def _make_event_listener(): + """ + Build an EventListener without touching the real master event bus. + """ + event_listener = saltnado_app.EventListener.__new__(saltnado_app.EventListener) + event_listener.tag_map = defaultdict(list) + event_listener.request_map = defaultdict(list) + event_listener.timeout_map = {} + event_listener.event = MagicMock() + return event_listener + + +def test_handle_event_socket_recv_delivers_to_all_waiters(): + """ + A single matching event must resolve every future waiting on that tag. + + Regression test for #35798: the delivery loop used to remove futures from + the very list it was iterating, skipping every other waiter so that only + some websocket clients received the event. + """ + event_listener = _make_event_listener() + matcher = saltnado_app.EventListener.exact_matcher + key = ("evt1", matcher) + + futures = [Future() for _ in range(4)] + for future in futures: + event_listener.tag_map[key].append(future) + + # event.unpack(raw) -> (mtag, data) + event_listener.event.unpack.return_value = ("evt1", {"data": "foo"}) + + event_listener._handle_event_socket_recv("raw") + + for future in futures: + assert future.done() + assert future.result() == {"data": {"data": "foo"}, "tag": "evt1"} + + # every delivered future should be removed from the tag_map list + assert event_listener.tag_map[key] == [] From 9e0f2d6cc0cb3bfa4a035f03428313e16f364df2 Mon Sep 17 00:00:00 2001 From: "Gary T. Giesen" Date: Sun, 5 Jul 2026 23:23:40 -0400 Subject: [PATCH 2/2] Add direct and inverse regression tests for saltnado websocket event delivery The direct test subscribes multiple clients through get_event(request) with the exact defaults the websocket handlers in saltnado_websockets.py pass (no tag, no matcher, so tag="" with prefix_matcher) and asserts a single event resolves every waiting future; it fails without the snapshot fix in _handle_event_socket_recv. The inverse test guards against overcorrection: an already-done future must keep its original result and stay in the tag_map, and a future waiting on a different exact tag (the SaltAPIHandler.get_minion_returns shape) must stay pending; it passes with and without the fix. --- .../netapi/saltnado/test_event_listener.py | 75 +++++++++++++++++++ 1 file changed, 75 insertions(+) diff --git a/tests/pytests/unit/netapi/saltnado/test_event_listener.py b/tests/pytests/unit/netapi/saltnado/test_event_listener.py index 19cde5286e0c..fab8dc10914a 100644 --- a/tests/pytests/unit/netapi/saltnado/test_event_listener.py +++ b/tests/pytests/unit/netapi/saltnado/test_event_listener.py @@ -44,3 +44,78 @@ def test_handle_event_socket_recv_delivers_to_all_waiters(): # every delivered future should be removed from the tag_map list assert event_listener.tag_map[key] == [] + + +def test_handle_event_socket_recv_websocket_default_subscription_35798(): + """ + One event must reach every concurrent websocket client subscribed through + the production entry point. + + This is the exact #35798 scenario: AllEventsHandler.on_message in + saltnado_websockets.py subscribes each client with + ``event_listener.get_event(self)`` and nothing else, so the decisive + arguments are the defaults, ``tag=""`` with ``prefix_matcher``, which + every event matches. All clients therefore share a single tag_map entry, + and a single incoming event must resolve all of their futures. + """ + event_listener = _make_event_listener() + + # one future per connected websocket client, registered exactly the way + # the websocket handlers do it: get_event(request) with no tag/matcher + requests = [MagicMock() for _ in range(3)] + futures = [event_listener.get_event(request) for request in requests] + + event_listener.event.unpack.return_value = ( + "salt/job/20260705000000000000/ret/minion1", + {"data": "foo"}, + ) + + event_listener._handle_event_socket_recv("raw") + + for future in futures: + assert future.done() + assert future.result() == { + "data": {"data": "foo"}, + "tag": "salt/job/20260705000000000000/ret/minion1", + } + + key = ("", saltnado_app.EventListener.prefix_matcher) + assert event_listener.tag_map[key] == [] + + +def test_handle_event_socket_recv_ignores_done_and_unmatched_35798(): + """ + Guard against overcorrection in the #35798 fix: iterating a snapshot of + the futures list must not widen delivery. A future that is already done + (for example one that timed out) must keep its original result and must + not be re-resolved, and a future waiting on a different exact tag must + stay pending and stay registered. This test passes with and without the + fix. + """ + event_listener = _make_event_listener() + # exact_matcher is what SaltAPIHandler.get_minion_returns passes in + # production (saltnado.py) for salt/job and syndic/job return tags + matcher = saltnado_app.EventListener.exact_matcher + matched_key = ("evt1", matcher) + other_key = ("evt2", matcher) + + done_future = Future() + done_future.set_result("already-done") + pending_future = Future() + other_future = Future() + event_listener.tag_map[matched_key].extend([done_future, pending_future]) + event_listener.tag_map[other_key].append(other_future) + + event_listener.event.unpack.return_value = ("evt1", {"data": "foo"}) + + event_listener._handle_event_socket_recv("raw") + + # an already-done future must not be re-resolved with the event payload + assert done_future.result() == "already-done" + # the pending waiter on the matching tag still receives the event + assert pending_future.result() == {"data": {"data": "foo"}, "tag": "evt1"} + # a waiter on a non-matching exact tag must not receive the event + assert not other_future.done() + assert event_listener.tag_map[other_key] == [other_future] + # done futures are skipped by the delivery loop, not removed + assert event_listener.tag_map[matched_key] == [done_future]