Fix inflated notification counts after /purge_history (#19834)

The sytest `After /purge_history users still get pushed for new
messages` is flaky. The flakiness exposes a real bug rather than a
test-timing issue.

Notification counts are stored in two places: `event_push_actions` (one
row per unread event) and `event_push_summary` (aggregate counts
populated periodically by `_rotate_notifs`, which runs on a 30-second
timer). `_purge_history_txn` deletes the purged events' rows from
`event_push_actions` but never adjusts `event_push_summary` (only the
full-room `purge_room` drops that table).

So the result depends on a race: if rotation hasn't fired, counts come
live from `event_push_actions`, the purge removes the right rows, and
the count is correct. If rotation fires before the purge — more likely
under the slower
multi-postgres/workers/asyncio CI config — the events get folded into
`event_push_summary`, the purge then deletes the underlying
`event_push_actions` rows but leaves the summary untouched, and the
count comes out inflated.

### Fix

Before deleting the rotated rows from `event_push_actions`, decrement
`event_push_summary` by the amount attributable to the events being
deleted. The decrement mirrors the counting logic in
`_rotate_notifs_before_txn`: only rows that were already rotated
(`stream_ordering <= event_push_summary_stream_ordering`) and that fall
after the summary's receipt are subtracted, so it stays correct in the
presence of read receipts and unread/highlight rows. The SQL avoids
`UPDATE ... FROM` and CTEs so it works on both SQLite and Postgres.
End-of-purge cache invalidation already covers
`get_unread_event_push_actions_by_room_for_user`.

### Tests

Adds `test_count_aggregation_after_purge`, which forces a rotation
before purging and asserts the aggregate count reflects only the
surviving events, covering read receipts and a subsequent re-rotation.
It fails (`3 != 1`) without the fix.

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

changelog.d/19834.bugfix

@@ -0,0 +1 @@
+Fix notification counts being inflated after a `/purge_history` when notifications had already been rotated into the summary table.

synapse/storage/databases/main/purge_events.py

@@ -326,6 +326,65 @@ class PurgeEventsStore(StateGroupWorkerStore, CacheInvalidationWorkerStore):
                 ")" % (table,)
             )
 
+        # Some of the `event_push_actions` we're about to delete may have already
+        # been rotated into the aggregate `event_push_summary` counts. Deleting
+        # the rows without adjusting those counts would leave the summary
+        # over-counting, inflating users' notification counts. So first work out
+        # how much of each summary is attributable to the events being deleted and
+        # decrement it.
+        #
+        # We only count rows that have already been rotated into the summary:
+        # those at or before the rotated-up-to position
+        # (`event_push_summary_stream_ordering`) and after the receipt used to
+        # compute the summary. Rows beyond that position aren't in the summary
+        # yet (they're still counted live from `event_push_actions`), so deleting
+        # them needs no adjustment here. This mirrors how rotation counts them in
+        # `_rotate_notifs_before_txn`.
+        logger.info("[purge] adjusting event_push_summary for deleted events")
+        txn.execute(
+            """
+            SELECT epa.user_id, epa.thread_id,
+                COUNT(CASE WHEN epa.notif = 1 THEN 1 END),
+                COUNT(CASE WHEN epa.unread = 1 THEN 1 END)
+            FROM event_push_actions AS epa
+            INNER JOIN event_push_summary AS eps USING (user_id, room_id, thread_id)
+            WHERE epa.room_id = ?
+                AND epa.event_id IN (
+                    SELECT event_id FROM events_to_purge WHERE should_delete
+                )
+                AND epa.stream_ordering <= (
+                    SELECT stream_ordering FROM event_push_summary_stream_ordering
+                )
+                AND (
+                    eps.last_receipt_stream_ordering IS NULL
+                    OR epa.stream_ordering > eps.last_receipt_stream_ordering
+                )
+            GROUP BY epa.user_id, epa.thread_id
+            """,
+            (room_id,),
+        )
+        summary_decrements = cast(list[tuple[str, str, int, int]], txn.fetchall())
+
+        # `unread_count` is nullable, so `COALESCE` it before subtracting (else
+        # the result would be NULL). Clamp both counts at 0 via `GREATEST`/`MAX`
+        # to guard against ever driving a count negative if the summary is
+        # somehow out of sync with `event_push_actions`.
+        greatest_func = (
+            "GREATEST" if isinstance(self.database_engine, PostgresEngine) else "MAX"
+        )
+        txn.execute_batch(
+            f"""
+            UPDATE event_push_summary
+            SET notif_count = {greatest_func}(notif_count - ?, 0),
+                unread_count = {greatest_func}(COALESCE(unread_count, 0) - ?, 0)
+            WHERE room_id = ? AND user_id = ? AND thread_id = ?
+            """,
+            [
+                (notif_count, unread_count, room_id, user_id, thread_id)
+                for user_id, thread_id, notif_count, unread_count in summary_decrements
+            ],
+        )
+
         # event_push_actions lacks an index on event_id, and has one on
         # (room_id, event_id) instead.
         for table in ("event_push_actions",):

tests/storage/test_event_push_actions.py

@@ -286,6 +286,83 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         _rotate()
         _assert_counts(0, 0)
 
+    def test_count_aggregation_after_purge(self) -> None:
+        """Purging history must not leave stale counts in event_push_summary.
+
+        Regression test: if notifications had been rotated into
+        `event_push_summary` before a history purge, deleting the purged rows
+        from `event_push_actions` left the summary counts unchanged, inflating
+        the notification count.
+        """
+        user_id, _, _, other_token, room_id = self._create_users_and_room()
+
+        def _assert_count(notif_count: int) -> None:
+            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)
+
+        def _create_event() -> str:
+            result = self.helper.send_event(
+                room_id,
+                type="m.room.message",
+                content={"msgtype": "m.text", "body": "msg"},
+                tok=other_token,
+            )
+            return result["event_id"]
+
+        def _mark_read(event_id: str) -> None:
+            self.get_success(
+                self.store.insert_receipt(
+                    room_id,
+                    "m.read",
+                    user_id=user_id,
+                    event_ids=[event_id],
+                    thread_id=None,
+                    data={},
+                )
+            )
+
+        def _purge_before(event_id: str) -> None:
+            token = self.get_success(
+                self.store.get_topological_token_for_event(event_id)
+            )
+            token_str = self.get_success(token.to_string(self.store))
+            self.get_success(
+                self.store.purge_history(room_id, token_str, delete_local_events=True)
+            )
+
+        # Mark an initial event as read so that the summary tracks a receipt.
+        read_event_id = _create_event()
+        _mark_read(read_event_id)
+        _assert_count(0)
+
+        # Send some events and rotate them into the summary table.
+        _create_event()
+        _create_event()
+        cutoff_event_id = _create_event()
+        self.get_success(self.store._rotate_notifs())
+        _assert_count(3)
+
+        # Purge history before the most recent event, which deletes the earlier
+        # events (including the one before the read receipt).
+        _purge_before(cutoff_event_id)
+
+        # Only the most recent event survives, so the count should be 1.
+        _assert_count(1)
+
+        # A subsequent rotation must not resurrect the purged counts.
+        self.get_success(self.store._rotate_notifs())
+        _assert_count(1)
+
+        # Reading the surviving event clears the count entirely.
+        _mark_read(cutoff_event_id)
+        _assert_count(0)
+
     def test_count_aggregation_threads(self) -> None:
         """
         This is essentially the same test as test_count_aggregation, but adds