Merge pull request #1088 from matrix-org/erikj/transaction_queue_check

Correctly guard against multiple concurrent transactions

synapse/federation/transaction_queue.py

@@ -170,8 +170,6 @@ class TransactionQueue(object):
 
     @defer.inlineCallbacks
     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
@@ -184,6 +182,12 @@ class TransactionQueue(object):
             )
             return
 
+        try:
+            self.pending_transactions[destination] = 1
+
+            yield run_on_reactor()
+
+            while True:
                     pending_pdus = self.pending_pdus_by_dest.pop(destination, [])
                     pending_edus = self.pending_edus_by_dest.pop(destination, [])
                     pending_failures = self.pending_failures_by_dest.pop(destination, [])
@@ -200,7 +204,9 @@ class TransactionQueue(object):
 
                     if not pending_pdus and not pending_edus and not pending_failures:
                         logger.debug("TX [%s] Nothing to send", destination)
-                self.last_device_stream_id_by_dest[destination] = device_stream_id
+                        self.last_device_stream_id_by_dest[destination] = (
+                            device_stream_id
+                        )
                         return
 
                     yield self._send_new_transaction(
@@ -208,6 +214,9 @@ class TransactionQueue(object):
                         device_stream_id,
                         should_delete_from_device_stream=bool(device_message_edus)
                     )
+        finally:
+            # We want to be *very* sure we delete this after we stop processing
+            self.pending_transactions.pop(destination, None)
 
     @defer.inlineCallbacks
     def _get_new_device_messages(self, destination):
@@ -240,8 +249,6 @@ class TransactionQueue(object):
             failures = [x.get_dict() for x in pending_failures]
 
             try:
-                self.pending_transactions[destination] = 1
-
                 logger.debug("TX [%s] _attempt_new_transaction", destination)
 
                 txn_id = str(self._next_txn_id)
@@ -375,7 +382,3 @@ class TransactionQueue(object):
 
                 for p in pdus:
                     logger.info("Failed to send event %s to %s", p.event_id, destination)
-
-            finally:
-                # We want to be *very* sure we delete this after we stop processing
-                self.pending_transactions.pop(destination, None)