Zellular

aggregator.py

@@ -27,6 +27,7 @@ def __init__(self, config):
         self.web3 = Web3(Web3.HTTPProvider(self.config["eth_rpc_url"]))
         self.__load_ecdsa_key()
         self.__load_clients()
+        self.__load_zellular()
         self.__load_task_manager()
         self.__load_bls_aggregation_service()
         self.tasks = {}
@@ -59,27 +60,12 @@ def start_server(self):
         self.app.run(host=host, port=port)
 
     def send_new_task(self, i):
-        tx = self.task_manager.functions.createNewTask(
-            i, 100, nums_to_bytes([0])
-        ).build_transaction({
-            "from": self.aggregator_address,
-            "gas": 2000000,
-            "gasPrice": self.web3.to_wei("20", "gwei"),
-            "nonce": self.web3.eth.get_transaction_count(
-                self.aggregator_address
-            ),
-            "chainId": self.web3.eth.chain_id,
-        })
-        signed_tx = self.web3.eth.account.sign_transaction(
-            tx, private_key=self.aggregator_ecdsa_private_key
-        )
-        tx_hash = self.web3.eth.send_raw_transaction(
-            signed_tx.rawTransaction
-        )
-        receipt = self.web3.eth.wait_for_transaction_receipt(tx_hash)
-        event = self.task_manager.events.NewTaskCreated().process_log(receipt['logs'][0])
-
-        task_index = event['args']['taskIndex']
+        task = {
+            "numberToBeSquared": i,
+            "quorumNumbers": [0],
+            "quorumThresholdPercentage": 100,
+        }
+        task_index = self.zellular.send([task], blocking=True)
         logger.info(f"Successfully sent the new task {task_index}")
         self.bls_aggregation_service.initialize_new_task(
             task_index=task_index,
@@ -165,6 +151,12 @@ def __load_clients(self):
         )
         self.clients = build_all(cfg, self.aggregator_ecdsa_private_key, logger)
 
+    def __load_zellular(self):
+        operators = zellular.get_operators()
+        base_url = random.choice(operators)["socket"]
+        app_name = "incredible-squaring"
+        self.zellular = zellular.Zellular(app_name, base_url)
+
     def __load_task_manager(self):
         service_manager_address = self.clients.avs_registry_writer.service_manager_addr
         with open("abis/IncredibleSquaringServiceManager.json") as f:

squaring_operator.py

@@ -11,6 +11,7 @@
 from eigensdk.chainio.clients.builder import BuildAllConfig, build_all
 from eigensdk.crypto.bls.attestation import KeyPair
 from eigensdk._types import Operator
+import zellular
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -21,7 +22,7 @@ def __init__(self, config):
         self.__load_bls_key()
         self.__load_ecdsa_key()
         self.__load_clients()
-        self.__load_task_manager()
+        self.__load_zellular()
         if config["register_operator_on_startup"] == 'true':
             self.register()
         # operator id can only be loaded after registration
@@ -47,18 +48,17 @@ def register(self):
 
     def start(self):
         logger.info("Starting Operator...")
-        event_filter = self.task_manager.events.NewTaskCreated.create_filter(
-            fromBlock="latest"
-        )
-        while True:
-            for event in event_filter.get_new_entries():
+        index = self.zellular.get_last_finalized()["index"]
+
+        for batch, index in self.zellular.batches(after=index):
+            events = json.loads(batch)
+            for i, event in enumerate(events):
                 logger.info(f"New task created: {event}")
                 self.process_task_event(event)
-            time.sleep(3)
 
     def process_task_event(self, event):
-        task_id = event["args"]["taskIndex"]
-        number_to_be_squared = event["args"]["task"]["numberToBeSquared"]
+        task_id = event["taskIndex"]
+        number_to_be_squared = event["numberToBeSquared"]
         number_squared = number_to_be_squared ** 2
         encoded = eth_abi.encode(["uint32", "uint256"], [task_id, number_squared])
         hash_bytes = Web3.keccak(encoded)
@@ -72,7 +72,6 @@ def process_task_event(self, event):
             "number_to_be_squared": number_to_be_squared,
             "number_squared": number_squared,
             "signature": signature,
-            "block_number": event['blockNumber'],
             "operator_id": "0x" + self.operator_id.hex(),
         }
         logger.info(f"Submitting result for task to aggregator {data}")
@@ -110,22 +109,11 @@ def __load_clients(self):
         )
         self.clients = build_all(cfg, self.operator_ecdsa_private_key, logger)
 
-    def __load_task_manager(self):
-        web3 = Web3(Web3.HTTPProvider(self.config["eth_rpc_url"]))
-
-        service_manager_address = self.clients.avs_registry_writer.service_manager_addr
-        with open("abis/IncredibleSquaringServiceManager.json") as f:
-            service_manager_abi = f.read()
-        service_manager = web3.eth.contract(
-            address=service_manager_address, abi=service_manager_abi
-        )
-
-        task_manager_address = (
-            service_manager.functions.incredibleSquaringTaskManager().call()
-        )
-        with open("abis/IncredibleSquaringTaskManager.json") as f:
-            task_manager_abi = f.read()
-        self.task_manager = web3.eth.contract(address=task_manager_address, abi=task_manager_abi)
+    def __load_zellular(self):
+        operators = zellular.get_operators()
+        base_url = random.choice(operators)["socket"]
+        app_name = "incredible-squaring"
+        self.zellular = zellular.Zellular(app_name, base_url)
 
     def __load_operator_id(self):
         self.operator_id = self.clients.avs_registry_reader.get_operator_id(