mirror of
https://github.com/element-hq/synapse.git
synced 2025-04-08 11:24:00 +00:00
Merge branch 'develop' into madlittlemods/sliding-sync-pre-populate-room-meta-data
This commit is contained in:
Eric Eastwood
commit
cc200ee9f5
40 changed files with 1813 additions and 1303 deletions
CHANGES.md
changelog.d
17483.bugfix17510.bugfix17514.misc17515.doc17531.misc17535.bugfix17536.misc17537.misc17538.bugfix17542.misc17545.bugfix17548.misc17557.misc17558.misc17559.doc17561.misc17562.misc17563.misc17564.misc17566.misc17567.misc17568.bugfix17569.misc17570.bugfix17571.misc17574.misc17575.misc17595.misc
debian
pyproject.tomlsynapse
handlers/sliding_sync
rest/client
storage
types/handlers
tests/rest/client/sliding_sync
54
CHANGES.md
54
CHANGES.md
|
|
@ -1,3 +1,57 @@
|
|||
# Synapse 1.114.0rc1 (2024-08-20)
|
||||
|
||||
### Features
|
||||
|
||||
- Add a flag to `/versions`, `org.matrix.simplified_msc3575`, to indicate whether experimental sliding sync support has been enabled. ([\#17571](https://github.com/element-hq/synapse/issues/17571))
|
||||
- Handle changes in `timeline_limit` in experimental sliding sync. ([\#17579](https://github.com/element-hq/synapse/issues/17579))
|
||||
- Correctly track read receipts that should be sent down in experimental sliding sync. ([\#17575](https://github.com/element-hq/synapse/issues/17575), [\#17589](https://github.com/element-hq/synapse/issues/17589), [\#17592](https://github.com/element-hq/synapse/issues/17592))
|
||||
|
||||
### Bugfixes
|
||||
|
||||
- Start handlers for new media endpoints when media resource configured. ([\#17483](https://github.com/element-hq/synapse/issues/17483))
|
||||
- Fix timeline ordering (using `stream_ordering` instead of topological ordering) in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17510](https://github.com/element-hq/synapse/issues/17510))
|
||||
- Fix experimental sliding sync implementation to remember any updates in rooms that were not sent down immediately. ([\#17535](https://github.com/element-hq/synapse/issues/17535))
|
||||
- Better exclude partially stated rooms if we must await full state in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17538](https://github.com/element-hq/synapse/issues/17538))
|
||||
- Handle lower-case http headers in `_Mulitpart_Parser_Protocol`. ([\#17545](https://github.com/element-hq/synapse/issues/17545))
|
||||
- Fix fetching federation signing keys from servers that omit `old_verify_keys`. Contributed by @tulir @ Beeper. ([\#17568](https://github.com/element-hq/synapse/issues/17568))
|
||||
- Fix bug where we would respond with an error when a remote server asked for media that had a length of 0, using the new multipart federation media endpoint. ([\#17570](https://github.com/element-hq/synapse/issues/17570))
|
||||
|
||||
### Improved Documentation
|
||||
|
||||
- Clarify default behaviour of the
|
||||
[`auto_accept_invites.worker_to_run_on`](https://element-hq.github.io/synapse/develop/usage/configuration/config_documentation.html#auto-accept-invites)
|
||||
option. ([\#17515](https://github.com/element-hq/synapse/issues/17515))
|
||||
- Improve docstrings for profile methods. ([\#17559](https://github.com/element-hq/synapse/issues/17559))
|
||||
|
||||
### Internal Changes
|
||||
|
||||
- Add more tracing to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17514](https://github.com/element-hq/synapse/issues/17514))
|
||||
- Fixup comment in sliding sync implementation. ([\#17531](https://github.com/element-hq/synapse/issues/17531))
|
||||
- Replace override of deprecated method `HTTPAdapter.get_connection` with `get_connection_with_tls_context`. ([\#17536](https://github.com/element-hq/synapse/issues/17536))
|
||||
- Fix performance of device lists in `/key/changes` and sliding sync. ([\#17537](https://github.com/element-hq/synapse/issues/17537), [\#17548](https://github.com/element-hq/synapse/issues/17548))
|
||||
- Bump setuptools from 67.6.0 to 72.1.0. ([\#17542](https://github.com/element-hq/synapse/issues/17542))
|
||||
- Add a utility function for generating random event IDs. ([\#17557](https://github.com/element-hq/synapse/issues/17557))
|
||||
- Speed up responding to media requests. ([\#17558](https://github.com/element-hq/synapse/issues/17558), [\#17561](https://github.com/element-hq/synapse/issues/17561), [\#17564](https://github.com/element-hq/synapse/issues/17564), [\#17566](https://github.com/element-hq/synapse/issues/17566), [\#17567](https://github.com/element-hq/synapse/issues/17567), [\#17569](https://github.com/element-hq/synapse/issues/17569))
|
||||
- Test github token before running release script steps. ([\#17562](https://github.com/element-hq/synapse/issues/17562))
|
||||
- Reduce log spam of multipart files. ([\#17563](https://github.com/element-hq/synapse/issues/17563))
|
||||
- Refactor per-connection state in experimental sliding sync handler. ([\#17574](https://github.com/element-hq/synapse/issues/17574))
|
||||
- Add histogram metrics for sliding sync processing time. ([\#17593](https://github.com/element-hq/synapse/issues/17593))
|
||||
|
||||
|
||||
|
||||
### Updates to locked dependencies
|
||||
|
||||
* Bump bytes from 1.6.1 to 1.7.1. ([\#17526](https://github.com/element-hq/synapse/issues/17526))
|
||||
* Bump lxml from 5.2.2 to 5.3.0. ([\#17550](https://github.com/element-hq/synapse/issues/17550))
|
||||
* Bump phonenumbers from 8.13.42 to 8.13.43. ([\#17551](https://github.com/element-hq/synapse/issues/17551))
|
||||
* Bump regex from 1.10.5 to 1.10.6. ([\#17527](https://github.com/element-hq/synapse/issues/17527))
|
||||
* Bump sentry-sdk from 2.10.0 to 2.12.0. ([\#17553](https://github.com/element-hq/synapse/issues/17553))
|
||||
* Bump serde from 1.0.204 to 1.0.206. ([\#17556](https://github.com/element-hq/synapse/issues/17556))
|
||||
* Bump serde_json from 1.0.122 to 1.0.124. ([\#17555](https://github.com/element-hq/synapse/issues/17555))
|
||||
* Bump sigstore/cosign-installer from 3.5.0 to 3.6.0. ([\#17549](https://github.com/element-hq/synapse/issues/17549))
|
||||
* Bump types-pyyaml from 6.0.12.20240311 to 6.0.12.20240808. ([\#17552](https://github.com/element-hq/synapse/issues/17552))
|
||||
* Bump types-requests from 2.31.0.20240406 to 2.32.0.20240712. ([\#17524](https://github.com/element-hq/synapse/issues/17524))
|
||||
|
||||
# Synapse 1.113.0 (2024-08-13)
|
||||
|
||||
No significant changes since 1.113.0rc1.
|
||||
|
|
|
|||
|
|
@ -1 +0,0 @@
|
|||
Start handlers for new media endpoints when media resource configured.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix timeline ordering (using `stream_ordering` instead of topological ordering) in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Add more tracing to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
Clarify default behaviour of the
|
||||
[`auto_accept_invites.worker_to_run_on`](https://element-hq.github.io/synapse/develop/usage/configuration/config_documentation.html#auto-accept-invites)
|
||||
option.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fixup comment in sliding sync implementation.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix experimental sliding sync implementation to remember any updates in rooms that were not sent down immediately.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Replace override of deprecated method `HTTPAdapter.get_connection` with `get_connection_with_tls_context`.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix performance of device lists in `/key/changes` and sliding sync.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Better exclude partially stated rooms if we must await full state in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Bump setuptools from 67.6.0 to 72.1.0.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Handle lower-case http headers in `_Mulitpart_Parser_Protocol`.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix performance of device lists in `/key/changes` and sliding sync.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Add a utility function for generating random event IDs.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Improve docstrings for profile methods.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Test github token before running release script steps.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Reduce log spam of multipart files.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix fetching federation signing keys from servers that omit `old_verify_keys`. Contributed by @tulir @ Beeper.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Speed up responding to media requests.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Fix bug where we would respond with an error when a remote server asked for media that had a length of 0, using the new multipart federation media endpoint.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Add a flag to `/versions`, `org.matrix.simplified_msc3575`, to indicate whether experimental sliding sync support has been enabled.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Refactor per-connection state in experimental sliding sync handler.
|
||||
|
|
@ -1 +0,0 @@
|
|||
Correctly track read receipts that should be sent down in experimental sliding sync.
|
||||
1
changelog.d/17595.misc
Normal file
1
changelog.d/17595.misc
Normal file
|
|
@ -0,0 +1 @@
|
|||
Refactor sliding sync class into multiple files.
|
||||
6
debian/changelog
vendored
6
debian/changelog
vendored
|
|
@ -1,3 +1,9 @@
|
|||
matrix-synapse-py3 (1.114.0~rc1) stable; urgency=medium
|
||||
|
||||
* New synapse release 1.114.0rc1.
|
||||
|
||||
-- Synapse Packaging team <packages@matrix.org> Tue, 20 Aug 2024 12:55:28 +0000
|
||||
|
||||
matrix-synapse-py3 (1.113.0) stable; urgency=medium
|
||||
|
||||
* New Synapse release 1.113.0.
|
||||
|
|
|
|||
|
|
@ -97,7 +97,7 @@ module-name = "synapse.synapse_rust"
|
|||
|
||||
[tool.poetry]
|
||||
name = "matrix-synapse"
|
||||
version = "1.113.0"
|
||||
version = "1.114.0rc1"
|
||||
description = "Homeserver for the Matrix decentralised comms protocol"
|
||||
authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
|
||||
license = "AGPL-3.0-or-later"
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
660
synapse/handlers/sliding_sync/extensions.py
Normal file
660
synapse/handlers/sliding_sync/extensions.py
Normal file
|
|
@ -0,0 +1,660 @@
|
|||
#
|
||||
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
||||
#
|
||||
# Copyright (C) 2023 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
|
||||
from typing import TYPE_CHECKING, Dict, List, Mapping, Optional, Sequence, Set
|
||||
|
||||
from typing_extensions import assert_never
|
||||
|
||||
from synapse.api.constants import AccountDataTypes
|
||||
from synapse.handlers.receipts import ReceiptEventSource
|
||||
from synapse.handlers.sliding_sync.types import (
|
||||
HaveSentRoomFlag,
|
||||
MutablePerConnectionState,
|
||||
PerConnectionState,
|
||||
)
|
||||
from synapse.logging.opentracing import trace
|
||||
from synapse.types import (
|
||||
DeviceListUpdates,
|
||||
JsonMapping,
|
||||
MultiWriterStreamToken,
|
||||
SlidingSyncStreamToken,
|
||||
StreamToken,
|
||||
)
|
||||
from synapse.types.handlers import OperationType, SlidingSyncConfig, SlidingSyncResult
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from synapse.server import HomeServer
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class SlidingSyncExtensionHandler:
|
||||
"""Handles the extensions to sliding sync."""
|
||||
|
||||
def __init__(self, hs: "HomeServer"):
|
||||
self.store = hs.get_datastores().main
|
||||
self.event_sources = hs.get_event_sources()
|
||||
self.device_handler = hs.get_device_handler()
|
||||
self.push_rules_handler = hs.get_push_rules_handler()
|
||||
|
||||
@trace
|
||||
async def get_extensions_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
previous_connection_state: "PerConnectionState",
|
||||
new_connection_state: "MutablePerConnectionState",
|
||||
actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
|
||||
actual_room_ids: Set[str],
|
||||
actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
|
||||
to_token: StreamToken,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> SlidingSyncResult.Extensions:
|
||||
"""Handle extension requests.
|
||||
|
||||
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
|
||||
state, used to record updates to the state during this request.
|
||||
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.
|
||||
to_token: The point in the stream to sync up to.
|
||||
from_token: The point in the stream to sync from.
|
||||
"""
|
||||
|
||||
if sync_config.extensions is None:
|
||||
return SlidingSyncResult.Extensions()
|
||||
|
||||
to_device_response = None
|
||||
if sync_config.extensions.to_device is not None:
|
||||
to_device_response = await self.get_to_device_extension_response(
|
||||
sync_config=sync_config,
|
||||
to_device_request=sync_config.extensions.to_device,
|
||||
to_token=to_token,
|
||||
)
|
||||
|
||||
e2ee_response = None
|
||||
if sync_config.extensions.e2ee is not None:
|
||||
e2ee_response = await self.get_e2ee_extension_response(
|
||||
sync_config=sync_config,
|
||||
e2ee_request=sync_config.extensions.e2ee,
|
||||
to_token=to_token,
|
||||
from_token=from_token,
|
||||
)
|
||||
|
||||
account_data_response = None
|
||||
if sync_config.extensions.account_data is not None:
|
||||
account_data_response = await self.get_account_data_extension_response(
|
||||
sync_config=sync_config,
|
||||
actual_lists=actual_lists,
|
||||
actual_room_ids=actual_room_ids,
|
||||
account_data_request=sync_config.extensions.account_data,
|
||||
to_token=to_token,
|
||||
from_token=from_token,
|
||||
)
|
||||
|
||||
receipts_response = None
|
||||
if sync_config.extensions.receipts is not None:
|
||||
receipts_response = await self.get_receipts_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,
|
||||
receipts_request=sync_config.extensions.receipts,
|
||||
to_token=to_token,
|
||||
from_token=from_token,
|
||||
)
|
||||
|
||||
typing_response = None
|
||||
if sync_config.extensions.typing is not None:
|
||||
typing_response = await self.get_typing_extension_response(
|
||||
sync_config=sync_config,
|
||||
actual_lists=actual_lists,
|
||||
actual_room_ids=actual_room_ids,
|
||||
actual_room_response_map=actual_room_response_map,
|
||||
typing_request=sync_config.extensions.typing,
|
||||
to_token=to_token,
|
||||
from_token=from_token,
|
||||
)
|
||||
|
||||
return SlidingSyncResult.Extensions(
|
||||
to_device=to_device_response,
|
||||
e2ee=e2ee_response,
|
||||
account_data=account_data_response,
|
||||
receipts=receipts_response,
|
||||
typing=typing_response,
|
||||
)
|
||||
|
||||
def find_relevant_room_ids_for_extension(
|
||||
self,
|
||||
requested_lists: Optional[List[str]],
|
||||
requested_room_ids: Optional[List[str]],
|
||||
actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
|
||||
actual_room_ids: Set[str],
|
||||
) -> Set[str]:
|
||||
"""
|
||||
Handle the reserved `lists`/`rooms` keys for extensions. Extensions should only
|
||||
return results for rooms in the Sliding Sync response. This matches up the
|
||||
requested rooms/lists with the actual lists/rooms in the Sliding Sync response.
|
||||
|
||||
{"lists": []} // Do not process any lists.
|
||||
{"lists": ["rooms", "dms"]} // Process only a subset of lists.
|
||||
{"lists": ["*"]} // Process all lists defined in the Sliding Window API. (This is the default.)
|
||||
|
||||
{"rooms": []} // Do not process any specific rooms.
|
||||
{"rooms": ["!a:b", "!c:d"]} // Process only a subset of room subscriptions.
|
||||
{"rooms": ["*"]} // Process all room subscriptions defined in the Room Subscription API. (This is the default.)
|
||||
|
||||
Args:
|
||||
requested_lists: The `lists` from the extension request.
|
||||
requested_room_ids: The `rooms` from the extension request.
|
||||
actual_lists: The actual lists from the Sliding Sync response.
|
||||
actual_room_ids: The actual room subscriptions from the Sliding Sync request.
|
||||
"""
|
||||
|
||||
# We only want to include account data for rooms that are already in the sliding
|
||||
# sync response AND that were requested in the account data request.
|
||||
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
|
||||
|
||||
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: Optional[SlidingSyncResult.SlidingWindowList] = 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:
|
||||
# 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
|
||||
async def get_to_device_extension_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
to_device_request: SlidingSyncConfig.Extensions.ToDeviceExtension,
|
||||
to_token: StreamToken,
|
||||
) -> Optional[SlidingSyncResult.Extensions.ToDeviceExtension]:
|
||||
"""Handle to-device extension (MSC3885)
|
||||
|
||||
Args:
|
||||
sync_config: Sync configuration
|
||||
to_device_request: The to-device extension from the request
|
||||
to_token: The point in the stream to sync up to.
|
||||
"""
|
||||
user_id = sync_config.user.to_string()
|
||||
device_id = sync_config.requester.device_id
|
||||
|
||||
# Skip if the extension is not enabled
|
||||
if not to_device_request.enabled:
|
||||
return None
|
||||
|
||||
# Check that this request has a valid device ID (not all requests have
|
||||
# to belong to a device, and so device_id is None)
|
||||
if device_id is None:
|
||||
return SlidingSyncResult.Extensions.ToDeviceExtension(
|
||||
next_batch=f"{to_token.to_device_key}",
|
||||
events=[],
|
||||
)
|
||||
|
||||
since_stream_id = 0
|
||||
if to_device_request.since is not None:
|
||||
# We've already validated this is an int.
|
||||
since_stream_id = int(to_device_request.since)
|
||||
|
||||
if to_token.to_device_key < since_stream_id:
|
||||
# The since token is ahead of our current token, so we return an
|
||||
# empty response.
|
||||
logger.warning(
|
||||
"Got to-device.since from the future. since token: %r is ahead of our current to_device stream position: %r",
|
||||
since_stream_id,
|
||||
to_token.to_device_key,
|
||||
)
|
||||
return SlidingSyncResult.Extensions.ToDeviceExtension(
|
||||
next_batch=to_device_request.since,
|
||||
events=[],
|
||||
)
|
||||
|
||||
# Delete everything before the given since token, as we know the
|
||||
# device must have received them.
|
||||
deleted = await self.store.delete_messages_for_device(
|
||||
user_id=user_id,
|
||||
device_id=device_id,
|
||||
up_to_stream_id=since_stream_id,
|
||||
)
|
||||
|
||||
logger.debug(
|
||||
"Deleted %d to-device messages up to %d for %s",
|
||||
deleted,
|
||||
since_stream_id,
|
||||
user_id,
|
||||
)
|
||||
|
||||
messages, stream_id = await self.store.get_messages_for_device(
|
||||
user_id=user_id,
|
||||
device_id=device_id,
|
||||
from_stream_id=since_stream_id,
|
||||
to_stream_id=to_token.to_device_key,
|
||||
limit=min(to_device_request.limit, 100), # Limit to at most 100 events
|
||||
)
|
||||
|
||||
return SlidingSyncResult.Extensions.ToDeviceExtension(
|
||||
next_batch=f"{stream_id}",
|
||||
events=messages,
|
||||
)
|
||||
|
||||
@trace
|
||||
async def get_e2ee_extension_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
e2ee_request: SlidingSyncConfig.Extensions.E2eeExtension,
|
||||
to_token: StreamToken,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> Optional[SlidingSyncResult.Extensions.E2eeExtension]:
|
||||
"""Handle E2EE device extension (MSC3884)
|
||||
|
||||
Args:
|
||||
sync_config: Sync configuration
|
||||
e2ee_request: The e2ee extension from the request
|
||||
to_token: The point in the stream to sync up to.
|
||||
from_token: The point in the stream to sync from.
|
||||
"""
|
||||
user_id = sync_config.user.to_string()
|
||||
device_id = sync_config.requester.device_id
|
||||
|
||||
# Skip if the extension is not enabled
|
||||
if not e2ee_request.enabled:
|
||||
return None
|
||||
|
||||
device_list_updates: Optional[DeviceListUpdates] = None
|
||||
if from_token is not None:
|
||||
# TODO: This should take into account the `from_token` and `to_token`
|
||||
device_list_updates = await self.device_handler.get_user_ids_changed(
|
||||
user_id=user_id,
|
||||
from_token=from_token.stream_token,
|
||||
)
|
||||
|
||||
device_one_time_keys_count: Mapping[str, int] = {}
|
||||
device_unused_fallback_key_types: Sequence[str] = []
|
||||
if device_id:
|
||||
# TODO: We should have a way to let clients differentiate between the states of:
|
||||
# * no change in OTK count since the provided since token
|
||||
# * the server has zero OTKs left for this device
|
||||
# Spec issue: https://github.com/matrix-org/matrix-doc/issues/3298
|
||||
device_one_time_keys_count = await self.store.count_e2e_one_time_keys(
|
||||
user_id, device_id
|
||||
)
|
||||
device_unused_fallback_key_types = (
|
||||
await self.store.get_e2e_unused_fallback_key_types(user_id, device_id)
|
||||
)
|
||||
|
||||
return SlidingSyncResult.Extensions.E2eeExtension(
|
||||
device_list_updates=device_list_updates,
|
||||
device_one_time_keys_count=device_one_time_keys_count,
|
||||
device_unused_fallback_key_types=device_unused_fallback_key_types,
|
||||
)
|
||||
|
||||
@trace
|
||||
async def get_account_data_extension_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
|
||||
actual_room_ids: Set[str],
|
||||
account_data_request: SlidingSyncConfig.Extensions.AccountDataExtension,
|
||||
to_token: StreamToken,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> Optional[SlidingSyncResult.Extensions.AccountDataExtension]:
|
||||
"""Handle Account Data extension (MSC3959)
|
||||
|
||||
Args:
|
||||
sync_config: Sync configuration
|
||||
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.
|
||||
account_data_request: The account_data extension from the request
|
||||
to_token: The point in the stream to sync up to.
|
||||
from_token: The point in the stream to sync from.
|
||||
"""
|
||||
user_id = sync_config.user.to_string()
|
||||
|
||||
# Skip if the extension is not enabled
|
||||
if not account_data_request.enabled:
|
||||
return None
|
||||
|
||||
global_account_data_map: Mapping[str, JsonMapping] = {}
|
||||
if from_token is not None:
|
||||
# TODO: This should take into account the `from_token` and `to_token`
|
||||
global_account_data_map = (
|
||||
await self.store.get_updated_global_account_data_for_user(
|
||||
user_id, from_token.stream_token.account_data_key
|
||||
)
|
||||
)
|
||||
|
||||
have_push_rules_changed = await self.store.have_push_rules_changed_for_user(
|
||||
user_id, from_token.stream_token.push_rules_key
|
||||
)
|
||||
if have_push_rules_changed:
|
||||
global_account_data_map = dict(global_account_data_map)
|
||||
# TODO: This should take into account the `from_token` and `to_token`
|
||||
global_account_data_map[AccountDataTypes.PUSH_RULES] = (
|
||||
await self.push_rules_handler.push_rules_for_user(sync_config.user)
|
||||
)
|
||||
else:
|
||||
# TODO: This should take into account the `to_token`
|
||||
all_global_account_data = await self.store.get_global_account_data_for_user(
|
||||
user_id
|
||||
)
|
||||
|
||||
global_account_data_map = dict(all_global_account_data)
|
||||
# TODO: This should take into account the `to_token`
|
||||
global_account_data_map[AccountDataTypes.PUSH_RULES] = (
|
||||
await self.push_rules_handler.push_rules_for_user(sync_config.user)
|
||||
)
|
||||
|
||||
# Fetch room account data
|
||||
account_data_by_room_map: Mapping[str, Mapping[str, JsonMapping]] = {}
|
||||
relevant_room_ids = self.find_relevant_room_ids_for_extension(
|
||||
requested_lists=account_data_request.lists,
|
||||
requested_room_ids=account_data_request.rooms,
|
||||
actual_lists=actual_lists,
|
||||
actual_room_ids=actual_room_ids,
|
||||
)
|
||||
if len(relevant_room_ids) > 0:
|
||||
if from_token is not None:
|
||||
# TODO: This should take into account the `from_token` and `to_token`
|
||||
account_data_by_room_map = (
|
||||
await self.store.get_updated_room_account_data_for_user(
|
||||
user_id, from_token.stream_token.account_data_key
|
||||
)
|
||||
)
|
||||
else:
|
||||
# TODO: This should take into account the `to_token`
|
||||
account_data_by_room_map = (
|
||||
await self.store.get_room_account_data_for_user(user_id)
|
||||
)
|
||||
|
||||
# Filter down to the relevant rooms
|
||||
account_data_by_room_map = {
|
||||
room_id: account_data_map
|
||||
for room_id, account_data_map in account_data_by_room_map.items()
|
||||
if room_id in relevant_room_ids
|
||||
}
|
||||
|
||||
return SlidingSyncResult.Extensions.AccountDataExtension(
|
||||
global_account_data_map=global_account_data_map,
|
||||
account_data_by_room_map=account_data_by_room_map,
|
||||
)
|
||||
|
||||
@trace
|
||||
async def get_receipts_extension_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
previous_connection_state: "PerConnectionState",
|
||||
new_connection_state: "MutablePerConnectionState",
|
||||
actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
|
||||
actual_room_ids: Set[str],
|
||||
actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
|
||||
receipts_request: SlidingSyncConfig.Extensions.ReceiptsExtension,
|
||||
to_token: StreamToken,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> Optional[SlidingSyncResult.Extensions.ReceiptsExtension]:
|
||||
"""Handle Receipts extension (MSC3960)
|
||||
|
||||
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.
|
||||
account_data_request: The account_data extension from the request
|
||||
to_token: The point in the stream to sync up to.
|
||||
from_token: The point in the stream to sync from.
|
||||
"""
|
||||
# Skip if the extension is not enabled
|
||||
if not receipts_request.enabled:
|
||||
return None
|
||||
|
||||
relevant_room_ids = self.find_relevant_room_ids_for_extension(
|
||||
requested_lists=receipts_request.lists,
|
||||
requested_room_ids=receipts_request.rooms,
|
||||
actual_lists=actual_lists,
|
||||
actual_room_ids=actual_room_ids,
|
||||
)
|
||||
|
||||
room_id_to_receipt_map: Dict[str, JsonMapping] = {}
|
||||
if len(relevant_room_ids) > 0:
|
||||
# We need to handle the different cases depending on if we have sent
|
||||
# down receipts previously or not, so we split the relevant rooms
|
||||
# up into different collections based on status.
|
||||
live_rooms = set()
|
||||
previously_rooms: Dict[str, MultiWriterStreamToken] = {}
|
||||
initial_rooms = set()
|
||||
|
||||
for room_id in relevant_room_ids:
|
||||
if not from_token:
|
||||
initial_rooms.add(room_id)
|
||||
continue
|
||||
|
||||
# If we're sending down the room from scratch again for some reason, we
|
||||
# should always resend the receipts as well (regardless of if
|
||||
# we've sent them down before). This is to mimic the behaviour
|
||||
# of what happens on initial sync, where you get a chunk of
|
||||
# timeline with all of the corresponding receipts for the events in the timeline.
|
||||
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.receipts.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)
|
||||
|
||||
# The set of receipts that we fetched. Private receipts need to be
|
||||
# filtered out before returning.
|
||||
fetched_receipts = []
|
||||
|
||||
# For live rooms we just fetch all receipts in those rooms since the
|
||||
# `since` token.
|
||||
if live_rooms:
|
||||
assert from_token is not None
|
||||
receipts = await self.store.get_linearized_receipts_for_rooms(
|
||||
room_ids=live_rooms,
|
||||
from_key=from_token.stream_token.receipt_key,
|
||||
to_key=to_token.receipt_key,
|
||||
)
|
||||
fetched_receipts.extend(receipts)
|
||||
|
||||
# For rooms we've previously sent down, but aren't up to date, we
|
||||
# need to use the from token from the room status.
|
||||
if previously_rooms:
|
||||
for room_id, receipt_token in previously_rooms.items():
|
||||
# TODO: Limit the number of receipts we're about to send down
|
||||
# for the room, if its too many we should TODO
|
||||
previously_receipts = (
|
||||
await self.store.get_linearized_receipts_for_room(
|
||||
room_id=room_id,
|
||||
from_key=receipt_token,
|
||||
to_key=to_token.receipt_key,
|
||||
)
|
||||
)
|
||||
fetched_receipts.extend(previously_receipts)
|
||||
|
||||
# For rooms we haven't previously sent down, we could send all receipts
|
||||
# from that room but we only want to include receipts for events
|
||||
# in the timeline to avoid bloating and blowing up the sync response
|
||||
# as the number of users in the room increases. (this behavior is part of the spec)
|
||||
initial_rooms_and_event_ids = [
|
||||
(room_id, event.event_id)
|
||||
for room_id in initial_rooms
|
||||
if room_id in actual_room_response_map
|
||||
for event in actual_room_response_map[room_id].timeline_events
|
||||
]
|
||||
if initial_rooms_and_event_ids:
|
||||
initial_receipts = await self.store.get_linearized_receipts_for_events(
|
||||
room_and_event_ids=initial_rooms_and_event_ids,
|
||||
)
|
||||
fetched_receipts.extend(initial_receipts)
|
||||
|
||||
fetched_receipts = ReceiptEventSource.filter_out_private_receipts(
|
||||
fetched_receipts, sync_config.user.to_string()
|
||||
)
|
||||
|
||||
for receipt in fetched_receipts:
|
||||
# These fields should exist for every receipt
|
||||
room_id = receipt["room_id"]
|
||||
type = receipt["type"]
|
||||
content = receipt["content"]
|
||||
|
||||
room_id_to_receipt_map[room_id] = {"type": type, "content": content}
|
||||
|
||||
# Now we update the per-connection state to track which receipts we have
|
||||
# and haven't sent down.
|
||||
new_connection_state.receipts.record_sent_rooms(relevant_room_ids)
|
||||
|
||||
if from_token:
|
||||
# Now find the set of rooms that may have receipts that we're not sending
|
||||
# down. We only need to check rooms that we have previously returned
|
||||
# receipts for (in `previous_connection_state`) because we only care about
|
||||
# updating `LIVE` rooms to `PREVIOUSLY`. The `PREVIOUSLY` rooms will just
|
||||
# stay pointing at their previous position so we don't need to waste time
|
||||
# checking those and since we default to `NEVER`, rooms that were `NEVER`
|
||||
# sent before don't need to be recorded as we'll handle them correctly when
|
||||
# they come into range for the first time.
|
||||
rooms_no_receipts = [
|
||||
room_id
|
||||
for room_id, room_status in previous_connection_state.receipts._statuses.items()
|
||||
if room_status.status == HaveSentRoomFlag.LIVE
|
||||
and room_id not in relevant_room_ids
|
||||
]
|
||||
changed_rooms = await self.store.get_rooms_with_receipts_between(
|
||||
rooms_no_receipts,
|
||||
from_key=from_token.stream_token.receipt_key,
|
||||
to_key=to_token.receipt_key,
|
||||
)
|
||||
new_connection_state.receipts.record_unsent_rooms(
|
||||
changed_rooms, from_token.stream_token.receipt_key
|
||||
)
|
||||
|
||||
return SlidingSyncResult.Extensions.ReceiptsExtension(
|
||||
room_id_to_receipt_map=room_id_to_receipt_map,
|
||||
)
|
||||
|
||||
async def get_typing_extension_response(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
|
||||
actual_room_ids: Set[str],
|
||||
actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
|
||||
typing_request: SlidingSyncConfig.Extensions.TypingExtension,
|
||||
to_token: StreamToken,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> Optional[SlidingSyncResult.Extensions.TypingExtension]:
|
||||
"""Handle Typing Notification extension (MSC3961)
|
||||
|
||||
Args:
|
||||
sync_config: Sync configuration
|
||||
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.
|
||||
account_data_request: The account_data extension from the request
|
||||
to_token: The point in the stream to sync up to.
|
||||
from_token: The point in the stream to sync from.
|
||||
"""
|
||||
# Skip if the extension is not enabled
|
||||
if not typing_request.enabled:
|
||||
return None
|
||||
|
||||
relevant_room_ids = self.find_relevant_room_ids_for_extension(
|
||||
requested_lists=typing_request.lists,
|
||||
requested_room_ids=typing_request.rooms,
|
||||
actual_lists=actual_lists,
|
||||
actual_room_ids=actual_room_ids,
|
||||
)
|
||||
|
||||
room_id_to_typing_map: Dict[str, JsonMapping] = {}
|
||||
if len(relevant_room_ids) > 0:
|
||||
# Note: We don't need to take connection tracking into account for typing
|
||||
# notifications because they'll get anything still relevant and hasn't timed
|
||||
# out when the room comes into range. We consider the gap where the room
|
||||
# fell out of range, as long enough for any typing notifications to have
|
||||
# timed out (it's not worth the 30 seconds of data we may have missed).
|
||||
typing_source = self.event_sources.sources.typing
|
||||
typing_notifications, _ = await typing_source.get_new_events(
|
||||
user=sync_config.user,
|
||||
from_key=(from_token.stream_token.typing_key if from_token else 0),
|
||||
to_key=to_token.typing_key,
|
||||
# This is a dummy value and isn't used in the function
|
||||
limit=0,
|
||||
room_ids=relevant_room_ids,
|
||||
is_guest=False,
|
||||
)
|
||||
|
||||
for typing_notification in typing_notifications:
|
||||
# These fields should exist for every typing notification
|
||||
room_id = typing_notification["room_id"]
|
||||
type = typing_notification["type"]
|
||||
content = typing_notification["content"]
|
||||
|
||||
room_id_to_typing_map[room_id] = {"type": type, "content": content}
|
||||
|
||||
return SlidingSyncResult.Extensions.TypingExtension(
|
||||
room_id_to_typing_map=room_id_to_typing_map,
|
||||
)
|
||||
200
synapse/handlers/sliding_sync/store.py
Normal file
200
synapse/handlers/sliding_sync/store.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
#
|
||||
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
||||
#
|
||||
# Copyright (C) 2023 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
|
||||
from typing import TYPE_CHECKING, Dict, Optional, Tuple
|
||||
|
||||
import attr
|
||||
|
||||
from synapse.api.errors import SlidingSyncUnknownPosition
|
||||
from synapse.handlers.sliding_sync.types import (
|
||||
MutablePerConnectionState,
|
||||
PerConnectionState,
|
||||
)
|
||||
from synapse.logging.opentracing import trace
|
||||
from synapse.types import SlidingSyncStreamToken
|
||||
from synapse.types.handlers import SlidingSyncConfig
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@attr.s(auto_attribs=True)
|
||||
class SlidingSyncConnectionStore:
|
||||
"""In-memory store of per-connection state, including what rooms we have
|
||||
previously sent down a sliding sync connection.
|
||||
|
||||
Note: This is NOT safe to run in a worker setup because connection positions will
|
||||
point to different sets of rooms on different workers. e.g. for the same connection,
|
||||
a connection position of 5 might have totally different states on worker A and
|
||||
worker B.
|
||||
|
||||
One complication that we need to deal with here is needing to handle requests being
|
||||
resent, i.e. if we sent down a room in a response that the client received, we must
|
||||
consider the room *not* sent when we get the request again.
|
||||
|
||||
This is handled by using an integer "token", which is returned to the client
|
||||
as part of the sync token. For each connection we store a mapping from
|
||||
tokens to the room states, and create a new entry when we send down new
|
||||
rooms.
|
||||
|
||||
Note that for any given sliding sync connection we will only store a maximum
|
||||
of two different tokens: the previous token from the request and a new token
|
||||
sent in the response. When we receive a request with a given token, we then
|
||||
clear out all other entries with a different token.
|
||||
|
||||
Attributes:
|
||||
_connections: Mapping from `(user_id, conn_id)` to mapping of `token`
|
||||
to mapping of room ID to `HaveSentRoom`.
|
||||
"""
|
||||
|
||||
# `(user_id, conn_id)` -> `connection_position` -> `PerConnectionState`
|
||||
_connections: Dict[Tuple[str, str], Dict[int, PerConnectionState]] = attr.Factory(
|
||||
dict
|
||||
)
|
||||
|
||||
async def is_valid_token(
|
||||
self, sync_config: SlidingSyncConfig, connection_token: int
|
||||
) -> bool:
|
||||
"""Return whether the connection token is valid/recognized"""
|
||||
if connection_token == 0:
|
||||
return True
|
||||
|
||||
conn_key = self._get_connection_key(sync_config)
|
||||
return connection_token in self._connections.get(conn_key, {})
|
||||
|
||||
async def get_per_connection_state(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> PerConnectionState:
|
||||
"""Fetch the per-connection state for the token.
|
||||
|
||||
Raises:
|
||||
SlidingSyncUnknownPosition if the connection_token is unknown
|
||||
"""
|
||||
if from_token is None:
|
||||
return PerConnectionState()
|
||||
|
||||
connection_position = from_token.connection_position
|
||||
if connection_position == 0:
|
||||
# Initial sync (request without a `from_token`) starts at `0` so
|
||||
# there is no existing per-connection state
|
||||
return PerConnectionState()
|
||||
|
||||
conn_key = self._get_connection_key(sync_config)
|
||||
sync_statuses = self._connections.get(conn_key, {})
|
||||
connection_state = sync_statuses.get(connection_position)
|
||||
|
||||
if connection_state is None:
|
||||
raise SlidingSyncUnknownPosition()
|
||||
|
||||
return connection_state
|
||||
|
||||
@trace
|
||||
async def record_new_state(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
new_connection_state: MutablePerConnectionState,
|
||||
) -> int:
|
||||
"""Record updated per-connection state, returning the connection
|
||||
position associated with the new state.
|
||||
If there are no changes to the state this may return the same token as
|
||||
the existing per-connection state.
|
||||
"""
|
||||
prev_connection_token = 0
|
||||
if from_token is not None:
|
||||
prev_connection_token = from_token.connection_position
|
||||
|
||||
if not new_connection_state.has_updates():
|
||||
return prev_connection_token
|
||||
|
||||
conn_key = self._get_connection_key(sync_config)
|
||||
sync_statuses = self._connections.setdefault(conn_key, {})
|
||||
|
||||
# Generate a new token, removing any existing entries in that token
|
||||
# (which can happen if requests get resent).
|
||||
new_store_token = prev_connection_token + 1
|
||||
sync_statuses.pop(new_store_token, None)
|
||||
|
||||
# We copy the `MutablePerConnectionState` so that the inner `ChainMap`s
|
||||
# don't grow forever.
|
||||
sync_statuses[new_store_token] = new_connection_state.copy()
|
||||
|
||||
return new_store_token
|
||||
|
||||
@trace
|
||||
async def mark_token_seen(
|
||||
self,
|
||||
sync_config: SlidingSyncConfig,
|
||||
from_token: Optional[SlidingSyncStreamToken],
|
||||
) -> None:
|
||||
"""We have received a request with the given token, so we can clear out
|
||||
any other tokens associated with the connection.
|
||||
|
||||
If there is no from token then we have started afresh, and so we delete
|
||||
all tokens associated with the device.
|
||||
"""
|
||||
# Clear out any tokens for the connection that doesn't match the one
|
||||
# from the request.
|
||||
|
||||
conn_key = self._get_connection_key(sync_config)
|
||||
sync_statuses = self._connections.pop(conn_key, {})
|
||||
if from_token is None:
|
||||
return
|
||||
|
||||
sync_statuses = {
|
||||
connection_token: room_statuses
|
||||
for connection_token, room_statuses in sync_statuses.items()
|
||||
if connection_token == from_token.connection_position
|
||||
}
|
||||
if sync_statuses:
|
||||
self._connections[conn_key] = sync_statuses
|
||||
|
||||
@staticmethod
|
||||
def _get_connection_key(sync_config: SlidingSyncConfig) -> Tuple[str, str]:
|
||||
"""Return a unique identifier for this connection.
|
||||
|
||||
The first part is simply the user ID.
|
||||
|
||||
The second part is generally a combination of device ID and conn_id.
|
||||
However, both these two are optional (e.g. puppet access tokens don't
|
||||
have device IDs), so this handles those edge cases.
|
||||
|
||||
We use this over the raw `conn_id` to avoid clashes between different
|
||||
clients that use the same `conn_id`. Imagine a user uses a web client
|
||||
that uses `conn_id: main_sync_loop` and an Android client that also has
|
||||
a `conn_id: main_sync_loop`.
|
||||
"""
|
||||
|
||||
user_id = sync_config.user.to_string()
|
||||
|
||||
# Only one sliding sync connection is allowed per given conn_id (empty
|
||||
# or not).
|
||||
conn_id = sync_config.conn_id or ""
|
||||
|
||||
if sync_config.requester.device_id:
|
||||
return (user_id, f"D/{sync_config.requester.device_id}/{conn_id}")
|
||||
|
||||
if sync_config.requester.access_token_id:
|
||||
# If we don't have a device, then the access token ID should be a
|
||||
# stable ID.
|
||||
return (user_id, f"A/{sync_config.requester.access_token_id}/{conn_id}")
|
||||
|
||||
# If we have neither then its likely an AS or some weird token. Either
|
||||
# way we can just fail here.
|
||||
raise Exception("Cannot use sliding sync with access token type")
|
||||
506
synapse/handlers/sliding_sync/types.py
Normal file
506
synapse/handlers/sliding_sync/types.py
Normal file
|
|
@ -0,0 +1,506 @@
|
|||
#
|
||||
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
||||
#
|
||||
# Copyright (C) 2024 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 typing
|
||||
from collections import ChainMap
|
||||
from enum import Enum
|
||||
from typing import (
|
||||
TYPE_CHECKING,
|
||||
Callable,
|
||||
Dict,
|
||||
Final,
|
||||
Generic,
|
||||
Mapping,
|
||||
MutableMapping,
|
||||
Optional,
|
||||
Set,
|
||||
TypeVar,
|
||||
cast,
|
||||
)
|
||||
|
||||
import attr
|
||||
|
||||
from synapse.api.constants import EventTypes
|
||||
from synapse.types import MultiWriterStreamToken, RoomStreamToken, StrCollection, UserID
|
||||
from synapse.types.handlers import SlidingSyncConfig
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class StateValues:
|
||||
"""
|
||||
Understood values of the (type, state_key) tuple in `required_state`.
|
||||
"""
|
||||
|
||||
# Include all state events of the given type
|
||||
WILDCARD: Final = "*"
|
||||
# Lazy-load room membership events (include room membership events for any event
|
||||
# `sender` in the timeline). We only give special meaning to this value when it's a
|
||||
# `state_key`.
|
||||
LAZY: Final = "$LAZY"
|
||||
# Subsitute with the requester's user ID. Typically used by clients to get
|
||||
# the user's membership.
|
||||
ME: Final = "$ME"
|
||||
|
||||
|
||||
# We can't freeze this class because we want to update it in place with the
|
||||
# de-duplicated data.
|
||||
@attr.s(slots=True, auto_attribs=True)
|
||||
class RoomSyncConfig:
|
||||
"""
|
||||
Holds the config for what data we should fetch for a room in the sync response.
|
||||
|
||||
Attributes:
|
||||
timeline_limit: The maximum number of events to return in the timeline.
|
||||
|
||||
required_state_map: Map from state event type to state_keys requested for the
|
||||
room. The values are close to `StateKey` but actually use a syntax where you
|
||||
can provide `*` wildcard and `$LAZY` for lazy-loading room members.
|
||||
"""
|
||||
|
||||
timeline_limit: int
|
||||
required_state_map: Dict[str, Set[str]]
|
||||
|
||||
@classmethod
|
||||
def from_room_config(
|
||||
cls,
|
||||
room_params: SlidingSyncConfig.CommonRoomParameters,
|
||||
) -> "RoomSyncConfig":
|
||||
"""
|
||||
Create a `RoomSyncConfig` from a `SlidingSyncList`/`RoomSubscription` config.
|
||||
|
||||
Args:
|
||||
room_params: `SlidingSyncConfig.SlidingSyncList` or `SlidingSyncConfig.RoomSubscription`
|
||||
"""
|
||||
required_state_map: Dict[str, Set[str]] = {}
|
||||
for (
|
||||
state_type,
|
||||
state_key,
|
||||
) in room_params.required_state:
|
||||
# If we already have a wildcard for this specific `state_key`, we don't need
|
||||
# to add it since the wildcard already covers it.
|
||||
if state_key in required_state_map.get(StateValues.WILDCARD, set()):
|
||||
continue
|
||||
|
||||
# If we already have a wildcard `state_key` for this `state_type`, we don't need
|
||||
# to add anything else
|
||||
if StateValues.WILDCARD in required_state_map.get(state_type, set()):
|
||||
continue
|
||||
|
||||
# If we're getting wildcards for the `state_type` and `state_key`, that's
|
||||
# all that matters so get rid of any other entries
|
||||
if state_type == StateValues.WILDCARD and state_key == StateValues.WILDCARD:
|
||||
required_state_map = {StateValues.WILDCARD: {StateValues.WILDCARD}}
|
||||
# We can break, since we don't need to add anything else
|
||||
break
|
||||
|
||||
# If we're getting a wildcard for the `state_type`, get rid of any other
|
||||
# entries with the same `state_key`, since the wildcard will cover it already.
|
||||
elif state_type == StateValues.WILDCARD:
|
||||
# Get rid of any entries that match the `state_key`
|
||||
#
|
||||
# Make a copy so we don't run into an error: `dictionary changed size
|
||||
# during iteration`, when we remove items
|
||||
for (
|
||||
existing_state_type,
|
||||
existing_state_key_set,
|
||||
) in list(required_state_map.items()):
|
||||
# Make a copy so we don't run into an error: `Set changed size during
|
||||
# iteration`, when we filter out and remove items
|
||||
for existing_state_key in existing_state_key_set.copy():
|
||||
if existing_state_key == state_key:
|
||||
existing_state_key_set.remove(state_key)
|
||||
|
||||
# If we've the left the `set()` empty, remove it from the map
|
||||
if existing_state_key_set == set():
|
||||
required_state_map.pop(existing_state_type, None)
|
||||
|
||||
# If we're getting a wildcard `state_key`, get rid of any other state_keys
|
||||
# for this `state_type` since the wildcard will cover it already.
|
||||
if state_key == StateValues.WILDCARD:
|
||||
required_state_map[state_type] = {state_key}
|
||||
# Otherwise, just add it to the set
|
||||
else:
|
||||
if required_state_map.get(state_type) is None:
|
||||
required_state_map[state_type] = {state_key}
|
||||
else:
|
||||
required_state_map[state_type].add(state_key)
|
||||
|
||||
return cls(
|
||||
timeline_limit=room_params.timeline_limit,
|
||||
required_state_map=required_state_map,
|
||||
)
|
||||
|
||||
def deep_copy(self) -> "RoomSyncConfig":
|
||||
required_state_map: Dict[str, Set[str]] = {
|
||||
state_type: state_key_set.copy()
|
||||
for state_type, state_key_set in self.required_state_map.items()
|
||||
}
|
||||
|
||||
return RoomSyncConfig(
|
||||
timeline_limit=self.timeline_limit,
|
||||
required_state_map=required_state_map,
|
||||
)
|
||||
|
||||
def combine_room_sync_config(
|
||||
self, other_room_sync_config: "RoomSyncConfig"
|
||||
) -> None:
|
||||
"""
|
||||
Combine this `RoomSyncConfig` with another `RoomSyncConfig` and take the
|
||||
superset union of the two.
|
||||
"""
|
||||
# Take the highest timeline limit
|
||||
if self.timeline_limit < other_room_sync_config.timeline_limit:
|
||||
self.timeline_limit = other_room_sync_config.timeline_limit
|
||||
|
||||
# Union the required state
|
||||
for (
|
||||
state_type,
|
||||
state_key_set,
|
||||
) in other_room_sync_config.required_state_map.items():
|
||||
# If we already have a wildcard for everything, we don't need to add
|
||||
# anything else
|
||||
if StateValues.WILDCARD in self.required_state_map.get(
|
||||
StateValues.WILDCARD, set()
|
||||
):
|
||||
break
|
||||
|
||||
# If we already have a wildcard `state_key` for this `state_type`, we don't need
|
||||
# to add anything else
|
||||
if StateValues.WILDCARD in self.required_state_map.get(state_type, set()):
|
||||
continue
|
||||
|
||||
# If we're getting wildcards for the `state_type` and `state_key`, that's
|
||||
# all that matters so get rid of any other entries
|
||||
if (
|
||||
state_type == StateValues.WILDCARD
|
||||
and StateValues.WILDCARD in state_key_set
|
||||
):
|
||||
self.required_state_map = {state_type: {StateValues.WILDCARD}}
|
||||
# We can break, since we don't need to add anything else
|
||||
break
|
||||
|
||||
for state_key in state_key_set:
|
||||
# If we already have a wildcard for this specific `state_key`, we don't need
|
||||
# to add it since the wildcard already covers it.
|
||||
if state_key in self.required_state_map.get(
|
||||
StateValues.WILDCARD, set()
|
||||
):
|
||||
continue
|
||||
|
||||
# If we're getting a wildcard for the `state_type`, get rid of any other
|
||||
# entries with the same `state_key`, since the wildcard will cover it already.
|
||||
if state_type == StateValues.WILDCARD:
|
||||
# Get rid of any entries that match the `state_key`
|
||||
#
|
||||
# Make a copy so we don't run into an error: `dictionary changed size
|
||||
# during iteration`, when we remove items
|
||||
for existing_state_type, existing_state_key_set in list(
|
||||
self.required_state_map.items()
|
||||
):
|
||||
# Make a copy so we don't run into an error: `Set changed size during
|
||||
# iteration`, when we filter out and remove items
|
||||
for existing_state_key in existing_state_key_set.copy():
|
||||
if existing_state_key == state_key:
|
||||
existing_state_key_set.remove(state_key)
|
||||
|
||||
# If we've the left the `set()` empty, remove it from the map
|
||||
if existing_state_key_set == set():
|
||||
self.required_state_map.pop(existing_state_type, None)
|
||||
|
||||
# If we're getting a wildcard `state_key`, get rid of any other state_keys
|
||||
# for this `state_type` since the wildcard will cover it already.
|
||||
if state_key == StateValues.WILDCARD:
|
||||
self.required_state_map[state_type] = {state_key}
|
||||
break
|
||||
# Otherwise, just add it to the set
|
||||
else:
|
||||
if self.required_state_map.get(state_type) is None:
|
||||
self.required_state_map[state_type] = {state_key}
|
||||
else:
|
||||
self.required_state_map[state_type].add(state_key)
|
||||
|
||||
def must_await_full_state(
|
||||
self,
|
||||
is_mine_id: Callable[[str], bool],
|
||||
) -> bool:
|
||||
"""
|
||||
Check if we have a we're only requesting `required_state` which is completely
|
||||
satisfied even with partial state, then we don't need to `await_full_state` before
|
||||
we can return it.
|
||||
|
||||
Also see `StateFilter.must_await_full_state(...)` for comparison
|
||||
|
||||
Partially-stated rooms should have all state events except for remote membership
|
||||
events so if we require a remote membership event anywhere, then we need to
|
||||
return `True` (requires full state).
|
||||
|
||||
Args:
|
||||
is_mine_id: a callable which confirms if a given state_key matches a mxid
|
||||
of a local user
|
||||
"""
|
||||
wildcard_state_keys = self.required_state_map.get(StateValues.WILDCARD)
|
||||
# Requesting *all* state in the room so we have to wait
|
||||
if (
|
||||
wildcard_state_keys is not None
|
||||
and StateValues.WILDCARD in wildcard_state_keys
|
||||
):
|
||||
return True
|
||||
|
||||
# If the wildcards don't refer to remote user IDs, then we don't need to wait
|
||||
# for full state.
|
||||
if wildcard_state_keys is not None:
|
||||
for possible_user_id in wildcard_state_keys:
|
||||
if not possible_user_id[0].startswith(UserID.SIGIL):
|
||||
# Not a user ID
|
||||
continue
|
||||
|
||||
localpart_hostname = possible_user_id.split(":", 1)
|
||||
if len(localpart_hostname) < 2:
|
||||
# Not a user ID
|
||||
continue
|
||||
|
||||
if not is_mine_id(possible_user_id):
|
||||
return True
|
||||
|
||||
membership_state_keys = self.required_state_map.get(EventTypes.Member)
|
||||
# We aren't requesting any membership events at all so the partial state will
|
||||
# cover us.
|
||||
if membership_state_keys is None:
|
||||
return False
|
||||
|
||||
# If we're requesting entirely local users, the partial state will cover us.
|
||||
for user_id in membership_state_keys:
|
||||
if user_id == StateValues.ME:
|
||||
continue
|
||||
# We're lazy-loading membership so we can just return the state we have.
|
||||
# Lazy-loading means we include membership for any event `sender` in the
|
||||
# timeline but since we had to auth those timeline events, we will have the
|
||||
# membership state for them (including from remote senders).
|
||||
elif user_id == StateValues.LAZY:
|
||||
continue
|
||||
elif user_id == StateValues.WILDCARD:
|
||||
return False
|
||||
elif not is_mine_id(user_id):
|
||||
return True
|
||||
|
||||
# Local users only so the partial state will cover us.
|
||||
return False
|
||||
|
||||
|
||||
class HaveSentRoomFlag(Enum):
|
||||
"""Flag for whether we have sent the room down a sliding sync connection.
|
||||
|
||||
The valid state changes here are:
|
||||
NEVER -> LIVE
|
||||
LIVE -> PREVIOUSLY
|
||||
PREVIOUSLY -> LIVE
|
||||
"""
|
||||
|
||||
# The room has never been sent down (or we have forgotten we have sent it
|
||||
# down).
|
||||
NEVER = "never"
|
||||
|
||||
# We have previously sent the room down, but there are updates that we
|
||||
# haven't sent down.
|
||||
PREVIOUSLY = "previously"
|
||||
|
||||
# We have sent the room down and the client has received all updates.
|
||||
LIVE = "live"
|
||||
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
@attr.s(auto_attribs=True, slots=True, frozen=True)
|
||||
class HaveSentRoom(Generic[T]):
|
||||
"""Whether we have sent the room data down a sliding sync connection.
|
||||
|
||||
We are generic over the type of token used, e.g. `RoomStreamToken` or
|
||||
`MultiWriterStreamToken`.
|
||||
|
||||
Attributes:
|
||||
status: Flag of if we have or haven't sent down the room
|
||||
last_token: If the flag is `PREVIOUSLY` then this is non-null and
|
||||
contains the last stream token of the last updates we sent down
|
||||
the room, i.e. we still need to send everything since then to the
|
||||
client.
|
||||
"""
|
||||
|
||||
status: HaveSentRoomFlag
|
||||
last_token: Optional[T]
|
||||
|
||||
@staticmethod
|
||||
def live() -> "HaveSentRoom[T]":
|
||||
return HaveSentRoom(HaveSentRoomFlag.LIVE, None)
|
||||
|
||||
@staticmethod
|
||||
def previously(last_token: T) -> "HaveSentRoom[T]":
|
||||
"""Constructor for `PREVIOUSLY` flag."""
|
||||
return HaveSentRoom(HaveSentRoomFlag.PREVIOUSLY, last_token)
|
||||
|
||||
@staticmethod
|
||||
def never() -> "HaveSentRoom[T]":
|
||||
return HaveSentRoom(HaveSentRoomFlag.NEVER, None)
|
||||
|
||||
|
||||
@attr.s(auto_attribs=True, slots=True, frozen=True)
|
||||
class RoomStatusMap(Generic[T]):
|
||||
"""For a given stream, e.g. events, records what we have or have not sent
|
||||
down for that stream in a given room."""
|
||||
|
||||
# `room_id` -> `HaveSentRoom`
|
||||
_statuses: Mapping[str, HaveSentRoom[T]] = attr.Factory(dict)
|
||||
|
||||
def have_sent_room(self, room_id: str) -> HaveSentRoom[T]:
|
||||
"""Return whether we have previously sent the room down"""
|
||||
return self._statuses.get(room_id, HaveSentRoom.never())
|
||||
|
||||
def get_mutable(self) -> "MutableRoomStatusMap[T]":
|
||||
"""Get a mutable copy of this state."""
|
||||
return MutableRoomStatusMap(
|
||||
statuses=self._statuses,
|
||||
)
|
||||
|
||||
def copy(self) -> "RoomStatusMap[T]":
|
||||
"""Make a copy of the class. Useful for converting from a mutable to
|
||||
immutable version."""
|
||||
|
||||
return RoomStatusMap(statuses=dict(self._statuses))
|
||||
|
||||
|
||||
class MutableRoomStatusMap(RoomStatusMap[T]):
|
||||
"""A mutable version of `RoomStatusMap`"""
|
||||
|
||||
# We use a ChainMap here so that we can easily track what has been updated
|
||||
# and what hasn't. Note that when we persist the per connection state this
|
||||
# will get flattened to a normal dict (via calling `.copy()`)
|
||||
_statuses: typing.ChainMap[str, HaveSentRoom[T]]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
statuses: Mapping[str, HaveSentRoom[T]],
|
||||
) -> None:
|
||||
# ChainMap requires a mutable mapping, but we're not actually going to
|
||||
# mutate it.
|
||||
statuses = cast(MutableMapping, statuses)
|
||||
|
||||
super().__init__(
|
||||
statuses=ChainMap({}, statuses),
|
||||
)
|
||||
|
||||
def get_updates(self) -> Mapping[str, HaveSentRoom[T]]:
|
||||
"""Return only the changes that were made"""
|
||||
return self._statuses.maps[0]
|
||||
|
||||
def record_sent_rooms(self, room_ids: StrCollection) -> None:
|
||||
"""Record that we have sent these rooms in the response"""
|
||||
for room_id in room_ids:
|
||||
current_status = self._statuses.get(room_id, HaveSentRoom.never())
|
||||
if current_status.status == HaveSentRoomFlag.LIVE:
|
||||
continue
|
||||
|
||||
self._statuses[room_id] = HaveSentRoom.live()
|
||||
|
||||
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.
|
||||
"""
|
||||
# Whether we add/update the entries for unsent rooms depends on the
|
||||
# existing entry:
|
||||
# - LIVE: We have previously sent down everything up to
|
||||
# `last_room_token, so we update the entry to be `PREVIOUSLY` with
|
||||
# `last_room_token`.
|
||||
# - PREVIOUSLY: We have previously sent down everything up to *a*
|
||||
# given token, so we don't need to update the entry.
|
||||
# - NEVER: We have never previously sent down the room, and we haven't
|
||||
# sent anything down this time either so we leave it as NEVER.
|
||||
|
||||
for room_id in room_ids:
|
||||
current_status = self._statuses.get(room_id, HaveSentRoom.never())
|
||||
if current_status.status != HaveSentRoomFlag.LIVE:
|
||||
continue
|
||||
|
||||
self._statuses[room_id] = HaveSentRoom.previously(from_token)
|
||||
|
||||
|
||||
@attr.s(auto_attribs=True)
|
||||
class PerConnectionState:
|
||||
"""The per-connection state. A snapshot of what we've sent down the
|
||||
connection before.
|
||||
|
||||
Currently, we track whether we've sent down various aspects of a given room
|
||||
before.
|
||||
|
||||
We use the `rooms` field to store the position in the events stream for each
|
||||
room that we've previously sent to the client before. On the next request
|
||||
that includes the room, we can then send only what's changed since that
|
||||
recorded position.
|
||||
|
||||
Same goes for the `receipts` field so we only need to send the new receipts
|
||||
since the last time you made a sync request.
|
||||
|
||||
Attributes:
|
||||
rooms: The status of each room for the events stream.
|
||||
receipts: The status of each room for the receipts stream.
|
||||
room_configs: Map from room_id to the `RoomSyncConfig` of all
|
||||
rooms that we have previously sent down.
|
||||
"""
|
||||
|
||||
rooms: RoomStatusMap[RoomStreamToken] = attr.Factory(RoomStatusMap)
|
||||
receipts: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap)
|
||||
|
||||
room_configs: Mapping[str, RoomSyncConfig] = attr.Factory(dict)
|
||||
|
||||
def get_mutable(self) -> "MutablePerConnectionState":
|
||||
"""Get a mutable copy of this state."""
|
||||
room_configs = cast(MutableMapping[str, RoomSyncConfig], self.room_configs)
|
||||
|
||||
return MutablePerConnectionState(
|
||||
rooms=self.rooms.get_mutable(),
|
||||
receipts=self.receipts.get_mutable(),
|
||||
room_configs=ChainMap({}, room_configs),
|
||||
)
|
||||
|
||||
def copy(self) -> "PerConnectionState":
|
||||
return PerConnectionState(
|
||||
rooms=self.rooms.copy(),
|
||||
receipts=self.receipts.copy(),
|
||||
room_configs=dict(self.room_configs),
|
||||
)
|
||||
|
||||
|
||||
@attr.s(auto_attribs=True)
|
||||
class MutablePerConnectionState(PerConnectionState):
|
||||
"""A mutable version of `PerConnectionState`"""
|
||||
|
||||
rooms: MutableRoomStatusMap[RoomStreamToken]
|
||||
receipts: MutableRoomStatusMap[MultiWriterStreamToken]
|
||||
|
||||
room_configs: typing.ChainMap[str, RoomSyncConfig]
|
||||
|
||||
def has_updates(self) -> bool:
|
||||
return (
|
||||
bool(self.rooms.get_updates())
|
||||
or bool(self.receipts.get_updates())
|
||||
or bool(self.get_room_config_updates())
|
||||
)
|
||||
|
||||
def get_room_config_updates(self) -> Mapping[str, RoomSyncConfig]:
|
||||
"""Get updates to the room sync config"""
|
||||
return self.room_configs.maps[0]
|
||||
|
|
@ -1044,6 +1044,11 @@ class SlidingSyncRestServlet(RestServlet):
|
|||
if room_result.initial:
|
||||
serialized_rooms[room_id]["initial"] = room_result.initial
|
||||
|
||||
if room_result.unstable_expanded_timeline:
|
||||
serialized_rooms[room_id][
|
||||
"unstable_expanded_timeline"
|
||||
] = room_result.unstable_expanded_timeline
|
||||
|
||||
# This will be omitted for invite/knock rooms with `stripped_state`
|
||||
if (
|
||||
room_result.required_state is not None
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ from synapse.storage.database import (
|
|||
DatabasePool,
|
||||
LoggingDatabaseConnection,
|
||||
LoggingTransaction,
|
||||
make_tuple_in_list_sql_clause,
|
||||
)
|
||||
from synapse.storage.engines._base import IsolationLevel
|
||||
from synapse.storage.util.id_generators import MultiWriterIdGenerator
|
||||
|
|
@ -481,6 +482,83 @@ class ReceiptsWorkerStore(SQLBaseStore):
|
|||
}
|
||||
return results
|
||||
|
||||
async def get_linearized_receipts_for_events(
|
||||
self,
|
||||
room_and_event_ids: Collection[Tuple[str, str]],
|
||||
) -> Sequence[JsonMapping]:
|
||||
"""Get all receipts for the given set of events.
|
||||
|
||||
Arguments:
|
||||
room_and_event_ids: A collection of 2-tuples of room ID and
|
||||
event IDs to fetch receipts for
|
||||
|
||||
Returns:
|
||||
A list of receipts, one per room.
|
||||
"""
|
||||
|
||||
def get_linearized_receipts_for_events_txn(
|
||||
txn: LoggingTransaction,
|
||||
room_id_event_id_tuples: Collection[Tuple[str, str]],
|
||||
) -> List[Tuple[str, str, str, str, Optional[str], str]]:
|
||||
clause, args = make_tuple_in_list_sql_clause(
|
||||
self.database_engine, ("room_id", "event_id"), room_id_event_id_tuples
|
||||
)
|
||||
|
||||
sql = f"""
|
||||
SELECT room_id, receipt_type, user_id, event_id, thread_id, data
|
||||
FROM receipts_linearized
|
||||
WHERE {clause}
|
||||
"""
|
||||
|
||||
txn.execute(sql, args)
|
||||
|
||||
return txn.fetchall()
|
||||
|
||||
# room_id -> event_id -> receipt_type -> user_id -> receipt data
|
||||
room_to_content: Dict[str, Dict[str, Dict[str, Dict[str, JsonMapping]]]] = {}
|
||||
for batch in batch_iter(room_and_event_ids, 1000):
|
||||
batch_results = await self.db_pool.runInteraction(
|
||||
"get_linearized_receipts_for_events",
|
||||
get_linearized_receipts_for_events_txn,
|
||||
batch,
|
||||
)
|
||||
|
||||
for (
|
||||
room_id,
|
||||
receipt_type,
|
||||
user_id,
|
||||
event_id,
|
||||
thread_id,
|
||||
data,
|
||||
) in batch_results:
|
||||
content = room_to_content.setdefault(room_id, {})
|
||||
user_receipts = content.setdefault(event_id, {}).setdefault(
|
||||
receipt_type, {}
|
||||
)
|
||||
|
||||
receipt_data = db_to_json(data)
|
||||
if thread_id is not None:
|
||||
receipt_data["thread_id"] = thread_id
|
||||
|
||||
# MSC4102: always replace threaded receipts with unthreaded ones
|
||||
# if there is a clash. Specifically:
|
||||
# - if there is no existing receipt, great, set the data.
|
||||
# - if there is an existing receipt, is it threaded (thread_id
|
||||
# present)? YES: replace if this receipt has no thread id.
|
||||
# NO: do not replace. This means we will drop some receipts, but
|
||||
# MSC4102 is designed to drop semantically meaningless receipts,
|
||||
# so this is okay. Previously, we would drop meaningful data!
|
||||
if user_id in user_receipts:
|
||||
if "thread_id" in user_receipts[user_id] and not thread_id:
|
||||
user_receipts[user_id] = receipt_data
|
||||
else:
|
||||
user_receipts[user_id] = receipt_data
|
||||
|
||||
return [
|
||||
{"type": EduTypes.RECEIPT, "room_id": room_id, "content": content}
|
||||
for room_id, content in room_to_content.items()
|
||||
]
|
||||
|
||||
@cached(
|
||||
num_args=2,
|
||||
)
|
||||
|
|
@ -996,6 +1074,12 @@ class ReceiptsBackgroundUpdateStore(SQLBaseStore):
|
|||
self.RECEIPTS_GRAPH_UNIQUE_INDEX_UPDATE_NAME,
|
||||
self._background_receipts_graph_unique_index,
|
||||
)
|
||||
self.db_pool.updates.register_background_index_update(
|
||||
update_name="receipts_room_id_event_id_index",
|
||||
index_name="receipts_linearized_event_id",
|
||||
table="receipts_linearized",
|
||||
columns=("room_id", "event_id"),
|
||||
)
|
||||
|
||||
async def _populate_receipt_event_stream_ordering(
|
||||
self, progress: JsonDict, batch_size: int
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
--
|
||||
-- This file is licensed under the Affero General Public License (AGPL) version 3.
|
||||
--
|
||||
-- Copyright (C) 2024 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>.
|
||||
|
||||
INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
|
||||
(8602, 'receipts_room_id_event_id_index', '{}');
|
||||
|
|
@ -184,6 +184,9 @@ class SlidingSyncResult:
|
|||
their local state. When there is an update, servers MUST omit this flag
|
||||
entirely and NOT send "initial":false as this is wasteful on bandwidth. The
|
||||
absence of this flag means 'false'.
|
||||
unstable_expanded_timeline: Flag which is set if we're returning more historic
|
||||
events due to the timeline limit having increased. See "XXX: Odd behavior"
|
||||
comment ing `synapse.handlers.sliding_sync`.
|
||||
required_state: The current state of the room
|
||||
timeline: Latest events in the room. The last event is the most recent.
|
||||
bundled_aggregations: A mapping of event ID to the bundled aggregations for
|
||||
|
|
@ -232,6 +235,7 @@ class SlidingSyncResult:
|
|||
heroes: Optional[List[StrippedHero]]
|
||||
is_dm: bool
|
||||
initial: bool
|
||||
unstable_expanded_timeline: bool
|
||||
# Should be empty for invite/knock rooms with `stripped_state`
|
||||
required_state: List[EventBase]
|
||||
# Should be empty for invite/knock rooms with `stripped_state`
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ from typing import List, Optional
|
|||
from twisted.test.proto_helpers import MemoryReactor
|
||||
|
||||
import synapse.rest.admin
|
||||
from synapse.api.constants import EventTypes
|
||||
from synapse.rest.client import login, room, sync
|
||||
from synapse.server import HomeServer
|
||||
from synapse.types import StreamToken, StrSequence
|
||||
|
|
@ -573,3 +574,138 @@ class SlidingSyncRoomsTimelineTestCase(SlidingSyncBase):
|
|||
|
||||
# Nothing to see for this banned user in the room in the token range
|
||||
self.assertIsNone(response_body["rooms"].get(room_id1))
|
||||
|
||||
def test_increasing_timeline_range_sends_more_messages(self) -> None:
|
||||
"""
|
||||
Test that increasing the timeline limit via room subscriptions sends the
|
||||
room down with more messages in a limited sync.
|
||||
"""
|
||||
|
||||
user1_id = self.register_user("user1", "pass")
|
||||
user1_tok = self.login(user1_id, "pass")
|
||||
|
||||
room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
|
||||
|
||||
sync_body = {
|
||||
"lists": {
|
||||
"foo-list": {
|
||||
"ranges": [[0, 1]],
|
||||
"required_state": [[EventTypes.Create, ""]],
|
||||
"timeline_limit": 1,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
message_events = []
|
||||
for _ in range(10):
|
||||
resp = self.helper.send(room_id1, "msg", tok=user1_tok)
|
||||
message_events.append(resp["event_id"])
|
||||
|
||||
# Make the first Sliding Sync request
|
||||
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
|
||||
room_response = response_body["rooms"][room_id1]
|
||||
|
||||
self.assertEqual(room_response["initial"], True)
|
||||
self.assertNotIn("unstable_expanded_timeline", room_response)
|
||||
self.assertEqual(room_response["limited"], True)
|
||||
|
||||
# We only expect the last message at first
|
||||
self._assertTimelineEqual(
|
||||
room_id=room_id1,
|
||||
actual_event_ids=[event["event_id"] for event in room_response["timeline"]],
|
||||
expected_event_ids=message_events[-1:],
|
||||
message=str(room_response["timeline"]),
|
||||
)
|
||||
|
||||
# We also expect to get the create event state.
|
||||
state_map = self.get_success(
|
||||
self.storage_controllers.state.get_current_state(room_id1)
|
||||
)
|
||||
self._assertRequiredStateIncludes(
|
||||
room_response["required_state"],
|
||||
{state_map[(EventTypes.Create, "")]},
|
||||
exact=True,
|
||||
)
|
||||
|
||||
# Now do another request with a room subscription with an increased timeline limit
|
||||
sync_body["room_subscriptions"] = {
|
||||
room_id1: {
|
||||
"required_state": [],
|
||||
"timeline_limit": 10,
|
||||
}
|
||||
}
|
||||
|
||||
response_body, from_token = self.do_sync(
|
||||
sync_body, since=from_token, tok=user1_tok
|
||||
)
|
||||
room_response = response_body["rooms"][room_id1]
|
||||
|
||||
self.assertNotIn("initial", room_response)
|
||||
self.assertEqual(room_response["unstable_expanded_timeline"], True)
|
||||
self.assertEqual(room_response["limited"], True)
|
||||
|
||||
# Now we expect all the messages
|
||||
self._assertTimelineEqual(
|
||||
room_id=room_id1,
|
||||
actual_event_ids=[event["event_id"] for event in room_response["timeline"]],
|
||||
expected_event_ids=message_events,
|
||||
message=str(room_response["timeline"]),
|
||||
)
|
||||
|
||||
# We don't expect to get the room create down, as nothing has changed.
|
||||
self.assertNotIn("required_state", room_response)
|
||||
|
||||
# Decreasing the timeline limit shouldn't resend any events
|
||||
sync_body["room_subscriptions"] = {
|
||||
room_id1: {
|
||||
"required_state": [],
|
||||
"timeline_limit": 5,
|
||||
}
|
||||
}
|
||||
|
||||
event_response = self.helper.send(room_id1, "msg", tok=user1_tok)
|
||||
latest_event_id = event_response["event_id"]
|
||||
|
||||
response_body, from_token = self.do_sync(
|
||||
sync_body, since=from_token, tok=user1_tok
|
||||
)
|
||||
room_response = response_body["rooms"][room_id1]
|
||||
|
||||
self.assertNotIn("initial", room_response)
|
||||
self.assertNotIn("unstable_expanded_timeline", room_response)
|
||||
self.assertEqual(room_response["limited"], False)
|
||||
|
||||
self._assertTimelineEqual(
|
||||
room_id=room_id1,
|
||||
actual_event_ids=[event["event_id"] for event in room_response["timeline"]],
|
||||
expected_event_ids=[latest_event_id],
|
||||
message=str(room_response["timeline"]),
|
||||
)
|
||||
|
||||
# Increasing the limit to what it was before also should not resend any
|
||||
# events
|
||||
sync_body["room_subscriptions"] = {
|
||||
room_id1: {
|
||||
"required_state": [],
|
||||
"timeline_limit": 10,
|
||||
}
|
||||
}
|
||||
|
||||
event_response = self.helper.send(room_id1, "msg", tok=user1_tok)
|
||||
latest_event_id = event_response["event_id"]
|
||||
|
||||
response_body, from_token = self.do_sync(
|
||||
sync_body, since=from_token, tok=user1_tok
|
||||
)
|
||||
room_response = response_body["rooms"][room_id1]
|
||||
|
||||
self.assertNotIn("initial", room_response)
|
||||
self.assertNotIn("unstable_expanded_timeline", room_response)
|
||||
self.assertEqual(room_response["limited"], False)
|
||||
|
||||
self._assertTimelineEqual(
|
||||
room_id=room_id1,
|
||||
actual_event_ids=[event["event_id"] for event in room_response["timeline"]],
|
||||
expected_event_ids=[latest_event_id],
|
||||
message=str(room_response["timeline"]),
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue