dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 11:34:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into madlittlemods/rust-db-access-using-python-db-pool-run-interaction-llm1

Browse Source
This commit is contained in:
Eric Eastwood
2026-06-23 21:22:06 -05:00
parent c21dbbdc1f 1c79397d7f
commit ce68c0f73b
22 changed files with 1486 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+6 -6
View File
@@ -564,9 +564,9 @@ dependencies = [
[[package]]
name = "http"
version = "1.4.0"
version = "1.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a"
checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
dependencies = [
"bytes",
"itoa",
@@ -946,9 +946,9 @@ dependencies = [
[[package]]
name = "log"
version = "0.4.29"
version = "0.4.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a"
[[package]]
name = "lru-slab"
@@ -1692,9 +1692,9 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.149"
version = "1.0.150"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86"
checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
dependencies = [
"itoa",
"memchr",
changelog.d/19591.feature
+1
View File
@@ -0,0 +1 @@
Expose [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) over [MSC4186 (Simplified) Sliding Sync](https://github.com/matrix-org/matrix-spec-proposals/pull/4186).
changelog.d/19785.bugfix
+1
View File
@@ -0,0 +1 @@
Fix a long-standing bug where the badge notification count for a room could become permanently inflated if a read receipt was sent before the room's notification counts were first summarised.
changelog.d/19829.doc
+1
View File
@@ -0,0 +1 @@
Tweak wording of Rust crate dependency update policy.
docs/deprecation_policy.md
+14 -8
View File
@@ -79,11 +79,17 @@ concerned about the criteria for selecting minimum versions. The only thing of c
is making sure we're not making it unnecessarily difficult for downstream package
maintainers. Generally, this just means avoiding the bleeding edge for a few months.
The situation for Rust dependencies is fundamentally different. For packagers, the
concerns around Python dependency versions do not apply. The `cargo` tool handles
downloading and building all libraries to satisfy dependencies, and these libraries are
statically linked into the final binary. This means that from a packager's perspective,
the Rust dependency versions are an internal build detail, not a runtime dependency to
be managed on the target system. Consequently, we have even greater flexibility to
upgrade Rust dependencies as needed for the project. Some distros (e.g. Fedora) do
package Rust libraries, but this appears to be the outlier rather than the norm.
The situation for Rust dependencies is typically different and
the concerns around Python dependency versions typically do not apply.
For example, for packagers of Debian, the packagers have the choice of either
using a crate packaged in the distro, or vendoring crates in the source package for the
application.
This freedom to vendor a dependency crate for a specific application consequently gives
us even greater flexibility to upgrade Rust dependencies as needed for the project.
(This is in contrast with Python dependencies, which are generally
installed system-wide by mainstream distributions' official packages.)
Some distros (e.g. Fedora) do not vendor Rust dependencies in their
official application packages, but these cases appear to be less common.
synapse/handlers/delayed_events.py
+3 -5
View File
@@ -36,6 +36,7 @@ from synapse.storage.databases.main.delayed_events import (
)
from synapse.storage.databases.main.state_deltas import StateDelta
from synapse.types import (
Absent,
JsonDict,
Requester,
RoomID,
@@ -45,7 +46,6 @@ from synapse.types import (
from synapse.util.duration import Duration
from synapse.util.events import generate_fake_event_id
from synapse.util.metrics import Measure
from synapse.util.sentinel import Sentinel
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -273,9 +273,7 @@ class DelayedEventsHandler:
)
continue
sender_str = event_id_and_sender_dict.get(
delta.event_id, Sentinel.UNSET_SENTINEL
)
sender_str = event_id_and_sender_dict.get(delta.event_id, Absent)
if sender_str is None:
# An event exists, but the `sender` field was "null" and Synapse
# incorrectly accepted the event. This is not expected.
@@ -285,7 +283,7 @@ class DelayedEventsHandler:
delta.event_id,
)
continue
if sender_str is Sentinel.UNSET_SENTINEL:
if sender_str is Absent:
# We have an event ID, but the event was not found in the
# datastore. This can happen if a room, or its history, is
# purged. State deltas related to the room are left behind, but
synapse/handlers/sliding_sync/__init__.py
+2 -2
View File
@@ -287,7 +287,6 @@ class SlidingSyncHandler:
lists = interested_rooms.lists
relevant_room_map = interested_rooms.relevant_room_map
all_rooms = interested_rooms.all_rooms
room_membership_for_user_map = interested_rooms.room_membership_for_user_map
relevant_rooms_to_send_map = interested_rooms.relevant_rooms_to_send_map
@@ -328,6 +327,7 @@ 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
@@ -350,7 +350,7 @@ class SlidingSyncHandler:
if from_token:
# The set of rooms that the client (may) care about, but aren't
# in any list range (or subscribed to).
missing_rooms = all_rooms - relevant_room_map.keys()
missing_rooms = interested_rooms.all_rooms - relevant_room_map.keys()
# We now just go and try fetching any events in the above rooms
# to see if anything has happened since the `from_token`.
synapse/handlers/sliding_sync/extensions.py
+92 -4
View File
@@ -11,7 +11,6 @@
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
import itertools
import logging
from collections import ChainMap
@@ -26,11 +25,13 @@ from typing import (
from typing_extensions import TypeAlias, assert_never
from synapse.api.constants import AccountDataTypes, EduTypes
from synapse.api.constants import AccountDataTypes, EduTypes, StickyEvent
from synapse.events.utils import FilteredEvent
from synapse.handlers.receipts import ReceiptEventSource
from synapse.logging.opentracing import trace
from synapse.storage.databases.main.receipts import ReceiptInRoom
from synapse.types import (
Absent,
DeviceListUpdates,
JsonMapping,
MultiWriterStreamToken,
@@ -47,10 +48,12 @@ from synapse.types.handlers.sliding_sync import (
SlidingSyncConfig,
SlidingSyncResult,
)
from synapse.types.rest.client import SlidingSyncStickyEventsToken
from synapse.util.async_helpers import (
concurrently_execute,
gather_optional_coroutines,
)
from synapse.visibility import filter_and_transform_events_for_client
_ThreadSubscription: TypeAlias = (
SlidingSyncResult.Extensions.ThreadSubscriptionsExtension.ThreadSubscription
@@ -73,7 +76,10 @@ class SlidingSyncExtensionHandler:
self.event_sources = hs.get_event_sources()
self.device_handler = hs.get_device_handler()
self.push_rules_handler = hs.get_push_rules_handler()
self.clock = hs.get_clock()
self._storage_controllers = hs.get_storage_controllers()
self._enable_thread_subscriptions = hs.config.experimental.msc4306_enabled
self._enable_sticky_events = hs.config.experimental.msc4354_enabled
@trace
async def get_extensions_response(
@@ -81,6 +87,7 @@ 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],
@@ -91,9 +98,12 @@ class SlidingSyncExtensionHandler:
Args:
sync_config: Sync configuration
new_connection_state: Snapshot of the current per-connection state
new_per_connection_state: A mutable copy of the per-connection
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.
@@ -174,6 +184,19 @@ class SlidingSyncExtensionHandler:
from_token=from_token,
)
sticky_events_coro = None
if (
sync_config.extensions.sticky_events is not Absent
and self._enable_sticky_events
):
sticky_events_coro = self.get_sticky_events_extension_response(
sync_config=sync_config,
sticky_events_request=sync_config.extensions.sticky_events,
all_interested_room_ids=all_interested_room_ids,
to_token=to_token,
from_token=from_token,
)
(
to_device_response,
e2ee_response,
@@ -181,6 +204,7 @@ class SlidingSyncExtensionHandler:
receipts_response,
typing_response,
thread_subs_response,
sticky_events_response,
) = await gather_optional_coroutines(
to_device_coro,
e2ee_coro,
@@ -188,6 +212,7 @@ class SlidingSyncExtensionHandler:
receipts_coro,
typing_coro,
thread_subs_coro,
sticky_events_coro,
)
return SlidingSyncResult.Extensions(
@@ -197,6 +222,7 @@ class SlidingSyncExtensionHandler:
receipts=receipts_response,
typing=typing_response,
thread_subscriptions=thread_subs_response,
sticky_events=sticky_events_response,
)
def find_relevant_room_ids_for_extension(
@@ -967,3 +993,65 @@ class SlidingSyncExtensionHandler:
unsubscribed=unsubscribed_threads,
prev_batch=prev_batch,
)
async def get_sticky_events_extension_response(
self,
sync_config: SlidingSyncConfig,
sticky_events_request: SlidingSyncConfig.Extensions.StickyEventsExtension,
all_interested_room_ids: set[str],
to_token: StreamToken,
from_token: SlidingSyncStreamToken | None,
) -> SlidingSyncResult.Extensions.StickyEventsExtension | None:
if not sticky_events_request.enabled:
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
(
sticky_events_to_id,
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,
now=now,
limit=min(sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC),
)
# No need to preserve sticky event order here because we will
# reassemble it in the right order after.
all_sticky_event_ids = {
ev_id for evs in room_to_event_ids.values() for ev_id in evs
}
unfiltered_events = await self.store.get_events_as_list(all_sticky_event_ids)
filtered_events = await filter_and_transform_events_for_client(
self._storage_controllers,
sync_config.user.to_string(),
unfiltered_events,
# As per MSC4354:
# > History visibility checks MUST NOT be applied to sticky events.
# > Any joined user is authorised to see sticky events for the duration they remain sticky.
always_include_ids=frozenset(all_sticky_event_ids),
)
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]
# This reintroduces the correct order
# (by the sticky events stream)
for event_id in sticky_event_ids
if event_id in filtered_event_map
]
if len(filtered_events_for_room) == 0:
continue
room_id_to_sticky_events[room_id] = filtered_events_for_room
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
),
)
synapse/handlers/sliding_sync/room_lists.py
+41 -19
View File
@@ -52,6 +52,7 @@ from synapse.storage.roommember import (
RoomsForUserStateReset,
)
from synapse.types import (
Absent,
MutableStateMap,
RoomStreamToken,
StateMap,
@@ -71,7 +72,6 @@ from synapse.types.handlers.sliding_sync import (
from synapse.types.state import StateFilter
from synapse.util import MutableOverlayMapping
from synapse.util.duration import Duration
from synapse.util.sentinel import Sentinel
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -112,31 +112,55 @@ class SlidingSyncInterestedRooms:
sliding sync request.
Returned by `compute_interested_rooms`.
Attributes:
lists: A mapping from list name to the list result for the response
relevant_room_map: A map from rooms that match the sync request to
their room sync config.
relevant_rooms_to_send_map: Subset of `relevant_room_map` that
includes the rooms that *may* have relevant updates. Rooms not
in this map will definitely not have room updates (though
extensions may have updates in these rooms).
newly_joined_rooms: The set of rooms that were joined in the token range
and the user is still joined to at the end of this range.
newly_left_rooms: The set of rooms that we left in the token range
and are still "leave" at the end of this range.
dm_room_ids: The set of rooms the user consider as direct-message (DM) rooms
"""
lists: Mapping[str, SlidingSyncResult.SlidingWindowList]
"""
A mapping from list name to the list result for the response
"""
relevant_room_map: Mapping[str, RoomSyncConfig]
"""
A map from rooms that match the sync request to
their room sync config.
"""
relevant_rooms_to_send_map: Mapping[str, RoomSyncConfig]
"""
Subset of `relevant_room_map` that
includes the rooms that *may* have relevant updates. Rooms not
in this map will definitely not have room updates (though
extensions may have updates in these rooms).
"""
all_rooms: set[str]
"""
The set of room IDs of all rooms that could appear in any list.
This set includes rooms that are outside the list ranges.
In other words, this is the set of all rooms that the client is
_interested_ in (in a pure sense),
even if these rooms are omitted from the current window (which
is, in a sense, just a computational optimisation).
"""
room_membership_for_user_map: Mapping[str, RoomsForUserType]
newly_joined_rooms: AbstractSet[str]
"""
The set of rooms that were joined in the token range
and the user is still joined to at the end of this range.
"""
newly_left_rooms: AbstractSet[str]
"""
The set of rooms that we left in the token range
and are still "leave" at the end of this range.
"""
dm_room_ids: AbstractSet[str]
"""
The set of rooms the user consider as direct-message (DM) rooms
"""
@staticmethod
def empty() -> "SlidingSyncInterestedRooms":
@@ -1703,10 +1727,8 @@ class SlidingSyncRoomLists:
# (applies to invite/knock rooms)
rooms_ids_without_stripped_state: set[str] = set()
for room_id in room_ids_without_results:
stripped_state_map = room_id_to_stripped_state_map.get(
room_id, Sentinel.UNSET_SENTINEL
)
assert stripped_state_map is not Sentinel.UNSET_SENTINEL, (
stripped_state_map = room_id_to_stripped_state_map.get(room_id, Absent)
assert stripped_state_map is not Absent, (
f"Stripped state left unset for room {room_id}. "
+ "Make sure you're calling `_bulk_get_stripped_state_for_rooms_from_sync_room_map(...)` "
+ "with that room_id. (this is a problem with Synapse itself)"
synapse/module_api/__init__.py
+4 -3
View File
@@ -142,6 +142,8 @@ from synapse.storage.background_updates import (
from synapse.storage.database import DatabasePool, LoggingTransaction
from synapse.storage.databases.main.roommember import ProfileInfo
from synapse.types import (
Absent,
AbsentType,
DomainSpecificString,
JsonDict,
JsonMapping,
@@ -160,7 +162,6 @@ from synapse.util.caches.descriptors import CachedFunction, cached as _cached
from synapse.util.clock import Clock
from synapse.util.duration import Duration
from synapse.util.frozenutils import freeze
from synapse.util.sentinel import Sentinel
if TYPE_CHECKING:
# Old versions don't have `LiteralString`
@@ -1990,7 +1991,7 @@ class ModuleApi:
self,
user_id: UserID,
new_displayname: str,
deactivation: bool | Sentinel = Sentinel.UNSET_SENTINEL,
deactivation: bool | AbsentType = Absent,
) -> None:
"""Sets a user's display name.
@@ -2020,7 +2021,7 @@ class ModuleApi:
"""
requester = create_requester(user_id)
if deactivation is not Sentinel.UNSET_SENTINEL:
if deactivation is not Absent:
logger.error(
"Deprecated `deactivation` parameter passed to `set_displayname` Module API (value: %r). This will break in 2027.",
deactivation,
synapse/rest/client/sync.py
+86 -3
View File
@@ -20,7 +20,7 @@
#
import logging
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Mapping
from typing import TYPE_CHECKING, Any, Literal, Mapping
import attr
@@ -672,6 +672,7 @@ class SlidingSyncRestServlet(RestServlet):
- receipts (MSC3960)
- account data (MSC3959)
- thread subscriptions (MSC4308)
- sticky events (MSC4354)
Request query parameters:
timeout: How long to wait for new events in milliseconds.
@@ -895,7 +896,7 @@ class SlidingSyncRestServlet(RestServlet):
requester, sliding_sync_result.rooms
)
response["extensions"] = await self.encode_extensions(
requester, sliding_sync_result.extensions
requester, sliding_sync_result.extensions, sliding_sync_result.rooms
)
return response
@@ -1045,8 +1046,18 @@ class SlidingSyncRestServlet(RestServlet):
@trace_with_opname("sliding_sync.encode_extensions")
async def encode_extensions(
self, requester: Requester, extensions: SlidingSyncResult.Extensions
self,
requester: Requester,
extensions: SlidingSyncResult.Extensions,
ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult],
) -> JsonDict:
"""
Args:
ref_rooms_results:
Map of room ID -> RoomResult that was serialised as the `room` section
of the Sliding Sync response.
Will not be mutated, only used for reading.
"""
serialized_extensions: JsonDict = {}
if extensions.to_device is not None:
@@ -1115,8 +1126,80 @@ class SlidingSyncRestServlet(RestServlet):
_serialise_thread_subscriptions(extensions.thread_subscriptions)
)
if extensions.sticky_events:
serialized_extensions[
"org.matrix.msc4354.sticky_events"
] = await self._serialise_sticky_events(
requester, extensions.sticky_events, ref_rooms_results
)
return serialized_extensions
async def _serialise_sticky_events(
self,
requester: Requester,
sticky_events: SlidingSyncResult.Extensions.StickyEventsExtension,
ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult],
) -> JsonDict:
"""
Serialise the sticky events extension response.
This includes deduplicating by filtering out sticky events
from this extension that already appeared in the timeline
section.
Args:
ref_rooms_results:
Map of room ID -> RoomResult that was serialised as the `room` section
of the Sliding Sync response.
Will not be mutated, only used for reading.
"""
time_now = self.clock.time_msec()
# Same as SSS timelines.
#
serialize_options = SerializeEventConfig(
event_format=format_event_for_client_v2_without_room_id,
requester=requester,
)
rooms_out: dict[str, dict[Literal["events"], list[JsonDict]]] = {}
for (
room_id,
possibly_duplicated_sticky_events,
) in sticky_events.room_id_to_sticky_events.items():
# As per MSC4354:
# Remove sticky events that are already in the timeline, else we will needlessly duplicate
# events.
# There is no purpose in including sticky events in the sticky section if they're already in
# the timeline, as either way the client becomes aware of them.
# This is particularly important given the risk of sticky events spam since
# anyone can send sticky events, so halving the bandwidth on average for each sticky
# event is helpful.
room_result = ref_rooms_results.get(room_id)
if room_result is None:
# Nothing to deduplicate
sticky_events_to_write = possibly_duplicated_sticky_events
else:
sent_event_ids_in_room_section = {
ev.event.event_id for ev in room_result.timeline_events
}
sticky_events_to_write = [
ev
for ev in possibly_duplicated_sticky_events
if ev.event.event_id not in sent_event_ids_in_room_section
]
rooms_out[room_id] = {
"events": await self.event_serializer.serialize_events(
sticky_events_to_write, time_now, config=serialize_options
)
}
return {
"rooms": rooms_out,
"next_batch": sticky_events.next_batch.serialise(),
}
def _serialise_thread_subscriptions(
thread_subscriptions: SlidingSyncResult.Extensions.ThreadSubscriptionsExtension,
synapse/storage/databases/main/event_push_actions.py
+41 -2
View File
@@ -1509,6 +1509,43 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
"last_receipt_stream_ordering": stream_ordering,
},
)
# If no summary row exists yet for a thread that has pending push
# actions (room active but not yet through a rotation cycle), the
# UPDATE above is a silent no-op for that thread and
# last_receipt_stream_ordering is never persisted.
# _rotate_notifs_before_txn would then INSERT the row with
# last_receipt_stream_ordering=NULL, causing the badge query to
# include every event before the receipt as unread. Pre-populate
# rows for every thread with pending push actions so rotation
# only counts events that arrive after this receipt.
txn.execute(
"""
SELECT DISTINCT thread_id
FROM event_push_actions
WHERE user_id = ? AND room_id = ?
""",
(user_id, room_id),
)
pending_thread_ids = [row[0] for row in txn]
self.db_pool.simple_upsert_many_txn(
txn,
table="event_push_summary",
key_names=("user_id", "room_id", "thread_id"),
key_values=[
(user_id, room_id, pending_thread_id)
for pending_thread_id in pending_thread_ids
],
value_names=(
"notif_count",
"unread_count",
"stream_ordering",
"last_receipt_stream_ordering",
),
value_values=[
(0, 0, old_rotate_stream_ordering, stream_ordering)
for _ in pending_thread_ids
],
)
# For a threaded receipt, we *always* want to update that receipt,
# event if there are no new notifications in that thread. This ensures
@@ -1517,8 +1554,10 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
unread_counts = [(0, 0, thread_id)]
# Then any updated threads get their notification count and unread
# count updated.
self.db_pool.simple_update_many_txn(
# count updated. Use upsert so that a row is created if none exists
# yet (same race as the unthreaded case above: without this, rotation
# would INSERT with last_receipt_stream_ordering=NULL).
self.db_pool.simple_upsert_many_txn(
txn,
table="event_push_summary",
key_names=("room_id", "user_id", "thread_id"),
synapse/types/__init__.py
+110
View File
@@ -27,8 +27,10 @@ from enum import Enum
from typing import (
TYPE_CHECKING,
AbstractSet,
Annotated,
Any,
ClassVar,
Final,
Literal,
Mapping,
Match,
@@ -41,8 +43,12 @@ from typing import (
overload,
)
import annotated_types
import attr
import pydantic_core.core_schema
from immutabledict import immutabledict
from pydantic import GetCoreSchemaHandler, StrictInt
from pydantic_core import CoreSchema
from signedjson.key import decode_verify_key_bytes
from signedjson.types import VerifyKey
from typing_extensions import Self
@@ -109,6 +115,110 @@ StrCollection = tuple[str, ...] | list[str] | AbstractSet[str]
StrSequence = tuple[str, ...] | list[str]
class AbsentType(Enum):
"""
Type of a sentinel to use as an alternative to `None`
for when we really mean 'absent' and not JSON null.
Generally suitable for distinguishing a default state from user-suppliable values.
Has no meaning on its own.
It is falsy (like None is), so shorthand forms like `x or 0` can be used.
"""
# Making this an Enum member makes this compatible with type narrowing,
# meaning `x is not Absent` will narrow `x: int | AbsentType` to `x: int` etc.
_Absent = object()
@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.
As the `Absent` type has no valid value when deserialising
from JSON (as that's the point; `Absent` is a marker representing
a lack of any JSON value), we always reject any value.
Instead of deserialising from this type, we rely on the struct class
we are in having field defaults that provide an `Absent`, which does not
go through the JSON validation.
When validating Python, we accept the absent marker itself.
"""
def _reject_from_json(v: object) -> "AbsentType":
"""
Reject the JSON value, no matter what it is, since absent values
are meant to be ... absent, thus have nothing they can be deserialised
from.
"""
raise ValueError("AbsentType cannot be deserialized from JSON")
# `json_or_python_schema` wrapper needed for Pydantic < 2.10
# but can be replaced with just the `is_instance_schema` after that version.
return pydantic_core.core_schema.json_or_python_schema(
json_schema=pydantic_core.core_schema.no_info_plain_validator_function(
_reject_from_json
),
python_schema=pydantic_core.core_schema.is_instance_schema(cls),
)
def __copy__(self) -> "AbsentType":
"""
Copy implementation used by `copy.copy()`.
Always use the same instance.
Without this and the deep version `__deepcopy__`,
`copy.copy(Absent)` on Python 3.10 (olddeps)
had a problem where it tried to construct a new Absent
as part of a deepcopy operation and resulted in:
ValueError: <object object at 0x7f64b3b6d930> is not a valid AbsentType
"""
return self
def __deepcopy__(self, memo: object) -> "AbsentType":
"""
Copy implementation used by `copy.deepcopy()`.
Always use the same instance.
"""
return self
def __bool__(self) -> Literal[False]:
return False
def __str__(self) -> str:
return "Absent"
def __repr__(self) -> str:
return "Absent"
Absent: Final = AbsentType._Absent
"""
Sentinel to use as an alternative to `None`
for when we really mean 'absent' and not JSON null.
Generally suitable for distinguishing a default state from user-suppliable values.
Has no meaning on its own.
It is falsy (like None is), so shorthand forms like `x or 0` can be used.
(Previously known as `Sentinel.UNSET_SENTINEL`.)
"""
NonNegativeStrictInt = Annotated[StrictInt, annotated_types.Ge(0)]
"""A strict integer that must be greater than or equal to zero.
Should be preferred in place of Pydantic's own (lax) NonNegativeInt,
which will coerce strings to integers in a way that does not agree with
the Matrix specification (and would risk backing us into a backward compatibility
hole where we had to support input forms we didn't intend).
"""
# Note that this seems to require inheriting *directly* from Interface in order
# for mypy-zope to realize it is an interface.
class ISynapseThreadlessReactor(
synapse/types/handlers/sliding_sync.py
+21 -1
View File
@@ -48,7 +48,7 @@ from synapse.types import (
ThreadSubscriptionsToken,
UserID,
)
from synapse.types.rest.client import SlidingSyncBody
from synapse.types.rest.client import SlidingSyncBody, SlidingSyncStickyEventsToken
from synapse.util.clock import Clock
from synapse.util.duration import Duration
@@ -424,12 +424,31 @@ class SlidingSyncResult:
or bool(self.prev_batch)
)
@attr.s(slots=True, frozen=True, auto_attribs=True)
class StickyEventsExtension:
"""The Sticky Events extension (MSC4354)
Attributes:
room_id_to_sticky_events: map (room_id -> [unexpired_sticky_events])
The events are ordered by the sticky events stream.
The events haven't yet been deduplicated to remove
events that also appear in the timeline.
"""
room_id_to_sticky_events: Mapping[str, list[FilteredEvent]]
next_batch: SlidingSyncStickyEventsToken
def __bool__(self) -> bool:
return bool(self.room_id_to_sticky_events)
to_device: ToDeviceExtension | None = None
e2ee: E2eeExtension | None = None
account_data: AccountDataExtension | None = None
receipts: ReceiptsExtension | None = None
typing: TypingExtension | None = None
thread_subscriptions: ThreadSubscriptionsExtension | None = None
sticky_events: StickyEventsExtension | None = None
def __bool__(self) -> bool:
"""Are there any updates that should be returned immediately to
@@ -441,6 +460,7 @@ class SlidingSyncResult:
or self.receipts
or self.typing
or self.thread_subscriptions
or self.sticky_events
)
next_pos: SlidingSyncStreamToken
synapse/types/rest/client/__init__.py
+99 -1
View File
@@ -18,9 +18,14 @@
# [This file includes modifications made by New Vector Limited]
#
#
import re
from typing import ClassVar
import pydantic_core.core_schema
from pydantic import (
ConfigDict,
Field,
GetCoreSchemaHandler,
StrictBool,
StrictInt,
StrictStr,
@@ -28,9 +33,10 @@ from pydantic import (
field_validator,
model_validator,
)
from pydantic_core import PydanticCustomError
from pydantic_core import CoreSchema, PydanticCustomError
from typing_extensions import Annotated, Self
from synapse.types import Absent, AbsentType, NonNegativeStrictInt
from synapse.types.rest import RequestBodyModel
from synapse.util.threepids import validate_email
@@ -107,6 +113,82 @@ 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_' DIGIT+
DIGIT ::= '0'-'9'
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.
"""
PATTERN = re.compile(r"^sticky_([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
@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(sticky_events_stream_id=int(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.sticky_events_stream_id}"
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(
sticky_events_stream_id=0
)
class SlidingSyncBody(RequestBodyModel):
"""
Sliding Sync API request body.
@@ -383,6 +465,19 @@ class SlidingSyncBody(RequestBodyModel):
enabled: StrictBool | None = False
limit: StrictInt = 100
class StickyEventsExtension(RequestBodyModel):
"""The Sticky Events extension (MSC4354)
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
"""
enabled: StrictBool = False
limit: NonNegativeStrictInt = 100
since: SlidingSyncStickyEventsToken | AbsentType = Absent
to_device: ToDeviceExtension | None = None
e2ee: E2eeExtension | None = None
account_data: AccountDataExtension | None = None
@@ -391,6 +486,9 @@ class SlidingSyncBody(RequestBodyModel):
thread_subscriptions: ThreadSubscriptionsExtension | None = Field(
None, alias="io.element.msc4308.thread_subscriptions"
)
sticky_events: StickyEventsExtension | AbsentType = Field(
Absent, alias="org.matrix.msc4354.sticky_events"
)
conn_id: StrictStr | None = None
lists: (
synapse/util/async_helpers.py
+25
View File
@@ -390,6 +390,7 @@ T3 = TypeVar("T3")
T4 = TypeVar("T4")
T5 = TypeVar("T5")
T6 = TypeVar("T6")
T7 = TypeVar("T7")
@overload
@@ -519,6 +520,30 @@ async def gather_optional_coroutines(
) -> tuple[T1 | None, T2 | None, T3 | None, T4 | None, T5 | None, T6 | None]: ...
@overload
async def gather_optional_coroutines(
*coroutines: Unpack[
tuple[
Coroutine[Any, Any, T1] | None,
Coroutine[Any, Any, T2] | None,
Coroutine[Any, Any, T3] | None,
Coroutine[Any, Any, T4] | None,
Coroutine[Any, Any, T5] | None,
Coroutine[Any, Any, T6] | None,
Coroutine[Any, Any, T7] | None,
]
],
) -> tuple[
T1 | None,
T2 | None,
T3 | None,
T4 | None,
T5 | None,
T6 | None,
T7 | None,
]: ...
async def gather_optional_coroutines(
*coroutines: Unpack[tuple[Coroutine[Any, Any, T1] | None, ...]],
) -> tuple[T1 | None, ...]:
synapse/util/sentinel.py
-21
View File
@@ -1,21 +0,0 @@
#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2025 New Vector, Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
import enum
class Sentinel(enum.Enum):
# defining a sentinel in this way allows mypy to correctly handle the
# type of a dictionary lookup and subsequent type narrowing.
UNSET_SENTINEL = object()
synapse/visibility.py
+1
View File
@@ -103,6 +103,7 @@ async def filter_and_transform_events_for_client(
Returns:
The filtered events, wrapped in FilteredEvent with the requesting user's
membership at each event annotated for use during serialization (MSC4115).
The events are returned in the same order.
"""
# Filter out events that have been soft failed so that we don't relay them
# to clients, unless they're a server admin and want that to happen.
tests/rest/client/sliding_sync/test_extension_sticky_events.py
+676
View File
@@ -0,0 +1,676 @@
#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2026 New Vector, Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl_3.0.html>.
#
import logging
import sqlite3
from twisted.internet.testing import MemoryReactor
import synapse.rest.admin
import synapse.rest.client.account_data
from synapse.api.constants import EventTypes, EventUnsignedContentFields
from synapse.rest.client import account_data, login, register, room, sync
from synapse.server import HomeServer
from synapse.types import JsonDict, StreamKeyType
from synapse.util.clock import Clock
from synapse.util.duration import Duration
from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
from tests.server import TimedOutException
from tests.utils import USE_POSTGRES_FOR_TESTS
logger = logging.getLogger(__name__)
DUMMY_LISTS = {
"main": {
# Don't include any rooms in the top-N window
"ranges": [[0, 0]],
"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.
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
"""
class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase):
"""Tests for the sticky events sliding sync extension"""
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,
login.register_servlets,
register.register_servlets,
room.register_servlets,
sync.register_servlets,
account_data.register_servlets,
]
def default_config(self) -> JsonDict:
config = super().default_config()
# Enable sliding sync and sticky events MSCs
config["experimental_features"] = {
"msc3575_enabled": True,
"msc4354_enabled": True,
}
return config
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
self.store = hs.get_datastores().main
super().prepare(reactor, clock, hs)
def _assert_sticky_events_response(
self,
response_body: JsonDict,
expected_events_by_room: dict[str, list[str]] | None,
) -> str | None:
"""Assert the sliding sync response was successful and has the expected
sticky events.
Args:
response_body: Sliding Sync response body
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")
# 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
self.assertIsNotNone(sticky_events)
self.assertIsInstance(sticky_events["next_batch"], str)
actual_rooms = sticky_events["rooms"]
# Check that we have the expected rooms
self.assertIncludes(
set(actual_rooms.keys()), set(expected_events_by_room.keys()), exact=True
)
# Check the events in each room
for room_id, expected_events in expected_events_by_room.items():
actual_events = actual_rooms[room_id]["events"]
actual_event_ids = [e["event_id"] for e in actual_events]
self.assertEqual(actual_event_ids, expected_events)
for actual_event in actual_events:
# Check the sticky TTL is sent
self.assertIn("unsigned", actual_event)
ttl = actual_event["unsigned"][EventUnsignedContentFields.STICKY_TTL]
self.assertIsInstance(ttl, int)
self.assertIn("next_batch", sticky_events)
return sticky_events["next_batch"]
def test_empty_sync(self) -> None:
"""Test that enabling sticky events extension works on initial and incremental sync,
even if there is no data.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
# No sticky events in initial sync.
self._assert_sticky_events_response(response_body, None)
# Incremental sync should also have no sticky events
response_body, _ = self.do_sync(
sync_body, since=response_body["pos"], tok=user1_tok
)
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).
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("u2", "pass")
user2_tok = self.login(user2_id, "pass")
# Create a room and join both users
room_id = self.helper.create_room_as(user2_id, tok=user2_tok)
self.helper.join(room_id, user1_id, tok=user1_tok)
# Send a sticky event from user2
sticky_event_id: str = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(minutes=5),
content={"body": "sticky message", "msgtype": "m.text"},
tok=user2_tok,
)["event_id"]
# Initial sync should return the sticky event
sync_body: JsonDict = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"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
# 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)
# Check we don't get that event again
self._assert_sticky_events_response(response_body, None)
# Send another sticky event
sticky_event_id2: str = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(minutes=5),
content={"body": "another sticky message", "msgtype": "m.text"},
tok=user1_tok,
)["event_id"]
# Now the incremental sync should give us that event
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(
response_body, {room_id: [sticky_event_id2]}
)
def test_expired_events_not_returned(self) -> None:
"""Test that expired sticky events are not returned."""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("u2", "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 a sticky event with a short duration
sticky_event_id = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(seconds=2),
content={"body": "sticky message", "msgtype": "m.text"},
tok=user2_tok,
)["event_id"]
# Initial sync should return the sticky event
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
# We should still get the event for now
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# Advance time past the sticky duration
self.reactor.advance(3)
# A second initial sync should not return the expired sticky event
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self._assert_sticky_events_response(response_body, None)
def test_wait_for_new_data(self) -> None:
"""Test that the sliding sync request waits for new sticky events to arrive.
(Only applies to incremental syncs with a `timeout` specified).
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("u2", "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)
# Initial sync with no sticky events
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
_, from_token = self.do_sync(sync_body, tok=user1_tok)
# Make the sliding sync request with a timeout
channel = self.make_request(
"POST",
self.sync_endpoint + "?timeout=10000" + f"&pos={from_token}",
content=sync_body,
access_token=user1_tok,
await_result=False,
)
# Block for 5 seconds to make sure we are in `notifier.wait_for_events(...)`
with self.assertRaises(TimedOutException):
channel.await_result(timeout_ms=5000)
# Send a sticky event to trigger new results
sticky_event_id = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(minutes=5),
content={"body": "sticky message", "msgtype": "m.text"},
tok=user2_tok,
)["event_id"]
# Should respond before the 10 second timeout
channel.await_result(timeout_ms=100)
self.assertEqual(channel.code, 200, channel.json_body)
self._assert_sticky_events_response(
channel.json_body,
{room_id: [sticky_event_id]},
)
def test_wait_for_new_data_timeout(self) -> None:
"""
Test that the sliding sync request waits for new sticky events to arrive
and times out when no data arrives before the deadline.
(Only applies to incremental syncs with a `timeout` specified).
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
user2_id = self.register_user("u2", "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)
# Initial sync with no sticky events
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
_, from_token = self.do_sync(sync_body, tok=user1_tok)
# Make the sliding sync request with a timeout
channel = self.make_request(
"POST",
self.sync_endpoint + "?timeout=10000" + f"&pos={from_token}",
content=sync_body,
access_token=user1_tok,
await_result=False,
)
# Block for 5 seconds to make sure we are `notifier.wait_for_events(...)`
with self.assertRaises(TimedOutException):
channel.await_result(timeout_ms=5000)
# Wake-up `notifier.wait_for_events(...)` that will cause us test
# `SlidingSyncResult.__bool__` for new results.
self._bump_notifier_wait_for_events(
# wake key is intentionally unrelated to sticky events
user1_id,
wake_stream_key=StreamKeyType.ACCOUNT_DATA,
)
# Block for a little bit more to ensure we don't see any new results.
with self.assertRaises(TimedOutException):
channel.await_result(timeout_ms=4000)
# Wait for the sync to complete (wait for the rest of the 10 second timeout,
# 5000 + 4000 + 1200 > 10000)
channel.await_result(timeout_ms=1200)
self.assertEqual(channel.code, 200, channel.json_body)
self._assert_sticky_events_response(
channel.json_body,
None,
)
def test_ignored_users_sticky_events(self) -> None:
"""
Test that sticky events from ignored users are not delivered to clients.
> As with normal events, sticky events sent by ignored users MUST NOT be
> delivered to clients.
> — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#sync-api-changes
"""
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)
# User1 ignores user2
channel = self.make_request(
"PUT",
f"/_matrix/client/v3/user/{user1_id}/account_data/m.ignored_user_list",
{"ignored_users": {user2_id: {}}},
access_token=user1_tok,
)
self.assertEqual(channel.code, 200, channel.result)
# User2 sends a sticky event
sticky_event_id = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(minutes=5),
content={"body": "sticky from ignored user", "msgtype": "m.text"},
tok=user2_tok,
)["event_id"]
# Initial sync for user1
sync_body = {
"lists": {
"main": {
"ranges": [[0, 10]],
"required_state": [],
# In this test we ask for 10 events of timeline.
"timeline_limit": 10,
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
# Timeline events should not include sticky event from ignored user
timeline_events = response_body["rooms"][room_id]["timeline"]
timeline_event_ids = [e["event_id"] for e in timeline_events]
self.assertNotIn(
sticky_event_id,
timeline_event_ids,
"Sticky event from ignored user should not be in timeline",
)
# Sticky events section should also not include the event from ignored user
self._assert_sticky_events_response(response_body, None)
def test_history_visibility_bypass_for_sticky_events(self) -> None:
"""
Test that joined users can see sticky events even when history visibility
is set to "joined" and they joined after the event was sent.
> History visibility checks MUST NOT be applied to sticky events.
> Any joined user is authorised to see sticky events for the duration they remain sticky.
> — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#proposal
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
# Create a room with restrictive history visibility
room_id = self.helper.create_room_as(
user1_id,
tok=user1_tok,
extra_content={
# Anyone can join
"preset": "public_chat",
# But you can't see history before you joined
"initial_state": [
{
"type": EventTypes.RoomHistoryVisibility,
"state_key": "",
"content": {"history_visibility": "joined"},
}
],
},
is_public=False,
)
# User1 sends a sticky event
sticky_event_id = self.helper.send_sticky_event(
room_id,
EventTypes.Message,
duration=Duration(minutes=5),
content={"body": "sticky message", "msgtype": "m.text"},
tok=user1_tok,
)["event_id"]
# User1 also sends a regular event, to verify our test setup
regular_event_id = self.helper.send(
room_id=room_id,
body="regular message",
tok=user1_tok,
)["event_id"]
# Register and join a second user after the sticky event was sent
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
self.helper.join(room_id, user2_id, tok=user2_tok)
# User2 syncs - they should see sticky event even though
# history visibility is "joined" and they joined after it was sent
sync_body = {
"lists": {
"main": {
"ranges": [[0, 10]],
"required_state": [],
# In this test, we ask for 10 events of timeline.
"timeline_limit": 10,
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
response_body, _ = self.do_sync(sync_body, tok=user2_tok)
# The sticky event is fully visible in its own right,
# but AFAICT the timeline only includes events since we join the room
# (regardless of history visibility),
# so this comes down in the sticky extension
self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]})
# Instead the sticky event is in the timeline
timeline_events = response_body["rooms"][room_id]["timeline"]
timeline_event_ids = [e["event_id"] for e in timeline_events]
self.assertNotIn(
regular_event_id,
timeline_event_ids,
f"Expecting to not see regular event ({regular_event_id}) before user1 joined.",
)
def test_sticky_event_pagination(self) -> None:
"""
Test that pagination works correctly when there are many sticky events.
Also check they are delivered in stream order.
"""
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 4 sticky events (more than our limit of 2)
sticky_event_ids: list[str] = []
for i in range(4):
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": 2}
},
}
response_body, _ = 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(
response_body, {room_id: sticky_event_ids[0:2]}
)
# Incremental sync to get remaining sticky events
sync_body = {
"lists": DUMMY_LISTS,
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
# This makes it incremental
"since": next_batch,
"limit": 3,
}
},
}
response_body, _ = self.do_sync(sync_body, 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]}
)
def test_deduplication_with_timeline(self) -> None:
"""
Test that sticky events are not included in the sticky event extension of sliding sync
if they are included in the main timeline section.
Send 3 events:
1. sticky
2. sticky
3. regular
We then will sync with a timeline limit of 2 and a sticky event limit of 2.
We should then see (2) and (3) included in the timeline
and (1) in the sticky event response (but not (2) because it's already
included in the timeline.)
1. sticky [in sticky section]
------------->>> Timeline section
2. sticky
3. regular
-------------<<<
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
room_id = self.helper.create_room_as(user1_id, tok=user1_tok)
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=user1_tok,
)["event_id"]
sticky_event_ids.append(event_id)
non_sticky_event_id = self.helper.send_event(
room_id,
EventTypes.Message,
content={"body": "regular message", "msgtype": "m.text"},
tok=user1_tok,
)["event_id"]
# Sync
sync_body = {
"lists": {
"main": {
"ranges": [[0, 10]],
"required_state": [],
# In this test, we want a timeline window of the 2 latest messages
"timeline_limit": 2,
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
"limit": 2,
}
},
}
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"]
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"]
event_ids_in_timeline_section = [
e["event_id"] for e in events_in_timeline_section
]
self.assertEqual(
event_ids_in_sticky_section,
[sticky_event_ids[0]],
)
self.assertEqual(
event_ids_in_timeline_section, [sticky_event_ids[1], non_sticky_event_id]
)
tests/rest/client/sliding_sync/test_sliding_sync.py
+1 -1
View File
@@ -374,7 +374,7 @@ class SlidingSyncBase(unittest.HomeserverTestCase):
user_id: The user ID to wake up the notifier for
wake_stream_key: The stream key to wake up. This will create an actual new
entity in that stream so it's best to choose one that won't affect the
Sliding Sync results you're testing for. In other words, if your testing
Sliding Sync results you're testing for. In other words, if you're testing
account data, choose `StreamKeyType.PRESENCE` instead. We support two
possible stream keys because you're probably testing one or the other so
one is always a "safe" option.
tests/storage/test_event_push_actions.py
+109
View File
@@ -363,6 +363,115 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
_mark_read(cutoff_event_id)
_assert_count(0)
def test_count_aggregation_receipt_before_first_rotation(self) -> None:
"""
Regression test: reading a highlight before the first rotation must not
permanently inflate the badge count.
Highlights survive the receipt-triggered DELETE (highlight=1), so if
last_receipt_stream_ordering is NULL when rotation creates the summary
row, the badge query re-counts them forever.
"""
user_id, token, _, other_token, room_id = self._create_users_and_room()
def _assert_badge(expected: int) -> None:
counts = self.get_success(
self.store.db_pool.runInteraction(
"get-aggregate-unread-counts",
self.store._get_unread_counts_by_room_for_user_txn,
user_id,
)
)
self.assertEqual(counts.get(room_id, 0), expected)
def _send(highlight: bool = False) -> str:
return self.helper.send_event(
room_id,
type="m.room.message",
content={"msgtype": "m.text", "body": user_id if highlight else "msg"},
tok=other_token,
)["event_id"]
def _read(event_id: str) -> None:
self.get_success(
self.store.insert_receipt(
room_id,
"m.read",
user_id=user_id,
event_ids=[event_id],
thread_id=None,
data={},
)
)
# Highlight arrives; user reads it before any rotation (no summary row exists).
_read(_send(highlight=True))
# One new event after the receipt makes stream_ordering > max_clause true.
_send()
# Without the fix: badge = 2 (highlight re-counted). With fix: badge = 1.
self.get_success(self.store._rotate_notifs())
_assert_badge(1)
def test_count_aggregation_receipt_before_first_rotation_in_thread(self) -> None:
"""
Same regression as test_count_aggregation_receipt_before_first_rotation,
but for a highlight inside a thread cleared by an unthreaded receipt.
The fix must pre-populate event_push_summary for every thread with
pending push actions, not just MAIN_TIMELINE.
"""
user_id, token, _, other_token, room_id = self._create_users_and_room()
def _assert_badge(expected: int) -> None:
counts = self.get_success(
self.store.db_pool.runInteraction(
"get-aggregate-unread-counts",
self.store._get_unread_counts_by_room_for_user_txn,
user_id,
)
)
self.assertEqual(counts.get(room_id, 0), expected)
def _send(thread_root: str | None = None, highlight: bool = False) -> str:
content: JsonDict = {
"msgtype": "m.text",
"body": user_id if highlight else "msg",
}
if thread_root is not None:
content["m.relates_to"] = {
"rel_type": RelationTypes.THREAD,
"event_id": thread_root,
}
return self.helper.send_event(
room_id,
type="m.room.message",
content=content,
tok=other_token,
)["event_id"]
def _read(event_id: str) -> None:
self.get_success(
self.store.insert_receipt(
room_id,
"m.read",
user_id=user_id,
event_ids=[event_id],
thread_id=None,
data={},
)
)
# A thread root, then a highlight inside that thread.
thread_root = _send()
thread_highlight = _send(thread_root=thread_root, highlight=True)
# User reads everything with an unthreaded receipt before any rotation.
_read(thread_highlight)
# A new event in the same thread makes stream_ordering > max_clause true.
_send(thread_root=thread_root)
# Without the fix: badge = 2 (thread highlight re-counted). With: badge = 1.
self.get_success(self.store._rotate_notifs())
_assert_badge(1)
def test_count_aggregation_threads(self) -> None:
"""
This is essentially the same test as test_count_aggregation, but adds
tests/test_types.py
+152 -1
View File
@@ -18,16 +18,20 @@
# [This file includes modifications made by New Vector Limited]
#
#
import copy
from unittest import skipUnless
from immutabledict import immutabledict
from parameterized import parameterized_class
from pydantic import BaseModel, PydanticInvalidForJsonSchema, ValidationError
from synapse.api.errors import SynapseError
from synapse.types import (
Absent,
AbsentType,
AbstractMultiWriterStreamToken,
MultiWriterStreamToken,
NonNegativeStrictInt,
RoomAlias,
RoomStreamToken,
UserID,
@@ -199,3 +203,150 @@ class MultiWriterTokenTestCase(unittest.HomeserverTestCase):
parsed_token = self.get_success(self.token_type.parse(store, "m5~"))
self.assertEqual(parsed_token, self.token_type(stream=5))
class AbsentTestCase(unittest.TestCase):
"""
Tests for the `Absent` utility, which is meant to be like `None` except
explicitly signalling absence rather than JSON null.
"""
def test_cant_create_second_absent(self) -> None:
"""
Tests that we aren't allowed to instantiate a second `Absent`.
"""
with self.assertRaises(TypeError):
AbsentType() # type: ignore[call-arg]
def test_is_falsy(self) -> None:
"""
Tests `Absent` is falsy and can therefore be used a bit like `None`.
"""
if Absent:
self.fail("Absent is truthy!")
self.assertEqual(Absent or "something", "something")
def test_pydantic_jsonschema(self) -> None:
"""
Tests that `Absent` can't be used to produce JSONSchema in Pydantic models.
In the future, it may be useful to produce correct JSONSchema, but for now
I was mostly interested in making sure we don't produce weird/invalid JSONSchema.
"""
class MyModel(BaseModel):
absent: AbsentType = Absent
with self.assertRaises(PydanticInvalidForJsonSchema):
MyModel.model_json_schema()
def test_pydantic_reject_null(self) -> None:
"""
Tests that `Absent` rejects `None` (JSON null) when used in Pydantic models.
"""
class MyModel(BaseModel):
absent: AbsentType = Absent
with self.assertRaises(ValidationError):
MyModel.model_validate({"absent": None})
with self.assertRaises(ValidationError):
MyModel.model_validate_json('{"absent": null}')
def test_pydantic_accept_absence(self) -> None:
"""
Tests that `Absent` accepts the absence of a value when used in Pydantic models.
"""
class MyModel(BaseModel):
absent: AbsentType = Absent
self.assertEqual(MyModel.model_validate({}), MyModel(absent=Absent))
self.assertEqual(MyModel.model_validate_json("{}"), MyModel(absent=Absent))
def test_copy(self) -> None:
"""
Tests that the `copy` module always uses the same instance of Absent.
"""
class MyModel(BaseModel):
absent: AbsentType = Absent
a = MyModel.model_validate({})
b = copy.deepcopy(a)
self.assertIs(copy.copy(Absent), Absent)
self.assertIs(a.absent, b.absent)
class NonNegativeStrictIntTestCase(unittest.TestCase):
"""
Tests for the `NonNegativeStrictInt` utility.
"""
def test_pydantic_jsonschema(self) -> None:
"""
Tests that `NonNegativeStrictInt` produces sensible JSONSchema.
"""
class MyModel(BaseModel):
limit: NonNegativeStrictInt = 100
self.assertEqual(
MyModel.model_json_schema(),
{
"properties": {
"limit": {
"default": 100,
"minimum": 0,
"title": "Limit",
"type": "integer",
}
},
"title": "MyModel",
"type": "object",
},
f"JSONSchema actually is:\n{MyModel.model_json_schema()!r}",
)
def test_pydantic_reject(self) -> None:
"""
Tests that `NonNegativeStrictInt` rejects negative numbers
and non-ints.
"""
class MyModel(BaseModel):
limit: NonNegativeStrictInt = 100
with self.assertRaises(ValidationError):
MyModel.model_validate({"limit": -1})
with self.assertRaises(ValidationError):
MyModel.model_validate_json('{"limit": -1}')
# StrictInt, so don't accept floats...
with self.assertRaises(ValidationError):
MyModel.model_validate({"limit": 1.5})
with self.assertRaises(ValidationError):
MyModel.model_validate_json('{"limit": 1.5}')
# ...and don't accept stringy ints either.
with self.assertRaises(ValidationError):
MyModel.model_validate({"limit": "42"})
with self.assertRaises(ValidationError):
MyModel.model_validate_json('{"limit": "42"}')
def test_pydantic_accept(self) -> None:
"""
Tests that `Absent` accepts the absence of a value when used in Pydantic models.
"""
class MyModel(BaseModel):
limit: NonNegativeStrictInt = 100
self.assertEqual(MyModel.model_validate_json('{"limit": 0}'), MyModel(limit=0))
self.assertEqual(MyModel.model_validate({"limit": 42}), MyModel(limit=42))