@defer.inlineCallbacks
    @log_function
    def _attempt_new_transaction(self, destination):
        # list of (pending_pdu, deferred, order)
        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.
            # we need application-layer timeouts of some flavour of these
            # requests
            logger.debug(
                "TX [%s] Transaction already in progress",
                destination
            )
            return

        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, [])

        if pending_pdus:
            logger.debug("TX [%s] len(pending_pdus_by_dest[dest]) = %d",
                         destination, len(pending_pdus))

        if not pending_pdus and not pending_edus and not pending_failures:
            logger.debug("TX [%s] Nothing to send", destination)
            return

        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(
                destination,
                self._clock,
                self.store,
            )

            logger.debug(
                "TX [%s] {%s} Attempting new transaction"
                " (pdus: %d, edus: %d, failures: %d)",
                destination, txn_id,
                len(pending_pdus),
                len(pending_edus),
                len(pending_failures)
            )

            logger.debug("TX [%s] Persisting transaction...", destination)

            transaction = Transaction.create_new(
                origin_server_ts=int(self._clock.time_msec()),
                transaction_id=txn_id,
                origin=self.server_name,
                destination=destination,
                pdus=pdus,
                edus=edus,
                pdu_failures=failures,
            )

            self._next_txn_id += 1

            yield self.transaction_actions.prepare_to_send(transaction)

            logger.debug("TX [%s] Persisted transaction", destination)
            logger.info(
                "TX [%s] {%s} Sending transaction [%s],"
                " (PDUs: %d, EDUs: %d, failures: %d)",
                destination, txn_id,
                transaction.transaction_id,
                len(pending_pdus),
                len(pending_edus),
                len(pending_failures),
            )

            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 data in "pdus":
                        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 response.get("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] {%s} got %d response",
                    destination, txn_id, 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:
                    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 - "
                "dropping transaction for now",
                destination,
            )
        except RuntimeError as e:
            # We capture this here as there as nothing actually listens
            # for this finishing functions deferred.
            logger.warn(
                "TX [%s] Problem in _attempt_transaction: %s",
                destination,
                e,
            )
        except Exception as e:
            # We capture this here as there as nothing actually listens
            # for this finishing functions deferred.
            logger.warn(
                "TX [%s] Problem in _attempt_transaction: %s",
                destination,
                e,
            )

            for deferred in deferreds:
                if not deferred.called:
                    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)
dataset/ETHPy150Open matrix-org/synapse/synapse/federation/transaction_queue.py/TransactionQueue._attempt_new_transaction/swapped operands of `in`