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. diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 28fbbfc1e17..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 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). + """ + 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) -> None: - """A user has started syncing. Send a UserSync to the presence writer, - unless they had recently stopped syncing. + 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). """ - 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()) + 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 @@ -1033,6 +1098,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 +1110,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 +1162,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 +1280,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 @@ -1274,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") @@ -1312,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 [] @@ -1322,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. @@ -1333,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) @@ -1372,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. @@ -1481,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, @@ -1514,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. @@ -2261,17 +2421,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 +2480,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 +2517,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 +2537,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 +2562,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 +2587,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/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 88d9d2aebba..e25291a0b14 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) @@ -1154,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. @@ -1278,21 +1421,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 +1504,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 +1547,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 +1591,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, @@ -1428,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}" @@ -1472,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)) @@ -1702,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)) @@ -2324,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}" @@ -2368,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)