From 55c7431edef7549e8f3c3725d9ea08f66d512457 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 10 Jul 2026 11:03:42 +0000 Subject: [PATCH 1/3] Treat an actively-syncing device as active in the presence timeouts A device that is currently syncing is now treated as active for as long as it keeps syncing: `handle_timeout` no longer auto-idles it, and it stays `currently_active`. A federation-ping wheel timer keeps a quietly-syncing user's keep-alives flowing once its other timers lapse. This matches the historic behaviour where every sync bumped the device's last active time, and is a prerequisite for driving presence over USER_SYNC (next commit): once the writer is no longer told a device's state on every sync, it must treat syncing-set membership itself as ongoing activity, or a continuously-syncing user would wrongly idle. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PRR4mYoQDuv2k1FwEyreL6 --- synapse/handlers/presence.py | 70 ++++++++++-- tests/handlers/test_presence.py | 195 ++++++++++++++++++++++++++++---- 2 files changed, 237 insertions(+), 28 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 28fbbfc1e17..5faf3c4c57f 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -1033,6 +1033,7 @@ async def _update_states( self, new_states: Iterable[UserPresenceState], force_notify: bool = False, + syncing_user_ids: AbstractSet[str] = frozenset(), ) -> None: """Updates presence of users. Sets the appropriate timeouts. Pokes the notifier and federation if and only if the changed presence state @@ -1044,6 +1045,9 @@ async def _update_states( even if it doesn't change the state of a user's presence (e.g online -> online). This is currently used to bump the max presence stream ID without changing any user's presence (see PresenceHandler.add_users_to_send_full_presence_to). + syncing_user_ids: Users with an active sync. Such users are treated as + currently active for as long as they keep syncing, matching the + historic behaviour where each sync bumped their last active time. """ if not self._presence_enabled: # We shouldn't get here if presence is disabled, but we check anyway @@ -1093,6 +1097,7 @@ async def _update_states( idle_timer=self._idle_timer, sync_online_timeout=self._sync_online_timeout, last_active_granularity=self._last_active_granularity, + is_syncing=user_id in syncing_user_ids, ) if force_notify: @@ -1210,7 +1215,8 @@ async def _handle_timeouts(self) -> None: last_active_granularity=self._last_active_granularity, ) - return await self._update_states(changes) + syncing_user_ids = {user_id for user_id, _ in syncing_user_devices} + return await self._update_states(changes, syncing_user_ids=syncing_user_ids) async def bump_presence_active_time( self, user: UserID, device_id: str | None @@ -2261,17 +2267,34 @@ def handle_timeout( # due to timeouts. device_changed = False offline_devices = [] + # Whether any of the user's devices is currently syncing. A syncing + # device is treated as active (see below), so this keeps the user + # `currently_active` without relying on a periodic sync bumping their + # last active time. + user_is_syncing = False for device_id, device_state in user_devices.items(): + device_is_syncing = (state.user_id, device_id) in syncing_device_ids + user_is_syncing = user_is_syncing or device_is_syncing + if device_state.state == PresenceState.ONLINE: - if now - device_state.last_active_ts > idle_timer: + if ( + not device_is_syncing + and now - device_state.last_active_ts > idle_timer + ): # Currently online, but last activity ages ago so auto - # idle + # idle. + # + # A device that is actively syncing is treated as active and + # never auto-idled: historically each sync bumped the device's + # last active time, so a syncing device never idled. (Clients + # that want to appear idle say so via the sync's presence + # state.) device_state.state = PresenceState.UNAVAILABLE device_changed = True # If there are have been no sync for a while (and none ongoing), # set presence to offline. - if (state.user_id, device_id) not in syncing_device_ids: + if not device_is_syncing: # If the user has done something recently but hasn't synced, # don't set them as offline. sync_or_active = max( @@ -2303,9 +2326,13 @@ def handle_timeout( state = state.copy_and_replace(state=new_presence) changed = True - if now - state.last_active_ts > last_active_granularity: - # So that we send down a notification that we've - # stopped updating. + if not user_is_syncing and now - state.last_active_ts > last_active_granularity: + # So that we send down a notification that we've stopped updating. + # + # Skipped while the user is syncing: they are treated as active (and + # kept `currently_active` in handle_update), so we neither notify that + # they've stopped nor busy-loop re-firing this timer against a last + # active time that no longer advances on every sync. changed = True if now - state.last_federation_update_ts > FEDERATION_PING_INTERVAL: @@ -2336,6 +2363,7 @@ def handle_update( idle_timer: int, sync_online_timeout: int, last_active_granularity: int, + is_syncing: bool = False, ) -> tuple[UserPresenceState, bool, bool]: """Given a presence update: 1. Add any appropriate timers. @@ -2355,6 +2383,9 @@ def handle_update( is marked as offline. last_active_granularity: How long in ms a user counts as "currently active" after their last activity. + is_syncing: Whether the user currently has an active sync. Such a user is + treated as currently active for as long as they keep syncing, matching + the historic behaviour where each sync bumped their last active time. Returns: 3-tuple: `(new_state, persist_and_notify, federation_ping)` where: @@ -2377,10 +2408,16 @@ def handle_update( now=now, obj=user_id, then=new_state.last_active_ts + idle_timer ) - active = now - new_state.last_active_ts < last_active_granularity + # A syncing user is treated as active for as long as they sync, so + # that they don't lose `currently_active` (and flip back and forth) + # now that a sync no longer bumps their last active time on every + # request. + active = ( + is_syncing or now - new_state.last_active_ts < last_active_granularity + ) new_state = new_state.copy_and_replace(currently_active=active) - if active and not persist: + if active and not persist and not is_syncing: wheel_timer.insert( now=now, obj=user_id, @@ -2396,6 +2433,21 @@ def handle_update( then=new_state.last_user_sync_ts + sync_online_timeout, ) + # A quietly-syncing user generates no other presence updates, so + # once their idle/sync timers lapse nothing would re-check them + # and we'd stop sending federation keep-alives. Insert a timer at + # the federation ping interval to keep re-checking them; it + # re-arms itself each time it fires (see handle_timeout). Only + # needed for syncing users — anyone else times out to offline + # within SYNC_ONLINE_TIMEOUT. + if is_syncing: + wheel_timer.insert( + now=now, + obj=user_id, + then=new_state.last_federation_update_ts + + FEDERATION_PING_INTERVAL, + ) + last_federate = new_state.last_federation_update_ts if now - last_federate > FEDERATION_PING_INTERVAL: # Been a while since we've poked remote servers diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 88d9d2aebba..353000df464 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -411,6 +411,51 @@ def test_online_to_idle(self) -> None: any_order=True, ) + def test_online_while_syncing_stays_currently_active(self) -> None: + """A syncing user is treated as active even if their last active time is + stale, so they stay `currently_active` (and don't flip back and forth) + now that a sync no longer bumps their last active time on every request. + """ + wheel_timer = Mock() + user_id = "@foo:bar" + now = 5000000 + + prev_state = UserPresenceState.default(user_id) + prev_state = prev_state.copy_and_replace( + state=PresenceState.ONLINE, + last_active_ts=now - DEFAULT_LAST_ACTIVE_GRANULARITY - 1, + currently_active=True, + ) + + new_state = prev_state.copy_and_replace(state=PresenceState.ONLINE) + + state, persist_and_notify, _ = handle_update( + prev_state, + new_state, + is_mine=True, + our_server_name=self.hs.hostname, + 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, + is_syncing=True, + ) + + # Despite the stale last active time, the user stays currently active + # because they are syncing, so there's nothing to notify about. + self.assertTrue(state.currently_active) + self.assertFalse(persist_and_notify) + + # A federation ping timer is inserted so a quietly-syncing user is still + # re-checked (and keeps sending keep-alives) once their other timers lapse. + wheel_timer.insert.assert_any_call( + now=now, + obj=user_id, + then=new_state.last_federation_update_ts + FEDERATION_PING_INTERVAL, + ) + def test_persisting_presence_updates(self) -> None: """Tests that the latest presence state for each user is persisted correctly""" # Create some test users and presence states for them @@ -630,6 +675,51 @@ def test_idle_timer(self) -> None: self.assertEqual(new_state.state, PresenceState.UNAVAILABLE) self.assertEqual(new_state.status_msg, status_msg) + def test_idle_timer_does_not_fire_while_syncing(self) -> None: + """A device that is actively syncing is treated as active and is not + auto-idled, even if its last active time is older than the idle timer. + + This matches a client that keeps re-syncing with the same state: under + the pure-USER_SYNC model the writer isn't told about every sync, so it + must treat syncing-set membership itself as ongoing activity. + """ + user_id = "@foo:bar" + device_id = "dev-1" + status_msg = "I'm here!" + now = 5000000 + + state = UserPresenceState.default(user_id) + state = state.copy_and_replace( + state=PresenceState.ONLINE, + last_active_ts=now - DEFAULT_IDLE_TIMER - 1, + last_user_sync_ts=now, + last_federation_update_ts=now, + status_msg=status_msg, + ) + device_state = UserDevicePresenceState( + user_id=user_id, + device_id=device_id, + state=state.state, + last_active_ts=state.last_active_ts, + last_sync_ts=state.last_user_sync_ts, + ) + + new_state = handle_timeout( + state, + is_mine=True, + # The device is still syncing. + 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, + ) + + # Nothing changed: the device stays online rather than idling. + self.assertIsNone(new_state) + self.assertEqual(device_state.state, PresenceState.ONLINE) + def test_busy_no_idle(self) -> None: """ Tests that a user setting their presence to busy but idling doesn't turn their @@ -1063,23 +1153,34 @@ def test_sync_online_timeout(self) -> None: @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. + """Idle is driven by the client, not a server-side timer. + + A device that keeps syncing with a presence state of "online" is treated + as active and is never auto-idled by the server, however long the + (configured) idle timeout is; the server trusts the client's declared + state. The client signals idle by syncing with a presence of + "unavailable" instead, which the server applies immediately. + """ + # Keep a sync open declaring the device online. 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) + # Even well past the configured idle timeout, a syncing client that keeps + # declaring itself online stays online. + self.reactor.advance(2 * 20 * 60) self.reactor.pump([5]) self.assertEqual(self._get_state().state, PresenceState.ONLINE) - # Past the configured timeout: idle. - self.reactor.advance(15 * 60) + # The client itself signals idle by syncing with presence "unavailable", + # which is applied straight away. + self.get_success( + self.presence_handler.user_syncing( + self.user_id, self.device_id, True, PresenceState.UNAVAILABLE + ) + ) self.reactor.pump([5]) self.assertEqual(self._get_state().state, PresenceState.UNAVAILABLE) @@ -1278,21 +1379,77 @@ def test_set_presence_from_syncing_is_set(self) -> None: # we should now be online self.assertEqual(state.state, PresenceState.ONLINE) + def test_syncing_does_not_idle(self) -> None: + """A device that keeps syncing stays online past the idle timer. + + Under pure USER_SYNC the writer is only told about the first sync, so it + must keep a syncing device active itself rather than idling it once the + (now un-refreshed) last active time gets old. + """ + # Use a local user so the presence timers apply. + user_id = f"@test:{self.hs.config.server.server_name}" + user_id_obj = UserID.from_string(user_id) + + # Start syncing and leave the sync open (context is never closed). + self.get_success( + self.presence_handler.user_syncing( + user_id, self.device_id, True, PresenceState.ONLINE + ) + ) + + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + # Advance well past the idle timer, pumping so _handle_timeouts runs. + self.reactor.advance(DEFAULT_IDLE_TIMER / 1000 * 2) + self.reactor.pump([0.1]) + + # The device is still syncing, so it stays online (and currently active). + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + self.assertTrue(state.currently_active) + + def test_sync_state_change_is_applied(self) -> None: + """Changing the requested presence state on a later sync for the same + device (without the first sync ending) is applied.""" + user_id = f"@test:{self.hs.config.server.server_name}" + user_id_obj = UserID.from_string(user_id) + + # First sync: online. + self.get_success( + self.presence_handler.user_syncing( + user_id, self.device_id, True, PresenceState.ONLINE + ) + ) + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + # Second sync on the same device requesting a different state. + self.get_success( + self.presence_handler.user_syncing( + user_id, self.device_id, True, PresenceState.UNAVAILABLE + ) + ) + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.UNAVAILABLE) + @parameterized.expand( # A list of tuples of 4 strings: # # * The presence state of device 1. # * The presence state of device 2. # * The expected user presence state after both devices have synced. - # * The expected user presence state after device 1 has idled. - # * The expected user presence state after device 2 has idled. + # * The expected user presence state once device 1 has been syncing past + # the idle timer (unchanged: a syncing device is treated as active). + # * The expected user presence state once device 2 has also been syncing + # past the idle timer (also unchanged). # * True to use workers, False a monolith. [ (*cases, workers) for workers in (False, True) for cases in [ - # If both devices have the same state, online should eventually idle. - # Otherwise, the state doesn't change. + # A device that keeps syncing is treated as active, so the + # combined state doesn't change as the idle timer elapses. ( PresenceState.BUSY, PresenceState.BUSY, @@ -1305,7 +1462,7 @@ def test_set_presence_from_syncing_is_set(self) -> None: PresenceState.ONLINE, PresenceState.ONLINE, PresenceState.ONLINE, - PresenceState.UNAVAILABLE, + PresenceState.ONLINE, ), ( PresenceState.UNAVAILABLE, @@ -1348,15 +1505,15 @@ def test_set_presence_from_syncing_is_set(self) -> None: PresenceState.ONLINE, PresenceState.UNAVAILABLE, PresenceState.ONLINE, - PresenceState.UNAVAILABLE, - PresenceState.UNAVAILABLE, + PresenceState.ONLINE, + PresenceState.ONLINE, ), ( PresenceState.ONLINE, PresenceState.OFFLINE, PresenceState.ONLINE, - PresenceState.UNAVAILABLE, - PresenceState.UNAVAILABLE, + PresenceState.ONLINE, + PresenceState.ONLINE, ), ( PresenceState.UNAVAILABLE, @@ -1392,14 +1549,14 @@ def test_set_presence_from_syncing_is_set(self) -> None: PresenceState.ONLINE, PresenceState.ONLINE, PresenceState.ONLINE, - PresenceState.UNAVAILABLE, + PresenceState.ONLINE, ), ( PresenceState.OFFLINE, PresenceState.ONLINE, PresenceState.ONLINE, PresenceState.ONLINE, - PresenceState.UNAVAILABLE, + PresenceState.ONLINE, ), ( PresenceState.OFFLINE, From 2684275e1af8fd4aaed3151e5795bdc8d6d60f48 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 10 Jul 2026 11:04:17 +0000 Subject: [PATCH 2/3] Drive presence over USER_SYNC instead of a per-sync set_state relay Previously a worker proxied a ReplicationPresenceSetState HTTP call to the presence writer on every sync request, which made the writer's load scale with the client sync rate and dominated its CPU on large deployments. Instead, carry presence over the existing USER_SYNC replication command: * UserSyncCommand gains a presence_state field. Workers edge-trigger it: a USER_SYNC is sent only when a device first starts syncing or changes the presence state it is syncing with, so a steady sync loop generates no presence traffic. The reconnect resync and an explicit REST set_state are handled so the writer's view stays correct. * The writer applies the state from USER_SYNC (factored into _apply_sync_state). Combined with the previous commit, a device is treated as active for as long as it stays in the syncing set, so it neither idles nor goes offline while syncing without the writer being told about every sync. * A new USER_SYNC_KEEPALIVE command lets a worker periodically remind the writer that its syncing users are still live, so a quiet worker isn't expired (which would otherwise mark all its users offline). * The sync-driven half of the relay throttle (#19941) is superseded: syncs no longer relay a set_state at all, so the throttle now only covers the per-user-action activity bumps. Its entries are reset whenever the device's state on the writer may have changed (an explicit set_state, a change of sync state, or the device stopping syncing), so a bump that may un-idle the device is always relayed immediately. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PRR4mYoQDuv2k1FwEyreL6 --- synapse/handlers/presence.py | 408 +++++++++++++++++++--------- synapse/replication/tcp/commands.py | 67 ++++- synapse/replication/tcp/handler.py | 37 ++- tests/handlers/test_presence.py | 223 ++++++++++----- 4 files changed, 531 insertions(+), 204 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 5faf3c4c57f..a53bee52796 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -205,6 +205,12 @@ # syncing. UPDATE_SYNCING_USERS = Duration(seconds=10) +# How often a worker reminds the presence writer that its set of syncing users +# is still live. Must be comfortably shorter than EXTERNAL_PROCESS_EXPIRY, +# otherwise a worker whose users are all quietly syncing (and so sends no +# USER_SYNC commands) would be expired and its users marked offline. +SYNC_PRESENCE_KEEPALIVE_INTERVAL = Duration(minutes=1) + class BasePresenceHandler(abc.ABC): """Parts of the PresenceHandler that are shared between workers and presence @@ -275,15 +281,16 @@ async def user_syncing( @abc.abstractmethod def get_currently_syncing_users_for_replication( self, - ) -> Iterable[tuple[str, str | None]]: + ) -> Iterable[tuple[str, str | None, str]]: """Get an iterable of syncing users and devices on this worker, to send to the presence handler This is called when a replication connection is established. It should return - a list of tuples of user ID & device ID, which are then sent as USER_SYNC commands - to inform the process handling presence about those users/devices. + a list of tuples of user ID, device ID & the presence state the device is + syncing with, which are then sent as USER_SYNC commands to inform the + process handling presence about those users/devices. Returns: - An iterable of tuples of user ID and device ID. + An iterable of tuples of user ID, device ID and presence state. """ async def get_state(self, target_user: UserID) -> UserPresenceState: @@ -376,6 +383,7 @@ async def update_external_syncs_row( # noqa: B027 (no-op by design) user_id: str, device_id: str | None, is_syncing: bool, + presence_state: str, sync_time_msec: int, ) -> None: """Update the syncing users for an external process as a delta. @@ -389,9 +397,20 @@ async def update_external_syncs_row( # noqa: B027 (no-op by design) user_id: The user who has started or stopped syncing device_id: The user's device that has started or stopped syncing is_syncing: Whether or not the user is now syncing + presence_state: The presence state the device is syncing with. Only + meaningful when the device is starting (or changing) its sync. sync_time_msec: Time in ms when the user was last syncing """ + async def update_external_syncs_keepalive( # noqa: B027 (no-op by design) + self, process_id: str + ) -> None: + """Refresh the "last updated" time for an external process, so that its + syncing users aren't expired while it is quiet. + + This is a no-op when presence is handled by a different worker. + """ + async def update_external_syncs_clear( # noqa: B027 (no-op by design) self, process_id: str ) -> None: @@ -530,25 +549,29 @@ 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 = ( + # How often to relay repeated activity bumps for a device to the + # presence writer. The relayed bumps are what feed the writer's device + # last_active_ts timer, so this must sit comfortably below the + # (configurable) last-active granularity — or users would drop out of + # "currently active" between relays. We use 5/6 of the tighter of it + # and the sync online timeout, i.e. the historic 25s at the defaults. + self._bump_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]] = {} + # (user_id, device_id) -> the time we last relayed an activity bump for + # the device to the presence writer. Used to suppress the per-user-action + # bump calls, which are no-ops on the writer at finer granularity than + # its timers: one bump per relay interval is enough to keep them fed. + # Entries older than the window are swept by `_sweep_last_relayed_bumps`. + self._last_relayed_bump_ms: dict[tuple[str, str | None], int] = {} + + # (user_id, device_id) -> presence state. The presence state we last told + # the presence writer each syncing device is in. Used to avoid re-sending + # a USER_SYNC when a sync repeats the same state, and to work out what to + # send on reconnect. An entry exists iff the writer believes the device is + # syncing. + self._user_device_last_sync_state: dict[tuple[str, str | None], str] = {} self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs) self._set_state_client = ReplicationPresenceSetState.make_client(hs) @@ -556,7 +579,10 @@ def __init__(self, hs: "HomeServer"): 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) + self.send_sync_keepalive, SYNC_PRESENCE_KEEPALIVE_INTERVAL + ) + self.clock.looping_call( + self._sweep_last_relayed_bumps, Duration(minutes=30) ) hs.register_async_shutdown_handler( @@ -577,22 +603,57 @@ def send_user_sync( user_id: str, device_id: str | None, is_syncing: bool, + presence_state: str, last_sync_ms: int, ) -> None: if self._track_presence: self.hs.get_replication_command_handler().send_user_sync( - self.instance_id, user_id, device_id, is_syncing, last_sync_ms + self.instance_id, + user_id, + device_id, + is_syncing, + presence_state, + last_sync_ms, ) - def mark_as_coming_online(self, user_id: str, device_id: str | None) -> None: - """A user has started syncing. Send a UserSync to the presence writer, - unless they had recently stopped syncing. + def send_sync_keepalive(self) -> None: + """Periodically remind the presence writer that our syncing users are + still live, so it doesn't expire us during quiet periods (when no device + is starting, stopping or changing its sync). """ - going_offline = self._user_devices_going_offline.pop((user_id, device_id), None) - if not going_offline: - # Safe to skip because we haven't yet told the presence writer they - # were offline - self.send_user_sync(user_id, device_id, True, self.clock.time_msec()) + if self._track_presence and any( + self._user_device_to_num_current_syncs.values() + ): + self.hs.get_replication_command_handler().send_user_sync_keepalive( + self.instance_id + ) + + def mark_as_coming_online( + self, user_id: str, device_id: str | None, presence_state: str + ) -> None: + """A user's device has started syncing, or changed the presence state it + is syncing with. Tell the presence writer, unless it already believes the + device is syncing in this exact state (e.g. the device only briefly + stopped syncing, or the state is unchanged since the last sync). + """ + key = (user_id, device_id) + # Cancel any pending "stopped syncing" notification: the device is (still) + # syncing. + self._user_devices_going_offline.pop(key, None) + + if self._user_device_last_sync_state.get(key) == presence_state: + # The writer already knows this device is syncing in this state, so + # there's nothing to send. + return + + self._user_device_last_sync_state[key] = presence_state + # The device's state on the writer is about to change; drop its + # bump-throttle entry so a following bump (which may un-idle the new + # state) is relayed immediately. + self._last_relayed_bump_ms.pop(key, None) + self.send_user_sync( + user_id, device_id, True, presence_state, self.clock.time_msec() + ) def mark_as_going_offline(self, user_id: str, device_id: str | None) -> None: """A user has stopped syncing. We wait before notifying the presence @@ -613,24 +674,29 @@ def send_stop_syncing(self) -> None: self._user_devices_going_offline.items() ): 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. + key = (user_id, device_id) + self._user_devices_going_offline.pop(key, None) + # The writer no longer believes this device is syncing; forget the + # state we last sent so a fresh sync re-sends it, and drop the + # device's bump-throttle entry. + last_state = self._user_device_last_sync_state.pop( + key, PresenceState.OFFLINE + ) + self._last_relayed_bump_ms.pop(key, None) + self.send_user_sync(user_id, device_id, False, last_state, last_sync_ms) + + def _sweep_last_relayed_bumps(self) -> None: + """Drop expired bump-throttling entries. + + Entries are dropped in `send_stop_syncing` when a device stops syncing, + but we add a safety net here to ensure that the dict doesn'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) + for key, last_relayed_ms in list(self._last_relayed_bump_ms.items()): + if now - last_relayed_ms >= self._bump_relay_interval: + self._last_relayed_bump_ms.pop(key, None) async def user_syncing( self, @@ -647,23 +713,19 @@ async def user_syncing( if not affect_presence or not self._track_presence: return _NullContextManager() - # Note that this causes last_active_ts to be incremented which is not - # 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, - state={"presence": presence_state}, - is_sync=True, - ) - curr_sync = self._user_device_to_num_current_syncs.get((user_id, device_id), 0) self._user_device_to_num_current_syncs[(user_id, device_id)] = curr_sync + 1 - # If this is the first in-flight sync, notify replication - if self._user_device_to_num_current_syncs[(user_id, device_id)] == 1: - self.mark_as_coming_online(user_id, device_id) + # Tell the presence writer that this device is syncing (and with what + # presence state). This is a no-op if the writer already believes the + # device is syncing in this state, so an ongoing sync loop that keeps + # requesting the same state generates no replication traffic; only the + # first sync and any change of state are sent. + # + # Note that, unlike the old behaviour, we no longer proxy a set_state to + # the writer on every sync: the writer treats a syncing device as active + # for as long as it is syncing (see PresenceHandler.handle_timeout). + self.mark_as_coming_online(user_id, device_id, presence_state) def _end() -> None: # We check that the user_id is in user_to_num_current_syncs because @@ -749,10 +811,18 @@ async def process_replication_rows( def get_currently_syncing_users_for_replication( self, - ) -> Iterable[tuple[str, str | None]]: + ) -> Iterable[tuple[str, str | None, str]]: return [ - user_id_device_id - for user_id_device_id, count in self._user_device_to_num_current_syncs.items() + ( + user_id, + device_id, + self._user_device_last_sync_state.get( + (user_id, device_id), PresenceState.ONLINE + ), + ) + for (user_id, device_id), count in ( + self._user_device_to_num_current_syncs.items() + ) if count > 0 ] @@ -787,27 +857,16 @@ 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) + if not is_sync: + # An explicit state change (e.g. a client PUT /presence) is applied + # on the writer directly below. Forget the state we last told the + # writer this device was syncing with, so that the device's next sync + # re-asserts its sync state (which may differ from the state just set). + self._user_device_last_sync_state.pop((user_id, device_id), None) + # Also forget the last relayed bump: the state just set may be one + # that a bump would un-idle, so the next bump must be relayed + # immediately rather than suppressed. + self._last_relayed_bump_ms.pop((user_id, device_id), None) # Proxy request to instance that writes presence await self._set_state_client( @@ -832,20 +891,20 @@ async def bump_presence_active_time( 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. + # flipping an idle device back online. If we relayed a bump within the + # relay window the writer's last_active_ts is already fresh and this + # bump is a no-op: skip it. (Entries are dropped whenever the device's + # state may have changed on the writer — an explicit set_state, a change + # of sync state, or the device stopping syncing — so a bump that might + # un-idle the device always goes through immediately.) now = self.clock.time_msec() - last_sent = self._last_sent_presence.get((user_id, device_id)) + last_relayed_ms = self._last_relayed_bump_ms.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 + last_relayed_ms is not None + and now - last_relayed_ms < self._bump_relay_interval ): return - self._last_sent_presence[(user_id, device_id)] = (PresenceState.ONLINE, now) + self._last_relayed_bump_ms[(user_id, device_id)] = now # Proxy request to instance that writes presence await self._bump_active_client( @@ -932,6 +991,12 @@ def __init__(self, hs: "HomeServer"): # this is non zero a user will never go offline. self._user_device_to_num_current_syncs: dict[tuple[str, str | None], int] = {} + # (user_id, device_id) -> presence state we last applied for a local sync. + # Used to edge-trigger: we only re-apply a device's sync state when it + # first starts syncing or changes the state it is syncing with, rather + # than on every sync request. + self._user_device_last_sync_state: dict[tuple[str, str | None], str] = {} + # Keeps track of the number of *ongoing* syncs on other processes. # # While any sync is ongoing on another process the user's device will never @@ -1280,30 +1345,44 @@ async def user_syncing( if not affect_presence or not self._track_presence: return _NullContextManager() - curr_sync = self._user_device_to_num_current_syncs.get((user_id, device_id), 0) - self._user_device_to_num_current_syncs[(user_id, device_id)] = curr_sync + 1 + key = (user_id, device_id) + curr_sync = self._user_device_to_num_current_syncs.get(key, 0) + self._user_device_to_num_current_syncs[key] = curr_sync + 1 - # Note that this causes last_active_ts to be incremented which is not - # what the spec wants. - await self.set_state( - UserID.from_string(user_id), - device_id, - state={"presence": presence_state}, - is_sync=True, - ) + # Edge-trigger the presence state: apply it when the device first starts + # syncing or when the requested state changes, rather than on every sync. + # While the device keeps syncing it is treated as active (see + # handle_timeout), so repeating the same state is a no-op. + if ( + curr_sync == 0 + or self._user_device_last_sync_state.get(key) != presence_state + ): + self._user_device_last_sync_state[key] = presence_state + await self._apply_sync_state(user_id, device_id, presence_state) async def _end() -> None: try: - self._user_device_to_num_current_syncs[(user_id, device_id)] -= 1 + # `_user_device_to_num_current_syncs` may have been cleared if we + # are shutting down. + if key not in self._user_device_to_num_current_syncs: + return - prev_state = await self.current_state_for_user(user_id) - await self._update_states( - [ - prev_state.copy_and_replace( - last_user_sync_ts=self.clock.time_msec() - ) - ] - ) + count = self._user_device_to_num_current_syncs[key] - 1 + self._user_device_to_num_current_syncs[key] = count + + # Only when the last in-flight sync for the device ends do we + # record it as no longer syncing (bumping last_user_sync_ts so it + # can time out) and forget the state we were tracking. + if count == 0: + self._user_device_last_sync_state.pop(key, None) + prev_state = await self.current_state_for_user(user_id) + await self._update_states( + [ + prev_state.copy_and_replace( + last_user_sync_ts=self.clock.time_msec() + ) + ] + ) except Exception: logger.exception("Error updating presence after sync") @@ -1318,7 +1397,7 @@ def _user_syncing() -> Generator[None, None, None]: def get_currently_syncing_users_for_replication( self, - ) -> Iterable[tuple[str, str | None]]: + ) -> Iterable[tuple[str, str | None, str]]: # since we are the process handling presence, there is nothing to do here. return [] @@ -1328,6 +1407,7 @@ async def update_external_syncs_row( user_id: str, device_id: str | None, is_syncing: bool, + presence_state: str, sync_time_msec: int, ) -> None: """Update the syncing users for an external process as a delta. @@ -1339,30 +1419,32 @@ async def update_external_syncs_row( user_id: The user who has started or stopped syncing device_id: The user's device that has started or stopped syncing is_syncing: Whether or not the user is now syncing + presence_state: The presence state the device is syncing with. Only + used when the device is starting (or changing) its sync. sync_time_msec: Time in ms when the user was last syncing """ async with self.external_sync_linearizer.queue(process_id): - prev_state = await self.current_state_for_user(user_id) - process_presence = self.external_process_to_current_syncs.setdefault( process_id, set() ) - # USER_SYNC is sent when a user's device starts or stops syncing on - # a remote # process. (But only for the initial and last sync for that - # device.) + # A worker sends USER_SYNC "start" when a device first starts syncing + # and whenever the presence state it is syncing with changes, and + # "end" when the device stops syncing (see WorkerPresenceHandler). # - # When a device *starts* syncing it also calls set_state(...) which - # will update the state, last_active_ts, and last_user_sync_ts. - # Simply ensure the user & device is tracked as syncing in this case. + # On start (or a state change) apply the presence state and mark the + # device as syncing. The device is then treated as active for as long + # as it remains in `process_presence` (see handle_timeout), so — unlike + # the old per-sync set_state relay — no further updates are needed + # until the state changes or the device stops. # - # When a device *stops* syncing, update the last_user_sync_ts and mark - # them as no longer syncing. Note this doesn't quite match the - # monolith behaviour, which updates last_user_sync_ts at the end of - # every sync, not just the last in-flight sync. - if is_syncing and (user_id, device_id) not in process_presence: + # On stop, update last_user_sync_ts and mark the device as no longer + # syncing, so it can time out. + if is_syncing: process_presence.add((user_id, device_id)) - elif not is_syncing and (user_id, device_id) in process_presence: + await self._apply_sync_state(user_id, device_id, presence_state) + elif (user_id, device_id) in process_presence: + prev_state = await self.current_state_for_user(user_id) devices = self._user_to_device_to_current_state.setdefault(user_id, {}) device_state = devices.setdefault( device_id, UserDevicePresenceState.default(user_id, device_id) @@ -1378,6 +1460,22 @@ async def update_external_syncs_row( self.external_process_last_updated_ms[process_id] = self.clock.time_msec() + async def update_external_syncs_keepalive(self, process_id: str) -> None: + """Refresh the "last updated" time for an external process so that its + syncing users aren't expired during quiet periods. + + Sent periodically by workers that have syncing users but haven't had any + starts/stops/state-changes to report (see + WorkerPresenceHandler.send_sync_keepalive). + """ + async with self.external_sync_linearizer.queue(process_id): + # Only refresh a process we're actually tracking; a keepalive for an + # unknown process (e.g. one we've already expired) is ignored. + if process_id in self.external_process_last_updated_ms: + self.external_process_last_updated_ms[process_id] = ( + self.clock.time_msec() + ) + async def update_external_syncs_clear(self, process_id: str) -> None: """Marks all users that had been marked as syncing by a given process as offline. @@ -1487,6 +1585,58 @@ async def incoming_presence(self, origin: str, content: JsonDict) -> None: ).inc(len(updates)) await self._update_states(updates) + async def _apply_sync_state( + self, user_id: str, device_id: str | None, presence_state: str + ) -> None: + """Mark a device as actively syncing in the given presence state. + + This is the equivalent of ``set_state(..., is_sync=True)`` but is driven + directly by USER_SYNC (rather than a per-sync ``set_state`` relay). It is + applied when a device first starts syncing and whenever the presence + state it is syncing with changes; the device is then treated as active + for as long as it keeps syncing (see ``handle_timeout``). + """ + presence = presence_state + if presence not in self.VALID_PRESENCE: + logger.warning( + "Ignoring invalid presence state %r from a sync for %s", + presence_state, + user_id, + ) + presence = PresenceState.ONLINE + + now = self.clock.time_msec() + prev_state = await self.current_state_for_user(user_id) + + # Syncs do not override a previous presence of busy. + # + # TODO: This is a hack for lack of multi-device support. Unfortunately + # removing this requires coordination with clients. + if prev_state.state == PresenceState.BUSY: + presence = PresenceState.BUSY + + # Update the device specific information. + devices = self._user_to_device_to_current_state.setdefault(user_id, {}) + device_state = devices.setdefault( + device_id, + UserDevicePresenceState.default(user_id, device_id), + ) + device_state.state = presence + device_state.last_active_ts = now + device_state.last_sync_ts = now + + # Based on the state of each user's device calculate the new presence state. + presence = _combine_device_states(devices.values()) + + new_fields: JsonDict = {"state": presence, "last_user_sync_ts": now} + if presence == PresenceState.ONLINE or presence == PresenceState.BUSY: + new_fields["last_active_ts"] = now + + await self._update_states( + [prev_state.copy_and_replace(**new_fields)], + syncing_user_ids={user_id}, + ) + async def set_state( self, target_user: UserID, @@ -1520,6 +1670,10 @@ async def set_state( user_id = target_user.to_string() now = self.clock.time_msec() + # Forget any state we were tracking for a sync on this device: an + # explicit state change means the next sync should re-assert its state. + self._user_device_last_sync_state.pop((user_id, device_id), None) + prev_state = await self.current_state_for_user(user_id) # Syncs do not override a previous presence of busy. diff --git a/synapse/replication/tcp/commands.py b/synapse/replication/tcp/commands.py index 0c85fb36fcc..175aa983b63 100644 --- a/synapse/replication/tcp/commands.py +++ b/synapse/replication/tcp/commands.py @@ -271,16 +271,26 @@ class UserSyncCommand(Command): This is used by the process handling presence (typically the master) to calculate who is online and who is not. - Includes a timestamp of when the last user sync was. + Includes the presence state the sync requested and a timestamp of when the + last user sync was. Format:: - USER_SYNC + USER_SYNC - Where is either "start" or "end" + Where is either "start" or "end", and is the + presence state the sync requested (e.g. "online" or "unavailable"). The + presence state is only meaningful when is "start". """ - __slots__ = ["instance_id", "user_id", "device_id", "is_syncing", "last_sync_ms"] + __slots__ = [ + "instance_id", + "user_id", + "device_id", + "is_syncing", + "presence_state", + "last_sync_ms", + ] NAME = "USER_SYNC" @@ -290,18 +300,22 @@ def __init__( user_id: str, device_id: str | None, is_syncing: bool, + presence_state: str, last_sync_ms: int, ): self.instance_id = instance_id self.user_id = user_id self.device_id = device_id self.is_syncing = is_syncing + self.presence_state = presence_state self.last_sync_ms = last_sync_ms @classmethod def from_line(cls: type["UserSyncCommand"], line: str) -> "UserSyncCommand": device_id: str | None - instance_id, user_id, device_id, state, last_sync_ms = line.split(" ", 4) + instance_id, user_id, device_id, state, presence_state, last_sync_ms = ( + line.split(" ", 5) + ) if device_id == "None": device_id = None @@ -309,7 +323,14 @@ def from_line(cls: type["UserSyncCommand"], line: str) -> "UserSyncCommand": if state not in ("start", "end"): raise Exception("Invalid USER_SYNC state %r" % (state,)) - return cls(instance_id, user_id, device_id, state == "start", int(last_sync_ms)) + return cls( + instance_id, + user_id, + device_id, + state == "start", + presence_state, + int(last_sync_ms), + ) def to_line(self) -> str: return " ".join( @@ -318,6 +339,7 @@ def to_line(self) -> str: self.user_id, str(self.device_id), "start" if self.is_syncing else "end", + self.presence_state, str(self.last_sync_ms), ) ) @@ -351,6 +373,37 @@ def to_line(self) -> str: return self.instance_id +class UserSyncKeepaliveCommand(Command): + """Sent periodically by a worker to let the presence writer know that its + set of syncing users is still live, even when no user has started or + stopped syncing recently. + + Without this the presence writer would expire the process (see + ``EXTERNAL_PROCESS_EXPIRY``) after a period of quiet and mark all of its + syncing users as offline. + + Format:: + + USER_SYNC_KEEPALIVE + """ + + __slots__ = ["instance_id"] + + NAME = "USER_SYNC_KEEPALIVE" + + def __init__(self, instance_id: str): + self.instance_id = instance_id + + @classmethod + def from_line( + cls: type["UserSyncKeepaliveCommand"], line: str + ) -> "UserSyncKeepaliveCommand": + return cls(line) + + def to_line(self) -> str: + return self.instance_id + + class FederationAckCommand(Command): """Sent by the client when it has processed up to a given point in the federation stream. This allows the master to drop in-memory caches of the @@ -530,6 +583,7 @@ class CancelTaskCommand(_SimpleCommand): UserIpCommand, RemoteServerUpCommand, ClearUserSyncsCommand, + UserSyncKeepaliveCommand, LockReleasedCommand, NewActiveTaskCommand, CancelTaskCommand, @@ -556,6 +610,7 @@ class CancelTaskCommand(_SimpleCommand): PingCommand.NAME, UserSyncCommand.NAME, ClearUserSyncsCommand.NAME, + UserSyncKeepaliveCommand.NAME, FederationAckCommand.NAME, UserIpCommand.NAME, ErrorCommand.NAME, diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py index ad9fed72dd8..fc1cbfd4a73 100644 --- a/synapse/replication/tcp/handler.py +++ b/synapse/replication/tcp/handler.py @@ -47,6 +47,7 @@ ReplicateCommand, UserIpCommand, UserSyncCommand, + UserSyncKeepaliveCommand, ) from synapse.replication.tcp.context import ClientContextFactory from synapse.replication.tcp.protocol import IReplicationConnection @@ -472,6 +473,7 @@ def on_USER_SYNC( cmd.user_id, cmd.device_id, cmd.is_syncing, + cmd.presence_state, cmd.last_sync_ms, ) else: @@ -485,6 +487,16 @@ def on_CLEAR_USER_SYNC( else: return None + def on_USER_SYNC_KEEPALIVE( + self, conn: IReplicationConnection, cmd: UserSyncKeepaliveCommand + ) -> Awaitable[None] | None: + if self._is_presence_writer: + return self._presence_handler.update_external_syncs_keepalive( + cmd.instance_id + ) + else: + return None + def on_FEDERATION_ACK( self, conn: IReplicationConnection, cmd: FederationAckCommand ) -> None: @@ -787,9 +799,16 @@ def new_connection(self, connection: IReplicationConnection) -> None: ) now = self._clock.time_msec() - for user_id, device_id in currently_syncing: + for user_id, device_id, presence_state in currently_syncing: connection.send_command( - UserSyncCommand(self._instance_id, user_id, device_id, True, now) + UserSyncCommand( + self._instance_id, + user_id, + device_id, + True, + presence_state, + now, + ) ) def lost_connection(self, connection: IReplicationConnection) -> None: @@ -846,13 +865,25 @@ def send_user_sync( user_id: str, device_id: str | None, is_syncing: bool, + presence_state: str, last_sync_ms: int, ) -> None: """Poke the master that a user has started/stopped syncing.""" self.send_command( - UserSyncCommand(instance_id, user_id, device_id, is_syncing, last_sync_ms) + UserSyncCommand( + instance_id, + user_id, + device_id, + is_syncing, + presence_state, + last_sync_ms, + ) ) + def send_user_sync_keepalive(self, instance_id: str) -> None: + """Poke the master that this worker's syncing users are still live.""" + self.send_command(UserSyncKeepaliveCommand(instance_id)) + def send_user_ip( self, user_id: str, diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 353000df464..e25291a0b14 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -1255,6 +1255,48 @@ def test_external_process_timeout(self) -> None: state = self.get_success(self.presence_handler.get_state(self.user_id_obj)) self.assertEqual(state.state, PresenceState.OFFLINE) + def test_external_process_keepalive(self) -> None: + """A process that keeps sending keep-alives should not have its syncing + users expired, even if no user starts/stops/changes syncing. + + This is what stops a worker whose users are all quietly syncing (and so + which sends no USER_SYNC commands under the pure-USER_SYNC model) from + being expired and having all of its users marked offline. + """ + user_id = f"@test:{self.hs.config.server.server_name}" + user_id_obj = UserID.from_string(user_id) + process_id = "proc-1" + + # A user starts syncing on the external process. + self.get_success( + self.presence_handler.update_external_syncs_row( + process_id, + user_id, + "dev-1", + True, + PresenceState.ONLINE, + self.clock.time_msec(), + ) + ) + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + # Advance past the expiry twice, sending a keep-alive each time in + # between. The user should stay online throughout. + for _ in range(3): + self.reactor.advance(EXTERNAL_PROCESS_EXPIRY * 0.75 / 1000) + self.get_success( + self.presence_handler.update_external_syncs_keepalive(process_id) + ) + + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + # Once the keep-alives stop, the process is expired and the user times out. + self.reactor.advance(EXTERNAL_PROCESS_EXPIRY * 2 / 1000) + state = self.get_success(self.presence_handler.get_state(user_id_obj)) + self.assertEqual(state.state, PresenceState.OFFLINE) + def test_user_goes_offline_by_timeout_status_msg_remain(self) -> None: """Test that if a user doesn't update the records for a while users presence goes `OFFLINE` because of timeout and `status_msg` remains. @@ -1585,12 +1627,15 @@ def test_set_presence_from_syncing_multi_device( Test the behaviour of multiple devices syncing at the same time. Roughly the user's presence state should be set to the "highest" priority - of all the devices. When a device then goes offline its state should be - discarded and the next highest should win. - - Note that these tests use the idle timer (and don't close the syncs), it - is unlikely that a *single* sync would last this long, but is close enough - to continually syncing with that current state. + of all the devices. + + The syncs are never closed, i.e. each device keeps syncing with a fixed + state. A device that is actively syncing is treated as active and is *not* + auto-idled (a client that wants to appear idle says so via the sync's + presence state), so the combined state does not change as time passes. + This matches a client that continually re-syncs with the same state; it is + only once a device stops syncing that its state is discarded (see + test_set_presence_from_non_syncing_multi_device). """ user_id = f"@test:{self.hs.config.server.server_name}" @@ -1629,6 +1674,12 @@ def test_set_presence_from_syncing_multi_device( ) ) + # On a worker the sync notifies the presence writer over replication + # (a USER_SYNC command); pump the reactor so it is delivered and applied + # before we assert. + if test_with_workers: + self.reactor.pump([0.1]) + # 4. Assert the expected presence state. state = self.get_success( self.presence_handler.get_state(UserID.from_string(user_id)) @@ -1859,6 +1910,12 @@ def test_set_presence_from_non_syncing_multi_device( ) ) + # On a worker the syncs notify the presence writer over replication + # (USER_SYNC commands); pump the reactor so they are delivered and applied + # before we assert. + if test_with_workers: + self.reactor.pump([0.1]) + # 3. Assert the expected presence state. state = self.get_success( self.presence_handler.get_state(UserID.from_string(user_id)) @@ -2481,9 +2538,10 @@ def create_fake_event_from_remote_server( 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.""" + """Tests that sync workers relay no per-sync-request presence updates to + the presence writer (presence is carried by edge-triggered USER_SYNC + commands instead), and that the per-user-action activity bumps are + throttled.""" def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.user_id = f"@throttled:{hs.config.server.server_name}" @@ -2525,133 +2583,162 @@ def _sync(self, presence: Any, state: str = PresenceState.ONLINE) -> Any: presence.user_syncing(self.user_id, self.device_id, True, state), ) - def test_repeated_syncs_are_throttled(self) -> None: + def test_syncs_relay_no_set_state(self) -> None: + """Syncs no longer proxy a set_state to the writer at all: presence is + carried by USER_SYNC, so the writer still sees the user come online.""" 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) + self.assertEqual(len(set_state_calls), 0) - # The user did come online on the writer. + # The user did come online on the writer (via USER_SYNC). Pump the + # reactor so the replicated command is delivered and applied before we + # assert. + self.reactor.pump([0.1]) 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: + def test_sync_state_changes_reach_writer(self) -> None: + """A change in the state a device is syncing with reaches the writer + immediately (via a fresh USER_SYNC), still without any set_state.""" presence, set_state_calls, _ = self._make_sync_worker() + # After each sync, pump the reactor so the replicated USER_SYNC command + # is delivered and applied before we assert on the writer's state. self._sync(presence, PresenceState.ONLINE) + self.reactor.pump([0.1]) + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + self._sync(presence, PresenceState.UNAVAILABLE) - self._sync(presence, PresenceState.ONLINE) - self.assertEqual(len(set_state_calls), 3) + self.reactor.pump([0.1]) + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.UNAVAILABLE) - # A repeat of the current state within the window is suppressed. self._sync(presence, PresenceState.ONLINE) - self.assertEqual(len(set_state_calls), 3) + self.reactor.pump([0.1]) + state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) + self.assertEqual(state.state, PresenceState.ONLINE) + + self.assertEqual(len(set_state_calls), 0) 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.""" + a device that reconnects must be told to the writer 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). + # sent. 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. + # Reconnecting sends a fresh USER_SYNC, even though the presence value + # is unchanged from the last one sent, and brings the user back online. self._sync(presence) - self.assertEqual(len(set_state_calls), 2) + self.reactor.pump([0.1]) state = self.get_success(self.writer_handler.get_state(self.user_id_obj)) self.assertEqual(state.state, PresenceState.ONLINE) + self.assertEqual(len(set_state_calls), 0) + def test_bumps_are_throttled(self) -> None: - presence, set_state_calls, bump_calls = self._make_sync_worker() + presence, _, bump_calls = self._make_sync_worker() - # While we recently relayed an online state, bumps are suppressed. self._sync(presence, PresenceState.ONLINE) + + # The first bump is relayed; repeats within the window are suppressed. 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) + self.assertEqual(len(bump_calls), 1) - # After the window passes, a bump goes through (and then suppresses - # further bumps). - self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1) + # After the window passes, the next bump goes through (and then + # suppresses further bumps). + self.reactor.advance(presence._bump_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) + self.assertEqual(len(bump_calls), 2) 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() + bump throttle: the state just set may be one a bump would un-idle, so + the next bump must be relayed afresh.""" + presence, set_state_calls, bump_calls = self._make_sync_worker() - self._sync(presence, PresenceState.ONLINE) - self.assertEqual(len(set_state_calls), 1) + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_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}, + {"presence": PresenceState.UNAVAILABLE}, ), ) - self.assertEqual(len(set_state_calls), 2) + self.assertEqual(len(set_state_calls), 1) - # ...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) + # A bump within the window is relayed rather than suppressed, as it + # un-idles the state just set. + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_calls), 2) - def test_bump_after_non_online_state_goes_through(self) -> None: - presence, set_state_calls, bump_calls = self._make_sync_worker() + def test_sync_state_change_resets_bump_throttle(self) -> None: + """A change in the state a device is syncing with resets the bump + throttle, so a bump that may un-idle the new state goes through.""" + presence, _, 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._sync(presence, PresenceState.ONLINE) self.get_success( presence.bump_presence_active_time(self.user_id_obj, self.device_id), ) self.assertEqual(len(bump_calls), 1) + # The device switches to syncing as unavailable; a following bump may + # un-idle it 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), 2) + @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() + so it stays comfortably below them rather than being a hardcoded 25s.""" + presence, _, bump_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) + self.assertEqual(presence._bump_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) + # A repeated bump within the (now shorter) window is still suppressed... + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_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) + # ...and once it passes, the next bump relays again. + self.reactor.advance(presence._bump_relay_interval / 1000 + 1) + self.get_success( + presence.bump_presence_active_time(self.user_id_obj, self.device_id), + ) + self.assertEqual(len(bump_calls), 2) From 351be7e63848a8107ce17e848f7492226df7b4ad Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 10 Jul 2026 10:59:19 +0100 Subject: [PATCH 3/3] Newsfile --- changelog.d/19946.misc | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19946.misc diff --git a/changelog.d/19946.misc b/changelog.d/19946.misc new file mode 100644 index 00000000000..e928f68bc58 --- /dev/null +++ b/changelog.d/19946.misc @@ -0,0 +1 @@ +Reduce replication traffic caused by presence.