Attempt to delete more duplicate rows in receipts_linearized table.

changelog.d/14915.bugfix

@@ -0,0 +1 @@
+Fix a bug introduced in Synapse 1.70.0 where the background updates to add non-thread unique indexes on receipts could fail when upgrading from 1.67.0 or earlier.

synapse/storage/databases/main/receipts.py

@@ -941,34 +941,39 @@ async def _background_receipts_linearized_unique_index(
         receipts."""
 
         def _remote_duplicate_receipts_txn(txn: LoggingTransaction) -> None:
+            if isinstance(self.database_engine, PostgresEngine):
+                ROW_ID_NAME = "ctid"
+            else:
+                ROW_ID_NAME = "rowid"
+
             # Identify any duplicate receipts arising from
             # https://github.com/matrix-org/synapse/issues/14406.
             # We expect the following query to use the per-thread receipt index and take
             # less than a minute.
             sql = """
-                SELECT MAX(stream_id), room_id, receipt_type, user_id
+                SELECT MAX(stream_id), ?, room_id, receipt_type, user_id
                 FROM receipts_linearized
                 WHERE thread_id IS NULL
                 GROUP BY room_id, receipt_type, user_id
                 HAVING COUNT(*) > 1
             """
-            txn.execute(sql)
+            txn.execute(sql, (ROW_ID_NAME,))
             duplicate_keys = cast(List[Tuple[int, str, str, str]], list(txn))
 
             # Then remove duplicate receipts, keeping the one with the highest
-            # `stream_id`. There should only be a single receipt with any given
-            # `stream_id`.
-            for max_stream_id, room_id, receipt_type, user_id in duplicate_keys:
+            # `stream_id`. Since there might be duplicate rows with the same
+            # `stream_id`, we delete by the rowid instead.
+            for _, row_id, room_id, receipt_type, user_id in duplicate_keys:
                 sql = """
                     DELETE FROM receipts_linearized
                     WHERE
                         room_id = ? AND
                         receipt_type = ? AND
                         user_id = ? AND
                         thread_id IS NULL AND
-                        stream_id < ?
+                        ? != ?
                 """
-                txn.execute(sql, (room_id, receipt_type, user_id, max_stream_id))
+                txn.execute(sql, (room_id, receipt_type, user_id, ROW_ID_NAME, row_id))
 
         await self.db_pool.runInteraction(
             self.RECEIPTS_LINEARIZED_UNIQUE_INDEX_UPDATE_NAME,