From 8750a01ffb5e17e77f57aa3262f2fe81da8b3ea1 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 9 Jun 2026 13:22:57 +0100 Subject: [PATCH 1/3] Make MSC4102 "prefer unthreaded receipt" durable at insert time Previously the "unthreaded receipt always wins over a clashing threaded one" behaviour was only applied at read time, in `ReceiptInRoom.merge_to_content`, which dedupes the pair within a single /sync response. When the two receipts end up at different stream positions and are served in separate /sync responses (e.g. when they arrive over federation as separate EDUs), the threaded receipt could be served on its own and incorrectly win, which is what caused the `TestThreadReceiptsInSyncMSC4102` Complement flake. Drop a threaded receipt at insert time if an unthreaded receipt for the same user already exists at the same or a later event, so it never gets persisted, streamed or federated. This is safe for notification counts, since the unthreaded receipt already acts as a floor across all threads. Co-Authored-By: Claude Opus 4.8 --- changelog.d/19838.bugfix | 1 + synapse/storage/databases/main/receipts.py | 82 +++++++++++++++------- tests/storage/test_receipts.py | 78 ++++++++++++++++++++ 3 files changed, 136 insertions(+), 25 deletions(-) create mode 100644 changelog.d/19838.bugfix diff --git a/changelog.d/19838.bugfix b/changelog.d/19838.bugfix new file mode 100644 index 00000000000..064bd343484 --- /dev/null +++ b/changelog.d/19838.bugfix @@ -0,0 +1 @@ +Fix a bug where a threaded read receipt could incorrectly win over a clashing unthreaded one (MSC4102) when the two were served in separate `/sync` responses, e.g. when received over federation as separate EDUs. diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index ba5e07a0510..8af2030d2fb 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -908,35 +908,67 @@ def _insert_linearized_receipt_txn( rx_ts = res[1] if res else 0 # We don't want to clobber receipts for more recent events, so we - # have to compare orderings of existing receipts + # have to compare orderings of existing receipts. + # + # We fetch the user's existing receipts for this room/type in a single + # query: every receipt whose event ordering we know (for the + # same-thread comparison below), plus the unthreaded receipt for *this* + # event even if we don't know its ordering (for the MSC4102 check + # below). if stream_ordering is not None: - if thread_id is None: - thread_clause = "r.thread_id IS NULL" - thread_args: tuple[str, ...] = () - else: - thread_clause = "r.thread_id = ?" - thread_args = (thread_id,) - - # If the receipt doesn't have a stream ordering it is because we - # don't have the associated event, and so must be a remote receipt. - # Hence it's safe to just allow new receipts to clobber it. - sql = f""" - SELECT r.event_stream_ordering, r.event_id FROM receipts_linearized AS r + sql = """ + SELECT r.event_stream_ordering, r.event_id, r.thread_id + FROM receipts_linearized AS r WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ? - AND r.event_stream_ordering IS NOT NULL AND {thread_clause} - """ - txn.execute( - sql, - ( - room_id, - receipt_type, - user_id, - ) - + thread_args, + AND ( + r.event_stream_ordering IS NOT NULL + OR (r.thread_id IS NULL AND r.event_id = ?) ) + """ + txn.execute(sql, (room_id, receipt_type, user_id, event_id)) + + for so, eid, existing_thread_id in txn: + # MSC4102: an unthreaded receipt always supersedes a threaded + # one for the *same* event. If we are inserting a threaded + # receipt but already have an unthreaded receipt for this user + # at the same event, then the threaded receipt is semantically + # meaningless, so we drop it rather than persisting it. + # + # We match on event id (mirroring the read-time dedup in + # `ReceiptInRoom.merge_to_content`) rather than on ordering, so + # this also covers a remote unthreaded receipt whose event we + # hadn't seen when it arrived (and so has a NULL + # event_stream_ordering). + # + # Doing this at insert time, as well as when serving receipts, + # makes the "prefer unthreaded" behaviour durable: otherwise the + # threaded receipt could be served on its own in a later /sync + # response (e.g. when the unthreaded and threaded receipts + # arrive in separate federation EDUs and so end up at different + # stream positions), causing the client to incorrectly see it + # win. + if ( + thread_id is not None + and existing_thread_id is None + and eid == event_id + ): + logger.debug( + "Ignoring threaded receipt for %s in favour of " + "existing unthreaded receipt", + event_id, + ) + return None + + # The remaining check compares stream orderings, so skip + # receipts for older events, and remote receipts whose event we + # don't have (NULL ordering) since it's safe to clobber those. + if so is None or int(so) < stream_ordering: + continue - for so, eid in txn: - if int(so) >= stream_ordering: + # Don't clobber a receipt for a more recent event in the same + # thread. (When inserting an unthreaded receipt, thread_id is + # None and this matches the existing unthreaded receipt.) + if existing_thread_id == thread_id: logger.debug( "Ignoring new receipt for %s in favour of existing " "one for later event %s", diff --git a/tests/storage/test_receipts.py b/tests/storage/test_receipts.py index 27875dcebb6..5d47ffb5a0c 100644 --- a/tests/storage/test_receipts.py +++ b/tests/storage/test_receipts.py @@ -237,6 +237,84 @@ def test_get_receipts_for_user(self) -> None: ) self.assertEqual(res, {self.room_id1: event1_2_id, self.room_id2: event2_1_id}) + def test_threaded_receipt_dropped_when_unthreaded_exists(self) -> None: + """MSC4102: a threaded receipt that clashes with an existing unthreaded + receipt *for the same event* should be dropped at insert time, so the + unthreaded receipt durably wins regardless of how the receipts are + later served down /sync. A threaded receipt for a *different* event is + not a clash and is kept (matching the read-time dedup in + `ReceiptInRoom.merge_to_content`, which keys off event id). + + Regression test for the Complement test TestThreadReceiptsInSyncMSC4102, + which flaked because the read-time dedup alone doesn't survive the + receipts being served in separate /sync responses. + """ + event1_id = self.create_and_send_event( + self.room_id1, UserID.from_string(OTHER_USER_ID) + ) + # A second event. event1 is used as the thread-root id for the + # threaded receipts below; storage doesn't check thread membership. + event2_id = self.create_and_send_event( + self.room_id1, UserID.from_string(OTHER_USER_ID) + ) + + # Insert an unthreaded receipt for the second event. + pos = self.get_success( + self.store.insert_receipt( + room_id=self.room_id1, + receipt_type=ReceiptTypes.READ, + user_id=OUR_USER_ID, + event_ids=[event2_id], + thread_id=None, + data={}, + ) + ) + self.assertIsNotNone(pos) + + # A threaded receipt that clashes with it (same event) should be + # dropped. + pos = self.get_success( + self.store.insert_receipt( + room_id=self.room_id1, + receipt_type=ReceiptTypes.READ, + user_id=OUR_USER_ID, + event_ids=[event2_id], + thread_id=event1_id, + data={}, + ) + ) + self.assertIsNone(pos) + + # A threaded receipt for a *different* event is not a clash, so it is + # kept. + pos = self.get_success( + self.store.insert_receipt( + room_id=self.room_id1, + receipt_type=ReceiptTypes.READ, + user_id=OUR_USER_ID, + event_ids=[event1_id], + thread_id=event1_id, + data={}, + ) + ) + self.assertIsNotNone(pos) + + # The unthreaded receipt at event2 wins (no thread_id), and the + # non-clashing threaded receipt at event1 is retained. + to_key = self.store.get_max_receipt_stream_id() + receipts = self.get_success( + self.store.get_linearized_receipts_for_rooms([self.room_id1], to_key=to_key) + ) + content = receipts[0]["content"] + self.assertEqual(set(content.keys()), {event1_id, event2_id}) + self.assertNotIn( + "thread_id", content[event2_id][ReceiptTypes.READ][OUR_USER_ID] + ) + self.assertEqual( + content[event1_id][ReceiptTypes.READ][OUR_USER_ID]["thread_id"], + event1_id, + ) + def test_get_last_receipt_event_id_for_user(self) -> None: # Send some events into the first room event1_1_id = self.create_and_send_event( From d5f8ea40c5b7287f31d6d11260a696899b6c9532 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 15 Jun 2026 10:35:30 +0100 Subject: [PATCH 2/3] Mention why its safe to clobber null stream ordering --- synapse/storage/databases/main/receipts.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index 8af2030d2fb..d9780ac1aa4 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -961,7 +961,9 @@ def _insert_linearized_receipt_txn( # The remaining check compares stream orderings, so skip # receipts for older events, and remote receipts whose event we - # don't have (NULL ordering) since it's safe to clobber those. + # don't have (NULL ordering). It's safe to clobber events with + # NULL ordering as they are pointing at unknown events that we + # haven't seen (and are likely from before we joined the room). if so is None or int(so) < stream_ordering: continue From dcc2f64988a3117c4c905d8336ea9f1ce894078e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Mon, 15 Jun 2026 10:38:39 +0100 Subject: [PATCH 3/3] Mention we're fetching both threaded and unthreaded receipts --- synapse/storage/databases/main/receipts.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py index d9780ac1aa4..4c1e88af652 100644 --- a/synapse/storage/databases/main/receipts.py +++ b/synapse/storage/databases/main/receipts.py @@ -907,14 +907,14 @@ def _insert_linearized_receipt_txn( stream_ordering = int(res[0]) if res else None rx_ts = res[1] if res else 0 - # We don't want to clobber receipts for more recent events, so we - # have to compare orderings of existing receipts. + # We don't want to clobber receipts for more recent events, so we have + # to compare orderings of existing receipts. # - # We fetch the user's existing receipts for this room/type in a single - # query: every receipt whose event ordering we know (for the - # same-thread comparison below), plus the unthreaded receipt for *this* - # event even if we don't know its ordering (for the MSC4102 check - # below). + # We fetch the user's existing receipts for this room and receipt type + # (including both threaded and unthreaded receipts) in a single query: + # every receipt whose event ordering we know (for the same-thread + # comparison below), plus the unthreaded receipt for *this* event even + # if we don't know its ordering (for the MSC4102 check below). if stream_ordering is not None: sql = """ SELECT r.event_stream_ordering, r.event_id, r.thread_id