Rework electrumx_rpc; add "query" command

docs/rpc-interface.rst

@@ -23,7 +23,7 @@ Add a peer to the peers list.  ElectrumX will schdule an immediate
 connection attempt.  This command takes a single argument: the peer's
 "real name" as it used to advertise itself on IRC::
 
-  $ ./electrumx_rpc add_peer "ecdsa.net v1.0 s110 t"
+  $ electrumx_rpc add_peer "ecdsa.net v1.0 s110 t"
   "peer 'ecdsa.net v1.0 s110 t' added"
 
 daemon_url
@@ -48,7 +48,7 @@ disconnect
 Disconnect the given session IDs.  Session IDs can be seen in the logs
 or with the `sessions`_ RPC command::
 
-  $ ./electrumx_rpc disconnect 2 3
+  $ electrumx_rpc disconnect 2 3
   [
       "disconnected 2",
       "disconnected 3"
@@ -161,6 +161,30 @@ Peer data is obtained via a peer discovery protocol documented
   [...]
   electroncash.checksum0.com     good   50001 50002 ElectrumX 1.2.1    0.9  1.1          07h 30m 40s 07h 30m 41s     0                 peer 149.56.198.233
 
+query
+-----
+
+Run a query of the UTXO and history databases against one or more
+addresses or hex scripts.  `--limit <N>` or `-l <N>` limits the output
+for each kind to that many entries.  History is printed in blockchain
+order; UTXOs in an arbitrary order.
+
+For example::
+
+  $ electrumx_rpc query --limit 5 76a91462e907b15cbf27d5425399ebf6f0fb50ebb88f1888ac
+  Script: 76a91462e907b15cbf27d5425399ebf6f0fb50ebb88f1888ac
+  History #1: height 123,723 tx_hash 3387418aaddb4927209c5032f515aa442a6587d6e54677f08a03b8fa7789e688
+  History #2: height 127,280 tx_hash 4574958d135e66a53abf9c61950aba340e9e140be50efeea9456aa9f92bf40b5
+  History #3: height 127,909 tx_hash 8b960c87f9f1a6e6910e214fcf5f9c69b60319ba58a39c61f299548412f5a1c6
+  History #4: height 127,943 tx_hash 8f6b63012753005236b1b76e4884e4dee7415e05ab96604d353001662cde6b53
+  History #5: height 127,943 tx_hash 60ff2dfdf67917040139903a0141f7525a7d152365b371b35fd1cf83f1d7f704
+  UTXO #1: tx_hash 9aa497bf000b20f5ec5dc512bb6c1b60b68fc584d38b292b434e839ea8807bf0 tx_pos 0 height 254,148 value 5,500
+  UTXO #2: tx_hash 1c998142a5a5aae6f8c1eab245351413fe8d4032a3f14345f9943a0d0bc90ec0 tx_pos 0 height 254,161 value 5,500
+  UTXO #3: tx_hash 53345491b4829140be53f30079c6e4556a18545343b122900ebbfa158f9ca97a tx_pos 0 height 254,163 value 5,500
+  UTXO #4: tx_hash c71ad947ac46af217da3cd5521113cbd03e36ddada2b4452afe6c15f944d2529 tx_pos 0 height 372,916 value 1,000
+  UTXO #5: tx_hash c944a6acac054275a5e294e746d9ce79f6dcae91f3b4f5a84561aee6404a55b3 tx_pos 0 height 254,148 value 5,500
+  Balance: 17.8983303 BCH
+
 reorg
 -----
 

electrumx/server/chain_state.py

@@ -11,6 +11,8 @@ import pylru
 
 from aiorpcx import run_in_thread
 
+from electrumx.lib.hash import hash_to_hex_str
+
 
 class ChainState(object):
     '''Used as an interface by servers to request information about
@@ -92,3 +94,50 @@ class ChainState(object):
     def set_daemon_url(self, daemon_url):
         self._daemon.set_urls(self._env.coin.daemon_urls(daemon_url))
         return self._daemon.logged_url()
+
+    async def query(self, args, limit):
+        coin = self._env.coin
+        db = self._bp
+        lines = []
+
+        def arg_to_hashX(arg):
+            try:
+                script = bytes.fromhex(arg)
+                lines.append(f'Script: {arg}')
+                return coin.hashX_from_script(script)
+            except ValueError:
+                pass
+
+            try:
+                hashX = coin.address_to_hashX(arg)
+                lines.append(f'Address: {arg}')
+                return hashX
+            except Base58Error:
+                print(f'Ingoring unknown arg: {arg}')
+                return None
+
+        for arg in args:
+            hashX = arg_to_hashX(arg)
+            if not hashX:
+                continue
+            n = None
+            for n, (tx_hash, height) in enumerate(
+                    db.get_history(hashX, limit), start=1):
+                lines.append(f'History #{n:,d}: height {height:,d} '
+                             f'tx_hash {hash_to_hex_str(tx_hash)}')
+            if n is None:
+                lines.append('No history found')
+            n = None
+            for n, utxo in enumerate(db.get_utxos(hashX, limit), start=1):
+                lines.append(f'UTXO #{n:,d}: tx_hash '
+                             f'{hash_to_hex_str(utxo.tx_hash)} '
+                             f'tx_pos {utxo.tx_pos:,d} height {utxo.height:,d} '
+                             f'value {utxo.value:,d}')
+            if n is None:
+                lines.append('No UTXOs found')
+
+            balance = db.get_balance(hashX)
+            lines.append(f'Balance: {coin.decimal_value(balance):,f} '
+                         f'{coin.SHORTNAME}')
+
+        return lines

electrumx/server/session.py

@@ -130,7 +130,7 @@ class SessionManager(object):
 
         # Set up the RPC request handlers
         cmds = ('add_peer daemon_url disconnect getinfo groups log peers '
-                'reorg sessions stop'.split())
+                'query reorg sessions stop'.split())
         self.rpc_handlers = {cmd: getattr(self, 'rpc_' + cmd) for cmd in cmds}
 
     async def _start_server(self, kind, *args, **kw_args):
@@ -353,7 +353,7 @@ class SessionManager(object):
 
         return self._for_each_session(session_ids, toggle_logging)
 
-    def rpc_daemon_url(self, daemon_url=None):
+    def rpc_daemon_url(self, daemon_url):
         '''Replace the daemon URL.'''
         daemon_url = daemon_url or self.env.daemon_url
         try:
@@ -379,19 +379,23 @@ class SessionManager(object):
         '''Return a list of data about server peers.'''
         return self.peer_mgr.rpc_data()
 
+    async def rpc_query(self, items, limit):
+        '''Return a list of data about server peers.'''
+        return await self.chain_state.query(items, limit)
+
     def rpc_sessions(self):
         '''Return statistics about connected sessions.'''
         return self._session_data(for_log=False)
 
-    def rpc_reorg(self, count=3):
+    def rpc_reorg(self, count):
         '''Force a reorg of the given number of blocks.
 
-        count: number of blocks to reorg (default 3)
+        count: number of blocks to reorg
         '''
         count = non_negative_integer(count)
         if not self.chain_state.force_chain_reorg(count):
             raise RPCError(BAD_REQUEST, 'still catching up with daemon')
-        return 'scheduled a reorg of {:,d} blocks'.format(count)
+        return f'scheduled a reorg of {count:,d} blocks'
 
     # --- External Interface
 

electrumx_rpc

@@ -10,6 +10,7 @@
 '''Script to send RPC commands to a running ElectrumX server.'''
 
 
+from aiorpcx import timeout_after
 import argparse
 import asyncio
 import json
@@ -19,38 +20,111 @@ import electrumx.lib.text as text
 
 from aiorpcx import ClientSession
 
+simple_commands = {
+    'getinfo': 'Print a summary of server state',
+    'groups': 'Print current session groups',
+    'peers': 'Print information about peer servers for the same coin',
+    'sessions': 'Print information about client sessions',
+    'stop': 'Shut down the server cleanly',
+}
+
+session_commands = {
+    'disconnect': 'Disconnect sessions',
+    'log': 'Toggle logging of sessions',
+}
+
+other_commands = {
+    'add_peer': (
+        'add a peer to the peers list',
+        [], {
+            'type': str,
+            'dest': 'real_name',
+            'help': 'e.g. "a.domain.name s995 t"',
+        },
+    ),
+    'daemon_url': (
+        "replace the daemon's URL at run-time, and forecefully rotate "
+        " to the first URL in the list",
+        [], {
+            'type': str,
+            'nargs': '?',
+            'default': '',
+            'dest': 'daemon_url',
+            'help': 'see documentation of DAEMON_URL envvar',
+        },
+    ),
+    'query': (
+        'query the UTXO and history databases',
+        ['-l', '--limit'], {
+            'type': int,
+            'default': 1000,
+            'help': 'UTXO and history output limit',
+        }, ['items'], {
+            'nargs': '+',
+            'type': str,
+            'help': 'hex scripts, or addresses, to query',
+        },
+    ),
+    'reorg': (
+        'simulate a chain reorganization',
+        [], {
+            'type': int,
+            'dest': 'count',
+            'default': 3,
+            'help': 'number of blocks to back up'
+        },
+    ),
+}
+
 
 def main():
     '''Send the RPC command to the server and print the result.'''
-    parser = argparse.ArgumentParser('Send electrumx an RPC command')
+    main_parser = argparse.ArgumentParser(
-    parser.add_argument('-p', '--port', metavar='port_num', type=int,
+        'elextrumx_rpc',
-                        help='RPC port number')
+        description='Send electrumx an RPC command'
-    parser.add_argument('command', nargs=1, default=[],
+    )
-                        help='command to send')
+    main_parser.add_argument('-p', '--port', metavar='port_num', type=int,
-    parser.add_argument('param', nargs='*', default=[],
+                             help='RPC port number')
-                        help='params to send')
-    args = parser.parse_args()
 
-    port = args.port
+    subparsers = main_parser.add_subparsers(help='sub-command help',
+                                            dest='command')
+
+    for command, help in simple_commands.items():
+        parser = subparsers.add_parser(command, help=help)
+
+    for command, help in session_commands.items():
+        parser = subparsers.add_parser(command, help=help)
+        parser.add_argument('session_ids', nargs='+', type=int,
+                            help='list of session ids')
+
+    for command, data in other_commands.items():
+        parser_help, *arguments = data
+        parser = subparsers.add_parser(command, help=parser_help)
+        for n in range(0, len(arguments), 2):
+            args, kwargs = arguments[n: n+2]
+            parser.add_argument(*args, **kwargs)
+
+    args = main_parser.parse_args()
+    args = vars(args)
+    port = args.pop('port')
     if port is None:
         port = int(environ.get('RPC_PORT', 8000))
+    method = args.pop('command')
 
-    # Get the RPC request.
+    # aiorpcX makes this so easy...
-    method = args.command[0]
-    params = args.param
-    if method in ('log', 'disconnect'):
-        params = [params]
-
     async def send_request():
-        # aiorpcX makes this so easy...
+        async with timeout_after(15):
-        async with ClientSession('localhost', port) as session:
+            async with ClientSession('localhost', port) as session:
-            result = await session.send_request(method, params, timeout=15)
+                result = await session.send_request(method, args)
-            if method in ('groups', 'peers', 'sessions'):
+                if method in ('query', ):
-                lines_func = getattr(text, f'{method}_lines')
+                    for line in result:
-                for line in lines_func(result):
+                        print(line)
-                    print(line)
+                elif method in ('groups', 'peers', 'sessions'):
-            else:
+                    lines_func = getattr(text, f'{method}_lines')
-                print(json.dumps(result, indent=4, sort_keys=True))
+                    for line in lines_func(result):
+                        print(line)
+                else:
+                    print(json.dumps(result, indent=4, sort_keys=True))
 
     loop = asyncio.get_event_loop()
     try: