added ledger.on_ready to know when ledger is finished syncing

lbry/tests/integration/test_resolve_command.py

@@ -90,6 +90,7 @@ class ResolveCommand(BaseResolveTestCase):
         await self.conductor.spv_node.stop()
         resolving_future = asyncio.ensure_future(self.resolve('foo'))
         await self.conductor.spv_node.start(self.conductor.blockchain_node)
+        await self.ledger.on_ready.first
         self.assertIsNotNone((await resolving_future)['foo']['claim_id'])
 
     async def test_winning_by_effective_amount(self):

torba/torba/client/baseledger.py

@@ -137,6 +137,9 @@ class BaseLedger(metaclass=LedgerRegistry):
             )
         )
 
+        self._on_ready_controller = StreamController()
+        self.on_ready = self._on_ready_controller.stream
+
         self._tx_cache = pylru.lrucache(100000)
         self._update_tasks = TaskGroup()
         self._utxo_reservation_lock = asyncio.Lock()
@@ -291,6 +294,7 @@ class BaseLedger(metaclass=LedgerRegistry):
             await self.update_headers()
         await self.subscribe_accounts()
         await self._update_tasks.done.wait()
+        self._on_ready_controller.add(True)
 
     async def stop(self):
         self._update_tasks.cancel()

torba/torba/tasks.py

@@ -17,7 +17,8 @@ class TaskGroup:
 
     def _remove(self, task):
         self._tasks.remove(task)
-        len(self._tasks) < 1 and self.done.set()
+        if len(self._tasks) < 1:
+            self.done.set()
 
     def cancel(self):
         for task in self._tasks: