Merge branch 'mine_subtask' into next_mine_spec

node/miner/src/mine.rs

@@ -1,4 +1,4 @@
-use contract_interface::zgs_flow::MineContext;
+use contract_interface::pora_mine::MineContext;
 use ethereum_types::{H256, U256};
 use rand::{self, Rng};
 use std::time;
@@ -35,16 +35,23 @@ pub struct PoraService {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) struct PoraPuzzle {
     context: MineContext,
-    target_quality: U256,
+    pora_target: U256,
     max_shards: u64,
+    subtask_digest: H256,
 }
 
 impl PoraPuzzle {
-    pub fn new(context: MineContext, target_quality: U256, max_shards: u64) -> Self {
+    pub fn new(
+        context: MineContext,
+        pora_target: U256,
+        max_shards: u64,
+        subtask_digest: H256,
+    ) -> Self {
         Self {
             context,
-            target_quality,
+            pora_target,
             max_shards,
+            subtask_digest,
         }
     }
 
@@ -255,7 +262,8 @@ impl PoraService {
             miner_id: &self.miner_id,
             mine_range_config: &self.mine_range,
             context: &puzzle.context,
-            target_quality: &puzzle.target_quality,
+            subtask_digest: &puzzle.subtask_digest,
+            pora_target: &puzzle.pora_target,
             loader: &*self.loader,
         })
     }

node/miner/src/pora.rs

@@ -2,7 +2,7 @@ use super::metrics::*;
 use crate::recall_range::RecallRange;
 use crate::{MineRangeConfig, PoraLoader};
 use blake2::{Blake2b512, Digest};
-use contract_interface::zgs_flow::MineContext;
+use contract_interface::pora_mine::MineContext;
 use ethereum_types::{H256, U256};
 use ethers::utils::keccak256;
 use lighthouse_metrics::inc_counter;
@@ -25,7 +25,8 @@ pub(crate) struct Miner<'a> {
     pub range: RecallRange,
     pub miner_id: &'a H256,
     pub context: &'a MineContext,
-    pub target_quality: &'a U256,
+    pub subtask_digest: &'a H256,
+    pub pora_target: &'a U256,
     pub loader: &'a dyn PoraLoader,
     pub mine_range_config: &'a MineRangeConfig,
 }
@@ -107,11 +108,11 @@ impl<'a> Miner<'a> {
                 .range
                 .difficulty_scale_x64(self.context.flow_length.as_u64());
 
-            if quality <= (self.target_quality / difficulty_scale_x64) << 64 {
+            if quality <= (self.pora_target / difficulty_scale_x64) << 64 {
                 debug!(
-                    "Find a PoRA valid answer, quality: {}, target_quality {}, scale {:.3}",
+                    "Find a PoRA valid answer, quality: {}, pora_target {}, scale {:.3}",
                     U256::MAX / quality,
-                    U256::MAX / self.target_quality,
+                    U256::MAX / self.pora_target,
                     difficulty_scale_x64.as_u128() as f64 / u64::MAX as f64
                 );
                 inc_counter(&HIT_COUNT);
@@ -139,7 +140,7 @@ impl<'a> Miner<'a> {
             let mut hasher = Blake2b512::new();
             hasher.update(self.miner_id);
             hasher.update(nonce);
-            hasher.update(self.context.digest);
+            hasher.update(self.subtask_digest);
             hasher.update(self.range.digest());
             hasher.finalize().into()
         };

node/miner/src/sealer.rs

@@ -87,7 +87,7 @@ impl Sealer {
 
     async fn update_flow_length(&mut self) -> Result<()> {
         let recent_context = self.flow_contract.make_context_with_result().call().await?;
-        debug!(target: "seal", "Recent context is {:?}", recent_context);
+        debug!("Recent context is {:?}", recent_context);
 
         let recent_flow_length = recent_context.flow_length.as_u64();
         if self.last_context_flow_length < recent_flow_length {

node/miner/src/service.rs

@@ -46,6 +46,7 @@ impl MineService {
             msg_recv.resubscribe(),
             provider.clone(),
             &config,
+            miner_id,
         );
 
         let mine_answer_receiver = PoraService::spawn(

node/miner/src/watcher.rs

@@ -1,6 +1,6 @@
 #![allow(unused)]
 
-use contract_interface::{zgs_flow::MineContext, PoraMine, ZgsFlow};
+use contract_interface::{zgs_flow::MineContext, PoraMine, WorkerContext, ZgsFlow};
 use ethereum_types::{Address, H256, U256};
 use ethers::{
     contract::Contract,
@@ -36,6 +36,7 @@ pub struct MineContextWatcher {
     mine_context_sender: broadcast::Sender<MineContextMessage>,
     last_report: MineContextMessage,
     query_interval: Duration,
+    miner_id: H256,
 
     msg_recv: broadcast::Receiver<MinerMessage>,
 }
@@ -46,6 +47,7 @@ impl MineContextWatcher {
         msg_recv: broadcast::Receiver<MinerMessage>,
         provider: Arc<Provider<RetryClient<Http>>>,
         config: &MinerConfig,
+        miner_id: H256,
     ) -> broadcast::Receiver<MineContextMessage> {
         let mine_contract = PoraMine::new(config.mine_address, provider.clone());
         let flow_contract = ZgsFlow::new(config.flow_address, provider.clone());
@@ -60,6 +62,7 @@ impl MineContextWatcher {
             msg_recv,
             last_report: None,
             query_interval: config.context_query_interval,
+            miner_id,
         };
         executor.spawn(
             async move { Box::pin(watcher.start()).await },
@@ -105,20 +108,26 @@ impl MineContextWatcher {
     }
 
     async fn query_recent_context(&mut self) -> Result<(), String> {
-        let context_call = self.flow_contract.make_context_with_result();
+        let miner_id = self.miner_id.0;
-        let valid_call = self.mine_contract.can_submit();
+        let WorkerContext {
-        let quality_call = self.mine_contract.pora_target();
+            context,
-        let shards_call = self.mine_contract.max_shards();
+            pora_target,
-
+            subtask_digest,
-        let (context, can_submit, quality, max_shards) = try_join!(
+            max_shards,
-            context_call.call(),
+        } = self
-            valid_call.call(),
+            .mine_contract
-            quality_call.call(),
+            .compute_worker_context(miner_id)
-            shards_call.call()
+            .call()
-        )
+            .await
             .map_err(|e| format!("Failed to query mining context: {:?}", e))?;
-        let report = if can_submit && context.digest != EMPTY_HASH.0 {
+
-            Some(PoraPuzzle::new(context, quality, max_shards))
+        let report = if pora_target > U256::zero() && context.digest != EMPTY_HASH.0 {
+            Some(PoraPuzzle::new(
+                context,
+                pora_target,
+                max_shards,
+                H256(subtask_digest),
+            ))
         } else {
             None
         };
@@ -127,6 +136,8 @@ impl MineContextWatcher {
             return Ok(());
         }
 
+        debug!("Update pora puzzle: {:?}", report);
+
         self.mine_context_sender
             .send(report.clone())
             .map_err(|e| format!("Failed to send out the most recent mine context: {:?}", e))?;

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

@@ -1 +1 @@
-bc826a18642023401040cf6dd4f2b10b0748075f
+1e931c7b168f9bc2b55f7b8fd96946e35b373048

tests/mine_with_market_test.py

@@ -75,7 +75,7 @@ class MineTest(TestFramework):
         wait_until(lambda: self.mine_contract.last_mined_epoch() == start_epoch + 1 and not self.mine_contract.can_submit(), timeout=120)
 
         rewards = self.reward_contract.reward_distributes()
-        assert_equal(len(rewards), 2)
+        assert_equal(len(rewards), 4)
         firstReward = rewards[0].args.amount
         self.log.info("Received reward %d Gwei", firstReward / (10**9))
 
@@ -96,8 +96,8 @@ class MineTest(TestFramework):
         assert_equal(self.contract.epoch(), start_epoch + 2)
 
         rewards = self.reward_contract.reward_distributes()
-        assert_equal(len(rewards), 4)
+        assert_equal(len(rewards), 8)
-        secondReward = rewards[2].args.amount
+        secondReward = rewards[4].args.amount
         self.log.info("Received reward %d Gwei", secondReward / (10**9))
 
         assert_greater_than(secondReward, 100 * firstReward / (start_epoch + 1))

tests/test_framework/contract_proxy.py

@@ -105,6 +105,9 @@ class MineContractProxy(ContractProxy):
     def can_submit(self, node_idx=0):
         return self._call("canSubmit", node_idx)
     
+    def current_submissions(self, node_idx=0):
+        return self._call("currentSubmissions", node_idx)
+
     def set_quality(self, quality, node_idx=0):
         return self._send("setQuality", node_idx, _targetQuality=quality)