Throttle sync-driven presence updates relayed to the presence writer (#19941)

Sync workers proxied a full `ReplicationPresenceSetState` call to the
presence writer on every sync request with `affect_presence=True` (via
`user_syncing`), and a `ReplicationBumpPresenceActiveTime` call on every
user action, even though the writer's presence timers only need feeding
every `SYNC_ONLINE_TIMEOUT` / `LAST_ACTIVE_GRANULARITY`. On busy clients
this amounts to tens of no-op replication calls per user per minute, and
the resulting per-update work is the dominant CPU cost on saturated
presence writers.

Track the last relayed `(state, timestamp)` per `(user, device)` on the
worker and suppress unchanged sync-driven repeats within a 25s relay
interval - deliberately below `SYNC_ONLINE_TIMEOUT` (30s) so the
writer's device `last_sync_ts`/`last_active_ts` timers stay fed and
users neither flap offline nor bounce `currently_active`. Genuine state
changes are relayed immediately, explicit (non-sync) set_state calls
always go through and reset the throttle, bumps that might un-idle a
device bypass it, and entries are evicted when a `USER_SYNC` stop is
sent so reconnecting devices are relayed afresh.

This gives the writer-CPU benefit that deployments currently obtain by
tightening `rc_presence` to ~1/29s, without dropping real presence
transitions.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Quentin Gliech <quenting@element.io>

changelog.d/19941.misc

@@ -0,0 +1 @@
+Reduce replication traffic caused by presence.

synapse/handlers/presence.py

@@ -530,10 +530,34 @@ class WorkerPresenceHandler(BasePresenceHandler):
         # syncing but we haven't notified the presence writer of that yet
         self._user_devices_going_offline: dict[tuple[str, str | None], int] = {}
 
+        # How often to relay an unchanged sync-driven presence state to the
+        # presence writer. The relayed updates are what feed the writer's device
+        # last_sync_ts/last_active_ts timers, so this must sit comfortably below
+        # the timers it feeds — the (configurable) sync online timeout and
+        # last-active granularity — or users would flap offline / lose
+        # "currently active" between relays. We use 5/6 of the tighter of the
+        # two, i.e. the historic 25s at the default 30s sync online timeout.
+        self._sync_presence_relay_interval = (
+            min(self._sync_online_timeout, self._last_active_granularity) * 5 // 6
+        )
+
+        # (user_id, device_id) -> (state, last_sent_ms) of the most recent
+        # sync-driven presence update we proxied to the presence writer. Used
+        # to suppress the per-sync-request set_state/bump calls, which are
+        # no-ops on the writer at finer granularity than its timers: while
+        # the state is unchanged there is no point relaying more than one
+        # update per relay interval. Entries older than the window are swept by
+        # `_sweep_last_sent_presence`.
+        self._last_sent_presence: dict[tuple[str, str | None], tuple[str, int]] = {}
+
         self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs)
         self._set_state_client = ReplicationPresenceSetState.make_client(hs)
 
-        self.clock.looping_call(self.send_stop_syncing, UPDATE_SYNCING_USERS)
+        if self._track_presence:
+            self.clock.looping_call(self.send_stop_syncing, UPDATE_SYNCING_USERS)
+            self.clock.looping_call(
+                self._sweep_last_sent_presence, Duration(minutes=30)
+            )
 
         hs.register_async_shutdown_handler(
             phase="before",
@@ -576,6 +600,8 @@ class WorkerPresenceHandler(BasePresenceHandler):
         sending a stopped syncing immediately followed by a started syncing
         notification to the presence writer
         """
+        if not self._track_presence:
+            return
         self._user_devices_going_offline[(user_id, device_id)] = self.clock.time_msec()
 
     def send_stop_syncing(self) -> None:
@@ -589,6 +615,22 @@ class WorkerPresenceHandler(BasePresenceHandler):
             if now - last_sync_ms > UPDATE_SYNCING_USERS.as_millis():
                 self._user_devices_going_offline.pop((user_id, device_id), None)
                 self.send_user_sync(user_id, device_id, False, last_sync_ms)
+                # Once the writer knows the device stopped syncing it may time
+                # the user out, so if the device comes back we must relay its
+                # state again rather than suppress it as a repeat.
+                self._last_sent_presence.pop((user_id, device_id), None)
+
+    def _sweep_last_sent_presence(self) -> None:
+        """Drop expired presence-throttling entries.
+
+        Entries should be dropped in `send_stop_syncing`, but we add a safety
+        net here to ensure that the dict deesn't grow unbounded.
+        """
+        now = self.clock.time_msec()
+
+        for key, (_, last_sent_ms) in list(self._last_sent_presence.items()):
+            if now - last_sent_ms >= self._sync_presence_relay_interval:
+                self._last_sent_presence.pop(key, None)
 
     async def user_syncing(
         self,
@@ -606,7 +648,9 @@ class WorkerPresenceHandler(BasePresenceHandler):
             return _NullContextManager()
 
         # Note that this causes last_active_ts to be incremented which is not
-        # what the spec wants.
+        # what the spec wants. (This call is throttled in `set_state`: while
+        # the state is unchanged, only one update per relay interval is relayed
+        # to the presence writer.)
         await self.set_state(
             UserID.from_string(user_id),
             device_id,
@@ -743,6 +787,28 @@ class WorkerPresenceHandler(BasePresenceHandler):
         if not self._track_presence:
             return
 
+        now = self.clock.time_msec()
+        if is_sync and not force_notify:
+            # Sync-driven updates arrive on every /sync request, which is far
+            # finer-grained than any of the writer's presence timers need:
+            # while the state is unchanged, relaying one update per relay
+            # interval is enough to keep them fed. State changes always go
+            # through immediately.
+            last_sent = self._last_sent_presence.get((user_id, device_id))
+            if last_sent is not None:
+                last_presence, last_sent_ms = last_sent
+                if (
+                    presence == last_presence
+                    and now - last_sent_ms < self._sync_presence_relay_interval
+                ):
+                    return
+            self._last_sent_presence[(user_id, device_id)] = (presence, now)
+        else:
+            # An explicit (non-sync) update doesn't refresh the writer's
+            # last_sync_ts, so it must not count as a recent relay: drop any
+            # entry so the next sync-driven update goes through.
+            self._last_sent_presence.pop((user_id, device_id), None)
+
         # Proxy request to instance that writes presence
         await self._set_state_client(
             instance_name=self._presence_writer_instance,
@@ -763,8 +829,25 @@ class WorkerPresenceHandler(BasePresenceHandler):
         if not self._track_presence:
             return
 
-        # Proxy request to instance that writes presence
         user_id = user.to_string()
+
+        # A bump's only effects on the writer are updating last_active_ts and
+        # flipping an idle device back online. Going idle takes far longer
+        # than the relay window, so if we relayed an *online* update within
+        # the window the user cannot have gone idle since, and this bump is a
+        # no-op: skip it. Bumps after any other state (or an unknown one) go
+        # through immediately, as they may un-idle the device.
+        now = self.clock.time_msec()
+        last_sent = self._last_sent_presence.get((user_id, device_id))
+        if (
+            last_sent is not None
+            and last_sent[0] == PresenceState.ONLINE
+            and now - last_sent[1] < self._sync_presence_relay_interval
+        ):
+            return
+        self._last_sent_presence[(user_id, device_id)] = (PresenceState.ONLINE, now)
+
+        # Proxy request to instance that writes presence
         await self._bump_active_client(
             instance_name=self._presence_writer_instance,
             user_id=user_id,

tests/handlers/test_presence.py

@@ -19,7 +19,7 @@
 #
 #
 import itertools
-from typing import cast
+from typing import Any, cast
 from unittest.mock import Mock, call
 
 from parameterized import parameterized
@@ -50,6 +50,7 @@ from synapse.handlers.presence import (
     FEDERATION_PING_INTERVAL,
     FEDERATION_TIMEOUT,
     PresenceHandler,
+    WorkerPresenceHandler,
     handle_timeout,
     handle_update,
 )
@@ -2320,3 +2321,180 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
         )
 
         return event
+
+
+class WorkerPresenceThrottleTestCase(BaseMultiWorkerStreamTestCase):
+    """Tests that sync workers suppress the per-sync-request presence updates
+    that the presence writer would discard anyway, while relaying genuine
+    state changes immediately."""
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.user_id = f"@throttled:{hs.config.server.server_name}"
+        self.user_id_obj = UserID.from_string(self.user_id)
+        self.device_id = "dev-1"
+        # In this test setup the main process is the presence writer.
+        self.writer_handler = hs.get_presence_handler()
+
+    def _make_sync_worker(self) -> tuple[Any, list, list]:
+        """Create a sync worker whose proxied presence calls are recorded."""
+        worker = self.make_worker_hs(
+            "synapse.app.generic_worker", {"worker_name": "synchrotron"}
+        )
+        presence = worker.get_presence_handler()
+        assert isinstance(presence, WorkerPresenceHandler)
+
+        set_state_calls: list = []
+        bump_calls: list = []
+        real_set_state = presence._set_state_client
+        real_bump = presence._bump_active_client
+
+        async def recording_set_state(**kwargs: Any) -> Any:
+            set_state_calls.append(kwargs)
+            return await real_set_state(**kwargs)
+
+        async def recording_bump(**kwargs: Any) -> Any:
+            bump_calls.append(kwargs)
+            return await real_bump(**kwargs)
+
+        presence._set_state_client = recording_set_state
+        presence._bump_active_client = recording_bump
+        return presence, set_state_calls, bump_calls
+
+    def _sync(self, presence: Any, state: str = PresenceState.ONLINE) -> Any:
+        # Note: `get_success` only advances the fake clock by tiny epsilon steps
+        # while pumping the replication traffic; the throttle window is
+        # time-sensitive and these tests advance time explicitly.
+        return self.get_success(
+            presence.user_syncing(self.user_id, self.device_id, True, state),
+        )
+
+    def test_repeated_syncs_are_throttled(self) -> None:
+        presence, set_state_calls, _ = self._make_sync_worker()
+
+        # Several syncs in quick succession only relay one set_state.
+        for _ in range(3):
+            self._sync(presence)
+        self.assertEqual(len(set_state_calls), 1)
+
+        # The user did come online on the writer.
+        state = self.get_success(self.writer_handler.get_state(self.user_id_obj))
+        self.assertEqual(state.state, PresenceState.ONLINE)
+
+        # Once the relay window has passed, the next sync relays again.
+        self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1)
+        self._sync(presence)
+        self.assertEqual(len(set_state_calls), 2)
+
+    def test_state_changes_are_relayed_immediately(self) -> None:
+        presence, set_state_calls, _ = self._make_sync_worker()
+
+        self._sync(presence, PresenceState.ONLINE)
+        self._sync(presence, PresenceState.UNAVAILABLE)
+        self._sync(presence, PresenceState.ONLINE)
+        self.assertEqual(len(set_state_calls), 3)
+
+        # A repeat of the current state within the window is suppressed.
+        self._sync(presence, PresenceState.ONLINE)
+        self.assertEqual(len(set_state_calls), 3)
+
+    def test_resends_after_device_stops_syncing(self) -> None:
+        """After a USER_SYNC stop is sent the writer may time the user out, so
+        a device that reconnects within the window must be relayed afresh."""
+        presence, set_state_calls, _ = self._make_sync_worker()
+
+        with self._sync(presence):
+            pass
+        self.assertEqual(len(set_state_calls), 1)
+
+        # Wait for the going-offline grace period to elapse: USER_SYNC stop is
+        # sent and the throttle entry evicted. The writer then times the user
+        # out to offline. Advance in steps (rather than one jump) so the
+        # replicated stop command is delivered before the writer's timeout
+        # loop fires (which only starts 30s after startup).
+        for _ in range(4):
+            self.reactor.advance(12)
+        state = self.get_success(self.writer_handler.get_state(self.user_id_obj))
+        self.assertEqual(state.state, PresenceState.OFFLINE)
+
+        # Reconnecting relays the state immediately, even though the presence
+        # value is unchanged from the last relayed one.
+        self._sync(presence)
+        self.assertEqual(len(set_state_calls), 2)
+        state = self.get_success(self.writer_handler.get_state(self.user_id_obj))
+        self.assertEqual(state.state, PresenceState.ONLINE)
+
+    def test_bumps_are_throttled(self) -> None:
+        presence, set_state_calls, bump_calls = self._make_sync_worker()
+
+        # While we recently relayed an online state, bumps are suppressed.
+        self._sync(presence, PresenceState.ONLINE)
+        for _ in range(3):
+            self.get_success(
+                presence.bump_presence_active_time(self.user_id_obj, self.device_id),
+            )
+        self.assertEqual(len(bump_calls), 0)
+
+        # After the window passes, a bump goes through (and then suppresses
+        # further bumps).
+        self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1)
+        for _ in range(2):
+            self.get_success(
+                presence.bump_presence_active_time(self.user_id_obj, self.device_id),
+            )
+        self.assertEqual(len(bump_calls), 1)
+
+    def test_explicit_set_state_always_relayed_and_resets(self) -> None:
+        """An explicit (non-sync) set_state is always relayed, and resets the
+        throttle so the next sync-driven update is relayed afresh."""
+        presence, set_state_calls, _ = self._make_sync_worker()
+
+        self._sync(presence, PresenceState.ONLINE)
+        self.assertEqual(len(set_state_calls), 1)
+
+        # An explicit update of the same state within the window still goes
+        # through (it isn't sync-driven)...
+        self.get_success(
+            presence.set_state(
+                self.user_id_obj,
+                self.device_id,
+                {"presence": PresenceState.ONLINE},
+            ),
+        )
+        self.assertEqual(len(set_state_calls), 2)
+
+        # ...and the following sync-driven update is relayed rather than
+        # suppressed, re-establishing the writer's sync timestamps.
+        self._sync(presence, PresenceState.ONLINE)
+        self.assertEqual(len(set_state_calls), 3)
+
+    def test_bump_after_non_online_state_goes_through(self) -> None:
+        presence, set_state_calls, bump_calls = self._make_sync_worker()
+
+        # The user is unavailable; a bump may un-idle them so it must not be
+        # suppressed.
+        self._sync(presence, PresenceState.UNAVAILABLE)
+        self.get_success(
+            presence.bump_presence_active_time(self.user_id_obj, self.device_id),
+        )
+        self.assertEqual(len(bump_calls), 1)
+
+    @override_config({"presence": {"sync_online_timeout": "12s"}})
+    def test_relay_interval_scales_with_config(self) -> None:
+        """The throttle window is derived from the configurable presence timers,
+        so it stays comfortably below a lowered sync online timeout rather than
+        being a hardcoded 25s (which would make users flap)."""
+        presence, set_state_calls, _ = self._make_sync_worker()
+
+        # 5/6 of min(12s sync online timeout, 60s default last-active
+        # granularity).
+        self.assertEqual(presence._sync_presence_relay_interval, 10 * 1000)
+
+        # A repeat within the (now shorter) window is still suppressed...
+        self._sync(presence)
+        self._sync(presence)
+        self.assertEqual(len(set_state_calls), 1)
+
+        # ...and once it passes, the next sync relays again.
+        self.reactor.advance(presence._sync_presence_relay_interval / 1000 + 1)
+        self._sync(presence)
+        self.assertEqual(len(set_state_calls), 2)