Take an optional unthreaded receipt position in _get_unread_counts_by_pos_txn

The caller passed either the user's latest unthreaded receipt or, when they
had none, the stream ordering of their join, leaving the callee unable to tell
the two apart. Pass `None` instead and do the membership lookup in the callee,
so the fallback is explicit.

changelog.d/20112.bugfix

@@ -0,0 +1 @@
+Fix a bug where push badge counts could ignore notifications in one room based on another room's summarised counts.

changelog.d/20113.bugfix

@@ -0,0 +1 @@
+Fix push badge counts on PostgreSQL for rooms where the user has no read receipt.

synapse/storage/databases/main/event_push_actions.py

@@ -407,15 +407,16 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
         # stream ordering of both threaded & unthreaded receipts to compare against the
         # summary table.
         #
-        # PostgreSQL and SQLite differ in comparing scalar numerics.
+        # PostgreSQL and SQLite differ in comparing scalar numerics. Both COALESCE to
+        # 0 first so that a user with no receipts at all in the room compares against
+        # 0 (rather than NULL, which would make every comparison below unknown) on
+        # both engines.
         if isinstance(self.database_engine, PostgresEngine):
-            # GREATEST ignores NULLs.
             max_clause = """GREATEST(
-                threaded_receipt_stream_ordering,
-                unthreaded_receipt_stream_ordering
+                COALESCE(threaded_receipt_stream_ordering, 0),
+                COALESCE(unthreaded_receipt_stream_ordering, 0)
             )"""
         else:
-            # MAX returns NULL if any are NULL, so COALESCE to 0 first.
             max_clause = """MAX(
                 COALESCE(threaded_receipt_stream_ordering, 0),
                 COALESCE(unthreaded_receipt_stream_ordering, 0)
@@ -435,12 +436,13 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
         """
         txn.execute(sql, args)
 
-        seen_thread_ids = set()
+        # The (room ID, thread ID) pairs we found an up-to-date summary for.
+        seen_room_thread_ids = set()
         room_to_count: dict[str, int] = defaultdict(int)
 
         for room_id, thread_id, notif_count in txn:
             room_to_count[room_id] += notif_count
-            seen_thread_ids.add(thread_id)
+            seen_room_thread_ids.add((room_id, thread_id))
 
         # Now get any event push actions that haven't been rotated using the same OR
         # join and filter by receipt and event push summary rotated up to stream ordering.
@@ -460,35 +462,31 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
 
         for room_id, thread_id, notif_count in txn:
             # Note: only count push actions we have valid summaries for with up to date receipt.
-            if thread_id not in seen_thread_ids:
+            if (room_id, thread_id) not in seen_room_thread_ids:
                 continue
             room_to_count[room_id] += notif_count
 
-        thread_id_clause, thread_ids_args = make_in_list_sql_clause(
-            self.database_engine, "epa.thread_id", seen_thread_ids
-        )
-
-        # Finally re-check event_push_actions for any rooms not in the summary, ignoring
-        # the rotated up-to position. This handles the case where a read receipt has arrived
-        # but not been rotated meaning the summary table is out of date, so we go back to
-        # the push actions table.
+        # Finally re-check event_push_actions for any room/threads not in the summary,
+        # ignoring the rotated up-to position. This handles the case where a read receipt
+        # has arrived but not been rotated meaning the summary table is out of date, so we
+        # go back to the push actions table.
         sql = f"""
             {receipts_cte}
-            SELECT epa.room_id, COUNT(CASE WHEN epa.notif = 1 THEN 1 END) AS notif_count
+            SELECT epa.room_id, epa.thread_id, COUNT(CASE WHEN epa.notif = 1 THEN 1 END) AS notif_count
             FROM event_push_actions AS epa
             {receipts_joins}
             WHERE user_id = ?
-            AND NOT {thread_id_clause}
             AND epa.notif = 1
             AND (threaded_receipt_stream_ordering IS NULL OR stream_ordering > threaded_receipt_stream_ordering)
             AND (unthreaded_receipt_stream_ordering IS NULL OR stream_ordering > unthreaded_receipt_stream_ordering)
-            GROUP BY epa.room_id
+            GROUP BY epa.room_id, epa.thread_id
         """
 
-        args.extend(thread_ids_args)
         txn.execute(sql, args)
 
-        for room_id, notif_count in txn:
+        for room_id, thread_id, notif_count in txn:
+            if (room_id, thread_id) in seen_room_thread_ids:
+                continue
             room_to_count[room_id] += notif_count
 
         return room_to_count
@@ -536,24 +534,8 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             receipt_types=(ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE),
         )
 
-        if result:
-            _, stream_ordering = result
-
-        else:
-            # If the user has no receipts in the room, retrieve the stream ordering for
-            # the latest membership event from this user in this room (which we assume is
-            # a join).
-            event_id = self.db_pool.simple_select_one_onecol_txn(
-                txn=txn,
-                table="local_current_membership",
-                keyvalues={"room_id": room_id, "user_id": user_id},
-                retcol="event_id",
-            )
-
-            stream_ordering = self.get_stream_id_for_event_txn(txn, event_id)
-
         return self._get_unread_counts_by_pos_txn(
-            txn, room_id, user_id, stream_ordering
+            txn, room_id, user_id, result[1] if result else None
         )
 
     def _get_unread_counts_by_pos_txn(
@@ -561,7 +543,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
         txn: LoggingTransaction,
         room_id: str,
         user_id: str,
-        unthreaded_receipt_stream_ordering: int,
+        unthreaded_receipt_stream_ordering: int | None,
     ) -> RoomNotifCounts:
         """Get the number of unread messages for a user/room that have happened
         since the given stream ordering.
@@ -571,8 +553,8 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             room_id: The room ID to get unread counts for.
             user_id: The user ID to get unread counts for.
             unthreaded_receipt_stream_ordering: The stream ordering of the user's latest
-                unthreaded receipt in the room. If there are no unthreaded receipts,
-                the stream ordering of the user's join event.
+                unthreaded receipt in the room, or None if they have none. In that case
+                we count from the user's latest membership event instead.
 
         Returns:
             A RoomNotifCounts object containing the notification count, the
@@ -580,6 +562,21 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             and threads.
         """
 
+        if unthreaded_receipt_stream_ordering is not None:
+            from_stream_ordering = unthreaded_receipt_stream_ordering
+        else:
+            # If the user has no unthreaded receipt in the room, retrieve the stream
+            # ordering for the latest membership event from this user in this room
+            # (which we assume is a join).
+            event_id = self.db_pool.simple_select_one_onecol_txn(
+                txn=txn,
+                table="local_current_membership",
+                keyvalues={"room_id": room_id, "user_id": user_id},
+                retcol="event_id",
+            )
+
+            from_stream_ordering = self.get_stream_id_for_event_txn(txn, event_id)
+
         main_counts = NotifCounts()
         thread_counts: dict[str, NotifCounts] = {}
 
@@ -634,12 +631,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 room_id,
                 user_id,
-                unthreaded_receipt_stream_ordering,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
+                from_stream_ordering,
             ),
         )
         summarised_threads = set()
@@ -673,11 +670,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
             ),
         )
         for highlight_count, thread_id in txn:
@@ -744,11 +741,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
             (
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *receipts_args,
                 user_id,
                 room_id,
-                unthreaded_receipt_stream_ordering,
+                from_stream_ordering,
                 *thread_id_args,
             ),
         )

tests/storage/test_event_push_actions.py

@@ -537,6 +537,123 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         self.assertEqual(counts.main_timeline, NotifCounts(notify_count=2))
         self.assertEqual(counts.threads, {"zzz-thread": NotifCounts(notify_count=1)})
 
+    def test_count_aggregation_no_receipt_summary_survives_pruning(self) -> None:
+        """
+        Regression test: a room the user has never sent a read receipt in must keep
+        its badge count once the push actions behind it have been deleted.
+
+        Note that the interesting engine here is Postgres. The badge query trusts a
+        summary row with a NULL `last_receipt_stream_ordering` when its
+        `stream_ordering` is past the user's latest receipt, but with no receipts at
+        all `GREATEST(NULL, NULL)` is NULL, so the comparison was unknown and the row
+        was rejected -- leaving the count to come from `event_push_actions`, which
+        has nothing left to offer. SQLite already COALESCEd to 0 and so was fine.
+        """
+        user_id, _, _, other_token, room_id = self._create_users_and_room()
+
+        self.helper.send_event(
+            room_id,
+            type="m.room.message",
+            content={"msgtype": "m.text", "body": "msg"},
+            tok=other_token,
+        )
+        self.get_success(self.store._rotate_notifs())
+
+        # Delete the (rotated, and now more than a day old) push actions, so that the
+        # summary row is the only remaining record of the notification.
+        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-aggregate-unread-counts",
+                self.store._get_unread_counts_by_room_for_user_txn,
+                user_id,
+            )
+        )
+        self.assertEqual(counts.get(room_id, 0), 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
+        from `event_push_actions`, even when another room has an up-to-date summary
+        for the same thread ID.
+
+        The set of threads a valid summary was found for used to be keyed on the
+        thread ID alone, so a single room with an up-to-date `main` summary excluded
+        `main` from the recount in *every* room, dropping those rooms' counts.
+        """
+        user_id, token, other_id, other_token, room_id = self._create_users_and_room()
+
+        stale_room_id = self.helper.create_room_as(user_id, tok=token)
+        self.helper.join(stale_room_id, other_id, tok=other_token)
+
+        def _send(room: str) -> str:
+            return self.helper.send_event(
+                room,
+                type="m.room.message",
+                content={"msgtype": "m.text", "body": "msg"},
+                tok=other_token,
+            )["event_id"]
+
+        def _read(room: str, event_id: str) -> None:
+            self.get_success(
+                self.store.insert_receipt(
+                    room,
+                    "m.read",
+                    user_id=user_id,
+                    event_ids=[event_id],
+                    thread_id=None,
+                    data={},
+                )
+            )
+
+        def _badge(room: str) -> int:
+            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,
+                )
+            )
+            return counts.get(room, 0)
+
+        # `room_id` keeps an up-to-date summary throughout, so its `main` thread is
+        # always one we found a valid summary for.
+        first = _send(room_id)
+        _send(room_id)
+
+        stale_first = _send(stale_room_id)
+        stale_second = _send(stale_room_id)
+        _send(stale_room_id)
+
+        # Read one event in each room and rotate, so that both summary rows record
+        # the receipt they were calculated against.
+        _read(room_id, first)
+        _read(stale_room_id, stale_first)
+        self.get_success(self.store._rotate_notifs())
+
+        self.assertEqual(_badge(room_id), 1)
+        self.assertEqual(_badge(stale_room_id), 2)
+
+        # A second receipt, which rotation has not processed yet: `stale_room_id`'s
+        # summary row no longer matches it, so its count has to be recovered from
+        # `event_push_actions`.
+        _read(stale_room_id, stale_second)
+
+        self.assertEqual(_badge(room_id), 1)
+        self.assertEqual(_badge(stale_room_id), 1)
+
     def test_count_aggregation_threads(self) -> None:
         """
         This is essentially the same test as test_count_aggregation, but adds