From 0ac2a79faa918280767c18e4db7ec29d7d3a3afb Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 19 Feb 2015 17:24:14 +0000
Subject: Initial stab at implementing a batched get_missing_pdus request

---
 synapse/federation/federation_server.py | 72 +++++++++++++++++++++++++++++++++
 1 file changed, 72 insertions(+)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 22b9663831..34bc397e8a 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -305,6 +305,78 @@ class FederationServer(FederationBase):
             (200, send_content)
         )
 
+    @defer.inlineCallbacks
+    def get_missing_events(self, origin, room_id, earliest_events,
+                           latest_events, limit, min_depth):
+        limit = max(limit, 50)
+        min_depth = max(min_depth, 0)
+
+        missing_events = yield self.store.get_missing_events(
+            room_id=room_id,
+            earliest_events=earliest_events,
+            latest_events=latest_events,
+            limit=limit,
+            min_depth=min_depth,
+        )
+
+        known_ids = {e.event_id for e in missing_events} | {earliest_events}
+
+        back_edges = {
+            e for e in missing_events
+            if {i for i, h in e.prev_events.items()} <= known_ids
+        }
+
+        decoded_auth_events = set()
+        state = {}
+        auth_events = set()
+        auth_and_state = {}
+        for event in back_edges:
+            state_pdus = yield self.handler.get_state_for_pdu(
+                origin, room_id, event.event_id,
+                do_auth=False,
+            )
+
+            state[event.event_id] = [s.event_id for s in state_pdus]
+
+            auth_and_state.update({
+                s.event_id: s for s in state_pdus
+            })
+
+            state_ids = {pdu.event_id for pdu in state_pdus}
+            prev_ids = {i for i, h in event.prev_events.items()}
+            partial_auth_chain = yield self.store.get_auth_chain(
+                state_ids | prev_ids, have_ids=decoded_auth_events.keys()
+            )
+
+            for p in partial_auth_chain:
+                p.signatures.update(
+                    compute_event_signature(
+                        p,
+                        self.hs.hostname,
+                        self.hs.config.signing_key[0]
+                    )
+                )
+
+            auth_events.update(
+                a.event_id for a in partial_auth_chain
+            )
+
+            auth_and_state.update({
+                a.event_id: a for a in partial_auth_chain
+            })
+
+        time_now = self._clock.time_msec()
+
+        defer.returnValue({
+            "events": [ev.get_pdu_json(time_now) for ev in missing_events],
+            "state_for_events": state,
+            "auth_events": auth_events,
+            "event_map": {
+                k: ev.get_pdu_json(time_now)
+                for k, ev in auth_and_state.items()
+            },
+        })
+
     @log_function
     def _get_persisted_pdu(self, origin, event_id, do_auth=True):
         """ Get a PDU from the database with given origin and id.
-- 
cgit 1.4.1


From db215b7e0007a207b8775d78c6693153e16f2731 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 23 Feb 2015 13:58:02 +0000
Subject: Implement and use new batched get missing pdu

---
 synapse/federation/federation_client.py |  19 ++++
 synapse/federation/federation_server.py | 150 +++++++++++---------------------
 synapse/federation/transaction_queue.py |   2 +-
 synapse/federation/transport/client.py  |  19 ++++
 synapse/federation/transport/server.py  |  31 +++++++
 synapse/handlers/federation.py          |  23 +++++
 6 files changed, 144 insertions(+), 100 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index cd3c962d50..ca89a0787c 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -439,6 +439,25 @@ class FederationClient(FederationBase):
 
         defer.returnValue(ret)
 
+    @defer.inlineCallbacks
+    def get_missing_events(self, destination, room_id, earliest_events,
+                           latest_events, limit, min_depth):
+        content = yield self.transport_layer.get_missing_events(
+            destination, room_id, earliest_events, latest_events, limit,
+            min_depth,
+        )
+
+        events = [
+            self.event_from_pdu_json(e)
+            for e in content.get("events", [])
+        ]
+
+        signed_events = yield self._check_sigs_and_hash_and_fetch(
+            destination, events, outlier=True
+        )
+
+        defer.returnValue(signed_events)
+
     def event_from_pdu_json(self, pdu_json, outlier=False):
         event = FrozenEvent(
             pdu_json
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 34bc397e8a..f74e16abd5 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -142,7 +142,15 @@ class FederationServer(FederationBase):
             if r[0]:
                 ret.append({})
             else:
-                logger.exception(r[1])
+                failure = r[1]
+                logger.error(
+                    "Failed to handle PDU",
+                    exc_info=(
+                        failure.type,
+                        failure.value,
+                        failure.getTracebackObject()
+                    )
+                )
                 ret.append({"error": str(r[1].value)})
 
         logger.debug("Returning: %s", str(ret))
@@ -306,75 +314,17 @@ class FederationServer(FederationBase):
         )
 
     @defer.inlineCallbacks
-    def get_missing_events(self, origin, room_id, earliest_events,
-                           latest_events, limit, min_depth):
-        limit = max(limit, 50)
-        min_depth = max(min_depth, 0)
-
-        missing_events = yield self.store.get_missing_events(
-            room_id=room_id,
-            earliest_events=earliest_events,
-            latest_events=latest_events,
-            limit=limit,
-            min_depth=min_depth,
+    @log_function
+    def on_get_missing_events(self, origin, room_id, earliest_events,
+                              latest_events, limit, min_depth):
+        missing_events = yield self.handler.on_get_missing_events(
+            origin, room_id, earliest_events, latest_events, limit, min_depth
         )
 
-        known_ids = {e.event_id for e in missing_events} | {earliest_events}
-
-        back_edges = {
-            e for e in missing_events
-            if {i for i, h in e.prev_events.items()} <= known_ids
-        }
-
-        decoded_auth_events = set()
-        state = {}
-        auth_events = set()
-        auth_and_state = {}
-        for event in back_edges:
-            state_pdus = yield self.handler.get_state_for_pdu(
-                origin, room_id, event.event_id,
-                do_auth=False,
-            )
-
-            state[event.event_id] = [s.event_id for s in state_pdus]
-
-            auth_and_state.update({
-                s.event_id: s for s in state_pdus
-            })
-
-            state_ids = {pdu.event_id for pdu in state_pdus}
-            prev_ids = {i for i, h in event.prev_events.items()}
-            partial_auth_chain = yield self.store.get_auth_chain(
-                state_ids | prev_ids, have_ids=decoded_auth_events.keys()
-            )
-
-            for p in partial_auth_chain:
-                p.signatures.update(
-                    compute_event_signature(
-                        p,
-                        self.hs.hostname,
-                        self.hs.config.signing_key[0]
-                    )
-                )
-
-            auth_events.update(
-                a.event_id for a in partial_auth_chain
-            )
-
-            auth_and_state.update({
-                a.event_id: a for a in partial_auth_chain
-            })
-
         time_now = self._clock.time_msec()
 
         defer.returnValue({
             "events": [ev.get_pdu_json(time_now) for ev in missing_events],
-            "state_for_events": state,
-            "auth_events": auth_events,
-            "event_map": {
-                k: ev.get_pdu_json(time_now)
-                for k, ev in auth_and_state.items()
-            },
         })
 
     @log_function
@@ -403,7 +353,7 @@ class FederationServer(FederationBase):
 
     @defer.inlineCallbacks
     @log_function
-    def _handle_new_pdu(self, origin, pdu, max_recursion=10):
+    def _handle_new_pdu(self, origin, pdu, get_missing=True):
         # We reprocess pdus when we have seen them only as outliers
         existing = yield self._get_persisted_pdu(
             origin, pdu.event_id, do_auth=False
@@ -455,48 +405,50 @@ class FederationServer(FederationBase):
                 pdu.room_id, min_depth
             )
 
+            prevs = {e_id for e_id, _ in pdu.prev_events}
+            seen = set(have_seen.keys())
+
             if min_depth and pdu.depth < min_depth:
                 # This is so that we don't notify the user about this
                 # message, to work around the fact that some events will
                 # reference really really old events we really don't want to
                 # send to the clients.
                 pdu.internal_metadata.outlier = True
-            elif min_depth and pdu.depth > min_depth and max_recursion > 0:
-                for event_id, hashes in pdu.prev_events:
-                    if event_id not in have_seen:
-                        logger.debug(
-                            "_handle_new_pdu requesting pdu %s",
-                            event_id
+            elif min_depth and pdu.depth > min_depth:
+                if get_missing and prevs - seen:
+                    latest_tuples = yield self.store.get_latest_events_in_room(
+                        pdu.room_id
+                    )
+
+                    # We add the prev events that we have seen to the latest
+                    # list to ensure the remote server doesn't give them to us
+                    latest = set(e_id for e_id, _, _ in latest_tuples)
+                    latest |= seen
+
+                    missing_events = yield self.get_missing_events(
+                        origin,
+                        pdu.room_id,
+                        earliest_events=list(latest),
+                        latest_events=[pdu.event_id],
+                        limit=10,
+                        min_depth=min_depth,
+                    )
+
+                    for e in missing_events:
+                        yield self._handle_new_pdu(
+                            origin,
+                            e,
+                            get_missing=False
                         )
 
-                        try:
-                            new_pdu = yield self.federation_client.get_pdu(
-                                [origin, pdu.origin],
-                                event_id=event_id,
-                            )
-
-                            if new_pdu:
-                                yield self._handle_new_pdu(
-                                    origin,
-                                    new_pdu,
-                                    max_recursion=max_recursion-1
-                                )
-
-                                logger.debug("Processed pdu %s", event_id)
-                            else:
-                                logger.warn("Failed to get PDU %s", event_id)
-                                fetch_state = True
-                        except:
-                            # TODO(erikj): Do some more intelligent retries.
-                            logger.exception("Failed to get PDU")
-                            fetch_state = True
-            else:
-                prevs = {e_id for e_id, _ in pdu.prev_events}
-                seen = set(have_seen.keys())
-                if prevs - seen:
-                    fetch_state = True
-        else:
-            fetch_state = True
+                    have_seen = yield self.store.have_events(
+                        [ev for ev, _ in pdu.prev_events]
+                    )
+
+            prevs = {e_id for e_id, _ in pdu.prev_events}
+            seen = set(have_seen.keys())
+            if prevs - seen:
+                fetch_state = True
 
         if fetch_state:
             # We need to get the state at this event, since we haven't
diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index 7d30c924d1..8f1acbe590 100644
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -287,7 +287,7 @@ class TransactionQueue(object):
                     code = 200
 
                     if response:
-                        for e_id, r in getattr(response, "pdus", {}).items():
+                        for e_id, r in response.get("pdus", {}).items():
                             if "error" in r:
                                 logger.warn(
                                     "Transaction returned error for %s: %s",
diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py
index 8b137e7128..80d03012b7 100644
--- a/synapse/federation/transport/client.py
+++ b/synapse/federation/transport/client.py
@@ -219,3 +219,22 @@ class TransportLayerClient(object):
         )
 
         defer.returnValue(content)
+
+    @defer.inlineCallbacks
+    @log_function
+    def get_missing_events(self, destination, room_id, earliest_events,
+                           latest_events, limit, min_depth):
+        path = PREFIX + "/get_missing_events/%s" % (room_id,)
+
+        content = yield self.client.post_json(
+            destination=destination,
+            path=path,
+            data={
+                "limit": int(limit),
+                "min_depth": int(min_depth),
+                "earliest_events": earliest_events,
+                "latest_events": latest_events,
+            }
+        )
+
+        defer.returnValue(content)
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 2ffb37aa18..ad75c8ddb7 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -234,6 +234,7 @@ class TransportLayerServer(object):
                 )
             )
         )
+
         self.server.register_path(
             "POST",
             re.compile("^" + PREFIX + "/query_auth/([^/]*)/([^/]*)$"),
@@ -245,6 +246,17 @@ class TransportLayerServer(object):
             )
         )
 
+        self.server.register_path(
+            "POST",
+            re.compile("^" + PREFIX + "/get_missing_events/([^/]*)/?$"),
+            self._with_authentication(
+                lambda origin, content, query, room_id:
+                self._get_missing_events(
+                    origin, content, room_id,
+                )
+            )
+        )
+
     @defer.inlineCallbacks
     @log_function
     def _on_send_request(self, origin, content, query, transaction_id):
@@ -344,3 +356,22 @@ class TransportLayerServer(object):
         )
 
         defer.returnValue((200, new_content))
+
+    @defer.inlineCallbacks
+    @log_function
+    def _get_missing_events(self, origin, content, room_id):
+        limit = int(content.get("limit", 10))
+        min_depth = int(content.get("min_depth", 0))
+        earliest_events = content.get("earliest_events", [])
+        latest_events = content.get("latest_events", [])
+
+        content = yield self.request_handler.on_get_missing_events(
+            origin,
+            room_id=room_id,
+            earliest_events=earliest_events,
+            latest_events=latest_events,
+            min_depth=min_depth,
+            limit=limit,
+        )
+
+        defer.returnValue((200, content))
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 26bdc6d1a7..628e62f8b1 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -789,6 +789,29 @@ class FederationHandler(BaseHandler):
 
         defer.returnValue(ret)
 
+    @defer.inlineCallbacks
+    def on_get_missing_events(self, origin, room_id, earliest_events,
+                              latest_events, limit, min_depth):
+        in_room = yield self.auth.check_host_in_room(
+            room_id,
+            origin
+        )
+        if not in_room:
+            raise AuthError(403, "Host not in room.")
+
+        limit = min(limit, 20)
+        min_depth = max(min_depth, 0)
+
+        missing_events = yield self.store.get_missing_events(
+            room_id=room_id,
+            earliest_events=earliest_events,
+            latest_events=latest_events,
+            limit=limit,
+            min_depth=min_depth,
+        )
+
+        defer.returnValue(missing_events)
+
     @defer.inlineCallbacks
     @log_function
     def do_auth(self, origin, event, context, auth_events):
-- 
cgit 1.4.1


From 59362454ddbcb3adf64d81da00dcd85cdb59a2e6 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 26 Feb 2015 15:47:35 +0000
Subject: Must update pending_transactions map before yield'ing

---
 synapse/federation/transaction_queue.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index 8f1acbe590..741a4e7a1a 100644
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -224,6 +224,8 @@ class TransactionQueue(object):
         ]
 
         try:
+            self.pending_transactions[destination] = 1
+
             limiter = yield get_retry_limiter(
                 destination,
                 self._clock,
@@ -239,8 +241,6 @@ class TransactionQueue(object):
                 len(pending_failures)
             )
 
-            self.pending_transactions[destination] = 1
-
             logger.debug("TX [%s] Persisting transaction...", destination)
 
             transaction = Transaction.create_new(
-- 
cgit 1.4.1


From 93d90765c4b09bc870fc91c6ddcf21fd4389659d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 26 Feb 2015 16:15:26 +0000
Subject: Initial implementation of federation server rate limiting

---
 synapse/federation/transport/__init__.py |  12 ++-
 synapse/federation/transport/server.py   | 175 ++++++++++++++++++++++++++++++-
 2 files changed, 182 insertions(+), 5 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/transport/__init__.py b/synapse/federation/transport/__init__.py
index 6800ac46c5..7028ca6947 100644
--- a/synapse/federation/transport/__init__.py
+++ b/synapse/federation/transport/__init__.py
@@ -21,7 +21,7 @@ support HTTPS), however individual pairings of servers may decide to
 communicate over a different (albeit still reliable) protocol.
 """
 
-from .server import TransportLayerServer
+from .server import TransportLayerServer, FederationRateLimiter
 from .client import TransportLayerClient
 
 
@@ -55,8 +55,18 @@ class TransportLayer(TransportLayerServer, TransportLayerClient):
                 send requests
         """
         self.keyring = homeserver.get_keyring()
+        self.clock = homeserver.get_clock()
         self.server_name = server_name
         self.server = server
         self.client = client
         self.request_handler = None
         self.received_handler = None
+
+        self.ratelimiter = FederationRateLimiter(
+            self.clock,
+            window_size=10000,
+            sleep_limit=10,
+            sleep_msec=500,
+            reject_limit=50,
+            concurrent_requests=3,
+        )
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 2ffb37aa18..a9e625f127 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -16,9 +16,11 @@
 from twisted.internet import defer
 
 from synapse.api.urls import FEDERATION_PREFIX as PREFIX
-from synapse.api.errors import Codes, SynapseError
+from synapse.api.errors import Codes, SynapseError, LimitExceededError
+from synapse.util.async import sleep
 from synapse.util.logutils import log_function
 
+import collections
 import logging
 import simplejson as json
 import re
@@ -27,6 +29,163 @@ import re
 logger = logging.getLogger(__name__)
 
 
+class FederationRateLimiter(object):
+    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
+                 reject_limit, concurrent_requests):
+        self.clock = clock
+
+        self.window_size = window_size
+        self.sleep_limit = sleep_limit
+        self.sleep_msec = sleep_msec
+        self.reject_limit = reject_limit
+        self.concurrent_requests = concurrent_requests
+
+        self.ratelimiters = {}
+
+    def ratelimit(self, host):
+        return self.ratelimiters.setdefault(
+            host,
+            PerHostRatelimiter(
+                clock=self.clock,
+                window_size=self.window_size,
+                sleep_limit=self.sleep_limit,
+                sleep_msec=self.sleep_msec,
+                reject_limit=self.reject_limit,
+                concurrent_requests=self.concurrent_requests,
+            )
+        ).ratelimit()
+
+
+class PerHostRatelimiter(object):
+    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
+                 reject_limit, concurrent_requests):
+        self.clock = clock
+
+        self.window_size = window_size
+        self.sleep_limit = sleep_limit
+        self.sleep_msec = sleep_msec
+        self.reject_limit = reject_limit
+        self.concurrent_requests = concurrent_requests
+
+        self.sleeping_requests = set()
+        self.ready_request_queue = collections.OrderedDict()
+        self.current_processing = set()
+        self.request_times = []
+
+    def is_empty(self):
+        time_now = self.clock.time_msec()
+        self.request_times[:] = [
+            r for r in self.request_times
+            if time_now - r < self.window_size
+        ]
+
+        return not (
+            self.ready_request_queue
+            or self.sleeping_requests
+            or self.current_processing
+            or self.request_times
+        )
+
+    def ratelimit(self):
+        request_id = object()
+
+        def on_enter():
+            return self._on_enter(request_id)
+
+        def on_exit(exc_type, exc_val, exc_tb):
+            return self._on_exit(request_id)
+
+        return ContextManagerFunction(on_enter, on_exit)
+
+    def _on_enter(self, request_id):
+        time_now = self.clock.time_msec()
+        self.request_times[:] = [
+            r for r in self.request_times
+            if time_now - r < self.window_size
+        ]
+
+        queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
+        if queue_size > self.reject_limit:
+            raise LimitExceededError(
+                retry_after_ms=int(
+                    self.window_size / self.sleep_limit
+                ),
+            )
+
+        self.request_times.append(time_now)
+
+        def queue_request():
+            if len(self.current_processing) > self.concurrent_requests:
+                logger.debug("Ratelimit [%s]: Queue req", id(request_id))
+                queue_defer = defer.Deferred()
+                self.ready_request_queue[request_id] = queue_defer
+                return queue_defer
+            else:
+                return defer.succeed(None)
+
+        logger.debug("Ratelimit [%s]: len(self.request_times)=%d", id(request_id), len(self.request_times))
+        logger.debug("Ratelimit [%s]: len(self.request_times)=%d", id(request_id), len(self.request_times))
+
+        if len(self.request_times) > self.sleep_limit:
+            logger.debug("Ratelimit [%s]: sleeping req", id(request_id))
+            ret_defer = sleep(self.sleep_msec/1000.0)
+
+            self.sleeping_requests.add(request_id)
+
+            def on_wait_finished(_):
+                logger.debug("Ratelimit [%s]: Finished sleeping", id(request_id))
+                self.sleeping_requests.discard(request_id)
+                queue_defer = queue_request()
+                return queue_defer
+
+            ret_defer.addBoth(on_wait_finished)
+        else:
+            ret_defer = queue_request()
+
+        def on_start(r):
+            logger.debug("Ratelimit [%s]: Processing req", id(request_id))
+            self.current_processing.add(request_id)
+            return r
+
+        def on_err(r):
+            self.current_processing.discard(request_id)
+            return r
+
+        def on_both(r):
+            # Ensure that we've properly cleaned up.
+            self.sleeping_requests.discard(request_id)
+            self.ready_request_queue.pop(request_id, None)
+            return r
+
+        ret_defer.addCallbacks(on_start, on_err)
+        ret_defer.addBoth(on_both)
+        return ret_defer
+
+    def _on_exit(self, request_id):
+        logger.debug("Ratelimit [%s]: Processed req", id(request_id))
+        self.current_processing.discard(request_id)
+        try:
+            request_id, deferred = self.ready_request_queue.popitem()
+            self.current_processing.add(request_id)
+            deferred.callback(None)
+        except KeyError:
+            pass
+
+
+class ContextManagerFunction(object):
+    def __init__(self, on_enter, on_exit):
+        self.on_enter = on_enter
+        self.on_exit = on_exit
+
+    def __enter__(self):
+        if self.on_enter:
+            return self.on_enter()
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.on_exit:
+            return self.on_exit(exc_type, exc_val, exc_tb)
+
+
 class TransportLayerServer(object):
     """Handles incoming federation HTTP requests"""
 
@@ -98,15 +257,23 @@ class TransportLayerServer(object):
         def new_handler(request, *args, **kwargs):
             try:
                 (origin, content) = yield self._authenticate_request(request)
-                response = yield handler(
-                    origin, content, request.args, *args, **kwargs
-                )
+                with self.ratelimiter.ratelimit(origin) as d:
+                    yield d
+                    response = yield handler(
+                        origin, content, request.args, *args, **kwargs
+                    )
             except:
                 logger.exception("_authenticate_request failed")
                 raise
             defer.returnValue(response)
         return new_handler
 
+    def rate_limit_origin(self, handler):
+        def new_handler(origin, *args, **kwargs):
+            response = yield handler(origin, *args, **kwargs)
+            defer.returnValue(response)
+        return new_handler()
+
     @log_function
     def register_received_handler(self, handler):
         """ Register a handler that will be fired when we receive data.
-- 
cgit 1.4.1


From 9dc9118e552bfddaa7579b4ded8b1a0da7eff0e6 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 27 Feb 2015 15:16:47 +0000
Subject: Document FederationRateLimiter

---
 synapse/federation/transport/server.py | 59 +++++++++++++++++++++++++++++-----
 1 file changed, 51 insertions(+), 8 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index a9e625f127..390e54b9fb 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -32,6 +32,21 @@ logger = logging.getLogger(__name__)
 class FederationRateLimiter(object):
     def __init__(self, clock, window_size, sleep_limit, sleep_msec,
                  reject_limit, concurrent_requests):
+        """
+        Args:
+            clock (Clock)
+            window_size (int): The window size in milliseconds.
+            sleep_limit (int): The number of requests received in the last
+                `window_size` milliseconds before we artificially start
+                delaying processing of requests.
+            sleep_msec (int): The number of milliseconds to delay processing
+                of incoming requests by.
+            reject_limit (int): The maximum number of requests that are can be
+                queued for processing before we start rejecting requests with
+                a 429 Too Many Requests response.
+            concurrent_requests (int): The number of concurrent requests to
+                process.
+        """
         self.clock = clock
 
         self.window_size = window_size
@@ -43,9 +58,23 @@ class FederationRateLimiter(object):
         self.ratelimiters = {}
 
     def ratelimit(self, host):
+        """Used to ratelimit an incoming request from given host
+
+        Example usage:
+
+            with rate_limiter.ratelimit(origin) as wait_deferred:
+                yield wait_deferred
+                # Handle request ...
+
+        Args:
+            host (str): Origin of incoming request.
+
+        Returns:
+            _PerHostRatelimiter
+        """
         return self.ratelimiters.setdefault(
             host,
-            PerHostRatelimiter(
+            _PerHostRatelimiter(
                 clock=self.clock,
                 window_size=self.window_size,
                 sleep_limit=self.sleep_limit,
@@ -56,7 +85,7 @@ class FederationRateLimiter(object):
         ).ratelimit()
 
 
-class PerHostRatelimiter(object):
+class _PerHostRatelimiter(object):
     def __init__(self, clock, window_size, sleep_limit, sleep_msec,
                  reject_limit, concurrent_requests):
         self.clock = clock
@@ -123,17 +152,25 @@ class PerHostRatelimiter(object):
             else:
                 return defer.succeed(None)
 
-        logger.debug("Ratelimit [%s]: len(self.request_times)=%d", id(request_id), len(self.request_times))
-        logger.debug("Ratelimit [%s]: len(self.request_times)=%d", id(request_id), len(self.request_times))
+        logger.debug(
+            "Ratelimit [%s]: len(self.request_times)=%d",
+            id(request_id), len(self.request_times),
+        )
 
         if len(self.request_times) > self.sleep_limit:
-            logger.debug("Ratelimit [%s]: sleeping req", id(request_id))
+            logger.debug(
+                "Ratelimit [%s]: sleeping req",
+                id(request_id),
+            )
             ret_defer = sleep(self.sleep_msec/1000.0)
 
             self.sleeping_requests.add(request_id)
 
             def on_wait_finished(_):
-                logger.debug("Ratelimit [%s]: Finished sleeping", id(request_id))
+                logger.debug(
+                    "Ratelimit [%s]: Finished sleeping",
+                    id(request_id),
+                )
                 self.sleeping_requests.discard(request_id)
                 queue_defer = queue_request()
                 return queue_defer
@@ -143,7 +180,10 @@ class PerHostRatelimiter(object):
             ret_defer = queue_request()
 
         def on_start(r):
-            logger.debug("Ratelimit [%s]: Processing req", id(request_id))
+            logger.debug(
+                "Ratelimit [%s]: Processing req",
+                id(request_id),
+            )
             self.current_processing.add(request_id)
             return r
 
@@ -162,7 +202,10 @@ class PerHostRatelimiter(object):
         return ret_defer
 
     def _on_exit(self, request_id):
-        logger.debug("Ratelimit [%s]: Processed req", id(request_id))
+        logger.debug(
+            "Ratelimit [%s]: Processed req",
+            id(request_id),
+        )
         self.current_processing.discard(request_id)
         try:
             request_id, deferred = self.ready_request_queue.popitem()
-- 
cgit 1.4.1


From 0554d0708225afe13d141bd00e3aaca2509f3f78 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 27 Feb 2015 15:41:52 +0000
Subject: Move federation rate limiting out of transport layer

---
 synapse/federation/transport/__init__.py |   4 +-
 synapse/federation/transport/server.py   | 204 +---------------------------
 synapse/util/ratelimitutils.py           | 226 +++++++++++++++++++++++++++++++
 3 files changed, 230 insertions(+), 204 deletions(-)
 create mode 100644 synapse/util/ratelimitutils.py

(limited to 'synapse/federation')

diff --git a/synapse/federation/transport/__init__.py b/synapse/federation/transport/__init__.py
index 7028ca6947..f0283b5105 100644
--- a/synapse/federation/transport/__init__.py
+++ b/synapse/federation/transport/__init__.py
@@ -21,9 +21,11 @@ support HTTPS), however individual pairings of servers may decide to
 communicate over a different (albeit still reliable) protocol.
 """
 
-from .server import TransportLayerServer, FederationRateLimiter
+from .server import TransportLayerServer
 from .client import TransportLayerClient
 
+from synapse.util.ratelimitutils import FederationRateLimiter
+
 
 class TransportLayer(TransportLayerServer, TransportLayerClient):
     """This is a basic implementation of the transport layer that translates
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 390e54b9fb..fce9c0195e 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -16,11 +16,9 @@
 from twisted.internet import defer
 
 from synapse.api.urls import FEDERATION_PREFIX as PREFIX
-from synapse.api.errors import Codes, SynapseError, LimitExceededError
-from synapse.util.async import sleep
+from synapse.api.errors import Codes, SynapseError
 from synapse.util.logutils import log_function
 
-import collections
 import logging
 import simplejson as json
 import re
@@ -29,206 +27,6 @@ import re
 logger = logging.getLogger(__name__)
 
 
-class FederationRateLimiter(object):
-    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
-                 reject_limit, concurrent_requests):
-        """
-        Args:
-            clock (Clock)
-            window_size (int): The window size in milliseconds.
-            sleep_limit (int): The number of requests received in the last
-                `window_size` milliseconds before we artificially start
-                delaying processing of requests.
-            sleep_msec (int): The number of milliseconds to delay processing
-                of incoming requests by.
-            reject_limit (int): The maximum number of requests that are can be
-                queued for processing before we start rejecting requests with
-                a 429 Too Many Requests response.
-            concurrent_requests (int): The number of concurrent requests to
-                process.
-        """
-        self.clock = clock
-
-        self.window_size = window_size
-        self.sleep_limit = sleep_limit
-        self.sleep_msec = sleep_msec
-        self.reject_limit = reject_limit
-        self.concurrent_requests = concurrent_requests
-
-        self.ratelimiters = {}
-
-    def ratelimit(self, host):
-        """Used to ratelimit an incoming request from given host
-
-        Example usage:
-
-            with rate_limiter.ratelimit(origin) as wait_deferred:
-                yield wait_deferred
-                # Handle request ...
-
-        Args:
-            host (str): Origin of incoming request.
-
-        Returns:
-            _PerHostRatelimiter
-        """
-        return self.ratelimiters.setdefault(
-            host,
-            _PerHostRatelimiter(
-                clock=self.clock,
-                window_size=self.window_size,
-                sleep_limit=self.sleep_limit,
-                sleep_msec=self.sleep_msec,
-                reject_limit=self.reject_limit,
-                concurrent_requests=self.concurrent_requests,
-            )
-        ).ratelimit()
-
-
-class _PerHostRatelimiter(object):
-    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
-                 reject_limit, concurrent_requests):
-        self.clock = clock
-
-        self.window_size = window_size
-        self.sleep_limit = sleep_limit
-        self.sleep_msec = sleep_msec
-        self.reject_limit = reject_limit
-        self.concurrent_requests = concurrent_requests
-
-        self.sleeping_requests = set()
-        self.ready_request_queue = collections.OrderedDict()
-        self.current_processing = set()
-        self.request_times = []
-
-    def is_empty(self):
-        time_now = self.clock.time_msec()
-        self.request_times[:] = [
-            r for r in self.request_times
-            if time_now - r < self.window_size
-        ]
-
-        return not (
-            self.ready_request_queue
-            or self.sleeping_requests
-            or self.current_processing
-            or self.request_times
-        )
-
-    def ratelimit(self):
-        request_id = object()
-
-        def on_enter():
-            return self._on_enter(request_id)
-
-        def on_exit(exc_type, exc_val, exc_tb):
-            return self._on_exit(request_id)
-
-        return ContextManagerFunction(on_enter, on_exit)
-
-    def _on_enter(self, request_id):
-        time_now = self.clock.time_msec()
-        self.request_times[:] = [
-            r for r in self.request_times
-            if time_now - r < self.window_size
-        ]
-
-        queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
-        if queue_size > self.reject_limit:
-            raise LimitExceededError(
-                retry_after_ms=int(
-                    self.window_size / self.sleep_limit
-                ),
-            )
-
-        self.request_times.append(time_now)
-
-        def queue_request():
-            if len(self.current_processing) > self.concurrent_requests:
-                logger.debug("Ratelimit [%s]: Queue req", id(request_id))
-                queue_defer = defer.Deferred()
-                self.ready_request_queue[request_id] = queue_defer
-                return queue_defer
-            else:
-                return defer.succeed(None)
-
-        logger.debug(
-            "Ratelimit [%s]: len(self.request_times)=%d",
-            id(request_id), len(self.request_times),
-        )
-
-        if len(self.request_times) > self.sleep_limit:
-            logger.debug(
-                "Ratelimit [%s]: sleeping req",
-                id(request_id),
-            )
-            ret_defer = sleep(self.sleep_msec/1000.0)
-
-            self.sleeping_requests.add(request_id)
-
-            def on_wait_finished(_):
-                logger.debug(
-                    "Ratelimit [%s]: Finished sleeping",
-                    id(request_id),
-                )
-                self.sleeping_requests.discard(request_id)
-                queue_defer = queue_request()
-                return queue_defer
-
-            ret_defer.addBoth(on_wait_finished)
-        else:
-            ret_defer = queue_request()
-
-        def on_start(r):
-            logger.debug(
-                "Ratelimit [%s]: Processing req",
-                id(request_id),
-            )
-            self.current_processing.add(request_id)
-            return r
-
-        def on_err(r):
-            self.current_processing.discard(request_id)
-            return r
-
-        def on_both(r):
-            # Ensure that we've properly cleaned up.
-            self.sleeping_requests.discard(request_id)
-            self.ready_request_queue.pop(request_id, None)
-            return r
-
-        ret_defer.addCallbacks(on_start, on_err)
-        ret_defer.addBoth(on_both)
-        return ret_defer
-
-    def _on_exit(self, request_id):
-        logger.debug(
-            "Ratelimit [%s]: Processed req",
-            id(request_id),
-        )
-        self.current_processing.discard(request_id)
-        try:
-            request_id, deferred = self.ready_request_queue.popitem()
-            self.current_processing.add(request_id)
-            deferred.callback(None)
-        except KeyError:
-            pass
-
-
-class ContextManagerFunction(object):
-    def __init__(self, on_enter, on_exit):
-        self.on_enter = on_enter
-        self.on_exit = on_exit
-
-    def __enter__(self):
-        if self.on_enter:
-            return self.on_enter()
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self.on_exit:
-            return self.on_exit(exc_type, exc_val, exc_tb)
-
-
 class TransportLayerServer(object):
     """Handles incoming federation HTTP requests"""
 
diff --git a/synapse/util/ratelimitutils.py b/synapse/util/ratelimitutils.py
new file mode 100644
index 0000000000..259d5f6f88
--- /dev/null
+++ b/synapse/util/ratelimitutils.py
@@ -0,0 +1,226 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from twisted.internet import defer
+
+from synapse.api.errors import LimitExceededError
+
+from synapse.util.async import sleep
+
+import collections
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class FederationRateLimiter(object):
+    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
+                 reject_limit, concurrent_requests):
+        """
+        Args:
+            clock (Clock)
+            window_size (int): The window size in milliseconds.
+            sleep_limit (int): The number of requests received in the last
+                `window_size` milliseconds before we artificially start
+                delaying processing of requests.
+            sleep_msec (int): The number of milliseconds to delay processing
+                of incoming requests by.
+            reject_limit (int): The maximum number of requests that are can be
+                queued for processing before we start rejecting requests with
+                a 429 Too Many Requests response.
+            concurrent_requests (int): The number of concurrent requests to
+                process.
+        """
+        self.clock = clock
+
+        self.window_size = window_size
+        self.sleep_limit = sleep_limit
+        self.sleep_msec = sleep_msec
+        self.reject_limit = reject_limit
+        self.concurrent_requests = concurrent_requests
+
+        self.ratelimiters = {}
+
+    def ratelimit(self, host):
+        """Used to ratelimit an incoming request from given host
+
+        Example usage:
+
+            with rate_limiter.ratelimit(origin) as wait_deferred:
+                yield wait_deferred
+                # Handle request ...
+
+        Args:
+            host (str): Origin of incoming request.
+
+        Returns:
+            _PerHostRatelimiter
+        """
+        return self.ratelimiters.setdefault(
+            host,
+            _PerHostRatelimiter(
+                clock=self.clock,
+                window_size=self.window_size,
+                sleep_limit=self.sleep_limit,
+                sleep_msec=self.sleep_msec,
+                reject_limit=self.reject_limit,
+                concurrent_requests=self.concurrent_requests,
+            )
+        ).ratelimit()
+
+
+class _PerHostRatelimiter(object):
+    def __init__(self, clock, window_size, sleep_limit, sleep_msec,
+                 reject_limit, concurrent_requests):
+        self.clock = clock
+
+        self.window_size = window_size
+        self.sleep_limit = sleep_limit
+        self.sleep_msec = sleep_msec
+        self.reject_limit = reject_limit
+        self.concurrent_requests = concurrent_requests
+
+        self.sleeping_requests = set()
+        self.ready_request_queue = collections.OrderedDict()
+        self.current_processing = set()
+        self.request_times = []
+
+    def is_empty(self):
+        time_now = self.clock.time_msec()
+        self.request_times[:] = [
+            r for r in self.request_times
+            if time_now - r < self.window_size
+        ]
+
+        return not (
+            self.ready_request_queue
+            or self.sleeping_requests
+            or self.current_processing
+            or self.request_times
+        )
+
+    def ratelimit(self):
+        request_id = object()
+
+        def on_enter():
+            return self._on_enter(request_id)
+
+        def on_exit(exc_type, exc_val, exc_tb):
+            return self._on_exit(request_id)
+
+        return ContextManagerFunction(on_enter, on_exit)
+
+    def _on_enter(self, request_id):
+        time_now = self.clock.time_msec()
+        self.request_times[:] = [
+            r for r in self.request_times
+            if time_now - r < self.window_size
+        ]
+
+        queue_size = len(self.ready_request_queue) + len(self.sleeping_requests)
+        if queue_size > self.reject_limit:
+            raise LimitExceededError(
+                retry_after_ms=int(
+                    self.window_size / self.sleep_limit
+                ),
+            )
+
+        self.request_times.append(time_now)
+
+        def queue_request():
+            if len(self.current_processing) > self.concurrent_requests:
+                logger.debug("Ratelimit [%s]: Queue req", id(request_id))
+                queue_defer = defer.Deferred()
+                self.ready_request_queue[request_id] = queue_defer
+                return queue_defer
+            else:
+                return defer.succeed(None)
+
+        logger.debug(
+            "Ratelimit [%s]: len(self.request_times)=%d",
+            id(request_id), len(self.request_times),
+        )
+
+        if len(self.request_times) > self.sleep_limit:
+            logger.debug(
+                "Ratelimit [%s]: sleeping req",
+                id(request_id),
+            )
+            ret_defer = sleep(self.sleep_msec/1000.0)
+
+            self.sleeping_requests.add(request_id)
+
+            def on_wait_finished(_):
+                logger.debug(
+                    "Ratelimit [%s]: Finished sleeping",
+                    id(request_id),
+                )
+                self.sleeping_requests.discard(request_id)
+                queue_defer = queue_request()
+                return queue_defer
+
+            ret_defer.addBoth(on_wait_finished)
+        else:
+            ret_defer = queue_request()
+
+        def on_start(r):
+            logger.debug(
+                "Ratelimit [%s]: Processing req",
+                id(request_id),
+            )
+            self.current_processing.add(request_id)
+            return r
+
+        def on_err(r):
+            self.current_processing.discard(request_id)
+            return r
+
+        def on_both(r):
+            # Ensure that we've properly cleaned up.
+            self.sleeping_requests.discard(request_id)
+            self.ready_request_queue.pop(request_id, None)
+            return r
+
+        ret_defer.addCallbacks(on_start, on_err)
+        ret_defer.addBoth(on_both)
+        return ret_defer
+
+    def _on_exit(self, request_id):
+        logger.debug(
+            "Ratelimit [%s]: Processed req",
+            id(request_id),
+        )
+        self.current_processing.discard(request_id)
+        try:
+            request_id, deferred = self.ready_request_queue.popitem()
+            self.current_processing.add(request_id)
+            deferred.callback(None)
+        except KeyError:
+            pass
+
+
+class ContextManagerFunction(object):
+    def __init__(self, on_enter, on_exit):
+        self.on_enter = on_enter
+        self.on_exit = on_exit
+
+    def __enter__(self):
+        if self.on_enter:
+            return self.on_enter()
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.on_exit:
+            return self.on_exit(exc_type, exc_val, exc_tb)
-- 
cgit 1.4.1


From 9d9b230501915c326136567349b0995623c48a21 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 2 Mar 2015 11:33:45 +0000
Subject: Make the federation server ratelimiting configurable.

---
 synapse/config/ratelimiting.py           | 36 ++++++++++++++++++++++++++++++++
 synapse/federation/transport/__init__.py | 10 ++++-----
 2 files changed, 41 insertions(+), 5 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py
index 17c7e64ce7..862c07ef8c 100644
--- a/synapse/config/ratelimiting.py
+++ b/synapse/config/ratelimiting.py
@@ -22,6 +22,12 @@ class RatelimitConfig(Config):
         self.rc_messages_per_second = args.rc_messages_per_second
         self.rc_message_burst_count = args.rc_message_burst_count
 
+        self.federation_rc_window_size = args.federation_rc_window_size
+        self.federation_rc_sleep_limit = args.federation_rc_sleep_limit
+        self.federation_rc_sleep_delay = args.federation_rc_sleep_delay
+        self.federation_rc_reject_limit = args.federation_rc_reject_limit
+        self.federation_rc_concurrent = args.federation_rc_concurrent
+
     @classmethod
     def add_arguments(cls, parser):
         super(RatelimitConfig, cls).add_arguments(parser)
@@ -34,3 +40,33 @@ class RatelimitConfig(Config):
             "--rc-message-burst-count", type=float, default=10,
             help="number of message a client can send before being throttled"
         )
+
+        rc_group.add_argument(
+            "--federation-rc-window-size", type=int, default=10000,
+            help="The federation window size in milliseconds",
+        )
+
+        rc_group.add_argument(
+            "--federation-rc-sleep-limit", type=int, default=10,
+            help="The number of federation requests from a single server"
+                 " in a window before the server will delay processing the"
+                 " request.",
+        )
+
+        rc_group.add_argument(
+            "--federation-rc-sleep-delay", type=int, default=500,
+            help="The duration in milliseconds to delay processing events from"
+                 " remote servers by if they go over the sleep limit.",
+        )
+
+        rc_group.add_argument(
+            "--federation-rc-reject-limit", type=int, default=50,
+            help="The maximum number of concurrent federation requests allowed"
+                 " from a single server",
+        )
+
+        rc_group.add_argument(
+            "--federation-rc-concurrent", type=int, default=3,
+            help="The number of federation requests to concurrently process"
+                 " from a single server",
+        )
diff --git a/synapse/federation/transport/__init__.py b/synapse/federation/transport/__init__.py
index f0283b5105..2a671b9aec 100644
--- a/synapse/federation/transport/__init__.py
+++ b/synapse/federation/transport/__init__.py
@@ -66,9 +66,9 @@ class TransportLayer(TransportLayerServer, TransportLayerClient):
 
         self.ratelimiter = FederationRateLimiter(
             self.clock,
-            window_size=10000,
-            sleep_limit=10,
-            sleep_msec=500,
-            reject_limit=50,
-            concurrent_requests=3,
+            window_size=homeserver.config.federation_rc_window_size,
+            sleep_limit=homeserver.config.federation_rc_sleep_limit,
+            sleep_msec=homeserver.config.federation_rc_sleep_delay,
+            reject_limit=homeserver.config.federation_rc_reject_limit,
+            concurrent_requests=homeserver.config.federation_rc_concurrent,
         )
-- 
cgit 1.4.1


From 23d9bd1d745a037202bb9a134cdb848eb65a01e9 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 2 Mar 2015 11:39:40 +0000
Subject: Process transactions serially.

Since the events received in a transaction are ordered, later events
might depend on earlier events and so we shouldn't blindly process them
in parellel.
---
 synapse/federation/federation_server.py | 24 ++++++++++--------------
 1 file changed, 10 insertions(+), 14 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 22b9663831..7ee37fb34d 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -112,17 +112,23 @@ class FederationServer(FederationBase):
         logger.debug("[%s] Transaction is new", transaction.transaction_id)
 
         with PreserveLoggingContext():
-            dl = []
+            results = []
+
             for pdu in pdu_list:
                 d = self._handle_new_pdu(transaction.origin, pdu)
 
                 def handle_failure(failure):
                     failure.trap(FederationError)
                     self.send_failure(failure.value, transaction.origin)
+                    return failure
 
                 d.addErrback(handle_failure)
 
-                dl.append(d)
+                try:
+                    yield d
+                    results.append({})
+                except Exception as e:
+                    results.append({"error": str(e)})
 
             if hasattr(transaction, "edus"):
                 for edu in [Edu(**x) for x in transaction.edus]:
@@ -135,21 +141,11 @@ class FederationServer(FederationBase):
             for failure in getattr(transaction, "pdu_failures", []):
                 logger.info("Got failure %r", failure)
 
-            results = yield defer.DeferredList(dl, consumeErrors=True)
-
-        ret = []
-        for r in results:
-            if r[0]:
-                ret.append({})
-            else:
-                logger.exception(r[1])
-                ret.append({"error": str(r[1].value)})
-
-        logger.debug("Returning: %s", str(ret))
+        logger.debug("Returning: %s", str(results))
 
         response = {
             "pdus": dict(zip(
-                (p.event_id for p in pdu_list), ret
+                (p.event_id for p in pdu_list), results
             )),
         }
 
-- 
cgit 1.4.1


From 29481690c5b296a1c8aee3068d32ef083ef227f3 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 2 Mar 2015 11:50:43 +0000
Subject: If we're yielding don't add errback

---
 synapse/federation/federation_server.py | 10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 7ee37fb34d..bc9bac809a 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -117,16 +117,12 @@ class FederationServer(FederationBase):
             for pdu in pdu_list:
                 d = self._handle_new_pdu(transaction.origin, pdu)
 
-                def handle_failure(failure):
-                    failure.trap(FederationError)
-                    self.send_failure(failure.value, transaction.origin)
-                    return failure
-
-                d.addErrback(handle_failure)
-
                 try:
                     yield d
                     results.append({})
+                except FederationError as e:
+                    self.send_failure(e, transaction.origin)
+                    results.append({"error": str(e)})
                 except Exception as e:
                     results.append({"error": str(e)})
 
-- 
cgit 1.4.1


From ae702d161ab6d518caa91759ec6bdec01b11954f Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 5 Mar 2015 16:08:02 +0000
Subject: Handle if get_missing_pdu returns 400 or not all events.

---
 synapse/federation/federation_client.py | 109 ++++++++++++++++++++++++++++----
 synapse/federation/federation_server.py |   6 +-
 2 files changed, 100 insertions(+), 15 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index ca89a0787c..b87c8a3bbb 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -19,14 +19,18 @@ from twisted.internet import defer
 from .federation_base import FederationBase
 from .units import Edu
 
-from synapse.api.errors import CodeMessageException, SynapseError
+from synapse.api.errors import (
+    CodeMessageException, HttpResponseException, SynapseError,
+)
 from synapse.util.expiringcache import ExpiringCache
 from synapse.util.logutils import log_function
 from synapse.events import FrozenEvent
 
 from synapse.util.retryutils import get_retry_limiter, NotRetryingDestination
 
+import itertools
 import logging
+import random
 
 
 logger = logging.getLogger(__name__)
@@ -440,21 +444,100 @@ class FederationClient(FederationBase):
         defer.returnValue(ret)
 
     @defer.inlineCallbacks
-    def get_missing_events(self, destination, room_id, earliest_events,
+    def get_missing_events(self, destination, room_id, earliest_events_ids,
                            latest_events, limit, min_depth):
-        content = yield self.transport_layer.get_missing_events(
-            destination, room_id, earliest_events, latest_events, limit,
-            min_depth,
-        )
+        try:
+            content = yield self.transport_layer.get_missing_events(
+                destination=destination,
+                room_id=room_id,
+                earliest_events=earliest_events_ids,
+                latest_events=[e.event_id for e in latest_events],
+                limit=limit,
+                min_depth=min_depth,
+            )
+
+            events = [
+                self.event_from_pdu_json(e)
+                for e in content.get("events", [])
+            ]
+
+            signed_events = yield self._check_sigs_and_hash_and_fetch(
+                destination, events, outlier=True
+            )
+
+            have_gotten_all_from_destination = True
+        except HttpResponseException as e:
+            if not e.code == 400:
+                raise
 
-        events = [
-            self.event_from_pdu_json(e)
-            for e in content.get("events", [])
-        ]
+            signed_events = []
+            have_gotten_all_from_destination = False
 
-        signed_events = yield self._check_sigs_and_hash_and_fetch(
-            destination, events, outlier=True
-        )
+        if len(signed_events) >= limit:
+            defer.returnValue(signed_events)
+
+        servers = yield self.store.get_joined_hosts_for_room(room_id)
+
+        servers = set(servers)
+        servers.discard(self.server_name)
+
+        failed_to_fetch = set()
+
+        while len(signed_events) < limit:
+            # Are we missing any?
+
+            seen_events = set(earliest_events_ids)
+            seen_events.update(e.event_id for e in signed_events)
+
+            missing_events = {}
+            for e in itertools.chain(latest_events, signed_events):
+                missing_events.update({
+                    e_id: e.depth for e_id, _ in e.prev_events
+                    if e_id not in seen_events and e_id not in failed_to_fetch
+                })
+
+            if not missing_events:
+                break
+
+            have_seen = yield self.store.have_events(missing_events)
+
+            for k in have_seen:
+                missing_events.pop(k, None)
+
+            if not missing_events:
+                break
+
+            # Okay, we haven't gotten everything yet. Lets get them.
+            ordered_missing = sorted(missing_events.items(), key=lambda x: x[0])
+
+            if have_gotten_all_from_destination:
+                servers.discard(destination)
+
+            def random_server_list():
+                srvs = list(servers)
+                random.shuffle(srvs)
+                return srvs
+
+            deferreds = [
+                self.get_pdu(
+                    destinations=random_server_list(),
+                    event_id=e_id,
+                )
+                for e_id, depth in ordered_missing[:limit - len(signed_events)]
+            ]
+
+            got_a_new_event = False
+
+            res = yield defer.DeferredList(deferreds, consumeErrors=True)
+            for (result, val), (e_id, _) in zip(res, ordered_missing):
+                if result:
+                    signed_events.append(val)
+                    got_a_new_event = True
+                else:
+                    failed_to_fetch.add(e_id)
+
+            if not got_a_new_event:
+                break
 
         defer.returnValue(signed_events)
 
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 4264d857be..dd4ca74ba6 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -413,12 +413,14 @@ class FederationServer(FederationBase):
                     missing_events = yield self.get_missing_events(
                         origin,
                         pdu.room_id,
-                        earliest_events=list(latest),
-                        latest_events=[pdu.event_id],
+                        earliest_events_ids=list(latest),
+                        latest_events=[pdu],
                         limit=10,
                         min_depth=min_depth,
                     )
 
+                    missing_events.sort(key=lambda x: x.depth)
+
                     for e in missing_events:
                         yield self._handle_new_pdu(
                             origin,
-- 
cgit 1.4.1


From 6dfd8c73fcdd727cd6589513e2b8059f779623ae Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 5 Mar 2015 16:31:13 +0000
Subject: Docs.

---
 synapse/federation/federation_client.py | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index b87c8a3bbb..11e2753fed 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -446,6 +446,20 @@ class FederationClient(FederationBase):
     @defer.inlineCallbacks
     def get_missing_events(self, destination, room_id, earliest_events_ids,
                            latest_events, limit, min_depth):
+        """Tries to fetch events we are missing. This is called when we receive
+        an event without having received all of its ancestors.
+
+        Args:
+            destination (str)
+            room_id (str)
+            earliest_events_ids (list): List of event ids. Effectively the
+                events we expected to receive, but haven't. `get_missing_events`
+                should only return events that didn't happen before these.
+            latest_events (list): List of events we have received that we don't
+                have all previous events for.
+            limit (int): Maximum number of events to return.
+            min_depth (int): Minimum depth of events tor return.
+        """
         try:
             content = yield self.transport_layer.get_missing_events(
                 destination=destination,
@@ -470,6 +484,8 @@ class FederationClient(FederationBase):
             if not e.code == 400:
                 raise
 
+            # We are probably hitting an old server that doesn't support
+            # get_missing_events
             signed_events = []
             have_gotten_all_from_destination = False
 
-- 
cgit 1.4.1


From 39aa968a764816632a05ac0e3cf9c865b7a3a68d Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 5 Mar 2015 16:31:32 +0000
Subject: Respect min_depth argument

---
 synapse/federation/federation_client.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 11e2753fed..75b6a7b46a 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -507,10 +507,12 @@ class FederationClient(FederationBase):
 
             missing_events = {}
             for e in itertools.chain(latest_events, signed_events):
-                missing_events.update({
-                    e_id: e.depth for e_id, _ in e.prev_events
-                    if e_id not in seen_events and e_id not in failed_to_fetch
-                })
+                if e.depth > min_depth:
+                    missing_events.update({
+                        e_id: e.depth for e_id, _ in e.prev_events
+                        if e_id not in seen_events
+                        and e_id not in failed_to_fetch
+                    })
 
             if not missing_events:
                 break
-- 
cgit 1.4.1


From 96fee64421a534787e9316a61ab407b43c782dc7 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 5 Mar 2015 16:31:47 +0000
Subject: Remove unecessary check

---
 synapse/federation/federation_client.py | 6 ------
 1 file changed, 6 deletions(-)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 75b6a7b46a..f131941f45 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -544,19 +544,13 @@ class FederationClient(FederationBase):
                 for e_id, depth in ordered_missing[:limit - len(signed_events)]
             ]
 
-            got_a_new_event = False
-
             res = yield defer.DeferredList(deferreds, consumeErrors=True)
             for (result, val), (e_id, _) in zip(res, ordered_missing):
                 if result:
                     signed_events.append(val)
-                    got_a_new_event = True
                 else:
                     failed_to_fetch.add(e_id)
 
-            if not got_a_new_event:
-                break
-
         defer.returnValue(signed_events)
 
     def event_from_pdu_json(self, pdu_json, outlier=False):
-- 
cgit 1.4.1


From 9708f49abfb5fa48c1190364093ab4ce5c4e6f23 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 5 Mar 2015 16:35:16 +0000
Subject: Docs

---
 synapse/federation/federation_server.py | 2 ++
 1 file changed, 2 insertions(+)

(limited to 'synapse/federation')

diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index dd4ca74ba6..9c7dcdba96 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -419,6 +419,8 @@ class FederationServer(FederationBase):
                         min_depth=min_depth,
                     )
 
+                    # We want to sort these by depth so we process them and
+                    # tell clients about them in order.
                     missing_events.sort(key=lambda x: x.depth)
 
                     for e in missing_events:
-- 
cgit 1.4.1