|
|
@@ -16,52 +16,250 @@
|
|
|
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
*/
|
|
|
|
|
|
-use std::{collections::HashMap, sync::Arc, time::Duration};
|
|
|
+use std::{collections::HashMap, io, str::FromStr, sync::Arc, time::Duration};
|
|
|
|
|
|
use darkfi::{
|
|
|
rpc::{
|
|
|
- jsonrpc::{ErrorCode, JsonError, JsonResponse, JsonResult, JsonSubscriber},
|
|
|
+ jsonrpc::{
|
|
|
+ ErrorCode::{InternalError, InvalidParams, ServerError},
|
|
|
+ JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber,
|
|
|
+ },
|
|
|
util::JsonValue,
|
|
|
},
|
|
|
- system::{timeout::timeout, StoppableTask},
|
|
|
+ system::{sleep, timeout::timeout, StoppableTask, StoppableTaskPtr},
|
|
|
Error, Result,
|
|
|
};
|
|
|
use log::{debug, error, info, warn};
|
|
|
-use rand::{rngs::OsRng, Rng};
|
|
|
+use monero::blockdata::transaction::{ExtraField, RawExtraField, SubField::MergeMining};
|
|
|
use smol::{channel, lock::RwLock};
|
|
|
+use url::Url;
|
|
|
use uuid::Uuid;
|
|
|
|
|
|
-use super::{error::RpcError, MiningProxy, Worker};
|
|
|
+use super::{error::RpcError, MiningProxy};
|
|
|
|
|
|
/// Algo string representing Monero's RandomX
|
|
|
pub const RANDOMX_ALGO: &str = "rx/0";
|
|
|
|
|
|
+/// A mining job instance
|
|
|
+#[derive(Clone)]
|
|
|
+struct MiningJob {
|
|
|
+ /// Current job ID for the worker
|
|
|
+ pub job_id: blake3::Hash,
|
|
|
+ /// Full block being mined
|
|
|
+ pub block: monero::Block,
|
|
|
+ /// Difficulty target,
|
|
|
+ pub target: String,
|
|
|
+ /// Block height
|
|
|
+ pub height: f64,
|
|
|
+ /// RandomX seed hash
|
|
|
+ pub seed_hash: String,
|
|
|
+}
|
|
|
+
|
|
|
+/// Single worker connected to the mining proxy
|
|
|
+pub struct Worker {
|
|
|
+ /// Wallet address
|
|
|
+ addr: monero::Address,
|
|
|
+ /// Miner useragent
|
|
|
+ _agent: String,
|
|
|
+ /// JSON-RPC notification subscriber, used to send new job notifications
|
|
|
+ job_sub: JsonSubscriber,
|
|
|
+ /// Background keepalive task reference
|
|
|
+ _ka_task: StoppableTaskPtr,
|
|
|
+ /// Keepalive sender channel, pinged from Stratum keepalived
|
|
|
+ ka_send: channel::Sender<()>,
|
|
|
+ /// Background mining job task reference
|
|
|
+ _job_task: StoppableTaskPtr,
|
|
|
+ /// Block submit trigger sender channel, pinged from Stratum submit
|
|
|
+ submit_send: channel::Sender<()>,
|
|
|
+ /// Current mining job
|
|
|
+ mining_job: MiningJob,
|
|
|
+}
|
|
|
+
|
|
|
+impl Worker {
|
|
|
+ async fn notify_job(&mut self, mining_job: MiningJob) -> Result<()> {
|
|
|
+ // Update the mining job
|
|
|
+ self.mining_job = mining_job.clone();
|
|
|
+
|
|
|
+ // Build notification params
|
|
|
+ let params: JsonValue = JsonValue::Object(HashMap::from([
|
|
|
+ ("blob".to_string(), hex::encode(mining_job.block.serialize_hashable()).into()),
|
|
|
+ ("job_id".to_string(), mining_job.job_id.to_string().into()),
|
|
|
+ ("target".to_string(), mining_job.target.into()),
|
|
|
+ ("height".to_string(), mining_job.height.into()),
|
|
|
+ ("seed_hash".to_string(), mining_job.seed_hash.into()),
|
|
|
+ ("algo".to_string(), RANDOMX_ALGO.to_string().into()),
|
|
|
+ ]));
|
|
|
+
|
|
|
+ info!(
|
|
|
+ target: "worker::notify_job",
|
|
|
+ "[STRATUM] Sending mining job notification to worker",
|
|
|
+ );
|
|
|
+ self.job_sub.notify(params).await;
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// Send a HTTP JSON-RPC request to the given monerod RPC endpoint
|
|
|
+async fn monerod_request(endpoint: &Url, req: JsonRequest) -> Result<JsonValue> {
|
|
|
+ let client = surf::Client::new();
|
|
|
+
|
|
|
+ let mut response = match client
|
|
|
+ .get(endpoint)
|
|
|
+ .header("Content-Type", "application/json")
|
|
|
+ .body(req.stringify().unwrap())
|
|
|
+ .send()
|
|
|
+ .await
|
|
|
+ {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::monerod_request",
|
|
|
+ "[STRATUM] Failed sending RPC request to monerod: {}", e,
|
|
|
+ );
|
|
|
+ return Err(io::Error::new(io::ErrorKind::Other, e).into())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ let response_bytes = match response.body_bytes().await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::monerod_request",
|
|
|
+ "[STRATUM] Failed reading monerod RPC response: {}", e,
|
|
|
+ );
|
|
|
+ return Err(io::Error::new(io::ErrorKind::Other, e).into())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ let response_string = match String::from_utf8(response_bytes) {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::monerod_request",
|
|
|
+ "[STRATUM] Failed parsing monerod RPC response: {}", e,
|
|
|
+ );
|
|
|
+ return Err(io::Error::new(io::ErrorKind::Other, e).into())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ let response_json: JsonValue = match response_string.parse() {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::monerod_request",
|
|
|
+ "[STRATUM] Failed parsing monerod RPC response JSON: {}", e,
|
|
|
+ );
|
|
|
+ return Err(io::Error::new(io::ErrorKind::Other, e).into())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ Ok(response_json)
|
|
|
+}
|
|
|
+
|
|
|
+/// Perform getblocktemplate from monerod and inject it with the
|
|
|
+/// necessary merge mining data.
|
|
|
+/// Returns data necessary to create a mining job
|
|
|
+async fn getblocktemplate(endpoint: &Url, wallet_address: &monero::Address) -> Result<MiningJob> {
|
|
|
+ // Create the Merge Mining Tag: (`depth`, `merkle_root`)
|
|
|
+ let mm_tag = MergeMining(Some(monero::VarInt(32)), monero::Hash([0_u8; 32]));
|
|
|
+
|
|
|
+ // Construct `tx_extra` from all the extra fields we have to
|
|
|
+ // add to the coinbase transaction in the block we're mining
|
|
|
+ let tx_extra: RawExtraField = ExtraField(vec![mm_tag]).into();
|
|
|
+
|
|
|
+ // Create the monerod JSON-RPC request. `reserve_size` is the space
|
|
|
+ // we need to create for the `tx_extra` field created above.
|
|
|
+ let req = JsonRequest::new(
|
|
|
+ "get_block_template",
|
|
|
+ HashMap::from([
|
|
|
+ ("wallet_address".to_string(), wallet_address.to_string().into()),
|
|
|
+ ("reserve_size".to_string(), (tx_extra.0.len() as f64).into()),
|
|
|
+ ])
|
|
|
+ .into(),
|
|
|
+ );
|
|
|
+
|
|
|
+ // Get block template from monerod
|
|
|
+ let rep = match monerod_request(endpoint, req).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::getblocktemplate",
|
|
|
+ "[STRATUM] Failed sending getblocktemplate to monerod: {}", e,
|
|
|
+ );
|
|
|
+ return Err(io::Error::new(io::ErrorKind::Other, e).into())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ // Now we have to modify the block template:
|
|
|
+ // * Update the coinbase tx with our tx_extra field
|
|
|
+ // * Update the `blockhashing_blob` in order to perform correct PoW
|
|
|
+
|
|
|
+ // Deserialize the block template
|
|
|
+ let mut block_template = monero::consensus::deserialize::<monero::Block>(
|
|
|
+ &hex::decode(rep["result"]["blocktemplate_blob"].get::<String>().unwrap()).unwrap(),
|
|
|
+ )
|
|
|
+ .unwrap();
|
|
|
+
|
|
|
+ // Modify the coinbase tx with our additional merge mining data
|
|
|
+ block_template.miner_tx.prefix.extra = tx_extra;
|
|
|
+
|
|
|
+ // TODO: Get the difficulty target
|
|
|
+ let target = "b88d0600".to_string();
|
|
|
+
|
|
|
+ // Get the remaining metadata
|
|
|
+ let height = *rep["result"]["height"].get::<f64>().unwrap();
|
|
|
+ let seed_hash = rep["result"]["seed_hash"].get::<String>().unwrap().to_string();
|
|
|
+
|
|
|
+ // Create a deterministic job id
|
|
|
+ let mut hasher = blake3::Hasher::new();
|
|
|
+ hasher.update(&wallet_address.as_bytes());
|
|
|
+ hasher.update(&height.to_le_bytes());
|
|
|
+ hasher.update(seed_hash.as_bytes());
|
|
|
+ let job_id = hasher.finalize();
|
|
|
+
|
|
|
+ // Return the necessary data
|
|
|
+ Ok(MiningJob { job_id, block: block_template, target, height, seed_hash })
|
|
|
+}
|
|
|
+
|
|
|
impl MiningProxy {
|
|
|
- /// Background task listening for keepalives from a worker, if timeout is reached
|
|
|
- /// the worker will be dropped.
|
|
|
+ /// Background task listening for keepalives from a worker.
|
|
|
+ /// If timeout is reached, the worker will be dropped.
|
|
|
async fn keepalive_task(
|
|
|
workers: Arc<RwLock<HashMap<Uuid, Worker>>>,
|
|
|
uuid: Uuid,
|
|
|
ka_recv: channel::Receiver<()>,
|
|
|
) -> Result<()> {
|
|
|
- debug!("Spawned keepalive_task for worker {}", uuid);
|
|
|
- const TIMEOUT: Duration = Duration::from_secs(60);
|
|
|
+ debug!(target: "stratum::keepalive_task", "Spawned keepalive_task for worker {}", uuid);
|
|
|
+ const TIMEOUT: Duration = Duration::from_secs(65);
|
|
|
|
|
|
loop {
|
|
|
let Ok(r) = timeout(TIMEOUT, ka_recv.recv()).await else {
|
|
|
// Timeout, remove worker
|
|
|
- warn!("keepalive_task {} worker timed out", uuid);
|
|
|
+ warn!(
|
|
|
+ target: "stratum::keepalive_task",
|
|
|
+ "keepalive_task for worker {} timed out", uuid,
|
|
|
+ );
|
|
|
workers.write().await.remove(&uuid);
|
|
|
break
|
|
|
};
|
|
|
|
|
|
match r {
|
|
|
Ok(()) => {
|
|
|
- debug!("keepalive_task {} got ping", uuid);
|
|
|
+ debug!(
|
|
|
+ target: "stratum::keepalive_task",
|
|
|
+ "keepalive_task for worker {} got ping", uuid,
|
|
|
+ );
|
|
|
continue
|
|
|
}
|
|
|
Err(e) => {
|
|
|
- error!("keepalive_task {} channel recv error: {}", uuid, e);
|
|
|
+ error!(
|
|
|
+ target: "stratum::keepalive_task",
|
|
|
+ "keepalive_task for worker {} channel recv error: {}", uuid, e,
|
|
|
+ );
|
|
|
+ warn!(
|
|
|
+ target: "stratum::keepalive_task",
|
|
|
+ "Dropping worker {}", uuid,
|
|
|
+ );
|
|
|
workers.write().await.remove(&uuid);
|
|
|
break
|
|
|
}
|
|
|
@@ -71,84 +269,207 @@ impl MiningProxy {
|
|
|
Ok(())
|
|
|
}
|
|
|
|
|
|
- /// Stratum login method. `darkfi-mmproxy` will check that it is a valid worker
|
|
|
- /// login, and will also search for `RANDOMX_ALGO`.
|
|
|
- /// TODO: More proper error codes
|
|
|
- pub async fn stratum_login(&self, id: u16, params: JsonValue) -> JsonResult {
|
|
|
- let params = params.get::<Vec<JsonValue>>().unwrap();
|
|
|
- if params.len() != 1 || !params[0].is_object() {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ /// Background task used to notify a worker about new mining jobs.
|
|
|
+ /// `keepalive_task` iis able to remove workers from the worker pool,
|
|
|
+ /// so this task can easily exit if the worker is not found.
|
|
|
+ async fn job_task(
|
|
|
+ workers: Arc<RwLock<HashMap<Uuid, Worker>>>,
|
|
|
+ uuid: Uuid,
|
|
|
+ endpoint: Url,
|
|
|
+ submit_recv: channel::Receiver<()>,
|
|
|
+ ) -> Result<()> {
|
|
|
+ debug!(target: "stratum::job_task", "Spawned job_task for worker {}", uuid);
|
|
|
+ const POLL_INTERVAL: Duration = Duration::from_secs(60);
|
|
|
+
|
|
|
+ // Comfy wait for settling the Stratum login RPC call
|
|
|
+ sleep(2).await;
|
|
|
+
|
|
|
+ // In this loop, we'll be getting the block template for mining.
|
|
|
+ // At the beginning of the loop, we'll perform a getblocktemplate,
|
|
|
+ // and then inject our Merge Mining stuff, and forward it to the
|
|
|
+ // miner. After the notification, we'll either poll or wait for a
|
|
|
+ // trigger for a submitted block and reiterate the loop again in
|
|
|
+ // order to get the next mining job.
|
|
|
+ loop {
|
|
|
+ // Get the workers lock and the worker reference
|
|
|
+ let mut workers_ptr = workers.write().await;
|
|
|
+ let Some(worker) = workers_ptr.get_mut(&uuid) else {
|
|
|
+ info!(
|
|
|
+ target: "stratum::job_task",
|
|
|
+ "[STRATUM] Worker {} disconnected, exiting job_task", uuid,
|
|
|
+ );
|
|
|
+ break
|
|
|
+ };
|
|
|
+
|
|
|
+ // Get the next mining job
|
|
|
+ let mining_job = match getblocktemplate(&endpoint, &worker.addr).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::job_task",
|
|
|
+ "[STRATUM] Failed fetching getblocktemplate for worker {}: {}", uuid, e,
|
|
|
+ );
|
|
|
+ warn!(
|
|
|
+ target: "stratum::job_task",
|
|
|
+ "[STRATUM] Exiting job_task for worker {}", uuid,
|
|
|
+ );
|
|
|
+ break
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ // In case it's the same job, we'll wait and try again
|
|
|
+ if worker.mining_job.job_id == mining_job.job_id {
|
|
|
+ match timeout(POLL_INTERVAL, submit_recv.recv()).await {
|
|
|
+ Ok(_) => continue,
|
|
|
+ Err(_) => continue,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // Notify the worker about the new job
|
|
|
+ if let Err(e) = worker.notify_job(mining_job).await {
|
|
|
+ error!(
|
|
|
+ target: "stratum::job_task",
|
|
|
+ "[STRATUM] Failed sending job to worker {}: {}", uuid, e,
|
|
|
+ );
|
|
|
+ warn!(
|
|
|
+ target: "stratum::job_task",
|
|
|
+ "[STRATUM] Exiting job_task for worker {}", uuid,
|
|
|
+ );
|
|
|
+ break
|
|
|
+ }
|
|
|
+
|
|
|
+ drop(workers_ptr);
|
|
|
+
|
|
|
+ // Now poll or wait for a trigger for a new job.
|
|
|
+ match timeout(POLL_INTERVAL, submit_recv.recv()).await {
|
|
|
+ Ok(_) => continue,
|
|
|
+ Err(_) => continue,
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- let params = params[0].get::<HashMap<String, JsonValue>>().unwrap();
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+
|
|
|
+ /// Stratum login method
|
|
|
+ ///
|
|
|
+ /// `darkfi-mmproxy` will check that the worker provided a valid
|
|
|
+ /// address as the username, and will enforce `RANDOMX_ALGO` to
|
|
|
+ /// be supported. Upon success, we will fetch the block template
|
|
|
+ /// from monerod, inject it with our necessary merge mining info,
|
|
|
+ /// and forward it to the worker.
|
|
|
+ /// Additionally, we will spawn background tasks for new job and
|
|
|
+ /// keepalive notifications for this worker.
|
|
|
+ pub async fn stratum_login(&self, id: u16, params: JsonValue) -> JsonResult {
|
|
|
+ let Some(params) = params.get::<HashMap<String, JsonValue>>() else {
|
|
|
+ return JsonError::new(InvalidParams, None, id).into()
|
|
|
+ };
|
|
|
|
|
|
if !params.contains_key("login") ||
|
|
|
!params.contains_key("pass") ||
|
|
|
!params.contains_key("agent") ||
|
|
|
!params.contains_key("algo")
|
|
|
{
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, None, id).into()
|
|
|
}
|
|
|
|
|
|
let Some(login) = params["login"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"login\" object".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- let Some(pass) = params["pass"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ let Some(_pass) = params["pass"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"pass\" object".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- let Some(_agent) = params["agent"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ let Some(agent) = params["agent"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"agent\" object".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
let Some(algos) = params["algo"].get::<Vec<JsonValue>>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"algo\" object".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- // We'll only support rx/0 algo.
|
|
|
- let mut found_xmr_algo = false;
|
|
|
- for algo in algos {
|
|
|
+ // We'll only support `RANDOMX_ALGO`
|
|
|
+ let mut found_randomx_algo = false;
|
|
|
+ for algo in algos.iter() {
|
|
|
if !algo.is_string() {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, Some("Algo is not a string".to_string()), id)
|
|
|
+ .into()
|
|
|
}
|
|
|
|
|
|
if algo.get::<String>().unwrap() == RANDOMX_ALGO {
|
|
|
- found_xmr_algo = true;
|
|
|
+ found_randomx_algo = true;
|
|
|
break
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- if !found_xmr_algo {
|
|
|
+ if !found_randomx_algo {
|
|
|
return JsonError::new(
|
|
|
RpcError::UnsupportedMiningAlgo.into(),
|
|
|
- Some("Unsupported mining algo".to_string()),
|
|
|
+ Some("Unsupported mining algos".to_string()),
|
|
|
id,
|
|
|
)
|
|
|
.into()
|
|
|
}
|
|
|
|
|
|
- // Check valid login
|
|
|
- let Some(known_pass) = self.logins.get(login) else {
|
|
|
+ // Check valid login. We will parse the username as a Monero
|
|
|
+ // address, and validate that it corresponds to the network
|
|
|
+ // we're mining on.
|
|
|
+ let addr_bytes = match bs58::decode(login).into_vec() {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ return JsonError::new(
|
|
|
+ RpcError::InvalidWorkerLogin.into(),
|
|
|
+ Some(format!("Invalid Monero address login: {}", e)),
|
|
|
+ id,
|
|
|
+ )
|
|
|
+ .into()
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ let addr = match monero::Address::from_bytes(&addr_bytes) {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ return JsonError::new(
|
|
|
+ RpcError::InvalidWorkerLogin.into(),
|
|
|
+ Some(format!("Invalid Monero address login: {}", e)),
|
|
|
+ id,
|
|
|
+ )
|
|
|
+ .into()
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ if addr.network != self.monerod_network {
|
|
|
return JsonError::new(
|
|
|
RpcError::InvalidWorkerLogin.into(),
|
|
|
- Some("Unknown worker username".to_string()),
|
|
|
+ Some(format!(
|
|
|
+ "Invalid Monero address network, expected \"{:?}\"",
|
|
|
+ self.monerod_network
|
|
|
+ )),
|
|
|
id,
|
|
|
)
|
|
|
.into()
|
|
|
- };
|
|
|
+ }
|
|
|
|
|
|
- if known_pass != pass {
|
|
|
+ if addr.addr_type != monero::AddressType::Standard {
|
|
|
return JsonError::new(
|
|
|
RpcError::InvalidWorkerLogin.into(),
|
|
|
- Some("Invalid worker password".to_string()),
|
|
|
+ Some(format!(
|
|
|
+ "Invalid Monero address type, expected \"{}\"",
|
|
|
+ monero::AddressType::Standard
|
|
|
+ )),
|
|
|
id,
|
|
|
)
|
|
|
.into()
|
|
|
}
|
|
|
|
|
|
- // Login success, generate UUID
|
|
|
- let uuid = Uuid::new_v4();
|
|
|
+ // Now we have a valid address for mining.
|
|
|
+ // Create a new UUID for the worker, and initialize the `Worker`
|
|
|
+ // struct that will live throughout the miner's lifetime.
|
|
|
+ let worker_uuid = Uuid::new_v4();
|
|
|
|
|
|
// Create job subscriber
|
|
|
let job_sub = JsonSubscriber::new("job");
|
|
|
@@ -156,127 +477,269 @@ impl MiningProxy {
|
|
|
// Create keepalive channel
|
|
|
let (ka_send, ka_recv) = channel::unbounded();
|
|
|
|
|
|
+ // Create submit trigger channel
|
|
|
+ let (submit_send, submit_recv) = channel::unbounded();
|
|
|
+
|
|
|
// Create background keepalive task
|
|
|
let ka_task = StoppableTask::new();
|
|
|
|
|
|
+ // Create background job task
|
|
|
+ let job_task = StoppableTask::new();
|
|
|
+
|
|
|
+ // Get the current mining job for the worker
|
|
|
+ let mining_job = match getblocktemplate(&self.monerod_rpc, &addr).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::login",
|
|
|
+ "[STRATUM] Failed fetching block template for worker: {}", e,
|
|
|
+ );
|
|
|
+ return JsonError::new(InternalError, None, id).into()
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
// Create worker
|
|
|
- let worker = Worker::new(job_sub.clone(), ka_send, ka_task.clone());
|
|
|
+ let worker = Worker {
|
|
|
+ addr,
|
|
|
+ _agent: agent.clone(),
|
|
|
+ job_sub: job_sub.clone(),
|
|
|
+ _ka_task: ka_task.clone(),
|
|
|
+ ka_send,
|
|
|
+ _job_task: job_task.clone(),
|
|
|
+ submit_send,
|
|
|
+ mining_job: mining_job.clone(),
|
|
|
+ };
|
|
|
|
|
|
- // Insert into connections map
|
|
|
- self.workers.write().await.insert(uuid, worker);
|
|
|
+ // Insert the worker into connections map
|
|
|
+ self.workers.write().await.insert(worker_uuid, worker);
|
|
|
|
|
|
- // Spawn background task
|
|
|
+ // Spawn keepalive background task
|
|
|
ka_task.start(
|
|
|
- Self::keepalive_task(self.workers.clone(), uuid, ka_recv),
|
|
|
- move |_| async move { debug!("keepalive_task for {} exited", uuid) },
|
|
|
+ Self::keepalive_task(self.workers.clone(), worker_uuid, ka_recv),
|
|
|
+ move |_| async move { debug!("keepalive_task for {} exited", worker_uuid) },
|
|
|
Error::DetachedTaskStopped,
|
|
|
self.executor.clone(),
|
|
|
);
|
|
|
|
|
|
- info!("Added worker {} ({})", login, uuid);
|
|
|
-
|
|
|
- // Get block template for mining
|
|
|
- let gbt_params: JsonValue = HashMap::from([
|
|
|
- ("wallet_address".to_string(), self.monerod.wallet_address.clone().into()),
|
|
|
- ("reserve_size".to_string(), (0_f64).into()),
|
|
|
- ])
|
|
|
- .into();
|
|
|
-
|
|
|
- let block_template = match self.monero_get_block_template(OsRng.gen(), gbt_params).await {
|
|
|
- JsonResult::Response(resp) => {
|
|
|
- resp.result.get::<HashMap<String, JsonValue>>().unwrap().clone()
|
|
|
- }
|
|
|
- _ => {
|
|
|
- error!("[STRATUM] Failed getting block template from monerod");
|
|
|
- return JsonError::new(ErrorCode::InternalError, None, id).into()
|
|
|
- }
|
|
|
- };
|
|
|
+ // Spawn job notification background task
|
|
|
+ job_task.start(
|
|
|
+ Self::job_task(
|
|
|
+ self.workers.clone(),
|
|
|
+ worker_uuid,
|
|
|
+ self.monerod_rpc.clone(),
|
|
|
+ submit_recv,
|
|
|
+ ),
|
|
|
+ move |_| async move { debug!("job_task for {} exited", worker_uuid) },
|
|
|
+ Error::DetachedTaskStopped,
|
|
|
+ self.executor.clone(),
|
|
|
+ );
|
|
|
|
|
|
- // Send the job to the worker
|
|
|
- let mut workers = self.workers.write().await;
|
|
|
- let worker = workers.get_mut(&uuid).unwrap();
|
|
|
- let blob = block_template["blockhashing_blob"].get::<String>().unwrap();
|
|
|
- let target = block_template["wide_difficulty"].get::<String>().unwrap();
|
|
|
- if let Err(e) = worker.send_job(blob.clone(), target.clone()).await {
|
|
|
- error!("[STRATUM] Failed sending job to {}: {}", uuid, e);
|
|
|
- return JsonError::new(ErrorCode::InternalError, None, id).into()
|
|
|
- }
|
|
|
+ info!("[STRATUM] Added worker {}", worker_uuid);
|
|
|
+
|
|
|
+ // Finally, we return the job notification subscriber, along with the
|
|
|
+ // initial job response as noted in:
|
|
|
+ // https://github.com/xmrig/xmrig-proxy/blob/master/doc/STRATUM.md#example-success-reply
|
|
|
+ let blob = hex::encode(mining_job.block.serialize_hashable());
|
|
|
+ let response = JsonResponse::new(
|
|
|
+ HashMap::from([
|
|
|
+ ("status".to_string(), "OK".to_string().into()),
|
|
|
+ ("id".to_string(), worker_uuid.to_string().into()),
|
|
|
+ (
|
|
|
+ "extensions".to_string(),
|
|
|
+ vec!["algo".to_string().into(), "keepalive".to_string().into()].into(),
|
|
|
+ ),
|
|
|
+ (
|
|
|
+ "job".to_string(),
|
|
|
+ HashMap::from([
|
|
|
+ ("blob".to_string(), blob.into()),
|
|
|
+ ("job_id".to_string(), mining_job.job_id.to_string().into()),
|
|
|
+ ("target".to_string(), mining_job.target.to_string().into()),
|
|
|
+ ("height".to_string(), mining_job.height.into()),
|
|
|
+ ("seed_hash".to_string(), mining_job.seed_hash.to_string().into()),
|
|
|
+ ("algo".to_string(), RANDOMX_ALGO.to_string().into()),
|
|
|
+ ])
|
|
|
+ .into(),
|
|
|
+ ),
|
|
|
+ ])
|
|
|
+ .into(),
|
|
|
+ id,
|
|
|
+ );
|
|
|
|
|
|
- job_sub.into()
|
|
|
+ JsonResult::SubscriberWithReply(job_sub, response)
|
|
|
}
|
|
|
|
|
|
+ /// Stratum submit method
|
|
|
+ ///
|
|
|
+ /// The miner submits the request after a share was found.
|
|
|
pub async fn stratum_submit(&self, id: u16, params: JsonValue) -> JsonResult {
|
|
|
- let params = params.get::<Vec<JsonValue>>().unwrap();
|
|
|
- if params.len() != 1 || !params[0].is_object() {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
- }
|
|
|
-
|
|
|
- let params = params[0].get::<HashMap<String, JsonValue>>().unwrap();
|
|
|
+ let Some(params) = params.get::<HashMap<String, JsonValue>>() else {
|
|
|
+ return JsonError::new(InvalidParams, None, id).into()
|
|
|
+ };
|
|
|
|
|
|
if !params.contains_key("id") ||
|
|
|
!params.contains_key("job_id") ||
|
|
|
!params.contains_key("nonce") ||
|
|
|
!params.contains_key("result")
|
|
|
{
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, None, id).into()
|
|
|
}
|
|
|
|
|
|
- let Some(_uuid) = params["id"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ // Validate all the parameters
|
|
|
+ let Some(worker_uuid) = params["id"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
+ };
|
|
|
+
|
|
|
+ let Ok(worker_uuid) = Uuid::try_from(worker_uuid.as_str()) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
+ };
|
|
|
+
|
|
|
+ let Some(job_id) = params["job_id"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"job_id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
+ };
|
|
|
+
|
|
|
+ let Ok(job_id) = blake3::Hash::from_str(job_id) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"job_id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- let Some(_job_id) = params["job_id"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ let Some(nonce) = params["nonce"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"nonce\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- let Some(_nonce) = params["nonce"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ let Ok(nonce) = u32::from_str_radix(nonce, 16) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"nonce\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
let Some(_result) = params["result"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"result\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- todo!()
|
|
|
- }
|
|
|
+ // Get the worker reference and confirm this is submitted for the current job
|
|
|
+ let workers_ptr = self.workers.read().await;
|
|
|
+ let Some(worker) = workers_ptr.get(&worker_uuid) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Unknown worker UUID".to_string()), id).into()
|
|
|
+ };
|
|
|
|
|
|
- /// Non standard but widely supported protocol extension. Miner sends `keepalived`
|
|
|
- /// to prevent connection timeout.
|
|
|
- pub async fn stratum_keepalived(&self, id: u16, params: JsonValue) -> JsonResult {
|
|
|
- let params = params.get::<Vec<JsonValue>>().unwrap();
|
|
|
- if params.len() != 1 || !params[0].is_object() {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ if worker.mining_job.job_id != job_id {
|
|
|
+ return JsonError::new(InvalidParams, Some("Job ID mismatch".to_string()), id).into()
|
|
|
}
|
|
|
|
|
|
- let params = params[0].get::<HashMap<String, JsonValue>>().unwrap();
|
|
|
+ // Get the block template from the worker reference and update the nonce
|
|
|
+ let mut block_template = worker.mining_job.block.clone();
|
|
|
+ block_template.header.nonce = nonce;
|
|
|
+
|
|
|
+ // Submit the block to monerod
|
|
|
+ let block = monero::consensus::serialize_hex(&block_template);
|
|
|
+ let params: JsonValue = vec![block.into()].into();
|
|
|
+ let req = JsonRequest::new("submit_block", params);
|
|
|
+
|
|
|
+ let resp = match monerod_request(&self.monerod_rpc, req).await {
|
|
|
+ Ok(v) => v,
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::submit",
|
|
|
+ "[STRATUM] Failed submitting block to monerod: {}", e,
|
|
|
+ );
|
|
|
+ return JsonError::new(
|
|
|
+ InternalError,
|
|
|
+ Some("Failed submitting block".to_string()),
|
|
|
+ id,
|
|
|
+ )
|
|
|
+ .into()
|
|
|
+ }
|
|
|
+ };
|
|
|
|
|
|
- if !params.contains_key("id") {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ // Ping the job_task to reiterate.
|
|
|
+ // We don't release the lock after this, so that we can hopefully first
|
|
|
+ // return the result of the `submit` call, and then unlock job_task for
|
|
|
+ // the new notification.
|
|
|
+ let _ = worker.submit_send.send(()).await;
|
|
|
+
|
|
|
+ match JsonResult::try_from_value(&resp) {
|
|
|
+ Ok(JsonResult::Response(r)) => {
|
|
|
+ info!(
|
|
|
+ target: "stratum::submit",
|
|
|
+ "[STRATUM] Sucessfully submitted block to monerod: {:?}", r,
|
|
|
+ );
|
|
|
+
|
|
|
+ let result = HashMap::from([("status".to_string(), "OK".to_string().into())]);
|
|
|
+ JsonResponse::new(result.into(), id).into()
|
|
|
+ }
|
|
|
+ Ok(JsonResult::Error(e)) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::submit",
|
|
|
+ "[STRATUM] Failed submitting block to monerod: {:?}", e,
|
|
|
+ );
|
|
|
+ JsonError::new(ServerError(e.error.code), Some(e.error.message), id).into()
|
|
|
+ }
|
|
|
+ Ok(x) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::submit",
|
|
|
+ "[STRATUM] Unexpected RPC reply from monerod: {:?}", x,
|
|
|
+ );
|
|
|
+ JsonError::new(InternalError, Some("Failed submitting block".to_string()), id)
|
|
|
+ .into()
|
|
|
+ }
|
|
|
+ Err(e) => {
|
|
|
+ error!(
|
|
|
+ target: "stratum::submit",
|
|
|
+ "[STRATUM] Unexpected RPC reply from monerod: {}", e,
|
|
|
+ );
|
|
|
+ JsonError::new(InternalError, Some("Failed submitting block".to_string()), id)
|
|
|
+ .into()
|
|
|
+ }
|
|
|
}
|
|
|
+ }
|
|
|
|
|
|
- let Some(uuid) = params["id"].get::<String>() else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ /// Nonstandard, but widely supported protocol extension.
|
|
|
+ /// The miner sends `keepalived` to prevent connection timeout.
|
|
|
+ /// `darkfi-mmproxy` makes having keepalived mandatory.
|
|
|
+ pub async fn stratum_keepalived(&self, id: u16, params: JsonValue) -> JsonResult {
|
|
|
+ let Some(params) = params.get::<HashMap<String, JsonValue>>() else {
|
|
|
+ return JsonError::new(InvalidParams, None, id).into()
|
|
|
};
|
|
|
|
|
|
- let Ok(uuid) = Uuid::try_from(uuid.as_str()) else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ if !params.contains_key("id") {
|
|
|
+ return JsonError::new(InvalidParams, Some("Missing \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
- // Ping the keepalive task
|
|
|
- let workers = self.workers.read().await;
|
|
|
- let Some(worker) = workers.get(&uuid) else {
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ let Some(worker_uuid) = params["id"].get::<String>() else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
+ };
|
|
|
+
|
|
|
+ let Ok(worker_uuid) = Uuid::try_from(worker_uuid.as_str()) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
};
|
|
|
|
|
|
+ // Get the worker reference
|
|
|
+ let workers_ptr = self.workers.read().await;
|
|
|
+ let Some(worker) = workers_ptr.get(&worker_uuid) else {
|
|
|
+ return JsonError::new(InvalidParams, Some("Invalid \"id\" field".to_string()), id)
|
|
|
+ .into()
|
|
|
+ };
|
|
|
+
|
|
|
+ // Ping the keepalive task
|
|
|
if let Err(e) = worker.ka_send.send(()).await {
|
|
|
- error!("stratum_keepalived: keepalive task ping error: {}", e);
|
|
|
- return JsonError::new(ErrorCode::InvalidParams, None, id).into()
|
|
|
+ error!(
|
|
|
+ target: "stratum::keepalived",
|
|
|
+ "[STRATUM] Keepalive task ping error for {}: {}", worker_uuid, e,
|
|
|
+ );
|
|
|
+ return JsonError::new(InternalError, None, id).into()
|
|
|
}
|
|
|
|
|
|
JsonResponse::new(
|
|
|
JsonValue::Object(HashMap::from([(
|
|
|
"status".to_string(),
|
|
|
- JsonValue::String("KEEPALIVED".to_string()),
|
|
|
+ "KEEPALIVED".to_string().into(),
|
|
|
)])),
|
|
|
id,
|
|
|
)
|
parazyd