From 7385d4c41ac9998b18dc99e0cb99fa62ebef2adf Mon Sep 17 00:00:00 2001 From: Quentin Gliech Date: Fri, 14 Aug 2026 15:15:01 +0200 Subject: [PATCH] Record the read receipt rotation counted against in `event_push_summary` `_rotate_notifs_before_txn` upserted summary rows without setting `last_receipt_stream_ordering`, so every row it inserted was left NULL. Only `_handle_new_receipts_for_notifs_txn` ever set it, and only in response to a *new* receipt, so a row created for a thread that first got push actions after the user's last receipt stayed NULL forever, indistinguishable from a row that predates the receipt entirely. Stamp it on insert with the user's latest read receipt for the thread (or the room, for an unthreaded one). Rows that already exist keep the value `_handle_new_receipts_for_notifs_txn` gave them, since their counts have not been recalculated since -- which is also what stops the below-receipt highlights it deliberately leaves in `event_push_actions` being counted twice. A row is now NULL only if the user has never sent a read receipt in the room. (cherry picked from commit 6e94c5befe1910ed3991cf64d8d0a0f2f2432a4b) --- .../databases/main/event_push_actions.py | 51 +++++++++++++++++-- 1 file changed, 47 insertions(+), 4 deletions(-) diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index 039a93ba7e..a8b60f912e 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -1653,12 +1653,39 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas rotate_to_stream_ordering: The new maximum event stream ordering to summarise. """ - # Calculate the new counts that should be upserted into event_push_summary + receipt_types_clause, receipt_types_args = make_in_list_sql_clause( + self.database_engine, + "receipt_type", + (ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE), + ) + + # Calculate the new counts that should be upserted into event_push_summary. + # + # A row we insert here has to record which receipt its counts are relative to, + # otherwise `_get_unread_counts_by_pos_txn` has no way of telling whether the + # row predates the user's receipt. `_handle_new_receipts_for_notifs_txn` ran + # first, so barring a receipt that landed in the gap between the two phases + # (which self-heals on the next rotation, at the cost of overcounting it until + # then) the user's latest receipt for the thread is what the counts below are + # relative to: rows we are updating keep the value it stamped (their counts + # have not been recalculated since), and rows we are inserting get the latest + # receipt, or NULL if the user has never sent one in the room. def _sql(count_column: str, action_column: str) -> str: return f""" SELECT user_id, room_id, thread_id, coalesce(old.{count_column}, 0) + upd.cnt, - upd.stream_ordering + upd.stream_ordering, + COALESCE( + old.last_receipt_stream_ordering, + ( + SELECT MAX(r.event_stream_ordering) + FROM receipts_linearized AS r + WHERE r.user_id = upd.user_id + AND r.room_id = upd.room_id + AND (r.thread_id IS NULL OR r.thread_id = upd.thread_id) + AND {receipt_types_clause} + ) + ) FROM ( SELECT user_id, room_id, thread_id, count(*) as cnt, max(ea.stream_ordering) as stream_ordering @@ -1675,7 +1702,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas LEFT JOIN event_push_summary AS old USING (user_id, room_id, thread_id) """ - args = (old_rotate_stream_ordering, rotate_to_stream_ordering) + args = ( + *receipt_types_args, + old_rotate_stream_ordering, + rotate_to_stream_ordering, + ) # First get the count of unread messages. txn.execute(_sql("unread_count", "unread"), args) @@ -1691,6 +1722,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas unread_count=row[3], stream_ordering=row[4], notif_count=0, + last_receipt_stream_ordering=row[5], ) # Then get the count of notifications. @@ -1708,6 +1740,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas unread_count=0, stream_ordering=row[4], notif_count=row[3], + last_receipt_stream_ordering=row[5], ) logger.info("Rotating notifications, handling %d rows", len(summaries)) @@ -1717,12 +1750,18 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas table="event_push_summary", key_names=("user_id", "room_id", "thread_id"), key_values=list(summaries), - value_names=("notif_count", "unread_count", "stream_ordering"), + value_names=( + "notif_count", + "unread_count", + "stream_ordering", + "last_receipt_stream_ordering", + ), value_values=[ ( summary.notif_count, summary.unread_count, summary.stream_ordering, + summary.last_receipt_stream_ordering, ) for summary in summaries.values() ], @@ -1966,3 +2005,7 @@ class _EventPushSummary: unread_count: int stream_ordering: int notif_count: int + # The stream ordering of the user's latest read receipt for the thread at the + # point these counts were calculated, or None if they have never sent one in the + # room. See `_get_unread_counts_by_pos_txn` for how this is used. + last_receipt_stream_ordering: int | None