Honour a read receipt for an event older than the last summary rotation

`event_push_summary` rows written by rotation for a room the user has never
sent a receipt in have a NULL `last_receipt_stream_ordering`, and the badge
queries read NULL as "already accounts for every receipt" -- a compatibility
branch for rows written before Synapse v1.62.0. It is only sound while the user
has no receipt: such a row counts from the user's join, so once a receipt
arrives it may include events at or before it. A receipt for an event older
than the row's high-water mark was therefore ignored until the next rotation,
up to 30 seconds later, leaving a stale unread count.

Only trust a NULL row while the user has no receipt covering the thread. The
manual count from `event_push_actions` that replaces it can only be complete
above the point `_remove_old_push_actions_that_have_rotated` deletes from, so
below that we keep serving the (possibly stale) summary and let the next
rotation correct it: distrusting the row there would drop the counts entirely
rather than merely delay them.

For the same reason, keep a row whose stamp cannot be reconciled with any
receipt at all. With no unthreaded receipt the comparison is against the user's
membership event, which a receipt from before their latest join can never
match -- and nothing recalculates such a row, since the receipt is not new.

(cherry picked from commit f18998e6ce)

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.

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

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