Newsfile

changelog.d/15519.bugfix

@@ -0,0 +1 @@
+Fix bug where we didn't clean out old highlight push notifications from the database. Introduced in v1.62.0.

synapse/storage/databases/main/event_push_actions.py

@@ -1817,33 +1817,41 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
         1 day ago).
         """
 
-        # We want to clear out anything that is older than a day that *has* already
-        # been rotated.
+        # Fetch the stream ordering that we've rotated up to, to ensure we don't
+        # delete anything after.
         rotated_upto_stream_ordering = await self.db_pool.simple_select_one_onecol(
             table="event_push_summary_stream_ordering",
             keyvalues={},
             retcol="stream_ordering",
         )
 
-        max_stream_ordering_to_delete = min(
-            rotated_upto_stream_ordering, self.stream_ordering_day_ago
-        )
-
         def remove_old_push_actions_that_have_rotated_txn(
             txn: LoggingTransaction,
+            highlight: bool,
+            max_stream_ordering: int,
         ) -> bool:
             # We don't want to clear out too much at a time, so we bound our
             # deletes.
             batch_size = self._rotate_count
 
+            if isinstance(self.database_engine, PostgresEngine):
+                # Temporarily disable sequential scans in this transaction. We
+                # need to do this as the postgres statistics don't take into
+                # account the `highlight = 0` part when estimating the
+                # distribution of `stream_ordering`. I.e. since we keep old
+                # highlight rows the query planner thinks there are way more old
+                # rows to delete than there actually are.
+                txn.execute("SET LOCAL enable_seqscan=off")
+
             txn.execute(
                 """
                 SELECT stream_ordering FROM event_push_actions
-                WHERE stream_ordering <= ? AND highlight = 0
+                WHERE stream_ordering <= ? AND highlight = ?
                 ORDER BY stream_ordering ASC LIMIT 1 OFFSET ?
                 """,
                 (
-                    max_stream_ordering_to_delete,
+                    max_stream_ordering,
+                    1 if highlight else 0,
                     batch_size,
                 ),
             )
@@ -1852,26 +1860,51 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             if stream_row:
                 (stream_ordering,) = stream_row
             else:
-                stream_ordering = max_stream_ordering_to_delete
+                stream_ordering = max_stream_ordering
 
             # We need to use a inclusive bound here to handle the case where a
             # single stream ordering has more than `batch_size` rows.
             txn.execute(
                 """
                 DELETE FROM event_push_actions
-                WHERE stream_ordering <= ? AND highlight = 0
+                WHERE stream_ordering <= ? AND highlight = ?
                 """,
-                (stream_ordering,),
+                (
+                    stream_ordering,
+                    1 if highlight else 0,
+                ),
             )
 
             logger.info("Rotating notifications, deleted %s push actions", txn.rowcount)
 
             return txn.rowcount < batch_size
 
+        # First we delete all rotated highlights older than a month...
+        max_stream_ordering_to_delete = min(
+            rotated_upto_stream_ordering, self.stream_ordering_month_ago
+        )
+
+        while True:
+            done = await self.db_pool.runInteraction(
+                "_remove_old_push_actions_that_have_rotated_highlights",
+                remove_old_push_actions_that_have_rotated_txn,
+                True,
+                max_stream_ordering_to_delete,
+            )
+            if done:
+                break
+
+        # ... and then delete all rotated notifs older than a day
+        max_stream_ordering_to_delete = min(
+            rotated_upto_stream_ordering, self.stream_ordering_day_ago
+        )
+
         while True:
             done = await self.db_pool.runInteraction(
                 "_remove_old_push_actions_that_have_rotated",
                 remove_old_push_actions_that_have_rotated_txn,
+                False,
+                max_stream_ordering_to_delete,
             )
             if done:
                 break