hub (wallet server) now uses Elasticsearch to handle claim_search and resolve requests for added performance and reliability

.github/workflows/main.yml

@@ -37,6 +37,17 @@ jobs:
           - blockchain
           - other
     steps:
+      - name: Configure sysctl limits
+        run: |
+                sudo swapoff -a
+                sudo sysctl -w vm.swappiness=1
+                sudo sysctl -w fs.file-max=262144
+                sudo sysctl -w vm.max_map_count=262144
+
+      - name: Runs Elasticsearch
+        uses: elastic/elastic-github-actions/elasticsearch@master
+        with:
+                stack-version: 7.6.0
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v1
         with:

docker/Dockerfile.wallet_server

@@ -1,4 +1,4 @@
-FROM ubuntu:20.04
+FROM debian:10-slim
 
 ARG user=lbry
 ARG db_dir=/database
@@ -13,7 +13,9 @@ RUN apt-get update && \
       wget \
       tar unzip \
       build-essential \
-      python3 \
+      pkg-config \
+      libleveldb-dev \
+      python3.7 \
       python3-dev \
       python3-pip \
       python3-wheel \

docker/docker-compose-wallet-server.yml

@@ -1,36 +1,40 @@
 version: "3"
 
 volumes:
-  lbrycrd:
   wallet_server:
+  es01:
 
 services:
-  lbrycrd:
-    image: lbry/lbrycrd:${LBRYCRD_TAG:-latest-release}
-    restart: always
-    ports: # accessible from host
-      - "9246:9246" # rpc port
-    expose: # internal to docker network. also this doesn't do anything. its for documentation only.
-      - "9245" # node-to-node comms port
-    volumes:
-      - "lbrycrd:/data/.lbrycrd"
-    environment:
-      - RUN_MODE=default
-      # Curently not snapshot provided
-      #- SNAPSHOT_URL=${LBRYCRD_SNAPSHOT_URL-https://lbry.com/snapshot/blockchain}
-      - RPC_ALLOW_IP=0.0.0.0/0
   wallet_server:
+    depends_on: 
+      - es01
     image: lbry/wallet-server:${WALLET_SERVER_TAG:-latest-release}
-    depends_on:
-      - lbrycrd
     restart: always
+    network_mode: host
     ports:
       - "50001:50001" # rpc port
-      - "50005:50005" # websocket port
-      #- "2112:2112"   # uncomment to enable prometheus
+      - "2112:2112"   # uncomment to enable prometheus
     volumes:
       - "wallet_server:/database"
+    env_file: [/home/lbry/wallet-server-env]
     environment:
-      # Curently not snapshot provided
-      # - SNAPSHOT_URL=${WALLET_SERVER_SNAPSHOT_URL-https://lbry.com/snapshot/wallet}
-      - DAEMON_URL=http://lbry:lbry@lbrycrd:9245
+      - DAEMON_URL=http://lbry:lbry@127.0.0.1:9245
+      - TCP_PORT=50001
+      - PROMETHEUS_PORT=2112
+  es01:
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.11.0
+    container_name: es01
+    environment:
+      - node.name=es01
+      - discovery.type=single-node
+      - indices.query.bool.max_clause_count=4096
+      - bootstrap.memory_lock=true
+      - "ES_JAVA_OPTS=-Xms8g -Xmx8g"  # no more than 32, remember to disable swap
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    volumes:
+      - es01:/usr/share/elasticsearch/data
+    ports:
+      - 127.0.0.1:9200:9200

docker/wallet_server_entrypoint.sh

@@ -20,4 +20,6 @@ if [[ -n "$SNAPSHOT_URL" ]] && [[ ! -f /database/claims.db ]]; then
   rm "$filename"
 fi
 
-/home/lbry/.local/bin/torba-server "$@"
+/home/lbry/.local/bin/lbry-hub-elastic-sync /database/claims.db
+echo 'starting server'
+/home/lbry/.local/bin/lbry-hub "$@"

lbry/schema/result.py

@@ -13,57 +13,45 @@
 BLOCKED = ErrorMessage.Code.Name(ErrorMessage.BLOCKED)
 
 
-def set_reference(reference, claim_hash, rows):
-    if claim_hash:
-        for txo in rows:
-            if claim_hash == txo['claim_hash']:
-                reference.tx_hash = txo['txo_hash'][:32]
-                reference.nout = struct.unpack('<I', txo['txo_hash'][32:])[0]
-                reference.height = txo['height']
-                return
+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']
 
 
 class Censor:
 
-    __slots__ = 'streams', 'channels', 'limit_claims_per_channel', 'censored', 'claims_in_channel', 'total'
+    NOT_CENSORED = 0
+    SEARCH = 1
+    RESOLVE = 2
 
-    def __init__(self, streams: dict = None, channels: dict = None, limit_claims_per_channel: int = None):
-        self.streams = streams or {}
-        self.channels = channels or {}
-        self.limit_claims_per_channel = limit_claims_per_channel  # doesn't count as censored
+    __slots__ = 'censor_type', 'censored'
+
+    def __init__(self, censor_type):
+        self.censor_type = censor_type
         self.censored = {}
-        self.claims_in_channel = {}
-        self.total = 0
+
+    def is_censored(self, row):
+        return (row.get('censor_type') or self.NOT_CENSORED) >= self.censor_type
+
+    def apply(self, rows):
+        return [row for row in rows if not self.censor(row)]
 
     def censor(self, row) -> bool:
-        was_censored = False
-        for claim_hash, lookup in (
-                (row['claim_hash'], self.streams),
-                (row['claim_hash'], self.channels),
-                (row['channel_hash'], self.channels),
-                (row['reposted_claim_hash'], self.streams),
-                (row['reposted_claim_hash'], self.channels)):
-            censoring_channel_hash = lookup.get(claim_hash)
-            if censoring_channel_hash:
-                was_censored = True
-                self.censored.setdefault(censoring_channel_hash, 0)
-                self.censored[censoring_channel_hash] += 1
-                break
-        if was_censored:
-            self.total += 1
-        if not was_censored and self.limit_claims_per_channel is not None and row['channel_hash']:
-            self.claims_in_channel.setdefault(row['channel_hash'], 0)
-            self.claims_in_channel[row['channel_hash']] += 1
-            if self.claims_in_channel[row['channel_hash']] > self.limit_claims_per_channel:
-                return True
-        return was_censored
-
-    def to_message(self, outputs: OutputsMessage, extra_txo_rows):
-        outputs.blocked_total = self.total
+        if self.is_censored(row):
+            censoring_channel_hash = row['censoring_channel_hash']
+            self.censored.setdefault(censoring_channel_hash, set())
+            self.censored[censoring_channel_hash].add(row['tx_hash'])
+            return True
+        return False
+
+    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 = count
-            set_reference(blocked.channel, censoring_channel_hash, extra_txo_rows)
+            blocked.count = len(count)
+            set_reference(blocked.channel, extra_txo_rows.get(censoring_channel_hash))
+            outputs.blocked_total += len(count)
 
 
 class Outputs:
@@ -168,6 +156,7 @@ 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:
@@ -176,12 +165,12 @@ def to_bytes(cls, txo_rows, extra_txo_rows, offset=0, total=None, blocked: Censo
             blocked.to_message(page, extra_txo_rows)
         for row in txo_rows:
             cls.row_to_message(row, page.txos.add(), extra_txo_rows)
-        for row in extra_txo_rows:
+        for row in extra_txo_rows.values():
             cls.row_to_message(row, page.extra_txos.add(), extra_txo_rows)
         return page.SerializeToString()
 
     @classmethod
-    def row_to_message(cls, txo, txo_message, extra_txo_rows):
+    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):
@@ -190,7 +179,7 @@ def row_to_message(cls, txo, txo_message, extra_txo_rows):
                 txo_message.error.code = ErrorMessage.NOT_FOUND
             elif isinstance(txo, ResolveCensoredError):
                 txo_message.error.code = ErrorMessage.BLOCKED
-                set_reference(txo_message.error.blocked.channel, txo.censor_hash, extra_txo_rows)
+                set_reference(txo_message.error.blocked.channel, extra_row_dict.get(txo.censor_hash))
             return
         txo_message.tx_hash = txo['txo_hash'][:32]
         txo_message.nout, = struct.unpack('<I', txo['txo_hash'][32:])
@@ -213,5 +202,5 @@ def row_to_message(cls, txo, txo_message, extra_txo_rows):
         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, txo['channel_hash'], extra_txo_rows)
-        set_reference(txo_message.claim.repost, txo['reposted_claim_hash'], extra_txo_rows)
+        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']))

lbry/schema/url.py

@@ -55,6 +55,14 @@ class PathSegment(NamedTuple):
     def normalized(self):
         return normalize_name(self.name)
 
+    @property
+    def is_shortid(self):
+        return self.claim_id is not None and len(self.claim_id) < 40
+
+    @property
+    def is_fullid(self):
+        return self.claim_id is not None and len(self.claim_id) == 40
+
     def to_dict(self):
         q = {'name': self.name}
         if self.claim_id is not None:

lbry/wallet/network.py

@@ -417,9 +417,6 @@ def unsubscribe_address(self, address):
     def get_server_features(self):
         return self.rpc('server.features', (), restricted=True)
 
-    def get_claims_by_ids(self, claim_ids):
-        return self.rpc('blockchain.claimtrie.getclaimsbyids', claim_ids)
-
     def resolve(self, urls, session_override=None):
         return self.rpc('blockchain.claimtrie.resolve', urls, False, session_override)
 

lbry/wallet/orchstr8/node.py

@@ -11,6 +11,7 @@
 from binascii import hexlify
 from typing import Type, Optional
 import urllib.request
+from uuid import uuid4
 
 import lbry
 from lbry.wallet.server.server import Server
@@ -187,7 +188,9 @@ async def start(self, blockchain_node: 'BlockchainNode', extraconf=None):
             'SESSION_TIMEOUT': str(self.session_timeout),
             'MAX_QUERY_WORKERS': '0',
             'INDIVIDUAL_TAG_INDEXES': '',
-            'RPC_PORT': self.rpc_port
+            'RPC_PORT': self.rpc_port,
+            'ES_INDEX_PREFIX': uuid4().hex,
+            'ES_MODE': 'writer',
         }
         if extraconf:
             conf.update(extraconf)
@@ -199,6 +202,8 @@ async def start(self, blockchain_node: 'BlockchainNode', extraconf=None):
 
     async def stop(self, cleanup=True):
         try:
+            await self.server.db.search_index.delete_index()
+            await self.server.db.search_index.stop()
             await self.server.stop()
         finally:
             cleanup and self.cleanup()

lbry/wallet/rpc/util.py

@@ -32,10 +32,13 @@
 
 # other_params: None means cannot be called with keyword arguments only
 # any means any name is good
+from functools import lru_cache
+
 SignatureInfo = namedtuple('SignatureInfo', 'min_args max_args '
                            'required_names other_names')
 
 
+@lru_cache(256)
 def signature_info(func):
     params = inspect.signature(func).parameters
     min_args = max_args = 0

lbry/wallet/server/block_processor.py

@@ -5,7 +5,6 @@
 from typing import Optional
 from prometheus_client import Gauge, Histogram
 import lbry
-from lbry.schema.claim import Claim
 from lbry.wallet.server.db.writer import SQLDB
 from lbry.wallet.server.daemon import DaemonError
 from lbry.wallet.server.hash import hash_to_hex_str, HASHX_LEN
@@ -215,6 +214,8 @@ async def check_and_advance_blocks(self, raw_blocks):
         if hprevs == chain:
             start = time.perf_counter()
             await self.run_in_thread_with_lock(self.advance_blocks, blocks)
+            if self.sql:
+                await self.db.search_index.claim_consumer(self.sql.claim_producer())
             for cache in self.search_cache.values():
                 cache.clear()
             self.history_cache.clear()
@@ -228,6 +229,9 @@ async def check_and_advance_blocks(self, raw_blocks):
                 s = '' if len(blocks) == 1 else 's'
                 self.logger.info('processed {:,d} block{} in {:.1f}s'.format(len(blocks), s, processed_time))
             if self._caught_up_event.is_set():
+                if self.sql:
+                    await self.db.search_index.apply_filters(self.sql.blocked_streams, self.sql.blocked_channels,
+                                                             self.sql.filtered_streams, self.sql.filtered_channels)
                 await self.notifications.on_block(self.touched, self.height)
             self.touched = set()
         elif hprevs[0] != chain[0]:
@@ -282,7 +286,6 @@ def flush_backup():
                 await self.run_in_thread_with_lock(flush_backup)
                 last -= len(raw_blocks)
 
-            await self.run_in_thread_with_lock(self.db.sql.delete_claims_above_height, self.height)
             await self.prefetcher.reset_height(self.height)
             self.reorg_count_metric.inc()
         except:
@@ -651,7 +654,11 @@ async def _process_prefetched_blocks(self):
                 self.reorg_count = 0
             else:
                 blocks = self.prefetcher.get_prefetched_blocks()
-                await self.check_and_advance_blocks(blocks)
+                try:
+                    await self.check_and_advance_blocks(blocks)
+                except Exception:
+                    self.logger.exception("error while processing txs")
+                    raise
 
     async def _first_caught_up(self):
         self.logger.info(f'caught up to height {self.height}')
@@ -782,15 +789,17 @@ def __init__(self, *args, **kwargs):
         self.timer = Timer('BlockProcessor')
 
     def advance_blocks(self, blocks):
-        self.sql.begin()
+        if self.sql:
+            self.sql.begin()
         try:
             self.timer.run(super().advance_blocks, blocks)
         except:
             self.logger.exception(f'Error while advancing transaction in new block.')
             raise
         finally:
-            self.sql.commit()
-        if self.db.first_sync and self.height == self.daemon.cached_height():
+            if self.sql:
+                self.sql.commit()
+        if self.sql and self.db.first_sync and self.height == self.daemon.cached_height():
             self.timer.run(self.sql.execute, self.sql.SEARCH_INDEXES, timer_name='executing SEARCH_INDEXES')
             if self.env.individual_tag_indexes:
                 self.timer.run(self.sql.execute, self.sql.TAG_INDEXES, timer_name='executing TAG_INDEXES')
@@ -799,22 +808,8 @@ def advance_blocks(self, blocks):
     def advance_txs(self, height, txs, header, block_hash):
         timer = self.timer.sub_timers['advance_blocks']
         undo = timer.run(super().advance_txs, height, txs, header, block_hash, timer_name='super().advance_txs')
-        timer.run(self.sql.advance_txs, height, txs, header, self.daemon.cached_height(), forward_timer=True)
+        if self.sql:
+            timer.run(self.sql.advance_txs, height, txs, header, self.daemon.cached_height(), forward_timer=True)
         if (height % 10000 == 0 or not self.db.first_sync) and self.logger.isEnabledFor(10):
             self.timer.show(height=height)
         return undo
-
-    def _checksig(self, value, address):
-        try:
-            claim_dict = Claim.from_bytes(value)
-            cert_id = claim_dict.signing_channel_hash
-            if not self.should_validate_signatures:
-                return cert_id
-            if cert_id:
-                cert_claim = self.db.get_claim_info(cert_id)
-                if cert_claim:
-                    certificate = Claim.from_bytes(cert_claim.value)
-                    claim_dict.validate_signature(address, certificate)
-                    return cert_id
-        except Exception:
-            pass