enhanced account and balance

TODO

@@ -1,5 +1,6 @@
 Non Exaustive list of ideas:
 
+- Add Readme
 - Scripting with rhai  []
 - Mouse Input          []
 - Http server          []

node/.rustfmt.toml

@@ -0,0 +1 @@
+tab_spaces = 2

node/Cargo.lock

@@ -177,6 +177,7 @@ dependencies = [
  "chrono",
  "clap",
  "crossterm 0.29.0",
+ "futures",
  "hex",
  "memory-stats",
  "once_cell",
@@ -398,6 +399,7 @@ dependencies = [
  "crossterm_winapi",
  "derive_more",
  "document-features",
+ "futures-core",
  "mio",
  "parking_lot 0.12.4",
  "rustix 1.0.8",
@@ -550,12 +552,65 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "futures"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+]
+
 [[package]]
 name = "futures-core"
 version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-io"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.31"
@@ -574,9 +629,13 @@ version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
 dependencies = [
+ "futures-channel",
  "futures-core",
+ "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "pin-utils",
  "slab",

node/Cargo.toml

@@ -16,7 +16,7 @@ tokio-tungstenite = "0.27.0"
 uuid = { version = "1.18.0", features = ["v4", "serde"] }
 vlogger = { path = "./lib/logger-rs" }
 ratatui = "0.29.0"
-crossterm = "0.29.0"
+crossterm = { version = "0.29.0", features = ["event-stream"] }
 once_cell = "1.21.3"
 async-trait = "0.1.89"
 anyhow = "1.0.99"
@@ -26,3 +26,4 @@ memory-stats = "1.2.0"
 textwrap = "0.16.2"
 sled = "0.34.7"
 bincode = { version = "2.0.1", features = ["derive", "serde"] }
+futures = "0.3.31"

node/proc/15:16:35_93227

@@ -0,0 +1,6 @@
+[INFO ] [watcher.rs     :145 ] 15:16:45: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:16:45: Virtual memory usage: 1691 MB
+[INFO ] [watcher.rs     :145 ] 15:16:55: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:16:55: Virtual memory usage: 1691 MB
+[INFO ] [watcher.rs     :145 ] 15:17:05: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:17:05: Virtual memory usage: 1691 MB

node/proc/15:17:19_93429

@@ -0,0 +1,4 @@
+[INFO ] [watcher.rs     :145 ] 15:17:29: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:17:29: Virtual memory usage: 1691 MB
+[INFO ] [watcher.rs     :145 ] 15:17:39: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:17:39: Virtual memory usage: 1691 MB

node/proc/15:40:52_98073

@@ -0,0 +1,6 @@
+[INFO ] [watcher.rs     :145 ] 15:41:02: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:41:02: Virtual memory usage: 1691 MB
+[INFO ] [watcher.rs     :145 ] 15:41:12: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:41:12: Virtual memory usage: 1691 MB
+[INFO ] [watcher.rs     :145 ] 15:41:22: Physical memory usage: 32 MB
+[INFO ] [watcher.rs     :154 ] 15:41:22: Virtual memory usage: 1691 MB

node/src/args.rs

@@ -94,14 +94,6 @@ pub enum CliBlockCommand {
   #[command(name = "create", aliases = ["c", "new"])]
   Create,
 
-  /// Export Blocks to file
-  #[command(name = "dump", aliases = ["export"])]
-  Dump {
-    /// Output file
-    #[arg(short, long)]
-    output: String,
-  },
-
   /// Display Block by Hash
   #[command(name = "display", aliases = ["d"])]
   #[group(multiple = false)]

node/src/bus/error.rs

@@ -0,0 +1,21 @@
+use once_cell::sync::Lazy;
+use std::sync::Arc;
+use tokio::sync::broadcast;
+
+use super::event_bus::EventBus;
+use crate::executor::ExecutorCommand;
+
+pub enum ErrorEvent {
+
+}
+
+static ERROR_BUS: Lazy<Arc<EventBus<ExecutorCommand>>> =
+  Lazy::new(|| Arc::new(EventBus::new()));
+
+pub fn publish_error(event: ExecutorCommand) {
+  ERROR_BUS.publish(event);
+}
+
+pub fn subscribe_error_bus() -> broadcast::Receiver<ExecutorCommand> {
+  ERROR_BUS.subscribe()
+}

node/src/bus/system.rs

@@ -9,7 +9,8 @@ pub enum SystemEvent {
   ExecutorStarted,
   RendererStarted,
   NodeStarted,
-  Exit,
+  Shutdown,
+  Abort,
 }
 
 static SYSTEM_EVENT_BUS: Lazy<Arc<EventBus<SystemEvent>>> = Lazy::new(|| Arc::new(EventBus::new()));

node/src/cli.rs

@@ -18,7 +18,6 @@ pub fn handle_peer_command(cmd: CliPeerCommand) -> NodeCommand {
 pub fn handle_block_command(cmd: CliBlockCommand) -> NodeCommand {
   match cmd {
     CliBlockCommand::List => NodeCommand::ListBlocks,
-    CliBlockCommand::Dump { output } => NodeCommand::DumpBlocks(output),
     CliBlockCommand::Create => NodeCommand::CreateBlock,
     CliBlockCommand::Display { key, height } => match (key, height) {
       (Some(k), _) => return NodeCommand::DisplayBlockByKey(k),

node/src/core/account.rs

@@ -0,0 +1,22 @@
+pub type Address = String;
+
+#[derive(Debug, bincode::Decode, bincode::Encode)]
+pub struct Account {
+  address: Address,
+  balance: u64,
+  nonce: u64,
+}
+
+impl Account {
+  pub fn nonce(&self) -> u64 {
+    self.nonce
+  }
+
+  pub fn balance(&self) -> u64 {
+    self.balance
+  }
+
+  pub fn address(&self) -> &Address {
+    &self.address
+  }
+}

node/src/core/blockchain.rs

@@ -1,4 +1,7 @@
 use std::sync::Arc;
+use std::time::UNIX_EPOCH;
+
+use thiserror::*;
 
 use crate::core;
 use crate::core::ChainData;
@@ -10,15 +13,10 @@ use crate::log;
 
 use super::hasher::Hasher;
 
-use std::collections::HashMap;
-use std::time::UNIX_EPOCH;
-use vlogger::*;
-
-use thiserror::*;
 #[allow(dead_code)]
 #[derive(Error, Debug)]
 pub enum BlockchainError {
-  #[error("Blockchain initialisation failed")]
+  #[error("Database operation failed")]
   Database(#[from] DatabaseError),
 
   #[error("invalid account creation")]
@@ -30,14 +28,15 @@ pub enum BlockchainError {
   #[error("Validation Error")]
   Validation(#[from] ValidationError),
 
+  #[error("Insufficient fonds on address {0}")]
+  InsufficientFunds(String),
+
   #[error("Block Creation Error")]
   BlockCreation,
 }
 
 const BLOCKCHAIN_ID: &str = "watch-chain";
 
-pub type Account = String;
-
 #[derive(Debug, thiserror::Error)]
 pub enum ValidationError {
   #[error("Invalid Block Hash Detected at height {0}")]
@@ -52,27 +51,12 @@ pub enum ValidationError {
 #[derive(Debug)]
 pub struct Blockchain {
   id: String,
-  balances: std::collections::HashMap<Account, u32>,
   mempool: Vec<ChainData>,
   db: database::ChainDb,
 }
 
 #[allow(dead_code)]
 impl Blockchain {
-  pub fn add(&mut self, data: ChainData) -> Result<(), BlockchainError> {
-    self.apply(data.clone())?;
-    self.mempool.push(data);
-    Ok(())
-  }
-
-  fn acc_exists(&self, acc: &Account) -> bool {
-    self.balances.iter().find(|(k, _)| *k == acc).is_some()
-  }
-
-  pub fn dump_blocks(&self, path: String) {
-    log(msg!(DEBUG, "TODO: implement block export"));
-  }
-
   fn hash_transaction_pool(&self) -> Vec<String> {
     self
       .mempool
@@ -125,32 +109,23 @@ impl Blockchain {
 
   fn apply_transaction(&mut self, tx: &core::Tx) -> Result<(), BlockchainError> {
     tx.validate()?;
-    return Ok(());
+    let from = tx.from();
-    if let Some(from_balance) = self.balances.get_mut(tx.from()) {
+    let to = tx.to();
-      if *from_balance > tx.value() {
+    let from_balance = self.db.get_balance(tx.from())?;
-        *from_balance -= tx.value();
+    let to_balance = self.db.get_balance(tx.to())?;
-      } else {
+
-        return Err(BlockchainError::Tx(TxError::FromInsuffitientFonds));
+    if tx.value() > from_balance {
-      }
+      return Err(BlockchainError::InsufficientFunds(tx.from().to_string()))
-    } else {
-      return Err(BlockchainError::Tx(TxError::UnknownAccount(
-        tx.from().to_string(),
-      )));
     }
 
-    if let Some(to_balance) = self.balances.get_mut(&tx.to().to_string()) {
+    let new_from_balance = from_balance - tx.value();
-      *to_balance += tx.value()
+    let new_to_balance = to_balance + tx.value();
-    } else {
+
-      if self.acc_exists(tx.to()) {
+    let changes = vec![(from, new_from_balance), (to, new_to_balance)];
-        *self.balances.get_mut(tx.to()).unwrap() += tx.value();
+    Ok(self.db.set_balance_batch(changes)?)
-      } else {
-        self.balances.insert(tx.to().clone(), tx.value());
-      }
-    }
-    Ok(())
   }
 
-  pub fn apply(&mut self, data: ChainData) -> Result<(), BlockchainError> {
+  pub fn apply_chain_data(&mut self, data: ChainData) -> Result<(), BlockchainError> {
     match &data {
       ChainData::Transaction(tx) => {
         self.apply_transaction(tx)?;
@@ -171,10 +146,6 @@ impl Blockchain {
     Ok(ret)
   }
 
-  pub fn get_balances(&self) -> &std::collections::HashMap<String, u32> {
-    &self.balances
-  }
-
   pub fn blocks(&self) -> Result<Vec<std::sync::Arc<core::Block>>, BlockchainError> {
     Ok(self.db.get_all_blocks()?)
   }
@@ -238,12 +209,18 @@ impl Blockchain {
     Ok(())
   }
 
+  pub async fn shutdown(&self) -> Result<(), BlockchainError> {
+    self.db.dump_mempool(&self.mempool)?;
+    self.db.shutdown().await?;
+    Ok(())
+  }
+
   pub fn build(path: Option<String>) -> Result<Blockchain, BlockchainError> {
     let db = db::ChainDb::new(path).or_else(|e| Err(BlockchainError::Database(e)))?;
+    let mempool = db.recover_mempool()?;
 
     let chain = Blockchain {
-      balances: HashMap::new(),
+      mempool,
-      mempool: vec![],
       id: BLOCKCHAIN_ID.to_string(),
       db,
     };

node/src/core/tx.rs

@@ -1,18 +1,18 @@
-use crate::core::Account;
 use crate::error::TxError;
+use super::Address;
 
 #[derive(
   serde::Deserialize, serde::Serialize, Debug, clap::Args, Clone, bincode::Encode, bincode::Decode,
 )]
 pub struct Tx {
-  from: Account,
+  from: String,
-  to: Account,
+  to: String,
-  value: u32,
+  value: u64,
   data: String,
 }
 
 impl Tx {
-  pub fn new(from: Account, to: Account, value: u32, data: String) -> Self {
+  pub fn new(from: String, to: String, value: u64, data: String) -> Self {
     Self {
       from,
       to,
@@ -37,13 +37,13 @@ impl Tx {
   pub fn is_reward(&self) -> bool {
     return self.data == "reward";
   }
-  pub fn from(&self) -> &Account {
+  pub fn from(&self) -> &Address {
     &self.from
   }
-  pub fn to(&self) -> &Account {
+  pub fn to(&self) -> &Address {
     &self.to
   }
-  pub fn value(&self) -> u32 {
+  pub fn value(&self) -> u64 {
     self.value
   }
   pub fn data(&self) -> &str {

node/src/db/database.rs

@@ -1,53 +1,73 @@
-use crate::{
-  core::{self, Block, ChainData, Hasher},
-  db::error::DatabaseError,
-  error::print_error_chain,
-  log,
-};
 use bincode::{self, config::Configuration};
 use sled::{self, Batch};
 use std::sync::Arc;
 use vlogger::*;
 
+use crate::{
+  core::{self, Block, ChainData, Hasher, Address, Account},
+  db::error::DatabaseError,
+  error::print_error_chain,
+  log,
+};
+
 static BINCODE_CONFIG: Configuration = bincode::config::standard();
 
 const DB_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/database");
 
 const DB_TREE: &str = "blocks";
 
-const BLOCK_INDEX: &str = "blocks";
+const BLOCK_PREFIX: &str = "blocks";
-const CHAIN_DATA_INDEX: &str = "chain_data";
+const CHAIN_DATA_PREFIX: &str = "chain_data";
-const DATA_TO_BLOCK_INDEX: &str = "data_to_block";
+const DATA_TO_BLOCK_PREFIX: &str = "data_to_block";
-const METADATA_INDEX: &str = "metadata";
+const METADATA_PREFIX: &str = "metadata";
+const BALANCE_PREFIX: &str = "balance";
+const ACCOUNT_PREFIX: &str = "account";
+
+const MEMPOOL_PREFIX: &str = "mempool";
 
 const TIP_KEY: &str = "chain_tip";
 const HEIGHT_KEY: &str = "chain_height";
-
+const HEIGHT_TO_HASH_PREFIX: &str = "height_to_hash";
-const HEIGHT_TO_HASH_INDEX: &str = "height_to_hash";
 
 #[derive(Debug)]
 pub struct ChainDb {
   db: sled::Tree,
 }
 
-fn data_index(key: &str) -> String {
+fn data_prefix(key: &str) -> String {
-  format!("{}:{}", CHAIN_DATA_INDEX, key)
+  format!("{}:{}", CHAIN_DATA_PREFIX, key)
 }
 
-fn data_to_block_index(key: &str) -> String {
+fn data_to_block_prefix(key: &str) -> String {
-  format!("{}:{}", DATA_TO_BLOCK_INDEX, key)
+  format!("{}:{}", DATA_TO_BLOCK_PREFIX, key)
 }
 
-fn block_index(key: &str) -> String {
+fn block_prefix(key: &str) -> String {
-  format!("{}:{}", BLOCK_INDEX, key)
+  format!("{}:{}", BLOCK_PREFIX, key)
 }
 
-fn metadata_index(key: &str) -> String {
+fn metadata_prefix(key: &str) -> String {
-  format!("{}:{}", METADATA_INDEX, key)
+  format!("{}:{}", METADATA_PREFIX, key)
 }
 
-fn height_to_hash_index(height: u64) -> String {
+fn height_to_hash_prefix(height: u64) -> String {
-  format!("{}:{:020}", HEIGHT_TO_HASH_INDEX, height)
+  format!("{}:{:020}", HEIGHT_TO_HASH_PREFIX, height)
+}
+
+fn balances_prefix(address: &str) -> String {
+  format!("{}:{}", BALANCE_PREFIX, address)
+}
+
+fn nonce_prefix(address: &str) -> String {
+  format!("{}:{}", ACCOUNT_PREFIX, address)
+}
+
+fn mempool_prefix(tx_hash: &str) -> String {
+  format!("{}:{}", MEMPOOL_PREFIX, tx_hash)
+}
+
+fn account_prefix(addr: &str) -> String {
+  format!("{}:{}", ACCOUNT_PREFIX, addr)
 }
 
 impl ChainDb {
@@ -83,12 +103,118 @@ impl ChainDb {
     }
   }
 
-  pub fn get_block_by_key(&self, block_hash: &str) -> Result<Option<core::Block>, DatabaseError> {
+  pub fn get_account(&self, addr: &Address) -> Result<Option<Account>, DatabaseError> {
-    let block_hash = block_index(block_hash);
+    if let Some(bin_acc) = self.db.get(account_prefix(addr))? {
+      let (acc, _) = bincode::decode_from_slice::<Account, _>(&bin_acc, BINCODE_CONFIG)?;
+      Ok(Some(acc))
+    } else {
+      Ok(None)
+    }
+  }
+
+  pub fn insert_account(&self, acc: &Account) -> Result<(), DatabaseError> {
+    let mut batch = Batch::default();
+    if let Some(_) = self.get_account(acc.address())? {
+      return Err(DatabaseError::AccountExists(acc.address().to_string()));
+    }
+    let bin_acc = bincode::encode_to_vec(acc, BINCODE_CONFIG)?;
+    let bin_nonce = bincode::encode_to_vec(&acc.nonce(), BINCODE_CONFIG)?;
+    let bin_balance = bincode::encode_to_vec(&acc.balance(), BINCODE_CONFIG)?;
+    batch.insert(account_prefix(acc.address()).as_str(), bin_acc);
+    batch.insert(nonce_prefix(acc.address()).as_str(), bin_nonce);
+    batch.insert(balances_prefix(acc.address()).as_str(), bin_balance);
+    Ok(self.db.apply_batch(batch)?)
+  }
+
+  pub fn recover_mempool(&self) -> Result<Vec<ChainData>, DatabaseError> {
+    let mem_tree = self.db.scan_prefix(MEMPOOL_PREFIX);
+    let mut mem_vec = vec![];
+    for mem in mem_tree {
+      if let Ok((_, value)) = mem {
+        let (value, _) = bincode::decode_from_slice::<ChainData, _>(&value, BINCODE_CONFIG)?;
+        mem_vec.push(value);
+      } else {
+        return Err(DatabaseError::Corruption("Failed to recover Mempool".to_string()))
+      }
+    }
+    Ok(mem_vec)
+  }
+
+  pub fn dump_mempool(&self, mempool: &[ChainData]) -> Result<(), DatabaseError> {
+    let mut batch = Batch::default();
+    for mem in mempool {
+      let data_hash = Hasher::hash_chain_data(&mem);
+      let bin_data = bincode::encode_to_vec(&mem, BINCODE_CONFIG)?;
+      let key = mempool_prefix(&data_hash);
+      batch.insert(key.as_str(), bin_data);
+    }
+    self.db.apply_batch(batch)?;
+    Ok(())
+  }
+
+  pub async fn shutdown(&self) -> Result<usize, DatabaseError> {
+    Ok(self.db.flush_async().await?)
+  }
+
+  pub fn set_balance(&self, address: &Address, new_balance: u64) -> Result<(), DatabaseError> {
+    let mut batch = Batch::default();
+    let account_nonce = self.get_nonce(address)?;
+    let account_nonce = account_nonce + 1;
+    let bin_balance = bincode::encode_to_vec(new_balance, BINCODE_CONFIG)?;
+    let bin_nonce = bincode::encode_to_vec(account_nonce, BINCODE_CONFIG)?;
+    batch.insert(balances_prefix(address).as_str(), bin_balance);
+    batch.insert(nonce_prefix(address).as_str(), bin_nonce);
+    Ok(self.db.apply_batch(batch)?)
+  }
+
+  fn set_nonce(&self, address: &Address, nonce: u64) -> Result<(), DatabaseError> {
+    let bin_nonce = bincode::encode_to_vec(&nonce, BINCODE_CONFIG)?;
+    let address = nonce_prefix(address);
+    self.db.insert(address, bin_nonce)?;
+    Ok(())
+  }
+
+  fn get_nonce(&self, address: &Address) -> Result<u64, DatabaseError> {
+    match self.db.get(nonce_prefix(address))? {
+      Some(n) => {
+        let (nonce, _) = bincode::decode_from_slice::<u64, _>(&n, BINCODE_CONFIG)?;
+        Ok(nonce)
+      },
+      None => Err(DatabaseError::AccountNotFound(address.to_string()))
+    }
+  }
+
+  pub fn set_balance_batch(&self, changes: Vec<(&Address, u64)>) -> Result<(), DatabaseError> {
+    let mut batch = Batch::default();
+
+    for (address, amount) in changes {
+      let account_nonce = self.get_nonce(address)?;
+      let account_nonce = account_nonce + 1;
+      let balance_key = balances_prefix(&address);
+      let bin_amount = bincode::encode_to_vec(amount, BINCODE_CONFIG)?;
+      let bin_nonce = bincode::encode_to_vec(&account_nonce, BINCODE_CONFIG)?;
+      batch.insert(nonce_prefix(address).as_str(), bin_nonce);
+      batch.insert(balance_key.as_str(), bin_amount);
+    }
+    Ok(self.db.apply_batch(batch)?)
+  }
+
+  pub fn get_balance(&self, address: &Address) -> Result<u64, DatabaseError> {
+    match self.db.get(balances_prefix(address))? {
+      Some(b) => {
+        let (balance, _) = bincode::decode_from_slice::<u64, _>(&b, BINCODE_CONFIG)?;
+        Ok(balance)
+      },
+      None => Err(DatabaseError::AccountNotFound(address.to_string())),
+    }
+  }
+
+  pub fn get_block_by_key(&self, block_hash: &str) -> Result<Option<Arc<core::Block>>, DatabaseError> {
+    let block_hash = block_prefix(block_hash);
     if let Some(bin_block) = self.db.get(block_hash)? {
       let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&bin_block, BINCODE_CONFIG)
         .map_err(|e| DatabaseError::Decode(e))?;
-      Ok(Some(block))
+      Ok(Some(block.into()))
     } else {
       Ok(None)
     }
@@ -98,22 +224,19 @@ impl ChainDb {
     &self,
     height: u64,
   ) -> Result<Option<Arc<core::Block>>, DatabaseError> {
-    for result in self.db.scan_prefix(BLOCK_INDEX) {
+    if let Some(hash) = self.db.get(height_to_hash_prefix(height))? {
-      let (_key, value) = result?;
+      let (hash_str, _) = bincode::decode_from_slice::<String, _>(&hash, BINCODE_CONFIG)?;
-      let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&value, BINCODE_CONFIG)?;
+      Ok(self.get_block_by_key(&hash_str)?)
-
+    } else {
-      if block.head().height == height {
-        return Ok(Some(Arc::new(block)));
-      }
-    }
       Ok(None)
     }
+  }
 
   pub fn get_block_data(&self, block: &Block) -> Result<Arc<Vec<ChainData>>, DatabaseError> {
     let mut chain_data = Vec::new();
 
     for data_hash in &block.data {
-      let data_hash = data_index(data_hash);
+      let data_hash = data_prefix(data_hash);
       if let Some(bin_data) = self.db.get(&data_hash)? {
         let (data, _) = bincode::decode_from_slice::<ChainData, _>(&bin_data, BINCODE_CONFIG)?;
         chain_data.push(data);
@@ -128,7 +251,7 @@ impl ChainDb {
   pub fn get_all_blocks(&self) -> Result<Vec<Arc<core::Block>>, DatabaseError> {
     self
       .db
-      .scan_prefix(BLOCK_INDEX)
+      .scan_prefix(BLOCK_PREFIX)
       .map(|res| -> Result<Arc<core::Block>, DatabaseError> {
         let (_key, value) = res?;
         let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&value, BINCODE_CONFIG)
@@ -140,7 +263,7 @@ impl ChainDb {
 
   pub fn add_data(&self, data: &core::ChainData) -> Result<(), DatabaseError> {
     let bin_data = bincode::encode_to_vec(data, BINCODE_CONFIG)?;
-    let data_hash = data_index(&Hasher::hash_chain_data(data));
+    let data_hash = data_prefix(&Hasher::hash_chain_data(data));
     self.db.insert(data_hash, bin_data)?;
     Ok(())
   }
@@ -148,21 +271,22 @@ impl ChainDb {
   pub fn add_block(&self, block: &core::Block) -> Result<(), DatabaseError> {
     let mut db_batch = Batch::default();
     let bin_block = bincode::encode_to_vec(block, BINCODE_CONFIG)?;
-    db_batch.insert(block_index(block.head().block_hash()).as_str(), bin_block);
+    db_batch.insert(block_prefix(block.head().block_hash()).as_str(), bin_block);
     db_batch.insert(
-      height_to_hash_index(block.head().height).as_str(),
+      height_to_hash_prefix(block.head().height).as_str(),
       block.head().block_hash(),
     );
     for data in block.data() {
       db_batch.insert(
-        data_to_block_index(data.as_str()).as_str(),
+        data_to_block_prefix(data.as_str()).as_str(),
         block.head().block_hash(),
       );
     }
-    db_batch.insert(metadata_index(TIP_KEY).as_str(), block.head().block_hash());
+    db_batch.insert(metadata_prefix(TIP_KEY).as_str(), block.head().block_hash());
+    let bin_head = bincode::encode_to_vec(&block.head().height, BINCODE_CONFIG)?;
     db_batch.insert(
-      metadata_index(HEIGHT_KEY).as_str(),
+      metadata_prefix(HEIGHT_KEY).as_str(),
-      &block.head().height.to_be_bytes(),
+      bin_head
     );
     self.db.apply_batch(db_batch)?;
     Ok(())

node/src/db/error.rs

@@ -2,21 +2,20 @@ use thiserror::Error;
 
 #[derive(Debug, Error)]
 pub enum DatabaseError {
-  #[error("Database initialization failed")]
+  #[error("Database initialization failed: {0}")]
   Init(#[from] std::io::Error),
-
+  #[error("Database operation failed: {0}")]
-  #[error("Database read failed")]
+  Operation(#[from] sled::Error),
-  Read(#[from] anyhow::Error),
+  #[error("Failed to serialize data: {0}")]
-
-  #[error("Sled Failed")]
-  Sled(#[from] sled::Error),
-
-  #[error("Database Encode failed")]
   Encode(#[from] bincode::error::EncodeError),
-
+  #[error("Failed to deserialize data: {0}")]
-  #[error("Database Decode failed")]
   Decode(#[from] bincode::error::DecodeError),
-
+  #[error("Chain data not found for hash: {0}")]
-  #[error("Missing chain data for hash: {0}")]
   MissingData(String),
+  #[error("Account already exists: {0}")]
+  AccountExists(String),
+  #[error("Account does not exist: {0}")]
+  AccountNotFound(String),
+  #[error("Database corruption detected: {0}")]
+  Corruption(String),
 }

node/src/executor/executor.rs

@@ -1,12 +1,12 @@
 use crate::{
-  bus::{SystemEvent, publish_watcher_event, publish_system_event},
+  bus::{publish_system_event, publish_watcher_event, subscribe_system_event, SystemEvent},
   log,
   node::NodeCommand,
   renderer::RenderTarget,
   watcher::WatcherCommand,
 };
 use thiserror::Error;
-use tokio::sync::mpsc;
+use tokio::{select, sync::mpsc};
 use vlogger::*;
 
 use super::ExecutorCommand;
@@ -35,8 +35,21 @@ impl Executor {
 
   pub async fn run(&mut self) {
     publish_system_event(SystemEvent::ExecutorStarted);
+    let mut sys_rx = subscribe_system_event();
     while !self.exit {
-      self.listen().await;
+      select! {
+        _ = self.listen() => {}
+        event_res = sys_rx.recv() => {
+          if let Ok(event) = event_res {
+            match event {
+              SystemEvent::Shutdown => {
+                self.exit().await;
+              }
+              _ => {}
+            }
+          }
+        }
+      }
     }
   }
 
@@ -62,7 +75,7 @@ impl Executor {
   async fn echo(&self, s: Vec<String>) {
     let mut str = s.join(" ");
     str.push_str("\n");
-    let rd_cmd = WatcherCommand::Render(RenderCommand::RenderStringToPaneId {
+    let rd_cmd = WatcherCommand::Render(RenderCommand::StringToPaneId {
       str,
       pane: RenderTarget::CliOutput,
     });
@@ -70,7 +83,7 @@ impl Executor {
   }
 
   async fn invalid_command(&self, str: String) {
-    let rd_cmd = WatcherCommand::Render(RenderCommand::RenderStringToPaneId {
+    let rd_cmd = WatcherCommand::Render(RenderCommand::StringToPaneId {
       str,
       pane: RenderTarget::CliOutput,
     });

node/src/lib.rs

@@ -12,6 +12,26 @@ pub mod args;
 
 pub mod error;
 
+pub mod core {
+  pub mod block;
+  pub use block::*;
+
+  pub mod blockchain;
+  pub use blockchain::*;
+
+  pub mod tx;
+  pub use tx::*;
+
+  pub mod data;
+  pub use data::*;
+
+  pub mod hasher;
+  pub use hasher::*;
+
+  pub mod account;
+  pub use account::*;
+}
+
 pub mod db {
   pub mod database;
   pub mod error;
@@ -23,9 +43,10 @@ pub mod bus {
   pub mod event_bus;
   pub mod network;
   pub mod watcher;
+  pub mod error;
   pub mod system;
 
-  pub use executor::*;
+  pub use error::*;
   pub mod executor;
   pub use network::*;
   pub use watcher::*;
@@ -72,29 +93,12 @@ pub mod protocol {
   pub use connector::*;
 }
 
-pub mod core {
-  pub mod block;
-  pub use block::*;
-
-  pub mod blockchain;
-  pub use blockchain::*;
-
-  pub mod tx;
-  pub use tx::*;
-
-  pub mod data;
-  pub use data::*;
-
-  pub mod hasher;
-  pub use hasher::*;
-}
-
 pub mod seeds_constants;
 
 use crate::renderer::{RenderCommand, RenderTarget};
 
 pub fn log(msg: String) {
-  crate::bus::publish_watcher_event(watcher::WatcherCommand::Render(RenderCommand::RenderStringToPaneId {
+  crate::bus::publish_watcher_event(watcher::WatcherCommand::Render(RenderCommand::StringToPaneId {
     pane: RenderTarget::CliOutput,
     str: msg,
   }))

node/src/node/node.rs

@@ -1,4 +1,4 @@
-use crate::bus::{publish_system_event, publish_watcher_event, SystemEvent};
+use crate::bus::{publish_system_event, publish_watcher_event, subscribe_system_event, SystemEvent};
 use crate::core::{self, Blockchain, BlockchainError, ChainData, ValidationError};
 use crate::error::print_error_chain;
 use crate::executor::ExecutorCommand;
@@ -13,6 +13,7 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use thiserror::*;
+use tokio::select;
 use tokio::sync::mpsc;
 use uuid::Uuid;
 use vlogger::*;
@@ -74,7 +75,6 @@ pub enum NodeCommand {
   ListBlocks,
   ListPeers,
   ShowId,
-  DumpBlocks(String),
   ConnectToSeeds,
   ConnectTcpPeer(String),
   BootStrap,
@@ -155,6 +155,16 @@ impl Node {
     }
   }
 
+  async fn shutdown(&mut self) {
+    if let Some(conn) = &self.tcp_connector {
+      let res = conn.send(ConnectorCommand::Shutdown).await;
+      if res.is_err() {
+        log(msg!(ERROR, "Failed to send shutdown signal to connector"));
+      }
+    }
+    let _ = self.chain.shutdown().await;
+  }
+
   fn get_blocks(&self) -> Result<Vec<Arc<core::Block>>, NodeError> {
     Ok(self.chain.blocks()?)
   }
@@ -226,7 +236,7 @@ impl Node {
       }
       ProtocolMessage::ChainData { data, .. } => {
         log(msg!(DEBUG, "Received ChainData from {peer_id}"));
-        self.chain.apply(data).unwrap()
+        self.chain.apply_chain_data(data).unwrap()
       }
       _ => {
         log(msg!(DEBUG, "TODO: implement this message type"));
@@ -312,13 +322,6 @@ impl Node {
     return self.exec_tx.clone();
   }
 
-  async fn network_data(&mut self, data: ChainData) {
-    match self.chain.apply(data) {
-      Ok(_) => log(msg!(DEBUG, "ChainData Applied")),
-      Err(e) => print_error_chain(&e.into()),
-    };
-  }
-
   async fn connector_cmd(&self, cmd: ConnectorCommand) {
     match &self.tcp_connector {
       Some(t) => match t.send(cmd).await {
@@ -348,20 +351,7 @@ impl Node {
       .await;
   }
 
-  pub async fn run(&mut self) {
+  async fn accept_command(&mut self) {
-    if let Some(addr) = self.addr {
-      self.start_connection_listner(addr).await;
-    } else {
-      self
-        .start_connection_listner(SocketAddr::new(
-          std::net::IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)),
-          8080,
-        ))
-        .await;
-    };
-
-    publish_system_event(SystemEvent::NodeStarted);
-
     while let Some(command) = self.rx.recv().await {
       match command {
         NodeCommand::BootStrap => {
@@ -409,7 +399,9 @@ impl Node {
           self.process_message(peer_id, message).await;
         }
         NodeCommand::ProcessChainData(data) => {
-          self.network_data(data.clone()).await;
+          if let Err(e) = self.chain.apply_chain_data(data.clone()) {
+            print_error_chain(&e.into());
+          }
           self.broadcast_network_data(data).await;
         }
         NodeCommand::CreateBlock => {
@@ -453,9 +445,6 @@ impl Node {
           log(msg!(DEBUG, "Received DebugListBlocks command"));
           self.show_id().await;
         }
-        NodeCommand::DumpBlocks(s) => {
-          self.chain.dump_blocks(s);
-        }
         NodeCommand::Exit => {
           log(msg!(DEBUG, "Node Exit"));
           break;
@@ -463,4 +452,41 @@ impl Node {
       }
     }
   }
+
+  pub async fn run(&mut self) {
+    if let Some(addr) = self.addr {
+      self.start_connection_listner(addr).await;
+    } else {
+      self
+        .start_connection_listner(SocketAddr::new(
+          std::net::IpAddr::V4(std::net::Ipv4Addr::new(0, 0, 0, 0)),
+          8080,
+        ))
+        .await;
+    };
+    let mut system_rx = subscribe_system_event();
+    publish_system_event(SystemEvent::NodeStarted);
+
+    loop {
+      select! {
+        _ = self.accept_command() => {
+
+        }
+        event_result = system_rx.recv() => {
+          match event_result {
+            Ok(e) => {
+              match e {
+                SystemEvent::Shutdown => {
+                  break;
+                }
+                _ => {}
+              }
+            }
+            _ => {}
+          }
+        }
+      }
+    }
+    self.shutdown().await;
+  }
 }

node/src/protocol/connector.rs

@@ -19,6 +19,7 @@ use thiserror::*;
 pub enum ConnectorCommand {
   ConnectToTcpPeer(SocketAddr),
   ConnectToTcpSeed(SocketAddr),
+  Shutdown,
 }
 
 pub struct Connector {
@@ -26,6 +27,7 @@ pub struct Connector {
   addr: SocketAddr,
   exec_tx: mpsc::Sender<ExecutorCommand>,
   rx: mpsc::Receiver<ConnectorCommand>,
+  exit: bool,
 }
 
 #[derive(Error, Debug)]
@@ -48,6 +50,7 @@ impl Connector {
       addr,
       exec_tx,
       rx,
+      exit: false,
     }
   }
 
@@ -68,7 +71,7 @@ impl Connector {
       };
     }
     if let Some(listener) = listner {
-      loop {
+      while !self.exit {
         tokio::select! {
           cmd_result = self.rx.recv() => {
             match cmd_result {
@@ -109,6 +112,9 @@ impl Connector {
       ConnectorCommand::ConnectToTcpSeed(addr) => {
         self.connect_to_seed(addr).await;
       }
+      ConnectorCommand::Shutdown => {
+        self.exit = true;
+      }
     }
   }
 

node/src/renderer/pane.rs

@@ -8,9 +8,6 @@ use ratatui::{
   symbols::border,
   widgets::{Block, List, Paragraph, Widget},
 };
-use vlogger::{msg, DEBUG};
-
-use crate::log;
 
 use super::center;
 
@@ -122,7 +119,6 @@ impl Widget for &mut Pane {
           0
         };
         let scroll_offset = self.max_scroll.saturating_sub(self.scroll) as u16;
-        log(msg!(DEBUG, "idx {idx}"));
         let list_w = List::new(
           list
             .iter()

node/src/renderer/renderer.rs

@@ -26,14 +26,14 @@ pub struct Renderer {
 
 #[derive(Clone, Debug)]
 pub enum RenderCommand {
-  RenderStringToPaneId {
+  StringToPaneId {
     str: String,
     pane: RenderTarget,
   },
-  RenderStringToPaneFocused {
+  StringToPaneFocused {
     str: String,
   },
-  RenderKeyInput(KeyCode),
+  KeyInput(KeyCode),
   ListMove {
     pane: RenderTarget,
     index: usize,
@@ -53,7 +53,6 @@ pub enum RenderCommand {
 #[allow(dead_code)]
 impl Renderer {
   pub fn new(layout: RenderLayoutKind) -> Self {
-    todo!("Fix renderer idx in select mode");
     Self {
       buffer: String::new(),
       exit: false,
@@ -151,15 +150,18 @@ impl Renderer {
   }
 
   pub fn handle_arrow_key(&mut self, key: KeyCode) {
-    match &mut self.mode {
+    if let Some(pane) = self.focused() {
-      InputMode::Input => {}
+      match &mut pane.target {
-      InputMode::PopUp(ref content, .., idx) => {
+        RenderTarget::CliInput => {}
-        log(msg!(DEBUG, "Received keycode: {key}"));
+        RenderTarget::CliOutput => {}
+        RenderTarget::PopUp => {
+          match &mut pane.buffer {
+            RenderBuffer::Select(content, idx) => {
               log(msg!(DEBUG, "idx before: {idx}"));
               match key {
                 KeyCode::Up => { *idx = idx.saturating_sub(1) }
                 KeyCode::Down => {
-            if *idx < content.len() {
+                  if *idx < content.len().saturating_sub(1) {
                     *idx += 1;
                   }
                 }
@@ -167,11 +169,9 @@ impl Renderer {
               }
               log(msg!(DEBUG, "idx after: {idx}"))
             }
+            _ => {}
+          }
         }
-    if let Some(pane) = self.focused() {
-      match &pane.target {
-        RenderTarget::CliInput => {}
-        RenderTarget::CliOutput => {}
         _ => {}
       }
     }
@@ -216,6 +216,12 @@ impl Renderer {
     }
   }
 
+  fn clear_focus(&mut self) {
+    while let Some(focused) = self.focused() {
+      focused.focused = false;
+    }
+  }
+
   pub fn clear_pane(&mut self, pane: RenderTarget) {
     if matches!(pane, RenderTarget::All) {
       for p in self.layout.panes.iter_mut() {
@@ -248,7 +254,7 @@ impl Renderer {
       RenderCommand::MouseClickLeft(x, y) => self.handle_mouse_click_left(x, y, rects),
       RenderCommand::MouseScrollUp => self.handle_scroll_up(),
       RenderCommand::MouseScrollDown => self.handle_scroll_down(),
-      RenderCommand::RenderKeyInput(k) => match k {
+      RenderCommand::KeyInput(k) => match k {
         KeyCode::Char(c) => self.handle_char_input(c),
         KeyCode::Backspace => self.handle_backspace(),
         KeyCode::Enter => self.handle_enter(),
@@ -256,8 +262,8 @@ impl Renderer {
         _ => {}
       },
       RenderCommand::ListMove { pane, index } => self.list_move(pane, index),
-      RenderCommand::RenderStringToPaneFocused { str } => self.render_string_to_focused(str),
+      RenderCommand::StringToPaneFocused { str } => self.render_string_to_focused(str),
-      RenderCommand::RenderStringToPaneId { str, pane } => self.render_string_to_id(str, pane),
+      RenderCommand::StringToPaneId { str, pane } => self.render_string_to_id(str, pane),
       RenderCommand::Exit => self.exit(),
       RenderCommand::ChangeLayout(l) => self.layout = RenderLayout::generate(l),
       RenderCommand::ClearPane => self.clear_pane(RenderTarget::All),
@@ -269,6 +275,7 @@ impl Renderer {
             }
           }
           InputMode::PopUp(content, title, ..) => {
+            self.clear_focus();
             let pane = Pane::new(
               Some(title.to_string()),
               RenderTarget::PopUp,

node/src/watcher/watcher.rs

@@ -1,4 +1,4 @@
-use crate::{cli::cli, error::print_error_chain, node::node::NodeCommand, watcher::WatcherMode};
+use crate::{bus::{publish_system_event, subscribe_system_event, SystemEvent}, cli::cli, error::print_error_chain, node::node::NodeCommand, watcher::WatcherMode};
 use crossterm::event::{Event, EventStream, KeyCode, KeyEventKind, MouseButton, MouseEventKind};
 use futures::StreamExt;
 use memory_stats::memory_stats;
@@ -64,8 +64,12 @@ impl Watcher {
     )
   }
 
-  fn shutdown(&self) -> io::Result<()> {
+  async fn shutdown(&mut self) -> io::Result<()> {
     ratatui::restore();
+    let handles = std::mem::take(&mut self.handles);
+    for handle in handles {
+      handle.await.unwrap()
+    }
     crossterm::execute!(
       std::io::stdout(),
       crossterm::event::DisableBracketedPaste,
@@ -98,6 +102,7 @@ impl Watcher {
     let mut ui_rx = subscribe_watcher_event();
     let mut render_interval = interval(Duration::from_millis(32));
     let mut terminal = ratatui::init();
+    let mut system_rx = subscribe_system_event();
 
     self.init()?;
 
@@ -124,12 +129,22 @@ impl Watcher {
             }
           }
         }
+        event_res = system_rx.recv() => {
+          if let Ok(event) = event_res {
+            match event {
+              SystemEvent::Shutdown => {
+                break ;
+              }
+              _ => {}
+            }
+          }
+        }
         _ = render_interval.tick() => {
           terminal.draw(|frame| self.renderer.draw(frame))?;
         }
       }
     }
-    self.shutdown()
+    self.shutdown().await
   }
 
   pub fn build() -> WatcherBuilder {
@@ -257,7 +272,7 @@ impl Watcher {
         _ => {}
       },
       Event::Key(k) if k.kind == KeyEventKind::Press => match k.code {
-        KeyCode::Esc => return Ok(false),
+        KeyCode::Esc => publish_system_event(SystemEvent::Shutdown),
         KeyCode::Char(c) => {
           self.cmd_buffer.push(c);
           self.renderer.handle_char_input(c)