Don't send outlier events to ASes

CHANGES.rst

@@ -1,58 +1,3 @@
-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.
-
-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)
 =======================================
 

contrib/graph/graph3.py

@@ -22,8 +22,6 @@ 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"
@@ -60,7 +58,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, string_types):
+            elif isinstance(value, basestring):
                 pass
             else:
                 value = json.dumps(value)

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_request_count_servlet"></div>
+<div id="synapse_http_server_requests_servlet"></div>
 <script>
 new PromConsole.Graph({
-  node: document.querySelector("#synapse_http_server_request_count_servlet"),
-  expr: "rate(synapse_http_server_request_count:servlet[2m])",
+  node: document.querySelector("#synapse_http_server_requests_servlet"),
+  expr: "rate(synapse_http_server_requests: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_request_count_servlet_minus_events"></div>
+<div id="synapse_http_server_requests_servlet_minus_events"></div>
 <script>
 new PromConsole.Graph({
-  node: document.querySelector("#synapse_http_server_request_count_servlet_minus_events"),
-  expr: "rate(synapse_http_server_request_count:servlet{servlet!=\"EventStreamRestServlet\", servlet!=\"SyncRestServlet\"}[2m])",
+  node: document.querySelector("#synapse_http_server_requests_servlet_minus_events"),
+  expr: "rate(synapse_http_server_requests: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_seconds[2m]) / rate(synapse_http_server_response_count[2m]) / 1000",
+  expr: "rate(synapse_http_server_response_time:total[2m]) / rate(synapse_http_server_response_time: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_seconds[2m])",
+  expr: "rate(synapse_http_server_response_ru_utime:total[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_seconds[2m])",
+  expr: "rate(synapse_http_server_response_db_txn_duration:total[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_second{servlet='RoomSendEventRestServlet'}[2m]) / rate(synapse_http_server_response_count{servlet='RoomSendEventRestServlet'}[2m]) / 1000",
+  expr: "rate(synapse_http_server_response_time:total{servlet='RoomSendEventRestServlet'}[2m]) / rate(synapse_http_server_response_time:count{servlet='RoomSendEventRestServlet'}[2m]) / 1000",
   name: "[[servlet]]",
   yAxisFormatter: PromConsole.NumberFormatter.humanize,
   yHoverFormatter: PromConsole.NumberFormatter.humanize,

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_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: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:total{servlet=""} = sum(synapse_http_server_request_count:by_method) by (servlet)
+synapse_http_server_requests:total{servlet=""} = sum(synapse_http_server_requests: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])

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_request_count:method'
+  - record: 'synapse_http_server_requests:method'
     labels:
       servlet: ""
-    expr: "sum(synapse_http_server_request_count) by (method)"
-  - record: 'synapse_http_server_request_count:servlet'
+    expr: "sum(synapse_http_server_requests) by (method)"
+  - record: 'synapse_http_server_requests:servlet'
     labels:
       method: ""
-    expr: 'sum(synapse_http_server_request_count) by (servlet)'
+    expr: 'sum(synapse_http_server_requests) by (servlet)'
 
-  - record: 'synapse_http_server_request_count:total'
+  - record: 'synapse_http_server_requests:total'
     labels:
       servlet: ""
-    expr: 'sum(synapse_http_server_request_count:by_method) by (servlet)'
+    expr: 'sum(synapse_http_server_requests: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])'

docs/workers.rst

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

scripts/synapse_port_db

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

synapse/__init__.py

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

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(
-                b"User-Agent",
-                default=[b""]
+                "User-Agent",
+                default=[""]
             )[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(b"Authorization")
+    auth_headers = request.requestHeaders.getRawHeaders("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(b"Authorization")
-    query_params = request.args.get(b"access_token")
+    auth_headers = request.requestHeaders.getRawHeaders("Authorization")
+    query_params = request.args.get("access_token")
     if auth_headers:
         # Try the get the access_token from a "Authorization: Bearer"
         # header

synapse/app/appservice.py

@@ -42,7 +42,7 @@ logger = logging.getLogger("synapse.app.appservice")
 
 
 class AppserviceSlaveStore(
-    DirectoryStore, SlavedEventStore, SlavedApplicationServiceStore,
+    DirectoryStore, SlavedApplicationServiceStore, SlavedEventStore,
     SlavedRegistrationStore,
 ):
     pass

synapse/app/client_reader.py

@@ -50,11 +50,11 @@ logger = logging.getLogger("synapse.app.client_reader")
 
 
 class ClientReaderSlavedStore(
+    SlavedApplicationServiceStore,
     SlavedEventStore,
     SlavedKeyStore,
     RoomStore,
     DirectoryStore,
-    SlavedApplicationServiceStore,
     SlavedRegistrationStore,
     TransactionStore,
     SlavedClientIpStore,

synapse/app/frontend_proxy.py

@@ -36,7 +36,6 @@ 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
@@ -50,35 +49,6 @@ 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>[^/]+))?$")
 
@@ -120,7 +90,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(b"Authorization", [])
+            auth_headers = request.requestHeaders.getRawHeaders("Authorization", [])
             headers = {
                 "Authorization": auth_headers,
             }
@@ -165,7 +135,6 @@ 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,

synapse/app/homeserver.py

@@ -430,10 +430,6 @@ 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

synapse/app/synchrotron.py

@@ -64,6 +64,7 @@ logger = logging.getLogger("synapse.app.synchrotron")
 class SynchrotronSlavedStore(
     SlavedReceiptsStore,
     SlavedAccountDataStore,
+    SlavedPushRuleStore,
     SlavedApplicationServiceStore,
     SlavedRegistrationStore,
     SlavedFilteringStore,
@@ -71,7 +72,6 @@ class SynchrotronSlavedStore(
     SlavedGroupServerStore,
     SlavedDeviceInboxStore,
     SlavedDeviceStore,
-    SlavedPushRuleStore,
     SlavedEventStore,
     SlavedClientIpStore,
     RoomStore,
@@ -113,7 +113,6 @@ 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):
@@ -211,8 +210,6 @@ 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

synapse/app/synctl.py

@@ -108,7 +108,7 @@ def stop(pidfile, app):
 
 
 Worker = collections.namedtuple("Worker", [
-    "app", "configfile", "pidfile", "cache_factor", "cache_factors",
+    "app", "configfile", "pidfile", "cache_factor"
 ])
 
 
@@ -171,10 +171,6 @@ 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
@@ -215,10 +211,6 @@ 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"
 
@@ -234,10 +226,8 @@ 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
@@ -262,7 +252,6 @@ def main():
             for running_pid in running_pids:
                 while pid_running(running_pid):
                     time.sleep(0.2)
-            write("All processes exited; now restarting...")
 
     if action == "start" or action == "restart":
         if start_stop_synapse:
@@ -272,19 +261,15 @@ 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)
 
-            # Reset env back to the original
-            os.environ.clear()
-            os.environ.update(env)
+            if cache_factor:
+                os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)
+            else:
+                os.environ.pop("SYNAPSE_CACHE_FACTOR", None)
 
 
 if __name__ == "__main__":

synapse/app/user_dir.py

@@ -49,8 +49,8 @@ logger = logging.getLogger("synapse.app.user_dir")
 
 
 class UserDirectorySlaveStore(
-    SlavedEventStore,
     SlavedApplicationServiceStore,
+    SlavedEventStore,
     SlavedRegistrationStore,
     SlavedClientIpStore,
     UserDirectoryStore,

synapse/appservice/__init__.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from synapse.api.constants import EventTypes
-from synapse.util.caches.descriptors import cachedInlineCallbacks
 from synapse.types import GroupID, get_domain_from_id
 
 from twisted.internet import defer
@@ -21,8 +20,6 @@ from twisted.internet import defer
 import logging
 import re
 
-from six import string_types
-
 logger = logging.getLogger(__name__)
 
 
@@ -148,7 +145,7 @@ class ApplicationService(object):
                         )
 
                 regex = regex_obj.get("regex")
-                if isinstance(regex, string_types):
+                if isinstance(regex, basestring):
                     regex_obj["regex"] = re.compile(regex)  # Pre-compile regex
                 else:
                     raise ValueError(
@@ -175,6 +172,7 @@ class ApplicationService(object):
 
         if self.is_interested_in_user(event.sender):
             defer.returnValue(True)
+
         # also check m.room.member state key
         if (event.type == EventTypes.Member and
                 self.is_interested_in_user(event.state_key)):
@@ -183,20 +181,18 @@ class ApplicationService(object):
         if not store:
             defer.returnValue(False)
 
-        does_match = yield self._matches_user_in_member_list(event.room_id, store)
+        does_match = yield self._matches_user_in_member_list(
+            event, store,
+        )
         defer.returnValue(does_match)
 
-    @cachedInlineCallbacks(num_args=1, cache_context=True)
-    def _matches_user_in_member_list(self, room_id, store, cache_context):
-        member_list = yield store.get_users_in_room(
-            room_id, on_invalidate=cache_context.invalidate
+    @defer.inlineCallbacks
+    def _matches_user_in_member_list(self, event, store):
+        ases = yield store.get_appservices_with_user_in_room(
+            event,
         )
 
-        # check joined member events
-        for user_id in member_list:
-            if self.is_interested_in_user(user_id):
-                defer.returnValue(True)
-        defer.returnValue(False)
+        defer.returnValue(self.id in ases)
 
     def _matches_room_id(self, event):
         if hasattr(event, "room_id"):

synapse/appservice/api.py

@@ -73,8 +73,7 @@ class ApplicationServiceApi(SimpleHttpClient):
         super(ApplicationServiceApi, self).__init__(hs)
         self.clock = hs.get_clock()
 
-        self.protocol_meta_cache = ResponseCache(hs, "as_protocol_meta",
-                                                 timeout_ms=HOUR_IN_MS)
+        self.protocol_meta_cache = ResponseCache(hs, timeout_ms=HOUR_IN_MS)
 
     @defer.inlineCallbacks
     def query_user(self, service, user_id):

synapse/config/_base.py

@@ -19,8 +19,6 @@ import os
 import yaml
 from textwrap import dedent
 
-from six import integer_types
-
 
 class ConfigError(Exception):
     pass
@@ -51,7 +49,7 @@ Missing mandatory `server_name` config option.
 class Config(object):
     @staticmethod
     def parse_size(value):
-        if isinstance(value, integer_types):
+        if isinstance(value, int) or isinstance(value, long):
             return value
         sizes = {"K": 1024, "M": 1024 * 1024}
         size = 1
@@ -63,7 +61,7 @@ class Config(object):
 
     @staticmethod
     def parse_duration(value):
-        if isinstance(value, integer_types):
+        if isinstance(value, int) or isinstance(value, long):
             return value
         second = 1000
         minute = 60 * second
@@ -290,22 +288,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(

synapse/config/appservice.py

@@ -21,8 +21,6 @@ import urllib
 import yaml
 import logging
 
-from six import string_types
-
 logger = logging.getLogger(__name__)
 
 
@@ -91,14 +89,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), string_types):
+        if not isinstance(as_info.get(field), basestring):
             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"), string_types) and
+    if (not isinstance(as_info.get("url"), basestring) and
             as_info.get("url", "") is not None):
         raise KeyError(
             "Required string field or explicit null: 'url' (%s)" % (config_filename,)
@@ -130,7 +128,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"), string_types):
+                if not isinstance(regex_obj.get("regex"), basestring):
                     raise ValueError(
                         "Missing/bad type 'regex' key in %s", regex_obj
                     )

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

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=100,
+                    last_token, self._last_poked_id, limit=20,
                 )
 
                 logger.debug("Handling %s -> %s", last_token, next_token)
@@ -177,33 +177,24 @@ class TransactionQueue(object):
                 if not events and next_token >= self._last_poked_id:
                     break
 
-                @defer.inlineCallbacks
-                def handle_event(event):
+                for event in events:
                     # 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:
-                        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
+                        continue
 
+                    # Get the state from before the event.
+                    # We need to make sure that this is the state from before
+                    # the event and not from after it.
+                    # Otherwise if the last member on a server in a room is
+                    # banned then it won't receive the event because it won't
+                    # be in the room after the ban.
+                    destinations = yield self.state.get_current_hosts_in_room(
+                        event.room_id, latest_event_ids=[
+                            prev_id for prev_id, _ in event.prev_events
+                        ],
+                    )
                     destinations = set(destinations)
 
                     if send_on_behalf_of is not None:
@@ -216,44 +207,12 @@ class TransactionQueue(object):
 
                     self._send_pdu(event, destinations)
 
-                @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
-                ))
+                events_processed_counter.inc_by(len(events))
 
                 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
 
@@ -295,7 +254,6 @@ class TransactionQueue(object):
         Args:
             states (list(UserPresenceState))
         """
-        return
 
         # First we queue up the new presence by user ID, so multiple presence
         # updates in quick successtion are correctly handled

synapse/federation/transport/client.py

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

synapse/federation/transport/server.py

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

synapse/groups/groups_server.py

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

synapse/handlers/appservice.py

@@ -18,9 +18,7 @@ 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, run_in_background,
-)
+from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
 
 import logging
 
@@ -86,16 +84,11 @@ 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:
-                            return  # no services need notifying
+                            continue  # 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.
@@ -115,33 +108,9 @@ class ApplicationServicesHandler(object):
                                 service, event
                             )
 
-                    @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",
-                    )
+                    yield self.store.set_appservice_last_pos(upper_bound)
             finally:
                 self.is_processing = False
 

synapse/handlers/groups_local.py

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

synapse/handlers/initial_sync.py

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

synapse/handlers/message.py

@@ -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 False and requester.app_service and user_id not in users_with_profile:
+        if 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,39 +454,40 @@ class EventCreationHandler(object):
         """
         builder = self.event_builder_factory.new(event_dict)
 
-        self.validator.validate_new(builder)
+        with (yield self.limiter.queue(builder.room_id)):
+            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))
 
@@ -556,34 +557,27 @@ 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
+        )
 
-        # 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
+        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
             )
 
-            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,
-            )
+        yield self.send_nonmember_event(
+            requester,
+            event,
+            context,
+            ratelimit=ratelimit,
+        )
         defer.returnValue(event)
 
     @measure_func("create_new_client_event")

synapse/handlers/presence.py

@@ -373,7 +373,6 @@ 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()
@@ -403,7 +402,6 @@ 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
@@ -446,8 +444,6 @@ 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
@@ -467,7 +463,6 @@ class PresenceHandler(object):
             syncing_user_ids(set(str)): The set of user_ids that are
                 currently syncing on that server.
         """
-        return
 
         # Grab the previous list of user_ids that were syncing on that process
         prev_syncing_user_ids = (

synapse/handlers/register.py

@@ -23,7 +23,7 @@ from synapse.api.errors import (
 )
 from synapse.http.client import CaptchaServerHttpClient
 from synapse import types
-from synapse.types import UserID, create_requester, RoomID, RoomAlias
+from synapse.types import UserID
 from synapse.util.async import run_on_reactor, Linearizer
 from synapse.util.threepids import check_3pid_allowed
 from ._base import BaseHandler
@@ -205,17 +205,10 @@ 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
@@ -490,28 +483,3 @@ 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",
-        )

synapse/handlers/room_list.py

@@ -44,9 +44,8 @@ EMTPY_THIRD_PARTY_ID = ThirdPartyInstanceID(None, None)
 class RoomListHandler(BaseHandler):
     def __init__(self, hs):
         super(RoomListHandler, self).__init__(hs)
-        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)
+        self.response_cache = ResponseCache(hs)
+        self.remote_response_cache = ResponseCache(hs, timeout_ms=30 * 1000)
 
     def get_local_public_room_list(self, limit=None, since_token=None,
                                    search_filter=None,

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, Limiter
+from synapse.util.async import Linearizer
 from synapse.util.distributor import user_left_room, user_joined_room
 
 
@@ -60,7 +60,6 @@ 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()
@@ -233,23 +232,18 @@ class RoomMemberHandler(object):
     ):
         key = (room_id,)
 
-        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,
-                )
+        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)
 
@@ -858,14 +852,6 @@ 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")
 

synapse/handlers/sync.py

@@ -30,7 +30,6 @@ import itertools
 
 logger = logging.getLogger(__name__)
 
-SYNC_RESPONSE_CACHE_MS = 2 * 60 * 1000
 
 SyncConfig = collections.namedtuple("SyncConfig", [
     "user",
@@ -170,10 +169,7 @@ 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, "sync",
-            timeout_ms=SYNC_RESPONSE_CACHE_MS,
-        )
+        self.response_cache = ResponseCache(hs)
         self.state = hs.get_state_handler()
 
     def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0,
@@ -603,7 +599,7 @@ class SyncHandler(object):
             since_token is None and
             sync_config.filter_collection.blocks_all_presence()
         )
-        if False and not block_all_presence_data:
+        if not block_all_presence_data:
             yield self._generate_sync_entry_for_presence(
                 sync_result_builder, newly_joined_rooms, newly_joined_users
             )

synapse/http/endpoint.py

@@ -12,6 +12,8 @@
 # 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
@@ -31,7 +33,7 @@ SERVER_CACHE = {}
 
 # our record of an individual server which can be tried to reach a destination.
 #
-# "host" is the hostname acquired from the SRV record. Except when there's
+# "host" is actually a dotted-quad or ipv6 address string. Except when there's
 # no SRV record, in which case it is the original hostname.
 _Server = collections.namedtuple(
     "_Server", "priority weight host port expires"
@@ -295,13 +297,20 @@ def resolve_service(service_name, dns_client=client, cache=SERVER_CACHE, clock=t
 
             payload = answer.payload
 
-            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,
-            ))
+            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.sort()
         cache[service_name] = list(servers)
@@ -319,3 +328,81 @@ 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)

synapse/http/matrixfederationclient.py

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

synapse/http/server.py

@@ -113,11 +113,6 @@ response_db_sched_duration = metrics.register_counter(
     "response_db_sched_duration_seconds", labels=["method", "servlet", "tag"]
 )
 
-# size in bytes of the response written
-response_size = metrics.register_counter(
-    "response_size", labels=["method", "servlet", "tag"]
-)
-
 _next_request_id = 0
 
 
@@ -329,7 +324,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 == b"OPTIONS":
+        if request.method == "OPTIONS":
             return _options_handler, {}
 
         # Loop through all the registered callbacks to check if the method
@@ -431,8 +426,6 @@ class RequestMetrics(object):
             context.db_sched_duration_ms / 1000., request.method, self.name, tag
         )
 
-        response_size.inc_by(request.sentLength, request.method, self.name, tag)
-
 
 class RootRedirect(resource.Resource):
     """Redirects the root '/' path to another path."""
@@ -543,7 +536,7 @@ def finish_request(request):
 
 def _request_user_agent_is_curl(request):
     user_agents = request.requestHeaders.getRawHeaders(
-        b"User-Agent", default=[]
+        "User-Agent", default=[]
     )
     for user_agent in user_agents:
         if "curl" in user_agent:

synapse/http/site.py

@@ -20,7 +20,7 @@ import logging
 import re
 import time
 
-ACCESS_TOKEN_RE = re.compile(br'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')
+ACCESS_TOKEN_RE = re.compile(r'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')
 
 
 class SynapseRequest(Request):
@@ -43,12 +43,12 @@ class SynapseRequest(Request):
 
     def get_redacted_uri(self):
         return ACCESS_TOKEN_RE.sub(
-            br'\1<redacted>\3',
+            r'\1<redacted>\3',
             self.uri
         )
 
     def get_user_agent(self):
-        return self.requestHeaders.getRawHeaders(b"User-Agent", [None])[-1]
+        return self.requestHeaders.getRawHeaders("User-Agent", [None])[-1]
 
     def started_processing(self):
         self.site.access_logger.info(

synapse/metrics/__init__.py

@@ -17,13 +17,12 @@ import logging
 import functools
 import time
 import gc
-import platform
 
 from twisted.internet import reactor
 
 from .metric import (
     CounterMetric, CallbackMetric, DistributionMetric, CacheMetric,
-    MemoryUsageMetric, GaugeMetric,
+    MemoryUsageMetric,
 )
 from .process_collector import register_process_collector
 
@@ -31,7 +30,6 @@ from .process_collector import register_process_collector
 logger = logging.getLogger(__name__)
 
 
-running_on_pypy = platform.python_implementation() == 'PyPy'
 all_metrics = []
 all_collectors = []
 
@@ -65,13 +63,6 @@ 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:
@@ -151,32 +142,6 @@ 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):
 
@@ -209,9 +174,6 @@ 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()
@@ -244,7 +206,6 @@ try:
 
     # We manually run the GC each reactor tick so that we can get some metrics
     # about time spent doing GC,
-    if not running_on_pypy:
-        gc.disable()
+    gc.disable()
 except AttributeError:
     pass

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 tuple).
+        # (if the metric is a scalar, the (single) key is the empty list).
         self.counts = {}
 
         # Scalar metrics are never empty
@@ -145,36 +145,6 @@ 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

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, "send_event", timeout_ms=30 * 60 * 1000)
+        self.response_cache = ResponseCache(hs, timeout_ms=30 * 60 * 1000)
 
     def on_PUT(self, request, event_id):
         result = self.response_cache.get(event_id)

synapse/replication/slave/storage/appservice.py

@@ -17,8 +17,10 @@
 from synapse.storage.appservice import (
     ApplicationServiceWorkerStore, ApplicationServiceTransactionWorkerStore,
 )
+from synapse.replication.slave.storage.events import SlavedEventStore
 
 
 class SlavedApplicationServiceStore(ApplicationServiceTransactionWorkerStore,
-                                    ApplicationServiceWorkerStore):
+                                    ApplicationServiceWorkerStore,
+                                    SlavedEventStore):
     pass

synapse/replication/slave/storage/client_ips.py

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

synapse/replication/tcp/streams.py

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

synapse/rest/client/v1/admin.py

@@ -296,6 +296,17 @@ 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)
@@ -333,17 +344,6 @@ 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(

synapse/rest/client/v1/logout.py

@@ -44,10 +44,7 @@ class LogoutRestServlet(ClientV1RestServlet):
             requester = yield self.auth.get_user_by_req(request)
         except AuthError:
             # this implies the access token has already been deleted.
-            defer.returnValue((401, {
-                "errcode": "M_UNKNOWN_TOKEN",
-                "error": "Access Token unknown or expired"
-            }))
+            pass
         else:
             if requester.device_id is None:
                 # the acccess token wasn't associated with a device.

synapse/rest/client/v1/presence.py

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

synapse/rest/client/v1/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 b"\x00" in user:
+        if "\x00" in user:
             raise SynapseError(400, "Invalid user")
-        if b"\x00" in password:
+        if "\x00" in password:
             raise SynapseError(400, "Invalid password")
 
         # str() because otherwise hmac complains that 'unicode' does not

synapse/rest/client/v1/room.py

@@ -165,12 +165,17 @@ class RoomStateEventRestServlet(ClientV1RestServlet):
                 content=content,
             )
         else:
-            event = yield self.event_creation_hander.create_and_send_nonmember_event(
+            event, context = yield self.event_creation_hander.create_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}
@@ -650,12 +655,7 @@ class RoomMembershipRestServlet(ClientV1RestServlet):
             content=event_content,
         )
 
-        return_value = {}
-
-        if membership_action == "join":
-            return_value["room_id"] = room_id
-
-        defer.returnValue((200, return_value))
+        defer.returnValue((200, {}))
 
     def _has_3pid_invite_keys(self, content):
         for key in {"id_server", "medium", "address"}:

synapse/rest/client/v2_alpha/groups.py

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

synapse/rest/client/v2_alpha/register.py

@@ -20,6 +20,7 @@ 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
@@ -404,6 +405,14 @@ 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(
@@ -436,6 +445,29 @@ 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(

synapse/rest/media/v1/media_storage.py

@@ -16,8 +16,6 @@
 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
@@ -121,7 +119,7 @@ class MediaStorage(object):
                 os.remove(fname)
             except Exception:
                 pass
-            six.reraise(t, v, tb)
+            raise t, v, tb
 
         if not finished_called:
             raise Exception("Finished callback not called")

synapse/storage/__init__.py

@@ -14,6 +14,8 @@
 # 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
@@ -242,12 +244,13 @@ class DataStore(RoomMemberStore, RoomStore,
 
         return [UserPresenceState(**row) for row in rows]
 
+    @defer.inlineCallbacks
     def count_daily_users(self):
         """
         Counts the number of users who used this homeserver in the last 24 hours.
         """
         def _count_users(txn):
-            yesterday = int(self._clock.time_msec()) - (1000 * 60 * 60 * 24)
+            yesterday = int(self._clock.time_msec()) - (1000 * 60 * 60 * 24),
 
             sql = """
                 SELECT COALESCE(count(*), 0) FROM (
@@ -261,91 +264,8 @@ class DataStore(RoomMemberStore, RoomStore,
             count, = txn.fetchone()
             return count
 
-        return self.runInteraction("count_users", _count_users)
-
-    def count_r30_users(self):
-        """
-        Counts the number of 30 day retained users, defined as:-
-         * Users who have created their accounts more than 30 days 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)
+        ret = yield self.runInteraction("count_users", _count_users)
+        defer.returnValue(ret)
 
     def get_users(self):
         """Function to reterive a list of users in users table.

synapse/storage/_base.py

@@ -376,7 +376,7 @@ class SQLBaseStore(object):
         Returns:
             A list of dicts where the key is the column header.
         """
-        col_headers = list(intern(str(column[0])) for column in cursor.description)
+        col_headers = list(intern(column[0]) for column in cursor.description)
         results = list(
             dict(zip(col_headers, row)) for row in cursor
         )

synapse/storage/appservice.py

@@ -18,9 +18,14 @@ import re
 import simplejson as json
 from twisted.internet import defer
 
+from synapse.api.constants import EventTypes, Membership
 from synapse.appservice import AppServiceTransaction
 from synapse.config.appservice import load_appservices
 from synapse.storage.events import EventsWorkerStore
+from synapse.storage.roommember import RoomMemberWorkerStore
+from synapse.storage.state import StateGroupWorkerStore
+from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
+from synapse.util.async import Linearizer
 from ._base import SQLBaseStore
 
 
@@ -46,7 +51,8 @@ def _make_exclusive_regex(services_cache):
     return exclusive_user_regex
 
 
-class ApplicationServiceWorkerStore(SQLBaseStore):
+class ApplicationServiceWorkerStore(RoomMemberWorkerStore, StateGroupWorkerStore,
+                                    SQLBaseStore):
     def __init__(self, db_conn, hs):
         self.services_cache = load_appservices(
             hs.hostname,
@@ -111,6 +117,38 @@ class ApplicationServiceWorkerStore(SQLBaseStore):
                 return service
         return None
 
+    @defer.inlineCallbacks
+    def get_appservices_with_user_in_room(self, event):
+        """Get the list of appservices in the room at the given event
+
+        Args:
+            event (Event)
+
+        Returns:
+            Deferred[set(str)]: The IDs of all ASes in the room
+        """
+        state_group = yield self._get_state_group_for_event(event.event_id)
+
+        if not state_group:
+            raise Exception("No state group for event %s", event.event_id)
+
+        ases_in_room = yield self._get_appservices_with_user_in_room(
+            event.room_id, state_group,
+        )
+
+        defer.returnValue(ases_in_room)
+
+    @cachedInlineCallbacks(num_args=2, max_entries=10000)
+    def _get_appservices_with_user_in_room(self, room_id, state_group):
+        cache = self._get_appservices_with_user_in_room_cache(room_id)
+        ases_in_room = yield cache.get_appservices_in_room_by_user(state_group)
+
+        defer.returnValue(ases_in_room)
+
+    @cached(max_entries=10000)
+    def _get_appservices_with_user_in_room_cache(self, room_id):
+        return _AppserviceUsersCache(self, room_id)
+
 
 class ApplicationServiceStore(ApplicationServiceWorkerStore):
     # This is currently empty due to there not being any AS storage functions
@@ -346,6 +384,7 @@ class ApplicationServiceTransactionWorkerStore(ApplicationServiceWorkerStore,
                 " (SELECT stream_ordering FROM appservice_stream_position)"
                 "     < e.stream_ordering"
                 " AND e.stream_ordering <= ?"
+                " AND NOT e.outlier"
                 " ORDER BY e.stream_ordering ASC"
                 " LIMIT ?"
             )
@@ -374,3 +413,119 @@ class ApplicationServiceTransactionStore(ApplicationServiceTransactionWorkerStor
     # to keep consistency with the other stores, we keep this empty class for
     # now.
     pass
+
+
+class _AppserviceUsersCache(object):
+    """Attempts to calculate which appservices have users in a given room by
+    looking at state groups and their delta_ids
+    """
+
+    def __init__(self, store, room_id):
+        self.store = store
+        self.room_id = room_id
+
+        self.linearizer = Linearizer("_AppserviceUsersCache")
+
+        # The last state group we calculated the ASes in the room for.
+        self.state_group = object()
+
+        # A dict of all appservices in the room at the above state group,
+        # along with a user_id of an AS user in the room.
+        # Dict of as_id -> user_id.
+        self.appservices_in_room = {}
+
+    @defer.inlineCallbacks
+    def get_appservices_in_room_by_user(self, state_group):
+        """
+        Args:
+            state_group(str)
+
+        Returns:
+            Deferred[set(str)]: The IDs of all ASes in the room
+        """
+        assert state_group is not None
+
+        if state_group == self.state_group:
+            defer.returnValue(frozenset(self.appservices_in_room))
+
+        with (yield self.linearizer.queue(())):
+            # Set of ASes that we need to recalculate their membership of
+            # the room
+            uhandled_ases = set()
+
+            # If the state groups match then there is nothing to do
+            if state_group == self.state_group:
+                defer.returnValue(frozenset(self.appservices_in_room))
+
+            prev_group, delta_ids = yield self.store.get_state_group_delta(state_group)
+
+            # If the prev_group matches the last state group we can calculate
+            # the new value by looking at the deltas
+            if prev_group and prev_group == self.state_group:
+                for (typ, state_key), event_id in delta_ids.iteritems():
+                    if typ != EventTypes.Member:
+                        continue
+
+                    user_id = state_key
+
+                    event = yield self.store.get_event(event_id)
+
+                    is_join = event.membership == Membership.JOIN
+                    for appservice in self.store.get_app_services():
+                        as_id = appservice.id
+
+                        # If this is a join and the appservice is already in
+                        # the room then its a noop
+                        if is_join:
+                            if as_id in self.appservices_in_room:
+                                continue
+                        # If this is not a join, then we only need to recalculate
+                        # if the AS is in the room and the cached joined AS user
+                        # matches this event.
+                        elif self.appservices_in_room.get(as_id, None) != user_id:
+                            continue
+
+                        # If the AS is not interested in the user then its a
+                        # noop.
+                        if not appservice.is_interested_in_user(user_id):
+                            continue
+
+                        if is_join:
+                            # If an AS user is joining then the AS is now
+                            # interested in the room
+                            self.appservices_in_room[as_id] = user_id
+                        else:
+                            # If an AS user has left then we need to
+                            # recalcualte if they're in the room.
+                            uhandled_ases.add(appservice)
+                            self.appservices_in_room.pop(as_id, None)
+                else:
+                    uhandled_ases = set(self.store.get_app_services())
+
+            if uhandled_ases:
+                # We need to recalculate which ASes are in the room, so lets
+                # get the current state and try and find a join event
+                # that the AS is interested in.
+
+                current_state_ids = yield self.store.get_state_ids_for_group(state_group)
+
+                for appservice in uhandled_ases:
+                    as_id = appservice.id
+
+                    self.appservices_in_room.pop(as_id, None)
+
+                    for (etype, state_key), event_id in current_state_ids.iteritems():
+                        if etype != EventTypes.Member:
+                            continue
+
+                        if not appservice.is_interested_in_user(state_key):
+                            continue
+
+                        event = yield self.store.get_event(event_id)
+                        if event.membership == Membership.JOIN:
+                            self.appservices_in_room[as_id] = state_key
+                            break
+
+            self.state_group = state_group
+
+        defer.returnValue(frozenset(self.appservices_in_room))

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 = 10 * 60 * 1000
+LAST_SEEN_GRANULARITY = 120 * 1000
 
 
 class ClientIpStore(background_updates.BackgroundUpdateStore):
@@ -48,13 +48,6 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             columns=["user_id", "device_id", "last_seen"],
         )
 
-        self.register_background_index_update(
-            "user_ips_last_seen_index",
-            index_name="user_ips_last_seen",
-            table="user_ips",
-            columns=["user_id", "last_seen"],
-        )
-
         # (user_id, access_token, ip) -> (user_agent, device_id, last_seen)
         self._batch_row_update = {}
 

synapse/storage/engines/__init__.py

@@ -18,7 +18,6 @@ from .postgres import PostgresEngine
 from .sqlite3 import Sqlite3Engine
 
 import importlib
-import platform
 
 
 SUPPORTED_MODULE = {
@@ -32,10 +31,6 @@ 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)
 

synapse/storage/event_push_actions.py

@@ -613,9 +613,6 @@ 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
@@ -812,7 +809,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
                 )
                 if caught_up:
                     break
-                yield sleep(self._rotate_delay)
+                yield sleep(5)
         finally:
             self._doing_notif_rotation = False
 
@@ -833,8 +830,8 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         txn.execute("""
             SELECT stream_ordering FROM event_push_actions
             WHERE stream_ordering > ?
-            ORDER BY stream_ordering ASC LIMIT 1 OFFSET ?
-        """, (old_rotate_stream_ordering, self._rotate_count))
+            ORDER BY stream_ordering ASC LIMIT 1 OFFSET 50000
+        """, (old_rotate_stream_ordering,))
         stream_row = txn.fetchone()
         if stream_row:
             offset_stream_ordering, = stream_row

synapse/storage/events.py

@@ -444,9 +444,6 @@ 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"
@@ -1013,6 +1010,7 @@ class EventsStore(EventsWorkerStore):
                 "event_edge_hashes",
                 "event_edges",
                 "event_forward_extremities",
+                "event_push_actions",
                 "event_reference_hashes",
                 "event_search",
                 "event_signatures",
@@ -1032,14 +1030,6 @@ 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
 

synapse/storage/events_worker.py

@@ -51,26 +51,6 @@ _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,

synapse/storage/group_server.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2017 Vector Creations Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -30,24 +29,6 @@ _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",
@@ -55,11 +36,10 @@ class GroupServerStore(SQLBaseStore):
                 "group_id": group_id,
             },
             retcols=(
-                "name", "short_description", "long_description",
-                "avatar_url", "is_public", "join_policy",
+                "name", "short_description", "long_description", "avatar_url", "is_public"
             ),
             allow_none=True,
-            desc="get_group",
+            desc="is_user_in_group",
         )
 
     def get_users_in_group(self, group_id, include_private=False):

synapse/storage/prepare_database.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014 - 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -26,7 +25,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 = 48
+SCHEMA_VERSION = 47
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 

synapse/storage/room.py

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

synapse/storage/roommember.py

@@ -65,7 +65,7 @@ class RoomMemberWorkerStore(EventsWorkerStore):
         hosts = frozenset(get_domain_from_id(user_id) for user_id in user_ids)
         defer.returnValue(hosts)
 
-    @cachedInlineCallbacks(max_entries=100000, iterable=True)
+    @cached(max_entries=100000, iterable=True)
     def get_users_in_room(self, room_id):
         def f(txn):
             sql = (
@@ -79,14 +79,7 @@ class RoomMemberWorkerStore(EventsWorkerStore):
 
             txn.execute(sql, (room_id, Membership.JOIN,))
             return [to_ascii(r[0]) for r in txn]
-        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)
+        return self.runInteraction("get_users_in_room", f)
 
     @cached()
     def get_invited_rooms_for_user(self, user_id):
@@ -447,7 +440,6 @@ class RoomMemberWorkerStore(EventsWorkerStore):
         )
 
     @cachedInlineCallbacks(num_args=2, max_entries=10000, iterable=True)
-    # @defer.inlineCallbacks
     def _get_joined_hosts(self, room_id, state_group, current_state_ids, state_entry):
         # We don't use `state_group`, its there so that we can cache based
         # on it. However, its important that its never None, since two current_state's
@@ -460,7 +452,7 @@ class RoomMemberWorkerStore(EventsWorkerStore):
 
         defer.returnValue(joined_hosts)
 
-    @cached(max_entries=10000)
+    @cached(max_entries=10000, iterable=True)
     def _get_joined_hosts_cache(self, room_id):
         return _JoinedHostsCache(self, room_id)
 
@@ -652,9 +644,8 @@ class RoomMemberStore(RoomMemberWorkerStore):
 
         def add_membership_profile_txn(txn):
             sql = ("""
-                SELECT stream_ordering, event_id, events.room_id, event_json.json
+                SELECT stream_ordering, event_id, events.room_id, content
                 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'
@@ -675,8 +666,7 @@ class RoomMemberStore(RoomMemberWorkerStore):
                 event_id = row["event_id"]
                 room_id = row["room_id"]
                 try:
-                    event_json = json.loads(row["json"])
-                    content = event_json['content']
+                    content = json.loads(row["content"])
                 except Exception:
                     continue
 

synapse/storage/schema/delta/48/add_user_ips_last_seen_index.sql

@@ -1,17 +0,0 @@
-/* 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', '{}');

synapse/storage/schema/delta/48/groups_joinable.sql

@@ -1,22 +0,0 @@
-/* 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';

synapse/storage/search.py

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

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, get_cache_factor_for
+from synapse.util.caches import intern_string, CACHE_SIZE_FACTOR
 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*", 500000 * get_cache_factor_for("stateGroupCache")
+            "*stateGroupCache*", 100000 * CACHE_SIZE_FACTOR
         )
 
     @cached(max_entries=100000, iterable=True)
@@ -566,7 +566,8 @@ class StateGroupWorkerStore(SQLBaseStore):
                 state_dict = results[group]
 
                 state_dict.update(
-                    group_state_dict.iteritems()
+                    ((intern_string(k[0]), intern_string(k[1])), to_ascii(v))
+                    for k, v in group_state_dict.iteritems()
                 )
 
                 self._state_group_cache.update(

synapse/types.py

@@ -169,7 +169,7 @@ class DomainSpecificString(
         except Exception:
             return False
 
-    __repr__ = to_string
+    __str__ = to_string
 
 
 class UserID(DomainSpecificString):

synapse/util/caches/__init__.py

@@ -18,16 +18,6 @@ import os
 
 CACHE_SIZE_FACTOR = float(os.environ.get("SYNAPSE_CACHE_FACTOR", 0.5))
 
-
-def get_cache_factor_for(cache_name):
-    env_var = "SYNAPSE_CACHE_FACTOR_" + cache_name.upper()
-    factor = os.environ.get(env_var)
-    if factor:
-        return float(factor)
-
-    return CACHE_SIZE_FACTOR
-
-
 metrics = synapse.metrics.get_metrics_for("synapse.util.caches")
 
 caches_by_name = {}

synapse/util/caches/descriptors.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2015, 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,7 +16,7 @@ import logging
 
 from synapse.util.async import ObservableDeferred
 from synapse.util import unwrapFirstError, logcontext
-from synapse.util.caches import get_cache_factor_for
+from synapse.util.caches import CACHE_SIZE_FACTOR
 from synapse.util.caches.lrucache import LruCache
 from synapse.util.caches.treecache import TreeCache, iterate_tree_cache_entry
 from synapse.util.stringutils import to_ascii
@@ -40,11 +39,12 @@ _CacheSentinel = object()
 
 class CacheEntry(object):
     __slots__ = [
-        "deferred", "callbacks", "invalidated"
+        "deferred", "sequence", "callbacks", "invalidated"
     ]
 
-    def __init__(self, deferred, callbacks):
+    def __init__(self, deferred, sequence, callbacks):
         self.deferred = deferred
+        self.sequence = sequence
         self.callbacks = set(callbacks)
         self.invalidated = False
 
@@ -62,6 +62,7 @@ class Cache(object):
         "max_entries",
         "name",
         "keylen",
+        "sequence",
         "thread",
         "metrics",
         "_pending_deferred_cache",
@@ -79,6 +80,7 @@ class Cache(object):
 
         self.name = name
         self.keylen = keylen
+        self.sequence = 0
         self.thread = None
         self.metrics = register_cache(name, self.cache)
 
@@ -111,10 +113,11 @@ class Cache(object):
         callbacks = [callback] if callback else []
         val = self._pending_deferred_cache.get(key, _CacheSentinel)
         if val is not _CacheSentinel:
-            val.callbacks.update(callbacks)
-            if update_metrics:
-                self.metrics.inc_hits()
-            return val.deferred
+            if val.sequence == self.sequence:
+                val.callbacks.update(callbacks)
+                if update_metrics:
+                    self.metrics.inc_hits()
+                return val.deferred
 
         val = self.cache.get(key, _CacheSentinel, callbacks=callbacks)
         if val is not _CacheSentinel:
@@ -134,9 +137,12 @@ class Cache(object):
         self.check_thread()
         entry = CacheEntry(
             deferred=value,
+            sequence=self.sequence,
             callbacks=callbacks,
         )
 
+        entry.callbacks.update(callbacks)
+
         existing_entry = self._pending_deferred_cache.pop(key, None)
         if existing_entry:
             existing_entry.invalidate()
@@ -144,25 +150,13 @@ class Cache(object):
         self._pending_deferred_cache[key] = entry
 
         def shuffle(result):
-            existing_entry = self._pending_deferred_cache.pop(key, None)
-            if existing_entry is entry:
-                self.cache.set(key, result, entry.callbacks)
+            if self.sequence == entry.sequence:
+                existing_entry = self._pending_deferred_cache.pop(key, None)
+                if existing_entry is entry:
+                    self.cache.set(key, result, entry.callbacks)
+                else:
+                    entry.invalidate()
             else:
-                # oops, the _pending_deferred_cache has been updated since
-                # we started our query, so we are out of date.
-                #
-                # Better put back whatever we took out. (We do it this way
-                # round, rather than peeking into the _pending_deferred_cache
-                # and then removing on a match, to make the common case faster)
-                if existing_entry is not None:
-                    self._pending_deferred_cache[key] = existing_entry
-
-                # we're not going to put this entry into the cache, so need
-                # to make sure that the invalidation callbacks are called.
-                # That was probably done when _pending_deferred_cache was
-                # updated, but it's possible that `set` was called without
-                # `invalidate` being previously called, in which case it may
-                # not have been. Either way, let's double-check now.
                 entry.invalidate()
             return result
 
@@ -174,29 +168,25 @@ class Cache(object):
 
     def invalidate(self, key):
         self.check_thread()
-        self.cache.pop(key, None)
 
-        # if we have a pending lookup for this key, remove it from the
-        # _pending_deferred_cache, which will (a) stop it being returned
-        # for future queries and (b) stop it being persisted as a proper entry
-        # in self.cache.
+        # Increment the sequence number so that any SELECT statements that
+        # raced with the INSERT don't update the cache (SYN-369)
+        self.sequence += 1
         entry = self._pending_deferred_cache.pop(key, None)
-
-        # run the invalidation callbacks now, rather than waiting for the
-        # deferred to resolve.
         if entry:
             entry.invalidate()
 
+        self.cache.pop(key, None)
+
     def invalidate_many(self, key):
         self.check_thread()
         if not isinstance(key, tuple):
             raise TypeError(
                 "The cache key must be a tuple not %r" % (type(key),)
             )
+        self.sequence += 1
         self.cache.del_multi(key)
 
-        # if we have a pending lookup for this key, remove it from the
-        # _pending_deferred_cache, as above
         entry_dict = self._pending_deferred_cache.pop(key, None)
         if entry_dict is not None:
             for entry in iterate_tree_cache_entry(entry_dict):
@@ -204,10 +194,8 @@ class Cache(object):
 
     def invalidate_all(self):
         self.check_thread()
+        self.sequence += 1
         self.cache.clear()
-        for entry in self._pending_deferred_cache.itervalues():
-            entry.invalidate()
-        self._pending_deferred_cache.clear()
 
 
 class _CacheDescriptorBase(object):
@@ -310,7 +298,7 @@ class CacheDescriptor(_CacheDescriptorBase):
             orig, num_args=num_args, inlineCallbacks=inlineCallbacks,
             cache_context=cache_context)
 
-        max_entries = int(max_entries * get_cache_factor_for(orig.__name__))
+        max_entries = int(max_entries * CACHE_SIZE_FACTOR)
 
         self.max_entries = max_entries
         self.tree = tree

synapse/util/caches/response_cache.py

@@ -14,7 +14,6 @@
 # limitations under the License.
 
 from synapse.util.async import ObservableDeferred
-from synapse.util.caches import metrics as cache_metrics
 
 
 class ResponseCache(object):
@@ -25,63 +24,20 @@ class ResponseCache(object):
     used rather than trying to compute a new response.
     """
 
-    def __init__(self, hs, name, timeout_ms=0):
+    def __init__(self, hs, timeout_ms=0):
         self.pending_result_cache = {}  # Requests that haven't finished yet.
 
         self.clock = hs.get_clock()
         self.timeout_sec = timeout_ms / 1000.
 
-        self._metrics = cache_metrics.register_cache(
-            "response_cache",
-            size_callback=lambda: self.size(),
-            cache_name=name,
-        )
-
-    def size(self):
-        return len(self.pending_result_cache)
-
     def get(self, key):
-        """Look up the given key.
-
-        Returns a deferred which doesn't follow the synapse logcontext rules,
-        so you'll probably want to make_deferred_yieldable it.
-
-        Args:
-            key (str):
-
-        Returns:
-            twisted.internet.defer.Deferred|None: None if there is no entry
-            for this key; otherwise a deferred result.
-        """
         result = self.pending_result_cache.get(key)
         if result is not None:
-            self._metrics.inc_hits()
             return result.observe()
         else:
-            self._metrics.inc_misses()
             return None
 
     def set(self, key, deferred):
-        """Set the entry for the given key to the given deferred.
-
-        *deferred* should run its callbacks in the sentinel logcontext (ie,
-        you should wrap normal synapse deferreds with
-        logcontext.run_in_background).
-
-        Returns a new Deferred which also doesn't follow the synapse logcontext
-        rules, so you will want to make_deferred_yieldable it
-
-        (TODO: before using this more widely, it might make sense to refactor
-        it and get() so that they do the necessary wrapping rather than having
-        to do it everywhere ResponseCache is used.)
-
-        Args:
-            key (str):
-            deferred (twisted.internet.defer.Deferred):
-
-        Returns:
-            twisted.internet.defer.Deferred
-        """
         result = ObservableDeferred(deferred, consumeErrors=True)
         self.pending_result_cache[key] = result
 

tests/appservice/test_appservice.py

@@ -55,7 +55,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
             _regex("@irc_.*")
         )
         self.event.sender = "@irc_foobar:matrix.org"
-        self.assertTrue((yield self.service.is_interested(self.event)))
+        self.assertTrue((yield self.service.is_interested(self.event, None)))
 
     @defer.inlineCallbacks
     def test_regex_user_id_prefix_no_match(self):
@@ -63,7 +63,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
             _regex("@irc_.*")
         )
         self.event.sender = "@someone_else:matrix.org"
-        self.assertFalse((yield self.service.is_interested(self.event)))
+        self.assertFalse((yield self.service.is_interested(self.event, None)))
 
     @defer.inlineCallbacks
     def test_regex_room_member_is_checked(self):
@@ -73,7 +73,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
         self.event.sender = "@someone_else:matrix.org"
         self.event.type = "m.room.member"
         self.event.state_key = "@irc_foobar:matrix.org"
-        self.assertTrue((yield self.service.is_interested(self.event)))
+        self.assertTrue((yield self.service.is_interested(self.event, None)))
 
     @defer.inlineCallbacks
     def test_regex_room_id_match(self):
@@ -81,7 +81,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
             _regex("!some_prefix.*some_suffix:matrix.org")
         )
         self.event.room_id = "!some_prefixs0m3th1nGsome_suffix:matrix.org"
-        self.assertTrue((yield self.service.is_interested(self.event)))
+        self.assertTrue((yield self.service.is_interested(self.event, None)))
 
     @defer.inlineCallbacks
     def test_regex_room_id_no_match(self):
@@ -89,7 +89,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
             _regex("!some_prefix.*some_suffix:matrix.org")
         )
         self.event.room_id = "!XqBunHwQIXUiqCaoxq:matrix.org"
-        self.assertFalse((yield self.service.is_interested(self.event)))
+        self.assertFalse((yield self.service.is_interested(self.event, None)))
 
     @defer.inlineCallbacks
     def test_regex_alias_match(self):
@@ -160,7 +160,7 @@ class ApplicationServiceTestCase(unittest.TestCase):
         self.store.get_aliases_for_room.return_value = [
             "#xmpp_foobar:matrix.org", "#athing:matrix.org"
         ]
-        self.store.get_users_in_room.return_value = []
+        self.store.get_appservices_with_user_in_room.return_value = []
         self.assertFalse((yield self.service.is_interested(
             self.event, self.store
         )))
@@ -193,20 +193,3 @@ class ApplicationServiceTestCase(unittest.TestCase):
         }
         self.event.state_key = self.service.sender
         self.assertTrue((yield self.service.is_interested(self.event)))
-
-    @defer.inlineCallbacks
-    def test_member_list_match(self):
-        self.service.namespaces[ApplicationService.NS_USERS].append(
-            _regex("@irc_.*")
-        )
-        self.store.get_users_in_room.return_value = [
-            "@alice:here",
-            "@irc_fo:here",  # AS user
-            "@bob:here",
-        ]
-        self.store.get_aliases_for_room.return_value = []
-
-        self.event.sender = "@xmpp_foobar:matrix.org"
-        self.assertTrue((yield self.service.is_interested(
-            event=self.event, store=self.store
-        )))

tests/handlers/test_appservice.py

@@ -31,7 +31,6 @@ class AppServiceHandlerTestCase(unittest.TestCase):
         self.mock_scheduler = Mock()
         hs = Mock()
         hs.get_datastore = Mock(return_value=self.mock_store)
-        self.mock_store.get_received_ts.return_value = 0
         hs.get_application_service_api = Mock(return_value=self.mock_as_api)
         hs.get_application_service_scheduler = Mock(return_value=self.mock_scheduler)
         hs.get_clock.return_value = MockClock()

tests/rest/client/v1/test_events.py

@@ -123,7 +123,6 @@ class EventStreamPermissionsTestCase(RestTestCase):
         self.ratelimiter.send_message.return_value = (True, 0)
         hs.config.enable_registration_captcha = False
         hs.config.enable_registration = True
-        hs.config.auto_join_rooms = []
 
         hs.get_handlers().federation_handler = Mock()
 

tests/rest/client/v1/test_rooms.py

@@ -984,13 +984,11 @@ class RoomInitialSyncTestCase(RestTestCase):
 
         self.assertTrue("presence" in response)
 
-        # presence is turned off on hotfixes
-
-        # presence_by_user = {
-        #     e["content"]["user_id"]: e for e in response["presence"]
-        # }
-        # self.assertTrue(self.user_id in presence_by_user)
-        # self.assertEquals("m.presence", presence_by_user[self.user_id]["type"])
+        presence_by_user = {
+            e["content"]["user_id"]: e for e in response["presence"]
+        }
+        self.assertTrue(self.user_id in presence_by_user)
+        self.assertEquals("m.presence", presence_by_user[self.user_id]["type"])
 
 
 class RoomMessageListTestCase(RestTestCase):

tests/test_dns.py

@@ -33,6 +33,8 @@ class DnsTestCase(unittest.TestCase):
 
         service_name = "test_service.example.com"
         host_name = "example.com"
+        ip_address = "127.0.0.1"
+        ip6_address = "::1"
 
         answer_srv = dns.RRHeader(
             type=dns.SRV,
@@ -41,9 +43,29 @@ class DnsTestCase(unittest.TestCase):
             )
         )
 
+        answer_a = dns.RRHeader(
+            type=dns.A,
+            payload=dns.Record_A(
+                address=ip_address,
+            )
+        )
+
+        answer_aaaa = dns.RRHeader(
+            type=dns.AAAA,
+            payload=dns.Record_AAAA(
+                address=ip6_address,
+            )
+        )
+
         dns_client_mock.lookupService.return_value = defer.succeed(
             ([answer_srv], None, None),
         )
+        dns_client_mock.lookupAddress.return_value = defer.succeed(
+            ([answer_a], None, None),
+        )
+        dns_client_mock.lookupIPV6Address.return_value = defer.succeed(
+            ([answer_aaaa], None, None),
+        )
 
         cache = {}
 
@@ -52,10 +74,13 @@ class DnsTestCase(unittest.TestCase):
         )
 
         dns_client_mock.lookupService.assert_called_once_with(service_name)
+        dns_client_mock.lookupAddress.assert_called_once_with(host_name)
+        dns_client_mock.lookupIPV6Address.assert_called_once_with(host_name)
 
-        self.assertEquals(len(servers), 1)
+        self.assertEquals(len(servers), 2)
         self.assertEquals(servers, cache[service_name])
-        self.assertEquals(servers[0].host, host_name)
+        self.assertEquals(servers[0].host, ip_address)
+        self.assertEquals(servers[1].host, ip6_address)
 
     @defer.inlineCallbacks
     def test_from_cache_expired_and_dns_fail(self):

tests/util/caches/test_descriptors.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2016 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -13,7 +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.
-from functools import partial
 import logging
 
 import mock
@@ -27,50 +25,6 @@ from tests import unittest
 logger = logging.getLogger(__name__)
 
 
-class CacheTestCase(unittest.TestCase):
-    def test_invalidate_all(self):
-        cache = descriptors.Cache("testcache")
-
-        callback_record = [False, False]
-
-        def record_callback(idx):
-            callback_record[idx] = True
-
-        # add a couple of pending entries
-        d1 = defer.Deferred()
-        cache.set("key1", d1, partial(record_callback, 0))
-
-        d2 = defer.Deferred()
-        cache.set("key2", d2, partial(record_callback, 1))
-
-        # lookup should return the deferreds
-        self.assertIs(cache.get("key1"), d1)
-        self.assertIs(cache.get("key2"), d2)
-
-        # let one of the lookups complete
-        d2.callback("result2")
-        self.assertEqual(cache.get("key2"), "result2")
-
-        # now do the invalidation
-        cache.invalidate_all()
-
-        # lookup should return none
-        self.assertIsNone(cache.get("key1", None))
-        self.assertIsNone(cache.get("key2", None))
-
-        # both callbacks should have been callbacked
-        self.assertTrue(
-            callback_record[0], "Invalidation callback for key1 not called",
-        )
-        self.assertTrue(
-            callback_record[1], "Invalidation callback for key2 not called",
-        )
-
-        # letting the other lookup complete should do nothing
-        d1.callback("result1")
-        self.assertIsNone(cache.get("key1", None))
-
-
 class DescriptorTestCase(unittest.TestCase):
     @defer.inlineCallbacks
     def test_cache(self):

tests/utils.py

@@ -212,7 +212,7 @@ class MockHttpResource(HttpServer):
 
         headers = {}
         if federation_auth:
-            headers[b"Authorization"] = ["X-Matrix origin=test,key=,sig="]
+            headers["Authorization"] = ["X-Matrix origin=test,key=,sig="]
         mock_request.requestHeaders.getRawHeaders = mock_getRawHeaders(headers)
 
         # return the right path if the event requires it