Merge pull request #8 from zellular-xyz/dev

v 0.0.1

common/bls.py

@@ -45,13 +45,13 @@ async def gather_and_aggregate_signatures(
     data: dict[str, Any], node_ids: set[str]
 ) -> dict[str, Any] | None:
     """Gather and aggregate signatures from nodes."""
-    if len(node_ids) < zconfig.THRESHOLD_NUMBER:
+    stake = sum([zconfig.NODES[node_id]['stake'] for node_id in node_ids])
+    if 100 * stake / zconfig.TOTAL_STAKE < zconfig.THRESHOLD_PERCENT:
         return None
 
     if not node_ids.issubset(set(zconfig.NODES.keys())):
         return None
 
-
     message: str = utils.gen_hash(json.dumps(data, sort_keys=True))
 
     tasks: list[asyncio.Task] = [
@@ -69,7 +69,9 @@ async def gather_and_aggregate_signatures(
     signatures: list[dict[str, Any] | None] = await asyncio.gather(*tasks)
     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]
-    if len(signatures) - len(nonsigners) + 1 < zconfig.THRESHOLD_NUMBER:
+    nonsigners += list(set(zconfig.NODES.keys()) - node_ids - set(zconfig.NODE["id"]))
+    nonsigners_stake = sum([zconfig.NODES[node_id]['stake'] for node_id in nonsigners])
+    if 100 * nonsigners_stake / zconfig.TOTAL_STAKE > 100 - zconfig.THRESHOLD_PERCENT:
         return None
 
     data["signature"] = bls_sign(message)
@@ -79,7 +81,6 @@ async def gather_and_aggregate_signatures(
         [sig for sig in signatures if sig]
     )
 
-    nonsigners += list(set(zconfig.NODES.keys()) - node_ids - set(zconfig.NODE["id"]))
     return {
         "message": message,
         "signature": aggregated_signature,

common/db.py

@@ -32,6 +32,7 @@ def _initialize(self) -> None:
 
         self.apps: dict[str, Any] = {}
         self.keys: dict[str, Any] = {}
+        self.is_sequencer_down: bool = False
         self.load_state()
 
     def load_state(self) -> None:

common/utils.py

@@ -90,8 +90,9 @@ def get_next_sequencer_id(old_sequencer_id: str) -> str:
 
 def is_switch_approved(proofs: list[dict[str, Any]]) -> bool:
     """Check if the switch to a new sequencer is approved."""
-    approvals: int = sum(1 for proof in proofs if is_dispute_approved(proof))
-    return approvals >= zconfig.THRESHOLD_NUMBER
+    node_ids = [proof['node_id'] for proof in proofs if is_dispute_approved(proof)]
+    stake = sum([zconfig.NODES[node_id]['stake'] for node_id in node_ids])
+    return 100 * stake / zconfig.TOTAL_STAKE >= zconfig.THRESHOLD_PERCENT
 
 
 def is_dispute_approved(proof: dict[str, Any]) -> bool:

config.py

@@ -47,7 +47,7 @@ def validate_env_variables() -> None:
             "ZSEQUENCER_SNAPSHOT_CHUNK",
             "ZSEQUENCER_REMOVE_CHUNK_BORDER",
             "ZSEQUENCER_SNAPSHOT_PATH",
-            "ZSEQUENCER_THRESHOLD_NUMBER",
+            "ZSEQUENCER_THRESHOLD_PERCENT",
             "ZSEQUENCER_SEND_TXS_INTERVAL",
             "ZSEQUENCER_SYNC_INTERVAL",
             "ZSEQUENCER_FINALIZATION_TIME_BORDER",
@@ -97,8 +97,8 @@ def load_environment_variables(self):
             node["public_key_g2"] = attestation.new_zero_g2_point()
             node["public_key_g2"].setStr(node["public_key"].encode("utf-8"))
 
-        self.THRESHOLD_NUMBER: int = int(
-            os.getenv("ZSEQUENCER_THRESHOLD_NUMBER", str(len(self.NODES)))
+        self.THRESHOLD_PERCENT: int = float(
+            os.getenv("ZSEQUENCER_THRESHOLD_PERCENT", str(100))
         )
         self.NODE: dict[str, Any] = next(
             (n for n in self.NODES.values() if n["public_key"] == self.PUBLIC_KEY), {}
@@ -111,6 +111,7 @@ def load_environment_variables(self):
         self.AGGREGATED_PUBLIC_KEY: attestation.G2Point = (
             self.get_aggregated_public_key()
         )
+        self.TOTAL_STAKE = sum([node['stake'] for node in self.NODES.values()])
 
         self.SEQUENCER: dict[str, Any] = self.NODES["1"]
 

examples/runner.py

@@ -14,7 +14,7 @@
 
 NUM_INSTANCES: int = 3
 BASE_PORT: int = 6000
-THRESHOLD_NUMBER: int = 2
+THRESHOLD_PERCENT: int = 60
 DST_DIR: str = "/tmp/zellular_dev_net"
 NODES_FILE: str = "/tmp/zellular_dev_net/nodes.json"
 APPS_FILE: str = "/tmp/zellular_dev_net/apps.json"
@@ -57,6 +57,7 @@ def generate_privates_and_nodes_info() -> tuple[list[str], dict[str, Any]]:
             "address": address,
             "host": "127.0.0.1",
             "port": str(BASE_PORT + i + 1),
+            "stake": 10,
         }
 
     return privates_list, nodes_info_dict
@@ -113,7 +114,7 @@ def main() -> None:
             "ZSEQUENCER_SNAPSHOT_CHUNK": str(ZSEQUENCER_SNAPSHOT_CHUNK),
             "ZSEQUENCER_REMOVE_CHUNK_BORDER": str(ZSEQUENCER_REMOVE_CHUNK_BORDER),
             "ZSEQUENCER_SNAPSHOT_PATH": data_dir,
-            "ZSEQUENCER_THRESHOLD_NUMBER": str(THRESHOLD_NUMBER),
+            "ZSEQUENCER_THRESHOLD_PERCENT": str(THRESHOLD_PERCENT),
             "ZSEQUENCER_SEND_TXS_INTERVAL": str(ZSEQUENCER_SEND_TXS_INTERVAL),
             "ZSEQUENCER_SYNC_INTERVAL": str(ZSEQUENCER_SYNC_INTERVAL),
             "ZSEQUENCER_FINALIZATION_TIME_BORDER": str(

node/routes.py

@@ -54,7 +54,6 @@ def put_transactions() -> Response:
     error_message: str = utils.validate_request(req_data, required_keys)
     if error_message:
         return error_response(ErrorCodes.INVALID_REQUEST, error_message)
-
     zdb.init_txs(req_data["app_name"], req_data["transactions"])
     return success_response(data={}, message="The transactions received successfully.")
 
@@ -86,8 +85,7 @@ def post_dispute() -> Response:
 
     if req_data["sequencer_id"] != zconfig.SEQUENCER["id"]:
         return error_response(ErrorCodes.INVALID_SEQUENCER)
-
-    if zdb.has_missed_txs():
+    if zdb.has_missed_txs() or zdb.is_sequencer_down:
         timestamp: int = int(time.time())
         data: dict[str, Any] = {
             "node_id": zconfig.NODE["id"],
@@ -97,10 +95,9 @@ def post_dispute() -> Response:
             "signature": utils.eth_sign(f'{zconfig.SEQUENCER["id"]}{timestamp}'),
         }
         return success_response(data=data)
-
     zdb.init_txs(req_data["app_name"], req_data["txs"])
     return error_response(ErrorCodes.ISSUE_NOT_FOUND)
-
+    
 
 @node_blueprint.route("/switch", methods=["POST"])
 def post_switch_sequencer() -> Response:

node/tasks.py

@@ -31,6 +31,7 @@ def send_txs() -> None:
     """Send transactions for all apps."""
     for app_name in zconfig.APPS:
         send_app_txs(app_name)
+    zdb.is_sequencer_down = False
 
 
 def send_app_txs(app_name: str) -> None:
@@ -81,6 +82,7 @@ def send_app_txs(app_name: str) -> None:
     except Exception:
         zlogger.exception("An unexpected error occurred:")
         zdb.add_missed_txs(app_name=app_name, txs=initialized_txs)
+        zdb.is_sequencer_down = True
 
     check_finalization()
 
@@ -195,14 +197,13 @@ def is_sync_point_signature_verified(
 
 def send_dispute_requests() -> None:
     """Send dispute requests if there are missed transactions."""
-    if not zdb.has_missed_txs() or zdb.pause_node.is_set():
+    if (not zdb.has_missed_txs() and not zdb.is_sequencer_down) or zdb.pause_node.is_set():
         return
 
     timestamp: int = int(time.time())
     new_sequencer_id: str = utils.get_next_sequencer_id(
         old_sequencer_id=zconfig.SEQUENCER["id"]
     )
-
     proofs: list[dict[str, Any]] = []
     proofs.append(
         {
@@ -220,7 +221,7 @@ def send_dispute_requests() -> None:
 
         for app_name in zconfig.APPS.keys():
             missed_txs: dict[str, Any] = zdb.get_missed_txs(app_name)
-            if not missed_txs:
+            if len(missed_txs) == 0:
                 continue
 
             try:
@@ -234,6 +235,17 @@ def send_dispute_requests() -> None:
             except Exception:
                 zlogger.exception("An unexpected error occurred:")
 
+        if zdb.is_sequencer_down:
+            try:
+                response: dict[str, Any] | None = send_dispute_request(
+                    node, app_name = '', missed_txs = []
+                )
+
+                if response:
+                    proofs.append(response)
+            except Exception:
+                zlogger.exception("An unexpected error occurred:")
+
     if utils.is_switch_approved(proofs):
         zdb.pause_node.set()
         old_sequencer_id, new_sequencer_id = utils.get_switch_parameter_from_proofs(

production.env.example

@@ -7,7 +7,7 @@ ZSEQUENCER_APPS_FILE=
 ZSEQUENCER_SNAPSHOT_CHUNK=
 ZSEQUENCER_REMOVE_CHUNK_BORDER=
 ZSEQUENCER_SNAPSHOT_PATH=
-ZSEQUENCER_THRESHOLD_NUMBER=
+ZSEQUENCER_THRESHOLD_PERCENT=
 ZSEQUENCER_SEND_TXS_INTERVAL=
 ZSEQUENCER_SYNC_INTERVAL=
 ZSEQUENCER_FINALIZATION_TIME_BORDER=

sequencer/tasks.py

@@ -27,7 +27,9 @@ def find_locked_sync_point(app_name: str) -> dict[str, Any] | None:
             for s in sorted_filtered_states
             if s["sequenced_index"] >= state["sequenced_index"]
         }
-        if len(party) >= zconfig.THRESHOLD_NUMBER:
+        stake = sum([zconfig.NODES[node_id]['stake'] for node_id in party])
+        stake += zconfig.NODE["stake"]
+        if 100 * stake / zconfig.TOTAL_STAKE >= zconfig.THRESHOLD_PERCENT:
             return {"state": state, "party": party}
     return None
 
@@ -50,7 +52,9 @@ def find_finalized_sync_point(app_name: str) -> dict[str, Any] | None:
             for s in sorted_filtered_states
             if s["locked_index"] >= state["locked_index"]
         }
-        if len(party) >= zconfig.THRESHOLD_NUMBER:
+        stake = sum([zconfig.NODES[node_id]['stake'] for node_id in party])
+        stake += zconfig.NODE["stake"]
+        if 100 * stake / zconfig.TOTAL_STAKE >= zconfig.THRESHOLD_PERCENT:
             return {"state": state, "party": party}
     return None