victorvobis/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lets goo

Browse Source
This commit is contained in:
victor 2025-08-31 00:28:08 +02:00
parent d63ae5aa55
commit 97878de2ef
21 changed files with 651 additions and 415 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
node/.gitignore vendored
View File

@ -1 +1,2 @@
/target
database/

33
node/Cargo.lock generated
View File

@ -126,6 +126,26 @@ dependencies = [
"windows-targets 0.52.6",
]
[[package]]
name = "bincode"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "36eaf5d7b090263e8150820482d5d93cd964a81e4019913c972f4edcc6edb740"
dependencies = [
"bincode_derive",
"serde",
"unty",
]
[[package]]
name = "bincode_derive"
version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf95709a440f45e986983918d0e8a1f30a9b1df04918fc828670606804ac3c09"
dependencies = [
"virtue",
]
[[package]]
name = "bitflags"
version = "1.3.2"
@ -153,6 +173,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"bincode",
"chrono",
"clap",
"crossterm 0.29.0",
@ -1386,6 +1407,12 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd"
[[package]]
name = "unty"
version = "0.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae"
[[package]]
name = "utf-8"
version = "0.7.6"
@ -1416,6 +1443,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
[[package]]
name = "virtue"
version = "0.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "051eb1abcf10076295e815102942cc58f9d5e3b4560e46e53c21e8ff6f3af7b1"
[[package]]
name = "vlogger"
version = "0.1.0"

1
node/Cargo.toml
View File

@ -25,3 +25,4 @@ memory-stats = "1.2.0"
# jemallocator = "0.5.4"
textwrap = "0.16.2"
sled = "0.34.7"
bincode = { version = "2.0.1", features = ["derive", "serde"] }

12
node/src/args.rs
View File

@ -100,6 +100,18 @@ pub enum CliBlockCommand {
#[arg(short, long)]
output: String,
},
/// Display Block by Hash
#[command(name = "display", aliases = ["d"])]
#[group(required = true, multiple = false)]
Display{
/// Block Hash
#[arg(long)]
key: Option<String>,
/// Block Height
#[arg(long)]
height: Option<u64>
}
}
#[derive(Subcommand)]

13
node/src/cli.rs
View File

@ -1,5 +1,5 @@
use crate::args::*;
use crate::core::NetworkData;
use crate::core::ChainData;
use crate::node::*;
use crate::watcher::{RenderCommand, ExecutorCommand};
use clap::Parser;
@ -19,6 +19,13 @@ pub fn handle_block_command(cmd: CliBlockCommand) -> NodeCommand {
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),
(_, Some(h)) => return NodeCommand::DisplayBlockByHeight(h),
(None, None) => panic!()
}
},
}
}
@ -46,8 +53,8 @@ pub fn cli(input: &[&str]) -> ExecutorCommand {
CliCommand::Block { block_cmd } => {
ExecutorCommand::Node(handle_block_command(block_cmd))
}
CliCommand::Transaction(tx) => ExecutorCommand::Node(NodeCommand::ProcessNetworkData(
NetworkData::Transaction(tx),
CliCommand::Transaction(tx) => ExecutorCommand::Node(NodeCommand::ProcessChainData(
ChainData::Transaction(tx),
)),
CliCommand::DebugShowId => ExecutorCommand::Node(NodeCommand::ShowId),
CliCommand::StartListner { addr } => {

12
node/src/core/block.rs
View File

@ -1,6 +1,4 @@
use crate::core::NetworkData;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, Default)]
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, Default, bincode::Decode, bincode::Encode)]
pub struct BlockHeader {
pub previous_hash: String,
pub timestamp: u64,
@ -10,10 +8,10 @@ pub struct BlockHeader {
pub height: u64,
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, Default)]
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize, Default, bincode::Decode, bincode::Encode)]
pub struct Block {
pub head: BlockHeader,
pub data: Vec<NetworkData>,
pub data: Vec<String>,
}
impl BlockHeader {
@ -35,13 +33,13 @@ impl BlockHeader {
}
impl Block {
pub fn new(head: BlockHeader, data: Vec<NetworkData>) -> Self {
pub fn new(head: BlockHeader, data: Vec<String>) -> Self {
Self { head, data }
}
pub fn head(&self) -> &BlockHeader {
&self.head
}
pub fn data(&self) -> &[NetworkData] {
pub fn data(&self) -> &[String] {
&self.data
}
}

436
node/src/core/blockchain.rs
View File

@ -1,290 +1,254 @@
use sha2::Digest;
use sha2::Sha256;
use crate::core::NetworkData;
use crate::error::print_error_chain;
use crate::log;
use vlogger::*;
use std::sync::Arc;
use crate::core::ChainData;
use crate::db::database;
use crate::db;
use crate::core;
use crate::error::{BlockchainError, TxError};
use crate::db::DatabaseError;
use crate::error::TxError;
use crate::log;
use super::hasher::Hasher;
use vlogger::*;
use std::collections::HashMap;
use std::io::Write;
use std::time::UNIX_EPOCH;
use thiserror::*;
#[allow(dead_code)]
#[derive(Error, Debug)]
pub enum BlockchainError {
#[error("Blockchain initialisation failed")]
Database(#[from] DatabaseError),
#[error("invalid account creation")]
InvalidAccountCreation,
#[error("Transactional error")]
Tx(#[from] TxError),
#[error("Validation Error")]
Validation(#[from] ValidationError),
#[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")]
InvalidBlockHash,
#[error("Previous Block Hash doesn't match")]
InvalidPreviousBlockHash,
#[error("Invalid Block JSON: {0}")]
InvalidBlockJson(#[from] serde_json::Error),
#[error("Invalid Block Hash Detected at height {0}")]
InvalidBlockHash(u64),
#[error("Previous Block Hash doesn't match at height {0}")]
InvalidPreviousBlockHash(u64),
#[error("Invalid Block JSON: {0}")]
InvalidBlockJson(#[from] serde_json::Error),
}
#[allow(dead_code)]
#[derive(Debug, Default)]
#[derive(Debug)]
pub struct Blockchain {
id: String,
balances: std::collections::HashMap<Account, u32>,
mempool: Vec<NetworkData>,
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: NetworkData) -> Result<(), BlockchainError> {
self.apply(data.clone())?;
self.mempool.push(data);
Ok(())
}
pub fn add(&mut self, data: ChainData) -> Result<(), BlockchainError> {
self.apply(data.clone())?;
self.mempool.push(data);
Ok(())
}
pub fn hash_network_data(data: &NetworkData) -> String {
match data {
NetworkData::Transaction(tx) => Self::hash_transaction(tx),
}
}
pub fn hash_transaction(tx: &core::Tx) -> String {
let mut hasher = Sha256::new();
hasher.update(tx.to());
hasher.update(tx.from());
hasher.update(tx.value().to_be_bytes());
hasher.update(tx.data());
fn acc_exists(&self, acc: &Account) -> bool {
self.balances.iter().find(|(k, _)| *k == acc).is_some()
}
let res = hasher.finalize();
hex::encode(res)
}
pub fn dump_blocks(&self, path: String) {
log(msg!(DEBUG, "TODO: implement block export"));
}
pub fn calculate_next_level(level: &[String]) -> Vec<String> {
let mut next_level = Vec::new();
fn hash_transaction_pool(&self) -> Vec<String> {
self.mempool
.iter()
.map(|tx| Hasher::hash_chain_data(tx))
.collect()
}
for chunk in level.chunks(2) {
let combined_hash = if chunk.len() == 2 {
Self::hash_pair(&chunk[0], &chunk[1])
} else {
Self::hash_pair(&chunk[0], &chunk[0])
};
next_level.push(combined_hash);
}
next_level
}
pub fn calculate_merkle_root(tx: &[NetworkData]) -> String {
let tx_hashes: Vec<String> = tx
.iter()
.map(|tx| Blockchain::hash_network_data(tx))
.collect();
if tx_hashes.is_empty() {
return Blockchain::hash_data("");
}
if tx_hashes.len() == 1 {
return tx_hashes[0].clone();
}
let mut current_level = tx_hashes.to_vec();
while current_level.len() > 1 {
current_level = Self::calculate_next_level(&current_level);
}
return current_level[0].clone();
}
fn hash_pair(left: &str, right: &str) -> String {
let combined = format!("{}{}", left, right);
Self::hash_data(&combined)
}
fn hash_data(data: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(data.as_bytes());
hex::encode(hasher.finalize())
}
fn acc_exists(&self, acc: &Account) -> bool {
self.balances.iter().find(|(k, _)| *k == acc).is_some()
}
pub fn dump_blocks(&self, path: String) {
let block_json = serde_json::to_string_pretty(&self.blocks).unwrap();
if let Ok(mut db_file) = std::fs::OpenOptions::new()
.truncate(true)
.create(true)
.write(true)
.open(path)
{
db_file.write_all(&block_json.as_bytes()).unwrap();
}
}
pub fn create_block(&mut self) -> core::Block {
let previous_hash = if self.blocks().len() > 0 {
self.blocks().last().unwrap().head().block_hash()
pub fn create_block(&mut self) -> Result<Arc<core::Block>, BlockchainError> {
match self.blocks() {
Ok(blocks) => {
let previous_hash = if blocks.len() > 0 {
blocks.last().unwrap().head().block_hash()
} else {
""
""
};
let merkle_root = Self::calculate_merkle_root(&self.mempool);
let tx_hashes = self.hash_transaction_pool();
let merkle_root = Hasher::calculate_merkle_root(&tx_hashes);
let timestamp = std::time::SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
let nonce = 0;
let mut new_head = core::BlockHeader {
previous_hash: previous_hash.to_string(),
merkle_root,
timestamp,
nonce,
height: self.blocks().len() as u64 + 1,
block_hash: "".to_string(),
previous_hash: previous_hash.to_string(),
merkle_root,
timestamp,
nonce,
height: blocks.len() as u64 + 1,
block_hash: "".to_string(),
};
let mut block_hash = String::new();
while !block_hash.starts_with("0") {
new_head.nonce += 1;
block_hash = calculate_block_hash(&new_head)
new_head.nonce += 1;
block_hash = Hasher::calculate_block_hash(&new_head)
}
new_head.block_hash = block_hash;
let new_block = core::Block::new(new_head, self.mempool.clone());
self.blocks.push(new_block);
self.blocks.last().unwrap().clone()
let new_block = Arc::new(core::Block::new(new_head, tx_hashes));
self.add_block(new_block.clone())?;
Ok(new_block)
}
Err(_) => Err(BlockchainError::BlockCreation)
}
}
fn apply_transaction(&mut self, tx: &core::Tx) -> Result<(), BlockchainError> {
tx.validate()?;
return Ok(());
if let Some(from_balance) = self.balances.get_mut(tx.from()) {
if *from_balance > tx.value() {
*from_balance -= tx.value();
} else {
return Err(BlockchainError::Tx(TxError::FromInsuffitientFonds));
}
} else {
return Err(BlockchainError::Tx(TxError::UnknownAccount(
tx.from().to_string(),
)));
}
fn apply_transaction(&mut self, tx: &core::Tx) -> Result<(), BlockchainError> {
tx.validate()?;
return Ok(());
if let Some(from_balance) = self.balances.get_mut(tx.from()) {
if *from_balance > tx.value() {
*from_balance -= tx.value();
} else {
return Err(BlockchainError::Tx(TxError::FromInsuffitientFonds));
}
} else {
return Err(BlockchainError::Tx(TxError::UnknownAccount(
tx.from().to_string(),
)));
}
if let Some(to_balance) = self.balances.get_mut(&tx.to().to_string()) {
*to_balance += tx.value()
} else {
if self.acc_exists(tx.to()) {
*self.balances.get_mut(tx.to()).unwrap() += tx.value();
} else {
self.balances.insert(tx.to().clone(), tx.value());
}
}
Ok(())
if let Some(to_balance) = self.balances.get_mut(&tx.to().to_string()) {
*to_balance += tx.value()
} else {
if self.acc_exists(tx.to()) {
*self.balances.get_mut(tx.to()).unwrap() += tx.value();
} else {
self.balances.insert(tx.to().clone(), tx.value());
}
}
Ok(())
}
pub fn apply(&mut self, data: NetworkData) -> Result<(), BlockchainError> {
match &data {
NetworkData::Transaction(tx) => {
self.apply_transaction(tx)?;
self.mempool.push(data);
}
}
Ok(())
pub fn apply(&mut self, data: ChainData) -> Result<(), BlockchainError> {
match &data {
ChainData::Transaction(tx) => {
self.apply_transaction(tx)?;
self.mempool.push(data);
}
}
pub fn new(balances: HashMap<Account, u32>, mempool: Vec<NetworkData>) -> Blockchain {
return Self {
id: BLOCKCHAIN_ID.to_string(),
balances,
mempool,
};
}
}
pub fn calculate_block_hash(head: &core::BlockHeader) -> String {
let mut hasher = sha2::Sha256::new();
hasher.update(head.nonce().to_be_bytes());
hasher.update(head.previous_hash());
hasher.update(head.timestamp().to_be_bytes());
hasher.update(head.merkle_root());
let res = hasher.finalize();
hex::encode(res)
Ok(())
}
}
impl Blockchain {
pub fn list_blocks(&self) -> String {
let mut ret = String::from("Blocks List\n-------------------\n");
for (i, b) in self.blocks.iter().enumerate() {
ret.push_str(format!("Block Hash #{i}: {}\n", b.head.block_hash()).as_str())
pub fn list_blocks(&self) -> Result<String, BlockchainError> {
let mut ret = String::from("Blocks List\n-------------------\n");
let blocks = self.blocks()?;
for (i, b) in blocks.iter().enumerate() {
ret.push_str(format!("Block Hash #{i}: {}\n", b.head.block_hash()).as_str())
}
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()?)
}
fn insert_block(&self, block: &core::Block) -> Result<(), BlockchainError> {
self.db.add_block(block)?;
Ok(())
}
pub fn add_block(&mut self, block: Arc<core::Block>) -> Result<(), BlockchainError>{
match self.validate_block(&block) {
Ok(()) => Ok(self.insert_block(&block)?),
Err(e) => Err(BlockchainError::Validation(e)),
}
}
pub fn display_block_by_key(&self, key: String) {
if let Ok(Some(b)) = self.db.get_block_by_key(&key) {
log(format!("{:#?}", b))
}
}
pub fn display_block_by_height(&self, height: u64) {
if let Ok(Some(b)) = self.db.get_block_by_height(height) {
log(format!("{:#?}", b))
}
}
fn validate_block(&self, block: &core::Block) -> Result<(), ValidationError> {
let head = block.head();
let hash = Hasher::calculate_block_hash(block.head());
if hash != head.block_hash() {
return Err(ValidationError::InvalidBlockHash(head.height));
}
if let Ok(blocks) = self.blocks() {
if let Some(last_block) = blocks.last() {
if head.previous_hash() != last_block.head().block_hash() {
return Err(ValidationError::InvalidPreviousBlockHash(head.height));
}
ret
}
}
Ok(())
}
pub fn get_balances(&self) -> &std::collections::HashMap<String, u32> {
&self.balances
}
fn validate_chain(&self) -> Result<(), ValidationError> {
if let Ok(blocks) = self.blocks() {
if let Some(mut prev_block) = blocks.first() {
for (i, block) in blocks.iter().enumerate() {
let head = block.head();
let hash = Hasher::calculate_block_hash(block.head());
pub fn blocks(&self) -> &[core::Block] {
&self.blocks
}
pub fn add_block(&mut self, block: core::Block) {
match self.validate_block(&block) {
Ok(()) => self.blocks.push(block),
Err(e) => match e {
ValidationError::InvalidBlockHash => log(msg!(ERROR, "Invalid Block Hash")),
ValidationError::InvalidPreviousBlockHash => {
log(msg!(ERROR, "Invalid Previos Block Hash"))
}
ValidationError::InvalidBlockJson(e) => print_error_chain(e.into()),
},
if hash != head.block_hash() {
return Err(ValidationError::InvalidBlockHash(i as u64));
} else if head.previous_hash() != prev_block.head().block_hash() && blocks.len() > 1 {
return Err(ValidationError::InvalidPreviousBlockHash(i as u64));
}
prev_block = block;
}
}
}
Ok(())
}
fn validate_block(&self, block: &core::Block) -> Result<(), ValidationError> {
let head = block.head();
let hash = calculate_block_hash(block.head());
if hash != head.block_hash() {
return Err(ValidationError::InvalidBlockHash);
}
if let Some(prev_block) = self.blocks().last() {
if head.previous_hash() != prev_block.head().block_hash() {
return Err(ValidationError::InvalidPreviousBlockHash);
}
}
Ok(())
}
pub fn build(path: Option<String>) -> Result<Blockchain, BlockchainError> {
let db = db::ChainDb::new(path).or_else(|e| Err(BlockchainError::Database(e)))?;
fn validate_chain(&self) -> Result<(), ValidationError> {
let blocks = self.blocks();
for block in blocks {
let head = block.head();
let hash = calculate_block_hash(block.head());
if hash != head.block_hash() {
return Err(ValidationError::InvalidBlockHash);
}
}
Ok(())
}
pub fn build(blocks: &str) -> Result<Blockchain, ValidationError> {
match serde_json::from_str::<Vec<core::Block>>(&blocks) {
Ok(blocks) => {
let chain = Blockchain {
blocks,
balances: HashMap::new(),
mempool: vec![],
id: BLOCKCHAIN_ID.to_string(),
};
chain.validate_chain()?;
Ok(chain)
}
Err(e) => Err(ValidationError::InvalidBlockJson(e)),
}
}
let chain = Blockchain {
balances: HashMap::new(),
mempool: vec![],
id: BLOCKCHAIN_ID.to_string(),
db
};
chain.validate_chain().or_else(|e| return Err(BlockchainError::Validation(e)))?;
Ok(chain)
}
}

6
node/src/core/data.rs
View File

@ -1,6 +1,8 @@
use bincode::{Decode, Encode};
use super::Tx;
#[derive(serde::Deserialize, serde::Serialize, Debug, Clone)]
pub enum NetworkData {
#[derive(serde::Deserialize, serde::Serialize, Encode, Decode, Debug, Clone)]
pub enum ChainData {
Transaction(Tx),
}

77
node/src/core/hasher.rs Normal file
View File

@ -0,0 +1,77 @@
use sha2::Sha256;
use sha2::Digest;
use super::{ChainData, BlockHeader};
pub struct Hasher {}
impl Hasher {
pub fn hash_chain_data(data: &ChainData) -> String {
let mut hasher = Sha256::new();
match data {
ChainData::Transaction(tx) => {
hasher.update(tx.to());
hasher.update(tx.from());
hasher.update(tx.value().to_be_bytes());
hasher.update(tx.data());
}
}
let res = hasher.finalize();
hex::encode(res)
}
pub fn calculate_next_level(level: &[String]) -> Vec<String> {
let mut next_level = Vec::new();
for chunk in level.chunks(2) {
let combined_hash = if chunk.len() == 2 {
Self::hash_pair(&chunk[0], &chunk[1])
} else {
Self::hash_pair(&chunk[0], &chunk[0])
};
next_level.push(combined_hash);
}
next_level
}
pub fn calculate_merkle_root(tx_hashes: &[String]) -> String {
if tx_hashes.is_empty() {
return Self::hash_data("");
}
if tx_hashes.len() == 1 {
return tx_hashes[0].clone();
}
let mut current_level = tx_hashes.to_vec();
while current_level.len() > 1 {
current_level = Self::calculate_next_level(&current_level);
}
return current_level[0].clone();
}
fn hash_pair(left: &str, right: &str) -> String {
let combined = format!("{}{}", left, right);
Self::hash_data(&combined)
}
fn hash_data(data: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(data.as_bytes());
hex::encode(hasher.finalize())
}
pub fn calculate_block_hash(head: &BlockHeader) -> String {
let mut hasher = sha2::Sha256::new();
hasher.update(head.nonce().to_be_bytes());
hasher.update(head.previous_hash());
hasher.update(head.timestamp().to_be_bytes());
hasher.update(head.merkle_root());
let res = hasher.finalize();
hex::encode(res)
}
}

2
node/src/core/tx.rs
View File

@ -1,7 +1,7 @@
use crate::core::Account;
use crate::error::TxError;
#[derive(serde::Deserialize, serde::Serialize, Debug, clap::Args, Clone)]
#[derive(serde::Deserialize, serde::Serialize, Debug, clap::Args, Clone, bincode::Encode, bincode::Decode)]
pub struct Tx {
from: Account,
to: Account,

139
node/src/db/database.rs
View File

@ -1,20 +1,139 @@
use sled;
use crate::log;
use bincode::{self, config::Configuration};
use sled::{self, Batch};
use crate::{core::{self, Block, ChainData, Hasher}, db::error::DatabaseError, error::print_error_chain, log};
use vlogger::*;
use std::sync::Arc;
static BINCODE_CONFIG: Configuration = bincode::config::standard();
const DB_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/database");
fn db_init() {
match sled::open(DB_PATH) {
Ok(db) => {
if db.was_recovered() {
log(msg!(INFO, "Loaded Database from Previous state at: {}", DB_PATH));
} else {
log(msg!(INFO, "Created Database at {}", DB_PATH));
const DB_TREE: &str = "blocks";
const BLOCK_INDEX: &str = "blocks:";
const CHAIN_DATA_INDEX: &str = "chain_data:";
const DATA_TO_BLOCK_INDEX: &str = "data_to_block:";
const METADATA_INDEX: &str= "metadata:";
const TIP_KEY: &str = "chain_tip";
const HEIGHT_KEY: &str = "chain_height";
#[derive(Debug)]
pub struct ChainDb {
db: sled::Tree,
}
fn data_index(key: &str) -> String {
format!("{}{}", CHAIN_DATA_INDEX, key)
}
fn data_to_block_index(key: &str) -> String {
format!("{}{}", DATA_TO_BLOCK_INDEX, key)
}
fn block_index(key: &str) -> String {
format!("{}{}", BLOCK_INDEX, key)
}
fn metadata_index(key: &str) -> String {
format!("{}{}", METADATA_INDEX, key)
}
impl ChainDb {
pub fn new(path: Option<String>) ->Result<ChainDb, DatabaseError> {
let path = if path.is_some() {
&path.unwrap()
} else {
DB_PATH
};
match sled::open(&path) {
Ok(db) => {
if db.was_recovered() {
log(msg!(INFO, "Loaded Database from Previous state at: {}", path));
} else {
log(msg!(INFO, "Created Database at {}", path));
}
let db = db
.open_tree(DB_TREE)?;
Ok(ChainDb {
db
})
}
Err(err) => {
print_error_chain(&err.clone().into());
Err(DatabaseError::Init(err.into()))
}
}
Err(e) => {
}
pub fn get_block_by_key(&self, block_hash: &str) -> Result<Option<core::Block>, DatabaseError> {
let block_hash = block_index(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))
} else {
Ok(None)
}
}
pub fn get_block_by_height(&self, height: u64) -> Result<Option<Arc<core::Block>>, DatabaseError> {
for result in self.db.scan_prefix(BLOCK_INDEX) {
let (_key, value) = result?;
let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&value, BINCODE_CONFIG)?;
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);
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);
} else {
return Err(DatabaseError::MissingData(data_hash.clone()));
}
}
Ok(Arc::new(chain_data))
}
pub fn get_all_blocks(&self) -> Result<Vec<Arc<core::Block>>, DatabaseError> {
self.db.scan_prefix(BLOCK_INDEX)
.map(|res| -> Result<Arc<core::Block>, DatabaseError> {
let (_key, value) = res?;
let (block, _size) = bincode::decode_from_slice::<core::Block, _>(&value, BINCODE_CONFIG)
.map_err(|e| DatabaseError::Decode(e))?;
Ok(Arc::new(block))
})
.collect()
}
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));
self.db.insert(data_hash, bin_data)?;
Ok(())
}
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);
for data in block.data() {
db_batch.insert(data_to_block_index(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_index(HEIGHT_KEY).as_str(), &block.head().height.to_be_bytes());
self.db.apply_batch(db_batch)?;
Ok(())
}
}

22
node/src/db/error.rs Normal file
View File

@ -0,0 +1,22 @@
use thiserror::Error;
#[derive(Debug, Error)]
pub enum DatabaseError {
#[error("Database initialization failed")]
Init(#[from] std::io::Error),
#[error("Database read failed")]
Read(#[from] anyhow::Error),
#[error("Sled Failed")]
Sled(#[from] sled::Error),
#[error("Database Encode failed")]
Encode(#[from] bincode::error::EncodeError),
#[error("Database Decode failed")]
Decode(#[from] bincode::error::DecodeError),
#[error("Missing chain data for hash: {0}")]
MissingData(String)
}

67
node/src/error.rs
View File

@ -1,58 +1,49 @@
use crate::log;
use thiserror::Error;
#[allow(dead_code)]
#[derive(Error, Debug)]
pub enum BlockchainError {
#[error("invalid account creation")]
InvalidAccountCreation,
#[error("Transactional error")]
Tx(#[from] TxError),
}
#[allow(dead_code)]
#[derive(Error, Debug)]
pub enum TxError {
#[error("from field is empty")]
FromEmpty,
#[error("to field is empty")]
ToEmpty,
#[error("insuffitient fonds")]
FromInsuffitientFonds,
#[error("0 value transaction")]
ValueEmpty,
#[error("account {0} not found in database")]
UnknownAccount(String),
#[error("from field is empty")]
FromEmpty,
#[error("to field is empty")]
ToEmpty,
#[error("insuffitient fonds")]
FromInsuffitientFonds,
#[error("0 value transaction")]
ValueEmpty,
#[error("account {0} not found in database")]
UnknownAccount(String),
}
pub fn print_error_chain(err: anyhow::Error) {
let mut err_string = String::from(format!("Error: {}\n", err));
pub fn print_error_chain(err: &anyhow::Error) {
let mut err_string = String::from(format!("Error: {}\n", err));
let mut source = err.source();
let mut level = 1;
let mut source = err.source();
let mut level = 1;
while let Some(err) = source {
err_string.push_str(format!(" {}: {}\n", level, err).as_str());
source = err.source();
level += 1;
}
log(err_string)
while let Some(err) = source {
err_string.push_str(format!(" {}: {}\n", level, err).as_str());
source = err.source();
level += 1;
}
log(err_string)
}
#[derive(Debug, Error)]
pub enum SystemError<T>
where
T: std::fmt::Debug,
T: std::fmt::Debug,
{
#[error("TODO")]
TODO,
#[error("TODO")]
TODO,
#[error("Failed to send message: {message:?}")]
ChannelSendError { message: T },
#[error("Failed to send message: {message:?}")]
ChannelSendError { message: T },
#[error("Channel {message:?} was closed")]
ChannelClosed { message: T },
#[error("Channel {message:?} was closed")]
ChannelClosed { message: T },
#[error("Event Bus closed")]
EventBusClosed(#[from] tokio::sync::broadcast::error::RecvError),
#[error("Event Bus closed")]
EventBusClosed(#[from] tokio::sync::broadcast::error::RecvError),
}

7
node/src/lib.rs
View File

@ -14,6 +14,10 @@ pub mod error;
pub mod db {
pub mod database;
pub mod error;
pub use database::*;
pub use error::*;
}
pub mod bus {
@ -64,6 +68,9 @@ pub mod core {
pub mod data;
pub use data::*;
pub mod hasher;
pub use hasher::*;
}
pub mod seeds_constants;

1
node/src/main.rs
View File

@ -13,7 +13,6 @@ async fn main() -> Result<(), std::io::Error> {
let args = args::CliArgs::parse();
let mut watcher = Watcher::build()
.file(args.seed_file)
.addr(args.addr)
.seed(args.seed)
.debug(args.debug)

15
node/src/node/error.rs
View File

@ -1,12 +1,7 @@
#[derive(Debug, Clone)]
pub struct NetworkError {
pub message: String,
}
use thiserror::Error;
impl std::fmt::Display for NetworkError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.message)
}
#[derive(Debug, Clone, Error)]
pub enum NetworkError {
#[error("Implement NetworkError Enum: ({})", file!())]
TODO
}
impl std::error::Error for NetworkError {}

104
node/src/node/node.rs
View File

@ -1,19 +1,20 @@
use crate::core::{self, Blockchain, NetworkData, ValidationError};
use crate::core::{self, Blockchain, BlockchainError, ChainData, ValidationError};
use crate::error::print_error_chain;
use crate::bus::{SystemEvent, publish_system_event};
use crate::protocol::ProtocolMessage;
use crate::protocol::{Connector, ConnectorCommand};
use crate::seeds_constants::SEED_NODES;
use crate::watcher::executor::ExecutorCommand;
use crate::log;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::mpsc;
use uuid::Uuid;
use vlogger::*;
use crate::log;
use thiserror::*;
#[derive(Debug, Clone)]
pub struct TcpPeer {
@ -38,14 +39,17 @@ pub struct Node {
pub id: Uuid,
pub addr: Option<SocketAddr>,
pub tcp_peers: HashMap<Uuid, TcpPeer>,
chain: Blockchain,
listner_handle: Option<tokio::task::JoinHandle<()>>,
exec_tx: mpsc::Sender<ExecutorCommand>,
rx: mpsc::Receiver<NodeCommand>,
tx: mpsc::Sender<NodeCommand>,
}
#[derive(Debug, Error)]
pub enum NodeError {
#[error("Block chain error")]
ChainError(#[from] BlockchainError)
}
#[derive(Debug, Clone)]
@ -58,11 +62,13 @@ pub enum NodeCommand {
peer_id: Uuid,
message: ProtocolMessage,
},
ProcessNetworkData(NetworkData),
ProcessChainData(ChainData),
StartListner(SocketAddr),
PingAddr(String),
PingId(String),
CreateBlock,
DisplayBlockByKey(String),
DisplayBlockByHeight(u64),
ListBlocks,
ListPeers,
ShowId,
@ -112,6 +118,7 @@ impl Node {
id: uuid::Uuid,
exec_tx: mpsc::Sender<ExecutorCommand>,
addr: Option<SocketAddr>,
chain: Blockchain
) -> Self {
let (tx, rx) = mpsc::channel::<NodeCommand>(100);
Self {
@ -119,6 +126,7 @@ impl Node {
tcp_peers: HashMap::new(),
addr,
exec_tx,
chain,
listner_handle: None,
tcp_connector: None,
tx,
@ -126,16 +134,12 @@ impl Node {
}
}
pub async fn new(
pub fn new(
addr: Option<SocketAddr>,
blocks_json: &str,
exec_tx: mpsc::Sender<ExecutorCommand>,
chain: Blockchain,
) -> Self {
let (tx, rx) = mpsc::channel::<NodeCommand>(100);
let chain = Blockchain::build(blocks_json).unwrap_or_else(|e| {
print_error_chain(e.into());
Default::default()
});
Self {
id: Uuid::new_v4(),
tcp_peers: HashMap::new(),
@ -143,13 +147,14 @@ impl Node {
exec_tx,
listner_handle: None,
tcp_connector: None,
chain,
tx,
rx,
}
}
async fn get_blocks(&mut self, ) -> Result<Vec<core::Block>, NodeError> {
Ok(vec![])
fn get_blocks(&self) -> Result<Vec<Arc<core::Block>>, NodeError> {
Ok(self.chain.blocks()?)
}
pub async fn process_message(&mut self, peer_id: uuid::Uuid, message: ProtocolMessage) {
@ -157,24 +162,36 @@ impl Node {
ProtocolMessage::BootstrapRequest { .. } => {
log(msg!(DEBUG, "Received BootstrapRequest from {peer_id}"));
let peer = &self.tcp_peers[&peer_id];
let blocks = self.get_blocks();
let resp = ProtocolMessage::BootstrapResponse {
blocks: serde_json::to_string(&self.chain.blocks().to_vec()).unwrap_or_else(
|e| {
log(msg!(
WARNING,
"Failed to serde Chain for BootstrapResponse: {e}"
));
Default::default()
},
),
blocks: {
if let Ok(blocks) = self.get_blocks() {
serde_json::to_string(&blocks
.iter()
.map(|f| (**f).clone())
.collect::<Vec<core::Block>>()
).map_err(
|e| {
log(msg!(
ERROR,
"Failed to serde Chain for BootstrapResponse: {e}"
));
e
},
).ok()
} else {
None
}
}
};
peer.sender.send(resp).await.unwrap();
log(msg!(DEBUG, "Send BootstrapResponse to {peer_id}"));
}
ProtocolMessage::BootstrapResponse { blocks } => {
log(msg!(DEBUG, "Received BootstrapResponse from seed"));
self.chain = core::Blockchain::build(&blocks).unwrap();
self.chain = core::Blockchain::build(blocks).unwrap();
}
ProtocolMessage::Pong { peer_id } => {
log(msg!(DEBUG, "Received Pong from {peer_id}"));
}
ProtocolMessage::Ping { peer_id } => {
log(msg!(DEBUG, "Received Ping from {peer_id}"));
@ -195,10 +212,12 @@ impl Node {
}
ProtocolMessage::Block { block, .. } => {
log(msg!(DEBUG, "Received Block from {peer_id}"));
self.chain.add_block(block.clone())
if let Err(_e) = self.chain.add_block(block.into()) {
log(msg!(DEBUG, "TODO: implement error handling in {}:{}", file!(), line!()));
}
}
ProtocolMessage::NetworkData { data, .. } => {
log(msg!(DEBUG, "Received NetworkData from {peer_id}"));
ProtocolMessage::ChainData { data, .. } => {
log(msg!(DEBUG, "Received ChainData from {peer_id}"));
self.chain.apply(data).unwrap()
}
_ => {
@ -254,9 +273,9 @@ impl Node {
Ok(())
}
async fn broadcast_network_data(&self, data: NetworkData) {
async fn broadcast_network_data(&self, data: ChainData) {
for (id, peer) in &self.tcp_peers {
let message = ProtocolMessage::NetworkData {
let message = ProtocolMessage::ChainData {
peer_id: self.id,
data: data.clone(),
};
@ -285,10 +304,10 @@ impl Node {
return self.exec_tx.clone();
}
async fn network_data(&mut self, data: NetworkData) {
async fn network_data(&mut self, data: ChainData) {
match self.chain.apply(data) {
Ok(_) => log(msg!(DEBUG, "NetworkData Applied")),
Err(e) => print_error_chain(e.into()),
Ok(_) => log(msg!(DEBUG, "ChainData Applied")),
Err(e) => print_error_chain(&e.into()),
};
}
@ -379,18 +398,29 @@ impl Node {
NodeCommand::ProcessMessage { peer_id, message } => {
self.process_message(peer_id, message).await;
}
NodeCommand::ProcessNetworkData(data) => {
NodeCommand::ProcessChainData(data) => {
self.network_data(data.clone()).await;
self.broadcast_network_data(data).await;
}
NodeCommand::CreateBlock => {
log(msg!(DEBUG, "Received CreateBlock Command"));
let block = self.chain.create_block();
self.broadcast_block(&block).await;
if let Ok(block) = self.chain.create_block() {
log(msg!(INFO, "Created Block with hash {}", block.head().block_hash()));
self.broadcast_block(&block).await;
}
}
NodeCommand::DisplayBlockByKey(key) => {
self.chain.display_block_by_key(key)
}
NodeCommand::DisplayBlockByHeight(height) => {
self.chain.display_block_by_height(height)
}
NodeCommand::ListBlocks => {
log(msg!(DEBUG, "Received DebugListBlocks command"));
log(self.chain.list_blocks());
match self.chain.list_blocks() {
Ok(s) => log(s),
Err(e) => print_error_chain(&e.into()),
}
}
NodeCommand::ListPeers => {
log(msg!(DEBUG, "Received DebugListPeers command"));

18
node/src/protocol/connection.rs
View File

@ -3,6 +3,7 @@ use crate::protocol::ProtocolMessage;
use crate::watcher::ExecutorCommand;
use tokio::net;
use tokio::sync::mpsc;
use crate::log;
use super::Connector;
@ -35,14 +36,9 @@ impl Connection {
}
}
async fn log(&self, msg: String) {
let _ = self.exec_tx.send(ExecutorCommand::Print(msg)).await;
}
pub async fn start(mut self) {
tokio::spawn(async move {
self.log(msg!(DEBUG, "Started Message Handler for {}", self.peer_id))
.await;
log(msg!(DEBUG, "Started Message Handler for {}", self.peer_id));
loop {
tokio::select! {
@ -50,12 +46,12 @@ impl Connection {
match response_result {
Some(response) => {
if let Err(e) = Connector::send_message(&mut self.stream, &response).await {
self.log(msg!(ERROR, "Failed to send response to {}: {}", self.peer_id, e)).await;
log(msg!(ERROR, "Failed to send response to {}: {}", self.peer_id, e));
break;
}
},
None => {
self.log(msg!(DEBUG, "Response channel closed for {}", self.peer_id)).await;
log(msg!(DEBUG, "Response channel closed for {}", self.peer_id));
break;
}
}
@ -64,7 +60,7 @@ impl Connection {
message_result = Connector::receive_message(&mut self.stream) => {
match message_result {
Ok(message) => {
self.log(msg!(DEBUG, "Received Message from {}", self.peer_id)).await;
log(msg!(DEBUG, "Received Message from {}", self.peer_id));
let command = ExecutorCommand::Node(node::NodeCommand::ProcessMessage {
peer_id: self.peer_id,
@ -72,12 +68,12 @@ impl Connection {
});
if self.exec_tx.send(command).await.is_err() {
self.log(msg!(ERROR, "Failed to send command to main thread from {}", self.peer_id)).await;
log(msg!(ERROR, "Failed to send command to main thread from {}", self.peer_id));
break;
}
},
Err(e) => {
self.log(msg!(WARNING, "Connection to {} closed: {}", self.peer_id, e.message)).await;
log(msg!(WARNING, "Connection to {} closed: {}", self.peer_id, e));
let cmd = ExecutorCommand::Node(node::NodeCommand::RemovePeer {
peer_id: self.peer_id
});

64
node/src/protocol/connector.rs
View File

@ -10,7 +10,7 @@ use crate::log;
use super::Connection;
use crate::bus::*;
use crate::node::error;
use crate::node::{error, NetworkError};
use crate::node::node;
use crate::protocol::ProtocolMessage;
use crate::watcher::ExecutorCommand;
@ -120,7 +120,7 @@ impl Connector {
Ok(stream) => self.establish_connection_to_seed(stream, addr).await,
Err(e) => {
// let err = ConnectorError::ConnectionError(e.into());
print_error_chain(e.into());
print_error_chain(&e.into());
}
}
}
@ -130,7 +130,7 @@ impl Connector {
Ok(stream) => self.establish_connection_outbound(stream, addr).await,
Err(e) => {
let err = ConnectorError::ConnectionError(e.into());
print_error_chain(err.into());
print_error_chain(&err.into());
}
}
}
@ -168,7 +168,7 @@ impl Connector {
.await;
}
}
Err(e) => print_error_chain(e.into()),
Err(e) => print_error_chain(&e.into()),
}
}
@ -204,7 +204,7 @@ impl Connector {
.await;
}
}
Err(e) => print_error_chain(e.into()),
Err(e) => print_error_chain(&e.into()),
}
}
@ -223,7 +223,7 @@ impl Connector {
};
match Connector::send_message(&mut stream, &ack).await {
Ok(()) => node::TcpPeer::new(peer_id, addr, ch_tx),
Err(e) => return print_error_chain(e.into()),
Err(e) => return print_error_chain(&e.into()),
}
}
_ => {
@ -245,9 +245,10 @@ impl Connector {
pub async fn send_message(
stream: &mut net::TcpStream,
message: &ProtocolMessage,
) -> Result<(), error::NetworkError> {
let json = serde_json::to_string(message).map_err(|e| error::NetworkError {
message: format!("Failed to serialize: {}", e),
) -> Result<(), NetworkError> {
let json = serde_json::to_string(message)
.map_err(|_e| {
NetworkError::TODO
})?;
let data = json.as_bytes();
@ -255,19 +256,20 @@ impl Connector {
stream
.write_all(&len.to_be_bytes())
.await
.map_err(|e| error::NetworkError {
message: format!("Failed to write data: {}", e),
.map_err(|_e| {
NetworkError::TODO
})?;
stream
.write_all(data)
.await
.map_err(|e| error::NetworkError {
message: format!("Failed to write data: {}", e),
.map_err(|_e| {
NetworkError::TODO
})?;
stream.flush().await
.map_err(|_e| {
NetworkError::TODO
})?;
stream.flush().await.map_err(|e| error::NetworkError {
message: format!("Failed to flush stream: {}", e),
})?;
Ok(())
}
@ -278,37 +280,33 @@ impl Connector {
stream
.read_exact(&mut len_bytes)
.await
.map_err(|e| error::NetworkError {
message: format!("Failed to read length: {}", e),
.map_err(|_e| {
NetworkError::TODO
})?;
let len = u32::from_be_bytes(len_bytes) as usize;
if len >= super::message::MAX_MESSAGE_SIZE {
return Err(error::NetworkError {
message: format!(
"NetworkError: Inbound Message too large: max = {}, got = {len}",
super::message::MAX_MESSAGE_SIZE
),
});
return Err(NetworkError::TODO);
}
let mut data = vec![0u8; len];
stream
.read_exact(&mut data)
.await
.map_err(|e| error::NetworkError {
message: format!("Failed to read data: {}", e),
.map_err(|_e| {
NetworkError::TODO
})?;
let json = String::from_utf8(data).map_err(|e| error::NetworkError {
message: format!("Invalid UTF-8: {}", e),
})?;
let json = String::from_utf8(data)
.map_err(|_e| {
NetworkError::TODO
})?;
let message: ProtocolMessage =
serde_json::from_str(&json).map_err(|e| error::NetworkError {
message: format!("JSON parse error: {}", e),
})?;
let message: ProtocolMessage = serde_json::from_str(&json)
.map_err(|_e| {
NetworkError::TODO
})?;
Ok(message)
}

14
node/src/protocol/message.rs
View File

@ -1,4 +1,4 @@
use crate::core::{self, NetworkData};
use crate::core::{self, ChainData};
use std::fmt;
use std::net::SocketAddr;
@ -11,7 +11,7 @@ pub enum ProtocolMessage {
version: String,
},
BootstrapResponse {
blocks: String,
blocks: Option<String>,
},
GetPeersRequest {
peer_id: uuid::Uuid,
@ -32,9 +32,9 @@ pub enum ProtocolMessage {
height: u64,
block: core::Block,
},
NetworkData {
ChainData {
peer_id: uuid::Uuid,
data: NetworkData,
data: ChainData,
},
Ping {
peer_id: uuid::Uuid,
@ -54,7 +54,7 @@ impl fmt::Display for ProtocolMessage {
write!(f, "BootstrapRequest from {} (v{})", peer_id, version)
}
ProtocolMessage::BootstrapResponse { blocks } => {
write!(f, "BootstrapResponse with {} blocks", blocks.len())
write!(f, "BootstrapResponse with {:?} blocks", blocks.clone().unwrap_or_default().len())
}
ProtocolMessage::GetPeersRequest { peer_id } => {
write!(f, "GetPeersRequest from {}", peer_id)
@ -75,8 +75,8 @@ impl fmt::Display for ProtocolMessage {
} => {
write!(f, "Block #{} from {}", height, peer_id)
}
ProtocolMessage::NetworkData { peer_id, data: _ } => {
write!(f, "NetworkData from {}", peer_id)
ProtocolMessage::ChainData { peer_id, data: _ } => {
write!(f, "ChainData from {}", peer_id)
}
ProtocolMessage::Ping { peer_id } => {
write!(f, "Ping from {}", peer_id)

22
node/src/watcher/watcher.rs
View File

@ -8,9 +8,7 @@ use std::{
use tokio::sync::mpsc;
use crate::{
error::print_error_chain,
bus::{NetworkEvent, SystemEvent, subscribe_system_event},
node::node::{Node, NodeCommand},
bus::{subscribe_system_event, NetworkEvent, SystemEvent}, core, node::node::{Node, NodeCommand}
};
use vlogger::*;
@ -248,26 +246,12 @@ impl WatcherBuilder {
});
log(msg!(DEBUG, "Database Location: {:?}", self.database));
let db = self
.database
.as_ref()
.and_then(|path| {
log(msg!(INFO, "Reading chain data from {path}"));
match std::fs::read_to_string(path) {
Ok(s) => Some(s),
Err(e) => {
print_error_chain(e.into());
None
}
}
})
.unwrap_or_default();
if self.seed {
self.addr = Some(crate::seeds_constants::SEED_NODES[0]);
}
let mut node = Node::new(self.addr.clone(), &blocks, exec_tx.clone()).await;
let chain = core::Blockchain::build(None).unwrap();
let mut node = Node::new(self.addr.clone(), exec_tx.clone(), chain);
log(msg!(INFO, "Built Node"));
let executor_handle = tokio::spawn({