Update with the most recent storage contract (#142)

* Update with the most recent storage contract

* fix nits

node/miner/src/lib.rs

@@ -9,6 +9,7 @@ mod loader;
 mod metrics;
 mod mine;
 mod miner_id;
+mod monitor;
 pub mod pora;
 mod recall_range;
 mod sealer;

node/miner/src/mine.rs

@@ -9,7 +9,6 @@ use tokio::time::{sleep, Duration, Instant};
 use storage::config::ShardConfig;
 use zgs_spec::{SECTORS_PER_LOAD, SECTORS_PER_MAX_MINING_RANGE, SECTORS_PER_PRICING};
 
-use super::metrics;
 use crate::recall_range::RecallRange;
 use crate::{
     pora::{AnswerWithoutProof, Miner},
@@ -20,7 +19,7 @@ use crate::{
 use std::sync::Arc;
 
 pub struct PoraService {
-    mine_context_receiver: mpsc::UnboundedReceiver<MineContextMessage>,
+    mine_context_receiver: broadcast::Receiver<MineContextMessage>,
     mine_answer_sender: mpsc::UnboundedSender<AnswerWithoutProof>,
     msg_recv: broadcast::Receiver<MinerMessage>,
     loader: Arc<dyn PoraLoader>,
@@ -33,9 +32,29 @@ pub struct PoraService {
     iter_batch: usize,
 }
 
-struct PoraPuzzle {
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) struct PoraPuzzle {
     context: MineContext,
     target_quality: U256,
+    max_shards: u64,
+}
+
+impl PoraPuzzle {
+    pub fn new(context: MineContext, target_quality: U256, max_shards: u64) -> Self {
+        Self {
+            context,
+            target_quality,
+            max_shards,
+        }
+    }
+
+    pub fn max_shards(&self) -> u64 {
+        self.max_shards
+    }
+
+    pub fn context_digest(&self) -> H256 {
+        H256(self.context.digest)
+    }
 }
 #[derive(Clone, Debug, Default)]
 pub struct MineRangeConfig {
@@ -89,7 +108,7 @@ impl PoraService {
     pub fn spawn(
         executor: TaskExecutor,
         msg_recv: broadcast::Receiver<MinerMessage>,
-        mine_context_receiver: mpsc::UnboundedReceiver<MineContextMessage>,
+        mine_context_receiver: broadcast::Receiver<MineContextMessage>,
         loader: Arc<dyn PoraLoader>,
         config: &MinerConfig,
         miner_id: H256,
@@ -138,13 +157,17 @@ impl PoraService {
                         Ok(MinerMessage::SetStartPosition(pos)) => {
                             info!("Change start position to: {:?}", pos);
                             self.mine_range.start_position = pos;
+                            self.report_reason_if_mine_stop("update mine range");
+
                         }
                         Ok(MinerMessage::SetEndPosition(pos)) => {
                             info!("Change end position to: {:?}", pos);
                             self.mine_range.end_position = pos;
+                            self.report_reason_if_mine_stop("update mine range");
                         }
                         Ok(MinerMessage::SetShardConfig(shard_config)) => {
                             self.mine_range.shard_config = shard_config;
+                            self.report_reason_if_mine_stop("update shard");
                         }
                         Err(broadcast::error::RecvError::Closed) => {
                             warn!("Unexpected: Mine service config channel closed.");
@@ -157,21 +180,26 @@ impl PoraService {
                 }
 
                 maybe_msg = self.mine_context_receiver.recv() => {
-                    if let Some(msg) = maybe_msg {
+                    match maybe_msg {
+                        Ok(msg) => {
                             info!("Update mine service: {:?}", msg);
-                        info!("Mine iterations statistics: {}", metrics::report());
-                        self.puzzle = msg.map(|(context, target_quality)| PoraPuzzle {
-                            context, target_quality
-                        });
-                    } else {
+                            self.puzzle = msg;
+                            self.report_reason_if_mine_stop("update mine context");
+                        },
+                        Err(broadcast::error::RecvError::Closed) => {
                             warn!("Mine context channel closed.");
+                        },
+                        Err(_) => {}
                     }
                 }
 
                 () = &mut diastole, if !diastole.is_elapsed() => {
                 }
 
-                _ = async {}, if mining_enabled && cpu_percent > 0 && self.as_miner().map_or(false, |miner| miner.range.mining_length > 0) && diastole.is_elapsed() => {
+                _ = async {}, if mining_enabled
+                                && cpu_percent > 0
+                                && self.as_miner().is_ok()
+                                && diastole.is_elapsed() => {
                     let nonce = H256(rand::thread_rng().gen());
                     let miner = self.as_miner().unwrap();
 
@@ -194,13 +222,27 @@ impl PoraService {
     }
 
     #[inline]
-    fn as_miner(&self) -> Option<Miner> {
-        let puzzle = self.puzzle.as_ref()?;
+    fn as_miner(&self) -> Result<Miner, &'static str> {
+        let puzzle = self.puzzle.as_ref().ok_or("no mine context")?;
 
-        let range = self.mine_range.to_valid_range(&puzzle.context)?;
-        (range.mining_length > 0).then_some(())?;
+        let range = self
+            .mine_range
+            .to_valid_range(&puzzle.context)
+            .ok_or("no mine range")?;
 
-        Some(Miner {
+        if range.mining_length == 0 {
+            return Err("mine range is zero");
+        }
+
+        if puzzle.max_shards() < self.mine_range.shard_config.num_shard as u64 {
+            return Err("too many mine shards");
+        }
+
+        if self.mine_range.shard_config.num_shard as u64 > puzzle.context.flow_length.as_u64() {
+            return Err("Not enough flow length to shard");
+        }
+
+        Ok(Miner {
             range,
             miner_id: &self.miner_id,
             mine_range_config: &self.mine_range,
@@ -209,4 +251,10 @@ impl PoraService {
             loader: &*self.loader,
         })
     }
+
+    fn report_reason_if_mine_stop(&self, event: &'static str) {
+        if let Err(reason) = self.as_miner() {
+            info!(reason, "Mine stopped on {}", event);
+        }
+    }
 }

node/miner/src/monitor.rs

@@ -0,0 +1,27 @@
+use std::time::Duration;
+
+use task_executor::TaskExecutor;
+use tokio::time::sleep;
+
+use super::metrics;
+
+pub struct Monitor {
+    period: Duration,
+}
+
+impl Monitor {
+    pub fn spawn(executor: TaskExecutor, period: Duration) {
+        let monitor = Monitor { period };
+        executor.spawn(
+            async move { Box::pin(monitor.start()).await },
+            "pora_master",
+        );
+    }
+
+    async fn start(&self) {
+        loop {
+            info!("Mine iterations statistics: {}", metrics::report());
+            let _ = sleep(self.period).await;
+        }
+    }
+}

node/miner/src/pora.rs

@@ -102,15 +102,16 @@ impl<'a> Miner<'a> {
             }
 
             let quality = self.pora(idx, &sealed_data, pad_seed);
-            let quality_scale = self.range.shard_mask.count_zeros();
-            if quality <= U256::MAX >> quality_scale
-                && quality << quality_scale <= *self.target_quality
-            {
+            let difficulty_scale_x64 = self
+                .range
+                .difficulty_scale_x64(self.context.flow_length.as_u64());
+
+            if quality <= (self.target_quality / difficulty_scale_x64) << 64 {
                 debug!(
-                    "Find a PoRA valid answer, quality: {}, target_quality {}, scale {}",
+                    "Find a PoRA valid answer, quality: {}, target_quality {}, scale {:.3}",
                     U256::MAX / quality,
                     U256::MAX / self.target_quality,
-                    quality_scale
+                    difficulty_scale_x64.as_u128() as f64 / (u64::MAX as f64 + 1.0)
                 );
                 inc_counter(&HIT_COUNT);
                 // Undo mix data when find a valid solition

node/miner/src/recall_range.rs

@@ -1,6 +1,6 @@
 use ethereum_types::U256;
 use tiny_keccak::{Hasher, Keccak};
-use zgs_spec::SECTORS_PER_LOAD;
+use zgs_spec::{SECTORS_PER_LOAD, SECTORS_PER_MAX_MINING_RANGE};
 
 #[derive(PartialEq, Eq, Clone, Copy, Debug)]
 pub struct RecallRange {
@@ -38,6 +38,16 @@ impl RecallRange {
 
         Some(self.start_position + recall_offset * SECTORS_PER_LOAD as u64)
     }
+
+    pub fn difficulty_scale_x64(&self, flow_length: u64) -> U256 {
+        let no_shard_mine_length = std::cmp::min(flow_length, SECTORS_PER_MAX_MINING_RANGE as u64);
+        let sharded_mine_length = std::cmp::min(
+            flow_length >> self.shard_mask.count_zeros(),
+            SECTORS_PER_MAX_MINING_RANGE as u64,
+        );
+
+        (U256::from(no_shard_mine_length) << 64) / sharded_mine_length
+    }
 }
 
 impl From<RecallRange> for contract_interface::RecallRange {

node/miner/src/service.rs

@@ -1,9 +1,11 @@
 use crate::miner_id::check_and_request_miner_id;
+use crate::monitor::Monitor;
 use crate::sealer::Sealer;
 use crate::submitter::Submitter;
 use crate::{config::MinerConfig, mine::PoraService, watcher::MineContextWatcher};
 use network::NetworkMessage;
 use std::sync::Arc;
+use std::time::Duration;
 use storage::config::ShardConfig;
 use storage_async::Store;
 use tokio::sync::broadcast;
@@ -48,7 +50,7 @@ impl MineService {
         let mine_answer_receiver = PoraService::spawn(
             executor.clone(),
             msg_recv.resubscribe(),
-            mine_context_receiver,
+            mine_context_receiver.resubscribe(),
             store.clone(),
             &config,
             miner_id,
@@ -57,12 +59,15 @@ impl MineService {
         Submitter::spawn(
             executor.clone(),
             mine_answer_receiver,
+            mine_context_receiver,
             provider.clone(),
             store.clone(),
             &config,
         );
 
-        Sealer::spawn(executor, provider, store, &config, miner_id);
+        Sealer::spawn(executor.clone(), provider, store, &config, miner_id);
+
+        Monitor::spawn(executor, Duration::from_secs(5));
 
         debug!("Starting miner service");
 

node/miner/src/submitter.rs

@@ -6,19 +6,23 @@ use ethers::providers::PendingTransaction;
 use hex::ToHex;
 use shared_types::FlowRangeProof;
 use std::sync::Arc;
+use std::time::Duration;
+use storage::H256;
 use storage_async::Store;
 use task_executor::TaskExecutor;
-use tokio::sync::mpsc;
+use tokio::sync::{broadcast, mpsc};
 
 use crate::config::{MineServiceMiddleware, MinerConfig};
 use crate::pora::AnswerWithoutProof;
+use crate::watcher::MineContextMessage;
 
 use zgs_spec::{BYTES_PER_SEAL, SECTORS_PER_SEAL};
 
-const SUBMISSION_RETIES: usize = 3;
+const SUBMISSION_RETIES: usize = 15;
 
 pub struct Submitter {
     mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
+    mine_context_receiver: broadcast::Receiver<MineContextMessage>,
     mine_contract: PoraMine<MineServiceMiddleware>,
     flow_contract: ZgsFlow<MineServiceMiddleware>,
     default_gas_limit: Option<U256>,
@@ -29,6 +33,7 @@ impl Submitter {
     pub fn spawn(
         executor: TaskExecutor,
         mine_answer_receiver: mpsc::UnboundedReceiver<AnswerWithoutProof>,
+        mine_context_receiver: broadcast::Receiver<MineContextMessage>,
         provider: Arc<MineServiceMiddleware>,
         store: Arc<Store>,
         config: &MinerConfig,
@@ -39,6 +44,7 @@ impl Submitter {
 
         let submitter = Submitter {
             mine_answer_receiver,
+            mine_context_receiver,
             mine_contract,
             flow_contract,
             store,
@@ -51,18 +57,39 @@ impl Submitter {
     }
 
     async fn start(mut self) {
+        let mut current_context_digest: Option<H256> = None;
         loop {
-            match self.mine_answer_receiver.recv().await {
+            tokio::select! {
+                answer_msg = self.mine_answer_receiver.recv() => {
+                    match answer_msg {
                         Some(answer) => {
+                            if Some(answer.context_digest) != current_context_digest {
+                                info!("Skip submission because of inconsistent context digest");
+                                continue;
+                            }
                             if let Err(e) = self.submit_answer(answer).await {
-                        warn!(e)
+                                warn!(e);
                             }
                         }
                         None => {
                             warn!("Mine submitter stopped because mine answer channel is closed.");
-                    break;
+                            return;
+                        }
+                    }
+                }
+
+                context_msg = self.mine_context_receiver.recv() => {
+                    match context_msg {
+                        Ok(puzzle) => {
+                            current_context_digest = puzzle.map(|p| p.context_digest());
+                        }
+                        Err(broadcast::error::RecvError::Closed) => {
+                            warn!("Mine context channel closed.");
+                        },
+                        Err(_) => {}
+                    }
+                }
             }
-            };
         }
     }
 
@@ -136,6 +163,7 @@ impl Submitter {
 
         let receipt = pending_transaction
             .retries(SUBMISSION_RETIES)
+            .interval(Duration::from_secs(2))
             .await
             .map_err(|e| format!("Fail to execute mine answer transaction: {:?}", e))?
             .ok_or(format!(

node/miner/src/watcher.rs

@@ -19,9 +19,9 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::{ops::DerefMut, str::FromStr};
 
-use crate::{config::MineServiceMiddleware, MinerConfig, MinerMessage};
+use crate::{config::MineServiceMiddleware, mine::PoraPuzzle, MinerConfig, MinerMessage};
 
-pub type MineContextMessage = Option<(MineContext, U256)>;
+pub type MineContextMessage = Option<PoraPuzzle>;
 
 lazy_static! {
     pub static ref EMPTY_HASH: H256 =
@@ -33,7 +33,7 @@ pub struct MineContextWatcher {
     flow_contract: ZgsFlow<MineServiceMiddleware>,
     mine_contract: PoraMine<MineServiceMiddleware>,
 
-    mine_context_sender: mpsc::UnboundedSender<MineContextMessage>,
+    mine_context_sender: broadcast::Sender<MineContextMessage>,
     last_report: MineContextMessage,
 
     msg_recv: broadcast::Receiver<MinerMessage>,
@@ -45,14 +45,14 @@ impl MineContextWatcher {
         msg_recv: broadcast::Receiver<MinerMessage>,
         provider: Arc<MineServiceMiddleware>,
         config: &MinerConfig,
-    ) -> mpsc::UnboundedReceiver<MineContextMessage> {
+    ) -> broadcast::Receiver<MineContextMessage> {
         let provider = provider;
 
         let mine_contract = PoraMine::new(config.mine_address, provider.clone());
         let flow_contract = ZgsFlow::new(config.flow_address, provider.clone());
 
         let (mine_context_sender, mine_context_receiver) =
-            mpsc::unbounded_channel::<MineContextMessage>();
+            broadcast::channel::<MineContextMessage>(4096);
         let watcher = MineContextWatcher {
             provider,
             flow_contract,
@@ -108,12 +108,17 @@ impl MineContextWatcher {
         let context_call = self.flow_contract.make_context_with_result();
         let valid_call = self.mine_contract.can_submit();
         let quality_call = self.mine_contract.pora_target();
+        let shards_call = self.mine_contract.max_shards();
 
-        let (context, can_submit, quality) =
-            try_join!(context_call.call(), valid_call.call(), quality_call.call())
+        let (context, can_submit, quality, max_shards) = try_join!(
+            context_call.call(),
+            valid_call.call(),
+            quality_call.call(),
+            shards_call.call()
+        )
         .map_err(|e| format!("Failed to query mining context: {:?}", e))?;
         let report = if can_submit && context.digest != EMPTY_HASH.0 {
-            Some((context, quality))
+            Some(PoraPuzzle::new(context, quality, max_shards))
         } else {
             None
         };

scripts/check_abis.sh

@@ -8,5 +8,5 @@ check_abis() {
         diff $(./scripts/search_abi.sh "$artifacts_path" "$contract_name.json") "storage-contracts-abis/$contract_name.json"
     done
 }
-check_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice OnePoolReward FixedPriceFlow
+check_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice ChunkLinearReward FixedPriceFlow
 

scripts/update_abis.sh

@@ -54,7 +54,7 @@ copy_abis() {
     done
 }
 
-copy_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice OnePoolReward FixedPriceFlow
+copy_abis DummyMarket DummyReward Flow PoraMine PoraMineTest FixedPrice ChunkLinearReward FixedPriceFlow
 
 
 # Step 4: Get the current Git revision and write it to a specified file

storage-contracts-abis/0g-storage-contracts-rev

@@ -1 +1 @@
-6ec57942cba3068d3373bcb2c88f8fd51d2322b1
+dbeff538b949599c203e43be6ecc05e9e997d09d

tests/mine_test.py

@@ -45,14 +45,14 @@ class MineTest(TestFramework):
         self.log.info("Submission done, current epoch is %d", start_epoch)
 
         self.log.info("Wait for the first mine context release")
-        wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 1, timeout=180)
+        wait_until(lambda: self.contract.epoch() >= start_epoch + 1, timeout=180)
         self.contract.update_context()
 
         self.log.info("Wait for the first mine answer")
         wait_until(lambda: self.mine_contract.last_mined_epoch() == start_epoch + 1 and not self.mine_contract.can_submit(), timeout=180)
 
         self.log.info("Wait for the second mine context release")
-        wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 2, timeout=180)
+        wait_until(lambda: self.contract.epoch() >= start_epoch + 2, timeout=180)
         self.contract.update_context()
 
         self.log.info("Wait for the second mine answer")
@@ -60,7 +60,7 @@ class MineTest(TestFramework):
 
         self.nodes[0].miner_stop()
         self.log.info("Wait for the third mine context release")
-        wait_until(lambda: int(blockchain.eth_blockNumber(), 16) >= start_epoch + 3, timeout=180)
+        wait_until(lambda: self.contract.epoch() >= start_epoch + 3, timeout=180)
         self.contract.update_context()
         
         self.log.info("Submit the second data chunk")

tests/mine_with_market_test.py

@@ -50,6 +50,7 @@ class MineTest(TestFramework):
 
         SECTORS_PER_PRICING = int(8 * ( 2 ** 30 ) / 256)
 
+
         self.log.info("Submit the actual data chunk (256 MB)")
         self.submit_data(b"\x11", int(SECTORS_PER_PRICING / 32))
 
@@ -74,7 +75,7 @@ class MineTest(TestFramework):
         firstReward = rewards[0].args.amount
         self.log.info("Received reward %d Gwei", firstReward / (10**9))
 
-        self.reward_contract.transfer(10000 * 10 ** 18)
+        self.reward_contract.donate(10000 * 10 ** 18)
         self.log.info("Donation Done")
         self.log.info("Submit the data hash only (8 GB)")
         self.submit_data(b"\x11", int(SECTORS_PER_PRICING), no_submit=True)

tests/test_framework/blockchain_node.py

@@ -303,7 +303,7 @@ class BlockchainNode(TestNode):
             mine_contract.functions.setTargetSubmissions(2).transact(TX_PARAMS)
             self.log.debug("Mine Initialized")
 
-            flow_initialize_hash = (flow_contract.get_function_by_signature('initialize(address)'))(dummy_market_contract.address).transact(TX_PARAMS)
+            flow_initialize_hash = flow_contract.functions.initialize(dummy_market_contract.address).transact(TX_PARAMS)
             self.log.debug("Flow Initialized")
 
             self.wait_for_transaction_receipt(w3, flow_initialize_hash)
@@ -325,7 +325,7 @@ class BlockchainNode(TestNode):
             market_contract, _ = deploy_contract("FixedPrice", [])
             self.log.debug("Market deployed")
             
-            reward_contract, _ =deploy_contract("OnePoolReward", [LIFETIME_MONTH])
+            reward_contract, _ =deploy_contract("ChunkLinearReward", [LIFETIME_MONTH * 31 * 86400])
             self.log.debug("Reward deployed")
             
             flow_contract, _ = deploy_contract("FixedPriceFlow", [mine_period, 0])
@@ -336,13 +336,15 @@ class BlockchainNode(TestNode):
             mine_contract.functions.setTargetSubmissions(2).transact(TX_PARAMS)
             self.log.debug("Mine Initialized")
             
-            market_contract.functions.initialize(LIFETIME_MONTH, flow_contract.address, reward_contract.address).transact(TX_PARAMS)
+            price_per_sector = int(LIFETIME_MONTH * 256 * 10 * 1_000_000_000_000_000_000 / 1024 / 1024 / 1024 / 12)
+            market_contract.functions.initialize(price_per_sector, flow_contract.address, reward_contract.address).transact(TX_PARAMS)
             self.log.debug("Market Initialized")
             
             reward_contract.functions.initialize(market_contract.address, mine_contract.address).transact(TX_PARAMS)
+            reward_contract.functions.setBaseReward(10 ** 18).transact(TX_PARAMS)
             self.log.debug("Reward Initialized")
             
-            flow_initialize_hash = (flow_contract.get_function_by_signature('initialize(address)'))(market_contract.address).transact(TX_PARAMS)
+            flow_initialize_hash = flow_contract.functions.initialize(market_contract.address).transact(TX_PARAMS)
             self.log.debug("Flow Initialized")
             
             self.wait_for_transaction_receipt(w3, flow_initialize_hash)

tests/test_framework/contract_proxy.py

@@ -29,6 +29,14 @@ class ContractProxy:
         contract = self._get_contract(node_idx)
         return getattr(contract.functions, fn_name)(**args).transact(copy(TX_PARAMS))
 
+    def _send_payable(self, fn_name, node_idx, value, **args):
+        assert node_idx < len(self.blockchain_nodes)
+
+        contract = self._get_contract(node_idx)
+        tx_params = copy(TX_PARAMS)
+        tx_params["value"] = value
+        return getattr(contract.functions, fn_name)(**args).transact(tx_params)
+    
     def _logs(self, event_name, node_idx, **args):
         assert node_idx < len(self.blockchain_nodes)
 
@@ -66,7 +74,6 @@ class FlowContractProxy(ContractProxy):
             contract.w3, tx_hash, parent_hash=parent_hash
         )
         if receipt["status"] != 1:
-            print(receipt)
             assert_equal(receipt["status"], 1)
         return tx_hash
 
@@ -98,6 +105,12 @@ class MineContractProxy(ContractProxy):
     
 
 
-class IRewardContractProxy(ContractProxy):
+class RewardContractProxy(ContractProxy):
     def reward_distributes(self, node_idx=0):
         return self._logs("DistributeReward", node_idx)
+
+    def donate(self, value, node_idx = 0):
+        return self._send_payable("donate", node_idx, value)
+    
+    def base_reward(self, node_idx = 0):
+        return self._call("baseReward", node_idx)

tests/test_framework/test_framework.py

@@ -15,7 +15,7 @@ from pathlib import Path
 
 from eth_utils import encode_hex
 from test_framework.bsc_node import BSCNode
-from test_framework.contract_proxy import FlowContractProxy, MineContractProxy, IRewardContractProxy
+from test_framework.contract_proxy import FlowContractProxy, MineContractProxy, RewardContractProxy
 from test_framework.zgs_node import ZgsNode
 from test_framework.blockchain_node import BlockChainNodeType
 from test_framework.conflux_node import ConfluxNode, connect_sample_nodes
@@ -174,7 +174,7 @@ class TestFramework:
         contract, tx_hash, mine_contract, reward_contract = self.blockchain_nodes[0].setup_contract(self.enable_market, self.mine_period)
         self.contract = FlowContractProxy(contract, self.blockchain_nodes)
         self.mine_contract = MineContractProxy(mine_contract, self.blockchain_nodes)
-        self.reward_contract = IRewardContractProxy(reward_contract, self.blockchain_nodes)
+        self.reward_contract = RewardContractProxy(reward_contract, self.blockchain_nodes)
 
 
         for node in self.blockchain_nodes[1:]: