Merge branch 'main' of github.com-bread:zero-gravity-labs/zerog-data-…

…avail

.gitmodules

@@ -13,3 +13,6 @@
 [submodule "contracts/lib/eigenlayer-middleware"]
 	path = contracts/lib/eigenlayer-middleware
 	url = https://github.com/Layr-Labs/eigenlayer-middleware
+[submodule "zerog_storage_kv"]
+	path = zerog_storage_kv
+	url = git@github.com:zero-gravity-labs/zerog-storage-kv.git

tests/da_hello_test.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import sys
+sys.path.append("../zerog_storage_kv/tests")
+import time
+
+from da_test_framework.da_test_framework import DATestFramework
+
+
+class DAHelloTest(DATestFramework):
+    def setup_params(self):
+        self.num_blockchain_nodes = 1
+        self.num_nodes = 2
+
+    def run_test(self):
+        self.log.debug("===============================================")
+        self.log.debug("root dir:" + self.root_dir)
+        self.log.debug("blockchain binary:" + self.blockchain_binary)
+        self.log.debug("zgs binary:" + self.zgs_binary)
+        self.log.debug("cli binary:" + self.cli_binary)
+        self.log.debug("kv binary:" + self.kv_binary)
+        self.log.debug("contract path:" + self.contract_path)
+        self.log.debug("token contract path:" + self.token_contract_path)
+        self.log.debug("mine contract path:" + self.mine_contract_path)
+        self.log.debug("localstack binary:" + self.localstack_binary)
+        self.log.debug("da encoder binary:" + self.da_encoder_binary)
+        self.log.debug("da batcher binary:" + self.da_batcher_binary)
+        self.log.debug("da server binary:" + self.da_server_binary)
+        self.log.debug("===============================================")
+
+        self.log.info("hello")
+        time.sleep(3)
+
+
+if __name__ == "__main__":
+    DAHelloTest().main()

tests/da_test_all.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python3
+import os
+import sys
+
+sys.path.append("../zerog_storage_kv/zerog-storage-rust/tests")
+
+from utility.run_all import run_all
+
+if __name__ == "__main__":
+    run_all(test_dir=os.path.dirname(__file__))

tests/da_test_framework/da_batcher.py

@@ -0,0 +1,86 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from utility.utils import blockchain_rpc_port
+from config.node_config import GENESIS_PRIV_KEY
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DABatcher(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log_contract_address,
+            log,
+    ):
+        local_conf = {
+            "log_config_file": "log_config",
+            "log_contract_address": log_contract_address,
+            "blockchain_rpc_endpoint": f"http://127.0.0.1:{blockchain_rpc_port(0)}",
+        }
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_batcher")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_BATCHER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--batcher.pull-interval", "10s",
+                     "--chain.rpc", local_conf['blockchain_rpc_endpoint'],
+                     "--chain.private-key", GENESIS_PRIV_KEY,
+                     "--batcher.finalizer-interval", "20s",
+                     "--batcher.aws.region", "us-east-1",
+                     "--batcher.aws.access-key-id", "localstack",
+                     "--batcher.aws.secret-access-key", "localstack",
+                     "--batcher.aws.endpoint-url", "http://0.0.0.0:4566",
+                     "--batcher.s3-bucket-name", "test-zgda-blobstore",
+                     "--batcher.dynamodb-table-name", "test-BlobMetadata",
+                     "--encoder-socket", "0.0.0.0:34000",
+                     "--batcher.batch-size-limit", "10000",
+                     "--batcher.srs-order", "300000",
+                     "--encoding-timeout", "10s",
+                     "--chain-read-timeout", "12s",
+                     "--chain-write-timeout", "13s",
+                     "--batcher.storage.node-url", "http://0.0.0.0:5678",
+                     "--batcher.storage.node-url", "http://0.0.0.0:6789",
+                     "--batcher.storage.kv-url", "http://0.0.0.0:7890",
+                     "--batcher.storage.kv-stream-id",
+                     "000000000000000000000000000000000000000000000000000000000000f2bd",
+                     "--batcher.storage.flow-contract", local_conf['log_contract_address']]
+
+    def start(self):
+        self.log.info("Start DA batcher")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA batcher")
+        try:
+            super().stop(kill=True, wait=False)
+        except AssertionError as e:
+            err = repr(e)
+            # The batcher will check return_code via rpc when error log exists
+            # that is written when the batcher starts normally.
+            # The exception handling can be removed when rpc is added or the error
+            # is not written when the batcher starts normally.
+            if "no RPC connection" in err:
+                self.log.debug(f"Stop DA encoder: no RPC connection")
+            else:
+                raise e

tests/da_test_framework/da_encoder.py

@@ -0,0 +1,66 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DAEncoder(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log,
+    ):
+        local_conf = dict(log_config_file="log_config")
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_encoder")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_ENCODER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--disperser-encoder.grpc-port", "34000",
+                     "--disperser-encoder.metrics-http-port", "9109",
+                     "--kzg.g1-path", f"{__file_path__}/../../inabox/resources/kzg/g1.point.300000",
+                     "--kzg.g2-path", f"{__file_path__}/../../inabox/resources/kzg/g2.point.300000",
+                     "--kzg.cache-path", f"{__file_path__}/../../inabox/resources/kzg/SRSTables",
+                     "--kzg.srs-order", "300000",
+                     "--kzg.num-workers", "12",
+                     "--disperser-encoder.log.level-std", "trace",
+                     "--disperser-encoder.log.level-file", "trace"]
+
+    def start(self):
+        self.log.info("Start DA encoder")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA encoder")
+        # The encoder will check return_code via rpc when error log exists
+        # that is written when the encoder starts normally.
+        # The exception handling can be removed when rpc is added or the error
+        # is not written when the encoder starts normally.
+        try:
+            super().stop(kill=True, wait=False)
+        except AssertionError as e:
+            err = repr(e)
+            if "no RPC connection" in err:
+                self.log.debug(f"Stop DA encoder: no RPC connection")
+            else:
+                raise e

tests/da_test_framework/da_node_type.py

@@ -0,0 +1,9 @@
+from enum import Enum, unique
+
+
+@unique
+class DANodeType(Enum):
+    DA_LOCAL_STACK = 3
+    DA_ENCODER = 4
+    DA_BATCHER = 5
+    DA_SERVER = 6

tests/da_test_framework/da_server.py

@@ -0,0 +1,53 @@
+import os
+import sys
+import time
+
+sys.path.append("../../zerog_storage_kv/tests")
+
+from test_framework.blockchain_node import TestNode
+from da_test_framework.da_node_type import DANodeType
+
+__file_path__ = os.path.dirname(os.path.realpath(__file__))
+
+
+class DAServer(TestNode):
+    def __init__(
+            self,
+            root_dir,
+            binary,
+            updated_config,
+            log,
+    ):
+        local_conf = dict(log_config_file="log_config")
+
+        local_conf.update(updated_config)
+        data_dir = os.path.join(root_dir, "da_server")
+        # rpc_url = "http://" + local_conf["rpc_listen_address"]
+        super().__init__(
+            DANodeType.DA_SERVER,
+            0,
+            data_dir,
+            None,
+            binary,
+            local_conf,
+            log,
+            None,
+        )
+        self.args = [binary, "--disperser-server.grpc-port", "51001",
+                     "--disperser-server.s3-bucket-name", "test-zgda-blobstore",
+                     "--disperser-server.dynamodb-table-name", "test-BlobMetadata",
+                     "--disperser-server.aws.region", "us-east-1",
+                     "--disperser-server.aws.access-key-id", "localstack",
+                     "--disperser-server.aws.secret-access-key", "localstack",
+                     "--disperser-server.aws.endpoint-url", "http://0.0.0.0:4566"]
+
+    def start(self):
+        self.log.info("Start DA server")
+        super().start()
+
+    def wait_for_rpc_connection(self):
+        time.sleep(1)
+
+    def stop(self):
+        self.log.info("Stop DA server")
+        super().stop(kill=True, wait=False)