Refactor the wallet server/hub database use only leveldb (drop sqlite) and safely handle reorgs

.github/workflows/main.yml

@@ -78,7 +78,11 @@ jobs:
         test:
           - datanetwork
           - blockchain
-          - blockchain_legacy_search
+          - claims
+          - takeovers
+          - transactions
+          - claims_legacy_search
+          - takeovers_legacy_search
           - other
     steps:
       - name: Configure sysctl limits

.gitignore

@@ -13,7 +13,7 @@ __pycache__
 _trial_temp/
 trending*.log
 
-/tests/integration/blockchain/files
+/tests/integration/claims/files
 /tests/.coverage.*
 
 /lbry/wallet/bin

lbry/error/__init__.py

@@ -252,9 +252,10 @@ def __init__(self, url):
 
 class ResolveCensoredError(WalletError):
 
-    def __init__(self, url, censor_id):
+    def __init__(self, url, censor_id, censor_row):
         self.url = url
         self.censor_id = censor_id
+        self.censor_row = censor_row
         super().__init__(f"Resolve of '{url}' was censored by channel with claim id '{censor_id}'.")
 
 

lbry/extras/daemon/daemon.py

@@ -2282,7 +2282,7 @@ async def jsonrpc_purchase_create(
         accounts = wallet.get_accounts_or_all(funding_account_ids)
         txo = None
         if claim_id:
-            txo = await self.ledger.get_claim_by_claim_id(accounts, claim_id, include_purchase_receipt=True)
+            txo = await self.ledger.get_claim_by_claim_id(claim_id, accounts, include_purchase_receipt=True)
             if not isinstance(txo, Output) or not txo.is_claim:
                 # TODO: use error from lbry.error
                 raise Exception(f"Could not find claim with claim_id '{claim_id}'.")
@@ -3616,7 +3616,7 @@ async def jsonrpc_stream_update(
             claim_address = old_txo.get_address(account.ledger)
 
         channel = None
-        if channel_id or channel_name:
+        if not clear_channel and (channel_id or channel_name):
             channel = await self.get_channel_or_error(
                 wallet, channel_account_id, channel_id, channel_name, for_signing=True)
         elif old_txo.claim.is_signed and not clear_channel and not replace:
@@ -3646,11 +3646,13 @@ async def jsonrpc_stream_update(
         else:
             claim = Claim.from_bytes(old_txo.claim.to_bytes())
             claim.stream.update(file_path=file_path, **kwargs)
+        if clear_channel:
+            claim.clear_signature()
         tx = await Transaction.claim_update(
-            old_txo, claim, amount, claim_address, funding_accounts, funding_accounts[0], channel
+            old_txo, claim, amount, claim_address, funding_accounts, funding_accounts[0],
+            channel if not clear_channel else None
         )
         new_txo = tx.outputs[0]
-
         stream_hash = None
         if not preview:
             old_stream = self.file_manager.get_filtered(sd_hash=old_txo.claim.stream.source.sd_hash)
@@ -4148,7 +4150,7 @@ async def jsonrpc_collection_resolve(
         wallet = self.wallet_manager.get_wallet_or_default(wallet_id)
 
         if claim_id:
-            txo = await self.ledger.get_claim_by_claim_id(wallet.accounts, claim_id)
+            txo = await self.ledger.get_claim_by_claim_id(claim_id, wallet.accounts)
             if not isinstance(txo, Output) or not txo.is_claim:
                 # TODO: use error from lbry.error
                 raise Exception(f"Could not find collection with claim_id '{claim_id}'.")
@@ -4215,7 +4217,7 @@ async def jsonrpc_support_create(
         funding_accounts = wallet.get_accounts_or_all(funding_account_ids)
         channel = await self.get_channel_or_none(wallet, channel_account_id, channel_id, channel_name, for_signing=True)
         amount = self.get_dewies_or_error("amount", amount)
-        claim = await self.ledger.get_claim_by_claim_id(wallet.accounts, claim_id)
+        claim = await self.ledger.get_claim_by_claim_id(claim_id)
         claim_address = claim.get_address(self.ledger)
         if not tip:
             account = wallet.get_account_or_default(account_id)

lbry/schema/result.py

@@ -1,23 +1,27 @@
 import base64
-import struct
-from typing import List
+from typing import List, TYPE_CHECKING, Union, Optional
 from binascii import hexlify
 from itertools import chain
 
 from lbry.error import ResolveCensoredError
 from lbry.schema.types.v2.result_pb2 import Outputs as OutputsMessage
 from lbry.schema.types.v2.result_pb2 import Error as ErrorMessage
+if TYPE_CHECKING:
+    from lbry.wallet.server.leveldb import ResolveResult
 
 INVALID = ErrorMessage.Code.Name(ErrorMessage.INVALID)
 NOT_FOUND = ErrorMessage.Code.Name(ErrorMessage.NOT_FOUND)
 BLOCKED = ErrorMessage.Code.Name(ErrorMessage.BLOCKED)
 
 
-def set_reference(reference, txo_row):
-    if txo_row:
-        reference.tx_hash = txo_row['txo_hash'][:32]
-        reference.nout = struct.unpack('<I', txo_row['txo_hash'][32:])[0]
-        reference.height = txo_row['height']
+def set_reference(reference, claim_hash, rows):
+    if claim_hash:
+        for txo in rows:
+            if claim_hash == txo.claim_hash:
+                reference.tx_hash = txo.tx_hash
+                reference.nout = txo.position
+                reference.height = txo.height
+                return
 
 
 class Censor:
@@ -38,19 +42,19 @@ def is_censored(self, row):
     def apply(self, rows):
         return [row for row in rows if not self.censor(row)]
 
-    def censor(self, row) -> bool:
+    def censor(self, row) -> Optional[bytes]:
         if self.is_censored(row):
             censoring_channel_hash = bytes.fromhex(row['censoring_channel_id'])[::-1]
             self.censored.setdefault(censoring_channel_hash, set())
             self.censored[censoring_channel_hash].add(row['tx_hash'])
-            return True
-        return False
+            return censoring_channel_hash
+        return None
 
     def to_message(self, outputs: OutputsMessage, extra_txo_rows: dict):
         for censoring_channel_hash, count in self.censored.items():
             blocked = outputs.blocked.add()
             blocked.count = len(count)
-            set_reference(blocked.channel, extra_txo_rows.get(censoring_channel_hash))
+            set_reference(blocked.channel, censoring_channel_hash, extra_txo_rows)
             outputs.blocked_total += len(count)
 
 
@@ -115,10 +119,10 @@ def message_to_txo(self, txo_message, tx_map):
                 'expiration_height': claim.expiration_height,
                 'effective_amount': claim.effective_amount,
                 'support_amount': claim.support_amount,
-                'trending_group': claim.trending_group,
-                'trending_mixed': claim.trending_mixed,
-                'trending_local': claim.trending_local,
-                'trending_global': claim.trending_global,
+                # 'trending_group': claim.trending_group,
+                # 'trending_mixed': claim.trending_mixed,
+                # 'trending_local': claim.trending_local,
+                # 'trending_global': claim.trending_global,
             }
             if claim.HasField('channel'):
                 txo.channel = tx_map[claim.channel.tx_hash].outputs[claim.channel.nout]
@@ -169,51 +173,54 @@ def to_base64(cls, txo_rows, extra_txo_rows, offset=0, total=None, blocked=None)
 
     @classmethod
     def to_bytes(cls, txo_rows, extra_txo_rows, offset=0, total=None, blocked: Censor = None) -> bytes:
-        extra_txo_rows = {row['claim_hash']: row for row in extra_txo_rows}
         page = OutputsMessage()
         page.offset = offset
         if total is not None:
             page.total = total
         if blocked is not None:
             blocked.to_message(page, extra_txo_rows)
+        for row in extra_txo_rows:
+            cls.encode_txo(page.extra_txos.add(), row)
+
         for row in txo_rows:
-            cls.row_to_message(row, page.txos.add(), extra_txo_rows)
-        for row in extra_txo_rows.values():
-            cls.row_to_message(row, page.extra_txos.add(), extra_txo_rows)
+            # cls.row_to_message(row, page.txos.add(), extra_txo_rows)
+            txo_message: 'OutputsMessage' = page.txos.add()
+            cls.encode_txo(txo_message, row)
+            if not isinstance(row, Exception):
+                if row.channel_hash:
+                    set_reference(txo_message.claim.channel, row.channel_hash, extra_txo_rows)
+                if row.reposted_claim_hash:
+                    set_reference(txo_message.claim.repost, row.reposted_claim_hash, extra_txo_rows)
+            elif isinstance(row, ResolveCensoredError):
+                set_reference(txo_message.error.blocked.channel, row.censor_id, extra_txo_rows)
         return page.SerializeToString()
 
     @classmethod
-    def row_to_message(cls, txo, txo_message, extra_row_dict: dict):
-        if isinstance(txo, Exception):
-            txo_message.error.text = txo.args[0]
-            if isinstance(txo, ValueError):
+    def encode_txo(cls, txo_message, resolve_result: Union['ResolveResult', Exception]):
+        if isinstance(resolve_result, Exception):
+            txo_message.error.text = resolve_result.args[0]
+            if isinstance(resolve_result, ValueError):
                 txo_message.error.code = ErrorMessage.INVALID
-            elif isinstance(txo, LookupError):
+            elif isinstance(resolve_result, LookupError):
                 txo_message.error.code = ErrorMessage.NOT_FOUND
-            elif isinstance(txo, ResolveCensoredError):
+            elif isinstance(resolve_result, ResolveCensoredError):
                 txo_message.error.code = ErrorMessage.BLOCKED
-                set_reference(txo_message.error.blocked.channel, extra_row_dict.get(bytes.fromhex(txo.censor_id)[::-1]))
             return
-        txo_message.tx_hash = txo['txo_hash'][:32]
-        txo_message.nout, = struct.unpack('<I', txo['txo_hash'][32:])
-        txo_message.height = txo['height']
-        txo_message.claim.short_url = txo['short_url']
-        txo_message.claim.reposted = txo['reposted']
-        if txo['canonical_url'] is not None:
-            txo_message.claim.canonical_url = txo['canonical_url']
-        txo_message.claim.is_controlling = bool(txo['is_controlling'])
-        if txo['last_take_over_height'] is not None:
-            txo_message.claim.take_over_height = txo['last_take_over_height']
-        txo_message.claim.creation_height = txo['creation_height']
-        txo_message.claim.activation_height = txo['activation_height']
-        txo_message.claim.expiration_height = txo['expiration_height']
-        if txo['claims_in_channel'] is not None:
-            txo_message.claim.claims_in_channel = txo['claims_in_channel']
-        txo_message.claim.effective_amount = txo['effective_amount']
-        txo_message.claim.support_amount = txo['support_amount']
-        txo_message.claim.trending_group = txo['trending_group']
-        txo_message.claim.trending_mixed = txo['trending_mixed']
-        txo_message.claim.trending_local = txo['trending_local']
-        txo_message.claim.trending_global = txo['trending_global']
-        set_reference(txo_message.claim.channel, extra_row_dict.get(txo['channel_hash']))
-        set_reference(txo_message.claim.repost, extra_row_dict.get(txo['reposted_claim_hash']))
+        txo_message.tx_hash = resolve_result.tx_hash
+        txo_message.nout = resolve_result.position
+        txo_message.height = resolve_result.height
+        txo_message.claim.short_url = resolve_result.short_url
+        txo_message.claim.reposted = resolve_result.reposted
+        txo_message.claim.is_controlling = resolve_result.is_controlling
+        txo_message.claim.creation_height = resolve_result.creation_height
+        txo_message.claim.activation_height = resolve_result.activation_height
+        txo_message.claim.expiration_height = resolve_result.expiration_height
+        txo_message.claim.effective_amount = resolve_result.effective_amount
+        txo_message.claim.support_amount = resolve_result.support_amount
+
+        if resolve_result.canonical_url is not None:
+            txo_message.claim.canonical_url = resolve_result.canonical_url
+        if resolve_result.last_takeover_height is not None:
+            txo_message.claim.take_over_height = resolve_result.last_takeover_height
+        if resolve_result.claims_in_channel is not None:
+            txo_message.claim.claims_in_channel = resolve_result.claims_in_channel