JAMNP-S

pyjamaz/app.py

@@ -144,6 +144,15 @@ async def import_block_from_bytes(self, data):
             logging.info(f"Syncing in progress, current timeslot={self.working_state.timeslot.number}")
 
 
+    async def import_queue_add_blocks(self, blocks: List[Block], process: bool=False, on_success: MESSAGE_TYPES=None, on_failure: MESSAGE_TYPES=None):
+        async with self.import_lock:
+            for block in blocks:
+                self.import_queue.append(block)
+
+        if process:
+            await self.process_import_queue(on_success=on_success, on_failure=on_failure)
+
+
     async def import_block_from_json(self, data):
         DEBUG and logging.debug(f"📦 Importing block from json")
         block = Block.from_json(data)
@@ -152,35 +161,47 @@ async def import_block_from_json(self, data):
 
     async def requested_blocks_from_json(self, data):
         block_list = [Block.from_json(block_data) for block_data in data]
-        for block in block_list:
-            self.import_queue.append(Block.from_codec_type(block))
-            logging.info(f"📦 Queue block requested #{block.header.timeslot}")
+        async with self.import_lock:
+            for block in block_list:
+                self.import_queue.append(Block.from_codec_type(block))
+                logging.info(f"📦 Queue block requested #{block.header.timeslot}")
         await self.process_import_queue()
 
 
     async def requested_blocks_from_bytes(self, data):
-        block_list = Vec(Block.to_codec_def()).new()
-        block_list.decode(JamBytes(data))
-        for block_bytes in block_list:
-            block = Block.from_codec_type(block_bytes)
-            self.import_queue.append(block)
+        async with self.import_lock:
+            block_list = Vec(Block.to_codec_def()).new()
+            block_list.decode(JamBytes(data))
+            for block_bytes in block_list:
+                block = Block.from_codec_type(block_bytes)
+                self.import_queue.append(block)
 
         await self.process_import_queue()
 
 
-    async def process_import_queue(self):
+    async def process_import_queue(self, on_success:MESSAGE_TYPES=None, on_failure:MESSAGE_TYPES=None):
         async with self.import_lock:
             sorted_blocks = sorted(self.import_queue, key=lambda x: x.header.timeslot)
             self.import_queue = []
 
-        for block in sorted_blocks:
-            # TODO: protocol should only import blocks from this point on -> fix the block_request
-            if self.working_state.timeslot.number >= block.header.timeslot:
-                DEBUG and logging.debug(f" TEMP BREAK block from process_import_queue: {block.header.timeslot}")
-                continue
+        try:
+            for block in sorted_blocks:
+                # TODO: protocol should only import blocks from this point on -> fix the block_request
+                if self.working_state.timeslot.number >= block.header.timeslot:
+                    DEBUG and logging.debug(f" TEMP BREAK block from process_import_queue: {block.header.timeslot}")
+                    continue
 
-            await self.import_block(block)
-            DEBUG and logging.debug(f'✅ Block {block.header.timeslot} successfully imported from process_import_queue.')
+                await self.import_block(block)
+                DEBUG and logging.debug(
+                    f'✅ Block {block.header.timeslot} successfully imported from process_import_queue.')
+
+            if on_success:
+                await self.pubsub.publish(PubSubSignal(topic=on_success, data=None))
+
+        except Exception as e:
+            if on_failure:
+                await self.pubsub.publish(PubSubSignal(topic=on_failure, data=None))
+            logging.error(f"Error processing import queue: {e}")
 
 
     async def initialize(self, header: Optional[Header] = None, produce=False):
@@ -211,8 +232,6 @@ async def initialize(self, header: Optional[Header] = None, produce=False):
             self.working_state = self.retrieve_jam_state()
             DEBUG and logging.debug(f"Updated working state to state_root={format_hash(self.working_state.state_root)}")
 
-        else:
-            DEBUG and logging.debug("StateStorage: State already matches requested state root, no updating required")
 
     def retrieve_ancestor_headers(self, block_hash: bytes) -> List[Header]:
         """
@@ -715,6 +734,9 @@ async def store_block_header(self, header: Header):
         self.block_db.put(
             b'block_number:' + header.hash, header.timeslot.to_bytes(length=4, byteorder='little')
         )
+        self.block_db.put(
+            b'block_child:' + header.parent, header.hash
+        )
 
     def retrieve_block(self, timeslot: int) -> Optional[Block]:
         block_data = self.block_db.get(b'block:' + timeslot.to_bytes(length=4, byteorder='little'))
@@ -743,6 +765,9 @@ async def store_finalized_head(self, block_hash: bytes):
     def retrieve_block_hash(self, timeslot: int) -> Optional[bytes]:
         return self.block_db.get(b'block_hash:' + timeslot.to_bytes(length=4, byteorder='little'))
 
+    def retrieve_block_child_hash(self, block_hash: bytes) -> Optional[bytes]:
+        return self.block_db.get(b'block_child:' + block_hash)
+
     def should_produce_block(self, timeslot: int, safrole_state: SafroleState) -> bool:
         slot_phase_index = timeslot % EPOCH_TIMESLOTS
 
@@ -926,24 +951,24 @@ async def produce_block(
                 ring_public_keys = [v.bandersnatch for v in safrole_state.validators]
                 ring_context = RingContext(self.config.ring_data, ring_public_keys)
 
-                self.block_extrinsic.add_own_ticket(
-                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index()
+                await self.block_extrinsic.add_own_ticket(
+                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index(), pubsub=self.pubsub
                 )
 
-                self.block_extrinsic.add_own_ticket(
-                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index()
+                await self.block_extrinsic.add_own_ticket(
+                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index(), pubsub=self.pubsub
                 )
 
-                self.block_extrinsic.add_own_ticket(
-                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index()
+                await self.block_extrinsic.add_own_ticket(
+                    ring_context, entropy, self.config.keys.bandersnatch, self.get_author_index(), pubsub=self.pubsub
                 )
 
         extrinsic = Extrinsic(
-            tickets=self.block_extrinsic.collect_tickets(),
+            tickets=await self.block_extrinsic.collect_tickets(),
             disputes=ExtrinsicDisputes(verdicts=[], culprits=[], faults=[]),
             preimages=self.block_extrinsic.collect_preimages(self.working_state.services),
             assurances=self.block_extrinsic.collect_assurances(),
-            guarantees=self.block_extrinsic.collect_guarantees(),
+            guarantees=await self.block_extrinsic.collect_guarantees(),
         )
 
         header = Header(

pyjamaz/cli.py

@@ -1,5 +1,6 @@
 import bisect
 import logging
+import re
 import traceback
 from asyncio import CancelledError
 from datetime import datetime, timezone
@@ -27,15 +28,14 @@
 from pyjamaz.logger import setup_logging
 from pyjamaz.models.app import Trace, TraceGenesis
 from pyjamaz.models.state import STORAGE_KEY_MAPPING, ServiceAccount
-from pyjamaz.rpc.ws_server import start_rpc_server, WebSocketServer
+from pyjamaz.transport.rpc.ws_server import start_rpc_server, WebSocketServer
 from pyjamaz.runtime.node_runtime import NodeRuntime
 from pyjamaz.settings import GP_VERSION, APP_VERSION, STORAGE_ENGINE, DEBUG
 from pyjamaz.storage import InMemoryStorageEngine, RocksDBStorageEngine
 from pyjamaz.models.block import Block, Header, Extrinsic
 from pyjamaz.fuzzer import FuzzerMessage, InitializeMessage, FuzzerTarget, FuzzerSession, AncestryItem
-from pyjamaz.transport.cert import generate_cert, write_cert
-from pyjamaz.transport.protocol_fs import FSProtocol
-from pyjamaz.transport.protocol_jamnp_s import JAMNPS
+from pyjamaz.transport.cert import generate_cert, read_cert_public_key, write_cert
+from pyjamaz.transport.jamnp_s.network import JAMNPS
 
 from pyjamaz.transport.pubsub import PubSub, PubSubSignal
 from pyjamaz.utils import format_hash, quic_peer_id
@@ -124,21 +124,6 @@ async def cli_import_block(self, block: Block, dry_run=False):
     return cli_import_block
 
 
-def wrap_produced_block_jamnp(app: PyjamazApp, traces_dir, np_protocol: JAMNPS):
-    async def produced_block_jamnp(block: Block):
-        await np_protocol.broadcast_block(block)
-
-    return produced_block_jamnp
-
-
-def wrap_produced_block_fs(app: PyjamazApp, traces_dir, fs_protocol: FSProtocol):
-    async def produced_block_fs(block: Block):
-        await app.import_block(block)
-        await fs_protocol.broadcast_block(block)
-
-    return produced_block_fs
-
-
 async def initialize_app(
         read_state=True,
         storage_engine='memory',
@@ -237,6 +222,7 @@ async def run(seed, port, ts, culprit, block_dir, record_traces, custom_db_path,
     # Note: Add packages that need a different logging level here
     log_package_overrides = {
         "pyjamaz.transport": log_level,
+        #"pyjamaz.transport.jamnp_s": logging.DEBUG,
         "quic": logging.WARNING,
         "numba": logging.WARNING,
         "numba.core": logging.WARNING
@@ -254,9 +240,7 @@ async def run(seed, port, ts, culprit, block_dir, record_traces, custom_db_path,
 
     db_path = custom_db_path or default_db_path
 
-    network_bootstrap = ts is None
-    if network_bootstrap:
-        ts = 0
+    ts = ts or 0
 
     #TODO: currently it is not possible to provide a hard unix timestamp (only deltas)
     current_time = time.time()
@@ -279,7 +263,6 @@ async def run(seed, port, ts, culprit, block_dir, record_traces, custom_db_path,
     for header in app.retrieve_ancestor_headers(app.state_storage.finalized_block_hash):
         app.state_storage.add_ancestor(header)
 
-    app.network_bootstrap = network_bootstrap
     common_era_time = datetime.fromtimestamp(app.config.common_era, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
 
     if replay_blocks:
@@ -296,6 +279,14 @@ async def run(seed, port, ts, culprit, block_dir, record_traces, custom_db_path,
     logging.info(f'🌐 Peer ID: {quic_peer_id(app.config.keys.ed25519.public_key)}')
     logging.info(f'🔑 Bandersnatch public: {format_hash(app.config.keys.bandersnatch.public_key)}')
     logging.info(f'🔑 Ed25519 public: {format_hash(app.config.keys.ed25519.public_key)}')
+    #TODO: for now we assume we are either in the validator pool or are a listening node
+    validator_index = app.get_validator_index()
+    if validator_index is None:
+        logging.warning(
+            "⚠️ Local key not in validator pool; acting as non participating node"
+        )
+    else:
+        logging.info(f"🧩 Validator pool index: #{validator_index}")
     logging.info(f'🗓️ Common Era: {app.config.common_era} ({common_era_time})')
     logging.info(f'🌲 State trie root: {format_hash(app.working_state.state_root)}')
     logging.info(f'📦 Finalized block: {format_hash(app.state_storage.finalized_block_hash)}')
@@ -313,54 +304,64 @@ async def run(seed, port, ts, culprit, block_dir, record_traces, custom_db_path,
     try:
         async with anyio.create_task_group() as tg:
 
+            # Create and start runtime
+            app.runtime = NodeRuntime(app)
+            app.runtime.start(tg)
+
             # TODO: we need to start this manually in all event loops, make an AppFactory that handles this in a generic way
             # Create a subscriber to process incoming messages (fx from a protocol)
             tg.start_soon(app.pubsub.process_messages)
 
             # Start WebSocket server
             tg.start_soon(start_rpc_server, rpc_server)
 
-            if False and block_dir:
-                # TODO remove
-                logging.info(f"👀 Watching directory: {block_dir} for new blocks...")
-                fs_protocol = FSProtocol(block_dir, app)
-                app.protocol = fs_protocol
-                app.pubsub.subscribe(MESSAGE_TYPES.PRODUCED_BLOCK, wrap_produced_block_fs(app, record_traces, fs_protocol))
-                app.pubsub.subscribe(MESSAGE_TYPES.RECEIVED_BLOCK, app.import_block_from_json)
-                app.pubsub.subscribe(MESSAGE_TYPES.REQUESTED_BLOCKS, app.requested_blocks_from_json)
-                tg.start_soon(fs_protocol.listen)
-            else:
-                certificate_file = os.path.join(db_path, "cert.pem")
-                pk_file = os.path.join(db_path, "cert.key")
-                nps_protocol = JAMNPS(host, port, certificate_file, pk_file, app)
-                app.protocol = nps_protocol
-                app.pubsub.subscribe(MESSAGE_TYPES.PRODUCED_BLOCK, wrap_produced_block_jamnp(app, record_traces, nps_protocol))
-                app.pubsub.subscribe(MESSAGE_TYPES.RECEIVED_BLOCK, app.import_block_from_bytes)
-                app.pubsub.subscribe(MESSAGE_TYPES.REQUESTED_BLOCKS, app.requested_blocks_from_bytes)
-                tg.start_soon(nps_protocol.listen)
-
-                for validator in app.working_state.safrole.validators:
-                    # The validators' IP-layer endpoints are given as IPv6/port combinations,
-                    # to be found in the first 18 bytes of validator metadata, with the first 16 bytes being the IPv6 address and
-                    # the latter 2 being a little endian representation of the port.
-
-                    validator_port = validator.get_metadata_port()
-                    validator_address = validator.get_metadata_ipaddress()
-
-                    if validator.ed25519 == app.config.keys.ed25519.public_key:
-                        DEBUG and logging.debug(
-                            f'Skipping own node ({validator_address}:{validator_port})'
-                        )
-                        continue
-
-                    DEBUG and logging.debug(f'Connecting to node {validator_address}:{validator_port}')
-                    tg.start_soon(nps_protocol.connect, validator_address, validator_port)
+            certificate_file = os.path.join(db_path, "cert.pem")
+            pk_file = os.path.join(db_path, "cert.key")
+            await ensure_certificate_matches_seed(db_path, seed)
+            nps_protocol = JAMNPS(host, port, certificate_file, pk_file, app)
+            app.protocol = nps_protocol
+            tg.start_soon(nps_protocol.listen)
+            tg.start_soon(nps_protocol.check_connections)
+
+            if bootnode:
+                # logging.debug(f'Connecting to node {validator_address}:{validator_port}')
+                # tg.start_soon(nps_protocol.connect, validator_address, validator_port)
+
+                # ecjn4brac2kgu25kiykefww6p6ai7noueo6p5af5tnwjgra4eisya@172.16.238.11:40001
+                conn = re.match(
+                    r"^(?P<key>[A-Za-z0-9]+)"
+                    r"@"
+                    r"(?P<addr>(?:\d{1,3}\.){3}\d{1,3})"
+                    r":"
+                    r"(?P<port>\d{1,5})$",
+                    bootnode,
+                )
 
-            await anyio.sleep(ts - time.time())
+                logging.debug(f'Connecting to bootnode {conn["key"]} at {conn["addr"]}:{conn["port"]}')
+                try:
+                    await nps_protocol.connect(conn["addr"], int(conn["port"]), None)
+                except Exception as exc:
+                    traceback.print_exc()
+            # else:
+            #     # for validator in app.state.safrole.validators:
+            #     #     # The validators' IP-layer endpoints are given as IPv6/port combinations,
+            #     #     # to be found in the first 18 bytes of validator metadata, with the first 16 bytes being the IPv6 address and
+            #     #     # the latter 2 being a little endian representation of the port.
+            #     #
+            #     #     validator_port = validator.get_metadata_port()
+            #     #     validator_address = validator.get_metadata_ipaddress()
+            #     #
+            #     #     if validator.ed25519 == app.config.keys.ed25519.public_key:
+            #     #         logging.debug(
+            #     #             f'Skipping own node ({validator_address}:{validator_port})'
+            #     #         )
+            #     #         continue
+            #     #
+            #     #     logging.debug(f'Connecting to node {validator_address}:{validator_port}')
+            #     #     tg.start_soon(nps_protocol.connect, validator_address, validator_port)
+            #     #tg.start_soon(nps_protocol.connect, "127.0.0.1", 40001)
 
-            # Create and start runtime
-            app.runtime = NodeRuntime(app)
-            app.runtime.start(tg)
+            await anyio.sleep(ts - time.time())
 
     except (KeyboardInterrupt, CancelledError):
         logging.info("Stopping node...")
@@ -414,13 +415,33 @@ async def init_certificate(db_path, seed):
     pk_pem, cert_pem = generate_cert(
         keys,
         ips="127.0.0.1",    #TODO: hardcoded for now
-        alternative_name="e3r2oc62zwfj3crnuifuvsxvbtlzetk4o5qyhetkhagsc2fgl2oka",
     )
     pk_file = os.path.join(db_path, "cert.key")
     pem_file = os.path.join(db_path, "cert.pem")
     write_cert(pk_pem, pk_file, cert_pem, pem_file)
 
 
+async def ensure_certificate_matches_seed(db_path, seed):
+    cert_file = os.path.join(db_path, "cert.pem")
+    pk_file = os.path.join(db_path, "cert.key")
+    expected_public_key = Keys.from_seed(bytes.fromhex(seed[2:])).ed25519.public_key
+
+    if not os.path.exists(cert_file) or not os.path.exists(pk_file):
+        logging.info("🔐 JAMNP-S certificate missing; generating a new certificate")
+        await init_certificate(db_path, seed)
+        return
+
+    actual_public_key = read_cert_public_key(cert_file)
+    if actual_public_key == expected_public_key:
+        return
+
+    logging.warning(
+        "🔐 JAMNP-S certificate key does not match the provided seed; regenerating certificate "
+        f"(cert={format_hash(actual_public_key)}, expected={format_hash(expected_public_key)})"
+    )
+    await init_certificate(db_path, seed)
+
+
 @main.command()
 @click.option('--seed', 'seed', type=str, help="Seed to use for validator keys")
 @click.option('--chainspec', 'chainspec', type=click.Choice(['dev', 'docker']), help="Chainspec to use as genesis", default='dev', show_default=True)

pyjamaz/constants.py

@@ -38,8 +38,6 @@
 
 class MESSAGE_TYPES(Enum):
     PRODUCED_BLOCK    = "PRODUCED_BLOCK"
-    RECEIVED_BLOCK    = "RECEIVED_BLOCK"
-    REQUESTED_BLOCKS    = "REQUESTED_BLOCKS"
 
     STATISTICS = "statistics"
     SERVICE_ACCOUNT = "service_account"
@@ -50,6 +48,12 @@ class MESSAGE_TYPES(Enum):
     FINALIZED_BLOCK = "finalized_block"
     WORK_PACKAGE_STATUS = "work_package_status"
 
+    TICKET_ADD = "ticket_add"
+
+    CE128_SUCCESS = "CE128_SUCCESS"
+    CE128_FAILURE = "CE128_FAILURE"
+
+
 PVM_MARSHALLING_OFFSET_ACCUMULATE = 5
 PVM_MARSHALLING_OFFSET_TRANSFER = 10
 PVM_MARSHALLING_OFFSET_REFINE = 0