Clean up TransactionQueue

synapse/federation/transaction_queue.py

@@ -21,8 +21,7 @@ from .units import Transaction
 
 from synapse.api.errors import HttpResponseException
 from synapse.util.async import run_on_reactor
-from synapse.util.logutils import log_function
+from synapse.util.logcontext import preserve_context_over_fn
-from synapse.util.logcontext import PreserveLoggingContext
 from synapse.util.retryutils import (
     get_retry_limiter, NotRetryingDestination,
 )
@@ -120,90 +119,46 @@ class TransactionQueue(object):
         if not destinations:
             return
 
-        deferreds = []
-
         for destination in destinations:
-            deferred = defer.Deferred()
             self.pending_pdus_by_dest.setdefault(destination, []).append(
-                (pdu, deferred, order)
+                (pdu, order)
             )
 
-            def chain(failure):
+            preserve_context_over_fn(
-                if not deferred.called:
+                self._attempt_new_transaction, destination
-                    deferred.errback(failure)
+            )
 
-            def log_failure(f):
-                logger.warn("Failed to send pdu to %s: %s", destination, f.value)
-
-            deferred.addErrback(log_failure)
-
-            with PreserveLoggingContext():
-                self._attempt_new_transaction(destination).addErrback(chain)
-
-            deferreds.append(deferred)
-
-    # NO inlineCallbacks
     def enqueue_edu(self, edu):
         destination = edu.destination
 
         if not self.can_send_to(destination):
             return
 
-        deferred = defer.Deferred()
+        self.pending_edus_by_dest.setdefault(destination, []).append(edu)
-        self.pending_edus_by_dest.setdefault(destination, []).append(
+
-            (edu, deferred)
+        preserve_context_over_fn(
+            self._attempt_new_transaction, destination
         )
 
-        def chain(failure):
-            if not deferred.called:
-                deferred.errback(failure)
-
-        def log_failure(f):
-            logger.warn("Failed to send edu to %s: %s", destination, f.value)
-
-        deferred.addErrback(log_failure)
-
-        with PreserveLoggingContext():
-            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(
+        ).append(failure)
-            (failure, deferred)
+
+        preserve_context_over_fn(
+            self._attempt_new_transaction, destination
         )
 
-        def chain(f):
-            if not deferred.called:
-                deferred.errback(f)
-
-        def log_failure(f):
-            logger.warn("Failed to send failure to %s: %s", destination, f.value)
-
-        deferred.addErrback(log_failure)
-
-        with PreserveLoggingContext():
-            self._attempt_new_transaction(destination).addErrback(chain)
-
-        yield deferred
-
-    @measure_func("attempt_new_transaction")
     @defer.inlineCallbacks
-    @log_function
     def _attempt_new_transaction(self, destination):
         yield run_on_reactor()
-
+        while True:
             # list of (pending_pdu, deferred, order)
             if destination in self.pending_transactions:
                 # XXX: pending_transactions can get stuck on by a never-ending
@@ -228,22 +183,26 @@ class TransactionQueue(object):
                 logger.debug("TX [%s] Nothing to send", destination)
                 return
 
+            yield self._send_new_transaction(
+                destination, pending_pdus, pending_edus, pending_failures
+            )
+
+    @measure_func("_send_new_transaction")
+    @defer.inlineCallbacks
+    def _send_new_transaction(self, destination, pending_pdus, pending_edus,
+                              pending_failures):
+
+            # Sort based on the order field
+            pending_pdus.sort(key=lambda t: t[1])
+            pdus = [x[0] for x in pending_pdus]
+            edus = pending_edus
+            failures = [x.get_dict() for x in pending_failures]
+
             try:
                 self.pending_transactions[destination] = 1
 
                 logger.debug("TX [%s] _attempt_new_transaction", destination)
 
-            # Sort based on the order field
-            pending_pdus.sort(key=lambda t: t[2])
-
-            pdus = [x[0] for x in pending_pdus]
-            edus = [x[0] for x in pending_edus]
-            failures = [x[0].get_dict() for x in pending_failures]
-            deferreds = [
-                x[1]
-                for x in pending_pdus + pending_edus + pending_failures
-            ]
-
                 txn_id = str(self._next_txn_id)
 
                 limiter = yield get_retry_limiter(
@@ -335,22 +294,9 @@ class TransactionQueue(object):
 
                 logger.debug("TX [%s] Marked as delivered", destination)
 
-            logger.debug("TX [%s] Yielding to callbacks...", destination)
+                if code != 200:
-
+                    for p in pdus:
-            for deferred in deferreds:
+                        logger.info("Failed to send event %s to %s", p.event_id, destination)
-                if code == 200:
-                    deferred.callback(None)
-                else:
-                    deferred.errback(RuntimeError("Got status %d" % code))
-
-                # Ensures we don't continue until all callbacks on that
-                # deferred have fired
-                try:
-                    yield deferred
-                except:
-                    pass
-
-            logger.debug("TX [%s] Yielded to callbacks", destination)
             except NotRetryingDestination:
                 logger.info(
                     "TX [%s] not ready for retry yet - "
@@ -365,6 +311,9 @@ class TransactionQueue(object):
                     destination,
                     e,
                 )
+
+                for p in pdus:
+                    logger.info("Failed to send event %s to %s", p.event_id, destination)
             except Exception as e:
                 # We capture this here as there as nothing actually listens
                 # for this finishing functions deferred.
@@ -374,13 +323,9 @@ class TransactionQueue(object):
                     e,
                 )
 
-            for deferred in deferreds:
+                for p in pdus:
-                if not deferred.called:
+                    logger.info("Failed to send event %s to %s", p.event_id, destination)
-                    deferred.errback(e)
 
             finally:
                 # We want to be *very* sure we delete this after we stop processing
                 self.pending_transactions.pop(destination, None)
-
-            # Check to see if there is anything else to send.
-            self._attempt_new_transaction(destination)

synapse/http/matrixfederationclient.py

@@ -155,9 +155,7 @@ class MatrixFederationHttpClient(object):
                             time_out=timeout / 1000. if timeout else 60,
                         )
 
-                    response = yield preserve_context_over_fn(
+                    response = yield preserve_context_over_fn(send_request)
-                        send_request,
-                    )
 
                     log_result = "%d %s" % (response.code, response.phrase,)
                     break

synapse/util/logcontext.py

@@ -317,7 +317,6 @@ def preserve_fn(f):
     def g(*args, **kwargs):
         with PreserveLoggingContext(current):
             return f(*args, **kwargs)
-
     return g
 
 

synapse/util/metrics.py

@@ -78,7 +78,6 @@ class Measure(object):
         self.start = self.clock.time_msec()
         self.start_context = LoggingContext.current_context()
         if not self.start_context:
-            logger.warn("Entered Measure without log context: %s", self.name)
             self.start_context = LoggingContext("Measure")
             self.start_context.__enter__()
             self.created_context = True
@@ -99,7 +98,7 @@ class Measure(object):
         if context != self.start_context:
             logger.warn(
                 "Context has unexpectedly changed from '%s' to '%s'. (%r)",
-                context, self.start_context, self.name
+                self.start_context, context, self.name
             )
             return