Merge pull request #8 from rainshowerLabs/0.3.2-bug-fixes-optimizations

0.3.2 bug fixes optimizations

Author: makemake-kbo

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "sothis"
-version = "0.3.1"
+version = "0.3.2"
 edition = "2021"
 authors = ["makemake <vukasin@gostovic.me>"]
 license = "GPL-3.0-or-later"
@@ -23,3 +23,4 @@ rlp = "0.5.2"
 serde = { version = "1.0.163", features = ["derive"] }
 serde_json = "1.0.96"
 tokio = { version = "1.28.1", features = ["full"] }
+url = "2.4.0"

Cargo.toml

@@ -106,10 +106,11 @@ The result is saved to a JSON file that looks like this:
 - `--source_rpc`: RPC of the node we are getting data from.
 - `--contract_address`: Address of the contract we are reading storage from.
 - `--storage_slot`: The storage slot of the contract.
+- `--terminal_block`(optional): Final block sothis will track. If not specified, sothis will track until terminated.
 - `--filename`(optional): Name of our output file. The default filename is formatted as: `address-{}-slot-{}-timestamp-{}.json`.
 - `--path`(optional): Path to our output file. The default path is the current directory.
 
-Once you are done tracking the slot, terminate the process via a `SIGTERM` or a `SIGINT` (ctrl-c), which will terminate execution and write the file.
+Once you are done tracking the slot, terminate the process via a `SIGTERM` or a `SIGINT` (ctrl-c), which will terminate execution and write the file. Keep in mind that sothis will check once per new block if you tried to terminate it. If no new block are produced on the source_rpc, sothis will not terminate and nothing will be written if you force close it.
 
 `sothis --mode track --source_rpc http://localhost:8545 --contract_address 0x1c479675ad559DC151F6Ec7ed3FbF8ceE79582B6 --storage_slot 0 --filename siuuu.json --path ~/Desktop
 `

README.md

@@ -0,0 +1,48 @@
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+
+
+class DummyRPCHandler(BaseHTTPRequestHandler):
+    def _set_response(self, status_code=200):
+        self.send_response(status_code)
+        self.send_header('Content-type', 'application/json')
+        self.end_headers()
+
+    def do_POST(self):
+        content_length = int(self.headers['Content-Length'])
+        request_body = self.rfile.read(content_length)
+        rpc_request = json.loads(request_body)
+
+        if rpc_request['method'] == 'eth_blockNumber':
+            response = {
+                "jsonrpc": "2.0",
+                "id": rpc_request['id'],
+                "result": "0x123456"  # Dummy block number
+            }
+        else:
+            response = {
+                "jsonrpc": "2.0",
+                "id": rpc_request['id'],
+                "error": {
+                    "code": -32601,
+                    "message": "Method not found"
+                }
+            }
+
+        self._set_response()
+        self.wfile.write(json.dumps(response).encode())
+
+    def do_GET(self):
+        self._set_response(404)
+        self.wfile.write(b'Not found')
+
+
+def run_server():
+    server_address = ('localhost', 8000)
+    httpd = HTTPServer(server_address, DummyRPCHandler)
+    print('Dummy RPC server is running on http://localhost:8000...')
+    httpd.serve_forever()
+
+
+if __name__ == '__main__':
+    run_server()

dummy_server.py

@@ -33,7 +33,7 @@ lazy_static! {
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let matches = Command::new("sothis")
-        .version("0.3.1")
+        .version("0.3.2")
         .author("makemake <vukasin@gostovic.me>")
         .about("Tool for replaying historical transactions. Designed to be used with anvil or hardhat.")
         .arg(Arg::new("source_rpc")
@@ -131,13 +131,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         "historic" => {
             println!("Replaying in historic mode...");
 
-            let block: String = matches.get_one::<String>("terminal_block").expect("required").to_string();
-            let block = format_number_input(&block);
+            let terminal_block: String = matches.get_one::<String>("terminal_block").expect("required").to_string();
+            let terminal_block = format_number_input(&terminal_block);
 
             let replay_rpc: String = matches.get_one::<String>("replay_rpc").expect("required").to_string();
             let replay_rpc = RpcConnection::new(replay_rpc);
 
-            replay_historic_blocks(source_rpc, replay_rpc, hex_to_decimal(&block)?).await?;
+            replay_historic_blocks(source_rpc, replay_rpc, hex_to_decimal(&terminal_block)?).await?;
         },
         "live" => {
             println!("Replaying live blocks...");
@@ -149,16 +149,22 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
         "track" => {
             println!("Tracking state variable...");
-            println!("Send SIGTERM or SIGINT to serialize to JSON, write and stop.");
+            println!("Send SIGTERM or SIGINT (ctrl-c) to serialize to JSON, write and stop.");
 
             let contract_address: String = matches.get_one::<String>("contract_address").expect("required").to_string();
             let storage_slot: String = matches.get_one::<String>("storage_slot").expect("required").to_string();
             let storage_slot = U256::from_dec_str(&storage_slot)?;
+            
+            // If terminal_block is set by the user use that, otherwise have it be none
+            let terminal_block: Option<u64> = matches.get_one::<String>("terminal_block").map(|x| x.parse().expect("Invalid terminal block"));
+            
+            if terminal_block == None {
+                println!("No terminal block set, tracking indefinitely.");
+            }
 
-            track_state(source_rpc, storage_slot, contract_address).await?;
+            track_state(source_rpc, storage_slot, contract_address, terminal_block).await?;
         }
         &_ => {
-            // handle this properly later
             panic!("Mode does not exist!");
         },
     }

src/main.rs

@@ -46,8 +46,8 @@ pub async fn replay_historic_blocks(
         let app_config = APP_CONFIG.lock()?;
         replay_delay = app_config.replay_delay;
     }
-
-    while until > replay_block {
+    
+    loop {
         // we write a bit of illegible code
         let hex_block = decimal_to_hex(replay_block + 1);
         // get block from historical node
@@ -71,8 +71,10 @@ pub async fn replay_historic_blocks(
 
         replay_block = hex_to_decimal(&replay_rpc.block_number().await?)?;
 
-        // TODO: For some godforsaken reason i cannot do an infinite loop and break here or else it crashes.
-        // I feel dirty doing 2 checks for the same thing so you have to wait a bit ig.
+        if replay_block >= until {
+            break;
+        }
+
         sleep(Duration::from_millis(replay_delay));
     }
     println!("Done replaying blocks");

src/replay/replay.rs

@@ -1,51 +1,52 @@
+use crate::rpc::error::RequestError;
 use crate::RpcConnection;
 use crate::rpc::types::Transaction;
 use crate::APP_CONFIG;
 
+// Abstract over the return types of send functions
+impl RpcConnection {
+    async fn send(&self, tx: Transaction, chain_id: u64) -> Result<String, RequestError> {
+        if APP_CONFIG.lock().unwrap().send_as_raw {
+            self.send_raw_transaction(tx, chain_id).await
+        } else {
+            self.send_unsigned_transaction(tx, chain_id).await
+        }
+    }
+}
+
 // Generic function we use to replay all tx in a block.
 pub async fn send_transactions(
     replay_rpc: RpcConnection,
     historical_txs: Vec<Transaction>,
     chain_id: u64,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    let app_config = APP_CONFIG.lock()?;
+    let exit_on_tx_fail;
+    let entropy_threshold;
+    {
+        let app_config = APP_CONFIG.lock()?;
+        exit_on_tx_fail = app_config.exit_on_tx_fail;
+        entropy_threshold = app_config.entropy_threshold;
+    }
 
     let tx_amount = historical_txs.len() as f32;
     let mut fail_tx_amount: f32 = 0.0;
 
-    // TODO: This is really bad, please reimplement this
-
-    if app_config.send_as_raw {
-        for tx in historical_txs {
-            // Gracefully handle errors so execution doesn't halt on error
-            match replay_rpc.send_raw_transaction(tx, chain_id).await {
-                Ok(_) => (),
-                Err(e) => if app_config.exit_on_tx_fail {
-                    return Err(e.into());
-                } else {
-                    fail_tx_amount += 1.0;
-                    println!("!!! \x1b[93mError sending transaction:\x1b[0m {} !!!", e)
-                }
-            }
-        }
-    } else {
-        for tx in historical_txs {
-            // Gracefully handle errors so execution doesn't halt on error
-            match replay_rpc.send_unsigned_transaction(tx, chain_id).await {
-                Ok(_) => (),
-                Err(e) => if app_config.exit_on_tx_fail {
-                    return Err(e.into());
-                } else {
-                    fail_tx_amount += 1.0;
-                    println!("!!! \x1b[93mError sending transaction:\x1b[0m {} !!!", e)
-                }
+    for tx in historical_txs {
+        // Gracefully handle errors so execution doesn't halt on error
+        match replay_rpc.send(tx, chain_id).await {
+            Ok(_) => (),
+            Err(e) => if exit_on_tx_fail {
+                return Err(e.into());
+            } else {
+                fail_tx_amount += 1.0;
+                println!("!!! \x1b[93mError sending transaction:\x1b[0m {} !!!", e)
             }
         }
     }
 
     // Calculate the percentage of failed transactions
     let fail_percent = fail_tx_amount / tx_amount;
-    if fail_percent > app_config.entropy_threshold {
+    if fail_percent > entropy_threshold {
         println!("!!! \x1b[91mHigh entropy detected!\x1b[0m Fail ratio: {}. Consider restarting the fork\x1b[0m !!!", format!("{:.2}%", fail_percent * 100.0));
     }
 

src/replay/send_transaction.rs

@@ -1,4 +1,6 @@
+use url::Url;
 use std::thread::sleep;
+use std::time::Instant;
 
 use tokio::time::Duration;
 use serde::{Deserialize, Serialize};
@@ -27,27 +29,19 @@ struct JsonRpcResponse {
     id: u32,
 }
 
+#[derive(Clone)]
 pub struct RpcConnection {
     client: Client,
     url: String,
 }
 
-impl Clone for RpcConnection {
-    fn clone(&self) -> Self {
-        RpcConnection {
-            client: self.client.clone(),
-            url: self.url.clone(),
-        }
-    }
-}
-
 #[allow(dead_code)]
 impl RpcConnection {
     // Create client and set url
     pub fn new(url: String) -> Self {
         Self {
             client: Client::new(),
-            url,
+            url: Url::parse(&url).expect("REASON").into(),
         }
     }
 
@@ -240,9 +234,23 @@ impl RpcConnection {
         let mut new_blocknumber = blocknumber.clone();
         println!("Listening for new blocks from block {}...", hex_to_decimal(&blocknumber).unwrap());
 
+        // Start timer for the *heartbeat*
+        let mut start_time = Instant::now();
+
         while blocknumber == new_blocknumber {
-            // sleep for 1 second
+            // sleep for set duration
             sleep(Duration::from_millis(time));
+            
+            // Add this as a *heartbeat* so users are less confused if nothing is happening
+            let elapsed_time = start_time.elapsed();
+
+            if elapsed_time >= Duration::from_secs(60) {
+                println!("!!! \x1b[93mNo new blocks have been detected in 60 seconds! Check your node(s)\x1b[0m !!!");
+                println!("Still listening...");
+                start_time = Instant::now();
+            }
+
+
             new_blocknumber = self.block_number().await?
         }
 

src/rpc/rpc.rs

@@ -76,8 +76,6 @@ impl Transaction {
         // features set to legacy, this is a legacy tx
         let mut typed_tx: TypedTransaction = Default::default();
 
-        //todo: fix this
-        // If to doesnt contain a value, set it
         match self.to {
             Some(_) => {
                 let address = H160::from_str(&self.to.clone().expect("Can't read `to` field"));