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 <noreply@anthropic.com>

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.

synapse/storage/databases/main/receipts.py

@@ -908,35 +908,67 @@ class ReceiptsWorkerStore(SQLBaseStore):
         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 in txn:
-                if int(so) >= stream_ordering:
+            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
+
+                # 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",

tests/storage/test_receipts.py

@@ -237,6 +237,84 @@ class ReceiptTestCase(HomeserverTestCase):
         )
         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(