node + config + ui: fork stratum server to handle state on node start/stop, complete stratum configuration, complete node server config, settings ui refactoring, update i18n lib

src/node/config.rs

@@ -100,6 +100,19 @@ impl NodeConfig {
         api_secret_path
     }
 
+    /// List of available IP addresses.
+    pub fn get_ip_addrs() -> Vec<IpAddr> {
+        let mut ip_addrs = Vec::new();
+        for net_if in pnet::datalink::interfaces() {
+            for ip in net_if.ips {
+                if ip.is_ipv4() {
+                    ip_addrs.push(ip.ip());
+                }
+            }
+        }
+        ip_addrs
+    }
+
     /// Check whether a port is available on the provided host.
     fn is_port_available(host: &String, port: &String) -> bool {
         if let Ok(p) = port.parse::<u16>() {
@@ -142,6 +155,17 @@ impl NodeConfig {
 
     /// Check if stratum server port is available across the system and config.
     pub fn is_stratum_port_available(ip: &String, port: &String) -> bool {
+        if Node::get_stratum_stats().is_running {
+            // Check if Stratum server with same address is running.
+            let (cur_ip, cur_port) = Self::get_stratum_address();
+            let same_running = ip == &cur_ip && port == &cur_port;
+            return same_running || Self::is_not_running_stratum_port_available(ip, port);
+        }
+        Self::is_not_running_stratum_port_available(&ip, &port)
+    }
+
+    /// Check if stratum port is available when server is not running.
+    fn is_not_running_stratum_port_available(ip: &String, port: &String) -> bool {
         if Self::is_port_available(&ip, &port) {
             if &Self::get_p2p_port().to_string() != port {
                 let (api_ip, api_port) = Self::get_api_address();
@@ -161,6 +185,54 @@ impl NodeConfig {
         r_config.members.clone().server.stratum_mining_config.unwrap().wallet_listener_url
     }
 
+    /// Get the amount of time in seconds to attempt to mine on a particular header.
+    pub fn get_stratum_attempt_time() -> String {
+        let r_config = Settings::node_config_to_read();
+        r_config.members
+            .clone()
+            .server
+            .stratum_mining_config
+            .unwrap()
+            .attempt_time_per_block
+            .to_string()
+    }
+
+    /// Save stratum attempt time value in seconds.
+    pub fn save_stratum_attempt_time(time: u32) {
+        let mut w_node_config = Settings::node_config_to_update();
+        w_node_config.members
+            .clone()
+            .server
+            .stratum_mining_config
+            .unwrap()
+            .attempt_time_per_block = time;
+        w_node_config.save();
+    }
+
+    /// Get minimum acceptable share difficulty to request from miners.
+    pub fn get_stratum_min_share_diff() -> String {
+        let r_config = Settings::node_config_to_read();
+        r_config.members
+            .clone()
+            .server
+            .stratum_mining_config
+            .unwrap()
+            .minimum_share_difficulty
+            .to_string()
+    }
+
+    /// Save minimum acceptable share difficulty.
+    pub fn save_stratum_min_share_diff(diff: u64) {
+        let mut w_node_config = Settings::node_config_to_update();
+        w_node_config.members
+            .clone()
+            .server
+            .stratum_mining_config
+            .unwrap()
+            .minimum_share_difficulty = diff;
+        w_node_config.save();
+    }
+
     /// Check if stratum mining server autorun is enabled.
     pub fn is_stratum_autorun_enabled() -> bool {
         let stratum_config = Settings::node_config_to_read()
@@ -212,17 +284,17 @@ impl NodeConfig {
     pub fn is_api_port_available(ip: &String, port: &String) -> bool {
         if Node::is_running() {
             // Check if API server with same address is running.
-            let same_running = if let Some(running_addr) =  Node::get_api_addr() {
+            let same_running = if let Some(running_addr) = Node::get_api_addr() {
                 running_addr == format!("{}:{}", ip, port)
             } else {
                 false
             };
-            if same_running || NodeConfig::is_port_available(&ip, &port) {
-                return &NodeConfig::get_p2p_port().to_string() != port;
+            if same_running || Self::is_port_available(&ip, &port) {
+                return &Self::get_p2p_port().to_string() != port;
             }
             return false;
-        } else if NodeConfig::is_port_available(&ip, &port) {
-            return &NodeConfig::get_p2p_port().to_string() != port;
+        } else if Self::is_port_available(&ip, &port) {
+            return &Self::get_p2p_port().to_string() != port;
         }
         false
     }
@@ -472,7 +544,7 @@ impl NodeConfig {
     }
 
     /// Set how long a banned peer should stay banned in ms.
-    pub fn set_ban_window(time: i64) {
+    pub fn save_ban_window(time: i64) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.p2p_config.ban_window = Some(time);
         w_node_config.save();
@@ -484,7 +556,7 @@ impl NodeConfig {
     }
 
     /// Set maximum number of inbound peer connections.
-    pub fn set_max_inbound_count(count: u32) {
+    pub fn save_max_inbound_count(count: u32) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.p2p_config.peer_max_inbound_count = Some(count);
         w_node_config.save();
@@ -496,7 +568,7 @@ impl NodeConfig {
     }
 
     /// Set maximum number of outbound peer connections.
-    pub fn set_max_outbound_count(count: u32) {
+    pub fn save_max_outbound_count(count: u32) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.p2p_config.peer_max_outbound_count = Some(count);
         w_node_config.save();
@@ -512,7 +584,7 @@ impl NodeConfig {
     }
 
     /// Set minimum number of outbound peer connections.
-    pub fn set_min_outbound_count(count: u32) {
+    pub fn save_min_outbound_count(count: u32) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.p2p_config.peer_min_preferred_outbound_count = Some(count);
         w_node_config.save();
@@ -526,7 +598,7 @@ impl NodeConfig {
     }
 
     /// Set base fee that's accepted into the pool.
-    pub fn set_base_fee(fee: u64) {
+    pub fn save_base_fee(fee: u64) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.pool_config.accept_fee_base = fee;
         w_node_config.save();
@@ -538,7 +610,7 @@ impl NodeConfig {
     }
 
     /// Set reorg cache retention period in minute.
-    pub fn set_reorg_cache_period(period: u32) {
+    pub fn save_reorg_cache_period(period: u32) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.pool_config.reorg_cache_period = period;
         w_node_config.save();
@@ -550,7 +622,7 @@ impl NodeConfig {
     }
 
     /// Set max amount of transactions at pool.
-    pub fn set_max_pool_size(amount: usize) {
+    pub fn save_max_pool_size(amount: usize) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.pool_config.max_pool_size = amount;
         w_node_config.save();
@@ -562,7 +634,7 @@ impl NodeConfig {
     }
 
     /// Set max amount of transactions at stem pool.
-    pub fn set_max_stempool_size(amount: usize) {
+    pub fn save_max_stempool_size(amount: usize) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.pool_config.max_stempool_size = amount;
         w_node_config.save();
@@ -574,7 +646,7 @@ impl NodeConfig {
     }
 
     /// Set max total weight of transactions that can get selected to build a block.
-    pub fn set_mineable_max_weight(weight: u64) {
+    pub fn save_mineable_max_weight(weight: u64) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.pool_config.mineable_max_weight = weight;
         w_node_config.save();
@@ -588,7 +660,7 @@ impl NodeConfig {
     }
 
     /// Set Dandelion epoch duration in secs.
-    pub fn set_epoch(secs: u16) {
+    pub fn save_epoch(secs: u16) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.dandelion_config.epoch_secs = secs;
         w_node_config.save();
@@ -601,7 +673,7 @@ impl NodeConfig {
     }
 
     /// Set Dandelion embargo timer.
-    pub fn set_embargo(secs: u16) {
+    pub fn save_embargo(secs: u16) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.dandelion_config.embargo_secs = secs;
         w_node_config.save();
@@ -613,7 +685,7 @@ impl NodeConfig {
     }
 
     /// Set Dandelion stem probability.
-    pub fn set_stem_probability(percent: u8) {
+    pub fn save_stem_probability(percent: u8) {
         let mut w_node_config = Settings::node_config_to_update();
         w_node_config.members.server.dandelion_config.stem_probability = percent;
         w_node_config.save();

src/node/mine_block.rs

@@ -0,0 +1,298 @@
+// Copyright 2023 The Grim Developers
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Build a block to mine: gathers transactions from the pool, assembles
+//! them into a block and returns it.
+
+use chrono::prelude::{DateTime, NaiveDateTime, Utc};
+use rand::{thread_rng, Rng};
+use serde_json::{json, Value};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+use grin_api;
+use grin_chain;
+use grin_servers::common::types::Error;
+use grin_core::core::{Output, TxKernel};
+use grin_core::libtx::secp_ser;
+use grin_core::libtx::ProofBuilder;
+use grin_core::{consensus, core, global};
+use grin_keychain::{ExtKeychain, Identifier, Keychain};
+use grin_servers::ServerTxPool;
+use log::{debug, error, trace, warn};
+use serde_derive::{Deserialize, Serialize};
+
+/// Fees in block to use for coinbase amount calculation
+/// (Duplicated from Grin wallet project)
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct BlockFees {
+    /// fees
+    #[serde(with = "secp_ser::string_or_u64")]
+    pub fees: u64,
+    /// height
+    #[serde(with = "secp_ser::string_or_u64")]
+    pub height: u64,
+    /// key id
+    pub key_id: Option<Identifier>,
+}
+
+impl BlockFees {
+    /// return key id
+    pub fn key_id(&self) -> Option<Identifier> {
+        self.key_id.clone()
+    }
+}
+
+/// Response to build a coinbase output.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct CbData {
+    /// Output
+    pub output: Output,
+    /// Kernel
+    pub kernel: TxKernel,
+    /// Key Id
+    pub key_id: Option<Identifier>,
+}
+
+// Ensure a block suitable for mining is built and returned
+// If a wallet listener URL is not provided the reward will be "burnt"
+// Warning: This call does not return until/unless a new block can be built
+pub fn get_block(
+    chain: &Arc<grin_chain::Chain>,
+    tx_pool: &ServerTxPool,
+    key_id: Option<Identifier>,
+    wallet_listener_url: Option<String>,
+) -> (core::Block, BlockFees) {
+    let wallet_retry_interval = 5;
+    // get the latest chain state and build a block on top of it
+    let mut result = build_block(chain, tx_pool, key_id.clone(), wallet_listener_url.clone());
+    while let Err(e) = result {
+        let mut new_key_id = key_id.to_owned();
+        match e {
+            self::Error::Chain(c) => match c {
+                grin_chain::Error::DuplicateCommitment(_) => {
+                    debug!(
+						"Duplicate commit for potential coinbase detected. Trying next derivation."
+					);
+                    // use the next available key to generate a different coinbase commitment
+                    new_key_id = None;
+                }
+                _ => {
+                    error!("Chain Error: {}", c);
+                }
+            },
+            self::Error::WalletComm(_) => {
+                error!(
+					"Error building new block: Can't connect to wallet listener at {:?}; will retry",
+					wallet_listener_url.as_ref().unwrap()
+				);
+                thread::sleep(Duration::from_secs(wallet_retry_interval));
+            }
+            ae => {
+                warn!("Error building new block: {:?}. Retrying.", ae);
+            }
+        }
+
+        // only wait if we are still using the same key: a different coinbase commitment is unlikely
+        // to have duplication
+        if new_key_id.is_some() {
+            thread::sleep(Duration::from_millis(100));
+        }
+
+        result = build_block(chain, tx_pool, new_key_id, wallet_listener_url.clone());
+    }
+    return result.unwrap();
+}
+
+/// Builds a new block with the chain head as previous and eligible
+/// transactions from the pool.
+fn build_block(
+    chain: &Arc<grin_chain::Chain>,
+    tx_pool: &ServerTxPool,
+    key_id: Option<Identifier>,
+    wallet_listener_url: Option<String>,
+) -> Result<(core::Block, BlockFees), Error> {
+    let head = chain.head_header()?;
+
+    // prepare the block header timestamp
+    let mut now_sec = Utc::now().timestamp();
+    let head_sec = head.timestamp.timestamp();
+    if now_sec <= head_sec {
+        now_sec = head_sec + 1;
+    }
+
+    // Determine the difficulty our block should be at.
+    // Note: do not keep the difficulty_iter in scope (it has an active batch).
+    let difficulty = consensus::next_difficulty(head.height + 1, chain.difficulty_iter()?);
+
+    // Extract current "mineable" transactions from the pool.
+    // If this fails for *any* reason then fallback to an empty vec of txs.
+    // This will allow us to mine an "empty" block if the txpool is in an
+    // invalid (and unexpected) state.
+    let txs = match tx_pool.read().prepare_mineable_transactions() {
+        Ok(txs) => txs,
+        Err(e) => {
+            error!(
+				"build_block: Failed to prepare mineable txs from txpool: {:?}",
+				e
+			);
+            warn!("build_block: Falling back to mining empty block.");
+            vec![]
+        }
+    };
+
+    // build the coinbase and the block itself
+    let fees = txs.iter().map(|tx| tx.fee()).sum();
+    let height = head.height + 1;
+    let block_fees = BlockFees {
+        fees,
+        key_id,
+        height,
+    };
+
+    let (output, kernel, block_fees) = get_coinbase(wallet_listener_url, block_fees)?;
+    let mut b = core::Block::from_reward(&head, &txs, output, kernel, difficulty.difficulty)?;
+
+    // making sure we're not spending time mining a useless block
+    b.validate(&head.total_kernel_offset)?;
+
+    b.header.pow.nonce = thread_rng().gen();
+    b.header.pow.secondary_scaling = difficulty.secondary_scaling;
+    b.header.timestamp = DateTime::<Utc>::from_utc(NaiveDateTime::from_timestamp(now_sec, 0), Utc);
+
+    debug!(
+		"Built new block with {} inputs and {} outputs, block difficulty: {}, cumulative difficulty {}",
+		b.inputs().len(),
+		b.outputs().len(),
+		difficulty.difficulty,
+		b.header.total_difficulty().to_num(),
+	);
+
+    // Now set txhashset roots and sizes on the header of the block being built.
+    match chain.set_txhashset_roots(&mut b) {
+        Ok(_) => Ok((b, block_fees)),
+        Err(e) => {
+            match e {
+                // If this is a duplicate commitment then likely trying to use
+                // a key that hass already been derived but not in the wallet
+                // for some reason, allow caller to retry.
+                grin_chain::Error::DuplicateCommitment(e) => {
+                    Err(Error::Chain(grin_chain::Error::DuplicateCommitment(e)))
+                }
+
+                // Some other issue, possibly duplicate kernel
+                _ => {
+                    error!("Error setting txhashset root to build a block: {:?}", e);
+                    Err(Error::Chain(grin_chain::Error::Other(format!("{:?}", e))))
+                }
+            }
+        }
+    }
+}
+
+///
+/// Probably only want to do this when testing.
+///
+fn burn_reward(block_fees: BlockFees) -> Result<(Output, TxKernel, BlockFees), Error> {
+    warn!("Burning block fees: {:?}", block_fees);
+    let keychain = ExtKeychain::from_random_seed(global::is_testnet())?;
+    let key_id = ExtKeychain::derive_key_id(1, 1, 0, 0, 0);
+    let (out, kernel) = grin_core::libtx::reward::output(
+        &keychain,
+        &ProofBuilder::new(&keychain),
+        &key_id,
+        block_fees.fees,
+        false,
+    )
+        .unwrap();
+    Ok((out, kernel, block_fees))
+}
+
+// Connect to the wallet listener and get coinbase.
+// Warning: If a wallet listener URL is not provided the reward will be "burnt"
+fn get_coinbase(
+    wallet_listener_url: Option<String>,
+    block_fees: BlockFees,
+) -> Result<(core::Output, core::TxKernel, BlockFees), Error> {
+    match wallet_listener_url {
+        None => {
+            // Burn it
+            return burn_reward(block_fees);
+        }
+        Some(wallet_listener_url) => {
+            let res = create_coinbase(&wallet_listener_url, &block_fees)?;
+            let output = res.output;
+            let kernel = res.kernel;
+            let key_id = res.key_id;
+            let block_fees = BlockFees {
+                key_id: key_id,
+                ..block_fees
+            };
+
+            debug!("get_coinbase: {:?}", block_fees);
+            return Ok((output, kernel, block_fees));
+        }
+    }
+}
+
+/// Call the wallet API to create a coinbase output for the given block_fees.
+/// Will retry based on default "retry forever with backoff" behavior.
+fn create_coinbase(dest: &str, block_fees: &BlockFees) -> Result<CbData, Error> {
+    let url = format!("{}/v2/foreign", dest);
+    let req_body = json!({
+		"jsonrpc": "2.0",
+		"method": "build_coinbase",
+		"id": 1,
+		"params": {
+			"block_fees": block_fees
+		}
+	});
+
+    trace!("Sending build_coinbase request: {}", req_body);
+    let req = grin_api::client::create_post_request(url.as_str(), None, &req_body)?;
+    let timeout = grin_api::client::TimeOut::default();
+    let res: String = grin_api::client::send_request(req, timeout).map_err(|e| {
+        let report = format!(
+            "Failed to get coinbase from {}. Is the wallet listening? {}",
+            dest, e
+        );
+        error!("{}", report);
+        Error::WalletComm(report)
+    })?;
+
+    let res: Value = serde_json::from_str(&res).unwrap();
+    trace!("Response: {}", res);
+    if res["error"] != json!(null) {
+        let report = format!(
+            "Failed to get coinbase from {}: Error: {}, Message: {}",
+            dest, res["error"]["code"], res["error"]["message"]
+        );
+        error!("{}", report);
+        return Err(Error::WalletComm(report));
+    }
+
+    let cb_data = res["result"]["Ok"].clone();
+    trace!("cb_data: {}", cb_data);
+    let ret_val = match serde_json::from_value::<CbData>(cb_data) {
+        Ok(r) => r,
+        Err(e) => {
+            let report = format!("Couldn't deserialize CbData: {}", e);
+            error!("{}", report);
+            return Err(Error::WalletComm(report));
+        }
+    };
+
+    Ok(ret_val)
+}

src/node/mod.rs

@@ -16,4 +16,8 @@ mod node;
 pub use node::Node;
 
 mod config;
+
+mod stratum;
+mod mine_block;
+
 pub use config::NodeConfig;

src/node/node.rs

@@ -22,11 +22,12 @@ use futures::channel::oneshot;
 use grin_chain::SyncStatus;
 use grin_core::global;
 use grin_core::global::ChainTypes;
-use grin_servers::{Server, ServerStats};
+use grin_servers::{Server, ServerStats, StratumServerConfig, StratumStats};
 use grin_servers::common::types::Error;
 use jni::sys::{jboolean, jstring};
 use lazy_static::lazy_static;
 use crate::node::NodeConfig;
+use crate::node::stratum::StratumServer;
 
 lazy_static! {
     /// Static thread-aware state of [`Node`] to be updated from another thread.
@@ -35,8 +36,10 @@ lazy_static! {
 
 /// Provides [`Server`] control, holds current status and statistics.
 pub struct Node {
-    /// Statistics data for UI.
+    /// The node [`Server`] statistics for UI.
     stats: Arc<RwLock<Option<ServerStats>>>,
+    /// Stratum server statistics.
+    stratum_stats: Arc<grin_util::RwLock<StratumStats>>,
     /// Running API server address.
     api_addr: Arc<RwLock<Option<String>>>,
     /// Running P2P server port.
@@ -49,8 +52,8 @@ pub struct Node {
     stop_needed: AtomicBool,
     /// Flag to check if app exit is needed after server stop.
     exit_after_stop: AtomicBool,
-    /// Thread flag to start stratum server at separate.
-    start_stratum_server: AtomicBool,
+    /// Thread flag to start stratum server.
+    start_stratum_needed: AtomicBool,
     /// Error on [`Server`] start.
     init_error: Option<Error>
 }
@@ -59,13 +62,14 @@ impl Default for Node {
     fn default() -> Self {
         Self {
             stats: Arc::new(RwLock::new(None)),
+            stratum_stats: Arc::new(grin_util::RwLock::new(StratumStats::default())),
             api_addr: Arc::new(RwLock::new(None)),
             p2p_port: Arc::new(RwLock::new(None)),
             starting: AtomicBool::new(false),
             restart_needed: AtomicBool::new(false),
             stop_needed: AtomicBool::new(false),
             exit_after_stop: AtomicBool::new(false),
-            start_stratum_server: AtomicBool::new(false),
+            start_stratum_needed: AtomicBool::new(false),
             init_error: None
         }
     }
@@ -114,14 +118,14 @@ impl Node {
         }
     }
 
-    /// Start stratum server.
+    /// Request to start stratum server.
     pub fn start_stratum_server() {
-        NODE_STATE.start_stratum_server.store(true, Ordering::Relaxed);
+        NODE_STATE.start_stratum_needed.store(true, Ordering::Relaxed);
     }
 
     /// Check if stratum server is starting.
     pub fn is_stratum_server_starting() -> bool {
-        NODE_STATE.start_stratum_server.load(Ordering::Relaxed)
+        NODE_STATE.start_stratum_needed.load(Ordering::Relaxed)
     }
 
     /// Check if node is starting.
@@ -149,6 +153,11 @@ impl Node {
         NODE_STATE.stats.read().unwrap()
     }
 
+    /// Get stratum server [`Server`] statistics.
+    pub fn get_stratum_stats() -> grin_util::RwLockReadGuard<'static, StratumStats> {
+        NODE_STATE.stratum_stats.read()
+    }
+
     /// Get synchronization status, empty when [`Server`] is not running.
     pub fn get_sync_status() -> Option<SyncStatus> {
         // Return Shutdown status when node is stopping.
@@ -169,13 +178,13 @@ impl Node {
         None
     }
 
-    /// Start node [`Server`] at separate thread to update [`NODE_STATE`] with [`ServerStats`].
+    /// Start the [`Server`] at separate thread to update state with stats and handle statuses.
     fn start_server_thread() {
         thread::spawn(move || {
             NODE_STATE.starting.store(true, Ordering::Relaxed);
 
             // Start the server.
-            match start_server() {
+            match start_node_server() {
                 Ok(mut server) => {
                     let mut first_start = true;
                     loop {
@@ -183,84 +192,85 @@ impl Node {
                             // Stop the server.
                             server.stop();
 
+                            // Reset stratum stats
+                            {
+                                let mut w_stratum_stats = NODE_STATE.stratum_stats.write();
+                                *w_stratum_stats = StratumStats::default();
+                            }
+
                             // Create new server.
-                            match start_server() {
+                            match start_node_server() {
                                 Ok(s) => {
                                     server = s;
                                     NODE_STATE.restart_needed.store(false, Ordering::Relaxed);
                                 }
                                 Err(e) => {
-                                    NODE_STATE.restart_needed.store(false, Ordering::Relaxed);
                                     Self::on_start_error(&e);
                                     break;
                                 }
                             }
                         } else if Self::is_stopping() {
-                            // Clean server stats.
-                            {
-                                let mut w_stats = NODE_STATE.stats.write().unwrap();
-                                *w_stats = None;
-                            }
-
                             // Stop the server.
                             server.stop();
-
-                            NODE_STATE.starting.store(false, Ordering::Relaxed);
-                            NODE_STATE.stop_needed.store(false, Ordering::Relaxed);
-                            NODE_STATE.start_stratum_server.store(false, Ordering::Relaxed);
-
-                            // Clean launched API server address.
-                            {
-                                let mut w_api_addr = NODE_STATE.api_addr.write().unwrap();
-                                *w_api_addr = None;
-                            }
-                            // Clean launched P2P server port.
-                            {
-                                let mut w_p2p_port = NODE_STATE.p2p_port.write().unwrap();
-                                *w_p2p_port = None;
-                            }
+                            // Clean stats and statuses.
+                            Self::on_thread_stop();
+                            // Exit thread loop.
                             break;
-                        } else {
-                            // Start stratum mining server.
-                            if Self::is_stratum_server_starting() {
+                        }
+
+                        // Start stratum mining server if requested.
+                        let stratum_start_requested = Self::is_stratum_server_starting();
+                        if stratum_start_requested {
+                            let (s_ip, s_port) = NodeConfig::get_stratum_address();
+                            if NodeConfig::is_stratum_port_available(&s_ip, &s_port) {
                                 let stratum_config = server
                                     .config
                                     .stratum_mining_config
                                     .clone()
                                     .unwrap();
-                                server.start_stratum_server(stratum_config);
-
-                                // Wait for mining server to start and update status.
-                                thread::sleep(Duration::from_millis(100));
-                                NODE_STATE.start_stratum_server.store(false, Ordering::Relaxed);
-                            }
-
-                            // Update server stats.
-                            if let Ok(stats) = server.get_server_stats() {
-                                {
-                                    let mut w_stats = NODE_STATE.stats.write().unwrap();
-                                    *w_stats = Some(stats);
-                                }
-
-                                if first_start {
-                                    NODE_STATE.starting.store(false, Ordering::Relaxed);
-                                    first_start = false;
-                                }
+                                start_stratum_mining_server(&server, stratum_config);
                             }
                         }
+
+                        // Update server stats.
+                        if let Ok(stats) = server.get_server_stats() {
+                            {
+                                let mut w_stats = NODE_STATE.stats.write().unwrap();
+                                *w_stats = Some(stats.clone());
+                            }
+
+                            if first_start {
+                                NODE_STATE.starting.store(false, Ordering::Relaxed);
+                                first_start = false;
+                            }
+                        }
+
+                        if stratum_start_requested {
+                            NODE_STATE.start_stratum_needed.store(false, Ordering::Relaxed);
+                        }
+
                         thread::sleep(Duration::from_millis(250));
                     }
                 }
                 Err(e) => {
-                    NODE_STATE.starting.store(false, Ordering::Relaxed);
                     Self::on_start_error(&e);
                 }
             }
         });
     }
 
-    /// Handle node [`Server`] error on start.
-    fn on_start_error(e: &Error) {
+    /// Reset stats and statuses on [`Server`] thread stop.
+    fn on_thread_stop() {
+        NODE_STATE.starting.store(false, Ordering::Relaxed);
+        NODE_STATE.restart_needed.store(false, Ordering::Relaxed);
+        NODE_STATE.start_stratum_needed.store(false, Ordering::Relaxed);
+        NODE_STATE.stop_needed.store(false, Ordering::Relaxed);
+
+        // Reset stratum stats.
+        {
+            let mut w_stratum_stats = NODE_STATE.stratum_stats.write();
+            *w_stratum_stats = StratumStats::default();
+        }
         // Clean server stats.
         {
             let mut w_stats = NODE_STATE.stats.write().unwrap();
@@ -276,6 +286,12 @@ impl Node {
             let mut w_p2p_port = NODE_STATE.p2p_port.write().unwrap();
             *w_p2p_port = None;
         }
+    }
+
+    /// Handle node [`Server`] error on start.
+    fn on_start_error(e: &Error) {
+        Self::on_thread_stop();
+
         //TODO: Create error
         // NODE_STATE.init_error = Some(e);
 
@@ -464,8 +480,8 @@ impl Node {
     }
 }
 
-/// Start the [`Server`] for node.
-fn start_server() -> Result<Server, Error>  {
+/// Start the node [`Server`].
+fn start_node_server() -> Result<Server, Error>  {
     // Get current global config
     let config = NodeConfig::get_members();
     let server_config = config.server.clone();
@@ -545,10 +561,34 @@ fn start_server() -> Result<Server, Error>  {
         *w_p2p_port = Some(config.server.p2p_config.port);
     }
 
+    // Put flag to start stratum server if autorun is available.
+    if NodeConfig::is_stratum_autorun_enabled() {
+        NODE_STATE.start_stratum_needed.store(true, Ordering::Relaxed);
+    }
+
     let server_result = Server::new(server_config.clone(), None, api_chan);
     server_result
 }
 
+/// Start stratum mining server on a separate thread.
+pub fn start_stratum_mining_server(server: &Server, config: StratumServerConfig) {
+    let proof_size = global::proofsize();
+    let sync_state = server.sync_state.clone();
+
+    let mut stratum_server = StratumServer::new(
+        config,
+        server.chain.clone(),
+        server.tx_pool.clone(),
+        NODE_STATE.stratum_stats.clone(),
+    );
+    let stop_state = server.stop_state.clone();
+    let _ = thread::Builder::new()
+        .name("stratum_server".to_string())
+        .spawn(move || {
+            stratum_server.run_loop(proof_size, sync_state, stop_state);
+        });
+}
+
 #[allow(dead_code)]
 #[cfg(target_os = "android")]
 #[allow(non_snake_case)]

src/node/stratum.rs

@@ -0,0 +1,934 @@
+// Copyright 2023 The Grim Developers
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Mining Stratum Server
+
+
+
+use futures::channel::mpsc;
+use futures::pin_mut;
+use futures::{SinkExt, StreamExt, TryStreamExt};
+use tokio::net::TcpListener;
+use tokio::runtime::Runtime;
+use tokio_util::codec::{Framed, LinesCodec};
+
+use grin_util::{RwLock, StopState};
+use chrono::prelude::Utc;
+use serde_json::Value;
+use std::collections::HashMap;
+use std::net::{SocketAddr, TcpStream};
+use std::panic::panic_any;
+use std::sync::Arc;
+use std::thread;
+use std::time::{Duration, SystemTime};
+
+use grin_chain::{self, SyncState};
+use grin_servers::common::stats::{StratumStats, WorkerStats};
+use grin_servers::common::types::StratumServerConfig;
+use grin_core::consensus::graph_weight;
+use grin_core::core::hash::Hashed;
+use grin_core::core::Block;
+use grin_core::global;
+use grin_core::{pow, ser};
+use crate::node::mine_block;
+use grin_util::ToHex;
+use grin_servers::ServerTxPool;
+use log::{debug, error, info, warn};
+use serde_derive::{Deserialize, Serialize};
+
+type Tx = mpsc::UnboundedSender<String>;
+
+
+// ----------------------------------------
+// http://www.jsonrpc.org/specification
+// RPC Methods
+
+/// Represents a compliant JSON RPC 2.0 id.
+/// Valid id: Integer, String.
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
+#[serde(untagged)]
+enum JsonId {
+    IntId(u32),
+    StrId(String),
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
+struct RpcRequest {
+    id: JsonId,
+    jsonrpc: String,
+    method: String,
+    params: Option<Value>,
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
+struct RpcResponse {
+    id: JsonId,
+    jsonrpc: String,
+    method: String,
+    result: Option<Value>,
+    error: Option<Value>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct RpcError {
+    code: i32,
+    message: String,
+}
+
+impl RpcError {
+    pub fn internal_error() -> Self {
+        RpcError {
+            code: 32603,
+            message: "Internal error".to_owned(),
+        }
+    }
+    pub fn node_is_syncing() -> Self {
+        RpcError {
+            code: -32000,
+            message: "Node is syncing - Please wait".to_owned(),
+        }
+    }
+    pub fn method_not_found() -> Self {
+        RpcError {
+            code: -32601,
+            message: "Method not found".to_owned(),
+        }
+    }
+    pub fn too_late() -> Self {
+        RpcError {
+            code: -32503,
+            message: "Solution submitted too late".to_string(),
+        }
+    }
+    pub fn cannot_validate() -> Self {
+        RpcError {
+            code: -32502,
+            message: "Failed to validate solution".to_string(),
+        }
+    }
+    pub fn too_low_difficulty() -> Self {
+        RpcError {
+            code: -32501,
+            message: "Share rejected due to low difficulty".to_string(),
+        }
+    }
+    pub fn invalid_request() -> Self {
+        RpcError {
+            code: -32600,
+            message: "Invalid Request".to_string(),
+        }
+    }
+}
+
+impl From<RpcError> for Value {
+    fn from(e: RpcError) -> Self {
+        serde_json::to_value(e).unwrap()
+    }
+}
+
+impl<T> From<T> for RpcError
+    where
+        T: std::error::Error,
+{
+    fn from(e: T) -> Self {
+        error!("Received unhandled error: {}", e);
+        RpcError::internal_error()
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct LoginParams {
+    login: String,
+    pass: String,
+    agent: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct SubmitParams {
+    height: u64,
+    job_id: u64,
+    nonce: u64,
+    edge_bits: u32,
+    pow: Vec<u64>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct JobTemplate {
+    height: u64,
+    job_id: u64,
+    difficulty: u64,
+    pre_pow: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct WorkerStatus {
+    id: String,
+    height: u64,
+    difficulty: u64,
+    accepted: u64,
+    rejected: u64,
+    stale: u64,
+}
+
+struct State {
+    current_block_versions: Vec<Block>,
+    // to prevent the wallet from generating a new HD key derivation for each
+    // iteration, we keep the returned derivation to provide it back when
+    // nothing has changed. We only want to create a key_id for each new block,
+    // and reuse it when we rebuild the current block to add new tx.
+    current_key_id: Option<grin_keychain::Identifier>,
+    current_difficulty: u64,       // scaled
+    minimum_share_difficulty: u64, // unscaled
+}
+
+impl State {
+    pub fn new(minimum_share_difficulty: u64) -> Self {
+        let blocks = vec![Block::default()];
+        State {
+            current_block_versions: blocks,
+            current_key_id: None,
+            current_difficulty: <u64>::max_value(),
+            minimum_share_difficulty: minimum_share_difficulty,
+        }
+    }
+}
+
+struct Handler {
+    id: String,
+    workers: Arc<WorkersList>,
+    sync_state: Arc<SyncState>,
+    chain: Arc<grin_chain::Chain>,
+    current_state: Arc<RwLock<State>>,
+}
+
+impl Handler {
+    pub fn new(
+        id: String,
+        stratum_stats: Arc<RwLock<StratumStats>>,
+        sync_state: Arc<SyncState>,
+        minimum_share_difficulty: u64,
+        chain: Arc<grin_chain::Chain>,
+    ) -> Self {
+        Handler {
+            id: id,
+            workers: Arc::new(WorkersList::new(stratum_stats)),
+            sync_state: sync_state,
+            chain: chain,
+            current_state: Arc::new(RwLock::new(State::new(minimum_share_difficulty))),
+        }
+    }
+    pub fn from_stratum(stratum: &StratumServer) -> Self {
+        Handler::new(
+            stratum.id.clone(),
+            stratum.stratum_stats.clone(),
+            stratum.sync_state.clone(),
+            stratum.config.minimum_share_difficulty,
+            stratum.chain.clone(),
+        )
+    }
+    fn handle_rpc_requests(&self, request: RpcRequest, worker_id: usize) -> String {
+        self.workers.last_seen(worker_id);
+
+        // Call the handler function for requested method
+        let response = match request.method.as_str() {
+            "login" => self.handle_login(request.params, worker_id),
+            "submit" => {
+                let res = self.handle_submit(request.params, worker_id);
+                // this key_id has been used now, reset
+                if let Ok((_, true)) = res {
+                    self.current_state.write().current_key_id = None;
+                }
+                res.map(|(v, _)| v)
+            }
+            "keepalive" => self.handle_keepalive(),
+            "getjobtemplate" => {
+                if self.sync_state.is_syncing() {
+                    Err(RpcError::node_is_syncing())
+                } else {
+                    self.handle_getjobtemplate()
+                }
+            }
+            "status" => self.handle_status(worker_id),
+            _ => {
+                // Called undefined method
+                Err(RpcError::method_not_found())
+            }
+        };
+
+        // Package the reply as RpcResponse json
+        let resp = match response {
+            Err(rpc_error) => RpcResponse {
+                id: request.id,
+                jsonrpc: String::from("2.0"),
+                method: request.method,
+                result: None,
+                error: Some(rpc_error.into()),
+            },
+            Ok(response) => RpcResponse {
+                id: request.id,
+                jsonrpc: String::from("2.0"),
+                method: request.method,
+                result: Some(response),
+                error: None,
+            },
+        };
+        serde_json::to_string(&resp).unwrap()
+    }
+    fn handle_login(&self, params: Option<Value>, worker_id: usize) -> Result<Value, RpcError> {
+        let params: LoginParams = parse_params(params)?;
+        self.workers.login(worker_id, params.login, params.agent)?;
+        return Ok("ok".into());
+    }
+
+    // Handle KEEPALIVE message
+    fn handle_keepalive(&self) -> Result<Value, RpcError> {
+        return Ok("ok".into());
+    }
+
+    fn handle_status(&self, worker_id: usize) -> Result<Value, RpcError> {
+        // Return worker status in json for use by a dashboard or healthcheck.
+        let stats = self.workers.get_stats(worker_id)?;
+        let status = WorkerStatus {
+            id: stats.id.clone(),
+            height: self
+                .current_state
+                .read()
+                .current_block_versions
+                .last()
+                .unwrap()
+                .header
+                .height,
+            difficulty: stats.pow_difficulty,
+            accepted: stats.num_accepted,
+            rejected: stats.num_rejected,
+            stale: stats.num_stale,
+        };
+        let response = serde_json::to_value(&status).unwrap();
+        return Ok(response);
+    }
+    // Handle GETJOBTEMPLATE message
+    fn handle_getjobtemplate(&self) -> Result<Value, RpcError> {
+        // Build a JobTemplate from a BlockHeader and return JSON
+        let job_template = self.build_block_template();
+        let response = serde_json::to_value(&job_template).unwrap();
+        debug!(
+			"(Server ID: {}) sending block {} with id {} to single worker",
+			self.id, job_template.height, job_template.job_id,
+		);
+        return Ok(response);
+    }
+
+    // Build and return a JobTemplate for mining the current block
+    fn build_block_template(&self) -> JobTemplate {
+        let bh = self
+            .current_state
+            .read()
+            .current_block_versions
+            .last()
+            .unwrap()
+            .header
+            .clone();
+        // Serialize the block header into pre and post nonce strings
+        let mut header_buf = vec![];
+        {
+            let mut writer = ser::BinWriter::default(&mut header_buf);
+            bh.write_pre_pow(&mut writer).unwrap();
+            bh.pow.write_pre_pow(&mut writer).unwrap();
+        }
+        let pre_pow = header_buf.to_hex();
+        let current_state = self.current_state.read();
+        let job_template = JobTemplate {
+            height: bh.height,
+            job_id: (current_state.current_block_versions.len() - 1) as u64,
+            difficulty: current_state.minimum_share_difficulty,
+            pre_pow,
+        };
+        return job_template;
+    }
+    // Handle SUBMIT message
+    // params contains a solved block header
+    // We accept and log valid shares of all difficulty above configured minimum
+    // Accepted shares that are full solutions will also be submitted to the
+    // network
+    fn handle_submit(
+        &self,
+        params: Option<Value>,
+        worker_id: usize,
+    ) -> Result<(Value, bool), RpcError> {
+        // Validate parameters
+        let params: SubmitParams = parse_params(params)?;
+
+        let state = self.current_state.read();
+        // Find the correct version of the block to match this header
+        let b: Option<&Block> = state.current_block_versions.get(params.job_id as usize);
+        if params.height != state.current_block_versions.last().unwrap().header.height
+            || b.is_none()
+        {
+            // Return error status
+            error!(
+					"(Server ID: {}) Share at height {}, edge_bits {}, nonce {}, job_id {} submitted too late",
+					self.id, params.height, params.edge_bits, params.nonce, params.job_id,
+				);
+            self.workers.update_stats(worker_id, |ws| ws.num_stale += 1);
+            return Err(RpcError::too_late());
+        }
+
+        let scaled_share_difficulty: u64;
+        let unscaled_share_difficulty: u64;
+        let mut share_is_block = false;
+
+        let mut b: Block = b.unwrap().clone();
+        // Reconstruct the blocks header with this nonce and pow added
+        b.header.pow.proof.edge_bits = params.edge_bits as u8;
+        b.header.pow.nonce = params.nonce;
+        b.header.pow.proof.nonces = params.pow;
+
+        if !b.header.pow.is_primary() && !b.header.pow.is_secondary() {
+            // Return error status
+            error!(
+					"(Server ID: {}) Failed to validate solution at height {}, hash {}, edge_bits {}, nonce {}, job_id {}: cuckoo size too small",
+					self.id, params.height, b.hash(), params.edge_bits, params.nonce, params.job_id,
+				);
+            self.workers
+                .update_stats(worker_id, |worker_stats| worker_stats.num_rejected += 1);
+            return Err(RpcError::cannot_validate());
+        }
+
+        // Get share difficulty values
+        scaled_share_difficulty = b.header.pow.to_difficulty(b.header.height).to_num();
+        unscaled_share_difficulty = b.header.pow.to_unscaled_difficulty().to_num();
+        // Note:  state.minimum_share_difficulty is unscaled
+        //        state.current_difficulty is scaled
+        // If the difficulty is too low its an error
+        if unscaled_share_difficulty < state.minimum_share_difficulty {
+            // Return error status
+            error!(
+					"(Server ID: {}) Share at height {}, hash {}, edge_bits {}, nonce {}, job_id {} rejected due to low difficulty: {}/{}",
+					self.id, params.height, b.hash(), params.edge_bits, params.nonce, params.job_id, unscaled_share_difficulty, state.minimum_share_difficulty,
+				);
+            self.workers
+                .update_stats(worker_id, |worker_stats| worker_stats.num_rejected += 1);
+            return Err(RpcError::too_low_difficulty());
+        }
+
+        // If the difficulty is high enough, submit it (which also validates it)
+        if scaled_share_difficulty >= state.current_difficulty {
+            // This is a full solution, submit it to the network
+            let res = self.chain.process_block(b.clone(), grin_chain::Options::MINE);
+            if let Err(e) = res {
+                // Return error status
+                error!(
+						"(Server ID: {}) Failed to validate solution at height {}, hash {}, edge_bits {}, nonce {}, job_id {}, {}",
+						self.id,
+						params.height,
+						b.hash(),
+						params.edge_bits,
+						params.nonce,
+						params.job_id,
+						e,
+					);
+                self.workers
+                    .update_stats(worker_id, |worker_stats| worker_stats.num_rejected += 1);
+                return Err(RpcError::cannot_validate());
+            }
+            share_is_block = true;
+            self.workers
+                .update_stats(worker_id, |worker_stats| worker_stats.num_blocks_found += 1);
+            self.workers.stratum_stats.write().blocks_found += 1;
+            // Log message to make it obvious we found a block
+            let stats = self.workers.get_stats(worker_id)?;
+            warn!(
+					"(Server ID: {}) Solution Found for block {}, hash {} - Yay!!! Worker ID: {}, blocks found: {}, shares: {}",
+					self.id, params.height,
+					b.hash(),
+					stats.id,
+					stats.num_blocks_found,
+					stats.num_accepted,
+				);
+        } else {
+            // Do some validation but dont submit
+            let res = pow::verify_size(&b.header);
+            if res.is_err() {
+                // Return error status
+                error!(
+						"(Server ID: {}) Failed to validate share at height {}, hash {}, edge_bits {}, nonce {}, job_id {}. {:?}",
+						self.id,
+						params.height,
+						b.hash(),
+						params.edge_bits,
+						b.header.pow.nonce,
+						params.job_id,
+						res,
+					);
+                self.workers
+                    .update_stats(worker_id, |worker_stats| worker_stats.num_rejected += 1);
+                return Err(RpcError::cannot_validate());
+            }
+        }
+        // Log this as a valid share
+        self.workers.update_edge_bits(params.edge_bits as u16);
+        let worker = self.workers.get_worker(worker_id)?;
+        let submitted_by = match worker.login {
+            None => worker.id.to_string(),
+            Some(login) => login,
+        };
+
+        info!(
+				"(Server ID: {}) Got share at height {}, hash {}, edge_bits {}, nonce {}, job_id {}, difficulty {}/{}, submitted by {}",
+				self.id,
+				b.header.height,
+				b.hash(),
+				b.header.pow.proof.edge_bits,
+				b.header.pow.nonce,
+				params.job_id,
+				scaled_share_difficulty,
+				state.current_difficulty,
+				submitted_by,
+			);
+        self.workers
+            .update_stats(worker_id, |worker_stats| worker_stats.num_accepted += 1);
+        let submit_response = if share_is_block {
+            format!("blockfound - {}", b.hash().to_hex())
+        } else {
+            "ok".to_string()
+        };
+        return Ok((
+            serde_json::to_value(submit_response).unwrap(),
+            share_is_block,
+        ));
+    } // handle submit a solution
+
+    fn broadcast_job(&self) {
+        debug!("broadcast job");
+        // Package new block into RpcRequest
+        let job_template = self.build_block_template();
+        let job_template_json = serde_json::to_string(&job_template).unwrap();
+        // Issue #1159 - use a serde_json Value type to avoid extra quoting
+        let job_template_value: Value = serde_json::from_str(&job_template_json).unwrap();
+        let job_request = RpcRequest {
+            id: JsonId::StrId(String::from("Stratum")),
+            jsonrpc: String::from("2.0"),
+            method: String::from("job"),
+            params: Some(job_template_value),
+        };
+        let job_request_json = serde_json::to_string(&job_request).unwrap();
+        debug!(
+			"(Server ID: {}) sending block {} with id {} to stratum clients",
+			self.id, job_template.height, job_template.job_id,
+		);
+        self.workers.broadcast(job_request_json);
+    }
+
+    pub fn run(&self, config: &StratumServerConfig, tx_pool: &ServerTxPool, stop_state: Arc<StopState>) {
+        debug!("Run main loop");
+        let mut deadline: i64 = 0;
+        let mut head = self.chain.head().unwrap();
+        let mut current_hash = head.prev_block_h;
+        loop {
+            // Ping stratum socket on stop to handle TcpListener unbind.
+            if stop_state.is_stopped() {
+                let listen_addr: SocketAddr = config
+                    .stratum_server_addr
+                    .clone()
+                    .unwrap()
+                    .parse()
+                    .expect("Stratum: Incorrect address ");
+                thread::spawn(move || {
+                    let _ = TcpStream::connect(listen_addr).unwrap();
+                });
+                break;
+            }
+
+            // get the latest chain state
+            head = self.chain.head().unwrap();
+            let latest_hash = head.last_block_h;
+
+            // Build a new block if there is at least one worker and
+            // There is a new block on the chain or its time to rebuild
+            // the current one to include new transactions
+            if (current_hash != latest_hash || Utc::now().timestamp() >= deadline)
+                && self.workers.count() > 0
+            {
+                {
+                    debug!("resend updated block");
+                    let mut state = self.current_state.write();
+                    let wallet_listener_url = if !config.burn_reward {
+                        Some(config.wallet_listener_url.clone())
+                    } else {
+                        None
+                    };
+                    // If this is a new block we will clear the current_block version history
+                    let clear_blocks = current_hash != latest_hash;
+
+                    // Build the new block (version)
+                    let (new_block, block_fees) = mine_block::get_block(
+                        &self.chain,
+                        tx_pool,
+                        state.current_key_id.clone(),
+                        wallet_listener_url,
+                    );
+
+                    // scaled difficulty
+                    state.current_difficulty =
+                        (new_block.header.total_difficulty() - head.total_difficulty).to_num();
+
+                    state.current_key_id = block_fees.key_id();
+
+                    current_hash = latest_hash;
+                    // set the minimum acceptable share unscaled difficulty for this block
+                    state.minimum_share_difficulty = config.minimum_share_difficulty;
+
+                    // set a new deadline for rebuilding with fresh transactions
+                    deadline = Utc::now().timestamp() + config.attempt_time_per_block as i64;
+
+                    // If this is a new block we will clear the current_block version history
+                    if clear_blocks {
+                        state.current_block_versions.clear();
+                    }
+
+                    // Update the mining stats
+                    self.workers.update_block_height(new_block.header.height);
+                    let difficulty = new_block.header.total_difficulty() - head.total_difficulty;
+                    self.workers.update_network_difficulty(difficulty.to_num());
+                    self.workers.update_network_hashrate();
+
+                    // Add this new block candidate onto our list of block versions for this height
+                    state.current_block_versions.push(new_block);
+                }
+                // Send this job to all connected workers
+                self.broadcast_job();
+            }
+
+            // sleep before restarting loop
+            thread::sleep(Duration::from_millis(5));
+        } // Main Loop
+    }
+}
+
+// ----------------------------------------
+// Worker Factory Thread Function
+fn accept_connections(listen_addr: SocketAddr, handler: Arc<Handler>, stop_state: Arc<StopState>) {
+    info!("Start tokio stratum server");
+
+    let task = async move {
+        let mut listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| {
+            panic!("Stratum: Failed to bind to listen address {}", listen_addr)
+        });
+        let state_socket = &stop_state.clone();
+        let server = listener
+            .incoming()
+            .filter_map(|s| async { s.map_err(|e| error!("accept error = {:?}", e)).ok() })
+            .for_each(move |socket| {
+                let handler = handler.clone();
+                async move {
+                    // Stop listener on node server stop.
+                    if state_socket.is_stopped() {
+                        panic_any("Stopped");
+                    }
+                    // Spawn a task to process the connection
+                    let (tx, mut rx) = mpsc::unbounded();
+
+                    let worker_id = handler.workers.add_worker(tx);
+                    info!("Worker {} connected", worker_id);
+
+                    let framed = Framed::new(socket, LinesCodec::new());
+                    let (mut writer, mut reader) = framed.split();
+
+                    let h = handler.clone();
+                    let read = async move {
+                        while let Some(line) = reader
+                            .try_next()
+                            .await
+                            .map_err(|e| error!("error reading line: {}", e))?
+                        {
+                            let request = serde_json::from_str(&line)
+                                .map_err(|e| error!("error serializing line: {}", e))?;
+                            let resp = h.handle_rpc_requests(request, worker_id);
+                            h.workers.send_to(worker_id, resp);
+                        }
+
+                        Result::<_, ()>::Ok(())
+                    };
+
+                    let write = async move {
+                        while let Some(line) = rx.next().await {
+                            writer
+                                .send(line)
+                                .await
+                                .map_err(|e| error!("error writing line: {}", e))?;
+                        }
+
+                        Result::<_, ()>::Ok(())
+                    };
+
+                    let task = async move {
+                        pin_mut!(read, write);
+                        futures::future::select(read, write).await;
+                        handler.workers.remove_worker(worker_id);
+                        info!("Worker {} disconnected", worker_id);
+                    };
+                    tokio::spawn(task);
+                }
+            });
+        server.await
+    };
+    let mut rt = Runtime::new().unwrap();
+    rt.block_on(task);
+}
+
+// ----------------------------------------
+// Worker Object - a connected stratum client - a miner, pool, proxy, etc...
+
+#[derive(Clone)]
+pub struct Worker {
+    id: usize,
+    agent: String,
+    login: Option<String>,
+    authenticated: bool,
+    tx: Tx,
+}
+
+impl Worker {
+    /// Creates a new Stratum Worker.
+    pub fn new(id: usize, tx: Tx) -> Worker {
+        Worker {
+            id: id,
+            agent: String::from(""),
+            login: None,
+            authenticated: false,
+            tx: tx,
+        }
+    }
+} // impl Worker
+
+struct WorkersList {
+    workers_list: Arc<RwLock<HashMap<usize, Worker>>>,
+    stratum_stats: Arc<RwLock<StratumStats>>,
+}
+
+impl WorkersList {
+    pub fn new(stratum_stats: Arc<RwLock<StratumStats>>) -> Self {
+        WorkersList {
+            workers_list: Arc::new(RwLock::new(HashMap::new())),
+            stratum_stats: stratum_stats,
+        }
+    }
+
+    pub fn add_worker(&self, tx: Tx) -> usize {
+        let mut stratum_stats = self.stratum_stats.write();
+        let worker_id = stratum_stats.worker_stats.len();
+        let worker = Worker::new(worker_id, tx);
+        let mut workers_list = self.workers_list.write();
+        workers_list.insert(worker_id, worker);
+
+        let mut worker_stats = WorkerStats::default();
+        worker_stats.is_connected = true;
+        worker_stats.id = worker_id.to_string();
+        worker_stats.pow_difficulty = stratum_stats.minimum_share_difficulty;
+        stratum_stats.worker_stats.push(worker_stats);
+        stratum_stats.num_workers = workers_list.len();
+        worker_id
+    }
+    pub fn remove_worker(&self, worker_id: usize) {
+        self.update_stats(worker_id, |ws| ws.is_connected = false);
+        let mut stratum_stats = self.stratum_stats.write();
+        let mut workers_list = self.workers_list.write();
+        workers_list
+            .remove(&worker_id)
+            .expect("Stratum: no such addr in map");
+
+        stratum_stats.num_workers = workers_list.len();
+    }
+
+    pub fn login(&self, worker_id: usize, login: String, agent: String) -> Result<(), RpcError> {
+        let mut wl = self.workers_list.write();
+        let mut worker = wl
+            .get_mut(&worker_id)
+            .ok_or_else(RpcError::internal_error)?;
+        worker.login = Some(login);
+        // XXX TODO Future - Validate password?
+        worker.agent = agent;
+        worker.authenticated = true;
+        Ok(())
+    }
+
+    pub fn get_worker(&self, worker_id: usize) -> Result<Worker, RpcError> {
+        self.workers_list
+            .read()
+            .get(&worker_id)
+            .ok_or_else(|| {
+                error!("Worker {} not found", worker_id);
+                RpcError::internal_error()
+            })
+            .map(|w| w.clone())
+    }
+
+    pub fn get_stats(&self, worker_id: usize) -> Result<WorkerStats, RpcError> {
+        self.stratum_stats
+            .read()
+            .worker_stats
+            .get(worker_id)
+            .ok_or_else(RpcError::internal_error)
+            .map(|ws| ws.clone())
+    }
+
+    pub fn last_seen(&self, worker_id: usize) {
+        //self.stratum_stats.write().worker_stats[worker_id].last_seen = SystemTime::now();
+        self.update_stats(worker_id, |ws| ws.last_seen = SystemTime::now());
+    }
+
+    pub fn update_stats(&self, worker_id: usize, f: impl FnOnce(&mut WorkerStats) -> ()) {
+        let mut stratum_stats = self.stratum_stats.write();
+        f(&mut stratum_stats.worker_stats[worker_id]);
+    }
+
+    pub fn send_to(&self, worker_id: usize, msg: String) {
+        let _ = self
+            .workers_list
+            .read()
+            .get(&worker_id)
+            .unwrap()
+            .tx
+            .unbounded_send(msg);
+    }
+
+    pub fn broadcast(&self, msg: String) {
+        for worker in self.workers_list.read().values() {
+            let _ = worker.tx.unbounded_send(msg.clone());
+        }
+    }
+
+    pub fn count(&self) -> usize {
+        self.workers_list.read().len()
+    }
+
+    pub fn update_edge_bits(&self, edge_bits: u16) {
+        {
+            let mut stratum_stats = self.stratum_stats.write();
+            stratum_stats.edge_bits = edge_bits;
+        }
+        self.update_network_hashrate();
+    }
+
+    pub fn update_block_height(&self, height: u64) {
+        let mut stratum_stats = self.stratum_stats.write();
+        stratum_stats.block_height = height;
+    }
+
+    pub fn update_network_difficulty(&self, difficulty: u64) {
+        let mut stratum_stats = self.stratum_stats.write();
+        stratum_stats.network_difficulty = difficulty;
+    }
+
+    pub fn update_network_hashrate(&self) {
+        let mut stratum_stats = self.stratum_stats.write();
+        stratum_stats.network_hashrate = 42.0
+            * (stratum_stats.network_difficulty as f64
+            / graph_weight(stratum_stats.block_height, stratum_stats.edge_bits as u8) as f64)
+            / 60.0;
+    }
+}
+
+// ----------------------------------------
+// Grin Stratum Server
+
+pub struct StratumServer {
+    id: String,
+    config: StratumServerConfig,
+    chain: Arc<grin_chain::Chain>,
+    pub tx_pool: ServerTxPool,
+    sync_state: Arc<SyncState>,
+    stratum_stats: Arc<RwLock<StratumStats>>,
+}
+
+impl StratumServer {
+    /// Creates a new Stratum Server.
+    pub fn new(
+        config: StratumServerConfig,
+        chain: Arc<grin_chain::Chain>,
+        tx_pool: ServerTxPool,
+        stratum_stats: Arc<RwLock<StratumStats>>,
+    ) -> StratumServer {
+        StratumServer {
+            id: String::from("0"),
+            config,
+            chain,
+            tx_pool,
+            sync_state: Arc::new(SyncState::new()),
+            stratum_stats: stratum_stats,
+        }
+    }
+
+    /// "main()" - Starts the stratum-server.  Creates a thread to Listens for
+    /// a connection, then enters a loop, building a new block on top of the
+    /// existing chain anytime required and sending that to the connected
+    /// stratum miner, proxy, or pool, and accepts full solutions to
+    /// be submitted.
+    pub fn run_loop(&mut self, proof_size: usize, sync_state: Arc<SyncState>, stop_state: Arc<StopState>) {
+        info!(
+			"(Server ID: {}) Starting stratum server with proof_size = {}",
+			self.id, proof_size
+		);
+
+        self.sync_state = sync_state;
+
+        let listen_addr = self
+            .config
+            .stratum_server_addr
+            .clone()
+            .unwrap()
+            .parse()
+            .expect("Stratum: Incorrect address ");
+
+        let handler = Arc::new(Handler::from_stratum(&self));
+        let h = handler.clone();
+
+        let s_state = stop_state.clone();
+
+        let _listener_th = thread::spawn(move || {
+            accept_connections(listen_addr, h, s_state);
+        });
+
+        // We have started
+        {
+            let mut stratum_stats = self.stratum_stats.write();
+            stratum_stats.is_running = true;
+            stratum_stats.edge_bits = (global::min_edge_bits() + 1) as u16;
+            stratum_stats.minimum_share_difficulty = self.config.minimum_share_difficulty;
+        }
+
+        warn!(
+			"Stratum server started on {}",
+			self.config.stratum_server_addr.clone().unwrap()
+		);
+
+        // Initial Loop. Waiting node complete syncing
+        while self.sync_state.is_syncing() {
+            thread::sleep(Duration::from_millis(50));
+        }
+
+        handler.run(&self.config, &self.tx_pool, stop_state.clone());
+    } // fn run_loop()
+} // StratumServer
+
+// Utility function to parse a JSON RPC parameter object, returning a proper
+// error if things go wrong.
+fn parse_params<T>(params: Option<Value>) -> Result<T, RpcError>
+    where
+            for<'de> T: serde::Deserialize<'de>,
+{
+    params
+        .and_then(|v| serde_json::from_value(v).ok())
+        .ok_or_else(RpcError::invalid_request)
+}