Merge pull request #2989 from lbryio/blocking-wallet-send

add `--blocking` to `account_send` and `wallet_send`

lbry/extras/daemon/daemon.py

@@ -1459,7 +1459,7 @@ class Daemon(metaclass=JSONRPCServerType):
     @requires(WALLET_COMPONENT)
     async def jsonrpc_wallet_send(
             self, amount, addresses, wallet_id=None,
-            change_account_id=None, funding_account_ids=None, preview=False):
+            change_account_id=None, funding_account_ids=None, preview=False, blocking=True):
         """
         Send the same number of credits to multiple addresses using all accounts in wallet to
         fund the transaction and the default account to receive any change.
@@ -1467,12 +1467,14 @@ class Daemon(metaclass=JSONRPCServerType):
         Usage:
             wallet_send <amount> <addresses>... [--wallet_id=<wallet_id>] [--preview]
                         [--change_account_id=None] [--funding_account_ids=<funding_account_ids>...]
+                        [--blocking]
 
         Options:
             --wallet_id=<wallet_id>         : (str) restrict operation to specific wallet
             --change_account_id=<wallet_id> : (str) account where change will go
             --funding_account_ids=<funding_account_ids> : (str) accounts to fund the transaction
-            --preview                  : (bool) do not broadcast the transaction
+            --preview                       : (bool) do not broadcast the transaction
+            --blocking                      : (bool) wait until tx has synced
 
         Returns: {Transaction}
         """
@@ -1498,13 +1500,11 @@ class Daemon(metaclass=JSONRPCServerType):
         tx = await Transaction.create(
             [], outputs, accounts, account
         )
-
         if not preview:
-            await self.ledger.broadcast(tx)
+            await self.broadcast_or_release(tx, blocking)
             self.component_manager.loop.create_task(self.analytics_manager.send_credits_sent())
         else:
             await self.ledger.release_tx(tx)
-
         return tx
 
     ACCOUNT_DOC = """
@@ -1782,24 +1782,26 @@ class Daemon(metaclass=JSONRPCServerType):
         )
 
     @requires(WALLET_COMPONENT)
-    def jsonrpc_account_send(self, amount, addresses, account_id=None, wallet_id=None, preview=False):
+    def jsonrpc_account_send(self, amount, addresses, account_id=None, wallet_id=None, preview=False, blocking=False):
         """
         Send the same number of credits to multiple addresses from a specific account (or default account).
 
         Usage:
             account_send <amount> <addresses>... [--account_id=<account_id>] [--wallet_id=<wallet_id>] [--preview]
+                                                 [--blocking]
 
         Options:
             --account_id=<account_id>  : (str) account to fund the transaction
             --wallet_id=<wallet_id>    : (str) restrict operation to specific wallet
             --preview                  : (bool) do not broadcast the transaction
+            --blocking                 : (bool) wait until tx has synced
 
         Returns: {Transaction}
         """
         return self.jsonrpc_wallet_send(
             amount=amount, addresses=addresses, wallet_id=wallet_id,
             change_account_id=account_id, funding_account_ids=[account_id] if account_id else [],
-            preview=preview
+            preview=preview, blocking=blocking
         )
 
     SYNC_DOC = """

lbry/wallet/ledger.py

@@ -133,7 +133,7 @@ class Ledger(metaclass=LedgerRegistry):
         self._on_transaction_controller = StreamController()
         self.on_transaction = self._on_transaction_controller.stream
         self.on_transaction.listen(
-            lambda e: log.debug(
+            lambda e: log.info(
                 '(%s) on_transaction: address=%s, height=%s, is_verified=%s, tx.id=%s',
                 self.get_id(), e.address, e.tx.height, e.tx.is_verified, e.tx.id
             )
@@ -682,20 +682,26 @@ class Ledger(metaclass=LedgerRegistry):
                 address_record['address']
             )) for address_record in records
         ], timeout=1)
-        if pending:
-            records = await self.db.get_addresses(address__in=addresses)
-            for record in records:
-                found = False
-                local_history = (await self.get_local_status_and_history(
-                    record['address'], history=record['history']
-                ))[1] if record['history'] else []
-                for txid, local_height in local_history:
-                    if txid == tx.id and local_height >= height:
-                        found = True
-                if not found:
-                    log.debug("timeout: %s, %s, %s", record['history'], addresses, tx.id)
+        if not pending:
+            return True
+        records = await self.db.get_addresses(address__in=addresses)
+        for record in records:
+            local_history = (await self.get_local_status_and_history(
+                record['address'], history=record['history']
+            ))[1] if record['history'] else []
+            for txid, local_height in local_history:
+                if txid == tx.id:
+                    if local_height >= height:
+                        return True
+                    log.warning(
+                        "local history has higher height than remote for %s (%i vs %i)", txid,
+                        local_height, height
+                    )
                     return False
-        return True
+            log.warning(
+                "local history does not contain %s, requested height %i", tx.id, height
+            )
+        return False
 
     async def _inflate_outputs(
             self, query, accounts,