Don't spuriously create new listeners

Change listener config.

CHANGES.rst

@@ -1,3 +1,31 @@
+Changes in synapse v0.9.2-r2 (2015-06-15)
+=========================================
+
+Fix packaging so that schema delta python files get included in the package.
+
+Changes in synapse v0.9.2 (2015-06-12)
+======================================
+
+General:
+
+* Use ultrajson for json (de)serialisation when a canonical encoding is not
+  required. Ultrajson is significantly faster than simplejson in certain
+  circumstances.
+* Use connection pools for outgoing HTTP connections.
+* Process thumbnails on separate threads.
+
+Configuration:
+
+* Add option, ``gzip_responses``, to disable HTTP response compression.
+
+Federation:
+
+* Improve resilience of backfill by ensuring we fetch any missing auth events.
+* Improve performance of backfill and joining remote rooms by removing
+  unnecessary computations. This included handling events we'd previously
+  handled as well as attempting to compute the current state for outliers.
+
+
 Changes in synapse v0.9.1 (2015-05-26)
 ======================================
 

MANIFEST.in

@@ -5,6 +5,7 @@ include *.rst
 include demo/README
 
 recursive-include synapse/storage/schema *.sql
+recursive-include synapse/storage/schema *.py
 
 recursive-include demo *.dh
 recursive-include demo *.py

contrib/systemd/log_config.yaml

@@ -21,3 +21,5 @@ handlers:
 root:
     level: INFO
     handlers: [journal]
+
+disable_existing_loggers: False

synapse/__init__.py

@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """
 
-__version__ = "0.9.1"
+__version__ = "0.9.2-r2"

synapse/app/homeserver.py

@@ -34,7 +34,7 @@ from twisted.application import service
 from twisted.enterprise import adbapi
 from twisted.web.resource import Resource, EncodingResourceWrapper
 from twisted.web.static import File
-from twisted.web.server import Site, GzipEncoderFactory
+from twisted.web.server import Site, GzipEncoderFactory, Request
 from twisted.web.http import proxiedLogFormatter, combinedLogFormatter
 from synapse.http.server import JsonResource, RootRedirect
 from synapse.rest.media.v0.content_repository import ContentRepoResource
@@ -54,6 +54,8 @@ from synapse.rest.client.v1 import ClientV1RestResource
 from synapse.rest.client.v2_alpha import ClientV2AlphaRestResource
 from synapse.metrics.resource import MetricsResource, METRICS_PREFIX
 
+from synapse import events
+
 from daemonize import Daemonize
 import twisted.manhole.telnet
 
@@ -61,7 +63,6 @@ import synapse
 
 import logging
 import os
-import re
 import resource
 import subprocess
 
@@ -85,10 +86,10 @@ class SynapseHomeServer(HomeServer):
         return MatrixFederationHttpClient(self)
 
     def build_resource_for_client(self):
-        return gz_wrap(ClientV1RestResource(self))
+        return ClientV1RestResource(self)
 
     def build_resource_for_client_v2_alpha(self):
-        return gz_wrap(ClientV2AlphaRestResource(self))
+        return ClientV2AlphaRestResource(self)
 
     def build_resource_for_federation(self):
         return JsonResource(self)
@@ -137,152 +138,102 @@ class SynapseHomeServer(HomeServer):
             **self.db_config.get("args", {})
         )
 
-    def create_resource_tree(self, redirect_root_to_web_client):
-        """Create the resource tree for this Home Server.
+    def _listener_http(self, config, listener_config):
+        port = listener_config["port"]
+        bind_address = listener_config.get("bind_address", "")
+        tls = listener_config.get("tls", False)
 
-        This in unduly complicated because Twisted does not support putting
-        child resources more than 1 level deep at a time.
-
-        Args:
-            web_client (bool): True to enable the web client.
-            redirect_root_to_web_client (bool): True to redirect '/' to the
-            location of the web client. This does nothing if web_client is not
-            True.
-        """
-        config = self.get_config()
-        web_client = config.web_client
-
-        # list containing (path_str, Resource) e.g:
-        # [ ("/aaa/bbb/cc", Resource1), ("/aaa/dummy", Resource2) ]
-        desired_tree = [
-            (CLIENT_PREFIX, self.get_resource_for_client()),
-            (CLIENT_V2_ALPHA_PREFIX, self.get_resource_for_client_v2_alpha()),
-            (FEDERATION_PREFIX, self.get_resource_for_federation()),
-            (CONTENT_REPO_PREFIX, self.get_resource_for_content_repo()),
-            (SERVER_KEY_PREFIX, self.get_resource_for_server_key()),
-            (SERVER_KEY_V2_PREFIX, self.get_resource_for_server_key_v2()),
-            (MEDIA_PREFIX, self.get_resource_for_media_repository()),
-            (STATIC_PREFIX, self.get_resource_for_static_content()),
-        ]
-
-        if web_client:
-            logger.info("Adding the web client.")
-            desired_tree.append((WEB_CLIENT_PREFIX,
-                                self.get_resource_for_web_client()))
-
-        if web_client and redirect_root_to_web_client:
-            self.root_resource = RootRedirect(WEB_CLIENT_PREFIX)
-        else:
-            self.root_resource = Resource()
+        if tls and config.no_tls:
+            return
 
         metrics_resource = self.get_resource_for_metrics()
-        if config.metrics_port is None and metrics_resource is not None:
-            desired_tree.append((METRICS_PREFIX, metrics_resource))
 
-        # ideally we'd just use getChild and putChild but getChild doesn't work
-        # unless you give it a Request object IN ADDITION to the name :/ So
-        # instead, we'll store a copy of this mapping so we can actually add
-        # extra resources to existing nodes. See self._resource_id for the key.
-        resource_mappings = {}
-        for full_path, res in desired_tree:
-            logger.info("Attaching %s to path %s", res, full_path)
-            last_resource = self.root_resource
-            for path_seg in full_path.split('/')[1:-1]:
-                if path_seg not in last_resource.listNames():
-                    # resource doesn't exist, so make a "dummy resource"
-                    child_resource = Resource()
-                    last_resource.putChild(path_seg, child_resource)
-                    res_id = self._resource_id(last_resource, path_seg)
-                    resource_mappings[res_id] = child_resource
-                    last_resource = child_resource
-                else:
-                    # we have an existing Resource, use that instead.
-                    res_id = self._resource_id(last_resource, path_seg)
-                    last_resource = resource_mappings[res_id]
+        resources = {}
+        for res in listener_config["resources"]:
+            for name in res["names"]:
+                if name == "client":
+                    if res["compress"]:
+                        client_v1 = gz_wrap(self.get_resource_for_client())
+                        client_v2 = gz_wrap(self.get_resource_for_client_v2_alpha())
+                    else:
+                        client_v1 = self.get_resource_for_client()
+                        client_v2 = self.get_resource_for_client_v2_alpha()
 
-            # ===========================
-            # now attach the actual desired resource
-            last_path_seg = full_path.split('/')[-1]
+                    resources.update({
+                        CLIENT_PREFIX: client_v1,
+                        CLIENT_V2_ALPHA_PREFIX: client_v2,
+                    })
 
-            # if there is already a resource here, thieve its children and
-            # replace it
-            res_id = self._resource_id(last_resource, last_path_seg)
-            if res_id in resource_mappings:
-                # there is a dummy resource at this path already, which needs
-                # to be replaced with the desired resource.
-                existing_dummy_resource = resource_mappings[res_id]
-                for child_name in existing_dummy_resource.listNames():
-                    child_res_id = self._resource_id(existing_dummy_resource,
-                                                     child_name)
-                    child_resource = resource_mappings[child_res_id]
-                    # steal the children
-                    res.putChild(child_name, child_resource)
+                if name == "federation":
+                    resources.update({
+                        FEDERATION_PREFIX: self.get_resource_for_federation(),
+                    })
 
-            # finally, insert the desired resource in the right place
-            last_resource.putChild(last_path_seg, res)
-            res_id = self._resource_id(last_resource, last_path_seg)
-            resource_mappings[res_id] = res
+                if name in ["static", "client"]:
+                    resources.update({
+                        STATIC_PREFIX: self.get_resource_for_static_content(),
+                    })
 
-        return self.root_resource
+                if name in ["media", "federation", "client"]:
+                    resources.update({
+                        MEDIA_PREFIX: self.get_resource_for_media_repository(),
+                        CONTENT_REPO_PREFIX: self.get_resource_for_content_repo(),
+                    })
 
-    def _resource_id(self, resource, path_seg):
-        """Construct an arbitrary resource ID so you can retrieve the mapping
-        later.
+                if name in ["keys", "federation"]:
+                    resources.update({
+                        SERVER_KEY_PREFIX: self.get_resource_for_server_key(),
+                        SERVER_KEY_V2_PREFIX: self.get_resource_for_server_key_v2(),
+                    })
 
-        If you want to represent resource A putChild resource B with path C,
-        the mapping should looks like _resource_id(A,C) = B.
+                if name == "webclient":
+                    resources[WEB_CLIENT_PREFIX] = self.get_resource_for_web_client()
 
-        Args:
-            resource (Resource): The *parent* Resource
-            path_seg (str): The name of the child Resource to be attached.
-        Returns:
-            str: A unique string which can be a key to the child Resource.
-        """
-        return "%s-%s" % (resource, path_seg)
+                if name == "metrics" and metrics_resource:
+                    resources[METRICS_PREFIX] = metrics_resource
+
+        root_resource = create_resource_tree(resources)
+        if tls:
+            reactor.listenSSL(
+                port,
+                SynapseSite(
+                    "synapse.access.https",
+                    listener_config,
+                    root_resource,
+                ),
+                self.tls_context_factory,
+                interface=bind_address
+            )
+        else:
+            reactor.listenTCP(
+                port,
+                SynapseSite(
+                    "synapse.access.https",
+                    listener_config,
+                    root_resource,
+                ),
+                interface=bind_address
+            )
+        logger.info("Synapse now listening on port %d", port)
 
     def start_listening(self):
         config = self.get_config()
 
-        if not config.no_tls and config.bind_port is not None:
-            reactor.listenSSL(
-                config.bind_port,
-                SynapseSite(
-                    "synapse.access.https",
-                    config,
-                    self.root_resource,
-                ),
-                self.tls_context_factory,
-                interface=config.bind_host
-            )
-            logger.info("Synapse now listening on port %d", config.bind_port)
-
-        if config.unsecure_port is not None:
-            reactor.listenTCP(
-                config.unsecure_port,
-                SynapseSite(
-                    "synapse.access.http",
-                    config,
-                    self.root_resource,
-                ),
-                interface=config.bind_host
-            )
-            logger.info("Synapse now listening on port %d", config.unsecure_port)
-
-        metrics_resource = self.get_resource_for_metrics()
-        if metrics_resource and config.metrics_port is not None:
-            reactor.listenTCP(
-                config.metrics_port,
-                SynapseSite(
-                    "synapse.access.metrics",
-                    config,
-                    metrics_resource,
-                ),
-                interface=config.metrics_bind_host,
-            )
-            logger.info(
-                "Metrics now running on %s port %d",
-                config.metrics_bind_host, config.metrics_port,
-            )
+        for listener in config.listeners:
+            if listener["type"] == "http":
+                self._listener_http(config, listener)
+            elif listener["type"] == "manhole":
+                f = twisted.manhole.telnet.ShellFactory()
+                f.username = "matrix"
+                f.password = "rabbithole"
+                f.namespace['hs'] = self
+                reactor.listenTCP(
+                    listener["port"],
+                    f,
+                    interface=listener.get("bind_address", '127.0.0.1')
+                )
+            else:
+                logger.warn("Unrecognized listener type: %s", listener["type"])
 
     def run_startup_checks(self, db_conn, database_engine):
         all_users_native = are_all_users_on_domain(
@@ -415,10 +366,7 @@ def setup(config_options):
     logger.info("Server hostname: %s", config.server_name)
     logger.info("Server version: %s", version_string)
 
-    if re.search(":[0-9]+$", config.server_name):
-        domain_with_port = config.server_name
-    else:
-        domain_with_port = "%s:%s" % (config.server_name, config.bind_port)
+    events.USE_FROZEN_DICTS = config.use_frozen_dicts
 
     tls_context_factory = context_factory.ServerContextFactory(config)
 
@@ -427,7 +375,6 @@ def setup(config_options):
 
     hs = SynapseHomeServer(
         config.server_name,
-        domain_with_port=domain_with_port,
         upload_dir=os.path.abspath("uploads"),
         db_config=config.database_config,
         tls_context_factory=tls_context_factory,
@@ -437,10 +384,6 @@ def setup(config_options):
         database_engine=database_engine,
     )
 
-    hs.create_resource_tree(
-        redirect_root_to_web_client=True,
-    )
-
     logger.info("Preparing database: %r...", config.database_config)
 
     try:
@@ -465,13 +408,6 @@ def setup(config_options):
 
     logger.info("Database prepared in %r.", config.database_config)
 
-    if config.manhole:
-        f = twisted.manhole.telnet.ShellFactory()
-        f.username = "matrix"
-        f.password = "rabbithole"
-        f.namespace['hs'] = hs
-        reactor.listenTCP(config.manhole, f, interface='127.0.0.1')
-
     hs.start_listening()
 
     hs.get_pusherpool().start()
@@ -497,6 +433,28 @@ class SynapseService(service.Service):
         return self._port.stopListening()
 
 
+class XForwardedForRequest(Request):
+    def __init__(self, *args, **kw):
+        Request.__init__(self, *args, **kw)
+
+    """
+    Add a layer on top of another request that only uses the value of an
+    X-Forwarded-For header as the result of C{getClientIP}.
+    """
+    def getClientIP(self):
+        """
+        @return: The client address (the first address) in the value of the
+            I{X-Forwarded-For header}.  If the header is not present, return
+            C{b"-"}.
+        """
+        return self.requestHeaders.getRawHeaders(
+            b"x-forwarded-for", [b"-"])[0].split(b",")[0].strip()
+
+
+def XForwardedFactory(*args, **kwargs):
+    return XForwardedForRequest(*args, **kwargs)
+
+
 class SynapseSite(Site):
     """
     Subclass of a twisted http Site that does access logging with python's
@@ -504,7 +462,8 @@ class SynapseSite(Site):
     """
     def __init__(self, logger_name, config, resource, *args, **kwargs):
         Site.__init__(self, resource, *args, **kwargs)
-        if config.captcha_ip_origin_is_x_forwarded:
+        if config.get("x_forwarded", False):
+            self.requestFactory = XForwardedFactory
             self._log_formatter = proxiedLogFormatter
         else:
             self._log_formatter = combinedLogFormatter
@@ -515,6 +474,87 @@ class SynapseSite(Site):
         self.access_logger.info(line)
 
 
+def create_resource_tree(desired_tree, redirect_root_to_web_client=True):
+    """Create the resource tree for this Home Server.
+
+    This in unduly complicated because Twisted does not support putting
+    child resources more than 1 level deep at a time.
+
+    Args:
+        web_client (bool): True to enable the web client.
+        redirect_root_to_web_client (bool): True to redirect '/' to the
+        location of the web client. This does nothing if web_client is not
+        True.
+    """
+    if redirect_root_to_web_client and WEB_CLIENT_PREFIX in desired_tree:
+        root_resource = RootRedirect(WEB_CLIENT_PREFIX)
+    else:
+        root_resource = Resource()
+
+    # ideally we'd just use getChild and putChild but getChild doesn't work
+    # unless you give it a Request object IN ADDITION to the name :/ So
+    # instead, we'll store a copy of this mapping so we can actually add
+    # extra resources to existing nodes. See self._resource_id for the key.
+    resource_mappings = {}
+    for full_path, res in desired_tree.items():
+        logger.info("Attaching %s to path %s", res, full_path)
+        last_resource = root_resource
+        for path_seg in full_path.split('/')[1:-1]:
+            if path_seg not in last_resource.listNames():
+                # resource doesn't exist, so make a "dummy resource"
+                child_resource = Resource()
+                last_resource.putChild(path_seg, child_resource)
+                res_id = _resource_id(last_resource, path_seg)
+                resource_mappings[res_id] = child_resource
+                last_resource = child_resource
+            else:
+                # we have an existing Resource, use that instead.
+                res_id = _resource_id(last_resource, path_seg)
+                last_resource = resource_mappings[res_id]
+
+        # ===========================
+        # now attach the actual desired resource
+        last_path_seg = full_path.split('/')[-1]
+
+        # if there is already a resource here, thieve its children and
+        # replace it
+        res_id = _resource_id(last_resource, last_path_seg)
+        if res_id in resource_mappings:
+            # there is a dummy resource at this path already, which needs
+            # to be replaced with the desired resource.
+            existing_dummy_resource = resource_mappings[res_id]
+            for child_name in existing_dummy_resource.listNames():
+                child_res_id = _resource_id(
+                    existing_dummy_resource, child_name
+                )
+                child_resource = resource_mappings[child_res_id]
+                # steal the children
+                res.putChild(child_name, child_resource)
+
+        # finally, insert the desired resource in the right place
+        last_resource.putChild(last_path_seg, res)
+        res_id = _resource_id(last_resource, last_path_seg)
+        resource_mappings[res_id] = res
+
+    return root_resource
+
+
+def _resource_id(resource, path_seg):
+    """Construct an arbitrary resource ID so you can retrieve the mapping
+    later.
+
+    If you want to represent resource A putChild resource B with path C,
+    the mapping should looks like _resource_id(A,C) = B.
+
+    Args:
+        resource (Resource): The *parent* Resource
+        path_seg (str): The name of the child Resource to be attached.
+    Returns:
+        str: A unique string which can be a key to the child Resource.
+    """
+    return "%s-%s" % (resource, path_seg)
+
+
 def run(hs):
     PROFILE_SYNAPSE = False
     if PROFILE_SYNAPSE:

synapse/config/captcha.py

@@ -21,11 +21,8 @@ class CaptchaConfig(Config):
         self.recaptcha_private_key = config["recaptcha_private_key"]
         self.recaptcha_public_key = config["recaptcha_public_key"]
         self.enable_registration_captcha = config["enable_registration_captcha"]
-        # XXX: This is used for more than just captcha
-        self.captcha_ip_origin_is_x_forwarded = (
-            config["captcha_ip_origin_is_x_forwarded"]
-        )
         self.captcha_bypass_secret = config.get("captcha_bypass_secret")
+        self.recaptcha_siteverify_api = config["recaptcha_siteverify_api"]
 
     def default_config(self, config_dir_path, server_name):
         return """\
@@ -42,10 +39,9 @@ class CaptchaConfig(Config):
         # public/private key.
         enable_registration_captcha: False
 
-        # When checking captchas, use the X-Forwarded-For (XFF) header
-        # as the client IP and not the actual client IP.
-        captcha_ip_origin_is_x_forwarded: False
-
         # A secret key used to bypass the captcha test entirely.
         #captcha_bypass_secret: "YOUR_SECRET_HERE"
+
+        # The API endpoint to use for verifying m.login.recaptcha responses.
+        recaptcha_siteverify_api: "https://www.google.com/recaptcha/api/siteverify"
         """

synapse/config/metrics.py

@@ -28,10 +28,4 @@ class MetricsConfig(Config):
 
         # Enable collection and rendering of performance metrics
         enable_metrics: False
-
-        # Separate port to accept metrics requests on
-        # metrics_port: 8081
-
-        # Which host to bind the metric listener to
-        # metrics_bind_host: 127.0.0.1
         """

synapse/config/registration.py

@@ -39,7 +39,7 @@ class RegistrationConfig(Config):
         ## Registration ##
 
         # Enable registration for new users.
-        enable_registration: True
+        enable_registration: False
 
         # If set, allows registration by anyone who also has the shared
         # secret, even if registration is otherwise disabled.

synapse/config/server.py

@@ -20,24 +20,97 @@ class ServerConfig(Config):
 
     def read_config(self, config):
         self.server_name = config["server_name"]
-        self.bind_port = config["bind_port"]
-        self.bind_host = config["bind_host"]
-        self.unsecure_port = config["unsecure_port"]
-        self.manhole = config.get("manhole")
         self.pid_file = self.abspath(config.get("pid_file"))
         self.web_client = config["web_client"]
         self.soft_file_limit = config["soft_file_limit"]
         self.daemonize = config.get("daemonize")
+        self.use_frozen_dicts = config.get("use_frozen_dicts", True)
+
+        self.listeners = config.get("listeners", [])
+
+        bind_port = config.get("bind_port")
+        if bind_port:
+            self.listeners = []
+            bind_host = config.get("bind_host", "")
+            gzip_responses = config.get("gzip_responses", True)
+
+            names = ["client", "webclient"] if self.web_client else ["client"]
+
+            self.listeners.append({
+                "port": bind_port,
+                "bind_address": bind_host,
+                "tls": True,
+                "type": "http",
+                "resources": [
+                    {
+                        "names": names,
+                        "compress": gzip_responses,
+                    },
+                    {
+                        "names": ["federation"],
+                        "compress": False,
+                    }
+                ]
+            })
+
+            unsecure_port = config.get("unsecure_port", bind_port - 400)
+            if unsecure_port:
+                self.listeners.append({
+                    "port": unsecure_port,
+                    "bind_address": bind_host,
+                    "tls": False,
+                    "type": "http",
+                    "resources": [
+                        {
+                            "names": names,
+                            "compress": gzip_responses,
+                        },
+                        {
+                            "names": ["federation"],
+                            "compress": False,
+                        }
+                    ]
+                })
+
+        manhole = config.get("manhole")
+        if manhole:
+            self.listeners.append({
+                "port": manhole,
+                "bind_address": "127.0.0.1",
+                "type": "manhole",
+            })
+
+        metrics_port = config.get("metrics_port")
+        if metrics_port:
+            self.listeners.append({
+                "port": metrics_port,
+                "bind_address": config.get("metrics_bind_host", "127.0.0.1"),
+                "tls": False,
+                "type": "http",
+                "resources": [
+                    {
+                        "names": ["metrics"],
+                        "compress": False,
+                    },
+                ]
+            })
 
         # Attempt to guess the content_addr for the v0 content repostitory
         content_addr = config.get("content_addr")
         if not content_addr:
+            for listener in self.listeners:
+                if listener["type"] == "http" and not listener.get("tls", False):
+                    unsecure_port = listener["port"]
+                    break
+            else:
+                raise RuntimeError("Could not determine 'content_addr'")
+
             host = self.server_name
             if ':' not in host:
-                host = "%s:%d" % (host, self.unsecure_port)
+                host = "%s:%d" % (host, unsecure_port)
             else:
                 host = host.split(':')[0]
-                host = "%s:%d" % (host, self.unsecure_port)
+                host = "%s:%d" % (host, unsecure_port)
             content_addr = "http://%s" % (host,)
 
         self.content_addr = content_addr
@@ -59,18 +132,6 @@ class ServerConfig(Config):
         # e.g. matrix.org, localhost:8080, etc.
         server_name: "%(server_name)s"
 
-        # The port to listen for HTTPS requests on.
-        # For when matrix traffic is sent directly to synapse.
-        bind_port: %(bind_port)s
-
-        # The port to listen for HTTP requests on.
-        # For when matrix traffic passes through loadbalancer that unwraps TLS.
-        unsecure_port: %(unsecure_port)s
-
-        # Local interface to listen on.
-        # The empty string will cause synapse to listen on all interfaces.
-        bind_host: ""
-
         # When running as a daemon, the file to store the pid in
         pid_file: %(pid_file)s
 
@@ -82,9 +143,64 @@ class ServerConfig(Config):
         # hard limit.
         soft_file_limit: 0
 
-        # Turn on the twisted telnet manhole service on localhost on the given
-        # port.
-        #manhole: 9000
+        # List of ports that Synapse should listen on, their purpose and their
+        # configuration.
+        listeners:
+          # Main HTTPS listener
+          # For when matrix traffic is sent directly to synapse.
+          -
+            # The port to listen for HTTPS requests on.
+            port: %(bind_port)s
+
+            # Local interface to listen on.
+            # The empty string will cause synapse to listen on all interfaces.
+            bind_address: ''
+
+            # This is a 'http' listener, allows us to specify 'resources'.
+            type: http
+
+            tls: true
+
+            # Use the X-Forwarded-For (XFF) header as the client IP and not the
+            # actual client IP.
+            x_forwarded: false
+
+            # List of HTTP resources to serve on this listener.
+            resources:
+              -
+                # List of resources to host on this listener.
+                names:
+                  - client     # The client-server APIs, both v1 and v2
+                  - webclient  # The bundled webclient.
+
+                # Should synapse compress HTTP responses to clients that support it?
+                # This should be disabled if running synapse behind a load balancer
+                # that can do automatic compression.
+                compress: true
+
+              - names: [federation]  # Federation APIs
+                compress: false
+
+          # Unsecure HTTP listener,
+          # For when matrix traffic passes through loadbalancer that unwraps TLS.
+          - port: %(unsecure_port)s
+            tls: false
+            bind_address: ''
+            type: http
+
+            x_forwarded: false
+
+            resources:
+              - names: [client, webclient]
+                compress: true
+              - names: [federation]
+                compress: false
+
+          # Turn on the twisted telnet manhole service on localhost on the given
+          # port.
+          # - port: 9000
+          #   bind_address: 127.0.0.1
+          #   type: manhole
         """ % locals()
 
     def read_arguments(self, args):

synapse/events/__init__.py

@@ -16,6 +16,12 @@
 from synapse.util.frozenutils import freeze
 
 
+# Whether we should use frozen_dict in FrozenEvent. Using frozen_dicts prevents
+# bugs where we accidentally share e.g. signature dicts. However, converting
+# a dict to frozen_dicts is expensive.
+USE_FROZEN_DICTS = True
+
+
 class _EventInternalMetadata(object):
     def __init__(self, internal_metadata_dict):
         self.__dict__ = dict(internal_metadata_dict)
@@ -122,7 +128,10 @@ class FrozenEvent(EventBase):
 
         unsigned = dict(event_dict.pop("unsigned", {}))
 
-        frozen_dict = freeze(event_dict)
+        if USE_FROZEN_DICTS:
+            frozen_dict = freeze(event_dict)
+        else:
+            frozen_dict = event_dict
 
         super(FrozenEvent, self).__init__(
             frozen_dict,

synapse/federation/federation_base.py

@@ -18,8 +18,6 @@ from twisted.internet import defer
 
 from synapse.events.utils import prune_event
 
-from syutil.jsonutil import encode_canonical_json
-
 from synapse.crypto.event_signing import check_event_content_hash
 
 from synapse.api.errors import SynapseError
@@ -120,16 +118,15 @@ class FederationBase(object):
             )
         except SynapseError:
             logger.warn(
-                "Signature check failed for %s redacted to %s",
-                encode_canonical_json(pdu.get_pdu_json()),
-                encode_canonical_json(redacted_pdu_json),
+                "Signature check failed for %s",
+                pdu.event_id,
             )
             raise
 
         if not check_event_content_hash(pdu):
             logger.warn(
-                "Event content has been tampered, redacting %s, %s",
-                pdu.event_id, encode_canonical_json(pdu.get_dict())
+                "Event content has been tampered, redacting.",
+                pdu.event_id,
             )
             defer.returnValue(redacted_event)
 

synapse/federation/transport/server.py

@@ -93,6 +93,8 @@ class TransportLayerServer(object):
 
         yield self.keyring.verify_json_for_server(origin, json_request)
 
+        logger.info("Request from %s", origin)
+
         defer.returnValue((origin, content))
 
     @log_function

synapse/handlers/_base.py

@@ -78,7 +78,9 @@ class BaseHandler(object):
         context = yield state_handler.compute_event_context(builder)
 
         if builder.is_state():
-            builder.prev_state = context.prev_state_events
+            builder.prev_state = yield self.store.add_event_hashes(
+                context.prev_state_events
+            )
 
         yield self.auth.add_auth_events(builder, context)
 

synapse/handlers/auth.py

@@ -187,8 +187,8 @@ class AuthHandler(BaseHandler):
         # each request
         try:
             client = SimpleHttpClient(self.hs)
-            data = yield client.post_urlencoded_get_json(
-                "https://www.google.com/recaptcha/api/siteverify",
+            resp_body = yield client.post_urlencoded_get_json(
+                self.hs.config.recaptcha_siteverify_api,
                 args={
                     'secret': self.hs.config.recaptcha_private_key,
                     'response': user_response,
@@ -198,7 +198,8 @@ class AuthHandler(BaseHandler):
         except PartialDownloadError as pde:
             # Twisted is silly
             data = pde.response
-        resp_body = simplejson.loads(data)
+            resp_body = simplejson.loads(data)
+
         if 'success' in resp_body and resp_body['success']:
             defer.returnValue(True)
         raise LoginError(401, "", errcode=Codes.UNAUTHORIZED)

synapse/handlers/federation.py

@@ -247,9 +247,15 @@ class FederationHandler(BaseHandler):
             if set(e_id for e_id, _ in ev.prev_events) - event_ids
         ]
 
+        logger.info(
+            "backfill: Got %d events with %d edges",
+            len(events), len(edges),
+        )
+
         # For each edge get the current state.
 
         auth_events = {}
+        state_events = {}
         events_to_state = {}
         for e_id in edges:
             state, auth = yield self.replication_layer.get_state_for_room(
@@ -258,12 +264,46 @@ class FederationHandler(BaseHandler):
                 event_id=e_id
             )
             auth_events.update({a.event_id: a for a in auth})
+            auth_events.update({s.event_id: s for s in state})
+            state_events.update({s.event_id: s for s in state})
             events_to_state[e_id] = state
 
+        seen_events = yield self.store.have_events(
+            set(auth_events.keys()) | set(state_events.keys())
+        )
+
+        all_events = events + state_events.values() + auth_events.values()
+        required_auth = set(
+            a_id for event in all_events for a_id, _ in event.auth_events
+        )
+
+        missing_auth = required_auth - set(auth_events)
+        results = yield defer.gatherResults(
+            [
+                self.replication_layer.get_pdu(
+                    [dest],
+                    event_id,
+                    outlier=True,
+                    timeout=10000,
+                )
+                for event_id in missing_auth
+            ],
+            consumeErrors=True
+        ).addErrback(unwrapFirstError)
+        auth_events.update({a.event_id: a for a in results})
+
         yield defer.gatherResults(
             [
-                self._handle_new_event(dest, a)
+                self._handle_new_event(
+                    dest, a,
+                    auth_events={
+                        (auth_events[a_id].type, auth_events[a_id].state_key):
+                        auth_events[a_id]
+                        for a_id, _ in a.auth_events
+                    },
+                )
                 for a in auth_events.values()
+                if a.event_id not in seen_events
             ],
             consumeErrors=True,
         ).addErrback(unwrapFirstError)
@@ -274,6 +314,11 @@ class FederationHandler(BaseHandler):
                     dest, event_map[e_id],
                     state=events_to_state[e_id],
                     backfilled=True,
+                    auth_events={
+                        (auth_events[a_id].type, auth_events[a_id].state_key):
+                        auth_events[a_id]
+                        for a_id, _ in event_map[e_id].auth_events
+                    },
                 )
                 for e_id in events_to_state
             ],
@@ -900,8 +945,10 @@ class FederationHandler(BaseHandler):
             event.event_id, event.signatures,
         )
 
+        outlier = event.internal_metadata.is_outlier()
+
         context = yield self.state_handler.compute_event_context(
-            event, old_state=state
+            event, old_state=state, outlier=outlier,
         )
 
         if not auth_events:
@@ -912,7 +959,7 @@ class FederationHandler(BaseHandler):
             event.event_id, auth_events,
         )
 
-        is_new_state = not event.internal_metadata.is_outlier()
+        is_new_state = not outlier
 
         # This is a hack to fix some old rooms where the initial join event
         # didn't reference the create event in its auth events.

synapse/http/client.py

@@ -20,7 +20,8 @@ import synapse.metrics
 
 from twisted.internet import defer, reactor
 from twisted.web.client import (
-    Agent, readBody, FileBodyProducer, PartialDownloadError
+    Agent, readBody, FileBodyProducer, PartialDownloadError,
+    HTTPConnectionPool,
 )
 from twisted.web.http_headers import Headers
 
@@ -55,7 +56,9 @@ class SimpleHttpClient(object):
         # The default context factory in Twisted 14.0.0 (which we require) is
         # BrowserLikePolicyForHTTPS which will do regular cert validation
         # 'like a browser'
-        self.agent = Agent(reactor)
+        pool = HTTPConnectionPool(reactor)
+        pool.maxPersistentPerHost = 10
+        self.agent = Agent(reactor, pool=pool)
         self.version_string = hs.version_string
 
     def request(self, method, *args, **kwargs):

synapse/http/matrixfederationclient.py

@@ -16,7 +16,7 @@
 
 from twisted.internet import defer, reactor, protocol
 from twisted.internet.error import DNSLookupError
-from twisted.web.client import readBody, _AgentBase, _URI
+from twisted.web.client import readBody, _AgentBase, _URI, HTTPConnectionPool
 from twisted.web.http_headers import Headers
 from twisted.web._newclient import ResponseDone
 
@@ -103,7 +103,9 @@ class MatrixFederationHttpClient(object):
         self.hs = hs
         self.signing_key = hs.config.signing_key[0]
         self.server_name = hs.hostname
-        self.agent = MatrixFederationHttpAgent(reactor)
+        pool = HTTPConnectionPool(reactor)
+        pool.maxPersistentPerHost = 10
+        self.agent = MatrixFederationHttpAgent(reactor, pool=pool)
         self.clock = hs.get_clock()
         self.version_string = hs.version_string
 

synapse/http/server.py

@@ -19,9 +19,10 @@ from synapse.api.errors import (
 )
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
 import synapse.metrics
+import synapse.events
 
 from syutil.jsonutil import (
-    encode_canonical_json, encode_pretty_printed_json
+    encode_canonical_json, encode_pretty_printed_json, encode_json
 )
 
 from twisted.internet import defer
@@ -168,9 +169,10 @@ class JsonResource(HttpServer, resource.Resource):
 
     _PathEntry = collections.namedtuple("_PathEntry", ["pattern", "callback"])
 
-    def __init__(self, hs):
+    def __init__(self, hs, canonical_json=True):
         resource.Resource.__init__(self)
 
+        self.canonical_json = canonical_json
         self.clock = hs.get_clock()
         self.path_regexs = {}
         self.version_string = hs.version_string
@@ -256,6 +258,7 @@ class JsonResource(HttpServer, resource.Resource):
             response_code_message=response_code_message,
             pretty_print=_request_user_agent_is_curl(request),
             version_string=self.version_string,
+            canonical_json=self.canonical_json,
         )
 
 
@@ -277,11 +280,16 @@ class RootRedirect(resource.Resource):
 
 def respond_with_json(request, code, json_object, send_cors=False,
                       response_code_message=None, pretty_print=False,
-                      version_string=""):
+                      version_string="", canonical_json=True):
     if pretty_print:
         json_bytes = encode_pretty_printed_json(json_object) + "\n"
     else:
-        json_bytes = encode_canonical_json(json_object)
+        if canonical_json:
+            json_bytes = encode_canonical_json(json_object)
+        else:
+            json_bytes = encode_json(
+                json_object, using_frozen_dicts=synapse.events.USE_FROZEN_DICTS
+            )
 
     return respond_with_json_bytes(
         request, code, json_bytes,

synapse/notifier.py

@@ -21,6 +21,7 @@ from synapse.types import StreamToken
 import synapse.metrics
 
 import logging
+import time
 
 
 logger = logging.getLogger(__name__)
@@ -46,8 +47,11 @@ class _NotificationListener(object):
     notify the handler it is sufficient to resolve the deferred.
     """
 
-    def __init__(self, deferred):
+    def __init__(self, deferred, timeout):
         self.deferred = deferred
+        self.created = int(time.time() * 1000)
+        self.timeout = timeout
+        self.have_notified = False
 
     def notified(self):
         return self.deferred.called
@@ -55,6 +59,7 @@ class _NotificationListener(object):
     def notify(self, token):
         """ Inform whoever is listening about the new events.
         """
+        self.have_notified = True
         try:
             self.deferred.callback(token)
         except defer.AlreadyCalledError:
@@ -96,7 +101,10 @@ class _NotifierUserStream(object):
             listeners = self.listeners
             self.listeners = set()
             for listener in listeners:
-                listener.notify(self.current_token)
+                try:
+                    listener.notify(self.current_token)
+                except:
+                    logger.exception("Failed to notify listener")
 
     def remove(self, notifier):
         """ Remove this listener from all the indexes in the Notifier
@@ -308,10 +316,10 @@ class Notifier(object):
         else:
             current_token = user_stream.current_token
 
-        listener = [_NotificationListener(deferred)]
+        listeners = [_NotificationListener(deferred, timeout)]
 
         if timeout and not current_token.is_after(from_token):
-            user_stream.listeners.add(listener[0])
+            user_stream.listeners.update(listeners)
 
         if current_token.is_after(from_token):
             result = yield callback(from_token, current_token)
@@ -321,7 +329,7 @@ class Notifier(object):
         timer = [None]
 
         if result:
-            user_stream.listeners.discard(listener[0])
+            user_stream.listeners.difference_update(listeners)
             defer.returnValue(result)
             return
 
@@ -331,8 +339,9 @@ class Notifier(object):
             def _timeout_listener():
                 timed_out[0] = True
                 timer[0] = None
-                user_stream.listeners.discard(listener[0])
-                listener[0].notify(current_token)
+                user_stream.listeners.difference_update(listeners)
+                for listener in listeners:
+                    listener.notify(current_token)
 
             # We create multiple notification listeners so we have to manage
             # canceling the timeout ourselves.
@@ -340,12 +349,16 @@ class Notifier(object):
 
             while not result and not timed_out[0]:
                 new_token = yield deferred
-                deferred = defer.Deferred()
-                listener[0] = _NotificationListener(deferred)
-                user_stream.listeners.add(listener[0])
+
                 result = yield callback(current_token, new_token)
                 current_token = new_token
 
+                if not result:
+                    deferred = defer.Deferred()
+                    listener = _NotificationListener(deferred, timeout)
+                    listeners.append(listener)
+                    user_stream.listeners.add(listener)
+
         if timer[0] is not None:
             try:
                 self.clock.cancel_call_later(timer[0])

synapse/push/__init__.py

@@ -24,6 +24,7 @@ import baserules
 import logging
 import simplejson as json
 import re
+import random
 
 logger = logging.getLogger(__name__)
 
@@ -256,134 +257,154 @@ class Pusher(object):
             logger.info("Pusher %s for user %s starting from token %s",
                         self.pushkey, self.user_name, self.last_token)
 
+        wait = 0
         while self.alive:
-            from_tok = StreamToken.from_string(self.last_token)
-            config = PaginationConfig(from_token=from_tok, limit='1')
-            chunk = yield self.evStreamHandler.get_stream(
-                self.user_name, config,
-                timeout=100*365*24*60*60*1000, affect_presence=False
-            )
-
-            # limiting to 1 may get 1 event plus 1 presence event, so
-            # pick out the actual event
-            single_event = None
-            for c in chunk['chunk']:
-                if 'event_id' in c:  # Hmmm...
-                    single_event = c
-                    break
-            if not single_event:
-                self.last_token = chunk['end']
-                continue
-
-            if not self.alive:
-                continue
-
-            processed = False
-            actions = yield self._actions_for_event(single_event)
-            tweaks = _tweaks_for_actions(actions)
-
-            if len(actions) == 0:
-                logger.warn("Empty actions! Using default action.")
-                actions = Pusher.DEFAULT_ACTIONS
-
-            if 'notify' not in actions and 'dont_notify' not in actions:
-                logger.warn("Neither notify nor dont_notify in actions: adding default")
-                actions.extend(Pusher.DEFAULT_ACTIONS)
-
-            if 'dont_notify' in actions:
-                logger.debug(
-                    "%s for %s: dont_notify",
-                    single_event['event_id'], self.user_name
+            try:
+                if wait > 0:
+                    yield synapse.util.async.sleep(wait)
+                yield self.get_and_dispatch()
+                wait = 0
+            except:
+                if wait == 0:
+                    wait = 1
+                else:
+                    wait = min(wait * 2, 1800)
+                logger.exception(
+                    "Exception in pusher loop for pushkey %s. Pausing for %ds",
+                    self.pushkey, wait
                 )
+
+    @defer.inlineCallbacks
+    def get_and_dispatch(self):
+        from_tok = StreamToken.from_string(self.last_token)
+        config = PaginationConfig(from_token=from_tok, limit='1')
+        timeout = (300 + random.randint(-60, 60)) * 1000
+        chunk = yield self.evStreamHandler.get_stream(
+            self.user_name, config,
+            timeout=timeout, affect_presence=False
+        )
+
+        # limiting to 1 may get 1 event plus 1 presence event, so
+        # pick out the actual event
+        single_event = None
+        for c in chunk['chunk']:
+            if 'event_id' in c:  # Hmmm...
+                single_event = c
+                break
+        if not single_event:
+            self.last_token = chunk['end']
+            logger.debug("Event stream timeout for pushkey %s", self.pushkey)
+            return
+
+        if not self.alive:
+            return
+
+        processed = False
+        actions = yield self._actions_for_event(single_event)
+        tweaks = _tweaks_for_actions(actions)
+
+        if len(actions) == 0:
+            logger.warn("Empty actions! Using default action.")
+            actions = Pusher.DEFAULT_ACTIONS
+
+        if 'notify' not in actions and 'dont_notify' not in actions:
+            logger.warn("Neither notify nor dont_notify in actions: adding default")
+            actions.extend(Pusher.DEFAULT_ACTIONS)
+
+        if 'dont_notify' in actions:
+            logger.debug(
+                "%s for %s: dont_notify",
+                single_event['event_id'], self.user_name
+            )
+            processed = True
+        else:
+            rejected = yield self.dispatch_push(single_event, tweaks)
+            self.has_unread = True
+            if isinstance(rejected, list) or isinstance(rejected, tuple):
                 processed = True
-            else:
-                rejected = yield self.dispatch_push(single_event, tweaks)
-                self.has_unread = True
-                if isinstance(rejected, list) or isinstance(rejected, tuple):
-                    processed = True
-                    for pk in rejected:
-                        if pk != self.pushkey:
-                            # for sanity, we only remove the pushkey if it
-                            # was the one we actually sent...
-                            logger.warn(
-                                ("Ignoring rejected pushkey %s because we"
-                                 " didn't send it"), pk
-                            )
-                        else:
-                            logger.info(
-                                "Pushkey %s was rejected: removing",
-                                pk
-                            )
-                            yield self.hs.get_pusherpool().remove_pusher(
-                                self.app_id, pk, self.user_name
-                            )
+                for pk in rejected:
+                    if pk != self.pushkey:
+                        # for sanity, we only remove the pushkey if it
+                        # was the one we actually sent...
+                        logger.warn(
+                            ("Ignoring rejected pushkey %s because we"
+                             " didn't send it"), pk
+                        )
+                    else:
+                        logger.info(
+                            "Pushkey %s was rejected: removing",
+                            pk
+                        )
+                        yield self.hs.get_pusherpool().remove_pusher(
+                            self.app_id, pk, self.user_name
+                        )
 
-            if not self.alive:
-                continue
+        if not self.alive:
+            return
 
-            if processed:
-                self.backoff_delay = Pusher.INITIAL_BACKOFF
-                self.last_token = chunk['end']
-                self.store.update_pusher_last_token_and_success(
+        if processed:
+            self.backoff_delay = Pusher.INITIAL_BACKOFF
+            self.last_token = chunk['end']
+            self.store.update_pusher_last_token_and_success(
+                self.app_id,
+                self.pushkey,
+                self.user_name,
+                self.last_token,
+                self.clock.time_msec()
+            )
+            if self.failing_since:
+                self.failing_since = None
+                self.store.update_pusher_failing_since(
                     self.app_id,
                     self.pushkey,
                     self.user_name,
-                    self.last_token,
-                    self.clock.time_msec()
+                    self.failing_since)
+        else:
+            if not self.failing_since:
+                self.failing_since = self.clock.time_msec()
+                self.store.update_pusher_failing_since(
+                    self.app_id,
+                    self.pushkey,
+                    self.user_name,
+                    self.failing_since
+                )
+
+            if (self.failing_since and
+               self.failing_since <
+               self.clock.time_msec() - Pusher.GIVE_UP_AFTER):
+                # we really only give up so that if the URL gets
+                # fixed, we don't suddenly deliver a load
+                # of old notifications.
+                logger.warn("Giving up on a notification to user %s, "
+                            "pushkey %s",
+                            self.user_name, self.pushkey)
+                self.backoff_delay = Pusher.INITIAL_BACKOFF
+                self.last_token = chunk['end']
+                self.store.update_pusher_last_token(
+                    self.app_id,
+                    self.pushkey,
+                    self.user_name,
+                    self.last_token
+                )
+
+                self.failing_since = None
+                self.store.update_pusher_failing_since(
+                    self.app_id,
+                    self.pushkey,
+                    self.user_name,
+                    self.failing_since
                 )
-                if self.failing_since:
-                    self.failing_since = None
-                    self.store.update_pusher_failing_since(
-                        self.app_id,
-                        self.pushkey,
-                        self.user_name,
-                        self.failing_since)
             else:
-                if not self.failing_since:
-                    self.failing_since = self.clock.time_msec()
-                    self.store.update_pusher_failing_since(
-                        self.app_id,
-                        self.pushkey,
-                        self.user_name,
-                        self.failing_since
-                    )
-
-                if (self.failing_since and
-                   self.failing_since <
-                   self.clock.time_msec() - Pusher.GIVE_UP_AFTER):
-                    # we really only give up so that if the URL gets
-                    # fixed, we don't suddenly deliver a load
-                    # of old notifications.
-                    logger.warn("Giving up on a notification to user %s, "
-                                "pushkey %s",
-                                self.user_name, self.pushkey)
-                    self.backoff_delay = Pusher.INITIAL_BACKOFF
-                    self.last_token = chunk['end']
-                    self.store.update_pusher_last_token(
-                        self.app_id,
-                        self.pushkey,
-                        self.user_name,
-                        self.last_token
-                    )
-
-                    self.failing_since = None
-                    self.store.update_pusher_failing_since(
-                        self.app_id,
-                        self.pushkey,
-                        self.user_name,
-                        self.failing_since
-                    )
-                else:
-                    logger.warn("Failed to dispatch push for user %s "
-                                "(failing for %dms)."
-                                "Trying again in %dms",
-                                self.user_name,
-                                self.clock.time_msec() - self.failing_since,
-                                self.backoff_delay)
-                    yield synapse.util.async.sleep(self.backoff_delay / 1000.0)
-                    self.backoff_delay *= 2
-                    if self.backoff_delay > Pusher.MAX_BACKOFF:
-                        self.backoff_delay = Pusher.MAX_BACKOFF
+                logger.warn("Failed to dispatch push for user %s "
+                            "(failing for %dms)."
+                            "Trying again in %dms",
+                            self.user_name,
+                            self.clock.time_msec() - self.failing_since,
+                            self.backoff_delay)
+                yield synapse.util.async.sleep(self.backoff_delay / 1000.0)
+                self.backoff_delay *= 2
+                if self.backoff_delay > Pusher.MAX_BACKOFF:
+                    self.backoff_delay = Pusher.MAX_BACKOFF
 
     def stop(self):
         self.alive = False

synapse/python_dependencies.py

@@ -18,7 +18,7 @@ from distutils.version import LooseVersion
 logger = logging.getLogger(__name__)
 
 REQUIREMENTS = {
-    "syutil>=0.0.6": ["syutil>=0.0.6"],
+    "syutil>=0.0.7": ["syutil>=0.0.7"],
     "Twisted==14.0.2": ["twisted==14.0.2"],
     "service_identity>=1.0.0": ["service_identity>=1.0.0"],
     "pyopenssl>=0.14": ["OpenSSL>=0.14"],
@@ -30,6 +30,7 @@ REQUIREMENTS = {
     "frozendict>=0.4": ["frozendict"],
     "pillow": ["PIL"],
     "pydenticon": ["pydenticon"],
+    "ujson": ["ujson"],
 }
 CONDITIONAL_REQUIREMENTS = {
     "web_client": {
@@ -52,8 +53,8 @@ def github_link(project, version, egg):
 DEPENDENCY_LINKS = [
     github_link(
         project="matrix-org/syutil",
-        version="v0.0.6",
-        egg="syutil-0.0.6",
+        version="v0.0.7",
+        egg="syutil-0.0.7",
     ),
     github_link(
         project="matrix-org/matrix-angular-sdk",

synapse/rest/client/v1/__init__.py

@@ -25,7 +25,7 @@ class ClientV1RestResource(JsonResource):
     """A resource for version 1 of the matrix client API."""
 
     def __init__(self, hs):
-        JsonResource.__init__(self, hs)
+        JsonResource.__init__(self, hs, canonical_json=False)
         self.register_servlets(self, hs)
 
     @staticmethod

synapse/rest/client/v2_alpha/__init__.py

@@ -28,7 +28,7 @@ class ClientV2AlphaRestResource(JsonResource):
     """A resource for version 2 alpha of the matrix client API."""
 
     def __init__(self, hs):
-        JsonResource.__init__(self, hs)
+        JsonResource.__init__(self, hs, canonical_json=False)
         self.register_servlets(self, hs)
 
     @staticmethod

synapse/rest/media/v1/base_resource.py

@@ -15,13 +15,14 @@
 
 from .thumbnailer import Thumbnailer
 
+from synapse.http.matrixfederationclient import MatrixFederationHttpClient
 from synapse.http.server import respond_with_json
 from synapse.util.stringutils import random_string
 from synapse.api.errors import (
     cs_error, Codes, SynapseError
 )
 
-from twisted.internet import defer
+from twisted.internet import defer, threads
 from twisted.web.resource import Resource
 from twisted.protocols.basic import FileSender
 
@@ -52,7 +53,7 @@ class BaseMediaResource(Resource):
     def __init__(self, hs, filepaths):
         Resource.__init__(self)
         self.auth = hs.get_auth()
-        self.client = hs.get_http_client()
+        self.client = MatrixFederationHttpClient(hs)
         self.clock = hs.get_clock()
         self.server_name = hs.hostname
         self.store = hs.get_datastore()
@@ -273,57 +274,65 @@ class BaseMediaResource(Resource):
         if not requirements:
             return
 
+        remote_thumbnails = []
+
         input_path = self.filepaths.remote_media_filepath(server_name, file_id)
         thumbnailer = Thumbnailer(input_path)
         m_width = thumbnailer.width
         m_height = thumbnailer.height
 
-        if m_width * m_height >= self.max_image_pixels:
-            logger.info(
-                "Image too large to thumbnail %r x %r > %r",
-                m_width, m_height, self.max_image_pixels
-            )
-            return
+        def generate_thumbnails():
+            if m_width * m_height >= self.max_image_pixels:
+                logger.info(
+                    "Image too large to thumbnail %r x %r > %r",
+                    m_width, m_height, self.max_image_pixels
+                )
+                return
 
-        scales = set()
-        crops = set()
-        for r_width, r_height, r_method, r_type in requirements:
-            if r_method == "scale":
-                t_width, t_height = thumbnailer.aspect(r_width, r_height)
-                scales.add((
-                    min(m_width, t_width), min(m_height, t_height), r_type,
-                ))
-            elif r_method == "crop":
-                crops.add((r_width, r_height, r_type))
+            scales = set()
+            crops = set()
+            for r_width, r_height, r_method, r_type in requirements:
+                if r_method == "scale":
+                    t_width, t_height = thumbnailer.aspect(r_width, r_height)
+                    scales.add((
+                        min(m_width, t_width), min(m_height, t_height), r_type,
+                    ))
+                elif r_method == "crop":
+                    crops.add((r_width, r_height, r_type))
 
-        for t_width, t_height, t_type in scales:
-            t_method = "scale"
-            t_path = self.filepaths.remote_media_thumbnail(
-                server_name, file_id, t_width, t_height, t_type, t_method
-            )
-            self._makedirs(t_path)
-            t_len = thumbnailer.scale(t_path, t_width, t_height, t_type)
-            yield self.store.store_remote_media_thumbnail(
-                server_name, media_id, file_id,
-                t_width, t_height, t_type, t_method, t_len
-            )
+            for t_width, t_height, t_type in scales:
+                t_method = "scale"
+                t_path = self.filepaths.remote_media_thumbnail(
+                    server_name, file_id, t_width, t_height, t_type, t_method
+                )
+                self._makedirs(t_path)
+                t_len = thumbnailer.scale(t_path, t_width, t_height, t_type)
+                remote_thumbnails.append([
+                    server_name, media_id, file_id,
+                    t_width, t_height, t_type, t_method, t_len
+                ])
 
-        for t_width, t_height, t_type in crops:
-            if (t_width, t_height, t_type) in scales:
-                # If the aspect ratio of the cropped thumbnail matches a purely
-                # scaled one then there is no point in calculating a separate
-                # thumbnail.
-                continue
-            t_method = "crop"
-            t_path = self.filepaths.remote_media_thumbnail(
-                server_name, file_id, t_width, t_height, t_type, t_method
-            )
-            self._makedirs(t_path)
-            t_len = thumbnailer.crop(t_path, t_width, t_height, t_type)
-            yield self.store.store_remote_media_thumbnail(
-                server_name, media_id, file_id,
-                t_width, t_height, t_type, t_method, t_len
-            )
+            for t_width, t_height, t_type in crops:
+                if (t_width, t_height, t_type) in scales:
+                    # If the aspect ratio of the cropped thumbnail matches a purely
+                    # scaled one then there is no point in calculating a separate
+                    # thumbnail.
+                    continue
+                t_method = "crop"
+                t_path = self.filepaths.remote_media_thumbnail(
+                    server_name, file_id, t_width, t_height, t_type, t_method
+                )
+                self._makedirs(t_path)
+                t_len = thumbnailer.crop(t_path, t_width, t_height, t_type)
+                remote_thumbnails.append([
+                    server_name, media_id, file_id,
+                    t_width, t_height, t_type, t_method, t_len
+                ])
+
+        yield threads.deferToThread(generate_thumbnails)
+
+        for r in remote_thumbnails:
+            yield self.store.store_remote_media_thumbnail(*r)
 
         defer.returnValue({
             "width": m_width,

synapse/server.py

@@ -132,16 +132,8 @@ class BaseHomeServer(object):
         setattr(BaseHomeServer, "get_%s" % (depname), _get)
 
     def get_ip_from_request(self, request):
-        # May be an X-Forwarding-For header depending on config
-        ip_addr = request.getClientIP()
-        if self.config.captcha_ip_origin_is_x_forwarded:
-            # use the header
-            if request.requestHeaders.hasHeader("X-Forwarded-For"):
-                ip_addr = request.requestHeaders.getRawHeaders(
-                    "X-Forwarded-For"
-                )[0]
-
-        return ip_addr
+        # X-Forwarded-For is handled by our custom request type.
+        return request.getClientIP()
 
     def is_mine(self, domain_specific_string):
         return domain_specific_string.domain == self.hostname

synapse/state.py

@@ -106,7 +106,7 @@ class StateHandler(object):
         defer.returnValue(state)
 
     @defer.inlineCallbacks
-    def compute_event_context(self, event, old_state=None):
+    def compute_event_context(self, event, old_state=None, outlier=False):
         """ Fills out the context with the `current state` of the graph. The
         `current state` here is defined to be the state of the event graph
         just before the event - i.e. it never includes `event`
@@ -119,9 +119,23 @@ class StateHandler(object):
         Returns:
             an EventContext
         """
+        yield run_on_reactor()
+
         context = EventContext()
 
-        yield run_on_reactor()
+        if outlier:
+            # If this is an outlier, then we know it shouldn't have any current
+            # state. Certainly store.get_current_state won't return any, and
+            # persisting the event won't store the state group.
+            if old_state:
+                context.current_state = {
+                    (s.type, s.state_key): s for s in old_state
+                }
+            else:
+                context.current_state = {}
+            context.prev_state_events = []
+            context.state_group = None
+            defer.returnValue(context)
 
         if old_state:
             context.current_state = {
@@ -155,10 +169,6 @@ class StateHandler(object):
         context.current_state = curr_state
         context.state_group = group if not event.is_state() else None
 
-        prev_state = yield self.store.add_event_hashes(
-            prev_state
-        )
-
         if event.is_state():
             key = (event.type, event.state_key)
             if key in context.current_state:

synapse/storage/__init__.py

@@ -51,7 +51,7 @@ logger = logging.getLogger(__name__)
 
 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 19
+SCHEMA_VERSION = 20
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 
@@ -348,7 +348,7 @@ def _upgrade_existing_database(cur, current_version, applied_delta_files,
                         module_name, absolute_path, python_file
                     )
                 logger.debug("Running script %s", relative_path)
-                module.run_upgrade(cur)
+                module.run_upgrade(cur, database_engine)
             elif ext == ".sql":
                 # A plain old .sql file, just read and execute it
                 logger.debug("Applying schema %s", relative_path)

synapse/storage/_base.py

@@ -127,7 +127,7 @@ class Cache(object):
         self.cache.clear()
 
 
-def cached(max_entries=1000, num_args=1, lru=False):
+class CacheDescriptor(object):
     """ A method decorator that applies a memoizing cache around the function.
 
     The function is presumed to take zero or more arguments, which are used in
@@ -141,25 +141,32 @@ def cached(max_entries=1000, num_args=1, lru=False):
     which can be used to insert values into the cache specifically, without
     calling the calculation function.
     """
-    def wrap(orig):
+    def __init__(self, orig, max_entries=1000, num_args=1, lru=False):
+        self.orig = orig
+
+        self.max_entries = max_entries
+        self.num_args = num_args
+        self.lru = lru
+
+    def __get__(self, obj, objtype=None):
         cache = Cache(
-            name=orig.__name__,
-            max_entries=max_entries,
-            keylen=num_args,
-            lru=lru,
+            name=self.orig.__name__,
+            max_entries=self.max_entries,
+            keylen=self.num_args,
+            lru=self.lru,
         )
 
-        @functools.wraps(orig)
+        @functools.wraps(self.orig)
         @defer.inlineCallbacks
-        def wrapped(self, *keyargs):
+        def wrapped(*keyargs):
             try:
-                cached_result = cache.get(*keyargs)
+                cached_result = cache.get(*keyargs[:self.num_args])
                 if DEBUG_CACHES:
-                    actual_result = yield orig(self, *keyargs)
+                    actual_result = yield self.orig(obj, *keyargs)
                     if actual_result != cached_result:
                         logger.error(
                             "Stale cache entry %s%r: cached: %r, actual %r",
-                            orig.__name__, keyargs,
+                            self.orig.__name__, keyargs,
                             cached_result, actual_result,
                         )
                         raise ValueError("Stale cache entry")
@@ -170,18 +177,28 @@ def cached(max_entries=1000, num_args=1, lru=False):
                 # while the SELECT is executing (SYN-369)
                 sequence = cache.sequence
 
-                ret = yield orig(self, *keyargs)
+                ret = yield self.orig(obj, *keyargs)
 
-                cache.update(sequence, *keyargs + (ret,))
+                cache.update(sequence, *keyargs[:self.num_args] + (ret,))
 
                 defer.returnValue(ret)
 
         wrapped.invalidate = cache.invalidate
         wrapped.invalidate_all = cache.invalidate_all
         wrapped.prefill = cache.prefill
+
+        obj.__dict__[self.orig.__name__] = wrapped
+
         return wrapped
 
-    return wrap
+
+def cached(max_entries=1000, num_args=1, lru=False):
+    return lambda orig: CacheDescriptor(
+        orig,
+        max_entries=max_entries,
+        num_args=num_args,
+        lru=lru
+    )
 
 
 class LoggingTransaction(object):

synapse/storage/events.py

@@ -17,7 +17,7 @@ from _base import SQLBaseStore, _RollbackButIsFineException
 
 from twisted.internet import defer, reactor
 
-from synapse.events import FrozenEvent
+from synapse.events import FrozenEvent, USE_FROZEN_DICTS
 from synapse.events.utils import prune_event
 
 from synapse.util.logcontext import preserve_context_over_deferred
@@ -26,11 +26,11 @@ from synapse.api.constants import EventTypes
 from synapse.crypto.event_signing import compute_event_reference_hash
 
 from syutil.base64util import decode_base64
-from syutil.jsonutil import encode_canonical_json
+from syutil.jsonutil import encode_json
 from contextlib import contextmanager
 
 import logging
-import simplejson as json
+import ujson as json
 
 logger = logging.getLogger(__name__)
 
@@ -166,8 +166,9 @@ class EventsStore(SQLBaseStore):
             allow_none=True,
         )
 
-        metadata_json = encode_canonical_json(
-            event.internal_metadata.get_dict()
+        metadata_json = encode_json(
+            event.internal_metadata.get_dict(),
+            using_frozen_dicts=USE_FROZEN_DICTS
         ).decode("UTF-8")
 
         # If we have already persisted this event, we don't need to do any
@@ -235,12 +236,14 @@ class EventsStore(SQLBaseStore):
                 "event_id": event.event_id,
                 "room_id": event.room_id,
                 "internal_metadata": metadata_json,
-                "json": encode_canonical_json(event_dict).decode("UTF-8"),
+                "json": encode_json(
+                    event_dict, using_frozen_dicts=USE_FROZEN_DICTS
+                ).decode("UTF-8"),
             },
         )
 
-        content = encode_canonical_json(
-            event.content
+        content = encode_json(
+            event.content, using_frozen_dicts=USE_FROZEN_DICTS
         ).decode("UTF-8")
 
         vals = {
@@ -266,8 +269,8 @@ class EventsStore(SQLBaseStore):
             ]
         }
 
-        vals["unrecognized_keys"] = encode_canonical_json(
-            unrec
+        vals["unrecognized_keys"] = encode_json(
+            unrec, using_frozen_dicts=USE_FROZEN_DICTS
         ).decode("UTF-8")
 
         sql = (
@@ -733,7 +736,8 @@ class EventsStore(SQLBaseStore):
 
             because = yield self.get_event(
                 redaction_id,
-                check_redacted=False
+                check_redacted=False,
+                allow_none=True,
             )
 
             if because:
@@ -743,6 +747,7 @@ class EventsStore(SQLBaseStore):
             prev = yield self.get_event(
                 ev.unsigned["replaces_state"],
                 get_prev_content=False,
+                allow_none=True,
             )
             if prev:
                 ev.unsigned["prev_content"] = prev.get_dict()["content"]

synapse/storage/schema/delta/14/upgrade_appservice_db.py

@@ -18,7 +18,7 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-def run_upgrade(cur):
+def run_upgrade(cur, *args, **kwargs):
     cur.execute("SELECT id, regex FROM application_services_regex")
     for row in cur.fetchall():
         try:

synapse/storage/schema/delta/20/dummy.sql

@@ -0,0 +1 @@
+SELECT 1;

synapse/storage/schema/delta/20/pushers.py

@@ -0,0 +1,76 @@
+# Copyright 2015 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+"""
+Main purpose of this upgrade is to change the unique key on the
+pushers table again (it was missed when the v16 full schema was
+made) but this also changes the pushkey and data columns to text.
+When selecting a bytea column into a text column, postgres inserts
+the hex encoded data, and there's no portable way of getting the
+UTF-8 bytes, so we have to do it in Python.
+"""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def run_upgrade(cur, database_engine, *args, **kwargs):
+    logger.info("Porting pushers table...")
+    cur.execute("""
+        CREATE TABLE IF NOT EXISTS pushers2 (
+          id BIGINT PRIMARY KEY,
+          user_name TEXT NOT NULL,
+          access_token BIGINT DEFAULT NULL,
+          profile_tag VARCHAR(32) NOT NULL,
+          kind VARCHAR(8) NOT NULL,
+          app_id VARCHAR(64) NOT NULL,
+          app_display_name VARCHAR(64) NOT NULL,
+          device_display_name VARCHAR(128) NOT NULL,
+          pushkey TEXT NOT NULL,
+          ts BIGINT NOT NULL,
+          lang VARCHAR(8),
+          data TEXT,
+          last_token TEXT,
+          last_success BIGINT,
+          failing_since BIGINT,
+          UNIQUE (app_id, pushkey, user_name)
+        )
+    """)
+    cur.execute("""SELECT
+        id, user_name, access_token, profile_tag, kind,
+        app_id, app_display_name, device_display_name,
+        pushkey, ts, lang, data, last_token, last_success,
+        failing_since
+        FROM pushers
+    """)
+    count = 0
+    for row in cur.fetchall():
+        row = list(row)
+        row[8] = bytes(row[8]).decode("utf-8")
+        row[11] = bytes(row[11]).decode("utf-8")
+        cur.execute(database_engine.convert_param_style("""
+            INSERT into pushers2 (
+            id, user_name, access_token, profile_tag, kind,
+            app_id, app_display_name, device_display_name,
+            pushkey, ts, lang, data, last_token, last_success,
+            failing_since
+            ) values (%s)""" % (','.join(['?' for _ in range(len(row))]))),
+            row
+        )
+        count += 1
+    cur.execute("DROP TABLE pushers")
+    cur.execute("ALTER TABLE pushers2 RENAME TO pushers")
+    logger.info("Moved %d pushers to new table", count)

synapse/storage/state.py

@@ -81,19 +81,23 @@ class StateStore(SQLBaseStore):
             f,
         )
 
-        @defer.inlineCallbacks
-        def c(vals):
-            vals[:] = yield self._get_events(vals, get_prev_content=False)
-
-        yield defer.gatherResults(
+        state_list = yield defer.gatherResults(
             [
-                c(vals)
-                for vals in states.values()
+                self._fetch_events_for_group(group, vals)
+                for group, vals in states.items()
             ],
             consumeErrors=True,
         )
 
-        defer.returnValue(states)
+        defer.returnValue(dict(state_list))
+
+    @cached(num_args=1)
+    def _fetch_events_for_group(self, state_group, events):
+        return self._get_events(
+            events, get_prev_content=False
+        ).addCallback(
+            lambda evs: (state_group, evs)
+        )
 
     def _store_state_groups_txn(self, txn, event, context):
         if context.current_state is None:

synapse/util/jsonobject.py

@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import copy
-
 
 class JsonEncodedObject(object):
     """ A common base class for defining protocol units that are represented
@@ -76,15 +74,7 @@ class JsonEncodedObject(object):
             if k in self.valid_keys and k not in self.internal_keys
         }
         d.update(self.unrecognized_keys)
-        return copy.deepcopy(d)
-
-    def get_full_dict(self):
-        d = {
-            k: _encode(v) for (k, v) in self.__dict__.items()
-            if k in self.valid_keys or k in self.internal_keys
-        }
-        d.update(self.unrecognized_keys)
-        return copy.deepcopy(d)
+        return d
 
     def __str__(self):
         return "(%s, %s)" % (self.__class__.__name__, repr(self.__dict__))

tests/handlers/test_federation.py

@@ -100,7 +100,7 @@ class FederationTestCase(unittest.TestCase):
             return defer.succeed({})
         self.datastore.have_events.side_effect = have_events
 
-        def annotate(ev, old_state=None):
+        def annotate(ev, old_state=None, outlier=False):
             context = Mock()
             context.current_state = {}
             context.auth_events = {}
@@ -120,7 +120,7 @@ class FederationTestCase(unittest.TestCase):
         )
 
         self.state_handler.compute_event_context.assert_called_once_with(
-            ANY, old_state=None,
+            ANY, old_state=None, outlier=False
         )
 
         self.auth.check.assert_called_once_with(ANY, auth_events={})

tests/handlers/test_room.py

@@ -42,6 +42,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
                 "get_room",
                 "store_room",
                 "get_latest_events_in_room",
+                "add_event_hashes",
             ]),
             resource_for_federation=NonCallableMock(),
             http_client=NonCallableMock(spec_set=[]),
@@ -88,6 +89,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
         self.ratelimiter.send_message.return_value = (True, 0)
 
         self.datastore.persist_event.return_value = (1,1)
+        self.datastore.add_event_hashes.return_value = []
 
     @defer.inlineCallbacks
     def test_invite(self):

tests/storage/test__base.py

@@ -96,73 +96,84 @@ class CacheDecoratorTestCase(unittest.TestCase):
 
     @defer.inlineCallbacks
     def test_passthrough(self):
-        @cached()
-        def func(self, key):
-            return key
+        class A(object):
+            @cached()
+            def func(self, key):
+                return key
 
-        self.assertEquals((yield func(self, "foo")), "foo")
-        self.assertEquals((yield func(self, "bar")), "bar")
+        a = A()
+
+        self.assertEquals((yield a.func("foo")), "foo")
+        self.assertEquals((yield a.func("bar")), "bar")
 
     @defer.inlineCallbacks
     def test_hit(self):
         callcount = [0]
 
-        @cached()
-        def func(self, key):
-            callcount[0] += 1
-            return key
+        class A(object):
+            @cached()
+            def func(self, key):
+                callcount[0] += 1
+                return key
 
-        yield func(self, "foo")
+        a = A()
+        yield a.func("foo")
 
         self.assertEquals(callcount[0], 1)
 
-        self.assertEquals((yield func(self, "foo")), "foo")
+        self.assertEquals((yield a.func("foo")), "foo")
         self.assertEquals(callcount[0], 1)
 
     @defer.inlineCallbacks
     def test_invalidate(self):
         callcount = [0]
 
-        @cached()
-        def func(self, key):
-            callcount[0] += 1
-            return key
+        class A(object):
+            @cached()
+            def func(self, key):
+                callcount[0] += 1
+                return key
 
-        yield func(self, "foo")
+        a = A()
+        yield a.func("foo")
 
         self.assertEquals(callcount[0], 1)
 
-        func.invalidate("foo")
+        a.func.invalidate("foo")
 
-        yield func(self, "foo")
+        yield a.func("foo")
 
         self.assertEquals(callcount[0], 2)
 
     def test_invalidate_missing(self):
-        @cached()
-        def func(self, key):
-            return key
+        class A(object):
+            @cached()
+            def func(self, key):
+                return key
 
-        func.invalidate("what")
+        A().func.invalidate("what")
 
     @defer.inlineCallbacks
     def test_max_entries(self):
         callcount = [0]
 
-        @cached(max_entries=10)
-        def func(self, key):
-            callcount[0] += 1
-            return key
+        class A(object):
+            @cached(max_entries=10)
+            def func(self, key):
+                callcount[0] += 1
+                return key
 
-        for k in range(0,12):
-            yield func(self, k)
+        a = A()
+
+        for k in range(0, 12):
+            yield a.func(k)
 
         self.assertEquals(callcount[0], 12)
 
         # There must have been at least 2 evictions, meaning if we calculate
         # all 12 values again, we must get called at least 2 more times
         for k in range(0,12):
-            yield func(self, k)
+            yield a.func(k)
 
         self.assertTrue(callcount[0] >= 14,
             msg="Expected callcount >= 14, got %d" % (callcount[0]))
@@ -171,12 +182,15 @@ class CacheDecoratorTestCase(unittest.TestCase):
     def test_prefill(self):
         callcount = [0]
 
-        @cached()
-        def func(self, key):
-            callcount[0] += 1
-            return key
+        class A(object):
+            @cached()
+            def func(self, key):
+                callcount[0] += 1
+                return key
 
-        func.prefill("foo", 123)
+        a = A()
 
-        self.assertEquals((yield func(self, "foo")), 123)
+        a.func.prefill("foo", 123)
+
+        self.assertEquals((yield a.func("foo")), 123)
         self.assertEquals(callcount[0], 0)

tests/storage/test_registration.py

@@ -46,7 +46,7 @@ class RegistrationStoreTestCase(unittest.TestCase):
             (yield self.store.get_user_by_id(self.user_id))
         )
 
-        result = yield self.store.get_user_by_token(self.tokens[1])
+        result = yield self.store.get_user_by_token(self.tokens[0])
 
         self.assertDictContainsSubset(
             {

tests/utils.py

@@ -114,6 +114,8 @@ class MockHttpResource(HttpServer):
         mock_request.method = http_method
         mock_request.uri = path
 
+        mock_request.getClientIP.return_value = "-"
+
         mock_request.requestHeaders.getRawHeaders.return_value=[
             "X-Matrix origin=test,key=,sig="
         ]