Merge branch 'rei/t2_msc4306_conflict' into rei/t2_pushrules

Author: reivilibre

synapse/api/errors.py

@@ -141,8 +141,10 @@ class Codes(str, Enum):
     INVITE_BLOCKED = "ORG.MATRIX.MSC4155.M_INVITE_BLOCKED"
 
     # Part of MSC4306: Thread Subscriptions
-    SKIPPED = "M_SKIPPED"
-    NOT_IN_THREAD = "M_NOT_IN_THREAD"
+    MSC4306_CONFLICTING_UNSUBSCRIPTION = (
+        "IO.ELEMENT.MSC4306.M_CONFLICTING_UNSUBSCRIPTION"
+    )
+    MSC4306_NOT_IN_THREAD = "IO.ELEMENT.MSC4306.M_NOT_IN_THREAD"
 
 
 class CodeMessageException(RuntimeError):

synapse/handlers/thread_subscriptions.py

@@ -9,7 +9,7 @@
     AutomaticSubscriptionConflicted,
     ThreadSubscription,
 )
-from synapse.types import UserID
+from synapse.types import EventOrderings, UserID
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -87,46 +87,40 @@ async def subscribe_user_to_thread(
         # First check that the user can access the thread root event
         # and that it exists
         try:
-            event = await self.event_handler.get_event(
+            thread_root_event = await self.event_handler.get_event(
                 user_id, room_id, thread_root_event_id
             )
-            if event is None:
+            if thread_root_event is None:
                 raise NotFoundError("No such thread root")
         except AuthError:
             logger.info("rejecting thread subscriptions change (thread not accessible)")
             raise NotFoundError("No such thread root")
 
         if automatic_event_id:
-            event = await self.event_handler.get_event(
+            autosub_cause_event = await self.event_handler.get_event(
                 user_id, room_id, automatic_event_id
             )
-            if event is None:
-                raise NotFoundError("No such automatic subscription cause event")
-            relation = relation_from_event(event)
+            if autosub_cause_event is None:
+                raise NotFoundError("Automatic subscription event not found")
+            relation = relation_from_event(autosub_cause_event)
             if (
                 relation is None
                 or relation.rel_type != RelationTypes.THREAD
                 or relation.parent_id != thread_root_event_id
             ):
                 raise SynapseError(
                     HTTPStatus.BAD_REQUEST,
-                    "Automatic subscription must be caused by an event in the thread",
-                    errcode=Codes.NOT_IN_THREAD,
+                    "Automatic subscription must use an event in the thread",
+                    errcode=Codes.MSC4306_NOT_IN_THREAD,
                 )
 
-            stream_ordering = event.internal_metadata.stream_ordering
-            assert stream_ordering is not None
-            automatic_event_orderings = (
-                stream_ordering,
-                # depth is topological_ordering
-                event.depth,
-            )
+            automatic_event_orderings = EventOrderings.from_event(autosub_cause_event)
         else:
             automatic_event_orderings = None
 
         outcome = await self.store.subscribe_user_to_thread(
             user_id.to_string(),
-            event.room_id,
+            room_id,
             thread_root_event_id,
             automatic_event_orderings=automatic_event_orderings,
         )
@@ -135,7 +129,7 @@ async def subscribe_user_to_thread(
             raise SynapseError(
                 HTTPStatus.CONFLICT,
                 "Automatic subscription obsoleted by an unsubscription request.",
-                errcode=Codes.SKIPPED,
+                errcode=Codes.MSC4306_CONFLICTING_UNSUBSCRIPTION,
             )
 
         return outcome

synapse/storage/databases/main/thread_subscriptions.py

@@ -33,6 +33,7 @@
 )
 from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.util.id_generators import MultiWriterIdGenerator
+from synapse.types import EventOrderings
 from synapse.util.caches.descriptors import cached
 
 if TYPE_CHECKING:
@@ -112,39 +113,48 @@ def process_replication_position(
 
     @staticmethod
     def _should_skip_autosubscription_after_unsubscription(
-        autosub_stream_ordering: int,
-        autosub_topological_ordering: int,
-        unsubscribed_at_stream_ordering: int,
-        unsubscribed_at_topological_ordering: int,
+        *,
+        autosub: EventOrderings,
+        unsubscribed_at: EventOrderings,
     ) -> bool:
         """
-        Returns whether an automatic subscription occurring following an unsubscription
+        Returns whether an automatic subscription occurring *after* an unsubscription
         should be skipped, because the unsubscription already 'acknowledges' the event
         causing the automatic subscription (the cause event).
 
+        To determine *after*, we use `stream_ordering` unless the event is backfilled
+        (negative `stream_ordering`) and fallback to topological ordering.
+
         Args:
-            autosub_stream_ordering: the stream_ordering of the cause event
-            autosub_topological_ordering: the topological_ordering of the cause event
-            unsubscribed_at_stream_ordering: the maximum stream ordering at the time of unsubscription
-            unsubscribed_at_topological_ordering: the maximum stream ordering at the time of unsubscription
+            autosub: the stream_ordering and topological_ordering of the cause event
+            unsubscribed_at:
+                the maximum stream ordering and the maximum topological ordering at the time of unsubscription
 
         Returns:
             True if the automatic subscription should be skipped
         """
-        # these two orderings should be positive, because they don't refer to a specific event
-        # but rather the maximum at the time of unsubscription
-        assert unsubscribed_at_stream_ordering > 0
-        assert unsubscribed_at_topological_ordering > 0
-
-        if unsubscribed_at_stream_ordering >= autosub_stream_ordering > 0:
+        # For normal rooms, these two orderings should be positive, because
+        # they don't refer to a specific event but rather the maximum at the
+        # time of unsubscription.
+        #
+        # However, for rooms that have never been joined and that are being peeked at,
+        # we might not have a single non-backfilled event and therefore the stream
+        # ordering might be negative, so we don't assert this case.
+        assert unsubscribed_at.topological > 0
+
+        unsubscribed_at_backfilled = unsubscribed_at.stream < 0
+        if (
+            not unsubscribed_at_backfilled
+            and unsubscribed_at.stream >= autosub.stream > 0
+        ):
             # non-backfilled events: the unsubscription is later according to
             # the stream
             return True
 
-        if autosub_stream_ordering < 0:
+        if autosub.stream < 0:
             # the auto-subscription cause event was backfilled, so fall back to
             # topological ordering
-            if unsubscribed_at_topological_ordering >= autosub_topological_ordering:
+            if unsubscribed_at.topological >= autosub.topological:
                 return True
 
         return False
@@ -155,20 +165,29 @@ async def subscribe_user_to_thread(
         room_id: str,
         thread_root_event_id: str,
         *,
-        automatic_event_orderings: Optional[Tuple[int, int]],
+        automatic_event_orderings: Optional[EventOrderings],
     ) -> Optional[Union[int, AutomaticSubscriptionConflicted]]:
         """Updates a user's subscription settings for a specific thread root.
 
         If no change would be made to the subscription, does not produce any database change.
 
+        Case-by-case:
+            - if we already have an automatic subscription:
+                - new automatic subscriptions will be no-ops (no database write),
+                - new manual subscriptions will overwrite the automatic subscription
+            - if we already have a manual subscription:
+              we don't update (no database write) in either case, because:
+                - the existing manual subscription wins over a new automatic subscription request
+                - there would be no need to write a manual subscription because we already have one
+
         Args:
             user_id: The ID of the user whose settings are being updated.
             room_id: The ID of the room the thread root belongs to.
             thread_root_event_id: The event ID of the thread root.
-            automatic_event_id:
+            automatic_event_orderings:
                 Value depends on whether the subscription was performed automatically by the user's client.
                 For manual subscriptions: None.
-                For automatic subscriptions: (stream_ordering, topological_ordering) of the event.
+                For automatic subscriptions: the orderings of the event.
 
         Returns:
             If a subscription is made: (int) the stream ID for this update.
@@ -243,20 +262,19 @@ def _subscribe_user_to_thread_txn(
                 # is good enough (either we already have a manual subscription,
                 # or we requested an automatic subscription)
                 # In that case, nothing to change here.
+                # (See docstring for case-by-case explanation)
                 return None
 
             if not subscribed and requested_automatic:
                 assert automatic_event_orderings is not None
                 # we previously unsubscribed and we are now automatically subscribing
                 # Check whether the new autosubscription should be skipped
-                autosub_stream_ordering, autosub_topological_ordering = (
-                    automatic_event_orderings
-                )
                 if ThreadSubscriptionsWorkerStore._should_skip_autosubscription_after_unsubscription(
-                    autosub_stream_ordering,
-                    autosub_topological_ordering,
-                    unsubscribed_at_stream_ordering,
-                    unsubscribed_at_topological_ordering,
+                    autosub=automatic_event_orderings,
+                    unsubscribed_at=EventOrderings(
+                        unsubscribed_at_stream_ordering,
+                        unsubscribed_at_topological_ordering,
+                    ),
                 ):
                     # skip the subscription
                     return AutomaticSubscriptionConflicted()
@@ -473,7 +491,7 @@ def get_max_thread_subscriptions_stream_id(self) -> int:
         return self._thread_subscriptions_id_gen.get_current_token()
 
     async def get_updated_thread_subscriptions(
-        self, from_id: int, to_id: int, limit: int
+        self, *, from_id: int, to_id: int, limit: int
     ) -> List[Tuple[int, str, str, str]]:
         """Get updates to thread subscriptions between two stream IDs.
 
@@ -506,7 +524,7 @@ def get_updated_thread_subscriptions_txn(
         )
 
     async def get_updated_thread_subscriptions_for_user(
-        self, user_id: str, from_id: int, to_id: int, limit: int
+        self, user_id: str, *, from_id: int, to_id: int, limit: int
     ) -> List[Tuple[int, str, str]]:
         """Get updates to thread subscriptions for a specific user.
 

synapse/types/__init__.py

@@ -73,6 +73,7 @@
     from typing_extensions import Self
 
     from synapse.appservice.api import ApplicationService
+    from synapse.events import EventBase
     from synapse.storage.databases.main import DataStore, PurgeEventsStore
     from synapse.storage.databases.main.appservice import ApplicationServiceWorkerStore
     from synapse.storage.util.id_generators import MultiWriterIdGenerator
@@ -1464,3 +1465,31 @@ class ScheduledTask:
     result: Optional[JsonMapping]
     # Optional error that should be assigned a value when the status is FAILED
     error: Optional[str]
+
+
+@attr.s(auto_attribs=True, frozen=True, slots=True)
+class EventOrderings:
+    stream: int
+    """
+    The stream_ordering of the event.
+    Negative numbers mean the event was backfilled.
+    """
+
+    topological: int
+    """
+    The topological_ordering of the event.
+    Currently this is equivalent to the `depth` attributes of
+    the PDU.
+    """
+
+    @staticmethod
+    def from_event(event: "EventBase") -> "EventOrderings":
+        """
+        Get the orderings from an event.
+
+        Preconditions:
+        - the event must have been persisted (otherwise it won't have a stream ordering)
+        """
+        stream = event.internal_metadata.stream_ordering
+        assert stream is not None
+        return EventOrderings(stream, event.depth)

synapse/util/pydantic_models.py

@@ -72,8 +72,8 @@ def validate_event_id(cls, value: str) -> str:
             # Room version 3 is base64, 4+ are base64Url
             # In both cases, the base64 is unpadded.
             # refs:
-            # - https://spec.matrix.org/v1.15/rooms/v3/
-            # - https://spec.matrix.org/v1.15/rooms/v4/
+            # - https://spec.matrix.org/v1.15/rooms/v3/ e.g. $acR1l0raoZnm60CBwAVgqbZqoO/mYU81xysh1u7XcJk
+            # - https://spec.matrix.org/v1.15/rooms/v4/ e.g. $Rqnc-F-dvnEYJTyHq_iKxU2bZ1CI92-kuZq3a5lr5Zg
             b64_hash = value[1:]
             if cls.EVENT_ID_HASH_ROOM_VERSION_3_PLUS.fullmatch(b64_hash) is None:
                 raise ValueError(

tests/push/test_bulk_push_rule_evaluator.py

@@ -453,7 +453,7 @@ def test_suppress_edits(self) -> None:
     @override_config({"experimental_features": {"msc4306_enabled": True}})
     def test_thread_subscriptions(self) -> None:
         bulk_evaluator = BulkPushRuleEvaluator(self.hs)
-        (thread_root_id,) = self.helper.send_events(self.room_id, 1, tok=self.token)
+        (thread_root_id,) = self.helper.send_messages(self.room_id, 1, tok=self.token)
 
         self.assertFalse(
             self._create_and_process(
@@ -501,7 +501,7 @@ def test_with_disabled_thread_subscriptions(self) -> None:
         FUTURE: If MSC4306 becomes enabled-by-default/accepted, this test is to be removed.
         """
         bulk_evaluator = BulkPushRuleEvaluator(self.hs)
-        (thread_root_id,) = self.helper.send_events(self.room_id, 1, tok=self.token)
+        (thread_root_id,) = self.helper.send_messages(self.room_id, 1, tok=self.token)
 
         # When MSC4306 is not enabled, a threaded message generates a notification
         # by default.

tests/rest/client/test_thread_subscriptions.py

@@ -15,6 +15,7 @@
 
 from twisted.internet.testing import MemoryReactor
 
+from synapse.api.errors import Codes
 from synapse.rest import admin
 from synapse.rest.client import login, profile, room, thread_subscriptions
 from synapse.server import HomeServer
@@ -49,10 +50,12 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         # Create a room and send a message to use as a thread root
         self.room_id = self.helper.create_room_as(self.user_id, tok=self.token)
         self.helper.join(self.room_id, self.other_user_id, tok=self.other_token)
-        (self.root_event_id,) = self.helper.send_events(self.room_id, 1, tok=self.token)
+        (self.root_event_id,) = self.helper.send_messages(
+            self.room_id, 1, tok=self.token
+        )
 
         # Send a message in the thread
-        self.threaded_events = self.helper.send_events(
+        self.threaded_events = self.helper.send_messages(
             self.room_id,
             2,
             content_fn=lambda idx: {
@@ -265,15 +268,17 @@ def test_auto_subscribe_cause_event_not_in_thread(self) -> None:
         actually in the thread.
         This is an error.
         """
-        (unrelated_event_id,) = self.helper.send_events(self.room_id, 1, tok=self.token)
+        (unrelated_event_id,) = self.helper.send_messages(
+            self.room_id, 1, tok=self.token
+        )
         channel = self.make_request(
             "PUT",
             f"{PREFIX}/{self.room_id}/thread/{self.root_event_id}/subscription",
             {"automatic": unrelated_event_id},
             access_token=self.token,
         )
         self.assertEqual(channel.code, HTTPStatus.BAD_REQUEST, channel.text_body)
-        self.assertEqual(channel.json_body["errcode"], "M_NOT_IN_THREAD")
+        self.assertEqual(channel.json_body["errcode"], Codes.MSC4306_NOT_IN_THREAD)
 
     def test_auto_resubscription_conflict(self) -> None:
         """
@@ -297,7 +302,9 @@ def test_auto_resubscription_conflict(self) -> None:
             )
             self.assertEqual(channel.code, HTTPStatus.CONFLICT, channel.text_body)
             self.assertEqual(
-                channel.json_body["errcode"], "M_SKIPPED", channel.text_body
+                channel.json_body["errcode"],
+                Codes.MSC4306_CONFLICTING_UNSUBSCRIPTION,
+                channel.text_body,
             )
 
             # Check the subscription was not made
@@ -310,7 +317,7 @@ def test_auto_resubscription_conflict(self) -> None:
 
         # But if a new event is sent after the unsubscription took place,
         # that one can be used for an automatic subscription
-        (later_event_id,) = self.helper.send_events(
+        (later_event_id,) = self.helper.send_messages(
             self.room_id,
             1,
             content_fn=lambda _: {

tests/rest/client/utils.py

@@ -47,7 +47,7 @@
 from twisted.internet.testing import MemoryReactorClock
 from twisted.web.server import Site
 
-from synapse.api.constants import Membership, ReceiptTypes
+from synapse.api.constants import EventTypes, Membership, ReceiptTypes
 from synapse.api.errors import Codes
 from synapse.server import HomeServer
 from synapse.types import JsonDict
@@ -396,7 +396,7 @@ def send(
             custom_headers=custom_headers,
         )
 
-    def send_events(
+    def send_messages(
         self,
         room_id: str,
         num_events: int,
@@ -414,7 +414,7 @@ def send_events(
         for event_index in range(num_events):
             response = self.send_event(
                 room_id,
-                "m.room.message",
+                EventTypes.Message,
                 content_fn(event_index),
                 tok=tok,
             )