|
@@ -17,14 +17,15 @@
|
|
|
import contextlib
|
|
|
import logging
|
|
|
import sys
|
|
|
+from typing import Dict, Iterable, Optional, Set
|
|
|
+
|
|
|
+from typing_extensions import ContextManager
|
|
|
|
|
|
from twisted.internet import defer, reactor
|
|
|
-from twisted.web.resource import NoResource
|
|
|
|
|
|
import synapse
|
|
|
import synapse.events
|
|
|
-from synapse.api.constants import EventTypes
|
|
|
-from synapse.api.errors import HttpResponseException, SynapseError
|
|
|
+from synapse.api.errors import HttpResponseException, RequestSendFailed, SynapseError
|
|
|
from synapse.api.urls import (
|
|
|
CLIENT_API_PREFIX,
|
|
|
FEDERATION_PREFIX,
|
|
@@ -38,14 +39,23 @@ from synapse.config.homeserver import HomeServerConfig
|
|
|
from synapse.config.logger import setup_logging
|
|
|
from synapse.federation import send_queue
|
|
|
from synapse.federation.transport.server import TransportLayerServer
|
|
|
-from synapse.handlers.presence import PresenceHandler, get_interested_parties
|
|
|
-from synapse.http.server import JsonResource
|
|
|
+from synapse.handlers.presence import (
|
|
|
+ BasePresenceHandler,
|
|
|
+ PresenceState,
|
|
|
+ get_interested_parties,
|
|
|
+)
|
|
|
+from synapse.http.server import JsonResource, OptionsResource
|
|
|
from synapse.http.servlet import RestServlet, parse_json_object_from_request
|
|
|
from synapse.http.site import SynapseSite
|
|
|
-from synapse.logging.context import LoggingContext, run_in_background
|
|
|
+from synapse.logging.context import LoggingContext
|
|
|
from synapse.metrics import METRICS_PREFIX, MetricsResource, RegistryProxy
|
|
|
from synapse.metrics.background_process_metrics import run_as_background_process
|
|
|
-from synapse.replication.slave.storage._base import BaseSlavedStore, __func__
|
|
|
+from synapse.replication.http import REPLICATION_PREFIX, ReplicationRestResource
|
|
|
+from synapse.replication.http.presence import (
|
|
|
+ ReplicationBumpPresenceActiveTime,
|
|
|
+ ReplicationPresenceSetState,
|
|
|
+)
|
|
|
+from synapse.replication.slave.storage._base import BaseSlavedStore
|
|
|
from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
|
|
|
from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
|
|
|
from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
|
|
@@ -64,13 +74,20 @@ from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
|
|
|
from synapse.replication.slave.storage.registration import SlavedRegistrationStore
|
|
|
from synapse.replication.slave.storage.room import RoomStore
|
|
|
from synapse.replication.slave.storage.transactions import SlavedTransactionStore
|
|
|
-from synapse.replication.tcp.client import ReplicationClientHandler
|
|
|
-from synapse.replication.tcp.streams._base import (
|
|
|
+from synapse.replication.tcp.client import ReplicationDataHandler
|
|
|
+from synapse.replication.tcp.commands import ClearUserSyncsCommand
|
|
|
+from synapse.replication.tcp.streams import (
|
|
|
+ AccountDataStream,
|
|
|
DeviceListsStream,
|
|
|
+ GroupServerStream,
|
|
|
+ PresenceStream,
|
|
|
+ PushersStream,
|
|
|
+ PushRulesStream,
|
|
|
ReceiptsStream,
|
|
|
+ TagAccountDataStream,
|
|
|
ToDeviceStream,
|
|
|
+ TypingStream,
|
|
|
)
|
|
|
-from synapse.replication.tcp.streams.events import EventsStreamEventRow, EventsStreamRow
|
|
|
from synapse.rest.admin import register_servlets_for_media_repo
|
|
|
from synapse.rest.client.v1 import events
|
|
|
from synapse.rest.client.v1.initial_sync import InitialSyncRestServlet
|
|
@@ -107,17 +124,19 @@ from synapse.rest.client.v2_alpha.register import RegisterRestServlet
|
|
|
from synapse.rest.client.versions import VersionsRestServlet
|
|
|
from synapse.rest.key.v2 import KeyApiV2Resource
|
|
|
from synapse.server import HomeServer
|
|
|
+from synapse.storage.data_stores.main.censor_events import CensorEventsStore
|
|
|
from synapse.storage.data_stores.main.media_repository import MediaRepositoryStore
|
|
|
from synapse.storage.data_stores.main.monthly_active_users import (
|
|
|
MonthlyActiveUsersWorkerStore,
|
|
|
)
|
|
|
from synapse.storage.data_stores.main.presence import UserPresenceState
|
|
|
+from synapse.storage.data_stores.main.search import SearchWorkerStore
|
|
|
+from synapse.storage.data_stores.main.ui_auth import UIAuthWorkerStore
|
|
|
from synapse.storage.data_stores.main.user_directory import UserDirectoryStore
|
|
|
from synapse.types import ReadReceipt
|
|
|
from synapse.util.async_helpers import Linearizer
|
|
|
from synapse.util.httpresourcetree import create_resource_tree
|
|
|
from synapse.util.manhole import manhole
|
|
|
-from synapse.util.stringutils import random_string
|
|
|
from synapse.util.versionstring import get_version_string
|
|
|
|
|
|
logger = logging.getLogger("synapse.app.generic_worker")
|
|
@@ -125,31 +144,18 @@ logger = logging.getLogger("synapse.app.generic_worker")
|
|
|
|
|
|
class PresenceStatusStubServlet(RestServlet):
|
|
|
"""If presence is disabled this servlet can be used to stub out setting
|
|
|
- presence status, while proxying the getters to the master instance.
|
|
|
+ presence status.
|
|
|
"""
|
|
|
|
|
|
PATTERNS = client_patterns("/presence/(?P<user_id>[^/]*)/status")
|
|
|
|
|
|
def __init__(self, hs):
|
|
|
super(PresenceStatusStubServlet, self).__init__()
|
|
|
- self.http_client = hs.get_simple_http_client()
|
|
|
self.auth = hs.get_auth()
|
|
|
- self.main_uri = hs.config.worker_main_http_uri
|
|
|
|
|
|
async def on_GET(self, request, user_id):
|
|
|
- # Pass through the auth headers, if any, in case the access token
|
|
|
- # is there.
|
|
|
- auth_headers = request.requestHeaders.getRawHeaders("Authorization", [])
|
|
|
- headers = {"Authorization": auth_headers}
|
|
|
-
|
|
|
- try:
|
|
|
- result = await self.http_client.get_json(
|
|
|
- self.main_uri + request.uri.decode("ascii"), headers=headers
|
|
|
- )
|
|
|
- except HttpResponseException as e:
|
|
|
- raise e.to_synapse_error()
|
|
|
-
|
|
|
- return 200, result
|
|
|
+ await self.auth.get_user_by_req(request)
|
|
|
+ return 200, {"presence": "offline"}
|
|
|
|
|
|
async def on_PUT(self, request, user_id):
|
|
|
await self.auth.get_user_by_req(request)
|
|
@@ -203,9 +209,14 @@ class KeyUploadServlet(RestServlet):
|
|
|
# is there.
|
|
|
auth_headers = request.requestHeaders.getRawHeaders(b"Authorization", [])
|
|
|
headers = {"Authorization": auth_headers}
|
|
|
- result = await self.http_client.post_json_get_json(
|
|
|
- self.main_uri + request.uri.decode("ascii"), body, headers=headers
|
|
|
- )
|
|
|
+ try:
|
|
|
+ result = await self.http_client.post_json_get_json(
|
|
|
+ self.main_uri + request.uri.decode("ascii"), body, headers=headers
|
|
|
+ )
|
|
|
+ except HttpResponseException as e:
|
|
|
+ raise e.to_synapse_error() from e
|
|
|
+ except RequestSendFailed as e:
|
|
|
+ raise SynapseError(502, "Failed to talk to master") from e
|
|
|
|
|
|
return 200, result
|
|
|
else:
|
|
@@ -214,37 +225,61 @@ class KeyUploadServlet(RestServlet):
|
|
|
return 200, {"one_time_key_counts": result}
|
|
|
|
|
|
|
|
|
+class _NullContextManager(ContextManager[None]):
|
|
|
+ """A context manager which does nothing."""
|
|
|
+
|
|
|
+ def __exit__(self, exc_type, exc_val, exc_tb):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
UPDATE_SYNCING_USERS_MS = 10 * 1000
|
|
|
|
|
|
|
|
|
-class GenericWorkerPresence(object):
|
|
|
+class GenericWorkerPresence(BasePresenceHandler):
|
|
|
def __init__(self, hs):
|
|
|
+ super().__init__(hs)
|
|
|
self.hs = hs
|
|
|
self.is_mine_id = hs.is_mine_id
|
|
|
self.http_client = hs.get_simple_http_client()
|
|
|
- self.store = hs.get_datastore()
|
|
|
- self.user_to_num_current_syncs = {}
|
|
|
- self.clock = hs.get_clock()
|
|
|
- self.notifier = hs.get_notifier()
|
|
|
|
|
|
- active_presence = self.store.take_presence_startup_info()
|
|
|
- self.user_to_current_state = {state.user_id: state for state in active_presence}
|
|
|
+ self._presence_enabled = hs.config.use_presence
|
|
|
+
|
|
|
+ # The number of ongoing syncs on this process, by user id.
|
|
|
+ # Empty if _presence_enabled is false.
|
|
|
+ self._user_to_num_current_syncs = {} # type: Dict[str, int]
|
|
|
+
|
|
|
+ self.notifier = hs.get_notifier()
|
|
|
+ self.instance_id = hs.get_instance_id()
|
|
|
|
|
|
# user_id -> last_sync_ms. Lists the users that have stopped syncing
|
|
|
# but we haven't notified the master of that yet
|
|
|
self.users_going_offline = {}
|
|
|
|
|
|
+ self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs)
|
|
|
+ self._set_state_client = ReplicationPresenceSetState.make_client(hs)
|
|
|
+
|
|
|
self._send_stop_syncing_loop = self.clock.looping_call(
|
|
|
self.send_stop_syncing, UPDATE_SYNCING_USERS_MS
|
|
|
)
|
|
|
|
|
|
- self.process_id = random_string(16)
|
|
|
- logger.info("Presence process_id is %r", self.process_id)
|
|
|
+ hs.get_reactor().addSystemEventTrigger(
|
|
|
+ "before",
|
|
|
+ "shutdown",
|
|
|
+ run_as_background_process,
|
|
|
+ "generic_presence.on_shutdown",
|
|
|
+ self._on_shutdown,
|
|
|
+ )
|
|
|
+
|
|
|
+ def _on_shutdown(self):
|
|
|
+ if self._presence_enabled:
|
|
|
+ self.hs.get_tcp_replication().send_command(
|
|
|
+ ClearUserSyncsCommand(self.instance_id)
|
|
|
+ )
|
|
|
|
|
|
def send_user_sync(self, user_id, is_syncing, last_sync_ms):
|
|
|
- if self.hs.config.use_presence:
|
|
|
+ if self._presence_enabled:
|
|
|
self.hs.get_tcp_replication().send_user_sync(
|
|
|
- user_id, is_syncing, last_sync_ms
|
|
|
+ self.instance_id, user_id, is_syncing, last_sync_ms
|
|
|
)
|
|
|
|
|
|
def mark_as_coming_online(self, user_id):
|
|
@@ -280,32 +315,33 @@ class GenericWorkerPresence(object):
|
|
|
self.users_going_offline.pop(user_id, None)
|
|
|
self.send_user_sync(user_id, False, last_sync_ms)
|
|
|
|
|
|
- def set_state(self, user, state, ignore_status_msg=False):
|
|
|
- # TODO Hows this supposed to work?
|
|
|
- return defer.succeed(None)
|
|
|
+ async def user_syncing(
|
|
|
+ self, user_id: str, affect_presence: bool
|
|
|
+ ) -> ContextManager[None]:
|
|
|
+ """Record that a user is syncing.
|
|
|
|
|
|
- get_states = __func__(PresenceHandler.get_states)
|
|
|
- get_state = __func__(PresenceHandler.get_state)
|
|
|
- current_state_for_users = __func__(PresenceHandler.current_state_for_users)
|
|
|
+ Called by the sync and events servlets to record that a user has connected to
|
|
|
+ this worker and is waiting for some events.
|
|
|
+ """
|
|
|
+ if not affect_presence or not self._presence_enabled:
|
|
|
+ return _NullContextManager()
|
|
|
|
|
|
- def user_syncing(self, user_id, affect_presence):
|
|
|
- if affect_presence:
|
|
|
- curr_sync = self.user_to_num_current_syncs.get(user_id, 0)
|
|
|
- self.user_to_num_current_syncs[user_id] = curr_sync + 1
|
|
|
+ curr_sync = self._user_to_num_current_syncs.get(user_id, 0)
|
|
|
+ self._user_to_num_current_syncs[user_id] = curr_sync + 1
|
|
|
|
|
|
- # If we went from no in flight sync to some, notify replication
|
|
|
- if self.user_to_num_current_syncs[user_id] == 1:
|
|
|
- self.mark_as_coming_online(user_id)
|
|
|
+ # If we went from no in flight sync to some, notify replication
|
|
|
+ if self._user_to_num_current_syncs[user_id] == 1:
|
|
|
+ self.mark_as_coming_online(user_id)
|
|
|
|
|
|
def _end():
|
|
|
# We check that the user_id is in user_to_num_current_syncs because
|
|
|
# user_to_num_current_syncs may have been cleared if we are
|
|
|
# shutting down.
|
|
|
- if affect_presence and user_id in self.user_to_num_current_syncs:
|
|
|
- self.user_to_num_current_syncs[user_id] -= 1
|
|
|
+ if user_id in self._user_to_num_current_syncs:
|
|
|
+ self._user_to_num_current_syncs[user_id] -= 1
|
|
|
|
|
|
# If we went from one in flight sync to non, notify replication
|
|
|
- if self.user_to_num_current_syncs[user_id] == 0:
|
|
|
+ if self._user_to_num_current_syncs[user_id] == 0:
|
|
|
self.mark_as_going_offline(user_id)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
@@ -315,7 +351,7 @@ class GenericWorkerPresence(object):
|
|
|
finally:
|
|
|
_end()
|
|
|
|
|
|
- return defer.succeed(_user_syncing())
|
|
|
+ return _user_syncing()
|
|
|
|
|
|
@defer.inlineCallbacks
|
|
|
def notify_from_replication(self, states, stream_id):
|
|
@@ -350,15 +386,48 @@ class GenericWorkerPresence(object):
|
|
|
stream_id = token
|
|
|
yield self.notify_from_replication(states, stream_id)
|
|
|
|
|
|
- def get_currently_syncing_users(self):
|
|
|
- if self.hs.config.use_presence:
|
|
|
- return [
|
|
|
- user_id
|
|
|
- for user_id, count in self.user_to_num_current_syncs.items()
|
|
|
- if count > 0
|
|
|
- ]
|
|
|
- else:
|
|
|
- return set()
|
|
|
+ def get_currently_syncing_users_for_replication(self) -> Iterable[str]:
|
|
|
+ return [
|
|
|
+ user_id
|
|
|
+ for user_id, count in self._user_to_num_current_syncs.items()
|
|
|
+ if count > 0
|
|
|
+ ]
|
|
|
+
|
|
|
+ async def set_state(self, target_user, state, ignore_status_msg=False):
|
|
|
+ """Set the presence state of the user.
|
|
|
+ """
|
|
|
+ presence = state["presence"]
|
|
|
+
|
|
|
+ valid_presence = (
|
|
|
+ PresenceState.ONLINE,
|
|
|
+ PresenceState.UNAVAILABLE,
|
|
|
+ PresenceState.OFFLINE,
|
|
|
+ )
|
|
|
+ if presence not in valid_presence:
|
|
|
+ raise SynapseError(400, "Invalid presence state")
|
|
|
+
|
|
|
+ user_id = target_user.to_string()
|
|
|
+
|
|
|
+ # If presence is disabled, no-op
|
|
|
+ if not self.hs.config.use_presence:
|
|
|
+ return
|
|
|
+
|
|
|
+ # Proxy request to master
|
|
|
+ await self._set_state_client(
|
|
|
+ user_id=user_id, state=state, ignore_status_msg=ignore_status_msg
|
|
|
+ )
|
|
|
+
|
|
|
+ async def bump_presence_active_time(self, user):
|
|
|
+ """We've seen the user do something that indicates they're interacting
|
|
|
+ with the app.
|
|
|
+ """
|
|
|
+ # If presence is disabled, no-op
|
|
|
+ if not self.hs.config.use_presence:
|
|
|
+ return
|
|
|
+
|
|
|
+ # Proxy request to master
|
|
|
+ user_id = user.to_string()
|
|
|
+ await self._bump_active_client(user_id=user_id)
|
|
|
|
|
|
|
|
|
class GenericWorkerTyping(object):
|
|
@@ -375,12 +444,6 @@ class GenericWorkerTyping(object):
|
|
|
# map room IDs to sets of users currently typing
|
|
|
self._room_typing = {}
|
|
|
|
|
|
- def stream_positions(self):
|
|
|
- # We must update this typing token from the response of the previous
|
|
|
- # sync. In particular, the stream id may "reset" back to zero/a low
|
|
|
- # value which we *must* use for the next replication request.
|
|
|
- return {"typing": self._latest_room_serial}
|
|
|
-
|
|
|
def process_replication_rows(self, token, rows):
|
|
|
if self._latest_room_serial > token:
|
|
|
# The master has gone backwards. To prevent inconsistent data, just
|
|
@@ -394,11 +457,15 @@ class GenericWorkerTyping(object):
|
|
|
self._room_serials[row.room_id] = token
|
|
|
self._room_typing[row.room_id] = row.user_ids
|
|
|
|
|
|
+ def get_current_token(self) -> int:
|
|
|
+ return self._latest_room_serial
|
|
|
+
|
|
|
|
|
|
class GenericWorkerSlavedStore(
|
|
|
# FIXME(#3714): We need to add UserDirectoryStore as we write directly
|
|
|
# rather than going via the correct worker.
|
|
|
UserDirectoryStore,
|
|
|
+ UIAuthWorkerStore,
|
|
|
SlavedDeviceInboxStore,
|
|
|
SlavedDeviceStore,
|
|
|
SlavedReceiptsStore,
|
|
@@ -406,6 +473,7 @@ class GenericWorkerSlavedStore(
|
|
|
SlavedGroupServerStore,
|
|
|
SlavedAccountDataStore,
|
|
|
SlavedPusherStore,
|
|
|
+ CensorEventsStore,
|
|
|
SlavedEventStore,
|
|
|
SlavedKeyStore,
|
|
|
RoomStore,
|
|
@@ -419,6 +487,7 @@ class GenericWorkerSlavedStore(
|
|
|
SlavedFilteringStore,
|
|
|
MonthlyActiveUsersWorkerStore,
|
|
|
MediaRepositoryStore,
|
|
|
+ SearchWorkerStore,
|
|
|
BaseSlavedStore,
|
|
|
):
|
|
|
def __init__(self, database, db_conn, hs):
|
|
@@ -536,7 +605,10 @@ class GenericWorkerServer(HomeServer):
|
|
|
if name in ["keys", "federation"]:
|
|
|
resources[SERVER_KEY_V2_PREFIX] = KeyApiV2Resource(self)
|
|
|
|
|
|
- root_resource = create_resource_tree(resources, NoResource())
|
|
|
+ if name == "replication":
|
|
|
+ resources[REPLICATION_PREFIX] = ReplicationRestResource(self)
|
|
|
+
|
|
|
+ root_resource = create_resource_tree(resources, OptionsResource())
|
|
|
|
|
|
_base.listen_tcp(
|
|
|
bind_addresses,
|
|
@@ -583,7 +655,7 @@ class GenericWorkerServer(HomeServer):
|
|
|
def remove_pusher(self, app_id, push_key, user_id):
|
|
|
self.get_tcp_replication().send_remove_pusher(app_id, push_key, user_id)
|
|
|
|
|
|
- def build_tcp_replication(self):
|
|
|
+ def build_replication_data_handler(self):
|
|
|
return GenericWorkerReplicationHandler(self)
|
|
|
|
|
|
def build_presence_handler(self):
|
|
@@ -593,105 +665,71 @@ class GenericWorkerServer(HomeServer):
|
|
|
return GenericWorkerTyping(self)
|
|
|
|
|
|
|
|
|
-class GenericWorkerReplicationHandler(ReplicationClientHandler):
|
|
|
+class GenericWorkerReplicationHandler(ReplicationDataHandler):
|
|
|
def __init__(self, hs):
|
|
|
- super(GenericWorkerReplicationHandler, self).__init__(hs.get_datastore())
|
|
|
+ super(GenericWorkerReplicationHandler, self).__init__(hs)
|
|
|
|
|
|
self.store = hs.get_datastore()
|
|
|
self.typing_handler = hs.get_typing_handler()
|
|
|
- # NB this is a SynchrotronPresence, not a normal PresenceHandler
|
|
|
- self.presence_handler = hs.get_presence_handler()
|
|
|
+ self.presence_handler = hs.get_presence_handler() # type: GenericWorkerPresence
|
|
|
self.notifier = hs.get_notifier()
|
|
|
|
|
|
self.notify_pushers = hs.config.start_pushers
|
|
|
self.pusher_pool = hs.get_pusherpool()
|
|
|
|
|
|
+ self.send_handler = None # type: Optional[FederationSenderHandler]
|
|
|
if hs.config.send_federation:
|
|
|
- self.send_handler = FederationSenderHandler(hs, self)
|
|
|
- else:
|
|
|
- self.send_handler = None
|
|
|
-
|
|
|
- async def on_rdata(self, stream_name, token, rows):
|
|
|
- await super(GenericWorkerReplicationHandler, self).on_rdata(
|
|
|
- stream_name, token, rows
|
|
|
- )
|
|
|
- run_in_background(self.process_and_notify, stream_name, token, rows)
|
|
|
-
|
|
|
- def get_streams_to_replicate(self):
|
|
|
- args = super(GenericWorkerReplicationHandler, self).get_streams_to_replicate()
|
|
|
- args.update(self.typing_handler.stream_positions())
|
|
|
- if self.send_handler:
|
|
|
- args.update(self.send_handler.stream_positions())
|
|
|
- return args
|
|
|
+ self.send_handler = FederationSenderHandler(hs)
|
|
|
|
|
|
- def get_currently_syncing_users(self):
|
|
|
- return self.presence_handler.get_currently_syncing_users()
|
|
|
+ async def on_rdata(self, stream_name, instance_name, token, rows):
|
|
|
+ await super().on_rdata(stream_name, instance_name, token, rows)
|
|
|
+ await self._process_and_notify(stream_name, instance_name, token, rows)
|
|
|
|
|
|
- async def process_and_notify(self, stream_name, token, rows):
|
|
|
+ async def _process_and_notify(self, stream_name, instance_name, token, rows):
|
|
|
try:
|
|
|
if self.send_handler:
|
|
|
- self.send_handler.process_replication_rows(stream_name, token, rows)
|
|
|
-
|
|
|
- if stream_name == "events":
|
|
|
- # We shouldn't get multiple rows per token for events stream, so
|
|
|
- # we don't need to optimise this for multiple rows.
|
|
|
- for row in rows:
|
|
|
- if row.type != EventsStreamEventRow.TypeId:
|
|
|
- continue
|
|
|
- assert isinstance(row, EventsStreamRow)
|
|
|
-
|
|
|
- event = await self.store.get_event(
|
|
|
- row.data.event_id, allow_rejected=True
|
|
|
- )
|
|
|
- if event.rejected_reason:
|
|
|
- continue
|
|
|
-
|
|
|
- extra_users = ()
|
|
|
- if event.type == EventTypes.Member:
|
|
|
- extra_users = (event.state_key,)
|
|
|
- max_token = self.store.get_room_max_stream_ordering()
|
|
|
- self.notifier.on_new_room_event(
|
|
|
- event, token, max_token, extra_users
|
|
|
- )
|
|
|
+ await self.send_handler.process_replication_rows(
|
|
|
+ stream_name, token, rows
|
|
|
+ )
|
|
|
|
|
|
- await self.pusher_pool.on_new_notifications(token, token)
|
|
|
- elif stream_name == "push_rules":
|
|
|
+ if stream_name == PushRulesStream.NAME:
|
|
|
self.notifier.on_new_event(
|
|
|
"push_rules_key", token, users=[row.user_id for row in rows]
|
|
|
)
|
|
|
- elif stream_name in ("account_data", "tag_account_data"):
|
|
|
+ elif stream_name in (AccountDataStream.NAME, TagAccountDataStream.NAME):
|
|
|
self.notifier.on_new_event(
|
|
|
"account_data_key", token, users=[row.user_id for row in rows]
|
|
|
)
|
|
|
- elif stream_name == "receipts":
|
|
|
+ elif stream_name == ReceiptsStream.NAME:
|
|
|
self.notifier.on_new_event(
|
|
|
"receipt_key", token, rooms=[row.room_id for row in rows]
|
|
|
)
|
|
|
await self.pusher_pool.on_new_receipts(
|
|
|
token, token, {row.room_id for row in rows}
|
|
|
)
|
|
|
- elif stream_name == "typing":
|
|
|
+ elif stream_name == TypingStream.NAME:
|
|
|
self.typing_handler.process_replication_rows(token, rows)
|
|
|
self.notifier.on_new_event(
|
|
|
"typing_key", token, rooms=[row.room_id for row in rows]
|
|
|
)
|
|
|
- elif stream_name == "to_device":
|
|
|
+ elif stream_name == ToDeviceStream.NAME:
|
|
|
entities = [row.entity for row in rows if row.entity.startswith("@")]
|
|
|
if entities:
|
|
|
self.notifier.on_new_event("to_device_key", token, users=entities)
|
|
|
- elif stream_name == "device_lists":
|
|
|
- all_room_ids = set()
|
|
|
+ elif stream_name == DeviceListsStream.NAME:
|
|
|
+ all_room_ids = set() # type: Set[str]
|
|
|
for row in rows:
|
|
|
- room_ids = await self.store.get_rooms_for_user(row.user_id)
|
|
|
- all_room_ids.update(room_ids)
|
|
|
+ if row.entity.startswith("@"):
|
|
|
+ room_ids = await self.store.get_rooms_for_user(row.entity)
|
|
|
+ all_room_ids.update(room_ids)
|
|
|
self.notifier.on_new_event("device_list_key", token, rooms=all_room_ids)
|
|
|
- elif stream_name == "presence":
|
|
|
+ elif stream_name == PresenceStream.NAME:
|
|
|
await self.presence_handler.process_replication_rows(token, rows)
|
|
|
- elif stream_name == "receipts":
|
|
|
+ elif stream_name == GroupServerStream.NAME:
|
|
|
self.notifier.on_new_event(
|
|
|
"groups_key", token, users=[row.user_id for row in rows]
|
|
|
)
|
|
|
- elif stream_name == "pushers":
|
|
|
+ elif stream_name == PushersStream.NAME:
|
|
|
for row in rows:
|
|
|
if row.deleted:
|
|
|
self.stop_pusher(row.user_id, row.app_id, row.pushkey)
|
|
@@ -730,24 +768,33 @@ class GenericWorkerReplicationHandler(ReplicationClientHandler):
|
|
|
|
|
|
|
|
|
class FederationSenderHandler(object):
|
|
|
- """Processes the replication stream and forwards the appropriate entries
|
|
|
- to the federation sender.
|
|
|
+ """Processes the fedration replication stream
|
|
|
+
|
|
|
+ This class is only instantiate on the worker responsible for sending outbound
|
|
|
+ federation transactions. It receives rows from the replication stream and forwards
|
|
|
+ the appropriate entries to the FederationSender class.
|
|
|
"""
|
|
|
|
|
|
- def __init__(self, hs: GenericWorkerServer, replication_client):
|
|
|
+ def __init__(self, hs: GenericWorkerServer):
|
|
|
self.store = hs.get_datastore()
|
|
|
self._is_mine_id = hs.is_mine_id
|
|
|
self.federation_sender = hs.get_federation_sender()
|
|
|
- self.replication_client = replication_client
|
|
|
-
|
|
|
+ self._hs = hs
|
|
|
+
|
|
|
+ # if the worker is restarted, we want to pick up where we left off in
|
|
|
+ # the replication stream, so load the position from the database.
|
|
|
+ #
|
|
|
+ # XXX is this actually worthwhile? Whenever the master is restarted, we'll
|
|
|
+ # drop some rows anyway (which is mostly fine because we're only dropping
|
|
|
+ # typing and presence notifications). If the replication stream is
|
|
|
+ # unreliable, why do we do all this hoop-jumping to store the position in the
|
|
|
+ # database? See also https://github.com/matrix-org/synapse/issues/7535.
|
|
|
+ #
|
|
|
self.federation_position = self.store.federation_out_pos_startup
|
|
|
- self._fed_position_linearizer = Linearizer(name="_fed_position_linearizer")
|
|
|
|
|
|
+ self._fed_position_linearizer = Linearizer(name="_fed_position_linearizer")
|
|
|
self._last_ack = self.federation_position
|
|
|
|
|
|
- self._room_serials = {}
|
|
|
- self._room_typing = {}
|
|
|
-
|
|
|
def on_start(self):
|
|
|
# There may be some events that are persisted but haven't been sent,
|
|
|
# so send them now.
|
|
@@ -758,15 +805,12 @@ class FederationSenderHandler(object):
|
|
|
def wake_destination(self, server: str):
|
|
|
self.federation_sender.wake_destination(server)
|
|
|
|
|
|
- def stream_positions(self):
|
|
|
- return {"federation": self.federation_position}
|
|
|
-
|
|
|
- def process_replication_rows(self, stream_name, token, rows):
|
|
|
+ async def process_replication_rows(self, stream_name, token, rows):
|
|
|
# The federation stream contains things that we want to send out, e.g.
|
|
|
# presence, typing, etc.
|
|
|
if stream_name == "federation":
|
|
|
send_queue.process_rows_for_federation(self.federation_sender, rows)
|
|
|
- run_in_background(self.update_token, token)
|
|
|
+ await self.update_token(token)
|
|
|
|
|
|
# We also need to poke the federation sender when new events happen
|
|
|
elif stream_name == "events":
|
|
@@ -774,13 +818,14 @@ class FederationSenderHandler(object):
|
|
|
|
|
|
# ... and when new receipts happen
|
|
|
elif stream_name == ReceiptsStream.NAME:
|
|
|
- run_as_background_process(
|
|
|
- "process_receipts_for_federation", self._on_new_receipts, rows
|
|
|
- )
|
|
|
+ await self._on_new_receipts(rows)
|
|
|
|
|
|
# ... as well as device updates and messages
|
|
|
elif stream_name == DeviceListsStream.NAME:
|
|
|
- hosts = {row.destination for row in rows}
|
|
|
+ # The entities are either user IDs (starting with '@') whose devices
|
|
|
+ # have changed, or remote servers that we need to tell about
|
|
|
+ # changes.
|
|
|
+ hosts = {row.entity for row in rows if not row.entity.startswith("@")}
|
|
|
for host in hosts:
|
|
|
self.federation_sender.send_device_messages(host)
|
|
|
|
|
@@ -795,7 +840,7 @@ class FederationSenderHandler(object):
|
|
|
async def _on_new_receipts(self, rows):
|
|
|
"""
|
|
|
Args:
|
|
|
- rows (iterable[synapse.replication.tcp.streams.ReceiptsStreamRow]):
|
|
|
+ rows (Iterable[synapse.replication.tcp.streams.ReceiptsStream.ReceiptsStreamRow]):
|
|
|
new receipts to be processed
|
|
|
"""
|
|
|
for receipt in rows:
|
|
@@ -812,22 +857,51 @@ class FederationSenderHandler(object):
|
|
|
await self.federation_sender.send_read_receipt(receipt_info)
|
|
|
|
|
|
async def update_token(self, token):
|
|
|
- try:
|
|
|
- self.federation_position = token
|
|
|
+ """Update the record of where we have processed to in the federation stream.
|
|
|
+
|
|
|
+ Called after we have processed a an update received over replication. Sends
|
|
|
+ a FEDERATION_ACK back to the master, and stores the token that we have processed
|
|
|
+ in `federation_stream_position` so that we can restart where we left off.
|
|
|
+ """
|
|
|
+ self.federation_position = token
|
|
|
+
|
|
|
+ # We save and send the ACK to master asynchronously, so we don't block
|
|
|
+ # processing on persistence. We don't need to do this operation for
|
|
|
+ # every single RDATA we receive, we just need to do it periodically.
|
|
|
+
|
|
|
+ if self._fed_position_linearizer.is_queued(None):
|
|
|
+ # There is already a task queued up to save and send the token, so
|
|
|
+ # no need to queue up another task.
|
|
|
+ return
|
|
|
|
|
|
+ run_as_background_process("_save_and_send_ack", self._save_and_send_ack)
|
|
|
+
|
|
|
+ async def _save_and_send_ack(self):
|
|
|
+ """Save the current federation position in the database and send an ACK
|
|
|
+ to master with where we're up to.
|
|
|
+ """
|
|
|
+ try:
|
|
|
# We linearize here to ensure we don't have races updating the token
|
|
|
+ #
|
|
|
+ # XXX this appears to be redundant, since the ReplicationCommandHandler
|
|
|
+ # has a linearizer which ensures that we only process one line of
|
|
|
+ # replication data at a time. Should we remove it, or is it doing useful
|
|
|
+ # service for robustness? Or could we replace it with an assertion that
|
|
|
+ # we're not being re-entered?
|
|
|
+
|
|
|
with (await self._fed_position_linearizer.queue(None)):
|
|
|
- if self._last_ack < self.federation_position:
|
|
|
- await self.store.update_federation_out_pos(
|
|
|
- "federation", self.federation_position
|
|
|
- )
|
|
|
+ # We persist and ack the same position, so we take a copy of it
|
|
|
+ # here as otherwise it can get modified from underneath us.
|
|
|
+ current_position = self.federation_position
|
|
|
|
|
|
- # We ACK this token over replication so that the master can drop
|
|
|
- # its in memory queues
|
|
|
- self.replication_client.send_federation_ack(
|
|
|
- self.federation_position
|
|
|
- )
|
|
|
- self._last_ack = self.federation_position
|
|
|
+ await self.store.update_federation_out_pos(
|
|
|
+ "federation", current_position
|
|
|
+ )
|
|
|
+
|
|
|
+ # We ACK this token over replication so that the master can drop
|
|
|
+ # its in memory queues
|
|
|
+ self._hs.get_tcp_replication().send_federation_ack(current_position)
|
|
|
+ self._last_ack = current_position
|
|
|
except Exception:
|
|
|
logger.exception("Error updating federation stream position")
|
|
|
|
|
@@ -920,17 +994,22 @@ def start(config_options):
|
|
|
|
|
|
synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
|
|
|
|
|
|
- ss = GenericWorkerServer(
|
|
|
+ hs = GenericWorkerServer(
|
|
|
config.server_name,
|
|
|
config=config,
|
|
|
version_string="Synapse/" + get_version_string(synapse),
|
|
|
)
|
|
|
|
|
|
- setup_logging(ss, config, use_worker_options=True)
|
|
|
+ setup_logging(hs, config, use_worker_options=True)
|
|
|
+
|
|
|
+ hs.setup()
|
|
|
+
|
|
|
+ # Ensure the replication streamer is always started in case we write to any
|
|
|
+ # streams. Will no-op if no streams can be written to by this worker.
|
|
|
+ hs.get_replication_streamer()
|
|
|
|
|
|
- ss.setup()
|
|
|
reactor.addSystemEventTrigger(
|
|
|
- "before", "startup", _base.start, ss, config.worker_listeners
|
|
|
+ "before", "startup", _base.start, hs, config.worker_listeners
|
|
|
)
|
|
|
|
|
|
_base.start_worker_reactor("synapse-generic-worker", config)
|