Merge pull request #20 from zellular-xyz/dev

v 0.0.2

common/bls.py

@@ -40,7 +40,26 @@ def is_bls_sig_verified(
     signature.setStr(signature_hex.encode("utf-8"))
     return signature.verify(public_key, message.encode("utf-8"))
 
-
+async def gather_signatures(
+        sign_tasks: dict[asyncio.Task, str]
+) -> dict[str, Any] | None:
+    """Gather signatures from nodes until the stake of nodes reaches the threshold"""
+    completed_results = []
+    pending_tasks = list(sign_tasks.keys())
+    stake_percent = 0
+    try:
+        while stake_percent < zconfig.THRESHOLD_PERCENT:
+            done, pending = await asyncio.wait(pending_tasks, return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                node_id = sign_tasks[task]
+                completed_results.append(task.result())
+                stake_percent += 100 * zconfig.NODES[node_id]['stake'] / zconfig.TOTAL_STAKE
+            pending_tasks = pending
+        return completed_results
+    except Exception as error:
+        zlogger.exception(f"An unexpected error occurred: {error}")
+        return completed_results
+
 async def gather_and_aggregate_signatures(
     data: dict[str, Any], node_ids: set[str]
 ) -> dict[str, Any] | None:
@@ -54,7 +73,7 @@ async def gather_and_aggregate_signatures(
 
     message: str = utils.gen_hash(json.dumps(data, sort_keys=True))
 
-    tasks: list[asyncio.Task] = [
+    sign_tasks: dict[asyncio.Task, str] = {
         asyncio.create_task(
             request_signature(
                 node_id=node_id,
@@ -63,10 +82,16 @@ async def gather_and_aggregate_signatures(
                 message=message,
                 timeout=120,
             )
-        )
+        ) : node_id 
         for node_id in node_ids
-    ]
-    signatures: list[dict[str, Any] | None] = await asyncio.gather(*tasks)
+    }
+    try:
+        completed_results = await asyncio.wait_for(gather_signatures(sign_tasks), timeout=zconfig.AGGREGATION_TIMEOUT)
+    except asyncio.TimeoutError:
+        zlogger.exception(f"Aggregation of signatures timed out after {zconfig.AGGREGATION_TIMEOUT} seconds.")
+        return None
+
+    signatures: list[dict[str, Any] | None] = completed_results
     signatures_dict: dict[str, dict[str, Any] | None] = dict(zip(node_ids, signatures))
     nonsigners = [k for k, v in signatures_dict.items() if v is None]
     nonsigners += list(set(zconfig.NODES.keys()) - node_ids - set(zconfig.NODE["id"]))

config.py

@@ -51,6 +51,7 @@ def validate_env_variables() -> None:
             "ZSEQUENCER_SEND_TXS_INTERVAL",
             "ZSEQUENCER_SYNC_INTERVAL",
             "ZSEQUENCER_FINALIZATION_TIME_BORDER",
+            "ZSEQUENCER_SIGNATURES_AGGREGATION_TIMEOUT"
         ]
 
         missing_vars: list[str] = [var for var in required_vars if not os.getenv(var)]
@@ -91,6 +92,9 @@ def load_environment_variables(self):
         self.FINALIZATION_TIME_BORDER: int = int(
             os.getenv("ZSEQUENCER_FINALIZATION_TIME_BORDER", "120")
         )
+        self.AGGREGATION_TIMEOUT: int = int(
+            os.getenv("ZSEQUENCER_SIGNATURES_AGGREGATION_TIMEOUT", "5")
+        )
         self.NODES_FILE: str = os.getenv("ZSEQUENCER_NODES_FILE", "nodes.json")
         self.NODES: dict[str, dict[str, Any]] = self.load_json_file(self.NODES_FILE)
         for node in self.NODES.values():

examples/runner.py

@@ -23,6 +23,7 @@
 ZSEQUENCER_SEND_TXS_INTERVAL: float = 0.01
 ZSEQUENCER_SYNC_INTERVAL: float = 0.01
 ZSEQUENCER_FINALIZATION_TIME_BORDER: int = 120
+ZSEQUENCER_SIGNATURES_AGGREGATION_TIMEOUT = 5
 APP_NAME: str = "simple_app"
 
 
@@ -118,8 +119,9 @@ def main() -> None:
             "ZSEQUENCER_SEND_TXS_INTERVAL": str(ZSEQUENCER_SEND_TXS_INTERVAL),
             "ZSEQUENCER_SYNC_INTERVAL": str(ZSEQUENCER_SYNC_INTERVAL),
             "ZSEQUENCER_FINALIZATION_TIME_BORDER": str(
-                ZSEQUENCER_FINALIZATION_TIME_BORDER
-            ),
+                ZSEQUENCER_FINALIZATION_TIME_BORDER),
+            "ZSEQUENCER_SIGNATURES_AGGREGATION_TIMEOUT": str(
+                ZSEQUENCER_SIGNATURES_AGGREGATION_TIMEOUT),
             "ZSEQUENCER_ENV_PATH": f"{DST_DIR}/node{i + 1}",
         }
 

node/tasks.py

@@ -175,6 +175,11 @@ def is_sync_point_signature_verified(
     nonsigners: list[str],
 ) -> bool:
     """Verify the BLS signature of a synchronization point."""
+    nonsigners_stake = sum([zconfig.NODES[node_id]['stake'] for node_id in nonsigners])
+    if 100 * nonsigners_stake / zconfig.TOTAL_STAKE > 100 - zconfig.THRESHOLD_PERCENT:
+        zlogger.exception("Invalid signature from sequencer")
+        return False
+
     public_key: attestation.G2Point = bls.get_signers_aggregated_public_key(nonsigners)
     message: str = utils.gen_hash(
         json.dumps(