dandri/synapse
1
0 Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-11-04 06:57:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move the sticky events sliding sync extension to the global pos and MSC4508 scoping (#20363)

Browse Source 
This includes the following:
    
- Implements ['Implicit Forward
Pagination'](https://github.com/matrix-org/matrix-spec/issues/2378).
Remove the extension's `since` and `next_batch` fields. The position is
now tracked via the global `pos`, with per-room connection state
recording how far each room has been sent so that `limit` paginates
without holding `pos` back and lost responses are re-delivered.
- Discussed in
https://github.com/matrix-org/matrix-spec-proposals/pull/4480#discussion_r4165887827
- Honour the `lists` and `rooms` fields.
- Send all unexpired sticky events when a room enters scope for the
first time or is newly joined, and catch up on events sent while a room
was out of scope when it re-enters.
- Enabling the extension partway through a connection returns all
unexpired sticky events for in-scope rooms, and re-enabling it after
disabling it returns the unexpired sticky events sent in the meantime.

I tried to split this up into standalone commits though the bulk of the
change is in the third commit.

---------

Co-authored-by: Olivier 'reivilibre <git.oo@emunest.net>
This commit is contained in:
Johannes MarbachandOlivier 'reivilibre authored and GitHub committed 2026-10-26 14:02:22 +00:00
1 parent d99e5fe783
commit 0d28e93a7a
11 files changed
+1140 -344
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

changelog.d/20363.feature
+1
View File
@@ -0,0 +1 @@
Move the sticky events sliding sync extension to the global `pos` and MSC4508 scoping.
synapse/handlers/sliding_sync/__init__.py
+1 -1
View File
@@ -328,7 +328,6 @@ class SlidingSyncHandler:
actual_lists=lists,
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
all_interested_room_ids=interested_rooms.all_rooms,
# We're purposely using `relevant_room_map` instead of
# `relevant_rooms_to_send_map` here. This needs to be all room_ids we could
# send regardless of whether they have an event update or not. The
@@ -336,6 +335,7 @@ class SlidingSyncHandler:
# account data, read receipts, typing indicators, to-device messages, etc).
actual_room_ids=set(relevant_room_map.keys()),
actual_room_response_map=rooms,
newly_joined_room_ids=interested_rooms.newly_joined_rooms,
from_token=from_token,
to_token=to_token,
)
synapse/handlers/sliding_sync/extensions.py
+255 -102
View File
@@ -32,7 +32,6 @@ 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
@@ -60,7 +59,6 @@ from synapse.types.handlers.sliding_sync import (
SlidingSyncResult,
StateValues,
)
from synapse.types.rest.client import SlidingSyncStickyEventsToken
from synapse.util.async_helpers import (
concurrently_execute,
gather_optional_coroutines,
@@ -100,10 +98,10 @@ class SlidingSyncExtensionHandler:
sync_config: SlidingSyncConfig,
previous_connection_state: "PerConnectionState",
new_connection_state: "MutablePerConnectionState",
all_interested_room_ids: set[str],
actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList],
actual_room_ids: set[str],
actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult],
newly_joined_room_ids: AbstractSet[str],
to_token: StreamToken,
from_token: SlidingSyncStreamToken | None,
) -> SlidingSyncResult.Extensions:
@@ -114,19 +112,30 @@ class SlidingSyncExtensionHandler:
previous_connection_state: Snapshot of the current per-connection state
new_connection_state: A mutable copy of the per-connection
state, used to record updates to the state during this request.
all_interested_room_ids: The IDs of all rooms that the client is interested in,
even if they don't appear in the current limited window.
See `SlidingSyncInterestedRooms.all_rooms`.
actual_lists: Sliding window API. A map of list key to list results in the
Sliding Sync response.
actual_room_ids: The actual room IDs in the the Sliding Sync response.
actual_room_response_map: A map of room ID to room results in the the
Sliding Sync response.
newly_joined_room_ids: The rooms the user newly joined in the token range.
to_token: The latest point in the stream to sync up to.
from_token: The point in the stream to sync from.
"""
if sync_config.extensions is None:
# If no extensions are enabled, shortcut out.
if self._enable_sticky_events:
# The sticky events extension tracks per-room state even while it
# isn't enabled on the request, so that re-enabling it doesn't miss anything.
# (But only if the MSC4354 experimental feature is enabled)
await self._record_sticky_events_in_out_of_scope_rooms(
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
relevant_room_ids=set(),
to_token=to_token,
from_token=from_token,
)
return SlidingSyncResult.Extensions()
to_device_coro = None
@@ -198,14 +207,16 @@ class SlidingSyncExtensionHandler:
)
sticky_events_coro = None
if (
sync_config.extensions.sticky_events is not Absent
and self._enable_sticky_events
):
if self._enable_sticky_events:
sticky_events_coro = self.get_sticky_events_extension_response(
sync_config=sync_config,
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
actual_lists=actual_lists,
actual_room_ids=actual_room_ids,
actual_room_response_map=actual_room_response_map,
newly_joined_room_ids=newly_joined_room_ids,
sticky_events_request=sync_config.extensions.sticky_events,
all_interested_room_ids=all_interested_room_ids,
to_token=to_token,
from_token=from_token,
)
@@ -255,8 +266,8 @@ class SlidingSyncExtensionHandler:
def find_relevant_room_ids_for_extension(
self,
requested_lists: StrCollection | None,
requested_room_ids: StrCollection | None,
requested_lists: StrCollection,
requested_room_ids: StrCollection,
actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList],
actual_room_ids: AbstractSet[str],
) -> set[str]:
@@ -285,36 +296,23 @@ class SlidingSyncExtensionHandler:
relevant_room_ids: set[str] = set()
# See what rooms from the room subscriptions we should get account data for
if requested_room_ids is not None:
for room_id in requested_room_ids:
# A wildcard means we process all rooms from the room subscriptions
if room_id == "*":
relevant_room_ids.update(actual_room_ids)
break
for room_id in requested_room_ids:
# A wildcard means we process all rooms from the room subscriptions
if room_id == "*":
relevant_room_ids.update(actual_room_ids)
break
if room_id in actual_room_ids:
relevant_room_ids.add(room_id)
if room_id in actual_room_ids:
relevant_room_ids.add(room_id)
# See what rooms from the sliding window lists we should get account data for
if requested_lists is not None:
for list_key in requested_lists:
# Just some typing because we share the variable name in multiple places
actual_list: SlidingSyncResult.SlidingWindowList | None = None
for list_key in requested_lists:
# Just some typing because we share the variable name in multiple places
actual_list: SlidingSyncResult.SlidingWindowList | None = None
# A wildcard means we process rooms from all lists
if list_key == "*":
for actual_list in actual_lists.values():
# We only expect a single SYNC operation for any list
assert len(actual_list.ops) == 1
sync_op = actual_list.ops[0]
assert sync_op.op == OperationType.SYNC
relevant_room_ids.update(sync_op.room_ids)
break
actual_list = actual_lists.get(list_key)
if actual_list is not None:
# A wildcard means we process rooms from all lists
if list_key == "*":
for actual_list in actual_lists.values():
# We only expect a single SYNC operation for any list
assert len(actual_list.ops) == 1
sync_op = actual_list.ops[0]
@@ -322,6 +320,17 @@ class SlidingSyncExtensionHandler:
relevant_room_ids.update(sync_op.room_ids)
break
actual_list = actual_lists.get(list_key)
if actual_list is not None:
# We only expect a single SYNC operation for any list
assert len(actual_list.ops) == 1
sync_op = actual_list.ops[0]
assert sync_op.op == OperationType.SYNC
relevant_room_ids.update(sync_op.room_ids)
return relevant_room_ids
@trace
@@ -1025,71 +1034,167 @@ class SlidingSyncExtensionHandler:
async def get_sticky_events_extension_response(
self,
sync_config: SlidingSyncConfig,
sticky_events_request: SlidingSyncConfig.Extensions.StickyEventsExtension,
all_interested_room_ids: set[str],
previous_connection_state: "PerConnectionState",
new_connection_state: "MutablePerConnectionState",
actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList],
actual_room_ids: set[str],
actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult],
newly_joined_room_ids: AbstractSet[str],
sticky_events_request: (
SlidingSyncConfig.Extensions.StickyEventsExtension | AbsentType
),
to_token: StreamToken,
from_token: SlidingSyncStreamToken | None,
) -> SlidingSyncResult.Extensions.StickyEventsExtension | None:
if not sticky_events_request.enabled:
"""Handle Sticky Events extension (MSC4354 / MSC4480)
Args:
sync_config: Sync configuration
previous_connection_state: The current per-connection state
new_connection_state: A mutable copy of the per-connection
state, used to record updates to the state.
actual_lists: Sliding window API. A map of list key to list results in the
Sliding Sync response.
actual_room_ids: The actual room IDs in the the Sliding Sync response.
actual_room_response_map: A map of room ID to room results in the the
Sliding Sync response.
newly_joined_room_ids: The rooms the user newly joined in the token range.
sticky_events_request: The sticky_events extension from the request,
or absent if the request didn't include it.
to_token: The point in the stream to sync up to.
from_token: The point in the stream to sync from.
"""
if sticky_events_request is Absent or not sticky_events_request.enabled:
# While the extension is disabled, no room is in scope for it. We still
# keep track of which rooms gain sticky events in the meantime, so that
# when the extension is re-enabled the client receives the (unexpired)
# sticky events sent while it was disabled, rather than only those
# sent after the request's `pos`.
await self._record_sticky_events_in_out_of_scope_rooms(
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
relevant_room_ids=set(),
to_token=to_token,
from_token=from_token,
)
return None
now = self.clock.time_msec()
# 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_token,
room_to_event_ids,
) = await self.store.get_sticky_events_in_rooms(
all_interested_room_ids,
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),
relevant_room_ids = self.find_relevant_room_ids_for_extension(
requested_lists=sticky_events_request.lists,
requested_room_ids=sticky_events_request.rooms,
actual_lists=actual_lists,
actual_room_ids=actual_room_ids,
)
if not relevant_room_ids:
# No rooms are relevant, so shortcut out after recording
# any sticky events we're missing in out-of-scope rooms.
await self._record_sticky_events_in_out_of_scope_rooms(
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
relevant_room_ids=set(),
to_token=to_token,
from_token=from_token,
)
return SlidingSyncResult.Extensions.StickyEventsExtension(
room_id_to_sticky_events={},
)
room_id_to_sticky_events: dict[str, list[FilteredEvent]] = {}
# We need to handle the different cases depending on if we have sent
# down sticky events previously or not, so we split the relevant rooms
# up into different collections based on status.
initial_rooms: set[str] = set()
previously_rooms: dict[str, MultiWriterStreamToken] = {}
live_rooms: set[str] = set()
for room_id in relevant_room_ids:
if not from_token or room_id in newly_joined_room_ids:
initial_rooms.add(room_id)
continue
# If we're sending down the room from scratch again for some
# reason, we should also resend all the sticky events.
room_result = actual_room_response_map.get(room_id)
if room_result is not None and room_result.initial:
initial_rooms.add(room_id)
continue
room_status = previous_connection_state.sticky_events.have_sent_room(
room_id
)
if room_status.status == HaveSentRoomFlag.LIVE:
live_rooms.add(room_id)
elif room_status.status == HaveSentRoomFlag.PREVIOUSLY:
assert room_status.last_token is not None
previously_rooms[room_id] = room_status.last_token
elif room_status.status == HaveSentRoomFlag.NEVER:
initial_rooms.add(room_id)
else:
assert_never(room_status.status)
# Build the groups of rooms to read sticky events for and their stream
# position to read from.
# [(set of appliable rooms, from_token), ...]
groups: list[tuple[AbstractSet[str], MultiWriterStreamToken]] = []
if initial_rooms:
# Read from the start of the stream to get all unexpired sticky
# events.
groups.append((initial_rooms, StreamToken.START.sticky_events_key))
previously_rooms_by_token: dict[MultiWriterStreamToken, set[str]] = {}
for room_id, last_token in previously_rooms.items():
previously_rooms_by_token.setdefault(last_token, set()).add(room_id)
groups.extend(
(room_ids, last_token)
for last_token, room_ids in previously_rooms_by_token.items()
)
if live_rooms:
assert from_token is not None
groups.append((live_rooms, from_token.stream_token.sticky_events_key))
now = self.clock.time_msec()
# Now build a list of events to send, limited by the requested limit
remaining_limit = min(
sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC
)
room_to_event_ids: dict[str, list[str]] = {}
for room_ids, group_from_token in groups:
if remaining_limit <= 0:
# We've run out of budget. Record that these rooms still need
# to be read from `group_from_token` on a subsequent request.
new_connection_state.sticky_events.record_partially_sent_rooms(
room_ids, group_from_token
)
continue
(
read_to_token,
group_room_to_event_ids,
) = await self.store.get_sticky_events_in_rooms(
room_ids,
from_token=group_from_token,
to_token=to_token.sticky_events_key,
now=now,
limit=remaining_limit,
)
num_read = 0
for room_id, event_ids in group_room_to_event_ids.items():
room_to_event_ids.setdefault(room_id, []).extend(event_ids)
num_read += len(event_ids)
if num_read >= remaining_limit:
# We hit the limit, so there may be more sticky events to send
# for these rooms. Record where we got up to so that the next
# request continues from there.
new_connection_state.sticky_events.record_partially_sent_rooms(
room_ids, read_to_token
)
else:
# We read everything up to `to_token` for these rooms.
new_connection_state.sticky_events.record_sent_rooms(room_ids)
remaining_limit -= num_read
# No need to preserve sticky event order here because we will
# reassemble it in the right order after.
all_sticky_event_ids = {
@@ -1103,7 +1208,6 @@ class SlidingSyncExtensionHandler:
)
filtered_event_map = {ev.event.event_id: ev for ev in filtered_events}
room_id_to_sticky_events: dict[str, list[FilteredEvent]] = {}
for room_id, sticky_event_ids in room_to_event_ids.items():
filtered_events_for_room = [
filtered_event_map[event_id]
@@ -1117,11 +1221,60 @@ class SlidingSyncExtensionHandler:
room_id_to_sticky_events[room_id] = filtered_events_for_room
await self._record_sticky_events_in_out_of_scope_rooms(
previous_connection_state=previous_connection_state,
new_connection_state=new_connection_state,
relevant_room_ids=relevant_room_ids,
to_token=to_token,
from_token=from_token,
)
return SlidingSyncResult.Extensions.StickyEventsExtension(
room_id_to_sticky_events=room_id_to_sticky_events,
next_batch=await SlidingSyncStickyEventsToken.from_stream_token(
self.store, sticky_events_to_token
),
)
async def _record_sticky_events_in_out_of_scope_rooms(
self,
previous_connection_state: "PerConnectionState",
new_connection_state: "MutablePerConnectionState",
relevant_room_ids: AbstractSet[str],
to_token: StreamToken,
from_token: SlidingSyncStreamToken | None,
) -> None:
"""Record which rooms (may) have sticky events that we're not sending
down because they are not in scope for the sticky events extension, so
that we can send them when the room comes back into scope.
We only need to check rooms that are `LIVE`: `PREVIOUSLY` rooms keep
pointing at their previous position and `NEVER` rooms will get everything
when they come into scope.
Args:
previous_connection_state: The current per-connection state
new_connection_state: A mutable copy of the per-connection
state, used to record updates to the state.
relevant_room_ids: The rooms that are in scope for the extension
(empty if the extension is disabled).
to_token: The point in the stream to sync up to.
from_token: The point in the stream to sync from.
"""
if from_token is None:
# No previous connection state, so nothing to record.
return
rooms_out_of_scope = [
room_id
for room_id, room_status in previous_connection_state.sticky_events._statuses.items()
if room_status.status == HaveSentRoomFlag.LIVE
and room_id not in relevant_room_ids
]
changed_rooms = await self.store.get_rooms_with_sticky_events_between(
rooms_out_of_scope,
from_token=from_token.stream_token.sticky_events_key,
to_token=to_token.sticky_events_key,
)
new_connection_state.sticky_events.record_unsent_rooms(
changed_rooms, from_token.stream_token.sticky_events_key
)
async def _get_profile_ids_for_profiles_extension(
synapse/rest/client/sync.py
-1
View File
@@ -1248,7 +1248,6 @@ class SlidingSyncRestServlet(RestServlet):
return {
"rooms": rooms_out,
"next_batch": sticky_events.next_batch.serialise(),
}
synapse/storage/databases/main/sliding_sync.py
+41
View File
@@ -374,6 +374,15 @@ class SlidingSyncStore(SQLBaseStore):
(have_sent_room.status.value, have_sent_room.last_token)
)
for (
room_id,
have_sent_room,
) in per_connection_state.sticky_events._statuses.items():
key_values.append((connection_position, "sticky_events", room_id))
value_values.append(
(have_sent_room.status.value, have_sent_room.last_token)
)
self.db_pool.simple_upsert_many_txn(
txn,
table="sliding_sync_connection_streams",
@@ -596,6 +605,7 @@ class SlidingSyncStore(SQLBaseStore):
rooms: dict[str, HaveSentRoom[str]] = {}
receipts: dict[str, HaveSentRoom[str]] = {}
account_data: dict[str, HaveSentRoom[str]] = {}
sticky_events: dict[str, HaveSentRoom[str]] = {}
receipt_rows = self.db_pool.simple_select_list_txn(
txn,
@@ -618,6 +628,8 @@ class SlidingSyncStore(SQLBaseStore):
receipts[room_id] = have_sent_room
elif stream == "account_data":
account_data[room_id] = have_sent_room
elif stream == "sticky_events":
sticky_events[room_id] = have_sent_room
else:
# For forwards compatibility we ignore unknown streams, as in
# future we want to be able to easily add more stream types.
@@ -628,6 +640,7 @@ class SlidingSyncStore(SQLBaseStore):
rooms=RoomStatusMap(rooms),
receipts=RoomStatusMap(receipts),
account_data=RoomStatusMap(account_data),
sticky_events=RoomStatusMap(sticky_events),
room_configs=room_configs,
room_lazy_membership={},
)
@@ -810,6 +823,7 @@ class PerConnectionStateDB:
rooms: "RoomStatusMap[str]"
receipts: "RoomStatusMap[str]"
account_data: "RoomStatusMap[str]"
sticky_events: "RoomStatusMap[str]"
room_configs: Mapping[str, "RoomSyncConfig"]
@@ -856,11 +870,24 @@ class PerConnectionStateDB:
for room_id, status in per_connection_state.account_data.get_updates().items()
}
sticky_events = {
room_id: HaveSentRoom(
status=status.status,
last_token=(
await status.last_token.to_string(store)
if status.last_token is not None
else None
),
)
for room_id, status in per_connection_state.sticky_events.get_updates().items()
}
log_kv(
{
"rooms": rooms,
"receipts": receipts,
"account_data": account_data,
"sticky_events": sticky_events,
"room_configs": per_connection_state.room_configs.maps[0],
}
)
@@ -870,6 +897,7 @@ class PerConnectionStateDB:
rooms=RoomStatusMap(rooms),
receipts=RoomStatusMap(receipts),
account_data=RoomStatusMap(account_data),
sticky_events=RoomStatusMap(sticky_events),
room_configs=per_connection_state.room_configs.maps[0],
room_lazy_membership=per_connection_state.room_lazy_membership,
)
@@ -910,10 +938,23 @@ class PerConnectionStateDB:
for room_id, status in self.account_data._statuses.items()
}
sticky_events = {
room_id: HaveSentRoom(
status=status.status,
last_token=(
await MultiWriterStreamToken.parse(store, status.last_token)
if status.last_token is not None
else None
),
)
for room_id, status in self.sticky_events._statuses.items()
}
return PerConnectionState(
last_used_ts=self.last_used_ts,
rooms=RoomStatusMap(rooms),
receipts=RoomStatusMap(receipts),
account_data=RoomStatusMap(account_data),
sticky_events=RoomStatusMap(sticky_events),
room_configs=self.room_configs,
)
synapse/storage/databases/main/sticky_events.py
+51
View File
@@ -273,6 +273,57 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
)
return cast(list[tuple[int, str, str]], txn.fetchall())
async def get_rooms_with_sticky_events_between(
self,
room_ids: Collection[str],
*,
from_token: MultiWriterStreamToken,
to_token: MultiWriterStreamToken,
) -> set[str]:
"""Given a set of room IDs, find out which of them have sticky events
in the `sticky_events` table (which may be expired and awaiting cleanup)
with sticky stream positions after `from_token` and at or before `to_token`.
Bounds: from_token < ... <= to_token
Args:
room_ids: The room IDs to check.
from_token: The sticky stream position to check from (exclusive).
to_token: The sticky stream position to check to (inclusive).
Returns:
The subset of `room_ids` that have sticky events in the range.
"""
if len(room_ids) == 0 or to_token.is_before_or_eq(from_token):
return set()
def _get_rooms_with_sticky_events_between_txn(
txn: LoggingTransaction,
) -> set[str]:
room_id_in_list_clause, room_id_in_list_values = make_in_list_sql_clause(
txn.database_engine, "room_id", room_ids
)
stream_clause, stream_values = make_multiwriter_sharded_token_bounds_sql(
self.database_engine,
stream_id_column="stream_id",
instance_name_column="instance_name",
from_token_exclusive=from_token,
to_token_inclusive=to_token,
)
txn.execute(
f"""
SELECT DISTINCT room_id
FROM sticky_events
WHERE {stream_clause} AND {room_id_in_list_clause}
""",
(*stream_values, *room_id_in_list_values),
)
return {room_id for (room_id,) in txn}
return await self.db_pool.runInteraction(
"get_rooms_with_sticky_events_between",
_get_rooms_with_sticky_events_between_txn,
)
async def get_updated_sticky_events(
self, *, from_id: int, to_id: int, limit: int
) -> list[StickyEventUpdate]:
synapse/types/handlers/sliding_sync.py
+25 -3
View File
@@ -48,7 +48,7 @@ from synapse.types import (
ThreadSubscriptionsToken,
UserID,
)
from synapse.types.rest.client import SlidingSyncBody, SlidingSyncStickyEventsToken
from synapse.types.rest.client import SlidingSyncBody
from synapse.util.clock import Clock
from synapse.util.duration import Duration
@@ -426,7 +426,7 @@ class SlidingSyncResult:
@attr.s(slots=True, frozen=True, auto_attribs=True)
class StickyEventsExtension:
"""The Sticky Events extension (MSC4354)
"""The Sticky Events extension (MSC4354 / MSC4480)
Attributes:
room_id_to_sticky_events: map (room_id -> [unexpired_sticky_events])
@@ -437,7 +437,6 @@ class SlidingSyncResult:
"""
room_id_to_sticky_events: Mapping[str, list[FilteredEvent]]
next_batch: SlidingSyncStickyEventsToken
def __bool__(self) -> bool:
return bool(self.room_id_to_sticky_events)
@@ -772,8 +771,10 @@ class HaveSentRoomFlag(Enum):
The valid state changes here are:
NEVER -> LIVE
NEVER -> PREVIOUSLY (when only part of the room's data was sent, e.g. due to a limit)
LIVE -> PREVIOUSLY
PREVIOUSLY -> LIVE
PREVIOUSLY -> PREVIOUSLY (with a later token, when more of the data was sent)
"""
# The room has never been sent down (or we have forgotten we have sent it
@@ -889,6 +890,20 @@ class MutableRoomStatusMap(RoomStatusMap[T]):
self._statuses[room_id] = HaveSentRoom.live()
def record_partially_sent_rooms(
self, room_ids: StrCollection, last_token: T
) -> None:
"""Record that we have sent these rooms in the response, but only up to
`last_token` (exclusive of anything after it).
Used for Sticky Events, which are implicitly paginated by the server.
But this is not necessarily a good pattern to follow for future work,
so consider alternatives.
See: https://github.com/matrix-org/matrix-spec-proposals/pull/4480#discussion_r4165887827
"""
for room_id in room_ids:
self._statuses[room_id] = HaveSentRoom.previously(last_token)
def record_unsent_rooms(self, room_ids: StrCollection, from_token: T) -> None:
"""Record that we have not sent these rooms in the response, but there
have been updates.
@@ -935,6 +950,7 @@ class PerConnectionState:
room_configs: Map from room_id to the `RoomSyncConfig` of all
rooms that we have previously sent down.
account_data: The status of each room for the account_data stream.
sticky_events: The status of each room for the sticky events stream.
"""
last_used_ts: int | None = None
@@ -942,6 +958,7 @@ class PerConnectionState:
rooms: RoomStatusMap[RoomStreamToken] = attr.Factory(RoomStatusMap)
receipts: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap)
account_data: RoomStatusMap[int] = attr.Factory(RoomStatusMap)
sticky_events: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap)
room_configs: Mapping[str, RoomSyncConfig] = attr.Factory(dict)
@@ -954,6 +971,7 @@ class PerConnectionState:
rooms=self.rooms.get_mutable(),
receipts=self.receipts.get_mutable(),
account_data=self.account_data.get_mutable(),
sticky_events=self.sticky_events.get_mutable(),
room_configs=ChainMap({}, room_configs),
)
@@ -963,6 +981,7 @@ class PerConnectionState:
rooms=self.rooms.copy(),
receipts=self.receipts.copy(),
account_data=self.account_data.copy(),
sticky_events=self.sticky_events.copy(),
room_configs=dict(self.room_configs),
)
@@ -971,6 +990,7 @@ class PerConnectionState:
len(self.account_data)
+ len(self.rooms)
+ len(self.receipts)
+ len(self.sticky_events)
+ len(self.room_configs)
)
@@ -1047,6 +1067,7 @@ class MutablePerConnectionState(PerConnectionState):
rooms: MutableRoomStatusMap[RoomStreamToken]
receipts: MutableRoomStatusMap[MultiWriterStreamToken]
account_data: MutableRoomStatusMap[int]
sticky_events: MutableRoomStatusMap[MultiWriterStreamToken]
room_configs: ChainMap[str, RoomSyncConfig]
@@ -1067,6 +1088,7 @@ class MutablePerConnectionState(PerConnectionState):
bool(self.rooms.get_updates())
or bool(self.receipts.get_updates())
or bool(self.account_data.get_updates())
or bool(self.sticky_events.get_updates())
or bool(self.get_room_config_updates())
or any(
change.has_updates(clock)
synapse/types/rest/client/__init__.py
+26 -131
View File
@@ -18,14 +18,10 @@
# [This file includes modifications made by New Vector Limited]
#
#
import re
from typing import TYPE_CHECKING, ClassVar
import pydantic_core.core_schema
from pydantic import (
ConfigDict,
Field,
GetCoreSchemaHandler,
GetPydanticSchema,
StrictBool,
StrictInt,
@@ -34,22 +30,12 @@ from pydantic import (
field_validator,
model_validator,
)
from pydantic_core import CoreSchema
from typing_extensions import Annotated, Self
from synapse.types import (
Absent,
AbsentType,
MultiWriterStreamToken,
NonNegativeStrictInt,
UnresolvedMultiWriterStreamToken,
)
from synapse.types import Absent, AbsentType, NonNegativeStrictInt
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):
"""
@@ -142,104 +128,6 @@ class MsisdnRequestTokenBody(ThreepidRequestTokenBody):
phone_number: StrictStr
class SlidingSyncStickyEventsToken:
"""
A token returned by `next_batch` of the MSC4354 Sticky Events extension to Sliding Sync
and then accepted as the `since` parameter in the requests of the same extension.
Current format:
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]+|m[0-9]+(~[0-9]+\.[0-9]+)*)$")
START: ClassVar["SlidingSyncStickyEventsToken"]
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__(
cls, source_type: object, handler: GetCoreSchemaHandler
) -> CoreSchema:
"""
This function is checked for and used by Pydantic when
attempting to deserialise/validate a field of this type.
This returns a schema that will parse a string into an
instance of `SlidingSyncStickyEventsToken`.
"""
return pydantic_core.core_schema.no_info_plain_validator_function(
cls._validate,
serialization=pydantic_core.core_schema.plain_serializer_function_ser_schema(
cls.serialise,
info_arg=False,
),
)
@classmethod
def _validate(cls, v: object) -> Self:
"""
Create an instance from serialised string form.
The inverse of `serialise`.
"""
if isinstance(v, cls):
return v
if isinstance(v, str):
match = cls.PATTERN.match(v)
if match is None:
raise ValueError(f"Invalid SlidingSyncStickyEventsToken format: {v!r}")
return cls(
unresolved_stream_token=UnresolvedMultiWriterStreamToken.parse(
match.group(1)
)
)
raise ValueError(f"Cannot parse SlidingSyncStickyEventsToken from {type(v)}")
def serialise(self) -> str:
"""
Convert this instance to string.
The inverse of `_validate`.
"""
return f"sticky_{self._unresolved_stream_token.to_string()}"
def __repr__(self) -> str:
# Use the serialised form as debug output.
return self.serialise()
# Starting reading a stream at 0 ensures all stream fact rows will be read
SlidingSyncStickyEventsToken.START = SlidingSyncStickyEventsToken(
unresolved_stream_token=UnresolvedMultiWriterStreamToken(stream=0)
)
class SlidingSyncBody(RequestBodyModel):
"""
Sliding Sync API request body.
@@ -452,10 +340,10 @@ class SlidingSyncBody(RequestBodyModel):
enabled
"""
enabled: StrictBool | None = False
enabled: StrictBool = False
class AccountDataExtension(RequestBodyModel):
"""The Account Data extension (MSC3959)
"""The Account Data extension (MSC4548, previously MSC3959)
Attributes:
enabled
@@ -465,14 +353,14 @@ class SlidingSyncBody(RequestBodyModel):
extension to.
"""
enabled: StrictBool | None = False
enabled: StrictBool = False
# Process all lists defined in the Sliding Window API. (This is the default.)
lists: list[StrictStr] | None = ["*"]
lists: list[StrictStr] = ["*"]
# Process all room subscriptions defined in the Room Subscription API. (This is the default.)
rooms: list[StrictStr] | None = ["*"]
rooms: list[StrictStr] = ["*"]
class ReceiptsExtension(RequestBodyModel):
"""The Receipts extension (MSC3960)
"""The Receipts extension (MSC4551, previously MSC3960)
Attributes:
enabled
@@ -482,14 +370,14 @@ class SlidingSyncBody(RequestBodyModel):
extension to.
"""
enabled: StrictBool | None = False
enabled: StrictBool = False
# Process all lists defined in the Sliding Window API. (This is the default.)
lists: list[StrictStr] | None = ["*"]
lists: list[StrictStr] = ["*"]
# Process all room subscriptions defined in the Room Subscription API. (This is the default.)
rooms: list[StrictStr] | None = ["*"]
rooms: list[StrictStr] = ["*"]
class TypingExtension(RequestBodyModel):
"""The Typing Notification extension (MSC3961)
"""The Typing Notification extension (MSC4508, previously MSC3961)
Attributes:
enabled
@@ -499,11 +387,11 @@ class SlidingSyncBody(RequestBodyModel):
extension to.
"""
enabled: StrictBool | None = False
enabled: StrictBool = False
# Process all lists defined in the Sliding Window API. (This is the default.)
lists: list[StrictStr] | None = ["*"]
lists: list[StrictStr] = ["*"]
# Process all room subscriptions defined in the Room Subscription API. (This is the default.)
rooms: list[StrictStr] | None = ["*"]
rooms: list[StrictStr] = ["*"]
class ThreadSubscriptionsExtension(RequestBodyModel):
"""The Thread Subscriptions extension (MSC4308)
@@ -513,21 +401,28 @@ class SlidingSyncBody(RequestBodyModel):
limit: maximum number of subscription changes to return (default 100)
"""
enabled: StrictBool | None = False
enabled: StrictBool = False
limit: StrictInt = 100
class StickyEventsExtension(RequestBodyModel):
"""The Sticky Events extension (MSC4354)
"""The Sticky Events extension (MSC4354 / MSC4480)
Attributes:
enabled
limit: maximum number of sticky events to return in the extension (default 100)
since: either a string with the Sticky Events since token or absent
limit: maximum number of sticky events to return in the extension
(default 100). The server may cap this to a lower number.
lists: List of list keys (from the Sliding Window API) to apply this
extension to.
rooms: List of room IDs (from the Room Subscription API) to apply this
extension to.
"""
enabled: StrictBool = False
limit: NonNegativeStrictInt = 100
since: SlidingSyncStickyEventsToken | AbsentType = Absent
# Process all lists defined in the Sliding Window API. (This is the default.)
lists: list[StrictStr] = ["*"]
# Process all room subscriptions defined in the Room Subscription API. (This is the default.)
rooms: list[StrictStr] = ["*"]
class ProfilesExtension(RequestBodyModel):
"""The Profile Updates extension (MSC4262)
tests/replication/test_sharded_sticky_events.py
+8 -11
View File
@@ -22,7 +22,7 @@ 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.types import JsonDict, SlidingSyncStreamToken
from synapse.util.clock import Clock
from synapse.util.duration import Duration
@@ -246,21 +246,18 @@ class ShardedStickyEventsTestCase(BaseMultiWorkerStreamTestCase):
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,
},
{"rooms": {self.room_id1: {"events": []}}},
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
)
# The sticky events position in `pos` should carry the per-writer
# positions, with worker1 ahead of the stalled worker2.
sticky_events_key = self.get_success(
SlidingSyncStreamToken.from_string(self.store, pos)
).stream_token.sticky_events_key
self.assertIn("worker1", sticky_events_key.instance_map, sticky_events_key)
# worker2 catches up, so an unsharded sticky events position is now free to
# advance past worker1's write.
tests/rest/client/sliding_sync/test_extension_sticky_events.py
+664 -95
View File
@@ -34,28 +34,28 @@ logger = logging.getLogger(__name__)
DUMMY_LISTS = {
"main": {
# Don't include any rooms in the top-N window
"ranges": [[0, 0]],
# Include (up to) the first 11 rooms in the window, but don't return any
# timeline events for them.
"ranges": [[0, 10]],
"required_state": [],
"timeline_limit": 0,
}
}
"""
Subscription lists that can be used in the Sliding Sync request `lists` field,
which sets up a subscription that is interested in all rooms but does not let any rooms into the window,
thus does not return any timelines.
which sets up a list whose window covers all of the (few) rooms used in these
tests, but does not return any timelines.
Sufficient to get sticky event updates as per MSC4354:
> The server MUST include sticky events across all rooms that would be matched by at least one subscription list
> (i.e. all rooms that the client is interested in), even if the room does not appear in top-N window for that
> subscription list at this time.
> Rooms that would not be matched by a list are not included, as this means the client is not interested
> in those rooms.
>
> — https://github.com/matrix-org/matrix-spec-proposals/pull/4354/changes#diff-d76bc1a1d612c6da37d024f5b57f7b8352939b8db8a7ee9c6b71c1a848359afdR213-R217
As per the common per-room extension semantics of MSC4508 (which MSC4480 follows),
only rooms within the range of a list (or room subscriptions) are in scope for
the sticky events extension.
"""
STICKY_EVENTS_EXT = "org.matrix.msc4354.sticky_events"
ROOM_SUBSCRIPTION = {"required_state": [], "timeline_limit": 0}
"""A room subscription that doesn't return any timeline events."""
class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
"""Tests for the sticky events sliding sync extension"""
@@ -90,7 +90,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
self,
response_body: JsonDict,
expected_events_by_room: dict[str, list[str]] | None,
) -> str | None:
) -> None:
"""Assert the sliding sync response was successful and has the expected
sticky events.
@@ -99,20 +99,16 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
expected_events_by_room:
map of room ID to list of event IDs to expect (in the order we expect them),
or None if we expect an empty sticky events extension response
Returns the next_batch token from the sticky events section,
unless we're expecting an empty response.
"""
extensions = response_body["extensions"]
sticky_events = extensions.get("org.matrix.msc4354.sticky_events")
sticky_events = extensions.get(STICKY_EVENTS_EXT)
# If there are no expected events, we shouldn't get anything in the response
if expected_events_by_room is None:
self.assertIsNone(sticky_events)
return None
return
self.assertIsNotNone(sticky_events)
self.assertIsInstance(sticky_events["next_batch"], str)
actual_rooms = sticky_events["rooms"]
# Check that we have the expected rooms
@@ -131,8 +127,23 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
ttl = actual_event["unsigned"][EventUnsignedContentFields.STICKY_TTL]
self.assertIsInstance(ttl, int)
self.assertIn("next_batch", sticky_events)
return sticky_events["next_batch"]
def _send_sticky_event(
self,
room_id: str,
tok: str,
*,
duration: Duration = Duration(minutes=5),
body: str = "sticky message",
event_type: str = EventTypes.Message,
) -> str:
"""Send a sticky event and return its event ID."""
return self.helper.send_sticky_event(
room_id,
event_type,
duration=duration,
content={"body": body, "msgtype": "m.text"},
tok=tok,
)["event_id"]
def test_empty_sync(self) -> None:
"""Test that enabling sticky events extension works on initial and incremental sync,
@@ -144,7 +155,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -160,8 +171,8 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
self._assert_sticky_events_response(response_body, None)
def test_initial_sync(self) -> None:
"""Test that we get sticky events when we don't specify a since token
(initial sync).
"""Test that we get all unexpired sticky events on an initial sync (no
`pos`), and only new ones on subsequent incremental syncs.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
@@ -185,30 +196,18 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body: JsonDict = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
# Assert the response and then get the next_batch for the next sliding sync request
next_batch = self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id]}
)
assert next_batch is not None
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# Do an incremental sync immediately again
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
"since": next_batch,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
# Check we don't get that event again
self._assert_sticky_events_response(response_body, None)
@@ -223,7 +222,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
)["event_id"]
# Now the incremental sync should give us that event
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id2]}
)
@@ -252,7 +251,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -286,7 +285,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -343,7 +342,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -428,7 +427,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -512,7 +511,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
}
},
@@ -562,15 +561,13 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
# Initial sync
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 2}
},
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
# We expect to see the first 2 sticky events by stream order
# and they should be in that stream order
next_batch = self._assert_sticky_events_response(
self._assert_sticky_events_response(
response_body, {room_id: sticky_event_ids[0:2]}
)
@@ -578,30 +575,37 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
# This makes it incremental
"since": next_batch,
"limit": 3,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
# Should get remaining events, in stream order again
self._assert_sticky_events_response(
response_body, {room_id: sticky_event_ids[2:4]}
)
# And nothing more after that
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# A new sticky event is delivered as normal afterwards
sticky_event_id = self._send_sticky_event(room_id, user2_tok)
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
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.)
sticky events and omits the extension from the response, rather than erroring,
and that doing so doesn't cause any sticky events to be skipped.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
@@ -613,47 +617,42 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
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)
sticky_event_ids = [
self._send_sticky_event(room_id, user2_tok, body=f"sticky message {i}")
for i in range(2)
]
# Initial sync
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 1}
},
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 1}},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
# We expect to see the first sticky event
next_batch = self._assert_sticky_events_response(
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,
}
},
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "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)
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# The second sticky event is still delivered once the limit allows it
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 1}},
}
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(
response_body, {room_id: sticky_event_ids[1:2]}
)
def test_deduplication_with_timeline(self) -> None:
"""
@@ -710,7 +709,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
STICKY_EVENTS_EXT: {
"enabled": True,
"limit": 2,
}
@@ -718,9 +717,9 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
events_in_sticky_section = response_body["extensions"][
"org.matrix.msc4354.sticky_events"
]["rooms"][room_id]["events"]
events_in_sticky_section = response_body["extensions"][STICKY_EVENTS_EXT][
"rooms"
][room_id]["events"]
event_ids_in_sticky_section = [e["event_id"] for e in events_in_sticky_section]
events_in_timeline_section = response_body["rooms"][room_id]["timeline"]
@@ -735,3 +734,573 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
self.assertEqual(
event_ids_in_timeline_section, [sticky_event_ids[1], non_sticky_event_id]
)
def test_retry_with_same_pos_redelivers(self) -> None:
"""
Test that retrying a request with the same `pos` (e.g. because the
response was lost) delivers the same sticky events again.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True}},
}
_, from_token = self.do_sync(sync_body, tok=user1_tok)
sticky_event_id = self._send_sticky_event(room_id, user2_tok)
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# Retry with the same `pos`
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
def test_enable_extension_partway_through_connection(self) -> None:
"""
Test that enabling the extension for the first time partway through a
connection returns all unexpired sticky events for the rooms in scope,
including ones sent before the request's `pos`.
This allows clients to defer enabling the extension past their initial
request, as required by MSC4508.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
sticky_event_id = self._send_sticky_event(room_id, user2_tok)
# Initial sync and an incremental sync, both without the extension
sync_body: JsonDict = {"lists": DUMMY_LISTS}
_, from_token = self.do_sync(sync_body, tok=user1_tok)
_, from_token = self.do_sync(sync_body, since=from_token, tok=user1_tok)
# Now enable the extension: we should get the sticky event sent before
# the connection even started.
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True}},
}
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# But not again on the next request
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
def test_reenable_extension_after_disabling(self) -> None:
"""
Test that re-enabling the extension after having disabled it (or omitted
it) on the same connection returns the unexpired sticky events that were
sent while it was disabled, as well as those sent after the request's
`pos`.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
enabled_sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True}},
}
disabled_sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": False}},
}
omitted_sync_body: JsonDict = {"lists": DUMMY_LISTS}
_, from_token = self.do_sync(enabled_sync_body, tok=user1_tok)
# Disable the extension and send a sticky event while it is disabled
_, from_token = self.do_sync(
disabled_sync_body, since=from_token, tok=user1_tok
)
sticky_event_id1 = self._send_sticky_event(
room_id, user2_tok, body="while disabled"
)
response_body, from_token = self.do_sync(
disabled_sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Omit the extension altogether and send another sticky event, plus one
# that will have expired by the time the extension is re-enabled.
self._send_sticky_event(
room_id, user2_tok, duration=Duration(seconds=2), body="expires soon"
)
sticky_event_id2 = self._send_sticky_event(
room_id, user2_tok, body="while omitted"
)
response_body, from_token = self.do_sync(
omitted_sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
self.reactor.advance(3)
# Re-enable: the unexpired sticky events sent in the meantime are replayed
response_body, from_token = self.do_sync(
enabled_sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id1, sticky_event_id2]}
)
# But not again afterwards, and new sticky events are delivered as normal
response_body, from_token = self.do_sync(
enabled_sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
sticky_event_id3 = self._send_sticky_event(room_id, user2_tok)
response_body, _ = self.do_sync(
enabled_sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id3]}
)
def test_newly_joined_room_gets_all_unexpired_sticky_events(self) -> None:
"""
Test that when the user joins a room on an incremental sync, all unexpired
sticky events in the room are returned, including ones sent before `pos`.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
sticky_event_id1 = self._send_sticky_event(room_id, user2_tok, body="one")
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True}},
}
# user1 isn't in the room yet, so sees nothing
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
# user1 joins and sees the historical sticky event
self.helper.join(room_id, user1_id, tok=user1_tok)
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id1]}
)
# user1 leaves, another sticky event is sent, and user1 re-joins: they
# should get all unexpired sticky events in the room again.
self.helper.leave(room_id, user1_id, tok=user1_tok)
_, from_token = self.do_sync(sync_body, since=from_token, tok=user1_tok)
sticky_event_id2 = self._send_sticky_event(room_id, user2_tok, body="two")
self.helper.join(room_id, user1_id, tok=user1_tok)
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id1, sticky_event_id2]}
)
def test_scoping_by_lists_and_rooms(self) -> None:
"""
Test that the common per-room extension fields `lists` and `rooms`
(MSC4508) control which rooms sticky events are returned for.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
sticky_event_id = self._send_sticky_event(room_id, user2_tok)
# No lists in scope, no room subscriptions: nothing is in scope
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
STICKY_EVENTS_EXT: {"enabled": True, "lists": [], "rooms": []}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
# A list key that doesn't exist in the request is ignored
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
STICKY_EVENTS_EXT: {"enabled": True, "lists": ["other"], "rooms": []}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
# Naming the list brings the room into scope
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "lists": ["main"]}},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# A room subscription brings the room into scope, even with no lists
sync_body = {
"room_subscriptions": {room_id: ROOM_SUBSCRIPTION},
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "lists": []}},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# ...unless the subscription isn't named in `rooms`
sync_body = {
"room_subscriptions": {room_id: ROOM_SUBSCRIPTION},
"extensions": {
STICKY_EVENTS_EXT: {"enabled": True, "lists": [], "rooms": []}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
def test_room_outside_list_range_is_not_in_scope(self) -> None:
"""
Test that a room that matches a list but is outside the list's range is
not in scope for the extension, and that it gets all of its unexpired
sticky events when it comes into range.
"""
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")
room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_a, user1_id, tok=user1_tok)
room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_b, user1_id, tok=user1_tok)
# A sticky event in room B, and then a regular message in room A so that
# room A is the most recent room (and therefore the only one in a range
# of [0, 0]).
sticky_event_id = self._send_sticky_event(room_id_b, user2_tok)
self.helper.send(room_id_a, body="bump A", tok=user2_tok)
sync_body = {
"lists": {
"main": {
"ranges": [[0, 0]],
"required_state": [],
"timeline_limit": 0,
}
},
"extensions": {STICKY_EVENTS_EXT: {"enabled": True}},
}
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
self.assertEqual(
response_body["lists"]["main"]["ops"][0]["room_ids"], [room_id_a]
)
# Room B is out of range, so its sticky event is not returned
self._assert_sticky_events_response(response_body, None)
# Bump room B into range: we now get its (old) sticky event
self.helper.send(room_id_b, body="bump B", tok=user2_tok)
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self.assertEqual(
response_body["lists"]["main"]["ops"][0]["room_ids"], [room_id_b]
)
self._assert_sticky_events_response(
response_body, {room_id_b: [sticky_event_id]}
)
# And not again afterwards
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
def test_room_leaving_and_reentering_scope(self) -> None:
"""
Test that when a room drops out of scope and later comes back into scope,
the sticky events sent in the meantime are returned (unless they have
expired by then).
"""
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")
room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_a, user1_id, tok=user1_tok)
room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_b, user1_id, tok=user1_tok)
def sync_body_with_scope(room_ids: list[str]) -> JsonDict:
return {
"room_subscriptions": {
room_id_a: ROOM_SUBSCRIPTION,
room_id_b: ROOM_SUBSCRIPTION,
},
"extensions": {
STICKY_EVENTS_EXT: {
"enabled": True,
"lists": [],
"rooms": room_ids,
}
},
}
# Room A is in scope to begin with
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_a]), tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Room A drops out of scope
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Sticky events are sent in room A while it is out of scope
expiring_event_id = self._send_sticky_event(
room_id_a,
user2_tok,
duration=Duration(seconds=2),
body="expires soon",
)
sticky_event_id = self._send_sticky_event(
room_id_a, user2_tok, body="still sticky"
)
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Let the short-lived sticky event expire
self.reactor.advance(3)
# Room A comes back into scope: we get the sticky event sent while it was
# out of scope, but not the one that has expired since.
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id_a: [sticky_event_id]}
)
self.assertNotIn(
expiring_event_id,
[
e["event_id"]
for e in response_body["extensions"][STICKY_EVENTS_EXT]["rooms"][
room_id_a
]["events"]
],
)
# And not again afterwards
response_body, _ = self.do_sync(
sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
def test_pagination_across_rooms_entering_scope(self) -> None:
"""
Test that the `limit` bounds the response even when a room with many
unexpired sticky events comes into scope, with the remainder following on
subsequent requests in stream order, and that this doesn't lose track of
sticky events in other rooms that are out of scope in the meantime.
"""
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")
room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_a, user1_id, tok=user1_tok)
room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id_b, user1_id, tok=user1_tok)
def sync_body_with_scope(room_ids: list[str]) -> JsonDict:
return {
"room_subscriptions": {
room_id_a: ROOM_SUBSCRIPTION,
room_id_b: ROOM_SUBSCRIPTION,
},
"extensions": {
STICKY_EVENTS_EXT: {
"enabled": True,
"limit": 2,
"lists": [],
"rooms": room_ids,
}
},
}
# 3 sticky events in room B, which is out of scope to begin with
sticky_event_ids_b = [
self._send_sticky_event(room_id_b, user2_tok, body=f"B {i}")
for i in range(3)
]
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_a]), tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Room B comes into scope and room A drops out of scope, and a new sticky
# event arrives in room A (which must not be returned while out of scope).
sticky_event_id_a = self._send_sticky_event(room_id_a, user2_tok, body="A")
# First page: the oldest 2 of room B's sticky events
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id_b: sticky_event_ids_b[0:2]}
)
# Second page: the remaining one
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id_b: sticky_event_ids_b[2:3]}
)
# Nothing more after that
response_body, from_token = self.do_sync(
sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, None)
# Room A comes back into scope and we get the sticky event that was sent
# while it was out of scope.
response_body, _ = self.do_sync(
sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id_a: [sticky_event_id_a]}
)
def test_live_room_pagination(self) -> None:
"""
Test that `limit` also bounds the sticky events of rooms that are already
live on the connection, with the remainder following on subsequent
requests in stream order and nothing being lost or duplicated.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}},
}
# The room is now live on the connection
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
sticky_event_ids = [
self._send_sticky_event(room_id, user2_tok, body=f"sticky {i}")
for i in range(5)
]
# The sticky events arrive in pages of 2, in stream order
for page in (sticky_event_ids[0:2], sticky_event_ids[2:4]):
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(response_body, {room_id: page})
# A new sticky event sent in the meantime is picked up after the backlog
sticky_event_id = self._send_sticky_event(room_id, user2_tok, body="newest")
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_ids[4], sticky_event_id]}
)
# And then nothing more
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
def test_per_room_state_is_persisted(self) -> None:
"""
Test that the per-room sticky events position is persisted in the
per-connection state in the database (rather than only held in memory),
so that it survives restarts and is shared between workers.
"""
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")
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
def get_persisted_sticky_events_state(pos: str) -> list[tuple[str, str, str]]:
"""Returns (room_id, room_status, last_token) rows for the given `pos`."""
connection_position = int(pos.split("/", 1)[0])
return self.get_success(
self.store.db_pool.simple_select_list(
table="sliding_sync_connection_streams",
keyvalues={
"connection_position": connection_position,
"stream": "sticky_events",
},
retcols=("room_id", "room_status", "last_token"),
desc="get_persisted_sticky_events_state",
)
)
for _ in range(3):
self._send_sticky_event(room_id, user2_tok)
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}},
}
# The first page is cut short, so the room is recorded as partially sent,
# along with the position we got up to.
_, from_token = self.do_sync(sync_body, tok=user1_tok)
rows = get_persisted_sticky_events_state(from_token)
self.assertEqual(len(rows), 1, rows)
persisted_room_id, room_status, last_token = rows[0]
self.assertEqual(persisted_room_id, room_id)
self.assertEqual(room_status, "previously")
self.assertIsNotNone(last_token)
# After the remainder is sent, the room is recorded as live (and the
# persisted position was correctly read back to continue from).
response_body, from_token = self.do_sync(
sync_body, since=from_token, tok=user1_tok
)
self.assertEqual(
len(
response_body["extensions"][STICKY_EVENTS_EXT]["rooms"][room_id][
"events"
]
),
1,
)
rows = get_persisted_sticky_events_state(from_token)
self.assertEqual(rows, [(room_id, "live", None)])
tests/storage/test_sticky_events.py
+68
View File
@@ -299,6 +299,74 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
),
)
def test_get_rooms_with_sticky_events_between(self) -> None:
"""
Tests `get_rooms_with_sticky_events_between` returns only the rooms with
sticky events in the given stream range.
"""
other_room_id = self.helper.create_room_as(self.user_id, tok=self.token)
room_ids = [self.room_id, other_room_id]
start_token = self.store.get_sticky_events_stream_token()
self.helper.send_sticky_event(
self.room_id,
EventTypes.Message,
duration=Duration(minutes=1),
content={"body": "message 1", "msgtype": "m.text"},
tok=self.token,
)
mid_token = self.store.get_sticky_events_stream_token()
self.helper.send_sticky_event(
other_room_id,
EventTypes.Message,
duration=Duration(minutes=1),
content={"body": "message 2", "msgtype": "m.text"},
tok=self.token,
)
end_token = self.store.get_sticky_events_stream_token()
# Both rooms changed over the whole range
self.assertEqual(
self.get_success(
self.store.get_rooms_with_sticky_events_between(
room_ids, from_token=start_token, to_token=end_token
)
),
{self.room_id, other_room_id},
)
# Only the other room changed in the second half of the range
self.assertEqual(
self.get_success(
self.store.get_rooms_with_sticky_events_between(
room_ids, from_token=mid_token, to_token=end_token
)
),
{other_room_id},
)
# Nothing changed in an empty range
self.assertEqual(
self.get_success(
self.store.get_rooms_with_sticky_events_between(
room_ids, from_token=end_token, to_token=end_token
)
),
set(),
)
# Rooms we didn't ask about are not returned
self.assertEqual(
self.get_success(
self.store.get_rooms_with_sticky_events_between(
[self.room_id], from_token=start_token, to_token=end_token
)
),
{self.room_id},
)
def test_outlier_events_not_in_table(self) -> None:
"""
Tests the behaviour of outliered and then de-outliered events in the