From 193e154361c3ec10c8de7cc658fd48e250de797b Mon Sep 17 00:00:00 2001 From: Chenxing Li Date: Sat, 27 Apr 2024 11:15:57 +0800 Subject: [PATCH] Change miner id logic & request miner id automatically (#60) * Change miner id logic & request miner id automatically * Not enable all features in the test workflow. * Auto configurable mining period * Adjust test params for ci --- .github/workflows/cc.yml | 2 +- .gitignore | 1 + 0g-storage-contracts | 2 +- common/contract-interface/Cargo.toml | 2 +- common/contract-interface/build.rs | 6 +- common/contract-interface/src/lib.rs | 16 +++- node/miner/src/config.rs | 25 +++--- node/miner/src/lib.rs | 2 + node/miner/src/mine.rs | 3 +- node/miner/src/miner_id.rs | 111 ++++++++++++++++++++++++ node/miner/src/sealer.rs | 3 +- node/miner/src/service.rs | 7 +- node/miner/src/submitter.rs | 2 +- tests/long_time_mine_test_local.py | 2 + tests/mine_test.py | 5 +- tests/mine_with_market_test.py | 7 +- tests/test_framework/blockchain_node.py | 15 +++- tests/test_framework/test_framework.py | 23 ++++- 18 files changed, 202 insertions(+), 32 deletions(-) create mode 100644 node/miner/src/miner_id.rs diff --git a/.github/workflows/cc.yml b/.github/workflows/cc.yml index c657505..25eedcd 100644 --- a/.github/workflows/cc.yml +++ b/.github/workflows/cc.yml @@ -37,7 +37,7 @@ jobs: uses: ./.github/actions/setup-rust - name: Run unittest - run: cargo test --all-features --no-fail-fast + run: cargo test --no-fail-fast env: CARGO_INCREMENTAL: '0' RUSTC_BOOTSTRAP: '1' diff --git a/.gitignore b/.gitignore index a2e98aa..7c761db 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ tests/**/__pycache__ tests/tmp/** .vscode/*.json +/0g-storage-contracts-dev \ No newline at end of file diff --git a/0g-storage-contracts b/0g-storage-contracts index d466311..6a9f52e 160000 --- a/0g-storage-contracts +++ b/0g-storage-contracts @@ -1 +1 @@ -Subproject commit d466311abb6f629a6489450f2e684b2c6e7b1089 +Subproject commit 6a9f52e8c10ff9b5cd7a5844c543c0951b97d395 diff --git a/common/contract-interface/Cargo.toml b/common/contract-interface/Cargo.toml index f9a1564..61dfa16 100644 --- a/common/contract-interface/Cargo.toml +++ b/common/contract-interface/Cargo.toml @@ -11,4 +11,4 @@ ethers = "^2" serde_json = "1.0.82" [features] -compile-contracts = [] \ No newline at end of file +dev = [] \ No newline at end of file diff --git a/common/contract-interface/build.rs b/common/contract-interface/build.rs index 2a6aa7b..277ded1 100644 --- a/common/contract-interface/build.rs +++ b/common/contract-interface/build.rs @@ -1,3 +1,7 @@ fn main() { - println!("cargo:rerun-if-changed=../../0g-storage-contracts/artifacts/"); + if cfg!(not(feature = "dev")) { + println!("cargo:rerun-if-changed=../../0g-storage-contracts/artifacts/"); + } else { + println!("cargo:rerun-if-changed=../../0g-storage-contracts-dev/artifacts/"); + } } diff --git a/common/contract-interface/src/lib.rs b/common/contract-interface/src/lib.rs index 54eb6b4..10333c2 100644 --- a/common/contract-interface/src/lib.rs +++ b/common/contract-interface/src/lib.rs @@ -2,12 +2,26 @@ use ethers::prelude::abigen; // run `cargo doc -p contract-interface --open` to read struct definition +#[cfg(not(feature = "dev"))] abigen!( ZgsFlow, "../../0g-storage-contracts/artifacts/contracts/dataFlow/Flow.sol/Flow.json" ); +#[cfg(not(feature = "dev"))] abigen!( PoraMine, - "../../0g-storage-contracts/artifacts/contracts/test/PoraMineTest.sol/PoraMineTest.json" + "../../0g-storage-contracts/artifacts/contracts/miner/Mine.sol/PoraMine.json" +); + +#[cfg(feature = "dev")] +abigen!( + ZgsFlow, + "../../0g-storage-contracts-dev/artifacts/contracts/dataFlow/Flow.sol/Flow.json" +); + +#[cfg(feature = "dev")] +abigen!( + PoraMine, + "../../0g-storage-contracts-dev/artifacts/contracts/miner/Mine.sol/PoraMine.json" ); diff --git a/node/miner/src/config.rs b/node/miner/src/config.rs index a0a85ba..49e9a1a 100644 --- a/node/miner/src/config.rs +++ b/node/miner/src/config.rs @@ -8,7 +8,7 @@ use ethers::signers::LocalWallet; use ethers::signers::Signer; pub struct MinerConfig { - pub(crate) miner_id: H256, + pub(crate) miner_id: Option, pub(crate) miner_key: H256, pub(crate) rpc_endpoint_url: String, pub(crate) mine_address: Address, @@ -32,19 +32,16 @@ impl MinerConfig { cpu_percentage: u64, iter_batch: usize, ) -> Option { - match (miner_id, miner_key) { - (Some(miner_id), Some(miner_key)) => Some(MinerConfig { - miner_id, - miner_key, - rpc_endpoint_url, - mine_address, - flow_address, - submission_gas, - cpu_percentage, - iter_batch, - }), - _ => None, - } + miner_key.map(|miner_key| MinerConfig { + miner_id, + miner_key, + rpc_endpoint_url, + mine_address, + flow_address, + submission_gas, + cpu_percentage, + iter_batch, + }) } pub(crate) async fn make_provider(&self) -> Result { diff --git a/node/miner/src/lib.rs b/node/miner/src/lib.rs index 0c6804c..ab8bb77 100644 --- a/node/miner/src/lib.rs +++ b/node/miner/src/lib.rs @@ -7,6 +7,7 @@ extern crate lazy_static; mod config; mod loader; mod mine; +mod miner_id; pub mod pora; mod sealer; mod service; @@ -16,4 +17,5 @@ mod watcher; pub use config::MinerConfig; pub use loader::PoraLoader; pub use mine::CustomMineRange; +pub use miner_id::load_miner_id; pub use service::{MineService, MinerMessage}; diff --git a/node/miner/src/mine.rs b/node/miner/src/mine.rs index efb8aa4..ccc47b5 100644 --- a/node/miner/src/mine.rs +++ b/node/miner/src/mine.rs @@ -82,6 +82,7 @@ impl PoraService { mine_context_receiver: mpsc::UnboundedReceiver, loader: Arc, config: &MinerConfig, + miner_id: H256, ) -> mpsc::UnboundedReceiver { let (mine_answer_sender, mine_answer_receiver) = mpsc::unbounded_channel::(); @@ -95,7 +96,7 @@ impl PoraService { msg_recv, puzzle: None, mine_range, - miner_id: config.miner_id, + miner_id, loader, cpu_percentage: config.cpu_percentage, iter_batch: config.iter_batch, diff --git a/node/miner/src/miner_id.rs b/node/miner/src/miner_id.rs new file mode 100644 index 0000000..7b50dcc --- /dev/null +++ b/node/miner/src/miner_id.rs @@ -0,0 +1,111 @@ +use crate::config::MineServiceMiddleware; +use crate::config::MinerConfig; +use contract_interface::{NewMinerIdFilter, PoraMine}; +use ethereum_types::Address; +use ethers::contract::ContractCall; +use ethers::contract::EthEvent; +use std::sync::Arc; +use storage::log_store::{config::ConfigurableExt, Store}; +use storage::H256; +use tokio::sync::RwLock; + +const MINER_ID: &str = "mine.miner_id"; + +pub fn load_miner_id(store: &dyn Store) -> storage::error::Result> { + store.get_config_decoded(&MINER_ID) +} + +fn set_miner_id(store: &dyn Store, miner_id: &H256) -> storage::error::Result<()> { + store.set_config_encoded(&MINER_ID, miner_id) +} + +pub(crate) async fn check_and_request_miner_id( + config: &MinerConfig, + store: &RwLock, + provider: &Arc, +) -> Result { + let db_miner_id = load_miner_id(&*store.read().await) + .map_err(|e| format!("miner_id on db corrupt: {:?}", e))?; + + let mine_contract = PoraMine::new(config.mine_address, provider.clone()); + + match (db_miner_id, config.miner_id) { + (Some(d_id), Some(c_id)) => { + if d_id != c_id { + Err(format!( + "database miner id {} != configuration miner id {}", + d_id, c_id + )) + } else { + Ok(d_id) + } + } + (None, Some(c_id)) => { + check_miner_id(&mine_contract, c_id).await?; + set_miner_id(&*store.write().await, &c_id) + .map_err(|e| format!("set miner id on db corrupt: {:?}", e))?; + Ok(c_id) + } + (Some(d_id), None) => { + check_miner_id(&mine_contract, d_id).await?; + Ok(d_id) + } + (None, None) => { + let beneficiary = provider.address(); + let id = request_miner_id(&mine_contract, beneficiary).await?; + set_miner_id(&*store.write().await, &id) + .map_err(|e| format!("set miner id on db corrupt: {:?}", e))?; + Ok(id) + } + } +} + +async fn check_miner_id( + mine_contract: &PoraMine, + miner_id: H256, +) -> Result { + debug!("Checking miner id on chain..."); + + let beneficiary = mine_contract + .beneficiaries(miner_id.0) + .call() + .await + .map_err(|e| format!("Fail to query miner id information: {:?}", e))?; + + if beneficiary == Address::zero() { + Err("candidate miner id is not registered".into()) + } else { + Ok(beneficiary) + } +} + +async fn request_miner_id( + mine_contract: &PoraMine, + beneficiary: Address, +) -> Result { + debug!("Requesting miner id on chain..."); + + let submission_call: ContractCall<_, _> = + mine_contract.request_miner_id(beneficiary, 0).legacy(); + + let pending_tx = submission_call + .send() + .await + .map_err(|e| format!("Fail to request miner id: {:?}", e))?; + + let receipt = pending_tx + .retries(3) + .await + .map_err(|e| format!("Fail to execute mine answer transaction: {:?}", e))? + .ok_or("Request miner id transaction dropped after 3 retires")?; + + let first_log = receipt + .logs + .first() + .ok_or("Fail to find minerId in receipt")?; + + let new_id_event = NewMinerIdFilter::decode_log(&first_log.clone().into()) + .map_err(|e| format!("Fail to decode NewMinerId event: {:?}", e))?; + + Ok(H256(new_id_event.miner_id)) +} diff --git a/node/miner/src/sealer.rs b/node/miner/src/sealer.rs index cb41891..59390d4 100644 --- a/node/miner/src/sealer.rs +++ b/node/miner/src/sealer.rs @@ -34,6 +34,7 @@ impl Sealer { provider: Arc, store: Arc>, config: &MinerConfig, + miner_id: H256, ) { let flow_contract = ZgsFlow::new(config.flow_address, provider); let sealer = Sealer { @@ -41,7 +42,7 @@ impl Sealer { store, context_cache: Default::default(), last_context_flow_length: 0, - miner_id: config.miner_id, + miner_id, }; executor.spawn(async move { Box::pin(sealer.start()).await }, "data_sealer"); diff --git a/node/miner/src/service.rs b/node/miner/src/service.rs index 5575597..213ccaf 100644 --- a/node/miner/src/service.rs +++ b/node/miner/src/service.rs @@ -1,3 +1,4 @@ +use crate::miner_id::check_and_request_miner_id; use crate::sealer::Sealer; use crate::submitter::Submitter; use crate::{config::MinerConfig, mine::PoraService, watcher::MineContextWatcher}; @@ -30,6 +31,9 @@ impl MineService { let (msg_send, msg_recv) = broadcast::channel(1024); + let miner_id = check_and_request_miner_id(&config, &store, &provider).await?; + debug!("miner id setting complete."); + let mine_context_receiver = MineContextWatcher::spawn( executor.clone(), msg_recv.resubscribe(), @@ -43,6 +47,7 @@ impl MineService { mine_context_receiver, Arc::new(store.clone()), &config, + miner_id, ); Submitter::spawn( @@ -53,7 +58,7 @@ impl MineService { &config, ); - Sealer::spawn(executor, provider, store, &config); + Sealer::spawn(executor, provider, store, &config, miner_id); debug!("Starting miner service"); diff --git a/node/miner/src/submitter.rs b/node/miner/src/submitter.rs index bd5ec03..3cd953a 100644 --- a/node/miner/src/submitter.rs +++ b/node/miner/src/submitter.rs @@ -97,7 +97,7 @@ impl Submitter { mine_length: mine_answer.mining_length.into(), recall_position: mine_answer.recall_position.into(), seal_offset: mine_answer.seal_offset.into(), - sealed_context_digest: sealed_context_digest.digest, // TODO(kevin): wait for implementation of data sealing. + sealed_context_digest: sealed_context_digest.digest, sealed_data: unsafe { std::mem::transmute(mine_answer.sealed_data) }, merkle_proof: flow_proof_to_pora_merkle_proof(flow_proof), }; diff --git a/tests/long_time_mine_test_local.py b/tests/long_time_mine_test_local.py index be4b649..f9cb532 100755 --- a/tests/long_time_mine_test_local.py +++ b/tests/long_time_mine_test_local.py @@ -16,6 +16,8 @@ class LongTimeMineTest(TestFramework): "mine_iter_batch_size": 50, } self.mine_period = 15 + self.launch_wait_seconds = 15 + def submit_data(self, item, size): submissions_before = self.contract.num_submissions() diff --git a/tests/mine_test.py b/tests/mine_test.py index a91d0b7..89c0d34 100755 --- a/tests/mine_test.py +++ b/tests/mine_test.py @@ -3,6 +3,7 @@ from test_framework.test_framework import TestFramework from config.node_config import MINER_ID, GENESIS_PRIV_KEY from utility.submission import create_submission, submit_data from utility.utils import wait_until +from test_framework.blockchain_node import BlockChainNodeType class MineTest(TestFramework): @@ -10,10 +11,10 @@ class MineTest(TestFramework): self.num_blockchain_nodes = 1 self.num_nodes = 1 self.zgs_node_configs[0] = { - "miner_id": MINER_ID, "miner_key": GENESIS_PRIV_KEY, } - self.mine_period = 15 + self.mine_period = int(45 / self.block_time) + self.launch_wait_seconds = 15 def submit_data(self, item, size): submissions_before = self.contract.num_submissions() diff --git a/tests/mine_with_market_test.py b/tests/mine_with_market_test.py index 40d5c8e..2400537 100755 --- a/tests/mine_with_market_test.py +++ b/tests/mine_with_market_test.py @@ -3,6 +3,8 @@ from test_framework.test_framework import TestFramework from config.node_config import MINER_ID, GENESIS_PRIV_KEY from utility.submission import create_submission, submit_data from utility.utils import wait_until, assert_equal, assert_greater_than +from test_framework.blockchain_node import BlockChainNodeType + import math @@ -13,11 +15,12 @@ class MineTest(TestFramework): self.num_blockchain_nodes = 1 self.num_nodes = 1 self.zgs_node_configs[0] = { - "miner_id": MINER_ID, "miner_key": GENESIS_PRIV_KEY, } self.enable_market = True - self.mine_period = 20 + self.mine_period = int(60 / self.block_time) + self.launch_wait_seconds = 15 + def submit_data(self, item, size, no_submit = False): submissions_before = self.contract.num_submissions() diff --git a/tests/test_framework/blockchain_node.py b/tests/test_framework/blockchain_node.py index 987de2b..1f806fa 100644 --- a/tests/test_framework/blockchain_node.py +++ b/tests/test_framework/blockchain_node.py @@ -28,6 +28,13 @@ class BlockChainNodeType(Enum): BSC = 1 Evmos = 2 + def block_time(self): + if self == BlockChainNodeType.Conflux: + return 0.5 + elif self == BlockChainNodeType.BSC: + return 0.25 + else: + return 3.0 @unique class NodeType(Enum): @@ -299,8 +306,8 @@ class BlockchainNode(TestNode): self.log.debug("Mine deployed") self.log.info("All contracts deployed") - tx_hash = mine_contract.functions.setMiner(decode_hex(MINER_ID)).transact(TX_PARAMS) - self.wait_for_transaction_receipt(w3, tx_hash) + # tx_hash = mine_contract.functions.setMiner(decode_hex(MINER_ID)).transact(TX_PARAMS) + # self.wait_for_transaction_receipt(w3, tx_hash) dummy_reward_contract = w3.eth.contract( address = book.functions.reward().call(), @@ -328,8 +335,8 @@ class BlockchainNode(TestNode): self.log.info("All contracts deployed") - tx_hash = mine_contract.functions.setMiner(decode_hex(MINER_ID)).transact(TX_PARAMS) - self.wait_for_transaction_receipt(w3, tx_hash) + # tx_hash = mine_contract.functions.setMiner(decode_hex(MINER_ID)).transact(TX_PARAMS) + # self.wait_for_transaction_receipt(w3, tx_hash) return flow_contract, flow_contract_hash, mine_contract, reward_contract diff --git a/tests/test_framework/test_framework.py b/tests/test_framework/test_framework.py index 1df6bd0..d3d1520 100644 --- a/tests/test_framework/test_framework.py +++ b/tests/test_framework/test_framework.py @@ -47,8 +47,10 @@ class TestFramework: self.blockchain_node_configs = {} self.zgs_node_configs = {} self.blockchain_node_type = blockchain_node_type + self.block_time = blockchain_node_type.block_time() self.enable_market = False self.mine_period = 100 + self.launch_wait_seconds = 1 # Set default binary path binary_ext = ".exe" if is_windows_platform() else "" @@ -203,7 +205,9 @@ class TestFramework: time.sleep(1) node.start() - time.sleep(1) + self.log.info("Wait the zgs_node launch for %d seconds", self.launch_wait_seconds) + time.sleep(self.launch_wait_seconds) + for node in self.nodes: node.wait_for_rpc_connection() @@ -268,6 +272,10 @@ class TestFramework: "--tmpdir", dest="tmpdir", help="Root directory for datadirs" ) + parser.add_argument( + "--devdir", dest="devdir", help="A softlink point to the last run" + ) + parser.add_argument( "--randomseed", dest="random_seed", type=int, help="Set a random seed" ) @@ -433,6 +441,19 @@ class TestFramework: self.__start_logging() self.log.info("Root dir: %s", self.root_dir) + if self.options.devdir: + dst = self.options.devdir + + if os.path.islink(dst): + os.remove(dst) + elif os.path.isdir(dst): + shutil.rmtree(dst) + elif os.path.exists(dst): + os.remove(dst) + + os.symlink(self.options.tmpdir, dst) + self.log.info("Symlink: %s", Path(dst).absolute()) + if self.blockchain_node_type == BlockChainNodeType.Conflux: self.blockchain_binary = os.path.abspath(self.options.conflux) elif self.blockchain_node_type == BlockChainNodeType.BSC: