Add EventOrderings dataclass

Author: reivilibre

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
@@ -114,13 +114,7 @@ async def subscribe_user_to_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(event)
         else:
             automatic_event_orderings = None
 

synapse/storage/databases/main/thread_subscriptions.py

@@ -32,6 +32,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:
@@ -110,10 +111,9 @@ 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 *after* an unsubscription
@@ -123,28 +123,27 @@ def _should_skip_autosubscription_after_unsubscription(
         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
+        assert unsubscribed_at.stream > 0
+        assert unsubscribed_at.topological > 0
 
-        if unsubscribed_at_stream_ordering >= autosub_stream_ordering > 0:
+        if 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,7 +154,7 @@ 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.
 
@@ -165,10 +164,10 @@ async def subscribe_user_to_thread(
             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.
@@ -242,14 +241,12 @@ def _subscribe_user_to_thread_txn(
                 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()

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)

tests/storage/test_thread_subscriptions.py

@@ -12,7 +12,7 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 
-from typing import Optional, Tuple, Union
+from typing import Optional, Union
 
 from twisted.internet.testing import MemoryReactor
 
@@ -23,6 +23,7 @@
     ThreadSubscriptionsWorkerStore,
 )
 from synapse.storage.engines.sqlite import Sqlite3Engine
+from synapse.types import EventOrderings
 from synapse.util import Clock
 
 from tests import unittest
@@ -101,7 +102,7 @@ def _subscribe(
         self,
         thread_root_id: str,
         *,
-        automatic: Optional[Tuple[int, int]],
+        automatic: Optional[EventOrderings],
         room_id: Optional[str] = None,
         user_id: Optional[str] = None,
     ) -> Optional[Union[int, AutomaticSubscriptionConflicted]]:
@@ -153,7 +154,7 @@ def test_set_and_get_thread_subscription(self) -> None:
         # Subscribe
         self._subscribe(
             self.thread_root_id,
-            automatic=(1, 1),
+            automatic=EventOrderings(1, 1),
         )
 
         # Assert subscription went through
@@ -185,7 +186,7 @@ def test_set_and_get_thread_subscription(self) -> None:
     def test_purge_thread_subscriptions_for_user(self) -> None:
         """Test purging all thread subscription settings for a user."""
         # Set subscription settings for multiple threads
-        self._subscribe(self.thread_root_id, automatic=(1, 1))
+        self._subscribe(self.thread_root_id, automatic=EventOrderings(1, 1))
         self._subscribe(self.other_thread_root_id, automatic=None)
 
         subscriptions = self.get_success(
@@ -224,8 +225,12 @@ def test_purge_thread_subscriptions_for_user(self) -> None:
     def test_get_updated_thread_subscriptions(self) -> None:
         """Test getting updated thread subscriptions since a stream ID."""
 
-        stream_id1 = self._subscribe(self.thread_root_id, automatic=(1, 1))
-        stream_id2 = self._subscribe(self.other_thread_root_id, automatic=(2, 2))
+        stream_id1 = self._subscribe(
+            self.thread_root_id, automatic=EventOrderings(1, 1)
+        )
+        stream_id2 = self._subscribe(
+            self.other_thread_root_id, automatic=EventOrderings(2, 2)
+        )
         assert stream_id1 is not None and not isinstance(
             stream_id1, AutomaticSubscriptionConflicted
         )
@@ -253,15 +258,17 @@ def test_get_updated_thread_subscriptions_for_user(self) -> None:
         other_user_id = "@other_user:test"
 
         # Set thread subscription for main user
-        stream_id1 = self._subscribe(self.thread_root_id, automatic=(1, 1))
+        stream_id1 = self._subscribe(
+            self.thread_root_id, automatic=EventOrderings(1, 1)
+        )
         assert stream_id1 is not None and not isinstance(
             stream_id1, AutomaticSubscriptionConflicted
         )
 
         # Set thread subscription for other user
         stream_id2 = self._subscribe(
             self.other_thread_root_id,
-            automatic=(1, 1),
+            automatic=EventOrderings(1, 1),
             user_id=other_user_id,
         )
         assert stream_id2 is not None and not isinstance(
@@ -299,13 +306,21 @@ def test_should_skip_autosubscription_after_unsubscription(self) -> None:
         # unsubscribe maximums: stream order, then tological order
 
         # both orderings agree that the unsub is after the cause event
-        self.assertTrue(func(1, 1, 2, 2))
+        self.assertTrue(
+            func(autosub=EventOrderings(1, 1), unsubscribed_at=EventOrderings(2, 2))
+        )
 
         # topological ordering is inconsistent with stream ordering,
         # in that case favour stream ordering because it's what /sync uses
-        self.assertTrue(func(1, 2, 2, 1))
+        self.assertTrue(
+            func(autosub=EventOrderings(1, 2), unsubscribed_at=EventOrderings(2, 1))
+        )
 
         # the automatic subscription is caused by a backfilled event here
         # unfortunately we must fall back to topological ordering here
-        self.assertTrue(func(-50, 2, 2, 3))
-        self.assertFalse(func(-50, 2, 2, 1))
+        self.assertTrue(
+            func(autosub=EventOrderings(-50, 2), unsubscribed_at=EventOrderings(2, 3))
+        )
+        self.assertFalse(
+            func(autosub=EventOrderings(-50, 2), unsubscribed_at=EventOrderings(2, 1))
+        )