Skip to content

Update WorkerLock tests to better stress the WORKER_LOCK_MAX_RETRY_INTERVAL #19772

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MadLittleMods merged 5 commits into from
May 12, 2026
Merged

Update WorkerLock tests to better stress the WORKER_LOCK_MAX_RETRY_INTERVAL #19772

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f24eafb
Update `_LOCK_TIMEOUT` to be a `Duration`
MadLittleMods May 11, 2026
e90a919
Better test that actually stresses retry timeout interval
MadLittleMods May 11, 2026
f3be850
Apply treatment to other test
MadLittleMods May 11, 2026
dcd3591
Add changelog
MadLittleMods May 11, 2026
1b57897
Update understanding on how locks notified about releases
MadLittleMods May 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/19772.misc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Update `WorkerLock` tests to better stress the `WORKER_LOCK_MAX_RETRY_INTERVAL`.
8 changes: 4 additions & 4 deletions synapse/handlers/worker_lock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,10 @@
Better to retry more quickly than have workers wait around. 5 seconds is still a
reasonable gap in time to not overwhelm the CPU/Database.

This matters most in cross-worker scenarios. When locks are on the same worker, when the
lock holder releases, we signal to other locks (with the same name/key) that they
should try reacquiring the lock immediately. But locks on other workers only re-check
based on their retry `_timeout_interval`.
This matters most when locks go stale as normally, when the lock holder releases, we
signal to other locks (with the same name/key) that they should try reacquiring the lock
immediately. But stale locks are never released and instead forcefully reaped behind the
scenes.
Comment on lines +64 to +67

@MadLittleMods MadLittleMods May 11, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The original reasoning here came from #19755

It was based on my flawed understanding on how the lock release notifications worked. It turns out we also notify_lock_released(...) over replication when other workers tell us about it.

"""
WORKER_LOCK_EXCESSIVE_WAITING_WARN_DURATION = Duration(minutes=10)

Expand Down
14 changes: 8 additions & 6 deletions synapse/storage/databases/main/lock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,17 +53,17 @@
_RENEWAL_INTERVAL = Duration(seconds=30)

# How long before an acquired lock times out.
_LOCK_TIMEOUT_MS = 2 * 60 * 1000
_LOCK_TIMEOUT = Duration(minutes=2)

@MadLittleMods MadLittleMods May 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a refactor to use Duration for _LOCK_TIMEOUT (no behavioral change)


_LOCK_REAP_INTERVAL = Duration(milliseconds=_LOCK_TIMEOUT_MS / 10.0)
_LOCK_REAP_INTERVAL = Duration(milliseconds=_LOCK_TIMEOUT.as_millis() / 10.0)


class LockStore(SQLBaseStore):
"""Provides a best effort distributed lock between worker instances.

Locks are identified by a name and key. A lock is acquired by inserting into
the `worker_locks` table if a) there is no existing row for the name/key or
b) the existing row has a `last_renewed_ts` older than `_LOCK_TIMEOUT_MS`.
b) the existing row has a `last_renewed_ts` older than `_LOCK_TIMEOUT`.

When a lock is taken out the instance inserts a random `token`, the instance
that holds that token holds the lock until it drops (or times out).
Expand Down Expand Up @@ -182,7 +182,7 @@ def _try_acquire_lock_txn(txn: LoggingTransaction) -> bool:
self._instance_name,
token,
now,
now - _LOCK_TIMEOUT_MS,
now - _LOCK_TIMEOUT.as_millis(),
),
)

Expand Down Expand Up @@ -340,7 +340,9 @@ async def _reap_stale_read_write_locks(self) -> None:
"""

def reap_stale_read_write_locks_txn(txn: LoggingTransaction) -> None:
txn.execute(delete_sql, (self.clock.time_msec() - _LOCK_TIMEOUT_MS,))
txn.execute(
delete_sql, (self.clock.time_msec() - _LOCK_TIMEOUT.as_millis(),)
)
if txn.rowcount:
logger.info("Reaped %d stale locks", txn.rowcount)

Expand Down Expand Up @@ -489,7 +491,7 @@ async def is_still_valid(self) -> bool:
)
return (
last_renewed_ts is not None
and self._clock.time_msec() - _LOCK_TIMEOUT_MS < last_renewed_ts
and self._clock.time_msec() - _LOCK_TIMEOUT.as_millis() < last_renewed_ts
)

async def __aenter__(self) -> None:
Expand Down
73 changes: 64 additions & 9 deletions tests/handlers/test_worker_lock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,13 @@
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 _RENEWAL_INTERVAL
from synapse.storage.databases.main.lock import (
_LOCK_REAP_INTERVAL,
_LOCK_TIMEOUT,
_RENEWAL_INTERVAL,
)
from synapse.util.clock import Clock
from synapse.util.duration import Duration

Expand Down Expand Up @@ -83,17 +88,42 @@ def test_timeouts_for_lock_locally(self) -> None:
# 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
# further out by 2 minutes (`_LOCK_TIMEOUT`). 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))

@MadLittleMods MadLittleMods May 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

# [...] The second lock(`lock2`) should be
# automatically acquired by the `pump()` inside `get_success()`

Basically, the behavior described by this comment circumvents the retry timeout interval logic we're trying to stress. And the previous tests actually pass without any of the fixes from #19394 because of this happy-path flow.

To explain further: When a lock is released, we immediately try to re-acquire the lock again

Notifier.notify_lock_released(...) -> calls any callbacks registered from Notifier.add_lock_released_callback(...) -> which we do in WorkerLocksHandler and will call release_lock() which resolves the deferred and wakes up the timeout_deferred(...) and loops around the while-loop again which tries to re-acquire the lock.

Instead, we want to avoid the lock released notification stuff and stress the retry interval which helps in situations where the lock holder goes stale, is reaped, and the other locks want to try to acquire the lock.

I've tested to make sure these new tests fail with a version of Synapse before #19394

  1. git checkout v1.152.0
  2. Paste the latest tests/handlers/test_worker_lock.py into the codebase
  3. Shim a couple values that don't exist in that Synapse version: 
    WORKER_LOCK_MAX_RETRY_INTERVAL = Duration(seconds=5)
_LOCK_TIMEOUT = Duration(minutes=2)
  4. poetry install --extras all
  5. SYNAPSE_POSTGRES=1 SYNAPSE_POSTGRES_USER=postgres SYNAPSE_TEST_LOG_LEVEL=INFO poetry run trial tests.handlers.test_worker_lock
  6. Notice the tests fail as expected: 
    tests.handlers.test_worker_lock
  WorkerLockTestCase
    test_lock_contention ...                                               [OK]
    test_timeouts_for_lock_locally ...                                   [FAIL]
    test_wait_for_lock_locally ...                                         [OK]
  WorkerLockWorkersTestCase
    test_timeouts_for_lock_worker ...                                    [FAIL]
    test_wait_for_lock_worker ...                                          [OK]

# Drop the lock without releasing it. If we just normally released the lock
# (`self.get_success(lock1.__aexit__(None, None, None))`), the
# `add_lock_released_callback`/`notify_lock_released` cycle would signal that we
# should re-aquire the lock right away (on the next reactor tick). And we want
# to avoid that as the point of this test is to stress the retry timeout
# interval and `WORKER_LOCK_MAX_RETRY_INTERVAL`.
del lock1

# Wait for `lock1` to go stale (it won't be renewed anymore because we deleted
# it just above)
self._pump_by(
amount=_LOCK_TIMEOUT,
by=_RENEWAL_INTERVAL,
)

# Wait just enough time so `lock1` is reaped (found stale and forcefully drops
# the lock its holding)
self._pump_by(
amount=_LOCK_REAP_INTERVAL,
by=_RENEWAL_INTERVAL,
)

# Wait just enough time so `lock2` tries re-acquiring the lock. Should be no
# longer than our `WORKER_LOCK_MAX_RETRY_INTERVAL`.
self._pump_by(
amount=WORKER_LOCK_MAX_RETRY_INTERVAL,
by=_RENEWAL_INTERVAL,
)

# We should now have the lock
self.successResultOf(d2)
Expand Down Expand Up @@ -219,17 +249,42 @@ def test_timeouts_for_lock_worker(self) -> None:
# 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
# further out by 2 minutes (`_LOCK_TIMEOUT`). 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))
# Drop the lock without releasing it. If we just normally released the lock
# (`self.get_success(lock1.__aexit__(None, None, None))`), the
# `add_lock_released_callback`/`notify_lock_released` cycle would signal that we
# should re-aquire the lock right away (on the next reactor tick). And we want
# to avoid that as the point of this test is to stress the retry timeout
# interval and `WORKER_LOCK_MAX_RETRY_INTERVAL`.
del lock1

# Wait for `lock1` to go stale (it won't be renewed anymore because we deleted
# it just above)
self._pump_by(
amount=_LOCK_TIMEOUT,
by=_RENEWAL_INTERVAL,
)

# Wait just enough time so `lock1` is reaped (found stale and forcefully drops
# the lock its holding)
self._pump_by(
amount=_LOCK_REAP_INTERVAL,
by=_RENEWAL_INTERVAL,
)

# Wait just enough time so `lock2` tries re-acquiring the lock. Should be no
# longer than our `WORKER_LOCK_MAX_RETRY_INTERVAL`.
self._pump_by(
amount=WORKER_LOCK_MAX_RETRY_INTERVAL,
by=_RENEWAL_INTERVAL,
)

# We should now have the lock
self.successResultOf(d2)
Expand Down
12 changes: 6 additions & 6 deletions tests/storage/databases/main/test_lock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
from twisted.internet.testing import MemoryReactor

from synapse.server import HomeServer
from synapse.storage.databases.main.lock import _LOCK_TIMEOUT_MS, _RENEWAL_INTERVAL
from synapse.storage.databases.main.lock import _LOCK_TIMEOUT, _RENEWAL_INTERVAL
from synapse.util.clock import Clock

from tests import unittest
Expand Down Expand Up @@ -117,7 +117,7 @@ def test_maintain_lock(self) -> None:
self.get_success(lock.__aenter__())

# Wait for ages with the lock, we should not be able to get the lock.
self.reactor.advance(5 * _LOCK_TIMEOUT_MS / 1000)
self.reactor.advance(5 * _LOCK_TIMEOUT.as_secs())

lock2 = self.get_success(self.store.try_acquire_lock("name", "key"))
self.assertIsNone(lock2)
Expand All @@ -138,7 +138,7 @@ def test_timeout_lock(self) -> None:
lock._looping_call.stop()

# Wait for the lock to timeout.
self.reactor.advance(2 * _LOCK_TIMEOUT_MS / 1000)
self.reactor.advance(2 * _LOCK_TIMEOUT.as_secs())

lock2 = self.get_success(self.store.try_acquire_lock("name", "key"))
self.assertIsNotNone(lock2)
Expand All @@ -154,7 +154,7 @@ def test_drop(self) -> None:
del lock

# Wait for the lock to timeout.
self.reactor.advance(2 * _LOCK_TIMEOUT_MS / 1000)
self.reactor.advance(2 * _LOCK_TIMEOUT.as_secs())

lock2 = self.get_success(self.store.try_acquire_lock("name", "key"))
self.assertIsNotNone(lock2)
Expand Down Expand Up @@ -402,7 +402,7 @@ def test_timeout_lock(self) -> None:
lock._looping_call.stop()

# Wait for the lock to timeout.
self.reactor.advance(2 * _LOCK_TIMEOUT_MS / 1000)
self.reactor.advance(2 * _LOCK_TIMEOUT.as_secs())

lock2 = self.get_success(
self.store.try_acquire_read_write_lock("name", "key", write=True)
Expand All @@ -422,7 +422,7 @@ def test_drop(self) -> None:
del lock

# Wait for the lock to timeout.
self.reactor.advance(2 * _LOCK_TIMEOUT_MS / 1000)
self.reactor.advance(2 * _LOCK_TIMEOUT.as_secs())

lock2 = self.get_success(
self.store.try_acquire_read_write_lock("name", "key", write=True)
Expand Down
Loading