Use sharded tokens in MSC4354: Sticky Events sync and sliding sync, preventing events appearing in multiple sync responses. (#20186)

Fixes: #19661
Part of: MSC4354

First introduced some helpers for querying a sharded stream, which were moved to
https://github.com/element-hq/synapse/pull/20108.

- SQL builder for 'give me rows between two sharded tokens'
- a function to return the 'new' from_token after a limited read


Then make the sticky events portions of both oldschool and sliding sync use
sharded tokens.

This prevents a 'torn sync' problem where, because the events and sticky
events streams are both sharded with the same writers:

1. persisting a sticky event leads to an event row and a `sticky_events`
row being created for event `e` (on writer `W`).
2. Suppose another writer is still persisting, so the position of the
stream for writer W advances, but the global
['linear'](https://element-hq.github.io/synapse/latest/development/synapse_architecture/streams.html?highlight=%E2%80%9Clinear%E2%80%9D#current-stream-id)
(I have been calling this 'baseline') position does not advance yet.
3. sync is sharding-aware for the regular events stream, so `e` appears
as a real-time timeline event, thus coming down sync
4. since sticky events were not sharding-aware, `e` is not detected as a
sticky event in this sync response. The sticky event position doesn't
advance in the sync token, either.
5. (assume the position advances because other writers have finished
writing and so the 'linear' position advances for the sticky events
stream)
6. the client then syncs again, using the new sync token. This time, `e`
appears in the sticky event section.

If we make sticky events sharding-aware in sync, then `e` would appear
for consideration in both the timeline section **and** the sticky event
section in the first sync.
It then gets deduplicated out of the sticky section since it is
appearing in the timeline section in the same sync.

---------

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

changelog.d/20186.bugfix

@@ -0,0 +1 @@
+Use sharded tokens in [MSC4354: Sticky Events] sync and sliding sync, preventing events appearing in multiple sync responses.

synapse/api/errors.py

@@ -1004,9 +1004,9 @@ class SlidingSyncUnknownPosition(SynapseError):
     param).
     """
 
-    def __init__(self) -> None:
+    def __init__(self, message: str = "Unknown position") -> None:
         super().__init__(
             HTTPStatus.BAD_REQUEST,
-            msg="Unknown position",
+            msg=message,
             errcode=Codes.UNKNOWN_POS,
         )

synapse/handlers/sliding_sync/extensions.py

@@ -32,6 +32,7 @@ from synapse.api.constants import (
     ProfileUpdateAction,
     StickyEvent,
 )
+from synapse.api.errors import SlidingSyncUnknownPosition
 from synapse.events.utils import FilteredEvent
 from synapse.handlers.receipts import ReceiptEventSource
 from synapse.logging.opentracing import trace
@@ -1035,13 +1036,57 @@ class SlidingSyncExtensionHandler:
         # If there is no `since` token specified, start from the beginning of the stream
         # to make sure the client receives all visible (unexpired) sticky events
         since_token = sticky_events_request.since or SlidingSyncStickyEventsToken.START
+
+        since_token_as_stream_token = await since_token.to_stream_token(self.store)
+
+        if not since_token_as_stream_token.is_before_or_eq(
+            self.event_sources.get_current_token().sticky_events_key
+        ):
+            # The `since_token` is *after* the current position as seen on this worker.
+            # This either means that this worker is lagging, or the client has a token
+            # from the future. The client having a future token could happen maliciously
+            # where someone intentionally messes with the token or innocently if a
+            # database was rolled back by restoring from backup.
+            #
+            # Get the max allocated token out of the database to see which case it is.
+            # (For efficiency, we only do this after the non-database common case check.)
+            max_token = await self.store.get_sticky_events_stream_id_generator().get_max_allocated_token()
+
+            # It may be tempting to wonder why we don't do anything about the 'worker lagging'
+            # case, such as waiting for streams to catch up.
+            # The case where this worker is lagging should not happen in practice because:
+            # assuming the client didn't mess with the `since` token, it must have been
+            # produced as a `to_token` in a previous request, where the response-wide `next_pos`
+            # is equal to or later than the token given by this extension.
+            #
+            # Then in this request, the client will present that `next_pos` as the `pos` for the
+            # entire request and `wait_for_stream_token` (which is called with the request-level pos tokens)
+            # will cause us to wait for worker lag on that token, including the sticky events stream.
+            #
+            # (It also doesn't actually matter since `get_sticky_events_in_rooms` will just return
+            # an empty result and is safe against rewinding the token.)
+
+            if max_token < since_token_as_stream_token.get_max_stream_pos():
+                # The client has a token from the future.
+                #
+                # We could wait until we reach the token but we might as well not waste
+                # our resources on an invalid state scenario. Reset the sliding sync
+                # connection.
+                raise SlidingSyncUnknownPosition(
+                    "The `org.matrix.msc4354.sticky_events` extension `since` token is considered "
+                    "invalid because it includes stream positions greater than the furthest "
+                    "persisted position across all of the workers. This indicates either a Synapse "
+                    "programming error (as we should never hand out invalid future tokens), database "
+                    "was rolled back, or a fabricated `from` token. If you've modified the token, "
+                    "you can try paginating from the beginning again.",
+                )
         (
-            sticky_events_to_id,
+            sticky_events_to_token,
             room_to_event_ids,
         ) = await self.store.get_sticky_events_in_rooms(
             all_interested_room_ids,
-            from_id=since_token.sticky_events_stream_id,
-            to_id=to_token.sticky_events_key,
+            from_token=since_token_as_stream_token,
+            to_token=to_token.sticky_events_key,
             now=now,
             limit=min(sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC),
         )
@@ -1078,8 +1123,8 @@ class SlidingSyncExtensionHandler:
 
         return SlidingSyncResult.Extensions.StickyEventsExtension(
             room_id_to_sticky_events=room_id_to_sticky_events,
-            next_batch=SlidingSyncStickyEventsToken(
-                sticky_events_stream_id=sticky_events_to_id
+            next_batch=await SlidingSyncStickyEventsToken.from_stream_token(
+                self.store, sticky_events_to_token
             ),
         )
 

synapse/handlers/sync.py

@@ -779,18 +779,24 @@ class SyncHandler:
         with Measure(
             self.clock, name="sticky_events_by_room", server_name=self.server_name
         ):
-            from_id = since_token.sticky_events_key if since_token else 0
+            from_token = (
+                since_token.sticky_events_key
+                if since_token
+                else StreamToken.START.sticky_events_key
+            )
 
             room_ids = sync_result_builder.joined_room_ids
 
-            to_id, sticky_by_room = await self.store.get_sticky_events_in_rooms(
+            to_token, sticky_by_room = await self.store.get_sticky_events_in_rooms(
                 room_ids,
-                from_id=from_id,
-                to_id=now_token.sticky_events_key,
+                from_token=from_token,
+                to_token=now_token.sticky_events_key,
                 now=now,
                 limit=StickyEvent.MAX_EVENTS_IN_SYNC,
             )
-            now_token = now_token.copy_and_replace(StreamKeyType.STICKY_EVENTS, to_id)
+            now_token = now_token.copy_and_replace(
+                StreamKeyType.STICKY_EVENTS, to_token
+            )
 
         return now_token, sticky_by_room
 
@@ -815,8 +821,8 @@ class SyncHandler:
             _, sticky_by_room = await self.store.get_sticky_events_in_rooms(
                 newly_joined_rooms,
                 # Since the start of time
-                from_id=0,
-                to_id=now_token.sticky_events_key,
+                from_token=StreamToken.START.sticky_events_key,
+                to_token=now_token.sticky_events_key,
                 now=now,
                 # Unfortunately, we're meant to return all sticky events in one go
                 # (we don't have a good alternative).

synapse/notifier.py

@@ -533,7 +533,7 @@ class Notifier:
     @overload
     def on_new_event(
         self,
-        stream_key: Literal[StreamKeyType.RECEIPT],
+        stream_key: Literal[StreamKeyType.RECEIPT, StreamKeyType.STICKY_EVENTS],
         new_token: MultiWriterStreamToken,
         users: Collection[str | UserID] | None = None,
         rooms: StrCollection | None = None,
@@ -551,7 +551,6 @@ class Notifier:
             StreamKeyType.TYPING,
             StreamKeyType.UN_PARTIAL_STATED_ROOMS,
             StreamKeyType.THREAD_SUBSCRIPTIONS,
-            StreamKeyType.STICKY_EVENTS,
             StreamKeyType.PROFILE_UPDATES,
         ],
         new_token: int,

synapse/replication/tcp/client.py

@@ -297,7 +297,7 @@ class ReplicationDataHandler:
         elif stream_name == StickyEventsStream.NAME:
             self.notifier.on_new_event(
                 StreamKeyType.STICKY_EVENTS,
-                token,
+                self.store.get_sticky_events_stream_token(),
                 rooms=[row.room_id for row in rows],
             )
 

synapse/storage/databases/main/sticky_events.py

@@ -40,8 +40,12 @@ from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.events import DeltaState
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine
-from synapse.storage.util.id_generators import MultiWriterIdGenerator
-from synapse.types import RoomID, StateKey
+from synapse.storage.util.id_generators import (
+    MultiWriterIdGenerator,
+    advance_multiwriter_sharded_token_after_partial_read,
+    make_multiwriter_sharded_token_bounds_sql,
+)
+from synapse.types import MultiWriterStreamToken, RoomID, StateKey
 from synapse.types.state import StateFilter
 from synapse.util.duration import Duration
 from synapse.util.stringutils import shortstr
@@ -141,13 +145,13 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             self._sticky_events_id_gen.advance(instance_name, token)
         super().process_replication_position(stream_name, instance_name, token)
 
-    def get_max_sticky_events_stream_id(self) -> int:
-        """Get the current maximum stream_id for sticky events.
+    def get_sticky_events_stream_token(self) -> MultiWriterStreamToken:
+        """Get the current position of the sticky events stream.
 
         Returns:
-            The maximum stream_id
+            A (potentially sharded) token for the sticky events stream.
         """
-        return self._sticky_events_id_gen.get_current_token()
+        return MultiWriterStreamToken.from_generator(self._sticky_events_id_gen)
 
     def get_sticky_events_stream_id_generator(self) -> MultiWriterIdGenerator:
         return self._sticky_events_id_gen
@@ -156,41 +160,54 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
         self,
         room_ids: Collection[str],
         *,
-        from_id: int,
-        to_id: int,
+        from_token: MultiWriterStreamToken,
+        to_token: MultiWriterStreamToken,
         now: int,
         limit: int | None,
-    ) -> tuple[int, dict[str, list[str]]]:
+    ) -> tuple[MultiWriterStreamToken, dict[str, list[str]]]:
         """
-        Fetch all the sticky events' IDs in the given rooms, with sticky stream IDs satisfying
-        from_id < sticky stream ID <= to_id.
+        Fetch all the sticky events' IDs in the given rooms, with sticky stream positions
+        after `from_token` and at or before `to_token`.
 
         The events are returned ordered by the sticky events stream.
 
         Args:
             room_ids: The room IDs to return sticky events in.
-            from_id: The sticky stream ID that sticky events should be returned from (exclusive).
-            to_id: The sticky stream ID that sticky events should end at (inclusive).
+            from_token: The sticky stream position to return sticky events from (exclusive).
+            to_token: The sticky stream position to end at (inclusive).
             now: The current time in unix millis, used for skipping expired events.
             limit: Max sticky events to return, or None to apply no limit.
         Returns:
-            to_id, dict[room_id, list[event_ids]]
+            The stream position that has been read up to (which may be behind
+            `to_token` if `limit` was hit), and a dict[room_id, list[event_ids]].
         """
+        if limit == 0:
+            # No rows to return, so don't advance.
+            return from_token, {}
+
         sticky_events_rows = await self.db_pool.runInteraction(
             "get_sticky_events_in_rooms",
             self._get_sticky_events_in_rooms_txn,
             room_ids,
-            from_id=from_id,
-            to_id=to_id,
+            from_token=from_token,
+            to_token=to_token,
             now=now,
             limit=limit,
         )
 
-        if not sticky_events_rows:
-            return to_id, {}
-
-        # Get stream_id of the last row, which is the highest
-        new_to_id, _, _ = sticky_events_rows[-1]
+        if limit is not None and len(sticky_events_rows) == limit:
+            # We hit the limit, so advance the `to_token` partially
+            last_stream_id, _, _ = sticky_events_rows[-1]
+            new_to_token = advance_multiwriter_sharded_token_after_partial_read(
+                from_token_exclusive=from_token,
+                to_token_inclusive=to_token,
+                last_read_stream_id=last_stream_id,
+            )
+        else:
+            # We didn't hit the limit, therefore we have read the whole range.
+            # We can skip ahead to `to_token`, using `copy_and_advance` to ensure
+            # we don't rewind `from_token` in the case this worker is behind.
+            new_to_token = from_token.copy_and_advance(to_token)
 
         # room ID -> event IDs
         room_id_to_event_ids: dict[str, list[str]] = {}
@@ -198,23 +215,36 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             events = room_id_to_event_ids.setdefault(room_id, [])
             events.append(event_id)
 
-        return (new_to_id, room_id_to_event_ids)
+        return (new_to_token, room_id_to_event_ids)
 
     def _get_sticky_events_in_rooms_txn(
         self,
         txn: LoggingTransaction,
         room_ids: Collection[str],
         *,
-        from_id: int,
-        to_id: int,
+        from_token: MultiWriterStreamToken,
+        to_token: MultiWriterStreamToken,
         now: int,
         limit: int | None,
     ) -> list[tuple[int, str, str]]:
-        if len(room_ids) == 0:
+        if (
+            len(room_ids) == 0
+            or
+            # Check `to_token <= from_token` as a client could give us a token that is ahead of our 'now' position,
+            # perhaps if this worker is lagging
+            to_token.is_before_or_eq(from_token)
+        ):
             return []
         room_id_in_list_clause, room_id_in_list_values = make_in_list_sql_clause(
             txn.database_engine, "se.room_id", room_ids
         )
+        stream_clause, stream_values = make_multiwriter_sharded_token_bounds_sql(
+            self.database_engine,
+            stream_id_column="se.stream_id",
+            instance_name_column="se.instance_name",
+            from_token_exclusive=from_token,
+            to_token_inclusive=to_token,
+        )
         limit_clause = ""
         limit_params: tuple[int, ...] = ()
         if limit is not None:
@@ -234,13 +264,12 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             WHERE
                 NOT {expr_soft_failed}
                 AND ? < expires_at
-                AND ? < stream_id
-                AND stream_id <= ?
+                AND {stream_clause}
                 AND {room_id_in_list_clause}
-            ORDER BY stream_id ASC
+            ORDER BY se.stream_id ASC
             {limit_clause}
             """,
-            (now, from_id, to_id, *room_id_in_list_values, *limit_params),
+            (now, *stream_values, *room_id_in_list_values, *limit_params),
         )
         return cast(list[tuple[int, str, str]], txn.fetchall())
 
@@ -249,6 +278,10 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
     ) -> list[StickyEventUpdate]:
         """Get updates to sticky events between two stream IDs.
 
+        You probably don't want to use this as it doesn't use sharded tokens.
+        Consider `get_sticky_events_in_rooms` if you are looking at a set of rooms.
+        This method is more or less internal machinery for replication.
+
         Bounds: from_id < ... <= to_id
 
         Args:

synapse/streams/events.py

@@ -84,7 +84,7 @@ class EventSources:
             self._instance_name
         )
         thread_subscriptions_key = self.store.get_max_thread_subscriptions_stream_id()
-        sticky_events_key = self.store.get_max_sticky_events_stream_id()
+        sticky_events_key = self.store.get_sticky_events_stream_token()
         quarantined_media_key = self.store.get_quarantined_media_stream_token()
         profile_updates_key = self.store.get_max_profile_updates_stream_id()
 

synapse/types/__init__.py

@@ -1020,65 +1020,74 @@ class RoomStreamToken(AbstractMultiWriterStreamToken):
 
 @attr.s(frozen=True, slots=True, order=False)
 class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
-    """A basic stream token class for streams that supports multiple writers."""
+    """
+    A basic stream token class for streams that supports multiple writers.
+
+    Serialised as either:
+        - the bare `stream` position (e.g. `42`); or
+        - if any writer is ahead of that position,
+          as `m{stream}~{id}.{pos}~{id}.{pos}...`, where:
+          each `id` is the writer's ID from the `instance_map` table
+          (see `DataStore.get_id_for_instance`)
+
+    For the latter (`m{}~{}.{}~...`) form, it is necessary to check the writer of a
+    given row to determine how it compares to the token.
+
+    See:
+        - `is_stream_position_in_range` for a single row
+        - `is_before_or_eq` for comparing two tokens
+        - `make_multiwriter_sharded_token_bounds_sql` for SQL
+            - and its companion `advance_multiwriter_sharded_token_after_partial_read`
+    """
 
     @classmethod
-    async def parse(cls, store: "DataStore", string: str) -> "MultiWriterStreamToken":
+    async def parse(cls, store: "DataStore", string: str) -> Self:
+        unresolved = UnresolvedMultiWriterStreamToken.parse(string)
+        return await cls.from_unresolved(store, unresolved)
+
+    @classmethod
+    async def from_unresolved(
+        cls, store: "DataStore", unresolved: "UnresolvedMultiWriterStreamToken"
+    ) -> Self:
         try:
-            if string[0].isdigit():
-                return cls(stream=int(string))
-            if string[0] == "m":
-                parts = string[1:].split("~")
-                stream = int(parts[0])
-
-                instance_map = {}
-                for part in parts[1:]:
-                    if not part:
-                        # Handle tokens of the form `m5~`, which were created by
-                        # a bug
-                        continue
-
-                    key, value = part.split(".")
-                    instance_id = int(key)
-                    pos = int(value)
-
-                    instance_name = await store.get_name_from_instance_id(instance_id)
-                    instance_map[instance_name] = pos
-
-                return cls(
-                    stream=stream,
-                    instance_map=immutabledict(instance_map),
-                )
+            return cls(
+                stream=unresolved.stream,
+                instance_map=immutabledict(
+                    {
+                        (await store.get_name_from_instance_id(instance_id)): pos
+                        for instance_id, pos in unresolved.instance_id_to_stream_pos.items()
+                    }
+                ),
+            )
         except CancelledError:
             raise
         except Exception:
             # We log an exception here as even though this *might* be a client
             # handing a bad token, its more likely that Synapse returned a bad
             # token (and we really want to catch those!).
+            # TODO(2026): reivilibre: I'm less convinced this makes sense nowadays.
             logger.exception("Failed to parse stream token: %r", string)
         raise SynapseError(400, "Invalid stream token %r" % (string,))
 
     async def to_string(self, store: "DataStore") -> str:
         """See class level docstring for information about the format."""
+        return (await self.to_unresolved(store)).to_string()
 
-        if self.instance_map:
-            entries = []
-            for name, pos in self.instance_map.items():
-                if pos <= self.stream:
-                    # Ignore instances who are below the minimum stream position
-                    # (we might know they've advanced without seeing a recent
-                    # write from them).
-                    continue
-
-                instance_id = await store.get_id_for_instance(name)
-                entries.append(f"{instance_id}.{pos}")
-
-            if entries:
-                encoded_map = "~".join(entries)
-                return f"m{self.stream}~{encoded_map}"
-            return str(self.stream)
-        else:
-            return str(self.stream)
+    async def to_unresolved(
+        self, store: "DataStore"
+    ) -> "UnresolvedMultiWriterStreamToken":
+        """
+        Converts this token to its unresolved form, so that it can be used without database access.
+        """
+        return UnresolvedMultiWriterStreamToken(
+            stream=self.stream,
+            instance_id_to_stream_pos=immutabledict(
+                {
+                    (await store.get_id_for_instance(instance_name)): pos
+                    for instance_name, pos in self.instance_map.items()
+                }
+            ),
+        )
 
     @staticmethod
     def is_stream_position_in_range(
@@ -1087,7 +1096,10 @@ class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
         instance_name: str | None,
         pos: int,
     ) -> bool:
-        """Checks if a given persisted position is between the two given tokens.
+        """
+        Checks if a given persisted position is between the two given tokens.
+
+        Bounds: low < (instance_name, pos) <= high
 
         If `instance_name` is None then the row was persisted before multi
         writer support.
@@ -1120,6 +1132,91 @@ class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
         )
 
 
+@attr.s(auto_attribs=True, frozen=True, slots=True, order=False)
+class UnresolvedMultiWriterStreamToken:
+    """
+    `MultiWriterStreamToken`, but unresolved raw form where the instance names are not available.
+    This unresolved form is suitable for (de)serialising without database access.
+    Converting to or from a `MultiWriterStreamToken` requires database access to resolve instance (worker) names.
+
+    Serialised as either:
+        - the bare `stream` position (e.g. `42`); or
+        - if any writer is ahead of that position,
+            as `m{stream}~{id}.{pos}~{id}.{pos}...`, where:
+            each `id` is the writer's ID from the `instance_map` table
+            (see `DataStore.get_id_for_instance`)
+    """
+
+    stream: int
+    """
+    The baseline stream position.
+
+    All stream facts with stream ID numerically lower than this are unconditionally before this token.
+    """
+
+    instance_id_to_stream_pos: Mapping[int, int] = immutabledict()
+    """
+    Positions by writing instance that are ahead of the baseline (`stream`) position.
+    """
+
+    @classmethod
+    def parse(cls, string: str) -> Self:
+        try:
+            if string[0].isdigit():
+                return cls(stream=int(string))
+            if string[0] == "m":
+                parts = string[1:].split("~")
+                stream = int(parts[0])
+                instance_id_to_stream_pos = {}
+
+                for part in parts[1:]:
+                    if not part:
+                        # Handle tokens of the form `m5~`, which were created by
+                        # a bug
+                        continue
+
+                    key, value = part.split(".")
+                    instance_id = int(key)
+                    pos = int(value)
+
+                    instance_id_to_stream_pos[instance_id] = pos
+
+                return cls(
+                    stream=stream,
+                    instance_id_to_stream_pos=immutabledict(instance_id_to_stream_pos),
+                )
+        except CancelledError:
+            raise
+        except Exception:
+            # We log an exception here as even though this *might* be a client
+            # handing a bad token, its more likely that Synapse returned a bad
+            # token (and we really want to catch those!).
+            # TODO(2026): reivilibre: I'm less convinced this makes sense nowadays.
+            logger.exception("Failed to parse stream token: %r", string)
+        raise SynapseError(400, "Invalid stream token %r" % (string,))
+
+    def to_string(self) -> str:
+        """See class level docstring for information about the format."""
+
+        if self.instance_id_to_stream_pos:
+            entries = []
+            for instance_id, pos in self.instance_id_to_stream_pos.items():
+                if pos <= self.stream:
+                    # Ignore instances who are below the minimum stream position
+                    # (we might know they've advanced without seeing a recent
+                    # write from them).
+                    continue
+
+                entries.append(f"{instance_id}.{pos}")
+
+            if entries:
+                encoded_map = "~".join(entries)
+                return f"m{self.stream}~{encoded_map}"
+            return str(self.stream)
+        else:
+            return str(self.stream)
+
+
 class StreamKeyType(Enum):
     """Known stream types.
 
@@ -1222,7 +1319,9 @@ class StreamToken:
     groups_key: int
     un_partial_stated_rooms_key: int
     thread_subscriptions_key: int
-    sticky_events_key: int
+    sticky_events_key: MultiWriterStreamToken = attr.ib(
+        validator=attr.validators.instance_of(MultiWriterStreamToken)
+    )
     quarantined_media_key: MultiWriterStreamToken = attr.ib(
         validator=attr.validators.instance_of(MultiWriterStreamToken)
     )
@@ -1274,7 +1373,9 @@ class StreamToken:
                 groups_key=int(groups_key),
                 un_partial_stated_rooms_key=int(un_partial_stated_rooms_key),
                 thread_subscriptions_key=int(thread_subscriptions_key),
-                sticky_events_key=int(sticky_events_key),
+                sticky_events_key=await MultiWriterStreamToken.parse(
+                    store, sticky_events_key
+                ),
                 quarantined_media_key=await MultiWriterStreamToken.parse(
                     store, quarantined_media_key
                 ),
@@ -1302,7 +1403,7 @@ class StreamToken:
                 str(self.groups_key),
                 str(self.un_partial_stated_rooms_key),
                 str(self.thread_subscriptions_key),
-                str(self.sticky_events_key),
+                await self.sticky_events_key.to_string(store),
                 await self.quarantined_media_key.to_string(store),
                 str(self.profile_updates_key),
             ]
@@ -1340,6 +1441,12 @@ class StreamToken:
                 self.quarantined_media_key.copy_and_advance(new_value),
             )
             return new_token
+        elif key == StreamKeyType.STICKY_EVENTS:
+            new_token = self.copy_and_replace(
+                StreamKeyType.STICKY_EVENTS,
+                self.sticky_events_key.copy_and_advance(new_value),
+            )
+            return new_token
 
         new_token = self.copy_and_replace(key, new_value)
         new_id = new_token.get_field(key)
@@ -1363,6 +1470,7 @@ class StreamToken:
             StreamKeyType.RECEIPT,
             StreamKeyType.DEVICE_LIST,
             StreamKeyType.QUARANTINED_MEDIA,
+            StreamKeyType.STICKY_EVENTS,
         ],
     ) -> MultiWriterStreamToken: ...
 
@@ -1377,7 +1485,6 @@ class StreamToken:
             StreamKeyType.TYPING,
             StreamKeyType.UN_PARTIAL_STATED_ROOMS,
             StreamKeyType.THREAD_SUBSCRIPTIONS,
-            StreamKeyType.STICKY_EVENTS,
             StreamKeyType.PROFILE_UPDATES,
         ],
     ) -> int: ...
@@ -1472,7 +1579,7 @@ StreamToken.START = StreamToken(
     groups_key=0,
     un_partial_stated_rooms_key=0,
     thread_subscriptions_key=0,
-    sticky_events_key=0,
+    sticky_events_key=MultiWriterStreamToken(stream=0),
     quarantined_media_key=MultiWriterStreamToken(stream=0),
     profile_updates_key=0,
 )

synapse/types/rest/client/__init__.py

@@ -19,7 +19,7 @@
 #
 #
 import re
-from typing import ClassVar
+from typing import TYPE_CHECKING, ClassVar
 
 import pydantic_core.core_schema
 from pydantic import (
@@ -37,10 +37,19 @@ from pydantic import (
 from pydantic_core import CoreSchema
 from typing_extensions import Annotated, Self
 
-from synapse.types import Absent, AbsentType, NonNegativeStrictInt
+from synapse.types import (
+    Absent,
+    AbsentType,
+    MultiWriterStreamToken,
+    NonNegativeStrictInt,
+    UnresolvedMultiWriterStreamToken,
+)
 from synapse.types.rest import RequestBodyModel
 from synapse.util.threepids import validate_email
 
+if TYPE_CHECKING:
+    from synapse.storage.databases.main import DataStore
+
 
 class AuthenticationData(RequestBodyModel):
     """
@@ -139,20 +148,37 @@ class SlidingSyncStickyEventsToken:
     and then accepted as the `since` parameter in the requests of the same extension.
 
     Current format:
-        SlidingSyncStickyEventsToken ::= 'sticky_' DIGIT+
-        DIGIT ::= '0'-'9'
+        SlidingSyncStickyEventsToken ::= 'sticky_' UnresolvedMultiWriterStreamToken
+
+    where `MultiWriterStreamToken` is the serialised form of a (potentially sharded)
+    `MultiWriterStreamToken` for the sticky events stream, e.g. `42` or `m42~1.45`.
 
     The `sticky_` prefix allows us to make sure it's not swapped for another token
     or to evolve the type of token accepted with backwards compatibility in the future.
+
+    Note that the inner stream token can only be interpreted with access to the
+    database (to resolve instance IDs to instance names), so this class holds the
+    serialised form and converts on demand.
     """
 
-    PATTERN = re.compile(r"^sticky_([0-9]+)$")
+    PATTERN = re.compile(r"^sticky_([0-9]+|m[0-9]+(~[0-9]+\.[0-9]+)*)$")
     START: ClassVar["SlidingSyncStickyEventsToken"]
 
-    def __init__(self, *, sticky_events_stream_id: int) -> None:
-        # FIXME: We should use MultiWriterStreamToken here
-        # Track: https://github.com/element-hq/synapse/issues/19661
-        self.sticky_events_stream_id = sticky_events_stream_id
+    def __init__(
+        self, *, unresolved_stream_token: UnresolvedMultiWriterStreamToken
+    ) -> None:
+        self._unresolved_stream_token = unresolved_stream_token
+
+    @classmethod
+    async def from_stream_token(
+        cls, store: "DataStore", token: MultiWriterStreamToken
+    ) -> "SlidingSyncStickyEventsToken":
+        return cls(unresolved_stream_token=await token.to_unresolved(store))
+
+    async def to_stream_token(self, store: "DataStore") -> MultiWriterStreamToken:
+        return await MultiWriterStreamToken.from_unresolved(
+            store, self._unresolved_stream_token
+        )
 
     @classmethod
     def __get_pydantic_core_schema__(
@@ -187,7 +213,12 @@ class SlidingSyncStickyEventsToken:
             match = cls.PATTERN.match(v)
             if match is None:
                 raise ValueError(f"Invalid SlidingSyncStickyEventsToken format: {v!r}")
-            return cls(sticky_events_stream_id=int(match.group(1)))
+
+            return cls(
+                unresolved_stream_token=UnresolvedMultiWriterStreamToken.parse(
+                    match.group(1)
+                )
+            )
         raise ValueError(f"Cannot parse SlidingSyncStickyEventsToken from {type(v)}")
 
     def serialise(self) -> str:
@@ -196,7 +227,7 @@ class SlidingSyncStickyEventsToken:
 
         The inverse of `_validate`.
         """
-        return f"sticky_{self.sticky_events_stream_id}"
+        return f"sticky_{self._unresolved_stream_token.to_string()}"
 
     def __repr__(self) -> str:
         # Use the serialised form as debug output.
@@ -205,7 +236,7 @@ class SlidingSyncStickyEventsToken:
 
 # Starting reading a stream at 0 ensures all stream fact rows will be read
 SlidingSyncStickyEventsToken.START = SlidingSyncStickyEventsToken(
-    sticky_events_stream_id=0
+    unresolved_stream_token=UnresolvedMultiWriterStreamToken(stream=0)
 )
 
 

tests/replication/test_sharded_sticky_events.py

@@ -0,0 +1,282 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations 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>.
+import logging
+import sqlite3
+from contextlib import contextmanager
+from typing import Iterator
+from unittest.mock import AsyncMock
+
+from twisted.internet.testing import MemoryReactor
+
+import synapse.rest.admin
+from synapse.api.constants import EventTypes
+from synapse.rest.client import login, room, sync
+from synapse.server import HomeServer
+from synapse.types import JsonDict
+from synapse.util.clock import Clock
+from synapse.util.duration import Duration
+
+from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.utils import USE_POSTGRES_FOR_TESTS
+
+logger = logging.getLogger(__name__)
+
+
+class ShardedStickyEventsTestCase(BaseMultiWorkerStreamTestCase):
+    """
+    Tests for sharded Sticky Events.
+    """
+
+    if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0):
+        # We need the JSON functionality in SQLite
+        skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)"
+
+    servlets = [
+        synapse.rest.admin.register_servlets_for_client_rest_resource,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        conf = super().default_config()
+        conf["experimental_features"] = {
+            # Sliding Sync
+            "msc3575_enabled": True,
+            # Sticky Events
+            "msc4354_enabled": True,
+        }
+        conf["stream_writers"] = {"events": ["worker1", "worker2"]}
+        conf["instance_map"] = {
+            "main": {"host": "testserv", "port": 8765},
+            "worker1": {"host": "testserv", "port": 1001},
+            "worker2": {"host": "testserv", "port": 1002},
+        }
+        return conf
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+        # FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION`, see
+        # `SlidingSyncBase`.
+        self.store.have_finished_sliding_sync_background_jobs = AsyncMock(  # type: ignore[method-assign]
+            return_value=True
+        )
+
+        self.user_id = self.register_user("user", "pass")
+        self.tok = self.login("user", "pass")
+
+    def _setup_workers_and_rooms(self) -> None:
+        """
+        Start the two event persisters and create a room on each of them.
+        """
+        self.make_worker_hs("synapse.app.generic_worker", {"worker_name": "worker1"})
+        self.worker_hs2 = self.make_worker_hs(
+            "synapse.app.generic_worker", {"worker_name": "worker2"}
+        )
+
+        # Create rooms until we have at least one on each of the workers.
+        # (We can't choose the room IDs directly, so we rely on the sharding
+        # hash to route the generated rooms to different workers.)
+        room_generation_results = self._generate_rooms_on_worker(self.user_id, self.tok)
+        self.room_id1 = room_generation_results["worker1"]
+        self.room_id2 = room_generation_results["worker2"]
+
+        # Verify that the expected worker is responsible for each room.
+        self.assertEqual(
+            self.hs.config.worker.events_shard_config.get_instance(self.room_id1),
+            "worker1",
+        )
+        self.assertEqual(
+            self.hs.config.worker.events_shard_config.get_instance(self.room_id2),
+            "worker2",
+        )
+
+    def _send_sticky_event(self, room_id: str, body: str) -> str:
+        return self.helper.send_sticky_event(
+            room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=5),
+            content={"body": body, "msgtype": "m.text"},
+            tok=self.tok,
+        )["event_id"]
+
+    @contextmanager
+    def _stalled_sticky_writer(self) -> Iterator[None]:
+        """
+        Manipulate worker2's sticky events `MultiWriterIdGenerator` to mimic
+        it getting stuck part way through persisting a sticky event.
+
+        Whilst the context manager is entered, worker2's sticky events stream position
+        will not advance.
+        This stops the 'all writers have persisted up to' position from advancing as well.
+        """
+        worker_store2 = self.worker_hs2.get_datastores().main
+        stalled_write = worker_store2._sticky_events_id_gen.get_next()
+
+        self.get_success(stalled_write.__aenter__())
+        try:
+            yield
+        finally:
+            self.get_success(stalled_write.__aexit__(None, None, None))
+
+    def test_torn_oldschool_sync(self) -> None:
+        """
+        Tests that a sticky event must not be presented in two sync responses
+        as a 'torn' sync response. This test is for oldschool sync.
+
+        Specifically, a sticky event must not arrive in the timeline of
+        the first sync response and then in the sticky events section in
+        a subsequent sync response.
+        (This is a specific failure mode that occurred when the reader of the
+        sticky events stream was not shard-aware.)
+        """
+        self._setup_workers_and_rooms()
+
+        # Get our initial sync out of the way
+        channel = self.make_request("GET", "/sync", access_token=self.tok)
+        self.assertEqual(channel.code, 200, channel.result)
+        since = channel.json_body["next_batch"]
+
+        with self._stalled_sticky_writer():
+            sticky_event_id = self._send_sticky_event(self.room_id1, "sticky message")
+
+            # First response of the torn syncs: the event comes down the timeline, because the
+            # events stream token covers worker1's write despite worker2 lagging.
+            channel = self.make_request(
+                "GET", f"/sync?since={since}", access_token=self.tok
+            )
+            self.assertEqual(channel.code, 200, channel.result)
+            room = channel.json_body["rooms"]["join"][self.room_id1]
+            self.assertIn(
+                sticky_event_id,
+                [event["event_id"] for event in room["timeline"]["events"]],
+                room,
+            )
+            # Having been sent it in the timeline, we shouldn't also be sent it in
+            # the sticky section of the same response.
+            self.assertNotIn("msc4354_sticky", room, room)
+            since = channel.json_body["next_batch"]
+
+        # worker2 catches up, so an unsharded sticky events position is now free to
+        # advance past worker1's write.
+        self.replicate()
+
+        # Second response of the torn syncs: the sticky event shouldn't come down
+        # in the sticky event section.
+        channel = self.make_request(
+            "GET", f"/sync?since={since}", access_token=self.tok
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        room = channel.json_body.get("rooms", {}).get("join", {}).get(self.room_id1, {})
+        self.assertNotIn(
+            "msc4354_sticky",
+            room,
+            f"sticky event {sticky_event_id} was sent down the timeline in the "
+            f"previous sync, so we should not expect to see a sticky section in this sync: "
+            f"{room}",
+        )
+
+    def test_torn_sliding_sync(self) -> None:
+        """
+        Tests that a sticky event must not be presented in two sync responses
+        as a 'torn' sync response. This test is for sliding sync.
+
+        Specifically, a sticky event must not arrive in the timeline of
+        the first sync response and then in the sticky events section in
+        a subsequent sync response.
+        (This is a specific failure mode that occurred when the reader of the
+        sticky events stream was not shard-aware.)
+        """
+        self._setup_workers_and_rooms()
+
+        sync_endpoint = "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync"
+        sync_body: JsonDict = {
+            "lists": {
+                "main": {
+                    "ranges": [[0, 99]],
+                    "required_state": [],
+                    "timeline_limit": 10,
+                }
+            },
+            "extensions": {"org.matrix.msc4354.sticky_events": {"enabled": True}},
+        }
+
+        # Initial sync
+        channel = self.make_request(
+            "POST", sync_endpoint, sync_body, access_token=self.tok
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        pos = channel.json_body["pos"]
+
+        # There are no sticky events yet
+        self.assertNotIn(
+            "org.matrix.msc4354.sticky_events",
+            channel.json_body["extensions"],
+            channel.json_body["extensions"],
+        )
+
+        with self._stalled_sticky_writer():
+            sticky_event_id = self._send_sticky_event(self.room_id1, "sticky message")
+
+            # First response of the torn syncs: the event comes down the room timeline.
+            channel = self.make_request(
+                "POST", f"{sync_endpoint}?pos={pos}", sync_body, access_token=self.tok
+            )
+            self.assertEqual(channel.code, 200, channel.result)
+            pos = channel.json_body["pos"]
+            room = channel.json_body["rooms"][self.room_id1]
+            self.assertIn(
+                sticky_event_id,
+                [event["event_id"] for event in room["timeline"]],
+                room,
+            )
+
+            # ... and so should have been deduplicated out of the sticky section of
+            # the same response, leaving an empty room entry behind.
+            sticky_extension = channel.json_body["extensions"][
+                "org.matrix.msc4354.sticky_events"
+            ]
+            next_batch = sticky_extension["next_batch"]
+            self.assertEqual(
+                sticky_extension,
+                {
+                    "rooms": {self.room_id1: {"events": []}},
+                    "next_batch": next_batch,
+                },
+                sticky_extension,
+            )
+            # The token should carry the per-writer positions, as worker2 is stalled.
+            self.assertTrue(next_batch.startswith("sticky_m"), next_batch)
+
+            sync_body["extensions"]["org.matrix.msc4354.sticky_events"]["since"] = (
+                next_batch
+            )
+
+        # worker2 catches up, so an unsharded sticky events position is now free to
+        # advance past worker1's write.
+        self.replicate()
+
+        # Second response of the torn syncs: the sticky event shouldn't come down
+        # in the sticky event section.
+        channel = self.make_request(
+            "POST", f"{sync_endpoint}?pos={pos}", sync_body, access_token=self.tok
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+        # There is nothing left to send, so the extension should be omitted entirely.
+        self.assertNotIn(
+            "org.matrix.msc4354.sticky_events",
+            channel.json_body["extensions"],
+            f"sticky event {sticky_event_id} was sent down the timeline in the "
+            f"previous sync, so no sticky events extension should be sent now: "
+            f"{channel.json_body['extensions']}",
+        )

tests/rest/client/sliding_sync/test_extension_sticky_events.py

@@ -594,6 +594,68 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
             response_body, {room_id: sticky_event_ids[2:4]}
         )
 
+    def test_sticky_event_limit_zero_does_not_advance(self) -> None:
+        """
+        Tests that sending a limit of 0 in the Sticky Events request extension returns no
+        sticky events and omits the extension from the response, rather than erroring.
+
+        'Regression' test for a bug that _almost_ existed.
+        (The `next_batch` token would be wrongly advanced internally,
+        but because no events were returned, the entire extension response was not serialised
+        and so the dodgy `next_batch` never actually got sent to a client.)
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # Create a room
+        room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id, user1_id, tok=user1_tok)
+
+        # Send 2 sticky events (more than our limit of 1)
+        sticky_event_ids: list[str] = []
+        for i in range(2):
+            event_id = self.helper.send_sticky_event(
+                room_id,
+                EventTypes.Message,
+                duration=Duration(minutes=5),
+                content={"body": f"sticky message {i}", "msgtype": "m.text"},
+                tok=user2_tok,
+            )["event_id"]
+            sticky_event_ids.append(event_id)
+
+        # Initial sync
+        sync_body = {
+            "lists": DUMMY_LISTS,
+            "extensions": {
+                "org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 1}
+            },
+        }
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok)
+
+        # We expect to see the first sticky event
+        next_batch = self._assert_sticky_events_response(
+            response_body, {room_id: sticky_event_ids[0:1]}
+        )
+
+        # Incremental sync with limit=0
+        sync_body = {
+            "lists": DUMMY_LISTS,
+            "extensions": {
+                "org.matrix.msc4354.sticky_events": {
+                    "enabled": True,
+                    # This makes it incremental
+                    "since": next_batch,
+                    "limit": 0,
+                }
+            },
+        }
+        response_body, _ = self.do_sync(sync_body, tok=user1_tok)
+        # You could imagine a buggy system returns a `next_batch` here with no events,
+        # but Synapse just omits the response extension altogether.
+        next_batch = self._assert_sticky_events_response(response_body, None)
+
     def test_deduplication_with_timeline(self) -> None:
         """
         Test that sticky events are not included in the sticky event extension of sliding sync

tests/storage/test_sticky_events.py

@@ -13,6 +13,8 @@
 import sqlite3
 from http import HTTPStatus
 
+from immutabledict import immutabledict
+
 from twisted.internet.testing import MemoryReactor
 
 from synapse.api.constants import (
@@ -27,7 +29,7 @@ from synapse.replication.tcp.streams._base import StickyEventStreamPosition
 from synapse.rest import admin
 from synapse.rest.client import login, register, room, sync
 from synapse.server import HomeServer
-from synapse.types import JsonDict, RoomID, create_requester
+from synapse.types import JsonDict, MultiWriterStreamToken, RoomID, create_requester
 from synapse.util.clock import Clock
 from synapse.util.duration import Duration
 
@@ -76,7 +78,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
     def test_get_updated_sticky_events(self) -> None:
         """Test getting updated sticky events between stream IDs."""
         # Get the starting stream_id
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         event_id_1 = self.helper.send_sticky_event(
             self.room_id,
@@ -86,7 +88,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             tok=self.token,
         )["event_id"]
 
-        mid_id = self.store.get_max_sticky_events_stream_id()
+        mid_id = self.store.get_sticky_events_stream_token().stream
 
         event_id_2 = self.helper.send_sticky_event(
             self.room_id,
@@ -96,7 +98,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             tok=self.token,
         )["event_id"]
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         # Get all updates
         updates = self.get_success(
@@ -139,7 +141,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             tok=self.token,
         )["event_id"]
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         # Delete expired events
         self.get_success(self.store._delete_expired_sticky_events())
@@ -160,7 +162,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
     def test_get_updated_sticky_events_with_limit(self) -> None:
         """Test that the limit parameter works correctly."""
         # Get the starting stream_id
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         event_id_1 = self.helper.send_sticky_event(
             self.room_id,
@@ -187,6 +189,67 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         self.assertEqual(len(updates), 1)
         self.assertEqual(updates[0].event_id, event_id_1)
 
+    def test_get_sticky_events_in_rooms_token_does_not_go_backwards(self) -> None:
+        """
+        Tests `get_sticky_events_in_rooms` to make sure that the token does not get rewound, for instance if a
+        different, lagging, sync worker handles a request.
+        """
+        instance_name = self.hs.get_instance_name()
+
+        self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "message 1", "msgtype": "m.text"},
+            tok=self.token,
+        )
+        first_id = self.store.get_sticky_events_stream_token().stream
+
+        event_id_2 = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "message 2", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+        second_id = self.store.get_sticky_events_stream_token().stream
+
+        from_token = MultiWriterStreamToken(
+            stream=first_id, instance_map=immutabledict({"someworker": 42})
+        )
+        # `to_token` has 2 components that are behind compared to `from_token`:
+        # the baseline position and the position of the advanced worker 'someworker'
+        to_token = MultiWriterStreamToken(
+            stream=first_id - 1,
+            instance_map=immutabledict({instance_name: second_id, "someworker": 36}),
+        )
+
+        new_token, sticky_by_room = self.get_success(
+            self.store.get_sticky_events_in_rooms(
+                [self.room_id],
+                from_token=from_token,
+                to_token=to_token,
+                now=self.clock.time_msec(),
+                limit=None,
+            )
+        )
+
+        self.assertEqual(sticky_by_room, {self.room_id: [event_id_2]})
+        self.assertEqual(
+            new_token,
+            MultiWriterStreamToken(
+                # The baseline position is not wound back
+                stream=first_id,
+                instance_map=immutabledict(
+                    {
+                        instance_name: second_id,
+                        # The position of this advanced worker is not wound back either
+                        "someworker": 42,
+                    }
+                ),
+            ),
+        )
+
     def test_outlier_events_not_in_table(self) -> None:
         """
         Tests the behaviour of outliered and then de-outliered events in the
@@ -199,7 +262,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         user2_id = self.register_user("user2", "pass")
         user2_tok = self.login(user2_id, "pass")
 
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         room_id = self.helper.create_room_as(
             user2_id, tok=user2_tok, room_version=RoomVersions.V10.identifier
@@ -277,7 +340,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             )
         )
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         # Check the event made it into the sticky_events table
         updates = self.get_success(
@@ -368,7 +431,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         token = self.login(user_id, "pass")
         room_id = self.helper.create_room_as(user_id, tok=token)
 
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         # Create and persist a sticky event that is soft-failed
         soft_failed_sticky_event = self.get_success(
@@ -386,7 +449,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             )
         )
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         updates = self.get_success(
             self.store.get_updated_sticky_events(
@@ -407,7 +470,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         token = self.login(user_id, "pass")
         room_id = self.helper.create_room_as(user_id, tok=token)
 
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         # Create and persist a sticky event that is marked policy_server_spammy
         # N.B. policy_server_spammy events are always soft-failed too
@@ -441,7 +504,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             )
         )
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         # Verify only the regular event was inserted
         updates = self.get_success(
@@ -463,7 +526,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
         token = self.login(user_id, "pass")
         room_id = self.helper.create_room_as(user_id, tok=token)
 
-        start_id = self.store.get_max_sticky_events_stream_id()
+        start_id = self.store.get_sticky_events_stream_token().stream
 
         # Create and persist a sticky event that is marked spam_checker_spammy
         # N.B. spam_checker_spammy events are always soft-failed too
@@ -497,7 +560,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             )
         )
 
-        end_id = self.store.get_max_sticky_events_stream_id()
+        end_id = self.store.get_sticky_events_stream_token().stream
 
         # Verify only the valid sticky event was inserted
         updates = self.get_success(