From ff23d0011f30988a043d2f9817d96b9ba483c266 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Fri, 2 Jan 2026 12:48:22 -0600 Subject: [PATCH 01/25] max() and min() were probably switched. Set max to arbitrary 15 minutes, continue logging at durations greater than 10 minutes --- synapse/handlers/worker_lock.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 1537a18cc05..82dd896d5a3 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -275,8 +275,8 @@ async def __aexit__( 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 + self._retry_interval = min(Duration(minutes=15).as_secs(), next * 2) + if self._retry_interval > Duration(minutes=10).as_secs(): # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", self._retry_interval, @@ -362,8 +362,8 @@ async def __aexit__( 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 + self._retry_interval = min(Duration(minutes=15).as_secs(), next * 2) + if self._retry_interval > Duration(minutes=10).as_secs(): # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", self._retry_interval, From 548c85b0fb604718ff52f987defba77ed3d21c10 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 20 Jan 2026 06:42:18 -0600 Subject: [PATCH 02/25] changelog --- changelog.d/19394.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19394.bugfix diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix new file mode 100644 index 00000000000..eb93fffe156 --- /dev/null +++ b/changelog.d/19394.bugfix @@ -0,0 +1 @@ +Prevent excessively long numbers for the retry interval of `WorkerLock`s. Contributed by Famedly. From 1d22f90bd3368caeb620d82f233fa1708eda4811 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Wed, 21 Jan 2026 06:57:34 -0600 Subject: [PATCH 03/25] Update changelog.d/19394.bugfix Co-authored-by: Eric Eastwood --- changelog.d/19394.bugfix | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix index eb93fffe156..3591c0a745c 100644 --- a/changelog.d/19394.bugfix +++ b/changelog.d/19394.bugfix @@ -1 +1 @@ -Prevent excessively long numbers for the retry interval of `WorkerLock`s. Contributed by Famedly. +Capped the `WorkerLock` retry interval to a maximum of 15 minutes to prevent dealing with excessively long numbers. Contributed by Famedly. From e864cfe25f590dd75b31ad6e8ca0dccccf48df60 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Mon, 26 Jan 2026 10:26:38 -0600 Subject: [PATCH 04/25] Adjust for the retry interval actually being a timeout interval, and only increase it when a timeout occurs --- synapse/handlers/worker_lock.py | 56 +++++++++-------- tests/handlers/test_worker_lock.py | 99 ++++++++++++++++++++++++++---- 2 files changed, 120 insertions(+), 35 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 82dd896d5a3..88ecfd6318a 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -208,7 +208,7 @@ class WaitingLock: write: bool | None 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") ) @@ -240,19 +240,23 @@ async def __aenter__(self) -> None: break try: - # Wait until the we get notified the lock might have been + # Wait until the notification 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 interval if this was an actual timeout + self._timeout_interval = self._increment_timeout_interval() + except Exception as e: + logger.warning( + "Caught an exception while waiting on WaitingLock: %r", e + ) return await self._inner_lock.__aenter__() @@ -273,13 +277,13 @@ async def __aexit__( return r - def _get_next_retry_interval(self) -> float: - next = self._retry_interval - self._retry_interval = min(Duration(minutes=15).as_secs(), next * 2) - if self._retry_interval > Duration(minutes=10).as_secs(): # >12 iterations + def _increment_timeout_interval(self) -> float: + next = self._timeout_interval + next = min(Duration(minutes=15).as_secs(), next * 2) + if next > Duration(minutes=10).as_secs(): # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", - self._retry_interval, + next, ) return next * random.uniform(0.9, 1.1) @@ -297,7 +301,7 @@ class WaitingMultiLock: 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") ) @@ -324,19 +328,23 @@ async def __aenter__(self) -> None: break try: - # Wait until the we get notified the lock might have been + # Wait until the notification 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 interval if this was an actual timeout + self._timeout_interval = self._increment_timeout_interval() + except Exception as e: + logger.warning( + "Caught an exception while waiting on WaitingMultiLock: %r", e + ) assert self._inner_lock_cm await self._inner_lock_cm.__aenter__() @@ -360,12 +368,12 @@ async def __aexit__( return r - def _get_next_retry_interval(self) -> float: - next = self._retry_interval - self._retry_interval = min(Duration(minutes=15).as_secs(), next * 2) - if self._retry_interval > Duration(minutes=10).as_secs(): # >12 iterations + def _increment_timeout_interval(self) -> float: + next = self._timeout_interval + next = min(Duration(minutes=15).as_secs(), next * 2) + if next > Duration(minutes=10).as_secs(): # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", - self._retry_interval, + next, ) return next * random.uniform(0.9, 1.1) diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 61ff51ff923..3ae44d48e26 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -21,6 +21,7 @@ import logging import platform +from unittest.mock import patch from twisted.internet import defer from twisted.internet.testing import MemoryReactor @@ -48,13 +49,47 @@ def test_wait_for_lock_locally(self) -> None: self.get_success(lock1.__aenter__()) lock2 = self.worker_lock_handler.acquire_lock("name", "key") - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) - - self.get_success(lock1.__aexit__(None, None, None)) + # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + with patch.object( + lock2, + "_increment_timeout_interval", + wraps=lock2._increment_timeout_interval, + ) as wrapped_lock2_increment_timeout_interval_method: + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) + + # The lock should not time out here + wrapped_lock2_increment_timeout_interval_method.assert_not_called() + self.get_success(lock1.__aexit__(None, None, None)) + + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) + + def test_timeouts_for_lock_locally(self) -> None: + """Test timeouts are incremented for a lock on a single worker""" + lock1 = self.worker_lock_handler.acquire_lock("name", "key") + self.get_success(lock1.__aenter__()) - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + lock2 = self.worker_lock_handler.acquire_lock("name", "key") + # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + with patch.object( + lock2, + "_increment_timeout_interval", + wraps=lock2._increment_timeout_interval, + ) as wrapped_lock2_increment_timeout_interval_method: + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) + + # Recall that pump() will advance time of the given amount 100 times, this + # amounts to about 10 seconds passing + self.pump(10.0) + + # Should be timed out 6 times, but do not fail on that exact count + wrapped_lock2_increment_timeout_interval_method.assert_called() + self.get_success(lock1.__aexit__(None, None, None)) + + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) def test_lock_contention(self) -> None: """Test lock contention when a lot of locks wait on a single worker""" @@ -117,10 +152,52 @@ def test_wait_for_lock_worker(self) -> None: self.get_success(lock1.__aenter__()) lock2 = worker_lock_handler.acquire_lock("name", "key") - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) + # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + with patch.object( + lock2, + "_increment_timeout_interval", + wraps=lock2._increment_timeout_interval, + ) as wrapped_lock2_increment_timeout_interval_method: + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) + + # The lock should not time out here + wrapped_lock2_increment_timeout_interval_method.assert_not_called() + self.get_success(lock1.__aexit__(None, None, None)) + + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) + + def test_timeouts_for_lock_worker(self) -> None: + """Test timeouts are incremented for a lock on another worker""" + worker = self.make_worker_hs( + "synapse.app.generic_worker", + extra_config={ + "redis": {"enabled": True}, + }, + ) + worker_lock_handler = worker.get_worker_locks_handler() - self.get_success(lock1.__aexit__(None, None, None)) + lock1 = self.main_worker_lock_handler.acquire_lock("name", "key") + self.get_success(lock1.__aenter__()) - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + lock2 = worker_lock_handler.acquire_lock("name", "key") + # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + with patch.object( + lock2, + "_increment_timeout_interval", + wraps=lock2._increment_timeout_interval, + ) as wrapped_lock2_increment_timeout_interval_method: + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) + + # Recall that pump() will advance time of the given amount 100 times, this + # amounts to about 10 seconds passing + self.pump(0.1) + + # Should be timed out 6 times, but do not fail on that exact count + wrapped_lock2_increment_timeout_interval_method.assert_called() + self.get_success(lock1.__aexit__(None, None, None)) + + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) From d416dc8c41ab43df323ae4ba16341623c55fff6f Mon Sep 17 00:00:00 2001 From: Jason Little Date: Mon, 26 Jan 2026 11:18:29 -0600 Subject: [PATCH 05/25] unecessarily long pump() in test, left over from testing logging --- tests/handlers/test_worker_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 3ae44d48e26..c5c3ce22efd 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -82,7 +82,7 @@ def test_timeouts_for_lock_locally(self) -> None: # Recall that pump() will advance time of the given amount 100 times, this # amounts to about 10 seconds passing - self.pump(10.0) + self.pump(0.1) # Should be timed out 6 times, but do not fail on that exact count wrapped_lock2_increment_timeout_interval_method.assert_called() From e555cd68f25d6f81f4a6cdaf28ecda74a043ae3e Mon Sep 17 00:00:00 2001 From: Jason Little Date: Mon, 26 Jan 2026 11:18:38 -0600 Subject: [PATCH 06/25] adjust changelog(again) --- changelog.d/19394.bugfix | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix index 3591c0a745c..02131d89baf 100644 --- a/changelog.d/19394.bugfix +++ b/changelog.d/19394.bugfix @@ -1 +1 @@ -Capped the `WorkerLock` retry interval to a maximum of 15 minutes to prevent dealing with excessively long numbers. Contributed by Famedly. +Capped the `WorkerLock` time out interval to a maximum of 15 minutes to prevent dealing with excessively long numbers and prevent logging when the retry is not an actual time out. Contributed by Famedly. From e0d2882c3aff65e377135c0d134263857fba79fa Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 7 Apr 2026 13:28:07 -0500 Subject: [PATCH 07/25] apply some feedback --- synapse/handlers/worker_lock.py | 39 +++++++++++++++++++----------- tests/handlers/test_worker_lock.py | 18 ++++---------- 2 files changed, 30 insertions(+), 27 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 88ecfd6318a..85ee1db7d9f 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(minutes=15).as_secs() +WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10).as_secs() + class WorkerLocksHandler: """A class for waiting on taking out locks, rather than using the storage @@ -240,7 +243,7 @@ async def __aenter__(self) -> None: break try: - # Wait until the notification 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 # weren't notified. @@ -252,10 +255,11 @@ async def __aenter__(self) -> None: ) except defer.TimeoutError: # Only increment the timeout interval if this was an actual timeout - self._timeout_interval = self._increment_timeout_interval() + self._increment_timeout_interval() except Exception as e: logger.warning( - "Caught an exception while waiting on WaitingLock: %r", e + "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__() @@ -278,14 +282,17 @@ async def __aexit__( return r def _increment_timeout_interval(self) -> float: - next = self._timeout_interval - next = min(Duration(minutes=15).as_secs(), next * 2) - if next > Duration(minutes=10).as_secs(): # >12 iterations + _next = self._timeout_interval + _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) + if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", next, ) - return next * random.uniform(0.9, 1.1) + # 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 * random.uniform(0.9, 1.1) + return self._timeout_interval @attr.s(auto_attribs=True, eq=False) @@ -328,7 +335,7 @@ async def __aenter__(self) -> None: break try: - # Wait until the notification 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 # weren't notified. @@ -340,10 +347,11 @@ async def __aenter__(self) -> None: ) except defer.TimeoutError: # Only increment the timeout interval if this was an actual timeout - self._timeout_interval = self._increment_timeout_interval() + self._increment_timeout_interval() except Exception as e: logger.warning( - "Caught an exception while waiting on WaitingMultiLock: %r", e + "Caught an exception while waiting on WaitingMultiLock(lock_names=%r): %r", + self.lock_names, e ) assert self._inner_lock_cm @@ -369,11 +377,14 @@ async def __aexit__( return r def _increment_timeout_interval(self) -> float: - next = self._timeout_interval - next = min(Duration(minutes=15).as_secs(), next * 2) - if next > Duration(minutes=10).as_secs(): # >12 iterations + _next = self._timeout_interval + _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) + if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( "Lock timeout is getting excessive: %ss. There may be a deadlock.", next, ) - return next * random.uniform(0.9, 1.1) + # 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 * 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 c5c3ce22efd..1fd808d70d3 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -49,21 +49,13 @@ def test_wait_for_lock_locally(self) -> None: self.get_success(lock1.__aenter__()) lock2 = self.worker_lock_handler.acquire_lock("name", "key") - # Wrap the WaitingLock object, so we can detect if the timeouts are being hit - with patch.object( - lock2, - "_increment_timeout_interval", - wraps=lock2._increment_timeout_interval, - ) as wrapped_lock2_increment_timeout_interval_method: - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) - # The lock should not time out here - wrapped_lock2_increment_timeout_interval_method.assert_not_called() - self.get_success(lock1.__aexit__(None, None, None)) + self.get_success(lock1.__aexit__(None, None, None)) - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) def test_timeouts_for_lock_locally(self) -> None: """Test timeouts are incremented for a lock on a single worker""" From 7b1ab26d7edfef2c9a732260fa3cf5f15b09a50a Mon Sep 17 00:00:00 2001 From: Jason Little Date: Mon, 27 Apr 2026 11:55:21 -0500 Subject: [PATCH 08/25] lint --- synapse/handlers/worker_lock.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 85ee1db7d9f..a7e893403d5 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -259,7 +259,9 @@ async def __aenter__(self) -> None: 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 + self.lock_name, + self.lock_key, + e, ) return await self._inner_lock.__aenter__() @@ -351,7 +353,8 @@ async def __aenter__(self) -> None: except Exception as e: logger.warning( "Caught an exception while waiting on WaitingMultiLock(lock_names=%r): %r", - self.lock_names, e + self.lock_names, + e, ) assert self._inner_lock_cm From 56372468b0a75eb459b931b7e20d8f7b7c79d49d Mon Sep 17 00:00:00 2001 From: Jason Little Date: Mon, 27 Apr 2026 13:00:32 -0500 Subject: [PATCH 09/25] reword comment and adjust logging warning to contain more info --- synapse/handlers/worker_lock.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index a7e893403d5..65e24db7a21 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -254,7 +254,11 @@ async def __aenter__(self) -> None: clock=self.clock, ) except defer.TimeoutError: - # Only increment the timeout interval if this was an actual timeout + # Actual timeouts are raised with a TimeoutError. Not to be confused + # with another lock being released and broadcasting its notification + # for other locks to retry. This is the circumstance that the retry + # interval should increase, otherwise a series of progressively + # larger timeout interval warnings occur self._increment_timeout_interval() except Exception as e: logger.warning( @@ -288,7 +292,9 @@ def _increment_timeout_interval(self) -> float: _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( - "Lock timeout is getting excessive: %ss. There may be a deadlock.", + "(WaitingLock (%s, %s) Lock timeout is getting excessive: %ss. There may be a deadlock.", + self.lock_name, + self.lock_key, next, ) # The jitter value is maintained for the timeout, to help avoid a "thundering @@ -348,7 +354,11 @@ async def __aenter__(self) -> None: clock=self.clock, ) except defer.TimeoutError: - # Only increment the timeout interval if this was an actual timeout + # Actual timeouts are raised with a TimeoutError. Not to be confused + # with another lock being released and broadcasting its notification + # for other locks to retry. This is the circumstance that the retry + # interval should increase, otherwise a series of progressively + # larger timeout interval warnings occur self._increment_timeout_interval() except Exception as e: logger.warning( @@ -384,8 +394,9 @@ def _increment_timeout_interval(self) -> float: _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( - "Lock timeout is getting excessive: %ss. There may be a deadlock.", - next, + "(WaitingMultiLock (%r) Lock timeout is getting excessive: %ss. There may be a deadlock.", + self.lock_names, + _next, ) # 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. From 3a77b0ddd57bcd128a1db1958f4c1468a3542eb0 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 07:54:56 -0500 Subject: [PATCH 10/25] rename _next to next_interval --- synapse/handlers/worker_lock.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 65e24db7a21..202fa8a23db 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -288,9 +288,9 @@ async def __aexit__( return r def _increment_timeout_interval(self) -> float: - _next = self._timeout_interval - _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) - if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations + next_interval = self._timeout_interval + next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) + if next_interval > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( "(WaitingLock (%s, %s) Lock timeout is getting excessive: %ss. There may be a deadlock.", self.lock_name, @@ -299,7 +299,7 @@ def _increment_timeout_interval(self) -> float: ) # 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 * random.uniform(0.9, 1.1) + self._timeout_interval = next_interval * random.uniform(0.9, 1.1) return self._timeout_interval @@ -390,15 +390,15 @@ async def __aexit__( return r def _increment_timeout_interval(self) -> float: - _next = self._timeout_interval - _next = min(WORKER_LOCK_MAX_RETRY_INTERVAL, _next * 2) - if _next > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations + next_interval = self._timeout_interval + next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) + if next_interval > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations logger.warning( "(WaitingMultiLock (%r) Lock timeout is getting excessive: %ss. There may be a deadlock.", self.lock_names, - _next, + next_interval, ) # 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 * random.uniform(0.9, 1.1) + self._timeout_interval = next_interval * random.uniform(0.9, 1.1) return self._timeout_interval From c734d9f1ed4a67ca98e8dd1b2a36ba9a6591bb6e Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 12:23:08 -0500 Subject: [PATCH 11/25] restore worker model test original --- tests/handlers/test_worker_lock.py | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 1fd808d70d3..1d276b31a7c 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -144,21 +144,13 @@ def test_wait_for_lock_worker(self) -> None: self.get_success(lock1.__aenter__()) lock2 = worker_lock_handler.acquire_lock("name", "key") - # Wrap the WaitingLock object, so we can detect if the timeouts are being hit - with patch.object( - lock2, - "_increment_timeout_interval", - wraps=lock2._increment_timeout_interval, - ) as wrapped_lock2_increment_timeout_interval_method: - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) - # The lock should not time out here - wrapped_lock2_increment_timeout_interval_method.assert_not_called() - self.get_success(lock1.__aexit__(None, None, None)) + self.get_success(lock1.__aexit__(None, None, None)) - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + self.get_success(d2) + self.get_success(lock2.__aexit__(None, None, None)) def test_timeouts_for_lock_worker(self) -> None: """Test timeouts are incremented for a lock on another worker""" From e281f98fb1e43c94d54fd94420c4a570a90f3f6c Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 14:42:39 -0500 Subject: [PATCH 12/25] missed a next being renamed --- synapse/handlers/worker_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 202fa8a23db..0b8f01bcef1 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -295,7 +295,7 @@ def _increment_timeout_interval(self) -> float: "(WaitingLock (%s, %s) Lock timeout is getting excessive: %ss. There may be a deadlock.", self.lock_name, self.lock_key, - next, + next_interval, ) # 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. From 680c49aaa6433b34b7c9545b13d57d669d2565f7 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 15:11:06 -0500 Subject: [PATCH 13/25] restore comment --- synapse/handlers/worker_lock.py | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 0b8f01bcef1..7085871f484 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -254,11 +254,7 @@ async def __aenter__(self) -> None: clock=self.clock, ) except defer.TimeoutError: - # Actual timeouts are raised with a TimeoutError. Not to be confused - # with another lock being released and broadcasting its notification - # for other locks to retry. This is the circumstance that the retry - # interval should increase, otherwise a series of progressively - # larger timeout interval warnings occur + # Only increment the timeout value if this was an actual timeout self._increment_timeout_interval() except Exception as e: logger.warning( @@ -354,11 +350,7 @@ async def __aenter__(self) -> None: clock=self.clock, ) except defer.TimeoutError: - # Actual timeouts are raised with a TimeoutError. Not to be confused - # with another lock being released and broadcasting its notification - # for other locks to retry. This is the circumstance that the retry - # interval should increase, otherwise a series of progressively - # larger timeout interval warnings occur + # Only increment the timeout value if this was an actual timeout self._increment_timeout_interval() except Exception as e: logger.warning( From 5c60cf4d55e390ffbf12daa87dbfb3d5a4f7e7ba Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 16:06:39 -0500 Subject: [PATCH 14/25] stop logging a warning when calculating a timeout interval and start warning when time to acquire a lock exceeds a given time --- synapse/handlers/worker_lock.py | 46 +++++++++++++++++++++++---------- 1 file changed, 32 insertions(+), 14 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 7085871f484..92e5b33285b 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -55,7 +55,7 @@ NEW_EVENT_DURING_PURGE_LOCK_NAME = "new_event_during_purge_lock" WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(minutes=15).as_secs() -WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10).as_secs() +WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10).as_millis() class WorkerLocksHandler: @@ -209,6 +209,7 @@ class WaitingLock: lock_name: str lock_key: str write: bool | None + first_attempt_at_acquiring_lock_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock: Lock | None = None _timeout_interval: float = 0.1 @@ -223,6 +224,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: + self.first_attempt_at_acquiring_lock_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): @@ -263,6 +265,19 @@ async def __aenter__(self) -> None: self.lock_key, e, ) + finally: + now_ms = self.clock.time_msec() + time_spent_trying_to_lock = ( + now_ms - self.first_attempt_at_acquiring_lock_ts_ms + ) + if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: + 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, + ) return await self._inner_lock.__aenter__() @@ -286,13 +301,7 @@ async def __aexit__( def _increment_timeout_interval(self) -> float: next_interval = self._timeout_interval next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) - if next_interval > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations - logger.warning( - "(WaitingLock (%s, %s) Lock timeout is getting excessive: %ss. There may be a deadlock.", - self.lock_name, - self.lock_key, - next_interval, - ) + # 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) @@ -309,6 +318,7 @@ class WaitingMultiLock: store: LockStore handler: WorkerLocksHandler + first_attempt_at_acquiring_lock_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock_cm: AsyncContextManager | None = None @@ -324,6 +334,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: + self.first_attempt_at_acquiring_lock_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): @@ -358,6 +369,18 @@ async def __aenter__(self) -> None: self.lock_names, e, ) + finally: + now_ms = self.clock.time_msec() + time_spent_trying_to_lock = ( + now_ms - self.first_attempt_at_acquiring_lock_ts_ms + ) + if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: + 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, + ) assert self._inner_lock_cm await self._inner_lock_cm.__aenter__() @@ -384,12 +407,7 @@ async def __aexit__( def _increment_timeout_interval(self) -> float: next_interval = self._timeout_interval next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) - if next_interval > WORKER_LOCK_WARN_RETRY_INTERVAL: # >12 iterations - logger.warning( - "(WaitingMultiLock (%r) Lock timeout is getting excessive: %ss. There may be a deadlock.", - self.lock_names, - next_interval, - ) + # 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) From 74f69d53ed18a317b9af0e4c7f83117dc7569404 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 16:06:58 -0500 Subject: [PATCH 15/25] adjust timeout tests --- tests/handlers/test_worker_lock.py | 135 +++++++++++++++++++++++------ 1 file changed, 107 insertions(+), 28 deletions(-) diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 1d276b31a7c..8bcfd8bdaeb 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -26,7 +26,9 @@ from twisted.internet import defer from twisted.internet.testing import MemoryReactor +from synapse.handlers.worker_lock import WORKER_LOCK_MAX_RETRY_INTERVAL from synapse.server import HomeServer +from synapse.storage.databases.main.lock import _LOCK_TIMEOUT_MS from synapse.util.clock import Clock from tests import unittest @@ -41,6 +43,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""" @@ -63,21 +66,59 @@ def test_timeouts_for_lock_locally(self) -> None: self.get_success(lock1.__aenter__()) lock2 = self.worker_lock_handler.acquire_lock("name", "key") - # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + + d2 = defer.ensureDeferred(lock2.__aenter__()) + + self.assertNoResult(d2) + + # Wrap the database call to attempt to acquire the lock, so we can detect how + # many times it is called on. With lock1 having already been entered, any other + # calls to try_acquire_lock() should only be for lock2 with patch.object( - lock2, - "_increment_timeout_interval", - wraps=lock2._increment_timeout_interval, - ) as wrapped_lock2_increment_timeout_interval_method: - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) - - # Recall that pump() will advance time of the given amount 100 times, this - # amounts to about 10 seconds passing - self.pump(0.1) - - # Should be timed out 6 times, but do not fail on that exact count - wrapped_lock2_increment_timeout_interval_method.assert_called() + self.store, + "try_acquire_lock", + wraps=self.store.try_acquire_lock, + ) as wrapped_try_acquire_lock_method: + # A Lock has an internal background looping call that runs every 30 seconds + # to renew the Lock and push it's "drop timeout" value further out by 2 + # minutes. The Lock will prematurely drop if this renewal is not allowed to + # run, which sours the test. + + # Advance time by a bit over 3 hours. _LOCK_TIMEOUT_MS is 2 minutes. Remove + # 1 second from that to give it the barest minimum time to still renew + # itself(the test fails if we do not remove that second and the Lock will + # drop) + # 2 * 60 = 120 seconds + # 120 - 1 = 119 seconds to actually advance + pump_fraction = (_LOCK_TIMEOUT_MS / 1000) - 1 + # 119 * 100 = 11_900 seconds for the entire pump call + # 11_900 / 60 = 180 minutes and 18.33 seconds + self.pump(pump_fraction) # iterates 100 times, see the math above + # The actual Lock should not exist still + assert lock2._inner_lock is None + + wrapped_try_acquire_lock_method.reset_mock() + + # By this point, the timeout on the WaitingLock should be maxed out at + # WORKER_LOCK_MAX_RETRY_INTERVAL. Wait twice that long using pump() so lock1 + # doesn't drop in the background causing an incorrect pass for the test. + # WORKER_LOCK_MAX_RETRY_INTERVAL = 900 seconds + # 900 * 2 = 1800 seconds + # 1800 seconds / 100 iterations = 18 seconds per iteration + pump_fraction = 2 * WORKER_LOCK_MAX_RETRY_INTERVAL / 100 + # In case later adjustments to constants causes a drift in the calculation, + # let future us know this is not necessarily a fault + assert pump_fraction < _LOCK_TIMEOUT_MS, ( + "Please adjust this test, the calculated pump() iteration exceeds the " + f"time the Lock will drop by: {pump_fraction} > {_LOCK_TIMEOUT_MS}" + ) + self.pump(pump_fraction) + + # Should be called 1 or 2 times, there is a jitter to account for + call_count = wrapped_try_acquire_lock_method.call_count + assert 0 < call_count < 3, ( + f"Count of times try_to_acquire() was called was out of presumed bounds(> 0 but < 3): {call_count}" + ) self.get_success(lock1.__aexit__(None, None, None)) self.get_success(d2) @@ -161,26 +202,64 @@ def test_timeouts_for_lock_worker(self) -> None: }, ) worker_lock_handler = worker.get_worker_locks_handler() + worker_data_store = worker.get_datastores().main lock1 = self.main_worker_lock_handler.acquire_lock("name", "key") self.get_success(lock1.__aenter__()) lock2 = worker_lock_handler.acquire_lock("name", "key") - # Wrap the WaitingLock object, so we can detect if the timeouts are being hit + + d2 = defer.ensureDeferred(lock2.__aenter__()) + self.assertNoResult(d2) + + # Wrap the database call to attempt to acquire the lock, so we can detect how + # many times it is called on. With lock1 having already been entered, any other + # calls to try_acquire_lock() should only be for lock2 with patch.object( - lock2, - "_increment_timeout_interval", - wraps=lock2._increment_timeout_interval, - ) as wrapped_lock2_increment_timeout_interval_method: - d2 = defer.ensureDeferred(lock2.__aenter__()) - self.assertNoResult(d2) - - # Recall that pump() will advance time of the given amount 100 times, this - # amounts to about 10 seconds passing - self.pump(0.1) - - # Should be timed out 6 times, but do not fail on that exact count - wrapped_lock2_increment_timeout_interval_method.assert_called() + worker_data_store, + "try_acquire_lock", + wraps=worker_data_store.try_acquire_lock, + ) as wrapped_worker_try_acquire_lock_method: + # A Lock has an internal background looping call that runs every 30 seconds + # to renew the Lock and push it's "drop timeout" value further out by 2 + # minutes. The Lock will prematurely drop if this renewal is not allowed to + # run, which sours the test. + + # Advance time by a bit over 3 hours. _LOCK_TIMEOUT_MS is 2 minutes. Remove + # 1 second from that to give it the barest minimum time to still renew + # itself(the test fails if we do not remove that second and the Lock will + # drop) + # 2 * 60 = 120 seconds + # 120 - 1 = 119 seconds to actually advance + pump_fraction = (_LOCK_TIMEOUT_MS / 1000) - 1 + # 119 * 100 = 11_900 seconds for the entire pump call + # 11_900 / 60 = 180 minutes and 18.33 seconds + self.pump(pump_fraction) # iterates 100 times, see the math above + # The actual Lock should not exist still + assert lock2._inner_lock is None + + wrapped_worker_try_acquire_lock_method.reset_mock() + + # By this point, the timeout on the WaitingLock should be maxed out at + # WORKER_LOCK_MAX_RETRY_INTERVAL. Wait twice that long using pump() so lock1 + # doesn't drop in the background causing an incorrect pass for the test. + # WORKER_LOCK_MAX_RETRY_INTERVAL = 900 seconds + # 900 * 2 = 1800 seconds + # 1800 seconds / 100 iterations = 18 seconds per iteration + pump_fraction = 2 * WORKER_LOCK_MAX_RETRY_INTERVAL / 100 + # In case later adjustments to constants causes a drift in the calculation, + # let future us know this is not necessarily a fault + assert pump_fraction < _LOCK_TIMEOUT_MS, ( + "Please adjust this test, the calculated pump() iteration exceeds the " + f"time the Lock will drop by: {pump_fraction} > {_LOCK_TIMEOUT_MS}" + ) + self.pump(pump_fraction) + + # Should be called 1 or 2 times, there is a jitter to account for + call_count = wrapped_worker_try_acquire_lock_method.call_count + assert 0 < call_count < 3, ( + f"Count of times try_to_acquire() was called was out of presumed bounds(> 0 but < 3): {call_count}" + ) self.get_success(lock1.__aexit__(None, None, None)) self.get_success(d2) From 35386265e846891f844543b8a10fe0ed1459eea5 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Thu, 30 Apr 2026 16:26:29 -0500 Subject: [PATCH 16/25] start capping timeout for retry at 60 seconds --- synapse/handlers/worker_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 92e5b33285b..1fb357e5e46 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -54,7 +54,7 @@ # 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(minutes=15).as_secs() +WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(seconds=60).as_secs() WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10).as_millis() From 701adc8c536412c2c56fa15f082f0ae907da8893 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Fri, 1 May 2026 07:52:21 -0500 Subject: [PATCH 17/25] Update synapse/handlers/worker_lock.py Co-authored-by: Eric Eastwood --- synapse/handlers/worker_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 1fb357e5e46..7a02423c0be 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -334,7 +334,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: - self.first_attempt_at_acquiring_lock_ts_ms = self.clock.time_msec() + self.start_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): From 5d7d5f2cfe7e39232e368a18e84e380acd99a138 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Fri, 1 May 2026 07:56:05 -0500 Subject: [PATCH 18/25] follow up on renaming variable suggestion --- synapse/handlers/worker_lock.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 7a02423c0be..fbfd8ce5cfd 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -209,7 +209,7 @@ class WaitingLock: lock_name: str lock_key: str write: bool | None - first_attempt_at_acquiring_lock_ts_ms: int = 0 + start_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock: Lock | None = None _timeout_interval: float = 0.1 @@ -224,7 +224,7 @@ def release_lock(self) -> None: self.deferred.callback(None) async def __aenter__(self) -> None: - self.first_attempt_at_acquiring_lock_ts_ms = self.clock.time_msec() + self.start_ts_ms = self.clock.time_msec() self._lock_span.__enter__() with start_active_span("WaitingLock.waiting_for_lock"): @@ -268,7 +268,7 @@ async def __aenter__(self) -> None: finally: now_ms = self.clock.time_msec() time_spent_trying_to_lock = ( - now_ms - self.first_attempt_at_acquiring_lock_ts_ms + now_ms - self.start_ts_ms ) if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: logger.warning( @@ -318,7 +318,7 @@ class WaitingMultiLock: store: LockStore handler: WorkerLocksHandler - first_attempt_at_acquiring_lock_ts_ms: int = 0 + start_ts_ms: int = 0 deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred) _inner_lock_cm: AsyncContextManager | None = None @@ -372,7 +372,7 @@ async def __aenter__(self) -> None: finally: now_ms = self.clock.time_msec() time_spent_trying_to_lock = ( - now_ms - self.first_attempt_at_acquiring_lock_ts_ms + now_ms - self.start_ts_ms ) if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: logger.warning( From 42b60500562020bc50d4fab667eee23da1cffbff Mon Sep 17 00:00:00 2001 From: Jason Little Date: Fri, 1 May 2026 07:56:54 -0500 Subject: [PATCH 19/25] change constants for retry intervals into basic Durations --- synapse/handlers/worker_lock.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index fbfd8ce5cfd..5e3b9be4c8f 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -54,8 +54,8 @@ # 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).as_secs() -WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10).as_millis() +WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(seconds=60) +WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10) class WorkerLocksHandler: @@ -270,7 +270,7 @@ async def __aenter__(self) -> None: time_spent_trying_to_lock = ( now_ms - self.start_ts_ms ) - if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: + if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis(): logger.warning( "(WaitingLock (%s, %s)) Time spent waiting to acquire lock " "is getting excessive: %ss. There may be a deadlock.", @@ -300,7 +300,7 @@ async def __aexit__( def _increment_timeout_interval(self) -> float: next_interval = self._timeout_interval - next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) + 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. @@ -374,7 +374,7 @@ async def __aenter__(self) -> None: time_spent_trying_to_lock = ( now_ms - self.start_ts_ms ) - if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL: + if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis(): logger.warning( "(WaitingMultiLock (%r)) Time spent waiting to acquire lock " "is getting excessive: %ss. There may be a deadlock.", @@ -406,7 +406,7 @@ async def __aexit__( def _increment_timeout_interval(self) -> float: next_interval = self._timeout_interval - next_interval = min(WORKER_LOCK_MAX_RETRY_INTERVAL, next_interval * 2) + 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. From b70da34e52f807fadb937af805f8ac0bc51b9f64 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Sat, 2 May 2026 21:03:51 -0500 Subject: [PATCH 20/25] move warning to TimeoutError exception block(and linting too I guess) --- synapse/handlers/worker_lock.py | 50 ++++++++++++++++++--------------- 1 file changed, 27 insertions(+), 23 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 5e3b9be4c8f..5d5dc95db96 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -258,19 +258,13 @@ async def __aenter__(self) -> None: except defer.TimeoutError: # Only increment the timeout value if this was an actual timeout self._increment_timeout_interval() - 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, - ) - finally: + now_ms = self.clock.time_msec() - time_spent_trying_to_lock = ( - now_ms - self.start_ts_ms - ) - if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis(): + time_spent_trying_to_lock = now_ms - self.start_ts_ms + if ( + time_spent_trying_to_lock + > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis() + ): logger.warning( "(WaitingLock (%s, %s)) Time spent waiting to acquire lock " "is getting excessive: %ss. There may be a deadlock.", @@ -279,6 +273,14 @@ async def __aenter__(self) -> None: time_spent_trying_to_lock, ) + 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__() async def __aexit__( @@ -363,18 +365,13 @@ async def __aenter__(self) -> None: except defer.TimeoutError: # Only increment the timeout value if this was an actual timeout self._increment_timeout_interval() - except Exception as e: - logger.warning( - "Caught an exception while waiting on WaitingMultiLock(lock_names=%r): %r", - self.lock_names, - e, - ) - finally: + now_ms = self.clock.time_msec() - time_spent_trying_to_lock = ( - now_ms - self.start_ts_ms - ) - if time_spent_trying_to_lock > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis(): + time_spent_trying_to_lock = now_ms - self.start_ts_ms + if ( + time_spent_trying_to_lock + > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis() + ): logger.warning( "(WaitingMultiLock (%r)) Time spent waiting to acquire lock " "is getting excessive: %ss. There may be a deadlock.", @@ -382,6 +379,13 @@ async def __aenter__(self) -> None: time_spent_trying_to_lock, ) + 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__() return From e12bb0842f08378d0201c7075b2ac4a1c40dac54 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Sat, 2 May 2026 21:06:24 -0500 Subject: [PATCH 21/25] adjust tests --- tests/handlers/test_worker_lock.py | 206 ++++++++++++++--------------- 1 file changed, 98 insertions(+), 108 deletions(-) diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 8bcfd8bdaeb..4891819ee64 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -21,15 +21,14 @@ import logging import platform -from unittest.mock import patch from twisted.internet import defer from twisted.internet.testing import MemoryReactor -from synapse.handlers.worker_lock import WORKER_LOCK_MAX_RETRY_INTERVAL from synapse.server import HomeServer -from synapse.storage.databases.main.lock import _LOCK_TIMEOUT_MS +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 @@ -61,68 +60,64 @@ def test_wait_for_lock_locally(self) -> None: self.get_success(lock2.__aexit__(None, None, None)) def test_timeouts_for_lock_locally(self) -> None: - """Test timeouts are incremented for a lock on a single worker""" + """ + 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 a day (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(days=1)) + self._pump_by(amount=Duration(days=1), by=_RENEWAL_INTERVAL) + + # Make sure we haven't acquired the `lock2` yet (`lock1` still holds it) self.assertNoResult(d2) - # Wrap the database call to attempt to acquire the lock, so we can detect how - # many times it is called on. With lock1 having already been entered, any other - # calls to try_acquire_lock() should only be for lock2 - with patch.object( - self.store, - "try_acquire_lock", - wraps=self.store.try_acquire_lock, - ) as wrapped_try_acquire_lock_method: - # A Lock has an internal background looping call that runs every 30 seconds - # to renew the Lock and push it's "drop timeout" value further out by 2 - # minutes. The Lock will prematurely drop if this renewal is not allowed to - # run, which sours the test. - - # Advance time by a bit over 3 hours. _LOCK_TIMEOUT_MS is 2 minutes. Remove - # 1 second from that to give it the barest minimum time to still renew - # itself(the test fails if we do not remove that second and the Lock will - # drop) - # 2 * 60 = 120 seconds - # 120 - 1 = 119 seconds to actually advance - pump_fraction = (_LOCK_TIMEOUT_MS / 1000) - 1 - # 119 * 100 = 11_900 seconds for the entire pump call - # 11_900 / 60 = 180 minutes and 18.33 seconds - self.pump(pump_fraction) # iterates 100 times, see the math above - # The actual Lock should not exist still - assert lock2._inner_lock is None - - wrapped_try_acquire_lock_method.reset_mock() - - # By this point, the timeout on the WaitingLock should be maxed out at - # WORKER_LOCK_MAX_RETRY_INTERVAL. Wait twice that long using pump() so lock1 - # doesn't drop in the background causing an incorrect pass for the test. - # WORKER_LOCK_MAX_RETRY_INTERVAL = 900 seconds - # 900 * 2 = 1800 seconds - # 1800 seconds / 100 iterations = 18 seconds per iteration - pump_fraction = 2 * WORKER_LOCK_MAX_RETRY_INTERVAL / 100 - # In case later adjustments to constants causes a drift in the calculation, - # let future us know this is not necessarily a fault - assert pump_fraction < _LOCK_TIMEOUT_MS, ( - "Please adjust this test, the calculated pump() iteration exceeds the " - f"time the Lock will drop by: {pump_fraction} > {_LOCK_TIMEOUT_MS}" - ) - self.pump(pump_fraction) + # 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)) - # Should be called 1 or 2 times, there is a jitter to account for - call_count = wrapped_try_acquire_lock_method.call_count - assert 0 < call_count < 3, ( - f"Count of times try_to_acquire() was called was out of presumed bounds(> 0 but < 3): {call_count}" - ) - 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() - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + 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""" @@ -194,7 +189,12 @@ def test_wait_for_lock_worker(self) -> None: self.get_success(lock2.__aexit__(None, None, None)) def test_timeouts_for_lock_worker(self) -> None: - """Test timeouts are incremented for a lock on another worker""" + """ + 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={ @@ -202,65 +202,55 @@ def test_timeouts_for_lock_worker(self) -> None: }, ) worker_lock_handler = worker.get_worker_locks_handler() - worker_data_store = worker.get_datastores().main + # 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) - # Wrap the database call to attempt to acquire the lock, so we can detect how - # many times it is called on. With lock1 having already been entered, any other - # calls to try_acquire_lock() should only be for lock2 - with patch.object( - worker_data_store, - "try_acquire_lock", - wraps=worker_data_store.try_acquire_lock, - ) as wrapped_worker_try_acquire_lock_method: - # A Lock has an internal background looping call that runs every 30 seconds - # to renew the Lock and push it's "drop timeout" value further out by 2 - # minutes. The Lock will prematurely drop if this renewal is not allowed to - # run, which sours the test. - - # Advance time by a bit over 3 hours. _LOCK_TIMEOUT_MS is 2 minutes. Remove - # 1 second from that to give it the barest minimum time to still renew - # itself(the test fails if we do not remove that second and the Lock will - # drop) - # 2 * 60 = 120 seconds - # 120 - 1 = 119 seconds to actually advance - pump_fraction = (_LOCK_TIMEOUT_MS / 1000) - 1 - # 119 * 100 = 11_900 seconds for the entire pump call - # 11_900 / 60 = 180 minutes and 18.33 seconds - self.pump(pump_fraction) # iterates 100 times, see the math above - # The actual Lock should not exist still - assert lock2._inner_lock is None - - wrapped_worker_try_acquire_lock_method.reset_mock() - - # By this point, the timeout on the WaitingLock should be maxed out at - # WORKER_LOCK_MAX_RETRY_INTERVAL. Wait twice that long using pump() so lock1 - # doesn't drop in the background causing an incorrect pass for the test. - # WORKER_LOCK_MAX_RETRY_INTERVAL = 900 seconds - # 900 * 2 = 1800 seconds - # 1800 seconds / 100 iterations = 18 seconds per iteration - pump_fraction = 2 * WORKER_LOCK_MAX_RETRY_INTERVAL / 100 - # In case later adjustments to constants causes a drift in the calculation, - # let future us know this is not necessarily a fault - assert pump_fraction < _LOCK_TIMEOUT_MS, ( - "Please adjust this test, the calculated pump() iteration exceeds the " - f"time the Lock will drop by: {pump_fraction} > {_LOCK_TIMEOUT_MS}" - ) - self.pump(pump_fraction) + # Advance time by a day (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(days=1)) + self._pump_by(amount=Duration(days=1), by=_RENEWAL_INTERVAL) + + # Make sure we haven't acquired the `lock2` yet (`lock1` still holds it) + self.assertNoResult(d2) - # Should be called 1 or 2 times, there is a jitter to account for - call_count = wrapped_worker_try_acquire_lock_method.call_count - assert 0 < call_count < 3, ( - f"Count of times try_to_acquire() was called was out of presumed bounds(> 0 but < 3): {call_count}" - ) - self.get_success(lock1.__aexit__(None, None, None)) + # 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() - self.get_success(d2) - self.get_success(lock2.__aexit__(None, None, None)) + while self.reactor.seconds() < end_time_s: + self.reactor.advance(by.as_secs()) From 36848ced10d4c64541d87b9b4be74e1e93400ada Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 5 May 2026 05:31:19 -0500 Subject: [PATCH 22/25] Update changelog.d/19394.bugfix Co-authored-by: Eric Eastwood --- changelog.d/19394.bugfix | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog.d/19394.bugfix b/changelog.d/19394.bugfix index 02131d89baf..4ca92cfb328 100644 --- a/changelog.d/19394.bugfix +++ b/changelog.d/19394.bugfix @@ -1 +1 @@ -Capped the `WorkerLock` time out interval to a maximum of 15 minutes to prevent dealing with excessively long numbers and prevent logging when the retry is not an actual time out. Contributed by Famedly. +Capped the `WorkerLock` time out interval to a maximum of 60 seconds to prevent dealing with excessively long numbers. Contributed by Famedly. From d8a1c09ba2e0bfb8af6c9239389a6b4946998a9a Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 5 May 2026 05:36:39 -0500 Subject: [PATCH 23/25] feedback --- synapse/handlers/worker_lock.py | 22 +++++++++++++--------- tests/handlers/test_worker_lock.py | 12 ++++++------ 2 files changed, 19 insertions(+), 15 deletions(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 5d5dc95db96..83f4fc598cc 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -55,7 +55,7 @@ NEW_EVENT_DURING_PURGE_LOCK_NAME = "new_event_during_purge_lock" WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(seconds=60) -WORKER_LOCK_WARN_RETRY_INTERVAL = Duration(minutes=10) +WORKER_LOCK_EXCESSIVE_WAITING_WARN_DURATION = Duration(minutes=10) class WorkerLocksHandler: @@ -257,20 +257,23 @@ async def __aenter__(self) -> None: ) 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 = now_ms - self.start_ts_ms + time_spent_trying_to_lock = Duration( + milliseconds=now_ms - self.start_ts_ms + ) if ( - time_spent_trying_to_lock - > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis() + 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, + time_spent_trying_to_lock.as_secs(), ) except Exception as e: @@ -364,19 +367,20 @@ async def __aenter__(self) -> None: ) 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 = now_ms - self.start_ts_ms + time_spent_trying_to_lock = Duration(now_ms - self.start_ts_ms) if ( - time_spent_trying_to_lock - > WORKER_LOCK_WARN_RETRY_INTERVAL.as_millis() + 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, + time_spent_trying_to_lock.as_secs(), ) except Exception as e: diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index 4891819ee64..74201f41515 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -77,7 +77,7 @@ def test_timeouts_for_lock_locally(self) -> None: # Make sure we haven't acquired the lock yet (`lock1` still holds it) self.assertNoResult(d2) - # Advance time by a day (some duration that would previously cause our timeout + # 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 @@ -85,8 +85,8 @@ def test_timeouts_for_lock_locally(self) -> None: # (`_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(days=1)) - self._pump_by(amount=Duration(days=1), by=_RENEWAL_INTERVAL) + # 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) @@ -213,7 +213,7 @@ def test_timeouts_for_lock_worker(self) -> None: # Make sure we haven't acquired the lock yet (`lock1` still holds it) self.assertNoResult(d2) - # Advance time by a day (some duration that would previously cause our timeout + # 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 @@ -221,8 +221,8 @@ def test_timeouts_for_lock_worker(self) -> None: # (`_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(days=1)) - self._pump_by(amount=Duration(days=1), by=_RENEWAL_INTERVAL) + # 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) From 90402c8a331dc6d0794a6e77b4dc591085f6c260 Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 5 May 2026 07:35:55 -0500 Subject: [PATCH 24/25] correct kwarg that was missing --- synapse/handlers/worker_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 83f4fc598cc..db0c94b2712 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -371,7 +371,7 @@ async def __aenter__(self) -> None: self._increment_timeout_interval() now_ms = self.clock.time_msec() - time_spent_trying_to_lock = Duration(now_ms - self.start_ts_ms) + 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() From a6d5b68a8ef925c9b5632317bcfdcd8beaf4951b Mon Sep 17 00:00:00 2001 From: Jason Little Date: Tue, 5 May 2026 07:38:02 -0500 Subject: [PATCH 25/25] lint --- synapse/handlers/worker_lock.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index db0c94b2712..51be3b5084a 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -371,7 +371,9 @@ async def __aenter__(self) -> None: self._increment_timeout_interval() now_ms = self.clock.time_msec() - time_spent_trying_to_lock = Duration(milliseconds=now_ms - self.start_ts_ms) + 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()