Support sending and receiving MSC4354 Sticky Event metadata. (#19365)

Part of: MSC4354 whose experimental feature tracking issue is
#19409

Follows: #19340 (a necessary bugfix for `/event/` to set this metadata)

Partially supersedes: #18968

This PR implements the first batch of work to support MSC4354 Sticky
Events.

Sticky events are events that have been configured with a finite
'stickiness' duration,
capped to 1 hour per current MSC draft.

Whilst an event is sticky, we provide stronger delivery guarantees for
the event, both to
our clients and to remote homeservers, essentially making it reliable
delivery as long as we
have a functional connection to the client/server and until the
stickiness expires.

This PR merely supports creating sticky events and receiving the sticky
TTL metadata in clients.
It is not suitable for trialling sticky events since none of the other
semantics are implemented.

Contains a temporary SQLite workaround due to a bug in our supported
version enforcement: #19452

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
Co-authored-by: Eric Eastwood <erice@element.io>

Author: reivilibre

changelog.d/19365.feature

@@ -0,0 +1 @@
+Support sending and receiving [MSC4354 Sticky Event](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) metadata.

docker/complement/conf/workers-shared-extra.yaml.j2

@@ -139,6 +139,8 @@ experimental_features:
   msc4155_enabled: true
   # Thread Subscriptions
   msc4306_enabled: true
+  # Sticky Events
+  msc4354_enabled: true
 
 server_notices:
   system_mxid_localpart: _server

synapse/api/constants.py

@@ -24,7 +24,9 @@
 """Contains constants from the specification."""
 
 import enum
-from typing import Final
+from typing import Final, TypedDict
+
+from synapse.util.duration import Duration
 
 # the max size of a (canonical-json-encoded) event
 MAX_PDU_SIZE = 65536
@@ -292,6 +294,8 @@ class EventUnsignedContentFields:
     # Requesting user's membership, per MSC4115
     MEMBERSHIP: Final = "membership"
 
+    STICKY_TTL: Final = "msc4354_sticky_duration_ttl_ms"
+
 
 class MTextFields:
     """Fields found inside m.text content blocks."""
@@ -377,3 +381,40 @@ class Direction(enum.Enum):
 class ProfileFields:
     DISPLAYNAME: Final = "displayname"
     AVATAR_URL: Final = "avatar_url"
+
+
+class StickyEventField(TypedDict):
+    """
+    Dict content of the `sticky` part of an event.
+    """
+
+    duration_ms: int
+
+
+class StickyEvent:
+    QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms"
+    """
+    Query parameter used by clients for setting the sticky duration of an event they are sending.
+
+    Applies to:
+        - /rooms/.../send/...
+        - /rooms/.../state/...
+    """
+
+    EVENT_FIELD_NAME: Final = "msc4354_sticky"
+    """
+    Name of the field in the top-level event dict that contains the sticky event dict.
+    """
+
+    MAX_DURATION: Duration = Duration(hours=1)
+    """
+    Maximum stickiness duration as specified in MSC4354.
+    Ensures that data in the /sync response can go down and not grow unbounded.
+    """
+
+    MAX_EVENTS_IN_SYNC: Final = 100
+    """
+    Maximum number of sticky events to include in /sync.
+
+    This is the default specified in the MSC. Chosen arbitrarily.
+    """

synapse/app/generic_worker.py

@@ -102,6 +102,7 @@
 from synapse.storage.databases.main.sliding_sync import SlidingSyncStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.databases.main.stats import StatsStore
+from synapse.storage.databases.main.sticky_events import StickyEventsWorkerStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
 from synapse.storage.databases.main.tags import TagsWorkerStore
 from synapse.storage.databases.main.task_scheduler import TaskSchedulerWorkerStore
@@ -137,6 +138,7 @@ class GenericWorkerStore(
     RoomWorkerStore,
     DirectoryWorkerStore,
     ThreadSubscriptionsWorkerStore,
+    StickyEventsWorkerStore,
     PushRulesWorkerStore,
     ApplicationServiceTransactionWorkerStore,
     ApplicationServiceWorkerStore,

synapse/config/experimental.py

@@ -580,5 +580,11 @@ def read_config(
         # (and MSC4308: Thread Subscriptions extension to Sliding Sync)
         self.msc4306_enabled: bool = experimental.get("msc4306_enabled", False)
 
+        # MSC4354: Sticky Events
+        # Tracked in: https://github.com/element-hq/synapse/issues/19409
+        # Note that sticky events persisted before this feature is enabled will not be
+        # considered sticky by the local homeserver.
+        self.msc4354_enabled: bool = experimental.get("msc4354_enabled", False)
+
         # MSC4380: Invite blocking
         self.msc4380_enabled: bool = experimental.get("msc4380_enabled", False)

synapse/config/workers.py

@@ -127,7 +127,9 @@ class WriterLocations:
     """Specifies the instances that write various streams.
 
     Attributes:
-        events: The instances that write to the event and backfill streams.
+        events: The instances that write to the event, backfill and `sticky_events` streams.
+            (`sticky_events` is written to during event persistence so must be handled by the
+            same stream writers.)
         typing: The instances that write to the typing stream. Currently
             can only be a single instance.
         to_device: The instances that write to the to_device stream. Currently

synapse/events/__init__.py

@@ -36,14 +36,20 @@
 import attr
 from unpaddedbase64 import encode_base64
 
-from synapse.api.constants import EventContentFields, EventTypes, RelationTypes
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    RelationTypes,
+    StickyEvent,
+)
 from synapse.api.room_versions import EventFormatVersions, RoomVersion, RoomVersions
 from synapse.synapse_rust.events import EventInternalMetadata
 from synapse.types import (
     JsonDict,
     StrCollection,
 )
 from synapse.util.caches import intern_dict
+from synapse.util.duration import Duration
 from synapse.util.frozenutils import freeze
 
 if TYPE_CHECKING:
@@ -318,6 +324,28 @@ def freeze(self) -> None:
         # this will be a no-op if the event dict is already frozen.
         self._dict = freeze(self._dict)
 
+    def sticky_duration(self) -> Duration | None:
+        """
+        Returns the effective sticky duration of this event, or None
+        if the event does not have a sticky duration.
+        (Sticky Events are a MSC4354 feature.)
+
+        Clamps the sticky duration to the maximum allowed duration.
+        """
+        sticky_obj = self.get_dict().get(StickyEvent.EVENT_FIELD_NAME, None)
+        if type(sticky_obj) is not dict:
+            return None
+        sticky_duration_ms = sticky_obj.get("duration_ms", None)
+        # MSC: Clamp to 0 and MAX_DURATION (1 hour)
+        # We use `type(...) is int` to avoid accepting bools as `isinstance(True, int)`
+        # (bool is a subclass of int)
+        if type(sticky_duration_ms) is int and sticky_duration_ms >= 0:
+            return min(
+                Duration(milliseconds=sticky_duration_ms),
+                StickyEvent.MAX_DURATION,
+            )
+        return None
+
     def __str__(self) -> str:
         return self.__repr__()
 

synapse/events/builder.py

@@ -24,7 +24,7 @@
 import attr
 from signedjson.types import SigningKey
 
-from synapse.api.constants import MAX_DEPTH, EventTypes
+from synapse.api.constants import MAX_DEPTH, EventTypes, StickyEvent, StickyEventField
 from synapse.api.room_versions import (
     KNOWN_EVENT_FORMAT_VERSIONS,
     EventFormatVersions,
@@ -89,6 +89,10 @@ class EventBuilder:
 
     content: JsonDict = attr.Factory(dict)
     unsigned: JsonDict = attr.Factory(dict)
+    sticky: StickyEventField | None = None
+    """
+    Fields for MSC4354: Sticky Events
+    """
 
     # These only exist on a subset of events, so they raise AttributeError if
     # someone tries to get them when they don't exist.
@@ -269,6 +273,9 @@ async def build(
         if self._origin_server_ts is not None:
             event_dict["origin_server_ts"] = self._origin_server_ts
 
+        if self.sticky is not None:
+            event_dict[StickyEvent.EVENT_FIELD_NAME] = self.sticky
+
         return create_local_event_from_event_dict(
             clock=self._clock,
             hostname=self._hostname,
@@ -318,6 +325,7 @@ def for_room_version(
             unsigned=key_values.get("unsigned", {}),
             redacts=key_values.get("redacts", None),
             origin_server_ts=key_values.get("origin_server_ts", None),
+            sticky=key_values.get(StickyEvent.EVENT_FIELD_NAME, None),
         )
 
 

synapse/handlers/delayed_events.py

@@ -17,7 +17,7 @@
 
 from twisted.internet.interfaces import IDelayedCall
 
-from synapse.api.constants import EventTypes
+from synapse.api.constants import EventTypes, StickyEvent, StickyEventField
 from synapse.api.errors import ShadowBanError, SynapseError
 from synapse.api.ratelimiting import Ratelimiter
 from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
@@ -333,6 +333,7 @@ async def add(
         origin_server_ts: int | None,
         content: JsonDict,
         delay: int,
+        sticky_duration_ms: int | None,
     ) -> str:
         """
         Creates a new delayed event and schedules its delivery.
@@ -346,7 +347,9 @@ async def add(
                 If None, the timestamp will be the actual time when the event is sent.
             content: The content of the event to be sent.
             delay: How long (in milliseconds) to wait before automatically sending the event.
-
+            sticky_duration_ms: If an MSC4354 sticky event: the sticky duration (in milliseconds).
+                The event will be attempted to be reliably delivered to clients and remote servers
+                during its sticky period.
         Returns: The ID of the added delayed event.
 
         Raises:
@@ -382,6 +385,7 @@ async def add(
             origin_server_ts=origin_server_ts,
             content=content,
             delay=delay,
+            sticky_duration_ms=sticky_duration_ms,
         )
 
         if self._repl_client is not None:
@@ -570,7 +574,10 @@ async def _send_event(
 
                 if event.state_key is not None:
                     event_dict["state_key"] = event.state_key
-
+                if event.sticky_duration_ms is not None:
+                    event_dict[StickyEvent.EVENT_FIELD_NAME] = StickyEventField(
+                        duration_ms=event.sticky_duration_ms
+                    )
                 (
                     sent_event,
                     _,

synapse/notifier.py

@@ -526,6 +526,7 @@ def on_new_event(
             StreamKeyType.TYPING,
             StreamKeyType.UN_PARTIAL_STATED_ROOMS,
             StreamKeyType.THREAD_SUBSCRIPTIONS,
+            StreamKeyType.STICKY_EVENTS,
         ],
         new_token: int,
         users: Collection[str | UserID] | None = None,