diff --git a/changelog.d/pr-id-2.bugfix b/changelog.d/pr-id-2.bugfix new file mode 100644 index 0000000000..5baa103c21 --- /dev/null +++ b/changelog.d/pr-id-2.bugfix @@ -0,0 +1 @@ +Fix unread/notification counts not immediately reflecting a read receipt for an event older than the latest push-summary rotation. diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index a8b60f912e..a78e32c0b1 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -80,6 +80,7 @@ receipt. """ import logging +import sys from collections import defaultdict from typing import ( TYPE_CHECKING, @@ -347,6 +348,29 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas ) return 0 + def _pruned_upto_stream_ordering(self) -> int: + """Approximately the highest stream ordering whose (rotated) rows in + `event_push_actions` may already have been deleted by + `_remove_old_push_actions_that_have_rotated`. + + Counting from `event_push_actions` only gives a complete answer above this + point; below it we have to keep relying on `event_push_summary`. + + This is a heuristic rather than an exact bound: `stream_ordering_day_ago` is + a per-process value refreshed every ten minutes, while the deletion may be + running on another worker, so this can lag reality by roughly that much. The + cost of getting it wrong is the same bounded staleness we already accept + elsewhere here -- a summary row trusted for up to one rotation cycle longer + than it strictly had to be -- so being approximate is fine. + """ + if self.stream_ordering_day_ago is None: + # `stream_ordering_day_ago` is populated synchronously in `__init__`, so + # in practice this doesn't happen; if it somehow did we'd rather never + # prefer `event_push_actions` over the summary. + return sys.maxsize + + return self.stream_ordering_day_ago + async def get_unread_counts_by_room_for_user(self, user_id: str) -> dict[str, int]: """Get the notification count by room for a user. Only considers notifications, not highlight or unread counts, and threads are currently aggregated under their room. @@ -421,6 +445,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas COALESCE(unthreaded_receipt_stream_ordering, 0) )""" + # A null `last_receipt_stream_ordering` means the row does not account for any + # read receipt (see `_get_unread_counts_by_pos_txn`), so it is only up-to-date + # while the user has no receipt applying to the room/thread. We only distrust + # it while the fallback below can still see the push actions it summarised, + # i.e. while the receipt is above the deletion horizon. With no receipts at + # all the max receipt clause is 0, so such a row is always trusted. sql = f""" {receipts_cte} SELECT eps.room_id, eps.thread_id, notif_count @@ -429,11 +459,15 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas WHERE user_id = ? AND notif_count != 0 AND ( - (last_receipt_stream_ordering IS NULL AND stream_ordering > {max_clause}) + ( + last_receipt_stream_ordering IS NULL + AND stream_ordering > {max_clause} + AND {max_clause} <= ? + ) OR last_receipt_stream_ordering = {max_clause} ) """ - txn.execute(sql, args) + txn.execute(sql, args + [self._pruned_upto_stream_ordering()]) # The (room ID, thread ID) pairs we found an up-to-date summary for. seen_room_thread_ids = set() @@ -603,10 +637,36 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas # We then do a manual count of all the rows in the `event_push_actions` table # for any user/room/thread which did not have a valid summary found. # - # If `last_receipt_stream_ordering` is null then that means it's up-to-date - # (as the row was written by an older version of Synapse that - # updated `event_push_summary` synchronously when persisting a new read - # receipt). + # A null `last_receipt_stream_ordering` means the row does not account for any + # read receipt: it was either written by a version of Synapse older than + # v1.62.0 (which updated `event_push_summary` synchronously when persisting a + # new read receipt), or by a `_rotate_notifs_before_txn` that ran while the + # user had no receipt in the room. Such a row counts from the user's join, so + # once the user does have a receipt it may include events at or before it and + # we must not trust it. + # + # We only do that while the manual count below can still see the push actions + # the row summarised: `_remove_old_push_actions_that_have_rotated` deletes + # rotated push actions older than a day, so for an older receipt we keep using + # the (possibly stale) summary and let the next rotation correct it instead. + if unthreaded_receipt_stream_ordering is None: + # `from_stream_ordering` is the user's membership event rather than a + # receipt, so a row stamped with a receipt at or before it can never + # match. That happens when the receipt is for an event from before the + # user's latest join, e.g. they read back into history they can see, or + # they rejoined the room. There is no newer receipt for the row to be out + # of date with respect to, and nothing will ever recalculate it, so keep + # it rather than recounting push actions that may already have been + # deleted. + unreconcilable_clause = """ + OR ( + last_receipt_stream_ordering IS NOT NULL + AND threaded_receipt_stream_ordering IS NULL + ) + """ + else: + unreconcilable_clause = "" + txn.execute( f""" SELECT notif_count, COALESCE(unread_count, 0), thread_id @@ -623,8 +683,13 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas ) AS receipts USING (thread_id) WHERE room_id = ? AND user_id = ? AND ( - (last_receipt_stream_ordering IS NULL AND stream_ordering > COALESCE(threaded_receipt_stream_ordering, ?)) + ( + last_receipt_stream_ordering IS NULL + AND stream_ordering > COALESCE(threaded_receipt_stream_ordering, ?) + AND COALESCE(threaded_receipt_stream_ordering, ?) <= ? + ) OR last_receipt_stream_ordering = COALESCE(threaded_receipt_stream_ordering, ?) + {unreconcilable_clause} ) AND (notif_count != 0 OR COALESCE(unread_count, 0) != 0) """, ( @@ -635,6 +700,10 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas room_id, user_id, from_stream_ordering, + # 0 rather than the join position: with no receipt at all there is + # nothing for the row to be out of date with respect to. + unthreaded_receipt_stream_ordering or 0, + self._pruned_upto_stream_ordering(), from_stream_ordering, ), ) @@ -1428,6 +1497,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas retcol="stream_ordering", ) + # NOTE: unlike `_rotate_notifs_before_txn` and the readers, this does not + # filter on `receipt_type`, so a receipt of some other type stamps a + # `last_receipt_stream_ordering` the readers will never match against. That + # predates this code; the consequence is that the row is treated as out of + # date (a bounded overcount), not that its counts are lost. sql = """ SELECT r.stream_id, r.room_id, r.user_id, r.thread_id, r.event_stream_ordering FROM receipts_linearized AS r diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index 158426f83a..6781cc037e 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -472,6 +472,343 @@ class EventPushActionsStoreTestCase(HomeserverTestCase): self.get_success(self.store._rotate_notifs()) _assert_badge(1) + def _send_message( + self, + room_id: str, + tok: str, + thread_root: str | None = None, + highlight_for: str | None = None, + ) -> str: + """Send a message, optionally into a thread and/or mentioning a user.""" + content: JsonDict = { + "msgtype": "m.text", + "body": highlight_for if highlight_for is not None else "msg", + } + if thread_root is not None: + content["m.relates_to"] = { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root, + } + return self.helper.send_event( + room_id, type="m.room.message", content=content, tok=tok + )["event_id"] + + def test_count_aggregation_threaded_receipt_for_old_event_after_rotation( + self, + ) -> None: + """ + Regression test: a threaded receipt for an event older than the summary's + high-water mark must be reflected in the counts straight away, rather than + only once the next rotation has run. + + `_rotate_notifs_before_txn` writes `event_push_summary` rows with a NULL + `last_receipt_stream_ordering`. Such a row counts from the user's join, so + it must not be treated as up-to-date once the user has a receipt covering + the thread it summarises. + """ + user_id, _, _, other_token, room_id = self._create_users_and_room() + + def _assert_counts( + notif_count: int, + highlight_count: int, + thread_notif_count: int, + ) -> None: + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-unread-counts", + self.store._get_unread_counts_by_receipt_txn, + room_id, + user_id, + ) + ) + self.assertEqual( + counts.main_timeline, + NotifCounts( + notify_count=notif_count, + unread_count=0, + highlight_count=highlight_count, + ), + ) + if thread_notif_count: + self.assertEqual( + counts.threads, + { + thread_root: NotifCounts( + notify_count=thread_notif_count, + unread_count=0, + highlight_count=0, + ) + }, + ) + else: + self.assertEqual(counts.threads, {}) + + aggregate_counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + self.assertEqual( + aggregate_counts.get(room_id, 0), notif_count + thread_notif_count + ) + + def _mark_read(event_id: str, thread_id: str) -> None: + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[event_id], + thread_id=thread_id, + data={}, + ) + ) + + # Two events on the main timeline (the second one a highlight) and two in a + # thread hanging off the first. + thread_root = self._send_message(room_id, other_token) + first_thread_event = self._send_message( + room_id, other_token, thread_root=thread_root + ) + self._send_message(room_id, other_token, thread_root=thread_root) + self._send_message(room_id, other_token, highlight_for=user_id) + + # Rotate *before* any receipt is sent: this writes summary rows with a NULL + # `last_receipt_stream_ordering`. + self.get_success(self.store._rotate_notifs()) + _assert_counts(2, 1, 2) + + # Read the very first event of the room, which is older than everything the + # summary rows cover. + _mark_read(thread_root, MAIN_TIMELINE) + _assert_counts(1, 1, 2) + + # Same again for the thread. + _mark_read(first_thread_event, thread_root) + _assert_counts(1, 1, 1) + + def test_count_aggregation_unthreaded_receipt_for_old_event_after_rotation( + self, + ) -> None: + """ + As above, but for an unthreaded receipt, which applies to every thread in + the room (including the main timeline). + + A second room with an up-to-date summary is used to check that rejecting one + room's summary row doesn't stop the counts being recovered from + `event_push_actions`. + """ + user_id, token, other_id, other_token, room_id = self._create_users_and_room() + + other_room_id = self.helper.create_room_as(user_id, tok=token) + self.helper.join(other_room_id, other_id, tok=other_token) + + def _assert_counts(notif_count: int, thread_notif_count: int) -> None: + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-unread-counts", + self.store._get_unread_counts_by_receipt_txn, + room_id, + user_id, + ) + ) + self.assertEqual( + counts.main_timeline, + NotifCounts( + notify_count=notif_count, unread_count=0, highlight_count=0 + ), + ) + if thread_notif_count: + self.assertEqual( + counts.threads, + { + thread_root: NotifCounts( + notify_count=thread_notif_count, + unread_count=0, + highlight_count=0, + ) + }, + ) + else: + self.assertEqual(counts.threads, {}) + + aggregate_counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + self.assertEqual( + aggregate_counts.get(room_id, 0), notif_count + thread_notif_count + ) + # The other room has no receipts at all, so its (also NULL) summary row + # is still up-to-date. + self.assertEqual(aggregate_counts.get(other_room_id, 0), 1) + + # The other room gets a single notification, which is never read. + self._send_message(other_room_id, other_token) + + thread_root = self._send_message(room_id, other_token) + first_thread_event = self._send_message( + room_id, other_token, thread_root=thread_root + ) + self._send_message(room_id, other_token) + + self.get_success(self.store._rotate_notifs()) + _assert_counts(2, 1) + + # An unthreaded receipt for the thread event clears everything at or before + # it, in the thread and on the main timeline. + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[first_thread_event], + thread_id=None, + data={}, + ) + ) + _assert_counts(1, 0) + + def test_count_aggregation_summary_written_after_receipt_survives_pruning( + self, + ) -> None: + """ + Regression test: rotation mints a summary row for every thread with new push + actions, including threads that only appear *after* the user's latest read + receipt. `_handle_new_receipts_for_notifs_txn` stamps + `last_receipt_stream_ordering` only in response to a *new* receipt, so such a + row must record the receipt it was calculated against as it is written -- + otherwise nothing can tell it apart from a row that predates the receipt, and + its counts are lost as soon as + `_remove_old_push_actions_that_have_rotated` deletes the backing push actions. + """ + user_id, _, _, other_token, room_id = self._create_users_and_room() + + # Read the first event of the room, and let the receipt be processed. This + # leaves the room with no summary rows at all. + thread_root = self._send_message(room_id, other_token) + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[thread_root], + thread_id=None, + data={}, + ) + ) + self.get_success(self.store._rotate_notifs()) + + # A thread which only starts after that receipt: rotation mints a brand new + # summary row for it. + self._send_message(room_id, other_token, thread_root=thread_root) + self._send_message(room_id, other_token, thread_root=thread_root) + self.get_success(self.store._rotate_notifs()) + + # Every row rotation wrote must record the receipt it counted against. + summary_rows = self.get_success( + self.store.db_pool.simple_select_list( + table="event_push_summary", + keyvalues={"user_id": user_id, "room_id": room_id}, + retcols=("thread_id", "notif_count", "last_receipt_stream_ordering"), + ) + ) + self.assertEqual( + [(thread_id, count) for thread_id, count, _ in summary_rows if count], + [(thread_root, 2)], + ) + for _, _, last_receipt_stream_ordering in summary_rows: + self.assertIsNotNone(last_receipt_stream_ordering) + + # Deleting the (rotated, and now more than a day old) push actions must not + # lose the counts: the summary is all that is left. + self.pump(60 * 60 * 24) + self.get_success(self.store._remove_old_push_actions_that_have_rotated()) + self.assertEqual( + self.get_success( + self.store.db_pool.simple_select_list( + table="event_push_actions", + keyvalues=None, + retcols=("event_id",), + ) + ), + [], + ) + + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-unread-counts", + self.store._get_unread_counts_by_receipt_txn, + room_id, + user_id, + ) + ) + self.assertEqual(counts.main_timeline, NotifCounts()) + self.assertEqual( + counts.threads, {thread_root: NotifCounts(notify_count=2, unread_count=0)} + ) + + aggregate_counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + self.assertEqual(aggregate_counts.get(room_id, 0), 2) + + def test_count_aggregation_receipt_before_join_survives_pruning(self) -> None: + """ + Regression test: a summary row stamped with a receipt from before the user's + latest membership event must keep its counts. + + With no unthreaded receipt the counts are taken from the user's join, so such + a stamp can never match and the row would be discarded forever -- losing the + counts entirely once the push actions behind them have been deleted. Nothing + recalculates the row either, as the receipt is not new. + """ + user_id, token, _, other_token, room_id = self._create_users_and_room() + + # A notification the user reads with a threaded receipt, so that the summary + # row gets stamped with its stream ordering... + old_event = self._send_message(room_id, other_token) + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[old_event], + thread_id=MAIN_TIMELINE, + data={}, + ) + ) + # ...and one they don't read, which is what the summary ends up counting. + self._send_message(room_id, other_token) + self.get_success(self.store._rotate_notifs()) + + # Now make the user's membership event newer than that receipt. + self.helper.leave(room_id, user_id, tok=token) + self.helper.join(room_id, user_id, tok=token) + + # Delete the push actions the summary row stands in for. + self.pump(60 * 60 * 24) + self.get_success(self.store._remove_old_push_actions_that_have_rotated()) + + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-unread-counts", + self.store._get_unread_counts_by_receipt_txn, + room_id, + user_id, + ) + ) + self.assertEqual(counts.main_timeline, NotifCounts(notify_count=1)) + def test_count_aggregation_badge_recount_is_scoped_per_room(self) -> None: """ Regression test: a room whose summary row is out of date must be recounted