Add a background job to automatically delete stale devices

changelog.d/12855.feature

@@ -0,0 +1 @@
+Add a configurable background job to delete stale devices.

docs/usage/configuration/config_documentation.md

@@ -567,6 +567,19 @@ Example configuration:
 dummy_events_threshold: 5
 ```
 ---
+Config option `delete_stale_devices_after`
+
+An optional duration. If set, Synapse will run an hourly background task to log out and
+delete any device that hasn't been accessed for more than the specified amount of time.
+
+Defaults to no duration, which means devices aren't pruned based on the time they were
+last accessed.
+
+Example configuration:
+```yaml
+delete_stale_devices_after: 1y
+```
+
 ## Homeserver blocking ##
 Useful options for Synapse admins.
 

synapse/config/server.py

@@ -679,6 +679,17 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
             config.get("exclude_rooms_from_sync") or []
         )
 
+        delete_stale_devices_after: Optional[str] = (
+            config.get("delete_stale_devices_after") or None
+        )
+
+        if delete_stale_devices_after is not None:
+            self.delete_stale_devices_after: Optional[int] = self.parse_duration(
+                delete_stale_devices_after
+            )
+        else:
+            self.delete_stale_devices_after = None
+
     def has_tls_listener(self) -> bool:
         return any(listener.tls for listener in self.listeners)
 

synapse/handlers/device.py

@@ -292,6 +292,19 @@ def __init__(self, hs: "HomeServer"):
         # On start up check if there are any updates pending.
         hs.get_reactor().callWhenRunning(self._handle_new_device_update_async)
 
+        self._delete_stale_devices_after = hs.config.server.delete_stale_devices_after
+
+        # Ideally we would run this on a worker and condition this on the
+        # "run_background_tasks_on" setting, but this would mean making the notification
+        # of device list changes over federation work on workers, which is nontrivial.
+        if self._delete_stale_devices_after is not None:
+            self.clock.looping_call(
+                run_as_background_process,
+                60 * 60 * 1000,
+                "delete_stale_devices",
+                self._delete_stale_devices,
+            )
+
     def _check_device_name_length(self, name: Optional[str]) -> None:
         """
         Checks whether a device name is longer than the maximum allowed length.
@@ -367,6 +380,20 @@ async def check_device_registered(
 
         raise errors.StoreError(500, "Couldn't generate a device ID.")
 
+    async def _delete_stale_devices(self) -> None:
+        """Background task that deletes devices which haven't been accessed for more than
+        a configured time period.
+        """
+        # We should only be running this job if the config option is defined.
+        assert self._delete_stale_devices_after is not None
+        now_ms = self.clock.time_msec()
+        devices = await self.store.get_devices_not_accessed_since(
+            now_ms - self._delete_stale_devices_after
+        )
+
+        for device in devices:
+            await self.delete_device(device["user_id"], device["device_id"])
+
     @trace
     async def delete_device(self, user_id: str, device_id: str) -> None:
         """Delete the given device
@@ -689,7 +716,8 @@ async def _handle_new_device_update_async(self) -> None:
                             )
                             # TODO: when called, this isn't in a logging context.
                             # This leads to log spam, sentry event spam, and massive
-                            # memory usage. See #12552.
+                            # memory usage.
+                            # See https://github.com/matrix-org/synapse/issues/12552.
                             # log_kv(
                             #     {"message": "sent device update to host", "host": host}
                             # )

synapse/storage/databases/main/devices.py

@@ -1153,6 +1153,35 @@ def _prune_txn(txn: LoggingTransaction) -> None:
             _prune_txn,
         )
 
+    async def get_devices_not_accessed_since(
+        self, since_ms: int
+    ) -> List[Dict[str, str]]:
+        """Retrieves a list of all devices that haven't been accessed since a given date.
+
+        Args:
+            since_ms: the timestamp to select on, every device which last access dates
+                from before that time is returned.
+
+        Returns:
+            A list of dictionary, each indicating the user ID and device ID of a device
+            that hasn't been accessed since the given date.
+        """
+
+        def get_devices_not_accessed_since_txn(
+            txn: LoggingTransaction,
+        ) -> List[Dict[str, str]]:
+            sql = """
+                SELECT user_id, device_id
+                FROM devices WHERE last_seen < ? AND hidden = FALSE
+            """
+            txn.execute(sql, (since_ms,))
+            return self.db_pool.cursor_to_dict(txn)
+
+        return await self.db_pool.runInteraction(
+            "get_devices_not_accessed_since",
+            get_devices_not_accessed_since_txn,
+        )
+
 
 class DeviceBackgroundUpdateStore(SQLBaseStore):
     def __init__(

tests/rest/client/test_device_lists.py → tests/rest/client/test_devices.py

@@ -13,8 +13,13 @@
 # limitations under the License.
 from http import HTTPStatus
 
+from twisted.internet.testing import MemoryReactor
+
+from synapse.api.errors import NotFoundError
 from synapse.rest import admin, devices, room, sync
 from synapse.rest.client import account, login, register
+from synapse.server import HomeServer
+from synapse.util import Clock
 
 from tests import unittest
 
@@ -157,3 +162,40 @@ def test_not_receiving_local_device_list_changes(self) -> None:
         self.assertNotIn(
             alice_user_id, changed_device_lists, bob_sync_channel.json_body
         )
+
+
+class DevicesTestCase(unittest.HomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        sync.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.handler = hs.get_device_handler()
+
+    @unittest.override_config({"delete_stale_devices_after": 3600000})
+    def test_delete_stale_devices(self) -> None:
+        """Tests that stale devices are automatically removed after a set time of
+        inactivity.
+        """
+        # Register a user and creates 2 devices for them.
+        user_id = self.register_user("user", "password")
+        tok1 = self.login("user", "password", device_id="abc")
+        tok2 = self.login("user", "password", device_id="def")
+
+        # Sync them so they have a last_seen value.
+        self.make_request("GET", "/sync", access_token=tok1)
+        self.make_request("GET", "/sync", access_token=tok2)
+
+        # Advance half an hour and sync again with one of the devices, so that the next
+        # time the background job runs we don't delete this device (since it will look
+        # for devices that haven't been used for over an hour).
+        self.reactor.advance(1800)
+        self.make_request("GET", "/sync", access_token=tok1)
+
+        # Advance another half an hour, and check that the device that has synced still
+        # exists but the one that hasn't has been removed.
+        self.reactor.advance(1800)
+        self.get_success(self.handler.get_device(user_id, "abc"))
+        self.get_failure(self.handler.get_device(user_id, "def"), NotFoundError)