Avoid doing presence updates on replication reconnect

Presence is supposed to be disabled on matrix.org, so we shouldn't send a load
of USER_SYNC commands every time the synchrotron reconnects to the master.

CHANGES.rst

@@ -1,89 +1,11 @@
-Changes in synapse v0.27.2 (2018-03-26)
-=======================================
+Unreleased
+==========
 
-Bug fixes:
-
-* Fix bug which broke TCP replication between workers (PR #3015)
-
-
-Changes in synapse v0.27.1 (2018-03-26)
-=======================================
-
-Meta release as v0.27.0 temporarily pointed to the wrong commit
-
-
-Changes in synapse v0.27.0 (2018-03-26)
-=======================================
-
-No changes since v0.27.0-rc2
-
-
-Changes in synapse v0.27.0-rc2 (2018-03-19)
-===========================================
-
-Pulls in v0.26.1
-
-Bug fixes:
-
-* Fix bug introduced in v0.27.0-rc1 that causes much increased memory usage in state cache (PR #3005)
-
-
-Changes in synapse v0.26.1 (2018-03-15)
-=======================================
-
-Bug fixes:
-
-* Fix bug where an invalid event caused server to stop functioning correctly,
-  due to parsing and serializing bugs in ujson library (PR #3008)
-
-
-Changes in synapse v0.27.0-rc1 (2018-03-14)
-===========================================
-
-The common case for running Synapse is not to run separate workers, but for those that do, be aware that synctl no longer starts the main synapse when using ``-a`` option with workers. A new worker file should be added with ``worker_app: synapse.app.homeserver``.
+synctl no longer starts the main synapse when using ``-a`` option with workers.
+A new worker file should be added with ``worker_app: synapse.app.homeserver``.
 
 This release also begins the process of renaming a number of the metrics
 reported to prometheus. See `docs/metrics-howto.rst <docs/metrics-howto.rst#block-and-response-metrics-renamed-for-0-27-0>`_.
-Note that the v0.28.0 release will remove the deprecated metric names.
-
-Features:
-
-* Add ability for ASes to override message send time (PR #2754)
-* Add support for custom storage providers for media repository (PR #2867, #2777, #2783, #2789, #2791, #2804, #2812, #2814, #2857, #2868, #2767)
-* Add purge API features, see `docs/admin_api/purge_history_api.rst <docs/admin_api/purge_history_api.rst>`_ for full details (PR #2858, #2867, #2882, #2946, #2962, #2943)
-* Add support for whitelisting 3PIDs that users can register. (PR #2813)
-* Add ``/room/{id}/event/{id}`` API (PR #2766)
-* Add an admin API to get all the media in a room (PR #2818) Thanks to @turt2live!
-* Add ``federation_domain_whitelist`` option (PR #2820, #2821)
-
-
-Changes:
-
-* Continue to factor out processing from main process and into worker processes. See updated `docs/workers.rst <docs/workers.rst>`_ (PR #2892 - #2904, #2913, #2920 - #2926, #2947, #2847, #2854, #2872, #2873, #2874, #2928, #2929, #2934, #2856, #2976 - #2984, #2987 - #2989, #2991 - #2993, #2995, #2784)
-* Ensure state cache is used when persisting events (PR #2864, #2871, #2802, #2835, #2836, #2841, #2842, #2849)
-* Change the default config to bind on both IPv4 and IPv6 on all platforms (PR #2435) Thanks to @silkeh!
-* No longer require a specific version of saml2 (PR #2695) Thanks to @okurz!
-* Remove ``verbosity``/``log_file`` from generated config (PR #2755)
-* Add and improve metrics and logging (PR #2770, #2778, #2785, #2786, #2787, #2793, #2794, #2795, #2809, #2810, #2833, #2834, #2844, #2965, #2927, #2975, #2790, #2796, #2838)
-* When using synctl with workers, don't start the main synapse automatically (PR #2774)
-* Minor performance improvements (PR #2773, #2792)
-* Use a connection pool for non-federation outbound connections (PR #2817)
-* Make it possible to run unit tests against postgres (PR #2829)
-* Update pynacl dependency to 1.2.1 or higher (PR #2888) Thanks to @bachp!
-* Remove ability for AS users to call /events and /sync (PR #2948)
-* Use bcrypt.checkpw (PR #2949) Thanks to @krombel!
-
-Bug fixes:
-
-* Fix broken ``ldap_config`` config option (PR #2683) Thanks to @seckrv!
-* Fix error message when user is not allowed to unban (PR #2761) Thanks to @turt2live!
-* Fix publicised groups GET API (singular) over federation (PR #2772)
-* Fix user directory when using ``user_directory_search_all_users`` config option (PR #2803, #2831)
-* Fix error on ``/publicRooms`` when no rooms exist (PR #2827)
-* Fix bug in quarantine_media (PR #2837)
-* Fix url_previews when no Content-Type is returned from URL (PR #2845)
-* Fix rare race in sync API when joining room (PR #2944)
-* Fix slow event search, switch back from GIST to GIN indexes (PR #2769, #2848)
 
 
 Changes in synapse v0.26.0 (2018-01-05)

CONTRIBUTING.rst

@@ -30,12 +30,8 @@ use github's pull request workflow to review the contribution, and either ask
 you to make any refinements needed or merge it and make them ourselves. The
 changes will then land on master when we next do a release.
 
-We use `Jenkins <http://matrix.org/jenkins>`_ and 
-`Travis <https://travis-ci.org/matrix-org/synapse>`_ for continuous
-integration. All pull requests to synapse get automatically tested by Travis; 
-the Jenkins builds require an adminstrator to start them. If your change 
-breaks the build, this will be shown in github, so please keep an eye on the 
-pull request for feedback.
+We use Jenkins for continuous integration (http://matrix.org/jenkins), and
+typically all pull requests get automatically tested Jenkins: if your change breaks the build, Jenkins will yell about it in #matrix-dev:matrix.org so please lurk there and keep an eye open.
 
 Code style
 ~~~~~~~~~~
@@ -119,4 +115,4 @@ can't be accepted. Git makes this trivial - just use the -s flag when you do
 Conclusion
 ~~~~~~~~~~
 
-That's it!  Matrix is a very open and collaborative project as you might expect given our obsession with open communication.  If we're going to successfully matrix together all the fragmented communication technologies out there we are reliant on contributions and collaboration from the community to do so.  So please get involved - and we hope you have as much fun hacking on Matrix as we do!
+That's it!  Matrix is a very open and collaborative project as you might expect given our obsession with open communication.  If we're going to successfully matrix together all the fragmented communication technologies out there we are reliant on contributions and collaboration from the community to do so.  So please get involved - and we hope you have as much fun hacking on Matrix as we do!

README.rst

@@ -354,10 +354,6 @@ https://matrix.org/docs/projects/try-matrix-now.html (or build your own with one
 Fedora
 ------
 
-Synapse is in the Fedora repositories as ``matrix-synapse``::
-
-    sudo dnf install matrix-synapse
-
 Oleg Girko provides Fedora RPMs at
 https://obs.infoserver.lv/project/monitor/matrix-synapse
 
@@ -894,17 +890,6 @@ This should end with a 'PASSED' result::
 
     PASSED (successes=143)
 
-Running the Integration Tests
-=============================
-
-Synapse is accompanied by `SyTest <https://github.com/matrix-org/sytest>`_,
-a Matrix homeserver integration testing suite, which uses HTTP requests to
-access the API as a Matrix client would. It is able to run Synapse directly from
-the source tree, so installation of the server is not required.
-
-Testing with SyTest is recommended for verifying that changes related to the
-Client-Server API are functioning correctly. See the `installation instructions
-<https://github.com/matrix-org/sytest#installing>`_ for details.
 
 Building Internal API Documentation
 ===================================

UPGRADE.rst

@@ -48,18 +48,6 @@ returned by the Client-Server API:
     # configured on port 443.
     curl -kv https://<host.name>/_matrix/client/versions 2>&1 | grep "Server:"
 
-Upgrading to $NEXT_VERSION
-====================
-
-This release expands the anonymous usage stats sent if the opt-in
-``report_stats`` configuration is set to ``true``. We now capture RSS memory 
-and cpu use at a very coarse level. This requires administrators to install
-the optional ``psutil`` python module.
-
-We would appreciate it if you could assist by ensuring this module is available
-and ``report_stats`` is enabled. This will let us see if performance changes to
-synapse are having an impact to the general community.
-
 Upgrading to v0.15.0
 ====================
 

docs/admin_api/purge_history_api.rst

@@ -8,56 +8,20 @@ Depending on the amount of history being purged a call to the API may take
 several minutes or longer. During this period users will not be able to
 paginate further back in the room from the point being purged from.
 
-The API is:
+The API is simply:
 
-``POST /_matrix/client/r0/admin/purge_history/<room_id>[/<event_id>]``
+``POST /_matrix/client/r0/admin/purge_history/<room_id>/<event_id>``
 
 including an ``access_token`` of a server admin.
 
 By default, events sent by local users are not deleted, as they may represent
 the only copies of this content in existence. (Events sent by remote users are
-deleted.)
+deleted, and room state data before the cutoff is always removed).
 
-Room state data (such as joins, leaves, topic) is always preserved.
-
-To delete local message events as well, set ``delete_local_events`` in the body:
+To delete local events as well, set ``delete_local_events`` in the body:
 
 .. code:: json
 
    {
        "delete_local_events": true
    }
-
-The caller must specify the point in the room to purge up to. This can be
-specified by including an event_id in the URI, or by setting a
-``purge_up_to_event_id`` or ``purge_up_to_ts`` in the request body. If an event
-id is given, that event (and others at the same graph depth) will be retained.
-If ``purge_up_to_ts`` is given, it should be a timestamp since the unix epoch,
-in milliseconds.
-
-The API starts the purge running, and returns immediately with a JSON body with
-a purge id:
-
-.. code:: json
-
-    {
-        "purge_id": "<opaque id>"
-    }
-
-Purge status query
-------------------
-
-It is possible to poll for updates on recent purges with a second API;
-
-``GET /_matrix/client/r0/admin/purge_history_status/<purge_id>``
-
-(again, with a suitable ``access_token``). This API returns a JSON body like
-the following:
-
-.. code:: json
-
-    {
-        "status": "active"
-    }
-
-The status will be one of ``active``, ``complete``, or ``failed``.

docs/log_contexts.rst

@@ -279,9 +279,9 @@ Obviously that option means that the operations done in
 that might be fixed by setting a different logcontext via a ``with
 LoggingContext(...)`` in ``background_operation``).
 
-The second option is to use ``logcontext.run_in_background``, which wraps a
-function so that it doesn't reset the logcontext even when it returns an
-incomplete deferred, and adds a callback to the returned deferred to reset the
+The second option is to use ``logcontext.preserve_fn``, which wraps a function
+so that it doesn't reset the logcontext even when it returns an incomplete
+deferred, and adds a callback to the returned deferred to reset the
 logcontext. In other words, it turns a function that follows the Synapse rules
 about logcontexts and Deferreds into one which behaves more like an external
 function — the opposite operation to that described in the previous section.
@@ -293,7 +293,7 @@ It can be used like this:
     def do_request_handling():
         yield foreground_operation()
 
-        logcontext.run_in_background(background_operation)
+        logcontext.preserve_fn(background_operation)()
 
         # this will now be logged against the request context
         logger.debug("Request handling complete")

docs/workers.rst

@@ -55,12 +55,7 @@ synapse process.)
 
 You then create a set of configs for the various worker processes.  These
 should be worker configuration files, and should be stored in a dedicated
-subdirectory, to allow synctl to manipulate them. An additional configuration
-for the master synapse process will need to be created because the process will
-not be started automatically. That configuration should look like this::
-
-    worker_app: synapse.app.homeserver
-    daemonize: true
+subdirectory, to allow synctl to manipulate them.
 
 Each worker configuration file inherits the configuration of the main homeserver
 configuration file.  You can then override configuration specific to that worker,
@@ -235,11 +230,9 @@ file. For example::
 ``synapse.app.event_creator``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-Handles some event creation. It can handle REST endpoints matching::
+Handles non-state event creation. It can handle REST endpoints matching:
 
     ^/_matrix/client/(api/v1|r0|unstable)/rooms/.*/send
-    ^/_matrix/client/(api/v1|r0|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$
-    ^/_matrix/client/(api/v1|r0|unstable)/join/
 
 It will create events locally and then send them on to the main synapse
 instance to be persisted and handled.

scripts/synapse_port_db

@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 # Copyright 2015, 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -251,12 +250,6 @@ class Porter(object):
     @defer.inlineCallbacks
     def handle_table(self, table, postgres_size, table_size, forward_chunk,
                      backward_chunk):
-        logger.info(
-            "Table %s: %i/%i (rows %i-%i) already ported",
-            table, postgres_size, table_size,
-            backward_chunk+1, forward_chunk-1,
-        )
-
         if not table_size:
             return
 
@@ -474,10 +467,31 @@ class Porter(object):
             self.progress.set_state("Preparing PostgreSQL")
             self.setup_db(postgres_config, postgres_engine)
 
-            self.progress.set_state("Creating port tables")
+            # Step 2. Get tables.
+            self.progress.set_state("Fetching tables")
+            sqlite_tables = yield self.sqlite_store._simple_select_onecol(
+                table="sqlite_master",
+                keyvalues={
+                    "type": "table",
+                },
+                retcol="name",
+            )
+
+            postgres_tables = yield self.postgres_store._simple_select_onecol(
+                table="information_schema.tables",
+                keyvalues={},
+                retcol="distinct table_name",
+            )
+
+            tables = set(sqlite_tables) & set(postgres_tables)
+
+            self.progress.set_state("Creating tables")
+
+            logger.info("Found %d tables", len(tables))
+
             def create_port_table(txn):
                 txn.execute(
-                    "CREATE TABLE IF NOT EXISTS port_from_sqlite3 ("
+                    "CREATE TABLE port_from_sqlite3 ("
                     " table_name varchar(100) NOT NULL UNIQUE,"
                     " forward_rowid bigint NOT NULL,"
                     " backward_rowid bigint NOT NULL"
@@ -503,33 +517,18 @@ class Porter(object):
                     "alter_table", alter_table
                 )
             except Exception as e:
-                pass
+                logger.info("Failed to create port table: %s", e)
 
-            yield self.postgres_store.runInteraction(
-                "create_port_table", create_port_table
-            )
+            try:
+                yield self.postgres_store.runInteraction(
+                    "create_port_table", create_port_table
+                )
+            except Exception as e:
+                logger.info("Failed to create port table: %s", e)
 
-            # Step 2. Get tables.
-            self.progress.set_state("Fetching tables")
-            sqlite_tables = yield self.sqlite_store._simple_select_onecol(
-                table="sqlite_master",
-                keyvalues={
-                    "type": "table",
-                },
-                retcol="name",
-            )
+            self.progress.set_state("Setting up")
 
-            postgres_tables = yield self.postgres_store._simple_select_onecol(
-                table="information_schema.tables",
-                keyvalues={},
-                retcol="distinct table_name",
-            )
-
-            tables = set(sqlite_tables) & set(postgres_tables)
-            logger.info("Found %d tables", len(tables))
-
-            # Step 3. Figure out what still needs copying
-            self.progress.set_state("Checking on port progress")
+            # Set up tables.
             setup_res = yield defer.gatherResults(
                 [
                     self.setup_table(table)
@@ -540,8 +539,7 @@ class Porter(object):
                 consumeErrors=True,
             )
 
-            # Step 4. Do the copying.
-            self.progress.set_state("Copying to postgres")
+            # Process tables.
             yield defer.gatherResults(
                 [
                     self.handle_table(*res)
@@ -550,9 +548,6 @@ class Porter(object):
                 consumeErrors=True,
             )
 
-            # Step 5. Do final post-processing
-            yield self._setup_state_group_id_seq()
-
             self.progress.done()
         except:
             global end_error_exec_info
@@ -712,16 +707,6 @@ class Porter(object):
 
         defer.returnValue((done, remaining + done))
 
-    def _setup_state_group_id_seq(self):
-        def r(txn):
-            txn.execute("SELECT MAX(id) FROM state_groups")
-            next_id = txn.fetchone()[0]+1
-            txn.execute(
-                "ALTER SEQUENCE state_group_id_seq RESTART WITH %s",
-                (next_id,),
-            )
-        return self.postgres_store.runInteraction("setup_state_group_id_seq", r)
-
 
 ##############################################
 ###### The following is simply UI stuff ######

synapse/__init__.py

@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """
 
-__version__ = "0.27.2"
+__version__ = "0.26.0"

synapse/api/errors.py

@@ -15,10 +15,9 @@
 
 """Contains exceptions and error codes."""
 
+import json
 import logging
 
-import simplejson as json
-
 logger = logging.getLogger(__name__)
 
 

synapse/api/filtering.py

@@ -17,7 +17,7 @@ from synapse.storage.presence import UserPresenceState
 from synapse.types import UserID, RoomID
 from twisted.internet import defer
 
-import simplejson as json
+import ujson as json
 import jsonschema
 from jsonschema import FormatChecker
 

synapse/app/appservice.py

@@ -36,7 +36,7 @@ from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.appservice")
 
@@ -64,7 +64,7 @@ class AppserviceServer(HomeServer):
                 if name == "metrics":
                     resources[METRICS_PREFIX] = MetricsResource(self)
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,

synapse/app/client_reader.py

@@ -44,7 +44,7 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.client_reader")
 
@@ -88,7 +88,7 @@ class ClientReaderServer(HomeServer):
                         "/_matrix/client/api/v1": resource,
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,
@@ -156,6 +156,7 @@ def start(config_options):
     )
 
     ss.setup()
+    ss.get_handlers()
     ss.start_listening(config.worker_listeners)
 
     def start():

synapse/app/event_creator.py

@@ -27,24 +27,14 @@ from synapse.http.server import JsonResource
 from synapse.http.site import SynapseSite
 from synapse.metrics.resource import METRICS_PREFIX, MetricsResource
 from synapse.replication.slave.storage._base import BaseSlavedStore
-from synapse.replication.slave.storage.account_data import SlavedAccountDataStore
 from synapse.replication.slave.storage.appservice import SlavedApplicationServiceStore
 from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
 from synapse.replication.slave.storage.devices import SlavedDeviceStore
-from synapse.replication.slave.storage.directory import DirectoryStore
 from synapse.replication.slave.storage.events import SlavedEventStore
-from synapse.replication.slave.storage.profile import SlavedProfileStore
-from synapse.replication.slave.storage.push_rule import SlavedPushRuleStore
-from synapse.replication.slave.storage.pushers import SlavedPusherStore
-from synapse.replication.slave.storage.receipts import SlavedReceiptsStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.slave.storage.room import RoomStore
-from synapse.replication.slave.storage.transactions import TransactionStore
 from synapse.replication.tcp.client import ReplicationClientHandler
-from synapse.rest.client.v1.room import (
-    RoomSendEventRestServlet, RoomMembershipRestServlet, RoomStateEventRestServlet,
-    JoinRoomAliasServlet,
-)
+from synapse.rest.client.v1.room import RoomSendEventRestServlet
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
 from synapse.util.httpresourcetree import create_resource_tree
@@ -52,19 +42,12 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.event_creator")
 
 
 class EventCreatorSlavedStore(
-    DirectoryStore,
-    TransactionStore,
-    SlavedProfileStore,
-    SlavedAccountDataStore,
-    SlavedPusherStore,
-    SlavedReceiptsStore,
-    SlavedPushRuleStore,
     SlavedDeviceStore,
     SlavedClientIpStore,
     SlavedApplicationServiceStore,
@@ -94,9 +77,6 @@ class EventCreatorServer(HomeServer):
                 elif name == "client":
                     resource = JsonResource(self, canonical_json=False)
                     RoomSendEventRestServlet(self).register(resource)
-                    RoomMembershipRestServlet(self).register(resource)
-                    RoomStateEventRestServlet(self).register(resource)
-                    JoinRoomAliasServlet(self).register(resource)
                     resources.update({
                         "/_matrix/client/r0": resource,
                         "/_matrix/client/unstable": resource,
@@ -104,7 +84,7 @@ class EventCreatorServer(HomeServer):
                         "/_matrix/client/api/v1": resource,
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,
@@ -173,6 +153,7 @@ def start(config_options):
     )
 
     ss.setup()
+    ss.get_handlers()
     ss.start_listening(config.worker_listeners)
 
     def start():

synapse/app/federation_reader.py

@@ -41,7 +41,7 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.federation_reader")
 
@@ -77,7 +77,7 @@ class FederationReaderServer(HomeServer):
                         FEDERATION_PREFIX: TransportLayerServer(self),
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,
@@ -144,6 +144,7 @@ def start(config_options):
     )
 
     ss.setup()
+    ss.get_handlers()
     ss.start_listening(config.worker_listeners)
 
     def start():

synapse/app/federation_sender.py

@@ -42,7 +42,7 @@ from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import defer, reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.federation_sender")
 
@@ -91,7 +91,7 @@ class FederationSenderServer(HomeServer):
                 if name == "metrics":
                     resources[METRICS_PREFIX] = MetricsResource(self)
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,

synapse/app/frontend_proxy.py

@@ -36,6 +36,7 @@ from synapse.replication.slave.storage.client_ips import SlavedClientIpStore
 from synapse.replication.slave.storage.devices import SlavedDeviceStore
 from synapse.replication.slave.storage.registration import SlavedRegistrationStore
 from synapse.replication.tcp.client import ReplicationClientHandler
+from synapse.rest.client.v1.base import ClientV1RestServlet, client_path_patterns
 from synapse.rest.client.v2_alpha._base import client_v2_patterns
 from synapse.server import HomeServer
 from synapse.storage.engines import create_engine
@@ -44,11 +45,40 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import defer, reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.frontend_proxy")
 
 
+class PresenceStatusStubServlet(ClientV1RestServlet):
+    PATTERNS = client_path_patterns("/presence/(?P<user_id>[^/]*)/status")
+
+    def __init__(self, hs):
+        super(PresenceStatusStubServlet, self).__init__(hs)
+        self.http_client = hs.get_simple_http_client()
+        self.auth = hs.get_auth()
+        self.main_uri = hs.config.worker_main_http_uri
+
+    @defer.inlineCallbacks
+    def on_GET(self, request, user_id):
+        # Pass through the auth headers, if any, in case the access token
+        # is there.
+        auth_headers = request.requestHeaders.getRawHeaders("Authorization", [])
+        headers = {
+            "Authorization": auth_headers,
+        }
+        result = yield self.http_client.get_json(
+            self.main_uri + request.uri,
+            headers=headers,
+        )
+        defer.returnValue((200, result))
+
+    @defer.inlineCallbacks
+    def on_PUT(self, request, user_id):
+        yield self.auth.get_user_by_req(request)
+        defer.returnValue((200, {}))
+
+
 class KeyUploadServlet(RestServlet):
     PATTERNS = client_v2_patterns("/keys/upload(/(?P<device_id>[^/]+))?$")
 
@@ -135,6 +165,7 @@ class FrontendProxyServer(HomeServer):
                 elif name == "client":
                     resource = JsonResource(self, canonical_json=False)
                     KeyUploadServlet(self).register(resource)
+                    PresenceStatusStubServlet(self).register(resource)
                     resources.update({
                         "/_matrix/client/r0": resource,
                         "/_matrix/client/unstable": resource,
@@ -142,7 +173,7 @@ class FrontendProxyServer(HomeServer):
                         "/_matrix/client/api/v1": resource,
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,
@@ -211,6 +242,7 @@ def start(config_options):
     )
 
     ss.setup()
+    ss.get_handlers()
     ss.start_listening(config.worker_listeners)
 
     def start():

synapse/app/homeserver.py

@@ -48,7 +48,6 @@ from synapse.server import HomeServer
 from synapse.storage import are_all_users_on_domain
 from synapse.storage.engines import IncorrectDatabaseSetup, create_engine
 from synapse.storage.prepare_database import UpgradeDatabaseException, prepare_database
-from synapse.util.caches import CACHE_SIZE_FACTOR
 from synapse.util.httpresourcetree import create_resource_tree
 from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
@@ -57,7 +56,7 @@ from synapse.util.rlimit import change_resource_limit
 from synapse.util.versionstring import get_version_string
 from twisted.application import service
 from twisted.internet import defer, reactor
-from twisted.web.resource import EncodingResourceWrapper, NoResource
+from twisted.web.resource import EncodingResourceWrapper, Resource
 from twisted.web.server import GzipEncoderFactory
 from twisted.web.static import File
 
@@ -127,7 +126,7 @@ class SynapseHomeServer(HomeServer):
         if WEB_CLIENT_PREFIX in resources:
             root_resource = RootRedirect(WEB_CLIENT_PREFIX)
         else:
-            root_resource = NoResource()
+            root_resource = Resource()
 
         root_resource = create_resource_tree(resources, root_resource)
 
@@ -349,7 +348,7 @@ def setup(config_options):
         hs.get_state_handler().start_caching()
         hs.get_datastore().start_profiling()
         hs.get_datastore().start_doing_background_updates()
-        hs.get_federation_client().start_get_pdu_cache()
+        hs.get_replication_layer().start_get_pdu_cache()
 
         register_memory_metrics(hs)
 
@@ -403,10 +402,6 @@ def run(hs):
 
     stats = {}
 
-    # Contains the list of processes we will be monitoring
-    # currently either 0 or 1
-    stats_process = []
-
     @defer.inlineCallbacks
     def phone_stats_home():
         logger.info("Gathering stats for reporting")
@@ -430,21 +425,8 @@ def run(hs):
         stats["daily_active_rooms"] = yield hs.get_datastore().count_daily_active_rooms()
         stats["daily_messages"] = yield hs.get_datastore().count_daily_messages()
 
-        r30_results = yield hs.get_datastore().count_r30_users()
-        for name, count in r30_results.iteritems():
-            stats["r30_users_" + name] = count
-
         daily_sent_messages = yield hs.get_datastore().count_daily_sent_messages()
         stats["daily_sent_messages"] = daily_sent_messages
-        stats["cache_factor"] = CACHE_SIZE_FACTOR
-        stats["event_cache_size"] = hs.config.event_cache_size
-
-        if len(stats_process) > 0:
-            stats["memory_rss"] = 0
-            stats["cpu_average"] = 0
-            for process in stats_process:
-                stats["memory_rss"] += process.memory_info().rss
-                stats["cpu_average"] += int(process.cpu_percent(interval=None))
 
         logger.info("Reporting stats to matrix.org: %s" % (stats,))
         try:
@@ -455,32 +437,10 @@ def run(hs):
         except Exception as e:
             logger.warn("Error reporting stats: %s", e)
 
-    def performance_stats_init():
-        try:
-            import psutil
-            process = psutil.Process()
-            # Ensure we can fetch both, and make the initial request for cpu_percent
-            # so the next request will use this as the initial point.
-            process.memory_info().rss
-            process.cpu_percent(interval=None)
-            logger.info("report_stats can use psutil")
-            stats_process.append(process)
-        except (ImportError, AttributeError):
-            logger.warn(
-                "report_stats enabled but psutil is not installed or incorrect version."
-                " Disabling reporting of memory/cpu stats."
-                " Ensuring psutil is available will help matrix.org track performance"
-                " changes across releases."
-            )
-
     if hs.config.report_stats:
         logger.info("Scheduling stats reporting for 3 hour intervals")
         clock.looping_call(phone_stats_home, 3 * 60 * 60 * 1000)
 
-        # We need to defer this init for the cases that we daemonize
-        # otherwise the process ID we get is that of the non-daemon process
-        clock.call_later(0, performance_stats_init)
-
         # We wait 5 minutes to send the first set of stats as the server can
         # be quite busy the first few minutes
         clock.call_later(5 * 60, phone_stats_home)

synapse/app/media_repository.py

@@ -43,7 +43,7 @@ from synapse.util.logcontext import LoggingContext
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.media_repository")
 
@@ -84,7 +84,7 @@ class MediaRepositoryServer(HomeServer):
                         ),
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,
@@ -158,6 +158,7 @@ def start(config_options):
     )
 
     ss.setup()
+    ss.get_handlers()
     ss.start_listening(config.worker_listeners)
 
     def start():

synapse/app/pusher.py

@@ -32,12 +32,13 @@ from synapse.replication.tcp.client import ReplicationClientHandler
 from synapse.server import HomeServer
 from synapse.storage import DataStore
 from synapse.storage.engines import create_engine
+from synapse.storage.roommember import RoomMemberStore
 from synapse.util.httpresourcetree import create_resource_tree
 from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import defer, reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.pusher")
 
@@ -74,6 +75,10 @@ class PusherSlaveStore(
         DataStore.get_profile_displayname.__func__
     )
 
+    who_forgot_in_room = (
+        RoomMemberStore.__dict__["who_forgot_in_room"]
+    )
+
 
 class PusherServer(HomeServer):
     def setup(self):
@@ -94,7 +99,7 @@ class PusherServer(HomeServer):
                 if name == "metrics":
                     resources[METRICS_PREFIX] = MetricsResource(self)
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,

synapse/app/synchrotron.py

@@ -56,12 +56,14 @@ from synapse.util.manhole import manhole
 from synapse.util.stringutils import random_string
 from synapse.util.versionstring import get_version_string
 from twisted.internet import defer, reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.synchrotron")
 
 
 class SynchrotronSlavedStore(
+    SlavedPushRuleStore,
+    SlavedEventStore,
     SlavedReceiptsStore,
     SlavedAccountDataStore,
     SlavedApplicationServiceStore,
@@ -71,12 +73,14 @@ class SynchrotronSlavedStore(
     SlavedGroupServerStore,
     SlavedDeviceInboxStore,
     SlavedDeviceStore,
-    SlavedPushRuleStore,
-    SlavedEventStore,
     SlavedClientIpStore,
     RoomStore,
     BaseSlavedStore,
 ):
+    who_forgot_in_room = (
+        RoomMemberStore.__dict__["who_forgot_in_room"]
+    )
+
     did_forget = (
         RoomMemberStore.__dict__["did_forget"]
     )
@@ -113,6 +117,7 @@ class SynchrotronPresence(object):
         logger.info("Presence process_id is %r", self.process_id)
 
     def send_user_sync(self, user_id, is_syncing, last_sync_ms):
+        return
         self.hs.get_tcp_replication().send_user_sync(user_id, is_syncing, last_sync_ms)
 
     def mark_as_coming_online(self, user_id):
@@ -210,6 +215,8 @@ class SynchrotronPresence(object):
         yield self.notify_from_replication(states, stream_id)
 
     def get_currently_syncing_users(self):
+        # presence is disabled on matrix.org, so we return the empty set
+        return set()
         return [
             user_id for user_id, count in self.user_to_num_current_syncs.iteritems()
             if count > 0
@@ -269,7 +276,7 @@ class SynchrotronServer(HomeServer):
                         "/_matrix/client/api/v1": resource,
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,

synapse/app/synctl.py

@@ -38,7 +38,7 @@ def pid_running(pid):
     try:
         os.kill(pid, 0)
         return True
-    except OSError as err:
+    except OSError, err:
         if err.errno == errno.EPERM:
             return True
         return False
@@ -98,7 +98,7 @@ def stop(pidfile, app):
         try:
             os.kill(pid, signal.SIGTERM)
             write("stopped %s" % (app,), colour=GREEN)
-        except OSError as err:
+        except OSError, err:
             if err.errno == errno.ESRCH:
                 write("%s not running" % (app,), colour=YELLOW)
             elif err.errno == errno.EPERM:
@@ -252,7 +252,6 @@ def main():
             for running_pid in running_pids:
                 while pid_running(running_pid):
                     time.sleep(0.2)
-            write("All processes exited; now restarting...")
 
     if action == "start" or action == "restart":
         if start_stop_synapse:

synapse/app/user_dir.py

@@ -43,7 +43,7 @@ from synapse.util.logcontext import LoggingContext, preserve_fn
 from synapse.util.manhole import manhole
 from synapse.util.versionstring import get_version_string
 from twisted.internet import reactor
-from twisted.web.resource import NoResource
+from twisted.web.resource import Resource
 
 logger = logging.getLogger("synapse.app.user_dir")
 
@@ -116,7 +116,7 @@ class UserDirectoryServer(HomeServer):
                         "/_matrix/client/api/v1": resource,
                     })
 
-        root_resource = create_resource_tree(resources, NoResource())
+        root_resource = create_resource_tree(resources, Resource())
 
         _base.listen_tcp(
             bind_addresses,

synapse/config/registration.py

@@ -77,9 +77,7 @@ class RegistrationConfig(Config):
 
         # Set the number of bcrypt rounds used to generate password hash.
         # Larger numbers increase the work factor needed to generate the hash.
-        # The default number is 12 (which equates to 2^12 rounds).
-        # N.B. that increasing this will exponentially increase the time required
-        # to register or login - e.g. 24 => 2^24 rounds which will take >20 mins.
+        # The default number of rounds is 12.
         bcrypt_rounds: 12
 
         # Allows users to register as guests without a password/email/etc, and

synapse/events/snapshot.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from twisted.internet import defer
 
 from frozendict import frozendict
 
@@ -52,6 +51,7 @@ class EventContext(object):
         "prev_state_ids",
         "state_group",
         "rejected",
+        "push_actions",
         "prev_group",
         "delta_ids",
         "prev_state_events",
@@ -66,6 +66,7 @@ class EventContext(object):
         self.state_group = None
 
         self.rejected = False
+        self.push_actions = []
 
         # A previously persisted state group and a delta between that
         # and this state.
@@ -76,32 +77,19 @@ class EventContext(object):
 
         self.app_service = None
 
-    def serialize(self, event):
+    def serialize(self):
         """Converts self to a type that can be serialized as JSON, and then
         deserialized by `deserialize`
 
-        Args:
-            event (FrozenEvent): The event that this context relates to
-
         Returns:
             dict
         """
-
-        # We don't serialize the full state dicts, instead they get pulled out
-        # of the DB on the other side. However, the other side can't figure out
-        # the prev_state_ids, so if we're a state event we include the event
-        # id that we replaced in the state.
-        if event.is_state():
-            prev_state_id = self.prev_state_ids.get((event.type, event.state_key))
-        else:
-            prev_state_id = None
-
         return {
-            "prev_state_id": prev_state_id,
-            "event_type": event.type,
-            "event_state_key": event.state_key if event.is_state() else None,
+            "current_state_ids": _encode_state_dict(self.current_state_ids),
+            "prev_state_ids": _encode_state_dict(self.prev_state_ids),
             "state_group": self.state_group,
             "rejected": self.rejected,
+            "push_actions": self.push_actions,
             "prev_group": self.prev_group,
             "delta_ids": _encode_state_dict(self.delta_ids),
             "prev_state_events": self.prev_state_events,
@@ -109,7 +97,6 @@ class EventContext(object):
         }
 
     @staticmethod
-    @defer.inlineCallbacks
     def deserialize(store, input):
         """Converts a dict that was produced by `serialize` back into a
         EventContext.
@@ -122,32 +109,20 @@ class EventContext(object):
             EventContext
         """
         context = EventContext()
+        context.current_state_ids = _decode_state_dict(input["current_state_ids"])
+        context.prev_state_ids = _decode_state_dict(input["prev_state_ids"])
         context.state_group = input["state_group"]
         context.rejected = input["rejected"]
+        context.push_actions = input["push_actions"]
         context.prev_group = input["prev_group"]
         context.delta_ids = _decode_state_dict(input["delta_ids"])
         context.prev_state_events = input["prev_state_events"]
 
-        # We use the state_group and prev_state_id stuff to pull the
-        # current_state_ids out of the DB and construct prev_state_ids.
-        prev_state_id = input["prev_state_id"]
-        event_type = input["event_type"]
-        event_state_key = input["event_state_key"]
-
-        context.current_state_ids = yield store.get_state_ids_for_group(
-            context.state_group,
-        )
-        if prev_state_id and event_state_key:
-            context.prev_state_ids = dict(context.current_state_ids)
-            context.prev_state_ids[(event_type, event_state_key)] = prev_state_id
-        else:
-            context.prev_state_ids = context.current_state_ids
-
         app_service_id = input["app_service_id"]
         if app_service_id:
             context.app_service = store.get_app_service_by_id(app_service_id)
 
-        defer.returnValue(context)
+        return context
 
 
 def _encode_state_dict(state_dict):

synapse/federation/__init__.py

@@ -15,3 +15,11 @@
 
 """ This package includes all the federation specific logic.
 """
+
+from .replication import ReplicationLayer
+
+
+def initialize_http_replication(hs):
+    transport = hs.get_federation_transport_client()
+
+    return ReplicationLayer(hs, transport)

synapse/federation/federation_base.py

@@ -27,13 +27,7 @@ logger = logging.getLogger(__name__)
 
 class FederationBase(object):
     def __init__(self, hs):
-        self.hs = hs
-
-        self.server_name = hs.hostname
-        self.keyring = hs.get_keyring()
         self.spam_checker = hs.get_spam_checker()
-        self.store = hs.get_datastore()
-        self._clock = hs.get_clock()
 
     @defer.inlineCallbacks
     def _check_sigs_and_hash_and_fetch(self, origin, pdus, outlier=False,

synapse/federation/federation_client.py

@@ -58,7 +58,6 @@ class FederationClient(FederationBase):
             self._clear_tried_cache, 60 * 1000,
         )
         self.state = hs.get_state_handler()
-        self.transport_layer = hs.get_federation_transport_client()
 
     def _clear_tried_cache(self):
         """Clear pdu_destination_tried cache"""

synapse/federation/federation_server.py

@@ -17,14 +17,12 @@ import logging
 import simplejson as json
 from twisted.internet import defer
 
-from synapse.api.errors import AuthError, FederationError, SynapseError, NotFoundError
+from synapse.api.errors import AuthError, FederationError, SynapseError
 from synapse.crypto.event_signing import compute_event_signature
 from synapse.federation.federation_base import (
     FederationBase,
     event_from_pdu_json,
 )
-
-from synapse.federation.persistence import TransactionActions
 from synapse.federation.units import Edu, Transaction
 import synapse.metrics
 from synapse.types import get_domain_from_id
@@ -54,19 +52,50 @@ class FederationServer(FederationBase):
         super(FederationServer, self).__init__(hs)
 
         self.auth = hs.get_auth()
-        self.handler = hs.get_handlers().federation_handler
 
         self._server_linearizer = async.Linearizer("fed_server")
         self._transaction_linearizer = async.Linearizer("fed_txn_handler")
 
-        self.transaction_actions = TransactionActions(self.store)
-
-        self.registry = hs.get_federation_registry()
-
         # We cache responses to state queries, as they take a while and often
         # come in waves.
         self._state_resp_cache = ResponseCache(hs, timeout_ms=30000)
 
+    def set_handler(self, handler):
+        """Sets the handler that the replication layer will use to communicate
+        receipt of new PDUs from other home servers. The required methods are
+        documented on :py:class:`.ReplicationHandler`.
+        """
+        self.handler = handler
+
+    def register_edu_handler(self, edu_type, handler):
+        if edu_type in self.edu_handlers:
+            raise KeyError("Already have an EDU handler for %s" % (edu_type,))
+
+        self.edu_handlers[edu_type] = handler
+
+    def register_query_handler(self, query_type, handler):
+        """Sets the handler callable that will be used to handle an incoming
+        federation Query of the given type.
+
+        Args:
+            query_type (str): Category name of the query, which should match
+                the string used by make_query.
+            handler (callable): Invoked to handle incoming queries of this type
+
+        handler is invoked as:
+            result = handler(args)
+
+        where 'args' is a dict mapping strings to strings of the query
+          arguments. It should return a Deferred that will eventually yield an
+          object to encode as JSON.
+        """
+        if query_type in self.query_handlers:
+            raise KeyError(
+                "Already have a Query handler for %s" % (query_type,)
+            )
+
+        self.query_handlers[query_type] = handler
+
     @defer.inlineCallbacks
     @log_function
     def on_backfill_request(self, origin, room_id, versions, limit):
@@ -200,7 +229,16 @@ class FederationServer(FederationBase):
     @defer.inlineCallbacks
     def received_edu(self, origin, edu_type, content):
         received_edus_counter.inc()
-        yield self.registry.on_edu(edu_type, origin, content)
+
+        if edu_type in self.edu_handlers:
+            try:
+                yield self.edu_handlers[edu_type](origin, content)
+            except SynapseError as e:
+                logger.info("Failed to handle edu %r: %r", edu_type, e)
+            except Exception as e:
+                logger.exception("Failed to handle edu %r", edu_type)
+        else:
+            logger.warn("Received EDU of type %s with no handler", edu_type)
 
     @defer.inlineCallbacks
     @log_function
@@ -290,8 +328,14 @@ class FederationServer(FederationBase):
     @defer.inlineCallbacks
     def on_query_request(self, query_type, args):
         received_queries_counter.inc(query_type)
-        resp = yield self.registry.on_query(query_type, args)
-        defer.returnValue((200, resp))
+
+        if query_type in self.query_handlers:
+            response = yield self.query_handlers[query_type](args)
+            defer.returnValue((200, response))
+        else:
+            defer.returnValue(
+                (404, "No handler for Query type '%s'" % (query_type,))
+            )
 
     @defer.inlineCallbacks
     def on_make_join_request(self, room_id, user_id):
@@ -563,66 +607,3 @@ class FederationServer(FederationBase):
             origin, room_id, event_dict
         )
         defer.returnValue(ret)
-
-
-class FederationHandlerRegistry(object):
-    """Allows classes to register themselves as handlers for a given EDU or
-    query type for incoming federation traffic.
-    """
-    def __init__(self):
-        self.edu_handlers = {}
-        self.query_handlers = {}
-
-    def register_edu_handler(self, edu_type, handler):
-        """Sets the handler callable that will be used to handle an incoming
-        federation EDU of the given type.
-
-        Args:
-            edu_type (str): The type of the incoming EDU to register handler for
-            handler (Callable[[str, dict]]): A callable invoked on incoming EDU
-                of the given type. The arguments are the origin server name and
-                the EDU contents.
-        """
-        if edu_type in self.edu_handlers:
-            raise KeyError("Already have an EDU handler for %s" % (edu_type,))
-
-        self.edu_handlers[edu_type] = handler
-
-    def register_query_handler(self, query_type, handler):
-        """Sets the handler callable that will be used to handle an incoming
-        federation query of the given type.
-
-        Args:
-            query_type (str): Category name of the query, which should match
-                the string used by make_query.
-            handler (Callable[[dict], Deferred[dict]]): Invoked to handle
-                incoming queries of this type. The return will be yielded
-                on and the result used as the response to the query request.
-        """
-        if query_type in self.query_handlers:
-            raise KeyError(
-                "Already have a Query handler for %s" % (query_type,)
-            )
-
-        self.query_handlers[query_type] = handler
-
-    @defer.inlineCallbacks
-    def on_edu(self, edu_type, origin, content):
-        handler = self.edu_handlers.get(edu_type)
-        if not handler:
-            logger.warn("No handler registered for EDU type %s", edu_type)
-
-        try:
-            yield handler(origin, content)
-        except SynapseError as e:
-            logger.info("Failed to handle edu %r: %r", edu_type, e)
-        except Exception as e:
-            logger.exception("Failed to handle edu %r", edu_type)
-
-    def on_query(self, query_type, args):
-        handler = self.query_handlers.get(query_type)
-        if not handler:
-            logger.warn("No handler registered for query type %s", query_type)
-            raise NotFoundError("No handler for Query type '%s'" % (query_type,))
-
-        return handler(args)

synapse/federation/replication.py

@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+# Copyright 2014-2016 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""This layer is responsible for replicating with remote home servers using
+a given transport.
+"""
+
+from .federation_client import FederationClient
+from .federation_server import FederationServer
+
+from .persistence import TransactionActions
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class ReplicationLayer(FederationClient, FederationServer):
+    """This layer is responsible for replicating with remote home servers over
+    the given transport. I.e., does the sending and receiving of PDUs to
+    remote home servers.
+
+    The layer communicates with the rest of the server via a registered
+    ReplicationHandler.
+
+    In more detail, the layer:
+        * Receives incoming data and processes it into transactions and pdus.
+        * Fetches any PDUs it thinks it might have missed.
+        * Keeps the current state for contexts up to date by applying the
+          suitable conflict resolution.
+        * Sends outgoing pdus wrapped in transactions.
+        * Fills out the references to previous pdus/transactions appropriately
+          for outgoing data.
+    """
+
+    def __init__(self, hs, transport_layer):
+        self.server_name = hs.hostname
+
+        self.keyring = hs.get_keyring()
+
+        self.transport_layer = transport_layer
+
+        self.federation_client = self
+
+        self.store = hs.get_datastore()
+
+        self.handler = None
+        self.edu_handlers = {}
+        self.query_handlers = {}
+
+        self._clock = hs.get_clock()
+
+        self.transaction_actions = TransactionActions(self.store)
+
+        self.hs = hs
+
+        super(ReplicationLayer, self).__init__(hs)
+
+    def __str__(self):
+        return "<ReplicationLayer(%s)>" % self.server_name

synapse/federation/transaction_queue.py

@@ -184,17 +184,22 @@ class TransactionQueue(object):
                     if not is_mine and send_on_behalf_of is None:
                         continue
 
-                    # Get the state from before the event.
-                    # We need to make sure that this is the state from before
-                    # the event and not from after it.
-                    # Otherwise if the last member on a server in a room is
-                    # banned then it won't receive the event because it won't
-                    # be in the room after the ban.
-                    destinations = yield self.state.get_current_hosts_in_room(
-                        event.room_id, latest_event_ids=[
-                            prev_id for prev_id, _ in event.prev_events
-                        ],
-                    )
+                    try:
+                        # Get the state from before the event.
+                        # We need to make sure that this is the state from before
+                        # the event and not from after it.
+                        # Otherwise if the last member on a server in a room is
+                        # banned then it won't receive the event because it won't
+                        # be in the room after the ban.
+                        destinations = yield self.state.get_current_hosts_in_room(
+                            event.room_id, latest_event_ids=[
+                                prev_id for prev_id, _ in event.prev_events
+                            ],
+                        )
+                    except Exception:
+                        logger.exception("Failed to calculate hosts in room")
+                        continue
+
                     destinations = set(destinations)
 
                     if send_on_behalf_of is not None:
@@ -254,6 +259,7 @@ class TransactionQueue(object):
         Args:
             states (list(UserPresenceState))
         """
+        return
 
         # First we queue up the new presence by user ID, so multiple presence
         # updates in quick successtion are correctly handled

synapse/federation/transport/client.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -21,7 +20,6 @@ from synapse.api.urls import FEDERATION_PREFIX as PREFIX
 from synapse.util.logutils import log_function
 
 import logging
-import urllib
 
 
 logger = logging.getLogger(__name__)
@@ -51,7 +49,7 @@ class TransportLayerClient(object):
         logger.debug("get_room_state dest=%s, room=%s",
                      destination, room_id)
 
-        path = _create_path(PREFIX, "/state/%s/", room_id)
+        path = PREFIX + "/state/%s/" % room_id
         return self.client.get_json(
             destination, path=path, args={"event_id": event_id},
         )
@@ -73,7 +71,7 @@ class TransportLayerClient(object):
         logger.debug("get_room_state_ids dest=%s, room=%s",
                      destination, room_id)
 
-        path = _create_path(PREFIX, "/state_ids/%s/", room_id)
+        path = PREFIX + "/state_ids/%s/" % room_id
         return self.client.get_json(
             destination, path=path, args={"event_id": event_id},
         )
@@ -95,7 +93,7 @@ class TransportLayerClient(object):
         logger.debug("get_pdu dest=%s, event_id=%s",
                      destination, event_id)
 
-        path = _create_path(PREFIX, "/event/%s/", event_id)
+        path = PREFIX + "/event/%s/" % (event_id, )
         return self.client.get_json(destination, path=path, timeout=timeout)
 
     @log_function
@@ -121,7 +119,7 @@ class TransportLayerClient(object):
             # TODO: raise?
             return
 
-        path = _create_path(PREFIX, "/backfill/%s/", room_id)
+        path = PREFIX + "/backfill/%s/" % (room_id,)
 
         args = {
             "v": event_tuples,
@@ -159,11 +157,9 @@ class TransportLayerClient(object):
         # generated by the json_data_callback.
         json_data = transaction.get_dict()
 
-        path = _create_path(PREFIX, "/send/%s/", transaction.transaction_id)
-
         response = yield self.client.put_json(
             transaction.destination,
-            path=path,
+            path=PREFIX + "/send/%s/" % transaction.transaction_id,
             data=json_data,
             json_data_callback=json_data_callback,
             long_retries=True,
@@ -181,7 +177,7 @@ class TransportLayerClient(object):
     @log_function
     def make_query(self, destination, query_type, args, retry_on_dns_fail,
                    ignore_backoff=False):
-        path = _create_path(PREFIX, "/query/%s", query_type)
+        path = PREFIX + "/query/%s" % query_type
 
         content = yield self.client.get_json(
             destination=destination,
@@ -226,7 +222,7 @@ class TransportLayerClient(object):
                 "make_membership_event called with membership='%s', must be one of %s" %
                 (membership, ",".join(valid_memberships))
             )
-        path = _create_path(PREFIX, "/make_%s/%s/%s", membership, room_id, user_id)
+        path = PREFIX + "/make_%s/%s/%s" % (membership, room_id, user_id)
 
         ignore_backoff = False
         retry_on_dns_fail = False
@@ -252,7 +248,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def send_join(self, destination, room_id, event_id, content):
-        path = _create_path(PREFIX, "/send_join/%s/%s", room_id, event_id)
+        path = PREFIX + "/send_join/%s/%s" % (room_id, event_id)
 
         response = yield self.client.put_json(
             destination=destination,
@@ -265,7 +261,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def send_leave(self, destination, room_id, event_id, content):
-        path = _create_path(PREFIX, "/send_leave/%s/%s", room_id, event_id)
+        path = PREFIX + "/send_leave/%s/%s" % (room_id, event_id)
 
         response = yield self.client.put_json(
             destination=destination,
@@ -284,7 +280,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def send_invite(self, destination, room_id, event_id, content):
-        path = _create_path(PREFIX, "/invite/%s/%s", room_id, event_id)
+        path = PREFIX + "/invite/%s/%s" % (room_id, event_id)
 
         response = yield self.client.put_json(
             destination=destination,
@@ -326,7 +322,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def exchange_third_party_invite(self, destination, room_id, event_dict):
-        path = _create_path(PREFIX, "/exchange_third_party_invite/%s", room_id,)
+        path = PREFIX + "/exchange_third_party_invite/%s" % (room_id,)
 
         response = yield self.client.put_json(
             destination=destination,
@@ -339,7 +335,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def get_event_auth(self, destination, room_id, event_id):
-        path = _create_path(PREFIX, "/event_auth/%s/%s", room_id, event_id)
+        path = PREFIX + "/event_auth/%s/%s" % (room_id, event_id)
 
         content = yield self.client.get_json(
             destination=destination,
@@ -351,7 +347,7 @@ class TransportLayerClient(object):
     @defer.inlineCallbacks
     @log_function
     def send_query_auth(self, destination, room_id, event_id, content):
-        path = _create_path(PREFIX, "/query_auth/%s/%s", room_id, event_id)
+        path = PREFIX + "/query_auth/%s/%s" % (room_id, event_id)
 
         content = yield self.client.post_json(
             destination=destination,
@@ -413,7 +409,7 @@ class TransportLayerClient(object):
         Returns:
             A dict containg the device keys.
         """
-        path = _create_path(PREFIX, "/user/devices/%s", user_id)
+        path = PREFIX + "/user/devices/" + user_id
 
         content = yield self.client.get_json(
             destination=destination,
@@ -463,7 +459,7 @@ class TransportLayerClient(object):
     @log_function
     def get_missing_events(self, destination, room_id, earliest_events,
                            latest_events, limit, min_depth, timeout):
-        path = _create_path(PREFIX, "/get_missing_events/%s", room_id,)
+        path = PREFIX + "/get_missing_events/%s" % (room_id,)
 
         content = yield self.client.post_json(
             destination=destination,
@@ -483,7 +479,7 @@ class TransportLayerClient(object):
     def get_group_profile(self, destination, group_id, requester_user_id):
         """Get a group profile
         """
-        path = _create_path(PREFIX, "/groups/%s/profile", group_id,)
+        path = PREFIX + "/groups/%s/profile" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -502,7 +498,7 @@ class TransportLayerClient(object):
             requester_user_id (str)
             content (dict): The new profile of the group
         """
-        path = _create_path(PREFIX, "/groups/%s/profile", group_id,)
+        path = PREFIX + "/groups/%s/profile" % (group_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -516,7 +512,7 @@ class TransportLayerClient(object):
     def get_group_summary(self, destination, group_id, requester_user_id):
         """Get a group summary
         """
-        path = _create_path(PREFIX, "/groups/%s/summary", group_id,)
+        path = PREFIX + "/groups/%s/summary" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -529,7 +525,7 @@ class TransportLayerClient(object):
     def get_rooms_in_group(self, destination, group_id, requester_user_id):
         """Get all rooms in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/rooms", group_id,)
+        path = PREFIX + "/groups/%s/rooms" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -542,7 +538,7 @@ class TransportLayerClient(object):
                           content):
         """Add a room to a group
         """
-        path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,)
+        path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -556,10 +552,7 @@ class TransportLayerClient(object):
                              config_key, content):
         """Update room in group
         """
-        path = _create_path(
-            PREFIX, "/groups/%s/room/%s/config/%s",
-            group_id, room_id, config_key,
-        )
+        path = PREFIX + "/groups/%s/room/%s/config/%s" % (group_id, room_id, config_key,)
 
         return self.client.post_json(
             destination=destination,
@@ -572,7 +565,7 @@ class TransportLayerClient(object):
     def remove_room_from_group(self, destination, group_id, requester_user_id, room_id):
         """Remove a room from a group
         """
-        path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,)
+        path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
 
         return self.client.delete_json(
             destination=destination,
@@ -585,7 +578,7 @@ class TransportLayerClient(object):
     def get_users_in_group(self, destination, group_id, requester_user_id):
         """Get users in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/users", group_id,)
+        path = PREFIX + "/groups/%s/users" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -598,7 +591,7 @@ class TransportLayerClient(object):
     def get_invited_users_in_group(self, destination, group_id, requester_user_id):
         """Get users that have been invited to a group
         """
-        path = _create_path(PREFIX, "/groups/%s/invited_users", group_id,)
+        path = PREFIX + "/groups/%s/invited_users" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -611,23 +604,7 @@ class TransportLayerClient(object):
     def accept_group_invite(self, destination, group_id, user_id, content):
         """Accept a group invite
         """
-        path = _create_path(
-            PREFIX, "/groups/%s/users/%s/accept_invite",
-            group_id, user_id,
-        )
-
-        return self.client.post_json(
-            destination=destination,
-            path=path,
-            data=content,
-            ignore_backoff=True,
-        )
-
-    @log_function
-    def join_group(self, destination, group_id, user_id, content):
-        """Attempts to join a group
-        """
-        path = _create_path(PREFIX, "/groups/%s/users/%s/join", group_id, user_id)
+        path = PREFIX + "/groups/%s/users/%s/accept_invite" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -640,7 +617,7 @@ class TransportLayerClient(object):
     def invite_to_group(self, destination, group_id, user_id, requester_user_id, content):
         """Invite a user to a group
         """
-        path = _create_path(PREFIX, "/groups/%s/users/%s/invite", group_id, user_id)
+        path = PREFIX + "/groups/%s/users/%s/invite" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -656,7 +633,7 @@ class TransportLayerClient(object):
         invited.
         """
 
-        path = _create_path(PREFIX, "/groups/local/%s/users/%s/invite", group_id, user_id)
+        path = PREFIX + "/groups/local/%s/users/%s/invite" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -670,7 +647,7 @@ class TransportLayerClient(object):
                                user_id, content):
         """Remove a user fron a group
         """
-        path = _create_path(PREFIX, "/groups/%s/users/%s/remove", group_id, user_id)
+        path = PREFIX + "/groups/%s/users/%s/remove" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -687,7 +664,7 @@ class TransportLayerClient(object):
         kicked from the group.
         """
 
-        path = _create_path(PREFIX, "/groups/local/%s/users/%s/remove", group_id, user_id)
+        path = PREFIX + "/groups/local/%s/users/%s/remove" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -702,7 +679,7 @@ class TransportLayerClient(object):
         the attestations
         """
 
-        path = _create_path(PREFIX, "/groups/%s/renew_attestation/%s", group_id, user_id)
+        path = PREFIX + "/groups/%s/renew_attestation/%s" % (group_id, user_id)
 
         return self.client.post_json(
             destination=destination,
@@ -717,12 +694,11 @@ class TransportLayerClient(object):
         """Update a room entry in a group summary
         """
         if category_id:
-            path = _create_path(
-                PREFIX, "/groups/%s/summary/categories/%s/rooms/%s",
+            path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
                 group_id, category_id, room_id,
             )
         else:
-            path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,)
+            path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -738,12 +714,11 @@ class TransportLayerClient(object):
         """Delete a room entry in a group summary
         """
         if category_id:
-            path = _create_path(
-                PREFIX + "/groups/%s/summary/categories/%s/rooms/%s",
+            path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
                 group_id, category_id, room_id,
             )
         else:
-            path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,)
+            path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
 
         return self.client.delete_json(
             destination=destination,
@@ -756,7 +731,7 @@ class TransportLayerClient(object):
     def get_group_categories(self, destination, group_id, requester_user_id):
         """Get all categories in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/categories", group_id,)
+        path = PREFIX + "/groups/%s/categories" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -769,7 +744,7 @@ class TransportLayerClient(object):
     def get_group_category(self, destination, group_id, requester_user_id, category_id):
         """Get category info in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)
+        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -783,7 +758,7 @@ class TransportLayerClient(object):
                               content):
         """Update a category in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)
+        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -798,7 +773,7 @@ class TransportLayerClient(object):
                               category_id):
         """Delete a category in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)
+        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
 
         return self.client.delete_json(
             destination=destination,
@@ -811,7 +786,7 @@ class TransportLayerClient(object):
     def get_group_roles(self, destination, group_id, requester_user_id):
         """Get all roles in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/roles", group_id,)
+        path = PREFIX + "/groups/%s/roles" % (group_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -824,7 +799,7 @@ class TransportLayerClient(object):
     def get_group_role(self, destination, group_id, requester_user_id, role_id):
         """Get a roles info
         """
-        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)
+        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
 
         return self.client.get_json(
             destination=destination,
@@ -838,7 +813,7 @@ class TransportLayerClient(object):
                           content):
         """Update a role in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)
+        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -852,7 +827,7 @@ class TransportLayerClient(object):
     def delete_group_role(self, destination, group_id, requester_user_id, role_id):
         """Delete a role in a group
         """
-        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)
+        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
 
         return self.client.delete_json(
             destination=destination,
@@ -867,12 +842,11 @@ class TransportLayerClient(object):
         """Update a users entry in a group
         """
         if role_id:
-            path = _create_path(
-                PREFIX, "/groups/%s/summary/roles/%s/users/%s",
+            path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
                 group_id, role_id, user_id,
             )
         else:
-            path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,)
+            path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
 
         return self.client.post_json(
             destination=destination,
@@ -882,33 +856,17 @@ class TransportLayerClient(object):
             ignore_backoff=True,
         )
 
-    @log_function
-    def set_group_join_policy(self, destination, group_id, requester_user_id,
-                              content):
-        """Sets the join policy for a group
-        """
-        path = _create_path(PREFIX, "/groups/%s/settings/m.join_policy", group_id,)
-
-        return self.client.put_json(
-            destination=destination,
-            path=path,
-            args={"requester_user_id": requester_user_id},
-            data=content,
-            ignore_backoff=True,
-        )
-
     @log_function
     def delete_group_summary_user(self, destination, group_id, requester_user_id,
                                   user_id, role_id):
         """Delete a users entry in a group
         """
         if role_id:
-            path = _create_path(
-                PREFIX, "/groups/%s/summary/roles/%s/users/%s",
+            path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
                 group_id, role_id, user_id,
             )
         else:
-            path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,)
+            path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
 
         return self.client.delete_json(
             destination=destination,
@@ -931,22 +889,3 @@ class TransportLayerClient(object):
             data=content,
             ignore_backoff=True,
         )
-
-
-def _create_path(prefix, path, *args):
-    """Creates a path from the prefix, path template and args. Ensures that
-    all args are url encoded.
-
-    Example:
-
-        _create_path(PREFIX, "/event/%s/", event_id)
-
-    Args:
-        prefix (str)
-        path (str): String template for the path
-        args: ([str]): Args to insert into path. Each arg will be url encoded
-
-    Returns:
-        str
-    """
-    return prefix + path % tuple(urllib.quote(arg, "") for arg in args)

synapse/federation/transport/server.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -803,23 +802,6 @@ class FederationGroupsAcceptInviteServlet(BaseFederationServlet):
         defer.returnValue((200, new_content))
 
 
-class FederationGroupsJoinServlet(BaseFederationServlet):
-    """Attempt to join a group
-    """
-    PATH = "/groups/(?P<group_id>[^/]*)/users/(?P<user_id>[^/]*)/join$"
-
-    @defer.inlineCallbacks
-    def on_POST(self, origin, content, query, group_id, user_id):
-        if get_domain_from_id(user_id) != origin:
-            raise SynapseError(403, "user_id doesn't match origin")
-
-        new_content = yield self.handler.join_group(
-            group_id, user_id, content,
-        )
-
-        defer.returnValue((200, new_content))
-
-
 class FederationGroupsRemoveUserServlet(BaseFederationServlet):
     """Leave or kick a user from the group
     """
@@ -1142,24 +1124,6 @@ class FederationGroupsBulkPublicisedServlet(BaseFederationServlet):
         defer.returnValue((200, resp))
 
 
-class FederationGroupsSettingJoinPolicyServlet(BaseFederationServlet):
-    """Sets whether a group is joinable without an invite or knock
-    """
-    PATH = "/groups/(?P<group_id>[^/]*)/settings/m.join_policy$"
-
-    @defer.inlineCallbacks
-    def on_PUT(self, origin, content, query, group_id):
-        requester_user_id = parse_string_from_args(query, "requester_user_id")
-        if get_domain_from_id(requester_user_id) != origin:
-            raise SynapseError(403, "requester_user_id doesn't match origin")
-
-        new_content = yield self.handler.set_group_join_policy(
-            group_id, requester_user_id, content
-        )
-
-        defer.returnValue((200, new_content))
-
-
 FEDERATION_SERVLET_CLASSES = (
     FederationSendServlet,
     FederationPullServlet,
@@ -1199,7 +1163,6 @@ GROUP_SERVER_SERVLET_CLASSES = (
     FederationGroupsInvitedUsersServlet,
     FederationGroupsInviteServlet,
     FederationGroupsAcceptInviteServlet,
-    FederationGroupsJoinServlet,
     FederationGroupsRemoveUserServlet,
     FederationGroupsSummaryRoomsServlet,
     FederationGroupsCategoriesServlet,
@@ -1209,7 +1172,6 @@ GROUP_SERVER_SERVLET_CLASSES = (
     FederationGroupsSummaryUsersServlet,
     FederationGroupsAddRoomsServlet,
     FederationGroupsAddRoomsConfigServlet,
-    FederationGroupsSettingJoinPolicyServlet,
 )
 
 
@@ -1228,7 +1190,7 @@ GROUP_ATTESTATION_SERVLET_CLASSES = (
 def register_servlets(hs, resource, authenticator, ratelimiter):
     for servletclass in FEDERATION_SERVLET_CLASSES:
         servletclass(
-            handler=hs.get_federation_server(),
+            handler=hs.get_replication_layer(),
             authenticator=authenticator,
             ratelimiter=ratelimiter,
             server_name=hs.hostname,

synapse/groups/groups_server.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2017 Vector Creations Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -206,28 +205,6 @@ class GroupsServerHandler(object):
 
         defer.returnValue({})
 
-    @defer.inlineCallbacks
-    def set_group_join_policy(self, group_id, requester_user_id, content):
-        """Sets the group join policy.
-
-        Currently supported policies are:
-         - "invite": an invite must be received and accepted in order to join.
-         - "open": anyone can join.
-        """
-        yield self.check_group_is_ours(
-            group_id, requester_user_id, and_exists=True, and_is_admin=requester_user_id
-        )
-
-        join_policy = _parse_join_policy_from_contents(content)
-        if join_policy is None:
-            raise SynapseError(
-                400, "No value specified for 'm.join_policy'"
-            )
-
-        yield self.store.set_group_join_policy(group_id, join_policy=join_policy)
-
-        defer.returnValue({})
-
     @defer.inlineCallbacks
     def get_group_categories(self, group_id, requester_user_id):
         """Get all categories in a group (as seen by user)
@@ -404,16 +381,9 @@ class GroupsServerHandler(object):
 
         yield self.check_group_is_ours(group_id, requester_user_id)
 
-        group = yield self.store.get_group(group_id)
-
-        if group:
-            cols = [
-                "name", "short_description", "long_description",
-                "avatar_url", "is_public",
-            ]
-            group_description = {key: group[key] for key in cols}
-            group_description["is_openly_joinable"] = group["join_policy"] == "open"
+        group_description = yield self.store.get_group(group_id)
 
+        if group_description:
             defer.returnValue(group_description)
         else:
             raise SynapseError(404, "Unknown group")
@@ -684,40 +654,6 @@ class GroupsServerHandler(object):
         else:
             raise SynapseError(502, "Unknown state returned by HS")
 
-    @defer.inlineCallbacks
-    def _add_user(self, group_id, user_id, content):
-        """Add a user to a group based on a content dict.
-
-        See accept_invite, join_group.
-        """
-        if not self.hs.is_mine_id(user_id):
-            local_attestation = self.attestations.create_attestation(
-                group_id, user_id,
-            )
-
-            remote_attestation = content["attestation"]
-
-            yield self.attestations.verify_attestation(
-                remote_attestation,
-                user_id=user_id,
-                group_id=group_id,
-            )
-        else:
-            local_attestation = None
-            remote_attestation = None
-
-        is_public = _parse_visibility_from_contents(content)
-
-        yield self.store.add_user_to_group(
-            group_id, user_id,
-            is_admin=False,
-            is_public=is_public,
-            local_attestation=local_attestation,
-            remote_attestation=remote_attestation,
-        )
-
-        defer.returnValue(local_attestation)
-
     @defer.inlineCallbacks
     def accept_invite(self, group_id, requester_user_id, content):
         """User tries to accept an invite to the group.
@@ -734,27 +670,30 @@ class GroupsServerHandler(object):
         if not is_invited:
             raise SynapseError(403, "User not invited to group")
 
-        local_attestation = yield self._add_user(group_id, requester_user_id, content)
+        if not self.hs.is_mine_id(requester_user_id):
+            local_attestation = self.attestations.create_attestation(
+                group_id, requester_user_id,
+            )
+            remote_attestation = content["attestation"]
 
-        defer.returnValue({
-            "state": "join",
-            "attestation": local_attestation,
-        })
+            yield self.attestations.verify_attestation(
+                remote_attestation,
+                user_id=requester_user_id,
+                group_id=group_id,
+            )
+        else:
+            local_attestation = None
+            remote_attestation = None
 
-    @defer.inlineCallbacks
-    def join_group(self, group_id, requester_user_id, content):
-        """User tries to join the group.
+        is_public = _parse_visibility_from_contents(content)
 
-        This will error if the group requires an invite/knock to join
-        """
-
-        group_info = yield self.check_group_is_ours(
-            group_id, requester_user_id, and_exists=True
+        yield self.store.add_user_to_group(
+            group_id, requester_user_id,
+            is_admin=False,
+            is_public=is_public,
+            local_attestation=local_attestation,
+            remote_attestation=remote_attestation,
         )
-        if group_info['join_policy'] != "open":
-            raise SynapseError(403, "Group is not publicly joinable")
-
-        local_attestation = yield self._add_user(group_id, requester_user_id, content)
 
         defer.returnValue({
             "state": "join",
@@ -896,31 +835,6 @@ class GroupsServerHandler(object):
         })
 
 
-def _parse_join_policy_from_contents(content):
-    """Given a content for a request, return the specified join policy or None
-    """
-
-    join_policy_dict = content.get("m.join_policy")
-    if join_policy_dict:
-        return _parse_join_policy_dict(join_policy_dict)
-    else:
-        return None
-
-
-def _parse_join_policy_dict(join_policy_dict):
-    """Given a dict for the "m.join_policy" config return the join policy specified
-    """
-    join_policy_type = join_policy_dict.get("type")
-    if not join_policy_type:
-        return "invite"
-
-    if join_policy_type not in ("invite", "open"):
-        raise SynapseError(
-            400, "Synapse only supports 'invite'/'open' join rule"
-        )
-    return join_policy_type
-
-
 def _parse_visibility_from_contents(content):
     """Given a content for a request parse out whether the entity should be
     public or not

synapse/handlers/__init__.py

@@ -17,6 +17,7 @@ from .register import RegistrationHandler
 from .room import (
     RoomCreationHandler, RoomContextHandler,
 )
+from .room_member import RoomMemberHandler
 from .message import MessageHandler
 from .federation import FederationHandler
 from .directory import DirectoryHandler
@@ -48,6 +49,7 @@ class Handlers(object):
         self.registration_handler = RegistrationHandler(hs)
         self.message_handler = MessageHandler(hs)
         self.room_creation_handler = RoomCreationHandler(hs)
+        self.room_member_handler = RoomMemberHandler(hs)
         self.federation_handler = FederationHandler(hs)
         self.directory_handler = DirectoryHandler(hs)
         self.admin_handler = AdminHandler(hs)

synapse/handlers/_base.py

@@ -158,7 +158,7 @@ class BaseHandler(object):
                 # homeserver.
                 requester = synapse.types.create_requester(
                     target_user, is_guest=True)
-                handler = self.hs.get_room_member_handler()
+                handler = self.hs.get_handlers().room_member_handler
                 yield handler.update_membership(
                     requester,
                     target_user,

synapse/handlers/auth.py

@@ -863,10 +863,8 @@ class AuthHandler(BaseHandler):
         """
 
         def _do_validate_hash():
-            return bcrypt.checkpw(
-                password.encode('utf8') + self.hs.config.password_pepper,
-                stored_hash.encode('utf8')
-            )
+            return bcrypt.hashpw(password.encode('utf8') + self.hs.config.password_pepper,
+                                 stored_hash.encode('utf8')) == stored_hash
 
         if stored_hash:
             return make_deferred_yieldable(threads.deferToThread(_do_validate_hash))

synapse/handlers/device.py

@@ -37,15 +37,14 @@ class DeviceHandler(BaseHandler):
         self.state = hs.get_state_handler()
         self._auth_handler = hs.get_auth_handler()
         self.federation_sender = hs.get_federation_sender()
+        self.federation = hs.get_replication_layer()
 
         self._edu_updater = DeviceListEduUpdater(hs, self)
 
-        federation_registry = hs.get_federation_registry()
-
-        federation_registry.register_edu_handler(
+        self.federation.register_edu_handler(
             "m.device_list_update", self._edu_updater.incoming_device_list_update,
         )
-        federation_registry.register_query_handler(
+        self.federation.register_query_handler(
             "user_devices", self.on_federation_query_user_devices,
         )
 
@@ -155,7 +154,7 @@ class DeviceHandler(BaseHandler):
 
         try:
             yield self.store.delete_device(user_id, device_id)
-        except errors.StoreError as e:
+        except errors.StoreError, e:
             if e.code == 404:
                 # no match
                 pass
@@ -204,7 +203,7 @@ class DeviceHandler(BaseHandler):
 
         try:
             yield self.store.delete_devices(user_id, device_ids)
-        except errors.StoreError as e:
+        except errors.StoreError, e:
             if e.code == 404:
                 # no match
                 pass
@@ -243,7 +242,7 @@ class DeviceHandler(BaseHandler):
                 new_display_name=content.get("display_name")
             )
             yield self.notify_device_update(user_id, [device_id])
-        except errors.StoreError as e:
+        except errors.StoreError, e:
             if e.code == 404:
                 raise errors.NotFoundError()
             else:
@@ -431,7 +430,7 @@ class DeviceListEduUpdater(object):
 
     def __init__(self, hs, device_handler):
         self.store = hs.get_datastore()
-        self.federation = hs.get_federation_client()
+        self.federation = hs.get_replication_layer()
         self.clock = hs.get_clock()
         self.device_handler = device_handler
 

synapse/handlers/devicemessage.py

@@ -37,7 +37,7 @@ class DeviceMessageHandler(object):
         self.is_mine = hs.is_mine
         self.federation = hs.get_federation_sender()
 
-        hs.get_federation_registry().register_edu_handler(
+        hs.get_replication_layer().register_edu_handler(
             "m.direct_to_device", self.on_direct_to_device_edu
         )
 

synapse/handlers/directory.py

@@ -36,8 +36,8 @@ class DirectoryHandler(BaseHandler):
         self.appservice_handler = hs.get_application_service_handler()
         self.event_creation_handler = hs.get_event_creation_handler()
 
-        self.federation = hs.get_federation_client()
-        hs.get_federation_registry().register_query_handler(
+        self.federation = hs.get_replication_layer()
+        self.federation.register_query_handler(
             "directory", self.on_directory_query
         )
 

synapse/handlers/e2e_keys.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import simplejson as json
+import ujson as json
 import logging
 
 from canonicaljson import encode_canonical_json
@@ -33,7 +32,7 @@ logger = logging.getLogger(__name__)
 class E2eKeysHandler(object):
     def __init__(self, hs):
         self.store = hs.get_datastore()
-        self.federation = hs.get_federation_client()
+        self.federation = hs.get_replication_layer()
         self.device_handler = hs.get_device_handler()
         self.is_mine = hs.is_mine
         self.clock = hs.get_clock()
@@ -41,7 +40,7 @@ class E2eKeysHandler(object):
         # doesn't really work as part of the generic query API, because the
         # query request requires an object POST, but we abuse the
         # "query handler" interface.
-        hs.get_federation_registry().register_query_handler(
+        self.federation.register_query_handler(
             "client_keys", self.on_federation_query_client_keys
         )
 
@@ -135,8 +134,23 @@ class E2eKeysHandler(object):
                     if user_id in destination_query:
                         results[user_id] = keys
 
+            except CodeMessageException as e:
+                failures[destination] = {
+                    "status": e.code, "message": e.message
+                }
+            except NotRetryingDestination as e:
+                failures[destination] = {
+                    "status": 503, "message": "Not ready for retry",
+                }
+            except FederationDeniedError as e:
+                failures[destination] = {
+                    "status": 403, "message": "Federation Denied",
+                }
             except Exception as e:
-                failures[destination] = _exception_to_failure(e)
+                # include ConnectionRefused and other errors
+                failures[destination] = {
+                    "status": 503, "message": e.message
+                }
 
         yield make_deferred_yieldable(defer.gatherResults([
             preserve_fn(do_remote_query)(destination)
@@ -238,8 +252,19 @@ class E2eKeysHandler(object):
                 for user_id, keys in remote_result["one_time_keys"].items():
                     if user_id in device_keys:
                         json_result[user_id] = keys
+            except CodeMessageException as e:
+                failures[destination] = {
+                    "status": e.code, "message": e.message
+                }
+            except NotRetryingDestination as e:
+                failures[destination] = {
+                    "status": 503, "message": "Not ready for retry",
+                }
             except Exception as e:
-                failures[destination] = _exception_to_failure(e)
+                # include ConnectionRefused and other errors
+                failures[destination] = {
+                    "status": 503, "message": e.message
+                }
 
         yield make_deferred_yieldable(defer.gatherResults([
             preserve_fn(claim_client_keys)(destination)
@@ -337,31 +362,6 @@ class E2eKeysHandler(object):
         )
 
 
-def _exception_to_failure(e):
-    if isinstance(e, CodeMessageException):
-        return {
-            "status": e.code, "message": e.message,
-        }
-
-    if isinstance(e, NotRetryingDestination):
-        return {
-            "status": 503, "message": "Not ready for retry",
-        }
-
-    if isinstance(e, FederationDeniedError):
-        return {
-            "status": 403, "message": "Federation Denied",
-        }
-
-    # include ConnectionRefused and other errors
-    #
-    # Note that some Exceptions (notably twisted's ResponseFailed etc) don't
-    # give a string for e.message, which simplejson then fails to serialize.
-    return {
-        "status": 503, "message": str(e.message),
-    }
-
-
 def _one_time_keys_match(old_key_json, new_key):
     old_key = json.loads(old_key_json)
 

synapse/handlers/federation.py

@@ -68,7 +68,7 @@ class FederationHandler(BaseHandler):
         self.hs = hs
 
         self.store = hs.get_datastore()
-        self.replication_layer = hs.get_federation_client()
+        self.replication_layer = hs.get_replication_layer()
         self.state_handler = hs.get_state_handler()
         self.server_name = hs.hostname
         self.keyring = hs.get_keyring()
@@ -78,6 +78,8 @@ class FederationHandler(BaseHandler):
         self.spam_checker = hs.get_spam_checker()
         self.event_creation_handler = hs.get_event_creation_handler()
 
+        self.replication_layer.set_handler(self)
+
         # When joining a room we need to queue any events for that room up
         self.room_queues = {}
         self._room_pdu_linearizer = Linearizer("fed_room_pdu")
@@ -1445,24 +1447,16 @@ class FederationHandler(BaseHandler):
             auth_events=auth_events,
         )
 
-        try:
-            if not event.internal_metadata.is_outlier() and not backfilled:
-                yield self.action_generator.handle_push_actions_for_event(
-                    event, context
-                )
-
-            event_stream_id, max_stream_id = yield self.store.persist_event(
-                event,
-                context=context,
-                backfilled=backfilled,
+        if not event.internal_metadata.is_outlier() and not backfilled:
+            yield self.action_generator.handle_push_actions_for_event(
+                event, context
             )
-        except:  # noqa: E722, as we reraise the exception this is fine.
-            # Ensure that we actually remove the entries in the push actions
-            # staging area
-            logcontext.preserve_fn(
-                self.store.remove_push_actions_from_staging
-            )(event.event_id)
-            raise
+
+        event_stream_id, max_stream_id = yield self.store.persist_event(
+            event,
+            context=context,
+            backfilled=backfilled,
+        )
 
         if not backfilled:
             # this intentionally does not yield: we don't care about the result
@@ -2151,7 +2145,7 @@ class FederationHandler(BaseHandler):
                 raise e
 
             yield self._check_signature(event, context)
-            member_handler = self.hs.get_room_member_handler()
+            member_handler = self.hs.get_handlers().room_member_handler
             yield member_handler.send_membership_event(None, event, context)
         else:
             destinations = set(x.split(":", 1)[-1] for x in (sender_user_id, room_id))
@@ -2195,7 +2189,7 @@ class FederationHandler(BaseHandler):
         # TODO: Make sure the signatures actually are correct.
         event.signatures.update(returned_invite.signatures)
 
-        member_handler = self.hs.get_room_member_handler()
+        member_handler = self.hs.get_handlers().room_member_handler
         yield member_handler.send_membership_event(None, event, context)
 
     @defer.inlineCallbacks

synapse/handlers/groups_local.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2017 Vector Creations Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -91,8 +90,6 @@ class GroupsLocalHandler(object):
     get_group_role = _create_rerouter("get_group_role")
     get_group_roles = _create_rerouter("get_group_roles")
 
-    set_group_join_policy = _create_rerouter("set_group_join_policy")
-
     @defer.inlineCallbacks
     def get_group_summary(self, group_id, requester_user_id):
         """Get the group summary for a group.
@@ -229,45 +226,7 @@ class GroupsLocalHandler(object):
     def join_group(self, group_id, user_id, content):
         """Request to join a group
         """
-        if self.is_mine_id(group_id):
-            yield self.groups_server_handler.join_group(
-                group_id, user_id, content
-            )
-            local_attestation = None
-            remote_attestation = None
-        else:
-            local_attestation = self.attestations.create_attestation(group_id, user_id)
-            content["attestation"] = local_attestation
-
-            res = yield self.transport_client.join_group(
-                get_domain_from_id(group_id), group_id, user_id, content,
-            )
-
-            remote_attestation = res["attestation"]
-
-            yield self.attestations.verify_attestation(
-                remote_attestation,
-                group_id=group_id,
-                user_id=user_id,
-                server_name=get_domain_from_id(group_id),
-            )
-
-        # TODO: Check that the group is public and we're being added publically
-        is_publicised = content.get("publicise", False)
-
-        token = yield self.store.register_user_group_membership(
-            group_id, user_id,
-            membership="join",
-            is_admin=False,
-            local_attestation=local_attestation,
-            remote_attestation=remote_attestation,
-            is_publicised=is_publicised,
-        )
-        self.notifier.on_new_event(
-            "groups_key", token, users=[user_id],
-        )
-
-        defer.returnValue({})
+        raise NotImplementedError()  # TODO
 
     @defer.inlineCallbacks
     def accept_invite(self, group_id, user_id, content):

synapse/handlers/identity.py

@@ -15,11 +15,6 @@
 # limitations under the License.
 
 """Utilities for interacting with Identity Servers"""
-
-import logging
-
-import simplejson as json
-
 from twisted.internet import defer
 
 from synapse.api.errors import (
@@ -29,6 +24,9 @@ from ._base import BaseHandler
 from synapse.util.async import run_on_reactor
 from synapse.api.errors import SynapseError, Codes
 
+import json
+import logging
+
 logger = logging.getLogger(__name__)
 
 

synapse/handlers/initial_sync.py

@@ -372,6 +372,7 @@ class InitialSyncHandler(BaseHandler):
 
         @defer.inlineCallbacks
         def get_presence():
+            defer.returnValue([])
             states = yield presence_handler.get_states(
                 [m.user_id for m in room_members],
                 as_event=True,

synapse/handlers/message.py

@@ -13,8 +13,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from twisted.internet import defer, reactor
-from twisted.python.failure import Failure
+from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.errors import AuthError, Codes, SynapseError
@@ -25,10 +24,9 @@ from synapse.types import (
     UserID, RoomAlias, RoomStreamToken,
 )
 from synapse.util.async import run_on_reactor, ReadWriteLock, Limiter
-from synapse.util.logcontext import preserve_fn, run_in_background
+from synapse.util.logcontext import preserve_fn
 from synapse.util.metrics import measure_func
-from synapse.util.frozenutils import frozendict_json_encoder
-from synapse.util.stringutils import random_string
+from synapse.util.frozenutils import unfreeze
 from synapse.visibility import filter_events_for_client
 from synapse.replication.http.send_event import send_event_to_master
 
@@ -38,41 +36,11 @@ from canonicaljson import encode_canonical_json
 
 import logging
 import random
-import simplejson
+import ujson
 
 logger = logging.getLogger(__name__)
 
 
-class PurgeStatus(object):
-    """Object tracking the status of a purge request
-
-    This class contains information on the progress of a purge request, for
-    return by get_purge_status.
-
-    Attributes:
-        status (int): Tracks whether this request has completed. One of
-            STATUS_{ACTIVE,COMPLETE,FAILED}
-    """
-
-    STATUS_ACTIVE = 0
-    STATUS_COMPLETE = 1
-    STATUS_FAILED = 2
-
-    STATUS_TEXT = {
-        STATUS_ACTIVE: "active",
-        STATUS_COMPLETE: "complete",
-        STATUS_FAILED: "failed",
-    }
-
-    def __init__(self):
-        self.status = PurgeStatus.STATUS_ACTIVE
-
-    def asdict(self):
-        return {
-            "status": PurgeStatus.STATUS_TEXT[self.status]
-        }
-
-
 class MessageHandler(BaseHandler):
 
     def __init__(self, hs):
@@ -82,87 +50,18 @@ class MessageHandler(BaseHandler):
         self.clock = hs.get_clock()
 
         self.pagination_lock = ReadWriteLock()
-        self._purges_in_progress_by_room = set()
-        # map from purge id to PurgeStatus
-        self._purges_by_id = {}
-
-    def start_purge_history(self, room_id, topological_ordering,
-                            delete_local_events=False):
-        """Start off a history purge on a room.
-
-        Args:
-            room_id (str): The room to purge from
-
-            topological_ordering (int): minimum topo ordering to preserve
-            delete_local_events (bool): True to delete local events as well as
-                remote ones
-
-        Returns:
-            str: unique ID for this purge transaction.
-        """
-        if room_id in self._purges_in_progress_by_room:
-            raise SynapseError(
-                400,
-                "History purge already in progress for %s" % (room_id, ),
-            )
-
-        purge_id = random_string(16)
-
-        # we log the purge_id here so that it can be tied back to the
-        # request id in the log lines.
-        logger.info("[purge] starting purge_id %s", purge_id)
-
-        self._purges_by_id[purge_id] = PurgeStatus()
-        run_in_background(
-            self._purge_history,
-            purge_id, room_id, topological_ordering, delete_local_events,
-        )
-        return purge_id
 
     @defer.inlineCallbacks
-    def _purge_history(self, purge_id, room_id, topological_ordering,
-                       delete_local_events):
-        """Carry out a history purge on a room.
+    def purge_history(self, room_id, event_id, delete_local_events=False):
+        event = yield self.store.get_event(event_id)
 
-        Args:
-            purge_id (str): The id for this purge
-            room_id (str): The room to purge from
-            topological_ordering (int): minimum topo ordering to preserve
-            delete_local_events (bool): True to delete local events as well as
-                remote ones
+        if event.room_id != room_id:
+            raise SynapseError(400, "Event is for wrong room.")
 
-        Returns:
-            Deferred
-        """
-        self._purges_in_progress_by_room.add(room_id)
-        try:
-            with (yield self.pagination_lock.write(room_id)):
-                yield self.store.purge_history(
-                    room_id, topological_ordering, delete_local_events,
-                )
-            logger.info("[purge] complete")
-            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_COMPLETE
-        except Exception:
-            logger.error("[purge] failed: %s", Failure().getTraceback().rstrip())
-            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED
-        finally:
-            self._purges_in_progress_by_room.discard(room_id)
+        depth = event.depth
 
-            # remove the purge from the list 24 hours after it completes
-            def clear_purge():
-                del self._purges_by_id[purge_id]
-            reactor.callLater(24 * 3600, clear_purge)
-
-    def get_purge_status(self, purge_id):
-        """Get the current status of an active purge
-
-        Args:
-            purge_id (str): purge_id returned by start_purge_history
-
-        Returns:
-            PurgeStatus|None
-        """
-        return self._purges_by_id.get(purge_id)
+        with (yield self.pagination_lock.write(room_id)):
+            yield self.store.purge_history(room_id, depth, delete_local_events)
 
     @defer.inlineCallbacks
     def get_messages(self, requester, room_id=None, pagin_config=None,
@@ -384,7 +283,7 @@ class MessageHandler(BaseHandler):
         # If this is an AS, double check that they are allowed to see the members.
         # This can either be because the AS user is in the room or becuase there
         # is a user in the room that the AS is "interested in"
-        if requester.app_service and user_id not in users_with_profile:
+        if False and requester.app_service and user_id not in users_with_profile:
             for uid in users_with_profile:
                 if requester.app_service.is_interested_in_user(uid):
                     break
@@ -654,21 +553,24 @@ class EventCreationHandler(object):
         event,
         context,
         ratelimit=True,
-        extra_users=[],
+        extra_users=[]
     ):
-        """Processes a new event. This includes checking auth, persisting it,
-        notifying users, sending to remote servers, etc.
+        # We now need to go and hit out to wherever we need to hit out to.
 
-        If called from a worker will hit out to the master process for final
-        processing.
+        # If we're a worker we need to hit out to the master.
+        if self.config.worker_app:
+            yield send_event_to_master(
+                self.http_client,
+                host=self.config.worker_replication_host,
+                port=self.config.worker_replication_http_port,
+                requester=requester,
+                event=event,
+                context=context,
+            )
+            return
 
-        Args:
-            requester (Requester)
-            event (FrozenEvent)
-            context (EventContext)
-            ratelimit (bool)
-            extra_users (list(UserID)): Any extra users to notify about event
-        """
+        if ratelimit:
+            yield self.base_handler.ratelimit(requester)
 
         try:
             yield self.auth.check_from_context(event, context)
@@ -678,63 +580,12 @@ class EventCreationHandler(object):
 
         # Ensure that we can round trip before trying to persist in db
         try:
-            dump = frozendict_json_encoder.encode(event.content)
-            simplejson.loads(dump)
+            dump = ujson.dumps(unfreeze(event.content))
+            ujson.loads(dump)
         except Exception:
             logger.exception("Failed to encode content: %r", event.content)
             raise
 
-        yield self.action_generator.handle_push_actions_for_event(
-            event, context
-        )
-
-        try:
-            # If we're a worker we need to hit out to the master.
-            if self.config.worker_app:
-                yield send_event_to_master(
-                    self.http_client,
-                    host=self.config.worker_replication_host,
-                    port=self.config.worker_replication_http_port,
-                    requester=requester,
-                    event=event,
-                    context=context,
-                    ratelimit=ratelimit,
-                    extra_users=extra_users,
-                )
-                return
-
-            yield self.persist_and_notify_client_event(
-                requester,
-                event,
-                context,
-                ratelimit=ratelimit,
-                extra_users=extra_users,
-            )
-        except:  # noqa: E722, as we reraise the exception this is fine.
-            # Ensure that we actually remove the entries in the push actions
-            # staging area, if we calculated them.
-            preserve_fn(self.store.remove_push_actions_from_staging)(event.event_id)
-            raise
-
-    @defer.inlineCallbacks
-    def persist_and_notify_client_event(
-        self,
-        requester,
-        event,
-        context,
-        ratelimit=True,
-        extra_users=[],
-    ):
-        """Called when we have fully built the event, have already
-        calculated the push actions for the event, and checked auth.
-
-        This should only be run on master.
-        """
-        assert not self.config.worker_app
-
-        if ratelimit:
-            yield self.base_handler.ratelimit(requester)
-
         yield self.base_handler.maybe_kick_guest_users(event, context)
 
         if event.type == EventTypes.CanonicalAlias:
@@ -828,6 +679,10 @@ class EventCreationHandler(object):
                 "Changing the room create event is forbidden",
             )
 
+        yield self.action_generator.handle_push_actions_for_event(
+            event, context
+        )
+
         (event_stream_id, max_stream_id) = yield self.store.persist_event(
             event, context=context
         )

synapse/handlers/presence.py

@@ -93,30 +93,29 @@ class PresenceHandler(object):
         self.store = hs.get_datastore()
         self.wheel_timer = WheelTimer()
         self.notifier = hs.get_notifier()
+        self.replication = hs.get_replication_layer()
         self.federation = hs.get_federation_sender()
 
         self.state = hs.get_state_handler()
 
-        federation_registry = hs.get_federation_registry()
-
-        federation_registry.register_edu_handler(
+        self.replication.register_edu_handler(
             "m.presence", self.incoming_presence
         )
-        federation_registry.register_edu_handler(
+        self.replication.register_edu_handler(
             "m.presence_invite",
             lambda origin, content: self.invite_presence(
                 observed_user=UserID.from_string(content["observed_user"]),
                 observer_user=UserID.from_string(content["observer_user"]),
             )
         )
-        federation_registry.register_edu_handler(
+        self.replication.register_edu_handler(
             "m.presence_accept",
             lambda origin, content: self.accept_presence(
                 observed_user=UserID.from_string(content["observed_user"]),
                 observer_user=UserID.from_string(content["observer_user"]),
             )
         )
-        federation_registry.register_edu_handler(
+        self.replication.register_edu_handler(
             "m.presence_deny",
             lambda origin, content: self.deny_presence(
                 observed_user=UserID.from_string(content["observed_user"]),
@@ -373,6 +372,7 @@ class PresenceHandler(object):
         """We've seen the user do something that indicates they're interacting
         with the app.
         """
+        return
         user_id = user.to_string()
 
         bump_active_time_counter.inc()
@@ -402,6 +402,7 @@ class PresenceHandler(object):
                 Useful for streams that are not associated with an actual
                 client that is being used by a user.
         """
+        affect_presence = False
         if affect_presence:
             curr_sync = self.user_to_num_current_syncs.get(user_id, 0)
             self.user_to_num_current_syncs[user_id] = curr_sync + 1
@@ -444,6 +445,8 @@ class PresenceHandler(object):
         Returns:
             set(str): A set of user_id strings.
         """
+        # presence is disabled on matrix.org, so we return the empty set
+        return set()
         syncing_user_ids = {
             user_id for user_id, count in self.user_to_num_current_syncs.items()
             if count
@@ -463,6 +466,7 @@ class PresenceHandler(object):
             syncing_user_ids(set(str)): The set of user_ids that are
                 currently syncing on that server.
         """
+        return
 
         # Grab the previous list of user_ids that were syncing on that process
         prev_syncing_user_ids = (

synapse/handlers/profile.py

@@ -31,17 +31,14 @@ class ProfileHandler(BaseHandler):
     def __init__(self, hs):
         super(ProfileHandler, self).__init__(hs)
 
-        self.federation = hs.get_federation_client()
-        hs.get_federation_registry().register_query_handler(
+        self.federation = hs.get_replication_layer()
+        self.federation.register_query_handler(
             "profile", self.on_profile_query
         )
 
         self.user_directory_handler = hs.get_user_directory_handler()
 
-        if hs.config.worker_app is None:
-            self.clock.looping_call(
-                self._update_remote_profile_cache, self.PROFILE_UPDATE_MS,
-            )
+        self.clock.looping_call(self._update_remote_profile_cache, self.PROFILE_UPDATE_MS)
 
     @defer.inlineCallbacks
     def get_profile(self, user_id):
@@ -236,7 +233,7 @@ class ProfileHandler(BaseHandler):
         )
 
         for room_id in room_ids:
-            handler = self.hs.get_room_member_handler()
+            handler = self.hs.get_handlers().room_member_handler
             try:
                 # Assume the target_user isn't a guest,
                 # because we don't let guests set profile or avatar data.

synapse/handlers/read_marker.py

@@ -41,9 +41,9 @@ class ReadMarkerHandler(BaseHandler):
         """
 
         with (yield self.read_marker_linearizer.queue((room_id, user_id))):
-            existing_read_marker = yield self.store.get_account_data_for_room_and_type(
-                user_id, room_id, "m.fully_read",
-            )
+            account_data = yield self.store.get_account_data_for_room(user_id, room_id)
+
+            existing_read_marker = account_data.get("m.fully_read", None)
 
             should_update = True
 

synapse/handlers/receipts.py

@@ -35,7 +35,7 @@ class ReceiptsHandler(BaseHandler):
         self.store = hs.get_datastore()
         self.hs = hs
         self.federation = hs.get_federation_sender()
-        hs.get_federation_registry().register_edu_handler(
+        hs.get_replication_layer().register_edu_handler(
             "m.receipt", self._received_remote_receipt
         )
         self.clock = self.hs.get_clock()

synapse/handlers/register.py

@@ -24,7 +24,7 @@ from synapse.api.errors import (
 from synapse.http.client import CaptchaServerHttpClient
 from synapse import types
 from synapse.types import UserID
-from synapse.util.async import run_on_reactor, Linearizer
+from synapse.util.async import run_on_reactor
 from synapse.util.threepids import check_3pid_allowed
 from ._base import BaseHandler
 
@@ -46,10 +46,6 @@ class RegistrationHandler(BaseHandler):
 
         self.macaroon_gen = hs.get_macaroon_generator()
 
-        self._generate_user_id_linearizer = Linearizer(
-            name="_generate_user_id_linearizer",
-        )
-
     @defer.inlineCallbacks
     def check_username(self, localpart, guest_access_token=None,
                        assigned_user_id=None):
@@ -349,11 +345,9 @@ class RegistrationHandler(BaseHandler):
     @defer.inlineCallbacks
     def _generate_user_id(self, reseed=False):
         if reseed or self._next_generated_user_id is None:
-            with (yield self._generate_user_id_linearizer.queue(())):
-                if reseed or self._next_generated_user_id is None:
-                    self._next_generated_user_id = (
-                        yield self.store.find_next_generated_user_id_localpart()
-                    )
+            self._next_generated_user_id = (
+                yield self.store.find_next_generated_user_id_localpart()
+            )
 
         id = self._next_generated_user_id
         self._next_generated_user_id += 1
@@ -452,34 +446,16 @@ class RegistrationHandler(BaseHandler):
         return self.hs.get_auth_handler()
 
     @defer.inlineCallbacks
-    def get_or_register_3pid_guest(self, medium, address, inviter_user_id):
-        """Get a guest access token for a 3PID, creating a guest account if
-        one doesn't already exist.
-
-        Args:
-            medium (str)
-            address (str)
-            inviter_user_id (str): The user ID who is trying to invite the
-                3PID
-
-        Returns:
-            Deferred[(str, str)]: A 2-tuple of `(user_id, access_token)` of the
-            3PID guest account.
-        """
+    def guest_access_token_for(self, medium, address, inviter_user_id):
         access_token = yield self.store.get_3pid_guest_access_token(medium, address)
         if access_token:
-            user_info = yield self.auth.get_user_by_access_token(
-                access_token
-            )
+            defer.returnValue(access_token)
 
-            defer.returnValue((user_info["user"].to_string(), access_token))
-
-        user_id, access_token = yield self.register(
+        _, access_token = yield self.register(
             generate_token=True,
             make_guest=True
         )
         access_token = yield self.store.save_or_get_3pid_guest_access_token(
             medium, address, access_token, inviter_user_id
         )
-
-        defer.returnValue((user_id, access_token))
+        defer.returnValue(access_token)

synapse/handlers/room.py

@@ -165,7 +165,7 @@ class RoomCreationHandler(BaseHandler):
 
         creation_content = config.get("creation_content", {})
 
-        room_member_handler = self.hs.get_room_member_handler()
+        room_member_handler = self.hs.get_handlers().room_member_handler
 
         yield self._send_events_for_new_room(
             requester,
@@ -224,7 +224,7 @@ class RoomCreationHandler(BaseHandler):
             id_server = invite_3pid["id_server"]
             address = invite_3pid["address"]
             medium = invite_3pid["medium"]
-            yield self.hs.get_room_member_handler().do_3pid_invite(
+            yield self.hs.get_handlers().room_member_handler.do_3pid_invite(
                 room_id,
                 requester.user,
                 medium,
@@ -475,9 +475,12 @@ class RoomEventSource(object):
             user.to_string()
         )
         if app_service:
-            # We no longer support AS users using /sync directly.
-            # See https://github.com/matrix-org/matrix-doc/issues/1144
-            raise NotImplementedError()
+            events, end_key = yield self.store.get_appservice_room_stream(
+                service=app_service,
+                from_key=from_key,
+                to_key=to_key,
+                limit=limit,
+            )
         else:
             room_events = yield self.store.get_membership_changes_for_user(
                 user.to_string(), from_key, to_key

synapse/handlers/room_list.py

@@ -44,7 +44,7 @@ EMTPY_THIRD_PARTY_ID = ThirdPartyInstanceID(None, None)
 class RoomListHandler(BaseHandler):
     def __init__(self, hs):
         super(RoomListHandler, self).__init__(hs)
-        self.response_cache = ResponseCache(hs)
+        self.response_cache = ResponseCache(hs, timeout_ms=10 * 60 * 1000)
         self.remote_response_cache = ResponseCache(hs, timeout_ms=30 * 1000)
 
     def get_local_public_room_list(self, limit=None, since_token=None,
@@ -409,7 +409,7 @@ class RoomListHandler(BaseHandler):
     def _get_remote_list_cached(self, server_name, limit=None, since_token=None,
                                 search_filter=None, include_all_networks=False,
                                 third_party_instance_id=None,):
-        repl_layer = self.hs.get_federation_client()
+        repl_layer = self.hs.get_replication_layer()
         if search_filter:
             # We can't cache when asking for search
             return repl_layer.get_public_rooms(

synapse/handlers/room_member.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import abc
+
 import logging
 
 from signedjson.key import decode_verify_key_bytes
@@ -28,123 +28,36 @@ from synapse.api.constants import (
 )
 from synapse.api.errors import AuthError, SynapseError, Codes
 from synapse.types import UserID, RoomID
-from synapse.util.async import Linearizer
+from synapse.util.async import Linearizer, Limiter
 from synapse.util.distributor import user_left_room, user_joined_room
-
+from ._base import BaseHandler
 
 logger = logging.getLogger(__name__)
 
 id_server_scheme = "https://"
 
 
-class RoomMemberHandler(object):
+class RoomMemberHandler(BaseHandler):
     # TODO(paul): This handler currently contains a messy conflation of
     #   low-level API that works on UserID objects and so on, and REST-level
     #   API that takes ID strings and returns pagination chunks. These concerns
     #   ought to be separated out a lot better.
 
-    __metaclass__ = abc.ABCMeta
-
     def __init__(self, hs):
-        self.hs = hs
-        self.store = hs.get_datastore()
-        self.auth = hs.get_auth()
-        self.state_handler = hs.get_state_handler()
-        self.config = hs.config
-        self.simple_http_client = hs.get_simple_http_client()
+        super(RoomMemberHandler, self).__init__(hs)
 
-        self.federation_handler = hs.get_handlers().federation_handler
-        self.directory_handler = hs.get_handlers().directory_handler
-        self.registration_handler = hs.get_handlers().registration_handler
         self.profile_handler = hs.get_profile_handler()
         self.event_creation_hander = hs.get_event_creation_handler()
 
         self.member_linearizer = Linearizer(name="member")
+        self.member_limiter = Limiter(3)
 
         self.clock = hs.get_clock()
         self.spam_checker = hs.get_spam_checker()
 
-    @abc.abstractmethod
-    def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
-        """Try and join a room that this server is not in
-
-        Args:
-            requester (Requester)
-            remote_room_hosts (list[str]): List of servers that can be used
-                to join via.
-            room_id (str): Room that we are trying to join
-            user (UserID): User who is trying to join
-            content (dict): A dict that should be used as the content of the
-                join event.
-
-        Returns:
-            Deferred
-        """
-        raise NotImplementedError()
-
-    @abc.abstractmethod
-    def _remote_reject_invite(self, remote_room_hosts, room_id, target):
-        """Attempt to reject an invite for a room this server is not in. If we
-        fail to do so we locally mark the invite as rejected.
-
-        Args:
-            requester (Requester)
-            remote_room_hosts (list[str]): List of servers to use to try and
-                reject invite
-            room_id (str)
-            target (UserID): The user rejecting the invite
-
-        Returns:
-            Deferred[dict]: A dictionary to be returned to the client, may
-            include event_id etc, or nothing if we locally rejected
-        """
-        raise NotImplementedError()
-
-    @abc.abstractmethod
-    def get_or_register_3pid_guest(self, requester, medium, address, inviter_user_id):
-        """Get a guest access token for a 3PID, creating a guest account if
-        one doesn't already exist.
-
-        Args:
-            requester (Requester)
-            medium (str)
-            address (str)
-            inviter_user_id (str): The user ID who is trying to invite the
-                3PID
-
-        Returns:
-            Deferred[(str, str)]: A 2-tuple of `(user_id, access_token)` of the
-            3PID guest account.
-        """
-        raise NotImplementedError()
-
-    @abc.abstractmethod
-    def _user_joined_room(self, target, room_id):
-        """Notifies distributor on master process that the user has joined the
-        room.
-
-        Args:
-            target (UserID)
-            room_id (str)
-
-        Returns:
-            Deferred|None
-        """
-        raise NotImplementedError()
-
-    @abc.abstractmethod
-    def _user_left_room(self, target, room_id):
-        """Notifies distributor on master process that the user has left the
-        room.
-
-        Args:
-            target (UserID)
-            room_id (str)
-
-        Returns:
-            Deferred|None
-        """
-        raise NotImplementedError()
+        self.distributor = hs.get_distributor()
+        self.distributor.declare("user_joined_room")
+        self.distributor.declare("user_left_room")
 
     @defer.inlineCallbacks
     def _local_membership_update(
@@ -208,15 +121,32 @@ class RoomMemberHandler(object):
                 prev_member_event = yield self.store.get_event(prev_member_event_id)
                 newly_joined = prev_member_event.membership != Membership.JOIN
             if newly_joined:
-                yield self._user_joined_room(target, room_id)
+                yield user_joined_room(self.distributor, target, room_id)
         elif event.membership == Membership.LEAVE:
             if prev_member_event_id:
                 prev_member_event = yield self.store.get_event(prev_member_event_id)
                 if prev_member_event.membership == Membership.JOIN:
-                    yield self._user_left_room(target, room_id)
+                    user_left_room(self.distributor, target, room_id)
 
         defer.returnValue(event)
 
+    @defer.inlineCallbacks
+    def remote_join(self, remote_room_hosts, room_id, user, content):
+        if len(remote_room_hosts) == 0:
+            raise SynapseError(404, "No known servers")
+
+        # We don't do an auth check if we are doing an invite
+        # join dance for now, since we're kinda implicitly checking
+        # that we are allowed to join when we decide whether or not we
+        # need to do the invite/join dance.
+        yield self.hs.get_handlers().federation_handler.do_invite_join(
+            remote_room_hosts,
+            room_id,
+            user.to_string(),
+            content,
+        )
+        yield user_joined_room(self.distributor, user, room_id)
+
     @defer.inlineCallbacks
     def update_membership(
             self,
@@ -232,18 +162,23 @@ class RoomMemberHandler(object):
     ):
         key = (room_id,)
 
-        with (yield self.member_linearizer.queue(key)):
-            result = yield self._update_membership(
-                requester,
-                target,
-                room_id,
-                action,
-                txn_id=txn_id,
-                remote_room_hosts=remote_room_hosts,
-                third_party_signed=third_party_signed,
-                ratelimit=ratelimit,
-                content=content,
-            )
+        as_id = object()
+        if requester.app_service:
+            as_id = requester.app_service.id
+
+        with (yield self.member_limiter.queue(as_id)):
+            with (yield self.member_linearizer.queue(key)):
+                result = yield self._update_membership(
+                    requester,
+                    target,
+                    room_id,
+                    action,
+                    txn_id=txn_id,
+                    remote_room_hosts=remote_room_hosts,
+                    third_party_signed=third_party_signed,
+                    ratelimit=ratelimit,
+                    content=content,
+                )
 
         defer.returnValue(result)
 
@@ -275,7 +210,8 @@ class RoomMemberHandler(object):
         # if this is a join with a 3pid signature, we may need to turn a 3pid
         # invite into a normal invite before we can handle the join.
         if third_party_signed is not None:
-            yield self.federation_handler.exchange_third_party_invite(
+            replication = self.hs.get_replication_layer()
+            yield replication.exchange_third_party_invite(
                 third_party_signed["sender"],
                 target.to_string(),
                 room_id,
@@ -296,7 +232,7 @@ class RoomMemberHandler(object):
                 requester.user,
             )
             if not is_requester_admin:
-                if self.config.block_non_admin_invites:
+                if self.hs.config.block_non_admin_invites:
                     logger.info(
                         "Blocking invite: user is not admin and non-admin "
                         "invites disabled"
@@ -355,7 +291,7 @@ class RoomMemberHandler(object):
                     raise AuthError(403, "Guest access not allowed")
 
             if not is_host_in_room:
-                inviter = yield self._get_inviter(target.to_string(), room_id)
+                inviter = yield self.get_inviter(target.to_string(), room_id)
                 if inviter and not self.hs.is_mine(inviter):
                     remote_room_hosts.append(inviter.domain)
 
@@ -369,15 +305,15 @@ class RoomMemberHandler(object):
                 if requester.is_guest:
                     content["kind"] = "guest"
 
-                ret = yield self._remote_join(
-                    requester, remote_room_hosts, room_id, target, content
+                ret = yield self.remote_join(
+                    remote_room_hosts, room_id, target, content
                 )
                 defer.returnValue(ret)
 
         elif effective_membership_state == Membership.LEAVE:
             if not is_host_in_room:
                 # perhaps we've been invited
-                inviter = yield self._get_inviter(target.to_string(), room_id)
+                inviter = yield self.get_inviter(target.to_string(), room_id)
                 if not inviter:
                     raise SynapseError(404, "Not a known room")
 
@@ -391,10 +327,28 @@ class RoomMemberHandler(object):
                 else:
                     # send the rejection to the inviter's HS.
                     remote_room_hosts = remote_room_hosts + [inviter.domain]
-                    res = yield self._remote_reject_invite(
-                        requester, remote_room_hosts, room_id, target,
-                    )
-                    defer.returnValue(res)
+                    fed_handler = self.hs.get_handlers().federation_handler
+                    try:
+                        ret = yield fed_handler.do_remotely_reject_invite(
+                            remote_room_hosts,
+                            room_id,
+                            target.to_string(),
+                        )
+                        defer.returnValue(ret)
+                    except Exception as e:
+                        # if we were unable to reject the exception, just mark
+                        # it as rejected on our end and plough ahead.
+                        #
+                        # The 'except' clause is very broad, but we need to
+                        # capture everything from DNS failures upwards
+                        #
+                        logger.warn("Failed to reject invite: %s", e)
+
+                        yield self.store.locally_reject_invite(
+                            target.to_string(), room_id
+                        )
+
+                        defer.returnValue({})
 
         res = yield self._local_membership_update(
             requester=requester,
@@ -490,12 +444,12 @@ class RoomMemberHandler(object):
                 prev_member_event = yield self.store.get_event(prev_member_event_id)
                 newly_joined = prev_member_event.membership != Membership.JOIN
             if newly_joined:
-                yield self._user_joined_room(target_user, room_id)
+                yield user_joined_room(self.distributor, target_user, room_id)
         elif event.membership == Membership.LEAVE:
             if prev_member_event_id:
                 prev_member_event = yield self.store.get_event(prev_member_event_id)
                 if prev_member_event.membership == Membership.JOIN:
-                    yield self._user_left_room(target_user, room_id)
+                    user_left_room(self.distributor, target_user, room_id)
 
     @defer.inlineCallbacks
     def _can_guest_join(self, current_state_ids):
@@ -529,7 +483,7 @@ class RoomMemberHandler(object):
         Raises:
             SynapseError if room alias could not be found.
         """
-        directory_handler = self.directory_handler
+        directory_handler = self.hs.get_handlers().directory_handler
         mapping = yield directory_handler.get_association(room_alias)
 
         if not mapping:
@@ -541,7 +495,7 @@ class RoomMemberHandler(object):
         defer.returnValue((RoomID.from_string(room_id), servers))
 
     @defer.inlineCallbacks
-    def _get_inviter(self, user_id, room_id):
+    def get_inviter(self, user_id, room_id):
         invite = yield self.store.get_invite_for_user_in_room(
             user_id=user_id,
             room_id=room_id,
@@ -560,7 +514,7 @@ class RoomMemberHandler(object):
             requester,
             txn_id
     ):
-        if self.config.block_non_admin_invites:
+        if self.hs.config.block_non_admin_invites:
             is_requester_admin = yield self.auth.is_server_admin(
                 requester.user,
             )
@@ -607,7 +561,7 @@ class RoomMemberHandler(object):
             str: the matrix ID of the 3pid, or None if it is not recognized.
         """
         try:
-            data = yield self.simple_http_client.get_json(
+            data = yield self.hs.get_simple_http_client().get_json(
                 "%s%s/_matrix/identity/api/v1/lookup" % (id_server_scheme, id_server,),
                 {
                     "medium": medium,
@@ -618,7 +572,7 @@ class RoomMemberHandler(object):
             if "mxid" in data:
                 if "signatures" not in data:
                     raise AuthError(401, "No signatures on 3pid binding")
-                yield self._verify_any_signature(data, id_server)
+                self.verify_any_signature(data, id_server)
                 defer.returnValue(data["mxid"])
 
         except IOError as e:
@@ -626,11 +580,11 @@ class RoomMemberHandler(object):
             defer.returnValue(None)
 
     @defer.inlineCallbacks
-    def _verify_any_signature(self, data, server_hostname):
+    def verify_any_signature(self, data, server_hostname):
         if server_hostname not in data["signatures"]:
             raise AuthError(401, "No signature from server %s" % (server_hostname,))
         for key_name, signature in data["signatures"][server_hostname].items():
-            key_data = yield self.simple_http_client.get_json(
+            key_data = yield self.hs.get_simple_http_client().get_json(
                 "%s%s/_matrix/identity/api/v1/pubkey/%s" %
                 (id_server_scheme, server_hostname, key_name,),
             )
@@ -655,7 +609,7 @@ class RoomMemberHandler(object):
             user,
             txn_id
     ):
-        room_state = yield self.state_handler.get_current_state(room_id)
+        room_state = yield self.hs.get_state_handler().get_current_state(room_id)
 
         inviter_display_name = ""
         inviter_avatar_url = ""
@@ -686,7 +640,6 @@ class RoomMemberHandler(object):
 
         token, public_keys, fallback_public_key, display_name = (
             yield self._ask_id_server_for_third_party_invite(
-                requester=requester,
                 id_server=id_server,
                 medium=medium,
                 address=address,
@@ -723,7 +676,6 @@ class RoomMemberHandler(object):
     @defer.inlineCallbacks
     def _ask_id_server_for_third_party_invite(
             self,
-            requester,
             id_server,
             medium,
             address,
@@ -740,7 +692,6 @@ class RoomMemberHandler(object):
         Asks an identity server for a third party invite.
 
         Args:
-            requester (Requester)
             id_server (str): hostname + optional port for the identity server.
             medium (str): The literal string "email".
             address (str): The third party address being invited.
@@ -782,20 +733,24 @@ class RoomMemberHandler(object):
             "sender_avatar_url": inviter_avatar_url,
         }
 
-        if self.config.invite_3pid_guest:
-            guest_access_token, guest_user_id = yield self.get_or_register_3pid_guest(
-                requester=requester,
+        if self.hs.config.invite_3pid_guest:
+            registration_handler = self.hs.get_handlers().registration_handler
+            guest_access_token = yield registration_handler.guest_access_token_for(
                 medium=medium,
                 address=address,
                 inviter_user_id=inviter_user_id,
             )
 
+            guest_user_info = yield self.hs.get_auth().get_user_by_access_token(
+                guest_access_token
+            )
+
             invite_config.update({
                 "guest_access_token": guest_access_token,
-                "guest_user_id": guest_user_id,
+                "guest_user_id": guest_user_info["user"].to_string(),
             })
 
-        data = yield self.simple_http_client.post_urlencoded_get_json(
+        data = yield self.hs.get_simple_http_client().post_urlencoded_get_json(
             is_url,
             invite_config
         )
@@ -817,6 +772,27 @@ class RoomMemberHandler(object):
         display_name = data["display_name"]
         defer.returnValue((token, public_keys, fallback_public_key, display_name))
 
+    @defer.inlineCallbacks
+    def forget(self, user, room_id):
+        user_id = user.to_string()
+
+        member = yield self.state_handler.get_current_state(
+            room_id=room_id,
+            event_type=EventTypes.Member,
+            state_key=user_id
+        )
+        membership = member.membership if member else None
+
+        if membership is not None and membership not in [
+            Membership.LEAVE, Membership.BAN
+        ]:
+            raise SynapseError(400, "User %s in room %s" % (
+                user_id, room_id
+            ))
+
+        if membership:
+            yield self.store.forget(user_id, room_id)
+
     @defer.inlineCallbacks
     def _is_host_in_room(self, current_state_ids):
         # Have we just created the room, and is this about to be the very
@@ -838,94 +814,3 @@ class RoomMemberHandler(object):
                 defer.returnValue(True)
 
         defer.returnValue(False)
-
-
-class RoomMemberMasterHandler(RoomMemberHandler):
-    def __init__(self, hs):
-        super(RoomMemberMasterHandler, self).__init__(hs)
-
-        self.distributor = hs.get_distributor()
-        self.distributor.declare("user_joined_room")
-        self.distributor.declare("user_left_room")
-
-    @defer.inlineCallbacks
-    def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
-        """Implements RoomMemberHandler._remote_join
-        """
-        if len(remote_room_hosts) == 0:
-            raise SynapseError(404, "No known servers")
-
-        # We don't do an auth check if we are doing an invite
-        # join dance for now, since we're kinda implicitly checking
-        # that we are allowed to join when we decide whether or not we
-        # need to do the invite/join dance.
-        yield self.federation_handler.do_invite_join(
-            remote_room_hosts,
-            room_id,
-            user.to_string(),
-            content,
-        )
-        yield self._user_joined_room(user, room_id)
-
-    @defer.inlineCallbacks
-    def _remote_reject_invite(self, requester, remote_room_hosts, room_id, target):
-        """Implements RoomMemberHandler._remote_reject_invite
-        """
-        fed_handler = self.federation_handler
-        try:
-            ret = yield fed_handler.do_remotely_reject_invite(
-                remote_room_hosts,
-                room_id,
-                target.to_string(),
-            )
-            defer.returnValue(ret)
-        except Exception as e:
-            # if we were unable to reject the exception, just mark
-            # it as rejected on our end and plough ahead.
-            #
-            # The 'except' clause is very broad, but we need to
-            # capture everything from DNS failures upwards
-            #
-            logger.warn("Failed to reject invite: %s", e)
-
-            yield self.store.locally_reject_invite(
-                target.to_string(), room_id
-            )
-            defer.returnValue({})
-
-    def get_or_register_3pid_guest(self, requester, medium, address, inviter_user_id):
-        """Implements RoomMemberHandler.get_or_register_3pid_guest
-        """
-        rg = self.registration_handler
-        return rg.get_or_register_3pid_guest(medium, address, inviter_user_id)
-
-    def _user_joined_room(self, target, room_id):
-        """Implements RoomMemberHandler._user_joined_room
-        """
-        return user_joined_room(self.distributor, target, room_id)
-
-    def _user_left_room(self, target, room_id):
-        """Implements RoomMemberHandler._user_left_room
-        """
-        return user_left_room(self.distributor, target, room_id)
-
-    @defer.inlineCallbacks
-    def forget(self, user, room_id):
-        user_id = user.to_string()
-
-        member = yield self.state_handler.get_current_state(
-            room_id=room_id,
-            event_type=EventTypes.Member,
-            state_key=user_id
-        )
-        membership = member.membership if member else None
-
-        if membership is not None and membership not in [
-            Membership.LEAVE, Membership.BAN
-        ]:
-            raise SynapseError(400, "User %s in room %s" % (
-                user_id, room_id
-            ))
-
-        if membership:
-            yield self.store.forget(user_id, room_id)

synapse/handlers/room_member_worker.py

@@ -1,102 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2018 New Vector 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.
-
-import logging
-
-from twisted.internet import defer
-
-from synapse.api.errors import SynapseError
-from synapse.handlers.room_member import RoomMemberHandler
-from synapse.replication.http.membership import (
-    remote_join, remote_reject_invite, get_or_register_3pid_guest,
-    notify_user_membership_change,
-)
-
-
-logger = logging.getLogger(__name__)
-
-
-class RoomMemberWorkerHandler(RoomMemberHandler):
-    @defer.inlineCallbacks
-    def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
-        """Implements RoomMemberHandler._remote_join
-        """
-        if len(remote_room_hosts) == 0:
-            raise SynapseError(404, "No known servers")
-
-        ret = yield remote_join(
-            self.simple_http_client,
-            host=self.config.worker_replication_host,
-            port=self.config.worker_replication_http_port,
-            requester=requester,
-            remote_room_hosts=remote_room_hosts,
-            room_id=room_id,
-            user_id=user.to_string(),
-            content=content,
-        )
-
-        yield self._user_joined_room(user, room_id)
-
-        defer.returnValue(ret)
-
-    def _remote_reject_invite(self, requester, remote_room_hosts, room_id, target):
-        """Implements RoomMemberHandler._remote_reject_invite
-        """
-        return remote_reject_invite(
-            self.simple_http_client,
-            host=self.config.worker_replication_host,
-            port=self.config.worker_replication_http_port,
-            requester=requester,
-            remote_room_hosts=remote_room_hosts,
-            room_id=room_id,
-            user_id=target.to_string(),
-        )
-
-    def _user_joined_room(self, target, room_id):
-        """Implements RoomMemberHandler._user_joined_room
-        """
-        return notify_user_membership_change(
-            self.simple_http_client,
-            host=self.config.worker_replication_host,
-            port=self.config.worker_replication_http_port,
-            user_id=target.to_string(),
-            room_id=room_id,
-            change="joined",
-        )
-
-    def _user_left_room(self, target, room_id):
-        """Implements RoomMemberHandler._user_left_room
-        """
-        return notify_user_membership_change(
-            self.simple_http_client,
-            host=self.config.worker_replication_host,
-            port=self.config.worker_replication_http_port,
-            user_id=target.to_string(),
-            room_id=room_id,
-            change="left",
-        )
-
-    def get_or_register_3pid_guest(self, requester, medium, address, inviter_user_id):
-        """Implements RoomMemberHandler.get_or_register_3pid_guest
-        """
-        return get_or_register_3pid_guest(
-            self.simple_http_client,
-            host=self.config.worker_replication_host,
-            port=self.config.worker_replication_http_port,
-            requester=requester,
-            medium=medium,
-            address=address,
-            inviter_user_id=inviter_user_id,
-        )

synapse/handlers/sync.py

@@ -235,10 +235,10 @@ class SyncHandler(object):
         defer.returnValue(rules)
 
     @defer.inlineCallbacks
-    def ephemeral_by_room(self, sync_result_builder, now_token, since_token=None):
+    def ephemeral_by_room(self, sync_config, now_token, since_token=None):
         """Get the ephemeral events for each room the user is in
         Args:
-            sync_result_builder(SyncResultBuilder)
+            sync_config (SyncConfig): The flags, filters and user for the sync.
             now_token (StreamToken): Where the server is currently up to.
             since_token (StreamToken): Where the server was when the client
                 last synced.
@@ -248,12 +248,10 @@ class SyncHandler(object):
             typing events for that room.
         """
 
-        sync_config = sync_result_builder.sync_config
-
         with Measure(self.clock, "ephemeral_by_room"):
             typing_key = since_token.typing_key if since_token else "0"
 
-            room_ids = sync_result_builder.joined_room_ids
+            room_ids = yield self.store.get_rooms_for_user(sync_config.user.to_string())
 
             typing_source = self.event_sources.sources["typing"]
             typing, typing_key = yield typing_source.get_new_events(
@@ -567,22 +565,10 @@ class SyncHandler(object):
         # Always use the `now_token` in `SyncResultBuilder`
         now_token = yield self.event_sources.get_current_token()
 
-        user_id = sync_config.user.to_string()
-        app_service = self.store.get_app_service_by_user_id(user_id)
-        if app_service:
-            # We no longer support AS users using /sync directly.
-            # See https://github.com/matrix-org/matrix-doc/issues/1144
-            raise NotImplementedError()
-        else:
-            joined_room_ids = yield self.get_rooms_for_user_at(
-                user_id, now_token.room_stream_id,
-            )
-
         sync_result_builder = SyncResultBuilder(
             sync_config, full_state,
             since_token=since_token,
             now_token=now_token,
-            joined_room_ids=joined_room_ids,
         )
 
         account_data_by_room = yield self._generate_sync_entry_for_account_data(
@@ -599,7 +585,7 @@ class SyncHandler(object):
             since_token is None and
             sync_config.filter_collection.blocks_all_presence()
         )
-        if not block_all_presence_data:
+        if False and not block_all_presence_data:
             yield self._generate_sync_entry_for_presence(
                 sync_result_builder, newly_joined_rooms, newly_joined_users
             )
@@ -617,6 +603,7 @@ class SyncHandler(object):
         device_id = sync_config.device_id
         one_time_key_counts = {}
         if device_id:
+            user_id = sync_config.user.to_string()
             one_time_key_counts = yield self.store.count_e2e_one_time_keys(
                 user_id, device_id
             )
@@ -904,7 +891,7 @@ class SyncHandler(object):
             ephemeral_by_room = {}
         else:
             now_token, ephemeral_by_room = yield self.ephemeral_by_room(
-                sync_result_builder,
+                sync_result_builder.sync_config,
                 now_token=sync_result_builder.now_token,
                 since_token=sync_result_builder.since_token,
             )
@@ -1009,8 +996,15 @@ class SyncHandler(object):
         if rooms_changed:
             defer.returnValue(True)
 
+        app_service = self.store.get_app_service_by_user_id(user_id)
+        if app_service:
+            rooms = yield self.store.get_app_service_rooms(app_service)
+            joined_room_ids = set(r.room_id for r in rooms)
+        else:
+            joined_room_ids = yield self.store.get_rooms_for_user(user_id)
+
         stream_id = RoomStreamToken.parse_stream_token(since_token.room_key).stream
-        for room_id in sync_result_builder.joined_room_ids:
+        for room_id in joined_room_ids:
             if self.store.has_room_changed_since(room_id, stream_id):
                 defer.returnValue(True)
         defer.returnValue(False)
@@ -1034,6 +1028,13 @@ class SyncHandler(object):
 
         assert since_token
 
+        app_service = self.store.get_app_service_by_user_id(user_id)
+        if app_service:
+            rooms = yield self.store.get_app_service_rooms(app_service)
+            joined_room_ids = set(r.room_id for r in rooms)
+        else:
+            joined_room_ids = yield self.store.get_rooms_for_user(user_id)
+
         # Get a list of membership change events that have happened.
         rooms_changed = yield self.store.get_membership_changes_for_user(
             user_id, since_token.room_key, now_token.room_key
@@ -1056,7 +1057,7 @@ class SyncHandler(object):
             # we do send down the room, and with full state, where necessary
 
             old_state_ids = None
-            if room_id in sync_result_builder.joined_room_ids and non_joins:
+            if room_id in joined_room_ids and non_joins:
                 # Always include if the user (re)joined the room, especially
                 # important so that device list changes are calculated correctly.
                 # If there are non join member events, but we are still in the room,
@@ -1066,7 +1067,7 @@ class SyncHandler(object):
                 # User is in the room so we don't need to do the invite/leave checks
                 continue
 
-            if room_id in sync_result_builder.joined_room_ids or has_join:
+            if room_id in joined_room_ids or has_join:
                 old_state_ids = yield self.get_state_at(room_id, since_token)
                 old_mem_ev_id = old_state_ids.get((EventTypes.Member, user_id), None)
                 old_mem_ev = None
@@ -1078,7 +1079,7 @@ class SyncHandler(object):
                     newly_joined_rooms.append(room_id)
 
             # If user is in the room then we don't need to do the invite/leave checks
-            if room_id in sync_result_builder.joined_room_ids:
+            if room_id in joined_room_ids:
                 continue
 
             if not non_joins:
@@ -1145,7 +1146,7 @@ class SyncHandler(object):
 
         # Get all events for rooms we're currently joined to.
         room_to_events = yield self.store.get_room_events_stream_for_rooms(
-            room_ids=sync_result_builder.joined_room_ids,
+            room_ids=joined_room_ids,
             from_key=since_token.room_key,
             to_key=now_token.room_key,
             limit=timeline_limit + 1,
@@ -1153,7 +1154,7 @@ class SyncHandler(object):
 
         # We loop through all room ids, even if there are no new events, in case
         # there are non room events taht we need to notify about.
-        for room_id in sync_result_builder.joined_room_ids:
+        for room_id in joined_room_ids:
             room_entry = room_to_events.get(room_id, None)
 
             if room_entry:
@@ -1361,54 +1362,6 @@ class SyncHandler(object):
         else:
             raise Exception("Unrecognized rtype: %r", room_builder.rtype)
 
-    @defer.inlineCallbacks
-    def get_rooms_for_user_at(self, user_id, stream_ordering):
-        """Get set of joined rooms for a user at the given stream ordering.
-
-        The stream ordering *must* be recent, otherwise this may throw an
-        exception if older than a month. (This function is called with the
-        current token, which should be perfectly fine).
-
-        Args:
-            user_id (str)
-            stream_ordering (int)
-
-        ReturnValue:
-            Deferred[frozenset[str]]: Set of room_ids the user is in at given
-            stream_ordering.
-        """
-        joined_rooms = yield self.store.get_rooms_for_user_with_stream_ordering(
-            user_id,
-        )
-
-        joined_room_ids = set()
-
-        # We need to check that the stream ordering of the join for each room
-        # is before the stream_ordering asked for. This might not be the case
-        # if the user joins a room between us getting the current token and
-        # calling `get_rooms_for_user_with_stream_ordering`.
-        # If the membership's stream ordering is after the given stream
-        # ordering, we need to go and work out if the user was in the room
-        # before.
-        for room_id, membership_stream_ordering in joined_rooms:
-            if membership_stream_ordering <= stream_ordering:
-                joined_room_ids.add(room_id)
-                continue
-
-            logger.info("User joined room after current token: %s", room_id)
-
-            extrems = yield self.store.get_forward_extremeties_for_room(
-                room_id, stream_ordering,
-            )
-            users_in_room = yield self.state.get_current_user_in_room(
-                room_id, extrems,
-            )
-            if user_id in users_in_room:
-                joined_room_ids.add(room_id)
-
-        joined_room_ids = frozenset(joined_room_ids)
-        defer.returnValue(joined_room_ids)
-
 
 def _action_has_highlight(actions):
     for action in actions:
@@ -1458,8 +1411,7 @@ def _calculate_state(timeline_contains, timeline_start, previous, current):
 
 class SyncResultBuilder(object):
     "Used to help build up a new SyncResult for a user"
-    def __init__(self, sync_config, full_state, since_token, now_token,
-                 joined_room_ids):
+    def __init__(self, sync_config, full_state, since_token, now_token):
         """
         Args:
             sync_config(SyncConfig)
@@ -1471,7 +1423,6 @@ class SyncResultBuilder(object):
         self.full_state = full_state
         self.since_token = since_token
         self.now_token = now_token
-        self.joined_room_ids = joined_room_ids
 
         self.presence = []
         self.account_data = []

synapse/handlers/typing.py

@@ -56,7 +56,7 @@ class TypingHandler(object):
 
         self.federation = hs.get_federation_sender()
 
-        hs.get_federation_registry().register_edu_handler("m.typing", self._recv_edu)
+        hs.get_replication_layer().register_edu_handler("m.typing", self._recv_edu)
 
         hs.get_distributor().observe("user_left_room", self.user_left_room)
 

synapse/http/matrixfederationclient.py

@@ -286,8 +286,7 @@ class MatrixFederationHttpClient(object):
         headers_dict[b"Authorization"] = auth_headers
 
     @defer.inlineCallbacks
-    def put_json(self, destination, path, args={}, data={},
-                 json_data_callback=None,
+    def put_json(self, destination, path, data={}, json_data_callback=None,
                  long_retries=False, timeout=None,
                  ignore_backoff=False,
                  backoff_on_404=False):
@@ -297,7 +296,6 @@ class MatrixFederationHttpClient(object):
             destination (str): The remote server to send the HTTP request
                 to.
             path (str): The HTTP path.
-            args (dict): query params
             data (dict): A dict containing the data that will be used as
                 the request body. This will be encoded as JSON.
             json_data_callback (callable): A callable returning the dict to
@@ -344,7 +342,6 @@ class MatrixFederationHttpClient(object):
             path,
             body_callback=body_callback,
             headers_dict={"Content-Type": ["application/json"]},
-            query_bytes=encode_query_args(args),
             long_retries=long_retries,
             timeout=timeout,
             ignore_backoff=ignore_backoff,
@@ -376,7 +373,6 @@ class MatrixFederationHttpClient(object):
                 giving up. None indicates no timeout.
             ignore_backoff (bool): true to ignore the historical backoff data and
                 try the request anyway.
-            args (dict): query params
         Returns:
             Deferred: Succeeds when we get a 2xx HTTP response. The result
             will be the decoded JSON body.

synapse/http/server.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -37,7 +36,7 @@ from twisted.web.util import redirectTo
 import collections
 import logging
 import urllib
-import simplejson
+import ujson
 
 logger = logging.getLogger(__name__)
 
@@ -60,11 +59,6 @@ response_count = metrics.register_counter(
     )
 )
 
-requests_counter = metrics.register_counter(
-    "requests_received",
-    labels=["method", "servlet", ],
-)
-
 outgoing_responses_counter = metrics.register_counter(
     "responses",
     labels=["method", "code"],
@@ -113,11 +107,6 @@ response_db_sched_duration = metrics.register_counter(
     "response_db_sched_duration_seconds", labels=["method", "servlet", "tag"]
 )
 
-# size in bytes of the response written
-response_size = metrics.register_counter(
-    "response_size", labels=["method", "servlet", "tag"]
-)
-
 _next_request_id = 0
 
 
@@ -156,8 +145,7 @@ def wrap_request_handler(request_handler, include_metrics=False):
                 # at the servlet name. For most requests that name will be
                 # JsonResource (or a subclass), and JsonResource._async_render
                 # will update it once it picks a servlet.
-                servlet_name = self.__class__.__name__
-                request_metrics.start(self.clock, name=servlet_name)
+                request_metrics.start(self.clock, name=self.__class__.__name__)
 
                 request_context.request = request_id
                 with request.processing():
@@ -166,7 +154,6 @@ def wrap_request_handler(request_handler, include_metrics=False):
                             if include_metrics:
                                 yield request_handler(self, request, request_metrics)
                             else:
-                                requests_counter.inc(request.method, servlet_name)
                                 yield request_handler(self, request)
                     except CodeMessageException as e:
                         code = e.code
@@ -242,7 +229,7 @@ class JsonResource(HttpServer, resource.Resource):
     """ This implements the HttpServer interface and provides JSON support for
     Resources.
 
-    Register callbacks via register_paths()
+    Register callbacks via register_path()
 
     Callbacks can return a tuple of status code and a dict in which case the
     the dict will automatically be sent to the client as a JSON object.
@@ -289,59 +276,49 @@ class JsonResource(HttpServer, resource.Resource):
             This checks if anyone has registered a callback for that method and
             path.
         """
-        callback, group_dict = self._get_handler_for_request(request)
-
-        servlet_instance = getattr(callback, "__self__", None)
-        if servlet_instance is not None:
-            servlet_classname = servlet_instance.__class__.__name__
-        else:
-            servlet_classname = "%r" % callback
-
-        request_metrics.name = servlet_classname
-        requests_counter.inc(request.method, servlet_classname)
-
-        # Now trigger the callback. If it returns a response, we send it
-        # here. If it throws an exception, that is handled by the wrapper
-        # installed by @request_handler.
-
-        kwargs = intern_dict({
-            name: urllib.unquote(value).decode("UTF-8") if value else value
-            for name, value in group_dict.items()
-        })
-
-        callback_return = yield callback(request, **kwargs)
-        if callback_return is not None:
-            code, response = callback_return
-            self._send_response(request, code, response)
-
-    def _get_handler_for_request(self, request):
-        """Finds a callback method to handle the given request
-
-        Args:
-            request (twisted.web.http.Request):
-
-        Returns:
-            Tuple[Callable, dict[str, str]]: callback method, and the dict
-                mapping keys to path components as specified in the handler's
-                path match regexp.
-
-                The callback will normally be a method registered via
-                register_paths, so will return (possibly via Deferred) either
-                None, or a tuple of (http code, response body).
-        """
         if request.method == "OPTIONS":
-            return _options_handler, {}
+            self._send_response(request, 200, {})
+            return
 
         # Loop through all the registered callbacks to check if the method
         # and path regex match
         for path_entry in self.path_regexs.get(request.method, []):
             m = path_entry.pattern.match(request.path)
-            if m:
-                # We found a match!
-                return path_entry.callback, m.groupdict()
+            if not m:
+                continue
+
+            # We found a match! First update the metrics object to indicate
+            # which servlet is handling the request.
+
+            callback = path_entry.callback
+
+            servlet_instance = getattr(callback, "__self__", None)
+            if servlet_instance is not None:
+                servlet_classname = servlet_instance.__class__.__name__
+            else:
+                servlet_classname = "%r" % callback
+
+            request_metrics.name = servlet_classname
+
+            # Now trigger the callback. If it returns a response, we send it
+            # here. If it throws an exception, that is handled by the wrapper
+            # installed by @request_handler.
+
+            kwargs = intern_dict({
+                name: urllib.unquote(value).decode("UTF-8") if value else value
+                for name, value in m.groupdict().items()
+            })
+
+            callback_return = yield callback(request, **kwargs)
+            if callback_return is not None:
+                code, response = callback_return
+                self._send_response(request, code, response)
+
+            return
 
         # Huh. No one wanted to handle that? Fiiiiiine. Send 400.
-        return _unrecognised_request_handler, {}
+        request_metrics.name = self.__class__.__name__ + ".UnrecognizedRequest"
+        raise UnrecognizedRequestError()
 
     def _send_response(self, request, code, response_json_object,
                        response_code_message=None):
@@ -358,34 +335,6 @@ class JsonResource(HttpServer, resource.Resource):
         )
 
 
-def _options_handler(request):
-    """Request handler for OPTIONS requests
-
-    This is a request handler suitable for return from
-    _get_handler_for_request. It returns a 200 and an empty body.
-
-    Args:
-        request (twisted.web.http.Request):
-
-    Returns:
-        Tuple[int, dict]: http code, response body.
-    """
-    return 200, {}
-
-
-def _unrecognised_request_handler(request):
-    """Request handler for unrecognised requests
-
-    This is a request handler suitable for return from
-    _get_handler_for_request. It actually just raises an
-    UnrecognizedRequestError.
-
-    Args:
-        request (twisted.web.http.Request):
-    """
-    raise UnrecognizedRequestError()
-
-
 class RequestMetrics(object):
     def start(self, clock, name):
         self.start = clock.time_msec()
@@ -431,8 +380,6 @@ class RequestMetrics(object):
             context.db_sched_duration_ms / 1000., request.method, self.name, tag
         )
 
-        response_size.inc_by(request.sentLength, request.method, self.name, tag)
-
 
 class RootRedirect(resource.Resource):
     """Redirects the root '/' path to another path."""
@@ -468,7 +415,8 @@ def respond_with_json(request, code, json_object, send_cors=False,
         if canonical_json or synapse.events.USE_FROZEN_DICTS:
             json_bytes = encode_canonical_json(json_object)
         else:
-            json_bytes = simplejson.dumps(json_object)
+            # ujson doesn't like frozen_dicts.
+            json_bytes = ujson.dumps(json_object, ensure_ascii=False)
 
     return respond_with_json_bytes(
         request, code, json_bytes,
@@ -495,7 +443,6 @@ def respond_with_json_bytes(request, code, json_bytes, send_cors=False,
     request.setHeader(b"Content-Type", b"application/json")
     request.setHeader(b"Server", version_string)
     request.setHeader(b"Content-Length", b"%d" % (len(json_bytes),))
-    request.setHeader(b"Cache-Control", b"no-cache, no-store, must-revalidate")
 
     if send_cors:
         set_cors_headers(request)

synapse/metrics/__init__.py

@@ -57,31 +57,15 @@ class Metrics(object):
         return metric
 
     def register_counter(self, *args, **kwargs):
-        """
-        Returns:
-            CounterMetric
-        """
         return self._register(CounterMetric, *args, **kwargs)
 
     def register_callback(self, *args, **kwargs):
-        """
-        Returns:
-            CallbackMetric
-        """
         return self._register(CallbackMetric, *args, **kwargs)
 
     def register_distribution(self, *args, **kwargs):
-        """
-        Returns:
-            DistributionMetric
-        """
         return self._register(DistributionMetric, *args, **kwargs)
 
     def register_cache(self, *args, **kwargs):
-        """
-        Returns:
-            CacheMetric
-        """
         return self._register(CacheMetric, *args, **kwargs)
 
 

synapse/push/action_generator.py

@@ -40,6 +40,10 @@ class ActionGenerator(object):
     @defer.inlineCallbacks
     def handle_push_actions_for_event(self, event, context):
         with Measure(self.clock, "action_for_event_by_user"):
-            yield self.bulk_evaluator.action_for_event_by_user(
+            actions_by_user = yield self.bulk_evaluator.action_for_event_by_user(
                 event, context
             )
+
+        context.push_actions = [
+            (uid, actions) for uid, actions in actions_by_user.iteritems()
+        ]

synapse/push/bulk_push_rule_evaluator.py

@@ -137,11 +137,11 @@ class BulkPushRuleEvaluator(object):
 
     @defer.inlineCallbacks
     def action_for_event_by_user(self, event, context):
-        """Given an event and context, evaluate the push rules and insert the
-        results into the event_push_actions_staging table.
+        """Given an event and context, evaluate the push rules and return
+        the results
 
         Returns:
-            Deferred
+            dict of user_id -> action
         """
         rules_by_user = yield self._get_rules_for_event(event, context)
         actions_by_user = {}
@@ -190,16 +190,9 @@ class BulkPushRuleEvaluator(object):
                 if matches:
                     actions = [x for x in rule['actions'] if x != 'dont_notify']
                     if actions and 'notify' in actions:
-                        # Push rules say we should notify the user of this event
                         actions_by_user[uid] = actions
                     break
-
-        # Mark in the DB staging area the push actions for users who should be
-        # notified for this event. (This will then get handled when we persist
-        # the event)
-        yield self.store.add_push_actions_to_staging(
-            event.event_id, actions_by_user,
-        )
+        defer.returnValue(actions_by_user)
 
 
 def _condition_checker(evaluator, conditions, uid, display_name, cache):

synapse/python_dependencies.py

@@ -24,16 +24,17 @@ REQUIREMENTS = {
     "unpaddedbase64>=1.1.0": ["unpaddedbase64>=1.1.0"],
     "canonicaljson>=1.0.0": ["canonicaljson>=1.0.0"],
     "signedjson>=1.0.0": ["signedjson>=1.0.0"],
-    "pynacl>=1.2.1": ["nacl>=1.2.1", "nacl.bindings"],
+    "pynacl==0.3.0": ["nacl==0.3.0", "nacl.bindings"],
     "service_identity>=1.0.0": ["service_identity>=1.0.0"],
     "Twisted>=16.0.0": ["twisted>=16.0.0"],
     "pyopenssl>=0.14": ["OpenSSL>=0.14"],
     "pyyaml": ["yaml"],
     "pyasn1": ["pyasn1"],
     "daemonize": ["daemonize"],
-    "bcrypt": ["bcrypt>=3.1.0"],
+    "bcrypt": ["bcrypt"],
     "pillow": ["PIL"],
     "pydenticon": ["pydenticon"],
+    "ujson": ["ujson"],
     "blist": ["blist"],
     "pysaml2>=3.0.0": ["saml2>=3.0.0"],
     "pymacaroons-pynacl": ["pymacaroons"],

synapse/replication/http/__init__.py

@@ -13,8 +13,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
+import send_event
+
 from synapse.http.server import JsonResource
-from synapse.replication.http import membership, send_event
 
 
 REPLICATION_PREFIX = "/_synapse/replication"
@@ -27,4 +29,3 @@ class ReplicationRestResource(JsonResource):
 
     def register_servlets(self, hs):
         send_event.register_servlets(hs, self)
-        membership.register_servlets(hs, self)

synapse/replication/http/membership.py

@@ -1,334 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2018 New Vector 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.
-
-import logging
-import re
-
-from twisted.internet import defer
-
-from synapse.api.errors import SynapseError, MatrixCodeMessageException
-from synapse.http.servlet import RestServlet, parse_json_object_from_request
-from synapse.types import Requester, UserID
-from synapse.util.distributor import user_left_room, user_joined_room
-
-logger = logging.getLogger(__name__)
-
-
-@defer.inlineCallbacks
-def remote_join(client, host, port, requester, remote_room_hosts,
-                room_id, user_id, content):
-    """Ask the master to do a remote join for the given user to the given room
-
-    Args:
-        client (SimpleHttpClient)
-        host (str): host of master
-        port (int): port on master listening for HTTP replication
-        requester (Requester)
-        remote_room_hosts (list[str]): Servers to try and join via
-        room_id (str)
-        user_id (str)
-        content (dict): The event content to use for the join event
-
-    Returns:
-        Deferred
-    """
-    uri = "http://%s:%s/_synapse/replication/remote_join" % (host, port)
-
-    payload = {
-        "requester": requester.serialize(),
-        "remote_room_hosts": remote_room_hosts,
-        "room_id": room_id,
-        "user_id": user_id,
-        "content": content,
-    }
-
-    try:
-        result = yield client.post_json_get_json(uri, payload)
-    except MatrixCodeMessageException as e:
-        # We convert to SynapseError as we know that it was a SynapseError
-        # on the master process that we should send to the client. (And
-        # importantly, not stack traces everywhere)
-        raise SynapseError(e.code, e.msg, e.errcode)
-    defer.returnValue(result)
-
-
-@defer.inlineCallbacks
-def remote_reject_invite(client, host, port, requester, remote_room_hosts,
-                         room_id, user_id):
-    """Ask master to reject the invite for the user and room.
-
-    Args:
-        client (SimpleHttpClient)
-        host (str): host of master
-        port (int): port on master listening for HTTP replication
-        requester (Requester)
-        remote_room_hosts (list[str]): Servers to try and reject via
-        room_id (str)
-        user_id (str)
-
-    Returns:
-        Deferred
-    """
-    uri = "http://%s:%s/_synapse/replication/remote_reject_invite" % (host, port)
-
-    payload = {
-        "requester": requester.serialize(),
-        "remote_room_hosts": remote_room_hosts,
-        "room_id": room_id,
-        "user_id": user_id,
-    }
-
-    try:
-        result = yield client.post_json_get_json(uri, payload)
-    except MatrixCodeMessageException as e:
-        # We convert to SynapseError as we know that it was a SynapseError
-        # on the master process that we should send to the client. (And
-        # importantly, not stack traces everywhere)
-        raise SynapseError(e.code, e.msg, e.errcode)
-    defer.returnValue(result)
-
-
-@defer.inlineCallbacks
-def get_or_register_3pid_guest(client, host, port, requester,
-                               medium, address, inviter_user_id):
-    """Ask the master to get/create a guest account for given 3PID.
-
-    Args:
-        client (SimpleHttpClient)
-        host (str): host of master
-        port (int): port on master listening for HTTP replication
-        requester (Requester)
-        medium (str)
-        address (str)
-        inviter_user_id (str): The user ID who is trying to invite the
-            3PID
-
-    Returns:
-        Deferred[(str, str)]: A 2-tuple of `(user_id, access_token)` of the
-        3PID guest account.
-    """
-
-    uri = "http://%s:%s/_synapse/replication/get_or_register_3pid_guest" % (host, port)
-
-    payload = {
-        "requester": requester.serialize(),
-        "medium": medium,
-        "address": address,
-        "inviter_user_id": inviter_user_id,
-    }
-
-    try:
-        result = yield client.post_json_get_json(uri, payload)
-    except MatrixCodeMessageException as e:
-        # We convert to SynapseError as we know that it was a SynapseError
-        # on the master process that we should send to the client. (And
-        # importantly, not stack traces everywhere)
-        raise SynapseError(e.code, e.msg, e.errcode)
-    defer.returnValue(result)
-
-
-@defer.inlineCallbacks
-def notify_user_membership_change(client, host, port, user_id, room_id, change):
-    """Notify master that a user has joined or left the room
-
-    Args:
-        client (SimpleHttpClient)
-        host (str): host of master
-        port (int): port on master listening for HTTP replication.
-        user_id (str)
-        room_id (str)
-        change (str): Either "join" or "left"
-
-    Returns:
-        Deferred
-    """
-    assert change in ("joined", "left")
-
-    uri = "http://%s:%s/_synapse/replication/user_%s_room" % (host, port, change)
-
-    payload = {
-        "user_id": user_id,
-        "room_id": room_id,
-    }
-
-    try:
-        result = yield client.post_json_get_json(uri, payload)
-    except MatrixCodeMessageException as e:
-        # We convert to SynapseError as we know that it was a SynapseError
-        # on the master process that we should send to the client. (And
-        # importantly, not stack traces everywhere)
-        raise SynapseError(e.code, e.msg, e.errcode)
-    defer.returnValue(result)
-
-
-class ReplicationRemoteJoinRestServlet(RestServlet):
-    PATTERNS = [re.compile("^/_synapse/replication/remote_join$")]
-
-    def __init__(self, hs):
-        super(ReplicationRemoteJoinRestServlet, self).__init__()
-
-        self.federation_handler = hs.get_handlers().federation_handler
-        self.store = hs.get_datastore()
-        self.clock = hs.get_clock()
-
-    @defer.inlineCallbacks
-    def on_POST(self, request):
-        content = parse_json_object_from_request(request)
-
-        remote_room_hosts = content["remote_room_hosts"]
-        room_id = content["room_id"]
-        user_id = content["user_id"]
-        event_content = content["content"]
-
-        requester = Requester.deserialize(self.store, content["requester"])
-
-        if requester.user:
-            request.authenticated_entity = requester.user.to_string()
-
-        logger.info(
-            "remote_join: %s into room: %s",
-            user_id, room_id,
-        )
-
-        yield self.federation_handler.do_invite_join(
-            remote_room_hosts,
-            room_id,
-            user_id,
-            event_content,
-        )
-
-        defer.returnValue((200, {}))
-
-
-class ReplicationRemoteRejectInviteRestServlet(RestServlet):
-    PATTERNS = [re.compile("^/_synapse/replication/remote_reject_invite$")]
-
-    def __init__(self, hs):
-        super(ReplicationRemoteRejectInviteRestServlet, self).__init__()
-
-        self.federation_handler = hs.get_handlers().federation_handler
-        self.store = hs.get_datastore()
-        self.clock = hs.get_clock()
-
-    @defer.inlineCallbacks
-    def on_POST(self, request):
-        content = parse_json_object_from_request(request)
-
-        remote_room_hosts = content["remote_room_hosts"]
-        room_id = content["room_id"]
-        user_id = content["user_id"]
-
-        requester = Requester.deserialize(self.store, content["requester"])
-
-        if requester.user:
-            request.authenticated_entity = requester.user.to_string()
-
-        logger.info(
-            "remote_reject_invite: %s out of room: %s",
-            user_id, room_id,
-        )
-
-        try:
-            event = yield self.federation_handler.do_remotely_reject_invite(
-                remote_room_hosts,
-                room_id,
-                user_id,
-            )
-            ret = event.get_pdu_json()
-        except Exception as e:
-            # if we were unable to reject the exception, just mark
-            # it as rejected on our end and plough ahead.
-            #
-            # The 'except' clause is very broad, but we need to
-            # capture everything from DNS failures upwards
-            #
-            logger.warn("Failed to reject invite: %s", e)
-
-            yield self.store.locally_reject_invite(
-                user_id, room_id
-            )
-            ret = {}
-
-        defer.returnValue((200, ret))
-
-
-class ReplicationRegister3PIDGuestRestServlet(RestServlet):
-    PATTERNS = [re.compile("^/_synapse/replication/get_or_register_3pid_guest$")]
-
-    def __init__(self, hs):
-        super(ReplicationRegister3PIDGuestRestServlet, self).__init__()
-
-        self.registeration_handler = hs.get_handlers().registration_handler
-        self.store = hs.get_datastore()
-        self.clock = hs.get_clock()
-
-    @defer.inlineCallbacks
-    def on_POST(self, request):
-        content = parse_json_object_from_request(request)
-
-        medium = content["medium"]
-        address = content["address"]
-        inviter_user_id = content["inviter_user_id"]
-
-        requester = Requester.deserialize(self.store, content["requester"])
-
-        if requester.user:
-            request.authenticated_entity = requester.user.to_string()
-
-        logger.info("get_or_register_3pid_guest: %r", content)
-
-        ret = yield self.registeration_handler.get_or_register_3pid_guest(
-            medium, address, inviter_user_id,
-        )
-
-        defer.returnValue((200, ret))
-
-
-class ReplicationUserJoinedLeftRoomRestServlet(RestServlet):
-    PATTERNS = [re.compile("^/_synapse/replication/user_(?P<change>joined|left)_room$")]
-
-    def __init__(self, hs):
-        super(ReplicationUserJoinedLeftRoomRestServlet, self).__init__()
-
-        self.registeration_handler = hs.get_handlers().registration_handler
-        self.store = hs.get_datastore()
-        self.clock = hs.get_clock()
-        self.distributor = hs.get_distributor()
-
-    def on_POST(self, request, change):
-        content = parse_json_object_from_request(request)
-
-        user_id = content["user_id"]
-        room_id = content["room_id"]
-
-        logger.info("user membership change: %s in %s", user_id, room_id)
-
-        user = UserID.from_string(user_id)
-
-        if change == "joined":
-            user_joined_room(self.distributor, user, room_id)
-        elif change == "left":
-            user_left_room(self.distributor, user, room_id)
-        else:
-            raise Exception("Unrecognized change: %r", change)
-
-        return (200, {})
-
-
-def register_servlets(hs, http_server):
-    ReplicationRemoteJoinRestServlet(hs).register(http_server)
-    ReplicationRemoteRejectInviteRestServlet(hs).register(http_server)
-    ReplicationRegister3PIDGuestRestServlet(hs).register(http_server)
-    ReplicationUserJoinedLeftRoomRestServlet(hs).register(http_server)

synapse/replication/http/send_event.py

@@ -15,17 +15,11 @@
 
 from twisted.internet import defer
 
-from synapse.api.errors import (
-    SynapseError, MatrixCodeMessageException, CodeMessageException,
-)
 from synapse.events import FrozenEvent
 from synapse.events.snapshot import EventContext
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
-from synapse.util.async import sleep
-from synapse.util.caches.response_cache import ResponseCache
-from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
 from synapse.util.metrics import Measure
-from synapse.types import Requester, UserID
+from synapse.types import Requester
 
 import logging
 import re
@@ -33,9 +27,7 @@ import re
 logger = logging.getLogger(__name__)
 
 
-@defer.inlineCallbacks
-def send_event_to_master(client, host, port, requester, event, context,
-                         ratelimit, extra_users):
+def send_event_to_master(client, host, port, requester, event, context):
     """Send event to be handled on the master
 
     Args:
@@ -45,46 +37,18 @@ def send_event_to_master(client, host, port, requester, event, context,
         requester (Requester)
         event (FrozenEvent)
         context (EventContext)
-        ratelimit (bool)
-        extra_users (list(UserID)): Any extra users to notify about event
     """
-    uri = "http://%s:%s/_synapse/replication/send_event/%s" % (
-        host, port, event.event_id,
-    )
+    uri = "http://%s:%s/_synapse/replication/send_event" % (host, port,)
 
     payload = {
         "event": event.get_pdu_json(),
         "internal_metadata": event.internal_metadata.get_dict(),
         "rejected_reason": event.rejected_reason,
-        "context": context.serialize(event),
+        "context": context.serialize(),
         "requester": requester.serialize(),
-        "ratelimit": ratelimit,
-        "extra_users": [u.to_string() for u in extra_users],
     }
 
-    try:
-        # We keep retrying the same request for timeouts. This is so that we
-        # have a good idea that the request has either succeeded or failed on
-        # the master, and so whether we should clean up or not.
-        while True:
-            try:
-                result = yield client.put_json(uri, payload)
-                break
-            except CodeMessageException as e:
-                if e.code != 504:
-                    raise
-
-            logger.warn("send_event request timed out")
-
-            # If we timed out we probably don't need to worry about backing
-            # off too much, but lets just wait a little anyway.
-            yield sleep(1)
-    except MatrixCodeMessageException as e:
-        # We convert to SynapseError as we know that it was a SynapseError
-        # on the master process that we should send to the client. (And
-        # importantly, not stack traces everywhere)
-        raise SynapseError(e.code, e.msg, e.errcode)
-    defer.returnValue(result)
+    return client.post_json_get_json(uri, payload)
 
 
 class ReplicationSendEventRestServlet(RestServlet):
@@ -93,7 +57,7 @@ class ReplicationSendEventRestServlet(RestServlet):
 
     The API looks like:
 
-        POST /_synapse/replication/send_event/:event_id
+        POST /_synapse/replication/send_event
 
         {
             "event": { .. serialized event .. },
@@ -101,11 +65,9 @@ class ReplicationSendEventRestServlet(RestServlet):
             "rejected_reason": ..,   // The event.rejected_reason field
             "context": { .. serialized event context .. },
             "requester": { .. serialized requester .. },
-            "ratelimit": true,
-            "extra_users": [],
         }
     """
-    PATTERNS = [re.compile("^/_synapse/replication/send_event/(?P<event_id>[^/]+)$")]
+    PATTERNS = [re.compile("^/_synapse/replication/send_event$")]
 
     def __init__(self, hs):
         super(ReplicationSendEventRestServlet, self).__init__()
@@ -114,23 +76,8 @@ class ReplicationSendEventRestServlet(RestServlet):
         self.store = hs.get_datastore()
         self.clock = hs.get_clock()
 
-        # The responses are tiny, so we may as well cache them for a while
-        self.response_cache = ResponseCache(hs, timeout_ms=30 * 60 * 1000)
-
-    def on_PUT(self, request, event_id):
-        result = self.response_cache.get(event_id)
-        if not result:
-            result = self.response_cache.set(
-                event_id,
-                self._handle_request(request)
-            )
-        else:
-            logger.warn("Returning cached response")
-        return make_deferred_yieldable(result)
-
-    @preserve_fn
     @defer.inlineCallbacks
-    def _handle_request(self, request):
+    def on_POST(self, request):
         with Measure(self.clock, "repl_send_event_parse"):
             content = parse_json_object_from_request(request)
 
@@ -140,10 +87,7 @@ class ReplicationSendEventRestServlet(RestServlet):
             event = FrozenEvent(event_dict, internal_metadata, rejected_reason)
 
             requester = Requester.deserialize(self.store, content["requester"])
-            context = yield EventContext.deserialize(self.store, content["context"])
-
-            ratelimit = content["ratelimit"]
-            extra_users = [UserID.from_string(u) for u in content["extra_users"]]
+            context = EventContext.deserialize(self.store, content["context"])
 
         if requester.user:
             request.authenticated_entity = requester.user.to_string()
@@ -153,10 +97,8 @@ class ReplicationSendEventRestServlet(RestServlet):
             event.event_id, event.room_id,
         )
 
-        yield self.event_creation_handler.persist_and_notify_client_event(
+        yield self.event_creation_handler.handle_new_client_event(
             requester, event, context,
-            ratelimit=ratelimit,
-            extra_users=extra_users,
         )
 
         defer.returnValue((200, {}))

synapse/replication/slave/storage/account_data.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,20 +13,50 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.replication.slave.storage._base import BaseSlavedStore
-from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
-from synapse.storage.account_data import AccountDataWorkerStore
-from synapse.storage.tags import TagsWorkerStore
+from ._base import BaseSlavedStore
+from ._slaved_id_tracker import SlavedIdTracker
+from synapse.storage import DataStore
+from synapse.storage.account_data import AccountDataStore
+from synapse.storage.tags import TagsStore
+from synapse.util.caches.stream_change_cache import StreamChangeCache
 
 
-class SlavedAccountDataStore(TagsWorkerStore, AccountDataWorkerStore, BaseSlavedStore):
+class SlavedAccountDataStore(BaseSlavedStore):
 
     def __init__(self, db_conn, hs):
+        super(SlavedAccountDataStore, self).__init__(db_conn, hs)
         self._account_data_id_gen = SlavedIdTracker(
             db_conn, "account_data_max_stream_id", "stream_id",
         )
+        self._account_data_stream_cache = StreamChangeCache(
+            "AccountDataAndTagsChangeCache",
+            self._account_data_id_gen.get_current_token(),
+        )
 
-        super(SlavedAccountDataStore, self).__init__(db_conn, hs)
+    get_account_data_for_user = (
+        AccountDataStore.__dict__["get_account_data_for_user"]
+    )
+
+    get_global_account_data_by_type_for_users = (
+        AccountDataStore.__dict__["get_global_account_data_by_type_for_users"]
+    )
+
+    get_global_account_data_by_type_for_user = (
+        AccountDataStore.__dict__["get_global_account_data_by_type_for_user"]
+    )
+
+    get_tags_for_user = TagsStore.__dict__["get_tags_for_user"]
+    get_tags_for_room = (
+        DataStore.get_tags_for_room.__func__
+    )
+    get_account_data_for_room = (
+        DataStore.get_account_data_for_room.__func__
+    )
+
+    get_updated_tags = DataStore.get_updated_tags.__func__
+    get_updated_account_data_for_user = (
+        DataStore.get_updated_account_data_for_user.__func__
+    )
 
     def get_max_account_data_stream_id(self):
         return self._account_data_id_gen.get_current_token()
@@ -56,10 +85,6 @@ class SlavedAccountDataStore(TagsWorkerStore, AccountDataWorkerStore, BaseSlaved
                         (row.data_type, row.user_id,)
                     )
                 self.get_account_data_for_user.invalidate((row.user_id,))
-                self.get_account_data_for_room.invalidate((row.user_id, row.room_id,))
-                self.get_account_data_for_room_and_type.invalidate(
-                    (row.user_id, row.room_id, row.data_type,),
-                )
                 self._account_data_stream_cache.entity_has_changed(
                     row.user_id, token
                 )

synapse/replication/slave/storage/appservice.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2015, 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,11 +13,33 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.storage.appservice import (
-    ApplicationServiceWorkerStore, ApplicationServiceTransactionWorkerStore,
-)
+from ._base import BaseSlavedStore
+from synapse.storage import DataStore
+from synapse.config.appservice import load_appservices
+from synapse.storage.appservice import _make_exclusive_regex
 
 
-class SlavedApplicationServiceStore(ApplicationServiceTransactionWorkerStore,
-                                    ApplicationServiceWorkerStore):
-    pass
+class SlavedApplicationServiceStore(BaseSlavedStore):
+    def __init__(self, db_conn, hs):
+        super(SlavedApplicationServiceStore, self).__init__(db_conn, hs)
+        self.services_cache = load_appservices(
+            hs.config.server_name,
+            hs.config.app_service_config_files
+        )
+        self.exclusive_user_regex = _make_exclusive_regex(self.services_cache)
+
+    get_app_service_by_token = DataStore.get_app_service_by_token.__func__
+    get_app_service_by_user_id = DataStore.get_app_service_by_user_id.__func__
+    get_app_services = DataStore.get_app_services.__func__
+    get_new_events_for_appservice = DataStore.get_new_events_for_appservice.__func__
+    create_appservice_txn = DataStore.create_appservice_txn.__func__
+    get_appservices_by_state = DataStore.get_appservices_by_state.__func__
+    get_oldest_unsent_txn = DataStore.get_oldest_unsent_txn.__func__
+    _get_last_txn = DataStore._get_last_txn.__func__
+    complete_appservice_txn = DataStore.complete_appservice_txn.__func__
+    get_appservice_state = DataStore.get_appservice_state.__func__
+    set_appservice_last_pos = DataStore.set_appservice_last_pos.__func__
+    set_appservice_state = DataStore.set_appservice_state.__func__
+    get_if_app_services_interested_in_user = (
+        DataStore.get_if_app_services_interested_in_user.__func__
+    )

synapse/replication/slave/storage/client_ips.py

@@ -42,6 +42,8 @@ class SlavedClientIpStore(BaseSlavedStore):
         if last_seen is not None and (now - last_seen) < LAST_SEEN_GRANULARITY:
             return
 
+        self.client_ip_last_seen.prefill(key, now)
+
         self.hs.get_tcp_replication().send_user_ip(
             user_id, access_token, ip, user_agent, device_id, now
         )

synapse/replication/slave/storage/directory.py

@@ -14,8 +14,10 @@
 # limitations under the License.
 
 from ._base import BaseSlavedStore
-from synapse.storage.directory import DirectoryWorkerStore
+from synapse.storage.directory import DirectoryStore
 
 
-class DirectoryStore(DirectoryWorkerStore, BaseSlavedStore):
-    pass
+class DirectoryStore(BaseSlavedStore):
+    get_aliases_for_room = DirectoryStore.__dict__[
+        "get_aliases_for_room"
+    ]

synapse/replication/slave/storage/events.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,13 +15,14 @@
 import logging
 
 from synapse.api.constants import EventTypes
-from synapse.storage.event_federation import EventFederationWorkerStore
-from synapse.storage.event_push_actions import EventPushActionsWorkerStore
-from synapse.storage.events_worker import EventsWorkerStore
-from synapse.storage.roommember import RoomMemberWorkerStore
+from synapse.storage import DataStore
+from synapse.storage.event_federation import EventFederationStore
+from synapse.storage.event_push_actions import EventPushActionsStore
+from synapse.storage.roommember import RoomMemberStore
 from synapse.storage.state import StateGroupWorkerStore
-from synapse.storage.stream import StreamWorkerStore
-from synapse.storage.signatures import SignatureWorkerStore
+from synapse.storage.stream import StreamStore
+from synapse.storage.signatures import SignatureStore
+from synapse.util.caches.stream_change_cache import StreamChangeCache
 from ._base import BaseSlavedStore
 from ._slaved_id_tracker import SlavedIdTracker
 
@@ -38,33 +38,157 @@ logger = logging.getLogger(__name__)
 # the method descriptor on the DataStore and chuck them into our class.
 
 
-class SlavedEventStore(EventFederationWorkerStore,
-                       RoomMemberWorkerStore,
-                       EventPushActionsWorkerStore,
-                       StreamWorkerStore,
-                       EventsWorkerStore,
-                       StateGroupWorkerStore,
-                       SignatureWorkerStore,
-                       BaseSlavedStore):
+class SlavedEventStore(StateGroupWorkerStore, BaseSlavedStore):
 
     def __init__(self, db_conn, hs):
+        super(SlavedEventStore, self).__init__(db_conn, hs)
         self._stream_id_gen = SlavedIdTracker(
             db_conn, "events", "stream_ordering",
         )
         self._backfill_id_gen = SlavedIdTracker(
             db_conn, "events", "stream_ordering", step=-1
         )
+        events_max = self._stream_id_gen.get_current_token()
+        event_cache_prefill, min_event_val = self._get_cache_dict(
+            db_conn, "events",
+            entity_column="room_id",
+            stream_column="stream_ordering",
+            max_value=events_max,
+        )
+        self._events_stream_cache = StreamChangeCache(
+            "EventsRoomStreamChangeCache", min_event_val,
+            prefilled_cache=event_cache_prefill,
+        )
+        self._membership_stream_cache = StreamChangeCache(
+            "MembershipStreamChangeCache", events_max,
+        )
 
-        super(SlavedEventStore, self).__init__(db_conn, hs)
+        self.stream_ordering_month_ago = 0
+        self._stream_order_on_start = self.get_room_max_stream_ordering()
 
     # Cached functions can't be accessed through a class instance so we need
     # to reach inside the __dict__ to extract them.
+    get_rooms_for_user = RoomMemberStore.__dict__["get_rooms_for_user"]
+    get_users_in_room = RoomMemberStore.__dict__["get_users_in_room"]
+    get_hosts_in_room = RoomMemberStore.__dict__["get_hosts_in_room"]
+    get_users_who_share_room_with_user = (
+        RoomMemberStore.__dict__["get_users_who_share_room_with_user"]
+    )
+    get_latest_event_ids_in_room = EventFederationStore.__dict__[
+        "get_latest_event_ids_in_room"
+    ]
+    get_invited_rooms_for_user = RoomMemberStore.__dict__[
+        "get_invited_rooms_for_user"
+    ]
+    get_unread_event_push_actions_by_room_for_user = (
+        EventPushActionsStore.__dict__["get_unread_event_push_actions_by_room_for_user"]
+    )
+    _get_unread_counts_by_receipt_txn = (
+        DataStore._get_unread_counts_by_receipt_txn.__func__
+    )
+    _get_unread_counts_by_pos_txn = (
+        DataStore._get_unread_counts_by_pos_txn.__func__
+    )
+    get_recent_event_ids_for_room = (
+        StreamStore.__dict__["get_recent_event_ids_for_room"]
+    )
+    _get_joined_hosts_cache = RoomMemberStore.__dict__["_get_joined_hosts_cache"]
+    has_room_changed_since = DataStore.has_room_changed_since.__func__
 
-    def get_room_max_stream_ordering(self):
-        return self._stream_id_gen.get_current_token()
+    get_unread_push_actions_for_user_in_range_for_http = (
+        DataStore.get_unread_push_actions_for_user_in_range_for_http.__func__
+    )
+    get_unread_push_actions_for_user_in_range_for_email = (
+        DataStore.get_unread_push_actions_for_user_in_range_for_email.__func__
+    )
+    get_push_action_users_in_range = (
+        DataStore.get_push_action_users_in_range.__func__
+    )
+    get_event = DataStore.get_event.__func__
+    get_events = DataStore.get_events.__func__
+    get_rooms_for_user_where_membership_is = (
+        DataStore.get_rooms_for_user_where_membership_is.__func__
+    )
+    get_membership_changes_for_user = (
+        DataStore.get_membership_changes_for_user.__func__
+    )
+    get_room_events_max_id = DataStore.get_room_events_max_id.__func__
+    get_room_events_stream_for_room = (
+        DataStore.get_room_events_stream_for_room.__func__
+    )
+    get_events_around = DataStore.get_events_around.__func__
+    get_joined_users_from_state = DataStore.get_joined_users_from_state.__func__
+    get_joined_users_from_context = DataStore.get_joined_users_from_context.__func__
+    _get_joined_users_from_context = (
+        RoomMemberStore.__dict__["_get_joined_users_from_context"]
+    )
 
-    def get_room_min_stream_ordering(self):
-        return self._backfill_id_gen.get_current_token()
+    get_joined_hosts = DataStore.get_joined_hosts.__func__
+    _get_joined_hosts = RoomMemberStore.__dict__["_get_joined_hosts"]
+
+    get_recent_events_for_room = DataStore.get_recent_events_for_room.__func__
+    get_room_events_stream_for_rooms = (
+        DataStore.get_room_events_stream_for_rooms.__func__
+    )
+    is_host_joined = RoomMemberStore.__dict__["is_host_joined"]
+    get_stream_token_for_event = DataStore.get_stream_token_for_event.__func__
+
+    _set_before_and_after = staticmethod(DataStore._set_before_and_after)
+
+    _get_events = DataStore._get_events.__func__
+    _get_events_from_cache = DataStore._get_events_from_cache.__func__
+
+    _invalidate_get_event_cache = DataStore._invalidate_get_event_cache.__func__
+    _enqueue_events = DataStore._enqueue_events.__func__
+    _do_fetch = DataStore._do_fetch.__func__
+    _fetch_event_rows = DataStore._fetch_event_rows.__func__
+    _get_event_from_row = DataStore._get_event_from_row.__func__
+    _get_rooms_for_user_where_membership_is_txn = (
+        DataStore._get_rooms_for_user_where_membership_is_txn.__func__
+    )
+    _get_events_around_txn = DataStore._get_events_around_txn.__func__
+
+    get_backfill_events = DataStore.get_backfill_events.__func__
+    _get_backfill_events = DataStore._get_backfill_events.__func__
+    get_missing_events = DataStore.get_missing_events.__func__
+    _get_missing_events = DataStore._get_missing_events.__func__
+
+    get_auth_chain = DataStore.get_auth_chain.__func__
+    get_auth_chain_ids = DataStore.get_auth_chain_ids.__func__
+    _get_auth_chain_ids_txn = DataStore._get_auth_chain_ids_txn.__func__
+
+    get_room_max_stream_ordering = DataStore.get_room_max_stream_ordering.__func__
+
+    get_forward_extremeties_for_room = (
+        DataStore.get_forward_extremeties_for_room.__func__
+    )
+    _get_forward_extremeties_for_room = (
+        EventFederationStore.__dict__["_get_forward_extremeties_for_room"]
+    )
+
+    get_all_new_events_stream = DataStore.get_all_new_events_stream.__func__
+
+    get_federation_out_pos = DataStore.get_federation_out_pos.__func__
+    update_federation_out_pos = DataStore.update_federation_out_pos.__func__
+
+    get_latest_event_ids_and_hashes_in_room = (
+        DataStore.get_latest_event_ids_and_hashes_in_room.__func__
+    )
+    _get_latest_event_ids_and_hashes_in_room = (
+        DataStore._get_latest_event_ids_and_hashes_in_room.__func__
+    )
+    _get_event_reference_hashes_txn = (
+        DataStore._get_event_reference_hashes_txn.__func__
+    )
+    add_event_hashes = (
+        DataStore.add_event_hashes.__func__
+    )
+    get_event_reference_hashes = (
+        SignatureStore.__dict__["get_event_reference_hashes"]
+    )
+    get_event_reference_hash = (
+        SignatureStore.__dict__["get_event_reference_hash"]
+    )
 
     def stream_positions(self):
         result = super(SlavedEventStore, self).stream_positions()

synapse/replication/slave/storage/profile.py

@@ -1,21 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2018 New Vector Ltd
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-from synapse.replication.slave.storage._base import BaseSlavedStore
-from synapse.storage.profile import ProfileWorkerStore
-
-
-class SlavedProfileStore(ProfileWorkerStore, BaseSlavedStore):
-    pass

synapse/replication/slave/storage/push_rule.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2015, 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,15 +15,29 @@
 
 from .events import SlavedEventStore
 from ._slaved_id_tracker import SlavedIdTracker
-from synapse.storage.push_rule import PushRulesWorkerStore
+from synapse.storage import DataStore
+from synapse.storage.push_rule import PushRuleStore
+from synapse.util.caches.stream_change_cache import StreamChangeCache
 
 
-class SlavedPushRuleStore(PushRulesWorkerStore, SlavedEventStore):
+class SlavedPushRuleStore(SlavedEventStore):
     def __init__(self, db_conn, hs):
+        super(SlavedPushRuleStore, self).__init__(db_conn, hs)
         self._push_rules_stream_id_gen = SlavedIdTracker(
             db_conn, "push_rules_stream", "stream_id",
         )
-        super(SlavedPushRuleStore, self).__init__(db_conn, hs)
+        self.push_rules_stream_cache = StreamChangeCache(
+            "PushRulesStreamChangeCache",
+            self._push_rules_stream_id_gen.get_current_token(),
+        )
+
+    get_push_rules_for_user = PushRuleStore.__dict__["get_push_rules_for_user"]
+    get_push_rules_enabled_for_user = (
+        PushRuleStore.__dict__["get_push_rules_enabled_for_user"]
+    )
+    have_push_rules_changed_for_user = (
+        DataStore.have_push_rules_changed_for_user.__func__
+    )
 
     def get_push_rules_stream_token(self):
         return (
@@ -32,9 +45,6 @@ class SlavedPushRuleStore(PushRulesWorkerStore, SlavedEventStore):
             self._stream_id_gen.get_current_token(),
         )
 
-    def get_max_push_rules_stream_id(self):
-        return self._push_rules_stream_id_gen.get_current_token()
-
     def stream_positions(self):
         result = super(SlavedPushRuleStore, self).stream_positions()
         result["push_rules"] = self._push_rules_stream_id_gen.get_current_token()

synapse/replication/slave/storage/pushers.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,10 +16,10 @@
 from ._base import BaseSlavedStore
 from ._slaved_id_tracker import SlavedIdTracker
 
-from synapse.storage.pusher import PusherWorkerStore
+from synapse.storage import DataStore
 
 
-class SlavedPusherStore(PusherWorkerStore, BaseSlavedStore):
+class SlavedPusherStore(BaseSlavedStore):
 
     def __init__(self, db_conn, hs):
         super(SlavedPusherStore, self).__init__(db_conn, hs)
@@ -29,6 +28,13 @@ class SlavedPusherStore(PusherWorkerStore, BaseSlavedStore):
             extra_tables=[("deleted_pushers", "stream_id")],
         )
 
+    get_all_pushers = DataStore.get_all_pushers.__func__
+    get_pushers_by = DataStore.get_pushers_by.__func__
+    get_pushers_by_app_id_and_pushkey = (
+        DataStore.get_pushers_by_app_id_and_pushkey.__func__
+    )
+    _decode_pushers_rows = DataStore._decode_pushers_rows.__func__
+
     def stream_positions(self):
         result = super(SlavedPusherStore, self).stream_positions()
         result["pushers"] = self._pushers_id_gen.get_current_token()

synapse/replication/slave/storage/receipts.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,7 +16,9 @@
 from ._base import BaseSlavedStore
 from ._slaved_id_tracker import SlavedIdTracker
 
-from synapse.storage.receipts import ReceiptsWorkerStore
+from synapse.storage import DataStore
+from synapse.storage.receipts import ReceiptsStore
+from synapse.util.caches.stream_change_cache import StreamChangeCache
 
 # So, um, we want to borrow a load of functions intended for reading from
 # a DataStore, but we don't want to take functions that either write to the
@@ -28,19 +29,36 @@ from synapse.storage.receipts import ReceiptsWorkerStore
 # the method descriptor on the DataStore and chuck them into our class.
 
 
-class SlavedReceiptsStore(ReceiptsWorkerStore, BaseSlavedStore):
+class SlavedReceiptsStore(BaseSlavedStore):
 
     def __init__(self, db_conn, hs):
-        # We instantiate this first as the ReceiptsWorkerStore constructor
-        # needs to be able to call get_max_receipt_stream_id
+        super(SlavedReceiptsStore, self).__init__(db_conn, hs)
+
         self._receipts_id_gen = SlavedIdTracker(
             db_conn, "receipts_linearized", "stream_id"
         )
 
-        super(SlavedReceiptsStore, self).__init__(db_conn, hs)
+        self._receipts_stream_cache = StreamChangeCache(
+            "ReceiptsRoomChangeCache", self._receipts_id_gen.get_current_token()
+        )
 
-    def get_max_receipt_stream_id(self):
-        return self._receipts_id_gen.get_current_token()
+    get_receipts_for_user = ReceiptsStore.__dict__["get_receipts_for_user"]
+    get_linearized_receipts_for_room = (
+        ReceiptsStore.__dict__["get_linearized_receipts_for_room"]
+    )
+    _get_linearized_receipts_for_rooms = (
+        ReceiptsStore.__dict__["_get_linearized_receipts_for_rooms"]
+    )
+    get_last_receipt_event_id_for_user = (
+        ReceiptsStore.__dict__["get_last_receipt_event_id_for_user"]
+    )
+
+    get_max_receipt_stream_id = DataStore.get_max_receipt_stream_id.__func__
+    get_all_updated_receipts = DataStore.get_all_updated_receipts.__func__
+
+    get_linearized_receipts_for_rooms = (
+        DataStore.get_linearized_receipts_for_rooms.__func__
+    )
 
     def stream_positions(self):
         result = super(SlavedReceiptsStore, self).stream_positions()
@@ -53,8 +71,6 @@ class SlavedReceiptsStore(ReceiptsWorkerStore, BaseSlavedStore):
         self.get_last_receipt_event_id_for_user.invalidate(
             (user_id, room_id, receipt_type)
         )
-        self._invalidate_get_users_with_receipts_in_room(room_id, receipt_type, user_id)
-        self.get_receipts_for_room.invalidate((room_id, receipt_type))
 
     def process_replication_rows(self, stream_name, token, rows):
         if stream_name == "receipts":

synapse/replication/slave/storage/registration.py

@@ -14,8 +14,20 @@
 # limitations under the License.
 
 from ._base import BaseSlavedStore
-from synapse.storage.registration import RegistrationWorkerStore
+from synapse.storage import DataStore
+from synapse.storage.registration import RegistrationStore
 
 
-class SlavedRegistrationStore(RegistrationWorkerStore, BaseSlavedStore):
-    pass
+class SlavedRegistrationStore(BaseSlavedStore):
+    def __init__(self, db_conn, hs):
+        super(SlavedRegistrationStore, self).__init__(db_conn, hs)
+
+    # TODO: use the cached version and invalidate deleted tokens
+    get_user_by_access_token = RegistrationStore.__dict__[
+        "get_user_by_access_token"
+    ]
+
+    _query_for_auth = DataStore._query_for_auth.__func__
+    get_user_by_id = RegistrationStore.__dict__[
+        "get_user_by_id"
+    ]

synapse/replication/slave/storage/room.py

@@ -14,19 +14,32 @@
 # limitations under the License.
 
 from ._base import BaseSlavedStore
-from synapse.storage.room import RoomWorkerStore
+from synapse.storage import DataStore
+from synapse.storage.room import RoomStore
 from ._slaved_id_tracker import SlavedIdTracker
 
 
-class RoomStore(RoomWorkerStore, BaseSlavedStore):
+class RoomStore(BaseSlavedStore):
     def __init__(self, db_conn, hs):
         super(RoomStore, self).__init__(db_conn, hs)
         self._public_room_id_gen = SlavedIdTracker(
             db_conn, "public_room_list_stream", "stream_id"
         )
 
-    def get_current_public_room_stream_id(self):
-        return self._public_room_id_gen.get_current_token()
+    get_public_room_ids = DataStore.get_public_room_ids.__func__
+    get_current_public_room_stream_id = (
+        DataStore.get_current_public_room_stream_id.__func__
+    )
+    get_public_room_ids_at_stream_id = (
+        RoomStore.__dict__["get_public_room_ids_at_stream_id"]
+    )
+    get_public_room_ids_at_stream_id_txn = (
+        DataStore.get_public_room_ids_at_stream_id_txn.__func__
+    )
+    get_published_at_stream_id_txn = (
+        DataStore.get_published_at_stream_id_txn.__func__
+    )
+    get_public_room_changes = DataStore.get_public_room_changes.__func__
 
     def stream_positions(self):
         result = super(RoomStore, self).stream_positions()

synapse/replication/tcp/commands.py

@@ -19,13 +19,11 @@ allowed to be sent by which side.
 """
 
 import logging
-import simplejson
+import ujson as json
 
 
 logger = logging.getLogger(__name__)
 
-_json_encoder = simplejson.JSONEncoder(namedtuple_as_object=False)
-
 
 class Command(object):
     """The base command class.
@@ -102,14 +100,14 @@ class RdataCommand(Command):
         return cls(
             stream_name,
             None if token == "batch" else int(token),
-            simplejson.loads(row_json)
+            json.loads(row_json)
         )
 
     def to_line(self):
         return " ".join((
             self.stream_name,
             str(self.token) if self.token is not None else "batch",
-            _json_encoder.encode(self.row),
+            json.dumps(self.row),
         ))
 
 
@@ -300,12 +298,10 @@ class InvalidateCacheCommand(Command):
     def from_line(cls, line):
         cache_func, keys_json = line.split(" ", 1)
 
-        return cls(cache_func, simplejson.loads(keys_json))
+        return cls(cache_func, json.loads(keys_json))
 
     def to_line(self):
-        return " ".join((
-            self.cache_func, _json_encoder.encode(self.keys),
-        ))
+        return " ".join((self.cache_func, json.dumps(self.keys)))
 
 
 class UserIpCommand(Command):
@@ -329,14 +325,14 @@ class UserIpCommand(Command):
     def from_line(cls, line):
         user_id, jsn = line.split(" ", 1)
 
-        access_token, ip, user_agent, device_id, last_seen = simplejson.loads(jsn)
+        access_token, ip, user_agent, device_id, last_seen = json.loads(jsn)
 
         return cls(
             user_id, access_token, ip, user_agent, device_id, last_seen
         )
 
     def to_line(self):
-        return self.user_id + " " + _json_encoder.encode((
+        return self.user_id + " " + json.dumps((
             self.access_token, self.ip, self.user_agent, self.device_id,
             self.last_seen,
         ))

synapse/replication/tcp/streams.py

@@ -33,7 +33,7 @@ import logging
 logger = logging.getLogger(__name__)
 
 
-MAX_EVENTS_BEHIND = 10000
+MAX_EVENTS_BEHIND = 500000
 
 
 EventStreamRow = namedtuple("EventStreamRow", (

synapse/rest/client/v1/admin.py

@@ -17,7 +17,7 @@
 from twisted.internet import defer
 
 from synapse.api.constants import Membership
-from synapse.api.errors import AuthError, SynapseError, Codes, NotFoundError
+from synapse.api.errors import AuthError, SynapseError
 from synapse.types import UserID, create_requester
 from synapse.http.servlet import parse_json_object_from_request
 
@@ -114,18 +114,12 @@ class PurgeMediaCacheRestServlet(ClientV1RestServlet):
 
 class PurgeHistoryRestServlet(ClientV1RestServlet):
     PATTERNS = client_path_patterns(
-        "/admin/purge_history/(?P<room_id>[^/]*)(/(?P<event_id>[^/]+))?"
+        "/admin/purge_history/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
     )
 
     def __init__(self, hs):
-        """
-
-        Args:
-            hs (synapse.server.HomeServer)
-        """
         super(PurgeHistoryRestServlet, self).__init__(hs)
         self.handlers = hs.get_handlers()
-        self.store = hs.get_datastore()
 
     @defer.inlineCallbacks
     def on_POST(self, request, room_id, event_id):
@@ -139,89 +133,12 @@ class PurgeHistoryRestServlet(ClientV1RestServlet):
 
         delete_local_events = bool(body.get("delete_local_events", False))
 
-        # establish the topological ordering we should keep events from. The
-        # user can provide an event_id in the URL or the request body, or can
-        # provide a timestamp in the request body.
-        if event_id is None:
-            event_id = body.get('purge_up_to_event_id')
-
-        if event_id is not None:
-            event = yield self.store.get_event(event_id)
-
-            if event.room_id != room_id:
-                raise SynapseError(400, "Event is for wrong room.")
-
-            depth = event.depth
-            logger.info(
-                "[purge] purging up to depth %i (event_id %s)",
-                depth, event_id,
-            )
-        elif 'purge_up_to_ts' in body:
-            ts = body['purge_up_to_ts']
-            if not isinstance(ts, int):
-                raise SynapseError(
-                    400, "purge_up_to_ts must be an int",
-                    errcode=Codes.BAD_JSON,
-                )
-
-            stream_ordering = (
-                yield self.store.find_first_stream_ordering_after_ts(ts)
-            )
-
-            (_, depth, _) = (
-                yield self.store.get_room_event_after_stream_ordering(
-                    room_id, stream_ordering,
-                )
-            )
-            logger.info(
-                "[purge] purging up to depth %i (received_ts %i => "
-                "stream_ordering %i)",
-                depth, ts, stream_ordering,
-            )
-        else:
-            raise SynapseError(
-                400,
-                "must specify purge_up_to_event_id or purge_up_to_ts",
-                errcode=Codes.BAD_JSON,
-            )
-
-        purge_id = yield self.handlers.message_handler.start_purge_history(
-            room_id, depth,
+        yield self.handlers.message_handler.purge_history(
+            room_id, event_id,
             delete_local_events=delete_local_events,
         )
 
-        defer.returnValue((200, {
-            "purge_id": purge_id,
-        }))
-
-
-class PurgeHistoryStatusRestServlet(ClientV1RestServlet):
-    PATTERNS = client_path_patterns(
-        "/admin/purge_history_status/(?P<purge_id>[^/]+)"
-    )
-
-    def __init__(self, hs):
-        """
-
-        Args:
-            hs (synapse.server.HomeServer)
-        """
-        super(PurgeHistoryStatusRestServlet, self).__init__(hs)
-        self.handlers = hs.get_handlers()
-
-    @defer.inlineCallbacks
-    def on_GET(self, request, purge_id):
-        requester = yield self.auth.get_user_by_req(request)
-        is_admin = yield self.auth.is_server_admin(requester.user)
-
-        if not is_admin:
-            raise AuthError(403, "You are not a server admin")
-
-        purge_status = self.handlers.message_handler.get_purge_status(purge_id)
-        if purge_status is None:
-            raise NotFoundError("purge id '%s' not found" % purge_id)
-
-        defer.returnValue((200, purge_status.asdict()))
+        defer.returnValue((200, {}))
 
 
 class DeactivateAccountRestServlet(ClientV1RestServlet):
@@ -263,7 +180,6 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
         self.handlers = hs.get_handlers()
         self.state = hs.get_state_handler()
         self.event_creation_handler = hs.get_event_creation_handler()
-        self.room_member_handler = hs.get_room_member_handler()
 
     @defer.inlineCallbacks
     def on_POST(self, request, room_id):
@@ -296,17 +212,6 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
         )
         new_room_id = info["room_id"]
 
-        yield self.event_creation_handler.create_and_send_nonmember_event(
-            room_creator_requester,
-            {
-                "type": "m.room.message",
-                "content": {"body": message, "msgtype": "m.text"},
-                "room_id": new_room_id,
-                "sender": new_room_user_id,
-            },
-            ratelimit=False,
-        )
-
         requester_user_id = requester.user.to_string()
 
         logger.info("Shutting down room %r", room_id)
@@ -322,7 +227,7 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
             logger.info("Kicking %r from %r...", user_id, room_id)
 
             target_requester = create_requester(user_id)
-            yield self.room_member_handler.update_membership(
+            yield self.handlers.room_member_handler.update_membership(
                 requester=target_requester,
                 target=target_requester.user,
                 room_id=room_id,
@@ -331,9 +236,9 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
                 ratelimit=False
             )
 
-            yield self.room_member_handler.forget(target_requester.user, room_id)
+            yield self.handlers.room_member_handler.forget(target_requester.user, room_id)
 
-            yield self.room_member_handler.update_membership(
+            yield self.handlers.room_member_handler.update_membership(
                 requester=target_requester,
                 target=target_requester.user,
                 room_id=new_room_id,
@@ -344,6 +249,17 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
 
             kicked_users.append(user_id)
 
+        yield self.event_creation_handler.create_and_send_nonmember_event(
+            room_creator_requester,
+            {
+                "type": "m.room.message",
+                "content": {"body": message, "msgtype": "m.text"},
+                "room_id": new_room_id,
+                "sender": new_room_user_id,
+            },
+            ratelimit=False,
+        )
+
         aliases_for_room = yield self.store.get_aliases_for_room(room_id)
 
         yield self.store.update_aliases_for_room(
@@ -592,7 +508,6 @@ class SearchUsersRestServlet(ClientV1RestServlet):
 def register_servlets(hs, http_server):
     WhoisRestServlet(hs).register(http_server)
     PurgeMediaCacheRestServlet(hs).register(http_server)
-    PurgeHistoryStatusRestServlet(hs).register(http_server)
     DeactivateAccountRestServlet(hs).register(http_server)
     PurgeHistoryRestServlet(hs).register(http_server)
     UsersRestServlet(hs).register(http_server)

synapse/rest/client/v1/presence.py

@@ -81,7 +81,7 @@ class PresenceStatusRestServlet(ClientV1RestServlet):
         except Exception:
             raise SynapseError(400, "Unable to parse state")
 
-        yield self.presence_handler.set_state(user, state)
+        # yield self.presence_handler.set_state(user, state)
 
         defer.returnValue((200, {}))
 

synapse/rest/client/v1/room.py

@@ -30,7 +30,7 @@ from synapse.http.servlet import (
 
 import logging
 import urllib
-import simplejson as json
+import ujson as json
 
 logger = logging.getLogger(__name__)
 
@@ -84,7 +84,6 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
         super(RoomStateEventRestServlet, self).__init__(hs)
         self.handlers = hs.get_handlers()
         self.event_creation_hander = hs.get_event_creation_handler()
-        self.room_member_handler = hs.get_room_member_handler()
 
     def register(self, http_server):
         # /room/$roomid/state/$eventtype
@@ -157,7 +156,7 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
 
         if event_type == EventTypes.Member:
             membership = content.get("membership", None)
-            event = yield self.room_member_handler.update_membership(
+            event = yield self.handlers.room_member_handler.update_membership(
                 requester,
                 target=UserID.from_string(state_key),
                 room_id=room_id,
@@ -230,7 +229,7 @@ class RoomSendEventRestServlet(ClientV1RestServlet):
 class JoinRoomAliasServlet(ClientV1RestServlet):
     def __init__(self, hs):
         super(JoinRoomAliasServlet, self).__init__(hs)
-        self.room_member_handler = hs.get_room_member_handler()
+        self.handlers = hs.get_handlers()
 
     def register(self, http_server):
         # /join/$room_identifier[/$txn_id]
@@ -258,7 +257,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
             except Exception:
                 remote_room_hosts = None
         elif RoomAlias.is_valid(room_identifier):
-            handler = self.room_member_handler
+            handler = self.handlers.room_member_handler
             room_alias = RoomAlias.from_string(room_identifier)
             room_id, remote_room_hosts = yield handler.lookup_room_alias(room_alias)
             room_id = room_id.to_string()
@@ -267,7 +266,7 @@ class JoinRoomAliasServlet(ClientV1RestServlet):
                 room_identifier,
             ))
 
-        yield self.room_member_handler.update_membership(
+        yield self.handlers.room_member_handler.update_membership(
             requester=requester,
             target=requester.user,
             room_id=room_id,
@@ -563,7 +562,7 @@ class RoomEventContextServlet(ClientV1RestServlet):
 class RoomForgetRestServlet(ClientV1RestServlet):
     def __init__(self, hs):
         super(RoomForgetRestServlet, self).__init__(hs)
-        self.room_member_handler = hs.get_room_member_handler()
+        self.handlers = hs.get_handlers()
 
     def register(self, http_server):
         PATTERNS = ("/rooms/(?P<room_id>[^/]*)/forget")
@@ -576,7 +575,7 @@ class RoomForgetRestServlet(ClientV1RestServlet):
             allow_guest=False,
         )
 
-        yield self.room_member_handler.forget(
+        yield self.handlers.room_member_handler.forget(
             user=requester.user,
             room_id=room_id,
         )
@@ -594,12 +593,12 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
 
     def __init__(self, hs):
         super(RoomMembershipRestServlet, self).__init__(hs)
-        self.room_member_handler = hs.get_room_member_handler()
+        self.handlers = hs.get_handlers()
 
     def register(self, http_server):
         # /rooms/$roomid/[invite|join|leave]
         PATTERNS = ("/rooms/(?P<room_id>[^/]*)/"
-                    "(?P<membership_action>join|invite|leave|ban|unban|kick)")
+                    "(?P<membership_action>join|invite|leave|ban|unban|kick|forget)")
         register_txn_path(self, PATTERNS, http_server)
 
     @defer.inlineCallbacks
@@ -623,7 +622,7 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
             content = {}
 
         if membership_action == "invite" and self._has_3pid_invite_keys(content):
-            yield self.room_member_handler.do_3pid_invite(
+            yield self.handlers.room_member_handler.do_3pid_invite(
                 room_id,
                 requester.user,
                 content["medium"],
@@ -645,7 +644,7 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
         if 'reason' in content and membership_action in ['kick', 'ban']:
             event_content = {'reason': content['reason']}
 
-        yield self.room_member_handler.update_membership(
+        yield self.handlers.room_member_handler.update_membership(
             requester=requester,
             target=target,
             room_id=room_id,
@@ -655,12 +654,7 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
             content=event_content,
         )
 
-        return_value = {}
-
-        if membership_action == "join":
-            return_value["room_id"] = room_id
-
-        defer.returnValue((200, return_value))
+        defer.returnValue((200, {}))
 
     def _has_3pid_invite_keys(self, content):
         for key in {"id_server", "medium", "address"}:

synapse/rest/client/v2_alpha/groups.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2017 Vector Creations Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -402,32 +401,6 @@ class GroupInvitedUsersServlet(RestServlet):
         defer.returnValue((200, result))
 
 
-class GroupSettingJoinPolicyServlet(RestServlet):
-    """Set group join policy
-    """
-    PATTERNS = client_v2_patterns("/groups/(?P<group_id>[^/]*)/settings/m.join_policy$")
-
-    def __init__(self, hs):
-        super(GroupSettingJoinPolicyServlet, self).__init__()
-        self.auth = hs.get_auth()
-        self.groups_handler = hs.get_groups_local_handler()
-
-    @defer.inlineCallbacks
-    def on_PUT(self, request, group_id):
-        requester = yield self.auth.get_user_by_req(request)
-        requester_user_id = requester.user.to_string()
-
-        content = parse_json_object_from_request(request)
-
-        result = yield self.groups_handler.set_group_join_policy(
-            group_id,
-            requester_user_id,
-            content,
-        )
-
-        defer.returnValue((200, result))
-
-
 class GroupCreateServlet(RestServlet):
     """Create a group
     """
@@ -765,7 +738,6 @@ def register_servlets(hs, http_server):
     GroupInvitedUsersServlet(hs).register(http_server)
     GroupUsersServlet(hs).register(http_server)
     GroupRoomServlet(hs).register(http_server)
-    GroupSettingJoinPolicyServlet(hs).register(http_server)
     GroupCreateServlet(hs).register(http_server)
     GroupAdminRoomsServlet(hs).register(http_server)
     GroupAdminRoomsConfigServlet(hs).register(http_server)

synapse/rest/client/v2_alpha/register.py

@@ -183,7 +183,7 @@ class RegisterRestServlet(RestServlet):
         self.auth_handler = hs.get_auth_handler()
         self.registration_handler = hs.get_handlers().registration_handler
         self.identity_handler = hs.get_handlers().identity_handler
-        self.room_member_handler = hs.get_room_member_handler()
+        self.room_member_handler = hs.get_handlers().room_member_handler
         self.device_handler = hs.get_device_handler()
         self.macaroon_gen = hs.get_macaroon_generator()
 

synapse/rest/client/v2_alpha/sync.py

@@ -33,7 +33,7 @@ from ._base import set_timeline_upper_limit
 import itertools
 import logging
 
-import simplejson as json
+import ujson as json
 
 logger = logging.getLogger(__name__)
 

synapse/rest/media/v1/media_storage.py

@@ -58,13 +58,23 @@ class MediaStorage(object):
         Returns:
             Deferred[str]: the file path written to in the primary media store
         """
+        path = self._file_info_to_path(file_info)
+        fname = os.path.join(self.local_media_directory, path)
 
-        with self.store_into_file(file_info) as (f, fname, finish_cb):
-            # Write to the main repository
-            yield make_deferred_yieldable(threads.deferToThread(
-                _write_file_synchronously, source, f,
-            ))
-            yield finish_cb()
+        dirname = os.path.dirname(fname)
+        if not os.path.exists(dirname):
+            os.makedirs(dirname)
+
+        # Write to the main repository
+        yield make_deferred_yieldable(threads.deferToThread(
+            _write_file_synchronously, source, fname,
+        ))
+
+        # Tell the storage providers about the new file. They'll decide
+        # if they should upload it and whether to do so synchronously
+        # or not.
+        for provider in self.storage_providers:
+            yield provider.store_file(path, file_info)
 
         defer.returnValue(fname)
 
@@ -230,16 +240,21 @@ class MediaStorage(object):
         )
 
 
-def _write_file_synchronously(source, dest):
-    """Write `source` to the file like `dest` synchronously. Should be called
+def _write_file_synchronously(source, fname):
+    """Write `source` to the path `fname` synchronously. Should be called
     from a thread.
 
     Args:
-        source: A file like object that's to be written
-        dest: A file like object to be written to
+        source: A file like object to be written
+        fname (str): Path to write to
     """
+    dirname = os.path.dirname(fname)
+    if not os.path.exists(dirname):
+        os.makedirs(dirname)
+
     source.seek(0)  # Ensure we read from the start of the file
-    shutil.copyfileobj(source, dest)
+    with open(fname, "wb") as f:
+        shutil.copyfileobj(source, f)
 
 
 class FileResponder(Responder):

synapse/rest/media/v1/preview_url_resource.py

@@ -23,7 +23,7 @@ import re
 import shutil
 import sys
 import traceback
-import simplejson as json
+import ujson as json
 import urlparse
 
 from twisted.web.server import NOT_DONE_YET

synapse/server.py

@@ -32,10 +32,8 @@ from synapse.appservice.scheduler import ApplicationServiceScheduler
 from synapse.crypto.keyring import Keyring
 from synapse.events.builder import EventBuilderFactory
 from synapse.events.spamcheck import SpamChecker
-from synapse.federation.federation_client import FederationClient
-from synapse.federation.federation_server import FederationServer
+from synapse.federation import initialize_http_replication
 from synapse.federation.send_queue import FederationRemoteSendQueue
-from synapse.federation.federation_server import FederationHandlerRegistry
 from synapse.federation.transport.client import TransportLayerClient
 from synapse.federation.transaction_queue import TransactionQueue
 from synapse.handlers import Handlers
@@ -47,8 +45,6 @@ from synapse.handlers.device import DeviceHandler
 from synapse.handlers.e2e_keys import E2eKeysHandler
 from synapse.handlers.presence import PresenceHandler
 from synapse.handlers.room_list import RoomListHandler
-from synapse.handlers.room_member import RoomMemberMasterHandler
-from synapse.handlers.room_member_worker import RoomMemberWorkerHandler
 from synapse.handlers.set_password import SetPasswordHandler
 from synapse.handlers.sync import SyncHandler
 from synapse.handlers.typing import TypingHandler
@@ -102,8 +98,7 @@ class HomeServer(object):
     DEPENDENCIES = [
         'http_client',
         'db_pool',
-        'federation_client',
-        'federation_server',
+        'replication_layer',
         'handlers',
         'v1auth',
         'auth',
@@ -150,8 +145,6 @@ class HomeServer(object):
         'groups_attestation_signing',
         'groups_attestation_renewer',
         'spam_checker',
-        'room_member_handler',
-        'federation_registry',
     ]
 
     def __init__(self, hostname, **kwargs):
@@ -200,11 +193,8 @@ class HomeServer(object):
     def get_ratelimiter(self):
         return self.ratelimiter
 
-    def build_federation_client(self):
-        return FederationClient(self)
-
-    def build_federation_server(self):
-        return FederationServer(self)
+    def build_replication_layer(self):
+        return initialize_http_replication(self)
 
     def build_handlers(self):
         return Handlers(self)
@@ -392,14 +382,6 @@ class HomeServer(object):
     def build_spam_checker(self):
         return SpamChecker(self)
 
-    def build_room_member_handler(self):
-        if self.config.worker_app:
-            return RoomMemberWorkerHandler(self)
-        return RoomMemberMasterHandler(self)
-
-    def build_federation_registry(self):
-        return FederationHandlerRegistry()
-
     def remove_pusher(self, app_id, push_key, user_id):
         return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
 

synapse/state.py

@@ -132,7 +132,7 @@ class StateHandler(object):
 
         state_map = yield self.store.get_events(state.values(), get_prev_content=False)
         state = {
-            key: state_map[e_id] for key, e_id in state.iteritems() if e_id in state_map
+            key: state_map[e_id] for key, e_id in state.items() if e_id in state_map
         }
 
         defer.returnValue(state)
@@ -378,7 +378,7 @@ class StateHandler(object):
             new_state = resolve_events_with_state_map(state_set_ids, state_map)
 
         new_state = {
-            key: state_map[ev_id] for key, ev_id in new_state.iteritems()
+            key: state_map[ev_id] for key, ev_id in new_state.items()
         }
 
         return new_state
@@ -458,15 +458,15 @@ class StateResolutionHandler(object):
             # build a map from state key to the event_ids which set that state.
             # dict[(str, str), set[str])
             state = {}
-            for st in state_groups_ids.itervalues():
-                for key, e_id in st.iteritems():
+            for st in state_groups_ids.values():
+                for key, e_id in st.items():
                     state.setdefault(key, set()).add(e_id)
 
             # build a map from state key to the event_ids which set that state,
             # including only those where there are state keys in conflict.
             conflicted_state = {
                 k: list(v)
-                for k, v in state.iteritems()
+                for k, v in state.items()
                 if len(v) > 1
             }
 
@@ -480,37 +480,36 @@ class StateResolutionHandler(object):
                     )
             else:
                 new_state = {
-                    key: e_ids.pop() for key, e_ids in state.iteritems()
+                    key: e_ids.pop() for key, e_ids in state.items()
                 }
 
-            with Measure(self.clock, "state.create_group_ids"):
-                # if the new state matches any of the input state groups, we can
-                # use that state group again. Otherwise we will generate a state_id
-                # which will be used as a cache key for future resolutions, but
-                # not get persisted.
-                state_group = None
-                new_state_event_ids = frozenset(new_state.itervalues())
-                for sg, events in state_groups_ids.iteritems():
-                    if new_state_event_ids == frozenset(e_id for e_id in events):
-                        state_group = sg
-                        break
+            # if the new state matches any of the input state groups, we can
+            # use that state group again. Otherwise we will generate a state_id
+            # which will be used as a cache key for future resolutions, but
+            # not get persisted.
+            state_group = None
+            new_state_event_ids = frozenset(new_state.values())
+            for sg, events in state_groups_ids.items():
+                if new_state_event_ids == frozenset(e_id for e_id in events):
+                    state_group = sg
+                    break
 
-                # TODO: We want to create a state group for this set of events, to
-                # increase cache hits, but we need to make sure that it doesn't
-                # end up as a prev_group without being added to the database
+            # TODO: We want to create a state group for this set of events, to
+            # increase cache hits, but we need to make sure that it doesn't
+            # end up as a prev_group without being added to the database
 
-                prev_group = None
-                delta_ids = None
-                for old_group, old_ids in state_groups_ids.iteritems():
-                    if not set(new_state) - set(old_ids):
-                        n_delta_ids = {
-                            k: v
-                            for k, v in new_state.iteritems()
-                            if old_ids.get(k) != v
-                        }
-                        if not delta_ids or len(n_delta_ids) < len(delta_ids):
-                            prev_group = old_group
-                            delta_ids = n_delta_ids
+            prev_group = None
+            delta_ids = None
+            for old_group, old_ids in state_groups_ids.iteritems():
+                if not set(new_state) - set(old_ids):
+                    n_delta_ids = {
+                        k: v
+                        for k, v in new_state.iteritems()
+                        if old_ids.get(k) != v
+                    }
+                    if not delta_ids or len(n_delta_ids) < len(delta_ids):
+                        prev_group = old_group
+                        delta_ids = n_delta_ids
 
             cache = _StateCacheEntry(
                 state=new_state,
@@ -703,7 +702,7 @@ def _resolve_with_state(unconflicted_state_ids, conflicted_state_ds, auth_event_
 
     auth_events = {
         key: state_map[ev_id]
-        for key, ev_id in auth_event_ids.iteritems()
+        for key, ev_id in auth_event_ids.items()
         if ev_id in state_map
     }
 
@@ -741,7 +740,7 @@ def _resolve_state_events(conflicted_state, auth_events):
 
     auth_events.update(resolved_state)
 
-    for key, events in conflicted_state.iteritems():
+    for key, events in conflicted_state.items():
         if key[0] == EventTypes.JoinRules:
             logger.debug("Resolving conflicted join rules %r", events)
             resolved_state[key] = _resolve_auth_events(
@@ -751,7 +750,7 @@ def _resolve_state_events(conflicted_state, auth_events):
 
     auth_events.update(resolved_state)
 
-    for key, events in conflicted_state.iteritems():
+    for key, events in conflicted_state.items():
         if key[0] == EventTypes.Member:
             logger.debug("Resolving conflicted member lists %r", events)
             resolved_state[key] = _resolve_auth_events(
@@ -761,7 +760,7 @@ def _resolve_state_events(conflicted_state, auth_events):
 
     auth_events.update(resolved_state)
 
-    for key, events in conflicted_state.iteritems():
+    for key, events in conflicted_state.items():
         if key not in resolved_state:
             logger.debug("Resolving conflicted state %r:%r", key, events)
             resolved_state[key] = _resolve_normal_events(

synapse/storage/__init__.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,10 +13,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer
+
 from synapse.storage.devices import DeviceStore
 from .appservice import (
     ApplicationServiceStore, ApplicationServiceTransactionStore
 )
+from ._base import LoggingTransaction
 from .directory import DirectoryStore
 from .events import EventsStore
 from .presence import PresenceStore, UserPresenceState
@@ -102,6 +104,12 @@ class DataStore(RoomMemberStore, RoomStore,
             db_conn, "events", "stream_ordering", step=-1,
             extra_tables=[("ex_outlier_stream", "event_stream_ordering")]
         )
+        self._receipts_id_gen = StreamIdGenerator(
+            db_conn, "receipts_linearized", "stream_id"
+        )
+        self._account_data_id_gen = StreamIdGenerator(
+            db_conn, "account_data_max_stream_id", "stream_id"
+        )
         self._presence_id_gen = StreamIdGenerator(
             db_conn, "presence_stream", "stream_id"
         )
@@ -138,6 +146,27 @@ class DataStore(RoomMemberStore, RoomStore,
         else:
             self._cache_id_gen = None
 
+        events_max = self._stream_id_gen.get_current_token()
+        event_cache_prefill, min_event_val = self._get_cache_dict(
+            db_conn, "events",
+            entity_column="room_id",
+            stream_column="stream_ordering",
+            max_value=events_max,
+        )
+        self._events_stream_cache = StreamChangeCache(
+            "EventsRoomStreamChangeCache", min_event_val,
+            prefilled_cache=event_cache_prefill,
+        )
+
+        self._membership_stream_cache = StreamChangeCache(
+            "MembershipStreamChangeCache", events_max,
+        )
+
+        account_max = self._account_data_id_gen.get_current_token()
+        self._account_data_stream_cache = StreamChangeCache(
+            "AccountDataAndTagsChangeCache", account_max,
+        )
+
         self._presence_on_startup = self._get_active_presence(db_conn)
 
         presence_cache_prefill, min_presence_val = self._get_cache_dict(
@@ -151,6 +180,18 @@ class DataStore(RoomMemberStore, RoomStore,
             prefilled_cache=presence_cache_prefill
         )
 
+        push_rules_prefill, push_rules_id = self._get_cache_dict(
+            db_conn, "push_rules_stream",
+            entity_column="user_id",
+            stream_column="stream_id",
+            max_value=self._push_rules_stream_id_gen.get_current_token()[0],
+        )
+
+        self.push_rules_stream_cache = StreamChangeCache(
+            "PushRulesStreamChangeCache", push_rules_id,
+            prefilled_cache=push_rules_prefill,
+        )
+
         max_device_inbox_id = self._device_inbox_id_gen.get_current_token()
         device_inbox_prefill, min_device_inbox_id = self._get_cache_dict(
             db_conn, "device_inbox",
@@ -185,7 +226,6 @@ class DataStore(RoomMemberStore, RoomStore,
             "DeviceListFederationStreamChangeCache", device_list_max,
         )
 
-        events_max = self._stream_id_gen.get_current_token()
         curr_state_delta_prefill, min_curr_state_delta_id = self._get_cache_dict(
             db_conn, "current_state_delta_stream",
             entity_column="room_id",
@@ -210,6 +250,20 @@ class DataStore(RoomMemberStore, RoomStore,
             prefilled_cache=_group_updates_prefill,
         )
 
+        cur = LoggingTransaction(
+            db_conn.cursor(),
+            name="_find_stream_orderings_for_times_txn",
+            database_engine=self.database_engine,
+            after_callbacks=[],
+            final_callbacks=[],
+        )
+        self._find_stream_orderings_for_times_txn(cur)
+        cur.close()
+
+        self.find_stream_orderings_looping_call = self._clock.looping_call(
+            self._find_stream_orderings_for_times, 10 * 60 * 1000
+        )
+
         self._stream_order_on_start = self.get_room_max_stream_ordering()
         self._min_stream_order_on_start = self.get_room_min_stream_ordering()
 
@@ -242,12 +296,13 @@ class DataStore(RoomMemberStore, RoomStore,
 
         return [UserPresenceState(**row) for row in rows]
 
+    @defer.inlineCallbacks
     def count_daily_users(self):
         """
         Counts the number of users who used this homeserver in the last 24 hours.
         """
         def _count_users(txn):
-            yesterday = int(self._clock.time_msec()) - (1000 * 60 * 60 * 24)
+            yesterday = int(self._clock.time_msec()) - (1000 * 60 * 60 * 24),
 
             sql = """
                 SELECT COALESCE(count(*), 0) FROM (
@@ -261,91 +316,8 @@ class DataStore(RoomMemberStore, RoomStore,
             count, = txn.fetchone()
             return count
 
-        return self.runInteraction("count_users", _count_users)
-
-    def count_r30_users(self):
-        """
-        Counts the number of 30 day retained users, defined as:-
-         * Users who have created their accounts more than 30 days
-         * Where last seen at most 30 days ago
-         * Where account creation and last_seen are > 30 days
-
-         Returns counts globaly for a given user as well as breaking
-         by platform
-        """
-        def _count_r30_users(txn):
-            thirty_days_in_secs = 86400 * 30
-            now = int(self._clock.time_msec())
-            thirty_days_ago_in_secs = now - thirty_days_in_secs
-
-            sql = """
-                SELECT platform, COALESCE(count(*), 0) FROM (
-                     SELECT
-                        users.name, platform, users.creation_ts * 1000,
-                        MAX(uip.last_seen)
-                     FROM users
-                     INNER JOIN (
-                         SELECT
-                         user_id,
-                         last_seen,
-                         CASE
-                             WHEN user_agent LIKE '%Android%' THEN 'android'
-                             WHEN user_agent LIKE '%iOS%' THEN 'ios'
-                             WHEN user_agent LIKE '%Electron%' THEN 'electron'
-                             WHEN user_agent LIKE '%Mozilla%' THEN 'web'
-                             WHEN user_agent LIKE '%Gecko%' THEN 'web'
-                             ELSE 'unknown'
-                         END
-                         AS platform
-                         FROM user_ips
-                     ) uip
-                     ON users.name = uip.user_id
-                     AND users.appservice_id is NULL
-                     AND users.creation_ts < ?
-                     AND uip.last_seen/1000 > ?
-                     AND (uip.last_seen/1000) - users.creation_ts > 86400 * 30
-                     GROUP BY users.name, platform, users.creation_ts
-                ) u GROUP BY platform
-            """
-
-            results = {}
-            txn.execute(sql, (thirty_days_ago_in_secs,
-                              thirty_days_ago_in_secs))
-
-            for row in txn:
-                if row[0] is 'unknown':
-                    pass
-                results[row[0]] = row[1]
-
-            sql = """
-                SELECT COALESCE(count(*), 0) FROM (
-                    SELECT users.name, users.creation_ts * 1000,
-                                                        MAX(uip.last_seen)
-                    FROM users
-                    INNER JOIN (
-                        SELECT
-                        user_id,
-                        last_seen
-                        FROM user_ips
-                    ) uip
-                    ON users.name = uip.user_id
-                    AND appservice_id is NULL
-                    AND users.creation_ts < ?
-                    AND uip.last_seen/1000 > ?
-                    AND (uip.last_seen/1000) - users.creation_ts > 86400 * 30
-                    GROUP BY users.name, users.creation_ts
-                ) u
-            """
-
-            txn.execute(sql, (thirty_days_ago_in_secs,
-                              thirty_days_ago_in_secs))
-
-            count, = txn.fetchone()
-            results['all'] = count
-
-            return results
-
-        return self.runInteraction("count_r30_users", _count_r30_users)
+        ret = yield self.runInteraction("count_users", _count_users)
+        defer.returnValue(ret)
 
     def get_users(self):
         """Function to reterive a list of users in users table.

synapse/storage/_base.py

@@ -48,16 +48,16 @@ class LoggingTransaction(object):
     passed to the constructor. Adds logging and metrics to the .execute()
     method."""
     __slots__ = [
-        "txn", "name", "database_engine", "after_callbacks", "exception_callbacks",
+        "txn", "name", "database_engine", "after_callbacks", "final_callbacks",
     ]
 
     def __init__(self, txn, name, database_engine, after_callbacks,
-                 exception_callbacks):
+                 final_callbacks):
         object.__setattr__(self, "txn", txn)
         object.__setattr__(self, "name", name)
         object.__setattr__(self, "database_engine", database_engine)
         object.__setattr__(self, "after_callbacks", after_callbacks)
-        object.__setattr__(self, "exception_callbacks", exception_callbacks)
+        object.__setattr__(self, "final_callbacks", final_callbacks)
 
     def call_after(self, callback, *args, **kwargs):
         """Call the given callback on the main twisted thread after the
@@ -66,8 +66,8 @@ class LoggingTransaction(object):
         """
         self.after_callbacks.append((callback, args, kwargs))
 
-    def call_on_exception(self, callback, *args, **kwargs):
-        self.exception_callbacks.append((callback, args, kwargs))
+    def call_finally(self, callback, *args, **kwargs):
+        self.final_callbacks.append((callback, args, kwargs))
 
     def __getattr__(self, name):
         return getattr(self.txn, name)
@@ -215,7 +215,7 @@ class SQLBaseStore(object):
 
         self._clock.looping_call(loop, 10000)
 
-    def _new_transaction(self, conn, desc, after_callbacks, exception_callbacks,
+    def _new_transaction(self, conn, desc, after_callbacks, final_callbacks,
                          logging_context, func, *args, **kwargs):
         start = time.time() * 1000
         txn_id = self._TXN_ID
@@ -236,7 +236,7 @@ class SQLBaseStore(object):
                     txn = conn.cursor()
                     txn = LoggingTransaction(
                         txn, name, self.database_engine, after_callbacks,
-                        exception_callbacks,
+                        final_callbacks,
                     )
                     r = func(txn, *args, **kwargs)
                     conn.commit()
@@ -308,11 +308,11 @@ class SQLBaseStore(object):
         current_context = LoggingContext.current_context()
 
         after_callbacks = []
-        exception_callbacks = []
+        final_callbacks = []
 
         def inner_func(conn, *args, **kwargs):
             return self._new_transaction(
-                conn, desc, after_callbacks, exception_callbacks, current_context,
+                conn, desc, after_callbacks, final_callbacks, current_context,
                 func, *args, **kwargs
             )
 
@@ -321,10 +321,9 @@ class SQLBaseStore(object):
 
             for after_callback, after_args, after_kwargs in after_callbacks:
                 after_callback(*after_args, **after_kwargs)
-        except:  # noqa: E722, as we reraise the exception this is fine.
-            for after_callback, after_args, after_kwargs in exception_callbacks:
+        finally:
+            for after_callback, after_args, after_kwargs in final_callbacks:
                 after_callback(*after_args, **after_kwargs)
-            raise
 
         defer.returnValue(result)
 
@@ -1001,8 +1000,7 @@ class SQLBaseStore(object):
             # __exit__ called after the transaction finishes.
             ctx = self._cache_id_gen.get_next()
             stream_id = ctx.__enter__()
-            txn.call_on_exception(ctx.__exit__, None, None, None)
-            txn.call_after(ctx.__exit__, None, None, None)
+            txn.call_finally(ctx.__exit__, None, None, None)
             txn.call_after(self.hs.get_notifier().on_new_replication_data)
 
             self._simple_insert_txn(

synapse/storage/account_data.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,46 +13,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from ._base import SQLBaseStore
 from twisted.internet import defer
 
-from synapse.storage._base import SQLBaseStore
-from synapse.storage.util.id_generators import StreamIdGenerator
-
-from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.caches.descriptors import cached, cachedList, cachedInlineCallbacks
 
-import abc
-import simplejson as json
+import ujson as json
 import logging
 
 logger = logging.getLogger(__name__)
 
 
-class AccountDataWorkerStore(SQLBaseStore):
-    """This is an abstract base class where subclasses must implement
-    `get_max_account_data_stream_id` which can be called in the initializer.
-    """
-
-    # This ABCMeta metaclass ensures that we cannot be instantiated without
-    # the abstract methods being implemented.
-    __metaclass__ = abc.ABCMeta
-
-    def __init__(self, db_conn, hs):
-        account_max = self.get_max_account_data_stream_id()
-        self._account_data_stream_cache = StreamChangeCache(
-            "AccountDataAndTagsChangeCache", account_max,
-        )
-
-        super(AccountDataWorkerStore, self).__init__(db_conn, hs)
-
-    @abc.abstractmethod
-    def get_max_account_data_stream_id(self):
-        """Get the current max stream ID for account data stream
-
-        Returns:
-            int
-        """
-        raise NotImplementedError()
+class AccountDataStore(SQLBaseStore):
 
     @cached()
     def get_account_data_for_user(self, user_id):
@@ -133,7 +104,6 @@ class AccountDataWorkerStore(SQLBaseStore):
             for row in rows
         })
 
-    @cached(num_args=2)
     def get_account_data_for_room(self, user_id, room_id):
         """Get all the client account_data for a user for a room.
 
@@ -157,38 +127,6 @@ class AccountDataWorkerStore(SQLBaseStore):
             "get_account_data_for_room", get_account_data_for_room_txn
         )
 
-    @cached(num_args=3, max_entries=5000)
-    def get_account_data_for_room_and_type(self, user_id, room_id, account_data_type):
-        """Get the client account_data of given type for a user for a room.
-
-        Args:
-            user_id(str): The user to get the account_data for.
-            room_id(str): The room to get the account_data for.
-            account_data_type (str): The account data type to get.
-        Returns:
-            A deferred of the room account_data for that type, or None if
-            there isn't any set.
-        """
-        def get_account_data_for_room_and_type_txn(txn):
-            content_json = self._simple_select_one_onecol_txn(
-                txn,
-                table="room_account_data",
-                keyvalues={
-                    "user_id": user_id,
-                    "room_id": room_id,
-                    "account_data_type": account_data_type,
-                },
-                retcol="content",
-                allow_none=True
-            )
-
-            return json.loads(content_json) if content_json else None
-
-        return self.runInteraction(
-            "get_account_data_for_room_and_type",
-            get_account_data_for_room_and_type_txn,
-        )
-
     def get_all_updated_account_data(self, last_global_id, last_room_id,
                                      current_id, limit):
         """Get all the client account_data that has changed on the server
@@ -271,36 +209,6 @@ class AccountDataWorkerStore(SQLBaseStore):
             "get_updated_account_data_for_user", get_updated_account_data_for_user_txn
         )
 
-    @cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)
-    def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
-        ignored_account_data = yield self.get_global_account_data_by_type_for_user(
-            "m.ignored_user_list", ignorer_user_id,
-            on_invalidate=cache_context.invalidate,
-        )
-        if not ignored_account_data:
-            defer.returnValue(False)
-
-        defer.returnValue(
-            ignored_user_id in ignored_account_data.get("ignored_users", {})
-        )
-
-
-class AccountDataStore(AccountDataWorkerStore):
-    def __init__(self, db_conn, hs):
-        self._account_data_id_gen = StreamIdGenerator(
-            db_conn, "account_data_max_stream_id", "stream_id"
-        )
-
-        super(AccountDataStore, self).__init__(db_conn, hs)
-
-    def get_max_account_data_stream_id(self):
-        """Get the current max stream id for the private user data stream
-
-        Returns:
-            A deferred int.
-        """
-        return self._account_data_id_gen.get_current_token()
-
     @defer.inlineCallbacks
     def add_account_data_to_room(self, user_id, room_id, account_data_type, content):
         """Add some account_data to a room for a user.
@@ -343,10 +251,6 @@ class AccountDataStore(AccountDataWorkerStore):
 
             self._account_data_stream_cache.entity_has_changed(user_id, next_id)
             self.get_account_data_for_user.invalidate((user_id,))
-            self.get_account_data_for_room.invalidate((user_id, room_id,))
-            self.get_account_data_for_room_and_type.prefill(
-                (user_id, room_id, account_data_type,), content,
-            )
 
         result = self._account_data_id_gen.get_current_token()
         defer.returnValue(result)
@@ -417,3 +321,16 @@ class AccountDataStore(AccountDataWorkerStore):
             "update_account_data_max_stream_id",
             _update,
         )
+
+    @cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)
+    def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
+        ignored_account_data = yield self.get_global_account_data_by_type_for_user(
+            "m.ignored_user_list", ignorer_user_id,
+            on_invalidate=cache_context.invalidate,
+        )
+        if not ignored_account_data:
+            defer.returnValue(False)
+
+        defer.returnValue(
+            ignored_user_id in ignored_account_data.get("ignored_users", {})
+        )

synapse/storage/appservice.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2015, 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -18,9 +17,10 @@ import re
 import simplejson as json
 from twisted.internet import defer
 
+from synapse.api.constants import Membership
 from synapse.appservice import AppServiceTransaction
 from synapse.config.appservice import load_appservices
-from synapse.storage.events import EventsWorkerStore
+from synapse.storage.roommember import RoomsForUser
 from ._base import SQLBaseStore
 
 
@@ -46,16 +46,17 @@ def _make_exclusive_regex(services_cache):
     return exclusive_user_regex
 
 
-class ApplicationServiceWorkerStore(SQLBaseStore):
+class ApplicationServiceStore(SQLBaseStore):
+
     def __init__(self, db_conn, hs):
+        super(ApplicationServiceStore, self).__init__(db_conn, hs)
+        self.hostname = hs.hostname
         self.services_cache = load_appservices(
             hs.hostname,
             hs.config.app_service_config_files
         )
         self.exclusive_user_regex = _make_exclusive_regex(self.services_cache)
 
-        super(ApplicationServiceWorkerStore, self).__init__(db_conn, hs)
-
     def get_app_services(self):
         return self.services_cache
 
@@ -111,17 +112,83 @@ class ApplicationServiceWorkerStore(SQLBaseStore):
                 return service
         return None
 
+    def get_app_service_rooms(self, service):
+        """Get a list of RoomsForUser for this application service.
 
-class ApplicationServiceStore(ApplicationServiceWorkerStore):
-    # This is currently empty due to there not being any AS storage functions
-    # that can't be run on the workers. Since this may change in future, and
-    # to keep consistency with the other stores, we keep this empty class for
-    # now.
-    pass
+        Application services may be "interested" in lots of rooms depending on
+        the room ID, the room aliases, or the members in the room. This function
+        takes all of these into account and returns a list of RoomsForUser which
+        represent the entire list of room IDs that this application service
+        wants to know about.
+
+        Args:
+            service: The application service to get a room list for.
+        Returns:
+            A list of RoomsForUser.
+        """
+        return self.runInteraction(
+            "get_app_service_rooms",
+            self._get_app_service_rooms_txn,
+            service,
+        )
+
+    def _get_app_service_rooms_txn(self, txn, service):
+        # get all rooms matching the room ID regex.
+        room_entries = self._simple_select_list_txn(
+            txn=txn, table="rooms", keyvalues=None, retcols=["room_id"]
+        )
+        matching_room_list = set([
+            r["room_id"] for r in room_entries if
+            service.is_interested_in_room(r["room_id"])
+        ])
+
+        # resolve room IDs for matching room alias regex.
+        room_alias_mappings = self._simple_select_list_txn(
+            txn=txn, table="room_aliases", keyvalues=None,
+            retcols=["room_id", "room_alias"]
+        )
+        matching_room_list |= set([
+            r["room_id"] for r in room_alias_mappings if
+            service.is_interested_in_alias(r["room_alias"])
+        ])
+
+        # get all rooms for every user for this AS. This is scoped to users on
+        # this HS only.
+        user_list = self._simple_select_list_txn(
+            txn=txn, table="users", keyvalues=None, retcols=["name"]
+        )
+        user_list = [
+            u["name"] for u in user_list if
+            service.is_interested_in_user(u["name"])
+        ]
+        rooms_for_user_matching_user_id = set()  # RoomsForUser list
+        for user_id in user_list:
+            # FIXME: This assumes this store is linked with RoomMemberStore :(
+            rooms_for_user = self._get_rooms_for_user_where_membership_is_txn(
+                txn=txn,
+                user_id=user_id,
+                membership_list=[Membership.JOIN]
+            )
+            rooms_for_user_matching_user_id |= set(rooms_for_user)
+
+        # make RoomsForUser tuples for room ids and aliases which are not in the
+        # main rooms_for_user_list - e.g. they are rooms which do not have AS
+        # registered users in it.
+        known_room_ids = [r.room_id for r in rooms_for_user_matching_user_id]
+        missing_rooms_for_user = [
+            RoomsForUser(r, service.sender, "join") for r in
+            matching_room_list if r not in known_room_ids
+        ]
+        rooms_for_user_matching_user_id |= set(missing_rooms_for_user)
+
+        return rooms_for_user_matching_user_id
 
 
-class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
-                                               EventsWorkerStore):
+class ApplicationServiceTransactionStore(SQLBaseStore):
+
+    def __init__(self, db_conn, hs):
+        super(ApplicationServiceTransactionStore, self).__init__(db_conn, hs)
+
     @defer.inlineCallbacks
     def get_appservices_by_state(self, state):
         """Get a list of application services based on their state.
@@ -366,11 +433,3 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
         events = yield self._get_events(event_ids)
 
         defer.returnValue((upper_bound, events))
-
-
-class ApplicationServiceTransactionStore(ApplicationServiceTransactionWorkerStore):
-    # This is currently empty due to there not being any AS storage functions
-    # that can't be run on the workers. Since this may change in future, and
-    # to keep consistency with the other stores, we keep this empty class for
-    # now.
-    pass

synapse/storage/background_updates.py

@@ -19,7 +19,7 @@ from . import engines
 
 from twisted.internet import defer
 
-import simplejson as json
+import ujson as json
 import logging
 
 logger = logging.getLogger(__name__)

synapse/storage/client_ips.py

@@ -28,7 +28,7 @@ logger = logging.getLogger(__name__)
 # Number of msec of granularity to store the user IP 'last seen' time. Smaller
 # times give more inserts into the database even for readonly API hits
 # 120 seconds == 2 minutes
-LAST_SEEN_GRANULARITY = 120 * 1000
+LAST_SEEN_GRANULARITY = 10 * 60 * 1000
 
 
 class ClientIpStore(background_updates.BackgroundUpdateStore):
@@ -48,13 +48,6 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             columns=["user_id", "device_id", "last_seen"],
         )
 
-        self.register_background_index_update(
-            "user_ips_last_seen_index",
-            index_name="user_ips_last_seen",
-            table="user_ips",
-            columns=["user_id", "last_seen"],
-        )
-
         # (user_id, access_token, ip) -> (user_agent, device_id, last_seen)
         self._batch_row_update = {}
 

synapse/storage/deviceinbox.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 import logging
-import simplejson
+import ujson
 
 from twisted.internet import defer
 
@@ -85,7 +85,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
             )
             rows = []
             for destination, edu in remote_messages_by_destination.items():
-                edu_json = simplejson.dumps(edu)
+                edu_json = ujson.dumps(edu)
                 rows.append((destination, stream_id, now_ms, edu_json))
             txn.executemany(sql, rows)
 
@@ -177,7 +177,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
                     " WHERE user_id = ?"
                 )
                 txn.execute(sql, (user_id,))
-                message_json = simplejson.dumps(messages_by_device["*"])
+                message_json = ujson.dumps(messages_by_device["*"])
                 for row in txn:
                     # Add the message for all devices for this user on this
                     # server.
@@ -199,7 +199,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
                     # Only insert into the local inbox if the device exists on
                     # this server
                     device = row[0]
-                    message_json = simplejson.dumps(messages_by_device[device])
+                    message_json = ujson.dumps(messages_by_device[device])
                     messages_json_for_user[device] = message_json
 
             if messages_json_for_user:
@@ -253,7 +253,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
             messages = []
             for row in txn:
                 stream_pos = row[0]
-                messages.append(simplejson.loads(row[1]))
+                messages.append(ujson.loads(row[1]))
             if len(messages) < limit:
                 stream_pos = current_stream_id
             return (messages, stream_pos)
@@ -389,7 +389,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
             messages = []
             for row in txn:
                 stream_pos = row[0]
-                messages.append(simplejson.loads(row[1]))
+                messages.append(ujson.loads(row[1]))
             if len(messages) < limit:
                 stream_pos = current_stream_id
             return (messages, stream_pos)

synapse/storage/devices.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-import simplejson as json
+import ujson as json
 
 from twisted.internet import defer
 

synapse/storage/directory.py

@@ -29,7 +29,8 @@ RoomAliasMapping = namedtuple(
 )
 
 
-class DirectoryWorkerStore(SQLBaseStore):
+class DirectoryStore(SQLBaseStore):
+
     @defer.inlineCallbacks
     def get_association_from_room_alias(self, room_alias):
         """ Get's the room_id and server list for a given room_alias
@@ -68,28 +69,6 @@ class DirectoryWorkerStore(SQLBaseStore):
             RoomAliasMapping(room_id, room_alias.to_string(), servers)
         )
 
-    def get_room_alias_creator(self, room_alias):
-        return self._simple_select_one_onecol(
-            table="room_aliases",
-            keyvalues={
-                "room_alias": room_alias,
-            },
-            retcol="creator",
-            desc="get_room_alias_creator",
-            allow_none=True
-        )
-
-    @cached(max_entries=5000)
-    def get_aliases_for_room(self, room_id):
-        return self._simple_select_onecol(
-            "room_aliases",
-            {"room_id": room_id},
-            "room_alias",
-            desc="get_aliases_for_room",
-        )
-
-
-class DirectoryStore(DirectoryWorkerStore):
     @defer.inlineCallbacks
     def create_room_alias_association(self, room_alias, room_id, servers, creator=None):
         """ Creates an associatin between  a room alias and room_id/servers
@@ -137,6 +116,17 @@ class DirectoryStore(DirectoryWorkerStore):
             )
         defer.returnValue(ret)
 
+    def get_room_alias_creator(self, room_alias):
+        return self._simple_select_one_onecol(
+            table="room_aliases",
+            keyvalues={
+                "room_alias": room_alias,
+            },
+            retcol="creator",
+            desc="get_room_alias_creator",
+            allow_none=True
+        )
+
     @defer.inlineCallbacks
     def delete_room_alias(self, room_alias):
         room_id = yield self.runInteraction(
@@ -145,6 +135,7 @@ class DirectoryStore(DirectoryWorkerStore):
             room_alias,
         )
 
+        self.get_aliases_for_room.invalidate((room_id,))
         defer.returnValue(room_id)
 
     def _delete_room_alias_txn(self, txn, room_alias):
@@ -169,12 +160,17 @@ class DirectoryStore(DirectoryWorkerStore):
             (room_alias.to_string(),)
         )
 
-        self._invalidate_cache_and_stream(
-            txn, self.get_aliases_for_room, (room_id,)
-        )
-
         return room_id
 
+    @cached(max_entries=5000)
+    def get_aliases_for_room(self, room_id):
+        return self._simple_select_onecol(
+            "room_aliases",
+            {"room_id": room_id},
+            "room_alias",
+            desc="get_aliases_for_room",
+        )
+
     def update_aliases_for_room(self, old_room_id, new_room_id, creator):
         def _update_aliases_for_room_txn(txn):
             sql = "UPDATE room_aliases SET room_id = ?, creator = ? WHERE room_id = ?"

synapse/storage/end_to_end_keys.py

@@ -17,7 +17,7 @@ from twisted.internet import defer
 from synapse.util.caches.descriptors import cached
 
 from canonicaljson import encode_canonical_json
-import simplejson as json
+import ujson as json
 
 from ._base import SQLBaseStore
 

synapse/storage/event_federation.py

@@ -15,10 +15,7 @@
 
 from twisted.internet import defer
 
-from synapse.storage._base import SQLBaseStore
-from synapse.storage.events import EventsWorkerStore
-from synapse.storage.signatures import SignatureWorkerStore
-
+from ._base import SQLBaseStore
 from synapse.api.errors import StoreError
 from synapse.util.caches.descriptors import cached
 from unpaddedbase64 import encode_base64
@@ -30,8 +27,30 @@ from Queue import PriorityQueue, Empty
 logger = logging.getLogger(__name__)
 
 
-class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore,
-                                 SQLBaseStore):
+class EventFederationStore(SQLBaseStore):
+    """ Responsible for storing and serving up the various graphs associated
+    with an event. Including the main event graph and the auth chains for an
+    event.
+
+    Also has methods for getting the front (latest) and back (oldest) edges
+    of the event graphs. These are used to generate the parents for new events
+    and backfilling from another server respectively.
+    """
+
+    EVENT_AUTH_STATE_ONLY = "event_auth_state_only"
+
+    def __init__(self, db_conn, hs):
+        super(EventFederationStore, self).__init__(db_conn, hs)
+
+        self.register_background_update_handler(
+            self.EVENT_AUTH_STATE_ONLY,
+            self._background_delete_non_state_event_auth,
+        )
+
+        hs.get_clock().looping_call(
+            self._delete_old_forward_extrem_cache, 60 * 60 * 1000
+        )
+
     def get_auth_chain(self, event_ids, include_given=False):
         """Get auth events for given event_ids. The events *must* be state events.
 
@@ -209,6 +228,88 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore,
 
         return int(min_depth) if min_depth is not None else None
 
+    def _update_min_depth_for_room_txn(self, txn, room_id, depth):
+        min_depth = self._get_min_depth_interaction(txn, room_id)
+
+        if min_depth and depth >= min_depth:
+            return
+
+        self._simple_upsert_txn(
+            txn,
+            table="room_depth",
+            keyvalues={
+                "room_id": room_id,
+            },
+            values={
+                "min_depth": depth,
+            },
+        )
+
+    def _handle_mult_prev_events(self, txn, events):
+        """
+        For the given event, update the event edges table and forward and
+        backward extremities tables.
+        """
+        self._simple_insert_many_txn(
+            txn,
+            table="event_edges",
+            values=[
+                {
+                    "event_id": ev.event_id,
+                    "prev_event_id": e_id,
+                    "room_id": ev.room_id,
+                    "is_state": False,
+                }
+                for ev in events
+                for e_id, _ in ev.prev_events
+            ],
+        )
+
+        self._update_backward_extremeties(txn, events)
+
+    def _update_backward_extremeties(self, txn, events):
+        """Updates the event_backward_extremities tables based on the new/updated
+        events being persisted.
+
+        This is called for new events *and* for events that were outliers, but
+        are now being persisted as non-outliers.
+
+        Forward extremities are handled when we first start persisting the events.
+        """
+        events_by_room = {}
+        for ev in events:
+            events_by_room.setdefault(ev.room_id, []).append(ev)
+
+        query = (
+            "INSERT INTO event_backward_extremities (event_id, room_id)"
+            " SELECT ?, ? WHERE NOT EXISTS ("
+            " SELECT 1 FROM event_backward_extremities"
+            " WHERE event_id = ? AND room_id = ?"
+            " )"
+            " AND NOT EXISTS ("
+            " SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
+            " AND outlier = ?"
+            " )"
+        )
+
+        txn.executemany(query, [
+            (e_id, ev.room_id, e_id, ev.room_id, e_id, ev.room_id, False)
+            for ev in events for e_id, _ in ev.prev_events
+            if not ev.internal_metadata.is_outlier()
+        ])
+
+        query = (
+            "DELETE FROM event_backward_extremities"
+            " WHERE event_id = ? AND room_id = ?"
+        )
+        txn.executemany(
+            query,
+            [
+                (ev.event_id, ev.room_id) for ev in events
+                if not ev.internal_metadata.is_outlier()
+            ]
+        )
+
     def get_forward_extremeties_for_room(self, room_id, stream_ordering):
         """For a given room_id and stream_ordering, return the forward
         extremeties of the room at that point in "time".
@@ -270,6 +371,28 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore,
             get_forward_extremeties_for_room_txn
         )
 
+    def _delete_old_forward_extrem_cache(self):
+        def _delete_old_forward_extrem_cache_txn(txn):
+            # Delete entries older than a month, while making sure we don't delete
+            # the only entries for a room.
+            sql = ("""
+                DELETE FROM stream_ordering_to_exterm
+                WHERE
+                room_id IN (
+                    SELECT room_id
+                    FROM stream_ordering_to_exterm
+                    WHERE stream_ordering > ?
+                ) AND stream_ordering < ?
+            """)
+            txn.execute(
+                sql,
+                (self.stream_ordering_month_ago, self.stream_ordering_month_ago,)
+            )
+        return self.runInteraction(
+            "_delete_old_forward_extrem_cache",
+            _delete_old_forward_extrem_cache_txn
+        )
+
     def get_backfill_events(self, room_id, event_list, limit):
         """Get a list of Events for a given topic that occurred before (and
         including) the events in event_list. Return a list of max size `limit`
@@ -399,135 +522,6 @@ class EventFederationWorkerStore(EventsWorkerStore, SignatureWorkerStore,
 
         return event_results
 
-
-class EventFederationStore(EventFederationWorkerStore):
-    """ Responsible for storing and serving up the various graphs associated
-    with an event. Including the main event graph and the auth chains for an
-    event.
-
-    Also has methods for getting the front (latest) and back (oldest) edges
-    of the event graphs. These are used to generate the parents for new events
-    and backfilling from another server respectively.
-    """
-
-    EVENT_AUTH_STATE_ONLY = "event_auth_state_only"
-
-    def __init__(self, db_conn, hs):
-        super(EventFederationStore, self).__init__(db_conn, hs)
-
-        self.register_background_update_handler(
-            self.EVENT_AUTH_STATE_ONLY,
-            self._background_delete_non_state_event_auth,
-        )
-
-        hs.get_clock().looping_call(
-            self._delete_old_forward_extrem_cache, 60 * 60 * 1000
-        )
-
-    def _update_min_depth_for_room_txn(self, txn, room_id, depth):
-        min_depth = self._get_min_depth_interaction(txn, room_id)
-
-        if min_depth and depth >= min_depth:
-            return
-
-        self._simple_upsert_txn(
-            txn,
-            table="room_depth",
-            keyvalues={
-                "room_id": room_id,
-            },
-            values={
-                "min_depth": depth,
-            },
-        )
-
-    def _handle_mult_prev_events(self, txn, events):
-        """
-        For the given event, update the event edges table and forward and
-        backward extremities tables.
-        """
-        self._simple_insert_many_txn(
-            txn,
-            table="event_edges",
-            values=[
-                {
-                    "event_id": ev.event_id,
-                    "prev_event_id": e_id,
-                    "room_id": ev.room_id,
-                    "is_state": False,
-                }
-                for ev in events
-                for e_id, _ in ev.prev_events
-            ],
-        )
-
-        self._update_backward_extremeties(txn, events)
-
-    def _update_backward_extremeties(self, txn, events):
-        """Updates the event_backward_extremities tables based on the new/updated
-        events being persisted.
-
-        This is called for new events *and* for events that were outliers, but
-        are now being persisted as non-outliers.
-
-        Forward extremities are handled when we first start persisting the events.
-        """
-        events_by_room = {}
-        for ev in events:
-            events_by_room.setdefault(ev.room_id, []).append(ev)
-
-        query = (
-            "INSERT INTO event_backward_extremities (event_id, room_id)"
-            " SELECT ?, ? WHERE NOT EXISTS ("
-            " SELECT 1 FROM event_backward_extremities"
-            " WHERE event_id = ? AND room_id = ?"
-            " )"
-            " AND NOT EXISTS ("
-            " SELECT 1 FROM events WHERE event_id = ? AND room_id = ? "
-            " AND outlier = ?"
-            " )"
-        )
-
-        txn.executemany(query, [
-            (e_id, ev.room_id, e_id, ev.room_id, e_id, ev.room_id, False)
-            for ev in events for e_id, _ in ev.prev_events
-            if not ev.internal_metadata.is_outlier()
-        ])
-
-        query = (
-            "DELETE FROM event_backward_extremities"
-            " WHERE event_id = ? AND room_id = ?"
-        )
-        txn.executemany(
-            query,
-            [
-                (ev.event_id, ev.room_id) for ev in events
-                if not ev.internal_metadata.is_outlier()
-            ]
-        )
-
-    def _delete_old_forward_extrem_cache(self):
-        def _delete_old_forward_extrem_cache_txn(txn):
-            # Delete entries older than a month, while making sure we don't delete
-            # the only entries for a room.
-            sql = ("""
-                DELETE FROM stream_ordering_to_exterm
-                WHERE
-                room_id IN (
-                    SELECT room_id
-                    FROM stream_ordering_to_exterm
-                    WHERE stream_ordering > ?
-                ) AND stream_ordering < ?
-            """)
-            txn.execute(
-                sql,
-                (self.stream_ordering_month_ago, self.stream_ordering_month_ago,)
-            )
-        return self.runInteraction(
-            "_delete_old_forward_extrem_cache",
-            _delete_old_forward_extrem_cache_txn
-        )
-
     def clean_room_for_join(self, room_id):
         return self.runInteraction(
             "clean_room_for_join",