From fdc015c6e9b023c5cb87491b7e64efd46eedd129 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 10 Jun 2016 16:30:26 +0100 Subject: [PATCH 01/20] Enable testing the synchrotron on jenkins --- jenkins-dendron-postgres.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/jenkins-dendron-postgres.sh b/jenkins-dendron-postgres.sh index d15836e6b..7e6f24aa7 100755 --- a/jenkins-dendron-postgres.sh +++ b/jenkins-dendron-postgres.sh @@ -80,6 +80,7 @@ echo >&2 "Running sytest with PostgreSQL"; --synapse-directory $WORKSPACE \ --dendron $WORKSPACE/dendron/bin/dendron \ --pusher \ + --synchrotron \ --port-base $PORT_BASE cd .. From a60169ea0987df41ee540eefbb77cf3ff53446bc Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 15 Jun 2016 16:57:48 +0100 Subject: [PATCH 02/20] Handle og props with not content --- synapse/rest/media/v1/preview_url_resource.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/synapse/rest/media/v1/preview_url_resource.py b/synapse/rest/media/v1/preview_url_resource.py index 37dd1de89..fc72896e0 100644 --- a/synapse/rest/media/v1/preview_url_resource.py +++ b/synapse/rest/media/v1/preview_url_resource.py @@ -252,7 +252,8 @@ class PreviewUrlResource(Resource): og = {} for tag in tree.xpath("//*/meta[starts-with(@property, 'og:')]"): - og[tag.attrib['property']] = tag.attrib['content'] + if 'content' in tag.attrib: + og[tag.attrib['property']] = tag.attrib['content'] # TODO: grab article: meta tags too, e.g.: From 1e9026e484be0f90256ae60c05eed9d1f87cf6b9 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 15 Jun 2016 16:58:05 +0100 Subject: [PATCH 03/20] Handle floats as img widths --- synapse/rest/media/v1/preview_url_resource.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/rest/media/v1/preview_url_resource.py b/synapse/rest/media/v1/preview_url_resource.py index fc72896e0..a6807df62 100644 --- a/synapse/rest/media/v1/preview_url_resource.py +++ b/synapse/rest/media/v1/preview_url_resource.py @@ -280,7 +280,7 @@ class PreviewUrlResource(Resource): # TODO: consider inlined CSS styles as well as width & height attribs images = tree.xpath("//img[@src][number(@width)>10][number(@height)>10]") images = sorted(images, key=lambda i: ( - -1 * int(i.attrib['width']) * int(i.attrib['height']) + -1 * float(i.attrib['width']) * float(i.attrib['height']) )) if not images: images = tree.xpath("//img[@src]") From 09a17f965cf55dca45983473ed744f539b9ec92e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 15 Jun 2016 16:58:12 +0100 Subject: [PATCH 04/20] Line lengths --- synapse/rest/media/v1/preview_url_resource.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/synapse/rest/media/v1/preview_url_resource.py b/synapse/rest/media/v1/preview_url_resource.py index a6807df62..74c64f137 100644 --- a/synapse/rest/media/v1/preview_url_resource.py +++ b/synapse/rest/media/v1/preview_url_resource.py @@ -288,9 +288,9 @@ class PreviewUrlResource(Resource): og['og:image'] = images[0].attrib['src'] # pre-cache the image for posterity - # FIXME: it might be cleaner to use the same flow as the main /preview_url request - # itself and benefit from the same caching etc. But for now we just rely on the - # caching on the master request to speed things up. + # FIXME: it might be cleaner to use the same flow as the main /preview_url + # request itself and benefit from the same caching etc. But for now we + # just rely on the caching on the master request to speed things up. if 'og:image' in og and og['og:image']: image_info = yield self._download_url( self._rebase_url(og['og:image'], media_info['uri']), requester.user From ed5f43a55accc8502a60b721871b208db704de3e Mon Sep 17 00:00:00 2001 From: Salvatore LaMendola Date: Thu, 16 Jun 2016 00:43:42 -0400 Subject: [PATCH 05/20] Fix TypeError in call to bcrypt.hashpw - At the very least, this TypeError caused logins to fail on my own running instance of Synapse, and the simple (explicit) UTF-8 conversion resolved login errors for me. Signed-off-by: Salvatore LaMendola --- synapse/handlers/auth.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/handlers/auth.py b/synapse/handlers/auth.py index 200793b5e..b38f81e99 100644 --- a/synapse/handlers/auth.py +++ b/synapse/handlers/auth.py @@ -626,6 +626,6 @@ class AuthHandler(BaseHandler): Whether self.hash(password) == stored_hash (bool). """ if stored_hash: - return bcrypt.hashpw(password, stored_hash) == stored_hash + return bcrypt.hashpw(password, stored_hash.encode('utf-8')) == stored_hash else: return False From 885ee861f7270fef1370a2d63e009a8fceaf8dd5 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 11:06:12 +0100 Subject: [PATCH 06/20] Inline the synchrotron and pusher configs into the main config --- synapse/app/pusher.py | 169 ++++++++++------------------------- synapse/app/synchrotron.py | 135 +++++++--------------------- synapse/config/homeserver.py | 4 +- synapse/config/logger.py | 86 +++++++++--------- synapse/config/server.py | 31 ++++--- 5 files changed, 144 insertions(+), 281 deletions(-) diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py index 4ec23d84c..6c8c02fb3 100644 --- a/synapse/app/pusher.py +++ b/synapse/app/pusher.py @@ -18,10 +18,9 @@ import synapse from synapse.server import HomeServer from synapse.config._base import ConfigError -from synapse.config.database import DatabaseConfig -from synapse.config.logger import LoggingConfig -from synapse.config.emailconfig import EmailConfig -from synapse.config.key import KeyConfig +from synapse.config.workers import clobber_with_worker_config +from synapse.config.logger import setup_logging +from synapse.config.homeserver import HomeServerConfig from synapse.http.site import SynapseSite from synapse.metrics.resource import MetricsResource, METRICS_PREFIX from synapse.storage.roommember import RoomMemberStore @@ -43,98 +42,12 @@ from twisted.web.resource import Resource from daemonize import Daemonize -import gc import sys import logging logger = logging.getLogger("synapse.app.pusher") -class SlaveConfig(DatabaseConfig): - def read_config(self, config): - self.replication_url = config["replication_url"] - self.server_name = config["server_name"] - self.use_insecure_ssl_client_just_for_testing_do_not_use = config.get( - "use_insecure_ssl_client_just_for_testing_do_not_use", False - ) - self.user_agent_suffix = None - self.start_pushers = True - self.listeners = config["listeners"] - self.soft_file_limit = config.get("soft_file_limit") - self.daemonize = config.get("daemonize") - self.pid_file = self.abspath(config.get("pid_file")) - self.public_baseurl = config["public_baseurl"] - - thresholds = config.get("gc_thresholds", None) - if thresholds is not None: - try: - assert len(thresholds) == 3 - self.gc_thresholds = ( - int(thresholds[0]), int(thresholds[1]), int(thresholds[2]), - ) - except: - raise ConfigError( - "Value of `gc_threshold` must be a list of three integers if set" - ) - else: - self.gc_thresholds = None - - # some things used by the auth handler but not actually used in the - # pusher codebase - self.bcrypt_rounds = None - self.ldap_enabled = None - self.ldap_server = None - self.ldap_port = None - self.ldap_tls = None - self.ldap_search_base = None - self.ldap_search_property = None - self.ldap_email_property = None - self.ldap_full_name_property = None - - # We would otherwise try to use the registration shared secret as the - # macaroon shared secret if there was no macaroon_shared_secret, but - # that means pulling in RegistrationConfig too. We don't need to be - # backwards compaitible in the pusher codebase so just make people set - # macaroon_shared_secret. We set this to None to prevent it referencing - # an undefined key. - self.registration_shared_secret = None - - def default_config(self, server_name, **kwargs): - pid_file = self.abspath("pusher.pid") - return """\ - # Slave configuration - - # The replication listener on the synapse to talk to. - #replication_url: https://localhost:{replication_port}/_synapse/replication - - server_name: "%(server_name)s" - - listeners: [] - # Enable a ssh manhole listener on the pusher. - # - type: manhole - # port: {manhole_port} - # bind_address: 127.0.0.1 - # Enable a metric listener on the pusher. - # - type: http - # port: {metrics_port} - # bind_address: 127.0.0.1 - # resources: - # - names: ["metrics"] - # compress: False - - report_stats: False - - daemonize: False - - pid_file: %(pid_file)s - - """ % locals() - - -class PusherSlaveConfig(SlaveConfig, LoggingConfig, EmailConfig, KeyConfig): - pass - - class PusherSlaveStore( SlavedEventStore, SlavedPusherStore, SlavedReceiptsStore, SlavedAccountDataStore @@ -232,8 +145,8 @@ class PusherServer(HomeServer): ) logger.info("Synapse pusher now listening on port %d", port) - def start_listening(self): - for listener in self.config.listeners: + def start_listening(self, listeners): + for listener in listeners: if listener["type"] == "http": self._listen_http(listener) elif listener["type"] == "manhole": @@ -329,19 +242,32 @@ class PusherServer(HomeServer): yield sleep(30) -def setup(config_options): +def setup(worker_name, config_options): try: - config = PusherSlaveConfig.load_config( + config = HomeServerConfig.load_config( "Synapse pusher", config_options ) except ConfigError as e: sys.stderr.write("\n" + e.message + "\n") sys.exit(1) - if not config: - sys.exit(0) + worker_config = config.workers[worker_name] - config.setup_logging() + setup_logging(worker_config.log_config, worker_config.log_file) + + clobber_with_worker_config(config, worker_config) + + if config.start_pushers: + sys.stderr.write( + "\nThe pushers must be disabled in the main synapse process" + "\nbefore they can be run in a separate worker." + "\nPlease add ``start_pushers: false`` to the main config" + "\n" + ) + sys.exit(1) + + # Force the pushers to start since they will be disabled in the main config + config.start_pushers = True database_engine = create_engine(config.database_config) @@ -354,11 +280,15 @@ def setup(config_options): ) ps.setup() - ps.start_listening() + ps.start_listening(worker_config.listeners) - change_resource_limit(ps.config.soft_file_limit) - if ps.config.gc_thresholds: - gc.set_threshold(*ps.config.gc_thresholds) + def run(): + with LoggingContext("run"): + logger.info("Running") + change_resource_limit(worker_config.soft_file_limit) + if worker_config.gc_thresholds: + ps.set_threshold(worker_config.gc_thresholds) + reactor.run() def start(): ps.replicate() @@ -367,30 +297,21 @@ def setup(config_options): reactor.callWhenRunning(start) - return ps + if worker_config.daemonize: + daemon = Daemonize( + app="synapse-pusher", + pid=worker_config.pid_file, + action=run, + auto_close_fds=False, + verbose=True, + logger=logger, + ) + daemon.start() + else: + run() if __name__ == '__main__': with LoggingContext("main"): - ps = setup(sys.argv[1:]) - - if ps.config.daemonize: - def run(): - with LoggingContext("run"): - change_resource_limit(ps.config.soft_file_limit) - if ps.config.gc_thresholds: - gc.set_threshold(*ps.config.gc_thresholds) - reactor.run() - - daemon = Daemonize( - app="synapse-pusher", - pid=ps.config.pid_file, - action=run, - auto_close_fds=False, - verbose=True, - logger=logger, - ) - - daemon.start() - else: - reactor.run() + worker_name = sys.argv[1] + ps = setup(worker_name, sys.argv[2:]) diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index 297e19945..7a607faef 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -18,9 +18,9 @@ import synapse from synapse.api.constants import EventTypes, PresenceState from synapse.config._base import ConfigError -from synapse.config.database import DatabaseConfig -from synapse.config.logger import LoggingConfig -from synapse.config.appservice import AppServiceConfig +from synapse.config.homeserver import HomeServerConfig +from synapse.config.logger import setup_logging +from synapse.config.workers import clobber_with_worker_config from synapse.events import FrozenEvent from synapse.handlers.presence import PresenceHandler from synapse.http.site import SynapseSite @@ -57,76 +57,11 @@ from daemonize import Daemonize import sys import logging import contextlib -import gc import ujson as json logger = logging.getLogger("synapse.app.synchrotron") -class SynchrotronConfig(DatabaseConfig, LoggingConfig, AppServiceConfig): - def read_config(self, config): - self.replication_url = config["replication_url"] - self.server_name = config["server_name"] - self.use_insecure_ssl_client_just_for_testing_do_not_use = config.get( - "use_insecure_ssl_client_just_for_testing_do_not_use", False - ) - self.user_agent_suffix = None - self.listeners = config["listeners"] - self.soft_file_limit = config.get("soft_file_limit") - self.daemonize = config.get("daemonize") - self.pid_file = self.abspath(config.get("pid_file")) - self.macaroon_secret_key = config["macaroon_secret_key"] - self.expire_access_token = config.get("expire_access_token", False) - - thresholds = config.get("gc_thresholds", None) - if thresholds is not None: - try: - assert len(thresholds) == 3 - self.gc_thresholds = ( - int(thresholds[0]), int(thresholds[1]), int(thresholds[2]), - ) - except: - raise ConfigError( - "Value of `gc_threshold` must be a list of three integers if set" - ) - else: - self.gc_thresholds = None - - def default_config(self, server_name, **kwargs): - pid_file = self.abspath("synchroton.pid") - return """\ - # Slave configuration - - # The replication listener on the synapse to talk to. - #replication_url: https://localhost:{replication_port}/_synapse/replication - - server_name: "%(server_name)s" - - listeners: - # Enable a /sync listener on the synchrontron - #- type: http - # port: {http_port} - # bind_address: "" - # Enable a ssh manhole listener on the synchrotron - # - type: manhole - # port: {manhole_port} - # bind_address: 127.0.0.1 - # Enable a metric listener on the synchrotron - # - type: http - # port: {metrics_port} - # bind_address: 127.0.0.1 - # resources: - # - names: ["metrics"] - # compress: False - - report_stats: False - - daemonize: False - - pid_file: %(pid_file)s - """ % locals() - - class SynchrotronSlavedStore( SlavedPushRuleStore, SlavedEventStore, @@ -350,8 +285,8 @@ class SynchrotronServer(HomeServer): ) logger.info("Synapse synchrotron now listening on port %d", port) - def start_listening(self): - for listener in self.config.listeners: + def start_listening(self, listeners): + for listener in listeners: if listener["type"] == "http": self._listen_http(listener) elif listener["type"] == "manhole": @@ -470,19 +405,20 @@ class SynchrotronServer(HomeServer): return SynchrotronTyping(self) -def setup(config_options): +def start(worker_name, config_options): try: - config = SynchrotronConfig.load_config( + config = HomeServerConfig.load_config( "Synapse synchrotron", config_options ) except ConfigError as e: sys.stderr.write("\n" + e.message + "\n") sys.exit(1) - if not config: - sys.exit(0) + worker_config = config.workers[worker_name] - config.setup_logging() + setup_logging(worker_config.log_config, worker_config.log_file) + + clobber_with_worker_config(config, worker_config) database_engine = create_engine(config.database_config) @@ -496,11 +432,15 @@ def setup(config_options): ) ss.setup() - ss.start_listening() + ss.start_listening(worker_config.listeners) - change_resource_limit(ss.config.soft_file_limit) - if ss.config.gc_thresholds: - ss.set_threshold(*ss.config.gc_thresholds) + def run(): + with LoggingContext("run"): + logger.info("Running") + change_resource_limit(worker_config.soft_file_limit) + if worker_config.gc_thresholds: + ss.set_threshold(worker_config.gc_thresholds) + reactor.run() def start(): ss.get_datastore().start_profiling() @@ -508,30 +448,21 @@ def setup(config_options): reactor.callWhenRunning(start) - return ss + if worker_config.daemonize: + daemon = Daemonize( + app="synapse-synchrotron", + pid=worker_config.pid_file, + action=run, + auto_close_fds=False, + verbose=True, + logger=logger, + ) + daemon.start() + else: + run() if __name__ == '__main__': with LoggingContext("main"): - ss = setup(sys.argv[1:]) - - if ss.config.daemonize: - def run(): - with LoggingContext("run"): - change_resource_limit(ss.config.soft_file_limit) - if ss.config.gc_thresholds: - gc.set_threshold(*ss.config.gc_thresholds) - reactor.run() - - daemon = Daemonize( - app="synapse-synchrotron", - pid=ss.config.pid_file, - action=run, - auto_close_fds=False, - verbose=True, - logger=logger, - ) - - daemon.start() - else: - reactor.run() + worker_name = sys.argv[1] + start(worker_name, sys.argv[2:]) diff --git a/synapse/config/homeserver.py b/synapse/config/homeserver.py index fc2445484..79b0534b3 100644 --- a/synapse/config/homeserver.py +++ b/synapse/config/homeserver.py @@ -32,13 +32,15 @@ from .password import PasswordConfig from .jwt import JWTConfig from .ldap import LDAPConfig from .emailconfig import EmailConfig +from .workers import WorkerConfig class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig, RatelimitConfig, ContentRepositoryConfig, CaptchaConfig, VoipConfig, RegistrationConfig, MetricsConfig, ApiConfig, AppServiceConfig, KeyConfig, SAML2Config, CasConfig, - JWTConfig, LDAPConfig, PasswordConfig, EmailConfig,): + JWTConfig, LDAPConfig, PasswordConfig, EmailConfig, + WorkerConfig,): pass diff --git a/synapse/config/logger.py b/synapse/config/logger.py index 5047db898..dc68683fb 100644 --- a/synapse/config/logger.py +++ b/synapse/config/logger.py @@ -126,54 +126,58 @@ class LoggingConfig(Config): ) def setup_logging(self): - log_format = ( - "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s" - " - %(message)s" - ) - if self.log_config is None: + setup_logging(self.log_config, self.log_file, self.verbosity) - level = logging.INFO - level_for_storage = logging.INFO - if self.verbosity: - level = logging.DEBUG - if self.verbosity > 1: - level_for_storage = logging.DEBUG - # FIXME: we need a logging.WARN for a -q quiet option - logger = logging.getLogger('') - logger.setLevel(level) +def setup_logging(log_config=None, log_file=None, verbosity=None): + log_format = ( + "%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s" + " - %(message)s" + ) + if log_config is None: - logging.getLogger('synapse.storage').setLevel(level_for_storage) + level = logging.INFO + level_for_storage = logging.INFO + if verbosity: + level = logging.DEBUG + if verbosity > 1: + level_for_storage = logging.DEBUG - formatter = logging.Formatter(log_format) - if self.log_file: - # TODO: Customisable file size / backup count - handler = logging.handlers.RotatingFileHandler( - self.log_file, maxBytes=(1000 * 1000 * 100), backupCount=3 - ) + # FIXME: we need a logging.WARN for a -q quiet option + logger = logging.getLogger('') + logger.setLevel(level) - def sighup(signum, stack): - logger.info("Closing log file due to SIGHUP") - handler.doRollover() - logger.info("Opened new log file due to SIGHUP") + logging.getLogger('synapse.storage').setLevel(level_for_storage) - # TODO(paul): obviously this is a terrible mechanism for - # stealing SIGHUP, because it means no other part of synapse - # can use it instead. If we want to catch SIGHUP anywhere - # else as well, I'd suggest we find a nicer way to broadcast - # it around. - if getattr(signal, "SIGHUP"): - signal.signal(signal.SIGHUP, sighup) - else: - handler = logging.StreamHandler() - handler.setFormatter(formatter) + formatter = logging.Formatter(log_format) + if log_file: + # TODO: Customisable file size / backup count + handler = logging.handlers.RotatingFileHandler( + log_file, maxBytes=(1000 * 1000 * 100), backupCount=3 + ) - handler.addFilter(LoggingContextFilter(request="")) + def sighup(signum, stack): + logger.info("Closing log file due to SIGHUP") + handler.doRollover() + logger.info("Opened new log file due to SIGHUP") - logger.addHandler(handler) + # TODO(paul): obviously this is a terrible mechanism for + # stealing SIGHUP, because it means no other part of synapse + # can use it instead. If we want to catch SIGHUP anywhere + # else as well, I'd suggest we find a nicer way to broadcast + # it around. + if getattr(signal, "SIGHUP"): + signal.signal(signal.SIGHUP, sighup) else: - with open(self.log_config, 'r') as f: - logging.config.dictConfig(yaml.load(f)) + handler = logging.StreamHandler() + handler.setFormatter(formatter) - observer = PythonLoggingObserver() - observer.start() + handler.addFilter(LoggingContextFilter(request="")) + + logger.addHandler(handler) + else: + with open(log_config, 'r') as f: + logging.config.dictConfig(yaml.load(f)) + + observer = PythonLoggingObserver() + observer.start() diff --git a/synapse/config/server.py b/synapse/config/server.py index 44b8d422e..f370b22c3 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -38,19 +38,7 @@ class ServerConfig(Config): self.listeners = config.get("listeners", []) - thresholds = config.get("gc_thresholds", None) - if thresholds is not None: - try: - assert len(thresholds) == 3 - self.gc_thresholds = ( - int(thresholds[0]), int(thresholds[1]), int(thresholds[2]), - ) - except: - raise ConfigError( - "Value of `gc_threshold` must be a list of three integers if set" - ) - else: - self.gc_thresholds = None + self.gc_thresholds = read_gc_thresholds(config.get("gc_thresholds", None)) bind_port = config.get("bind_port") if bind_port: @@ -264,3 +252,20 @@ class ServerConfig(Config): type=int, help="Turn on the twisted telnet manhole" " service on the given port.") + + +def read_gc_thresholds(thresholds): + """Reads the three integer thresholds for garbage collection. Ensures that + the thresholds are integers if thresholds are supplied. + """ + if thresholds is None: + return None + try: + assert len(thresholds) == 3 + return ( + int(thresholds[0]), int(thresholds[1]), int(thresholds[2]), + ) + except: + raise ConfigError( + "Value of `gc_threshold` must be a list of three integers if set" + ) From dbb5a39b64e3b52978ecb98f8f64b7b50acf9b59 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 11:09:15 +0100 Subject: [PATCH 07/20] Add worker config module --- synapse/config/workers.py | 71 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 synapse/config/workers.py diff --git a/synapse/config/workers.py b/synapse/config/workers.py new file mode 100644 index 000000000..fd19e38b8 --- /dev/null +++ b/synapse/config/workers.py @@ -0,0 +1,71 @@ +# -*- coding: utf-8 -*- +# Copyright 2016 matrix.org +# +# 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. + +import collections + +from ._base import Config +from .server import read_gc_thresholds + + +Worker = collections.namedtuple("Worker", [ + "app", + "listeners", + "pid_file", + "daemonize", + "log_file", + "log_config", + "event_cache_size", + "soft_file_limit", + "gc_thresholds", + "replication_url", +]) + + +def clobber_with_worker_config(config, worker_config): + """Overrides some of the keys of the main config with worker-specific + values.""" + config.event_cache_size = worker_config.event_cache_size + config.replication_url = worker_config.replication_url + + +def read_worker_config(config): + return Worker( + app=config["app"], + listeners=config.get("listeners", []), + pid_file=config.get("pid_file"), + daemonize=config["daemonize"], + log_file=config.get("log_file"), + log_config=config.get("log_config"), + event_cache_size=Config.parse_size(config.get("event_cache_size", "10K")), + soft_file_limit=config.get("soft_file_limit"), + gc_thresholds=read_gc_thresholds(config.get("gc_thresholds")), + replication_url=config.get("replication_url"), + ) + + +class WorkerConfig(Config): + """The workers are processes run separately to the main synapse process. + Each worker has a name that identifies it within the config file. + They have their own pid_file and listener configuration. They use the + replication_url to talk to the main synapse process. They have their + own cache size tuning, gc threshold tuning and open file limits.""" + + def read_config(self, config): + workers = config.get("workers", {}) + + self.workers = { + worker_name: read_worker_config(worker_config) + for worker_name, worker_config in workers.items() + } From 80a1bc7db517298baec24c1f11a144552719fb7b Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 11:29:45 +0100 Subject: [PATCH 08/20] Comment on what's going on in clobber_with_worker_config --- synapse/config/workers.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/synapse/config/workers.py b/synapse/config/workers.py index fd19e38b8..4f4658c0a 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -35,8 +35,19 @@ Worker = collections.namedtuple("Worker", [ def clobber_with_worker_config(config, worker_config): """Overrides some of the keys of the main config with worker-specific - values.""" + values. We only need to override the keys that are accessed deep + withing synapse code. Most of the keys that we want to override in + the workers are accessed in setup code that is rewritten specifically + for the workers. In that new code we can access the worker config directly, + so we don't need to override the values in the main config.""" + + # TODO: The event_cache_size is accessed in the db setup. It should be + # possible to rejigg that code so that the cache size is pulled from the + # worker config directly. config.event_cache_size = worker_config.event_cache_size + + # TODO: The replication_url should only be accessed within worker specific + # code so it really shouldn't need to be clobbered in the main config. config.replication_url = worker_config.replication_url From bde13833cb42fc6e09928ffb4f4efad9244abffa Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 12:44:40 +0100 Subject: [PATCH 09/20] Access replication_url from the worker config directly --- synapse/app/pusher.py | 5 +++-- synapse/app/synchrotron.py | 5 +++-- synapse/config/workers.py | 4 ---- 3 files changed, 6 insertions(+), 8 deletions(-) diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py index 6c8c02fb3..a26a3bd39 100644 --- a/synapse/app/pusher.py +++ b/synapse/app/pusher.py @@ -112,7 +112,7 @@ class PusherServer(HomeServer): def remove_pusher(self, app_id, push_key, user_id): http_client = self.get_simple_http_client() - replication_url = self.config.replication_url + replication_url = self.worker_config.replication_url url = replication_url + "/remove_pushers" return http_client.post_json_get_json(url, { "remove": [{ @@ -166,7 +166,7 @@ class PusherServer(HomeServer): def replicate(self): http_client = self.get_simple_http_client() store = self.get_datastore() - replication_url = self.config.replication_url + replication_url = self.worker_config.replication_url pusher_pool = self.get_pusherpool() clock = self.get_clock() @@ -275,6 +275,7 @@ def setup(worker_name, config_options): config.server_name, db_config=config.database_config, config=config, + worker_config=worker_config, version_string=get_version_string("Synapse", synapse), database_engine=database_engine, ) diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index 7a607faef..4443c73e6 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -98,7 +98,7 @@ class SynchrotronPresence(object): self.http_client = hs.get_simple_http_client() self.store = hs.get_datastore() self.user_to_num_current_syncs = {} - self.syncing_users_url = hs.config.replication_url + "/syncing_users" + self.syncing_users_url = hs.worker_config.replication_url + "/syncing_users" self.clock = hs.get_clock() active_presence = self.store.take_presence_startup_info() @@ -306,7 +306,7 @@ class SynchrotronServer(HomeServer): def replicate(self): http_client = self.get_simple_http_client() store = self.get_datastore() - replication_url = self.config.replication_url + replication_url = self.worker_config.replication_url clock = self.get_clock() notifier = self.get_notifier() presence_handler = self.get_presence_handler() @@ -426,6 +426,7 @@ def start(worker_name, config_options): config.server_name, db_config=config.database_config, config=config, + worker_config=worker_config, version_string=get_version_string("Synapse", synapse), database_engine=database_engine, application_service_handler=SynchrotronApplicationService(), diff --git a/synapse/config/workers.py b/synapse/config/workers.py index 4f4658c0a..f2c77ef59 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -46,10 +46,6 @@ def clobber_with_worker_config(config, worker_config): # worker config directly. config.event_cache_size = worker_config.event_cache_size - # TODO: The replication_url should only be accessed within worker specific - # code so it really shouldn't need to be clobbered in the main config. - config.replication_url = worker_config.replication_url - def read_worker_config(config): return Worker( From 364d6167926d5d8b2a312e3d35623d2e05330e0a Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 12:53:15 +0100 Subject: [PATCH 10/20] Access the event_cache_size directly from the server object. This means that the workers can override the event_cache_size directly without clobbering the value in the main synapse config. --- synapse/app/pusher.py | 6 +++--- synapse/app/synchrotron.py | 6 +++--- synapse/config/workers.py | 14 -------------- synapse/server.py | 3 +++ synapse/storage/_base.py | 2 +- 5 files changed, 10 insertions(+), 21 deletions(-) diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py index a26a3bd39..5d4db4f89 100644 --- a/synapse/app/pusher.py +++ b/synapse/app/pusher.py @@ -18,7 +18,6 @@ import synapse from synapse.server import HomeServer from synapse.config._base import ConfigError -from synapse.config.workers import clobber_with_worker_config from synapse.config.logger import setup_logging from synapse.config.homeserver import HomeServerConfig from synapse.http.site import SynapseSite @@ -241,6 +240,9 @@ class PusherServer(HomeServer): logger.exception("Error replicating from %r", replication_url) yield sleep(30) + def get_event_cache_size(self): + return self.worker_config.event_cache_size + def setup(worker_name, config_options): try: @@ -255,8 +257,6 @@ def setup(worker_name, config_options): setup_logging(worker_config.log_config, worker_config.log_file) - clobber_with_worker_config(config, worker_config) - if config.start_pushers: sys.stderr.write( "\nThe pushers must be disabled in the main synapse process" diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index 4443c73e6..d10bb2b3f 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -20,7 +20,6 @@ from synapse.api.constants import EventTypes, PresenceState from synapse.config._base import ConfigError from synapse.config.homeserver import HomeServerConfig from synapse.config.logger import setup_logging -from synapse.config.workers import clobber_with_worker_config from synapse.events import FrozenEvent from synapse.handlers.presence import PresenceHandler from synapse.http.site import SynapseSite @@ -404,6 +403,9 @@ class SynchrotronServer(HomeServer): def build_typing_handler(self): return SynchrotronTyping(self) + def get_event_cache_size(self): + return self.worker_config.event_cache_size + def start(worker_name, config_options): try: @@ -418,8 +420,6 @@ def start(worker_name, config_options): setup_logging(worker_config.log_config, worker_config.log_file) - clobber_with_worker_config(config, worker_config) - database_engine = create_engine(config.database_config) ss = SynchrotronServer( diff --git a/synapse/config/workers.py b/synapse/config/workers.py index f2c77ef59..503358e03 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -33,20 +33,6 @@ Worker = collections.namedtuple("Worker", [ ]) -def clobber_with_worker_config(config, worker_config): - """Overrides some of the keys of the main config with worker-specific - values. We only need to override the keys that are accessed deep - withing synapse code. Most of the keys that we want to override in - the workers are accessed in setup code that is rewritten specifically - for the workers. In that new code we can access the worker config directly, - so we don't need to override the values in the main config.""" - - # TODO: The event_cache_size is accessed in the db setup. It should be - # possible to rejigg that code so that the cache size is pulled from the - # worker config directly. - config.event_cache_size = worker_config.event_cache_size - - def read_worker_config(config): return Worker( app=config["app"], diff --git a/synapse/server.py b/synapse/server.py index dd4b81c65..b3c31ece7 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -236,6 +236,9 @@ class HomeServer(object): def remove_pusher(self, app_id, push_key, user_id): return self.get_pusherpool().remove_pusher(app_id, push_key, user_id) + def get_event_cache_size(self): + return self.config.event_cache_size + def _make_dependency_method(depname): def _get(hs): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 32c6677d4..2932880cc 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -166,7 +166,7 @@ class SQLBaseStore(object): self._get_event_counters = PerformanceCounters() self._get_event_cache = Cache("*getEvent*", keylen=3, lru=True, - max_entries=hs.config.event_cache_size) + max_entries=hs.get_event_cache_size()) self._state_group_cache = DictionaryCache( "*stateGroupCache*", 2000 * CACHE_SIZE_FACTOR From a352b68acf473f59012340b7f481f3dfd6544ac6 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 16 Jun 2016 17:29:50 +0100 Subject: [PATCH 11/20] Use worker_ prefixes for worker config, use existing support for multiple config files --- synapse/app/pusher.py | 29 ++++++++++------------ synapse/app/synchrotron.py | 29 ++++++++++------------ synapse/config/workers.py | 49 +++++++------------------------------- synapse/server.py | 3 --- synapse/storage/_base.py | 2 +- 5 files changed, 33 insertions(+), 79 deletions(-) diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py index 5d4db4f89..9ac26d52c 100644 --- a/synapse/app/pusher.py +++ b/synapse/app/pusher.py @@ -111,7 +111,7 @@ class PusherServer(HomeServer): def remove_pusher(self, app_id, push_key, user_id): http_client = self.get_simple_http_client() - replication_url = self.worker_config.replication_url + replication_url = self.config.worker_replication_url url = replication_url + "/remove_pushers" return http_client.post_json_get_json(url, { "remove": [{ @@ -165,7 +165,7 @@ class PusherServer(HomeServer): def replicate(self): http_client = self.get_simple_http_client() store = self.get_datastore() - replication_url = self.worker_config.replication_url + replication_url = self.config.worker_replication_url pusher_pool = self.get_pusherpool() clock = self.get_clock() @@ -240,11 +240,8 @@ class PusherServer(HomeServer): logger.exception("Error replicating from %r", replication_url) yield sleep(30) - def get_event_cache_size(self): - return self.worker_config.event_cache_size - -def setup(worker_name, config_options): +def start(config_options): try: config = HomeServerConfig.load_config( "Synapse pusher", config_options @@ -253,9 +250,9 @@ def setup(worker_name, config_options): sys.stderr.write("\n" + e.message + "\n") sys.exit(1) - worker_config = config.workers[worker_name] + assert config.worker_app == "synapse.app.pusher" - setup_logging(worker_config.log_config, worker_config.log_file) + setup_logging(config.worker_log_config, config.worker_log_file) if config.start_pushers: sys.stderr.write( @@ -275,20 +272,19 @@ def setup(worker_name, config_options): config.server_name, db_config=config.database_config, config=config, - worker_config=worker_config, version_string=get_version_string("Synapse", synapse), database_engine=database_engine, ) ps.setup() - ps.start_listening(worker_config.listeners) + ps.start_listening(config.worker_listeners) def run(): with LoggingContext("run"): logger.info("Running") - change_resource_limit(worker_config.soft_file_limit) - if worker_config.gc_thresholds: - ps.set_threshold(worker_config.gc_thresholds) + change_resource_limit(config.soft_file_limit) + if config.gc_thresholds: + ps.set_threshold(config.gc_thresholds) reactor.run() def start(): @@ -298,10 +294,10 @@ def setup(worker_name, config_options): reactor.callWhenRunning(start) - if worker_config.daemonize: + if config.worker_daemonize: daemon = Daemonize( app="synapse-pusher", - pid=worker_config.pid_file, + pid=config.worker_pid_file, action=run, auto_close_fds=False, verbose=True, @@ -314,5 +310,4 @@ def setup(worker_name, config_options): if __name__ == '__main__': with LoggingContext("main"): - worker_name = sys.argv[1] - ps = setup(worker_name, sys.argv[2:]) + ps = start(sys.argv[1:]) diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index d10bb2b3f..160db8637 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -97,7 +97,7 @@ class SynchrotronPresence(object): self.http_client = hs.get_simple_http_client() self.store = hs.get_datastore() self.user_to_num_current_syncs = {} - self.syncing_users_url = hs.worker_config.replication_url + "/syncing_users" + self.syncing_users_url = hs.config.worker_replication_url + "/syncing_users" self.clock = hs.get_clock() active_presence = self.store.take_presence_startup_info() @@ -305,7 +305,7 @@ class SynchrotronServer(HomeServer): def replicate(self): http_client = self.get_simple_http_client() store = self.get_datastore() - replication_url = self.worker_config.replication_url + replication_url = self.config.worker_replication_url clock = self.get_clock() notifier = self.get_notifier() presence_handler = self.get_presence_handler() @@ -403,11 +403,8 @@ class SynchrotronServer(HomeServer): def build_typing_handler(self): return SynchrotronTyping(self) - def get_event_cache_size(self): - return self.worker_config.event_cache_size - -def start(worker_name, config_options): +def start(config_options): try: config = HomeServerConfig.load_config( "Synapse synchrotron", config_options @@ -416,9 +413,9 @@ def start(worker_name, config_options): sys.stderr.write("\n" + e.message + "\n") sys.exit(1) - worker_config = config.workers[worker_name] + assert config.worker_app == "synapse.app.synchrotron" - setup_logging(worker_config.log_config, worker_config.log_file) + setup_logging(config.worker_log_config, config.worker_log_file) database_engine = create_engine(config.database_config) @@ -426,21 +423,20 @@ def start(worker_name, config_options): config.server_name, db_config=config.database_config, config=config, - worker_config=worker_config, version_string=get_version_string("Synapse", synapse), database_engine=database_engine, application_service_handler=SynchrotronApplicationService(), ) ss.setup() - ss.start_listening(worker_config.listeners) + ss.start_listening(config.worker_listeners) def run(): with LoggingContext("run"): logger.info("Running") - change_resource_limit(worker_config.soft_file_limit) - if worker_config.gc_thresholds: - ss.set_threshold(worker_config.gc_thresholds) + change_resource_limit(config.soft_file_limit) + if config.gc_thresholds: + ss.set_threshold(config.gc_thresholds) reactor.run() def start(): @@ -449,10 +445,10 @@ def start(worker_name, config_options): reactor.callWhenRunning(start) - if worker_config.daemonize: + if config.worker_daemonize: daemon = Daemonize( app="synapse-synchrotron", - pid=worker_config.pid_file, + pid=config.worker_pid_file, action=run, auto_close_fds=False, verbose=True, @@ -465,5 +461,4 @@ def start(worker_name, config_options): if __name__ == '__main__': with LoggingContext("main"): - worker_name = sys.argv[1] - start(worker_name, sys.argv[2:]) + start(sys.argv[1:]) diff --git a/synapse/config/workers.py b/synapse/config/workers.py index 503358e03..904789d15 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -13,52 +13,19 @@ # See the License for the specific language governing permissions and # limitations under the License. -import collections - from ._base import Config -from .server import read_gc_thresholds - - -Worker = collections.namedtuple("Worker", [ - "app", - "listeners", - "pid_file", - "daemonize", - "log_file", - "log_config", - "event_cache_size", - "soft_file_limit", - "gc_thresholds", - "replication_url", -]) - - -def read_worker_config(config): - return Worker( - app=config["app"], - listeners=config.get("listeners", []), - pid_file=config.get("pid_file"), - daemonize=config["daemonize"], - log_file=config.get("log_file"), - log_config=config.get("log_config"), - event_cache_size=Config.parse_size(config.get("event_cache_size", "10K")), - soft_file_limit=config.get("soft_file_limit"), - gc_thresholds=read_gc_thresholds(config.get("gc_thresholds")), - replication_url=config.get("replication_url"), - ) class WorkerConfig(Config): """The workers are processes run separately to the main synapse process. - Each worker has a name that identifies it within the config file. They have their own pid_file and listener configuration. They use the - replication_url to talk to the main synapse process. They have their - own cache size tuning, gc threshold tuning and open file limits.""" + replication_url to talk to the main synapse process.""" def read_config(self, config): - workers = config.get("workers", {}) - - self.workers = { - worker_name: read_worker_config(worker_config) - for worker_name, worker_config in workers.items() - } + self.worker_app = config.get("worker_app") + self.worker_listeners = config.get("worker_listeners") + self.worker_daemonize = config.get("worker_daemonize") + self.worker_pid_file = config.get("worker_pid_file") + self.worker_log_file = config.get("worker_log_file") + self.worker_log_config = config.get("worker_log_config") + self.worker_replication_url = config.get("worker_replication_url") diff --git a/synapse/server.py b/synapse/server.py index b3c31ece7..dd4b81c65 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -236,9 +236,6 @@ class HomeServer(object): def remove_pusher(self, app_id, push_key, user_id): return self.get_pusherpool().remove_pusher(app_id, push_key, user_id) - def get_event_cache_size(self): - return self.config.event_cache_size - def _make_dependency_method(depname): def _get(hs): diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 2932880cc..32c6677d4 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -166,7 +166,7 @@ class SQLBaseStore(object): self._get_event_counters = PerformanceCounters() self._get_event_cache = Cache("*getEvent*", keylen=3, lru=True, - max_entries=hs.get_event_cache_size()) + max_entries=hs.config.event_cache_size) self._state_group_cache = DictionaryCache( "*stateGroupCache*", 2000 * CACHE_SIZE_FACTOR From 8c75040c25495bf29f4c76ca0fcc032975210012 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 17 Jun 2016 11:48:12 +0100 Subject: [PATCH 12/20] Fix setting gc thresholds in the workers --- synapse/app/pusher.py | 3 ++- synapse/app/synchrotron.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py index 9ac26d52c..4f1d18ab5 100644 --- a/synapse/app/pusher.py +++ b/synapse/app/pusher.py @@ -43,6 +43,7 @@ from daemonize import Daemonize import sys import logging +import gc logger = logging.getLogger("synapse.app.pusher") @@ -284,7 +285,7 @@ def start(config_options): logger.info("Running") change_resource_limit(config.soft_file_limit) if config.gc_thresholds: - ps.set_threshold(config.gc_thresholds) + gc.set_threshold(*config.gc_thresholds) reactor.run() def start(): diff --git a/synapse/app/synchrotron.py b/synapse/app/synchrotron.py index 160db8637..8cf5bbbb6 100644 --- a/synapse/app/synchrotron.py +++ b/synapse/app/synchrotron.py @@ -56,6 +56,7 @@ from daemonize import Daemonize import sys import logging import contextlib +import gc import ujson as json logger = logging.getLogger("synapse.app.synchrotron") @@ -436,7 +437,7 @@ def start(config_options): logger.info("Running") change_resource_limit(config.soft_file_limit) if config.gc_thresholds: - ss.set_threshold(config.gc_thresholds) + gc.set_threshold(*config.gc_thresholds) reactor.run() def start(): From ded01c3bf65fd6bb83c9d3546ea44859208e4578 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 17 Jun 2016 13:49:16 +0100 Subject: [PATCH 13/20] Fix ``KeyError: 'msgtype'``. Use ``.get`` Fixes a key error where the mailer tried to get the ``msgtype`` of an event that was missing a ``msgtype``. ``` File "synapse/push/mailer.py", line 264, in get_notif_vars File "synapse/push/mailer.py", line 285, in get_message_vars File ".../frozendict/__init__.py", line 10, in __getitem__ return self.__dict[key] KeyError: 'msgtype' ``` --- synapse/push/mailer.py | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/synapse/push/mailer.py b/synapse/push/mailer.py index e5c3929cd..1028731bc 100644 --- a/synapse/push/mailer.py +++ b/synapse/push/mailer.py @@ -273,16 +273,16 @@ class Mailer(object): sender_state_event = room_state[("m.room.member", event.sender)] sender_name = name_from_member_event(sender_state_event) - sender_avatar_url = None - if "avatar_url" in sender_state_event.content: - sender_avatar_url = sender_state_event.content["avatar_url"] + sender_avatar_url = sender_state_event.content.get("avatar_url") # 'hash' for deterministically picking default images: use # sender_hash % the number of default images to choose from sender_hash = string_ordinal_total(event.sender) + msgtype = event.content.get("msgtype") + ret = { - "msgtype": event.content["msgtype"], + "msgtype": msgtype, "is_historical": event.event_id != notif['event_id'], "id": event.event_id, "ts": event.origin_server_ts, @@ -291,9 +291,9 @@ class Mailer(object): "sender_hash": sender_hash, } - if event.content["msgtype"] == "m.text": + if msgtype == "m.text": self.add_text_message_vars(ret, event) - elif event.content["msgtype"] == "m.image": + elif msgtype == "m.image": self.add_image_message_vars(ret, event) if "body" in event.content: @@ -302,16 +302,17 @@ class Mailer(object): return ret def add_text_message_vars(self, messagevars, event): - if "format" in event.content: - msgformat = event.content["format"] - else: - msgformat = None + msgformat = event.content.get("format") + messagevars["format"] = msgformat - if msgformat == "org.matrix.custom.html": - messagevars["body_text_html"] = safe_markup(event.content["formatted_body"]) - else: - messagevars["body_text_html"] = safe_text(event.content["body"]) + formatted_body = event.content.get("formatted_body") + body = event.content.get("body") + + if msgformat == "org.matrix.custom.html" and formatted_body: + messagevars["body_text_html"] = safe_markup(formatted_body) + elif body: + messagevars["body_text_html"] = safe_text(body) return messagevars From 2884712ca733f45d32468ecf2ede7a1518e85be4 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 17 Jun 2016 14:47:33 +0100 Subject: [PATCH 14/20] Only re-sign our own events --- synapse/federation/federation_server.py | 15 +++++++++------ synapse/handlers/federation.py | 15 +++++++++------ 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index fe92457ba..2a589524a 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -193,13 +193,16 @@ class FederationServer(FederationBase): ) for event in auth_chain: - event.signatures.update( - compute_event_signature( - event, - self.hs.hostname, - self.hs.config.signing_key[0] + # We sign these again because there was a bug where we + # incorrectly signed things the first time round + if self.hs.is_mine_id(event.event_id): + event.signatures.update( + compute_event_signature( + event, + self.hs.hostname, + self.hs.config.signing_key[0] + ) ) - ) else: raise NotImplementedError("Specify an event") diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py index c2df43e2f..6c0bc7eaf 100644 --- a/synapse/handlers/federation.py +++ b/synapse/handlers/federation.py @@ -1018,13 +1018,16 @@ class FederationHandler(BaseHandler): res = results.values() for event in res: - event.signatures.update( - compute_event_signature( - event, - self.hs.hostname, - self.hs.config.signing_key[0] + # We sign these again because there was a bug where we + # incorrectly signed things the first time round + if self.hs.is_mine_id(event.event_id): + event.signatures.update( + compute_event_signature( + event, + self.hs.hostname, + self.hs.config.signing_key[0] + ) ) - ) defer.returnValue(res) else: From 3e41de05cc13220f5cd88ae78002adf782728322 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 17 Jun 2016 15:11:22 +0100 Subject: [PATCH 15/20] Turn use_frozen_events off by default --- synapse/config/server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/config/server.py b/synapse/config/server.py index f370b22c3..7840dc3ad 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -27,7 +27,7 @@ class ServerConfig(Config): self.daemonize = config.get("daemonize") self.print_pidfile = config.get("print_pidfile") self.user_agent_suffix = config.get("user_agent_suffix") - self.use_frozen_dicts = config.get("use_frozen_dicts", True) + self.use_frozen_dicts = config.get("use_frozen_dicts", False) self.public_baseurl = config.get("public_baseurl") self.secondary_directory_servers = config.get("secondary_directory_servers", []) From 0113ad36ee7bc315aa162c42277b90764825f219 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 17 Jun 2016 15:13:13 +0100 Subject: [PATCH 16/20] Enable use_frozen_events in tests --- tests/utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/utils.py b/tests/utils.py index e19ae581e..6e41ae1ff 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -54,6 +54,7 @@ def setup_test_homeserver(name="test", datastore=None, config=None, **kargs): config.trusted_third_party_id_servers = [] config.room_invite_state_types = [] + config.use_frozen_dicts = True config.database_config = {"name": "sqlite3"} if "clock" not in kargs: From 120c2387053bdc30824d6b15931532664f739192 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 17 Jun 2016 16:10:37 +0100 Subject: [PATCH 17/20] Disable responding with canonical json for federation --- synapse/federation/transport/server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py index 6fc3e2207..8a1965f45 100644 --- a/synapse/federation/transport/server.py +++ b/synapse/federation/transport/server.py @@ -37,7 +37,7 @@ class TransportLayerServer(JsonResource): self.hs = hs self.clock = hs.get_clock() - super(TransportLayerServer, self).__init__(hs) + super(TransportLayerServer, self).__init__(hs, canonical_json=False) self.authenticator = Authenticator(hs) self.ratelimiter = FederationRateLimiter( From 9f1800fba852314332d7e682484e456d28838619 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 17 Jun 2016 19:14:16 +0100 Subject: [PATCH 18/20] Remove registered_users from the distributor. The only place that was observed was to set the profile. I've made it so that the profile is set within store.register in the same transaction that creates the user. This required some slight changes to the registration code for upgrading guest users, since it previously relied on the distributor swallowing errors if the profile already existed. --- synapse/handlers/profile.py | 7 ------- synapse/handlers/register.py | 23 ++++++++++------------- synapse/storage/profile.py | 6 ------ synapse/storage/registration.py | 17 ++++++++++++++--- synapse/util/distributor.py | 4 ---- 5 files changed, 24 insertions(+), 33 deletions(-) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index e37409170..711a6a567 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -36,13 +36,6 @@ class ProfileHandler(BaseHandler): "profile", self.on_profile_query ) - distributor = hs.get_distributor() - - distributor.observe("registered_user", self.registered_user) - - def registered_user(self, user): - return self.store.create_profile(user.localpart) - @defer.inlineCallbacks def get_displayname(self, target_user): if self.hs.is_mine(target_user): diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index e0aaefe7b..4fb12915d 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -23,7 +23,6 @@ from synapse.api.errors import ( from ._base import BaseHandler from synapse.util.async import run_on_reactor from synapse.http.client import CaptchaServerHttpClient -from synapse.util.distributor import registered_user import logging import urllib @@ -37,8 +36,6 @@ class RegistrationHandler(BaseHandler): super(RegistrationHandler, self).__init__(hs) self.auth = hs.get_auth() - self.distributor = hs.get_distributor() - self.distributor.declare("registered_user") self.captcha_client = CaptchaServerHttpClient(hs) self._next_generated_user_id = None @@ -140,9 +137,10 @@ class RegistrationHandler(BaseHandler): password_hash=password_hash, was_guest=was_guest, make_guest=make_guest, + create_profile_with_localpart=( + None if was_guest else user.localpart + ), ) - - yield registered_user(self.distributor, user) else: # autogen a sequential user ID attempts = 0 @@ -160,7 +158,8 @@ class RegistrationHandler(BaseHandler): user_id=user_id, token=token, password_hash=password_hash, - make_guest=make_guest + make_guest=make_guest, + create_profile_with_localpart=user.localpart, ) except SynapseError: # if user id is taken, just generate another @@ -168,7 +167,6 @@ class RegistrationHandler(BaseHandler): user_id = None token = None attempts += 1 - yield registered_user(self.distributor, user) # We used to generate default identicons here, but nowadays # we want clients to generate their own as part of their branding @@ -201,8 +199,8 @@ class RegistrationHandler(BaseHandler): token=token, password_hash="", appservice_id=service_id, + create_profile_with_localpart=user.localpart, ) - yield registered_user(self.distributor, user) defer.returnValue((user_id, token)) @defer.inlineCallbacks @@ -248,9 +246,9 @@ class RegistrationHandler(BaseHandler): yield self.store.register( user_id=user_id, token=token, - password_hash=None + password_hash=None, + create_profile_with_localpart=user.localpart, ) - yield registered_user(self.distributor, user) except Exception as e: yield self.store.add_access_token_to_user(user_id, token) # Ignore Registration errors @@ -395,10 +393,9 @@ class RegistrationHandler(BaseHandler): yield self.store.register( user_id=user_id, token=token, - password_hash=None + password_hash=None, + create_profile_with_localpart=user.localpart, ) - - yield registered_user(self.distributor, user) else: yield self.store.user_delete_access_tokens(user_id=user_id) yield self.store.add_access_token_to_user(user_id=user_id, token=token) diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py index 26a40905a..c3c3f9ffd 100644 --- a/synapse/storage/profile.py +++ b/synapse/storage/profile.py @@ -17,12 +17,6 @@ from ._base import SQLBaseStore class ProfileStore(SQLBaseStore): - def create_profile(self, user_localpart): - return self._simple_insert( - table="profiles", - values={"user_id": user_localpart}, - desc="create_profile", - ) def get_profile_displayname(self, user_localpart): return self._simple_select_one_onecol( diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index bda84a744..3de9e0f70 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -76,7 +76,8 @@ class RegistrationStore(SQLBaseStore): @defer.inlineCallbacks def register(self, user_id, token, password_hash, - was_guest=False, make_guest=False, appservice_id=None): + was_guest=False, make_guest=False, appservice_id=None, + create_profile_with_localpart=None): """Attempts to register an account. Args: @@ -88,6 +89,8 @@ class RegistrationStore(SQLBaseStore): make_guest (boolean): True if the the new user should be guest, false to add a regular user account. appservice_id (str): The ID of the appservice registering the user. + create_profile_with_localpart (str): Optionally create a profile for + the given localpart. Raises: StoreError if the user_id could not be registered. """ @@ -99,7 +102,8 @@ class RegistrationStore(SQLBaseStore): password_hash, was_guest, make_guest, - appservice_id + appservice_id, + create_profile_with_localpart, ) self.get_user_by_id.invalidate((user_id,)) self.is_guest.invalidate((user_id,)) @@ -112,7 +116,8 @@ class RegistrationStore(SQLBaseStore): password_hash, was_guest, make_guest, - appservice_id + appservice_id, + create_profile_with_localpart, ): now = int(self.clock.time()) @@ -157,6 +162,12 @@ class RegistrationStore(SQLBaseStore): (next_id, user_id, token,) ) + if create_profile_with_localpart: + txn.execute( + "INSERT INTO profiles(user_id) VALUES (?)", + (create_profile_with_localpart,) + ) + @cached() def get_user_by_id(self, user_id): return self._simple_select_one( diff --git a/synapse/util/distributor.py b/synapse/util/distributor.py index d7cccc06b..e68f94ce7 100644 --- a/synapse/util/distributor.py +++ b/synapse/util/distributor.py @@ -27,10 +27,6 @@ import logging logger = logging.getLogger(__name__) -def registered_user(distributor, user): - return distributor.fire("registered_user", user) - - def user_left_room(distributor, user, room_id): return preserve_context_over_fn( distributor.fire, From 0c13d45522c5c8c0b68322498a220969eb894ad5 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 17 Jun 2016 19:18:53 +0100 Subject: [PATCH 19/20] Add a comment on why we don't create a profile for upgrading users --- synapse/handlers/register.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index 4fb12915d..0b7517221 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -138,6 +138,7 @@ class RegistrationHandler(BaseHandler): was_guest=was_guest, make_guest=make_guest, create_profile_with_localpart=( + # If the user was a guest then they already have a profile None if was_guest else user.localpart ), ) From 41e4b2efeafa6e2f4cbfef4f30620b9f58b020a4 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 17 Jun 2016 19:20:47 +0100 Subject: [PATCH 20/20] Add the create_profile method back since the tests use it --- synapse/storage/profile.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py index c3c3f9ffd..26a40905a 100644 --- a/synapse/storage/profile.py +++ b/synapse/storage/profile.py @@ -17,6 +17,12 @@ from ._base import SQLBaseStore class ProfileStore(SQLBaseStore): + def create_profile(self, user_localpart): + return self._simple_insert( + table="profiles", + values={"user_id": user_localpart}, + desc="create_profile", + ) def get_profile_displayname(self, user_localpart): return self._simple_select_one_onecol(