Merge branch 'release-v0.7.1' of github.com:matrix-org/synapse

1 files changed, 120 insertions, 96 deletions

diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py
index 731019ad9f..7d30c924d1 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
 
@@ -63,6 +66,26 @@ class TransactionQueue(object):
         # HACK to get unique tx id
         self._next_txn_id = int(self._clock.time_msec())
 
+    def can_send_to(self, destination):
+        """Can we send messages to the given server?
+
+        We can't send messages to ourselves. If we are running on localhost
+        then we can only federation with other servers running on localhost.
+        Otherwise we only federate with servers on a public domain.
+
+        Args:
+            destination(str): The server we are possibly trying to send to.
+        Returns:
+            bool: True if we can send to the server.
+        """
+
+        if destination == self.server_name:
+            return False
+        if self.server_name.startswith("localhost"):
+            return destination.startswith("localhost")
+        else:
+            return not destination.startswith("localhost")
+
     @defer.inlineCallbacks
     @log_function
     def enqueue_pdu(self, pdu, destinations, order):
@@ -71,8 +94,9 @@ class TransactionQueue(object):
         # table and we'll get back to it later.
 
         destinations = set(destinations)
-        destinations.discard(self.server_name)
-        destinations.discard("localhost")
+        destinations = set(
+            dest for dest in destinations if self.can_send_to(dest)
+        )
 
         logger.debug("Sending to: %s", str(destinations))
 
@@ -87,24 +111,27 @@ class TransactionQueue(object):
                 (pdu, deferred, order)
             )
 
-            def eb(failure):
+            def chain(failure):
                 if not deferred.called:
                     deferred.errback(failure)
-                else:
-                    logger.warn("Failed to send pdu", failure)
+
+            def log_failure(failure):
+                logger.warn("Failed to send pdu", failure.value)
+
+            deferred.addErrback(log_failure)
 
             with PreserveLoggingContext():
-                self._attempt_new_transaction(destination).addErrback(eb)
+                self._attempt_new_transaction(destination).addErrback(chain)
 
             deferreds.append(deferred)
 
-        yield defer.DeferredList(deferreds)
+        yield defer.DeferredList(deferreds, consumeErrors=True)
 
     # NO inlineCallbacks
     def enqueue_edu(self, edu):
         destination = edu.destination
 
-        if destination == self.server_name:
+        if not self.can_send_to(destination):
             return
 
         deferred = defer.Deferred()
@@ -112,51 +139,53 @@ class TransactionQueue(object):
             (edu, deferred)
         )
 
-        def eb(failure):
+        def chain(failure):
             if not deferred.called:
                 deferred.errback(failure)
-            else:
-                logger.warn("Failed to send edu", failure)
+
+        def log_failure(failure):
+            logger.warn("Failed to send pdu", failure.value)
+
+        deferred.addErrback(log_failure)
 
         with PreserveLoggingContext():
-            self._attempt_new_transaction(destination).addErrback(eb)
+            self._attempt_new_transaction(destination).addErrback(chain)
 
         return deferred
 
     @defer.inlineCallbacks
     def enqueue_failure(self, failure, destination):
+        if destination == self.server_name or destination == "localhost":
+            return
+
         deferred = defer.Deferred()
 
+        if not self.can_send_to(destination):
+            return
+
         self.pending_failures_by_dest.setdefault(
             destination, []
         ).append(
             (failure, deferred)
         )
 
+        def chain(f):
+            if not deferred.called:
+                deferred.errback(f)
+
+        def log_failure(f):
+            logger.warn("Failed to send pdu", f.value)
+
+        deferred.addErrback(log_failure)
+
+        with PreserveLoggingContext():
+            self._attempt_new_transaction(destination).addErrback(chain)
+
         yield deferred
 
     @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.
@@ -183,15 +212,6 @@ class TransactionQueue(object):
             logger.info("TX [%s] Nothing to send", destination)
             return
 
-        logger.debug(
-            "TX [%s] Attempting new transaction"
-            " (pdus: %d, edus: %d, failures: %d)",
-            destination,
-            len(pending_pdus),
-            len(pending_edus),
-            len(pending_failures)
-        )
-
         # Sort based on the order field
         pending_pdus.sort(key=lambda t: t[2])
 
@@ -204,6 +224,21 @@ class TransactionQueue(object):
         ]
 
         try:
+            limiter = yield get_retry_limiter(
+                destination,
+                self._clock,
+                self.store,
+            )
+
+            logger.debug(
+                "TX [%s] Attempting new transaction"
+                " (pdus: %d, edus: %d, failures: %d)",
+                destination,
+                len(pending_pdus),
+                len(pending_edus),
+                len(pending_failures)
+            )
+
             self.pending_transactions[destination] = 1
 
             logger.debug("TX [%s] Persisting transaction...", destination)
@@ -229,52 +264,56 @@ class TransactionQueue(object):
                 transaction.transaction_id,
             )
 
-            # 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.info("TX [%s] got %d response", destination, code)
-
-            logger.debug("TX [%s] Sent transaction", destination)
-            logger.debug("TX [%s] Marking as delivered...", destination)
+            with limiter:
+                # 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
+
+                    if response:
+                        for e_id, r in getattr(response, "pdus", {}).items():
+                            if "error" in r:
+                                logger.warn(
+                                    "Transaction returned error for %s: %s",
+                                    e_id, r,
+                                )
+                except HttpResponseException as e:
+                    code = e.code
+                    response = e.response
+
+                logger.info("TX [%s] got %d response", destination, code)
+
+                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] 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 +324,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.
@@ -296,14 +341,12 @@ class TransactionQueue(object):
         except Exception as e:
             # We capture this here as there as nothing actually listens
             # for this finishing functions deferred.
-            logger.exception(
+            logger.warn(
                 "TX [%s] Problem in _attempt_transaction: %s",
                 destination,
                 e,
             )
 
-            self.set_retrying(destination, retry_interval)
-
             for deferred in deferreds:
                 if not deferred.called:
                     deferred.errback(e)
@@ -314,22 +357,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
-        )