Add per server retry limiting.

Factor out the pre destination retry logic from TransactionQueue so it
can be reused in both get_pdu and crypto.keyring

4 files changed, 205 insertions, 122 deletions

diff --git a/synapse/crypto/keyring.py b/synapse/crypto/keyring.py
index 3250cff595..ea00c830c0 100644
--- a/synapse/crypto/keyring.py
+++ b/synapse/crypto/keyring.py
@@ -22,6 +22,8 @@ from syutil.crypto.signing_key import (
 from syutil.base64util import decode_base64, encode_base64
 from synapse.api.errors import SynapseError, Codes
 
+from synapse.util.retryutils import get_retry_limiter
+
 from OpenSSL import crypto
 
 import logging
@@ -88,19 +90,13 @@ class Keyring(object):
 
         # Try to fetch the key from the remote server.
 
-        retry_last_ts, retry_interval = (0, 0)
-        retry_timings = yield self.store.get_destination_retry_timings(
-            server_name
+        limiter = yield get_retry_limiter(
+            server_name,
+            self.clock,
+            self.store,
         )
-        if retry_timings:
-            retry_last_ts, retry_interval = (
-                retry_timings.retry_last_ts, retry_timings.retry_interval
-            )
-            if retry_last_ts + retry_interval > int(self.clock.time_msec()):
-                logger.info("%s not ready for retry", server_name)
-                raise ValueError("No verification key found for given key ids")
 
-        try:
+        with limiter:
             (response, tls_certificate) = yield fetch_server_key(
                 server_name, self.hs.tls_context_factory
             )
@@ -165,7 +161,3 @@ class Keyring(object):
                     return
 
             raise ValueError("No verification key found for given key ids")
-
-        except:
-            self.set_retrying(server_name, retry_interval)
-            raise
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index 70c9a6f46b..c5b0274c2f 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -23,6 +23,8 @@ from synapse.api.errors import CodeMessageException
 from synapse.util.logutils import log_function
 from synapse.events import FrozenEvent
 
+from synapse.util.retryutils import get_retry_limiter, NotRetryingDestination
+
 import logging
 
 
@@ -163,24 +165,34 @@ class FederationClient(FederationBase):
         pdu = None
         for destination in destinations:
             try:
-                transaction_data = yield self.transport_layer.get_event(
-                    destination, event_id
+                limiter = yield get_retry_limiter(
+                    destination,
+                    self._clock,
+                    self.store,
                 )
 
-                logger.debug("transaction_data %r", transaction_data)
+                with limiter:
+                    transaction_data = yield self.transport_layer.get_event(
+                        destination, event_id
+                    )
 
-                pdu_list = [
-                    self.event_from_pdu_json(p, outlier=outlier)
-                    for p in transaction_data["pdus"]
-                ]
+                    logger.debug("transaction_data %r", transaction_data)
 
-                if pdu_list:
-                    pdu = pdu_list[0]
+                    pdu_list = [
+                        self.event_from_pdu_json(p, outlier=outlier)
+                        for p in transaction_data["pdus"]
+                    ]
 
-                    # Check signatures are correct.
-                    pdu = yield self._check_sigs_and_hash(pdu)
+                    if pdu_list:
+                        pdu = pdu_list[0]
 
-                    break
+                        # Check signatures are correct.
+                        pdu = yield self._check_sigs_and_hash(pdu)
+
+                        break
+            except NotRetryingDestination as e:
+                logger.info(e.message)
+                continue
             except CodeMessageException:
                 raise
             except Exception as e:
diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index bb20f2ebab..7d02afe163 100644
--- a/synapse/federation/transaction_queue.py
+++ b/synapse/federation/transaction_queue.py
@@ -22,6 +22,9 @@ from .units import Transaction
 from synapse.api.errors import HttpResponseException
 from synapse.util.logutils import log_function
 from synapse.util.logcontext import PreserveLoggingContext
+from synapse.util.retryutils import (
+    get_retry_limiter, NotRetryingDestination,
+)
 
 import logging
 
@@ -138,25 +141,6 @@ class TransactionQueue(object):
     @defer.inlineCallbacks
     @log_function
     def _attempt_new_transaction(self, destination):
-
-        (retry_last_ts, retry_interval) = (0, 0)
-        retry_timings = yield self.store.get_destination_retry_timings(
-            destination
-        )
-        if retry_timings:
-            (retry_last_ts, retry_interval) = (
-                retry_timings.retry_last_ts, retry_timings.retry_interval
-            )
-            if retry_last_ts + retry_interval > int(self._clock.time_msec()):
-                logger.info(
-                    "TX [%s] not ready for retry yet - "
-                    "dropping transaction for now",
-                    destination,
-                )
-                return
-            else:
-                logger.info("TX [%s] is ready for retry", destination)
-
         if destination in self.pending_transactions:
             # XXX: pending_transactions can get stuck on by a never-ending
             # request at which point pending_pdus_by_dest just keeps growing.
@@ -204,77 +188,79 @@ class TransactionQueue(object):
         ]
 
         try:
-            self.pending_transactions[destination] = 1
-
-            logger.debug("TX [%s] Persisting transaction...", destination)
-
-            transaction = Transaction.create_new(
-                origin_server_ts=int(self._clock.time_msec()),
-                transaction_id=str(self._next_txn_id),
-                origin=self.server_name,
-                destination=destination,
-                pdus=pdus,
-                edus=edus,
-                pdu_failures=failures,
+            limiter = yield get_retry_limiter(
+                destination,
+                self._clock,
+                self.store,
             )
 
-            self._next_txn_id += 1
+            with limiter:
+                self.pending_transactions[destination] = 1
 
-            yield self.transaction_actions.prepare_to_send(transaction)
+                logger.debug("TX [%s] Persisting transaction...", destination)
 
-            logger.debug("TX [%s] Persisted transaction", destination)
-            logger.info(
-                "TX [%s] Sending transaction [%s]",
-                destination,
-                transaction.transaction_id,
-            )
+                transaction = Transaction.create_new(
+                    origin_server_ts=int(self._clock.time_msec()),
+                    transaction_id=str(self._next_txn_id),
+                    origin=self.server_name,
+                    destination=destination,
+                    pdus=pdus,
+                    edus=edus,
+                    pdu_failures=failures,
+                )
+
+                self._next_txn_id += 1
 
-            # Actually send the transaction
-
-            # FIXME (erikj): This is a bit of a hack to make the Pdu age
-            # keys work
-            def json_data_cb():
-                data = transaction.get_dict()
-                now = int(self._clock.time_msec())
-                if "pdus" in data:
-                    for p in data["pdus"]:
-                        if "age_ts" in p:
-                            unsigned = p.setdefault("unsigned", {})
-                            unsigned["age"] = now - int(p["age_ts"])
-                            del p["age_ts"]
-                return data
-
-            try:
-                response = yield self.transport_layer.send_transaction(
-                    transaction, json_data_cb
+                yield self.transaction_actions.prepare_to_send(transaction)
+
+                logger.debug("TX [%s] Persisted transaction", destination)
+                logger.info(
+                    "TX [%s] Sending transaction [%s]",
+                    destination,
+                    transaction.transaction_id,
                 )
-                code = 200
-            except HttpResponseException as e:
-                code = e.code
-                response = e.response
 
-            logger.info("TX [%s] got %d response", destination, code)
+                # Actually send the transaction
+
+                # FIXME (erikj): This is a bit of a hack to make the Pdu age
+                # keys work
+                def json_data_cb():
+                    data = transaction.get_dict()
+                    now = int(self._clock.time_msec())
+                    if "pdus" in data:
+                        for p in data["pdus"]:
+                            if "age_ts" in p:
+                                unsigned = p.setdefault("unsigned", {})
+                                unsigned["age"] = now - int(p["age_ts"])
+                                del p["age_ts"]
+                    return data
+
+                try:
+                    response = yield self.transport_layer.send_transaction(
+                        transaction, json_data_cb
+                    )
+                    code = 200
+                except HttpResponseException as e:
+                    code = e.code
+                    response = e.response
 
-            logger.debug("TX [%s] Sent transaction", destination)
-            logger.debug("TX [%s] Marking as delivered...", destination)
+                logger.info("TX [%s] got %d response", destination, code)
 
-            yield self.transaction_actions.delivered(
-                transaction, code, response
-            )
+                logger.debug("TX [%s] Sent transaction", destination)
+                logger.debug("TX [%s] Marking as delivered...", destination)
+
+                yield self.transaction_actions.delivered(
+                    transaction, code, response
+                )
+
+                logger.debug("TX [%s] Marked as delivered", destination)
 
-            logger.debug("TX [%s] Marked as delivered", destination)
             logger.debug("TX [%s] Yielding to callbacks...", destination)
 
             for deferred in deferreds:
                 if code == 200:
-                    if retry_last_ts:
-                        # this host is alive! reset retry schedule
-                        yield self.store.set_destination_retry_timings(
-                            destination, 0, 0
-                        )
                     deferred.callback(None)
                 else:
-                    self.set_retrying(destination, retry_interval)
                     deferred.errback(RuntimeError("Got status %d" % code))
 
                 # Ensures we don't continue until all callbacks on that
@@ -285,6 +271,12 @@ class TransactionQueue(object):
                     pass
 
             logger.debug("TX [%s] Yielded to callbacks", destination)
+        except NotRetryingDestination:
+            logger.info(
+                "TX [%s] not ready for retry yet - "
+                "dropping transaction for now",
+                destination,
+            )
         except RuntimeError as e:
             # We capture this here as there as nothing actually listens
             # for this finishing functions deferred.
@@ -302,8 +294,6 @@ class TransactionQueue(object):
                 e,
             )
 
-            self.set_retrying(destination, retry_interval)
-
             for deferred in deferreds:
                 if not deferred.called:
                     deferred.errback(e)
@@ -314,22 +304,3 @@ class TransactionQueue(object):
 
             # Check to see if there is anything else to send.
             self._attempt_new_transaction(destination)
-
-    @defer.inlineCallbacks
-    def set_retrying(self, destination, retry_interval):
-        # track that this destination is having problems and we should
-        # give it a chance to recover before trying it again
-
-        if retry_interval:
-            retry_interval *= 2
-            # plateau at hourly retries for now
-            if retry_interval >= 60 * 60 * 1000:
-                retry_interval = 60 * 60 * 1000
-        else:
-            retry_interval = 2000  # try again at first after 2 seconds
-
-        yield self.store.set_destination_retry_timings(
-            destination,
-            int(self._clock.time_msec()),
-            retry_interval
-        )
diff --git a/synapse/util/retryutils.py b/synapse/util/retryutils.py
new file mode 100644
index 0000000000..bba524ebd8
--- /dev/null
+++ b/synapse/util/retryutils.py
@@ -0,0 +1,108 @@
+# -*- 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
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class NotRetryingDestination(Exception):
+    def __init__(self, retry_last_ts, retry_interval, destination):
+        msg = "Not retrying server %s." % (destination,)
+        super(NotRetryingDestination, self).__init__(msg)
+
+        self.retry_last_ts = retry_last_ts
+        self.retry_interval = retry_interval
+        self.destination = destination
+
+
+@defer.inlineCallbacks
+def get_retry_limiter(destination, clock, store, **kwargs):
+    retry_last_ts, retry_interval = (0, 0)
+
+    retry_timings = yield store.get_destination_retry_timings(
+        destination
+    )
+
+    if retry_timings:
+        retry_last_ts, retry_interval = (
+            retry_timings.retry_last_ts, retry_timings.retry_interval
+        )
+
+        now = int(clock.time_msec())
+
+        if retry_last_ts + retry_interval > now:
+            raise NotRetryingDestination(
+                retry_last_ts=retry_last_ts,
+                retry_interval=retry_interval,
+                destination=destination,
+            )
+
+    defer.returnValue(
+        RetryDestinationLimiter(
+            destination,
+            clock,
+            store,
+            retry_interval,
+            **kwargs
+        )
+    )
+
+
+class RetryDestinationLimiter(object):
+    def __init__(self, destination, clock, store, retry_interval,
+                 min_retry_interval=20000, max_retry_interval=60 * 60 * 1000,
+                 multiplier_retry_interval=2):
+        self.clock = clock
+        self.store = store
+        self.destination = destination
+
+        self.retry_interval = retry_interval
+        self.min_retry_interval = min_retry_interval
+        self.max_retry_interval = max_retry_interval
+        self.multiplier_retry_interval = multiplier_retry_interval
+
+    def __enter__(self):
+        pass
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        def err(self, failure):
+            logger.exception(
+                "Failed to store set_destination_retry_timings",
+                failure.value
+            )
+
+        if exc_type is None and exc_val is None and exc_tb is None:
+            # We connected successfully.
+            retry_last_ts = 0
+            self.retry_interval = 0
+        else:
+            # We couldn't connect.
+            if self.retry_interval:
+                self.retry_interval *= self.multiplier_retry_interval
+
+                if self.retry_interval >= self.max_retry_interval:
+                    self.retry_interval = self.max_retry_interval
+            else:
+                self.retry_interval = self.min_retry_interval
+
+            retry_last_ts = int(self._clock.time_msec()),
+
+        self.store.set_destination_retry_timings(
+            self.destination, retry_last_ts, self.retry_interval
+        ).addErrback(err)