From 63bade4325499202aba2df982116a4bddbd7f6fe Mon Sep 17 00:00:00 2001 From: Will Hunt <2072976+Half-Shot@users.noreply.github.com> Date: Wed, 7 Oct 2026 14:57:56 +0100 Subject: [PATCH] Skip running delayed event background job if the room event stream has not advanced. (#20316) In a similar vein to #20315, this attempts to speed up Synapse when delayed events aren't in use. Here we skip doing any work if the room event stream has not advanced. --- changelog.d/20316.bugfix | 1 + synapse/handlers/delayed_events.py | 10 +++++++ tests/rest/client/test_delayed_events.py | 37 ++++++++++++++++++++++++ 3 files changed, 48 insertions(+) create mode 100644 changelog.d/20316.bugfix diff --git a/changelog.d/20316.bugfix b/changelog.d/20316.bugfix new file mode 100644 index 0000000000..3c49f942a9 --- /dev/null +++ b/changelog.d/20316.bugfix @@ -0,0 +1 @@ +Only run delayed event background jobs when a new room event is processed. diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index c86194f77c..7ce3bfdd37 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -146,6 +146,16 @@ class DelayedEventsHandler: if self._event_processing: return + # This is called for updates on any stream (typing, receipts, to-device, + # etc), but only new room events can produce the state deltas we + # process. If the room stream hasn't advanced past what we've already + # handled, there's nothing to do, so skip the database round-trips. + if ( + self._event_pos is not None + and self._event_pos >= self._store.get_room_max_stream_ordering() + ): + return + self._event_processing = True async def process() -> None: diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 93e933d991..6877656e7e 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -15,6 +15,7 @@ """Tests REST events for /delayed_events paths.""" from http import HTTPStatus +from unittest.mock import AsyncMock from parameterized import parameterized @@ -774,6 +775,42 @@ class DelayedEventsTestCase(HomeserverTestCase): self._find_sent_delayed_event(self.user1_access_token, delay_id, False) self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + def test_new_state_processing_skipped_without_new_room_events(self) -> None: + """Ensure that we only process state deltas (and hit the database) when + new room events have been persisted, rather than on every replication + notification.""" + state_key = "to_be_cancelled_by_other_user" + + channel = self.make_request( + "PUT", + _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), + {}, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + self.assertEqual(1, len(self._get_delayed_events())) + + store = self.hs.get_datastores().main + store.get_count_of_delayed_events = AsyncMock(return_value=1) # type: ignore[method-assign] + + # Notifications for other streams (typing, receipts, to-device, etc) + # don't advance the room stream, so shouldn't cause any processing. + self.hs.get_notifier().notify_replication() + self.pump() + store.get_count_of_delayed_events.assert_not_called() + + # A new state event from another user should still be processed, and + # cancel the pending delayed event. + self.helper.send_state( + self.room_id, + _EVENT_TYPE, + {}, + self.user2_access_token, + state_key=state_key, + ) + store.get_count_of_delayed_events.assert_called() + self.assertListEqual([], self._get_delayed_events()) + @parameterized.expand((("client_api", False), ("admin_api", True))) def test_delayed_events_are_cancelled_on_deactivation( self, _name: str, by_admin: bool