Merge d66f46e into 6343771

.github/workflows/main.yml

@@ -94,7 +94,7 @@ jobs:
       - name: Runs Elasticsearch
         uses: elastic/elastic-github-actions/elasticsearch@master
         with:
-          stack-version: 7.6.0
+          stack-version: 7.12.1
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v1
         with:
@@ -110,6 +110,8 @@ jobs:
           key: tox-integration-${{ matrix.test }}-${{ hashFiles('setup.py') }}
           restore-keys: txo-integration-${{ matrix.test }}-
       - run: pip install tox coverage coveralls
+      - if: matrix.test == 'claims'
+        run: rm -rf .tox
       - run: tox -e ${{ matrix.test }}
       - name: submit coverage report
         env:

lbry/extras/daemon/daemon.py

@@ -205,12 +205,15 @@ def fix_kwargs_for_hub(**kwargs):
         elif key == "not_channel_ids":
             kwargs["channel_id"] = {"invert": True, "value": kwargs.pop("not_channel_ids")}
         elif key in MY_RANGE_FIELDS:
-            operator = '='
-            if isinstance(value, str) and value[0] in opcodes:
-                operator_length = 2 if value[:2] in opcodes else 1
-                operator, value = value[:operator_length], value[operator_length:]
-            value = [str(value if key != 'fee_amount' else Decimal(value)*1000)]
-            kwargs[key] = {"op": opcodes[operator], "value": value}
+            constraints = []
+            for val in value if isinstance(value, list) else [value]:
+                operator = '='
+                if isinstance(val, str) and val[0] in opcodes:
+                    operator_length = 2 if val[:2] in opcodes else 1
+                    operator, val = val[:operator_length], val[operator_length:]
+                val = [int(val if key != 'fee_amount' else Decimal(val)*1000)]
+                constraints.append({"op": opcodes[operator], "value": val})
+            kwargs[key] = constraints
         elif key == 'order_by':  # TODO: remove this after removing support for old trending args from the api
             value = value if isinstance(value, list) else [value]
             new_value = []
@@ -2408,6 +2411,9 @@ async def jsonrpc_claim_search(self, **kwargs):
         value with an equality constraint such as '>', '>=', '<' and '<='
         eg. --height=">400000" would limit results to only claims above 400k block height.
 
+        They also support multiple constraints passed as a list of the args described above.
+        eg. --release_time=[">1000000", "<2000000"]
+
         Usage:
             claim_search [<name> | --name=<name>] [--text=<text>] [--txid=<txid>] [--nout=<nout>]
                          [--claim_id=<claim_id> | --claim_ids=<claim_ids>...]

lbry/schema/types/v2/hub_pb2_grpc.py

@@ -20,6 +20,41 @@ def __init__(self, channel):
                 request_serializer=hub__pb2.SearchRequest.SerializeToString,
                 response_deserializer=result__pb2.Outputs.FromString,
                 )
+        self.Ping = channel.unary_unary(
+                '/pb.Hub/Ping',
+                request_serializer=hub__pb2.EmptyMessage.SerializeToString,
+                response_deserializer=hub__pb2.StringValue.FromString,
+                )
+        self.Hello = channel.unary_unary(
+                '/pb.Hub/Hello',
+                request_serializer=hub__pb2.HelloMessage.SerializeToString,
+                response_deserializer=hub__pb2.HelloMessage.FromString,
+                )
+        self.AddPeer = channel.unary_unary(
+                '/pb.Hub/AddPeer',
+                request_serializer=hub__pb2.ServerMessage.SerializeToString,
+                response_deserializer=hub__pb2.StringValue.FromString,
+                )
+        self.PeerSubscribe = channel.unary_unary(
+                '/pb.Hub/PeerSubscribe',
+                request_serializer=hub__pb2.ServerMessage.SerializeToString,
+                response_deserializer=hub__pb2.StringValue.FromString,
+                )
+        self.Version = channel.unary_unary(
+                '/pb.Hub/Version',
+                request_serializer=hub__pb2.EmptyMessage.SerializeToString,
+                response_deserializer=hub__pb2.StringValue.FromString,
+                )
+        self.Features = channel.unary_unary(
+                '/pb.Hub/Features',
+                request_serializer=hub__pb2.EmptyMessage.SerializeToString,
+                response_deserializer=hub__pb2.StringValue.FromString,
+                )
+        self.Broadcast = channel.unary_unary(
+                '/pb.Hub/Broadcast',
+                request_serializer=hub__pb2.EmptyMessage.SerializeToString,
+                response_deserializer=hub__pb2.UInt32Value.FromString,
+                )
 
 
 class HubServicer(object):
@@ -31,6 +66,48 @@ def Search(self, request, context):
         context.set_details('Method not implemented!')
         raise NotImplementedError('Method not implemented!')
 
+    def Ping(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Hello(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def AddPeer(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def PeerSubscribe(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Version(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Features(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def Broadcast(self, request, context):
+        """Missing associated documentation comment in .proto file."""
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
 
 def add_HubServicer_to_server(servicer, server):
     rpc_method_handlers = {
@@ -39,6 +116,41 @@ def add_HubServicer_to_server(servicer, server):
                     request_deserializer=hub__pb2.SearchRequest.FromString,
                     response_serializer=result__pb2.Outputs.SerializeToString,
             ),
+            'Ping': grpc.unary_unary_rpc_method_handler(
+                    servicer.Ping,
+                    request_deserializer=hub__pb2.EmptyMessage.FromString,
+                    response_serializer=hub__pb2.StringValue.SerializeToString,
+            ),
+            'Hello': grpc.unary_unary_rpc_method_handler(
+                    servicer.Hello,
+                    request_deserializer=hub__pb2.HelloMessage.FromString,
+                    response_serializer=hub__pb2.HelloMessage.SerializeToString,
+            ),
+            'AddPeer': grpc.unary_unary_rpc_method_handler(
+                    servicer.AddPeer,
+                    request_deserializer=hub__pb2.ServerMessage.FromString,
+                    response_serializer=hub__pb2.StringValue.SerializeToString,
+            ),
+            'PeerSubscribe': grpc.unary_unary_rpc_method_handler(
+                    servicer.PeerSubscribe,
+                    request_deserializer=hub__pb2.ServerMessage.FromString,
+                    response_serializer=hub__pb2.StringValue.SerializeToString,
+            ),
+            'Version': grpc.unary_unary_rpc_method_handler(
+                    servicer.Version,
+                    request_deserializer=hub__pb2.EmptyMessage.FromString,
+                    response_serializer=hub__pb2.StringValue.SerializeToString,
+            ),
+            'Features': grpc.unary_unary_rpc_method_handler(
+                    servicer.Features,
+                    request_deserializer=hub__pb2.EmptyMessage.FromString,
+                    response_serializer=hub__pb2.StringValue.SerializeToString,
+            ),
+            'Broadcast': grpc.unary_unary_rpc_method_handler(
+                    servicer.Broadcast,
+                    request_deserializer=hub__pb2.EmptyMessage.FromString,
+                    response_serializer=hub__pb2.UInt32Value.SerializeToString,
+            ),
     }
     generic_handler = grpc.method_handlers_generic_handler(
             'pb.Hub', rpc_method_handlers)
@@ -65,3 +177,122 @@ def Search(request,
             result__pb2.Outputs.FromString,
             options, channel_credentials,
             insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Ping(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/Ping',
+            hub__pb2.EmptyMessage.SerializeToString,
+            hub__pb2.StringValue.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Hello(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/Hello',
+            hub__pb2.HelloMessage.SerializeToString,
+            hub__pb2.HelloMessage.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def AddPeer(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/AddPeer',
+            hub__pb2.ServerMessage.SerializeToString,
+            hub__pb2.StringValue.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def PeerSubscribe(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/PeerSubscribe',
+            hub__pb2.ServerMessage.SerializeToString,
+            hub__pb2.StringValue.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Version(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/Version',
+            hub__pb2.EmptyMessage.SerializeToString,
+            hub__pb2.StringValue.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Features(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/Features',
+            hub__pb2.EmptyMessage.SerializeToString,
+            hub__pb2.StringValue.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+
+    @staticmethod
+    def Broadcast(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(request, target, '/pb.Hub/Broadcast',
+            hub__pb2.EmptyMessage.SerializeToString,
+            hub__pb2.UInt32Value.FromString,
+            options, channel_credentials,
+            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

lbry/wallet/orchstr8/__init__.py

@@ -1,5 +1,5 @@
 __hub_url__ = (
-    "https://github.com/lbryio/hub/releases/download/leveldb-hub/hub"
+    "https://github.com/lbryio/hub/releases/download/range-queries/hub"
 )
 from .node import Conductor
 from .service import ConductorService

lbry/wallet/orchstr8/node.py

@@ -473,6 +473,10 @@ def pipe_data_received(self, fd, data):
             raise SystemError(data.decode())
         if b'listening on' in data:
             self.ready.set()
+        str_lines = str(data.decode()).split("\n")
+        for line in str_lines:
+            if 'releaseTime' in line:
+                print(line)
 
     def process_exited(self):
         self.stopped.set()

lbry/wallet/server/db/elasticsearch/constants.py

@@ -42,7 +42,7 @@
     'tx_id', 'tx_nout', 'tx_position',
     'short_url', 'canonical_url',
     'is_controlling', 'last_take_over_height',
-    'public_key_bytes', 'public_key_id', 'claims_in_channel', 'channel_join_height',
+    'public_key_bytes', 'public_key_id', 'claims_in_channel',
     'channel_id', 'signature', 'signature_digest', 'is_signature_valid',
     'amount', 'effective_amount', 'support_amount',
     'fee_amount', 'fee_currency',
@@ -64,7 +64,7 @@
 RANGE_FIELDS = {
     'height', 'creation_height', 'activation_height', 'expiration_height',
     'timestamp', 'creation_timestamp', 'duration', 'release_time', 'fee_amount',
-    'tx_position', 'channel_join', 'repost_count', 'limit_claims_per_channel',
+    'tx_position', 'repost_count', 'limit_claims_per_channel',
     'amount', 'effective_amount', 'support_amount',
     'trending_score', 'censor_type', 'tx_num'
 }

lbry/wallet/server/db/elasticsearch/search.py

@@ -574,6 +574,15 @@ def expand_query(**kwargs):
                 if key == 'fee_amount':
                     value = str(Decimal(value)*1000)
                 query['must'].append({"range": {key: {ops[operator]: value}}})
+            elif key in RANGE_FIELDS and isinstance(value, list) and all(v[0] in ops for v in value):
+                range_constraints = []
+                for v in value:
+                    operator_length = 2 if v[:2] in ops else 1
+                    operator, stripped_op_v = v[:operator_length], v[operator_length:]
+                    if key == 'fee_amount':
+                        stripped_op_v = str(Decimal(stripped_op_v)*1000)
+                    range_constraints.append((operator, stripped_op_v))
+                query['must'].append({"range": {key: {ops[operator]: v for operator, v in range_constraints}}})
             elif many:
                 query['must'].append({"terms": {key: value}})
             else:

lbry/wallet/server/session.py

@@ -981,8 +981,9 @@ async def claimtrie_search(self, **kwargs):
         start = time.perf_counter()
         if 'release_time' in kwargs:
             release_time = kwargs.pop('release_time')
+            release_times = release_time if isinstance(release_time, list) else [release_time]
             try:
-                kwargs['release_time'] = format_release_time(release_time)
+                kwargs['release_time'] = [format_release_time(release_time) for release_time in release_times]
             except ValueError:
                 pass
         try:

tests/integration/claims/test_claim_commands.py

@@ -1789,13 +1789,21 @@ async def test_setting_stream_fields(self):
         self.assertEqual(3, len(await self.claim_search(release_time='>0', order_by=['release_time'])))
         self.assertEqual(3, len(await self.claim_search(release_time='>=0', order_by=['release_time'])))
         self.assertEqual(4, len(await self.claim_search(order_by=['release_time'])))
-        self.assertEqual(4, len(await self.claim_search(release_time='<derp', order_by=['release_time'])))
         self.assertEqual(3, len(await self.claim_search(claim_type='stream', order_by=['release_time'])))
         self.assertEqual(1, len(await self.claim_search(claim_type='channel', order_by=['release_time'])))
         self.assertEqual(1, len(await self.claim_search(release_time='>=123456', order_by=['release_time'])))
         self.assertEqual(1, len(await self.claim_search(release_time='>123456', order_by=['release_time'])))
         self.assertEqual(2, len(await self.claim_search(release_time='<123457', order_by=['release_time'])))
 
+        self.assertEqual(2, len(await self.claim_search(release_time=['<123457'], order_by=['release_time'])))
+        self.assertEqual(2, len(await self.claim_search(release_time=['>0', '<123457'], order_by=['release_time'])))
+        self.assertEqual(
+            2, len(await self.claim_search(release_time=['>=123097', '<123457'], order_by=['release_time']))
+        )
+        self.assertEqual(
+            2, len(await self.claim_search(release_time=['<123457', '>0'], order_by=['release_time']))
+        )
+
     async def test_setting_fee_fields(self):
         tx = await self.out(self.stream_create('paid-stream'))
         txo = tx['outputs'][0]