Merge pull request #1 from JLerxky/main-cita

Main cita

Cargo.toml

@@ -1,30 +1,32 @@
 [package]
 name = "controller"
-version = "6.7.0"
+version = "6.7.2"
 authors = ["Rivtower Technologies <[email protected]>"]
 license = "Apache-2.0"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-clap = { version = "4.2", features = ["derive"] }
-tonic = "0.9"
-prost = "0.11"
-tokio = { version = "1.27", features = ["full"] }
+clap = { version = "4.4", features = ["derive"] }
+tonic = "0.10"
+prost = "0.12"
+tokio = { version = "1.32", features = ["full"] }
 rand = "0.8"
-toml = "0.7"
+toml = "0.8"
 serde = "1.0"
 serde_derive = "1.0"
 hex = "0.4"
 tower = "0.4"
 tracing = "0.1"
-tonic-reflection = "0.9"
-tonic-web = "0.9"
+tonic-reflection = "0.10"
+tonic-web = "0.10"
 statig = { version = "0.3", features = ["async"] }
-flume = "0.10"
-rayon = "1.0"
+flume = "0.11"
+rayon = "1.8"
 cfg-if = "1.0"
+futures = "0.3"
+indexmap = "2.1"
 
 cloud-util = { package = "cloud-util", git = "https://github.com/cita-cloud/cloud-common-rs" }
 cita_cloud_proto = { package = "cita_cloud_proto", git = "https://github.com/cita-cloud/cloud-common-rs" }
@@ -33,7 +35,7 @@ crypto_sm = { git = "https://github.com/cita-cloud/crypto_sm", package = "crypto
 crypto_eth = { git = "https://github.com/cita-cloud/crypto_eth", package = "crypto", optional = true }
 
 [build-dependencies]
-tonic-build = "0.9"
+tonic-build = "0.10"
 
 [profile.release.package."*"]
 # Set the default for dependencies.

src/config.rs

@@ -92,6 +92,8 @@ pub struct ControllerConfig {
     pub is_danger: bool,
     /// log config
     pub log_config: LogConfig,
+    /// tx pool persistence
+    pub tx_persistence: bool,
 }
 
 impl Default for ControllerConfig {
@@ -129,6 +131,7 @@ impl Default for ControllerConfig {
             buffer_duration: 300,
             is_danger: false,
             log_config: Default::default(),
+            tx_persistence: false,
         }
     }
 }

src/core/auditor.rs

@@ -24,6 +24,7 @@ use cita_cloud_proto::{
     },
     status_code::StatusCodeEnum,
 };
+use futures::{stream::FuturesUnordered, StreamExt};
 
 use crate::{
     core::system_config::{SystemConfig, LOCK_ID_BUTTON, LOCK_ID_VERSION},
@@ -32,7 +33,7 @@ use crate::{
 
 #[derive(Clone)]
 pub struct Auditor {
-    history_hashes: HashMap<u64, HashSet<Vec<u8>>>,
+    pub history_hashes: HashMap<u64, HashSet<Vec<u8>>>,
     current_block_number: u64,
     sys_config: SystemConfig,
 }
@@ -62,13 +63,34 @@ impl Auditor {
             1u64
         };
 
+        let mut history_hashes_num = 0;
+        let mut futures = FuturesUnordered::new();
+
         for h in begin_block_number..=init_block_number {
-            let block = get_compact_block(h).await.unwrap();
-            let block_body = block.body.unwrap();
-            self.history_hashes
-                .insert(h, HashSet::from_iter(block_body.tx_hashes));
+            futures.push(get_compact_block(h));
+        }
+
+        while let Some(result) = futures.next().await {
+            match result {
+                Ok(block) => {
+                    let block_body = block.body.unwrap();
+                    history_hashes_num += block_body.tx_hashes.len();
+                    self.history_hashes.insert(
+                        block.header.unwrap().height,
+                        HashSet::from_iter(block_body.tx_hashes),
+                    );
+                }
+                Err(e) => {
+                    error!("Failed to get compact block: {}", e);
+                }
+            }
         }
+
         self.current_block_number = init_block_number;
+        info!(
+            "Auditor init finished: history_hashes({})",
+            history_hashes_num
+        );
     }
 
     pub fn insert_tx_hash(&mut self, h: u64, hash_list: Vec<Vec<u8>>) {

src/core/chain.rs

@@ -285,7 +285,7 @@ impl Chain {
             };
 
             store_data(
-                i32::from(Regions::AllBlockData) as u32,
+                Regions::AllBlockData as u32,
                 block_height.to_be_bytes().to_vec(),
                 block_bytes,
             )
@@ -298,7 +298,7 @@ impl Chain {
             );
         }
 
-        // update auditor pool and systemconfig
+        // update auditor pool and system_config
         // even empty block, we also need update current height of auditor
         {
             if let Some(raw_txs) = block.body.clone() {
@@ -312,7 +312,7 @@ impl Chain {
                             // if sys_config changed, store utxo tx hash into global region
                             let lock_id = utxo_tx.transaction.as_ref().unwrap().lock_id;
                             store_data(
-                                i32::from(Regions::Global) as u32,
+                                Regions::Global as u32,
                                 lock_id.to_be_bytes().to_vec(),
                                 utxo_tx.transaction_hash,
                             )
@@ -396,7 +396,7 @@ impl Chain {
 
             if prev_hash != self.block_hash {
                 warn!(
-                    "commit block({}) failed: get prehash: 0x{}, correct prehash: 0x{}. hash: 0x{}",
+                    "commit block({}) failed: get prevhash: 0x{}, correct prevhash: 0x{}. hash: 0x{}",
                     height,
                     hex::encode(&prev_hash),
                     hex::encode(&self.block_hash),
@@ -491,7 +491,7 @@ impl Chain {
 
         if header.prevhash != self.block_hash {
             warn!(
-                "process block({}) failed: get prehash: 0x{}, correct prehash: 0x{}. hash: 0x{}",
+                "process block({}) failed: get prevhash: 0x{}, correct prevhash: 0x{}. hash: 0x{}",
                 height,
                 hex::encode(&header.prevhash),
                 hex::encode(&self.block_hash),
@@ -513,7 +513,10 @@ impl Chain {
                 .auditor_check_batch(block.body.as_ref().ok_or(StatusCodeEnum::NoneBlockBody)?)?
         }
 
-        crypto_check_batch_async(block.body.clone().ok_or(StatusCodeEnum::NoneBlockBody)?).await?;
+        crypto_check_batch_async(Arc::new(
+            block.body.clone().ok_or(StatusCodeEnum::NoneBlockBody)?,
+        ))
+        .await?;
 
         self.finalize_block(block, block_hash.clone()).await?;
 
@@ -546,7 +549,9 @@ impl Chain {
     }
 
     pub fn next_step(&self, global_status: &ChainStatus) -> ChainStep {
-        if global_status.height > self.block_number && self.candidates.is_empty() {
+        if global_status.height > self.block_number + 1
+            || (global_status.height > self.block_number && self.candidates.is_empty())
+        {
             debug!("sync mode");
             ChainStep::SyncStep
         } else {