Utilize HDF5 for state

lamden/network.py

@@ -180,3 +180,6 @@ def all_vks_found(self, vks):
             if self.peers.get(vk) is None:
                 return False
         return True
+
+def discover_peer(vk, ip):
+    pass

lamden/nodes/base.py

@@ -245,7 +245,8 @@ async def catchup(self, mn_seed, mn_vk):
                     wallet=self.wallet,
                     ctx=self.ctx
                 )
-            self.process_new_block(block)
+            if not block.get('error'):
+                self.process_new_block(block)
 
         # Process any blocks that were made while we were catching up
         while len(self.new_block_processor.q) > 0:
@@ -318,8 +319,6 @@ def update_state(self, block):
                 client=self.client
             )
 
-        #self.nonces.flush_pending()
-
         self.log.info('Updating metadata.')
         self.current_height = storage.get_latest_block_height(self.driver)
         self.current_hash = storage.get_latest_block_hash(self.driver)
@@ -335,15 +334,14 @@ def process_new_block(self, block):
         if self.store:
             encoded_block = encode(block)
             encoded_block = json.loads(encoded_block)
-
-            self.blocks.store_block(encoded_block)
-
             # create Event File
             self.event_writer.write_event(Event(
                 topics=[NEW_BLOCK_EVENT],
                 data=encoded_block
             ))
 
+            self.blocks.store_block(block)
+
         # Prepare for the next block by flushing out driver and notification state
         # self.new_block_processor.clean()
 

lamden/nodes/delegate/delegate.py

@@ -38,9 +38,11 @@ def __init__(self, client: ContractingClient, nonces: storage.NonceStorage, debu
 
     async def process_message(self, msg):
         self.log.info(f'Received work from {msg["sender"][:8]}')
-        # if msg['sender'] not in self.masters:
-        #     self.log.error(f'TX Batch received from non-master {msg["sender"][:8]}')
-        #     return
+        self.log.info(msg)
+
+        if msg['sender'] not in self.masters:
+            self.log.error(f'TX Batch received from non-master {msg["sender"][:8]}')
+            return
 
         shim = {
             'transactions': [],

lamden/nodes/events.py

@@ -20,6 +20,7 @@
 from sanic import Sanic
 import json
 import argparse
+from contracting.db.encoder import encode, decode
 
 EVENTS_HOME = pathlib.Path().home().joinpath('.lamden').joinpath('events')
 EXTENSION = '.e'
@@ -51,8 +52,7 @@ def get_events(self) -> List[Event]:
             with open(file, 'r') as f:
                 try:
                     e = json.load(f)
-                    event = Event(e['topics'], e['data'])
-                    events.append(event)
+                    events.append(Event(e['topics'], e['data']))
                 except:
                     # TODO(nikita): proper handling
                     print('failed to load event')

lamden/nodes/masternode/masternode.py

@@ -43,7 +43,7 @@ def get_block(self, command):
         block = self.blocks.get_block(num)
 
         if block is None:
-            return None
+            return {"error": "block does not exist"}
 
         return block
 

lamden/nodes/masternode/webserver.py

@@ -153,7 +153,7 @@ async def disconnect():
         @self.sio.event
         async def event(data):
             for client in self.ws_clients:
-                await client.send(json.dumps(data))
+                await client.send(encode(data))
 
     def __register_app_listeners(self):
         @self.app.listener('after_server_start')
@@ -180,7 +180,7 @@ async def ws_handler(self, request, ws):
                 'data': block
             }
 
-            await ws.send(json.dumps(eventData))
+            await ws.send(encode(eventData))
 
             async for message in ws:
                 pass
@@ -386,7 +386,7 @@ async def get_latest_block(self, request):
 
         num = storage.get_latest_block_height(self.driver)
         block = self.blocks.get_block(int(num))
-        return response.json(block, dumps=NonceEncoder().encode, headers={'Access-Control-Allow-Origin': '*'})
+        return response.json(block, dumps=encode, headers={'Access-Control-Allow-Origin': '*'})
 
     async def get_latest_block_number(self, request):
         self.driver.clear_pending_state()
@@ -417,7 +417,7 @@ async def get_block(self, request):
             return response.json({'error': 'Block not found.'}, status=400,
                                  headers={'Access-Control-Allow-Origin': '*'})
 
-        return response.json(block, dumps=NonceEncoder().encode, headers={'Access-Control-Allow-Origin': '*'})
+        return response.json(block, dumps=encode, headers={'Access-Control-Allow-Origin': '*'})
 
     async def get_tx(self, request):
         _hash = request.args.get('hash')
@@ -437,7 +437,7 @@ async def get_tx(self, request):
             return response.json({'error': 'Transaction not found.'}, status=400,
                                  headers={'Access-Control-Allow-Origin': '*'})
 
-        return response.json(tx, dumps=NonceEncoder().encode, headers={'Access-Control-Allow-Origin': '*'})
+        return response.json(tx, dumps=encode, headers={'Access-Control-Allow-Origin': '*'})
 
     async def get_constitution(self, request):
         self.client.raw_driver.clear_pending_state()

lamden/router.py

@@ -277,3 +277,32 @@ async def secure_multicast(msg: dict, service, wallet: Wallet, peer_map: dict, c
         )
 
     await asyncio.gather(*coroutines)
+
+
+def build_secure_socket(wallet: Wallet, vk: str, ip: str, ctx: zmq.asyncio.Context, linger=500, cert_dir=DEFAULT_DIR):
+    socket = ctx.socket(zmq.DEALER)
+    socket.setsockopt(zmq.LINGER, linger)
+    socket.setsockopt(zmq.TCP_KEEPALIVE, 1)
+
+    socket.curve_secretkey = wallet.curve_sk
+    socket.curve_publickey = wallet.curve_vk
+
+    filename = str(cert_dir / f'{vk}.key')
+    if not os.path.exists(filename):
+        return None
+
+    server_pub, _ = load_certificate(filename)
+
+    socket.curve_serverkey = server_pub
+
+    try:
+        socket.connect(ip)
+    except ZMQBaseError:
+        socket.close()
+        return None
+
+    return socket
+
+
+def discover_peer():
+    pass