Merge branch 'develop' into matrix-org-hotfixes

Revert "Use sortedcontainers instead of blist"
This reverts commit 9fbe70a7dc. It turns out that sortedcontainers.SortedDict is not an exact match for blist.sorteddict; in particular, `popitem()` removes things from the opposite end of the dict. This is trivial to fix, but I want to add some unit tests, and potentially some more thought about it, before we do so.
2018-04-13 11:27:29 +01:00 · 2018-04-13 11:16:43 +01:00 · 2018-04-12 13:35:56 +01:00 · 2018-04-12 12:59:36 +01:00 · 2018-04-12 12:18:57 +01:00 · 2018-04-12 12:07:09 +01:00
115 changed files with 1679 additions and 673 deletions
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -1,11 +1,144 @@
-Unreleased
-==========
+Changes in synapse v0.27.3-rc2 (2018-04-09)
+==========================================
+v0.27.3-rc1 used a stale version of the develop branch so the changelog overstates
+the functionality. v0.27.3-rc2 is up to date, rc1 should be ignored.

-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``.
+Changes in synapse v0.27.3-rc1 (2018-04-09)
+=======================================
+
+Notable changes include API support for joinability of groups. Also new metrics 
+and phone home stats. Phone home stats include better visibility of system usage
+so we can tweak synpase to work better for all users rather than our own experience
+with matrix.org. Also, recording 'r30' stat which is the measure we use to track 
+overal growth of the Matrix ecosystem. It is defined as:-
+
+Counts the number of native 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"
+
+
+Features:
+
+* Add joinability for groups (PR #3045)
+* Implement group join API (PR #3046)
+* Add counter metrics for calculating state delta (PR #3033)
+* R30 stats (PR #3041)
+* Measure time it takes to calculate state group ID (PR #3043)
+* Add basic performance statistics to phone home (PR #3044)
+* Add response size metrics (PR #3071)
+* phone home cache size configurations (PR #3063)
+
+Changes:
+
+* Add a blurb explaining the main synapse worker (PR #2886) Thanks to @turt2live!
+* Replace old style error catching with 'as' keyword (PR #3000) Thanks to @NotAFile!
+* Use .iter* to avoid copies in StateHandler (PR #3006)
+* Linearize calls to _generate_user_id (PR #3029)
+* Remove last usage of ujson (PR #3030)
+* Use simplejson throughout (PR #3048)
+* Use static JSONEncoders (PR #3049)
+* Remove uses of events.content (PR #3060)
+* Improve database cache performance (PR #3068)
+
+Bug fixes:
+
+* Add room_id to the response of `rooms/{roomId}/join` (PR #2986) Thanks to @jplatte!
+* Fix replication after switch to simplejson (PR #3015)
+* Fix replication after switch to simplejson (PR #3015)
+* 404 correctly on missing paths via NoResource (PR #3022)
+* Fix error when claiming e2e keys from offline servers (PR #3034)
+* fix tests/storage/test_user_directory.py (PR #3042)
+* use PUT instead of POST for federating groups/m.join_policy (PR #3070) Thanks to @krombel!
+* postgres port script: fix state_groups_pkey error (PR #3072)
+
+
+Changes in synapse v0.27.2 (2018-03-26)
+=======================================
+
+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``.

 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)
--- a/CONTRIBUTING.rst
+++ b/CONTRIBUTING.rst
@@ -30,8 +30,12 @@ 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 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.
+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.

 Code style
 ~~~~~~~~~~
@@ -115,4 +119,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!
--- a/README.rst
+++ b/README.rst
@@ -354,6 +354,10 @@ 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

@@ -890,6 +894,17 @@ 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
 ===================================
--- a/UPGRADE.rst
+++ b/UPGRADE.rst
@@ -48,6 +48,18 @@ 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
 ====================

--- a/contrib/graph/graph3.py
+++ b/contrib/graph/graph3.py
@@ -22,6 +22,8 @@ import argparse
 from synapse.events import FrozenEvent
 from synapse.util.frozenutils import unfreeze

+from six import string_types
+

 def make_graph(file_name, room_id, file_prefix, limit):
    print "Reading lines"
@@ -58,7 +60,7 @@ def make_graph(file_name, room_id, file_prefix, limit):
        for key, value in unfreeze(event.get_dict()["content"]).items():
            if value is None:
                value = "<null>"
-            elif isinstance(value, basestring):
+            elif isinstance(value, string_types):
                pass
            else:
                value = json.dumps(value)
--- a/contrib/prometheus/consoles/synapse.html
+++ b/contrib/prometheus/consoles/synapse.html
@@ -202,11 +202,11 @@ new PromConsole.Graph({
 <h1>Requests</h1>

 <h3>Requests by Servlet</h3>
-<div id="synapse_http_server_requests_servlet"></div>
+<div id="synapse_http_server_request_count_servlet"></div>
 <script>
 new PromConsole.Graph({
-  node: document.querySelector("#synapse_http_server_requests_servlet"),
-  expr: "rate(synapse_http_server_requests:servlet[2m])",
+  node: document.querySelector("#synapse_http_server_request_count_servlet"),
+  expr: "rate(synapse_http_server_request_count:servlet[2m])",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
@@ -215,11 +215,11 @@ new PromConsole.Graph({
 })
 </script>
 <h4>&nbsp;(without <tt>EventStreamRestServlet</tt> or <tt>SyncRestServlet</tt>)</h4>
-<div id="synapse_http_server_requests_servlet_minus_events"></div>
+<div id="synapse_http_server_request_count_servlet_minus_events"></div>
 <script>
 new PromConsole.Graph({
-  node: document.querySelector("#synapse_http_server_requests_servlet_minus_events"),
-  expr: "rate(synapse_http_server_requests:servlet{servlet!=\"EventStreamRestServlet\", servlet!=\"SyncRestServlet\"}[2m])",
+  node: document.querySelector("#synapse_http_server_request_count_servlet_minus_events"),
+  expr: "rate(synapse_http_server_request_count:servlet{servlet!=\"EventStreamRestServlet\", servlet!=\"SyncRestServlet\"}[2m])",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
@@ -233,7 +233,7 @@ new PromConsole.Graph({
 <script>
 new PromConsole.Graph({
  node: document.querySelector("#synapse_http_server_response_time_avg"),
-  expr: "rate(synapse_http_server_response_time:total[2m]) / rate(synapse_http_server_response_time:count[2m]) / 1000",
+  expr: "rate(synapse_http_server_response_time_seconds[2m]) / rate(synapse_http_server_response_count[2m]) / 1000",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
@@ -276,7 +276,7 @@ new PromConsole.Graph({
 <script>
 new PromConsole.Graph({
  node: document.querySelector("#synapse_http_server_response_ru_utime"),
-  expr: "rate(synapse_http_server_response_ru_utime:total[2m])",
+  expr: "rate(synapse_http_server_response_ru_utime_seconds[2m])",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
@@ -291,7 +291,7 @@ new PromConsole.Graph({
 <script>
 new PromConsole.Graph({
  node: document.querySelector("#synapse_http_server_response_db_txn_duration"),
-  expr: "rate(synapse_http_server_response_db_txn_duration:total[2m])",
+  expr: "rate(synapse_http_server_response_db_txn_duration_seconds[2m])",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
@@ -306,7 +306,7 @@ new PromConsole.Graph({
 <script>
 new PromConsole.Graph({
  node: document.querySelector("#synapse_http_server_send_time_avg"),
-  expr: "rate(synapse_http_server_response_time:total{servlet='RoomSendEventRestServlet'}[2m]) / rate(synapse_http_server_response_time:count{servlet='RoomSendEventRestServlet'}[2m]) / 1000",
+  expr: "rate(synapse_http_server_response_time_second{servlet='RoomSendEventRestServlet'}[2m]) / rate(synapse_http_server_response_count{servlet='RoomSendEventRestServlet'}[2m]) / 1000",
  name: "[[servlet]]",
  yAxisFormatter: PromConsole.NumberFormatter.humanize,
  yHoverFormatter: PromConsole.NumberFormatter.humanize,
--- a/contrib/prometheus/synapse-v1.rules
+++ b/contrib/prometheus/synapse-v1.rules
@@ -1,10 +1,10 @@
 synapse_federation_transaction_queue_pendingEdus:total = sum(synapse_federation_transaction_queue_pendingEdus or absent(synapse_federation_transaction_queue_pendingEdus)*0)
 synapse_federation_transaction_queue_pendingPdus:total = sum(synapse_federation_transaction_queue_pendingPdus or absent(synapse_federation_transaction_queue_pendingPdus)*0)

-synapse_http_server_requests:method{servlet=""} = sum(synapse_http_server_requests) by (method)
-synapse_http_server_requests:servlet{method=""} = sum(synapse_http_server_requests) by (servlet)
+synapse_http_server_request_count:method{servlet=""} = sum(synapse_http_server_request_count) by (method)
+synapse_http_server_request_count:servlet{method=""} = sum(synapse_http_server_request_count) by (servlet)

-synapse_http_server_requests:total{servlet=""} = sum(synapse_http_server_requests:by_method) by (servlet)
+synapse_http_server_request_count:total{servlet=""} = sum(synapse_http_server_request_count:by_method) by (servlet)

 synapse_cache:hit_ratio_5m = rate(synapse_util_caches_cache:hits[5m]) / rate(synapse_util_caches_cache:total[5m])
 synapse_cache:hit_ratio_30s = rate(synapse_util_caches_cache:hits[30s]) / rate(synapse_util_caches_cache:total[30s])
--- a/contrib/prometheus/synapse-v2.rules
+++ b/contrib/prometheus/synapse-v2.rules
@@ -5,19 +5,19 @@ groups:
    expr: "sum(synapse_federation_transaction_queue_pendingEdus or absent(synapse_federation_transaction_queue_pendingEdus)*0)"
  - record: "synapse_federation_transaction_queue_pendingPdus:total"
    expr:   "sum(synapse_federation_transaction_queue_pendingPdus or absent(synapse_federation_transaction_queue_pendingPdus)*0)"
-  - record: 'synapse_http_server_requests:method'
+  - record: 'synapse_http_server_request_count:method'
    labels:
      servlet: ""
-    expr: "sum(synapse_http_server_requests) by (method)"
-  - record: 'synapse_http_server_requests:servlet'
+    expr: "sum(synapse_http_server_request_count) by (method)"
+  - record: 'synapse_http_server_request_count:servlet'
    labels:
      method: ""
-    expr: 'sum(synapse_http_server_requests) by (servlet)'
+    expr: 'sum(synapse_http_server_request_count) by (servlet)'

-  - record: 'synapse_http_server_requests:total'
+  - record: 'synapse_http_server_request_count:total'
    labels:
      servlet: ""
-    expr: 'sum(synapse_http_server_requests:by_method) by (servlet)'
+    expr: 'sum(synapse_http_server_request_count:by_method) by (servlet)'

  - record: 'synapse_cache:hit_ratio_5m'
    expr: 'rate(synapse_util_caches_cache:hits[5m]) / rate(synapse_util_caches_cache:total[5m])'
--- a/docs/admin_api/purge_history_api.rst
+++ b/docs/admin_api/purge_history_api.rst
@@ -16,9 +16,11 @@ 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, and room state data before the cutoff is always removed).
+deleted.)

-To delete local events as well, set ``delete_local_events`` in the body:
+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:

 .. code:: json

--- a/docs/workers.rst
+++ b/docs/workers.rst
@@ -55,7 +55,12 @@ 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.
+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

 Each worker configuration file inherits the configuration of the main homeserver
 configuration file.  You can then override configuration specific to that worker,
@@ -230,9 +235,11 @@ file. For example::
 ``synapse.app.event_creator``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

-Handles non-state event creation. It can handle REST endpoints matching:
+Handles some 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.
--- a/scripts/synapse_port_db
+++ b/scripts/synapse_port_db
@@ -1,6 +1,7 @@
 #!/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.
@@ -29,6 +30,8 @@ import time
 import traceback
 import yaml

+from six import string_types
+

 logger = logging.getLogger("synapse_port_db")

@@ -250,6 +253,12 @@ 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

@@ -467,31 +476,10 @@ class Porter(object):
            self.progress.set_state("Preparing PostgreSQL")
            self.setup_db(postgres_config, postgres_engine)

-            # 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))
-
+            self.progress.set_state("Creating port tables")
            def create_port_table(txn):
                txn.execute(
-                    "CREATE TABLE port_from_sqlite3 ("
+                    "CREATE TABLE IF NOT EXISTS port_from_sqlite3 ("
                    " table_name varchar(100) NOT NULL UNIQUE,"
                    " forward_rowid bigint NOT NULL,"
                    " backward_rowid bigint NOT NULL"
@@ -517,18 +505,33 @@ class Porter(object):
                    "alter_table", alter_table
                )
            except Exception as e:
-                logger.info("Failed to create port table: %s", e)
+                pass

-            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)
+            yield self.postgres_store.runInteraction(
+                "create_port_table", create_port_table
+            )

-            self.progress.set_state("Setting up")
+            # 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",
+            )

-            # Set up tables.
+            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")
            setup_res = yield defer.gatherResults(
                [
                    self.setup_table(table)
@@ -539,7 +542,8 @@ class Porter(object):
                consumeErrors=True,
            )

-            # Process tables.
+            # Step 4. Do the copying.
+            self.progress.set_state("Copying to postgres")
            yield defer.gatherResults(
                [
                    self.handle_table(*res)
@@ -548,6 +552,9 @@ 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
@@ -569,7 +576,7 @@ class Porter(object):
        def conv(j, col):
            if j in bool_cols:
                return bool(col)
-            elif isinstance(col, basestring) and "\0" in col:
+            elif isinstance(col, string_types) and "\0" in col:
                logger.warn("DROPPING ROW: NUL value in table %s col %s: %r", table, headers[j], col)
                raise BadValueException();
            return col
@@ -707,6 +714,16 @@ 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 ######
--- a/synapse/init.py
+++ b/synapse/init.py
@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """

-__version__ = "0.26.0"
+__version__ = "0.27.3-rc2"
--- a/synapse/api/auth.py
+++ b/synapse/api/auth.py
@@ -204,8 +204,8 @@ class Auth(object):

            ip_addr = self.hs.get_ip_from_request(request)
            user_agent = request.requestHeaders.getRawHeaders(
-                "User-Agent",
-                default=[""]
+                b"User-Agent",
+                default=[b""]
            )[0]
            if user and access_token and ip_addr:
                self.store.insert_client_ip(
@@ -672,7 +672,7 @@ def has_access_token(request):
        bool: False if no access_token was given, True otherwise.
    """
    query_params = request.args.get("access_token")
-    auth_headers = request.requestHeaders.getRawHeaders("Authorization")
+    auth_headers = request.requestHeaders.getRawHeaders(b"Authorization")
    return bool(query_params) or bool(auth_headers)


@@ -692,8 +692,8 @@ def get_access_token_from_request(request, token_not_found_http_status=401):
        AuthError: If there isn't an access_token in the request.
    """

-    auth_headers = request.requestHeaders.getRawHeaders("Authorization")
-    query_params = request.args.get("access_token")
+    auth_headers = request.requestHeaders.getRawHeaders(b"Authorization")
+    query_params = request.args.get(b"access_token")
    if auth_headers:
        # Try the get the access_token from a "Authorization: Bearer"
        # header
--- a/synapse/api/errors.py
+++ b/synapse/api/errors.py
@@ -15,9 +15,10 @@

 """Contains exceptions and error codes."""

-import json
 import logging

+import simplejson as json
+
 logger = logging.getLogger(__name__)


--- a/synapse/api/filtering.py
+++ b/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 ujson as json
+import simplejson as json
 import jsonschema
 from jsonschema import FormatChecker

--- a/synapse/app/appservice.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 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, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/client_reader.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 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, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/event_creator.py
+++ b/synapse/app/event_creator.py
@@ -52,7 +52,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 Resource
+from twisted.web.resource import NoResource

 logger = logging.getLogger("synapse.app.event_creator")

@@ -104,7 +104,7 @@ class EventCreatorServer(HomeServer):
                        "/_matrix/client/api/v1": resource,
                    })

-        root_resource = create_resource_tree(resources, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/federation_reader.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 logger = logging.getLogger("synapse.app.federation_reader")

@@ -77,7 +77,7 @@ class FederationReaderServer(HomeServer):
                        FEDERATION_PREFIX: TransportLayerServer(self),
                    })

-        root_resource = create_resource_tree(resources, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/federation_sender.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 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, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/frontend_proxy.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 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>[^/]+))?$")

@@ -90,7 +120,7 @@ class KeyUploadServlet(RestServlet):
            # They're actually trying to upload something, proxy to main synapse.
            # Pass through the auth headers, if any, in case the access token
            # is there.
-            auth_headers = request.requestHeaders.getRawHeaders("Authorization", [])
+            auth_headers = request.requestHeaders.getRawHeaders(b"Authorization", [])
            headers = {
                "Authorization": auth_headers,
            }
@@ -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, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -48,6 +48,7 @@ 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
@@ -56,7 +57,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, Resource
+from twisted.web.resource import EncodingResourceWrapper, NoResource
 from twisted.web.server import GzipEncoderFactory
 from twisted.web.static import File

@@ -126,7 +127,7 @@ class SynapseHomeServer(HomeServer):
        if WEB_CLIENT_PREFIX in resources:
            root_resource = RootRedirect(WEB_CLIENT_PREFIX)
        else:
-            root_resource = Resource()
+            root_resource = NoResource()

        root_resource = create_resource_tree(resources, root_resource)

@@ -402,6 +403,10 @@ 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")
@@ -425,8 +430,21 @@ 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:
@@ -437,10 +455,32 @@ 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)
--- a/synapse/app/media_repository.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 logger = logging.getLogger("synapse.app.media_repository")

@@ -84,7 +84,7 @@ class MediaRepositoryServer(HomeServer):
                        ),
                    })

-        root_resource = create_resource_tree(resources, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/pusher.py
+++ b/synapse/app/pusher.py
@@ -37,7 +37,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 Resource
+from twisted.web.resource import NoResource

 logger = logging.getLogger("synapse.app.pusher")

@@ -94,7 +94,7 @@ class PusherServer(HomeServer):
                if name == "metrics":
                    resources[METRICS_PREFIX] = MetricsResource(self)

-        root_resource = create_resource_tree(resources, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/synchrotron.py
+++ b/synapse/app/synchrotron.py
@@ -56,7 +56,7 @@ 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 Resource
+from twisted.web.resource import NoResource

 logger = logging.getLogger("synapse.app.synchrotron")

@@ -113,6 +113,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 +211,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 +272,7 @@ class SynchrotronServer(HomeServer):
                        "/_matrix/client/api/v1": resource,
                    })

-        root_resource = create_resource_tree(resources, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/app/synctl.py
+++ b/synapse/app/synctl.py
@@ -38,7 +38,7 @@ def pid_running(pid):
    try:
        os.kill(pid, 0)
        return True
-    except OSError, err:
+    except OSError as 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, err:
+        except OSError as err:
            if err.errno == errno.ESRCH:
                write("%s not running" % (app,), colour=YELLOW)
            elif err.errno == errno.EPERM:
@@ -108,7 +108,7 @@ def stop(pidfile, app):


 Worker = collections.namedtuple("Worker", [
-    "app", "configfile", "pidfile", "cache_factor"
+    "app", "configfile", "pidfile", "cache_factor", "cache_factors",
 ])


@@ -171,6 +171,10 @@ def main():
    if cache_factor:
        os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)

+    cache_factors = config.get("synctl_cache_factors", {})
+    for cache_name, factor in cache_factors.iteritems():
+        os.environ["SYNAPSE_CACHE_FACTOR_" + cache_name.upper()] = str(factor)
+
    worker_configfiles = []
    if options.worker:
        start_stop_synapse = False
@@ -211,6 +215,10 @@ def main():
                or pidfile
            )
            worker_cache_factor = worker_config.get("synctl_cache_factor") or cache_factor
+            worker_cache_factors = (
+                worker_config.get("synctl_cache_factors")
+                or cache_factors
+            )
            daemonize = worker_config.get("daemonize") or config.get("daemonize")
            assert daemonize, "Main process must have daemonize set to true"

@@ -226,8 +234,10 @@ def main():
            assert worker_daemonize, "In config %r: expected '%s' to be True" % (
                worker_configfile, "worker_daemonize")
            worker_cache_factor = worker_config.get("synctl_cache_factor")
+            worker_cache_factors = worker_config.get("synctl_cache_factors", {})
        workers.append(Worker(
            worker_app, worker_configfile, worker_pidfile, worker_cache_factor,
+            worker_cache_factors,
        ))

    action = options.action
@@ -252,6 +262,7 @@ 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:
@@ -261,15 +272,19 @@ def main():
            start(configfile)

        for worker in workers:
+            env = os.environ.copy()
+
            if worker.cache_factor:
                os.environ["SYNAPSE_CACHE_FACTOR"] = str(worker.cache_factor)

+            for cache_name, factor in worker.cache_factors.iteritems():
+                os.environ["SYNAPSE_CACHE_FACTOR_" + cache_name.upper()] = str(factor)
+
            start_worker(worker.app, configfile, worker.configfile)

-            if cache_factor:
-                os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)
-            else:
-                os.environ.pop("SYNAPSE_CACHE_FACTOR", None)
+            # Reset env back to the original
+            os.environ.clear()
+            os.environ.update(env)


 if __name__ == "__main__":
--- a/synapse/app/user_dir.py
+++ b/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 Resource
+from twisted.web.resource import NoResource

 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, Resource())
+        root_resource = create_resource_tree(resources, NoResource())

        _base.listen_tcp(
            bind_addresses,
--- a/synapse/appservice/init.py
+++ b/synapse/appservice/init.py
@@ -21,6 +21,8 @@ from twisted.internet import defer
 import logging
 import re

+from six import string_types
+
 logger = logging.getLogger(__name__)


@@ -146,7 +148,7 @@ class ApplicationService(object):
                        )

                regex = regex_obj.get("regex")
-                if isinstance(regex, basestring):
+                if isinstance(regex, string_types):
                    regex_obj["regex"] = re.compile(regex)  # Pre-compile regex
                else:
                    raise ValueError(
--- a/synapse/appservice/api.py
+++ b/synapse/appservice/api.py
@@ -73,7 +73,8 @@ class ApplicationServiceApi(SimpleHttpClient):
        super(ApplicationServiceApi, self).__init__(hs)
        self.clock = hs.get_clock()

-        self.protocol_meta_cache = ResponseCache(hs, timeout_ms=HOUR_IN_MS)
+        self.protocol_meta_cache = ResponseCache(hs, "as_protocol_meta",
+                                                 timeout_ms=HOUR_IN_MS)

    @defer.inlineCallbacks
    def query_user(self, service, user_id):
--- a/synapse/config/_base.py
+++ b/synapse/config/_base.py
@@ -19,6 +19,8 @@ import os
 import yaml
 from textwrap import dedent

+from six import integer_types
+

 class ConfigError(Exception):
    pass
@@ -49,7 +51,7 @@ Missing mandatory `server_name` config option.
 class Config(object):
    @staticmethod
    def parse_size(value):
-        if isinstance(value, int) or isinstance(value, long):
+        if isinstance(value, integer_types):
            return value
        sizes = {"K": 1024, "M": 1024 * 1024}
        size = 1
@@ -61,7 +63,7 @@ class Config(object):

    @staticmethod
    def parse_duration(value):
-        if isinstance(value, int) or isinstance(value, long):
+        if isinstance(value, integer_types):
            return value
        second = 1000
        minute = 60 * second
@@ -288,22 +290,22 @@ class Config(object):
                    )
                    obj.invoke_all("generate_files", config)
                    config_file.write(config_bytes)
-                print (
+                print((
                    "A config file has been generated in %r for server name"
                    " %r with corresponding SSL keys and self-signed"
                    " certificates. Please review this file and customise it"
                    " to your needs."
-                ) % (config_path, server_name)
-                print (
+                ) % (config_path, server_name))
+                print(
                    "If this server name is incorrect, you will need to"
                    " regenerate the SSL certificates"
                )
                return
            else:
-                print (
+                print((
                    "Config file %r already exists. Generating any missing key"
                    " files."
-                ) % (config_path,)
+                ) % (config_path,))
                generate_keys = True

        parser = argparse.ArgumentParser(
--- a/synapse/config/appservice.py
+++ b/synapse/config/appservice.py
@@ -21,6 +21,8 @@ import urllib
 import yaml
 import logging

+from six import string_types
+
 logger = logging.getLogger(__name__)


@@ -89,14 +91,14 @@ def _load_appservice(hostname, as_info, config_filename):
        "id", "as_token", "hs_token", "sender_localpart"
    ]
    for field in required_string_fields:
-        if not isinstance(as_info.get(field), basestring):
+        if not isinstance(as_info.get(field), string_types):
            raise KeyError("Required string field: '%s' (%s)" % (
                field, config_filename,
            ))

    # 'url' must either be a string or explicitly null, not missing
    # to avoid accidentally turning off push for ASes.
-    if (not isinstance(as_info.get("url"), basestring) and
+    if (not isinstance(as_info.get("url"), string_types) and
            as_info.get("url", "") is not None):
        raise KeyError(
            "Required string field or explicit null: 'url' (%s)" % (config_filename,)
@@ -128,7 +130,7 @@ def _load_appservice(hostname, as_info, config_filename):
                        "Expected namespace entry in %s to be an object,"
                        " but got %s", ns, regex_obj
                    )
-                if not isinstance(regex_obj.get("regex"), basestring):
+                if not isinstance(regex_obj.get("regex"), string_types):
                    raise ValueError(
                        "Missing/bad type 'regex' key in %s", regex_obj
                    )
--- a/synapse/config/registration.py
+++ b/synapse/config/registration.py
@@ -77,7 +77,9 @@ 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 of rounds is 12.
+        # 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.
        bcrypt_rounds: 12

        # Allows users to register as guests without a password/email/etc, and
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -65,7 +65,7 @@ class FederationServer(FederationBase):

        # 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)
+        self._state_resp_cache = ResponseCache(hs, "state_resp", timeout_ms=30000)

    @defer.inlineCallbacks
    @log_function
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -169,7 +169,7 @@ class TransactionQueue(object):
            while True:
                last_token = yield self.store.get_federation_out_pos("events")
                next_token, events = yield self.store.get_all_new_events_stream(
-                    last_token, self._last_poked_id, limit=20,
+                    last_token, self._last_poked_id, limit=100,
                )

                logger.debug("Handling %s -> %s", last_token, next_token)
@@ -177,24 +177,33 @@ class TransactionQueue(object):
                if not events and next_token >= self._last_poked_id:
                    break

-                for event in events:
+                @defer.inlineCallbacks
+                def handle_event(event):
                    # Only send events for this server.
                    send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of()
                    is_mine = self.is_mine_id(event.event_id)
                    if not is_mine and send_on_behalf_of is None:
-                        continue
+                        return
+
+                    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 for event: %s",
+                            event.event_id,
+                        )
+                        return

-                    # 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
-                        ],
-                    )
                    destinations = set(destinations)

                    if send_on_behalf_of is not None:
@@ -207,12 +216,44 @@ class TransactionQueue(object):

                    self._send_pdu(event, destinations)

-                events_processed_counter.inc_by(len(events))
+                @defer.inlineCallbacks
+                def handle_room_events(events):
+                    for event in events:
+                        yield handle_event(event)
+
+                events_by_room = {}
+                for event in events:
+                    events_by_room.setdefault(event.room_id, []).append(event)
+
+                yield logcontext.make_deferred_yieldable(defer.gatherResults(
+                    [
+                        logcontext.run_in_background(handle_room_events, evs)
+                        for evs in events_by_room.itervalues()
+                    ],
+                    consumeErrors=True
+                ))

                yield self.store.update_federation_out_pos(
                    "events", next_token
                )

+                if events:
+                    now = self.clock.time_msec()
+                    ts = yield self.store.get_received_ts(events[-1].event_id)
+
+                    synapse.metrics.event_processing_lag.set(
+                        now - ts, "federation_sender",
+                    )
+                    synapse.metrics.event_processing_last_ts.set(
+                        ts, "federation_sender",
+                    )
+
+                events_processed_counter.inc_by(len(events))
+
+                synapse.metrics.event_processing_positions.set(
+                    next_token, "federation_sender",
+                )
+
        finally:
            self._is_processing = False

@@ -254,6 +295,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
--- a/synapse/federation/transport/client.py
+++ b/synapse/federation/transport/client.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -20,6 +21,7 @@ from synapse.api.urls import FEDERATION_PREFIX as PREFIX
 from synapse.util.logutils import log_function

 import logging
+import urllib


 logger = logging.getLogger(__name__)
@@ -49,7 +51,7 @@ class TransportLayerClient(object):
        logger.debug("get_room_state dest=%s, room=%s",
                     destination, room_id)

-        path = PREFIX + "/state/%s/" % room_id
+        path = _create_path(PREFIX, "/state/%s/", room_id)
        return self.client.get_json(
            destination, path=path, args={"event_id": event_id},
        )
@@ -71,7 +73,7 @@ class TransportLayerClient(object):
        logger.debug("get_room_state_ids dest=%s, room=%s",
                     destination, room_id)

-        path = PREFIX + "/state_ids/%s/" % room_id
+        path = _create_path(PREFIX, "/state_ids/%s/", room_id)
        return self.client.get_json(
            destination, path=path, args={"event_id": event_id},
        )
@@ -93,7 +95,7 @@ class TransportLayerClient(object):
        logger.debug("get_pdu dest=%s, event_id=%s",
                     destination, event_id)

-        path = PREFIX + "/event/%s/" % (event_id, )
+        path = _create_path(PREFIX, "/event/%s/", event_id)
        return self.client.get_json(destination, path=path, timeout=timeout)

    @log_function
@@ -119,7 +121,7 @@ class TransportLayerClient(object):
            # TODO: raise?
            return

-        path = PREFIX + "/backfill/%s/" % (room_id,)
+        path = _create_path(PREFIX, "/backfill/%s/", room_id)

        args = {
            "v": event_tuples,
@@ -157,9 +159,11 @@ 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=PREFIX + "/send/%s/" % transaction.transaction_id,
+            path=path,
            data=json_data,
            json_data_callback=json_data_callback,
            long_retries=True,
@@ -177,7 +181,7 @@ class TransportLayerClient(object):
    @log_function
    def make_query(self, destination, query_type, args, retry_on_dns_fail,
                   ignore_backoff=False):
-        path = PREFIX + "/query/%s" % query_type
+        path = _create_path(PREFIX, "/query/%s", query_type)

        content = yield self.client.get_json(
            destination=destination,
@@ -222,7 +226,7 @@ class TransportLayerClient(object):
                "make_membership_event called with membership='%s', must be one of %s" %
                (membership, ",".join(valid_memberships))
            )
-        path = PREFIX + "/make_%s/%s/%s" % (membership, room_id, user_id)
+        path = _create_path(PREFIX, "/make_%s/%s/%s", membership, room_id, user_id)

        ignore_backoff = False
        retry_on_dns_fail = False
@@ -248,7 +252,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def send_join(self, destination, room_id, event_id, content):
-        path = PREFIX + "/send_join/%s/%s" % (room_id, event_id)
+        path = _create_path(PREFIX, "/send_join/%s/%s", room_id, event_id)

        response = yield self.client.put_json(
            destination=destination,
@@ -261,7 +265,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def send_leave(self, destination, room_id, event_id, content):
-        path = PREFIX + "/send_leave/%s/%s" % (room_id, event_id)
+        path = _create_path(PREFIX, "/send_leave/%s/%s", room_id, event_id)

        response = yield self.client.put_json(
            destination=destination,
@@ -280,7 +284,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def send_invite(self, destination, room_id, event_id, content):
-        path = PREFIX + "/invite/%s/%s" % (room_id, event_id)
+        path = _create_path(PREFIX, "/invite/%s/%s", room_id, event_id)

        response = yield self.client.put_json(
            destination=destination,
@@ -322,7 +326,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def exchange_third_party_invite(self, destination, room_id, event_dict):
-        path = PREFIX + "/exchange_third_party_invite/%s" % (room_id,)
+        path = _create_path(PREFIX, "/exchange_third_party_invite/%s", room_id,)

        response = yield self.client.put_json(
            destination=destination,
@@ -335,7 +339,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def get_event_auth(self, destination, room_id, event_id):
-        path = PREFIX + "/event_auth/%s/%s" % (room_id, event_id)
+        path = _create_path(PREFIX, "/event_auth/%s/%s", room_id, event_id)

        content = yield self.client.get_json(
            destination=destination,
@@ -347,7 +351,7 @@ class TransportLayerClient(object):
    @defer.inlineCallbacks
    @log_function
    def send_query_auth(self, destination, room_id, event_id, content):
-        path = PREFIX + "/query_auth/%s/%s" % (room_id, event_id)
+        path = _create_path(PREFIX, "/query_auth/%s/%s", room_id, event_id)

        content = yield self.client.post_json(
            destination=destination,
@@ -409,7 +413,7 @@ class TransportLayerClient(object):
        Returns:
            A dict containg the device keys.
        """
-        path = PREFIX + "/user/devices/" + user_id
+        path = _create_path(PREFIX, "/user/devices/%s", user_id)

        content = yield self.client.get_json(
            destination=destination,
@@ -459,7 +463,7 @@ class TransportLayerClient(object):
    @log_function
    def get_missing_events(self, destination, room_id, earliest_events,
                           latest_events, limit, min_depth, timeout):
-        path = PREFIX + "/get_missing_events/%s" % (room_id,)
+        path = _create_path(PREFIX, "/get_missing_events/%s", room_id,)

        content = yield self.client.post_json(
            destination=destination,
@@ -479,7 +483,7 @@ class TransportLayerClient(object):
    def get_group_profile(self, destination, group_id, requester_user_id):
        """Get a group profile
        """
-        path = PREFIX + "/groups/%s/profile" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/profile", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -498,7 +502,7 @@ class TransportLayerClient(object):
            requester_user_id (str)
            content (dict): The new profile of the group
        """
-        path = PREFIX + "/groups/%s/profile" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/profile", group_id,)

        return self.client.post_json(
            destination=destination,
@@ -512,7 +516,7 @@ class TransportLayerClient(object):
    def get_group_summary(self, destination, group_id, requester_user_id):
        """Get a group summary
        """
-        path = PREFIX + "/groups/%s/summary" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/summary", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -525,7 +529,7 @@ class TransportLayerClient(object):
    def get_rooms_in_group(self, destination, group_id, requester_user_id):
        """Get all rooms in a group
        """
-        path = PREFIX + "/groups/%s/rooms" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/rooms", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -538,7 +542,7 @@ class TransportLayerClient(object):
                          content):
        """Add a room to a group
        """
-        path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
+        path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,)

        return self.client.post_json(
            destination=destination,
@@ -552,7 +556,10 @@ class TransportLayerClient(object):
                             config_key, content):
        """Update room in group
        """
-        path = PREFIX + "/groups/%s/room/%s/config/%s" % (group_id, room_id, config_key,)
+        path = _create_path(
+            PREFIX, "/groups/%s/room/%s/config/%s",
+            group_id, room_id, config_key,
+        )

        return self.client.post_json(
            destination=destination,
@@ -565,7 +572,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 = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,)
+        path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,)

        return self.client.delete_json(
            destination=destination,
@@ -578,7 +585,7 @@ class TransportLayerClient(object):
    def get_users_in_group(self, destination, group_id, requester_user_id):
        """Get users in a group
        """
-        path = PREFIX + "/groups/%s/users" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/users", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -591,7 +598,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 = PREFIX + "/groups/%s/invited_users" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/invited_users", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -604,7 +611,23 @@ class TransportLayerClient(object):
    def accept_group_invite(self, destination, group_id, user_id, content):
        """Accept a group invite
        """
-        path = PREFIX + "/groups/%s/users/%s/accept_invite" % (group_id, user_id)
+        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)

        return self.client.post_json(
            destination=destination,
@@ -617,7 +640,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 = PREFIX + "/groups/%s/users/%s/invite" % (group_id, user_id)
+        path = _create_path(PREFIX, "/groups/%s/users/%s/invite", group_id, user_id)

        return self.client.post_json(
            destination=destination,
@@ -633,7 +656,7 @@ class TransportLayerClient(object):
        invited.
        """

-        path = PREFIX + "/groups/local/%s/users/%s/invite" % (group_id, user_id)
+        path = _create_path(PREFIX, "/groups/local/%s/users/%s/invite", group_id, user_id)

        return self.client.post_json(
            destination=destination,
@@ -647,7 +670,7 @@ class TransportLayerClient(object):
                               user_id, content):
        """Remove a user fron a group
        """
-        path = PREFIX + "/groups/%s/users/%s/remove" % (group_id, user_id)
+        path = _create_path(PREFIX, "/groups/%s/users/%s/remove", group_id, user_id)

        return self.client.post_json(
            destination=destination,
@@ -664,7 +687,7 @@ class TransportLayerClient(object):
        kicked from the group.
        """

-        path = PREFIX + "/groups/local/%s/users/%s/remove" % (group_id, user_id)
+        path = _create_path(PREFIX, "/groups/local/%s/users/%s/remove", group_id, user_id)

        return self.client.post_json(
            destination=destination,
@@ -679,7 +702,7 @@ class TransportLayerClient(object):
        the attestations
        """

-        path = PREFIX + "/groups/%s/renew_attestation/%s" % (group_id, user_id)
+        path = _create_path(PREFIX, "/groups/%s/renew_attestation/%s", group_id, user_id)

        return self.client.post_json(
            destination=destination,
@@ -694,11 +717,12 @@ class TransportLayerClient(object):
        """Update a room entry in a group summary
        """
        if category_id:
-            path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
+            path = _create_path(
+                PREFIX, "/groups/%s/summary/categories/%s/rooms/%s",
                group_id, category_id, room_id,
            )
        else:
-            path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
+            path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,)

        return self.client.post_json(
            destination=destination,
@@ -714,11 +738,12 @@ class TransportLayerClient(object):
        """Delete a room entry in a group summary
        """
        if category_id:
-            path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % (
+            path = _create_path(
+                PREFIX + "/groups/%s/summary/categories/%s/rooms/%s",
                group_id, category_id, room_id,
            )
        else:
-            path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,)
+            path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,)

        return self.client.delete_json(
            destination=destination,
@@ -731,7 +756,7 @@ class TransportLayerClient(object):
    def get_group_categories(self, destination, group_id, requester_user_id):
        """Get all categories in a group
        """
-        path = PREFIX + "/groups/%s/categories" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/categories", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -744,7 +769,7 @@ class TransportLayerClient(object):
    def get_group_category(self, destination, group_id, requester_user_id, category_id):
        """Get category info in a group
        """
-        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
+        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)

        return self.client.get_json(
            destination=destination,
@@ -758,7 +783,7 @@ class TransportLayerClient(object):
                              content):
        """Update a category in a group
        """
-        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
+        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)

        return self.client.post_json(
            destination=destination,
@@ -773,7 +798,7 @@ class TransportLayerClient(object):
                              category_id):
        """Delete a category in a group
        """
-        path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,)
+        path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,)

        return self.client.delete_json(
            destination=destination,
@@ -786,7 +811,7 @@ class TransportLayerClient(object):
    def get_group_roles(self, destination, group_id, requester_user_id):
        """Get all roles in a group
        """
-        path = PREFIX + "/groups/%s/roles" % (group_id,)
+        path = _create_path(PREFIX, "/groups/%s/roles", group_id,)

        return self.client.get_json(
            destination=destination,
@@ -799,7 +824,7 @@ class TransportLayerClient(object):
    def get_group_role(self, destination, group_id, requester_user_id, role_id):
        """Get a roles info
        """
-        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
+        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)

        return self.client.get_json(
            destination=destination,
@@ -813,7 +838,7 @@ class TransportLayerClient(object):
                          content):
        """Update a role in a group
        """
-        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
+        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)

        return self.client.post_json(
            destination=destination,
@@ -827,7 +852,7 @@ class TransportLayerClient(object):
    def delete_group_role(self, destination, group_id, requester_user_id, role_id):
        """Delete a role in a group
        """
-        path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,)
+        path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,)

        return self.client.delete_json(
            destination=destination,
@@ -842,11 +867,12 @@ class TransportLayerClient(object):
        """Update a users entry in a group
        """
        if role_id:
-            path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
+            path = _create_path(
+                PREFIX, "/groups/%s/summary/roles/%s/users/%s",
                group_id, role_id, user_id,
            )
        else:
-            path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
+            path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,)

        return self.client.post_json(
            destination=destination,
@@ -856,17 +882,33 @@ 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 = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % (
+            path = _create_path(
+                PREFIX, "/groups/%s/summary/roles/%s/users/%s",
                group_id, role_id, user_id,
            )
        else:
-            path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,)
+            path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,)

        return self.client.delete_json(
            destination=destination,
@@ -889,3 +931,22 @@ 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)
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -802,6 +803,23 @@ 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
    """
@@ -1124,6 +1142,24 @@ 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,
@@ -1163,6 +1199,7 @@ GROUP_SERVER_SERVLET_CLASSES = (
    FederationGroupsInvitedUsersServlet,
    FederationGroupsInviteServlet,
    FederationGroupsAcceptInviteServlet,
+    FederationGroupsJoinServlet,
    FederationGroupsRemoveUserServlet,
    FederationGroupsSummaryRoomsServlet,
    FederationGroupsCategoriesServlet,
@@ -1172,6 +1209,7 @@ GROUP_SERVER_SERVLET_CLASSES = (
    FederationGroupsSummaryUsersServlet,
    FederationGroupsAddRoomsServlet,
    FederationGroupsAddRoomsConfigServlet,
+    FederationGroupsSettingJoinPolicyServlet,
 )


--- a/synapse/groups/groups_server.py
+++ b/synapse/groups/groups_server.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -205,6 +206,28 @@ 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)
@@ -381,9 +404,16 @@ class GroupsServerHandler(object):

        yield self.check_group_is_ours(group_id, requester_user_id)

-        group_description = yield self.store.get_group(group_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"

-        if group_description:
            defer.returnValue(group_description)
        else:
            raise SynapseError(404, "Unknown group")
@@ -654,6 +684,40 @@ 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.
@@ -670,30 +734,27 @@ class GroupsServerHandler(object):
        if not is_invited:
            raise SynapseError(403, "User not invited to group")

-        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"]
+        local_attestation = yield self._add_user(group_id, requester_user_id, content)

-            yield self.attestations.verify_attestation(
-                remote_attestation,
-                user_id=requester_user_id,
-                group_id=group_id,
-            )
-        else:
-            local_attestation = None
-            remote_attestation = None
+        defer.returnValue({
+            "state": "join",
+            "attestation": local_attestation,
+        })

-        is_public = _parse_visibility_from_contents(content)
+    @defer.inlineCallbacks
+    def join_group(self, group_id, requester_user_id, content):
+        """User tries to join the group.

-        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,
+        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
        )
+        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",
@@ -835,6 +896,31 @@ 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
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -18,7 +18,9 @@ from twisted.internet import defer
 import synapse
 from synapse.api.constants import EventTypes
 from synapse.util.metrics import Measure
-from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
+from synapse.util.logcontext import (
+    make_deferred_yieldable, preserve_fn, run_in_background,
+)

 import logging

@@ -84,11 +86,16 @@ class ApplicationServicesHandler(object):
                    if not events:
                        break

+                    events_by_room = {}
                    for event in events:
+                        events_by_room.setdefault(event.room_id, []).append(event)
+
+                    @defer.inlineCallbacks
+                    def handle_event(event):
                        # Gather interested services
                        services = yield self._get_services_for_event(event)
                        if len(services) == 0:
-                            continue  # no services need notifying
+                            return  # no services need notifying

                        # Do we know this user exists? If not, poke the user
                        # query API for all services which match that user regex.
@@ -108,9 +115,33 @@ class ApplicationServicesHandler(object):
                                service, event
                            )

-                    events_processed_counter.inc_by(len(events))
+                    @defer.inlineCallbacks
+                    def handle_room_events(events):
+                        for event in events:
+                            yield handle_event(event)
+
+                    yield make_deferred_yieldable(defer.gatherResults([
+                        run_in_background(handle_room_events, evs)
+                        for evs in events_by_room.itervalues()
+                    ], consumeErrors=True))

                    yield self.store.set_appservice_last_pos(upper_bound)
+
+                    now = self.clock.time_msec()
+                    ts = yield self.store.get_received_ts(events[-1].event_id)
+
+                    synapse.metrics.event_processing_positions.set(
+                        upper_bound, "appservice_sender",
+                    )
+
+                    events_processed_counter.inc_by(len(events))
+
+                    synapse.metrics.event_processing_lag.set(
+                        now - ts, "appservice_sender",
+                    )
+                    synapse.metrics.event_processing_last_ts.set(
+                        ts, "appservice_sender",
+                    )
            finally:
                self.is_processing = False

--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -155,7 +155,7 @@ class DeviceHandler(BaseHandler):

        try:
            yield self.store.delete_device(user_id, device_id)
-        except errors.StoreError, e:
+        except errors.StoreError as e:
            if e.code == 404:
                # no match
                pass
@@ -204,7 +204,7 @@ class DeviceHandler(BaseHandler):

        try:
            yield self.store.delete_devices(user_id, device_ids)
-        except errors.StoreError, e:
+        except errors.StoreError as e:
            if e.code == 404:
                # no match
                pass
@@ -243,7 +243,7 @@ class DeviceHandler(BaseHandler):
                new_display_name=content.get("display_name")
            )
            yield self.notify_device_update(user_id, [device_id])
-        except errors.StoreError, e:
+        except errors.StoreError as e:
            if e.code == 404:
                raise errors.NotFoundError()
            else:
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -13,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-import ujson as json
+import simplejson as json
 import logging

 from canonicaljson import encode_canonical_json
@@ -134,23 +135,8 @@ 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:
-                # include ConnectionRefused and other errors
-                failures[destination] = {
-                    "status": 503, "message": e.message
-                }
+                failures[destination] = _exception_to_failure(e)

        yield make_deferred_yieldable(defer.gatherResults([
            preserve_fn(do_remote_query)(destination)
@@ -252,19 +238,8 @@ 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:
-                # include ConnectionRefused and other errors
-                failures[destination] = {
-                    "status": 503, "message": e.message
-                }
+                failures[destination] = _exception_to_failure(e)

        yield make_deferred_yieldable(defer.gatherResults([
            preserve_fn(claim_client_keys)(destination)
@@ -362,6 +337,31 @@ 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)

--- a/synapse/handlers/groups_local.py
+++ b/synapse/handlers/groups_local.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -90,6 +91,8 @@ 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.
@@ -226,7 +229,45 @@ class GroupsLocalHandler(object):
    def join_group(self, group_id, user_id, content):
        """Request to join a group
        """
-        raise NotImplementedError()  # TODO
+        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({})

    @defer.inlineCallbacks
    def accept_invite(self, group_id, user_id, content):
--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -15,6 +15,11 @@
 # 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 (
@@ -24,9 +29,6 @@ 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__)


--- a/synapse/handlers/initial_sync.py
+++ b/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,
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -27,7 +27,7 @@ from synapse.types import (
 from synapse.util.async import run_on_reactor, ReadWriteLock, Limiter
 from synapse.util.logcontext import preserve_fn, run_in_background
 from synapse.util.metrics import measure_func
-from synapse.util.frozenutils import unfreeze
+from synapse.util.frozenutils import frozendict_json_encoder
 from synapse.util.stringutils import random_string
 from synapse.visibility import filter_events_for_client
 from synapse.replication.http.send_event import send_event_to_master
@@ -38,7 +38,7 @@ from canonicaljson import encode_canonical_json

 import logging
 import random
-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -384,7 +384,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
@@ -454,40 +454,39 @@ class EventCreationHandler(object):
        """
        builder = self.event_builder_factory.new(event_dict)

-        with (yield self.limiter.queue(builder.room_id)):
-            self.validator.validate_new(builder)
+        self.validator.validate_new(builder)

-            if builder.type == EventTypes.Member:
-                membership = builder.content.get("membership", None)
-                target = UserID.from_string(builder.state_key)
+        if builder.type == EventTypes.Member:
+            membership = builder.content.get("membership", None)
+            target = UserID.from_string(builder.state_key)

-                if membership in {Membership.JOIN, Membership.INVITE}:
-                    # If event doesn't include a display name, add one.
-                    profile = self.profile_handler
-                    content = builder.content
+            if membership in {Membership.JOIN, Membership.INVITE}:
+                # If event doesn't include a display name, add one.
+                profile = self.profile_handler
+                content = builder.content

-                    try:
-                        if "displayname" not in content:
-                            content["displayname"] = yield profile.get_displayname(target)
-                        if "avatar_url" not in content:
-                            content["avatar_url"] = yield profile.get_avatar_url(target)
-                    except Exception as e:
-                        logger.info(
-                            "Failed to get profile information for %r: %s",
-                            target, e
-                        )
+                try:
+                    if "displayname" not in content:
+                        content["displayname"] = yield profile.get_displayname(target)
+                    if "avatar_url" not in content:
+                        content["avatar_url"] = yield profile.get_avatar_url(target)
+                except Exception as e:
+                    logger.info(
+                        "Failed to get profile information for %r: %s",
+                        target, e
+                    )

-            if token_id is not None:
-                builder.internal_metadata.token_id = token_id
+        if token_id is not None:
+            builder.internal_metadata.token_id = token_id

-            if txn_id is not None:
-                builder.internal_metadata.txn_id = txn_id
+        if txn_id is not None:
+            builder.internal_metadata.txn_id = txn_id

-            event, context = yield self.create_new_client_event(
-                builder=builder,
-                requester=requester,
-                prev_event_ids=prev_event_ids,
-            )
+        event, context = yield self.create_new_client_event(
+            builder=builder,
+            requester=requester,
+            prev_event_ids=prev_event_ids,
+        )

        defer.returnValue((event, context))

@@ -557,27 +556,34 @@ class EventCreationHandler(object):

        See self.create_event and self.send_nonmember_event.
        """
-        event, context = yield self.create_event(
-            requester,
-            event_dict,
-            token_id=requester.access_token_id,
-            txn_id=txn_id
-        )

-        spam_error = self.spam_checker.check_event_for_spam(event)
-        if spam_error:
-            if not isinstance(spam_error, basestring):
-                spam_error = "Spam is not permitted here"
-            raise SynapseError(
-                403, spam_error, Codes.FORBIDDEN
+        # We limit the number of concurrent event sends in a room so that we
+        # don't fork the DAG too much. If we don't limit then we can end up in
+        # a situation where event persistence can't keep up, causing
+        # extremities to pile up, which in turn leads to state resolution
+        # taking longer.
+        with (yield self.limiter.queue(event_dict["room_id"])):
+            event, context = yield self.create_event(
+                requester,
+                event_dict,
+                token_id=requester.access_token_id,
+                txn_id=txn_id
            )

-        yield self.send_nonmember_event(
-            requester,
-            event,
-            context,
-            ratelimit=ratelimit,
-        )
+            spam_error = self.spam_checker.check_event_for_spam(event)
+            if spam_error:
+                if not isinstance(spam_error, basestring):
+                    spam_error = "Spam is not permitted here"
+                raise SynapseError(
+                    403, spam_error, Codes.FORBIDDEN
+                )
+
+            yield self.send_nonmember_event(
+                requester,
+                event,
+                context,
+                ratelimit=ratelimit,
+            )
        defer.returnValue(event)

    @measure_func("create_new_client_event")
@@ -678,8 +684,8 @@ class EventCreationHandler(object):

        # Ensure that we can round trip before trying to persist in db
        try:
-            dump = ujson.dumps(unfreeze(event.content))
-            ujson.loads(dump)
+            dump = frozendict_json_encoder.encode(event.content)
+            simplejson.loads(dump)
        except Exception:
            logger.exception("Failed to encode content: %r", event.content)
            raise
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -373,6 +373,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 +403,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 +446,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 +467,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 = (
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -23,8 +23,8 @@ 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
+from synapse.types import UserID, create_requester, RoomID, RoomAlias
+from synapse.util.async import run_on_reactor, Linearizer
 from synapse.util.threepids import check_3pid_allowed
 from ._base import BaseHandler

@@ -46,6 +46,10 @@ 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):
@@ -201,10 +205,17 @@ class RegistrationHandler(BaseHandler):
                    token = None
                    attempts += 1

+        # auto-join the user to any rooms we're supposed to dump them into
+        fake_requester = create_requester(user_id)
+        for r in self.hs.config.auto_join_rooms:
+            try:
+                yield self._join_user_to_room(fake_requester, r)
+            except Exception as e:
+                logger.error("Failed to join new user to %r: %r", r, e)
+
        # We used to generate default identicons here, but nowadays
        # we want clients to generate their own as part of their branding
        # rather than there being consistent matrix-wide ones, so we don't.
-
        defer.returnValue((user_id, token))

    @defer.inlineCallbacks
@@ -345,9 +356,11 @@ class RegistrationHandler(BaseHandler):
    @defer.inlineCallbacks
    def _generate_user_id(self, reseed=False):
        if reseed or self._next_generated_user_id is None:
-            self._next_generated_user_id = (
-                yield self.store.find_next_generated_user_id_localpart()
-            )
+            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()
+                    )

        id = self._next_generated_user_id
        self._next_generated_user_id += 1
@@ -477,3 +490,28 @@ class RegistrationHandler(BaseHandler):
        )

        defer.returnValue((user_id, access_token))
+
+    @defer.inlineCallbacks
+    def _join_user_to_room(self, requester, room_identifier):
+        room_id = None
+        room_member_handler = self.hs.get_room_member_handler()
+        if RoomID.is_valid(room_identifier):
+            room_id = room_identifier
+        elif RoomAlias.is_valid(room_identifier):
+            room_alias = RoomAlias.from_string(room_identifier)
+            room_id, remote_room_hosts = (
+                yield room_member_handler.lookup_room_alias(room_alias)
+            )
+            room_id = room_id.to_string()
+        else:
+            raise SynapseError(400, "%s was not legal room ID or room alias" % (
+                room_identifier,
+            ))
+
+        yield room_member_handler.update_membership(
+            requester=requester,
+            target=requester.user,
+            room_id=room_id,
+            remote_room_hosts=remote_room_hosts,
+            action="join",
+        )
--- a/synapse/handlers/room_list.py
+++ b/synapse/handlers/room_list.py
@@ -44,8 +44,9 @@ 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.remote_response_cache = ResponseCache(hs, timeout_ms=30 * 1000)
+        self.response_cache = ResponseCache(hs, "room_list", timeout_ms=10 * 60 * 1000)
+        self.remote_response_cache = ResponseCache(hs, "remote_room_list",
+                                                   timeout_ms=30 * 1000)

    def get_local_public_room_list(self, limit=None, since_token=None,
                                   search_filter=None,
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -28,7 +28,7 @@ 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


@@ -60,6 +60,7 @@ class RoomMemberHandler(object):
        self.event_creation_hander = hs.get_event_creation_handler()

        self.member_linearizer = Linearizer(name="member")
+        self.member_limiter = Limiter(20)

        self.clock = hs.get_clock()
        self.spam_checker = hs.get_spam_checker()
@@ -232,18 +233,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)

@@ -852,6 +858,14 @@ class RoomMemberMasterHandler(RoomMemberHandler):
    def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
        """Implements RoomMemberHandler._remote_join
        """
+        # filter ourselves out of remote_room_hosts: do_invite_join ignores it
+        # and if it is the only entry we'd like to return a 404 rather than a
+        # 500.
+
+        remote_room_hosts = [
+            host for host in remote_room_hosts if host != self.hs.hostname
+        ]
+
        if len(remote_room_hosts) == 0:
            raise SynapseError(404, "No known servers")

--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -30,6 +30,7 @@ import itertools

 logger = logging.getLogger(__name__)

+SYNC_RESPONSE_CACHE_MS = 2 * 60 * 1000

 SyncConfig = collections.namedtuple("SyncConfig", [
    "user",
@@ -169,7 +170,10 @@ class SyncHandler(object):
        self.presence_handler = hs.get_presence_handler()
        self.event_sources = hs.get_event_sources()
        self.clock = hs.get_clock()
-        self.response_cache = ResponseCache(hs)
+        self.response_cache = ResponseCache(
+            hs, "sync",
+            timeout_ms=SYNC_RESPONSE_CACHE_MS,
+        )
        self.state = hs.get_state_handler()

    def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0,
@@ -599,7 +603,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
            )
--- a/synapse/http/endpoint.py
+++ b/synapse/http/endpoint.py
@@ -12,8 +12,6 @@
 # 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 socket
-
 from twisted.internet.endpoints import HostnameEndpoint, wrapClientTLS
 from twisted.internet import defer, reactor
 from twisted.internet.error import ConnectError
@@ -33,7 +31,7 @@ SERVER_CACHE = {}

 # our record of an individual server which can be tried to reach a destination.
 #
-# "host" is actually a dotted-quad or ipv6 address string. Except when there's
+# "host" is the hostname acquired from the SRV record. Except when there's
 # no SRV record, in which case it is the original hostname.
 _Server = collections.namedtuple(
    "_Server", "priority weight host port expires"
@@ -297,20 +295,13 @@ def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=t

            payload = answer.payload

-            hosts = yield _get_hosts_for_srv_record(
-                dns_client, str(payload.target)
-            )
-
-            for (ip, ttl) in hosts:
-                host_ttl = min(answer.ttl, ttl)
-
-                servers.append(_Server(
-                    host=ip,
-                    port=int(payload.port),
-                    priority=int(payload.priority),
-                    weight=int(payload.weight),
-                    expires=int(clock.time()) + host_ttl,
-                ))
+            servers.append(_Server(
+                host=str(payload.target),
+                port=int(payload.port),
+                priority=int(payload.priority),
+                weight=int(payload.weight),
+                expires=int(clock.time()) + answer.ttl,
+            ))

        servers.sort()
        cache[service_name] = list(servers)
@@ -328,81 +319,3 @@ def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=t
            raise e

    defer.returnValue(servers)
-
-
-@defer.inlineCallbacks
-def _get_hosts_for_srv_record(dns_client, host):
-    """Look up each of the hosts in a SRV record
-
-    Args:
-        dns_client (twisted.names.dns.IResolver):
-        host (basestring): host to look up
-
-    Returns:
-        Deferred[list[(str, int)]]: a list of (host, ttl) pairs
-
-    """
-    ip4_servers = []
-    ip6_servers = []
-
-    def cb(res):
-        # lookupAddress and lookupIP6Address return a three-tuple
-        # giving the answer, authority, and additional sections of the
-        # response.
-        #
-        # we only care about the answers.
-
-        return res[0]
-
-    def eb(res, record_type):
-        if res.check(DNSNameError):
-            return []
-        logger.warn("Error looking up %s for %s: %s", record_type, host, res)
-        return res
-
-    # no logcontexts here, so we can safely fire these off and gatherResults
-    d1 = dns_client.lookupAddress(host).addCallbacks(
-        cb, eb, errbackArgs=("A", ))
-    d2 = dns_client.lookupIPV6Address(host).addCallbacks(
-        cb, eb, errbackArgs=("AAAA", ))
-    results = yield defer.DeferredList(
-        [d1, d2], consumeErrors=True)
-
-    # if all of the lookups failed, raise an exception rather than blowing out
-    # the cache with an empty result.
-    if results and all(s == defer.FAILURE for (s, _) in results):
-        defer.returnValue(results[0][1])
-
-    for (success, result) in results:
-        if success == defer.FAILURE:
-            continue
-
-        for answer in result:
-            if not answer.payload:
-                continue
-
-            try:
-                if answer.type == dns.A:
-                    ip = answer.payload.dottedQuad()
-                    ip4_servers.append((ip, answer.ttl))
-                elif answer.type == dns.AAAA:
-                    ip = socket.inet_ntop(
-                        socket.AF_INET6, answer.payload.address,
-                    )
-                    ip6_servers.append((ip, answer.ttl))
-                else:
-                    # the most likely candidate here is a CNAME record.
-                    # rfc2782 says srvs may not point to aliases.
-                    logger.warn(
-                        "Ignoring unexpected DNS record type %s for %s",
-                        answer.type, host,
-                    )
-                    continue
-            except Exception as e:
-                logger.warn("Ignoring invalid DNS response for %s: %s",
-                            host, e)
-                continue
-
-    # keep the ipv4 results before the ipv6 results, mostly to match historical
-    # behaviour.
-    defer.returnValue(ip4_servers + ip6_servers)
--- a/synapse/http/matrixfederationclient.py
+++ b/synapse/http/matrixfederationclient.py
@@ -286,7 +286,8 @@ class MatrixFederationHttpClient(object):
        headers_dict[b"Authorization"] = auth_headers

    @defer.inlineCallbacks
-    def put_json(self, destination, path, data={}, json_data_callback=None,
+    def put_json(self, destination, path, args={}, data={},
+                 json_data_callback=None,
                 long_retries=False, timeout=None,
                 ignore_backoff=False,
                 backoff_on_404=False):
@@ -296,6 +297,7 @@ 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
@@ -342,6 +344,7 @@ 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,
@@ -373,6 +376,7 @@ 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.
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -37,7 +37,7 @@ from twisted.web.util import redirectTo
 import collections
 import logging
 import urllib
-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -113,6 +113,11 @@ 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


@@ -324,7 +329,7 @@ class JsonResource(HttpServer, resource.Resource):
                register_paths, so will return (possibly via Deferred) either
                None, or a tuple of (http code, response body).
        """
-        if request.method == "OPTIONS":
+        if request.method == b"OPTIONS":
            return _options_handler, {}

        # Loop through all the registered callbacks to check if the method
@@ -426,6 +431,8 @@ 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."""
@@ -461,8 +468,7 @@ 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:
-            # ujson doesn't like frozen_dicts.
-            json_bytes = ujson.dumps(json_object, ensure_ascii=False)
+            json_bytes = simplejson.dumps(json_object)

    return respond_with_json_bytes(
        request, code, json_bytes,
@@ -489,6 +495,7 @@ 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)
@@ -536,7 +543,7 @@ def finish_request(request):

 def _request_user_agent_is_curl(request):
    user_agents = request.requestHeaders.getRawHeaders(
-        "User-Agent", default=[]
+        b"User-Agent", default=[]
    )
    for user_agent in user_agents:
        if "curl" in user_agent:
--- a/synapse/http/site.py
+++ b/synapse/http/site.py
@@ -20,7 +20,7 @@ import logging
 import re
 import time

-ACCESS_TOKEN_RE = re.compile(r'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')
+ACCESS_TOKEN_RE = re.compile(br'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')


 class SynapseRequest(Request):
@@ -43,12 +43,12 @@ class SynapseRequest(Request):

    def get_redacted_uri(self):
        return ACCESS_TOKEN_RE.sub(
-            r'\1<redacted>\3',
+            br'\1<redacted>\3',
            self.uri
        )

    def get_user_agent(self):
-        return self.requestHeaders.getRawHeaders("User-Agent", [None])[-1]
+        return self.requestHeaders.getRawHeaders(b"User-Agent", [None])[-1]

    def started_processing(self):
        self.site.access_logger.info(
--- a/synapse/metrics/init.py
+++ b/synapse/metrics/init.py
@@ -17,12 +17,13 @@ import logging
 import functools
 import time
 import gc
+import platform

 from twisted.internet import reactor

 from .metric import (
    CounterMetric, CallbackMetric, DistributionMetric, CacheMetric,
-    MemoryUsageMetric,
+    MemoryUsageMetric, GaugeMetric,
 )
 from .process_collector import register_process_collector

@@ -30,6 +31,7 @@ from .process_collector import register_process_collector
 logger = logging.getLogger(__name__)


+running_on_pypy = platform.python_implementation() == 'PyPy'
 all_metrics = []
 all_collectors = []

@@ -63,6 +65,13 @@ class Metrics(object):
        """
        return self._register(CounterMetric, *args, **kwargs)

+    def register_gauge(self, *args, **kwargs):
+        """
+        Returns:
+            GaugeMetric
+        """
+        return self._register(GaugeMetric, *args, **kwargs)
+
    def register_callback(self, *args, **kwargs):
        """
        Returns:
@@ -142,6 +151,32 @@ reactor_metrics = get_metrics_for("python.twisted.reactor")
 tick_time = reactor_metrics.register_distribution("tick_time")
 pending_calls_metric = reactor_metrics.register_distribution("pending_calls")

+synapse_metrics = get_metrics_for("synapse")
+
+# Used to track where various components have processed in the event stream,
+# e.g. federation sending, appservice sending, etc.
+event_processing_positions = synapse_metrics.register_gauge(
+    "event_processing_positions", labels=["name"],
+)
+
+# Used to track the current max events stream position
+event_persisted_position = synapse_metrics.register_gauge(
+    "event_persisted_position",
+)
+
+# Used to track the received_ts of the last event processed by various
+# components
+event_processing_last_ts = synapse_metrics.register_gauge(
+    "event_processing_last_ts", labels=["name"],
+)
+
+# Used to track the lag processing events. This is the time difference
+# between the last processed event's received_ts and the time it was
+# finished being processed.
+event_processing_lag = synapse_metrics.register_gauge(
+    "event_processing_lag", labels=["name"],
+)
+

 def runUntilCurrentTimer(func):

@@ -174,6 +209,9 @@ def runUntilCurrentTimer(func):
        tick_time.inc_by(end - start)
        pending_calls_metric.inc_by(num_pending)

+        if running_on_pypy:
+            return ret
+
        # Check if we need to do a manual GC (since its been disabled), and do
        # one if necessary.
        threshold = gc.get_threshold()
@@ -206,6 +244,7 @@ try:

    # We manually run the GC each reactor tick so that we can get some metrics
    # about time spent doing GC,
-    gc.disable()
+    if not running_on_pypy:
+        gc.disable()
 except AttributeError:
    pass
--- a/synapse/metrics/metric.py
+++ b/synapse/metrics/metric.py
@@ -115,7 +115,7 @@ class CounterMetric(BaseMetric):
        # dict[list[str]]: value for each set of label values. the keys are the
        # label values, in the same order as the labels in self.labels.
        #
-        # (if the metric is a scalar, the (single) key is the empty list).
+        # (if the metric is a scalar, the (single) key is the empty tuple).
        self.counts = {}

        # Scalar metrics are never empty
@@ -145,6 +145,36 @@ class CounterMetric(BaseMetric):
        )


+class GaugeMetric(BaseMetric):
+    """A metric that can go up or down
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(GaugeMetric, self).__init__(*args, **kwargs)
+
+        # dict[list[str]]: value for each set of label values. the keys are the
+        # label values, in the same order as the labels in self.labels.
+        #
+        # (if the metric is a scalar, the (single) key is the empty tuple).
+        self.guages = {}
+
+    def set(self, v, *values):
+        if len(values) != self.dimension():
+            raise ValueError(
+                "Expected as many values to inc() as labels (%d)" % (self.dimension())
+            )
+
+        # TODO: should assert that the tag values are all strings
+
+        self.guages[values] = v
+
+    def render(self):
+        return flatten(
+            self._render_for_labels(k, self.guages[k])
+            for k in sorted(self.guages.keys())
+        )
+
+
 class CallbackMetric(BaseMetric):
    """A metric that returns the numeric value returned by a callback whenever
    it is rendered. Typically this is used to implement gauges that yield the
--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -34,7 +34,6 @@ REQUIREMENTS = {
    "bcrypt": ["bcrypt>=3.1.0"],
    "pillow": ["PIL"],
    "pydenticon": ["pydenticon"],
-    "ujson": ["ujson"],
    "blist": ["blist"],
    "pysaml2>=3.0.0": ["saml2>=3.0.0"],
    "pymacaroons-pynacl": ["pymacaroons"],
--- a/synapse/replication/http/send_event.py
+++ b/synapse/replication/http/send_event.py
@@ -115,7 +115,7 @@ class ReplicationSendEventRestServlet(RestServlet):
        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)
+        self.response_cache = ResponseCache(hs, "send_event", timeout_ms=30 * 60 * 1000)

    def on_PUT(self, request, event_id):
        result = self.response_cache.get(event_id)
--- a/synapse/replication/slave/storage/client_ips.py
+++ b/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
        )
--- a/synapse/replication/tcp/commands.py
+++ b/synapse/replication/tcp/commands.py
@@ -19,11 +19,13 @@ allowed to be sent by which side.
 """

 import logging
-import ujson as json
+import simplejson


 logger = logging.getLogger(__name__)

+_json_encoder = simplejson.JSONEncoder(namedtuple_as_object=False)
+

 class Command(object):
    """The base command class.
@@ -100,14 +102,14 @@ class RdataCommand(Command):
        return cls(
            stream_name,
            None if token == "batch" else int(token),
-            json.loads(row_json)
+            simplejson.loads(row_json)
        )

    def to_line(self):
        return " ".join((
            self.stream_name,
            str(self.token) if self.token is not None else "batch",
-            json.dumps(self.row),
+            _json_encoder.encode(self.row),
        ))


@@ -298,10 +300,12 @@ class InvalidateCacheCommand(Command):
    def from_line(cls, line):
        cache_func, keys_json = line.split(" ", 1)

-        return cls(cache_func, json.loads(keys_json))
+        return cls(cache_func, simplejson.loads(keys_json))

    def to_line(self):
-        return " ".join((self.cache_func, json.dumps(self.keys)))
+        return " ".join((
+            self.cache_func, _json_encoder.encode(self.keys),
+        ))


 class UserIpCommand(Command):
@@ -325,14 +329,14 @@ class UserIpCommand(Command):
    def from_line(cls, line):
        user_id, jsn = line.split(" ", 1)

-        access_token, ip, user_agent, device_id, last_seen = json.loads(jsn)
+        access_token, ip, user_agent, device_id, last_seen = simplejson.loads(jsn)

        return cls(
            user_id, access_token, ip, user_agent, device_id, last_seen
        )

    def to_line(self):
-        return self.user_id + " " + json.dumps((
+        return self.user_id + " " + _json_encoder.encode((
            self.access_token, self.ip, self.user_agent, self.device_id,
            self.last_seen,
        ))
--- a/synapse/replication/tcp/streams.py
+++ b/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", (
--- a/synapse/rest/client/v1/admin.py
+++ b/synapse/rest/client/v1/admin.py
@@ -296,17 +296,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)
@@ -344,6 +333,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(
--- a/synapse/rest/client/v1/logout.py
+++ b/synapse/rest/client/v1/logout.py
@@ -44,7 +44,10 @@ class LogoutRestServlet(ClientV1RestServlet):
            requester = yield self.auth.get_user_by_req(request)
        except AuthError:
            # this implies the access token has already been deleted.
-            pass
+            defer.returnValue((401, {
+                "errcode": "M_UNKNOWN_TOKEN",
+                "error": "Access Token unknown or expired"
+            }))
        else:
            if requester.device_id is None:
                # the acccess token wasn't associated with a device.
--- a/synapse/rest/client/v1/presence.py
+++ b/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, {}))

--- a/synapse/rest/client/v1/register.py
+++ b/synapse/rest/client/v1/register.py
@@ -348,9 +348,9 @@ class RegisterRestServlet(ClientV1RestServlet):
        admin = register_json.get("admin", None)

        # Its important to check as we use null bytes as HMAC field separators
-        if "\x00" in user:
+        if b"\x00" in user:
            raise SynapseError(400, "Invalid user")
-        if "\x00" in password:
+        if b"\x00" in password:
            raise SynapseError(400, "Invalid password")

        # str() because otherwise hmac complains that 'unicode' does not
--- a/synapse/rest/client/v1/room.py
+++ b/synapse/rest/client/v1/room.py
@@ -30,7 +30,7 @@ from synapse.http.servlet import (

 import logging
 import urllib
-import ujson as json
+import simplejson as json

 logger = logging.getLogger(__name__)

@@ -165,17 +165,12 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
                content=content,
            )
        else:
-            event, context = yield self.event_creation_hander.create_event(
+            event = yield self.event_creation_hander.create_and_send_nonmember_event(
                requester,
                event_dict,
-                token_id=requester.access_token_id,
                txn_id=txn_id,
            )

-            yield self.event_creation_hander.send_nonmember_event(
-                requester, event, context,
-            )
-
        ret = {}
        if event:
            ret = {"event_id": event.event_id}
@@ -655,7 +650,12 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
            content=event_content,
        )

-        defer.returnValue((200, {}))
+        return_value = {}
+
+        if membership_action == "join":
+            return_value["room_id"] = room_id
+
+        defer.returnValue((200, return_value))

    def _has_3pid_invite_keys(self, content):
        for key in {"id_server", "medium", "address"}:
--- a/synapse/rest/client/v2_alpha/groups.py
+++ b/synapse/rest/client/v2_alpha/groups.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -401,6 +402,32 @@ 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
    """
@@ -738,6 +765,7 @@ 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)
--- a/synapse/rest/client/v2_alpha/register.py
+++ b/synapse/rest/client/v2_alpha/register.py
@@ -20,7 +20,6 @@ import synapse
 import synapse.types
 from synapse.api.auth import get_access_token_from_request, has_access_token
 from synapse.api.constants import LoginType
-from synapse.types import RoomID, RoomAlias
 from synapse.api.errors import SynapseError, Codes, UnrecognizedRequestError
 from synapse.http.servlet import (
    RestServlet, parse_json_object_from_request, assert_params_in_request, parse_string
@@ -405,14 +404,6 @@ class RegisterRestServlet(RestServlet):
                generate_token=False,
            )

-            # auto-join the user to any rooms we're supposed to dump them into
-            fake_requester = synapse.types.create_requester(registered_user_id)
-            for r in self.hs.config.auto_join_rooms:
-                try:
-                    yield self._join_user_to_room(fake_requester, r)
-                except Exception as e:
-                    logger.error("Failed to join new user to %r: %r", r, e)
-
            # remember that we've now registered that user account, and with
            #  what user ID (since the user may not have specified)
            self.auth_handler.set_session_data(
@@ -445,29 +436,6 @@ class RegisterRestServlet(RestServlet):
    def on_OPTIONS(self, _):
        return 200, {}

-    @defer.inlineCallbacks
-    def _join_user_to_room(self, requester, room_identifier):
-        room_id = None
-        if RoomID.is_valid(room_identifier):
-            room_id = room_identifier
-        elif RoomAlias.is_valid(room_identifier):
-            room_alias = RoomAlias.from_string(room_identifier)
-            room_id, remote_room_hosts = (
-                yield self.room_member_handler.lookup_room_alias(room_alias)
-            )
-            room_id = room_id.to_string()
-        else:
-            raise SynapseError(400, "%s was not legal room ID or room alias" % (
-                room_identifier,
-            ))
-
-        yield self.room_member_handler.update_membership(
-            requester=requester,
-            target=requester.user,
-            room_id=room_id,
-            action="join",
-        )
-
    @defer.inlineCallbacks
    def _do_appservice_registration(self, username, as_token, body):
        user_id = yield self.registration_handler.appservice_register(
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -33,7 +33,7 @@ from ._base import set_timeline_upper_limit
 import itertools
 import logging

-import ujson as json
+import simplejson as json

 logger = logging.getLogger(__name__)

--- a/synapse/rest/media/v1/media_storage.py
+++ b/synapse/rest/media/v1/media_storage.py
@@ -16,6 +16,8 @@
 from twisted.internet import defer, threads
 from twisted.protocols.basic import FileSender

+import six
+
 from ._base import Responder

 from synapse.util.file_consumer import BackgroundFileConsumer
@@ -119,7 +121,7 @@ class MediaStorage(object):
                os.remove(fname)
            except Exception:
                pass
-            raise t, v, tb
+            six.reraise(t, v, tb)

        if not finished_called:
            raise Exception("Finished callback not called")
--- a/synapse/rest/media/v1/preview_url_resource.py
+++ b/synapse/rest/media/v1/preview_url_resource.py
@@ -23,7 +23,7 @@ import re
 import shutil
 import sys
 import traceback
-import ujson as json
+import simplejson as json
 import urlparse

 from twisted.web.server import NOT_DONE_YET
--- a/synapse/state.py
+++ b/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.items() if e_id in state_map
+            key: state_map[e_id] for key, e_id in state.iteritems() 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.items()
+            key: state_map[ev_id] for key, ev_id in new_state.iteritems()
        }

        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.values():
-                for key, e_id in st.items():
+            for st in state_groups_ids.itervalues():
+                for key, e_id in st.iteritems():
                    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.items()
+                for k, v in state.iteritems()
                if len(v) > 1
            }

@@ -480,36 +480,37 @@ class StateResolutionHandler(object):
                    )
            else:
                new_state = {
-                    key: e_ids.pop() for key, e_ids in state.items()
+                    key: e_ids.pop() for key, e_ids in state.iteritems()
                }

-            # 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
+            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

-            # 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,
@@ -702,7 +703,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.items()
+        for key, ev_id in auth_event_ids.iteritems()
        if ev_id in state_map
    }

@@ -740,7 +741,7 @@ def _resolve_state_events(conflicted_state, auth_events):

    auth_events.update(resolved_state)

-    for key, events in conflicted_state.items():
+    for key, events in conflicted_state.iteritems():
        if key[0] == EventTypes.JoinRules:
            logger.debug("Resolving conflicted join rules %r", events)
            resolved_state[key] = _resolve_auth_events(
@@ -750,7 +751,7 @@ def _resolve_state_events(conflicted_state, auth_events):

    auth_events.update(resolved_state)

-    for key, events in conflicted_state.items():
+    for key, events in conflicted_state.iteritems():
        if key[0] == EventTypes.Member:
            logger.debug("Resolving conflicted member lists %r", events)
            resolved_state[key] = _resolve_auth_events(
@@ -760,7 +761,7 @@ def _resolve_state_events(conflicted_state, auth_events):

    auth_events.update(resolved_state)

-    for key, events in conflicted_state.items():
+    for key, events in conflicted_state.iteritems():
        if key not in resolved_state:
            logger.debug("Resolving conflicted state %r:%r", key, events)
            resolved_state[key] = _resolve_normal_events(
--- a/synapse/storage/init.py
+++ b/synapse/storage/init.py
@@ -14,8 +14,6 @@
 # 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
@@ -244,13 +242,12 @@ 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 (
@@ -264,8 +261,91 @@ class DataStore(RoomMemberStore, RoomStore,
            count, = txn.fetchone()
            return count

-        ret = yield self.runInteraction("count_users", _count_users)
-        defer.returnValue(ret)
+        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 ago
+         * Where last seen at most 30 days ago
+         * Where account creation and last_seen are > 30 days apart
+
+         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())
+            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)

    def get_users(self):
        """Function to reterive a list of users in users table.
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -376,7 +376,7 @@ class SQLBaseStore(object):
        Returns:
            A list of dicts where the key is the column header.
        """
-        col_headers = list(intern(column[0]) for column in cursor.description)
+        col_headers = list(intern(str(column[0])) for column in cursor.description)
        results = list(
            dict(zip(col_headers, row)) for row in cursor
        )
--- a/synapse/storage/account_data.py
+++ b/synapse/storage/account_data.py
@@ -23,7 +23,7 @@ from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.caches.descriptors import cached, cachedList, cachedInlineCallbacks

 import abc
-import ujson as json
+import simplejson as json
 import logging

 logger = logging.getLogger(__name__)
--- a/synapse/storage/background_updates.py
+++ b/synapse/storage/background_updates.py
@@ -19,7 +19,7 @@ from . import engines

 from twisted.internet import defer

-import ujson as json
+import simplejson as json
 import logging

 logger = logging.getLogger(__name__)
--- a/synapse/storage/client_ips.py
+++ b/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,6 +48,13 @@ 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 = {}

--- a/synapse/storage/deviceinbox.py
+++ b/synapse/storage/deviceinbox.py
@@ -14,7 +14,7 @@
 # limitations under the License.

 import logging
-import ujson
+import simplejson

 from twisted.internet import defer

@@ -85,7 +85,7 @@ class DeviceInboxStore(BackgroundUpdateStore):
            )
            rows = []
            for destination, edu in remote_messages_by_destination.items():
-                edu_json = ujson.dumps(edu)
+                edu_json = simplejson.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 = ujson.dumps(messages_by_device["*"])
+                message_json = simplejson.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 = ujson.dumps(messages_by_device[device])
+                    message_json = simplejson.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(ujson.loads(row[1]))
+                messages.append(simplejson.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(ujson.loads(row[1]))
+                messages.append(simplejson.loads(row[1]))
            if len(messages) < limit:
                stream_pos = current_stream_id
            return (messages, stream_pos)
--- a/synapse/storage/devices.py
+++ b/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 ujson as json
+import simplejson as json

 from twisted.internet import defer

--- a/synapse/storage/end_to_end_keys.py
+++ b/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 ujson as json
+import simplejson as json

 from ._base import SQLBaseStore

--- a/synapse/storage/engines/init.py
+++ b/synapse/storage/engines/init.py
@@ -18,6 +18,7 @@ from .postgres import PostgresEngine
 from .sqlite3 import Sqlite3Engine

 import importlib
+import platform


 SUPPORTED_MODULE = {
@@ -31,6 +32,10 @@ def create_engine(database_config):
    engine_class = SUPPORTED_MODULE.get(name, None)

    if engine_class:
+        # pypy requires psycopg2cffi rather than psycopg2
+        if (name == "psycopg2" and
+                platform.python_implementation() == "PyPy"):
+            name = "psycopg2cffi"
        module = importlib.import_module(name)
        return engine_class(module, database_config)

--- a/synapse/storage/event_push_actions.py
+++ b/synapse/storage/event_push_actions.py
@@ -22,7 +22,7 @@ from synapse.types import RoomStreamToken
 from .stream import lower_bound

 import logging
-import ujson as json
+import simplejson as json

 logger = logging.getLogger(__name__)

@@ -613,6 +613,9 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
            self._rotate_notifs, 30 * 60 * 1000
        )

+        self._rotate_delay = 3
+        self._rotate_count = 10000
+
    def _set_push_actions_for_event_and_users_txn(self, txn, events_and_contexts,
                                                  all_events_and_contexts):
        """Handles moving push actions from staging table to main
@@ -809,7 +812,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                )
                if caught_up:
                    break
-                yield sleep(5)
+                yield sleep(self._rotate_delay)
        finally:
            self._doing_notif_rotation = False

@@ -830,8 +833,8 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
        txn.execute("""
            SELECT stream_ordering FROM event_push_actions
            WHERE stream_ordering > ?
-            ORDER BY stream_ordering ASC LIMIT 1 OFFSET 50000
-        """, (old_rotate_stream_ordering,))
+            ORDER BY stream_ordering ASC LIMIT 1 OFFSET ?
+        """, (old_rotate_stream_ordering, self._rotate_count))
        stream_row = txn.fetchone()
        if stream_row:
            offset_stream_ordering, = stream_row
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -14,15 +14,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-from synapse.storage.events_worker import EventsWorkerStore
+from collections import OrderedDict, deque, namedtuple
+from functools import wraps
+import logging

+import simplejson as json
 from twisted.internet import defer

-from synapse.events import USE_FROZEN_DICTS

+from synapse.storage.events_worker import EventsWorkerStore
 from synapse.util.async import ObservableDeferred
+from synapse.util.frozenutils import frozendict_json_encoder
 from synapse.util.logcontext import (
-    PreserveLoggingContext, make_deferred_yieldable
+    PreserveLoggingContext, make_deferred_yieldable,
 )
 from synapse.util.logutils import log_function
 from synapse.util.metrics import Measure
@@ -30,16 +34,8 @@ from synapse.api.constants import EventTypes
 from synapse.api.errors import SynapseError
 from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
 from synapse.types import get_domain_from_id
-
-from canonicaljson import encode_canonical_json
-from collections import deque, namedtuple, OrderedDict
-from functools import wraps
-
 import synapse.metrics

-import logging
-import ujson as json
-
 # these are only included to make the type annotations work
 from synapse.events import EventBase    # noqa: F401
 from synapse.events.snapshot import EventContext   # noqa: F401
@@ -53,13 +49,25 @@ event_counter = metrics.register_counter(
    "persisted_events_sep", labels=["type", "origin_type", "origin_entity"]
 )

+# The number of times we are recalculating the current state
+state_delta_counter = metrics.register_counter(
+    "state_delta",
+)
+# The number of times we are recalculating state when there is only a
+# single forward extremity
+state_delta_single_event_counter = metrics.register_counter(
+    "state_delta_single_event",
+)
+# The number of times we are reculating state when we could have resonably
+# calculated the delta when we calculated the state for an event we were
+# persisting.
+state_delta_reuse_delta_counter = metrics.register_counter(
+    "state_delta_reuse_delta",
+)
+

 def encode_json(json_object):
-    if USE_FROZEN_DICTS:
-        # ujson doesn't like frozen_dicts
-        return encode_canonical_json(json_object)
-    else:
-        return json.dumps(json_object, ensure_ascii=False)
+    return frozendict_json_encoder.encode(json_object)


 class _EventPeristenceQueue(object):
@@ -369,7 +377,8 @@ class EventsStore(EventsWorkerStore):
                                room_id, ev_ctx_rm, latest_event_ids
                            )

-                            if new_latest_event_ids == set(latest_event_ids):
+                            latest_event_ids = set(latest_event_ids)
+                            if new_latest_event_ids == latest_event_ids:
                                # No change in extremities, so no change in state
                                continue

@@ -390,6 +399,26 @@ class EventsStore(EventsWorkerStore):
                                if all_single_prev_not_state:
                                    continue

+                            state_delta_counter.inc()
+                            if len(new_latest_event_ids) == 1:
+                                state_delta_single_event_counter.inc()
+
+                                # This is a fairly handwavey check to see if we could
+                                # have guessed what the delta would have been when
+                                # processing one of these events.
+                                # What we're interested in is if the latest extremities
+                                # were the same when we created the event as they are
+                                # now. When this server creates a new event (as opposed
+                                # to receiving it over federation) it will use the
+                                # forward extremities as the prev_events, so we can
+                                # guess this by looking at the prev_events and checking
+                                # if they match the current forward extremities.
+                                for ev, _ in ev_ctx_rm:
+                                    prev_event_ids = set(e for e, _ in ev.prev_events)
+                                    if latest_event_ids == prev_event_ids:
+                                        state_delta_reuse_delta_counter.inc()
+                                        break
+
                            logger.info(
                                "Calculating state delta for room %s", room_id,
                            )
@@ -415,6 +444,9 @@ class EventsStore(EventsWorkerStore):
                    new_forward_extremeties=new_forward_extremeties,
                )
                persist_event_counter.inc_by(len(chunk))
+                synapse.metrics.event_persisted_position.set(
+                    chunk[-1][0].internal_metadata.stream_ordering,
+                )
                for event, context in chunk:
                    if context.app_service:
                        origin_type = "local"
@@ -981,7 +1013,6 @@ class EventsStore(EventsWorkerStore):
                "event_edge_hashes",
                "event_edges",
                "event_forward_extremities",
-                "event_push_actions",
                "event_reference_hashes",
                "event_search",
                "event_signatures",
@@ -1001,6 +1032,14 @@ class EventsStore(EventsWorkerStore):
                [(ev.event_id,) for ev, _ in events_and_contexts]
            )

+        for table in (
+            "event_push_actions",
+        ):
+            txn.executemany(
+                "DELETE FROM %s WHERE room_id = ? AND event_id = ?" % (table,),
+                [(ev.event_id,) for ev, _ in events_and_contexts]
+            )
+
    def _store_event_txn(self, txn, events_and_contexts):
        """Insert new events into the event and event_json tables

--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -28,7 +28,7 @@ from synapse.api.errors import SynapseError
 from collections import namedtuple

 import logging
-import ujson as json
+import simplejson as json

 # these are only included to make the type annotations work
 from synapse.events import EventBase    # noqa: F401
@@ -51,6 +51,26 @@ _EventCacheEntry = namedtuple("_EventCacheEntry", ("event", "redacted_event"))


 class EventsWorkerStore(SQLBaseStore):
+    def get_received_ts(self, event_id):
+        """Get received_ts (when it was persisted) for the event.
+
+        Raises an exception for unknown events.
+
+        Args:
+            event_id (str)
+
+        Returns:
+            Deferred[int|None]: Timestamp in milliseconds, or None for events
+            that were persisted before received_ts was implemented.
+        """
+        return self._simple_select_one_onecol(
+            table="events",
+            keyvalues={
+                "event_id": event_id,
+            },
+            retcol="received_ts",
+            desc="get_received_ts",
+        )

    @defer.inlineCallbacks
    def get_event(self, event_id, check_redacted=True,
--- a/synapse/storage/group_server.py
+++ b/synapse/storage/group_server.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -19,7 +20,7 @@ from synapse.api.errors import SynapseError

 from ._base import SQLBaseStore

-import ujson as json
+import simplejson as json


 # The category ID for the "default" category. We don't store as null in the
@@ -29,6 +30,24 @@ _DEFAULT_ROLE_ID = ""


 class GroupServerStore(SQLBaseStore):
+    def set_group_join_policy(self, group_id, join_policy):
+        """Set the join policy of a group.
+
+        join_policy can be one of:
+         * "invite"
+         * "open"
+        """
+        return self._simple_update_one(
+            table="groups",
+            keyvalues={
+                "group_id": group_id,
+            },
+            updatevalues={
+                "join_policy": join_policy,
+            },
+            desc="set_group_join_policy",
+        )
+
    def get_group(self, group_id):
        return self._simple_select_one(
            table="groups",
@@ -36,10 +55,11 @@ class GroupServerStore(SQLBaseStore):
                "group_id": group_id,
            },
            retcols=(
-                "name", "short_description", "long_description", "avatar_url", "is_public"
+                "name", "short_description", "long_description",
+                "avatar_url", "is_public", "join_policy",
            ),
            allow_none=True,
-            desc="is_user_in_group",
+            desc="get_group",
        )

    def get_users_in_group(self, group_id, include_private=False):
--- a/synapse/storage/prepare_database.py
+++ b/synapse/storage/prepare_database.py
@@ -1,5 +1,6 @@
 # -*- 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.
@@ -25,7 +26,7 @@ logger = logging.getLogger(__name__)

 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 47
+SCHEMA_VERSION = 48

 dir_path = os.path.abspath(os.path.dirname(__file__))

--- a/synapse/storage/receipts.py
+++ b/synapse/storage/receipts.py
@@ -23,7 +23,7 @@ from twisted.internet import defer

 import abc
 import logging
-import ujson as json
+import simplejson as json


 logger = logging.getLogger(__name__)
--- a/synapse/storage/registration.py
+++ b/synapse/storage/registration.py
@@ -460,14 +460,12 @@ class RegistrationStore(RegistrationWorkerStore,
        """
        def _find_next_generated_user_id(txn):
            txn.execute("SELECT name FROM users")
-            rows = self.cursor_to_dict(txn)

            regex = re.compile("^@(\d+):")

            found = set()

-            for r in rows:
-                user_id = r["name"]
+            for user_id, in txn:
                match = regex.search(user_id)
                if match:
                    found.add(int(match.group(1)))
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -22,7 +22,7 @@ from synapse.util.caches.descriptors import cached, cachedInlineCallbacks

 import collections
 import logging
-import ujson as json
+import simplejson as json
 import re

 logger = logging.getLogger(__name__)
@@ -594,7 +594,8 @@ class RoomStore(RoomWorkerStore, SearchStore):

        while next_token:
            sql = """
-                SELECT stream_ordering, content FROM events
+                SELECT stream_ordering, json FROM events
+                JOIN event_json USING (event_id)
                WHERE room_id = ?
                    AND stream_ordering < ?
                    AND contains_url = ? AND outlier = ?
@@ -606,8 +607,8 @@ class RoomStore(RoomWorkerStore, SearchStore):
            next_token = None
            for stream_ordering, content_json in txn:
                next_token = stream_ordering
-                content = json.loads(content_json)
-
+                event_json = json.loads(content_json)
+                content = event_json["content"]
                content_url = content.get("url")
                thumbnail_url = content.get("info", {}).get("thumbnail_url")

--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -28,7 +28,7 @@ from synapse.api.constants import Membership, EventTypes
 from synapse.types import get_domain_from_id

 import logging
-import ujson as json
+import simplejson as json

 logger = logging.getLogger(__name__)

@@ -65,7 +65,7 @@ class RoomMemberWorkerStore(EventsWorkerStore):
        hosts = frozenset(get_domain_from_id(user_id) for user_id in user_ids)
        defer.returnValue(hosts)

-    @cached(max_entries=100000, iterable=True)
+    @cachedInlineCallbacks(max_entries=100000, iterable=True)
    def get_users_in_room(self, room_id):
        def f(txn):
            sql = (
@@ -79,7 +79,14 @@ class RoomMemberWorkerStore(EventsWorkerStore):

            txn.execute(sql, (room_id, Membership.JOIN,))
            return [to_ascii(r[0]) for r in txn]
-        return self.runInteraction("get_users_in_room", f)
+        start_time = self._clock.time_msec()
+        result = yield self.runInteraction("get_users_in_room", f)
+        end_time = self._clock.time_msec()
+        logger.info(
+            "Fetched room membership for %s (%i users) in %i ms",
+            room_id, len(result), end_time - start_time,
+        )
+        defer.returnValue(result)

    @cached()
    def get_invited_rooms_for_user(self, user_id):
@@ -453,7 +460,7 @@ class RoomMemberWorkerStore(EventsWorkerStore):

        defer.returnValue(joined_hosts)

-    @cached(max_entries=10000, iterable=True)
+    @cached(max_entries=10000)
    def _get_joined_hosts_cache(self, room_id):
        return _JoinedHostsCache(self, room_id)

@@ -645,8 +652,9 @@ class RoomMemberStore(RoomMemberWorkerStore):

        def add_membership_profile_txn(txn):
            sql = ("""
-                SELECT stream_ordering, event_id, events.room_id, content
+                SELECT stream_ordering, event_id, events.room_id, event_json.json
                FROM events
+                INNER JOIN event_json USING (event_id)
                INNER JOIN room_memberships USING (event_id)
                WHERE ? <= stream_ordering AND stream_ordering < ?
                AND type = 'm.room.member'
@@ -667,7 +675,8 @@ class RoomMemberStore(RoomMemberWorkerStore):
                event_id = row["event_id"]
                room_id = row["room_id"]
                try:
-                    content = json.loads(row["content"])
+                    event_json = json.loads(row["json"])
+                    content = event_json['content']
                except Exception:
                    continue

--- a/synapse/storage/schema/delta/14/upgrade_appservice_db.py
+++ b/synapse/storage/schema/delta/14/upgrade_appservice_db.py
@@ -12,9 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-import json
 import logging

+import simplejson as json
+
 logger = logging.getLogger(__name__)


--- a/synapse/storage/schema/delta/25/fts.py
+++ b/synapse/storage/schema/delta/25/fts.py
@@ -17,7 +17,7 @@ import logging
 from synapse.storage.prepare_database import get_statements
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine

-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -66,7 +66,7 @@ def run_create(cur, database_engine, *args, **kwargs):
            "max_stream_id_exclusive": max_stream_id + 1,
            "rows_inserted": 0,
        }
-        progress_json = ujson.dumps(progress)
+        progress_json = simplejson.dumps(progress)

        sql = (
            "INSERT into background_updates (update_name, progress_json)"
--- a/synapse/storage/schema/delta/27/ts.py
+++ b/synapse/storage/schema/delta/27/ts.py
@@ -16,7 +16,7 @@ import logging

 from synapse.storage.prepare_database import get_statements

-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -45,7 +45,7 @@ def run_create(cur, database_engine, *args, **kwargs):
            "max_stream_id_exclusive": max_stream_id + 1,
            "rows_inserted": 0,
        }
-        progress_json = ujson.dumps(progress)
+        progress_json = simplejson.dumps(progress)

        sql = (
            "INSERT into background_updates (update_name, progress_json)"
--- a/synapse/storage/schema/delta/31/search_update.py
+++ b/synapse/storage/schema/delta/31/search_update.py
@@ -16,7 +16,7 @@ from synapse.storage.engines import PostgresEngine
 from synapse.storage.prepare_database import get_statements

 import logging
-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -49,7 +49,7 @@ def run_create(cur, database_engine, *args, **kwargs):
            "rows_inserted": 0,
            "have_added_indexes": False,
        }
-        progress_json = ujson.dumps(progress)
+        progress_json = simplejson.dumps(progress)

        sql = (
            "INSERT into background_updates (update_name, progress_json)"
--- a/synapse/storage/schema/delta/33/event_fields.py
+++ b/synapse/storage/schema/delta/33/event_fields.py
@@ -15,7 +15,7 @@
 from synapse.storage.prepare_database import get_statements

 import logging
-import ujson
+import simplejson

 logger = logging.getLogger(__name__)

@@ -44,7 +44,7 @@ def run_create(cur, database_engine, *args, **kwargs):
            "max_stream_id_exclusive": max_stream_id + 1,
            "rows_inserted": 0,
        }
-        progress_json = ujson.dumps(progress)
+        progress_json = simplejson.dumps(progress)

        sql = (
            "INSERT into background_updates (update_name, progress_json)"
--- a/synapse/storage/schema/delta/48/add_user_ips_last_seen_index.sql
+++ b/synapse/storage/schema/delta/48/add_user_ips_last_seen_index.sql
@@ -0,0 +1,17 @@
+/* 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.
+ */
+
+INSERT into background_updates (update_name, progress_json)
+    VALUES ('user_ips_last_seen_index', '{}');
--- a/synapse/storage/schema/delta/48/groups_joinable.sql
+++ b/synapse/storage/schema/delta/48/groups_joinable.sql
@@ -0,0 +1,22 @@
+/* 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.
+ */
+
+/* 
+ * This isn't a real ENUM because sqlite doesn't support it
+ * and we use a default of NULL for inserted rows and interpret
+ * NULL at the python store level as necessary so that existing
+ * rows are given the correct default policy.
+ */
+ALTER TABLE groups ADD COLUMN join_policy TEXT NOT NULL DEFAULT 'invite';
--- a/synapse/storage/search.py
+++ b/synapse/storage/search.py
@@ -16,7 +16,7 @@
 from collections import namedtuple
 import logging
 import re
-import ujson as json
+import simplejson as json

 from twisted.internet import defer

@@ -75,8 +75,9 @@ class SearchStore(BackgroundUpdateStore):

        def reindex_search_txn(txn):
            sql = (
-                "SELECT stream_ordering, event_id, room_id, type, content, "
+                "SELECT stream_ordering, event_id, room_id, type, json, "
                " origin_server_ts FROM events"
+                " JOIN event_json USING (event_id)"
                " WHERE ? <= stream_ordering AND stream_ordering < ?"
                " AND (%s)"
                " ORDER BY stream_ordering DESC"
@@ -104,7 +105,8 @@ class SearchStore(BackgroundUpdateStore):
                    stream_ordering = row["stream_ordering"]
                    origin_server_ts = row["origin_server_ts"]
                    try:
-                        content = json.loads(row["content"])
+                        event_json = json.loads(row["json"])
+                        content = event_json["content"]
                    except Exception:
                        continue

@@ -719,7 +721,7 @@ def _parse_query(database_engine, search_term):
    results = re.findall(r"([\w\-]+)", search_term, re.UNICODE)

    if isinstance(database_engine, PostgresEngine):
-        return " & ".join(result + ":*" for result in results)
+        return " & ".join(result for result in results)
    elif isinstance(database_engine, Sqlite3Engine):
        return " & ".join(result + "*" for result in results)
    else:
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -20,7 +20,7 @@ from twisted.internet import defer

 from synapse.storage.background_updates import BackgroundUpdateStore
 from synapse.storage.engines import PostgresEngine
-from synapse.util.caches import intern_string, CACHE_SIZE_FACTOR
+from synapse.util.caches import intern_string, get_cache_factor_for
 from synapse.util.caches.descriptors import cached, cachedList
 from synapse.util.caches.dictionary_cache import DictionaryCache
 from synapse.util.stringutils import to_ascii
@@ -54,7 +54,7 @@ class StateGroupWorkerStore(SQLBaseStore):
        super(StateGroupWorkerStore, self).__init__(db_conn, hs)

        self._state_group_cache = DictionaryCache(
-            "*stateGroupCache*", 100000 * CACHE_SIZE_FACTOR
+            "*stateGroupCache*", 500000 * get_cache_factor_for("stateGroupCache")
        )

    @cached(max_entries=100000, iterable=True)
@@ -566,8 +566,7 @@ class StateGroupWorkerStore(SQLBaseStore):
                state_dict = results[group]

                state_dict.update(
-                    ((intern_string(k[0]), intern_string(k[1])), to_ascii(v))
-                    for k, v in group_state_dict.iteritems()
+                    group_state_dict.iteritems()
                )

                self._state_group_cache.update(
--- a/synapse/storage/tags.py
+++ b/synapse/storage/tags.py
@@ -19,7 +19,7 @@ from synapse.storage.account_data import AccountDataWorkerStore
 from synapse.util.caches.descriptors import cached
 from twisted.internet import defer

-import ujson as json
+import simplejson as json
 import logging

 logger = logging.getLogger(__name__)
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -23,7 +23,7 @@ from canonicaljson import encode_canonical_json
 from collections import namedtuple

 import logging
-import ujson as json
+import simplejson as json

 logger = logging.getLogger(__name__)

--- a/Show More
+++ b/Show More