diff --git a/changelog.d/19948.bugfix b/changelog.d/19948.bugfix new file mode 100644 index 0000000000..5cd34d1abf --- /dev/null +++ b/changelog.d/19948.bugfix @@ -0,0 +1 @@ +Fix presence states being shown to clients forever after presence is disabled, by marking any previously only users as offline. diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 6ee517786b..55dd5ffb59 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -122,6 +122,7 @@ from synapse.types import ( ) from synapse.util.async_helpers import Linearizer from synapse.util.duration import Duration +from synapse.util.iterutils import batch_iter from synapse.util.metrics import Measure from synapse.util.wheel_timer import WheelTimer @@ -984,6 +985,14 @@ class PresenceHandler(BasePresenceHandler): Duration(minutes=1), ) + if not self._presence_enabled and self.user_to_current_state: + # Presence is disabled but the database still contains non-offline + # presence states, i.e. presence used to be enabled. Nothing writes + # to the presence stream while presence is disabled, so without + # intervention clients would show the stale states forever. Send + # out one final round of updates marking everyone as offline. + self.clock.call_when_running(self._mark_stale_presence_as_offline) + presence_wheel_timer_size_gauge.register_hook( homeserver_instance_id=hs.get_instance_id(), hook=lambda: {(self.server_name,): len(self.wheel_timer)}, @@ -1041,6 +1050,36 @@ class PresenceHandler(BasePresenceHandler): [self.user_to_current_state[user_id] for user_id in unpersisted] ) + @wrap_as_background_process("PresenceHandler._mark_stale_presence_as_offline") + async def _mark_stale_presence_as_offline(self) -> None: + """One-off job, run at startup when presence is disabled, that marks + any non-offline presence states left over from when presence was + enabled as offline, and streams the changes out to clients. + """ + states = [ + state.copy_and_replace( + state=PresenceState.OFFLINE, + status_msg=None, + currently_active=False, + ) + for state in self.user_to_current_state.values() + if state.state != PresenceState.OFFLINE + ] + if not states: + return + + logger.info( + "Presence is disabled: marking %d stale presence states as offline", + len(states), + ) + + self.user_to_current_state.update({state.user_id: state for state in states}) + + # There may be a lot of stale states (e.g. everyone that was online + # when presence was disabled), so persist them in batches. + for batch in batch_iter(states, 500): + await self._persist_and_notify(list(batch)) + async def _update_states( self, new_states: Iterable[UserPresenceState], diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 9ecfe0da0f..a05d6c6e59 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -1759,9 +1759,19 @@ class SyncHandler: await self._generate_sync_entry_for_account_data(sync_result_builder) # Presence data is included if the server has it enabled and not filtered out. - include_presence_data = bool( - self.hs_config.server.presence_enabled - and not sync_config.filter_collection.blocks_all_presence() + presence_enabled = bool(self.hs_config.server.presence_enabled) + if not presence_enabled and since_token is not None: + # Even with presence disabled we send down any presence updates the + # client hasn't yet seen, so that the "mark everyone as offline" + # updates written when presence was disabled reach clients that + # would otherwise show the old presence states forever. The stream + # doesn't advance while presence is disabled, so once clients have + # caught up this check stops any further presence work. + presence_enabled = ( + since_token.presence_key < sync_result_builder.now_token.presence_key + ) + include_presence_data = ( + presence_enabled and not sync_config.filter_collection.blocks_all_presence() ) # Device list updates are sent if a since token is provided. include_device_list_updates = bool(since_token and since_token.device_list_key) diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index 0b638b2115..a562050842 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -1002,6 +1002,125 @@ class PresenceHandlerInitTestCase(unittest.HomeserverTestCase): ) self.assertEqual(state.state, sync_state) + @unittest.override_config({"presence": {"enabled": False}}) + def test_restored_presence_flushed_offline_when_presence_disabled(self) -> None: + """If presence is disabled, any non-offline presence states left in the + database from when presence was enabled should be marked as offline at + startup, and the updates streamed out to clients. + """ + main_store = self.hs.get_datastores().main + before_token = main_store.get_current_presence_token() + + # Get the handler, which schedules the startup flush. + presence_handler = self.hs.get_presence_handler() + + # Fire pending `call_when_running` hooks and let the flush complete. + self.reactor.run() + self.reactor.advance(0) + + # The user should now be offline, both in memory and in the database. + state = self.get_success( + presence_handler.get_state(UserID.from_string(self.user_id)) + ) + self.assertEqual(state.state, PresenceState.OFFLINE) + + db_state = self.get_success(main_store.get_presence_for_users([self.user_id]))[ + self.user_id + ] + self.assertEqual(db_state.state, PresenceState.OFFLINE) + + # The flush must advance the presence stream so that syncing clients + # are sent the offline updates. + self.assertGreater(main_store.get_current_presence_token(), before_token) + + +class PresenceDisabledSyncTestCase(unittest.HomeserverTestCase): + """Tests that stale presence states left over from when presence was + enabled reach clients over /sync, and that the startup flush marks them as + offline and sends the offline updates down /sync too. + """ + + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + ] + + @unittest.override_config({"presence": {"enabled": False}}) + def test_stale_presence_flushed_offline_and_sent_on_sync(self) -> None: + user1 = self.register_user("alice", "pass") + user1_tok = self.login(user1, "pass") + user2 = self.register_user("bob", "pass") + user2_tok = self.login(user2, "pass") + + room_id = self.helper.create_room_as(user1, tok=user1_tok) + self.helper.join(room_id, user2, tok=user2_tok) + + channel = self.make_request("GET", "/sync", access_token=user2_tok) + self.assertEqual(channel.code, 200, channel.json_body) + next_batch = channel.json_body["next_batch"] + + # Seed a stale online presence state for user1, left over from when + # presence was enabled: in the database, and in the presence handler's + # in-memory state (which at startup is preloaded from the database). + now = self.clock.time_msec() + stale_state = UserPresenceState( + user_id=user1, + state=PresenceState.ONLINE, + last_active_ts=now, + last_federation_update_ts=now, + last_user_sync_ts=now, + status_msg=None, + currently_active=True, + ) + main_store = self.hs.get_datastores().main + self.get_success(main_store.update_presence([stale_state])) + + presence_handler = self.hs.get_presence_handler() + assert isinstance(presence_handler, PresenceHandler) + presence_handler.user_to_current_state[user1] = stale_state + + # The stale state comes down user2's incremental sync, even though + # presence is disabled. + channel = self.make_request( + "GET", f"/sync?since={next_batch}", access_token=user2_tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + presence_events = channel.json_body["presence"]["events"] + self.assertEqual( + [(e["sender"], e["content"]["presence"]) for e in presence_events], + [(user1, PresenceState.ONLINE)], + ) + next_batch = channel.json_body["next_batch"] + + # Run the startup flush, as scheduled when the presence writer starts + # up with presence disabled. + self.get_success(presence_handler._mark_stale_presence_as_offline()) + + # The stale state should have been marked offline in the database... + db_state = self.get_success(main_store.get_presence_for_users([user1]))[user1] + self.assertEqual(db_state.state, PresenceState.OFFLINE) + + # ... and the offline update also comes down user2's sync. + channel = self.make_request( + "GET", f"/sync?since={next_batch}", access_token=user2_tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + presence_events = channel.json_body["presence"]["events"] + self.assertEqual( + [(e["sender"], e["content"]["presence"]) for e in presence_events], + [(user1, PresenceState.OFFLINE)], + ) + + # Once caught up, further syncs include no presence. + next_batch = channel.json_body["next_batch"] + channel = self.make_request( + "GET", f"/sync?since={next_batch}", access_token=user2_tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + self.assertEqual(channel.json_body.get("presence", {}).get("events", []), []) + # Timer values used by `PresenceConfigurableTimersTestCase`, all larger than # the corresponding defaults.