Persist sticky events in sticky_events table

This only works on postgres for now

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
@@ -360,3 +360,12 @@ 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 = "msc4354_stick_duration_ms"
+    FIELD_NAME: Final = "msc4354_sticky"

synapse/events/builder.py

@@ -24,7 +24,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
 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,7 @@ class EventBuilder:
 
     content: JsonDict = attr.Factory(dict)
     unsigned: JsonDict = attr.Factory(dict)
+    sticky: Optional[StickyEventField] = None
 
     # 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 +270,9 @@ class EventBuilder:
         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 +322,7 @@ class EventBuilderFactory:
             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/rest/client/room.py

@@ -33,7 +33,7 @@ from prometheus_client.core import Histogram
 from twisted.web.server import Request
 
 from synapse import event_auth
-from synapse.api.constants import Direction, EventTypes, Membership
+from synapse.api.constants import Direction, EventTypes, Membership, StickyEvent
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -82,9 +82,6 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-MSC4354_STICKY_DURATION_QUERY_PARAM = "msc4354_stick_duration_ms"
-MSC4354_STICKY_EVENT_KEY = "msc4354_sticky"
-
 
 class _RoomSize(Enum):
     """
@@ -370,10 +367,10 @@ class RoomStateEventRestServlet(RestServlet):
                 }
                 if self.msc4354_enabled:
                     sticky_duration_ms = parse_integer(
-                        request, MSC4354_STICKY_DURATION_QUERY_PARAM
+                        request, StickyEvent.QUERY_PARAM_NAME
                     )
                     if sticky_duration_ms is not None:
-                        event_dict[MSC4354_STICKY_EVENT_KEY] = {
+                        event_dict[StickyEvent.FIELD_NAME] = {
                             "duration_ms": sticky_duration_ms,
                         }
 
@@ -456,11 +453,9 @@ class RoomSendEventRestServlet(TransactionRestServlet):
             event_dict["origin_server_ts"] = origin_server_ts
 
         if self.msc4354_enabled:
-            sticky_duration_ms = parse_integer(
-                request, MSC4354_STICKY_DURATION_QUERY_PARAM
-            )
+            sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME)
             if sticky_duration_ms is not None:
-                event_dict[MSC4354_STICKY_EVENT_KEY] = {
+                event_dict[StickyEvent.FIELD_NAME] = {
                     "duration_ms": sticky_duration_ms,
                 }
 

synapse/storage/databases/main/events.py

@@ -49,6 +49,7 @@ from synapse.api.constants import (
     EventTypes,
     Membership,
     RelationTypes,
+    StickyEvent,
 )
 from synapse.api.errors import PartialStateConflictError
 from synapse.api.room_versions import RoomVersions
@@ -2982,7 +2983,7 @@ class PersistEventsStore:
             if ev.rejected_reason is not None:
                 continue
             # MSC: The presence of sticky.duration_ms with a valid value makes the event “sticky”
-            sticky_obj = ev.get("sticky", None)
+            sticky_obj = ev.get_dict().get(StickyEvent.FIELD_NAME, None)
             if type(sticky_obj) is dict:
                 sticky_duration_ms = sticky_obj.get("duration_ms", None)
                 # MSC: Valid values are the integer range 0-3600000 (1 hour).
@@ -2993,8 +2994,15 @@ class PersistEventsStore:
                 ):
                     sticky_events.append(ev)
 
+        # TODO: filter out already expired sticky events.
+
         if len(sticky_events) == 0:
             return
+        logger.info(
+            "inserting %d sticky events in room %s",
+            len(sticky_events),
+            sticky_events[0].room_id,
+        )
         now_ms = round(time.time() * 1000)
         self.db_pool.simple_insert_many_txn(
             txn,
@@ -3009,8 +3017,10 @@ class PersistEventsStore:
                     # This ensures that malicious origin timestamps cannot specify start times in the future.
                     # Calculate the end time as start_time + min(sticky.duration_ms, 3600000).
                     min(ev.origin_server_ts, now_ms)
-                    + min(ev.get_dict()["sticky"]["duration_ms"], 3600000),
-                    ev.internal_metadata.soft_failed,
+                    + min(
+                        ev.get_dict()[StickyEvent.FIELD_NAME]["duration_ms"], 3600000
+                    ),
+                    ev.internal_metadata.is_soft_failed(),
                 )
                 for ev in sticky_events
             ],

synapse/storage/schema/__init__.py

@@ -19,7 +19,7 @@
 #
 #
 
-SCHEMA_VERSION = 92  # remember to update the list below when updating
+SCHEMA_VERSION = 93  # remember to update the list below when updating
 """Represents the expectations made by the codebase about the database schema
 
 This should be incremented whenever the codebase changes its requirements on the

synapse/storage/schema/main/delta/93/01_sticky_events.sql

@@ -0,0 +1,26 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2025 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+CREATE SEQUENCE IF NOT EXISTS sticky_events_seq;
+
+CREATE TABLE IF NOT EXISTS sticky_events(
+  id BIGINT PRIMARY KEY DEFAULT nextval('sticky_events_seq'),
+  room_id TEXT NOT NULL,
+  event_id TEXT NOT NULL,
+  sender TEXT NOT NULL,
+  expires_at BIGINT NOT NULL,
+  soft_failed BOOLEAN NOT NULL
+);
+
+-- for pulling out soft failed events by room
+CREATE INDEX IF NOT EXISTS sticky_events_room_idx ON sticky_events(room_id, soft_failed);