From f346048a6e9ac798b742d939a38e0cfe71475f38 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 10:34:10 +0100
Subject: [PATCH 01/45] Handle exceptions thrown in handling remote device list
 updates

---
 synapse/handlers/device.py | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index c22f65ce5..72915b85d 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -17,6 +17,7 @@ from synapse.api.constants import EventTypes
 from synapse.util import stringutils
 from synapse.util.async import Linearizer
 from synapse.util.caches.expiringcache import ExpiringCache
+from synapse.util.retryutils import NotRetryingDestination
 from synapse.util.metrics import measure_func
 from synapse.types import get_domain_from_id, RoomStreamToken
 from twisted.internet import defer
@@ -430,7 +431,21 @@ class DeviceListEduUpdater(object):
             if resync:
                 # Fetch all devices for the user.
                 origin = get_domain_from_id(user_id)
-                result = yield self.federation.query_user_devices(origin, user_id)
+                try:
+                    result = yield self.federation.query_user_devices(origin, user_id)
+                except NotRetryingDestination:
+                    logger.warn(
+                        "Failed to handle device list update for %s,"
+                        " we're not retrying the remote",
+                        user_id,
+                    )
+                    return
+                except Exception:
+                    logger.exception(
+                        "Failed to handle device list update for %s", user_id
+                    )
+                    return
+
                 stream_id = result["stream_id"]
                 devices = result["devices"]
                 yield self.store.update_remote_device_list_cache(

From db7d0c31272954f01d89b2dbd653d54db0cbb040 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 10:34:53 +0100
Subject: [PATCH 02/45] Always mark remotes as up if we receive a signed
 request from them

---
 synapse/federation/transport/server.py | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index c840da834..828dcd01a 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -79,6 +79,7 @@ class Authenticator(object):
     def __init__(self, hs):
         self.keyring = hs.get_keyring()
         self.server_name = hs.hostname
+        self.store = hs.get_datastore()
 
     # A method just so we can pass 'self' as the authenticator to the Servlets
     @defer.inlineCallbacks
@@ -138,6 +139,12 @@ class Authenticator(object):
         logger.info("Request from %s", origin)
         request.authenticated_entity = origin
 
+        # If we get a valid signed request from the other side, its probably
+        # alive
+        retry_timings = yield self.store.get_destination_retry_timings(origin)
+        if retry_timings and retry_timings["retry_last_ts"]:
+            self.store.set_destination_retry_timings(origin, 0, 0)
+
         defer.returnValue(origin)
 
 

From b843631d7157f0beb64db62a86c691369aa49b14 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 10:59:32 +0100
Subject: [PATCH 03/45] Add comment and TODO

---
 synapse/handlers/device.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index 72915b85d..187af03fb 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -426,6 +426,8 @@ class DeviceListEduUpdater(object):
                 # This can happen since we batch updates
                 return
 
+            # Given a list of updates we check if we need to resync. This
+            # happens if we've missed updates.
             resync = yield self._need_to_do_resync(user_id, pending_updates)
 
             if resync:
@@ -434,6 +436,8 @@ class DeviceListEduUpdater(object):
                 try:
                     result = yield self.federation.query_user_devices(origin, user_id)
                 except NotRetryingDestination:
+                    # TODO: Remember that we are now out of sync and try again
+                    # later
                     logger.warn(
                         "Failed to handle device list update for %s,"
                         " we're not retrying the remote",
@@ -441,6 +445,8 @@ class DeviceListEduUpdater(object):
                     )
                     return
                 except Exception:
+                    # TODO: Remember that we are now out of sync and try again
+                    # later
                     logger.exception(
                         "Failed to handle device list update for %s", user_id
                     )

From 7b222fc56e9854d1f8f084d996d9ca694e91dd6c Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 11:14:09 +0100
Subject: [PATCH 04/45] Remove redundant reset of destination timers

---
 synapse/handlers/federation.py | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 2af9849ed..52d97dfbf 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -380,13 +380,6 @@ class FederationHandler(BaseHandler):
                     affected=event.event_id,
                 )
 
-        # if we're receiving valid events from an origin,
-        # it's probably a good idea to mark it as not in retry-state
-        # for sending (although this is a bit of a leap)
-        retry_timings = yield self.store.get_destination_retry_timings(origin)
-        if retry_timings and retry_timings["retry_last_ts"]:
-            self.store.set_destination_retry_timings(origin, 0, 0)
-
         room = yield self.store.get_room(event.room_id)
 
         if not room:

From 310b1ccdc1e80164811d4b1287c0a504d0a33c77 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 13:41:19 +0100
Subject: [PATCH 05/45] Use preserve_fn and add logs

---
 synapse/federation/transport/server.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 828dcd01a..3d676e7d8 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -24,6 +24,7 @@ from synapse.http.servlet import (
 )
 from synapse.util.ratelimitutils import FederationRateLimiter
 from synapse.util.versionstring import get_version_string
+from synapse.util.logcontext import preserve_fn
 from synapse.types import ThirdPartyInstanceID
 
 import functools
@@ -143,7 +144,8 @@ class Authenticator(object):
         # alive
         retry_timings = yield self.store.get_destination_retry_timings(origin)
         if retry_timings and retry_timings["retry_last_ts"]:
-            self.store.set_destination_retry_timings(origin, 0, 0)
+            logger.info("Marking origin %r as up", origin)
+            preserve_fn(self.store.set_destination_retry_timings)(origin, 0, 0)
 
         defer.returnValue(origin)
 

From 653d90c1a529ff553d506ac806ab0403f34955ad Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 May 2017 14:01:17 +0100
Subject: [PATCH 06/45] Comment

---
 synapse/handlers/device.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index 187af03fb..982cda3ed 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -443,6 +443,12 @@ class DeviceListEduUpdater(object):
                         " we're not retrying the remote",
                         user_id,
                     )
+                    # We abort on exceptions rather than accepting the update
+                    # as otherwise synapse will 'forget' that its device list
+                    # is out of date. If we bail then we will retry the resync
+                    # next time we get a device list update for this user_id.
+                    # This makes it more likely that the device lists will
+                    # eventually become consistent.
                     return
                 except Exception:
                     # TODO: Remember that we are now out of sync and try again

From 78f306a6f70552672f8c70171b2d9d79f20f8f8d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 13:07:41 +0100
Subject: [PATCH 07/45] Revert "Speed up filtering of a single event in push"

This reverts commit 421fdf74609439edaaffce117436e6a6df147841.
---
 synapse/push/bulk_push_rule_evaluator.py | 27 +++++++++++++++++-------
 synapse/storage/account_data.py          | 13 ------------
 synapse/storage/push_rule.py             |  5 ++---
 synapse/visibility.py                    | 19 +++++++++++++++++
 4 files changed, 40 insertions(+), 24 deletions(-)

diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py
index cb13874cc..f943ff640 100644
--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -20,6 +20,7 @@ from twisted.internet import defer
 from .push_rule_evaluator import PushRuleEvaluatorForEvent
 
 from synapse.api.constants import EventTypes
+from synapse.visibility import filter_events_for_clients_context
 
 
 logger = logging.getLogger(__name__)
@@ -66,6 +67,17 @@ class BulkPushRuleEvaluator:
     def action_for_event_by_user(self, event, context):
         actions_by_user = {}
 
+        # None of these users can be peeking since this list of users comes
+        # from the set of users in the room, so we know for sure they're all
+        # actually in the room.
+        user_tuples = [
+            (u, False) for u in self.rules_by_user.keys()
+        ]
+
+        filtered_by_user = yield filter_events_for_clients_context(
+            self.store, user_tuples, [event], {event.event_id: context}
+        )
+
         room_members = yield self.store.get_joined_users_from_context(
             event, context
         )
@@ -75,14 +87,6 @@ class BulkPushRuleEvaluator:
         condition_cache = {}
 
         for uid, rules in self.rules_by_user.items():
-            if event.sender == uid:
-                continue
-
-            if not event.is_state():
-                is_ignored = yield self.store.is_ignored_by(event.sender, uid)
-                if is_ignored:
-                    continue
-
             display_name = None
             profile_info = room_members.get(uid)
             if profile_info:
@@ -94,6 +98,13 @@ class BulkPushRuleEvaluator:
                 if event.type == EventTypes.Member and event.state_key == uid:
                     display_name = event.content.get("displayname", None)
 
+            filtered = filtered_by_user[uid]
+            if len(filtered) == 0:
+                continue
+
+            if filtered[0].sender == uid:
+                continue
+
             for rule in rules:
                 if 'enabled' in rule and not rule['enabled']:
                     continue
diff --git a/synapse/storage/account_data.py b/synapse/storage/account_data.py
index ff14e54c1..aa84ffc2b 100644
--- a/synapse/storage/account_data.py
+++ b/synapse/storage/account_data.py
@@ -308,16 +308,3 @@ class AccountDataStore(SQLBaseStore):
             " WHERE stream_id < ?"
         )
         txn.execute(update_max_id_sql, (next_id, next_id))
-
-    @cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)
-    def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
-        ignored_account_data = yield self.get_global_account_data_by_type_for_user(
-            "m.ignored_user_list", ignorer_user_id,
-            on_invalidate=cache_context.invalidate,
-        )
-        if not ignored_account_data:
-            defer.returnValue(False)
-
-        defer.returnValue(
-            ignored_user_id in ignored_account_data.get("ignored_users", {})
-        )
diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py
index 353a135c4..cbec25596 100644
--- a/synapse/storage/push_rule.py
+++ b/synapse/storage/push_rule.py
@@ -188,7 +188,7 @@ class PushRuleStore(SQLBaseStore):
             user_ids, on_invalidate=cache_context.invalidate,
         )
 
-        rules_by_user = {k: v for k, v in rules_by_user.iteritems() if v is not None}
+        rules_by_user = {k: v for k, v in rules_by_user.items() if v is not None}
 
         defer.returnValue(rules_by_user)
 
@@ -398,8 +398,7 @@ class PushRuleStore(SQLBaseStore):
         with self._push_rules_stream_id_gen.get_next() as ids:
             stream_id, event_stream_ordering = ids
             yield self.runInteraction(
-                "delete_push_rule", delete_push_rule_txn, stream_id,
-                event_stream_ordering,
+                "delete_push_rule", delete_push_rule_txn, stream_id, event_stream_ordering
             )
 
     @defer.inlineCallbacks
diff --git a/synapse/visibility.py b/synapse/visibility.py
index 5590b866e..c4dd9ae2c 100644
--- a/synapse/visibility.py
+++ b/synapse/visibility.py
@@ -188,6 +188,25 @@ def filter_events_for_clients(store, user_tuples, events, event_id_to_state):
     })
 
 
+@defer.inlineCallbacks
+def filter_events_for_clients_context(store, user_tuples, events, event_id_to_context):
+    user_ids = set(u[0] for u in user_tuples)
+    event_id_to_state = {}
+    for event_id, context in event_id_to_context.items():
+        state = yield store.get_events([
+            e_id
+            for key, e_id in context.current_state_ids.iteritems()
+            if key == (EventTypes.RoomHistoryVisibility, "")
+            or (key[0] == EventTypes.Member and key[1] in user_ids)
+        ])
+        event_id_to_state[event_id] = state
+
+    res = yield filter_events_for_clients(
+        store, user_tuples, events, event_id_to_state
+    )
+    defer.returnValue(res)
+
+
 @defer.inlineCallbacks
 def filter_events_for_client(store, user_id, events, is_peeking=False):
     """

From fe7c1b969c5b0f51b6fe86e78e96350224fd0fb1 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 13:07:43 +0100
Subject: [PATCH 08/45] Revert "We don't care about forgotten rooms"

This reverts commit ad8b316939d59230526e60660caf9094cff62c8f.
---
 synapse/storage/push_rule.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py
index cbec25596..5467ba51c 100644
--- a/synapse/storage/push_rule.py
+++ b/synapse/storage/push_rule.py
@@ -184,6 +184,18 @@ class PushRuleStore(SQLBaseStore):
             if uid in local_users_in_room:
                 user_ids.add(uid)
 
+        forgotten = yield self.who_forgot_in_room(
+            event.room_id, on_invalidate=cache_context.invalidate,
+        )
+
+        for row in forgotten:
+            user_id = row["user_id"]
+            event_id = row["event_id"]
+
+            mem_id = current_state_ids.get((EventTypes.Member, user_id), None)
+            if event_id == mem_id:
+                user_ids.discard(user_id)
+
         rules_by_user = yield self.bulk_get_push_rules(
             user_ids, on_invalidate=cache_context.invalidate,
         )

From e0f20e9425f5fa0aecf0b8bf5b58ce72c2363d8b Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 13:07:43 +0100
Subject: [PATCH 09/45] Revert "Remove unused import"

This reverts commit ab37bef83bebd7cdaeb7cfd98553d18883d09103.
---
 synapse/storage/push_rule.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py
index 5467ba51c..0a819d32c 100644
--- a/synapse/storage/push_rule.py
+++ b/synapse/storage/push_rule.py
@@ -16,6 +16,7 @@
 from ._base import SQLBaseStore
 from synapse.util.caches.descriptors import cachedInlineCallbacks, cachedList
 from synapse.push.baserules import list_with_base_rules
+from synapse.api.constants import EventTypes
 from twisted.internet import defer
 
 import logging

From dcabef952c0c75ec756a364fc225a72eea391e1b Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 15:09:19 +0100
Subject: [PATCH 10/45] Increase client_ip cache size

---
 synapse/storage/client_ips.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/synapse/storage/client_ips.py b/synapse/storage/client_ips.py
index b01f0046e..747d2df62 100644
--- a/synapse/storage/client_ips.py
+++ b/synapse/storage/client_ips.py
@@ -33,6 +33,7 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
         self.client_ip_last_seen = Cache(
             name="client_ip_last_seen",
             keylen=4,
+            max_entries=5000,
         )
 
         super(ClientIpStore, self).__init__(hs)

From 738ccf61c01df04e1aef521ea7d1ae2844784214 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 15:32:18 +0100
Subject: [PATCH 11/45] Cache check to see if device exists

---
 synapse/storage/devices.py | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index c8d5f5ba8..fc87c9218 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -18,7 +18,7 @@ import ujson as json
 from twisted.internet import defer
 
 from synapse.api.errors import StoreError
-from ._base import SQLBaseStore
+from ._base import SQLBaseStore, Cache
 from synapse.util.caches.descriptors import cached, cachedList, cachedInlineCallbacks
 
 
@@ -29,6 +29,12 @@ class DeviceStore(SQLBaseStore):
     def __init__(self, hs):
         super(DeviceStore, self).__init__(hs)
 
+        self.device_id_exists_cache = Cache(
+            name="device_id_exists",
+            keylen=2,
+            max_entries=10000,
+        )
+
         self._clock.looping_call(
             self._prune_old_outbound_device_pokes, 60 * 60 * 1000
         )
@@ -54,6 +60,10 @@ class DeviceStore(SQLBaseStore):
             defer.Deferred: boolean whether the device was inserted or an
                 existing device existed with that ID.
         """
+        key = (user_id, device_id)
+        if self.device_id_exists_cache.get(key, None):
+            defer.returnValue(False)
+
         try:
             inserted = yield self._simple_insert(
                 "devices",
@@ -65,6 +75,7 @@ class DeviceStore(SQLBaseStore):
                 desc="store_device",
                 or_ignore=True,
             )
+            self.device_id_exists_cache.prefill(key, True)
             defer.returnValue(inserted)
         except Exception as e:
             logger.error("store_device with device_id=%s(%r) user_id=%s(%r)"

From fc6d4974a60a0d47492f5c5c8dff45abbf9abe03 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 15:33:57 +0100
Subject: [PATCH 12/45] Comment

---
 synapse/storage/devices.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index fc87c9218..6727861eb 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -29,6 +29,8 @@ class DeviceStore(SQLBaseStore):
     def __init__(self, hs):
         super(DeviceStore, self).__init__(hs)
 
+        # Map of (user_id, device_id) -> bool. If there is an entry that implies
+        # the device exists.
         self.device_id_exists_cache = Cache(
             name="device_id_exists",
             keylen=2,

From 8571f864d2fc20986341b7e9d6e18c3e061e48e0 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 15:34:27 +0100
Subject: [PATCH 13/45] Cache one time key counts

---
 synapse/storage/end_to_end_keys.py | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/synapse/storage/end_to_end_keys.py b/synapse/storage/end_to_end_keys.py
index 7cbc1470f..c96dae352 100644
--- a/synapse/storage/end_to_end_keys.py
+++ b/synapse/storage/end_to_end_keys.py
@@ -15,6 +15,7 @@
 from twisted.internet import defer
 
 from synapse.api.errors import SynapseError
+from synapse.util.caches.descriptors import cached
 
 from canonicaljson import encode_canonical_json
 import ujson as json
@@ -177,10 +178,14 @@ class EndToEndKeyStore(SQLBaseStore):
                     for algorithm, key_id, json_bytes in new_keys
                 ],
             )
+            txn.call_after(
+                self.count_e2e_one_time_keys.invalidate, (user_id, device_id,)
+            )
         yield self.runInteraction(
             "add_e2e_one_time_keys_insert", _add_e2e_one_time_keys
         )
 
+    @cached(max_entries=10000)
     def count_e2e_one_time_keys(self, user_id, device_id):
         """ Count the number of one time keys the server has for a device
         Returns:
@@ -225,6 +230,9 @@ class EndToEndKeyStore(SQLBaseStore):
             )
             for user_id, device_id, algorithm, key_id in delete:
                 txn.execute(sql, (user_id, device_id, algorithm, key_id))
+                txn.call_after(
+                    self.count_e2e_one_time_keys.invalidate, (user_id, device_id,)
+                )
             return result
         return self.runInteraction(
             "claim_e2e_one_time_keys", _claim_e2e_one_time_keys
@@ -242,3 +250,4 @@ class EndToEndKeyStore(SQLBaseStore):
             keyvalues={"user_id": user_id, "device_id": device_id},
             desc="delete_e2e_one_time_keys_by_device"
         )
+        self.count_e2e_one_time_keys.invalidate((user_id, device_id,))

From 94e6ad71f5445e014f3c9f6c260ab664635c7b59 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 15:55:59 +0100
Subject: [PATCH 14/45] Invalidate cache on device deletion

---
 synapse/storage/devices.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index 6727861eb..75c30abc2 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -115,12 +115,14 @@ class DeviceStore(SQLBaseStore):
         Returns:
             defer.Deferred
         """
-        return self._simple_delete_one(
+        self._simple_delete_one(
             table="devices",
             keyvalues={"user_id": user_id, "device_id": device_id},
             desc="delete_device",
         )
 
+        self.device_id_exists_cache.invalidate((user_id, device_id))
+
     def delete_devices(self, user_id, device_ids):
         """Deletes several devices.
 
@@ -130,13 +132,15 @@ class DeviceStore(SQLBaseStore):
         Returns:
             defer.Deferred
         """
-        return self._simple_delete_many(
+        self._simple_delete_many(
             table="devices",
             column="device_id",
             iterable=device_ids,
             keyvalues={"user_id": user_id},
             desc="delete_devices",
         )
+        for device_id in device_ids:
+            self.device_id_exists_cache.invalidate((user_id, device_id))
 
     def update_device(self, user_id, device_id, new_display_name=None):
         """Update a device.

From ffad4fe35be3baba5b2fffaa4e9b31f3008d09af Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 16:06:17 +0100
Subject: [PATCH 15/45] Don't update event cache hit ratio from
 get_joined_users

Otherwise the hit ration of plain get_events gets completely skewed by
calls to get_joined_users* functions.
---
 synapse/storage/events.py          | 13 +++++++++++--
 synapse/storage/roommember.py      |  4 ++++
 synapse/util/caches/descriptors.py |  9 ++++++---
 3 files changed, 21 insertions(+), 5 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 98707d40e..d944984d6 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -1343,11 +1343,20 @@ class EventsStore(SQLBaseStore):
     def _invalidate_get_event_cache(self, event_id):
             self._get_event_cache.invalidate((event_id,))
 
-    def _get_events_from_cache(self, events, allow_rejected):
+    def _get_events_from_cache(self, events, allow_rejected, update_metrics=True):
+        """
+        Args:
+            events (list(str)): list of event_ids to fetch
+            allow_rejected (bool): Whether to teturn events that were rejected
+            update_metrics (bool): Whether to update the cache hit ratio metrics
+        """
         event_map = {}
 
         for event_id in events:
-            ret = self._get_event_cache.get((event_id,), None)
+            ret = self._get_event_cache.get(
+                (event_id,), None,
+                update_metrics=update_metrics,
+            )
             if not ret:
                 continue
 
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index ad3c9b06d..2fa20bd87 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -421,9 +421,13 @@ class RoomMemberStore(SQLBaseStore):
         # We check if we have any of the member event ids in the event cache
         # before we ask the DB
 
+        # We don't update the event cache hit ratio as it completely throws off
+        # the hit ratio counts. After all, we don't populate the cache if we
+        # miss it here
         event_map = self._get_events_from_cache(
             member_event_ids,
             allow_rejected=False,
+            update_metrics=False,
         )
 
         missing_member_event_ids = []
diff --git a/synapse/util/caches/descriptors.py b/synapse/util/caches/descriptors.py
index aa182eeac..48dcbafee 100644
--- a/synapse/util/caches/descriptors.py
+++ b/synapse/util/caches/descriptors.py
@@ -96,7 +96,7 @@ class Cache(object):
                     "Cache objects can only be accessed from the main thread"
                 )
 
-    def get(self, key, default=_CacheSentinel, callback=None):
+    def get(self, key, default=_CacheSentinel, callback=None, update_metrics=True):
         """Looks the key up in the caches.
 
         Args:
@@ -104,6 +104,7 @@ class Cache(object):
             default: What is returned if key is not in the caches. If not
                 specified then function throws KeyError instead
             callback(fn): Gets called when the entry in the cache is invalidated
+            update_metrics (bool): whether to update the cache hit rate metrics
 
         Returns:
             Either a Deferred or the raw result
@@ -113,7 +114,8 @@ class Cache(object):
         if val is not _CacheSentinel:
             if val.sequence == self.sequence:
                 val.callbacks.update(callbacks)
-                self.metrics.inc_hits()
+                if update_metrics:
+                    self.metrics.inc_hits()
                 return val.deferred
 
         val = self.cache.get(key, _CacheSentinel, callbacks=callbacks)
@@ -121,7 +123,8 @@ class Cache(object):
             self.metrics.inc_hits()
             return val
 
-        self.metrics.inc_misses()
+        if update_metrics:
+            self.metrics.inc_misses()
 
         if default is _CacheSentinel:
             raise KeyError()

From 6a12998a83791137db0b7988646cfc4bff572427 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 16:10:51 +0100
Subject: [PATCH 16/45] Add missing yields

---
 synapse/storage/devices.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index 75c30abc2..d9936c88b 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -106,6 +106,7 @@ class DeviceStore(SQLBaseStore):
             desc="get_device",
         )
 
+    @defer.inlineCallbacks
     def delete_device(self, user_id, device_id):
         """Delete a device.
 
@@ -115,7 +116,7 @@ class DeviceStore(SQLBaseStore):
         Returns:
             defer.Deferred
         """
-        self._simple_delete_one(
+        yield self._simple_delete_one(
             table="devices",
             keyvalues={"user_id": user_id, "device_id": device_id},
             desc="delete_device",
@@ -123,6 +124,7 @@ class DeviceStore(SQLBaseStore):
 
         self.device_id_exists_cache.invalidate((user_id, device_id))
 
+    @defer.inlineCallbacks
     def delete_devices(self, user_id, device_ids):
         """Deletes several devices.
 
@@ -132,7 +134,7 @@ class DeviceStore(SQLBaseStore):
         Returns:
             defer.Deferred
         """
-        self._simple_delete_many(
+        yield self._simple_delete_many(
             table="devices",
             column="device_id",
             iterable=device_ids,

From 093f7e47ccf318181c262c79bb60ffd3b83edaee Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 May 2017 16:13:51 +0100
Subject: [PATCH 17/45] Expand docstring a bit

---
 synapse/storage/events.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index d944984d6..2ab44ceaa 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -1344,11 +1344,17 @@ class EventsStore(SQLBaseStore):
             self._get_event_cache.invalidate((event_id,))
 
     def _get_events_from_cache(self, events, allow_rejected, update_metrics=True):
-        """
+        """Fetch events from the caches
+
         Args:
             events (list(str)): list of event_ids to fetch
             allow_rejected (bool): Whether to teturn events that were rejected
             update_metrics (bool): Whether to update the cache hit ratio metrics
+
+        Returns:
+            dict of event_id -> _EventCacheEntry for each event_id in cache. If
+            allow_rejected is `False` then there will still be an entry but it
+            will be `None`
         """
         event_map = {}
 

From a7e9d8762ddbcea0fcb7ab87c2c4f4e0d91e639a Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Tue, 9 May 2017 18:26:54 +0100
Subject: [PATCH 18/45] Allow clients to upload one-time-keys with new sigs

When a client retries a key upload, don't give an error if the signature has
changed (but the key is the same).

Fixes https://github.com/vector-im/riot-android/issues/1208, hopefully.
---
 synapse/handlers/e2e_keys.py       | 70 +++++++++++++++++----
 synapse/storage/end_to_end_keys.py | 45 ++++++++------
 tests/handlers/test_e2e_keys.py    | 98 ++++++++++++++++++++++++++++++
 3 files changed, 181 insertions(+), 32 deletions(-)

diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py
index c2b38d72a..9d994a8f7 100644
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -288,19 +288,8 @@ class E2eKeysHandler(object):
 
         one_time_keys = keys.get("one_time_keys", None)
         if one_time_keys:
-            logger.info(
-                "Adding %d one_time_keys for device %r for user %r at %d",
-                len(one_time_keys), device_id, user_id, time_now
-            )
-            key_list = []
-            for key_id, key_json in one_time_keys.items():
-                algorithm, key_id = key_id.split(":")
-                key_list.append((
-                    algorithm, key_id, encode_canonical_json(key_json)
-                ))
-
-            yield self.store.add_e2e_one_time_keys(
-                user_id, device_id, time_now, key_list
+            yield self._upload_one_time_keys_for_user(
+                user_id, device_id, time_now, one_time_keys,
             )
 
         # the device should have been registered already, but it may have been
@@ -313,3 +302,58 @@ class E2eKeysHandler(object):
         result = yield self.store.count_e2e_one_time_keys(user_id, device_id)
 
         defer.returnValue({"one_time_key_counts": result})
+
+    @defer.inlineCallbacks
+    def _upload_one_time_keys_for_user(self, user_id, device_id, time_now,
+                                       one_time_keys):
+        logger.info(
+            "Adding one_time_keys %r for device %r for user %r at %d",
+            one_time_keys.keys(), device_id, user_id, time_now,
+        )
+
+        # make a list of (alg, id, key) tuples
+        key_list = []
+        for key_id, key_obj in one_time_keys.items():
+            algorithm, key_id = key_id.split(":")
+            key_list.append((
+                algorithm, key_id, key_obj
+            ))
+
+        # First we check if we have already persisted any of the keys.
+        existing_key_map = yield self.store.get_e2e_one_time_keys(
+            user_id, device_id, [k_id for _, k_id, _ in key_list]
+        )
+
+        new_keys = []  # Keys that we need to insert. (alg, id, json) tuples.
+        for algorithm, key_id, key in key_list:
+            ex_json = existing_key_map.get((algorithm, key_id), None)
+            if ex_json:
+                if not _one_time_keys_match(ex_json, key):
+                    raise SynapseError(
+                        400,
+                        ("One time key %s:%s already exists. "
+                         "Old key: %s; new key: %r") %
+                        (algorithm, key_id, ex_json, key)
+                    )
+            else:
+                new_keys.append((algorithm, key_id, encode_canonical_json(key)))
+
+        yield self.store.add_e2e_one_time_keys(
+            user_id, device_id, time_now, new_keys
+        )
+
+
+def _one_time_keys_match(old_key_json, new_key):
+    old_key = json.loads(old_key_json)
+
+    # if either is a string rather than an object, they must match exactly
+    if not isinstance(old_key, dict) or not isinstance(new_key, dict):
+        return old_key == new_key
+
+    # otherwise, we strip off the 'signatures' if any, because it's legitimate
+    # for different upload attempts to have different signatures.
+    old_key.pop("signatures", None)
+    new_key_copy = dict(new_key)
+    new_key_copy.pop("signatures", None)
+
+    return old_key == new_key_copy
diff --git a/synapse/storage/end_to_end_keys.py b/synapse/storage/end_to_end_keys.py
index c96dae352..e00f31da2 100644
--- a/synapse/storage/end_to_end_keys.py
+++ b/synapse/storage/end_to_end_keys.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 from twisted.internet import defer
 
-from synapse.api.errors import SynapseError
 from synapse.util.caches.descriptors import cached
 
 from canonicaljson import encode_canonical_json
@@ -124,18 +123,24 @@ class EndToEndKeyStore(SQLBaseStore):
         return result
 
     @defer.inlineCallbacks
-    def add_e2e_one_time_keys(self, user_id, device_id, time_now, key_list):
-        """Insert some new one time keys for a device.
+    def get_e2e_one_time_keys(self, user_id, device_id, key_ids):
+        """Retrieve a number of one-time keys for a user
 
-        Checks if any of the keys are already inserted, if they are then check
-        if they match. If they don't then we raise an error.
+        Args:
+            user_id(str): id of user to get keys for
+            device_id(str): id of device to get keys for
+            key_ids(list[str]): list of key ids (excluding algorithm) to
+                retrieve
+
+        Returns:
+            deferred resolving to Dict[(str, str), str]: map from (algorithm,
+            key_id) to json string for key
         """
 
-        # First we check if we have already persisted any of the keys.
         rows = yield self._simple_select_many_batch(
             table="e2e_one_time_keys_json",
             column="key_id",
-            iterable=[key_id for _, key_id, _ in key_list],
+            iterable=key_ids,
             retcols=("algorithm", "key_id", "key_json",),
             keyvalues={
                 "user_id": user_id,
@@ -144,20 +149,22 @@ class EndToEndKeyStore(SQLBaseStore):
             desc="add_e2e_one_time_keys_check",
         )
 
-        existing_key_map = {
+        defer.returnValue({
             (row["algorithm"], row["key_id"]): row["key_json"] for row in rows
-        }
+        })
 
-        new_keys = []  # Keys that we need to insert
-        for algorithm, key_id, json_bytes in key_list:
-            ex_bytes = existing_key_map.get((algorithm, key_id), None)
-            if ex_bytes:
-                if json_bytes != ex_bytes:
-                    raise SynapseError(
-                        400, "One time key with key_id %r already exists" % (key_id,)
-                    )
-            else:
-                new_keys.append((algorithm, key_id, json_bytes))
+    @defer.inlineCallbacks
+    def add_e2e_one_time_keys(self, user_id, device_id, time_now, new_keys):
+        """Insert some new one time keys for a device. Errors if any of the
+        keys already exist.
+
+        Args:
+            user_id(str): id of user to get keys for
+            device_id(str): id of device to get keys for
+            time_now(long): insertion time to record (ms since epoch)
+            new_keys(iterable[(str, str, str)]: keys to add - each a tuple of
+                (algorithm, key_id, key json)
+        """
 
         def _add_e2e_one_time_keys(txn):
             # We are protected from race between lookup and insertion due to
diff --git a/tests/handlers/test_e2e_keys.py b/tests/handlers/test_e2e_keys.py
index 878a54dc3..f10a80a8e 100644
--- a/tests/handlers/test_e2e_keys.py
+++ b/tests/handlers/test_e2e_keys.py
@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import mock
+from synapse.api import errors
 from twisted.internet import defer
 
 import synapse.api.errors
@@ -44,3 +45,100 @@ class E2eKeysHandlerTestCase(unittest.TestCase):
         local_user = "@boris:" + self.hs.hostname
         res = yield self.handler.query_local_devices({local_user: None})
         self.assertDictEqual(res, {local_user: {}})
+
+    @defer.inlineCallbacks
+    def test_reupload_one_time_keys(self):
+        """we should be able to re-upload the same keys"""
+        local_user = "@boris:" + self.hs.hostname
+        device_id = "xyz"
+        keys = {
+            "alg1:k1": "key1",
+            "alg2:k2": {
+                "key": "key2",
+                "signatures": {"k1": "sig1"}
+            },
+            "alg2:k3": {
+                "key": "key3",
+            },
+        }
+
+        res = yield self.handler.upload_keys_for_user(
+            local_user, device_id, {"one_time_keys": keys},
+        )
+        self.assertDictEqual(res, {
+            "one_time_key_counts": {"alg1": 1, "alg2": 2}
+        })
+
+        # we should be able to change the signature without a problem
+        keys["alg2:k2"]["signatures"]["k1"] = "sig2"
+        res = yield self.handler.upload_keys_for_user(
+            local_user, device_id, {"one_time_keys": keys},
+        )
+        self.assertDictEqual(res, {
+            "one_time_key_counts": {"alg1": 1, "alg2": 2}
+        })
+
+    @defer.inlineCallbacks
+    def test_change_one_time_keys(self):
+        """attempts to change one-time-keys should be rejected"""
+
+        local_user = "@boris:" + self.hs.hostname
+        device_id = "xyz"
+        keys = {
+            "alg1:k1": "key1",
+            "alg2:k2": {
+                "key": "key2",
+                "signatures": {"k1": "sig1"}
+            },
+            "alg2:k3": {
+                "key": "key3",
+            },
+        }
+
+        res = yield self.handler.upload_keys_for_user(
+            local_user, device_id, {"one_time_keys": keys},
+        )
+        self.assertDictEqual(res, {
+            "one_time_key_counts": {"alg1": 1, "alg2": 2}
+        })
+
+        try:
+            yield self.handler.upload_keys_for_user(
+                local_user, device_id, {"one_time_keys": {"alg1:k1": "key2"}},
+            )
+            self.fail("No error when changing string key")
+        except errors.SynapseError:
+            pass
+
+        try:
+            yield self.handler.upload_keys_for_user(
+                local_user, device_id, {"one_time_keys": {"alg2:k3": "key2"}},
+            )
+            self.fail("No error when replacing dict key with string")
+        except errors.SynapseError:
+            pass
+
+        try:
+            yield self.handler.upload_keys_for_user(
+                local_user, device_id, {
+                    "one_time_keys": {"alg1:k1": {"key": "key"}}
+                },
+            )
+            self.fail("No error when replacing string key with dict")
+        except errors.SynapseError:
+            pass
+
+        try:
+            yield self.handler.upload_keys_for_user(
+                local_user, device_id, {
+                    "one_time_keys": {
+                        "alg2:k2": {
+                            "key": "key3",
+                            "signatures": {"k1": "sig1"},
+                        }
+                    },
+                },
+            )
+            self.fail("No error when replacing dict key")
+        except errors.SynapseError:
+            pass

From de042b3b885aba6b1508ca50e033fb7a95893553 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Tue, 9 May 2017 19:01:39 +0100
Subject: [PATCH 19/45] Do some logging when one-time-keys get claimed

might help us figure out if https://github.com/vector-im/riot-web/issues/3868
has happened.
---
 synapse/federation/federation_server.py | 10 ++++++++
 synapse/handlers/e2e_keys.py            | 10 ++++++++
 tests/handlers/test_e2e_keys.py         | 34 +++++++++++++++++++++++++
 3 files changed, 54 insertions(+)

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index bc20b9c20..51e3fdea0 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -440,6 +440,16 @@ class FederationServer(FederationBase):
                         key_id: json.loads(json_bytes)
                     }
 
+        logger.info(
+            "Claimed one-time-keys: %s",
+            ",".join((
+                "%s for %s:%s" % (key_id, user_id, device_id)
+                for user_id, user_keys in json_result.iteritems()
+                for device_id, device_keys in user_keys.iteritems()
+                for key_id, _ in device_keys.iteritems()
+            )),
+        )
+
         defer.returnValue({"one_time_keys": json_result})
 
     @defer.inlineCallbacks
diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py
index 9d994a8f7..73921a530 100644
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -262,6 +262,16 @@ class E2eKeysHandler(object):
             for destination in remote_queries
         ]))
 
+        logger.info(
+            "Claimed one-time-keys: %s",
+            ",".join((
+                "%s for %s:%s" % (key_id, user_id, device_id)
+                for user_id, user_keys in json_result.iteritems()
+                for device_id, device_keys in user_keys.iteritems()
+                for key_id, _ in device_keys.iteritems()
+            )),
+        )
+
         defer.returnValue({
             "one_time_keys": json_result,
             "failures": failures
diff --git a/tests/handlers/test_e2e_keys.py b/tests/handlers/test_e2e_keys.py
index f10a80a8e..19f5ed6bc 100644
--- a/tests/handlers/test_e2e_keys.py
+++ b/tests/handlers/test_e2e_keys.py
@@ -142,3 +142,37 @@ class E2eKeysHandlerTestCase(unittest.TestCase):
             self.fail("No error when replacing dict key")
         except errors.SynapseError:
             pass
+
+    @unittest.DEBUG
+    @defer.inlineCallbacks
+    def test_claim_one_time_key(self):
+        local_user = "@boris:" + self.hs.hostname
+        device_id = "xyz"
+        keys = {
+            "alg1:k1": "key1",
+        }
+
+        res = yield self.handler.upload_keys_for_user(
+            local_user, device_id, {"one_time_keys": keys},
+        )
+        self.assertDictEqual(res, {
+            "one_time_key_counts": {"alg1": 1}
+        })
+
+        res2 = yield self.handler.claim_one_time_keys({
+            "one_time_keys": {
+                local_user: {
+                    device_id: "alg1"
+                }
+            }
+        }, timeout=None)
+        self.assertEqual(res2, {
+            "failures": {},
+            "one_time_keys": {
+                local_user: {
+                    device_id: {
+                        "alg1:k1": "key1"
+                    }
+                }
+            }
+        })

From aedaba018f97202ee69b8715b34d6fc1bac39566 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Tue, 9 May 2017 19:02:32 +0100
Subject: [PATCH 20/45] Replace some instances of
 preserve_context_over_deferred

---
 synapse/handlers/e2e_keys.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py
index 73921a530..668a90e49 100644
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -21,7 +21,7 @@ from twisted.internet import defer
 
 from synapse.api.errors import SynapseError, CodeMessageException
 from synapse.types import get_domain_from_id
-from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
+from synapse.util.logcontext import preserve_fn, make_deferred_yieldable
 from synapse.util.retryutils import NotRetryingDestination
 
 logger = logging.getLogger(__name__)
@@ -145,7 +145,7 @@ class E2eKeysHandler(object):
                     "status": 503, "message": e.message
                 }
 
-        yield preserve_context_over_deferred(defer.gatherResults([
+        yield make_deferred_yieldable(defer.gatherResults([
             preserve_fn(do_remote_query)(destination)
             for destination in remote_queries_not_in_cache
         ]))
@@ -257,7 +257,7 @@ class E2eKeysHandler(object):
                     "status": 503, "message": e.message
                 }
 
-        yield preserve_context_over_deferred(defer.gatherResults([
+        yield make_deferred_yieldable(defer.gatherResults([
             preserve_fn(claim_client_keys)(destination)
             for destination in remote_queries
         ]))

From b990b2fce5bb96c1c8eebcb0525ffcd011a22556 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 10 May 2017 11:05:43 +0100
Subject: [PATCH 21/45] Add per user ratelimiting overrides

---
 synapse/handlers/_base.py                     | 34 ++++++++++++++++--
 synapse/handlers/message.py                   | 16 ++-------
 synapse/handlers/profile.py                   |  2 +-
 synapse/handlers/room.py                      |  2 +-
 synapse/storage/room.py                       | 36 ++++++++++++++++++-
 synapse/storage/schema/delta/41/ratelimit.sql | 22 ++++++++++++
 6 files changed, 93 insertions(+), 19 deletions(-)
 create mode 100644 synapse/storage/schema/delta/41/ratelimit.sql

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index e83adc833..faa5609c0 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -53,7 +53,20 @@ class BaseHandler(object):
 
         self.event_builder_factory = hs.get_event_builder_factory()
 
-    def ratelimit(self, requester):
+    @defer.inlineCallbacks
+    def ratelimit(self, requester, update=True):
+        """Ratelimits requests.
+
+        Args:
+            requester (Requester)
+            update (bool): Whether to record that a request is being processed.
+                Set to False when doing multiple checks for one request (e.g.
+                to check up front if we would reject the request), and set to
+                True for the last call for a given request.
+
+        Raises:
+            LimitExceededError if the request should be ratelimited
+        """
         time_now = self.clock.time()
         user_id = requester.user.to_string()
 
@@ -67,10 +80,25 @@ class BaseHandler(object):
         if requester.app_service and not requester.app_service.is_rate_limited():
             return
 
+        # Check if there is a per user override in the DB.
+        override = yield self.store.get_ratelimit_for_user(user_id)
+        if override:
+            # If overriden with a null Hz then ratelimiting has been entirely
+            # disabled for the user
+            if not override.messages_per_second:
+                return
+
+            messages_per_second = override.messages_per_second
+            burst_count = override.burst_count
+        else:
+            messages_per_second = self.hs.config.rc_messages_per_second
+            burst_count = self.hs.config.rc_message_burst_count
+
         allowed, time_allowed = self.ratelimiter.send_message(
             user_id, time_now,
-            msg_rate_hz=self.hs.config.rc_messages_per_second,
-            burst_count=self.hs.config.rc_message_burst_count,
+            msg_rate_hz=messages_per_second,
+            burst_count=burst_count,
+            update=update,
         )
         if not allowed:
             raise LimitExceededError(
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 57265c6d7..196925eda 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -16,7 +16,7 @@
 from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
-from synapse.api.errors import AuthError, Codes, SynapseError, LimitExceededError
+from synapse.api.errors import AuthError, Codes, SynapseError
 from synapse.crypto.event_signing import add_hashes_and_signatures
 from synapse.events.utils import serialize_event
 from synapse.events.validator import EventValidator
@@ -254,17 +254,7 @@ class MessageHandler(BaseHandler):
         # We check here if we are currently being rate limited, so that we
         # don't do unnecessary work. We check again just before we actually
         # send the event.
-        time_now = self.clock.time()
-        allowed, time_allowed = self.ratelimiter.send_message(
-            event.sender, time_now,
-            msg_rate_hz=self.hs.config.rc_messages_per_second,
-            burst_count=self.hs.config.rc_message_burst_count,
-            update=False,
-        )
-        if not allowed:
-            raise LimitExceededError(
-                retry_after_ms=int(1000 * (time_allowed - time_now)),
-            )
+        yield self.ratelimit(requester, update=False)
 
         user = UserID.from_string(event.sender)
 
@@ -499,7 +489,7 @@ class MessageHandler(BaseHandler):
         # We now need to go and hit out to wherever we need to hit out to.
 
         if ratelimit:
-            self.ratelimit(requester)
+            yield self.ratelimit(requester)
 
         try:
             yield self.auth.check_from_context(event, context)
diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py
index 9bf638f81..7abee98de 100644
--- a/synapse/handlers/profile.py
+++ b/synapse/handlers/profile.py
@@ -156,7 +156,7 @@ class ProfileHandler(BaseHandler):
         if not self.hs.is_mine(user):
             return
 
-        self.ratelimit(requester)
+        yield self.ratelimit(requester)
 
         room_ids = yield self.store.get_rooms_for_user(
             user.to_string(),
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 99cb7db0d..d2a0d6520 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -75,7 +75,7 @@ class RoomCreationHandler(BaseHandler):
         """
         user_id = requester.user.to_string()
 
-        self.ratelimit(requester)
+        yield self.ratelimit(requester)
 
         if "room_alias_name" in config:
             for wchar in string.whitespace:
diff --git a/synapse/storage/room.py b/synapse/storage/room.py
index e4c56cc17..5d543652b 100644
--- a/synapse/storage/room.py
+++ b/synapse/storage/room.py
@@ -16,7 +16,7 @@
 from twisted.internet import defer
 
 from synapse.api.errors import StoreError
-from synapse.util.caches.descriptors import cached
+from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
 
 from ._base import SQLBaseStore
 from .engines import PostgresEngine, Sqlite3Engine
@@ -33,6 +33,11 @@ OpsLevel = collections.namedtuple(
     ("ban_level", "kick_level", "redact_level",)
 )
 
+RatelimitOverride = collections.namedtuple(
+    "RatelimitOverride",
+    ("messages_per_second", "burst_count",)
+)
+
 
 class RoomStore(SQLBaseStore):
 
@@ -473,3 +478,32 @@ class RoomStore(SQLBaseStore):
         return self.runInteraction(
             "get_all_new_public_rooms", get_all_new_public_rooms
         )
+
+    @cachedInlineCallbacks(max_entries=10000)
+    def get_ratelimit_for_user(self, user_id):
+        """Check if there are any overrides for ratelimiting for the given
+        user
+
+        Args:
+            user_id (str)
+
+        Returns:
+            RatelimitOverride if there is an override, else None. If the contents
+            of RatelimitOverride are None or 0 then ratelimitng has been
+            disabled for that user entirely.
+        """
+        row = yield self._simple_select_one(
+            table="ratelimit_override",
+            keyvalues={"user_id": user_id},
+            retcols=("messages_per_second", "burst_count"),
+            allow_none=True,
+            desc="get_ratelimit_for_user",
+        )
+
+        if row:
+            defer.returnValue(RatelimitOverride(
+                messages_per_second=row["messages_per_second"],
+                burst_count=row["burst_count"],
+            ))
+        else:
+            defer.returnValue(None)
diff --git a/synapse/storage/schema/delta/41/ratelimit.sql b/synapse/storage/schema/delta/41/ratelimit.sql
new file mode 100644
index 000000000..a194bf023
--- /dev/null
+++ b/synapse/storage/schema/delta/41/ratelimit.sql
@@ -0,0 +1,22 @@
+/* Copyright 2017 Vector Creations 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.
+ */
+
+CREATE TABLE ratelimit_override (
+    user_id TEXT NOT NULL,
+    messages_per_second BIGINT,
+    burst_count BIGINT
+);
+
+CREATE UNIQUE INDEX ratelimit_override_idx ON ratelimit_override(user_id);

From f7278e612e0238d918c2a0d71ca3fb8fb499c765 Mon Sep 17 00:00:00 2001
From: Luke Barnard <lukeb@openmarket.com>
Date: Wed, 10 May 2017 11:40:18 +0100
Subject: [PATCH 22/45] Change register/available to POST (from GET)

---
 synapse/rest/client/v2_alpha/register.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/synapse/rest/client/v2_alpha/register.py b/synapse/rest/client/v2_alpha/register.py
index 38a739f2f..6a7cd96ea 100644
--- a/synapse/rest/client/v2_alpha/register.py
+++ b/synapse/rest/client/v2_alpha/register.py
@@ -142,7 +142,7 @@ class UsernameAvailabilityRestServlet(RestServlet):
         )
 
     @defer.inlineCallbacks
-    def on_GET(self, request):
+    def on_POST(self, request):
         ip = self.hs.get_ip_from_request(request)
         with self.ratelimiter.ratelimit(ip) as wait_deferred:
             yield wait_deferred

From 57ed7f6772870b4afc833a5231894cd287def9dd Mon Sep 17 00:00:00 2001
From: hamber-dick <stevensander@web.de>
Date: Wed, 10 May 2017 18:01:39 +0200
Subject: [PATCH 23/45] Documantation to chek synapse version

I've added some Documentation, how to get the running Version of a
Synapse homeserver. This should help the HS-Owners to check whether the
Upgrade was successful.
---
 UPGRADE.rst | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/UPGRADE.rst b/UPGRADE.rst
index 9f044719a..6164df883 100644
--- a/UPGRADE.rst
+++ b/UPGRADE.rst
@@ -28,6 +28,15 @@ running:
     git pull
     # Update the versions of synapse's python dependencies.
     python synapse/python_dependencies.py | xargs -n1 pip install --upgrade
+	
+To check whether your update was sucessfull, run:
+
+.. code:: bash
+
+	 # replace your.server.domain with ther domain of your synaspe homeserver
+	 curl https://<your.server.domain>/_matrix/federation/v1/version 
+
+So for the Matrix.org HS server the URL would be: https://matrix.org/_matrix/federation/v1/version.
 
 
 Upgrading to v0.15.0

From 369195caa5f52fc67c2496507fc99fccf4b0ede8 Mon Sep 17 00:00:00 2001
From: Luke Barnard <lukeb@openmarket.com>
Date: Wed, 10 May 2017 17:17:12 +0100
Subject: [PATCH 24/45] Modify register/available to be GET with query param

- GET is now the method for register/available
- a query parameter "username" is now used

Also, empty usernames are now handled with an error message on registration or via register/available: `User ID cannot be empty`
---
 synapse/handlers/register.py             | 7 +++++++
 synapse/rest/client/v2_alpha/register.py | 9 ++++-----
 2 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index 03c6a85fc..dd84c5f5e 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -54,6 +54,13 @@ class RegistrationHandler(BaseHandler):
                 Codes.INVALID_USERNAME
             )
 
+        if len(localpart) == 0:
+            raise SynapseError(
+                400,
+                "User ID cannot be empty",
+                Codes.INVALID_USERNAME
+            )
+
         if localpart[0] == '_':
             raise SynapseError(
                 400,
diff --git a/synapse/rest/client/v2_alpha/register.py b/synapse/rest/client/v2_alpha/register.py
index 6a7cd96ea..1421c1815 100644
--- a/synapse/rest/client/v2_alpha/register.py
+++ b/synapse/rest/client/v2_alpha/register.py
@@ -21,7 +21,7 @@ from synapse.api.auth import get_access_token_from_request, has_access_token
 from synapse.api.constants import LoginType
 from synapse.api.errors import SynapseError, Codes, UnrecognizedRequestError
 from synapse.http.servlet import (
-    RestServlet, parse_json_object_from_request, assert_params_in_request
+    RestServlet, parse_json_object_from_request, assert_params_in_request, parse_string
 )
 from synapse.util.msisdn import phone_number_to_msisdn
 
@@ -142,15 +142,14 @@ class UsernameAvailabilityRestServlet(RestServlet):
         )
 
     @defer.inlineCallbacks
-    def on_POST(self, request):
+    def on_GET(self, request):
         ip = self.hs.get_ip_from_request(request)
         with self.ratelimiter.ratelimit(ip) as wait_deferred:
             yield wait_deferred
 
-            body = parse_json_object_from_request(request)
-            assert_params_in_request(body, ['username'])
+            username = parse_string(request, "username", required=True)
 
-            yield self.registration_handler.check_username(body['username'])
+            yield self.registration_handler.check_username(username)
 
             defer.returnValue((200, {"available": True}))
 

From ccad2ed824f1601d2cc29aafe5dff6ef46668f5e Mon Sep 17 00:00:00 2001
From: Luke Barnard <lukeb@openmarket.com>
Date: Wed, 10 May 2017 17:34:30 +0100
Subject: [PATCH 25/45] Modify condition on empty localpart

---
 synapse/handlers/register.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index dd84c5f5e..ee3a2269a 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -54,7 +54,7 @@ class RegistrationHandler(BaseHandler):
                 Codes.INVALID_USERNAME
             )
 
-        if len(localpart) == 0:
+        if not localpart:
             raise SynapseError(
                 400,
                 "User ID cannot be empty",

From b64d312421976162a8d41246f11652b5003bb66f Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Wed, 10 May 2017 17:46:41 +0100
Subject: [PATCH 26/45] add some logging to purge_history

---
 synapse/storage/events.py | 25 +++++++++++++++++++++----
 1 file changed, 21 insertions(+), 4 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 2ab44ceaa..512828cf3 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -2033,6 +2033,8 @@ class EventsStore(SQLBaseStore):
         for event_id, state_key in event_rows:
             txn.call_after(self._get_state_group_for_event.invalidate, (event_id,))
 
+        logger.debug("[purge] Finding new backward extremities")
+
         # We calculate the new entries for the backward extremeties by finding
         # all events that point to events that are to be purged
         txn.execute(
@@ -2045,6 +2047,8 @@ class EventsStore(SQLBaseStore):
         )
         new_backwards_extrems = txn.fetchall()
 
+        logger.debug("[purge] replacing backward extremities: %r", new_backwards_extrems)
+
         txn.execute(
             "DELETE FROM event_backward_extremities WHERE room_id = ?",
             (room_id,)
@@ -2059,6 +2063,8 @@ class EventsStore(SQLBaseStore):
             ]
         )
 
+        logger.debug("[purge] finding redundant state groups")
+
         # Get all state groups that are only referenced by events that are
         # to be deleted.
         txn.execute(
@@ -2076,6 +2082,10 @@ class EventsStore(SQLBaseStore):
         state_rows = txn.fetchall()
         state_groups_to_delete = [sg for sg, in state_rows]
 
+        logger.debug(
+            "[purge] finding state groups which depend on redundant state groups"
+        )
+
         # Now we get all the state groups that rely on these state groups
         new_state_edges = []
         chunks = [
@@ -2096,6 +2106,8 @@ class EventsStore(SQLBaseStore):
         # Now we turn the state groups that reference to-be-deleted state groups
         # to non delta versions.
         for new_state_edge in new_state_edges:
+            logger.debug("[purge] de-delta-ing remaining state group %s",
+                         new_state_edge)
             curr_state = self._get_state_groups_from_groups_txn(
                 txn, [new_state_edge], types=None
             )
@@ -2132,6 +2144,7 @@ class EventsStore(SQLBaseStore):
                 ],
             )
 
+        logger.debug("[purge] removing redundant state groups")
         txn.executemany(
             "DELETE FROM state_groups_state WHERE state_group = ?",
             state_rows
@@ -2140,12 +2153,15 @@ class EventsStore(SQLBaseStore):
             "DELETE FROM state_groups WHERE id = ?",
             state_rows
         )
+
         # Delete all non-state
+        logger.debug("[purge] removing events from event_to_state_groups")
         txn.executemany(
             "DELETE FROM event_to_state_groups WHERE event_id = ?",
             [(event_id,) for event_id, _ in event_rows]
         )
 
+        logger.debug("[purge] updating room_depth")
         txn.execute(
             "UPDATE room_depth SET min_depth = ? WHERE room_id = ?",
             (topological_ordering, room_id,)
@@ -2171,16 +2187,15 @@ class EventsStore(SQLBaseStore):
             "event_signatures",
             "rejections",
         ):
+            logger.debug("[purge] removing non-state events from %s", table)
+
             txn.executemany(
                 "DELETE FROM %s WHERE event_id = ?" % (table,),
                 to_delete
             )
 
-        txn.executemany(
-            "DELETE FROM events WHERE event_id = ?",
-            to_delete
-        )
         # Mark all state and own events as outliers
+        logger.debug("[purge] marking events as outliers")
         txn.executemany(
             "UPDATE events SET outlier = ?"
             " WHERE event_id = ?",
@@ -2190,6 +2205,8 @@ class EventsStore(SQLBaseStore):
             ]
         )
 
+        logger.debug("[purge] done")
+
     @defer.inlineCallbacks
     def is_event_after(self, event_id1, event_id2):
         """Returns True if event_id1 is after event_id2 in the stream

From 8e345ce46532974aac08c15cf4c90924ec4496d5 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Wed, 10 May 2017 18:17:41 +0100
Subject: [PATCH 27/45] Don't de-delta state groups we're about to delete

---
 synapse/storage/events.py | 20 +++++++++-----------
 1 file changed, 9 insertions(+), 11 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 512828cf3..2a37e6f1a 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -2080,19 +2080,14 @@ class EventsStore(SQLBaseStore):
         )
 
         state_rows = txn.fetchall()
-        state_groups_to_delete = [sg for sg, in state_rows]
-
-        logger.debug(
-            "[purge] finding state groups which depend on redundant state groups"
-        )
+        state_groups_to_delete = set([sg for sg, in state_rows])
 
         # Now we get all the state groups that rely on these state groups
+        logger.debug("[purge] finding state groups which depend on redundant"
+                     " state groups")
         new_state_edges = []
-        chunks = [
-            state_groups_to_delete[i:i + 100]
-            for i in xrange(0, len(state_groups_to_delete), 100)
-        ]
-        for chunk in chunks:
+        for i in xrange(0, len(state_rows), 100):
+            chunk = [sg for sg, in state_rows[i:i + 100]]
             rows = self._simple_select_many_txn(
                 txn,
                 table="state_group_edges",
@@ -2101,7 +2096,10 @@ class EventsStore(SQLBaseStore):
                 retcols=["state_group"],
                 keyvalues={},
             )
-            new_state_edges.extend(row["state_group"] for row in rows)
+            new_state_edges.extend(
+                row["state_group"] for row in rows
+                if row["state_group"] not in state_groups_to_delete
+            )
 
         # Now we turn the state groups that reference to-be-deleted state groups
         # to non delta versions.

From dc026bb16ff552e9424be217ec5c64104c8b193f Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Thu, 11 May 2017 10:56:12 +0100
Subject: [PATCH 28/45] Tidy purge code and add some comments

Try to make this clearer with more comments and some variable renames
---
 synapse/storage/events.py | 31 +++++++++++++++++++------------
 1 file changed, 19 insertions(+), 12 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 2a37e6f1a..dbd63078c 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -2080,14 +2080,19 @@ class EventsStore(SQLBaseStore):
         )
 
         state_rows = txn.fetchall()
+
+        # make a set of the redundant state groups, so that we can look them up
+        # efficiently
         state_groups_to_delete = set([sg for sg, in state_rows])
 
         # Now we get all the state groups that rely on these state groups
         logger.debug("[purge] finding state groups which depend on redundant"
                      " state groups")
-        new_state_edges = []
+        remaining_state_groups = []
         for i in xrange(0, len(state_rows), 100):
             chunk = [sg for sg, in state_rows[i:i + 100]]
+            # look for state groups whose prev_state_group is one we are about
+            # to delete
             rows = self._simple_select_many_txn(
                 txn,
                 table="state_group_edges",
@@ -2096,26 +2101,28 @@ class EventsStore(SQLBaseStore):
                 retcols=["state_group"],
                 keyvalues={},
             )
-            new_state_edges.extend(
+            remaining_state_groups.extend(
                 row["state_group"] for row in rows
+
+                # exclude state groups we are about to delete: no point in
+                # updating them
                 if row["state_group"] not in state_groups_to_delete
             )
 
-        # Now we turn the state groups that reference to-be-deleted state groups
-        # to non delta versions.
-        for new_state_edge in new_state_edges:
-            logger.debug("[purge] de-delta-ing remaining state group %s",
-                         new_state_edge)
+        # Now we turn the state groups that reference to-be-deleted state
+        # groups to non delta versions.
+        for sg in remaining_state_groups:
+            logger.debug("[purge] de-delta-ing remaining state group %s", sg)
             curr_state = self._get_state_groups_from_groups_txn(
-                txn, [new_state_edge], types=None
+                txn, [sg], types=None
             )
-            curr_state = curr_state[new_state_edge]
+            curr_state = curr_state[sg]
 
             self._simple_delete_txn(
                 txn,
                 table="state_groups_state",
                 keyvalues={
-                    "state_group": new_state_edge,
+                    "state_group": sg,
                 }
             )
 
@@ -2123,7 +2130,7 @@ class EventsStore(SQLBaseStore):
                 txn,
                 table="state_group_edges",
                 keyvalues={
-                    "state_group": new_state_edge,
+                    "state_group": sg,
                 }
             )
 
@@ -2132,7 +2139,7 @@ class EventsStore(SQLBaseStore):
                 table="state_groups_state",
                 values=[
                     {
-                        "state_group": new_state_edge,
+                        "state_group": sg,
                         "room_id": room_id,
                         "type": key[0],
                         "state_key": key[1],

From baafb85ba461b7d5073de94422fed0c43a417f46 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Thu, 11 May 2017 11:57:02 +0100
Subject: [PATCH 29/45] Add an index to event_search

- to make the purge API quicker
---
 synapse/storage/background_updates.py           | 10 +++++++---
 synapse/storage/events.py                       | 11 +++++++++++
 .../delta/41/event_search_event_id_idx.sql      | 17 +++++++++++++++++
 3 files changed, 35 insertions(+), 3 deletions(-)
 create mode 100644 synapse/storage/schema/delta/41/event_search_event_id_idx.sql

diff --git a/synapse/storage/background_updates.py b/synapse/storage/background_updates.py
index d4cf0fc59..12a8b8259 100644
--- a/synapse/storage/background_updates.py
+++ b/synapse/storage/background_updates.py
@@ -210,7 +210,8 @@ class BackgroundUpdateStore(SQLBaseStore):
         self._background_update_handlers[update_name] = update_handler
 
     def register_background_index_update(self, update_name, index_name,
-                                         table, columns, where_clause=None):
+                                         table, columns, where_clause=None,
+                                         unique=False):
         """Helper for store classes to do a background index addition
 
         To use:
@@ -245,9 +246,11 @@ class BackgroundUpdateStore(SQLBaseStore):
                 c.execute(sql)
 
                 sql = (
-                    "CREATE INDEX CONCURRENTLY %(name)s ON %(table)s"
+                    "CREATE %(unique)s INDEX CONCURRENTLY %(name)s"
+                    " ON %(table)s"
                     " (%(columns)s) %(where_clause)s"
                 ) % {
+                    "unique": "UNIQUE" if unique else "",
                     "name": index_name,
                     "table": table,
                     "columns": ", ".join(columns),
@@ -270,9 +273,10 @@ class BackgroundUpdateStore(SQLBaseStore):
             # down at the wrong moment - hance we use IF NOT EXISTS. (SQLite
             # has supported CREATE TABLE|INDEX IF NOT EXISTS since 3.3.0.)
             sql = (
-                "CREATE INDEX IF NOT EXISTS %(name)s ON %(table)s"
+                "CREATE %(unique)s INDEX IF NOT EXISTS %(name)s ON %(table)s"
                 " (%(columns)s)"
             ) % {
+                "unique": "UNIQUE" if unique else "",
                 "name": index_name,
                 "table": table,
                 "columns": ", ".join(columns),
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index dbd63078c..1fae1aeac 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -207,6 +207,17 @@ class EventsStore(SQLBaseStore):
             where_clause="contains_url = true AND outlier = false",
         )
 
+        # an event_id index on event_search is useful for the purge_history
+        # api. Plus it means we get to enforce some integrity with a UNIQUE
+        # clause
+        self.register_background_index_update(
+            "event_search_event_id_idx",
+            index_name="event_search_event_id_idx",
+            table="event_search",
+            columns=["event_id"],
+            unique=True,
+        )
+
         self._event_persist_queue = _EventPeristenceQueue()
 
     def persist_events(self, events_and_contexts, backfilled=False):
diff --git a/synapse/storage/schema/delta/41/event_search_event_id_idx.sql b/synapse/storage/schema/delta/41/event_search_event_id_idx.sql
new file mode 100644
index 000000000..5d9cfecf3
--- /dev/null
+++ b/synapse/storage/schema/delta/41/event_search_event_id_idx.sql
@@ -0,0 +1,17 @@
+/* Copyright 2017 Vector Creations 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 ('event_search_event_id_idx', '{}');

From 114f2909479b4396f3da8cff651990f075b4bfba Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Thu, 11 May 2017 12:06:28 +0100
Subject: [PATCH 30/45] Add more logging for purging

Log the number of events we will be deleting at info.
---
 synapse/storage/events.py | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 1fae1aeac..627e91a52 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -2033,6 +2033,8 @@ class EventsStore(SQLBaseStore):
                 400, "topological_ordering is greater than forward extremeties"
             )
 
+        logger.debug("[purge] looking for events to delete")
+
         txn.execute(
             "SELECT event_id, state_key FROM events"
             " LEFT JOIN state_events USING (room_id, event_id)"
@@ -2041,6 +2043,14 @@ class EventsStore(SQLBaseStore):
         )
         event_rows = txn.fetchall()
 
+        to_delete = [
+            (event_id,) for event_id, state_key in event_rows
+            if state_key is None and not self.hs.is_mine_id(event_id)
+        ]
+        logger.info(
+            "[purge] found %i events before cutoff, of which %i are remote"
+            " non-state events to delete", len(event_rows), len(to_delete))
+
         for event_id, state_key in event_rows:
             txn.call_after(self._get_state_group_for_event.invalidate, (event_id,))
 
@@ -2091,6 +2101,7 @@ class EventsStore(SQLBaseStore):
         )
 
         state_rows = txn.fetchall()
+        logger.debug("[purge] found %i redundant state groups", len(state_rows))
 
         # make a set of the redundant state groups, so that we can look them up
         # efficiently
@@ -2184,10 +2195,6 @@ class EventsStore(SQLBaseStore):
         )
 
         # Delete all remote non-state events
-        to_delete = [
-            (event_id,) for event_id, state_key in event_rows
-            if state_key is None and not self.hs.is_mine_id(event_id)
-        ]
         for table in (
             "events",
             "event_json",
@@ -2203,7 +2210,7 @@ class EventsStore(SQLBaseStore):
             "event_signatures",
             "rejections",
         ):
-            logger.debug("[purge] removing non-state events from %s", table)
+            logger.debug("[purge] removing remote non-state events from %s", table)
 
             txn.executemany(
                 "DELETE FROM %s WHERE event_id = ?" % (table,),
@@ -2211,7 +2218,7 @@ class EventsStore(SQLBaseStore):
             )
 
         # Mark all state and own events as outliers
-        logger.debug("[purge] marking events as outliers")
+        logger.debug("[purge] marking remaining events as outliers")
         txn.executemany(
             "UPDATE events SET outlier = ?"
             " WHERE event_id = ?",
@@ -2221,7 +2228,7 @@ class EventsStore(SQLBaseStore):
             ]
         )
 
-        logger.debug("[purge] done")
+        logger.info("[purge] done")
 
     @defer.inlineCallbacks
     def is_event_after(self, event_id1, event_id2):

From 34194aaff7c86d57c6dabfb016b7597d999b2001 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Thu, 11 May 2017 12:46:55 +0100
Subject: [PATCH 31/45] Don't create event_search index on sqlite

... because the table is virtual
---
 synapse/storage/background_updates.py | 13 ++++++++++---
 synapse/storage/events.py             |  1 +
 2 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/synapse/storage/background_updates.py b/synapse/storage/background_updates.py
index 12a8b8259..7157fb1df 100644
--- a/synapse/storage/background_updates.py
+++ b/synapse/storage/background_updates.py
@@ -211,7 +211,8 @@ class BackgroundUpdateStore(SQLBaseStore):
 
     def register_background_index_update(self, update_name, index_name,
                                          table, columns, where_clause=None,
-                                         unique=False):
+                                         unique=False,
+                                         psql_only=False):
         """Helper for store classes to do a background index addition
 
         To use:
@@ -227,6 +228,9 @@ class BackgroundUpdateStore(SQLBaseStore):
             index_name (str): name of index to add
             table (str): table to add index to
             columns (list[str]): columns/expressions to include in index
+            unique (bool): true to make a UNIQUE index
+            psql_only: true to only create this index on psql databases (useful
+                for virtual sqlite tables)
         """
 
         def create_index_psql(conn):
@@ -288,13 +292,16 @@ class BackgroundUpdateStore(SQLBaseStore):
 
         if isinstance(self.database_engine, engines.PostgresEngine):
             runner = create_index_psql
+        elif psql_only:
+            runner = None
         else:
             runner = create_index_sqlite
 
         @defer.inlineCallbacks
         def updater(progress, batch_size):
-            logger.info("Adding index %s to %s", index_name, table)
-            yield self.runWithConnection(runner)
+            if runner is not None:
+                logger.info("Adding index %s to %s", index_name, table)
+                yield self.runWithConnection(runner)
             yield self._end_background_update(update_name)
             defer.returnValue(1)
 
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 627e91a52..ea6879c61 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -216,6 +216,7 @@ class EventsStore(SQLBaseStore):
             table="event_search",
             columns=["event_id"],
             unique=True,
+            psql_only=True,
         )
 
         self._event_persist_queue = _EventPeristenceQueue()

From ff3d810ea8e84a48508a08e6246c7d70739c94ea Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <richard@matrix.org>
Date: Thu, 11 May 2017 12:48:50 +0100
Subject: [PATCH 32/45] Add a comment to old delta

---
 synapse/storage/schema/delta/37/remove_auth_idx.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/synapse/storage/schema/delta/37/remove_auth_idx.py b/synapse/storage/schema/delta/37/remove_auth_idx.py
index 784f3b348..20ad8bd5a 100644
--- a/synapse/storage/schema/delta/37/remove_auth_idx.py
+++ b/synapse/storage/schema/delta/37/remove_auth_idx.py
@@ -36,6 +36,10 @@ DROP INDEX IF EXISTS transactions_have_ref;
 -- and is used incredibly rarely.
 DROP INDEX IF EXISTS events_order_topo_stream_room;
 
+-- an equivalent index to this actually gets re-created in delta 41, because it
+-- turned out that deleting it wasn't a great plan :/. In any case, let's
+-- delete it here, and delta 41 will create a new one with an added UNIQUE
+-- constraint
 DROP INDEX IF EXISTS event_search_ev_idx;
 """
 

From 9da4316ca574f2d552136941d92ce722fabfe29e Mon Sep 17 00:00:00 2001
From: Pablo Saavedra <psaavedra@igalia.com>
Date: Sat, 13 May 2017 18:17:54 +0200
Subject: [PATCH 33/45] Configurable maximum number of events requested by
 /sync and /messages (#2220)

Set the limit on the returned events in the timeline in the get and sync
operations. The default value is -1, means no upper limit.

For example, using `filter_timeline_limit: 5000`:

POST /_matrix/client/r0/user/user:id/filter
{
room: {
    timeline: {
      limit: 1000000000000000000
    }
}
}

GET /_matrix/client/r0/user/user:id/filter/filter:id

{
room: {
    timeline: {
      limit: 5000
    }
}
}

The server cuts down the room.timeline.limit.
---
 synapse/config/server.py               | 6 ++++++
 synapse/rest/client/v2_alpha/_base.py  | 8 ++++++++
 synapse/rest/client/v2_alpha/filter.py | 4 ++++
 synapse/rest/client/v2_alpha/sync.py   | 3 +++
 4 files changed, 21 insertions(+)

diff --git a/synapse/config/server.py b/synapse/config/server.py
index 25e666623..3910b9dc3 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -35,6 +35,8 @@ class ServerConfig(Config):
         # "disable" federation
         self.send_federation = config.get("send_federation", True)
 
+        self.filter_timeline_limit = config.get("filter_timeline_limit", -1)
+
         if self.public_baseurl is not None:
             if self.public_baseurl[-1] != '/':
                 self.public_baseurl += '/'
@@ -161,6 +163,10 @@ class ServerConfig(Config):
         # The GC threshold parameters to pass to `gc.set_threshold`, if defined
         # gc_thresholds: [700, 10, 10]
 
+        # Set the limit on the returned events in the timeline in the get
+        # and sync operations. The default value is -1, means no upper limit.
+        # filter_timeline_limit: 5000
+
         # List of ports that Synapse should listen on, their purpose and their
         # configuration.
         listeners:
diff --git a/synapse/rest/client/v2_alpha/_base.py b/synapse/rest/client/v2_alpha/_base.py
index 20e765f48..279e2a775 100644
--- a/synapse/rest/client/v2_alpha/_base.py
+++ b/synapse/rest/client/v2_alpha/_base.py
@@ -47,3 +47,11 @@ def client_v2_patterns(path_regex, releases=(0,),
         new_prefix = CLIENT_V2_ALPHA_PREFIX.replace("/v2_alpha", "/r%d" % release)
         patterns.append(re.compile("^" + new_prefix + path_regex))
     return patterns
+
+
+def set_timeline_upper_limit(filter_json, filter_timeline_limit):
+    if filter_timeline_limit < 0:
+        return  # no upper limits
+    if 'room' in filter_json and 'limit' in filter_json['room']:
+        filter_json['room']["limit"] = min(filter_json['room']["limit"],
+                                           filter_timeline_limit)
diff --git a/synapse/rest/client/v2_alpha/filter.py b/synapse/rest/client/v2_alpha/filter.py
index b4084fec6..364d20d8e 100644
--- a/synapse/rest/client/v2_alpha/filter.py
+++ b/synapse/rest/client/v2_alpha/filter.py
@@ -20,6 +20,7 @@ from synapse.http.servlet import RestServlet, parse_json_object_from_request
 from synapse.types import UserID
 
 from ._base import client_v2_patterns
+from ._base import set_timeline_upper_limit
 
 import logging
 
@@ -85,6 +86,9 @@ class CreateFilterRestServlet(RestServlet):
             raise AuthError(403, "Can only create filters for local users")
 
         content = parse_json_object_from_request(request)
+        set_timeline_upper_limit(content,
+                                 self.hs.config.filter_timeline_limit)
+
         filter_id = yield self.filtering.add_user_filter(
             user_localpart=target_user.localpart,
             user_filter=content,
diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py
index f30eab76f..f5e7349c5 100644
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -28,6 +28,7 @@ from synapse.api.filtering import FilterCollection, DEFAULT_FILTER_COLLECTION
 from synapse.api.errors import SynapseError
 from synapse.api.constants import PresenceState
 from ._base import client_v2_patterns
+from ._base import set_timeline_upper_limit
 
 import itertools
 import logging
@@ -121,6 +122,8 @@ class SyncRestServlet(RestServlet):
             if filter_id.startswith('{'):
                 try:
                     filter_object = json.loads(filter_id)
+                    set_timeline_upper_limit(filter_object,
+                                             self.hs.config.filter_timeline_limit)
                 except:
                     raise SynapseError(400, "Invalid filter JSON")
                 self.filtering.check_valid_filter(filter_object)

From 627e6ea2b0b941c67f8752736993f82a5f123e76 Mon Sep 17 00:00:00 2001
From: Pablo Saavedra <psaavedra@igalia.com>
Date: Mon, 15 May 2017 14:51:43 +0200
Subject: [PATCH 34/45] Fixed implementation errors
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Added HS as property in SyncRestServlet
* Fixed set_timeline_upper_limit function implementat¡ion
---
 synapse/rest/client/v2_alpha/_base.py | 9 ++++++---
 synapse/rest/client/v2_alpha/sync.py  | 1 +
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/synapse/rest/client/v2_alpha/_base.py b/synapse/rest/client/v2_alpha/_base.py
index 279e2a775..fd8f91565 100644
--- a/synapse/rest/client/v2_alpha/_base.py
+++ b/synapse/rest/client/v2_alpha/_base.py
@@ -52,6 +52,9 @@ def client_v2_patterns(path_regex, releases=(0,),
 def set_timeline_upper_limit(filter_json, filter_timeline_limit):
     if filter_timeline_limit < 0:
         return  # no upper limits
-    if 'room' in filter_json and 'limit' in filter_json['room']:
-        filter_json['room']["limit"] = min(filter_json['room']["limit"],
-                                           filter_timeline_limit)
+    if 'room' in filter_json \
+            and 'timeline' in filter_json['room'] \
+            and 'limit' in filter_json['room']['timeline']:
+        filter_json['room']['timeline']["limit"] = min(
+            filter_json['room']['timeline']['limit'],
+            filter_timeline_limit)
diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py
index f5e7349c5..771e127ab 100644
--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -79,6 +79,7 @@ class SyncRestServlet(RestServlet):
 
     def __init__(self, hs):
         super(SyncRestServlet, self).__init__()
+        self.hs = hs
         self.auth = hs.get_auth()
         self.sync_handler = hs.get_sync_handler()
         self.clock = hs.get_clock()

From bfbc907cec96ce9a64730930f63ed400c1aa3b5b Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 2 May 2017 10:40:31 +0100
Subject: [PATCH 35/45] Prefill state caches

---
 synapse/storage/_base.py  |  8 ++++----
 synapse/storage/events.py | 10 ++++++++--
 synapse/storage/state.py  |  8 ++++++++
 3 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index c659004e8..58b73af7d 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -60,12 +60,12 @@ class LoggingTransaction(object):
         object.__setattr__(self, "database_engine", database_engine)
         object.__setattr__(self, "after_callbacks", after_callbacks)
 
-    def call_after(self, callback, *args):
+    def call_after(self, callback, *args, **kwargs):
         """Call the given callback on the main twisted thread after the
         transaction has finished. Used to invalidate the caches on the
         correct thread.
         """
-        self.after_callbacks.append((callback, args))
+        self.after_callbacks.append((callback, args, kwargs))
 
     def __getattr__(self, name):
         return getattr(self.txn, name)
@@ -319,8 +319,8 @@ class SQLBaseStore(object):
                     inner_func, *args, **kwargs
                 )
         finally:
-            for after_callback, after_args in after_callbacks:
-                after_callback(*after_args)
+            for after_callback, after_args, after_kwargs in after_callbacks:
+                after_callback(*after_args, **after_kwargs)
         defer.returnValue(result)
 
     @defer.inlineCallbacks
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index dbd63078c..0dffafd90 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -374,6 +374,7 @@ class EventsStore(SQLBaseStore):
                     new_forward_extremeties=new_forward_extremeties,
                 )
                 persist_event_counter.inc_by(len(chunk))
+
                 for event, context in chunk:
                     if context.app_service:
                         origin_type = "local"
@@ -387,6 +388,11 @@ class EventsStore(SQLBaseStore):
 
                     event_counter.inc(event.type, origin_type, origin_entity)
 
+                for room_id, (_, _, new_state) in current_state_for_room.iteritems():
+                    self.get_current_state_ids.prefill(
+                        (room_id, ), new_state
+                    )
+
     @defer.inlineCallbacks
     def _calculate_new_extremeties(self, room_id, event_contexts, latest_event_ids):
         """Calculates the new forward extremeties for a room given events to
@@ -545,7 +551,7 @@ class EventsStore(SQLBaseStore):
             if ev_id in events_to_insert
         }
 
-        defer.returnValue((to_delete, to_insert))
+        defer.returnValue((to_delete, to_insert, current_state))
 
     @defer.inlineCallbacks
     def get_event(self, event_id, check_redacted=True,
@@ -698,7 +704,7 @@ class EventsStore(SQLBaseStore):
 
     def _update_current_state_txn(self, txn, state_delta_by_room):
         for room_id, current_state_tuple in state_delta_by_room.iteritems():
-                to_delete, to_insert = current_state_tuple
+                to_delete, to_insert, _ = current_state_tuple
                 txn.executemany(
                     "DELETE FROM current_state_events WHERE event_id = ?",
                     [(ev_id,) for ev_id in to_delete.itervalues()],
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index a16afa8df..1e1ce87e0 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -227,6 +227,14 @@ class StateStore(SQLBaseStore):
                     ],
                 )
 
+            txn.call_after(
+                self._state_group_cache.update,
+                self._state_group_cache.sequence,
+                key=context.state_group,
+                value=context.current_state_ids,
+                full=True,
+            )
+
         self._simple_insert_many_txn(
             txn,
             table="event_to_state_groups",

From e0d2f6d5b02dd208bc55434b5c2d386827486e9f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 2 May 2017 11:36:11 +0100
Subject: [PATCH 36/45] Add more granular event send metrics

---
 synapse/storage/events.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 0dffafd90..36574f78b 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -374,6 +374,18 @@ class EventsStore(SQLBaseStore):
                     new_forward_extremeties=new_forward_extremeties,
                 )
                 persist_event_counter.inc_by(len(chunk))
+                for event, context in chunk:
+                    if context.app_service:
+                        origin_type = "local"
+                        origin_entity = context.app_service.id
+                    elif self.hs.is_mine_id(event.sender):
+                        origin_type = "local"
+                        origin_entity = "*client*"
+                    else:
+                        origin_type = "remote"
+                        origin_entity = get_domain_from_id(event.sender)
+
+                    event_counter.inc(event.type, origin_type, origin_entity)
 
                 for event, context in chunk:
                     if context.app_service:

From 871605f4e20cce3f093b2eae0f3d2ad7fb43a640 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 3 May 2017 09:56:05 +0100
Subject: [PATCH 37/45] Comments

---
 synapse/storage/events.py | 6 +++---
 synapse/storage/state.py  | 3 +++
 2 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 36574f78b..5db7ec162 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -453,10 +453,10 @@ class EventsStore(SQLBaseStore):
         Assumes that we are only persisting events for one room at a time.
 
         Returns:
-            2-tuple (to_delete, to_insert) where both are state dicts, i.e.
-            (type, state_key) -> event_id. `to_delete` are the entries to
+            3-tuple (to_delete, to_insert, new_state) where both are state dicts,
+            i.e. (type, state_key) -> event_id. `to_delete` are the entries to
             first be deleted from current_state_events, `to_insert` are entries
-            to insert.
+            to insert. `new_state` is the full set of state.
             May return None if there are no changes to be applied.
         """
         # Now we need to work out the different state sets for
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 1e1ce87e0..5d6f7dfa2 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -227,6 +227,9 @@ class StateStore(SQLBaseStore):
                     ],
                 )
 
+            # Prefill the state group cache with this group.
+            # It's fine to use the sequence like this as the state group map
+            # is immutable.
             txn.call_after(
                 self._state_group_cache.update,
                 self._state_group_cache.sequence,

From e4435b014e50a10ad89c201d6f91b6be35a9b02f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Wed, 3 May 2017 10:00:29 +0100
Subject: [PATCH 38/45] Update comment

---
 synapse/storage/state.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 5d6f7dfa2..03981f5d2 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -229,7 +229,8 @@ class StateStore(SQLBaseStore):
 
             # Prefill the state group cache with this group.
             # It's fine to use the sequence like this as the state group map
-            # is immutable.
+            # is immutable. (If the map wasn't immutable then this prefill could
+            # race with another update)
             txn.call_after(
                 self._state_group_cache.update,
                 self._state_group_cache.sequence,

From 224137fcf98b424c6b5d1ab4b76c24b6213c7174 Mon Sep 17 00:00:00 2001
From: Pablo Saavedra <psaavedra@igalia.com>
Date: Mon, 15 May 2017 16:21:02 +0200
Subject: [PATCH 39/45] Fixed syntax nits

---
 synapse/rest/client/v2_alpha/_base.py  | 5 ++---
 synapse/rest/client/v2_alpha/filter.py | 6 ++++--
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/synapse/rest/client/v2_alpha/_base.py b/synapse/rest/client/v2_alpha/_base.py
index fd8f91565..1f5bc24cc 100644
--- a/synapse/rest/client/v2_alpha/_base.py
+++ b/synapse/rest/client/v2_alpha/_base.py
@@ -52,9 +52,8 @@ def client_v2_patterns(path_regex, releases=(0,),
 def set_timeline_upper_limit(filter_json, filter_timeline_limit):
     if filter_timeline_limit < 0:
         return  # no upper limits
-    if 'room' in filter_json \
-            and 'timeline' in filter_json['room'] \
-            and 'limit' in filter_json['room']['timeline']:
+    timeline = filter_json.get('room', {}).get('timeline', {})
+    if 'limit' in timeline:
         filter_json['room']['timeline']["limit"] = min(
             filter_json['room']['timeline']['limit'],
             filter_timeline_limit)
diff --git a/synapse/rest/client/v2_alpha/filter.py b/synapse/rest/client/v2_alpha/filter.py
index 364d20d8e..d2b2fd66e 100644
--- a/synapse/rest/client/v2_alpha/filter.py
+++ b/synapse/rest/client/v2_alpha/filter.py
@@ -86,8 +86,10 @@ class CreateFilterRestServlet(RestServlet):
             raise AuthError(403, "Can only create filters for local users")
 
         content = parse_json_object_from_request(request)
-        set_timeline_upper_limit(content,
-                                 self.hs.config.filter_timeline_limit)
+        set_timeline_upper_limit(
+            content,
+            self.hs.config.filter_timeline_limit
+        )
 
         filter_id = yield self.filtering.add_user_filter(
             user_localpart=target_user.localpart,

From d12ae7fd1c213c2505e3904aab98604fa86f42f5 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 15 May 2017 15:42:18 +0100
Subject: [PATCH 40/45] Don't log exceptions for NotRetryingDestination

---
 synapse/rest/media/v1/media_repository.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/synapse/rest/media/v1/media_repository.py b/synapse/rest/media/v1/media_repository.py
index c43b185e0..caca96c22 100644
--- a/synapse/rest/media/v1/media_repository.py
+++ b/synapse/rest/media/v1/media_repository.py
@@ -34,6 +34,7 @@ from synapse.api.errors import SynapseError, HttpResponseException, \
 from synapse.util.async import Linearizer
 from synapse.util.stringutils import is_ascii
 from synapse.util.logcontext import preserve_context_over_fn
+from synapse.util.retryutils import NotRetryingDestination
 
 import os
 import errno
@@ -181,7 +182,8 @@ class MediaRepository(object):
                     logger.exception("Failed to fetch remote media %s/%s",
                                      server_name, media_id)
                     raise
-
+                except NotRetryingDestination:
+                    logger.warn("Not retrying destination %r", server_name)
                 except Exception:
                     logger.exception("Failed to fetch remote media %s/%s",
                                      server_name, media_id)

From 608b5a6317ce3797ff279f6d1a8a39f475b55736 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 16 May 2017 12:55:29 +0100
Subject: [PATCH 41/45] Take a copy before prefilling, as it may be a
 frozendict

---
 synapse/storage/state.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 03981f5d2..85acf2ad1 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -235,7 +235,7 @@ class StateStore(SQLBaseStore):
                 self._state_group_cache.update,
                 self._state_group_cache.sequence,
                 key=context.state_group,
-                value=context.current_state_ids,
+                value=dict(context.current_state_ids),
                 full=True,
             )
 

From 331570ea6f97d570cf2774cd0700eb588e9fb1d7 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 16 May 2017 15:33:07 +0100
Subject: [PATCH 42/45] Remove spurious merge artifacts

---
 synapse/storage/events.py | 13 -------------
 1 file changed, 13 deletions(-)

diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 5db7ec162..12dd74daa 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -387,19 +387,6 @@ class EventsStore(SQLBaseStore):
 
                     event_counter.inc(event.type, origin_type, origin_entity)
 
-                for event, context in chunk:
-                    if context.app_service:
-                        origin_type = "local"
-                        origin_entity = context.app_service.id
-                    elif self.hs.is_mine_id(event.sender):
-                        origin_type = "local"
-                        origin_entity = "*client*"
-                    else:
-                        origin_type = "remote"
-                        origin_entity = get_domain_from_id(event.sender)
-
-                    event_counter.inc(event.type, origin_type, origin_entity)
-
                 for room_id, (_, _, new_state) in current_state_for_room.iteritems():
                     self.get_current_state_ids.prefill(
                         (room_id, ), new_state

From f2a5b6dbfd0f21919a36f779026e041900b998a9 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 16 May 2017 14:07:08 +0100
Subject: [PATCH 43/45] Speed up get_domain_from_id

---
 synapse/types.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/synapse/types.py b/synapse/types.py
index c87ed813b..445bdcb4d 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -56,10 +56,10 @@ def create_requester(user_id, access_token_id=None, is_guest=False,
 
 
 def get_domain_from_id(string):
-    try:
-        return string.split(":", 1)[1]
-    except IndexError:
+    idx = string.find(":")
+    if idx == -1:
         raise SynapseError(400, "Invalid ID: %r" % (string,))
+    return string[idx + 1:]
 
 
 class DomainSpecificString(

From ec5c4499f4ab24445c6df7310007353b466020ce Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 16 May 2017 14:46:16 +0100
Subject: [PATCH 44/45] Make presence use cached users/hosts in room

---
 synapse/federation/transaction_queue.py | 2 +-
 synapse/handlers/presence.py            | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index 695f1a737..a15198e05 100644
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -285,7 +285,7 @@ class TransactionQueue(object):
         Args:
             states (list(UserPresenceState))
         """
-        hosts_and_states = yield get_interested_remotes(self.store, states)
+        hosts_and_states = yield get_interested_remotes(self.store, states, self.state)
 
         for destinations, states in hosts_and_states:
             for destination in destinations:
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index f3707afcd..c7c0b0a1e 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -780,12 +780,12 @@ class PresenceHandler(object):
         # don't need to send to local clients here, as that is done as part
         # of the event stream/sync.
         # TODO: Only send to servers not already in the room.
-        user_ids = yield self.store.get_users_in_room(room_id)
         if self.is_mine(user):
             state = yield self.current_state_for_user(user.to_string())
 
             self._push_to_remotes([state])
         else:
+            user_ids = yield self.store.get_users_in_room(room_id)
             user_ids = filter(self.is_mine_id, user_ids)
 
             states = yield self.current_state_for_users(user_ids)
@@ -1322,7 +1322,7 @@ def get_interested_parties(store, states):
 
 
 @defer.inlineCallbacks
-def get_interested_remotes(store, states):
+def get_interested_remotes(store, states, state_handler):
     """Given a list of presence states figure out which remote servers
     should be sent which.
 
@@ -1345,7 +1345,7 @@ def get_interested_remotes(store, states):
     room_ids_to_states, users_to_states = yield get_interested_parties(store, states)
 
     for room_id, states in room_ids_to_states.iteritems():
-        hosts = yield store.get_hosts_in_room(room_id)
+        hosts = yield state_handler.get_current_hosts_in_room(room_id)
         hosts_and_states.append((hosts, states))
 
     for user_id, states in users_to_states.iteritems():

From 13f540ef1b94e6173bdd4f2d84d90e0948cf5bf2 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Tue, 16 May 2017 14:07:24 +0100
Subject: [PATCH 45/45] Speed up get_joined_hosts

---
 synapse/handlers/room_member.py | 3 ++-
 synapse/storage/roommember.py   | 3 ++-
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index ab87632d9..1ca88517a 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -739,10 +739,11 @@ class RoomMemberHandler(BaseHandler):
         if len(current_state_ids) == 1 and create_event_id:
             defer.returnValue(self.hs.is_mine_id(create_event_id))
 
-        for (etype, state_key), event_id in current_state_ids.items():
+        for etype, state_key in current_state_ids:
             if etype != EventTypes.Member or not self.hs.is_mine_id(state_key):
                 continue
 
+            event_id = current_state_ids[(etype, state_key)]
             event = yield self.store.get_event(event_id, allow_none=True)
             if not event:
                 continue
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 2fa20bd87..404f3583e 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -534,7 +534,7 @@ class RoomMemberStore(SQLBaseStore):
         assert state_group is not None
 
         joined_hosts = set()
-        for (etype, state_key), event_id in current_state_ids.items():
+        for etype, state_key in current_state_ids:
             if etype == EventTypes.Member:
                 try:
                     host = get_domain_from_id(state_key)
@@ -545,6 +545,7 @@ class RoomMemberStore(SQLBaseStore):
                 if host in joined_hosts:
                     continue
 
+                event_id = current_state_ids[(etype, state_key)]
                 event = yield self.get_event(event_id, allow_none=True)
                 if event and event.content["membership"] == Membership.JOIN:
                     joined_hosts.add(intern_string(host))