feat: cli

Cargo.toml

@@ -38,6 +38,8 @@ toml = "0.5"
 tokio = { version = "0.2", features = ["macros", "sync", "rt-core", "rt-util", "signal", "time"] }
 muta-apm = "0.1.0-alpha.7"
 futures-timer="3.0"
+cita_trie = "2.0"
+fs_extra = "1.2.0"
 
 [dev-dependencies]
 cita_trie = "2.0"

built-in-services/asset/src/tests/mod.rs

@@ -301,4 +301,16 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _ctx: Context) -> ProtocolResult<Bytes> {
         unimplemented!()
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
 }

built-in-services/metadata/src/tests/mod.rs

@@ -167,4 +167,16 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _ctx: Context) -> ProtocolResult<Bytes> {
         unimplemented!()
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
 }

built-in-services/multi-signature/src/tests/mod.rs

@@ -98,6 +98,18 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _ctx: Context) -> ProtocolResult<Bytes> {
         unimplemented!()
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
 }
 
 fn new_multi_signature_service() -> MultiSignatureService<

built-in-services/util/src/tests/mod.rs

@@ -193,4 +193,16 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _: Context) -> ProtocolResult<Bytes> {
         unimplemented!()
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
 }

core/api/src/adapter/mod.rs

@@ -4,9 +4,9 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use derive_more::Display;
 
-use protocol::traits::ExecutorFactory;
 use protocol::traits::{
-    APIAdapter, Context, ExecutorParams, MemPool, ServiceMapping, ServiceResponse, Storage,
+    APIAdapter, Context, ExecutorFactory, ExecutorParams, MemPool, ServiceMapping, ServiceResponse,
+    Storage,
 };
 use protocol::types::{Address, Block, Hash, Receipt, SignedTransaction, TransactionRequest};
 use protocol::{ProtocolError, ProtocolErrorKind, ProtocolResult};

core/consensus/src/lib.rs

@@ -30,7 +30,7 @@ use protocol::{ProtocolError, ProtocolErrorKind};
 pub use crate::adapter::OverlordConsensusAdapter;
 pub use crate::consensus::OverlordConsensus;
 pub use crate::synchronization::{OverlordSynchronization, RichBlock};
-pub use crate::wal::SignedTxsWAL;
+pub use crate::wal::{ConsensusWal, SignedTxsWAL};
 pub use overlord::{types::Node, DurationConfig};
 
 pub const DEFAULT_OVERLORD_GAP: usize = 5;

core/consensus/src/wal.rs

@@ -70,6 +70,26 @@ impl SignedTxsWAL {
         Ok(())
     }
 
+    pub fn available_height(&self) -> ProtocolResult<Vec<u64>> {
+        let dir_path = self.path.clone();
+        let mut availables = Vec::<u64>::new();
+        for item in fs::read_dir(dir_path).map_err(ConsensusError::WALErr)? {
+            let item = item.map_err(ConsensusError::WALErr)?;
+
+            if item.path().is_dir() {
+                availables.push(item.file_name().to_str().unwrap().parse().unwrap())
+            }
+        }
+        Ok(availables)
+    }
+
+    pub fn remove_all(&self) -> ProtocolResult<()> {
+        for height in self.available_height()? {
+            self.remove(height)?
+        }
+        Ok(())
+    }
+
     pub fn load(
         &self,
         height: u64,
@@ -219,20 +239,6 @@ impl ConsensusWal {
             if item.file_name() != OsString::from(OVERLORD_WAL.clone() + ".txt")
                 && item.file_name() != OsString::from(check_sum.as_hex() + ".txt")
             {
-                println!("{}", item.file_name().to_str().unwrap());
-                println!(
-                    "{}",
-                    OsString::from(OVERLORD_WAL.clone() + ".txt")
-                        .to_str()
-                        .unwrap()
-                );
-                println!(
-                    "{}",
-                    OsString::from(check_sum.as_hex() + ".txt")
-                        .to_str()
-                        .unwrap()
-                );
-
                 fs::remove_file(item.path()).map_err(ConsensusError::WALErr)?;
             }
         }
@@ -280,6 +286,20 @@ impl ConsensusWal {
             Err(ConsensusError::CheckSumMismatch.into())
         }
     }
+
+    pub fn clear(&self) -> ProtocolResult<()> {
+        let dir_path = self.path.clone();
+        if !dir_path.exists() {
+            return Ok(());
+        }
+
+        for item in fs::read_dir(dir_path).map_err(ConsensusError::WALErr)? {
+            let item = item.map_err(ConsensusError::WALErr)?;
+
+            fs::remove_file(item.path()).map_err(ConsensusError::WALErr)?;
+        }
+        Ok(())
+    }
 }
 
 #[rustfmt::skip]

core/storage/src/lib.rs

@@ -373,9 +373,7 @@ impl<Adapter: StorageAdapter> Storage for ImplStorage<Adapter> {
 
     #[muta_apm::derive::tracing_span(kind = "storage")]
     async fn insert_block(&self, ctx: Context, block: Block) -> ProtocolResult<()> {
-        self.adapter
-            .insert::<BlockSchema>(BlockKey::new(block.header.height), block.clone())
-            .await?;
+        self.set_block(ctx.clone(), block.clone()).await?;
 
         self.set_latest_block(ctx, block).await?;
 
@@ -386,6 +384,22 @@ impl<Adapter: StorageAdapter> Storage for ImplStorage<Adapter> {
         self.adapter.get::<BlockSchema>(BlockKey::new(height)).await
     }
 
+    // !!!be careful, only upsert the block, the prev_hash may mismatch and latest
+    // block may diverse!!!
+    async fn set_block(&self, _ctx: Context, block: Block) -> ProtocolResult<()> {
+        self.adapter
+            .insert::<BlockSchema>(BlockKey::new(block.header.height), block.clone())
+            .await?;
+        Ok(())
+    }
+
+    // !be careful, only call this function in maintenance mode!
+    async fn remove_block(&self, _ctx: Context, height: u64) -> ProtocolResult<()> {
+        self.adapter
+            .remove::<BlockSchema>(BlockKey::new(height))
+            .await
+    }
+
     async fn get_latest_block(&self, _ctx: Context) -> ProtocolResult<Block> {
         let opt_block = { self.latest_block.read().await.clone() };
 

examples/muta-chain.rs

@@ -52,8 +52,8 @@ impl ServiceMapping for DefaultServiceMapping {
     }
 }
 
-fn main() {
-    cli::cli_main(DefaultServiceMapping {});
+pub fn main() {
+    cli::start(DefaultServiceMapping {}, None).expect("error happens")
 }
 
 #[derive(Debug, Display, From)]

framework/src/binding/tests/sdk.rs

@@ -186,6 +186,18 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _ctx: Context) -> ProtocolResult<Bytes> {
         Err(StoreError::GetNone.into())
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        Ok(())
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        Ok(())
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        Ok(())
+    }
 }
 
 // #####################

framework/src/executor/tests/framework.rs

@@ -137,6 +137,14 @@ impl Storage for MockStorage {
         unimplemented!()
     }
 
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
     async fn insert_receipts(&self, _: Context, _: u64, _: Vec<Receipt>) -> ProtocolResult<()> {
         unimplemented!()
     }
@@ -166,6 +174,10 @@ impl Storage for MockStorage {
         unimplemented!()
     }
 
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
     async fn update_overlord_wal(&self, _: Context, _: Bytes) -> ProtocolResult<()> {
         unimplemented!()
     }

framework/src/executor/tests/mod.rs

@@ -667,4 +667,16 @@ impl Storage for MockStorage {
     async fn load_overlord_wal(&self, _ctx: Context) -> ProtocolResult<Bytes> {
         unimplemented!()
     }
+
+    async fn set_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn remove_block(&self, _ctx: Context, _height: u64) -> ProtocolResult<()> {
+        unimplemented!()
+    }
+
+    async fn set_latest_block(&self, _ctx: Context, _block: Block) -> ProtocolResult<()> {
+        unimplemented!()
+    }
 }

protocol/src/traits/storage.rs

@@ -59,6 +59,10 @@ pub trait Storage: Send + Sync {
 
     async fn get_block(&self, ctx: Context, height: u64) -> ProtocolResult<Option<Block>>;
 
+    async fn set_block(&self, _ctx: Context, block: Block) -> ProtocolResult<()>;
+
+    async fn remove_block(&self, ctx: Context, height: u64) -> ProtocolResult<()>;
+
     async fn insert_receipts(
         &self,
         ctx: Context,

protocol/src/types/block.rs

@@ -7,13 +7,13 @@ use crate::fixed_codec::{FixedCodec, FixedCodecError};
 use crate::types::{Address, Hash, MerkleRoot};
 use crate::ProtocolResult;
 
-#[derive(RlpFixedCodec, Clone, Debug, PartialEq, Eq)]
+#[derive(RlpFixedCodec, Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
 pub struct Block {
     pub header:            BlockHeader,
     pub ordered_tx_hashes: Vec<Hash>,
 }
 
-#[derive(RlpFixedCodec, Clone, Debug, Display, PartialEq, Eq)]
+#[derive(RlpFixedCodec, Clone, Debug, Display, PartialEq, Eq, Deserialize, Serialize)]
 #[display(
     fmt = "chain id {:?}, height {}, exec height {}, previous hash {:?},
     ordered root {:?}, order_signed_transactions_hash {:?}, confirm root {:?}, state root {:?},

src/.gitignore

@@ -0,0 +1,3 @@
+tests/data
+
+tests/save