diff --git a/changelog.d/19941.misc b/changelog.d/19941.misc new file mode 100644 index 00000000000..e928f68bc58 --- /dev/null +++ b/changelog.d/19941.misc @@ -0,0 +1 @@ +Reduce replication traffic caused by presence. diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index c1994fe488b..28fbbfc1e17 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -530,10 +530,34 @@ def __init__(self, hs: "HomeServer"): # syncing but we haven't notified the presence writer of that yet self._user_devices_going_offline: dict[tuple[str, str | None], int] = {} + # How often to relay an unchanged sync-driven presence state to the + # presence writer. The relayed updates are what feed the writer's device + # last_sync_ts/last_active_ts timers, so this must sit comfortably below + # the timers it feeds — the (configurable) sync online timeout and + # last-active granularity — or users would flap offline / lose + # "currently active" between relays. We use 5/6 of the tighter of the + # two, i.e. the historic 25s at the default 30s sync online timeout. + self._sync_presence_relay_interval = ( + min(self._sync_online_timeout, self._last_active_granularity) * 5 // 6 + ) + + # (user_id, device_id) -> (state, last_sent_ms) of the most recent + # sync-driven presence update we proxied to the presence writer. Used + # to suppress the per-sync-request set_state/bump calls, which are + # no-ops on the writer at finer granularity than its timers: while + # the state is unchanged there is no point relaying more than one + # update per relay interval. Entries older than the window are swept by + # `_sweep_last_sent_presence`. + self._last_sent_presence: dict[tuple[str, str | None], tuple[str, int]] = {} + self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs) self._set_state_client = ReplicationPresenceSetState.make_client(hs) - self.clock.looping_call(self.send_stop_syncing, UPDATE_SYNCING_USERS) + if self._track_presence: + self.clock.looping_call(self.send_stop_syncing, UPDATE_SYNCING_USERS) + self.clock.looping_call( + self._sweep_last_sent_presence, Duration(minutes=30) + ) hs.register_async_shutdown_handler( phase="before", @@ -576,6 +600,8 @@ def mark_as_going_offline(self, user_id: str, device_id: str | None) -> None: sending a stopped syncing immediately followed by a started syncing notification to the presence writer """ + if not self._track_presence: + return self._user_devices_going_offline[(user_id, device_id)] = self.clock.time_msec() def send_stop_syncing(self) -> None: @@ -589,6 +615,22 @@ def send_stop_syncing(self) -> None: if now - last_sync_ms > UPDATE_SYNCING_USERS.as_millis(): self._user_devices_going_offline.pop((user_id, device_id), None) self.send_user_sync(user_id, device_id, False, last_sync_ms) + # Once the writer knows the device stopped syncing it may time + # the user out, so if the device comes back we must relay its + # state again rather than suppress it as a repeat. + self._last_sent_presence.pop((user_id, device_id), None) + + def _sweep_last_sent_presence(self) -> None: + """Drop expired presence-throttling entries. + + Entries should be dropped in `send_stop_syncing`, but we add a safety + net here to ensure that the dict deesn't grow unbounded. + """ + now = self.clock.time_msec() + + for key, (_, last_sent_ms) in list(self._last_sent_presence.items()): + if now - last_sent_ms >= self._sync_presence_relay_interval: + self._last_sent_presence.pop(key, None) async def user_syncing( self, @@ -606,7 +648,9 @@ async def user_syncing( return _NullContextManager() # Note that this causes last_active_ts to be incremented which is not - # what the spec wants. + # what the spec wants. (This call is throttled in `set_state`: while + # the state is unchanged, only one update per relay interval is relayed + # to the presence writer.) await self.set_state( UserID.from_string(user_id), device_id, @@ -743,6 +787,28 @@ async def set_state( if not self._track_presence: return + now = self.clock.time_msec() + if is_sync and not force_notify: + # Sync-driven updates arrive on every /sync request, which is far + # finer-grained than any of the writer's presence timers need: + # while the state is unchanged, relaying one update per relay + # interval is enough to keep them fed. State changes always go + # through immediately. + last_sent = self._last_sent_presence.get((user_id, device_id)) + if last_sent is not None: + last_presence, last_sent_ms = last_sent + if ( + presence == last_presence + and now - last_sent_ms < self._sync_presence_relay_interval + ): + return + self._last_sent_presence[(user_id, device_id)] = (presence, now) + else: + # An explicit (non-sync) update doesn't refresh the writer's + # last_sync_ts, so it must not count as a recent relay: drop any + # entry so the next sync-driven update goes through. + self._last_sent_presence.pop((user_id, device_id), None) + # Proxy request to instance that writes presence await self._set_state_client( instance_name=self._presence_writer_instance, @@ -763,8 +829,25 @@ async def bump_presence_active_time( if not self._track_presence: return - # Proxy request to instance that writes presence user_id = user.to_string() + + # A bump's only effects on the writer are updating last_active_ts and + # flipping an idle device back online. Going idle takes far longer + # than the relay window, so if we relayed an *online* update within + # the window the user cannot have gone idle since, and this bump is a + # no-op: skip it. Bumps after any other state (or an unknown one) go + # through immediately, as they may un-idle the device. + now = self.clock.time_msec() + last_sent = self._last_sent_presence.get((user_id, device_id)) + if ( + last_sent is not None + and last_sent[0] == PresenceState.ONLINE + and now - last_sent[1] < self._sync_presence_relay_interval + ): + return + self._last_sent_presence[(user_id, device_id)] = (PresenceState.ONLINE, now) + + # Proxy request to instance that writes presence await self._bump_active_client( instance_name=self._presence_writer_instance, user_id=user_id, diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 624754d0ccb..88d9d2aebba 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -19,7 +19,7 @@ # # import itertools -from typing import cast +from typing import Any, cast from unittest.mock import Mock, call from parameterized import parameterized @@ -50,6 +50,7 @@ FEDERATION_PING_INTERVAL, FEDERATION_TIMEOUT, PresenceHandler, + WorkerPresenceHandler, handle_timeout, handle_update, ) @@ -2320,3 +2321,180 @@ def create_fake_event_from_remote_server( ) return event + + +class WorkerPresenceThrottleTestCase(BaseMultiWorkerStreamTestCase): + """Tests that sync workers suppress the per-sync-request presence updates + that the presence writer would discard anyway, while relaying genuine + state changes immediately.""" + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.user_id = f"@throttled:{hs.config.server.server_name}" + self.user_id_obj = UserID.from_string(self.user_id) + self.device_id = "dev-1" + # In this test setup the main process is the presence writer. + self.writer_handler = hs.get_presence_handler() + + def _make_sync_worker(self) -> tuple[Any, list, list]: + """Create a sync worker whose proxied presence calls are recorded.""" + worker = self.make_worker_hs( + "synapse.app.generic_worker", {"worker_name": "synchrotron"} + ) + presence = worker.get_presence_handler() + assert isinstance(presence, WorkerPresenceHandler) + + set_state_calls: list = [] + bump_calls: list = [] + real_set_state = presence._set_state_client + real_bump = presence._bump_active_client + + async def recording_set_state(**kwargs: Any) -> Any: + set_state_calls.append(kwargs) + return await real_set_state(**kwargs) + + async def recording_bump(**kwargs: Any) -> Any: + bump_calls.append(kwargs) + return await real_bump(**kwargs) + + presence._set_state_client = recording_set_state + presence._bump_active_client = recording_bump + return presence, set_state_calls, bump_calls + + def _sync(self, presence: Any, state: str = PresenceState.ONLINE) -> Any: + # Note: `get_success` only advances the fake clock by tiny epsilon steps + # while pumping the replication traffic; the throttle window is + # time-sensitive and these tests advance time explicitly. + return self.get_success( + presence.user_syncing(self.user_id, self.device_id, True, state), + ) + + def test_repeated_syncs_are_throttled(self) -> None: + presence, set_state_calls, _ = self._make_sync_worker() + + # Several syncs in quick succession only relay one set_state. + for _ in range(3): + self._sync(presence) + self.assertEqual(len(set_state_calls), 1) + + # The user did come online on the writer. + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + # Once the relay window has passed, the next sync relays again. + self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1) + self._sync(presence) + self.assertEqual(len(set_state_calls), 2) + + def test_state_changes_are_relayed_immediately(self) -> None: + presence, set_state_calls, _ = self._make_sync_worker() + + self._sync(presence, PresenceState.ONLINE) + self._sync(presence, PresenceState.UNAVAILABLE) + self._sync(presence, PresenceState.ONLINE) + self.assertEqual(len(set_state_calls), 3) + + # A repeat of the current state within the window is suppressed. + self._sync(presence, PresenceState.ONLINE) + self.assertEqual(len(set_state_calls), 3) + + def test_resends_after_device_stops_syncing(self) -> None: + """After a USER_SYNC stop is sent the writer may time the user out, so + a device that reconnects within the window must be relayed afresh.""" + presence, set_state_calls, _ = self._make_sync_worker() + + with self._sync(presence): + pass + self.assertEqual(len(set_state_calls), 1) + + # Wait for the going-offline grace period to elapse: USER_SYNC stop is + # sent and the throttle entry evicted. The writer then times the user + # out to offline. Advance in steps (rather than one jump) so the + # replicated stop command is delivered before the writer's timeout + # loop fires (which only starts 30s after startup). + for _ in range(4): + self.reactor.advance(12) + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.OFFLINE) + + # Reconnecting relays the state immediately, even though the presence + # value is unchanged from the last relayed one. + self._sync(presence) + self.assertEqual(len(set_state_calls), 2) + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + def test_bumps_are_throttled(self) -> None: + presence, set_state_calls, bump_calls = self._make_sync_worker() + + # While we recently relayed an online state, bumps are suppressed. + self._sync(presence, PresenceState.ONLINE) + for _ in range(3): + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_calls), 0) + + # After the window passes, a bump goes through (and then suppresses + # further bumps). + self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1) + for _ in range(2): + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_calls), 1) + + def test_explicit_set_state_always_relayed_and_resets(self) -> None: + """An explicit (non-sync) set_state is always relayed, and resets the + throttle so the next sync-driven update is relayed afresh.""" + presence, set_state_calls, _ = self._make_sync_worker() + + self._sync(presence, PresenceState.ONLINE) + self.assertEqual(len(set_state_calls), 1) + + # An explicit update of the same state within the window still goes + # through (it isn't sync-driven)... + self.get_success( + presence.set_state( + self.user_id_obj, + self.device_id, + {"presence": PresenceState.ONLINE}, + ), + ) + self.assertEqual(len(set_state_calls), 2) + + # ...and the following sync-driven update is relayed rather than + # suppressed, re-establishing the writer's sync timestamps. + self._sync(presence, PresenceState.ONLINE) + self.assertEqual(len(set_state_calls), 3) + + def test_bump_after_non_online_state_goes_through(self) -> None: + presence, set_state_calls, bump_calls = self._make_sync_worker() + + # The user is unavailable; a bump may un-idle them so it must not be + # suppressed. + self._sync(presence, PresenceState.UNAVAILABLE) + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_calls), 1) + + @override_config({"presence": {"sync_online_timeout": "12s"}}) + def test_relay_interval_scales_with_config(self) -> None: + """The throttle window is derived from the configurable presence timers, + so it stays comfortably below a lowered sync online timeout rather than + being a hardcoded 25s (which would make users flap).""" + presence, set_state_calls, _ = self._make_sync_worker() + + # 5/6 of min(12s sync online timeout, 60s default last-active + # granularity). + self.assertEqual(presence._sync_presence_relay_interval, 10 * 1000) + + # A repeat within the (now shorter) window is still suppressed... + self._sync(presence) + self._sync(presence) + self.assertEqual(len(set_state_calls), 1) + + # ...and once it passes, the next sync relays again. + self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1) + self._sync(presence) + self.assertEqual(len(set_state_calls), 2)