mirror of
https://mau.dev/maunium/synapse.git
synced 2024-10-01 01:36:05 -04:00
80d44060c9
* Allow `AbstractSet` in `StrCollection` Or else frozensets are excluded. This will be useful in an upcoming commit where I plan to change a function that accepts `List[str]` to accept `StrCollection` instead. * `rooms_to_exclude` -> `rooms_to_exclude_globally` I am about to make use of this exclusion mechanism to exclude rooms for a specific user and a specific sync. This rename helps to clarify the distinction between the global config and the rooms to exclude for a specific sync. * Better function names for internal sync methods * Track a list of excluded rooms on SyncResultBuilder I plan to feed a list of partially stated rooms for this sync to ignore * Exclude partial state rooms during eager sync using the mechanism established in the previous commit * Track un-partial-state stream in sync tokens So that we can work out which rooms have become fully-stated during a given sync period. * Fix mutation of `@cached` return value This was fouling up a complement test added alongside this PR. Excluding a room would mean the set of forgotten rooms in the cache would be extended. This means that room could be erroneously considered forgotten in the future. Introduced in #12310, Synapse 1.57.0. I don't think this had any user-visible side effects (until now). * SyncResultBuilder: track rooms to force as newly joined Similar plan as before. We've omitted rooms from certain sync responses; now we establish the mechanism to reintroduce them into future syncs. * Read new field, to present rooms as newly joined * Force un-partial-stated rooms to be newly-joined for eager incremental syncs only, provided they're still fully stated * Notify user stream listeners to wake up long polling syncs * Changelog * Typo fix Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com> * Unnecessary list cast Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com> * Rephrase comment Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com> * Another comment Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com> * Fixup merge(?) * Poke notifier when receiving un-partial-stated msg over replication * Fixup merge whoops Thanks MV :) Co-authored-by: Mathieu Velen <mathieuv@matrix.org> Co-authored-by: Mathieu Velten <mathieuv@matrix.org> Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com>
118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
# Copyright 2014-2016 OpenMarket Ltd
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from typing import TYPE_CHECKING, Iterator, Tuple
|
|
|
|
import attr
|
|
|
|
from synapse.handlers.account_data import AccountDataEventSource
|
|
from synapse.handlers.presence import PresenceEventSource
|
|
from synapse.handlers.receipts import ReceiptEventSource
|
|
from synapse.handlers.room import RoomEventSource
|
|
from synapse.handlers.typing import TypingNotificationEventSource
|
|
from synapse.logging.opentracing import trace
|
|
from synapse.streams import EventSource
|
|
from synapse.types import StreamToken
|
|
|
|
if TYPE_CHECKING:
|
|
from synapse.server import HomeServer
|
|
|
|
|
|
@attr.s(frozen=True, slots=True, auto_attribs=True)
|
|
class _EventSourcesInner:
|
|
room: RoomEventSource
|
|
presence: PresenceEventSource
|
|
typing: TypingNotificationEventSource
|
|
receipt: ReceiptEventSource
|
|
account_data: AccountDataEventSource
|
|
|
|
def get_sources(self) -> Iterator[Tuple[str, EventSource]]:
|
|
for attribute in attr.fields(_EventSourcesInner):
|
|
yield attribute.name, getattr(self, attribute.name)
|
|
|
|
|
|
class EventSources:
|
|
def __init__(self, hs: "HomeServer"):
|
|
self.sources = _EventSourcesInner(
|
|
# mypy previously warned that attribute.type is `Optional`, but we know it's
|
|
# never `None` here since all the attributes of `_EventSourcesInner` are
|
|
# annotated.
|
|
# As of the stubs in attrs 22.1.0, `attr.fields()` now returns Any,
|
|
# so the call to `attribute.type` is not checked.
|
|
*(attribute.type(hs) for attribute in attr.fields(_EventSourcesInner))
|
|
)
|
|
self.store = hs.get_datastores().main
|
|
self._instance_name = hs.get_instance_name()
|
|
|
|
def get_current_token(self) -> StreamToken:
|
|
push_rules_key = self.store.get_max_push_rules_stream_id()
|
|
to_device_key = self.store.get_to_device_stream_token()
|
|
device_list_key = self.store.get_device_stream_token()
|
|
un_partial_stated_rooms_key = self.store.get_un_partial_stated_rooms_token(
|
|
self._instance_name
|
|
)
|
|
|
|
token = StreamToken(
|
|
room_key=self.sources.room.get_current_key(),
|
|
presence_key=self.sources.presence.get_current_key(),
|
|
typing_key=self.sources.typing.get_current_key(),
|
|
receipt_key=self.sources.receipt.get_current_key(),
|
|
account_data_key=self.sources.account_data.get_current_key(),
|
|
push_rules_key=push_rules_key,
|
|
to_device_key=to_device_key,
|
|
device_list_key=device_list_key,
|
|
# Groups key is unused.
|
|
groups_key=0,
|
|
un_partial_stated_rooms_key=un_partial_stated_rooms_key,
|
|
)
|
|
return token
|
|
|
|
@trace
|
|
async def get_start_token_for_pagination(self, room_id: str) -> StreamToken:
|
|
"""Get the start token for a given room to be used to paginate
|
|
events.
|
|
|
|
The returned token does not have the current values for fields other
|
|
than `room`, since they are not used during pagination.
|
|
|
|
Returns:
|
|
The start token for pagination.
|
|
"""
|
|
return StreamToken.START
|
|
|
|
@trace
|
|
async def get_current_token_for_pagination(self, room_id: str) -> StreamToken:
|
|
"""Get the current token for a given room to be used to paginate
|
|
events.
|
|
|
|
The returned token does not have the current values for fields other
|
|
than `room`, since they are not used during pagination.
|
|
|
|
Returns:
|
|
The current token for pagination.
|
|
"""
|
|
token = StreamToken(
|
|
room_key=await self.sources.room.get_current_key_for_room(room_id),
|
|
presence_key=0,
|
|
typing_key=0,
|
|
receipt_key=0,
|
|
account_data_key=0,
|
|
push_rules_key=0,
|
|
to_device_key=0,
|
|
device_list_key=0,
|
|
groups_key=0,
|
|
un_partial_stated_rooms_key=0,
|
|
)
|
|
return token
|