From 0eefdbcb9535d6595ca2b419217404acf57623ac Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 5 May 2026 08:40:17 -0500 Subject: [PATCH 1/3] fix: Cap `WorkerLock` timeout intervals to 60 seconds (#19394) Fixes the symptoms of https://github.com/element-hq/synapse/issues/19315 / https://github.com/element-hq/synapse/issues/19588 but not the underlying reason causing the number to grow so large in the first place. ``` ValueError: Exceeds the limit (4300 digits) for integer string conversion; use sys.set_int_max_str_digits() to increase the limit ``` Copied from the original pull request on [Famedly's Synapse repo](https://github.com/famedly/synapse/pull/221) (with some edits): Basing the time interval around a 5 seconds leaves a big window of waiting especially as this window is doubled each retry, when another worker could be making progress but can not. Right now, the retry interval in seconds looks like `[0.2, 5, 10, 20, 40, 80, 160, 320, (continues to double)]` after which logging should start about excessive times and (relatively quickly) end up with an extremely large retry interval with an unrealistic expectation past the heat death of the universe. 1 year in seconds = 31,536,000. With this change, retry intervals in seconds should look more like: ``` [ 0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 12.8, 25.6, 51.2, 60, < never goes higher than this ] ``` Logging about excessive wait times will start at 10 minutes.
Previous breakdown when we were using 15 minutes ``` [ 0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 12.8, 25.6, 51.2, 102.4, # 1.7 minutes 204.8, # 3.41 minutes 409.6, # 6.83 minutes 819.2, # 13.65 minutes < logging about excessive times will start here, 13th iteration 900, # 15 minutes < never goes higher than this ] ```
Further suggested work in this area could be to define the cap, the retry interval starting point and the multiplier depending on how frequently this lock should be checked. See data below for reasons why. Increasing the jitter range may also be a good idea --------- Co-authored-by: Eric Eastwood (cherry picked from commit 3f58bc50dfba5768ee43ce48c5e74c25ba0b078a) --- changelog.d/19394.bugfix | 1 + synapse/handlers/worker_lock.py | 117 +++++++++++++++++++------- tests/handlers/test_worker_lock.py | 130 +++++++++++++++++++++++++++++ 3 files changed, 216 insertions(+), 32 deletions(-) create mode 100644 changelog.d/19394.bugfix diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix new file mode 100644 index 0000000000..4ca92cfb32 --- /dev/null +++ b/changelog.d/19394.bugfix @@ -0,0 +1 @@ +Capped the `WorkerLock` time out interval to a maximum of 60 seconds to prevent dealing with excessively long numbers. Contributed by Famedly. diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 1537a18cc0..51be3b5084 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -54,6 +54,9 @@ # will not disappear under our feet as long as we don't delete the room. NEW_EVENT_DURING_PURGE_LOCK_NAME = "new_event_during_purge_lock" +WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(seconds=60) +WORKER_LOCK_EXCESSIVE_WAITING_WARN_DURATION = Duration(minutes=10) + class WorkerLocksHandler: """A class for waiting on taking out locks, rather than using the storage @@ -206,9 +209,10 @@ class WaitingLock: lock_name: str lock_key: str write: bool | None + start_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock: Lock | None = None - _retry_interval: float = 0.1 + _timeout_interval: float = 0.1 _lock_span: "opentracing.Scope" = attr.Factory( lambda: start_active_span("WaitingLock.lock") ) @@ -220,6 +224,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: + self.start_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): @@ -240,19 +245,44 @@ async def __aenter__(self) -> None: break try: - # Wait until the we get notified the lock might have been + # Wait until the notification that the lock might have been # released (by the deferred being resolved). We also - # periodically wake up in case the lock was released but we + # periodically wake up in case the lock was released, but we # weren't notified. with PreserveLoggingContext(): - timeout = self._get_next_retry_interval() await timeout_deferred( deferred=self.deferred, - timeout=timeout, + timeout=self._timeout_interval, clock=self.clock, ) - except Exception: - pass + except defer.TimeoutError: + # Only increment the timeout value if this was an actual timeout + # (defer.TimeoutError) + self._increment_timeout_interval() + + now_ms = self.clock.time_msec() + time_spent_trying_to_lock = Duration( + milliseconds=now_ms - self.start_ts_ms + ) + if ( + time_spent_trying_to_lock.as_millis() + > WORKER_LOCK_EXCESSIVE_WAITING_WARN_DURATION.as_millis() + ): + logger.warning( + "(WaitingLock (%s, %s)) Time spent waiting to acquire lock " + "is getting excessive: %ss. There may be a deadlock.", + self.lock_name, + self.lock_key, + time_spent_trying_to_lock.as_secs(), + ) + + except Exception as e: + logger.warning( + "Caught an exception while waiting on WaitingLock(lock_name=%s, lock_key=%s): %r", + self.lock_name, + self.lock_key, + e, + ) return await self._inner_lock.__aenter__() @@ -273,15 +303,14 @@ async def __aexit__( return r - def _get_next_retry_interval(self) -> float: - next = self._retry_interval - self._retry_interval = max(5, next * 2) - if self._retry_interval > Duration(minutes=10).as_secs(): # >7 iterations - logger.warning( - "Lock timeout is getting excessive: %ss. There may be a deadlock.", - self._retry_interval, - ) - return next * random.uniform(0.9, 1.1) + def _increment_timeout_interval(self) -> float: + next_interval = self._timeout_interval + next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL.as_secs(), next_interval * 2) + + # The jitter value is maintained for the timeout, to help avoid a "thundering + # herd" situation when all locks may time out at the same time. + self._timeout_interval = next_interval * random.uniform(0.9, 1.1) + return self._timeout_interval @attr.s(auto_attribs=True, eq=False) @@ -294,10 +323,11 @@ class WaitingMultiLock: store: LockStore handler: WorkerLocksHandler + start_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock_cm: AsyncContextManager | None = None - _retry_interval: float = 0.1 + _timeout_interval: float = 0.1 _lock_span: "opentracing.Scope" = attr.Factory( lambda: start_active_span("WaitingLock.lock") ) @@ -309,6 +339,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: + self.start_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): @@ -324,19 +355,42 @@ async def __aenter__(self) -> None: break try: - # Wait until the we get notified the lock might have been + # Wait until the notification that the lock might have been # released (by the deferred being resolved). We also - # periodically wake up in case the lock was released but we + # periodically wake up in case the lock was released, but we # weren't notified. with PreserveLoggingContext(): - timeout = self._get_next_retry_interval() await timeout_deferred( deferred=self.deferred, - timeout=timeout, + timeout=self._timeout_interval, clock=self.clock, ) - except Exception: - pass + except defer.TimeoutError: + # Only increment the timeout value if this was an actual timeout + # (defer.TimeoutError) + self._increment_timeout_interval() + + now_ms = self.clock.time_msec() + time_spent_trying_to_lock = Duration( + milliseconds=now_ms - self.start_ts_ms + ) + if ( + time_spent_trying_to_lock.as_millis() + > WORKER_LOCK_EXCESSIVE_WAITING_WARN_DURATION.as_millis() + ): + logger.warning( + "(WaitingMultiLock (%r)) Time spent waiting to acquire lock " + "is getting excessive: %ss. There may be a deadlock.", + self.lock_names, + time_spent_trying_to_lock.as_secs(), + ) + + except Exception as e: + logger.warning( + "Caught an exception while waiting on WaitingMultiLock(lock_names=%r): %r", + self.lock_names, + e, + ) assert self._inner_lock_cm await self._inner_lock_cm.__aenter__() @@ -360,12 +414,11 @@ async def __aexit__( return r - def _get_next_retry_interval(self) -> float: - next = self._retry_interval - self._retry_interval = max(5, next * 2) - if self._retry_interval > Duration(minutes=10).as_secs(): # >7 iterations - logger.warning( - "Lock timeout is getting excessive: %ss. There may be a deadlock.", - self._retry_interval, - ) - return next * random.uniform(0.9, 1.1) + def _increment_timeout_interval(self) -> float: + next_interval = self._timeout_interval + next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL.as_secs(), next_interval * 2) + + # The jitter value is maintained for the timeout, to help avoid a "thundering + # herd" situation when all locks may time out at the same time. + self._timeout_interval = next_interval * random.uniform(0.9, 1.1) + return self._timeout_interval diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 61ff51ff92..74201f4151 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -26,7 +26,9 @@ from twisted.internet.testing import MemoryReactor from synapse.server import HomeServer +from synapse.storage.databases.main.lock import _RENEWAL_INTERVAL from synapse.util.clock import Clock +from synapse.util.duration import Duration from tests import unittest from tests.replication._base import BaseMultiWorkerStreamTestCase @@ -40,6 +42,7 @@ def prepare( self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer ) -> None: self.worker_lock_handler = self.hs.get_worker_locks_handler() + self.store = self.hs.get_datastores().main def test_wait_for_lock_locally(self) -> None: """Test waiting for a lock on a single worker""" @@ -56,6 +59,66 @@ def test_wait_for_lock_locally(self) -> None: self.get_success(d2) self.get_success(lock2.__aexit__(None, None, None)) + def test_timeouts_for_lock_locally(self) -> None: + """ + Test that we regularly retry to reacquire locks. + + This is a regression test to make sure the lock retry time doesn't balloon to a value + so large it can't even be printed reliably anymore. + """ + + # Create and acquire the first lock + lock1 = self.worker_lock_handler.acquire_lock("name", "key") + self.get_success(lock1.__aenter__()) + + # Create and try to acquire the second lock + lock2 = self.worker_lock_handler.acquire_lock("name", "key") + d2 = defer.ensureDeferred(lock2.__aenter__()) + # Make sure we haven't acquired the lock yet (`lock1` still holds it) + self.assertNoResult(d2) + + # Advance time by an hour (some duration that would previously cause our timeout + # to balloon if it weren't constrained). Max back-off (saturate) + # + # Note: We use `_pump_by` instead of `pump`/`advance` as the `Lock` has an + # internal background looping call that runs every 30 seconds + # (`_RENEWAL_INTERVAL`) to renew the `Lock` and push it's "drop timeout" value + # further out by 2 minutes (`_LOCK_TIMEOUT_MS`). The `Lock` will prematurely + # drop if this renewal is not allowed to run, which sours the test. + # self.pump(amount=Duration(hours=1)) + self._pump_by(amount=Duration(hours=1), by=_RENEWAL_INTERVAL) + + # Make sure we haven't acquired the `lock2` yet (`lock1` still holds it) + self.assertNoResult(d2) + + # Release the first lock (`lock1`). The second lock(`lock2`) should be + # automatically acquired by the `pump()` inside `get_success()` + self.get_success(lock1.__aexit__(None, None, None)) + + # We should now have the lock + self.successResultOf(d2) + + def _pump_by( + self, + *, + amount: Duration = Duration(seconds=0), + by: Duration = Duration(seconds=0.1), + ) -> None: + """ + Like `self.pump()` but you can specify the time increment to advance with until + you reach the time amount. + + Unlike `self.pump()`, this doesn't multiply the time at all. + + Args: + amount: The amount of time to advance + by: The time increment in seconds to advance time by until we reach the `amount` + """ + end_time_s = self.reactor.seconds() + amount.as_secs() + + while self.reactor.seconds() < end_time_s: + self.reactor.advance(by.as_secs()) + def test_lock_contention(self) -> None: """Test lock contention when a lot of locks wait on a single worker""" nb_locks_to_test = 500 @@ -124,3 +187,70 @@ def test_wait_for_lock_worker(self) -> None: self.get_success(d2) self.get_success(lock2.__aexit__(None, None, None)) + + def test_timeouts_for_lock_worker(self) -> None: + """ + Test that we regularly retry to reacquire locks. + + This is a regression test to make sure the lock retry time doesn't balloon to a value + so large it can't even be printed reliably anymore. + """ + worker = self.make_worker_hs( + "synapse.app.generic_worker", + extra_config={ + "redis": {"enabled": True}, + }, + ) + worker_lock_handler = worker.get_worker_locks_handler() + + # Create and acquire the first lock on the main process + lock1 = self.main_worker_lock_handler.acquire_lock("name", "key") + self.get_success(lock1.__aenter__()) + + # Create and try to acquire the second lock on the worker + lock2 = worker_lock_handler.acquire_lock("name", "key") + d2 = defer.ensureDeferred(lock2.__aenter__()) + # Make sure we haven't acquired the lock yet (`lock1` still holds it) + self.assertNoResult(d2) + + # Advance time by an hour (some duration that would previously cause our timeout + # to balloon if it weren't constrained). Max back-off (saturate) + # + # Note: We use `_pump_by` instead of `pump`/`advance` as the `Lock` has an + # internal background looping call that runs every 30 seconds + # (`_RENEWAL_INTERVAL`) to renew the `Lock` and push it's "drop timeout" value + # further out by 2 minutes (`_LOCK_TIMEOUT_MS`). The `Lock` will prematurely + # drop if this renewal is not allowed to run, which sours the test. + # self.pump(amount=Duration(hours=1)) + self._pump_by(amount=Duration(hours=1), by=_RENEWAL_INTERVAL) + + # Make sure we haven't acquired the `lock2` yet (`lock1` still holds it) + self.assertNoResult(d2) + + # Release the first lock (`lock1`). The second lock(`lock2`) should be + # automatically acquired by the `pump()` inside `get_success()` + self.get_success(lock1.__aexit__(None, None, None)) + + # We should now have the lock + self.successResultOf(d2) + + def _pump_by( + self, + *, + amount: Duration = Duration(seconds=0), + by: Duration = Duration(seconds=0.1), + ) -> None: + """ + Like `self.pump()` but you can specify the time increment to advance with until + you reach the time amount. + + Unlike `self.pump()`, this doesn't multiply the time at all. + + Args: + amount: The amount of time to advance + by: The time increment in seconds to advance time by until we reach the `amount` + """ + end_time_s = self.reactor.seconds() + amount.as_secs() + + while self.reactor.seconds() < end_time_s: + self.reactor.advance(by.as_secs()) From 2d48851438d41bf6e248c7a743c8ffd61c2e1ba3 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Tue, 5 May 2026 17:12:11 +0100 Subject: [PATCH 2/3] Prevent pagination ending when a page is full of rejected events (ELEMENTSEC-2025-1636) Fixes: https://github.com/element-hq/synapse/security/advisories/GHSA-6qf2-7x63-mm6v Reviewed-on: https://github.com/element-hq/synapse-private/pull/117 --- synapse/handlers/pagination.py | 11 +-- synapse/storage/databases/main/stream.py | 13 +++- tests/rest/client/test_rooms.py | 86 +++++++++++++++++++++++- 3 files changed, 101 insertions(+), 9 deletions(-) diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py index 8cbe4b63c8..2bc7efeb5e 100644 --- a/synapse/handlers/pagination.py +++ b/synapse/handlers/pagination.py @@ -566,7 +566,7 @@ async def get_messages( ( events, next_key, - _, + limited, ) = await self.store.paginate_room_events_by_topological_ordering( room_id=room_id, from_key=from_token.room_key, @@ -645,7 +645,7 @@ async def get_messages( ( events, next_key, - _, + limited, ) = await self.store.paginate_room_events_by_topological_ordering( room_id=room_id, from_key=from_token.room_key, @@ -668,11 +668,12 @@ async def get_messages( next_token = from_token.copy_and_replace(StreamKeyType.ROOM, next_key) - # if no events are returned from pagination, that implies - # we have reached the end of the available events. + # if no events are returned from pagination (this page is empty) + # and there aren't any more pages (not limited), + # that implies we have reached the end of the available events. # In that case we do not return end, to tell the client # there is no need for further queries. - if not events: + if not limited and not events: return GetMessagesResult( messages_chunk=[], bundled_aggregations={}, diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 8fa1e2e5a9..7d14f9f4d8 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -2425,12 +2425,19 @@ async def paginate_room_events_by_topological_ordering( event_filter: If provided filters the events to those that match the filter. Returns: - The results as a list of events, a token that points to the end of - the result set, and a boolean to indicate if there were more events - but we hit the limit. If no events are returned then the end of the + - The results as a list of events; + - a token that points to the end of the result set; and + - a boolean to indicate if there were more events + but we hit the limit (`limited`) + + If no events are returned and `limited` is false, then the end of the stream has been reached (i.e. there are no events between `from_key` and `to_key`). + When `limited` is true, that means that more pagination can be attempted. + Note that `limited` can be true even if no events are returned, + because rejected events are filtered out after the limit check. + When Direction.FORWARDS: from_key < x <= to_key, (ascending order) When Direction.BACKWARDS: from_key >= x > to_key, (descending order) """ diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index 221121007d..61e7e87f62 100644 --- a/tests/rest/client/test_rooms.py +++ b/tests/rest/client/test_rooms.py @@ -66,7 +66,10 @@ from tests import unittest from tests.http.server._base import make_request_with_cancellation_test from tests.storage.test_stream import PaginationTestCase -from tests.test_utils.event_injection import create_event +from tests.test_utils.event_injection import ( + create_event, + inject_event, +) from tests.unittest import override_config from tests.utils import default_config @@ -2371,6 +2374,87 @@ def test_room_message_filter_query_validation(self) -> None: channel.json_body["errcode"], Codes.NOT_JSON, channel.json_body ) + def test_room_messages_paginate_through_rejected_events( + self, + ) -> None: + """Test that pagination continues past a batch of rejected events. + + Regression test for https://github.com/element-hq/synapse/security/advisories/GHSA-6qf2-7x63-mm6v + + Synapse before 1.152.1 had a bug meaning that a batch full of only + rejected events would cause `/messages` to not return any more + pagination tokens, falsely signalling the end of backpagination. + """ + # Send an early message that should not be filtered. + early_event_id = self.helper.send(self.room_id, "early message")["event_id"] + + # Inject a batch of events and mark them as rejected in the database. + # We create more events than a single pagination request would fetch, + # so that one page of backward pagination request would only see rejected events. + for _ in range(3): + event = self.get_success( + inject_event( + self.hs, + room_id=self.room_id, + sender=self.user_id, + type=EventTypes.Message, + content={"body": "filtered event", "msgtype": "m.text"}, + ) + ) + self.get_success( + self.hs.get_datastores().main.db_pool.runInteraction( + "mark_rejected", + self.hs.get_datastores().main.mark_event_rejected_txn, + event.event_id, + "testing", + ) + ) + + # Send a message after all the rejected events. + latest_event_id = self.helper.send(self.room_id, "latest message")["event_id"] + + # Start backpaginating. + channel = self.make_request( + "GET", f"/rooms/{self.room_id}/messages?dir=b&limit=2" + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body) + + events_in_page = [e["event_id"] for e in channel.json_body["chunk"]] + end_token: str | None = channel.json_body["end"] + + self.assertEqual( + events_in_page, + [latest_event_id], + "The latest event should be included in the first page we see whilst backpaginating", + ) + + event_ids_in_pages: list[list[str]] = [events_in_page] + + # Bound the number of backpagination attempts to 2 + for _ in range(2): + channel = self.make_request( + "GET", f"/rooms/{self.room_id}/messages?from={end_token}&dir=b&limit=2" + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body) + events_in_page = [e["event_id"] for e in channel.json_body["chunk"]] + event_ids_in_pages.append(events_in_page) + + if early_event_id in events_in_page: + # We have found the event we were looking for + return + + self.assertIn( + "end", + channel.json_body, + f"No `end` token received. Did not find {early_event_id} whilst backpaginating ({latest_event_id = }, {event_ids_in_pages = })", + ) + # Use the end_token in the next iteration + end_token = channel.json_body["end"] + + self.fail( + f"Exhausted backpagination attempts. Did not find {early_event_id} whilst backpaginating ({latest_event_id = }, {event_ids_in_pages = })" + ) + class RoomMessageFilterTestCase(RoomBase): """Tests /rooms/$room_id/messages REST events.""" From d97b5b9e21681792d78c5d58243fc483c664b66e Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Thu, 7 May 2026 13:49:49 +0100 Subject: [PATCH 3/3] 1.152.1 --- CHANGES.md | 8 ++++++++ changelog.d/19394.bugfix | 1 - debian/changelog | 6 ++++++ pyproject.toml | 2 +- 4 files changed, 15 insertions(+), 2 deletions(-) delete mode 100644 changelog.d/19394.bugfix diff --git a/CHANGES.md b/CHANGES.md index 3425bcca8c..d9b3f8b2c1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,3 +1,11 @@ +# Synapse 1.152.1 (2026-05-07) + +## Security Fixes + +- Prevent CPU starvation (Denial of Service) under worker lock contention, additionally capping the `WorkerLock` time out interval to a maximum of 60 seconds. Contributed by Famedly. ([\#19394](https://github.com/element-hq/synapse/issues/19394), ELEMENTSEC-2026-1706, [GHSA-8q93-326v-3m7g](https://github.com/element-hq/synapse/security/advisories/GHSA-8q93-326v-3m7g), CVE pending) +- Prevent pagination ending when a page is full of rejected events. (ELEMENTSEC-2025-1636, [GHSA-6qf2-7x63-mm6v](https://github.com/element-hq/synapse/security/advisories/GHSA-6qf2-7x63-mm6v), CVE pending) + + # Synapse 1.152.0 (2026-04-28) No significant changes since 1.152.0rc1. diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix deleted file mode 100644 index 4ca92cfb32..0000000000 --- a/changelog.d/19394.bugfix +++ /dev/null @@ -1 +0,0 @@ -Capped the `WorkerLock` time out interval to a maximum of 60 seconds to prevent dealing with excessively long numbers. Contributed by Famedly. diff --git a/debian/changelog b/debian/changelog index ff9dfe3e13..cfefe953e3 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +matrix-synapse-py3 (1.152.1) stable; urgency=medium + + * New Synapse release 1.152.1. + + -- Synapse Packaging team Thu, 07 May 2026 13:29:05 +0100 + matrix-synapse-py3 (1.152.0) stable; urgency=medium * New Synapse release 1.152.0. diff --git a/pyproject.toml b/pyproject.toml index 9509609ff3..e9ae6e204d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "matrix-synapse" -version = "1.152.0" +version = "1.152.1" description = "Homeserver for the Matrix decentralised comms protocol" readme = "README.rst" authors = [