diff --git a/changelog.d/20186.bugfix b/changelog.d/20186.bugfix new file mode 100644 index 0000000000..686d441a09 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/synapse/api/errors.py b/synapse/api/errors.py index 75cf8add27..bce77ffba1 100644 --- a/synapse/api/errors.py +++ b/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, ) diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 0e193cae36..3d49e585a4 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/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 ), ) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 2eb5eb79c6..3bc43e081c 100644 --- a/synapse/handlers/sync.py +++ b/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). diff --git a/synapse/notifier.py b/synapse/notifier.py index 3ee0fc60d6..be24cf4058 100644 --- a/synapse/notifier.py +++ b/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, diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py index 8a60f8df6f..6c897bd93a 100644 --- a/synapse/replication/tcp/client.py +++ b/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], ) diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index 1ff584a4a7..a45d7d9363 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/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: diff --git a/synapse/streams/events.py b/synapse/streams/events.py index 24120eb736..5d66b55ac4 100644 --- a/synapse/streams/events.py +++ b/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() diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 599ed5e3e4..79a09ce860 100644 --- a/synapse/types/__init__.py +++ b/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, ) diff --git a/synapse/types/rest/client/__init__.py b/synapse/types/rest/client/__init__.py index cd1bd1f057..28a48127f1 100644 --- a/synapse/types/rest/client/__init__.py +++ b/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) ) diff --git a/tests/replication/test_sharded_sticky_events.py b/tests/replication/test_sharded_sticky_events.py new file mode 100644 index 0000000000..b3a3b2d3ac --- /dev/null +++ b/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: +# . +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']}", + ) diff --git a/tests/rest/client/sliding_sync/test_extension_sticky_events.py b/tests/rest/client/sliding_sync/test_extension_sticky_events.py index aaea6d6793..cd48838648 100644 --- a/tests/rest/client/sliding_sync/test_extension_sticky_events.py +++ b/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 diff --git a/tests/storage/test_sticky_events.py b/tests/storage/test_sticky_events.py index 7058344123..a2880f239c 100644 --- a/tests/storage/test_sticky_events.py +++ b/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(