Send offline presence for stale states when presence is disabled (#19948)

If presence is disabled after having been enabled, the presence states
in the database (and hence on clients) are frozen at whatever they were
when presence was last enabled: nothing writes to the presence stream
any more and /sync omits the presence section entirely, so clients show
the old presence states forever.

Fix this in two parts:

1. At startup, if presence is disabled but the database still contains
non-offline presence states, the presence writer sends out one final
round of updates marking those users as offline.

2. /sync no longer unconditionally omits presence when presence is
disabled: incremental syncs whose since token is behind the presence
stream still get the straggling updates. As the stream doesn't advance
while presence is disabled, clients catch up once and the check then
short-circuits to a token comparison.

Remote servers already handle this themselves by timing out our users
([`FEDERATION_TIMEOUT`](https://github.com/element-hq/synapse/blob/4d8905a15a417ed0054ec2533d243932d890bbbd/synapse/handlers/presence.py#L194-L198)),
so no federation changes are needed.

Note that this only fixes the issue if presence is fully disabled. If
set to `untracked` we still have the same issue, however since modules
would still write to presence we can't just clobber everything like we
do in this patch.

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.

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],

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)

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.