From b4deb24be6754e4cbec0c0cc8dff24802e689475 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 9 Jul 2026 10:40:53 +0100 Subject: [PATCH 1/6] Use a CPU-time budget for `test_lock_contention` to fix postgres flakiness (#19929) `test_lock_contention` is a performance-regression canary (#16840): the pathological behaviour it guards against spent ~30s spinning the CPU, vs ~0.5s when healthy. The 5s wall-clock alarm it used was calibrated on SQLite, but against PostgreSQL a healthy run already takes 3-4s of wall-clock time (500 sequential acquire/release cycles, each a real database round-trip), so any CI load pushed it over the limit. Add a `cpu_time` mode to `tests/utils.py`'s test_timeout, implemented with [`setitimer(ITIMER_PROF)`](https://docs.python.org/3/library/signal.html#signal.setitimer), which budgets process CPU time instead of wall-clock time. Time spent blocked on the database or lost to a loaded CI runner no longer counts, while a regression to CPU-spinning still trips the alarm mid-spin. A healthy run costs <1s of CPU on either database engine; the budget is 10s. This also subsumes the RISC-V wall-clock carve-out from #18430, which is removed. --------- Co-authored-by: Claude Fable 5 --- changelog.d/19929.misc | 1 + tests/handlers/test_worker_lock.py | 37 ++++++++---------------------- tests/utils.py | 34 +++++++++++++++++++++++---- 3 files changed, 40 insertions(+), 32 deletions(-) create mode 100644 changelog.d/19929.misc diff --git a/changelog.d/19929.misc b/changelog.d/19929.misc new file mode 100644 index 0000000000..4778f9fa32 --- /dev/null +++ b/changelog.d/19929.misc @@ -0,0 +1 @@ +Fix `test_lock_contention` being flaky when running against PostgreSQL by budgeting CPU time rather than wall-clock time. diff --git a/tests/handlers/test_worker_lock.py b/tests/handlers/test_worker_lock.py index a38adcd4d4..5563ec4683 100644 --- a/tests/handlers/test_worker_lock.py +++ b/tests/handlers/test_worker_lock.py @@ -19,9 +19,6 @@ # # -import logging -import platform - from twisted.internet import defer from twisted.internet.testing import MemoryReactor @@ -39,8 +36,6 @@ from tests import unittest from tests.replication._base import BaseMultiWorkerStreamTestCase from tests.utils import test_timeout -logger = logging.getLogger(__name__) - class WorkerLockTestCase(unittest.HomeserverTestCase): def prepare( @@ -152,28 +147,16 @@ class WorkerLockTestCase(unittest.HomeserverTestCase): def test_lock_contention(self) -> None: """Test lock contention when a lot of locks wait on a single worker""" nb_locks_to_test = 500 - current_machine = platform.machine().lower() - if current_machine.startswith("riscv"): - # RISC-V specific settings - timeout_seconds = 15 # Increased timeout for RISC-V - # add a print or log statement here for visibility in CI logs - logger.info( # use logger.info - "Detected RISC-V architecture (%s). " - "Adjusting test_lock_contention: timeout=%ss", - current_machine, - timeout_seconds, - ) - else: - # Settings for other architectures - timeout_seconds = 5 - # It takes around 0.5s on a 5+ years old laptop - with test_timeout(timeout_seconds): # Use the dynamically set timeout - d = self._take_locks( - nb_locks_to_test - ) # Use the (potentially adjusted) number of locks - self.assertEqual( - self.get_success(d), nb_locks_to_test - ) # Assert against the used number of locks + + # This test is a performance-regression canary: before #16840 taking the + # locks below spent ~30s spinning the CPU, afterwards ~0.5s. We budget + # CPU time rather than wall-clock time so that time spent waiting on + # database round-trips (significant on PostgreSQL) or lost to a loaded + # CI machine doesn't make the test flaky: a healthy run costs well + # under 1s of CPU on either database engine. + with test_timeout(5, cpu_time=True): + d = self._take_locks(nb_locks_to_test) + self.assertEqual(self.get_success(d), nb_locks_to_test) async def _take_locks(self, nb_locks: int) -> int: locks = [ diff --git a/tests/utils.py b/tests/utils.py index f3d5129097..7ea3a2aded 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -318,20 +318,44 @@ class test_timeout: my_checking_func() time.sleep(0.1) ``` + + Args: + seconds: How long to allow the block to run for before raising + `TestTimeout`. + error_message: Extra text to append to the `TestTimeout` message. + cpu_time: If `True`, `seconds` is a budget of CPU time (user + system, + across all threads) consumed by the process rather than wall-clock + time. Useful for performance-regression tests, as time spent + blocked on I/O (e.g. waiting on the database) or lost to a loaded + CI machine doesn't count against the budget. Note that a block + which hangs while consuming *no* CPU will never trip this variant. """ - def __init__(self, seconds: int, error_message: str | None = None) -> None: - self.error_message = f"Test timed out after {seconds}s" + def __init__( + self, + seconds: float, + error_message: str | None = None, + *, + cpu_time: bool = False, + ) -> None: + self.error_message = f"Test timed out after {seconds}s of {'CPU' if cpu_time else 'wall-clock'} time" if error_message is not None: self.error_message += f": {error_message}" self.seconds = seconds + self.cpu_time = cpu_time def handle_timeout(self, signum: int, frame: FrameType | None) -> None: raise TestTimeout(self.error_message) def __enter__(self) -> None: - signal.signal(signal.SIGALRM, self.handle_timeout) - signal.alarm(self.seconds) + if self.cpu_time: + # `ITIMER_PROF` counts down against process CPU time (user + + # system) and delivers `SIGPROF` when it expires. + signal.signal(signal.SIGPROF, self.handle_timeout) + signal.setitimer(signal.ITIMER_PROF, self.seconds) + else: + signal.signal(signal.SIGALRM, self.handle_timeout) + signal.setitimer(signal.ITIMER_REAL, self.seconds) def __exit__( self, @@ -339,4 +363,4 @@ class test_timeout: exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: - signal.alarm(0) + signal.setitimer(signal.ITIMER_PROF if self.cpu_time else signal.ITIMER_REAL, 0) From 9c6cca1ac9b141a499c21bd41d3408ebfdbb999a Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 9 Jul 2026 11:09:59 -0500 Subject: [PATCH 2/6] Add clean deploy `FIXME` note for `TestOIDCProviderUnavailable` (#19938) Tracked by https://github.com/element-hq/synapse/issues/19937 --- changelog.d/19938.misc | 1 + complement/tests/oidc_test.go | 4 ++++ 2 files changed, 5 insertions(+) create mode 100644 changelog.d/19938.misc diff --git a/changelog.d/19938.misc b/changelog.d/19938.misc new file mode 100644 index 0000000000..d02c9bf3b7 --- /dev/null +++ b/changelog.d/19938.misc @@ -0,0 +1 @@ +Add clean deploy `FIXME` note for `TestOIDCProviderUnavailable` (problem tracked by [#19937](https://github.com/element-hq/synapse/issues/19937)). diff --git a/complement/tests/oidc_test.go b/complement/tests/oidc_test.go index deabf49950..59d8bb435f 100644 --- a/complement/tests/oidc_test.go +++ b/complement/tests/oidc_test.go @@ -52,6 +52,10 @@ oidc_providers: // `/_matrix/client/v3/login/sso/redirect/oidc-test_provider` endpoint. func TestOIDCProviderUnavailable(t *testing.T) { // Deploy a single homeserver + // + // FIXME: Since we're modifying the homeserver config, this should be using a clean + // deploy that won't affect subsequent tests because Complement will re-use the + // deployment, see https://github.com/element-hq/synapse/issues/19937 deployment := complement.Deploy(t, 1) defer deployment.Destroy(t) From 365df865367efeb25cf86ae77a8af985a119367a Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 9 Jul 2026 11:12:28 -0500 Subject: [PATCH 3/6] Fix Complement test flake when restarting Synapse workers (cross-test pollution caused by nginx upstreams being temporarily unavailable) (#19936) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fix https://github.com/element-hq/synapse/issues/19907 The flake manifested as a failure pointing at `TestMessagesOverFederation/Backfill_from_nearby_backward_extremities_past_token` but was actually caused by some cross-test pollution from an earlier test (`TestOIDCProviderUnavailable`) causing some workers to be temporarily unavailable. As explained in https://github.com/element-hq/synapse/issues/19907#issuecomment-4917362461, > ### Cross-test pollution > > When I point an LLM at the logs, it points to `TestOIDCProviderUnavailable` being the culprit because of the server restarts polluting this test. When this flake happens, we can indeed see that `TestOIDCProviderUnavailable` runs before `TestMessagesOverFederation`. > > ``` > PASS TestEventBetweenMakeJoinAndSendJoinIsNotLost 15.08s > PASS TestFederation/parallel/HS2_->_HS1 1.5s > PASS TestFederation/parallel/HS1_->_HS2 1.51s > PASS TestFederation/parallel 0s > PASS TestFederation 15.12s > PASS TestOIDCProviderUnavailable//login/sso/redirect_shows_HTML_error 0.02s > PASS TestOIDCProviderUnavailable 8.62s > FAIL TestMessagesOverFederation/Backfill_from_nearby_backward_extremities_past_token 0.89s > FAIL TestMessagesOverFederation 1.1s > PASS TestSynapseVersion/Synapse_version_matches_current_git_checkout 0.97s > PASS TestSynapseVersion 0.97s > ``` > > The pollution happens because we enable [`COMPLEMENT_ENABLE_DIRTY_RUNS`](https://github.com/element-hq/synapse/blob/c63d77a79d7157f26f849684520ba9e99f4d07c0/scripts-dev/complement.sh#L309-L311) ([docs](https://github.com/matrix-org/complement/blob/0e6f8552ff0c99fddb97222399efed3e1f0cb91a/ENVIRONMENT.md#complement_enable_dirty_runs)) which means Complement will reuse deployments (shares homeservers between tests). > > During the `TestOIDCProviderUnavailable` test, there are some stray federation requests that hit the homeserver while it's still booting which marks the nginx upstream as unavailable for 10 seconds. nginx has a default of [`max_fails=1`](https://nginx.org/en/docs/http/ngx_http_upstream_module.html#max_fails) and [`fail_timeout=10s`](https://nginx.org/en/docs/http/ngx_http_upstream_module.html#fail_timeout). Then when `TestMessagesOverFederation` starts, we're still in the 10 second unavailable window and nginx doesn't even try to connect at all. > >
> LLM summary of the logs and how this happens in practice > > 1. **19:52:10–11** — the previous federation test finishes: `user-3:hs2` does a faster-join (`send_join?omit_members=true`) to `!YnyCRpCLIpimppIreR:hs1`, so `hs2` kicks off a `sync_partial_state_room` background resync that is still running when the test ends. > 2. **19:52:11.6** — `TestOIDCProviderUnavailable` starts and calls `deployment.StopServer(t, "hs1")` / `StartServer` (`complement/tests/oidc_test.go:78-80`) to apply an OIDC config fragment. `hs1` gets `SIGTERM`; new supervisord at 19:52:13.3. `hs2` is not restarted and keeps retrying its unfinished work. > 3. **19:52:14** — `hs1`'s nginx is up, but the Synapse workers aren't: `federation_inbound` only listens on `18015` at 19:52:18.7, `federation_reader` on `18016` at 19:52:19.0. > 4. **19:52:15–16** — `hs2`'s retries arrive in that gap: `PUT /_matrix/federation/v1/send/…` → `18015` refused; `GET /state_ids/!YnyCRpCLIpimppIreR:hs1` → `18016` refused. With nginx defaults (`max_fails=1`, `fail_timeout=10s`) and only one server per upstream block, both upstreams are now marked down until ~19:52:25/26. (Side casualty: `hs2`'s partial-state resync gives up — "We can't get valid state history" — and puts `hs1` on a 10-minute federation backoff.) > 5. **19:52:21.5** — the failing test's `make_join` for `@user-5-bob:hs2` reaches `hs1`'s nginx → `no live upstreams` → `502` → the join fails, even though the worker has been listening for 2.5 seconds by then. > 6. **19:52:22** — `TestSynapseVersion`'s `GET /_matrix/federation/v1/version` hits the same dead upstream → `502` → it fails too. > > So it's a flake caused by a race between the OIDC test's container restart, hs2's background federation retries, and nginx's passive health-check — not anything wrong with the backfill logic under test. > >
> > > We can indeed confirm this suspicion with these logs > > :x: https://github.com/element-hq/synapse/actions/runs/28888070856/job/85701936736 (archive: [85701936736.log](https://github.com/user-attachments/files/29809986/85701936736.log)): > ``` > Error: 2026/07/07 19:52:20 [error] 34#34: *1 no live upstreams while connecting to upstream, client: 172.18.0.3, server: localhost, request: "PUT /_matrix/federation/v1/send/1783453927718 HTTP/1.1", upstream: "http://federation_inbound/_matrix/federation/v1/send/1783453927718", host: "hs1" > ... > > Error: 2026/07/07 19:52:21 [error] 33#33: *4 no live upstreams while connecting to upstream, client: 172.18.0.3, server: localhost, request: "GET /_matrix/federation/v1/make_join/%21scAbyTQDdaauGYicpU%3Ahs1/%40user-5-bob%3Ahs2?ver=1&ver=2&ver=3&ver=4&ver=5&ver=6&ver=7&ver=8&ver=9&ver=10&ver=11&ver=12&ver=org.matrix.msc3757.10&ver=org.matrix.msc3757.11&ver=org.matrix.hydra.11 HTTP/1.1", upstream: "http://federation_reader/_matrix/federation/v1/make_join/%21scAbyTQDdaauGYicpU%3Ahs1/%40user-5-bob%3Ahs2?ver=1&ver=2&ver=3&ver=4&ver=5&ver=6&ver=7&ver=8&ver=9&ver=10&ver=11&ver=12&ver=org.matrix.msc3757.10&ver=org.matrix.msc3757.11&ver=org.matrix.hydra.11", host: "hs1" > ``` > > The fix here would be to disable nginx's unavailable upstream behavior by configuring `max_fails=0` in the upstream block: https://github.com/element-hq/synapse/blob/c63d77a79d7157f26f849684520ba9e99f4d07c0/docker/configure_workers_and_start.py#L374-L378 --- changelog.d/19936.misc | 1 + docker/configure_workers_and_start.py | 13 ++++++++++--- 2 files changed, 11 insertions(+), 3 deletions(-) create mode 100644 changelog.d/19936.misc diff --git a/changelog.d/19936.misc b/changelog.d/19936.misc new file mode 100644 index 0000000000..5c1f7eef52 --- /dev/null +++ b/changelog.d/19936.misc @@ -0,0 +1 @@ +Fix Complement test flake when restarting Synapse workers (cross-test pollution caused by nginx upstreams being temporarily unavailable). diff --git a/docker/configure_workers_and_start.py b/docker/configure_workers_and_start.py index 26c8556eff..38f8649b44 100755 --- a/docker/configure_workers_and_start.py +++ b/docker/configure_workers_and_start.py @@ -1070,14 +1070,21 @@ def generate_worker_files( # Determine the load-balancing upstreams to configure nginx_upstream_config = "" for upstream_worker_base_name, upstream_worker_ports in nginx_upstreams.items(): + # We use `max_fails=0` to prevent nginx from marking an upstream as unavailable + # after it fails to contact the Synapse worker. Otherwise, if nginx sees a + # Synapse worker as unavailable once, it will be marked as unavailable for 10 + # seconds (`fail_timeout` default). + # + # This is necessary because we use `COMPLEMENT_ENABLE_DIRTY_RUNS` (re-uses + # deployments/homeservers) and we don't want any cross-test pollution from + # stopping/starting homeservers. body = "" if using_unix_sockets: for port in upstream_worker_ports: - body += f" server unix:/run/worker.{port};\n" - + body += f" server unix:/run/worker.{port} max_fails=0;\n" else: for port in upstream_worker_ports: - body += f" server localhost:{port};\n" + body += f" server localhost:{port} max_fails=0;\n" # Add to the list of configured upstreams nginx_upstream_config += NGINX_UPSTREAM_CONFIG_BLOCK.format( From 3f3edaf0293c9b85ce25451daeb9acd926820a3a Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 9 Jul 2026 17:21:50 +0100 Subject: [PATCH 4/6] Make the presence state-machine timers configurable (#19942) Adds `last_active_granularity`, `sync_online_timeout` and `idle_timeout` options to the `presence` config section, controlling the previously hard-coded `LAST_ACTIVE_GRANULARITY`, `SYNC_ONLINE_TIMEOUT` and `IDLE_TIMER` constants (which remain as the defaults). This is mainly useful on deployments that ratelimit how often syncs can affect presence (`rc_presence`): the sync timeout must exceed the ratelimit interval or users flap offline between syncs, so tuning down presence traffic currently requires squeezing under the fixed 30s limit. --------- Co-authored-by: Claude Fable 5 --- changelog.d/19942.misc | 1 + .../configuration/config_documentation.md | 18 ++ schema/synapse-config.schema.yaml | 40 +++ synapse/config/server.py | 39 +++ synapse/handlers/presence.py | 107 ++++++-- tests/handlers/test_presence.py | 259 +++++++++++++++--- 6 files changed, 401 insertions(+), 63 deletions(-) create mode 100644 changelog.d/19942.misc diff --git a/changelog.d/19942.misc b/changelog.d/19942.misc new file mode 100644 index 0000000000..bdd95c3c0f --- /dev/null +++ b/changelog.d/19942.misc @@ -0,0 +1 @@ +Add `last_active_granularity`, `sync_online_timeout` and `idle_timeout` options to the `presence` config section to allow tuning the presence state machine timers. diff --git a/docs/usage/configuration/config_documentation.md b/docs/usage/configuration/config_documentation.md index 92eca4a7ff..ec4098fdf6 100644 --- a/docs/usage/configuration/config_documentation.md +++ b/docs/usage/configuration/config_documentation.md @@ -284,6 +284,24 @@ This setting has the following sub-options: * `include_offline_users_on_sync` (boolean): When clients perform an initial or `full_state` sync, presence results for offline users are not included by default. Setting `include_offline_users_on_sync` to `true` will always include offline users in the results. Defaults to `false`. +* `last_active_granularity` (duration): How long after a user was last active that they are still shown as "currently active" to other users. Larger values reduce the rate of presence updates sent to other users and servers. + + *Added in Synapse 1.156.0.* + + Defaults to `"1m"`. + +* `sync_online_timeout` (duration): How long after a client's last sync request their presence is switched to offline. Clients are expected to keep a sync request open at (almost) all times while online, so this only needs to cover the gap between two consecutive sync requests. Note that if `rc_presence` is set to ratelimit how often syncs can affect presence, this must be greater than the ratelimit's interval or users will incorrectly be marked as offline in between syncs. + + *Added in Synapse 1.156.0.* + + Defaults to `"30s"`. + +* `idle_timeout` (duration): How long after a user was last active that their presence is switched to "unavailable" (idle) while they remain connected. Must be greater than `last_active_granularity`. + + *Added in Synapse 1.156.0.* + + Defaults to `"5m"`. + Example configuration: ```yaml presence: diff --git a/schema/synapse-config.schema.yaml b/schema/synapse-config.schema.yaml index 0e345b7b69..0d1aff071b 100644 --- a/schema/synapse-config.schema.yaml +++ b/schema/synapse-config.schema.yaml @@ -202,6 +202,46 @@ properties: `include_offline_users_on_sync` to `true` will always include offline users in the results. default: false + last_active_granularity: + $ref: "#/$defs/duration" + description: >- + How long after a user was last active that they are still shown as + "currently active" to other users. Larger values reduce the rate of + presence updates sent to other users and servers. + + + *Added in Synapse 1.156.0.* + default: 1m + examples: + - 5m + sync_online_timeout: + $ref: "#/$defs/duration" + description: >- + How long after a client's last sync request their presence is + switched to offline. Clients are expected to keep a sync request + open at (almost) all times while online, so this only needs to + cover the gap between two consecutive sync requests. Note that if + `rc_presence` is set to ratelimit how often syncs can affect + presence, this must be greater than the ratelimit's interval or + users will incorrectly be marked as offline in between syncs. + + + *Added in Synapse 1.156.0.* + default: 30s + examples: + - 1m + idle_timeout: + $ref: "#/$defs/duration" + description: >- + How long after a user was last active that their presence is + switched to "unavailable" (idle) while they remain connected. Must + be greater than `last_active_granularity`. + + + *Added in Synapse 1.156.0.* + default: 5m + examples: + - 10m examples: - enabled: false include_offline_users_on_sync: false diff --git a/synapse/config/server.py b/synapse/config/server.py index ca94c224ea..282679aabe 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -177,6 +177,19 @@ DEFAULT_IP_RANGE_BLOCKLIST = [ DEFAULT_ROOM_VERSION = "10" +# Defaults for the presence state machine timers, in milliseconds. Overridden +# by the corresponding options in the `presence` config section. +# +# How long after a user was last active that they are still considered +# "currently_active". +DEFAULT_LAST_ACTIVE_GRANULARITY = 60 * 1000 +# How long to wait until a new /events or /sync request before assuming the +# client has gone. +DEFAULT_SYNC_ONLINE_TIMEOUT = 30 * 1000 +# How long to wait before marking the user as idle. Compared against last +# active. +DEFAULT_IDLE_TIMER = 5 * 60 * 1000 + ROOM_COMPLEXITY_TOO_GREAT = ( "Your homeserver is unable to join rooms this large or complex. " "Please speak to your server administrator, or upgrade your instance " @@ -505,6 +518,32 @@ class ServerConfig(Config): "include_offline_users_on_sync", False ) + # Timers controlling the presence state machine. + self.presence_last_active_granularity = self.parse_duration( + presence_config.get( + "last_active_granularity", DEFAULT_LAST_ACTIVE_GRANULARITY + ) + ) + self.presence_sync_online_timeout = self.parse_duration( + presence_config.get("sync_online_timeout", DEFAULT_SYNC_ONLINE_TIMEOUT) + ) + self.presence_idle_timeout = self.parse_duration( + presence_config.get("idle_timeout", DEFAULT_IDLE_TIMER) + ) + if self.presence_last_active_granularity <= 0: + raise ConfigError( + "'presence.last_active_granularity' must be a positive duration" + ) + if self.presence_sync_online_timeout <= 0: + raise ConfigError( + "'presence.sync_online_timeout' must be a positive duration" + ) + if self.presence_idle_timeout <= self.presence_last_active_granularity: + raise ConfigError( + "'presence.idle_timeout' must be greater than " + "'presence.last_active_granularity'" + ) + # Custom presence router module # This is the legacy way of configuring it (the config should now be put in the modules section) self.presence_router_module_class = None diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 4c3adca46e..c1994fe488 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -179,19 +179,18 @@ presence_wheel_timer_size_gauge = LaterGauge( labelnames=[SERVER_NAME_LABEL], ) -# If a user was last active in the last LAST_ACTIVE_GRANULARITY, consider them -# "currently_active" -LAST_ACTIVE_GRANULARITY = 60 * 1000 +# Note: the timers deciding when a user goes idle or offline and how long +# they count as "currently_active" are configurable, via the +# `last_active_granularity`, `sync_online_timeout` and `idle_timeout` options +# in the `presence` config section (see +# `synapse.config.server.DEFAULT_LAST_ACTIVE_GRANULARITY` and friends for the +# defaults). -# How long to wait until a new /events or /sync request before assuming -# the client has gone. -SYNC_ONLINE_TIMEOUT = 30 * 1000 -# Busy status waits longer, but does eventually go offline. +# How long to wait until a device with busy status stops syncing before it +# goes offline. Busy status waits longer than the (configurable) sync online +# timeout, but does eventually go offline. BUSY_ONLINE_TIMEOUT = 60 * 60 * 1000 -# How long to wait before marking the user as idle. Compared against last active -IDLE_TIMER = 5 * 60 * 1000 - # How often we expect remote servers to resend us presence. FEDERATION_TIMEOUT = 30 * 60 * 1000 @@ -206,8 +205,6 @@ EXTERNAL_PROCESS_EXPIRY = 5 * 60 * 1000 # syncing. UPDATE_SYNCING_USERS = Duration(seconds=10) -assert LAST_ACTIVE_GRANULARITY < IDLE_TIMER - class BasePresenceHandler(abc.ABC): """Parts of the PresenceHandler that are shared between workers and presence @@ -225,6 +222,13 @@ class BasePresenceHandler(abc.ABC): self._presence_enabled = hs.config.server.presence_enabled self._track_presence = hs.config.server.track_presence + # The (configurable) presence state machine timers. + self._last_active_granularity = ( + hs.config.server.presence_last_active_granularity + ) + self._sync_online_timeout = hs.config.server.presence_sync_online_timeout + self._idle_timer = hs.config.server.presence_idle_timeout + self._federation = None if hs.should_send_federation(): self._federation = hs.get_federation_sender() @@ -685,7 +689,11 @@ class WorkerPresenceHandler(BasePresenceHandler): self.user_to_current_state[new_state.user_id] = new_state is_mine = self.is_mine_id(new_state.user_id) if not old_state or should_notify( - old_state, new_state, is_mine, self.server_name + old_state, + new_state, + is_mine, + self.server_name, + last_active_granularity=self._last_active_granularity, ): state_to_notify.append(new_state) @@ -791,7 +799,8 @@ class PresenceHandler(BasePresenceHandler): if self._track_presence: for state in self.user_to_current_state.values(): # Create a psuedo-device to properly handle time outs. This will - # be overridden by any "real" devices within SYNC_ONLINE_TIMEOUT. + # be overridden by any "real" devices within the sync online + # timeout. pseudo_device_id = None self._user_to_device_to_current_state[state.user_id] = { pseudo_device_id: UserDevicePresenceState( @@ -804,12 +813,14 @@ class PresenceHandler(BasePresenceHandler): } self.wheel_timer.insert( - now=now, obj=state.user_id, then=state.last_active_ts + IDLE_TIMER + now=now, + obj=state.user_id, + then=state.last_active_ts + self._idle_timer, ) self.wheel_timer.insert( now=now, obj=state.user_id, - then=state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=state.last_user_sync_ts + self._sync_online_timeout, ) if self.is_mine_id(state.user_id): self.wheel_timer.insert( @@ -996,6 +1007,9 @@ class PresenceHandler(BasePresenceHandler): # When overriding disabled presence, don't kick off all the # wheel timers. persist=not self._track_presence, + idle_timer=self._idle_timer, + sync_online_timeout=self._sync_online_timeout, + last_active_granularity=self._last_active_granularity, ) if force_notify: @@ -1108,6 +1122,9 @@ class PresenceHandler(BasePresenceHandler): syncing_user_devices=syncing_user_devices, user_to_devices=self._user_to_device_to_current_state, now=now, + idle_timer=self._idle_timer, + sync_online_timeout=self._sync_online_timeout, + last_active_granularity=self._last_active_granularity, ) return await self._update_states(changes) @@ -1695,6 +1712,8 @@ def should_notify( new_state: UserPresenceState, is_mine: bool, our_server_name: str, + *, + last_active_granularity: int, ) -> bool: """Decides if a presence state change should be sent to interested parties.""" user_location = "remote" @@ -1741,7 +1760,7 @@ def should_notify( if ( new_state.last_active_ts - old_state.last_active_ts - > LAST_ACTIVE_GRANULARITY + > last_active_granularity ): # Only notify about last active bumps if we're not currently active if not new_state.currently_active: @@ -1752,7 +1771,7 @@ def should_notify( ).inc() return True - elif new_state.last_active_ts - old_state.last_active_ts > LAST_ACTIVE_GRANULARITY: + elif new_state.last_active_ts - old_state.last_active_ts > last_active_granularity: # Always notify for a transition where last active gets bumped. notify_reason_counter.labels( locality=user_location, @@ -2075,6 +2094,10 @@ def handle_timeouts( syncing_user_devices: AbstractSet[tuple[str, str | None]], user_to_devices: dict[str, dict[str | None, UserDevicePresenceState]], now: int, + *, + idle_timer: int, + sync_online_timeout: int, + last_active_granularity: int, ) -> list[UserPresenceState]: """Checks the presence of users that have timed out and updates as appropriate. @@ -2085,6 +2108,11 @@ def handle_timeouts( syncing_user_devices: A set of (user ID, device ID) tuples with active syncs.. user_to_devices: A map of user ID to device ID to UserDevicePresenceState. now: Current time in ms. + idle_timer: How long in ms before an inactive device is marked as idle. + sync_online_timeout: How long in ms after the last sync before a device + is marked as offline. + last_active_granularity: How long in ms a user counts as + "currently active" after their last activity. Returns: List of UserPresenceState updates @@ -2101,6 +2129,9 @@ def handle_timeouts( syncing_user_devices, user_to_devices.get(user_id, {}), now, + idle_timer=idle_timer, + sync_online_timeout=sync_online_timeout, + last_active_granularity=last_active_granularity, ) if new_state: changes[state.user_id] = new_state @@ -2114,6 +2145,10 @@ def handle_timeout( syncing_device_ids: AbstractSet[tuple[str, str | None]], user_devices: dict[str | None, UserDevicePresenceState], now: int, + *, + idle_timer: int, + sync_online_timeout: int, + last_active_granularity: int, ) -> UserPresenceState | None: """Checks the presence of the user to see if any of the timers have elapsed @@ -2123,6 +2158,11 @@ def handle_timeout( syncing_user_devices: A set of (user ID, device ID) tuples with active syncs.. user_devices: A map of device ID to UserDevicePresenceState. now: Current time in ms. + idle_timer: How long in ms before an inactive device is marked as idle. + sync_online_timeout: How long in ms after the last sync before a device + is marked as offline. + last_active_granularity: How long in ms a user counts as + "currently active" after their last activity. Returns: A UserPresenceState update or None if no update. @@ -2140,7 +2180,7 @@ def handle_timeout( offline_devices = [] for device_id, device_state in user_devices.items(): if device_state.state == PresenceState.ONLINE: - if now - device_state.last_active_ts > IDLE_TIMER: + if now - device_state.last_active_ts > idle_timer: # Currently online, but last activity ages ago so auto # idle device_state.state = PresenceState.UNAVAILABLE @@ -2161,7 +2201,7 @@ def handle_timeout( online_timeout = ( BUSY_ONLINE_TIMEOUT if device_state.state == PresenceState.BUSY - else SYNC_ONLINE_TIMEOUT + else sync_online_timeout ) if now - sync_or_active > online_timeout: # Mark the device as going offline. @@ -2180,7 +2220,7 @@ def handle_timeout( state = state.copy_and_replace(state=new_presence) changed = True - if now - state.last_active_ts > LAST_ACTIVE_GRANULARITY: + if now - state.last_active_ts > last_active_granularity: # So that we send down a notification that we've # stopped updating. changed = True @@ -2209,6 +2249,10 @@ def handle_update( wheel_timer: WheelTimer, now: int, persist: bool, + *, + idle_timer: int, + sync_online_timeout: int, + last_active_granularity: int, ) -> tuple[UserPresenceState, bool, bool]: """Given a presence update: 1. Add any appropriate timers. @@ -2223,6 +2267,11 @@ def handle_update( now: Time now in ms persist: True if this state should persist until another update occurs. Skips insertion into wheel timers. + idle_timer: How long in ms before an inactive device is marked as idle. + sync_online_timeout: How long in ms after the last sync before a device + is marked as offline. + last_active_granularity: How long in ms a user counts as + "currently active" after their last activity. Returns: 3-tuple: `(new_state, persist_and_notify, federation_ping)` where: @@ -2242,17 +2291,17 @@ def handle_update( # Idle timer if not persist: wheel_timer.insert( - now=now, obj=user_id, then=new_state.last_active_ts + IDLE_TIMER + now=now, obj=user_id, then=new_state.last_active_ts + idle_timer ) - active = now - new_state.last_active_ts < LAST_ACTIVE_GRANULARITY + active = now - new_state.last_active_ts < last_active_granularity new_state = new_state.copy_and_replace(currently_active=active) if active and not persist: wheel_timer.insert( now=now, obj=user_id, - then=new_state.last_active_ts + LAST_ACTIVE_GRANULARITY, + then=new_state.last_active_ts + last_active_granularity, ) if new_state.state != PresenceState.OFFLINE: @@ -2261,7 +2310,7 @@ def handle_update( wheel_timer.insert( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_user_sync_ts + sync_online_timeout, ) last_federate = new_state.last_federation_update_ts @@ -2287,7 +2336,13 @@ def handle_update( ) # Check whether the change was something worth notifying about - if should_notify(prev_state, new_state, is_mine, our_server_name): + if should_notify( + prev_state, + new_state, + is_mine, + our_server_name, + last_active_granularity=last_active_granularity, + ): new_state = new_state.copy_and_replace(last_federation_update_ts=now) persist_and_notify = True diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 5d02c70161..624754d0cc 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -36,6 +36,11 @@ from synapse.api.presence import UserDevicePresenceState, UserPresenceState from synapse.api.room_versions import ( RoomVersion, ) +from synapse.config.server import ( + DEFAULT_IDLE_TIMER, + DEFAULT_LAST_ACTIVE_GRANULARITY, + DEFAULT_SYNC_ONLINE_TIMEOUT, +) from synapse.crypto.event_signing import add_hashes_and_signatures from synapse.events import EventBase, make_event_from_dict from synapse.federation.sender import FederationSender @@ -44,9 +49,6 @@ from synapse.handlers.presence import ( EXTERNAL_PROCESS_EXPIRY, FEDERATION_PING_INTERVAL, FEDERATION_TIMEOUT, - IDLE_TIMER, - LAST_ACTIVE_GRANULARITY, - SYNC_ONLINE_TIMEOUT, PresenceHandler, handle_timeout, handle_update, @@ -94,6 +96,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertTrue(persist_and_notify) @@ -105,16 +110,20 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): self.assertEqual(wheel_timer.insert.call_count, 3) wheel_timer.insert.assert_has_calls( [ - call(now=now, obj=user_id, then=new_state.last_active_ts + IDLE_TIMER), call( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_active_ts + DEFAULT_IDLE_TIMER, ), call( now=now, obj=user_id, - then=new_state.last_active_ts + LAST_ACTIVE_GRANULARITY, + then=new_state.last_user_sync_ts + DEFAULT_SYNC_ONLINE_TIMEOUT, + ), + call( + now=now, + obj=user_id, + then=new_state.last_active_ts + DEFAULT_LAST_ACTIVE_GRANULARITY, ), ], any_order=True, @@ -142,6 +151,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertFalse(persist_and_notify) @@ -154,16 +166,20 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): self.assertEqual(wheel_timer.insert.call_count, 3) wheel_timer.insert.assert_has_calls( [ - call(now=now, obj=user_id, then=new_state.last_active_ts + IDLE_TIMER), call( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_active_ts + DEFAULT_IDLE_TIMER, ), call( now=now, obj=user_id, - then=new_state.last_active_ts + LAST_ACTIVE_GRANULARITY, + then=new_state.last_user_sync_ts + DEFAULT_SYNC_ONLINE_TIMEOUT, + ), + call( + now=now, + obj=user_id, + then=new_state.last_active_ts + DEFAULT_LAST_ACTIVE_GRANULARITY, ), ], any_order=True, @@ -177,7 +193,7 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): prev_state = UserPresenceState.default(user_id) prev_state = prev_state.copy_and_replace( state=PresenceState.ONLINE, - last_active_ts=now - LAST_ACTIVE_GRANULARITY - 10, + last_active_ts=now - DEFAULT_LAST_ACTIVE_GRANULARITY - 10, currently_active=True, ) @@ -193,6 +209,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertFalse(persist_and_notify) @@ -205,16 +224,20 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): self.assertEqual(wheel_timer.insert.call_count, 3) wheel_timer.insert.assert_has_calls( [ - call(now=now, obj=user_id, then=new_state.last_active_ts + IDLE_TIMER), call( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_active_ts + DEFAULT_IDLE_TIMER, ), call( now=now, obj=user_id, - then=new_state.last_active_ts + LAST_ACTIVE_GRANULARITY, + then=new_state.last_user_sync_ts + DEFAULT_SYNC_ONLINE_TIMEOUT, + ), + call( + now=now, + obj=user_id, + then=new_state.last_active_ts + DEFAULT_LAST_ACTIVE_GRANULARITY, ), ], any_order=True, @@ -228,7 +251,7 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): prev_state = UserPresenceState.default(user_id) prev_state = prev_state.copy_and_replace( state=PresenceState.ONLINE, - last_active_ts=now - LAST_ACTIVE_GRANULARITY - 1, + last_active_ts=now - DEFAULT_LAST_ACTIVE_GRANULARITY - 1, currently_active=True, ) @@ -242,6 +265,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertTrue(persist_and_notify) @@ -253,11 +279,15 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): self.assertEqual(wheel_timer.insert.call_count, 2) wheel_timer.insert.assert_has_calls( [ - call(now=now, obj=user_id, then=new_state.last_active_ts + IDLE_TIMER), call( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_active_ts + DEFAULT_IDLE_TIMER, + ), + call( + now=now, + obj=user_id, + then=new_state.last_user_sync_ts + DEFAULT_SYNC_ONLINE_TIMEOUT, ), ], any_order=True, @@ -283,6 +313,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertFalse(persist_and_notify) @@ -323,6 +356,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertTrue(persist_and_notify) @@ -351,6 +387,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertTrue(persist_and_notify) @@ -365,7 +404,7 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): call( now=now, obj=user_id, - then=new_state.last_user_sync_ts + SYNC_ONLINE_TIMEOUT, + then=new_state.last_user_sync_ts + DEFAULT_SYNC_ONLINE_TIMEOUT, ) ], any_order=True, @@ -442,6 +481,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=True, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) wheel_timer.insert.assert_not_called() @@ -506,6 +548,9 @@ class PresenceUpdateTestCase(unittest.HomeserverTestCase): wheel_timer=wheel_timer, now=now, persist=False, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) # Check that the user is offline. @@ -556,7 +601,7 @@ class PresenceTimeoutTestCase(unittest.TestCase): state = UserPresenceState.default(user_id) state = state.copy_and_replace( state=PresenceState.ONLINE, - last_active_ts=now - IDLE_TIMER - 1, + last_active_ts=now - DEFAULT_IDLE_TIMER - 1, last_user_sync_ts=now, status_msg=status_msg, ) @@ -574,6 +619,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -594,7 +642,7 @@ class PresenceTimeoutTestCase(unittest.TestCase): state = UserPresenceState.default(user_id) state = state.copy_and_replace( state=PresenceState.BUSY, - last_active_ts=now - IDLE_TIMER - 1, + last_active_ts=now - DEFAULT_IDLE_TIMER - 1, last_user_sync_ts=now, status_msg=status_msg, ) @@ -612,6 +660,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -629,7 +680,7 @@ class PresenceTimeoutTestCase(unittest.TestCase): state = state.copy_and_replace( state=PresenceState.ONLINE, last_active_ts=0, - last_user_sync_ts=now - SYNC_ONLINE_TIMEOUT - 1, + last_user_sync_ts=now - DEFAULT_SYNC_ONLINE_TIMEOUT - 1, status_msg=status_msg, ) device_state = UserDevicePresenceState( @@ -646,6 +697,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -662,8 +716,8 @@ class PresenceTimeoutTestCase(unittest.TestCase): state = UserPresenceState.default(user_id) state = state.copy_and_replace( state=PresenceState.ONLINE, - last_active_ts=now - SYNC_ONLINE_TIMEOUT - 1, - last_user_sync_ts=now - SYNC_ONLINE_TIMEOUT - 1, + last_active_ts=now - DEFAULT_SYNC_ONLINE_TIMEOUT - 1, + last_user_sync_ts=now - DEFAULT_SYNC_ONLINE_TIMEOUT - 1, status_msg=status_msg, ) device_state = UserDevicePresenceState( @@ -680,6 +734,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids={(user_id, device_id)}, user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -715,6 +772,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -746,6 +806,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNone(new_state) @@ -766,7 +829,14 @@ class PresenceTimeoutTestCase(unittest.TestCase): # Note that this is a remote user so we do not have their device information. new_state = handle_timeout( - state, is_mine=False, syncing_device_ids=set(), user_devices={}, now=now + state, + is_mine=False, + syncing_device_ids=set(), + user_devices={}, + now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -783,7 +853,7 @@ class PresenceTimeoutTestCase(unittest.TestCase): state = UserPresenceState.default(user_id) state = state.copy_and_replace( state=PresenceState.ONLINE, - last_active_ts=now - LAST_ACTIVE_GRANULARITY - 1, + last_active_ts=now - DEFAULT_LAST_ACTIVE_GRANULARITY - 1, last_user_sync_ts=now, last_federation_update_ts=now, status_msg=status_msg, @@ -802,6 +872,9 @@ class PresenceTimeoutTestCase(unittest.TestCase): syncing_device_ids=set(), user_devices={device_id: device_state}, now=now, + idle_timer=DEFAULT_IDLE_TIMER, + sync_online_timeout=DEFAULT_SYNC_ONLINE_TIMEOUT, + last_active_granularity=DEFAULT_LAST_ACTIVE_GRANULARITY, ) self.assertIsNotNone(new_state) @@ -860,7 +933,7 @@ class PresenceHandlerInitTestCase(unittest.HomeserverTestCase): self.assertEqual(state.state, PresenceState.ONLINE) # Advance such that the user should timeout. - self.reactor.advance(SYNC_ONLINE_TIMEOUT / 1000) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT / 1000) self.reactor.pump([5]) # Check that the user is now offline. @@ -900,7 +973,7 @@ class PresenceHandlerInitTestCase(unittest.HomeserverTestCase): self.assertEqual(state.state, PresenceState.ONLINE) # Advance slightly and sync. - self.reactor.advance(SYNC_ONLINE_TIMEOUT / 1000 / 2) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT / 1000 / 2) self.get_success( presence_handler.user_syncing( self.user_id, @@ -917,7 +990,7 @@ class PresenceHandlerInitTestCase(unittest.HomeserverTestCase): self.assertEqual(state.state, expected_state) # Advance such that the user's preloaded data times out, but not the new sync. - self.reactor.advance(SYNC_ONLINE_TIMEOUT / 1000 / 2) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT / 1000 / 2) self.reactor.pump([5]) # Check that the user is in the sync state (as the client is currently syncing still). @@ -927,6 +1000,118 @@ class PresenceHandlerInitTestCase(unittest.HomeserverTestCase): self.assertEqual(state.state, sync_state) +# Timer values used by `PresenceConfigurableTimersTestCase`, all larger than +# the corresponding defaults. +_CUSTOM_TIMERS_CONFIG = { + "presence": { + "last_active_granularity": "2m", + "sync_online_timeout": "3m", + "idle_timeout": "20m", + } +} + + +class PresenceConfigurableTimersTestCase(unittest.HomeserverTestCase): + """Tests that the presence state machine timers can be changed via the + `presence` config section. + + Each test checks that nothing happens where the default timer would have + fired, and that the transition then occurs once the configured timer + elapses. + """ + + device_id = "dev-1" + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.presence_handler = hs.get_presence_handler() + self.user_id = f"@test:{hs.config.server.server_name}" + self.user_id_obj = UserID.from_string(self.user_id) + + def _get_state(self) -> UserPresenceState: + return self.get_success(self.presence_handler.get_state(self.user_id_obj)) + + @override_config(_CUSTOM_TIMERS_CONFIG) + def test_config_parsing(self) -> None: + config = self.hs.config.server + self.assertEqual(config.presence_last_active_granularity, 2 * 60 * 1000) + self.assertEqual(config.presence_sync_online_timeout, 3 * 60 * 1000) + self.assertEqual(config.presence_idle_timeout, 20 * 60 * 1000) + + @override_config(_CUSTOM_TIMERS_CONFIG) + def test_sync_online_timeout(self) -> None: + """A user only goes offline once the configured sync timeout passes.""" + with self.get_success( + self.presence_handler.user_syncing( + self.user_id, self.device_id, True, PresenceState.ONLINE + ) + ): + pass + + self.assertEqual(self._get_state().state, PresenceState.ONLINE) + + # Well past the DEFAULT_SYNC_ONLINE_TIMEOUT, but short of the + # configured 3m: still online. + self.reactor.advance(2 * DEFAULT_SYNC_ONLINE_TIMEOUT / 1000) + self.reactor.pump([5]) + self.assertEqual(self._get_state().state, PresenceState.ONLINE) + + # Past the configured timeout: offline. + self.reactor.advance(3 * 60) + self.reactor.pump([5]) + self.assertEqual(self._get_state().state, PresenceState.OFFLINE) + + @override_config(_CUSTOM_TIMERS_CONFIG) + def test_idle_timeout(self) -> None: + """A continuously syncing but inactive user only goes idle once the + configured idle timeout passes.""" + # Leave the sync open so the device never times out. + self.get_success( + self.presence_handler.user_syncing( + self.user_id, self.device_id, True, PresenceState.ONLINE + ) + ) + + # Well past the DEFAULT_IDLE_TIMER, but short of the configured 20m: + # still online. + self.reactor.advance(2 * DEFAULT_IDLE_TIMER / 1000) + self.reactor.pump([5]) + self.assertEqual(self._get_state().state, PresenceState.ONLINE) + + # Past the configured timeout: idle. + self.reactor.advance(15 * 60) + self.reactor.pump([5]) + self.assertEqual(self._get_state().state, PresenceState.UNAVAILABLE) + + @override_config(_CUSTOM_TIMERS_CONFIG) + def test_last_active_granularity(self) -> None: + """A user remains "currently active" for the configured duration + after their last activity.""" + with self.get_success( + self.presence_handler.user_syncing( + self.user_id, self.device_id, True, PresenceState.ONLINE + ) + ): + pass + + self.assertTrue(self._get_state().currently_active) + + # Past the DEFAULT_LAST_ACTIVE_GRANULARITY, but short of the + # configured 2m: still currently active. + self.reactor.advance(90) + self.reactor.pump([5]) + state = self._get_state() + self.assertEqual(state.state, PresenceState.ONLINE) + self.assertTrue(state.currently_active) + + # Past the configured granularity (but short of the 3m sync timeout): + # no longer currently active, but still online. + self.reactor.advance(60) + self.reactor.pump([5]) + state = self._get_state() + self.assertEqual(state.state, PresenceState.ONLINE) + self.assertFalse(state.currently_active) + + class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): user_id = "@test:server" user_id_obj = UserID.from_string(user_id) @@ -979,14 +1164,14 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # Check that if we wait a while without telling the handler the user has # stopped syncing that their presence state doesn't get timed out. - self.reactor.advance(SYNC_ONLINE_TIMEOUT / 2) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT / 2) state = self.get_success(self.presence_handler.get_state(self.user_id_obj)) self.assertEqual(state.state, PresenceState.ONLINE) self.assertEqual(state.status_msg, status_msg) # Check that if the timeout fires, then the syncing user gets timed out - self.reactor.advance(SYNC_ONLINE_TIMEOUT) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT) state = self.get_success(self.presence_handler.get_state(self.user_id_obj)) # status_msg should remain even after going offline @@ -1273,7 +1458,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): ) # 2. Wait half the idle timer. - self.reactor.advance(IDLE_TIMER / 1000 / 2) + self.reactor.advance(DEFAULT_IDLE_TIMER / 1000 / 2) self.reactor.pump([0.1]) # 3. Sync with the second device. @@ -1300,7 +1485,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # When testing with workers, make another random sync (with any *different* # user) to keep the process information from expiring. # - # This is due to EXTERNAL_PROCESS_EXPIRY being equivalent to IDLE_TIMER. + # This is due to EXTERNAL_PROCESS_EXPIRY being equivalent to DEFAULT_IDLE_TIMER. if test_with_workers: with self.get_success( worker_presence_handler.user_syncing( @@ -1314,7 +1499,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # 5. Advance such that the first device should be discarded (the idle timer), # then pump so _handle_timeouts function to called. - self.reactor.advance(IDLE_TIMER / 1000 / 2) + self.reactor.advance(DEFAULT_IDLE_TIMER / 1000 / 2) self.reactor.pump([0.01]) # 6. Assert the expected presence state. @@ -1330,7 +1515,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # 7. Advance such that the second device should be discarded (half the idle timer), # then pump so _handle_timeouts function to called. - self.reactor.advance(IDLE_TIMER / 1000 / 2) + self.reactor.advance(DEFAULT_IDLE_TIMER / 1000 / 2) self.reactor.pump([0.1]) # 8. The devices are still "syncing" (the sync context managers were never @@ -1533,7 +1718,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # 5. Advance such that the first device should be discarded (the sync timeout), # then pump so _handle_timeouts function to called. - self.reactor.advance(SYNC_ONLINE_TIMEOUT / 1000) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT / 1000) self.reactor.pump([5]) # 6. Assert the expected presence state. @@ -1556,7 +1741,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): if dev_1_state == PresenceState.BUSY or dev_2_state == PresenceState.BUSY: timeout = BUSY_ONLINE_TIMEOUT else: - timeout = SYNC_ONLINE_TIMEOUT + timeout = DEFAULT_SYNC_ONLINE_TIMEOUT self.reactor.advance(timeout / 1000) self.reactor.pump([5]) @@ -1629,7 +1814,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): # Advance such that the device would be discarded if it was not busy, # then pump so _handle_timeouts function to called. - self.reactor.advance(IDLE_TIMER / 1000) + self.reactor.advance(DEFAULT_IDLE_TIMER / 1000) self.reactor.pump([5]) # The account should still be busy. @@ -1682,7 +1867,7 @@ class PresenceHandlerTestCase(BaseMultiWorkerStreamTestCase): self.assertEqual(state.state, PresenceState.ONLINE) # The timeout should not fire and the state should be the same. - self.reactor.advance(SYNC_ONLINE_TIMEOUT) + self.reactor.advance(DEFAULT_SYNC_ONLINE_TIMEOUT) state = self.get_success(self.presence_handler.get_state(self.user_id_obj)) self.assertEqual(state.state, PresenceState.ONLINE) From 9bbd974b3ec6cae2cd56b0e3bff7df19201a98eb Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 10 Jul 2026 10:37:38 +0100 Subject: [PATCH 5/6] Add index to `sliding_sync_connection_lazy_members` on `connection_position` (#19923) This speeds up the cascading delete from `sliding_sync_connection_positions`, which without an index on `connection_position` requires a sequential scan of the whole table for each deleted position. Co-authored-by: Claude Opus 4.8 --- changelog.d/19923.misc | 1 + .../storage/databases/main/sliding_sync.py | 8 +++++++ ...iding_sync_lazy_members_position_index.sql | 24 +++++++++++++++++++ 3 files changed, 33 insertions(+) create mode 100644 changelog.d/19923.misc create mode 100644 synapse/storage/schema/main/delta/94/07_sliding_sync_lazy_members_position_index.sql diff --git a/changelog.d/19923.misc b/changelog.d/19923.misc new file mode 100644 index 0000000000..4a52d1cca2 --- /dev/null +++ b/changelog.d/19923.misc @@ -0,0 +1 @@ +Add an index to `sliding_sync_connection_lazy_members` to speed up deleting old sliding sync connection positions. diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index dbb9efc2ac..a5d6cd2548 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/synapse/storage/databases/main/sliding_sync.py @@ -104,6 +104,14 @@ class SlidingSyncStore(SQLBaseStore): where_clause="last_used_ts IS NOT NULL", ) + self.db_pool.updates.register_background_index_update( + update_name="sliding_sync_connection_lazy_members_conn_pos_idx", + index_name="sliding_sync_connection_lazy_members_conn_pos_idx", + table="sliding_sync_connection_lazy_members", + columns=("connection_position",), + where_clause="connection_position IS NOT NULL", + ) + if self.hs.config.worker.run_background_tasks: self.clock.looping_call( self.delete_old_sliding_sync_connections, diff --git a/synapse/storage/schema/main/delta/94/07_sliding_sync_lazy_members_position_index.sql b/synapse/storage/schema/main/delta/94/07_sliding_sync_lazy_members_position_index.sql new file mode 100644 index 0000000000..7c55763828 --- /dev/null +++ b/synapse/storage/schema/main/delta/94/07_sliding_sync_lazy_members_position_index.sql @@ -0,0 +1,24 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + + +-- Add an index on `sliding_sync_connection_lazy_members(connection_position)` +-- so that deleting from `sliding_sync_connection_positions` is efficient. This +-- is needed because `connection_position` has an `ON DELETE CASCADE` foreign key +-- constraint, and without this index Postgres has to sequentially scan the whole +-- table for each deleted position. +-- +-- This is a partial index as we only ever need to find rows with a non-NULL +-- `connection_position`. +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (9407, 'sliding_sync_connection_lazy_members_conn_pos_idx', '{}'); From 87439e55b5dd90c1e0192085e4ec54f7f8a12a55 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 10 Jul 2026 10:38:43 +0100 Subject: [PATCH 6/6] Fix storage type mismatches where values didn't match their column types (#19911) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A handful of places in the storage layer bound a value whose Python type didn't match the declared column type — an `int` into a `TEXT` column, or a `str` into a `BIGINT` column — and relied on psycopg2's loose coercion to paper over the mismatch. These are latent correctness bugs: they only work because the driver silently converts, and a stricter driver that binds typed parameters rejects them outright. Found during a Rust port of the database pool where the driver does not coerce automatically. Each fix binds the value with the type the column actually declares, rather than depending on driver-specific coercion. All changes are behaviour-preserving on psycopg2 and sqlite. There is also a fix to the multi-writer id-gen tests where we forgot to commit. This is tangential, but was found during the same effort. ### Changes - **`device_lists_remote_extremeties.stream_id (TEXT)`** — `_update_remote_device_list_cache_txn` (typed `stream_id: int`) bound an int; store it as a string, matching the column and the sibling `_update_remote_device_list_cache_entry_txn` (typed `stream_id: str`). The old mismatch, when rejected, was swallowed inside the device-list resync and hung `query_devices`. - **`user_filters.filter_id (BIGINT)`** — `get_user_filter` bound the raw `int | str` (a string, from sync requests). It already validates via `int(filter_id)`; bind that int so it matches the column. - **`rejections.last_check (TEXT)`** — both writers stored `clock.time_msec()` (an int); store the timestamp as a string. - **user-directory temp position** — the `populate_user_directory` background update's staging column `_temp_populate_user_directory_position.position` was `TEXT` but held an int (read back into a `BIGINT` column). Declare it `BIGINT`. The temp table is created and dropped within the background update, so there's no migration. - **`test_batched_state_group_storing`** — selected from `state_group_edges` with a stringified `state_group`; bind the int directly (the column is an integer). - **Multi-writer id-generator tests** — constructing a `MultiWriterIdGenerator` prunes stale `stream_positions` rows, but the harness never committed, so the cleanup only survived because adbapi keeps one connection per thread with its transaction open. Commit after construction so it persists regardless of pool semantics (the delete is legitimate work that should be committed anyway). --- changelog.d/19911.misc | 1 + synapse/storage/databases/main/devices.py | 5 ++++- synapse/storage/databases/main/events.py | 5 ++++- synapse/storage/databases/main/events_worker.py | 5 ++++- synapse/storage/databases/main/filtering.py | 2 +- synapse/storage/databases/main/user_directory.py | 2 +- tests/storage/test_id_generators.py | 11 ++++++++++- tests/storage/test_state.py | 4 +--- 8 files changed, 26 insertions(+), 9 deletions(-) create mode 100644 changelog.d/19911.misc diff --git a/changelog.d/19911.misc b/changelog.d/19911.misc new file mode 100644 index 0000000000..11331455aa --- /dev/null +++ b/changelog.d/19911.misc @@ -0,0 +1 @@ +Fix storage type mismatches where values were bound with a type that didn't match their database column. diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py index 8670d68f38..b293b47fea 100644 --- a/synapse/storage/databases/main/devices.py +++ b/synapse/storage/databases/main/devices.py @@ -2019,7 +2019,10 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore): txn, table="device_lists_remote_extremeties", keyvalues={"user_id": user_id}, - values={"stream_id": stream_id}, + # `stream_id` is a TEXT column, so store it as a string (this method + # takes an int) rather than relying on the driver to coerce it. + # (Ideally we'd fix the schema, but that is non-trivial) + values={"stream_id": str(stream_id)}, ) async def add_device_change_to_streams( diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 84b38f4bf2..d92bbeeae3 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -3513,7 +3513,10 @@ class PersistEventsStore: values={ "event_id": event_id, "reason": reason, - "last_check": self._clock.time_msec(), + # `last_check` is a TEXT column, so store the timestamp as a + # string rather than relying on the driver to coerce an int. + # (Ideally we'd fix the schema, but that is non-trivial) + "last_check": str(self._clock.time_msec()), }, ) diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index d40240a0a5..27dab290b3 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/synapse/storage/databases/main/events_worker.py @@ -2702,7 +2702,10 @@ class EventsWorkerStore(SQLBaseStore): keyvalues={"event_id": event_id}, values={ "reason": rejection_reason, - "last_check": self.clock.time_msec(), + # `last_check` is a TEXT column, so store the timestamp as a + # string rather than relying on the driver to coerce an int. + # (Ideally we'd fix the schema, but that is non-trivial) + "last_check": str(self.clock.time_msec()), }, ) self.db_pool.simple_update_txn( diff --git a/synapse/storage/databases/main/filtering.py b/synapse/storage/databases/main/filtering.py index 2019ad9904..c334457af3 100644 --- a/synapse/storage/databases/main/filtering.py +++ b/synapse/storage/databases/main/filtering.py @@ -156,7 +156,7 @@ class FilteringWorkerStore(SQLBaseStore): # filter_id is BIGINT UNSIGNED, so if it isn't a number, fail # with a coherent error message rather than 500 M_UNKNOWN. try: - int(filter_id) + filter_id = int(filter_id) except ValueError: raise SynapseError(400, "Invalid filter ID", Codes.INVALID_PARAM) diff --git a/synapse/storage/databases/main/user_directory.py b/synapse/storage/databases/main/user_directory.py index 6c5abc71ae..e3f16ef754 100644 --- a/synapse/storage/databases/main/user_directory.py +++ b/synapse/storage/databases/main/user_directory.py @@ -127,7 +127,7 @@ class UserDirectoryBackgroundUpdateStore(StateDeltasStore): sql = f""" CREATE TABLE IF NOT EXISTS {TEMP_TABLE}_position ( - position TEXT NOT NULL + position BIGINT NOT NULL ) """ txn.execute(sql) diff --git a/tests/storage/test_id_generators.py b/tests/storage/test_id_generators.py index 051c5de44d..9a338607ee 100644 --- a/tests/storage/test_id_generators.py +++ b/tests/storage/test_id_generators.py @@ -78,7 +78,7 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase): writers: list[str] | None = None, ) -> MultiWriterIdGenerator: def _create(conn: LoggingDatabaseConnection) -> MultiWriterIdGenerator: - return MultiWriterIdGenerator( + id_gen = MultiWriterIdGenerator( db_conn=conn, db=self.db_pool, notifier=self.hs.get_replication_notifier(), @@ -90,6 +90,15 @@ class MultiWriterIdGeneratorBase(HomeserverTestCase): writers=writers or ["master"], positive=self.positive, ) + # Constructing the generator prunes stale `stream_positions` rows + # (writers no longer in the config); commit so that persists for the + # next generator we create. + # + # Note we need to commit manually here as the generator is created + # in a `runWithConnection` call, which doesn't automatically + # commit/rollback. + conn.commit() + return id_gen self.instances[instance_name] = self.get_success_or_raise( self.db_pool.runWithConnection(_create) diff --git a/tests/storage/test_state.py b/tests/storage/test_state.py index dbbede812d..6c3506f348 100644 --- a/tests/storage/test_state.py +++ b/tests/storage/test_state.py @@ -637,9 +637,7 @@ class StateStoreTestCase(HomeserverTestCase): self.get_success( self.store.db_pool.simple_select_list( table="state_group_edges", - keyvalues={ - "state_group": str(context.state_group_after_event) - }, + keyvalues={"state_group": context.state_group_after_event}, retcols=("prev_state_group",), ) ),