Delete push actions on receipt

changelog.d/13834.misc

@@ -0,0 +1 @@
+Clear out pending push actions when processing read receipts, removing extra checks during push action processing. Contributed by Nick @ Beeper (@fizzadar).

synapse/storage/databases/main/event_push_actions.py

@@ -90,7 +90,7 @@
 
 from synapse.api.constants import ReceiptTypes
 from synapse.metrics.background_process_metrics import wrap_as_background_process
-from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
+from synapse.storage._base import SQLBaseStore, db_to_json
 from synapse.storage.database import (
     DatabasePool,
     LoggingDatabaseConnection,
@@ -557,31 +557,6 @@ def f(txn: LoggingTransaction) -> List[str]:
 
         return await self.db_pool.runInteraction("get_push_action_users_in_range", f)
 
-    def _get_receipts_by_room_txn(
-        self, txn: LoggingTransaction, user_id: str
-    ) -> List[Tuple[str, int]]:
-        receipt_types_clause, args = make_in_list_sql_clause(
-            self.database_engine,
-            "receipt_type",
-            (
-                ReceiptTypes.READ,
-                ReceiptTypes.READ_PRIVATE,
-            ),
-        )
-
-        sql = f"""
-            SELECT room_id, MAX(stream_ordering)
-            FROM receipts_linearized
-            INNER JOIN events USING (room_id, event_id)
-            WHERE {receipt_types_clause}
-            AND user_id = ?
-            GROUP BY room_id
-        """
-
-        args.extend((user_id,))
-        txn.execute(sql, args)
-        return cast(List[Tuple[str, int]], txn.fetchall())
-
     async def get_unread_push_actions_for_user_in_range_for_http(
         self,
         user_id: str,
@@ -605,14 +580,6 @@ async def get_unread_push_actions_for_user_in_range_for_http(
             The list will have between 0~limit entries.
         """
 
-        receipts_by_room = dict(
-            await self.db_pool.runInteraction(
-                "get_unread_push_actions_for_user_in_range_http_receipts",
-                self._get_receipts_by_room_txn,
-                user_id=user_id,
-            ),
-        )
-
         def get_push_actions_txn(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool]]:
@@ -641,9 +608,6 @@ def get_push_actions_txn(
                 actions=_deserialize_action(actions, highlight),
             )
             for event_id, room_id, stream_ordering, actions, highlight in push_actions
-            # Only include push actions with a stream ordering after any receipt, or without any
-            # receipt present (invited to but never read rooms).
-            if stream_ordering > receipts_by_room.get(room_id, 0)
         ]
 
         # Now sort it so it's ordered correctly, since currently it will
@@ -679,14 +643,6 @@ async def get_unread_push_actions_for_user_in_range_for_email(
             The list will have between 0~limit entries.
         """
 
-        receipts_by_room = dict(
-            await self.db_pool.runInteraction(
-                "get_unread_push_actions_for_user_in_range_email_receipts",
-                self._get_receipts_by_room_txn,
-                user_id=user_id,
-            ),
-        )
-
         def get_push_actions_txn(
             txn: LoggingTransaction,
         ) -> List[Tuple[str, str, int, str, bool, int]]:
@@ -719,9 +675,6 @@ def get_push_actions_txn(
                 received_ts=received_ts,
             )
             for event_id, room_id, stream_ordering, actions, highlight, received_ts in push_actions
-            # Only include push actions with a stream ordering after any receipt, or without any
-            # receipt present (invited to but never read rooms).
-            if stream_ordering > receipts_by_room.get(room_id, 0)
         ]
 
         # Now sort it so it's ordered correctly, since currently it will

synapse/storage/databases/main/receipts.py

@@ -26,7 +26,7 @@
     cast,
 )
 
-from synapse.api.constants import EduTypes
+from synapse.api.constants import EduTypes, ReceiptTypes
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
 from synapse.replication.tcp.streams import ReceiptsStream
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
@@ -702,6 +702,18 @@ def _insert_linearized_receipt_txn(
             lock=False,
         )
 
+        if stream_ordering is not None and receipt_type in (
+            ReceiptTypes.READ,
+            ReceiptTypes.READ_PRIVATE,
+        ):
+            sql = """
+                DELETE FROM event_push_actions
+                WHERE room_id = ?
+                AND user_id = ?
+                AND stream_ordering <= ?
+            """
+            txn.execute(sql, (room_id, user_id, stream_ordering))
+
         return rx_ts
 
     def _graph_to_linear(