From b855e100cc0f9f98f5ba9305963636bcf9a31857 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 13:20:14 +0100 Subject: [PATCH 1/9] Fix unread counts on large servers --- .../databases/main/event_push_actions.py | 40 +++++++++---------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index 80ca2fd0b612..7912e6fd0def 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -861,11 +861,13 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: retcol="stream_id", ) + max_receipts_stream_id = self._receipts_id_gen.get_current_token() + sql = """ SELECT r.stream_id, r.room_id, r.user_id, e.stream_ordering FROM receipts_linearized AS r INNER JOIN events AS e USING (event_id) - WHERE r.stream_id > ? AND user_id LIKE ? + WHERE ? < r.stream_id AND r.stream_id <= ? AND user_id LIKE ? ORDER BY r.stream_id ASC LIMIT ? """ @@ -884,6 +886,13 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: ) rows = txn.fetchall() + old_rotate_stream_ordering = self.db_pool.simple_select_one_onecol_txn( + txn, + table="event_push_summary_stream_ordering", + keyvalues={}, + retcol="stream_ordering", + ) + # For each new read receipt we delete push actions from before it and # recalculate the summary. for _, room_id, user_id, stream_ordering in rows: @@ -902,13 +911,6 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: (room_id, user_id, stream_ordering), ) - old_rotate_stream_ordering = self.db_pool.simple_select_one_onecol_txn( - txn, - table="event_push_summary_stream_ordering", - keyvalues={}, - retcol="stream_ordering", - ) - notif_count, unread_count = self._get_notif_unread_count_for_user_room( txn, room_id, user_id, stream_ordering, old_rotate_stream_ordering ) @@ -927,18 +929,16 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: # We always update `event_push_summary_last_receipt_stream_id` to # ensure that we don't rescan the same receipts for remote users. - # - # This requires repeatable read to be safe, as we need the - # `MAX(stream_id)` to not include any new rows that have been committed - # since the start of the transaction (since those rows won't have been - # returned by the query above). Alternatively we could query the max - # stream ID at the start of the transaction and bound everything by - # that. - txn.execute( - """ - UPDATE event_push_summary_last_receipt_stream_id - SET stream_id = (SELECT COALESCE(MAX(stream_id), 0) FROM receipts_linearized) - """ + + upper_limit = max_receipts_stream_id + if len(rows) >= limit: + upper_limit = rows[-1][0] + + self.db_pool.simple_update_txn( + txn, + table="event_push_summary_last_receipt_stream_id", + keyvalues={}, + updatevalues={"stream_id": upper_limit}, ) return len(rows) < limit From 479ad280ae5f8d61e9e9ca23631cfc5e4e7daf45 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 13:22:53 +0100 Subject: [PATCH 2/9] Newsfile --- changelog.d/13140.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/13140.bugfix diff --git a/changelog.d/13140.bugfix b/changelog.d/13140.bugfix new file mode 100644 index 000000000000..cb0586e39e88 --- /dev/null +++ b/changelog.d/13140.bugfix @@ -0,0 +1 @@ +Fix unread counts for users on large servers. Introduced in v1.62.0rc1. From 6da1660bc97e40d0c789610870da8593b76e2f83 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 13:29:34 +0100 Subject: [PATCH 3/9] Fix sql --- synapse/storage/databases/main/event_push_actions.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index 7912e6fd0def..ffa4e7c4dab7 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -880,6 +880,7 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: sql, ( min_stream_id, + max_receipts_stream_id, user_filter, limit, ), From 2a3f686919de71072b21d3c5e975b9e823d4bdfd Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 13:29:45 +0100 Subject: [PATCH 4/9] Fix test --- tests/storage/test_event_push_actions.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index ef069a8110b8..d52f9dfe5634 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -133,6 +133,10 @@ def _rotate(stream: int) -> None: def _mark_read(stream: int, depth: int) -> None: last_read_stream_ordering[0] = stream + # We need to update the receipts ID gen. We should rewrite this test + # using proper rooms and receipts. + self.store._receipts_id_gen._current = stream + self.get_success( self.store.db_pool.runInteraction( "", From a8c158347566cdc9f09348d76a116690bffb27d0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 13:38:09 +0100 Subject: [PATCH 5/9] Fix lint --- tests/storage/test_event_push_actions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index d52f9dfe5634..2acd5718c735 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -135,7 +135,7 @@ def _mark_read(stream: int, depth: int) -> None: # We need to update the receipts ID gen. We should rewrite this test # using proper rooms and receipts. - self.store._receipts_id_gen._current = stream + self.store._receipts_id_gen._current = stream # type: ignore self.get_success( self.store.db_pool.runInteraction( From 5cd2dcea8a915ef1c409e159e08bf5872b916ba6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 14:17:29 +0100 Subject: [PATCH 6/9] Properly fix tests --- tests/storage/test_event_push_actions.py | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index 2acd5718c735..38b6231483cf 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -20,7 +20,7 @@ from synapse.storage.databases.main.event_push_actions import NotifCounts from synapse.util import Clock -from tests.unittest import HomeserverTestCase +from tests.unittest import HomeserverTestCase, override_config USER_ID = "@user:example.com" @@ -133,20 +133,13 @@ def _rotate(stream: int) -> None: def _mark_read(stream: int, depth: int) -> None: last_read_stream_ordering[0] = stream - # We need to update the receipts ID gen. We should rewrite this test - # using proper rooms and receipts. - self.store._receipts_id_gen._current = stream # type: ignore - self.get_success( - self.store.db_pool.runInteraction( - "", - self.store._insert_linearized_receipt_txn, + self.store.insert_receipt( room_id, "m.read", - user_id, - f"$test{stream}:example.com", - {}, - stream, + user_id=user_id, + event_ids=[f"$test{stream}:example.com"], + data={}, ) ) @@ -170,6 +163,7 @@ def _mark_read(stream: int, depth: int) -> None: _inject_actions(6, PlAIN_NOTIF) _rotate(7) + _assert_counts(1, 0) self.get_success( self.store.db_pool.simple_delete( From 9edec7e6dc5824268453dab7c3dbf232fbddbfad Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 14:23:02 +0100 Subject: [PATCH 7/9] Fix lint --- tests/storage/test_event_push_actions.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index 38b6231483cf..684485ae0629 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -20,7 +20,7 @@ from synapse.storage.databases.main.event_push_actions import NotifCounts from synapse.util import Clock -from tests.unittest import HomeserverTestCase, override_config +from tests.unittest import HomeserverTestCase USER_ID = "@user:example.com" From 4421a0e2e260a48b04265a9ac045376e3ad1534e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 14:28:35 +0100 Subject: [PATCH 8/9] Sensible var name --- synapse/storage/databases/main/event_push_actions.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index ffa4e7c4dab7..a42b30cb05fc 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -854,7 +854,7 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: limit = 100 - min_stream_id = self.db_pool.simple_select_one_onecol_txn( + min_receipts_stream_id = self.db_pool.simple_select_one_onecol_txn( txn, table="event_push_summary_last_receipt_stream_id", keyvalues={}, @@ -879,7 +879,7 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: txn.execute( sql, ( - min_stream_id, + min_receipts_stream_id, max_receipts_stream_id, user_filter, limit, From e81fa1fa0ceb85f10512dbeb96157a9ac73c5935 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 30 Jun 2022 14:39:50 +0100 Subject: [PATCH 9/9] Add explicit comment --- synapse/storage/databases/main/event_push_actions.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index a42b30cb05fc..07c87a30c9be 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -933,6 +933,9 @@ def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool: upper_limit = max_receipts_stream_id if len(rows) >= limit: + # If we pulled out a limited number of rows we only update the + # position to the last receipt we processed, so we continue + # processing the rest next iteration. upper_limit = rows[-1][0] self.db_pool.simple_update_txn(