Track sync ts per-device.

Author: clokep

synapse/api/presence.py

@@ -26,6 +26,7 @@ class UserDevicePresenceState:
     device_id: Optional[str]
     state: str
     last_active_ts: int
+    last_sync_ts: int
 
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)

synapse/handlers/presence.py

@@ -264,6 +264,7 @@ async def set_state(
         state: JsonDict,
         ignore_status_msg: bool = False,
         force_notify: bool = False,
+        is_sync: bool = False,
     ) -> None:
         """Set the presence state of the user.
 
@@ -273,6 +274,7 @@ async def set_state(
             ignore_status_msg: True to ignore the "status_msg" field of the `state` dict.
                 If False, the user's current status will be updated.
             force_notify: Whether to force notification of the update to clients.
+            is_sync: True if this update was from a sync
         """
 
     @abc.abstractmethod
@@ -516,6 +518,7 @@ async def user_syncing(
                 device_id,
                 state={"presence": presence_state},
                 ignore_status_msg=True,
+                is_sync=True,
             )
 
         curr_sync = self._user_to_num_current_syncs.get(user_id, 0)
@@ -615,6 +618,7 @@ async def set_state(
         state: JsonDict,
         ignore_status_msg: bool = False,
         force_notify: bool = False,
+        is_sync: bool = False,
     ) -> None:
         """Set the presence state of the user.
 
@@ -624,6 +628,7 @@ async def set_state(
             ignore_status_msg: True to ignore the "status_msg" field of the `state` dict.
                 If False, the user's current status will be updated.
             force_notify: Whether to force notification of the update to clients.
+            is_sync: True if this update was from a sync
         """
         presence = state["presence"]
 
@@ -644,6 +649,7 @@ async def set_state(
             state=state,
             ignore_status_msg=ignore_status_msg,
             force_notify=force_notify,
+            is_sync=is_sync,
         )
 
     async def bump_presence_active_time(self, user: UserID) -> None:
@@ -1035,6 +1041,7 @@ async def user_syncing(
                 device_id,
                 state={"presence": presence_state},
                 ignore_status_msg=True,
+                is_sync=True,
             )
             # Retrieve the new state for the logic below. This should come from the
             # in-memory cache.
@@ -1236,6 +1243,7 @@ async def set_state(
         state: JsonDict,
         ignore_status_msg: bool = False,
         force_notify: bool = False,
+        is_sync: bool = False,
     ) -> None:
         """Set the presence state of the user.
 
@@ -1246,6 +1254,7 @@ async def set_state(
             ignore_status_msg: True to ignore the "status_msg" field of the `state` dict.
                 If False, the user's current status will be updated.
             force_notify: Whether to force notification of the update to clients.
+            is_sync: True if this update was from a sync
         """
         status_msg = state.get("status_msg", None)
         presence = state["presence"]
@@ -1258,6 +1267,7 @@ async def set_state(
             return
 
         user_id = target_user.to_string()
+        now = self.clock.time_msec()
 
         prev_state = await self.current_state_for_user(user_id)
 
@@ -1271,10 +1281,13 @@ async def set_state(
                 device_id,
                 presence,
                 last_active_ts=self.clock.time_msec(),
+                last_sync_ts=0,
             ),
         )
         device_state.state = presence
-        device_state.last_active_ts = self.clock.time_msec()
+        device_state.last_active_ts = now
+        if is_sync:
+            device_state.last_sync_ts = now
 
         # Based on (all) the user's devices calculate the new presence state.
         presence = _combine_device_states(
@@ -1290,7 +1303,7 @@ async def set_state(
         if presence == PresenceState.ONLINE or (
             presence == PresenceState.BUSY and self._busy_presence_enabled
         ):
-            new_fields["last_active_ts"] = self.clock.time_msec()
+            new_fields["last_active_ts"] = now
 
         await self._update_states(
             [prev_state.copy_and_replace(**new_fields)], force_notify=force_notify
@@ -1887,7 +1900,7 @@ def handle_timeouts(
     Args:
         user_states: List of UserPresenceState's to check.
         is_mine_fn: Function that returns if a user_id is ours
-        syncing_user_devices: Set of user_ids with active syncs.
+        syncing_user_devices: A map of user ID to device_ids with active syncs.
         user_to_devices: A map of user ID to device ID to UserDevicePresenceState.
         now: Current time in ms.
 
@@ -1955,15 +1968,16 @@ def handle_timeout(
             if device_id not in syncing_device_ids:
                 # If the user has done something recently but hasn't synced,
                 # don't set them as offline.
-                #
-                # XXX last_user_sync_ts needs to be per-device.
                 sync_or_active = max(
-                    state.last_user_sync_ts, device_state.last_active_ts
+                    device_state.last_sync_ts, device_state.last_active_ts
                 )
                 if now - sync_or_active > SYNC_ONLINE_TIMEOUT:
                     device_state.state = PresenceState.OFFLINE
                     device_changed = True
 
+        # XXX How will this work when restoring from the database if device information
+        # is not kept?
+
         # If the presence state of any of the devices changed, then (maybe) update
         # the user's overall presence state.
         if device_changed:

synapse/replication/http/presence.py

@@ -99,23 +99,27 @@ async def _serialize_payload(  # type: ignore[override]
         state: JsonDict,
         ignore_status_msg: bool = False,
         force_notify: bool = False,
+        is_sync: bool = False,
     ) -> JsonDict:
         return {
             "device_id": device_id,
             "state": state,
             "ignore_status_msg": ignore_status_msg,
             "force_notify": force_notify,
+            "is_sync": is_sync,
         }
 
     async def _handle_request(  # type: ignore[override]
         self, request: Request, content: JsonDict, user_id: str
     ) -> Tuple[int, JsonDict]:
+        # Older versions of Synapse won't set device_id and is_sync.
         await self._presence_handler.set_state(
             UserID.from_string(user_id),
-            content["device_id"],
+            content.get("device_id"),
             content["state"],
             content["ignore_status_msg"],
             content["force_notify"],
+            content.get("is_sync", False),
         )
 
         return (200, {})

tests/handlers/test_presence.py

@@ -367,6 +367,7 @@ def test_idle_timer(self) -> None:
             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(
@@ -404,6 +405,7 @@ def test_busy_no_idle(self) -> None:
             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(
@@ -437,6 +439,7 @@ def test_sync_timeout(self) -> None:
             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(
@@ -470,6 +473,7 @@ def test_sync_online(self) -> None:
             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(
@@ -504,6 +508,7 @@ def test_federation_ping(self) -> None:
             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(
@@ -534,6 +539,7 @@ def test_no_timeout(self) -> None:
             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(
@@ -589,6 +595,7 @@ def test_last_active(self) -> None:
             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(
@@ -801,6 +808,7 @@ def test_set_presence_from_syncing_is_set(self) -> None:
         # we should now be online
         self.assertEqual(state.state, PresenceState.ONLINE)
 
+    # TODO Add tests with BUSY.
     @parameterized.expand(
         # A list of tuples of 4 strings:
         #
@@ -952,6 +960,7 @@ def test_set_presence_from_syncing_multi_device(
         )
         self.assertEqual(state.state, expected_state_3)
 
+    # TODO Add tests with BUSY.
     @parameterized.expand(
         # A list of tuples of 4 strings:
         #