diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py
index 1924636c4d..b05e32f457 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -15,29 +15,30 @@
import logging
import random
+from typing import TYPE_CHECKING, Iterable, List, Optional
from synapse.api.constants import EventTypes, Membership
from synapse.api.errors import AuthError, SynapseError
from synapse.events import EventBase
from synapse.handlers.presence import format_user_presence_state
from synapse.logging.utils import log_function
-from synapse.types import UserID
+from synapse.streams.config import PaginationConfig
+from synapse.types import JsonDict, UserID
from synapse.visibility import filter_events_for_client
from ._base import BaseHandler
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
+
logger = logging.getLogger(__name__)
class EventStreamHandler(BaseHandler):
- def __init__(self, hs):
+ def __init__(self, hs: "HomeServer"):
super(EventStreamHandler, self).__init__(hs)
- # Count of active streams per user
- self._streams_per_user = {}
- # Grace timers per user to delay the "stopped" signal
- self._stop_timer_per_user = {}
-
self.distributor = hs.get_distributor()
self.distributor.declare("started_user_eventstream")
self.distributor.declare("stopped_user_eventstream")
@@ -52,14 +53,14 @@ class EventStreamHandler(BaseHandler):
@log_function
async def get_stream(
self,
- auth_user_id,
- pagin_config,
- timeout=0,
- as_client_event=True,
- affect_presence=True,
- room_id=None,
- is_guest=False,
- ):
+ auth_user_id: str,
+ pagin_config: PaginationConfig,
+ timeout: int = 0,
+ as_client_event: bool = True,
+ affect_presence: bool = True,
+ room_id: Optional[str] = None,
+ is_guest: bool = False,
+ ) -> JsonDict:
"""Fetches the events stream for a given user.
"""
@@ -98,7 +99,7 @@ class EventStreamHandler(BaseHandler):
# When the user joins a new room, or another user joins a currently
# joined room, we need to send down presence for those users.
- to_add = []
+ to_add = [] # type: List[JsonDict]
for event in events:
if not isinstance(event, EventBase):
continue
@@ -110,7 +111,7 @@ class EventStreamHandler(BaseHandler):
# Send down presence for everyone in the room.
users = await self.state.get_current_users_in_room(
event.room_id
- )
+ ) # type: Iterable[str]
else:
users = [event.state_key]
@@ -144,20 +145,22 @@ class EventStreamHandler(BaseHandler):
class EventHandler(BaseHandler):
- def __init__(self, hs):
+ def __init__(self, hs: "HomeServer"):
super(EventHandler, self).__init__(hs)
self.storage = hs.get_storage()
- async def get_event(self, user, room_id, event_id):
+ async def get_event(
+ self, user: UserID, room_id: Optional[str], event_id: str
+ ) -> Optional[EventBase]:
"""Retrieve a single specified event.
Args:
- user (synapse.types.UserID): The user requesting the event
- room_id (str|None): The expected room id. We'll return None if the
+ user: The user requesting the event
+ room_id: The expected room id. We'll return None if the
event's room does not match.
- event_id (str): The event ID to obtain.
+ event_id: The event ID to obtain.
Returns:
- dict: An event, or None if there is no event matching this ID.
+ An event, or None if there is no event matching this ID.
Raises:
SynapseError if there was a problem retrieving this event, or
AuthError if the user does not have the rights to inspect this
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index bd8efbb768..43f2986f89 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -440,11 +440,11 @@ class FederationHandler(BaseHandler):
if not prevs - seen:
return
- latest = await self.store.get_latest_event_ids_in_room(room_id)
+ latest_list = await self.store.get_latest_event_ids_in_room(room_id)
# We add the prev events that we have seen to the latest
# list to ensure the remote server doesn't give them to us
- latest = set(latest)
+ latest = set(latest_list)
latest |= seen
logger.info(
@@ -781,7 +781,7 @@ class FederationHandler(BaseHandler):
# keys across all devices.
current_keys = [
key
- for device in cached_devices
+ for device in cached_devices.values()
for key in device.get("keys", {}).get("keys", {}).values()
]
@@ -923,8 +923,7 @@ class FederationHandler(BaseHandler):
)
)
- if ev_infos:
- await self._handle_new_events(dest, room_id, ev_infos, backfilled=True)
+ await self._handle_new_events(dest, ev_infos, backfilled=True)
# Step 2: Persist the rest of the events in the chunk one by one
events.sort(key=lambda e: e.depth)
@@ -1217,7 +1216,7 @@ class FederationHandler(BaseHandler):
event_infos.append(_NewEventInfo(event, None, auth))
await self._handle_new_events(
- destination, room_id, event_infos,
+ destination, event_infos,
)
def _sanity_check_event(self, ev):
@@ -1364,15 +1363,15 @@ class FederationHandler(BaseHandler):
)
max_stream_id = await self._persist_auth_tree(
- origin, room_id, auth_chain, state, event, room_version_obj
+ origin, auth_chain, state, event, room_version_obj
)
# We wait here until this instance has seen the events come down
# replication (if we're using replication) as the below uses caches.
+ #
+ # TODO: Currently the events stream is written to from master
await self._replication.wait_for_stream_position(
- self.config.worker.events_shard_config.get_instance(room_id),
- "events",
- max_stream_id,
+ self.config.worker.writers.events, "events", max_stream_id
)
# Check whether this room is the result of an upgrade of a room we already know
@@ -1626,7 +1625,7 @@ class FederationHandler(BaseHandler):
)
context = await self.state_handler.compute_event_context(event)
- await self.persist_events_and_notify(event.room_id, [(event, context)])
+ await self.persist_events_and_notify([(event, context)])
return event
@@ -1653,9 +1652,7 @@ class FederationHandler(BaseHandler):
await self.federation_client.send_leave(host_list, event)
context = await self.state_handler.compute_event_context(event)
- stream_id = await self.persist_events_and_notify(
- event.room_id, [(event, context)]
- )
+ stream_id = await self.persist_events_and_notify([(event, context)])
return event, stream_id
@@ -1903,7 +1900,7 @@ class FederationHandler(BaseHandler):
)
await self.persist_events_and_notify(
- event.room_id, [(event, context)], backfilled=backfilled
+ [(event, context)], backfilled=backfilled
)
except Exception:
run_in_background(
@@ -1916,7 +1913,6 @@ class FederationHandler(BaseHandler):
async def _handle_new_events(
self,
origin: str,
- room_id: str,
event_infos: Iterable[_NewEventInfo],
backfilled: bool = False,
) -> None:
@@ -1948,7 +1944,6 @@ class FederationHandler(BaseHandler):
)
await self.persist_events_and_notify(
- room_id,
[
(ev_info.event, context)
for ev_info, context in zip(event_infos, contexts)
@@ -1959,7 +1954,6 @@ class FederationHandler(BaseHandler):
async def _persist_auth_tree(
self,
origin: str,
- room_id: str,
auth_events: List[EventBase],
state: List[EventBase],
event: EventBase,
@@ -1974,7 +1968,6 @@ class FederationHandler(BaseHandler):
Args:
origin: Where the events came from
- room_id,
auth_events
state
event
@@ -2049,20 +2042,17 @@ class FederationHandler(BaseHandler):
events_to_context[e.event_id].rejected = RejectedReason.AUTH_ERROR
await self.persist_events_and_notify(
- room_id,
[
(e, events_to_context[e.event_id])
for e in itertools.chain(auth_events, state)
- ],
+ ]
)
new_event_context = await self.state_handler.compute_event_context(
event, old_state=state
)
- return await self.persist_events_and_notify(
- room_id, [(event, new_event_context)]
- )
+ return await self.persist_events_and_notify([(event, new_event_context)])
async def _prep_event(
self,
@@ -2119,8 +2109,8 @@ class FederationHandler(BaseHandler):
if backfilled or event.internal_metadata.is_outlier():
return
- extrem_ids = await self.store.get_latest_event_ids_in_room(event.room_id)
- extrem_ids = set(extrem_ids)
+ extrem_ids_list = await self.store.get_latest_event_ids_in_room(event.room_id)
+ extrem_ids = set(extrem_ids_list)
prev_event_ids = set(event.prev_event_ids())
if extrem_ids == prev_event_ids:
@@ -2913,7 +2903,6 @@ class FederationHandler(BaseHandler):
async def persist_events_and_notify(
self,
- room_id: str,
event_and_contexts: Sequence[Tuple[EventBase, EventContext]],
backfilled: bool = False,
) -> int:
@@ -2921,19 +2910,14 @@ class FederationHandler(BaseHandler):
necessary.
Args:
- room_id: The room ID of events being persisted.
- event_and_contexts: Sequence of events with their associated
- context that should be persisted. All events must belong to
- the same room.
+ event_and_contexts:
backfilled: Whether these events are a result of
backfilling or not
"""
- instance = self.config.worker.events_shard_config.get_instance(room_id)
- if instance != self._instance_name:
+ if self.config.worker.writers.events != self._instance_name:
result = await self._send_events(
- instance_name=instance,
+ instance_name=self.config.worker.writers.events,
store=self.store,
- room_id=room_id,
event_and_contexts=event_and_contexts,
backfilled=backfilled,
)
diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py
index ae6bd1d352..d5ddc583ad 100644
--- a/synapse/handlers/initial_sync.py
+++ b/synapse/handlers/initial_sync.py
@@ -14,6 +14,7 @@
# limitations under the License.
import logging
+from typing import TYPE_CHECKING
from twisted.internet import defer
@@ -22,8 +23,9 @@ from synapse.api.errors import SynapseError
from synapse.events.validator import EventValidator
from synapse.handlers.presence import format_user_presence_state
from synapse.logging.context import make_deferred_yieldable, run_in_background
+from synapse.storage.roommember import RoomsForUser
from synapse.streams.config import PaginationConfig
-from synapse.types import StreamToken, UserID
+from synapse.types import JsonDict, Requester, StreamToken, UserID
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import concurrently_execute
from synapse.util.caches.response_cache import ResponseCache
@@ -31,11 +33,15 @@ from synapse.visibility import filter_events_for_client
from ._base import BaseHandler
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
+
logger = logging.getLogger(__name__)
class InitialSyncHandler(BaseHandler):
- def __init__(self, hs):
+ def __init__(self, hs: "HomeServer"):
super(InitialSyncHandler, self).__init__(hs)
self.hs = hs
self.state = hs.get_state_handler()
@@ -48,27 +54,25 @@ class InitialSyncHandler(BaseHandler):
def snapshot_all_rooms(
self,
- user_id=None,
- pagin_config=None,
- as_client_event=True,
- include_archived=False,
- ):
+ user_id: str,
+ pagin_config: PaginationConfig,
+ as_client_event: bool = True,
+ include_archived: bool = False,
+ ) -> JsonDict:
"""Retrieve a snapshot of all rooms the user is invited or has joined.
This snapshot may include messages for all rooms where the user is
joined, depending on the pagination config.
Args:
- user_id (str): The ID of the user making the request.
- pagin_config (synapse.api.streams.PaginationConfig): The pagination
- config used to determine how many messages *PER ROOM* to return.
- as_client_event (bool): True to get events in client-server format.
- include_archived (bool): True to get rooms that the user has left
+ user_id: The ID of the user making the request.
+ pagin_config: The pagination config used to determine how many
+ messages *PER ROOM* to return.
+ as_client_event: True to get events in client-server format.
+ include_archived: True to get rooms that the user has left
Returns:
- A list of dicts with "room_id" and "membership" keys for all rooms
- the user is currently invited or joined in on. Rooms where the user
- is joined on, may return a "messages" key with messages, depending
- on the specified PaginationConfig.
+ A JsonDict with the same format as the response to `/intialSync`
+ API
"""
key = (
user_id,
@@ -91,11 +95,11 @@ class InitialSyncHandler(BaseHandler):
async def _snapshot_all_rooms(
self,
- user_id=None,
- pagin_config=None,
- as_client_event=True,
- include_archived=False,
- ):
+ user_id: str,
+ pagin_config: PaginationConfig,
+ as_client_event: bool = True,
+ include_archived: bool = False,
+ ) -> JsonDict:
memberships = [Membership.INVITE, Membership.JOIN]
if include_archived:
@@ -134,7 +138,7 @@ class InitialSyncHandler(BaseHandler):
if limit is None:
limit = 10
- async def handle_room(event):
+ async def handle_room(event: RoomsForUser):
d = {
"room_id": event.room_id,
"membership": event.membership,
@@ -251,17 +255,18 @@ class InitialSyncHandler(BaseHandler):
return ret
- async def room_initial_sync(self, requester, room_id, pagin_config=None):
+ async def room_initial_sync(
+ self, requester: Requester, room_id: str, pagin_config: PaginationConfig
+ ) -> JsonDict:
"""Capture the a snapshot of a room. If user is currently a member of
the room this will be what is currently in the room. If the user left
the room this will be what was in the room when they left.
Args:
- requester(Requester): The user to get a snapshot for.
- room_id(str): The room to get a snapshot of.
- pagin_config(synapse.streams.config.PaginationConfig):
- The pagination config used to determine how many messages to
- return.
+ requester: The user to get a snapshot for.
+ room_id: The room to get a snapshot of.
+ pagin_config: The pagination config used to determine how many
+ messages to return.
Raises:
AuthError if the user wasn't in the room.
Returns:
@@ -305,8 +310,14 @@ class InitialSyncHandler(BaseHandler):
return result
async def _room_initial_sync_parted(
- self, user_id, room_id, pagin_config, membership, member_event_id, is_peeking
- ):
+ self,
+ user_id: str,
+ room_id: str,
+ pagin_config: PaginationConfig,
+ membership: Membership,
+ member_event_id: str,
+ is_peeking: bool,
+ ) -> JsonDict:
room_state = await self.state_store.get_state_for_events([member_event_id])
room_state = room_state[member_event_id]
@@ -350,8 +361,13 @@ class InitialSyncHandler(BaseHandler):
}
async def _room_initial_sync_joined(
- self, user_id, room_id, pagin_config, membership, is_peeking
- ):
+ self,
+ user_id: str,
+ room_id: str,
+ pagin_config: PaginationConfig,
+ membership: Membership,
+ is_peeking: bool,
+ ) -> JsonDict:
current_state = await self.state.get_current_state(room_id=room_id)
# TODO: These concurrently
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 0016af44be..b6f400e9c3 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -376,8 +376,9 @@ class EventCreationHandler(object):
self.notifier = hs.get_notifier()
self.config = hs.config
self.require_membership_for_aliases = hs.config.require_membership_for_aliases
- self._events_shard_config = self.config.worker.events_shard_config
- self._instance_name = hs.get_instance_name()
+ self._is_event_writer = (
+ self.config.worker.writers.events == hs.get_instance_name()
+ )
self.room_invite_state_types = self.hs.config.room_invite_state_types
@@ -905,10 +906,9 @@ class EventCreationHandler(object):
try:
# If we're a worker we need to hit out to the master.
- writer_instance = self._events_shard_config.get_instance(event.room_id)
- if writer_instance != self._instance_name:
+ if not self._is_event_writer:
result = await self.send_event(
- instance_name=writer_instance,
+ instance_name=self.config.worker.writers.events,
event_id=event.event_id,
store=self.store,
requester=requester,
@@ -976,9 +976,7 @@ class EventCreationHandler(object):
This should only be run on the instance in charge of persisting events.
"""
- assert self._events_shard_config.should_handle(
- self._instance_name, event.room_id
- )
+ assert self._is_event_writer
if ratelimit:
# We check if this is a room admin redacting an event so that we
diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py
index 5a1aa7d830..63d7edff87 100644
--- a/synapse/handlers/pagination.py
+++ b/synapse/handlers/pagination.py
@@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Optional, Set
from twisted.python.failure import Failure
@@ -30,6 +30,10 @@ from synapse.util.async_helpers import ReadWriteLock
from synapse.util.stringutils import random_string
from synapse.visibility import filter_events_for_client
+if TYPE_CHECKING:
+ from synapse.server import HomeServer
+
+
logger = logging.getLogger(__name__)
@@ -68,7 +72,7 @@ class PaginationHandler(object):
paginating during a purge.
"""
- def __init__(self, hs):
+ def __init__(self, hs: "HomeServer"):
self.hs = hs
self.auth = hs.get_auth()
self.store = hs.get_datastore()
@@ -78,9 +82,9 @@ class PaginationHandler(object):
self._server_name = hs.hostname
self.pagination_lock = ReadWriteLock()
- self._purges_in_progress_by_room = set()
+ self._purges_in_progress_by_room = set() # type: Set[str]
# map from purge id to PurgeStatus
- self._purges_by_id = {}
+ self._purges_by_id = {} # type: Dict[str, PurgeStatus]
self._event_serializer = hs.get_event_client_serializer()
self._retention_default_max_lifetime = hs.config.retention_default_max_lifetime
@@ -102,7 +106,9 @@ class PaginationHandler(object):
job["longest_max_lifetime"],
)
- async def purge_history_for_rooms_in_range(self, min_ms, max_ms):
+ async def purge_history_for_rooms_in_range(
+ self, min_ms: Optional[int], max_ms: Optional[int]
+ ):
"""Purge outdated events from rooms within the given retention range.
If a default retention policy is defined in the server's configuration and its
@@ -110,10 +116,10 @@ class PaginationHandler(object):
retention policy.
Args:
- min_ms (int|None): Duration in milliseconds that define the lower limit of
+ min_ms: Duration in milliseconds that define the lower limit of
the range to handle (exclusive). If None, it means that the range has no
lower limit.
- max_ms (int|None): Duration in milliseconds that define the upper limit of
+ max_ms: Duration in milliseconds that define the upper limit of
the range to handle (inclusive). If None, it means that the range has no
upper limit.
"""
@@ -220,18 +226,19 @@ class PaginationHandler(object):
"_purge_history", self._purge_history, purge_id, room_id, token, True,
)
- def start_purge_history(self, room_id, token, delete_local_events=False):
+ def start_purge_history(
+ self, room_id: str, token: str, delete_local_events: bool = False
+ ) -> str:
"""Start off a history purge on a room.
Args:
- room_id (str): The room to purge from
-
- token (str): topological token to delete events before
- delete_local_events (bool): True to delete local events as well as
+ room_id: The room to purge from
+ token: topological token to delete events before
+ delete_local_events: True to delete local events as well as
remote ones
Returns:
- str: unique ID for this purge transaction.
+ unique ID for this purge transaction.
"""
if room_id in self._purges_in_progress_by_room:
raise SynapseError(
@@ -284,14 +291,11 @@ class PaginationHandler(object):
self.hs.get_reactor().callLater(24 * 3600, clear_purge)
- def get_purge_status(self, purge_id):
+ def get_purge_status(self, purge_id: str) -> Optional[PurgeStatus]:
"""Get the current status of an active purge
Args:
- purge_id (str): purge_id returned by start_purge_history
-
- Returns:
- PurgeStatus|None
+ purge_id: purge_id returned by start_purge_history
"""
return self._purges_by_id.get(purge_id)
@@ -312,8 +316,8 @@ class PaginationHandler(object):
async def get_messages(
self,
requester: Requester,
- room_id: Optional[str] = None,
- pagin_config: Optional[PaginationConfig] = None,
+ room_id: str,
+ pagin_config: PaginationConfig,
as_client_event: bool = True,
event_filter: Optional[Filter] = None,
) -> Dict[str, Any]:
@@ -368,11 +372,15 @@ class PaginationHandler(object):
# If they have left the room then clamp the token to be before
# they left the room, to save the effort of loading from the
# database.
+
+ # This is only None if the room is world_readable, in which
+ # case "JOIN" would have been returned.
+ assert member_event_id
+
leave_token = await self.store.get_topological_token_for_event(
member_event_id
)
- leave_token = RoomStreamToken.parse(leave_token)
- if leave_token.topological < max_topo:
+ if RoomStreamToken.parse(leave_token).topological < max_topo:
source_config.from_key = str(leave_token)
await self.hs.get_handlers().federation_handler.maybe_backfill(
@@ -419,8 +427,8 @@ class PaginationHandler(object):
)
if state_ids:
- state = await self.store.get_events(list(state_ids.values()))
- state = state.values()
+ state_dict = await self.store.get_events(list(state_ids.values()))
+ state = state_dict.values()
time_now = self.clock.time_msec()
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 55794c3057..9d5b1828df 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -804,9 +804,7 @@ class RoomCreationHandler(BaseHandler):
# Always wait for room creation to progate before returning
await self._replication.wait_for_stream_position(
- self.hs.config.worker.events_shard_config.get_instance(room_id),
- "events",
- last_stream_id,
+ self.hs.config.worker.writers.events, "events", last_stream_id
)
return result, last_stream_id
@@ -1262,10 +1260,10 @@ class RoomShutdownHandler(object):
# We now wait for the create room to come back in via replication so
# that we can assume that all the joins/invites have propogated before
# we try and auto join below.
+ #
+ # TODO: Currently the events stream is written to from master
await self._replication.wait_for_stream_position(
- self.hs.config.worker.events_shard_config.get_instance(new_room_id),
- "events",
- stream_id,
+ self.hs.config.worker.writers.events, "events", stream_id
)
else:
new_room_id = None
@@ -1295,9 +1293,7 @@ class RoomShutdownHandler(object):
# Wait for leave to come in over replication before trying to forget.
await self._replication.wait_for_stream_position(
- self.hs.config.worker.events_shard_config.get_instance(room_id),
- "events",
- stream_id,
+ self.hs.config.worker.writers.events, "events", stream_id
)
await self.room_member_handler.forget(target_requester.user, room_id)
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index ed1d1bd83d..1258376f20 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -83,6 +83,13 @@ class RoomMemberHandler(object):
self._enable_lookup = hs.config.enable_3pid_lookup
self.allow_per_room_profiles = self.config.allow_per_room_profiles
+ self._event_stream_writer_instance = hs.config.worker.writers.events
+ self._is_on_event_persistence_instance = (
+ self._event_stream_writer_instance == hs.get_instance_name()
+ )
+ if self._is_on_event_persistence_instance:
+ self.persist_event_storage = hs.get_storage().persistence
+
self._join_rate_limiter_local = Ratelimiter(
clock=self.clock,
rate_hz=hs.config.ratelimiting.rc_joins_local.per_second,
|
