convert to background task and add tests

Author: dklimpel

synapse/storage/databases/main/devices.py

@@ -955,6 +955,11 @@ def __init__(self, database: DatabasePool, db_conn, hs):
             self._remove_duplicate_outbound_pokes,
         )
 
+        self.db_pool.updates.register_background_update_handler(
+            "remove_deleted_devices_from_device_inbox",
+            self._remove_deleted_devices_from_device_inbox,
+        )
+
         # a pair of background updates that were added during the 1.14 release cycle,
         # but replaced with 58/06dlols_unique_idx.py
         self.db_pool.updates.register_noop_background_update(
@@ -1045,6 +1050,63 @@ def _txn(txn):
 
         return rows
 
+    async def _remove_deleted_devices_from_device_inbox(
+        self, progress: dict, batch_size: int
+    ) -> int:
+        """A background update that deletes all device_inboxes for deleted devices.
+
+        This should only need to be run once (when users upgrade to v1.45.0)
+
+        Args:
+            progress: dict used to store progress of this background update
+            batch_size: the maximum number of rows to retrieve in a single select query
+
+        Returns:
+            The number of deleted rows
+        """
+
+        def _remove_deleted_devices_from_device_inbox_txn(
+            txn: LoggingTransaction,
+        ) -> int:
+
+            sql = """
+                SELECT user_id, device_id, stream_id
+                    FROM device_inbox
+                    WHERE device_id
+                        NOT IN (SELECT device_id FROM devices)
+                    LIMIT ?;
+            """
+
+            txn.execute(sql, (batch_size,))
+            rows = txn.fetchall()
+
+            row = None
+            for row in rows:
+                self.db_pool.simple_delete_txn(
+                    txn,
+                    "device_inbox",
+                    {"user_id": row[0], "device_id": row[1], "stream_id": row[2]},
+                )
+
+            if row:
+                self.db_pool.updates._background_update_progress_txn(
+                    txn, "remove_deleted_devices_from_device_inbox", row
+                )
+
+            return len(rows)
+
+        number_deleted = await self.db_pool.runInteraction(
+            "_remove_deleted_devices_from_device_inbox",
+            _remove_deleted_devices_from_device_inbox_txn,
+        )
+
+        if number_deleted < batch_size:
+            await self.db_pool.updates._end_background_update(
+                "remove_deleted_devices_from_device_inbox"
+            )
+
+        return number_deleted
+
 
 class DeviceStore(DeviceWorkerStore, DeviceBackgroundUpdateStore):
     def __init__(self, database: DatabasePool, db_conn, hs):
@@ -1121,7 +1183,7 @@ async def store_device(
             raise StoreError(500, "Problem storing device.")
 
     async def delete_device(self, user_id: str, device_id: str) -> None:
-        """Delete a device.
+        """Delete a device and this device_inbox.
 
         Args:
             user_id: The ID of the user which owns the device
@@ -1133,10 +1195,10 @@ async def delete_device(self, user_id: str, device_id: str) -> None:
             desc="delete_device",
         )
 
-        await self.db_pool.simple_delete_one(
+        await self.db_pool.simple_delete(
             table="device_inbox",
             keyvalues={"user_id": user_id, "device_id": device_id},
-            desc="delete_device",
+            desc="delete_device_inbox",
         )
 
         self.device_id_exists_cache.invalidate((user_id, device_id))
@@ -1161,7 +1223,7 @@ async def delete_devices(self, user_id: str, device_ids: List[str]) -> None:
             column="device_id",
             iterable=device_ids,
             keyvalues={"user_id": user_id},
-            desc="delete_devices",
+            desc="delete_devices_inbox",
         )
 
         for device_id in device_ids:

synapse/storage/schema/main/delta/64/02remove_deleted_devices_from_device_inbox.sql

@@ -0,0 +1,23 @@
+/* Copyright 2021 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+--- Remove messages from the device_inbox table which where sent to an
+--- allready deleted device.
+--- This schould run as background task, it may take a little bit longer
+--- to finish.
+
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (6402, 'remove_deleted_devices_from_device_inbox', '{}');

tests/handlers/test_device.py

@@ -160,6 +160,36 @@ def test_delete_device(self):
         # we'd like to check the access token was invalidated, but that's a
         # bit of a PITA.
 
+    def test_delete_device_and_device_inbox(self):
+        self._record_users()
+
+        # add an device_inbox
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                "device_inbox",
+                {
+                    "user_id": user1,
+                    "device_id": "abc",
+                    "stream_id": 1,
+                    "message_json": "{}",
+                },
+            )
+        )
+
+        # delete the device
+        self.get_success(self.handler.delete_device(user1, "abc"))
+
+        # check that the device_inbox was deleted
+        self.get_failure(
+            self.store.db_pool.simple_select_one(
+                table="device_inbox",
+                keyvalues={"user_id": user1, "device_id": "abc"},
+                retcols=("user_id", "device_id"),
+                desc="get_device_id_from_device_inbox",
+            ),
+            synapse.api.errors.StoreError,
+        )
+
     def test_update_device(self):
         self._record_users()
 

tests/storage/databases/main/test_devices.py

@@ -0,0 +1,95 @@
+# Copyright 2021 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from synapse.rest import admin
+from synapse.rest.client import devices
+
+from tests.unittest import HomeserverTestCase
+
+
+class DevicesBackgroundUpdateStoreTestCase(HomeserverTestCase):
+
+    servlets = [
+        admin.register_servlets,
+        devices.register_servlets,
+    ]
+
+    def prepare(self, reactor, clock, hs):
+        self.store = hs.get_datastore()
+        self.user_id = self.register_user("foo", "pass")
+
+    def test_background_remove_deleted_devices_from_device_inbox(self):
+        """Test that the background task to delete old device_inboxes works properly."""
+
+        # create a valid device
+        self.get_success(
+            self.store.store_device(self.user_id, "cur_device", "display_name")
+        )
+
+        # Add device_inbox to devices
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                "device_inbox",
+                {
+                    "user_id": self.user_id,
+                    "device_id": "cur_device",
+                    "stream_id": 1,
+                    "message_json": "{}",
+                },
+            )
+        )
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                "device_inbox",
+                {
+                    "user_id": self.user_id,
+                    "device_id": "old_device",
+                    "stream_id": 2,
+                    "message_json": "{}",
+                },
+            )
+        )
+
+        # Insert and run the background update.
+        self.get_success(
+            self.store.db_pool.simple_insert(
+                "background_updates",
+                {
+                    "update_name": "remove_deleted_devices_from_device_inbox",
+                    "progress_json": "{}",
+                },
+            )
+        )
+
+        # ... and tell the DataStore that it hasn't finished all updates yet
+        self.store.db_pool.updates._all_done = False
+
+        # Now let's actually drive the updates to completion
+        while not self.get_success(
+            self.store.db_pool.updates.has_completed_background_updates()
+        ):
+            self.get_success(
+                self.store.db_pool.updates.do_next_background_update(100), by=0.1
+            )
+
+        # Make sure the background task deleted old device_inbox
+        res = self.get_success(
+            self.store.db_pool.simple_select_onecol(
+                table="device_inbox",
+                keyvalues={},
+                retcol="device_id",
+                desc="get_device_id_from_device_inbox",
+            )
+        )
+        self.assertEqual(1, len(res))