Fix performance of history trimming (#3595)

* Fix performance of history trimming

* Align version

* CONCURRENTLY cannot run inside a transaction block

* Fix missing coverage

Author: leplatrem

kinto/core/storage/__init__.py

@@ -385,6 +385,28 @@ def collection_timestamp(self, collection_id, parent_id):
         warnings.warn(message, DeprecationWarning)
         return self.resource_timestamp(resource_name=collection_id, parent_id=parent_id)
 
+    def trim_objects(
+        self,
+        resource_name: str,
+        parent_id: str,
+        filters: list,
+        max_objects: int,
+        id_field: str = DEFAULT_ID_FIELD,
+        modified_field: str = DEFAULT_MODIFIED_FIELD,
+    ) -> int:
+        """
+        Trim the last N objects in the specified resource matching the filters.
+
+        :param str resource_name: the resource name.
+        :param str parent_id: the resource parent.
+        :param list filters: list of :class:`kinto.core.storage.Filter` to filter objects.
+        :param int max_objects: maximum number of objects to keep.
+        :param str id_field: the id field name.
+        :param str modified_field: the modified field name.
+        :returns: the number of deleted objects.
+        :rtype: int"""
+        raise NotImplementedError
+
 
 def heartbeat(backend):
     def ping(request):

kinto/core/storage/memory.py

@@ -10,6 +10,7 @@
     DEFAULT_ID_FIELD,
     DEFAULT_MODIFIED_FIELD,
     MISSING,
+    Sort,
     StorageBase,
     exceptions,
 )
@@ -438,6 +439,36 @@ def delete_all(
         ]
         return deleted
 
+    @synchronized
+    def trim_objects(
+        self,
+        resource_name: str,
+        parent_id: str,
+        filters: list,
+        max_objects: int,
+        id_field: str = DEFAULT_ID_FIELD,
+        modified_field: str = DEFAULT_MODIFIED_FIELD,
+    ) -> int:
+        objects = _get_objects_by_parent_id(self._store, parent_id, resource_name, with_meta=True)
+        objects, _ = self.extract_object_set(
+            objects=objects,
+            filters=filters,
+            sorting=[Sort(modified_field, -1)],
+            id_field=id_field,
+            deleted_field=DEFAULT_DELETED_FIELD,
+        )
+
+        to_delete = objects[max_objects:]
+        for r in to_delete:
+            self.delete(
+                resource_name,
+                parent_id,
+                r[id_field],
+                id_field=id_field,
+                modified_field=modified_field,
+            )
+        return len(to_delete)
+
 
 def extract_object_set(
     objects,

kinto/core/storage/postgresql/__init__.py

@@ -79,7 +79,7 @@ class Storage(StorageBase, MigratorMixin):
 
     # MigratorMixin attributes.
     name = "storage"
-    schema_version = 23
+    schema_version = 24
     schema_file = os.path.join(HERE, "schema.sql")
     migrations_directory = os.path.join(HERE, "migrations")
 
@@ -753,6 +753,45 @@ def count_all(
         )
         return rows[0].total_count
 
+    def trim_objects(
+        self,
+        resource_name: str,
+        parent_id: str,
+        filters: list,
+        max_objects: int,
+        id_field: str = DEFAULT_ID_FIELD,
+        modified_field: str = DEFAULT_MODIFIED_FIELD,
+    ) -> int:
+        query = """
+            WITH to_delete AS (
+                SELECT {id_field}
+                FROM objects
+                WHERE parent_id = :parent_id
+                  AND resource_name = :resource_name
+                  {conditions_filter}
+                ORDER BY {modified_field} DESC
+                OFFSET :max_objects
+            )
+            DELETE FROM objects o
+            USING to_delete d
+            WHERE o.id = d.id
+            RETURNING 1;
+        """
+
+        placeholders = dict(
+            parent_id=parent_id, resource_name=resource_name, max_objects=max_objects
+        )
+        safe_sql, holders = self._format_conditions(filters, id_field, modified_field)
+        placeholders.update(**holders)
+
+        safeholders = dict(id_field=id_field, modified_field=modified_field)
+        safeholders["conditions_filter"] = f"AND {safe_sql}"
+
+        with self.client.connect() as conn:
+            result = conn.execute(sa.text(query.format_map(safeholders)), placeholders)
+            # Using RETURNING so rowcount reflects the number deleted
+            return result.rowcount
+
     def _get_rows(
         self,
         query,
@@ -1040,7 +1079,7 @@ def _format_sorting(self, sorting, id_field, modified_field):
             if sort.field == id_field:
                 sql_field = "id"
             elif sort.field == modified_field:
-                sql_field = "last_modified"
+                sql_field = "objects.last_modified"
             else:
                 # Subfields: ``person.name`` becomes ``data->person->name``
                 subfields = sort.field.split(".")

kinto/core/storage/postgresql/migrations/migration_023_024.sql

@@ -0,0 +1,6 @@
+CREATE INDEX idx_objects_history_userid_and_resourcename
+  ON objects ((data->'user_id'), (data->'resource_name'))
+  WHERE resource_name = 'history';
+
+-- Bump storage schema version.
+INSERT INTO metadata (name, value) VALUES ('storage_schema_version', '24');

kinto/core/storage/postgresql/schema.sql

@@ -49,6 +49,10 @@ CREATE INDEX IF NOT EXISTS idx_objects_last_modified_epoch
     ON objects(as_epoch(last_modified));
 CREATE INDEX IF NOT EXISTS idx_objects_resource_name_parent_id_deleted
     ON objects(resource_name, parent_id, deleted);
+-- Index for history plugin trimming.
+CREATE INDEX IF NOT EXISTS idx_objects_history_userid_and_resourcename
+  ON objects ((data->'user_id'), (data->'resource_name'))
+  WHERE resource_name = 'history';
 
 CREATE TABLE IF NOT EXISTS timestamps (
   parent_id TEXT NOT NULL COLLATE "C",
@@ -132,4 +136,4 @@ INSERT INTO metadata (name, value) VALUES ('created_at', NOW()::TEXT);
 
 -- Set storage schema version.
 -- Should match ``kinto.core.storage.postgresql.PostgreSQL.schema_version``
-INSERT INTO metadata (name, value) VALUES ('storage_schema_version', '23');
+INSERT INTO metadata (name, value) VALUES ('storage_schema_version', '24');

kinto/core/storage/testing.py

@@ -1765,6 +1765,22 @@ def test_update_bytes_raises(self):
         )
 
 
+class TrimObjectsTest:
+    def test_trims_n_of_same_resource(self):
+        for i in range(20):
+            self.create_object({"num": i, "group": "a" if i % 2 == 0 else "b"})
+
+        num_removed = self.storage.trim_objects(
+            resource_name=self.storage_kw["resource_name"],
+            parent_id=self.storage_kw["parent_id"],
+            filters=[Filter("group", "a", utils.COMPARISON.EQ)],
+            max_objects=3,
+        )
+        self.assertEqual(num_removed, 20 / 2 - 3)  # it kept all 'b' and the 3 latest of 'a'
+        count = self.storage.count_all(**self.storage_kw)
+        self.assertEqual(count, 20 - num_removed)
+
+
 class DeprecatedCoreNotionsTest:
     def setUp(self):
         super().setUp()
@@ -1834,6 +1850,7 @@ class StorageTest(
     SerializationTest,
     DeprecatedCoreNotionsTest,
     BaseTestStorage,
+    TrimObjectsTest,
 ):
     """Compound of all storage tests."""
 

kinto/plugins/history/listener.py

@@ -3,7 +3,7 @@
 
 from pyramid.settings import asbool, aslist
 
-from kinto.core.storage import Filter, Sort
+from kinto.core.storage import Filter
 from kinto.core.utils import COMPARISON, instance_uri
 
 
@@ -145,27 +145,17 @@ def on_resource_changed(event):
             if is_trim_by_user_enabled:
                 filters.append(Filter("user_id", user_id, COMPARISON.EQ))
 
-            # Identify the oldest entry to keep.
-            previous_entries = storage.list_all(
+            count_deleted = storage.trim_objects(
                 parent_id=bucket_uri,
                 resource_name="history",
                 filters=filters,
-                sorting=[Sort("last_modified", -1)],
-                limit=trim_history_max + 1,
+                max_objects=trim_history_max,
             )
-            # And delete all older ones.
-            if len(previous_entries) > trim_history_max:
-                trim_before_timestamp = previous_entries[-1]["last_modified"]
-                deleted = storage.delete_all(
-                    parent_id=bucket_uri,
-                    resource_name="history",
-                    filters=filters
-                    + [Filter("last_modified", trim_before_timestamp, COMPARISON.MAX)],
-                )
-                logger.info(f"Trimmed {len(deleted)} old history entries.")
+            if count_deleted > 0:
+                logger.info(f"Trimmed {count_deleted} old history entries.")
             else:
                 logger.info(
-                    "No old history to trim for {user_id!r} on {resource_name!r} in {bucket_uri!r}."
+                    f"No old history to trim for {user_id!r} on {resource_name!r} in {bucket_uri!r}."
                 )
         else:
             logger.info(

tests/core/test_storage.py

@@ -75,6 +75,7 @@ def test_mandatory_overrides(self):
             (self.storage.purge_deleted, "", ""),
             (self.storage.list_all, "", ""),
             (self.storage.count_all, "", ""),
+            (self.storage.trim_objects, "", "", [], 0),
         ]
         for call in calls:
             self.assertRaises(NotImplementedError, *call)