wip sync fixing

tests/client_tests/integration/test_transactions.py

@@ -4,10 +4,45 @@ from torba.testcase import IntegrationTestCase
 from torba.client.constants import COIN
 
 
+log = logging.getLogger(__name__)
+log.setLevel(logging.INFO)
+
+
 class BasicTransactionTests(IntegrationTestCase):
 
     VERBOSITY = logging.WARN
 
+    async def test_stressing(self):
+        await self.blockchain.generate(1000)
+        await self.assertBalance(self.account, '0.0')
+        address1 = await self.account.receiving.get_or_create_usable_address()
+        hash1 = self.ledger.address_to_hash160(address1)
+
+        tasks = []
+        for _ in range(10):
+            sendtxid = await self.blockchain.send_to_address(address1, 100)
+            tasks.append(self.on_transaction_id(sendtxid))
+        await asyncio.wait(tasks)
+        await self.assertBalance(self.account, '1000.0')
+
+        tasks = []
+        for _ in range(10):
+            tx = await self.ledger.transaction_class.create(
+                [],
+                [self.ledger.transaction_class.output_class.pay_pubkey_hash(1*COIN, hash1)],
+                [self.account], self.account
+            )
+            await self.broadcast(tx)
+            tasks.append(asyncio.create_task(self.ledger.wait(tx)))
+
+        await asyncio.wait(tasks)
+
+        #await asyncio.sleep(5)
+
+        await self.assertBalance(self.account, '1000.0')
+
+        await self.blockchain.generate(1)
+
     async def test_sending_and_receiving(self):
         account1, account2 = self.account, self.wallet.generate_account(self.ledger)
         await self.ledger.subscribe_account(account2)

torba/client/baseledger.py

@@ -500,7 +500,7 @@ class BaseLedger(metaclass=LedgerRegistry):
     def broadcast(self, tx):
         return self.network.broadcast(hexlify(tx.raw).decode())
 
-    async def wait(self, tx: basetransaction.BaseTransaction, height=0):
+    async def wait(self, tx: basetransaction.BaseTransaction, height=-1):
         addresses = set()
         for txi in tx.inputs:
             if txi.txo_ref.txo is not None:

torba/client/basetransaction.py

@@ -247,7 +247,7 @@ class BaseTransaction:
     output_class = BaseOutput
 
     def __init__(self, raw=None, version: int = 1, locktime: int = 0, is_verified: bool = False,
-                 height: int = -1, position: int = -1) -> None:
+                 height: int = -2, position: int = -1) -> None:
         self._raw = raw
         self.ref = TXRefMutable(self)
         self.version = version
@@ -255,11 +255,28 @@ class BaseTransaction:
         self._inputs: List[BaseInput] = []
         self._outputs: List[BaseOutput] = []
         self.is_verified = is_verified
+        # Height Progression
+        #   -2: not broadcast
+        #   -1: in mempool but has unconfirmed inputs
+        #    0: in mempool and all inputs confirmed
+        # +num: confirmed in a specific block (height)
         self.height = height
         self.position = position
         if raw is not None:
             self._deserialize()
 
+    @property
+    def is_broadcast(self):
+        return self.height > -2
+
+    @property
+    def is_mempool(self):
+        return self.height in (-1, 0)
+
+    @property
+    def is_confirmed(self):
+        return self.height > 0
+
     @property
     def id(self):
         return self.ref.id

torba/orchstr8/node.py

@@ -49,6 +49,7 @@ def get_blockchain_node_from_ledger(ledger_module):
 
 def set_logging(ledger_module, level):
     logging.getLogger('torba').setLevel(level)
+    logging.getLogger('torba.client').setLevel(logging.INFO)
     logging.getLogger('torba.server').setLevel(level)
     #logging.getLogger('asyncio').setLevel(level)
     logging.getLogger('blockchain').setLevel(level)
@@ -213,8 +214,6 @@ class BlockchainProcess(asyncio.SubprocessProtocol):
             raise SystemError(data.decode('ascii'))
         elif b'Done loading' in data:
             self.ready.set()
-        elif b'Shutdown: done' in data:
-            self.stopped.set()
 
     def process_exited(self):
         self.stopped.set()
@@ -298,6 +297,7 @@ class BlockchainNode:
         try:
             self.transport.terminate()
             await self.protocol.stopped.wait()
+            self.transport.close()
         finally:
             if cleanup:
                 self.cleanup()

torba/testcase.py

@@ -1,6 +1,8 @@
 import sys
 import logging
 import unittest
+import asyncio
+from asyncio.runners import _cancel_all_tasks  # type: ignore
 from unittest.case import _Outcome
 from typing import Optional
 from torba.orchstr8 import Conductor
@@ -12,17 +14,34 @@ from torba.client.wallet import Wallet
 from torba.client.util import satoshis_to_coins
 
 
-try:
-    import asyncio
-    from asyncio.runners import _cancel_all_tasks  # type: ignore
-except ImportError:
-    import asyncio
+class ColorHandler(logging.StreamHandler):
 
-    # this is only available in py3.7
-    def _cancel_all_tasks(loop):
-        pass
+    level_color = {
+        logging.DEBUG: "black",
+        logging.INFO: "black",
+        logging.WARNING: "yellow",
+        logging.ERROR: "red"
+    }
 
-HANDLER = logging.StreamHandler(sys.stdout)
+    color_code = dict(
+        black=30, red=31, green=32, yellow=33,
+        blue=34, magenta=35, cyan=36, white=37
+    )
+
+    def emit(self, record):
+        try:
+            msg = self.format(record)
+            color_name = self.level_color.get(record.levelno, "black")
+            color_code = self.color_code[color_name]
+            stream = self.stream
+            stream.write('\x1b[%sm%s\x1b[0m' % (color_code, msg))
+            stream.write(self.terminator)
+            self.flush()
+        except Exception:
+            self.handleError(record)
+
+
+HANDLER = ColorHandler(sys.stdout)
 HANDLER.setFormatter(
     logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 )