snap_coin/node/
node.rs

1use num_bigint::BigUint;
2use tokio::net::TcpStream;
3use std::io::Write;
4use std::{
5    fs,
6    net::SocketAddr,
7    sync::{Arc, OnceLock},
8};
9use thiserror::Error;
10use tokio::{
11    sync::RwLock,
12    task::{JoinError, JoinHandle},
13};
14
15use crate::crypto::Hash;
16use crate::node::auto_peer::auto_peer;
17use crate::node::server::ServerError;
18use crate::{
19    core::{
20        block::Block,
21        blockchain::{
22            Blockchain, BlockchainError, validate_block_timestamp, validate_transaction_timestamp,
23        },
24        transaction::Transaction,
25    },
26    node::{
27        mempool::MemPool,
28        message::{Command, Message},
29        peer::{Peer, PeerError},
30        server::Server,
31    },
32};
33
34/// Path at which this blockchain is being read and written from and too. This enforces a singleton rule where only one node can exist in one program at once
35static NODE_PATH: OnceLock<String> = OnceLock::new();
36
37#[derive(Error, Debug)]
38pub enum NodeError {
39    #[error("{0}")]
40    PeerError(#[from] PeerError),
41
42    #[error("TCP error: {0}")]
43    IOError(#[from] std::io::Error),
44
45    #[error("Join error: {0}")]
46    JoinError(#[from] JoinError),
47
48    #[error("Server error: {0}")]
49    ServerError(#[from] super::server::ServerError),
50}
51
52/// Handles incoming connections and outbound peers
53pub struct Node {
54    pub peers: Vec<Arc<RwLock<Peer>>>,
55    pub blockchain: Blockchain,
56    pub mempool: MemPool,
57    pub last_seen_block: Hash,
58
59    // Synchronization flag
60    pub is_syncing: bool,
61
62    pub target_peers: usize,
63
64    pub port: u16,
65}
66
67impl Node {
68    /// Create a new blockchain (load / create) with default 12 nodes target
69    /// WARNING: Only one instance of this struct can exist in one program
70    pub fn new(node_path: &str, port: u16) -> Arc<RwLock<Self>> {
71        NODE_PATH
72            .set(String::from(node_path))
73            .expect("Only one node can exist at once!");
74        // Clear log file
75        if !fs::exists(node_path).expect("failed to check if blockchain dir exists") {
76            fs::create_dir(node_path).expect("Could not create blockchain directory");
77        }
78        fs::OpenOptions::new()
79            .write(true)
80            .truncate(true)
81            .create(true)
82            .open(
83                NODE_PATH
84                    .get()
85                    .expect("One blockchain instance must exist before logging")
86                    .to_owned()
87                    + "/info.log",
88            )
89            .expect("Could not open logging file!");
90        Arc::new(RwLock::new(Node {
91            peers: vec![],
92            blockchain: Blockchain::new(node_path),
93            mempool: MemPool::new(),
94            is_syncing: false,
95            target_peers: 12,
96            port,
97            last_seen_block: Hash::new_from_buf([0u8; 32]),
98        }))
99    }
100
101    /// Connect to a specified peer
102    pub async fn connect_peer(
103        node: Arc<RwLock<Node>>,
104        address: SocketAddr,
105    ) -> Result<(Arc<RwLock<Peer>>, JoinHandle<Result<(), PeerError>>), NodeError> {
106        let peer: Arc<RwLock<Peer>> = Arc::new(RwLock::new(Peer::new(address, false)));
107        let stream = TcpStream::connect(address).await?;
108
109        let handle = Peer::connect(peer.clone(), node, stream).await;
110
111        Ok((peer, handle))
112    }
113
114    /// Initialize this node, with a array of seed nodes which this node will use to connect to
115    /// Starts all handlers
116    /// WARNING: Can only be called once
117    pub async fn init(
118        node: Arc<RwLock<Node>>,
119        seed_nodes: Vec<SocketAddr>,
120    ) -> Result<JoinHandle<Result<(), ServerError>>, NodeError> {
121        let mut peers = Vec::new();
122
123        for addr in seed_nodes {
124            let (peer, _) = Self::connect_peer(node.clone(), addr).await?;
125            peers.push(peer);
126        }
127
128        node.write().await.peers = peers;
129
130        let server_handle: JoinHandle<Result<(), super::server::ServerError>> =
131            Server.init(node.clone(), node.read().await.port).await;
132
133        let node = node.clone();
134        auto_peer(node.clone());
135
136        Ok(server_handle)
137    }
138
139    /// Send some message to all peers
140    pub async fn send_to_peers(node: Arc<RwLock<Node>>, message: Message) {
141        for i_peer in &node.read().await.peers {
142            Peer::send(Arc::clone(i_peer), message.clone()).await;
143        }
144    }
145
146    /// Submit a new block to the network
147    pub async fn submit_block(
148        node: Arc<RwLock<Node>>,
149        new_block: Block,
150    ) -> Result<(), BlockchainError> {
151        node.write().await.blockchain.add_block(new_block.clone())?;
152        validate_block_timestamp(&new_block)?;
153
154        // Remove transactions from mempool
155        node.write()
156            .await
157            .mempool
158            .spend_transactions(
159                new_block
160                    .transactions
161                    .iter()
162                    .map(|block_transaction| block_transaction.transaction_id.unwrap())
163                    .collect(),
164            )
165            .await;
166
167        Node::send_to_peers(
168            node.clone(),
169            Message::new(Command::NewBlock {
170                block: new_block.clone(),
171            }),
172        )
173        .await;
174        {
175            node.write().await.last_seen_block = new_block.hash.unwrap();
176        }
177        Ok(())
178    }
179
180    /// Submit a new transaction to the network to be mined
181    pub async fn submit_transaction(
182        node: Arc<RwLock<Node>>,
183        new_transaction: Transaction,
184    ) -> Result<(), BlockchainError> {
185        let tx_difficulty =
186            BigUint::from_bytes_be(&node.read().await.blockchain.get_transaction_difficulty());
187
188        node.read()
189            .await
190            .blockchain
191            .get_utxos()
192            .validate_transaction(&new_transaction.clone(), &tx_difficulty)?;
193
194        if !node
195            .read()
196            .await
197            .mempool
198            .validate_transaction(&new_transaction)
199            .await
200        {
201            return Err(BlockchainError::DoubleSpend);
202        }
203        validate_transaction_timestamp(&new_transaction)?;
204
205        node.write()
206            .await
207            .mempool
208            .add_transaction(new_transaction.clone())
209            .await;
210
211        Node::send_to_peers(
212            node.clone(),
213            Message::new(Command::NewTransaction {
214                transaction: new_transaction.clone(),
215            }),
216        )
217        .await;
218        Ok(())
219    }
220
221    /// Log a message to the node log
222    pub fn log(msg: String) {
223        let mut log_file = fs::OpenOptions::new()
224            .append(true)
225            .create(true)
226            .open(
227                NODE_PATH
228                    .get()
229                    .expect("One blockchain instance must exist before logging")
230                    .to_owned()
231                    + "/info.log",
232            )
233            .expect("Could not open logging file!");
234        writeln!(
235            log_file,
236            "[{}] {}",
237            chrono::Local::now().format("%Y-%m-%d %H:%M:%S"),
238            msg
239        )
240        .expect("Failed to write to logging file");
241    }
242}
snap_coin/node/node.rs

snap_coin/node/
node.rs
