From d7e9a3ff8321fa820f38c17a4484f0437233cebb Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 12 Jun 2026 16:31:26 +0100 Subject: [PATCH] Fix inflated notification counts after `/purge_history` (#19834) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- changelog.d/19834.bugfix | 1 + .../storage/databases/main/purge_events.py | 59 ++++++++++++++ tests/storage/test_event_push_actions.py | 77 +++++++++++++++++++ 3 files changed, 137 insertions(+) create mode 100644 changelog.d/19834.bugfix diff --git a/changelog.d/19834.bugfix b/changelog.d/19834.bugfix new file mode 100644 index 0000000000..097a09bac6 --- /dev/null +++ b/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. diff --git a/synapse/storage/databases/main/purge_events.py b/synapse/storage/databases/main/purge_events.py index fe8079c201..9041cd232b 100644 --- a/synapse/storage/databases/main/purge_events.py +++ b/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",): diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index d5ed947094..85ebf26128 100644 --- a/tests/storage/test_event_push_actions.py +++ b/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