MSC4354: Sticky Events

changelog.d/18968.feature

@@ -0,0 +1 @@
+Implement support for MSC4354: Sticky Events.

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

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

scripts-dev/complement.sh

@@ -229,6 +229,7 @@ main() {
     ./tests/msc4140
     ./tests/msc4155
     ./tests/msc4306
+    ./tests/msc4354
   )
 
   # Export the list of test packages as a space-separated environment variable, so other

synapse/_scripts/synapse_port_db.py

@@ -132,6 +132,7 @@
         "has_known_state",
         "is_encrypted",
     ],
+    "sticky_events": ["soft_failed"],
     "thread_subscriptions": ["subscribed", "automatic"],
     "users": ["shadow_banned", "approved", "locked", "suspended"],
     "un_partial_stated_event_stream": ["rejection_status_changed"],

synapse/api/constants.py

@@ -24,7 +24,7 @@
 """Contains constants from the specification."""
 
 import enum
-from typing import Final
+from typing import Final, TypedDict
 
 # the max size of a (canonical-json-encoded) event
 MAX_PDU_SIZE = 65536
@@ -279,6 +279,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."""
@@ -364,3 +366,18 @@ class Direction(enum.Enum):
 class ProfileFields:
     DISPLAYNAME: Final = "displayname"
     AVATAR_URL: Final = "avatar_url"
+
+
+class StickyEventField(TypedDict):
+    duration_ms: int
+
+
+class StickyEvent:
+    QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms"
+    FIELD_NAME: Final = "msc4354_sticky"
+    MAX_DURATION_MS: Final = 3600000  # 1 hour
+    """
+    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

synapse/app/generic_worker.py

@@ -101,6 +101,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
@@ -136,6 +137,7 @@ class GenericWorkerStore(
     RoomWorkerStore,
     DirectoryWorkerStore,
     ThreadSubscriptionsWorkerStore,
+    StickyEventsWorkerStore,
     PushRulesWorkerStore,
     ApplicationServiceTransactionWorkerStore,
     ApplicationServiceWorkerStore,

synapse/config/experimental.py

@@ -597,5 +597,8 @@ def read_config(
         # (and MSC4308: Thread Subscriptions extension to Sliding Sync)
         self.msc4306_enabled: bool = experimental.get("msc4306_enabled", False)
 
+        # MSC4354: Sticky Events
+        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,7 @@ 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.
         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/event_auth.py

@@ -66,6 +66,7 @@
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.types import (
     MutableStateMap,
+    StateKey,
     StateMap,
     StrCollection,
     UserID,
@@ -1200,8 +1201,8 @@ def get_public_keys(invite_event: "EventBase") -> list[dict[str, Any]]:
 
 def auth_types_for_event(
     room_version: RoomVersion, event: Union["EventBase", "EventBuilder"]
-) -> set[tuple[str, str]]:
-    """Given an event, return a list of (EventType, StateKey) that may be
+) -> set[StateKey]:
+    """Given an event, return a list of (state event type, state key) that may be
     needed to auth the event. The returned list may be a superset of what
     would actually be required depending on the full state of the room.
 

synapse/events/__init__.py

@@ -36,7 +36,12 @@
 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 (
@@ -318,6 +323,23 @@ 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) -> int | 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.FIELD_NAME, None)
+        if type(sticky_obj) is not dict:
+            return None
+        sticky_duration_ms = sticky_obj.get("duration_ms", None)
+        # MSC: Valid values are the integer range 0-MAX_DURATION_MS
+        if type(sticky_duration_ms) is int and sticky_duration_ms >= 0:
+            return min(sticky_duration_ms, StickyEvent.MAX_DURATION_MS)
+        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.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.FIELD_NAME, None),
         )
 
 

synapse/federation/federation_base.py

@@ -194,6 +194,8 @@ async def _check_sigs_and_hash(
             # using the event in prev_events).
             redacted_event = prune_event(pdu)
             redacted_event.internal_metadata.soft_failed = True
+            # Mark this as spam so we don't re-evaluate soft-failure status.
+            redacted_event.internal_metadata.policy_server_spammy = True
             return redacted_event
 
         return pdu

synapse/federation/send_queue.py

@@ -212,6 +212,11 @@ def notify_new_events(self, max_token: RoomStreamToken) -> None:
         # This should never get called.
         raise NotImplementedError()
 
+    def notify_new_server_joined(self, server: str, room_id: str) -> None:
+        """As per FederationSender"""
+        # This should never get called.
+        raise NotImplementedError()
+
     def build_and_send_edu(
         self,
         destination: str,

synapse/federation/sender/__init__.py

@@ -177,6 +177,7 @@
 from synapse.util.duration import Duration
 from synapse.util.metrics import Measure
 from synapse.util.retryutils import filter_destinations_by_retry_limiter
+from synapse.visibility import filter_events_for_server
 
 if TYPE_CHECKING:
     from synapse.events.presence_router import PresenceRouter
@@ -240,6 +241,13 @@ def notify_new_events(self, max_token: RoomStreamToken) -> None:
         """
         raise NotImplementedError()
 
+    @abc.abstractmethod
+    def notify_new_server_joined(self, server: str, room_id: str) -> None:
+        """This gets called when we a new server has joined a room. We might
+        want to send out some events to this server.
+        """
+        raise NotImplementedError()
+
     @abc.abstractmethod
     async def send_read_receipt(self, receipt: ReadReceipt) -> None:
         """Send a RR to any other servers in the room
@@ -502,6 +510,66 @@ def _get_per_destination_queue(
             self._per_destination_queues[destination] = queue
         return queue
 
+    def notify_new_server_joined(self, server: str, room_id: str) -> None:
+        # We currently only use this notification for MSC4354: Sticky Events.
+        if not self.hs.config.experimental.msc4354_enabled:
+            return
+        # fire off a processing loop in the background
+        self.hs.run_as_background_process(
+            "process_new_server_joined_over_federation",
+            self._process_new_server_joined_over_federation,
+            server,
+            room_id,
+        )
+
+    async def _process_new_server_joined_over_federation(
+        self, new_server: str, room_id: str
+    ) -> None:
+        sticky_event_ids = await self.store.get_sticky_event_ids_sent_by_self(
+            room_id,
+            from_stream_pos=0,
+        )
+        sticky_events = await self.store.get_events_as_list(sticky_event_ids)
+
+        # We must not send events that are outliers / lack a stream ordering, else we won't be able to
+        # satisfy /get_missing_events requests
+        sticky_events = [
+            ev
+            for ev in sticky_events
+            if ev.internal_metadata.stream_ordering is not None
+            and not ev.internal_metadata.is_outlier()
+        ]
+        # order by stream ordering so we present things in the right timeline order on the receiver
+        sticky_events.sort(
+            key=lambda ev: ev.internal_metadata.stream_ordering
+            or 0,  # not possible to be 0
+        )
+
+        sticky_events = await filter_events_for_server(
+            self._storage_controllers,
+            new_server,
+            self.server_name,
+            sticky_events,
+            redact=False,
+            filter_out_erased_senders=True,
+            filter_out_remote_partial_state_events=True,
+        )
+
+        if not sticky_events:
+            return
+
+        logger.info(
+            "sending %d sticky events to newly joined server %s in room %s",
+            len(sticky_events),
+            new_server,
+            room_id,
+        )
+        # we don't track that we sent up to this stream position since it won't make any difference
+        # since notify_new_server_joined is only called initially.
+        await self._transaction_manager.send_new_transaction(
+            new_server, sticky_events, []
+        )
+
     def notify_new_events(self, max_token: RoomStreamToken) -> None:
         """This gets called when we have some new events we might want to
         send out to other servers.

synapse/federation/sender/per_destination_queue.py

@@ -105,6 +105,7 @@ def __init__(
         self._instance_name = hs.get_instance_name()
         self._federation_shard_config = hs.config.worker.federation_shard_config
         self._state = hs.get_state_handler()
+        self._msc4354_enabled = hs.config.experimental.msc4354_enabled
 
         self._should_send_on_this_instance = True
         if not self._federation_shard_config.should_handle(
@@ -583,6 +584,33 @@ async def _catch_up_transmission_loop(self) -> None:
                     # send.
                     extrem_events = await self._store.get_events_as_list(extrems)
 
+                    if self._msc4354_enabled:
+                        # we also want to send sticky events that are still active in this room
+                        sticky_event_ids = (
+                            await self._store.get_sticky_event_ids_sent_by_self(
+                                pdu.room_id,
+                                last_successful_stream_ordering,
+                            )
+                        )
+                        # skip any that are actually the forward extremities we want to send anyway
+                        sticky_events = await self._store.get_events_as_list(
+                            [
+                                event_id
+                                for event_id in sticky_event_ids
+                                if event_id not in extrems
+                            ]
+                        )
+                        if sticky_events:
+                            # *prepend* these to the extrem list, so they are processed first.
+                            # This ensures they will show up before the forward extrem in stream order
+                            extrem_events = sticky_events + extrem_events
+                            logger.info(
+                                "Sending %d missed sticky events to %s: %r",
+                                len(sticky_events),
+                                self._destination,
+                                pdu.room_id,
+                            )
+
                     new_pdus = []
                     for p in extrem_events:
                         # We pulled this from the DB, so it'll be non-null

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
 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.FIELD_NAME] = {
+                        "duration_ms": event.sticky_duration_ms,
+                    }
                 (
                     sent_event,
                     _,